Merge pull request #35 from Lamparter/connections

Initial implementation of virtual event stream handler for Connections

Author: Arlodotexe

src/IReadOnlyConnection.cs

@@ -13,10 +13,10 @@ public interface IReadOnlyConnection
     /// <summary>
     /// Gets the value of the connection.
     /// </summary>
-    string Value { get; }
+    Task<string> GetValueAsync(CancellationToken cancellationToken = default);
 
     /// <summary>
-    /// Raised when <see cref="Value"/> is updated.
+    /// Raised when the value is updated.
     /// </summary>
     event EventHandler<string>? ValueUpdated;
 }

src/IReadOnlyConnectionsCollection.cs

@@ -1,3 +1,5 @@
+using System.Collections.Generic;
+
 namespace WindowsAppCommunity.Sdk;
 
 /// <summary>
@@ -6,17 +8,17 @@ namespace WindowsAppCommunity.Sdk;
 public interface IReadOnlyConnectionsCollection
 {
     /// <summary>
-    /// The connections associated with this entity.
+    /// Gets the connections associated with this entity.
     /// </summary>
-    IReadOnlyConnection[] Connections { get; }
+    IAsyncEnumerable<IReadOnlyConnection> GetConnectionsAsync(CancellationToken cancellationToken = default);
 
     /// <summary>
-    /// Raised when items are added to the <see cref="Connections"/> collection.
+    /// Raised when items are added to the collection.
     /// </summary>
     event EventHandler<IReadOnlyConnection[]>? ConnectionsAdded;
 
     /// <summary>
-    /// Raised when items are removed from the <see cref="Connections"/> collection.
+    /// Raised when items are removed from the collection.
     /// </summary>
     event EventHandler<IReadOnlyConnection[]>? ConnectionsRemoved;
 }

src/Models/IConnections.cs

@@ -1,5 +1,4 @@
-using System.Collections.Generic;
-using Ipfs;
+using Ipfs;
 
 namespace WindowsAppCommunity.Sdk.Models;
 
@@ -11,5 +10,21 @@ public interface IConnections
     /// <summary>
     /// Represents data about external application connections.
     /// </summary>
-    Dictionary<string, DagCid> Connections { get; set; }
+    Connection[] Connections { get; set; }
+}
+
+/// <summary>
+/// Represents a published collection of key-value connection pairs.
+/// </summary>
+public record Connection
+{
+    /// <summary>
+    /// A unique identifier for this connection.
+    /// </summary>
+    public required string Id { get; init; }
+
+    /// <summary>
+    /// A <see cref="DagCid"/> of the connection value.
+    /// </summary>
+    public required DagCid Value { get; set; }
 }

src/Models/Project.cs

@@ -67,7 +67,7 @@ public record Project : IEntity, IUserRoleCollection, IAccentColor, IProjectColl
     /// <summary>
     /// Holds information about project assets that have been published for consumption by an end user, such as a Microsoft Store app, a package on nuget.org, a git repo, etc.
     /// </summary>
-    public Dictionary<string, DagCid> Connections { get; set; } = new();
+    public Connection[] Connections { get; set; } = [];
 
     /// <summary>
     /// A flag that indicates whether the profile has requested to be forgotten.

src/Models/Publisher.cs

@@ -62,7 +62,7 @@ public record Publisher : IEntity, ILinkCollection, IProjectCollection, IUserRol
     /// <summary>
     /// Holds information about publisher assets that have been published for consumption by an end user, such as a Microsoft Store app, a package on nuget.org, a git repo, etc.
     /// </summary>
-    public Dictionary<string, DagCid> Connections { get; set; } = [];
+    public Connection[] Connections { get; set; } = [];
 
     /// <summary>
     /// A flag that indicates whether the profile has requested to be forgotten.

src/Models/User.cs

@@ -27,7 +27,7 @@ public record User : IEntity, IConnections, ILinkCollection, IProjectRoleCollect
     /// <summary>
     /// Represents application connections added by the user.
     /// </summary>
-    public Dictionary<string, DagCid> Connections { get; set; } = [];
+    public Connection[] Connections { get; set; } = [];
 
     /// <summary>
     /// Represents links to external profiles or resources added by the user.

src/Nomad/ModifiableAccentColor.cs

@@ -24,6 +24,24 @@ public class ModifiableAccentColor : NomadKuboEventStreamHandler<ValueUpdateEven
     /// <inheritdoc />
     public event EventHandler<string?>? AccentColorUpdated;
 
+    /// <inheritdoc />
+    public async Task UpdateAccentColorAsync(string? accentColor, CancellationToken cancellationToken)
+    {
+        DagCid? valueCid = null;
+        if (accentColor is not null)
+        {
+            Cid cid = await Client.Dag.PutAsync(accentColor, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
+            valueCid = (DagCid)cid;
+        }
+
+        var updateEvent = new ValueUpdateEvent(null, valueCid, accentColor is null);
+
+        var appendedEntry = await AppendNewEntryAsync(targetId: Id, eventId: nameof(UpdateAccentColorAsync), updateEvent, DateTime.UtcNow, cancellationToken);
+        await ApplyEntryUpdateAsync(appendedEntry, updateEvent, accentColor, cancellationToken);
+
+        EventStreamPosition = appendedEntry;
+    }
+
     /// <inheritdoc />
     public override async Task ApplyEntryUpdateAsync(EventStreamEntry<DagCid> eventStreamEntry, ValueUpdateEvent updateEvent, CancellationToken cancellationToken)
     {
@@ -70,22 +88,4 @@ public override Task ResetEventStreamPositionAsync(CancellationToken cancellatio
 
         return Task.CompletedTask;
     }
-
-    /// <inheritdoc />
-    public async Task UpdateAccentColorAsync(string? accentColor, CancellationToken cancellationToken)
-    {
-        DagCid? valueCid = null;
-        if (accentColor is not null)
-        {
-            Cid cid = await Client.Dag.PutAsync(accentColor, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
-            valueCid = (DagCid)cid;
-        }
-
-        var updateEvent = new ValueUpdateEvent(null, valueCid, accentColor is null);
-
-        var appendedEntry = await AppendNewEntryAsync(targetId: Id, eventId: nameof(UpdateAccentColorAsync), updateEvent, DateTime.UtcNow, cancellationToken);
-        await ApplyEntryUpdateAsync(appendedEntry, updateEvent, accentColor, cancellationToken);
-
-        EventStreamPosition = appendedEntry;
-    }
 }

src/Nomad/ModifiableConnection.cs

@@ -0,0 +1,85 @@
+using CommunityToolkit.Diagnostics;
+using Ipfs;
+using OwlCore.ComponentModel;
+using OwlCore.Nomad;
+using OwlCore.Nomad.Kubo;
+using OwlCore.Nomad.Kubo.Events;
+
+namespace WindowsAppCommunity.Sdk.Nomad
+{
+    /// <summary>
+    /// Represents a modifiable connection.
+    /// </summary>
+    public class ModifiableConnection : NomadKuboEventStreamHandler<ValueUpdateEvent>, IModifiableConnection, IDelegable<ReadOnlyConnection>
+    {
+        /// <inheritdoc/>
+        public required ReadOnlyConnection Inner { get; init; }
+
+        /// <inheritdoc/>
+        public string Id => Inner.Id;
+
+        /// <inheritdoc/>
+        public event EventHandler<string>? ValueUpdated;
+
+        /// <inheritdoc/>
+        public async Task UpdateValueAsync(string newValue, CancellationToken cancellationToken = default)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            var dagCid = (DagCid)await Client.Dag.PutAsync(newValue, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
+            var updateEvent = new ValueUpdateEvent(null, dagCid, false);
+
+            var appendedEntry = await AppendNewEntryAsync(targetId: Id, eventId: nameof(UpdateValueAsync), updateEvent, DateTime.UtcNow, cancellationToken);
+            await ApplyEntryUpdateAsync(appendedEntry, updateEvent, newValue, cancellationToken);
+
+            EventStreamPosition = appendedEntry;
+        }
+
+        /// <inheritdoc/>
+        public Task<string> GetValueAsync(CancellationToken cancellationToken = default) => Inner.GetValueAsync(cancellationToken);
+
+        /// <inheritdoc />
+        public override async Task ApplyEntryUpdateAsync(EventStreamEntry<DagCid> eventStreamEntry, ValueUpdateEvent updateEvent, CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (eventStreamEntry.TargetId != Id)
+                return;
+
+            Guard.IsNotNull(updateEvent.Value);
+            var updatedValue = await Client.Dag.GetAsync<string>(updateEvent.Value, cancel: cancellationToken);
+
+            Guard.IsNotNull(updatedValue);
+            await ApplyEntryUpdateAsync(eventStreamEntry, updateEvent, updatedValue, cancellationToken);
+        }
+
+        /// <summary>
+        /// Applies an event stream update event and raises the relevant events.
+        /// </summary>
+        /// <param name="eventStreamEntry">The event stream entry to apply.</param>
+        /// <param name="updateEvent">The update event to apply.</param>
+        /// <param name="newValue">The resolved new value data for this event.</param>
+        /// <param name="cancellationToken">A token that can be used to cancel the ongoing operation.</param>
+        public Task ApplyEntryUpdateAsync(EventStreamEntry<DagCid> eventStreamEntry, ValueUpdateEvent updateEvent, string newValue, CancellationToken cancellationToken)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            if (eventStreamEntry.EventId is not nameof(UpdateValueAsync))
+                return Task.CompletedTask;
+
+            Guard.IsNotNull(updateEvent.Value);
+
+            Inner.Inner.Value = updateEvent.Value;
+            ValueUpdated?.Invoke(this, newValue);
+
+            return Task.CompletedTask;
+        }
+
+        /// <inheritdoc />
+        public override Task ResetEventStreamPositionAsync(CancellationToken cancellationToken)
+        {
+            EventStreamPosition = null;
+            return Task.CompletedTask;
+        }
+    }
+}

src/Nomad/ModifiableConnectionCollection.cs

@@ -0,0 +1,147 @@
+using System.Collections.Generic;
+using System.Linq;
+using CommunityToolkit.Diagnostics;
+using Ipfs;
+using OwlCore.ComponentModel;
+using OwlCore.Kubo;
+using OwlCore.Nomad;
+using OwlCore.Nomad.Kubo;
+using OwlCore.Nomad.Kubo.Events;
+using WindowsAppCommunity.Sdk.Models;
+
+namespace WindowsAppCommunity.Sdk.Nomad;
+
+/// <summary>
+/// Represents a modifiable connection collection.
+/// </summary>
+public class ModifiableConnectionCollection : NomadKuboEventStreamHandler<ValueUpdateEvent>, IModifiableConnectionsCollection, IDelegable<ReadOnlyConnectionCollection>
+{
+    /// <summary>
+    /// Raised when items are added to the collection.
+    /// </summary>
+    public event EventHandler<IReadOnlyConnection[]>? ConnectionsAdded;
+
+    /// <summary>
+    /// Raised when items are removed from the collection.
+    /// </summary>
+    public event EventHandler<IReadOnlyConnection[]>? ConnectionsRemoved;
+
+    /// <summary>
+    /// The unique identifier for this collection.
+    /// </summary>
+    public required string Id { get; init; }
+
+    /// <summary>
+    /// The inner read-only connections collection.
+    /// </summary>
+    public required ReadOnlyConnectionCollection Inner { get; init; }
+
+    /// <inheritdoc />
+    public IAsyncEnumerable<IReadOnlyConnection> GetConnectionsAsync(CancellationToken cancellationToken = default) => Inner.GetConnectionsAsync(cancellationToken);
+
+    /// <summary>
+    /// Adds a connection to this entity.
+    /// </summary>
+    public async Task AddConnectionAsync(IReadOnlyConnection connection, CancellationToken cancellationToken)
+    {
+        var value = await connection.GetValueAsync(cancellationToken);
+
+        var keyCid = (DagCid)await Client.Dag.PutAsync(connection.Id, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
+        var valueCid = (DagCid)await Client.Dag.PutAsync(value, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
+
+        var updateEvent = new ValueUpdateEvent(keyCid, valueCid, Unset:false);
+
+        var appendedEntry = await AppendNewEntryAsync(targetId: Id, eventId: nameof(AddConnectionAsync), updateEvent, DateTime.UtcNow, cancellationToken);
+        await ApplyEntryUpdateAsync(appendedEntry, updateEvent, new Connection { Id = connection.Id, Value = valueCid }, connection, cancellationToken);
+
+        EventStreamPosition = appendedEntry;
+    }
+
+    /// <summary>
+    /// Removes a connection from this entity.
+    /// </summary>
+    public async Task RemoveConnectionAsync(IReadOnlyConnection connection, CancellationToken cancellationToken)
+    {
+        var existing = Inner.Inner.Connections.FirstOrDefault(x => x.Id == connection.Id);
+        if (existing == null)
+            throw new ArgumentException("Connection not found in the collection.", nameof(connection));
+
+        var value = await connection.GetValueAsync(cancellationToken);
+
+        var keyCid = (DagCid)await Client.Dag.PutAsync(connection.Id, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
+        var valueCid = (DagCid)await Client.Dag.PutAsync(value, pin: KuboOptions.ShouldPin, cancel: cancellationToken);
+
+        var updateEvent = new ValueUpdateEvent(keyCid, valueCid, Unset: true);
+
+        var appendedEntry = await AppendNewEntryAsync(targetId: Id, eventId: nameof(RemoveConnectionAsync), updateEvent, DateTime.UtcNow, cancellationToken);
+        await ApplyEntryUpdateAsync(appendedEntry, updateEvent, new Connection { Id = connection.Id, Value = valueCid }, connection, cancellationToken);
+
+        EventStreamPosition = appendedEntry;
+    }
+
+    /// <summary>
+    /// Applies an event stream update event and raises the relevant events.
+    /// </summary>
+    /// <remarks>
+    /// This method will call <see cref="ReadOnlyConnectionCollection.GetAsync(string, CancellationToken)"/> and create a new instance to pass to the event handlers.
+    /// <para/>
+    /// If already have a resolved instance of <see cref="Connection"/>, you should call <see cref="ApplyEntryUpdateAsync(EventStreamEntry{DagCid}, ValueUpdateEvent, Connection, IReadOnlyConnection?, CancellationToken)"/> instead.
+    /// </remarks>
+    /// <param name="eventStreamEntry">The event stream entry to apply.</param>
+    /// <param name="updateEvent">The update event to apply.</param>
+    /// <param name="cancellationToken">A token that can be used to cancel the ongoing operation.</param>
+    public override async Task ApplyEntryUpdateAsync(EventStreamEntry<DagCid> eventStreamEntry, ValueUpdateEvent updateEvent, CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        if (eventStreamEntry.TargetId != Id)
+            return;
+
+        Guard.IsNotNull(updateEvent.Value);
+        var (connection, _) = await Client.ResolveDagCidAsync<Connection>(updateEvent.Value.Value, nocache: !KuboOptions.UseCache, cancellationToken);
+
+        Guard.IsNotNull(connection);
+        await ApplyEntryUpdateAsync(eventStreamEntry, updateEvent, connection, null, cancellationToken);
+    }
+
+    /// <summary>
+    /// Applies an event stream update event and raises the relevant events.
+    /// </summary>
+    /// <param name="eventStreamEntry">The event stream entry to apply.</param>
+    /// <param name="updateEvent">The update event to apply.</param>
+    /// <param name="connection">The resolved connection data for this event.</param>
+    /// <param name="connectionInstance">The existing instance of <see cref="IReadOnlyConnection"/> to emit on raised collection add/remove events.</param>
+    /// <param name="cancellationToken">A token that can be used to cancel the ongoing operation.</param>
+    public async Task ApplyEntryUpdateAsync(EventStreamEntry<DagCid> eventStreamEntry, ValueUpdateEvent updateEvent, Connection connection, IReadOnlyConnection? connectionInstance, CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        switch (eventStreamEntry.EventId)
+        {
+            case nameof(AddConnectionAsync):
+                {
+                    Inner.Inner.Connections = [.. Inner.Inner.Connections, connection];
+                    var connectionFile = connectionInstance ??= await Inner.GetAsync(connection.Id, cancellationToken);
+                    ConnectionsAdded?.Invoke(this, [connectionFile]);
+                    break;
+                }
+            case nameof(RemoveConnectionAsync):
+                {
+                    Inner.Inner.Connections = [.. Inner.Inner.Connections.Except([connection])];
+                    var connectionFile = connectionInstance ??= await Inner.GetAsync(connection.Id, cancellationToken);
+                    ConnectionsRemoved?.Invoke(this, [connectionFile]);
+                    break;
+                }
+        }
+    }
+
+    /// <inheritdoc/>
+    public override Task ResetEventStreamPositionAsync(CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        // TODO: Reset inner virtual event stream handlers
+        // Connections, Links, Images
+        throw new NotImplementedException();
+    }
+}