refactor: sync repository (#19581)

server/src/repositories/index.ts

@@ -34,6 +34,7 @@ import { SessionRepository } from 'src/repositories/session.repository';
 import { SharedLinkRepository } from 'src/repositories/shared-link.repository';
 import { StackRepository } from 'src/repositories/stack.repository';
 import { StorageRepository } from 'src/repositories/storage.repository';
+import { SyncCheckpointRepository } from 'src/repositories/sync-checkpoint.repository';
 import { SyncRepository } from 'src/repositories/sync.repository';
 import { SystemMetadataRepository } from 'src/repositories/system-metadata.repository';
 import { TagRepository } from 'src/repositories/tag.repository';
@@ -81,6 +82,7 @@ export const repositories = [
   StackRepository,
   StorageRepository,
   SyncRepository,
+  SyncCheckpointRepository,
   SystemMetadataRepository,
   TagRepository,
   TelemetryRepository,

server/src/repositories/sync-checkpoint.repository.ts

@@ -0,0 +1,41 @@
+import { Injectable } from '@nestjs/common';
+import { Insertable, Kysely } from 'kysely';
+import { InjectKysely } from 'nestjs-kysely';
+import { DB, SessionSyncCheckpoints } from 'src/db';
+import { DummyValue, GenerateSql } from 'src/decorators';
+import { SyncEntityType } from 'src/enum';
+
+@Injectable()
+export class SyncCheckpointRepository {
+  constructor(@InjectKysely() private db: Kysely<DB>) {}
+
+  @GenerateSql({ params: [DummyValue.UUID] })
+  getAll(sessionId: string) {
+    return this.db
+      .selectFrom('session_sync_checkpoints')
+      .select(['type', 'ack'])
+      .where('sessionId', '=', sessionId)
+      .execute();
+  }
+
+  upsertAll(items: Insertable<SessionSyncCheckpoints>[]) {
+    return this.db
+      .insertInto('session_sync_checkpoints')
+      .values(items)
+      .onConflict((oc) =>
+        oc.columns(['sessionId', 'type']).doUpdateSet((eb) => ({
+          ack: eb.ref('excluded.ack'),
+        })),
+      )
+      .execute();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID] })
+  deleteAll(sessionId: string, types?: SyncEntityType[]) {
+    return this.db
+      .deleteFrom('session_sync_checkpoints')
+      .where('sessionId', '=', sessionId)
+      .$if(!!types, (qb) => qb.where('type', 'in', types!))
+      .execute();
+  }
+}

server/src/repositories/sync.repository.ts

@@ -1,10 +1,9 @@
 import { Injectable } from '@nestjs/common';
-import { Insertable, Kysely, SelectQueryBuilder, sql } from 'kysely';
+import { Kysely, SelectQueryBuilder, sql } from 'kysely';
 import { InjectKysely } from 'nestjs-kysely';
 import { columns } from 'src/database';
-import { DB, SessionSyncCheckpoints } from 'src/db';
+import { DB } from 'src/db';
 import { DummyValue, GenerateSql } from 'src/decorators';
-import { SyncEntityType } from 'src/enum';
 import { SyncAck } from 'src/types';
 
 type AuditTables =
@@ -28,194 +27,78 @@ type UpsertTables =
 
 @Injectable()
 export class SyncRepository {
-  constructor(@InjectKysely() private db: Kysely<DB>) {}
+  album: AlbumSync;
+  albumAsset: AlbumAssetSync;
+  albumAssetExif: AlbumAssetExifSync;
+  albumToAsset: AlbumToAssetSync;
+  albumUser: AlbumUserSync;
+  asset: AssetSync;
+  assetExif: AssetExifSync;
+  memory: MemorySync;
+  memoryToAsset: MemoryToAssetSync;
+  partner: PartnerSync;
+  partnerAsset: PartnerAssetsSync;
+  partnerAssetExif: PartnerAssetExifsSync;
+  user: UserSync;
 
-  @GenerateSql({ params: [DummyValue.UUID] })
-  getCheckpoints(sessionId: string) {
-    return this.db
-      .selectFrom('session_sync_checkpoints')
-      .select(['type', 'ack'])
-      .where('sessionId', '=', sessionId)
-      .execute();
+  constructor(@InjectKysely() private db: Kysely<DB>) {
+    this.album = new AlbumSync(this.db);
+    this.albumAsset = new AlbumAssetSync(this.db);
+    this.albumAssetExif = new AlbumAssetExifSync(this.db);
+    this.albumToAsset = new AlbumToAssetSync(this.db);
+    this.albumUser = new AlbumUserSync(this.db);
+    this.asset = new AssetSync(this.db);
+    this.assetExif = new AssetExifSync(this.db);
+    this.memory = new MemorySync(this.db);
+    this.memoryToAsset = new MemoryToAssetSync(this.db);
+    this.partner = new PartnerSync(this.db);
+    this.partnerAsset = new PartnerAssetsSync(this.db);
+    this.partnerAssetExif = new PartnerAssetExifsSync(this.db);
+    this.user = new UserSync(this.db);
+  }
+}
+
+class BaseSync {
+  constructor(protected db: Kysely<DB>) {}
+
+  protected auditTableFilters<T extends keyof Pick<DB, AuditTables>, D>(
+    qb: SelectQueryBuilder<DB, T, D>,
+    ack?: SyncAck,
+  ) {
+    const builder = qb as SelectQueryBuilder<DB, AuditTables, D>;
+    return builder
+      .where('deletedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .$if(!!ack, (qb) => qb.where('id', '>', ack!.updateId))
+      .orderBy('id', 'asc') as SelectQueryBuilder<DB, T, D>;
   }
 
-  upsertCheckpoints(items: Insertable<SessionSyncCheckpoints>[]) {
-    return this.db
-      .insertInto('session_sync_checkpoints')
-      .values(items)
-      .onConflict((oc) =>
-        oc.columns(['sessionId', 'type']).doUpdateSet((eb) => ({
-          ack: eb.ref('excluded.ack'),
-        })),
-      )
-      .execute();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID] })
-  deleteCheckpoints(sessionId: string, types?: SyncEntityType[]) {
-    return this.db
-      .deleteFrom('session_sync_checkpoints')
-      .where('sessionId', '=', sessionId)
-      .$if(!!types, (qb) => qb.where('type', 'in', types!))
-      .execute();
-  }
-
-  @GenerateSql({ params: [], stream: true })
-  getUserUpserts(ack?: SyncAck) {
-    return this.db
-      .selectFrom('users')
-      .select(['id', 'name', 'email', 'deletedAt', 'updateId'])
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [], stream: true })
-  getUserDeletes(ack?: SyncAck) {
-    return this.db
-      .selectFrom('users_audit')
-      .select(['id', 'userId'])
-      .$call((qb) => this.auditTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getPartnerUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('partners')
-      .select(['sharedById', 'sharedWithId', 'inTimeline', 'updateId'])
-      .where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getPartnerDeletes(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('partners_audit')
-      .select(['id', 'sharedById', 'sharedWithId'])
-      .where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
-      .$call((qb) => this.auditTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAssetUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('assets')
-      .select(columns.syncAsset)
-      .select('assets.updateId')
-      .where('ownerId', '=', userId)
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
+  protected upsertTableFilters<T extends keyof Pick<DB, UpsertTables>, D>(
+    qb: SelectQueryBuilder<DB, T, D>,
+    ack?: SyncAck,
+  ) {
+    const builder = qb as SelectQueryBuilder<DB, UpsertTables, D>;
+    return builder
+      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .$if(!!ack, (qb) => qb.where('updateId', '>', ack!.updateId))
+      .orderBy('updateId', 'asc') as SelectQueryBuilder<DB, T, D>;
   }
+}
 
+class AlbumSync extends BaseSync {
   @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
-  getPartnerBackfill(userId: string, afterCreateId?: string) {
+  getCreatedAfter(userId: string, afterCreateId?: string) {
     return this.db
-      .selectFrom('partners')
-      .select(['sharedById', 'createId'])
-      .where('sharedWithId', '=', userId)
+      .selectFrom('albums_shared_users_users')
+      .select(['albumsId as id', 'createId'])
+      .where('usersId', '=', userId)
       .$if(!!afterCreateId, (qb) => qb.where('createId', '>=', afterCreateId!))
       .where('createdAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .orderBy('partners.createId', 'asc')
+      .orderBy('createId', 'asc')
       .execute();
   }
 
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
-  getPartnerAssetsBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
-    return this.db
-      .selectFrom('assets')
-      .select(columns.syncAsset)
-      .select('assets.updateId')
-      .where('ownerId', '=', partnerId)
-      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .where('updateId', '<=', beforeUpdateId)
-      .$if(!!afterUpdateId, (eb) => eb.where('updateId', '>=', afterUpdateId!))
-      .orderBy('updateId', 'asc')
-      .stream();
-  }
-
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getPartnerAssetsUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('assets')
-      .select(columns.syncAsset)
-      .select('assets.updateId')
-      .where('ownerId', 'in', (eb) =>
-        eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
-      )
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAssetDeletes(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('assets_audit')
-      .select(['id', 'assetId'])
-      .where('ownerId', '=', userId)
-      .$call((qb) => this.auditTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getPartnerAssetDeletes(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('assets_audit')
-      .select(['id', 'assetId'])
-      .where('ownerId', 'in', (eb) =>
-        eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
-      )
-      .$call((qb) => this.auditTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAssetExifsUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('exif')
-      .select(columns.syncAssetExif)
-      .select('exif.updateId')
-      .where('assetId', 'in', (eb) => eb.selectFrom('assets').select('id').where('ownerId', '=', userId))
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
-  getPartnerAssetExifsBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
-    return this.db
-      .selectFrom('exif')
-      .select(columns.syncAssetExif)
-      .select('exif.updateId')
-      .innerJoin('assets', 'assets.id', 'exif.assetId')
-      .where('assets.ownerId', '=', partnerId)
-      .where('exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .where('exif.updateId', '<=', beforeUpdateId)
-      .$if(!!afterUpdateId, (eb) => eb.where('exif.updateId', '>=', afterUpdateId!))
-      .orderBy('exif.updateId', 'asc')
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getPartnerAssetExifsUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('exif')
-      .select(columns.syncAssetExif)
-      .select('exif.updateId')
-      .where('assetId', 'in', (eb) =>
-        eb
-          .selectFrom('assets')
-          .select('id')
-          .where('ownerId', 'in', (eb) =>
-            eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
-          ),
-      )
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumDeletes(userId: string, ack?: SyncAck) {
+  getDeletes(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('albums_audit')
       .select(['id', 'albumId'])
@@ -225,7 +108,7 @@ export class SyncRepository {
   }
 
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumUpserts(userId: string, ack?: SyncAck) {
+  getUpserts(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('albums')
       .distinctOn(['albums.id', 'albums.updateId'])
@@ -248,9 +131,90 @@ export class SyncRepository {
       ])
       .stream();
   }
+}
+
+class AlbumAssetSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('assets')
+      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'assets.id')
+      .select(columns.syncAsset)
+      .select('assets.updateId')
+      .where('album_assets.albumsId', '=', albumId)
+      .where('assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('assets.updateId', '<=', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('assets.updateId', '>=', afterUpdateId!))
+      .orderBy('assets.updateId', 'asc')
+      .stream();
+  }
 
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumToAssetDeletes(userId: string, ack?: SyncAck) {
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets')
+      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'assets.id')
+      .select(columns.syncAsset)
+      .select('assets.updateId')
+      .where('assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .$if(!!ack, (qb) => qb.where('assets.updateId', '>', ack!.updateId))
+      .orderBy('assets.updateId', 'asc')
+      .innerJoin('albums', 'albums.id', 'album_assets.albumsId')
+      .leftJoin('albums_shared_users_users as album_users', 'album_users.albumsId', 'album_assets.albumsId')
+      .where((eb) => eb.or([eb('albums.ownerId', '=', userId), eb('album_users.usersId', '=', userId)]))
+      .stream();
+  }
+}
+
+class AlbumAssetExifSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('exif')
+      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'exif.assetId')
+      .select(columns.syncAssetExif)
+      .select('exif.updateId')
+      .where('album_assets.albumsId', '=', albumId)
+      .where('exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('exif.updateId', '<=', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('exif.updateId', '>=', afterUpdateId!))
+      .orderBy('exif.updateId', 'asc')
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('exif')
+      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'exif.assetId')
+      .select(columns.syncAssetExif)
+      .select('exif.updateId')
+      .where('exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .$if(!!ack, (qb) => qb.where('exif.updateId', '>', ack!.updateId))
+      .orderBy('exif.updateId', 'asc')
+      .innerJoin('albums', 'albums.id', 'album_assets.albumsId')
+      .leftJoin('albums_shared_users_users as album_users', 'album_users.albumsId', 'album_assets.albumsId')
+      .where((eb) => eb.or([eb('albums.ownerId', '=', userId), eb('album_users.usersId', '=', userId)]))
+      .stream();
+  }
+}
+
+class AlbumToAssetSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('albums_assets_assets as album_assets')
+      .select(['album_assets.assetsId as assetId', 'album_assets.albumsId as albumId', 'album_assets.updateId'])
+      .where('album_assets.albumsId', '=', albumId)
+      .where('album_assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('album_assets.updateId', '<=', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('album_assets.updateId', '>=', afterUpdateId!))
+      .orderBy('album_assets.updateId', 'asc')
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getDeletes(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('album_assets_audit')
       .select(['id', 'assetId', 'albumId'])
@@ -277,7 +241,37 @@ export class SyncRepository {
   }
 
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumUserDeletes(userId: string, ack?: SyncAck) {
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('albums_assets_assets as album_assets')
+      .select(['album_assets.assetsId as assetId', 'album_assets.albumsId as albumId', 'album_assets.updateId'])
+      .where('album_assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .$if(!!ack, (qb) => qb.where('album_assets.updateId', '>', ack!.updateId))
+      .orderBy('album_assets.updateId', 'asc')
+      .innerJoin('albums', 'albums.id', 'album_assets.albumsId')
+      .leftJoin('albums_shared_users_users as album_users', 'album_users.albumsId', 'album_assets.albumsId')
+      .where((eb) => eb.or([eb('albums.ownerId', '=', userId), eb('album_users.usersId', '=', userId)]))
+      .stream();
+  }
+}
+
+class AlbumUserSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('albums_shared_users_users as album_users')
+      .select(columns.syncAlbumUser)
+      .select('album_users.updateId')
+      .where('albumsId', '=', albumId)
+      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('updateId', '<=', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('updateId', '>=', afterUpdateId!))
+      .orderBy('updateId', 'asc')
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getDeletes(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('album_users_audit')
       .select(['id', 'userId', 'albumId'])
@@ -303,34 +297,8 @@ export class SyncRepository {
       .stream();
   }
 
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
-  getAlbumBackfill(userId: string, afterCreateId?: string) {
-    return this.db
-      .selectFrom('albums_shared_users_users')
-      .select(['albumsId as id', 'createId'])
-      .where('usersId', '=', userId)
-      .$if(!!afterCreateId, (qb) => qb.where('createId', '>=', afterCreateId!))
-      .where('createdAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .orderBy('createId', 'asc')
-      .execute();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
-  getAlbumUsersBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
-    return this.db
-      .selectFrom('albums_shared_users_users as album_users')
-      .select(columns.syncAlbumUser)
-      .select('album_users.updateId')
-      .where('albumsId', '=', albumId)
-      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .where('updateId', '<=', beforeUpdateId)
-      .$if(!!afterUpdateId, (eb) => eb.where('updateId', '>=', afterUpdateId!))
-      .orderBy('updateId', 'asc')
-      .stream();
-  }
-
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumUserUpserts(userId: string, ack?: SyncAck) {
+  getUpserts(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('albums_shared_users_users as album_users')
       .select(columns.syncAlbumUser)
@@ -358,98 +326,57 @@ export class SyncRepository {
       )
       .stream();
   }
+}
 
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
-  getAlbumAssetsBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+class AssetSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getDeletes(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets_audit')
+      .select(['id', 'assetId'])
+      .where('ownerId', '=', userId)
+      .$call((qb) => this.auditTableFilters(qb, ack))
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getUpserts(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('assets')
-      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'assets.id')
       .select(columns.syncAsset)
       .select('assets.updateId')
-      .where('album_assets.albumsId', '=', albumId)
-      .where('assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .where('assets.updateId', '<=', beforeUpdateId)
-      .$if(!!afterUpdateId, (eb) => eb.where('assets.updateId', '>=', afterUpdateId!))
-      .orderBy('assets.updateId', 'asc')
+      .where('ownerId', '=', userId)
+      .$call((qb) => this.upsertTableFilters(qb, ack))
       .stream();
   }
+}
 
+class AssetExifSync extends BaseSync {
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumAssetsUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('assets')
-      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'assets.id')
-      .select(columns.syncAsset)
-      .select('assets.updateId')
-      .where('assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .$if(!!ack, (qb) => qb.where('assets.updateId', '>', ack!.updateId))
-      .orderBy('assets.updateId', 'asc')
-      .innerJoin('albums', 'albums.id', 'album_assets.albumsId')
-      .leftJoin('albums_shared_users_users as album_users', 'album_users.albumsId', 'album_assets.albumsId')
-      .where((eb) => eb.or([eb('albums.ownerId', '=', userId), eb('album_users.usersId', '=', userId)]))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
-  getAlbumToAssetBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
-    return this.db
-      .selectFrom('albums_assets_assets as album_assets')
-      .select(['album_assets.assetsId as assetId', 'album_assets.albumsId as albumId', 'album_assets.updateId'])
-      .where('album_assets.albumsId', '=', albumId)
-      .where('album_assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .where('album_assets.updateId', '<=', beforeUpdateId)
-      .$if(!!afterUpdateId, (eb) => eb.where('album_assets.updateId', '>=', afterUpdateId!))
-      .orderBy('album_assets.updateId', 'asc')
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumToAssetUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('albums_assets_assets as album_assets')
-      .select(['album_assets.assetsId as assetId', 'album_assets.albumsId as albumId', 'album_assets.updateId'])
-      .where('album_assets.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .$if(!!ack, (qb) => qb.where('album_assets.updateId', '>', ack!.updateId))
-      .orderBy('album_assets.updateId', 'asc')
-      .innerJoin('albums', 'albums.id', 'album_assets.albumsId')
-      .leftJoin('albums_shared_users_users as album_users', 'album_users.albumsId', 'album_assets.albumsId')
-      .where((eb) => eb.or([eb('albums.ownerId', '=', userId), eb('album_users.usersId', '=', userId)]))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
-  getAlbumAssetExifsBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+  getUpserts(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('exif')
-      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'exif.assetId')
       .select(columns.syncAssetExif)
       .select('exif.updateId')
-      .where('album_assets.albumsId', '=', albumId)
-      .where('exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .where('exif.updateId', '<=', beforeUpdateId)
-      .$if(!!afterUpdateId, (eb) => eb.where('exif.updateId', '>=', afterUpdateId!))
-      .orderBy('exif.updateId', 'asc')
+      .where('assetId', 'in', (eb) => eb.selectFrom('assets').select('id').where('ownerId', '=', userId))
+      .$call((qb) => this.upsertTableFilters(qb, ack))
       .stream();
   }
+}
 
+class MemorySync extends BaseSync {
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getAlbumAssetExifsUpserts(userId: string, ack?: SyncAck) {
+  getDeletes(userId: string, ack?: SyncAck) {
     return this.db
-      .selectFrom('exif')
-      .innerJoin('albums_assets_assets as album_assets', 'album_assets.assetsId', 'exif.assetId')
-      .select(columns.syncAssetExif)
-      .select('exif.updateId')
-      .where('exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .$if(!!ack, (qb) => qb.where('exif.updateId', '>', ack!.updateId))
-      .orderBy('exif.updateId', 'asc')
-      .innerJoin('albums', 'albums.id', 'album_assets.albumsId')
-      .leftJoin('albums_shared_users_users as album_users', 'album_users.albumsId', 'album_assets.albumsId')
-      .where((eb) => eb.or([eb('albums.ownerId', '=', userId), eb('album_users.usersId', '=', userId)]))
+      .selectFrom('memories_audit')
+      .select(['id', 'memoryId'])
+      .where('userId', '=', userId)
+      .$call((qb) => this.auditTableFilters(qb, ack))
       .stream();
   }
 
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getMemoryUpserts(userId: string, ack?: SyncAck) {
+  getUpserts(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('memories')
       .select([
@@ -471,30 +398,11 @@ export class SyncRepository {
       .$call((qb) => this.upsertTableFilters(qb, ack))
       .stream();
   }
+}
 
+class MemoryToAssetSync extends BaseSync {
   @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getMemoryDeletes(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('memories_audit')
-      .select(['id', 'memoryId'])
-      .where('userId', '=', userId)
-      .$call((qb) => this.auditTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getMemoryAssetUpserts(userId: string, ack?: SyncAck) {
-    return this.db
-      .selectFrom('memories_assets_assets')
-      .select(['memoriesId as memoryId', 'assetsId as assetId'])
-      .select('updateId')
-      .where('memoriesId', 'in', (eb) => eb.selectFrom('memories').select('id').where('ownerId', '=', userId))
-      .$call((qb) => this.upsertTableFilters(qb, ack))
-      .stream();
-  }
-
-  @GenerateSql({ params: [DummyValue.UUID], stream: true })
-  getMemoryAssetDeletes(userId: string, ack?: SyncAck) {
+  getDeletes(userId: string, ack?: SyncAck) {
     return this.db
       .selectFrom('memory_assets_audit')
       .select(['id', 'memoryId', 'assetId'])
@@ -503,22 +411,144 @@ export class SyncRepository {
       .stream();
   }
 
-  private auditTableFilters<T extends keyof Pick<DB, AuditTables>, D>(qb: SelectQueryBuilder<DB, T, D>, ack?: SyncAck) {
-    const builder = qb as SelectQueryBuilder<DB, AuditTables, D>;
-    return builder
-      .where('deletedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .$if(!!ack, (qb) => qb.where('id', '>', ack!.updateId))
-      .orderBy('id', 'asc') as SelectQueryBuilder<DB, T, D>;
-  }
-
-  private upsertTableFilters<T extends keyof Pick<DB, UpsertTables>, D>(
-    qb: SelectQueryBuilder<DB, T, D>,
-    ack?: SyncAck,
-  ) {
-    const builder = qb as SelectQueryBuilder<DB, UpsertTables, D>;
-    return builder
-      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .$if(!!ack, (qb) => qb.where('updateId', '>', ack!.updateId))
-      .orderBy('updateId', 'asc') as SelectQueryBuilder<DB, T, D>;
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('memories_assets_assets')
+      .select(['memoriesId as memoryId', 'assetsId as assetId'])
+      .select('updateId')
+      .where('memoriesId', 'in', (eb) => eb.selectFrom('memories').select('id').where('ownerId', '=', userId))
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+}
+
+class PartnerSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
+  getCreatedAfter(userId: string, afterCreateId?: string) {
+    return this.db
+      .selectFrom('partners')
+      .select(['sharedById', 'createId'])
+      .where('sharedWithId', '=', userId)
+      .$if(!!afterCreateId, (qb) => qb.where('createId', '>=', afterCreateId!))
+      .where('createdAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .orderBy('partners.createId', 'asc')
+      .execute();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getDeletes(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('partners_audit')
+      .select(['id', 'sharedById', 'sharedWithId'])
+      .where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
+      .$call((qb) => this.auditTableFilters(qb, ack))
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('partners')
+      .select(['sharedById', 'sharedWithId', 'inTimeline', 'updateId'])
+      .where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+}
+
+class PartnerAssetsSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('assets')
+      .select(columns.syncAsset)
+      .select('assets.updateId')
+      .where('ownerId', '=', partnerId)
+      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('updateId', '<=', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('updateId', '>=', afterUpdateId!))
+      .orderBy('updateId', 'asc')
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getDeletes(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets_audit')
+      .select(['id', 'assetId'])
+      .where('ownerId', 'in', (eb) =>
+        eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
+      )
+      .$call((qb) => this.auditTableFilters(qb, ack))
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets')
+      .select(columns.syncAsset)
+      .select('assets.updateId')
+      .where('ownerId', 'in', (eb) =>
+        eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
+      )
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+}
+
+class PartnerAssetExifsSync extends BaseSync {
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
+  getBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
+    return this.db
+      .selectFrom('exif')
+      .select(columns.syncAssetExif)
+      .select('exif.updateId')
+      .innerJoin('assets', 'assets.id', 'exif.assetId')
+      .where('assets.ownerId', '=', partnerId)
+      .where('exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
+      .where('exif.updateId', '<=', beforeUpdateId)
+      .$if(!!afterUpdateId, (eb) => eb.where('exif.updateId', '>=', afterUpdateId!))
+      .orderBy('exif.updateId', 'asc')
+      .stream();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID], stream: true })
+  getUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('exif')
+      .select(columns.syncAssetExif)
+      .select('exif.updateId')
+      .where('assetId', 'in', (eb) =>
+        eb
+          .selectFrom('assets')
+          .select('id')
+          .where('ownerId', 'in', (eb) =>
+            eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
+          ),
+      )
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+}
+
+class UserSync extends BaseSync {
+  @GenerateSql({ params: [], stream: true })
+  getDeletes(ack?: SyncAck) {
+    return this.db
+      .selectFrom('users_audit')
+      .select(['id', 'userId'])
+      .$call((qb) => this.auditTableFilters(qb, ack))
+      .stream();
+  }
+
+  @GenerateSql({ params: [], stream: true })
+  getUpserts(ack?: SyncAck) {
+    return this.db
+      .selectFrom('users')
+      .select(['id', 'name', 'email', 'deletedAt', 'updateId'])
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
   }
 }