Merge branch 'canary' into stable

packages/backend/server/package.json

@@ -41,11 +41,11 @@
     "@opentelemetry/core": "^1.18.1",
     "@opentelemetry/exporter-prometheus": "^0.45.1",
     "@opentelemetry/exporter-zipkin": "^1.18.1",
-    "@opentelemetry/host-metrics": "^0.33.2",
+    "@opentelemetry/host-metrics": "^0.34.0",
     "@opentelemetry/instrumentation": "^0.45.1",
     "@opentelemetry/instrumentation-graphql": "^0.36.0",
     "@opentelemetry/instrumentation-http": "^0.45.1",
-    "@opentelemetry/instrumentation-ioredis": "^0.35.3",
+    "@opentelemetry/instrumentation-ioredis": "^0.36.0",
     "@opentelemetry/instrumentation-nestjs-core": "^0.33.3",
     "@opentelemetry/instrumentation-socket.io": "^0.34.3",
     "@opentelemetry/resources": "^1.18.1",
@@ -102,7 +102,7 @@
     "@types/sinon": "^17.0.2",
     "@types/supertest": "^2.0.16",
     "@types/ws": "^8.5.10",
-    "ava": "^5.3.1",
+    "ava": "^6.0.0",
     "c8": "^8.0.1",
     "nodemon": "^3.0.1",
     "sinon": "^17.0.1",

packages/backend/server/src/app.ts

@@ -3,6 +3,7 @@ import { Module } from '@nestjs/common';
 import { AppController } from './app.controller';
 import { CacheModule } from './cache';
 import { ConfigModule } from './config';
+import { EventModule } from './event';
 import { BusinessModules } from './modules';
 import { AuthModule } from './modules/auth';
 import { PrismaModule } from './prisma';
@@ -14,6 +15,7 @@ const BasicModules = [
   PrismaModule,
   ConfigModule.forRoot(),
   CacheModule,
+  EventModule,
   StorageModule.forRoot(),
   SessionModule,
   RateLimiterModule,

packages/backend/server/src/event/events.ts

@@ -0,0 +1,23 @@
+import type { Snapshot, Workspace } from '@prisma/client';
+
+import { Flatten, Payload } from './types';
+
+interface EventDefinitions {
+  workspace: {
+    deleted: Payload<Workspace['id']>;
+  };
+
+  snapshot: {
+    updated: Payload<
+      Pick<Snapshot, 'id' | 'workspaceId'> & {
+        previous: Pick<Snapshot, 'blob' | 'state' | 'updatedAt'>;
+      }
+    >;
+    deleted: Payload<Pick<Snapshot, 'id' | 'workspaceId'>>;
+  };
+}
+
+export type EventKV = Flatten<EventDefinitions>;
+
+export type Event = keyof EventKV;
+export type EventPayload<E extends Event> = EventKV[E];

packages/backend/server/src/event/index.ts

@@ -0,0 +1,45 @@
+import { Global, Injectable, Module } from '@nestjs/common';
+import {
+  EventEmitter2,
+  EventEmitterModule,
+  OnEvent as RawOnEvent,
+} from '@nestjs/event-emitter';
+
+import type { Event, EventPayload } from './events';
+
+@Injectable()
+export class EventEmitter {
+  constructor(private readonly emitter: EventEmitter2) {}
+
+  emit<E extends Event>(event: E, payload: EventPayload<E>) {
+    return this.emitter.emit(event, payload);
+  }
+
+  emitAsync<E extends Event>(event: E, payload: EventPayload<E>) {
+    return this.emitter.emitAsync(event, payload);
+  }
+
+  on<E extends Event>(event: E, handler: (payload: EventPayload<E>) => void) {
+    return this.emitter.on(event, handler);
+  }
+
+  once<E extends Event>(event: E, handler: (payload: EventPayload<E>) => void) {
+    return this.emitter.once(event, handler);
+  }
+}
+
+export const OnEvent = (
+  event: Event,
+  opts?: Parameters<typeof RawOnEvent>[1]
+) => {
+  return RawOnEvent(event, opts);
+};
+
+@Global()
+@Module({
+  imports: [EventEmitterModule.forRoot()],
+  providers: [EventEmitter],
+  exports: [EventEmitter],
+})
+export class EventModule {}
+export { EventPayload };

packages/backend/server/src/event/types.ts

@@ -0,0 +1,33 @@
+export type Payload<T> = {
+  __payload: true;
+  data: T;
+};
+
+export type Join<A extends string, B extends string> = A extends ''
+  ? B
+  : `${A}.${B}`;
+
+export type PathType<T, Path extends string> = string extends Path
+  ? unknown
+  : Path extends keyof T
+    ? T[Path]
+    : Path extends `${infer K}.${infer R}`
+      ? K extends keyof T
+        ? PathType<T[K], R>
+        : unknown
+      : unknown;
+
+export type Leaves<T, P extends string = ''> = T extends Payload<any>
+  ? P
+  : T extends Record<string, any>
+    ? {
+        [K in keyof T]: K extends string ? Leaves<T[K], Join<P, K>> : never;
+      }[keyof T]
+    : never;
+
+export type Flatten<T> = Leaves<T> extends infer R
+  ? {
+      // @ts-expect-error yo, ts can't make it
+      [K in R]: PathType<T, K> extends Payload<infer U> ? U : never;
+    }
+  : never;

packages/backend/server/src/modules/doc/history.ts

@@ -1,15 +1,15 @@
 import { isDeepStrictEqual } from 'node:util';
 
 import { Injectable, Logger } from '@nestjs/common';
-import { OnEvent } from '@nestjs/event-emitter';
 import { Cron, CronExpression } from '@nestjs/schedule';
-import type { Snapshot } from '@prisma/client';
 
 import { Config } from '../../config';
+import { type EventPayload, OnEvent } from '../../event';
 import { metrics } from '../../metrics';
 import { PrismaService } from '../../prisma';
 import { SubscriptionStatus } from '../payment/service';
 import { Permission } from '../workspaces/types';
+import { isEmptyBuffer } from './manager';
 
 @Injectable()
 export class DocHistoryManager {
@@ -19,16 +19,38 @@ export class DocHistoryManager {
     private readonly db: PrismaService
   ) {}
 
-  @OnEvent('doc:manager:snapshot:beforeUpdate')
-  async onDocUpdated(snapshot: Snapshot, forceCreate = false) {
-    const last = await this.last(snapshot.workspaceId, snapshot.id);
+  @OnEvent('workspace.deleted')
+  onWorkspaceDeleted(workspaceId: EventPayload<'workspace.deleted'>) {
+    return this.db.snapshotHistory.deleteMany({
+      where: {
+        workspaceId,
+      },
+    });
+  }
+
+  @OnEvent('snapshot.deleted')
+  onSnapshotDeleted({ workspaceId, id }: EventPayload<'snapshot.deleted'>) {
+    return this.db.snapshotHistory.deleteMany({
+      where: {
+        workspaceId,
+        id,
+      },
+    });
+  }
+
+  @OnEvent('snapshot.updated')
+  async onDocUpdated(
+    { workspaceId, id, previous }: EventPayload<'snapshot.updated'>,
+    forceCreate = false
+  ) {
+    const last = await this.last(workspaceId, id);
 
     let shouldCreateHistory = false;
 
     if (!last) {
       // never created
       shouldCreateHistory = true;
-    } else if (last.timestamp === snapshot.updatedAt) {
+    } else if (last.timestamp === previous.updatedAt) {
       // no change
       shouldCreateHistory = false;
     } else if (
@@ -36,16 +58,23 @@ export class DocHistoryManager {
       forceCreate ||
       // last history created before interval in configs
       last.timestamp.getTime() <
-        snapshot.updatedAt.getTime() - this.config.doc.history.interval
+        previous.updatedAt.getTime() - this.config.doc.history.interval
     ) {
       shouldCreateHistory = true;
     }
 
     if (shouldCreateHistory) {
       // skip the history recording when no actual update on snapshot happended
-      if (last && isDeepStrictEqual(last.state, snapshot.state)) {
+      if (last && isDeepStrictEqual(last.state, previous.state)) {
         this.logger.debug(
-          `State matches, skip creating history record for ${snapshot.id} in workspace ${snapshot.workspaceId}`
+          `State matches, skip creating history record for ${id} in workspace ${workspaceId}`
+        );
+        return;
+      }
+
+      if (isEmptyBuffer(previous.blob)) {
+        this.logger.debug(
+          `Doc is empty, skip creating history record for ${id} in workspace ${workspaceId}`
         );
         return;
       }
@@ -56,12 +85,12 @@ export class DocHistoryManager {
             timestamp: true,
           },
           data: {
-            workspaceId: snapshot.workspaceId,
-            id: snapshot.id,
-            timestamp: snapshot.updatedAt,
-            blob: snapshot.blob,
-            state: snapshot.state,
-            expiredAt: await this.getExpiredDateFromNow(snapshot.workspaceId),
+            workspaceId,
+            id,
+            timestamp: previous.updatedAt,
+            blob: previous.blob,
+            state: previous.state,
+            expiredAt: await this.getExpiredDateFromNow(workspaceId),
           },
         })
         .catch(() => {
@@ -73,9 +102,7 @@ export class DocHistoryManager {
           description: 'How many times the snapshot history created',
         })
         .add(1);
-      this.logger.log(
-        `History created for ${snapshot.id} in workspace ${snapshot.workspaceId}.`
-      );
+      this.logger.log(`History created for ${id} in workspace ${workspaceId}.`);
     }
   }
 
@@ -180,7 +207,7 @@ export class DocHistoryManager {
     }
 
     // save old snapshot as one history record
-    await this.onDocUpdated(oldSnapshot, true);
+    await this.onDocUpdated({ workspaceId, id, previous: oldSnapshot }, true);
     // WARN:
     //  we should never do the snapshot updating in recovering,
     //  which is not the solution in CRDT.

packages/backend/server/src/modules/doc/manager.ts

@@ -5,12 +5,12 @@ import {
   OnModuleDestroy,
   OnModuleInit,
 } from '@nestjs/common';
-import { EventEmitter2 } from '@nestjs/event-emitter';
 import { Snapshot, Update } from '@prisma/client';
 import { chunk } from 'lodash-es';
 import { defer, retry } from 'rxjs';
 import {
   applyUpdate,
+  decodeStateVector,
   Doc,
   encodeStateAsUpdate,
   encodeStateVector,
@@ -19,6 +19,7 @@ import {
 
 import { Cache } from '../../cache';
 import { Config } from '../../config';
+import { EventEmitter, type EventPayload, OnEvent } from '../../event';
 import { metrics } from '../../metrics/metrics';
 import { PrismaService } from '../../prisma';
 import { mergeUpdatesInApplyWay as jwstMergeUpdates } from '../../storage';
@@ -40,7 +41,37 @@ function compare(yBinary: Buffer, jwstBinary: Buffer, strict = false): boolean {
   return compare(yBinary, yBinary2, true);
 }
 
-function isEmptyBuffer(buf: Buffer): boolean {
+/**
+ * Detect whether rhs state is newer than lhs state.
+ *
+ * How could we tell a state is newer:
+ *
+ *   i. if the state vector size is larger, it's newer
+ *  ii. if the state vector size is same, compare each client's state
+ */
+function isStateNewer(lhs: Buffer, rhs: Buffer): boolean {
+  const lhsVector = decodeStateVector(lhs);
+  const rhsVector = decodeStateVector(rhs);
+
+  if (lhsVector.size < rhsVector.size) {
+    return true;
+  }
+
+  for (const [client, state] of lhsVector) {
+    const rstate = rhsVector.get(client);
+    if (!rstate) {
+      return false;
+    }
+
+    if (state < rstate) {
+      return true;
+    }
+  }
+
+  return false;
+}
+
+export function isEmptyBuffer(buf: Buffer): boolean {
   return (
     buf.length === 0 ||
     // 0x0000
@@ -71,7 +102,7 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
     private readonly db: PrismaService,
     private readonly config: Config,
     private readonly cache: Cache,
-    private readonly event: EventEmitter2
+    private readonly event: EventEmitter
   ) {}
 
   onModuleInit() {
@@ -193,6 +224,33 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
     }
   }
 
+  @OnEvent('workspace.deleted')
+  async onWorkspaceDeleted(workspaceId: string) {
+    await this.db.snapshot.deleteMany({
+      where: {
+        workspaceId,
+      },
+    });
+    await this.db.update.deleteMany({
+      where: {
+        workspaceId,
+      },
+    });
+  }
+
+  @OnEvent('snapshot.deleted')
+  async onSnapshotDeleted({
+    id,
+    workspaceId,
+  }: EventPayload<'snapshot.deleted'>) {
+    await this.db.update.deleteMany({
+      where: {
+        id,
+        workspaceId,
+      },
+    });
+  }
+
   /**
    * add update to manager for later processing.
    */
@@ -374,23 +432,17 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
     }
 
     const { id, workspaceId } = candidate;
-    // acquire lock
-    const ok = await this.lockUpdatesForAutoSquash(workspaceId, id);
 
-    if (!ok) {
-      return;
-    }
-
-    try {
-      await this._get(workspaceId, id);
-    } catch (e) {
-      this.logger.error(
-        `Failed to apply updates for workspace: ${workspaceId}, guid: ${id}`
-      );
-      this.logger.error(e);
-    } finally {
-      await this.unlockUpdatesForAutoSquash(workspaceId, id);
-    }
+    await this.lockUpdatesForAutoSquash(workspaceId, id, async () => {
+      try {
+        await this._get(workspaceId, id);
+      } catch (e) {
+        this.logger.error(
+          `Failed to apply updates for workspace: ${workspaceId}, guid: ${id}`
+        );
+        this.logger.error(e);
+      }
+    });
   }
 
   private async getAutoSquashCandidate() {
@@ -414,34 +466,67 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
     doc: Doc,
     initialSeq?: number
   ) {
-    const blob = Buffer.from(encodeStateAsUpdate(doc));
-    const state = Buffer.from(encodeStateVector(doc));
+    return this.lockSnapshotForUpsert(workspaceId, guid, async () => {
+      const blob = Buffer.from(encodeStateAsUpdate(doc));
 
-    if (isEmptyBuffer(blob)) {
-      return;
-    }
+      if (isEmptyBuffer(blob)) {
+        return false;
+      }
 
-    await this.db.snapshot.upsert({
-      select: {
-        seq: true,
-      },
-      where: {
-        id_workspaceId: {
-          id: guid,
-          workspaceId,
-        },
-      },
-      create: {
-        id: guid,
-        workspaceId,
-        blob,
-        state,
-        seq: initialSeq,
-      },
-      update: {
-        blob,
-        state,
-      },
+      const state = Buffer.from(encodeStateVector(doc));
+
+      return await this.db.$transaction(async db => {
+        const snapshot = await db.snapshot.findUnique({
+          where: {
+            id_workspaceId: {
+              id: guid,
+              workspaceId,
+            },
+          },
+        });
+
+        // update
+        if (snapshot) {
+          // only update if state is newer
+          if (isStateNewer(snapshot.state ?? Buffer.from([0]), state)) {
+            await db.snapshot.update({
+              select: {
+                seq: true,
+              },
+              where: {
+                id_workspaceId: {
+                  workspaceId,
+                  id: guid,
+                },
+              },
+              data: {
+                blob,
+                state,
+              },
+            });
+
+            return true;
+          } else {
+            return false;
+          }
+        } else {
+          // create
+          await db.snapshot.create({
+            select: {
+              seq: true,
+            },
+            data: {
+              id: guid,
+              workspaceId,
+              blob,
+              state,
+              seq: initialSeq,
+            },
+          });
+
+          return true;
+        }
+      });
     });
   }
 
@@ -480,25 +565,39 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
       ...updates.map(u => u.blob)
     );
 
-    if (snapshot) {
-      this.event.emit('doc:manager:snapshot:beforeUpdate', snapshot);
-    }
-
     await this.upsert(workspaceId, id, doc, last.seq);
-    this.logger.debug(
-      `Squashed ${updates.length} updates for ${id} in workspace ${workspaceId}`
-    );
-    await this.db.update.deleteMany({
-      where: {
+    if (snapshot) {
+      this.event.emit('snapshot.updated', {
         id,
         workspaceId,
-        seq: {
-          in: updates.map(u => u.seq),
+        previous: {
+          blob: snapshot.blob,
+          state: snapshot.state,
+          updatedAt: snapshot.updatedAt,
         },
-      },
-    });
+      });
+    }
+
+    const done = await this.upsert(workspaceId, id, doc, last.seq);
+
+    if (done) {
+      this.logger.debug(
+        `Squashed ${updates.length} updates for ${id} in workspace ${workspaceId}`
+      );
+
+      await this.db.update.deleteMany({
+        where: {
+          id,
+          workspaceId,
+          seq: {
+            in: updates.map(u => u.seq),
+          },
+        },
+      });
+
+      await this.updateCachedUpdatesCount(workspaceId, id, -updates.length);
+    }
 
-    await this.updateCachedUpdatesCount(workspaceId, id, -updates.length);
     return doc;
   }
 
@@ -581,22 +680,44 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
     return null;
   }
 
-  private async lockUpdatesForAutoSquash(workspaceId: string, guid: string) {
-    return this.cache.setnx(
+  private async doWithLock<T>(lock: string, job: () => Promise<T>) {
+    const acquired = await this.cache.setnx(lock, 1, {
+      ttl: 60 * 1000,
+    });
+
+    if (!acquired) {
+      return;
+    }
+
+    try {
+      return await job();
+    } finally {
+      await this.cache.delete(lock).catch(e => {
+        // safe, the lock will be expired when ttl ends
+        this.logger.error(`Failed to release lock ${lock}`, e);
+      });
+    }
+  }
+
+  private async lockUpdatesForAutoSquash<T>(
+    workspaceId: string,
+    guid: string,
+    job: () => Promise<T>
+  ) {
+    return this.doWithLock(
       `doc:manager:updates-lock:${workspaceId}::${guid}`,
-      1,
-      {
-        ttl: 60 * 1000,
-      }
+      job
     );
   }
 
-  private async unlockUpdatesForAutoSquash(workspaceId: string, guid: string) {
-    return this.cache
-      .delete(`doc:manager:updates-lock:${workspaceId}::${guid}`)
-      .catch(e => {
-        // safe, the lock will be expired when ttl ends
-        this.logger.error('Failed to release updates lock', e);
-      });
+  async lockSnapshotForUpsert<T>(
+    workspaceId: string,
+    guid: string,
+    job: () => Promise<T>
+  ) {
+    return this.doWithLock(
+      `doc:manager:snapshot-lock:${workspaceId}::${guid}`,
+      job
+    );
   }
 }

packages/backend/server/src/modules/index.ts

@@ -1,5 +1,4 @@
 import { DynamicModule, Type } from '@nestjs/common';
-import { EventEmitterModule } from '@nestjs/event-emitter';
 import { ScheduleModule } from '@nestjs/schedule';
 
 import { GqlModule } from '../graphql.module';
@@ -11,11 +10,7 @@ import { SyncModule } from './sync';
 import { UsersModule } from './users';
 import { WorkspaceModule } from './workspaces';
 
-const BusinessModules: (Type | DynamicModule)[] = [
-  EventEmitterModule.forRoot({
-    global: true,
-  }),
-];
+const BusinessModules: (Type | DynamicModule)[] = [];
 
 switch (SERVER_FLAVOR) {
   case 'sync':

packages/backend/server/src/modules/sync/events/events.gateway.ts

@@ -25,7 +25,6 @@ import {
   EventError,
   InternalError,
   NotInWorkspaceError,
-  WorkspaceNotFoundError,
 } from './error';
 
 export const GatewayErrorWrapper = (): MethodDecorator => {
@@ -319,9 +318,7 @@ export class EventsGateway implements OnGatewayConnection, OnGatewayDisconnect {
 
     if (!doc) {
       return {
-        error: docId.isWorkspace
-          ? new WorkspaceNotFoundError(workspaceId)
-          : new DocNotFoundError(workspaceId, docId.guid),
+        error: new DocNotFoundError(workspaceId, docId.guid),
       };
     }
 

packages/backend/server/src/modules/workspaces/resolver.ts

@@ -33,6 +33,7 @@ import type {
 import GraphQLUpload from 'graphql-upload/GraphQLUpload.mjs';
 import { applyUpdate, Doc } from 'yjs';
 
+import { EventEmitter } from '../../event';
 import { PrismaService } from '../../prisma';
 import { StorageProvide } from '../../storage';
 import { CloudThrottlerGuard, Throttle } from '../../throttler';
@@ -146,6 +147,7 @@ export class WorkspaceResolver {
     private readonly prisma: PrismaService,
     private readonly permissions: PermissionService,
     private readonly users: UsersService,
+    private readonly event: EventEmitter,
     @Inject(StorageProvide) private readonly storage: Storage
   ) {}
 
@@ -308,22 +310,11 @@ export class WorkspaceResolver {
   })
   async createWorkspace(
     @CurrentUser() user: UserType,
-    @Args({ name: 'init', type: () => GraphQLUpload })
-    update: FileUpload
+    // we no longer support init workspace with a preload file
+    // use sync system to uploading them once created
+    @Args({ name: 'init', type: () => GraphQLUpload, nullable: true })
+    init: FileUpload | null
   ) {
-    // convert stream to buffer
-    const buffer = await new Promise<Buffer>((resolve, reject) => {
-      const stream = update.createReadStream();
-      const chunks: Uint8Array[] = [];
-      stream.on('data', chunk => {
-        chunks.push(chunk);
-      });
-      stream.on('error', reject);
-      stream.on('end', () => {
-        resolve(Buffer.concat(chunks));
-      });
-    });
-
     const workspace = await this.prisma.workspace.create({
       data: {
         public: false,
@@ -341,14 +332,31 @@ export class WorkspaceResolver {
       },
     });
 
-    if (buffer.length) {
-      await this.prisma.snapshot.create({
-        data: {
-          id: workspace.id,
-          workspaceId: workspace.id,
-          blob: buffer,
-        },
+    if (init) {
+      // convert stream to buffer
+      const buffer = await new Promise<Buffer>(resolve => {
+        const stream = init.createReadStream();
+        const chunks: Uint8Array[] = [];
+        stream.on('data', chunk => {
+          chunks.push(chunk);
+        });
+        stream.on('error', () => {
+          resolve(Buffer.from([]));
+        });
+        stream.on('end', () => {
+          resolve(Buffer.concat(chunks));
+        });
       });
+
+      if (buffer.length) {
+        await this.prisma.snapshot.create({
+          data: {
+            id: workspace.id,
+            workspaceId: workspace.id,
+            blob: buffer,
+          },
+        });
+      }
     }
 
     return workspace;
@@ -382,18 +390,7 @@ export class WorkspaceResolver {
       },
     });
 
-    await this.prisma.$transaction([
-      this.prisma.update.deleteMany({
-        where: {
-          workspaceId: id,
-        },
-      }),
-      this.prisma.snapshot.deleteMany({
-        where: {
-          workspaceId: id,
-        },
-      }),
-    ]);
+    this.event.emit('workspace.deleted', id);
 
     return true;
   }

packages/backend/server/src/schema.gql

@@ -292,7 +292,7 @@ type Mutation {
   sendVerifyChangeEmail(token: String!, email: String!, callbackUrl: String!): Boolean!
 
   """Create a new workspace"""
-  createWorkspace(init: Upload!): WorkspaceType!
+  createWorkspace(init: Upload): WorkspaceType!
 
   """Update workspace"""
   updateWorkspace(input: UpdateWorkspaceInput!): WorkspaceType!

packages/backend/server/tests/doc.spec.ts

@@ -1,15 +1,20 @@
 import { mock } from 'node:test';
 
 import type { INestApplication } from '@nestjs/common';
-import { EventEmitterModule } from '@nestjs/event-emitter';
 import { Test, TestingModule } from '@nestjs/testing';
 import test from 'ava';
 import { register } from 'prom-client';
 import * as Sinon from 'sinon';
-import { Doc as YDoc, encodeStateAsUpdate } from 'yjs';
+import {
+  applyUpdate,
+  decodeStateVector,
+  Doc as YDoc,
+  encodeStateAsUpdate,
+} from 'yjs';
 
 import { CacheModule } from '../src/cache';
 import { Config, ConfigModule } from '../src/config';
+import { EventModule } from '../src/event';
 import { DocManager, DocModule } from '../src/modules/doc';
 import { PrismaModule, PrismaService } from '../src/prisma';
 import { flushDB } from './utils';
@@ -19,7 +24,7 @@ const createModule = () => {
     imports: [
       PrismaModule,
       CacheModule,
-      EventEmitterModule.forRoot(),
+      EventModule,
       ConfigModule.forRoot(),
       DocModule.forRoot(),
     ],
@@ -283,3 +288,73 @@ test('should throw if meet max retry times', async t => {
   );
   t.is(stub.callCount, 5);
 });
+
+test('should not update snapshot if state is outdated', async t => {
+  const db = m.get(PrismaService);
+  const manager = m.get(DocManager);
+
+  await db.snapshot.create({
+    data: {
+      id: '2',
+      workspaceId: '2',
+      blob: Buffer.from([0, 0]),
+      seq: 1,
+    },
+  });
+  const doc = new YDoc();
+  const text = doc.getText('content');
+  const updates: Buffer[] = [];
+
+  doc.on('update', update => {
+    updates.push(Buffer.from(update));
+  });
+
+  text.insert(0, 'hello');
+  text.insert(5, 'world');
+  text.insert(5, ' ');
+
+  await Promise.all(updates.map(update => manager.push('2', '2', update)));
+
+  const updateWith3Records = await manager.getUpdates('2', '2');
+  text.insert(11, '!');
+  await manager.push('2', '2', updates[3]);
+  const updateWith4Records = await manager.getUpdates('2', '2');
+
+  // Simulation:
+  //   Node A get 3 updates and squash them at time 1, will finish at time 10
+  //   Node B get 4 updates and squash them at time 3, will finish at time 8
+  //   Node B finish the squash first, and update the snapshot
+  //   Node A finish the squash later, and update the snapshot to an outdated state
+  // Time: ---------------------->
+  //    A:   ^get           ^upsert
+  //    B:      ^get    ^upsert
+  //
+  // We should avoid such situation
+  // @ts-expect-error private
+  await manager.squash(updateWith4Records, null);
+  // @ts-expect-error private
+  await manager.squash(updateWith3Records, null);
+
+  const result = await db.snapshot.findUnique({
+    where: {
+      id_workspaceId: {
+        id: '2',
+        workspaceId: '2',
+      },
+    },
+  });
+
+  if (!result) {
+    t.fail('snapshot not found');
+    return;
+  }
+
+  const state = decodeStateVector(result.state!);
+  t.is(state.get(doc.clientID), 12);
+
+  const d = new YDoc();
+  applyUpdate(d, result.blob!);
+
+  const dtext = d.getText('content');
+  t.is(dtext.toString(), 'hello world!');
+});

packages/backend/server/tests/history.spec.ts

@@ -6,6 +6,7 @@ import test from 'ava';
 import * as Sinon from 'sinon';
 
 import { ConfigModule } from '../src/config';
+import type { EventPayload } from '../src/event';
 import { DocHistoryManager } from '../src/modules/doc';
 import { PrismaModule, PrismaService } from '../src/prisma';
 import { flushDB } from './utils';
@@ -41,21 +42,28 @@ test.afterEach(async () => {
 const snapshot: Snapshot = {
   workspaceId: '1',
   id: 'doc1',
-  blob: Buffer.from([0, 0]),
-  state: Buffer.from([0, 0]),
+  blob: Buffer.from([1, 0]),
+  state: Buffer.from([0]),
   seq: 0,
   updatedAt: new Date(),
   createdAt: new Date(),
 };
 
+function getEventData(
+  timestamp: Date = new Date()
+): EventPayload<'snapshot.updated'> {
+  return {
+    workspaceId: snapshot.workspaceId,
+    id: snapshot.id,
+    previous: { ...snapshot, updatedAt: timestamp },
+  };
+}
+
 test('should create doc history if never created before', async t => {
   Sinon.stub(manager, 'last').resolves(null);
 
   const timestamp = new Date();
-  await manager.onDocUpdated({
-    ...snapshot,
-    updatedAt: timestamp,
-  });
+  await manager.onDocUpdated(getEventData(timestamp));
 
   const history = await db.snapshotHistory.findFirst({
     where: {
@@ -72,10 +80,7 @@ test('should not create history if timestamp equals to last record', async t =>
   const timestamp = new Date();
   Sinon.stub(manager, 'last').resolves({ timestamp, state: null });
 
-  await manager.onDocUpdated({
-    ...snapshot,
-    updatedAt: timestamp,
-  });
+  await manager.onDocUpdated(getEventData(timestamp));
 
   const history = await db.snapshotHistory.findFirst({
     where: {
@@ -94,10 +99,7 @@ test('should not create history if state equals to last record', async t => {
     state: snapshot.state,
   });
 
-  await manager.onDocUpdated({
-    ...snapshot,
-    updatedAt: timestamp,
-  });
+  await manager.onDocUpdated(getEventData(timestamp));
 
   const history = await db.snapshotHistory.findFirst({
     where: {
@@ -116,10 +118,7 @@ test('should not create history if time diff is less than interval config', asyn
     state: Buffer.from([0, 1]),
   });
 
-  await manager.onDocUpdated({
-    ...snapshot,
-    updatedAt: timestamp,
-  });
+  await manager.onDocUpdated(getEventData(timestamp));
 
   const history = await db.snapshotHistory.findFirst({
     where: {
@@ -138,10 +137,7 @@ test('should create history if time diff is larger than interval config and stat
     state: Buffer.from([0, 1]),
   });
 
-  await manager.onDocUpdated({
-    ...snapshot,
-    updatedAt: timestamp,
-  });
+  await manager.onDocUpdated(getEventData(timestamp));
 
   const history = await db.snapshotHistory.findFirst({
     where: {
@@ -160,13 +156,7 @@ test('should create history with force flag even if time diff in small', async t
     state: Buffer.from([0, 1]),
   });
 
-  await manager.onDocUpdated(
-    {
-      ...snapshot,
-      updatedAt: timestamp,
-    },
-    true
-  );
+  await manager.onDocUpdated(getEventData(timestamp), true);
 
   const history = await db.snapshotHistory.findFirst({
     where: {
@@ -224,13 +214,7 @@ test('should correctly list all history records', async t => {
 test('should be able to get history data', async t => {
   const timestamp = new Date();
 
-  await manager.onDocUpdated(
-    {
-      ...snapshot,
-      updatedAt: timestamp,
-    },
-    true
-  );
+  await manager.onDocUpdated(getEventData(timestamp), true);
 
   const history = await manager.get(
     snapshot.workspaceId,
@@ -274,10 +258,7 @@ test('should be able to recover from history', async t => {
     },
   });
   const history1Timestamp = snapshot.updatedAt.getTime() - 10;
-  await manager.onDocUpdated({
-    ...snapshot,
-    updatedAt: new Date(history1Timestamp),
-  });
+  await manager.onDocUpdated(getEventData(new Date(history1Timestamp)));
 
   await manager.recover(
     snapshot.workspaceId,