feat(server): impl doc history (#5004)

packages/backend/server/migrations/20231121033532_history/migration.sql

@@ -5,10 +5,10 @@ ALTER TABLE "blobs" ADD COLUMN "deleted_at" TIMESTAMPTZ(6);
 CREATE TABLE "snapshot_histories" (
     "workspace_id" VARCHAR(36) NOT NULL,
     "guid" VARCHAR(36) NOT NULL,
-    "seq" INTEGER NOT NULL,
+    "timestamp" TIMESTAMPTZ(6) NOT NULL,
     "blob" BYTEA NOT NULL,
     "state" BYTEA,
-    "created_at" TIMESTAMPTZ(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "expired_at" TIMESTAMPTZ(6) NOT NULL,
 
-    CONSTRAINT "snapshot_histories_pkey" PRIMARY KEY ("workspace_id","guid","seq")
+    CONSTRAINT "snapshot_histories_pkey" PRIMARY KEY ("workspace_id","guid","timestamp")
 );

packages/backend/server/package.json

@@ -31,6 +31,7 @@
     "@nestjs/graphql": "^12.0.9",
     "@nestjs/platform-express": "^10.2.7",
     "@nestjs/platform-socket.io": "^10.2.7",
+    "@nestjs/schedule": "^4.0.0",
     "@nestjs/throttler": "^5.0.0",
     "@nestjs/websockets": "^10.2.7",
     "@node-rs/argon2": "^1.5.2",

packages/backend/server/schema.prisma

@@ -219,12 +219,12 @@ model Update {
 model SnapshotHistory {
   workspaceId String   @map("workspace_id") @db.VarChar(36)
   id          String   @map("guid") @db.VarChar(36)
-  seq         Int      @db.Integer
+  timestamp   DateTime @db.Timestamptz(6)
   blob        Bytes    @db.ByteA
   state       Bytes?   @db.ByteA
-  createdAt   DateTime @default(now()) @map("created_at") @db.Timestamptz(6)
+  expiredAt   DateTime @map("expired_at") @db.Timestamptz(6)
 
-  @@id([workspaceId, id, seq])
+  @@id([workspaceId, id, timestamp])
   @@map("snapshot_histories")
 }
 

packages/backend/server/src/config/def.ts

@@ -362,6 +362,14 @@ export interface AFFiNEConfig {
        */
       experimentalMergeWithJwstCodec: boolean;
     };
+    history: {
+      /**
+       * How long the buffer time of creating a new history snapshot when doc get updated.
+       *
+       * in {ms}
+       */
+      interval: number;
+    };
   };
 
   payment: {

packages/backend/server/src/config/default.ts

@@ -209,6 +209,9 @@ export const getDefaultAFFiNEConfig: () => AFFiNEConfig = () => {
         updatePollInterval: 3000,
         experimentalMergeWithJwstCodec: false,
       },
+      history: {
+        interval: 1000 * 60 * 10 /* 10 mins */,
+      },
     },
     payment: {
       stripe: {

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

@@ -10,3 +10,4 @@ import { Metrics } from './metrics';
   controllers: [MetricsController],
 })
 export class MetricsModule {}
+export { Metrics };

packages/backend/server/src/metrics/metrics.ts

@@ -25,4 +25,7 @@ export class Metrics implements OnModuleDestroy {
 
   authCounter = metricsCreator.counter('auth');
   authFailCounter = metricsCreator.counter('auth_fail', ['reason']);
+
+  docHistoryCounter = metricsCreator.counter('doc_history_created');
+  docRecoverCounter = metricsCreator.counter('doc_history_recovered');
 }

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

@@ -0,0 +1,230 @@
+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 { Metrics } from '../../metrics';
+import { PrismaService } from '../../prisma';
+import { SubscriptionStatus } from '../payment/service';
+import { Permission } from '../workspaces/types';
+
+@Injectable()
+export class DocHistoryManager {
+  private readonly logger = new Logger(DocHistoryManager.name);
+  constructor(
+    private readonly config: Config,
+    private readonly db: PrismaService,
+    private readonly metrics: Metrics
+  ) {}
+
+  @OnEvent('doc:manager:snapshot:beforeUpdate')
+  async onDocUpdated(snapshot: Snapshot, forceCreate = false) {
+    const last = await this.last(snapshot.workspaceId, snapshot.id);
+
+    let shouldCreateHistory = false;
+
+    if (!last) {
+      // never created
+      shouldCreateHistory = true;
+    } else if (last.timestamp === snapshot.updatedAt) {
+      // no change
+      shouldCreateHistory = false;
+    } else if (
+      // force
+      forceCreate ||
+      // last history created before interval in configs
+      last.timestamp.getTime() <
+        snapshot.updatedAt.getTime() - this.config.doc.history.interval
+    ) {
+      shouldCreateHistory = true;
+    }
+
+    if (shouldCreateHistory) {
+      await this.db.snapshotHistory
+        .create({
+          select: {
+            timestamp: true,
+          },
+          data: {
+            workspaceId: snapshot.workspaceId,
+            id: snapshot.id,
+            timestamp: snapshot.updatedAt,
+            blob: snapshot.blob,
+            state: snapshot.state,
+            expiredAt: await this.getExpiredDateFromNow(snapshot.workspaceId),
+          },
+        })
+        .catch(() => {
+          // safe to ignore
+          // only happens when duplicated history record created in multi processes
+        });
+      this.metrics.docHistoryCounter(1, {});
+      this.logger.log(
+        `History created for ${snapshot.id} in workspace ${snapshot.workspaceId}.`
+      );
+    }
+  }
+
+  async list(
+    workspaceId: string,
+    id: string,
+    before: Date = new Date(),
+    take: number = 10
+  ) {
+    return this.db.snapshotHistory.findMany({
+      select: {
+        timestamp: true,
+      },
+      where: {
+        workspaceId,
+        id,
+        timestamp: {
+          lte: before,
+        },
+        // only include the ones has not expired
+        expiredAt: {
+          gt: new Date(),
+        },
+      },
+      orderBy: {
+        timestamp: 'desc',
+      },
+      take,
+    });
+  }
+
+  async count(workspaceId: string, id: string) {
+    return this.db.snapshotHistory.count({
+      where: {
+        workspaceId,
+        id,
+        expiredAt: {
+          gt: new Date(),
+        },
+      },
+    });
+  }
+
+  async get(workspaceId: string, id: string, timestamp: Date) {
+    return this.db.snapshotHistory.findUnique({
+      where: {
+        workspaceId_id_timestamp: {
+          workspaceId,
+          id,
+          timestamp,
+        },
+        expiredAt: {
+          gt: new Date(),
+        },
+      },
+    });
+  }
+
+  async last(workspaceId: string, id: string) {
+    return this.db.snapshotHistory.findFirst({
+      where: {
+        workspaceId,
+        id,
+      },
+      select: {
+        timestamp: true,
+      },
+      orderBy: {
+        timestamp: 'desc',
+      },
+    });
+  }
+
+  async recover(workspaceId: string, id: string, timestamp: Date) {
+    const history = await this.db.snapshotHistory.findUnique({
+      where: {
+        workspaceId_id_timestamp: {
+          workspaceId,
+          id,
+          timestamp,
+        },
+      },
+    });
+
+    if (!history) {
+      throw new Error('Given history not found');
+    }
+
+    const oldSnapshot = await this.db.snapshot.findUnique({
+      where: {
+        id_workspaceId: {
+          id,
+          workspaceId,
+        },
+      },
+    });
+
+    if (!oldSnapshot) {
+      // unreachable actually
+      throw new Error('Given Doc not found');
+    }
+
+    // save old snapshot as one history record
+    await this.onDocUpdated(oldSnapshot, true);
+    // WARN:
+    //  we should never do the snapshot updating in recovering,
+    //  which is not the solution in CRDT.
+    //  let user revert in client and update the data in sync system
+    //    `await this.db.snapshot.update();`
+    this.metrics.docRecoverCounter(1, {});
+
+    return history.timestamp;
+  }
+
+  /**
+   * @todo(@darkskygit) refactor with [Usage Control] system
+   */
+  async getExpiredDateFromNow(workspaceId: string) {
+    const permission = await this.db.workspaceUserPermission.findFirst({
+      select: {
+        userId: true,
+      },
+      where: {
+        workspaceId,
+        type: Permission.Owner,
+      },
+    });
+
+    if (!permission) {
+      // unreachable actually
+      throw new Error('Workspace owner not found');
+    }
+
+    const sub = await this.db.userSubscription.findFirst({
+      select: {
+        id: true,
+      },
+      where: {
+        userId: permission.userId,
+        status: SubscriptionStatus.Active,
+      },
+    });
+
+    return new Date(
+      Date.now() +
+        1000 *
+          60 *
+          60 *
+          24 *
+          // 30 days for subscription user, 7 days for free user
+          (sub ? 30 : 7)
+    );
+  }
+
+  @Cron(CronExpression.EVERY_DAY_AT_MIDNIGHT /* everyday at 12am */)
+  async cleanupExpiredHistory() {
+    await this.db.snapshotHistory.deleteMany({
+      where: {
+        expiredAt: {
+          lte: new Date(),
+        },
+      },
+    });
+  }
+}

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

@@ -1,5 +1,6 @@
 import { DynamicModule } from '@nestjs/common';
 
+import { DocHistoryManager } from './history';
 import { DocManager } from './manager';
 
 export class DocModule {
@@ -14,12 +15,10 @@ export class DocModule {
           provide: 'DOC_MANAGER_AUTOMATION',
           useValue: automation,
         },
-        {
+        DocManager,
-          provide: DocManager,
+        DocHistoryManager,
-          useClass: DocManager,
-        },
       ],
-      exports: [DocManager],
+      exports: [DocManager, DocHistoryManager],
     };
   }
 
@@ -36,4 +35,4 @@ export class DocModule {
   }
 }
 
-export { DocManager };
+export { DocHistoryManager, DocManager };

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

@@ -1,5 +1,6 @@
 import { DynamicModule, Type } from '@nestjs/common';
 import { EventEmitterModule } from '@nestjs/event-emitter';
+import { ScheduleModule } from '@nestjs/schedule';
 
 import { GqlModule } from '../graphql.module';
 import { AuthModule } from './auth';
@@ -23,6 +24,7 @@ switch (SERVER_FLAVOR) {
     break;
   case 'graphql':
     BusinessModules.push(
+      ScheduleModule.forRoot(),
       GqlModule,
       WorkspaceModule,
       UsersModule,
@@ -34,6 +36,7 @@ switch (SERVER_FLAVOR) {
   case 'allinone':
   default:
     BusinessModules.push(
+      ScheduleModule.forRoot(),
       GqlModule,
       WorkspaceModule,
       UsersModule,

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

@@ -16,9 +16,10 @@ import { PrismaService } from '../../prisma';
 import { StorageProvide } from '../../storage';
 import { DocID } from '../../utils/doc';
 import { Auth, CurrentUser, Publicable } from '../auth';
-import { DocManager } from '../doc';
+import { DocHistoryManager, DocManager } from '../doc';
 import { UserType } from '../users';
 import { PermissionService, PublicPageMode } from './permission';
+import { Permission } from './types';
 
 @Controller('/api/workspaces')
 export class WorkspacesController {
@@ -28,6 +29,7 @@ export class WorkspacesController {
     @Inject(StorageProvide) private readonly storage: Storage,
     private readonly permission: PermissionService,
     private readonly docManager: DocManager,
+    private readonly historyManager: DocHistoryManager,
     private readonly prisma: PrismaService
   ) {}
 
@@ -104,4 +106,47 @@ export class WorkspacesController {
     res.send(update);
     this.logger.debug(`workspaces doc api: ${format(process.hrtime(start))}`);
   }
+
+  @Get('/:id/docs/:guid/histories/:timestamp')
+  @Auth()
+  async history(
+    @CurrentUser() user: UserType,
+    @Param('id') ws: string,
+    @Param('guid') guid: string,
+    @Param('timestamp') timestamp: string,
+    @Res() res: Response
+  ) {
+    const docId = new DocID(guid, ws);
+    let ts;
+    try {
+      const timeNum = parseInt(timestamp);
+      if (Number.isNaN(timeNum)) {
+        throw new Error('Invalid timestamp');
+      }
+
+      ts = new Date(timeNum);
+    } catch (e) {
+      throw new Error('Invalid timestamp');
+    }
+
+    await this.permission.checkPagePermission(
+      docId.workspace,
+      docId.guid,
+      user.id,
+      Permission.Write
+    );
+
+    const history = await this.historyManager.get(
+      docId.workspace,
+      docId.guid,
+      ts
+    );
+
+    if (history) {
+      res.setHeader('content-type', 'application/octet-stream');
+      res.send(history.blob);
+    } else {
+      throw new NotFoundException('Doc history not found');
+    }
+  }
 }

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

@@ -0,0 +1,92 @@
+import {
+  Args,
+  Field,
+  GraphQLISODateTime,
+  Int,
+  Mutation,
+  ObjectType,
+  Parent,
+  ResolveField,
+  Resolver,
+} from '@nestjs/graphql';
+import type { SnapshotHistory } from '@prisma/client';
+
+import { DocID } from '../../utils/doc';
+import { Auth, CurrentUser } from '../auth';
+import { DocHistoryManager } from '../doc/history';
+import { UserType } from '../users';
+import { PermissionService } from './permission';
+import { WorkspaceType } from './resolver';
+import { Permission } from './types';
+
+@ObjectType()
+class DocHistoryType implements Partial<SnapshotHistory> {
+  @Field()
+  workspaceId!: string;
+
+  @Field()
+  id!: string;
+
+  @Field(() => GraphQLISODateTime)
+  timestamp!: Date;
+}
+
+@Resolver(() => WorkspaceType)
+export class DocHistoryResolver {
+  constructor(
+    private readonly historyManager: DocHistoryManager,
+    private readonly permission: PermissionService
+  ) {}
+
+  @ResolveField(() => [DocHistoryType])
+  async histories(
+    @Parent() workspace: WorkspaceType,
+    @Args('guid') guid: string,
+    @Args({ name: 'before', type: () => GraphQLISODateTime, nullable: true })
+    timestamp: Date = new Date(),
+    @Args({ name: 'take', type: () => Int, nullable: true })
+    take?: number
+  ): Promise<DocHistoryType[]> {
+    const docId = new DocID(guid, workspace.id);
+
+    if (docId.isWorkspace) {
+      throw new Error('Invalid guid for listing doc histories.');
+    }
+
+    return this.historyManager
+      .list(workspace.id, docId.guid, timestamp, take)
+      .then(rows =>
+        rows.map(({ timestamp }) => {
+          return {
+            workspaceId: workspace.id,
+            id: docId.guid,
+            timestamp,
+          };
+        })
+      );
+  }
+
+  @Auth()
+  @Mutation(() => Date)
+  async recoverDoc(
+    @CurrentUser() user: UserType,
+    @Args('workspaceId') workspaceId: string,
+    @Args('guid') guid: string,
+    @Args({ name: 'timestamp', type: () => GraphQLISODateTime }) timestamp: Date
+  ): Promise<Date> {
+    const docId = new DocID(guid, workspaceId);
+
+    if (docId.isWorkspace) {
+      throw new Error('Invalid guid for recovering doc from history.');
+    }
+
+    await this.permission.checkPagePermission(
+      docId.workspace,
+      docId.guid,
+      user.id,
+      Permission.Write
+    );
+
+    return this.historyManager.recover(docId.workspace, docId.guid, timestamp);
+  }
+}

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

@@ -3,6 +3,7 @@ import { Module } from '@nestjs/common';
 import { DocModule } from '../doc';
 import { UsersService } from '../users';
 import { WorkspacesController } from './controller';
+import { DocHistoryResolver } from './history.resolver';
 import { PermissionService } from './permission';
 import { PagePermissionResolver, WorkspaceResolver } from './resolver';
 
@@ -14,6 +15,7 @@ import { PagePermissionResolver, WorkspaceResolver } from './resolver';
     PermissionService,
     UsersService,
     PagePermissionResolver,
+    DocHistoryResolver,
   ],
   exports: [PermissionService],
 })

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

@@ -244,6 +244,7 @@ export class PermissionService {
     permission = Permission.Read
   ) {
     // check whether page is public
+    if (permission === Permission.Read) {
       const count = await this.prisma.workspacePage.count({
         where: {
           workspaceId: ws,
@@ -257,6 +258,7 @@ export class PermissionService {
       if (count > 0) {
         return true;
       }
+    }
 
     if (user) {
       const count = await this.prisma.workspacePageUserPermission.count({

packages/backend/server/src/schema.gql

@@ -192,6 +192,7 @@ type WorkspaceType {
 
   """Public pages of a workspace"""
   publicPages: [WorkspacePage!]!
+  histories(guid: String!, before: DateTime, take: Int): [DocHistoryType!]!
 }
 
 type InvitationWorkspaceType {
@@ -232,6 +233,12 @@ enum PublicPageMode {
   Edgeless
 }
 
+type DocHistoryType {
+  workspaceId: String!
+  id: String!
+  timestamp: DateTime!
+}
+
 type Query {
   """Get is owner of workspace"""
   isOwner(workspaceId: String!): Boolean!
@@ -288,6 +295,7 @@ type Mutation {
   publishPage(workspaceId: String!, pageId: String!, mode: PublicPageMode = Page): WorkspacePage!
   revokePage(workspaceId: String!, pageId: String!): Boolean! @deprecated(reason: "use revokePublicPage")
   revokePublicPage(workspaceId: String!, pageId: String!): WorkspacePage!
+  recoverDoc(workspaceId: String!, guid: String!, timestamp: DateTime!): DateTime!
 
   """Upload user avatar"""
   uploadAvatar(avatar: Upload!): UserType!

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

@@ -0,0 +1,312 @@
+import { INestApplication } from '@nestjs/common';
+import { ScheduleModule } from '@nestjs/schedule';
+import { Test, TestingModule } from '@nestjs/testing';
+import type { Snapshot } from '@prisma/client';
+import test from 'ava';
+import * as Sinon from 'sinon';
+
+import { ConfigModule } from '../src/config';
+import { MetricsModule } from '../src/metrics';
+import { DocHistoryManager } from '../src/modules/doc';
+import { PrismaModule, PrismaService } from '../src/prisma';
+import { flushDB } from './utils';
+
+let app: INestApplication;
+let m: TestingModule;
+let manager: DocHistoryManager;
+let db: PrismaService;
+
+// cleanup database before each test
+test.beforeEach(async () => {
+  await flushDB();
+  m = await Test.createTestingModule({
+    imports: [
+      PrismaModule,
+      MetricsModule,
+      ScheduleModule.forRoot(),
+      ConfigModule.forRoot(),
+    ],
+    providers: [DocHistoryManager],
+  }).compile();
+
+  app = m.createNestApplication();
+  await app.init();
+  manager = m.get(DocHistoryManager);
+  Sinon.stub(manager, 'getExpiredDateFromNow').resolves(
+    new Date(Date.now() + 1000)
+  );
+  db = m.get(PrismaService);
+});
+
+test.afterEach(async () => {
+  await app.close();
+  await m.close();
+  Sinon.restore();
+});
+
+const snapshot: Snapshot = {
+  workspaceId: '1',
+  id: 'doc1',
+  blob: Buffer.from([0, 0]),
+  state: Buffer.from([0, 0]),
+  seq: 0,
+  updatedAt: new Date(),
+  createdAt: new Date(),
+};
+
+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,
+  });
+
+  const history = await db.snapshotHistory.findFirst({
+    where: {
+      workspaceId: '1',
+      id: 'doc1',
+    },
+  });
+
+  t.truthy(history);
+  t.is(history?.timestamp.getTime(), timestamp.getTime());
+});
+
+test('should not create history is timestamp equals to last record', async t => {
+  const timestamp = new Date();
+  Sinon.stub(manager, 'last').resolves({ timestamp });
+
+  await manager.onDocUpdated({
+    ...snapshot,
+    updatedAt: timestamp,
+  });
+
+  const history = await db.snapshotHistory.findFirst({
+    where: {
+      workspaceId: '1',
+      id: 'doc1',
+    },
+  });
+
+  t.falsy(history);
+});
+
+test('should create history if time diff is larger than interval config', async t => {
+  const timestamp = new Date();
+  Sinon.stub(manager, 'last').resolves({
+    timestamp: new Date(timestamp.getTime() - 1000 * 60 * 20),
+  });
+
+  await manager.onDocUpdated({
+    ...snapshot,
+    updatedAt: timestamp,
+  });
+
+  const history = await db.snapshotHistory.findFirst({
+    where: {
+      workspaceId: '1',
+      id: 'doc1',
+    },
+  });
+
+  t.truthy(history);
+});
+
+test('should not create history if time diff is less than interval config', async t => {
+  const timestamp = new Date();
+  Sinon.stub(manager, 'last').resolves({
+    timestamp: new Date(timestamp.getTime() - 1000),
+  });
+
+  await manager.onDocUpdated({
+    ...snapshot,
+    updatedAt: timestamp,
+  });
+
+  const history = await db.snapshotHistory.findFirst({
+    where: {
+      workspaceId: '1',
+      id: 'doc1',
+    },
+  });
+
+  t.falsy(history);
+});
+
+test('should create history with force flag even if time diff in small', async t => {
+  const timestamp = new Date();
+  Sinon.stub(manager, 'last').resolves({
+    timestamp: new Date(timestamp.getTime() - 1),
+  });
+
+  await manager.onDocUpdated(
+    {
+      ...snapshot,
+      updatedAt: timestamp,
+    },
+    true
+  );
+
+  const history = await db.snapshotHistory.findFirst({
+    where: {
+      workspaceId: '1',
+      id: 'doc1',
+    },
+  });
+
+  t.truthy(history);
+});
+
+test('should correctly list all history records', async t => {
+  const timestamp = Date.now();
+
+  // insert expired data
+  await db.snapshotHistory.createMany({
+    data: new Array(10).fill(0).map((_, i) => ({
+      workspaceId: snapshot.workspaceId,
+      id: snapshot.id,
+      blob: snapshot.blob,
+      state: snapshot.state,
+      timestamp: new Date(timestamp - 10 - i),
+      expiredAt: new Date(timestamp - 1),
+    })),
+  });
+
+  // insert available data
+  await db.snapshotHistory.createMany({
+    data: new Array(10).fill(0).map((_, i) => ({
+      workspaceId: snapshot.workspaceId,
+      id: snapshot.id,
+      blob: snapshot.blob,
+      state: snapshot.state,
+      timestamp: new Date(timestamp + i),
+      expiredAt: new Date(timestamp + 1000),
+    })),
+  });
+
+  const list = await manager.list(
+    snapshot.workspaceId,
+    snapshot.id,
+    new Date(timestamp + 20),
+    8
+  );
+  const count = await manager.count(snapshot.workspaceId, snapshot.id);
+
+  t.is(list.length, 8);
+  t.is(count, 10);
+});
+
+test('should be able to get history data', async t => {
+  const timestamp = new Date();
+
+  await manager.onDocUpdated(
+    {
+      ...snapshot,
+      updatedAt: timestamp,
+    },
+    true
+  );
+
+  const history = await manager.get(
+    snapshot.workspaceId,
+    snapshot.id,
+    timestamp
+  );
+
+  t.truthy(history);
+  t.deepEqual(history?.blob, snapshot.blob);
+});
+
+test('should be able to get last history record', async t => {
+  const timestamp = Date.now();
+
+  // insert available data
+  await db.snapshotHistory.createMany({
+    data: new Array(10).fill(0).map((_, i) => ({
+      workspaceId: snapshot.workspaceId,
+      id: snapshot.id,
+      blob: snapshot.blob,
+      state: snapshot.state,
+      timestamp: new Date(timestamp + i),
+      expiredAt: new Date(timestamp + 1000),
+    })),
+  });
+
+  const history = await manager.last(snapshot.workspaceId, snapshot.id);
+
+  t.truthy(history);
+  t.is(history?.timestamp.getTime(), timestamp + 9);
+});
+
+test('should be able to recover from history', async t => {
+  await db.snapshot.create({
+    data: {
+      ...snapshot,
+      blob: Buffer.from([1, 1]),
+      state: Buffer.from([1, 1]),
+    },
+  });
+  const history1Timestamp = snapshot.updatedAt.getTime() - 10;
+  await manager.onDocUpdated({
+    ...snapshot,
+    updatedAt: new Date(history1Timestamp),
+  });
+
+  await manager.recover(
+    snapshot.workspaceId,
+    snapshot.id,
+    new Date(history1Timestamp)
+  );
+
+  const [history1, history2] = await db.snapshotHistory.findMany({
+    where: {
+      workspaceId: snapshot.workspaceId,
+      id: snapshot.id,
+    },
+  });
+
+  t.is(history1.timestamp.getTime(), history1Timestamp);
+  t.is(history2.timestamp.getTime(), snapshot.updatedAt.getTime());
+
+  // new history data force created with snapshot state before recovered
+  t.deepEqual(history2?.blob, Buffer.from([1, 1]));
+  t.deepEqual(history2?.state, Buffer.from([1, 1]));
+});
+
+test('should be able to cleanup expired history', async t => {
+  const timestamp = Date.now();
+
+  // insert expired data
+  await db.snapshotHistory.createMany({
+    data: new Array(10).fill(0).map((_, i) => ({
+      workspaceId: snapshot.workspaceId,
+      id: snapshot.id,
+      blob: snapshot.blob,
+      state: snapshot.state,
+      timestamp: new Date(timestamp - 10 - i),
+      expiredAt: new Date(timestamp - 1),
+    })),
+  });
+
+  // insert available data
+  await db.snapshotHistory.createMany({
+    data: new Array(10).fill(0).map((_, i) => ({
+      workspaceId: snapshot.workspaceId,
+      id: snapshot.id,
+      blob: snapshot.blob,
+      state: snapshot.state,
+      timestamp: new Date(timestamp + i),
+      expiredAt: new Date(timestamp + 1000),
+    })),
+  });
+
+  let count = await db.snapshotHistory.count();
+  t.is(count, 20);
+
+  await manager.cleanupExpiredHistory();
+
+  count = await db.snapshotHistory.count();
+  t.is(count, 10);
+});

yarn.lock

@@ -719,6 +719,7 @@ __metadata:
     "@nestjs/graphql": "npm:^12.0.9"
     "@nestjs/platform-express": "npm:^10.2.7"
     "@nestjs/platform-socket.io": "npm:^10.2.7"
+    "@nestjs/schedule": "npm:^4.0.0"
     "@nestjs/testing": "npm:^10.2.7"
     "@nestjs/throttler": "npm:^5.0.0"
     "@nestjs/websockets": "npm:^10.2.7"
@@ -7340,6 +7341,20 @@ __metadata:
   languageName: node
   linkType: hard
 
+"@nestjs/schedule@npm:^4.0.0":
+  version: 4.0.0
+  resolution: "@nestjs/schedule@npm:4.0.0"
+  dependencies:
+    cron: "npm:3.1.3"
+    uuid: "npm:9.0.1"
+  peerDependencies:
+    "@nestjs/common": ^8.0.0 || ^9.0.0 || ^10.0.0
+    "@nestjs/core": ^8.0.0 || ^9.0.0 || ^10.0.0
+    reflect-metadata: ^0.1.12
+  checksum: 85598ef37e80b6dd511ae12a5ad9016ef223b4d1e3a5369d9226cc7449c9b1991049ad79b6cc76615da9f15f87de078200344a7afacf8d8fd0af05ea529cca11
+  languageName: node
+  linkType: hard
+
 "@nestjs/testing@npm:^10.2.7":
   version: 10.2.7
   resolution: "@nestjs/testing@npm:10.2.7"
@@ -13235,6 +13250,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"@types/luxon@npm:~3.3.0":
+  version: 3.3.5
+  resolution: "@types/luxon@npm:3.3.5"
+  checksum: be2aede1787f437e0ec3e2d1b964c5831fed1838d10cc60d824f814d0c0659dfa8874ffa81bec116004845279bdee2e5127046bb4fd64dc71cce8c0c25f6c25f
+  languageName: node
+  linkType: hard
+
 "@types/marked@npm:^6.0.0":
   version: 6.0.0
   resolution: "@types/marked@npm:6.0.0"
@@ -17625,6 +17647,16 @@ __metadata:
   languageName: node
   linkType: hard
 
+"cron@npm:3.1.3":
+  version: 3.1.3
+  resolution: "cron@npm:3.1.3"
+  dependencies:
+    "@types/luxon": "npm:~3.3.0"
+    luxon: "npm:~3.4.0"
+  checksum: 1cf7c9176c380239af093943ba72bee631bac561e4e02bae137c7508cb07e2fff93e18ef7ed2003f469762794e01d98616af95e1e5df900724171db30fa9299b
+  languageName: node
+  linkType: hard
+
 "cross-env@npm:^7.0.3":
   version: 7.0.3
   resolution: "cross-env@npm:7.0.3"
@@ -25809,6 +25841,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"luxon@npm:~3.4.0":
+  version: 3.4.4
+  resolution: "luxon@npm:3.4.4"
+  checksum: c14164bc338987349075a08e63ea3ff902866735f7f5553a355b27be22667919765ff96fde4d3413d0e9a0edc4ff9e2e74ebcb8f86eae0ce8b14b27330d87d6e
+  languageName: node
+  linkType: hard
+
 "lz-string@npm:^1.5.0":
   version: 1.5.0
   resolution: "lz-string@npm:1.5.0"
@@ -34327,6 +34366,15 @@ __metadata:
   languageName: node
   linkType: hard
 
+"uuid@npm:9.0.1, uuid@npm:^9.0.0, uuid@npm:^9.0.1":
+  version: 9.0.1
+  resolution: "uuid@npm:9.0.1"
+  bin:
+    uuid: dist/bin/uuid
+  checksum: 9d0b6adb72b736e36f2b1b53da0d559125ba3e39d913b6072f6f033e0c87835b414f0836b45bcfaf2bdf698f92297fea1c3cc19b0b258bc182c9c43cc0fab9f2
+  languageName: node
+  linkType: hard
+
 "uuid@npm:^8.0.0, uuid@npm:^8.3.2":
   version: 8.3.2
   resolution: "uuid@npm:8.3.2"
@@ -34336,15 +34384,6 @@ __metadata:
   languageName: node
   linkType: hard
 
-"uuid@npm:^9.0.0, uuid@npm:^9.0.1":
-  version: 9.0.1
-  resolution: "uuid@npm:9.0.1"
-  bin:
-    uuid: dist/bin/uuid
-  checksum: 9d0b6adb72b736e36f2b1b53da0d559125ba3e39d913b6072f6f033e0c87835b414f0836b45bcfaf2bdf698f92297fea1c3cc19b0b258bc182c9c43cc0fab9f2
-  languageName: node
-  linkType: hard
-
 "v8-compile-cache-lib@npm:^3.0.1":
   version: 3.0.1
   resolution: "v8-compile-cache-lib@npm:3.0.1"