feat: add editor record (#7938)

fix CLOUD-58, CLOUD-61, CLOUD-62, PD-1607, PD-1608

Author: darkskygit

packages/backend/server/migrations/20240826033024_editor_record/migration.sql

@@ -0,0 +1,21 @@
+-- AlterTable
+ALTER TABLE "snapshot_histories" ADD COLUMN     "created_by" VARCHAR;
+
+-- AlterTable
+ALTER TABLE "snapshots" ADD COLUMN     "created_by" VARCHAR,
+ADD COLUMN     "updated_by" VARCHAR;
+
+-- AlterTable
+ALTER TABLE "updates" ADD COLUMN     "created_by" VARCHAR DEFAULT 'system';
+
+-- AddForeignKey
+ALTER TABLE "snapshots" ADD CONSTRAINT "snapshots_created_by_fkey" FOREIGN KEY ("created_by") REFERENCES "users"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "snapshots" ADD CONSTRAINT "snapshots_updated_by_fkey" FOREIGN KEY ("updated_by") REFERENCES "users"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "updates" ADD CONSTRAINT "updates_created_by_fkey" FOREIGN KEY ("created_by") REFERENCES "users"("id") ON DELETE SET NULL ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "snapshot_histories" ADD CONSTRAINT "snapshot_histories_created_by_fkey" FOREIGN KEY ("created_by") REFERENCES "users"("id") ON DELETE SET NULL ON UPDATE CASCADE;

packages/backend/server/schema.prisma

@@ -33,6 +33,10 @@ model User {
   aiSessions            AiSession[]
   updatedRuntimeConfigs RuntimeConfig[]
   userSnapshots         UserSnapshot[]
+  createdSnapshot       Snapshot[]                    @relation("createdSnapshot")
+  updatedSnapshot       Snapshot[]                    @relation("updatedSnapshot")
+  createdUpdate         Update[]                      @relation("createdUpdate")
+  createdHistory        SnapshotHistory[]             @relation("createdHistory")
 
   @@index([email])
   @@map("users")
@@ -241,9 +245,16 @@ model Snapshot {
   // the `updated_at` field will not record the time of record changed,
   // but the created time of last seen update that has been merged into snapshot.
   updatedAt   DateTime @map("updated_at") @db.Timestamptz(3)
+  createdBy   String?  @map("created_by") @db.VarChar
+  updatedBy   String?  @map("updated_by") @db.VarChar
+
+  // should not delete origin snapshot even if user is deleted
+  // we only delete the snapshot if the workspace is deleted
+  createdByUser User? @relation(name: "createdSnapshot", fields: [createdBy], references: [id], onDelete: SetNull)
+  updatedByUser User? @relation(name: "updatedSnapshot", fields: [updatedBy], references: [id], onDelete: SetNull)
 
   // @deprecated use updatedAt only
-  seq         Int?      @default(0) @db.Integer
+  seq Int? @default(0) @db.Integer
 
   // we need to clear all hanging updates and snapshots before enable the foreign key on workspaceId
   // workspace Workspace @relation(fields: [workspaceId], references: [id], onDelete: Cascade)
@@ -274,9 +285,14 @@ model Update {
   id          String   @map("guid") @db.VarChar
   blob        Bytes    @db.ByteA
   createdAt   DateTime @map("created_at") @db.Timestamptz(3)
+  // TODO(@darkskygit): fullfill old update, remove default value in next release
+  createdBy   String?  @default("system") @map("created_by") @db.VarChar
+
+  // will delete createor record if createor's account is deleted
+  createdByUser User? @relation(name: "createdUpdate", fields: [createdBy], references: [id], onDelete: SetNull)
 
   // @deprecated use createdAt only
-  seq         Int?      @db.Integer
+  seq Int? @db.Integer
 
   @@id([workspaceId, id, createdAt])
   @@map("updates")
@@ -289,6 +305,10 @@ model SnapshotHistory {
   blob        Bytes    @db.ByteA
   state       Bytes?   @db.ByteA
   expiredAt   DateTime @map("expired_at") @db.Timestamptz(3)
+  createdBy   String?  @map("created_by") @db.VarChar
+
+  // will delete createor record if creator's account is deleted
+  createdByUser User? @relation(name: "createdHistory", fields: [createdBy], references: [id], onDelete: SetNull)
 
   @@id([workspaceId, id, timestamp])
   @@map("snapshot_histories")

packages/backend/server/src/core/doc/adapters/userspace.ts

@@ -45,7 +45,12 @@ export class PgUserspaceDocStorageAdapter extends DocStorageAdapter {
     return this.getDocSnapshot(spaceId, docId);
   }
 
-  async pushDocUpdates(userId: string, docId: string, updates: Uint8Array[]) {
+  async pushDocUpdates(
+    userId: string,
+    docId: string,
+    updates: Uint8Array[],
+    editorId?: string
+  ) {
     if (!updates.length) {
       return 0;
     }
@@ -67,6 +72,7 @@ export class PgUserspaceDocStorageAdapter extends DocStorageAdapter {
       docId,
       bin,
       timestamp,
+      editor: editorId,
     });
 
     return timestamp;
@@ -135,6 +141,7 @@ export class PgUserspaceDocStorageAdapter extends DocStorageAdapter {
       docId,
       bin: snapshot.blob,
       timestamp: snapshot.updatedAt.getTime(),
+      editor: snapshot.userId,
     };
   }
 

packages/backend/server/src/core/doc/adapters/workspace.ts

@@ -38,7 +38,8 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
   async pushDocUpdates(
     workspaceId: string,
     docId: string,
-    updates: Uint8Array[]
+    updates: Uint8Array[],
+    editorId?: string
   ) {
     if (!updates.length) {
       return 0;
@@ -82,6 +83,7 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
                 blob: Buffer.from(update),
                 seq,
                 createdAt: new Date(createdAt),
+                createdBy: editorId || null,
               };
             }),
           });
@@ -113,6 +115,7 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
     return rows.map(row => ({
       bin: row.blob,
       timestamp: row.createdAt.getTime(),
+      editor: row.createdBy || undefined,
     }));
   }
 
@@ -216,6 +219,12 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
     const histories = await this.db.snapshotHistory.findMany({
       select: {
         timestamp: true,
+        createdByUser: {
+          select: {
+            name: true,
+            avatarUrl: true,
+          },
+        },
       },
       where: {
         workspaceId,
@@ -230,7 +239,10 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
       take: query.limit,
     });
 
-    return histories.map(h => h.timestamp.getTime());
+    return histories.map(h => ({
+      timestamp: h.timestamp.getTime(),
+      editor: h.createdByUser,
+    }));
   }
 
   async getDocHistory(workspaceId: string, docId: string, timestamp: number) {
@@ -253,13 +265,15 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
       docId,
       bin: history.blob,
       timestamp,
+      editor: history.createdBy || undefined,
     };
   }
 
   override async rollbackDoc(
     spaceId: string,
     docId: string,
-    timestamp: number
+    timestamp: number,
+    editorId?: string
   ): Promise<void> {
     await using _lock = await this.lockDocForUpdate(spaceId, docId);
     const toSnapshot = await this.getDocHistory(spaceId, docId, timestamp);
@@ -274,7 +288,14 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
     }
 
     // force create a new history record after rollback
-    await this.createDocHistory(fromSnapshot, true);
+    await this.createDocHistory(
+      {
+        ...fromSnapshot,
+        // override the editor to the one who requested the rollback
+        editor: editorId,
+      },
+      true
+    );
     // WARN:
     //  we should never do the snapshot updating in recovering,
     //  which is not the solution in CRDT.
@@ -331,6 +352,7 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
             id: snapshot.docId,
             timestamp: new Date(snapshot.timestamp),
             blob: Buffer.from(snapshot.bin),
+            createdBy: snapshot.editor,
             expiredAt: new Date(
               Date.now() + (await this.options.historyMaxAge(snapshot.spaceId))
             ),
@@ -374,6 +396,8 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
       docId,
       bin: snapshot.blob,
       timestamp: snapshot.updatedAt.getTime(),
+      // creator and editor may null if their account is deleted
+      editor: snapshot.updatedBy || snapshot.createdBy || undefined,
     };
   }
 
@@ -396,10 +420,10 @@ export class PgWorkspaceDocStorageAdapter extends DocStorageAdapter {
     //                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     try {
       const result: { updatedAt: Date }[] = await this.db.$queryRaw`
-        INSERT INTO "snapshots" ("workspace_id", "guid", "blob", "created_at", "updated_at")
-        VALUES (${spaceId}, ${docId}, ${bin}, DEFAULT, ${updatedAt})
+        INSERT INTO "snapshots" ("workspace_id", "guid", "blob", "created_at", "updated_at", "created_by", "updated_by")
+        VALUES (${spaceId}, ${docId}, ${bin}, DEFAULT, ${updatedAt}, ${snapshot.editor}, ${snapshot.editor})
         ON CONFLICT ("workspace_id", "guid")
-        DO UPDATE SET "blob" = ${bin}, "updated_at" = ${updatedAt}
+        DO UPDATE SET "blob" = ${bin}, "updated_at" = ${updatedAt}, "updated_by" = ${snapshot.editor}
         WHERE "snapshots"."workspace_id" = ${spaceId} AND "snapshots"."guid" = ${docId} AND "snapshots"."updated_at" <= ${updatedAt}
         RETURNING "snapshots"."workspace_id" as "workspaceId", "snapshots"."guid" as "id", "snapshots"."updated_at" as "updatedAt"
       `;

packages/backend/server/src/core/doc/index.ts

@@ -22,4 +22,4 @@ import { DocStorageOptions } from './options';
 export class DocStorageModule {}
 export { PgUserspaceDocStorageAdapter, PgWorkspaceDocStorageAdapter };
 
-export { DocStorageAdapter } from './storage';
+export { DocStorageAdapter, type Editor } from './storage';

packages/backend/server/src/core/doc/storage/doc.ts

@@ -16,18 +16,25 @@ export interface DocRecord {
   docId: string;
   bin: Uint8Array;
   timestamp: number;
+  editor?: string;
 }
 
 export interface DocUpdate {
   bin: Uint8Array;
   timestamp: number;
+  editor?: string;
 }
 
 export interface HistoryFilter {
   before?: number;
   limit?: number;
 }
 
+export interface Editor {
+  name: string;
+  avatarUrl: string | null;
+}
+
 export interface DocStorageOptions {
   mergeUpdates?: (updates: Uint8Array[]) => Promise<Uint8Array> | Uint8Array;
 }
@@ -61,7 +68,7 @@ export abstract class DocStorageAdapter extends Connection {
     const updates = await this.getDocUpdates(spaceId, docId);
 
     if (updates.length) {
-      const { timestamp, bin } = await this.squash(
+      const { timestamp, bin, editor } = await this.squash(
         snapshot ? [snapshot, ...updates] : updates
       );
 
@@ -70,6 +77,7 @@ export abstract class DocStorageAdapter extends Connection {
         docId,
         bin,
         timestamp,
+        editor,
       };
 
       const success = await this.setDocSnapshot(newSnapshot);
@@ -91,15 +99,17 @@ export abstract class DocStorageAdapter extends Connection {
   abstract pushDocUpdates(
     spaceId: string,
     docId: string,
-    updates: Uint8Array[]
+    updates: Uint8Array[],
+    editorId?: string
   ): Promise<number>;
 
   abstract deleteDoc(spaceId: string, docId: string): Promise<void>;
   abstract deleteSpace(spaceId: string): Promise<void>;
   async rollbackDoc(
     spaceId: string,
     docId: string,
-    timestamp: number
+    timestamp: number,
+    editorId?: string
   ): Promise<void> {
     await using _lock = await this.lockDocForUpdate(spaceId, docId);
     const toSnapshot = await this.getDocHistory(spaceId, docId, timestamp);
@@ -114,7 +124,7 @@ export abstract class DocStorageAdapter extends Connection {
     }
 
     const change = this.generateChangeUpdate(fromSnapshot.bin, toSnapshot.bin);
-    await this.pushDocUpdates(spaceId, docId, [change]);
+    await this.pushDocUpdates(spaceId, docId, [change], editorId);
     // force create a new history record after rollback
     await this.createDocHistory(fromSnapshot, true);
   }
@@ -127,7 +137,7 @@ export abstract class DocStorageAdapter extends Connection {
     spaceId: string,
     docId: string,
     query: { skip?: number; limit?: number }
-  ): Promise<number[]>;
+  ): Promise<{ timestamp: number; editor: Editor | null }[]>;
   abstract getDocHistory(
     spaceId: string,
     docId: string,
@@ -173,6 +183,7 @@ export abstract class DocStorageAdapter extends Connection {
     return {
       bin: finalUpdate,
       timestamp: lastUpdate.timestamp,
+      editor: lastUpdate.editor,
     };
   }
 

packages/backend/server/src/core/doc/storage/index.ts

@@ -28,5 +28,6 @@ export {
   DocStorageAdapter,
   type DocStorageOptions,
   type DocUpdate,
+  type Editor,
   type HistoryFilter,
 } from './doc';

packages/backend/server/src/core/sync/gateway.ts

@@ -264,9 +264,11 @@ export class SpaceSyncGateway
     };
   }
 
+  @Auth()
   @SubscribeMessage('space:push-doc-updates')
   async onReceiveDocUpdates(
     @ConnectedSocket() client: Socket,
+    @CurrentUser() user: CurrentUser,
     @MessageBody()
     message: PushDocUpdatesMessage
   ): Promise<EventResponse<{ accepted: true; timestamp?: number }>> {
@@ -277,7 +279,8 @@ export class SpaceSyncGateway
     const timestamp = await adapter.push(
       spaceId,
       docId,
-      updates.map(update => Buffer.from(update, 'base64'))
+      updates.map(update => Buffer.from(update, 'base64')),
+      user.id
     );
 
     // could be put in [adapter.push]
@@ -448,8 +451,10 @@ export class SpaceSyncGateway
     });
   }
 
+  @Auth()
   @SubscribeMessage('client-update-v2')
   async handleClientUpdateV2(
+    @CurrentUser() user: CurrentUser,
     @MessageBody()
     {
       workspaceId,
@@ -462,7 +467,7 @@ export class SpaceSyncGateway
     },
     @ConnectedSocket() client: Socket
   ): Promise<EventResponse<{ accepted: true; timestamp?: number }>> {
-    return this.onReceiveDocUpdates(client, {
+    return this.onReceiveDocUpdates(client, user, {
       spaceType: SpaceType.Workspace,
       spaceId: workspaceId,
       docId: guid,
@@ -596,9 +601,9 @@ abstract class SyncSocketAdapter {
     permission?: Permission
   ): Promise<void>;
 
-  push(spaceId: string, docId: string, updates: Buffer[]) {
+  push(spaceId: string, docId: string, updates: Buffer[], editorId: string) {
     this.assertIn(spaceId);
-    return this.storage.pushDocUpdates(spaceId, docId, updates);
+    return this.storage.pushDocUpdates(spaceId, docId, updates, editorId);
   }
 
   get(spaceId: string, docId: string) {
@@ -621,9 +626,14 @@ class WorkspaceSyncAdapter extends SyncSocketAdapter {
     super(SpaceType.Workspace, client, storage);
   }
 
-  override push(spaceId: string, docId: string, updates: Buffer[]) {
+  override push(
+    spaceId: string,
+    docId: string,
+    updates: Buffer[],
+    editorId: string
+  ) {
     const id = new DocID(docId, spaceId);
-    return super.push(spaceId, id.guid, updates);
+    return super.push(spaceId, id.guid, updates, editorId);
   }
 
   override get(spaceId: string, docId: string) {