feat: added registerListener

api/src/modules/DocumentDrive/model.ts

@@ -2,14 +2,23 @@ import type { Prisma } from "@prisma/client";
 import {
   DocumentDriveServer,
   DriveInput,
+  Listener,
   ListenerRevision,
   MemoryStorage,
   PrismaStorage,
+  generateUUID,
 } from "document-drive";
 import * as DocumentModelsLibs from "document-model-libs/document-models";
 import { module as DocumentModelLib } from "document-model/document-model";
 import { DocumentModel, Operation } from "document-model/dist/browser/document";
 import { PullResponderTransmitter } from "document-drive/src/transmitter/pull-responder";
+import {
+  utils as DocumentDriveUtils,
+  ListenerFilter,
+  actions,
+  reducer,
+} from "document-model-libs/dist/document-drive";
+
 export function getDocumentDriveCRUD(prisma: Prisma.TransactionClient) {
   const documentModels = [
     DocumentModelLib,
@@ -144,5 +153,27 @@ export function getDocumentDriveCRUD(prisma: Prisma.TransactionClient) {
         revisions
       );
     },
+
+    registerListener: async (driveId: string, filter: ListenerFilter) => {
+      const uuid = generateUUID();
+      const listener: Listener = {
+        block: false,
+        callInfo: {
+          data: "",
+          name: "PullResponder",
+          transmitterType: "PullResponder",
+        },
+        filter,
+        label: `Pullresponder #${uuid}`,
+        listenerId: uuid,
+        system: false,
+      };
+      const action = actions.addListener({
+        listener,
+      });
+
+      await driveServer.addDriveOperations(driveId, [action]);
+      return listener;
+    },
   };
 }

api/src/modules/DocumentDrive/mutations/registerListener.ts

@@ -0,0 +1,54 @@
+import { list, mutationField, nonNull } from "nexus";
+import { InputStrandUpdate, ListenerRevision } from "../definitions";
+import { OperationScope } from "document-model/document";
+import {
+  ListenerRevision as IListenerRevision,
+  UpdateStatus,
+} from "document-drive";
+
+export const registerListener = mutationField("registerListener", {
+  type: Listener,
+  args: {
+    filter: list(nonNull(InputStrandUpdate)),
+  },
+  resolve: async (_parent, { filter }, ctx) => {
+    //@todo: get connect drive server from ctx and apply updates
+
+    if (!strands || strands?.length === 0) return [];
+    const listenerRevisions: IListenerRevision[] = [];
+    const results = await Promise.all(
+      strands.map(async (s) => {
+        const operations = s.operations?.map((o) => {
+          const op = {
+            scope: s.scope as OperationScope,
+            branch: s.branch,
+            ...o,
+            input: JSON.parse(o.input),
+          };
+
+          return op;
+        });
+        try {
+          const result = await ctx.prisma.document.registerListener(
+            s.driveId,
+            s.documentId,
+            operations
+          );
+
+          listenerRevisions.push({
+            branch: s.branch,
+            documentId: s.documentId,
+            driveId: s.driveId,
+            revision: result.operations.pop().revision,
+            scope: s.scope as OperationScope,
+            status: (result.error ? "ERROR" : "SUCCESS") as UpdateStatus,
+          });
+        } catch (e) {
+          console.log(e);
+        }
+      })
+    );
+
+    return listenerRevisions;
+  },
+});