feat: sync assets, partner assets, exif, and partner exif (#16658)

* feat: sync assets, partner assets, exif, and partner exif

Co-authored-by: Zack Pollard <zack@futo.org>
Co-authored-by: Alex Tran <alex.tran1502@gmail.com>

* refactor: remove duplicate where clause and orderBy statements in sync queries

* fix: asset deletes not filtering by ownerId

---------

Co-authored-by: Zack Pollard <zack@futo.org>
Co-authored-by: Alex Tran <alex.tran1502@gmail.com>
Co-authored-by: Zack Pollard <zackpollard@ymail.com>

server/src/repositories/asset.repository.ts

@@ -551,7 +551,7 @@ export class AssetRepository {
     return this.getById(asset.id, { exifInfo: true, faces: { person: true } }) as Promise<AssetEntity>;
   }
 
-  async remove(asset: AssetEntity): Promise<void> {
+  async remove(asset: { id: string }): Promise<void> {
     await this.db.deleteFrom('assets').where('id', '=', asUuid(asset.id)).execute();
   }
 
@@ -968,8 +968,8 @@ export class AssetRepository {
       )
       .select((eb) => eb.fn.toJson(eb.table('stacked_assets')).as('stack'))
       .where('assets.ownerId', '=', asUuid(ownerId))
-      .where('isVisible', '=', true)
-      .where('updatedAt', '<=', updatedUntil)
+      .where('assets.isVisible', '=', true)
+      .where('assets.updatedAt', '<=', updatedUntil)
       .$if(!!lastId, (qb) => qb.where('assets.id', '>', lastId!))
       .orderBy('assets.id')
       .limit(limit)
@@ -996,8 +996,8 @@ export class AssetRepository {
       )
       .select((eb) => eb.fn.toJson(eb.table('stacked_assets')).as('stack'))
       .where('assets.ownerId', '=', anyUuid(options.userIds))
-      .where('isVisible', '=', true)
-      .where('updatedAt', '>', options.updatedAfter)
+      .where('assets.isVisible', '=', true)
+      .where('assets.updatedAt', '>', options.updatedAfter)
       .limit(options.limit)
       .execute() as any as Promise<AssetEntity[]>;
   }

server/src/repositories/sync.repository.ts

@@ -1,10 +1,14 @@
 import { Injectable } from '@nestjs/common';
-import { Insertable, Kysely, sql } from 'kysely';
+import { Insertable, Kysely, SelectQueryBuilder, sql } from 'kysely';
 import { InjectKysely } from 'nestjs-kysely';
+import { columns } from 'src/database';
 import { DB, SessionSyncCheckpoints } from 'src/db';
 import { SyncEntityType } from 'src/enum';
 import { SyncAck } from 'src/types';
 
+type auditTables = 'users_audit' | 'partners_audit' | 'assets_audit';
+type upsertTables = 'users' | 'partners' | 'assets' | 'exif';
+
 @Injectable()
 export class SyncRepository {
   constructor(@InjectKysely() private db: Kysely<DB>) {}
@@ -41,9 +45,7 @@ export class SyncRepository {
     return this.db
       .selectFrom('users')
       .select(['id', 'name', 'email', 'deletedAt', 'updateId'])
-      .$if(!!ack, (qb) => qb.where('updateId', '>', ack!.updateId))
-      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .orderBy(['updateId asc'])
+      .$call((qb) => this.upsertTableFilters(qb, ack))
       .stream();
   }
 
@@ -51,9 +53,7 @@ export class SyncRepository {
     return this.db
       .selectFrom('users_audit')
       .select(['id', 'userId'])
-      .$if(!!ack, (qb) => qb.where('id', '>', ack!.updateId))
-      .where('deletedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .orderBy(['id asc'])
+      .$call((qb) => this.auditTableFilters(qb, ack))
       .stream();
   }
 
@@ -61,10 +61,8 @@ export class SyncRepository {
     return this.db
       .selectFrom('partners')
       .select(['sharedById', 'sharedWithId', 'inTimeline', 'updateId'])
-      .$if(!!ack, (qb) => qb.where('updateId', '>', ack!.updateId))
       .where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
-      .where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .orderBy(['updateId asc'])
+      .$call((qb) => this.upsertTableFilters(qb, ack))
       .stream();
   }
 
@@ -72,10 +70,93 @@ export class SyncRepository {
     return this.db
       .selectFrom('partners_audit')
       .select(['id', 'sharedById', 'sharedWithId'])
-      .$if(!!ack, (qb) => qb.where('id', '>', ack!.updateId))
       .where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
-      .where('deletedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
-      .orderBy(['id asc'])
+      .$call((qb) => this.auditTableFilters(qb, ack))
       .stream();
   }
+
+  getAssetUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets')
+      .select(columns.syncAsset)
+      .where('ownerId', '=', userId)
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+
+  getPartnerAssetsUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets')
+      .select(columns.syncAsset)
+      .where('ownerId', 'in', (eb) =>
+        eb.selectFrom('partners').select(['sharedById']).where('sharedWithId', '=', userId),
+      )
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+
+  getAssetDeletes(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('assets_audit')
+      .select(['id', 'assetId'])
+      .where('ownerId', '=', userId)
+      .$if(!!ack, (qb) => qb.where('id', '>', ack!.updateId))
+      .$call((qb) => this.auditTableFilters(qb, ack))
+      .stream();
+  }
+
+  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();
+  }
+
+  getAssetExifsUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('exif')
+      .select(columns.syncAssetExif)
+      .where('assetId', 'in', (eb) => eb.selectFrom('assets').select('id').where('ownerId', '=', userId))
+      .$call((qb) => this.upsertTableFilters(qb, ack))
+      .stream();
+  }
+
+  getPartnerAssetExifsUpserts(userId: string, ack?: SyncAck) {
+    return this.db
+      .selectFrom('exif')
+      .select(columns.syncAssetExif)
+      .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();
+  }
+
+  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>;
+  }
 }