feat: ack sync reset (#20703)

server/src/database.ts

@@ -202,7 +202,6 @@ export type Album = Selectable<AlbumTable> & {
 
 export type AuthSession = {
   id: string;
-  isPendingSyncReset: boolean;
   hasElevatedPermission: boolean;
 };
 
@@ -309,7 +308,7 @@ export const columns = {
   assetFiles: ['asset_file.id', 'asset_file.path', 'asset_file.type'],
   authUser: ['user.id', 'user.name', 'user.email', 'user.isAdmin', 'user.quotaUsageInBytes', 'user.quotaSizeInBytes'],
   authApiKey: ['api_key.id', 'api_key.permissions'],
-  authSession: ['session.id', 'session.isPendingSyncReset', 'session.updatedAt', 'session.pinExpiresAt'],
+  authSession: ['session.id', 'session.updatedAt', 'session.pinExpiresAt'],
   authSharedLink: [
     'shared_link.id',
     'shared_link.userId',

server/src/dtos/session.dto.ts

@@ -1,4 +1,4 @@
-import { IsInt, IsPositive, IsString } from 'class-validator';
+import { Equals, IsInt, IsPositive, IsString } from 'class-validator';
 import { Session } from 'src/database';
 import { Optional, ValidateBoolean } from 'src/validation';
 
@@ -22,7 +22,8 @@ export class SessionCreateDto {
 
 export class SessionUpdateDto {
   @ValidateBoolean({ optional: true })
-  isPendingSyncReset?: boolean;
+  @Equals(true)
+  isPendingSyncReset?: true;
 }
 
 export class SessionResponseDto {

server/src/queries/session.repository.sql

@@ -10,10 +10,17 @@ from
 where
   "id" = $1
 
+-- SessionRepository.isPendingSyncReset
+select
+  "isPendingSyncReset"
+from
+  "session"
+where
+  "id" = $1
+
 -- SessionRepository.getByToken
 select
   "session"."id",
-  "session"."isPendingSyncReset",
   "session"."updatedAt",
   "session"."pinExpiresAt",
   (

server/src/repositories/session.repository.ts

@@ -37,6 +37,16 @@ export class SessionRepository {
       .executeTakeFirst();
   }
 
+  @GenerateSql({ params: [DummyValue.UUID] })
+  async isPendingSyncReset(id: string) {
+    const result = await this.db
+      .selectFrom('session')
+      .select(['isPendingSyncReset'])
+      .where('id', '=', id)
+      .executeTakeFirst();
+    return result?.isPendingSyncReset ?? false;
+  }
+
   @GenerateSql({ params: [DummyValue.STRING] })
   getByToken(token: string) {
     return this.db

server/src/services/auth.service.spec.ts

@@ -241,7 +241,6 @@ describe(AuthService.name, () => {
       const sessionWithToken = {
         id: session.id,
         updatedAt: session.updatedAt,
-        isPendingSyncReset: false,
         user: factory.authUser(),
         pinExpiresAt: null,
       };
@@ -259,7 +258,6 @@ describe(AuthService.name, () => {
         session: {
           id: session.id,
           hasElevatedPermission: false,
-          isPendingSyncReset: session.isPendingSyncReset,
         },
       });
     });
@@ -409,7 +407,6 @@ describe(AuthService.name, () => {
         id: session.id,
         updatedAt: session.updatedAt,
         user: factory.authUser(),
-        isPendingSyncReset: false,
         pinExpiresAt: null,
       };
 
@@ -426,7 +423,6 @@ describe(AuthService.name, () => {
         session: {
           id: session.id,
           hasElevatedPermission: false,
-          isPendingSyncReset: session.isPendingSyncReset,
         },
       });
     });

server/src/services/auth.service.ts

@@ -487,7 +487,6 @@ export class AuthService extends BaseService {
         user: session.user,
         session: {
           id: session.id,
-          isPendingSyncReset: session.isPendingSyncReset,
           hasElevatedPermission,
         },
       };

server/src/services/sync.service.ts

@@ -23,7 +23,7 @@ import { SyncAck } from 'src/types';
 import { getMyPartnerIds } from 'src/utils/asset.util';
 import { hexOrBufferToBase64 } from 'src/utils/bytes';
 import { setIsEqual } from 'src/utils/set';
-import { fromAck, mapJsonLine, serialize, SerializeOptions, toAck } from 'src/utils/sync';
+import { fromAck, serialize, SerializeOptions, toAck } from 'src/utils/sync';
 
 type CheckpointMap = Partial<Record<SyncEntityType, SyncAck>>;
 type AssetLike = Omit<SyncAssetV1, 'checksum' | 'thumbhash'> & {
@@ -100,6 +100,10 @@ export class SyncService extends BaseService {
     const checkpoints: Record<string, Insertable<SessionSyncCheckpointTable>> = {};
     for (const ack of dto.acks) {
       const { type } = fromAck(ack);
+      if (type === SyncEntityType.SyncResetV1) {
+        await this.sessionRepository.resetSyncProgress(sessionId);
+        return;
+      }
       // TODO proper ack validation via class validator
       if (!Object.values(SyncEntityType).includes(type)) {
         throw new BadRequestException(`Invalid ack type: ${type}`);
@@ -129,11 +133,12 @@ export class SyncService extends BaseService {
 
     if (dto.reset) {
       await this.sessionRepository.resetSyncProgress(session.id);
-      session.isPendingSyncReset = false;
     }
 
-    if (session.isPendingSyncReset) {
-      response.write(mapJsonLine({ type: SyncEntityType.SyncResetV1, data: {} }));
+    const isPendingSyncReset = await this.sessionRepository.isPendingSyncReset(session.id);
+
+    if (isPendingSyncReset) {
+      send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
       response.end();
       return;
     }

server/test/medium/specs/sync/sync-reset.spec.ts

@@ -1,5 +1,6 @@
 import { Kysely } from 'kysely';
 import { SyncEntityType, SyncRequestType } from 'src/enum';
+import { SessionRepository } from 'src/repositories/session.repository';
 import { DB } from 'src/schema';
 import { SyncTestContext } from 'test/medium.factory';
 import { getKyselyDB } from 'test/utils';
@@ -27,10 +28,12 @@ describe(SyncEntityType.SyncResetV1, () => {
   it('should detect a pending sync reset', async () => {
     const { auth, ctx } = await setup();
 
-    auth.session!.isPendingSyncReset = true;
+    await ctx.get(SessionRepository).update(auth.session!.id, {
+      isPendingSyncReset: true,
+    });
 
     const response = await ctx.syncStream(auth, [SyncRequestType.AssetsV1]);
-    expect(response).toEqual([{ type: SyncEntityType.SyncResetV1, data: {} }]);
+    expect(response).toEqual([{ type: SyncEntityType.SyncResetV1, data: {}, ack: 'SyncResetV1|reset' }]);
   });
 
   it('should not send other dtos when a reset is pending', async () => {
@@ -40,10 +43,12 @@ describe(SyncEntityType.SyncResetV1, () => {
 
     await expect(ctx.syncStream(auth, [SyncRequestType.AssetsV1])).resolves.toHaveLength(1);
 
-    auth.session!.isPendingSyncReset = true;
+    await ctx.get(SessionRepository).update(auth.session!.id, {
+      isPendingSyncReset: true,
+    });
 
     await expect(ctx.syncStream(auth, [SyncRequestType.AssetsV1])).resolves.toEqual([
-      { type: SyncEntityType.SyncResetV1, data: {} },
+      { type: SyncEntityType.SyncResetV1, data: {}, ack: 'SyncResetV1|reset' },
     ]);
   });
 
@@ -52,7 +57,9 @@ describe(SyncEntityType.SyncResetV1, () => {
 
     await ctx.newAsset({ ownerId: user.id });
 
-    auth.session!.isPendingSyncReset = true;
+    await ctx.get(SessionRepository).update(auth.session!.id, {
+      isPendingSyncReset: true,
+    });
 
     await expect(ctx.syncStream(auth, [SyncRequestType.AssetsV1], true)).resolves.toEqual([
       expect.objectContaining({
@@ -60,4 +67,28 @@ describe(SyncEntityType.SyncResetV1, () => {
       }),
     ]);
   });
+
+  it('should reset the sync progress', async () => {
+    const { auth, user, ctx } = await setup();
+
+    await ctx.newAsset({ ownerId: user.id });
+
+    const response = await ctx.syncStream(auth, [SyncRequestType.AssetsV1]);
+    await ctx.syncAckAll(auth, response);
+
+    await ctx.get(SessionRepository).update(auth.session!.id, {
+      isPendingSyncReset: true,
+    });
+
+    const resetResponse = await ctx.syncStream(auth, [SyncRequestType.AssetsV1]);
+
+    await ctx.syncAckAll(auth, resetResponse);
+
+    const postResetResponse = await ctx.syncStream(auth, [SyncRequestType.AssetsV1]);
+    expect(postResetResponse).toEqual([
+      expect.objectContaining({
+        type: SyncEntityType.AssetV1,
+      }),
+    ]);
+  });
 });

server/test/small.factory.ts

@@ -60,7 +60,6 @@ const authFactory = ({
   if (session) {
     auth.session = {
       id: session.id,
-      isPendingSyncReset: false,
       hasElevatedPermission: false,
     };
   }