Implement GraphQL Subscriptions Support

csharp/Platform.Data.Doublets.Gql.Schema/LinksMutation.cs

@@ -9,7 +9,7 @@ namespace Platform.Data.Doublets.Gql.Schema
 {
     public class LinksMutation : ObjectGraphType<object>
     {
-        public LinksMutation(ILinks<ulong> links)
+        public LinksMutation(ILinks<ulong> links, LinksSubscription subscription)
         {
             Name = "mutation_root";
             Field<LinksMutationResponseType>("delete_links", arguments: new QueryArguments(new QueryArgument<NonNullGraphType<LinksBooleanExpressionInputType>> { Name = "where" }), resolve: context =>
@@ -19,6 +19,7 @@ public LinksMutation(ILinks<ulong> links)
                 foreach (var linkToDelete in response.returning)
                 {
                     links.Delete((ulong)linkToDelete.id);
+                    subscription.OnLinkDeleted((ulong)linkToDelete.id);
                 }
                 return response;
             });
@@ -28,12 +29,19 @@ public LinksMutation(ILinks<ulong> links)
                 var response = new LinksMutationResponse { returning = new List<Links>() };
                 foreach (var link in context.GetArgument<List<LinksInsert>>("objects"))
                 {
-                    response.returning.Add(InsertLink(links, link));
+                    var insertedLink = InsertLink(links, link);
+                    response.returning.Add(insertedLink);
+                    subscription.OnLinkCreated(insertedLink);
                 }
                 response.affected_rows = response.returning.Count;
                 return response;
             });
-            Field<LinksType>("insert_links_one", arguments: new QueryArguments(new QueryArgument<NonNullGraphType<LinksInsertInputType>> { Name = "object" }, new QueryArgument<LinksOnConflictInputType> { Name = "on_conflict" }), resolve: context => InsertLink(links, context.GetArgument<LinksInsert>("object")));
+            Field<LinksType>("insert_links_one", arguments: new QueryArguments(new QueryArgument<NonNullGraphType<LinksInsertInputType>> { Name = "object" }, new QueryArgument<LinksOnConflictInputType> { Name = "on_conflict" }), resolve: context => 
+            {
+                var insertedLink = InsertLink(links, context.GetArgument<LinksInsert>("object"));
+                subscription.OnLinkCreated(insertedLink);
+                return insertedLink;
+            });
             Field<LinksMutationResponseType>("update_links", arguments: new QueryArguments(new QueryArgument<LinksIncInputType> { Name = "_inc" }, new QueryArgument<LinksSetInputType> { Name = "_set" }, new QueryArgument<NonNullGraphType<LinksBooleanExpressionInputType>> { Name = "where" }), resolve: context =>
             {
                 var set = context.GetArgument<LinksSet>("_set");
@@ -50,7 +58,9 @@ public LinksMutation(ILinks<ulong> links)
                     {
                         updatedLink = links.Update((ulong)link.id, (ulong)set.from_id.Value, (ulong)set.to_id.Value);
                     }
-                    response.returning.Add(new Links(links.GetLink(updatedLink)));
+                    var updatedLinkModel = new Links(links.GetLink(updatedLink));
+                    response.returning.Add(updatedLinkModel);
+                    subscription.OnLinkUpdated(updatedLinkModel);
                 }
                 response.affected_rows = response.returning.Count;
                 return response;

csharp/Platform.Data.Doublets.Gql.Schema/LinksSchema.cs

@@ -6,9 +6,10 @@ public class LinksSchema : GraphQL.Types.Schema
     {
         public LinksSchema(ILinks<ulong> links, IServiceProvider provider) : base(provider)
         {
+            var subscription = new LinksSubscription(links);
             Query = new LinksQuery(links);
-            Mutation = new LinksMutation(links);
-            Subscription = new LinksSubscription(links);
+            Mutation = new LinksMutation(links, subscription);
+            Subscription = subscription;
         }
     }
 }

csharp/Platform.Data.Doublets.Gql.Schema/LinksSubscription.cs

@@ -1,16 +1,113 @@
-using GraphQL.Types;
+using GraphQL;
+using GraphQL.Types;
 using Platform.Data.Doublets.Gql.Schema.Types;
+using Platform.Data.Doublets.Gql.Schema.Types.Input;
+using System;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
 
 namespace Platform.Data.Doublets.Gql.Schema
 {
     public class LinksSubscription : ObjectGraphType
     {
+        private readonly ISubject<Links> _linkCreated = new Subject<Links>();
+        private readonly ISubject<Links> _linkUpdated = new Subject<Links>();
+        private readonly ISubject<ulong> _linkDeleted = new Subject<ulong>();
+
         public LinksSubscription(ILinks<ulong> links)
         {
             Name = "subscription_root";
-            Field<NonNullGraphType<ListGraphType<NonNullGraphType<LinksType>>>>("links", arguments: LinksQuery.Arguments, resolve: context => { return LinksQuery.GetLinks(context, links); });
-            Field<NonNullGraphType<LinksAggregateType>>("links_aggregate", arguments: LinksQuery.Arguments, resolve: context => "");
-            Field<LinksType>("links_by_pk", arguments: new QueryArguments(new QueryArgument<NonNullGraphType<LongGraphType>> { Name = "id" }));
+
+            // Subscribe to link creation events
+            Field<NonNullGraphType<LinksType>>(
+                "link_created",
+                arguments: new QueryArguments(
+                    new QueryArgument<LinksBooleanExpressionInputType> { Name = "where" }
+                ),
+                resolve: context =>
+                {
+                    var whereFilter = context.GetArgument<LinksBooleanExpression>("where");
+                    return _linkCreated.AsObservable()
+                        .Where(link => FilterLink(link, whereFilter));
+                });
+
+            // Subscribe to link update events
+            Field<NonNullGraphType<LinksType>>(
+                "link_updated",
+                arguments: new QueryArguments(
+                    new QueryArgument<LinksBooleanExpressionInputType> { Name = "where" }
+                ),
+                resolve: context =>
+                {
+                    var whereFilter = context.GetArgument<LinksBooleanExpression>("where");
+                    return _linkUpdated.AsObservable()
+                        .Where(link => FilterLink(link, whereFilter));
+                });
+
+            // Subscribe to link deletion events
+            Field<NonNullGraphType<LongGraphType>>(
+                "link_deleted",
+                arguments: new QueryArguments(
+                    new QueryArgument<LongGraphType> { Name = "id" }
+                ),
+                resolve: context =>
+                {
+                    var idFilter = context.GetArgument<long?>("id");
+                    return _linkDeleted.AsObservable()
+                        .Where(linkId => idFilter == null || (long)linkId == idFilter)
+                        .Select(linkId => (long)linkId);
+                });
+
+            // Subscribe to all link changes (created, updated, deleted)
+            Field<NonNullGraphType<LinksType>>(
+                "link_changed",
+                arguments: new QueryArguments(
+                    new QueryArgument<LinksBooleanExpressionInputType> { Name = "where" }
+                ),
+                resolve: context =>
+                {
+                    var whereFilter = context.GetArgument<LinksBooleanExpression>("where");
+                    var created = _linkCreated.AsObservable().Where(link => FilterLink(link, whereFilter));
+                    var updated = _linkUpdated.AsObservable().Where(link => FilterLink(link, whereFilter));
+
+                    return created.Merge(updated);
+                });
+        }
+
+        private bool FilterLink(Links link, LinksBooleanExpression? whereFilter)
+        {
+            if (whereFilter == null)
+                return true;
+
+            // Apply basic filters
+            if (whereFilter.id?._eq != null && link.id != whereFilter.id._eq)
+                return false;
+
+            if (whereFilter.from_id?._eq != null && (long)link.from_id != whereFilter.from_id._eq)
+                return false;
+
+            if (whereFilter.to_id?._eq != null && (long)link.to_id != whereFilter.to_id._eq)
+                return false;
+
+            // Note: type_id filtering would need the actual structure
+            // For now, we'll skip it as LinksBooleanExpression.type_id is a complex object
+
+            return true;
+        }
+
+        public void OnLinkCreated(Links link)
+        {
+            _linkCreated.OnNext(link);
+        }
+
+        public void OnLinkUpdated(Links link)
+        {
+            _linkUpdated.OnNext(link);
+        }
+
+        public void OnLinkDeleted(ulong linkId)
+        {
+            _linkDeleted.OnNext(linkId);
         }
     }
 }

csharp/Platform.Data.Doublets.Gql.Schema/Platform.Data.Doublets.Gql.Schema.csproj

@@ -13,6 +13,7 @@
     <ItemGroup>
         <PackageReference Include="Platform.Data.Doublets" Version="0.13.3" />
         <PackageReference Include="GraphQL.Server.Transports.Subscriptions.Abstractions" Version="5.0.2" />
+        <PackageReference Include="System.Reactive" Version="5.0.0" />
     </ItemGroup>
 
 </Project>

examples/subscription_test.md

@@ -0,0 +1,92 @@
+# GraphQL Subscriptions Test
+
+This file demonstrates how to test the newly implemented GraphQL subscriptions in the Data.Doublets.Gql project.
+
+## Available Subscriptions
+
+### 1. Link Created Subscription
+Subscribe to new link creation events:
+
+```graphql
+subscription LinkCreated {
+  link_created {
+    id
+    from_id
+    to_id
+    type_id
+  }
+}
+```
+
+With filtering:
+```graphql
+subscription LinkCreatedFiltered {
+  link_created(where: { from_id: { _eq: 1 } }) {
+    id
+    from_id
+    to_id
+    type_id
+  }
+}
+```
+
+### 2. Link Updated Subscription
+Subscribe to link update events:
+
+```graphql
+subscription LinkUpdated {
+  link_updated {
+    id
+    from_id
+    to_id
+    type_id
+  }
+}
+```
+
+### 3. Link Deleted Subscription
+Subscribe to link deletion events:
+
+```graphql
+subscription LinkDeleted {
+  link_deleted
+}
+```
+
+With ID filtering:
+```graphql
+subscription LinkDeletedFiltered {
+  link_deleted(id: 123)
+}
+```
+
+### 4. Link Changed Subscription
+Subscribe to all link changes (created and updated):
+
+```graphql
+subscription LinkChanged {
+  link_changed {
+    id
+    from_id
+    to_id
+    type_id
+  }
+}
+```
+
+## Testing Instructions
+
+1. Start the GraphQL server
+2. Open GraphQL Playground or similar tool
+3. Set up a subscription using one of the queries above
+4. In another tab/window, perform mutations:
+   - Insert new links
+   - Update existing links
+   - Delete links
+5. Observe that the subscription receives real-time notifications
+
+## WebSocket Connection
+
+Subscriptions require a WebSocket connection. The server is already configured to support WebSocket subscriptions through the `GraphQL.Server.Transports.Subscriptions.WebSockets` package.
+
+The WebSocket endpoint is available at the same GraphQL endpoint with the WebSocket protocol.