mirror of
https://github.com/immich-app/immich.git
synced 2025-08-11 09:16:31 -04:00
fix: use a consistent point in time for all updates in a single sync
This commit is contained in:
Zack Pollard
parent
220ee82c18
commit
7724277416
@ -13,3 +13,7 @@ where
|
||||
delete from "session_sync_checkpoint"
|
||||
where
|
||||
"sessionId" = $1
|
||||
|
||||
-- SyncCheckpointRepository.getNow
|
||||
select
|
||||
immich_uuid_v7 (now() - interval '1 millisecond') as "nowId"
|
||||
|
||||
@ -9,7 +9,7 @@ from
|
||||
where
|
||||
"usersId" = $1
|
||||
and "createId" >= $2
|
||||
and "createdAt" < now() - interval '1 millisecond'
|
||||
and "createId" < $3
|
||||
order by
|
||||
"createId" asc
|
||||
|
||||
@ -21,7 +21,7 @@ from
|
||||
"album_audit"
|
||||
where
|
||||
"userId" = $1
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -41,10 +41,10 @@ from
|
||||
"album"
|
||||
left join "album_user" as "album_users" on "album"."id" = "album_users"."albumsId"
|
||||
where
|
||||
"album"."updatedAt" < now() - interval '1 millisecond'
|
||||
"album"."updateId" < $1
|
||||
and (
|
||||
"album"."ownerId" = $1
|
||||
or "album_users"."usersId" = $2
|
||||
"album"."ownerId" = $2
|
||||
or "album_users"."usersId" = $3
|
||||
)
|
||||
order by
|
||||
"album"."updateId" asc
|
||||
@ -67,20 +67,21 @@ select
|
||||
"asset"."livePhotoVideoId",
|
||||
"asset"."stackId",
|
||||
"asset"."libraryId",
|
||||
"asset"."updateId"
|
||||
"album_asset"."updateId"
|
||||
from
|
||||
"asset"
|
||||
inner join "album_asset" on "album_asset"."assetsId" = "asset"."id"
|
||||
"album_asset"
|
||||
inner join "asset" on "asset"."id" = "album_asset"."assetsId"
|
||||
where
|
||||
"album_asset"."albumsId" = $1
|
||||
and "asset"."updatedAt" < now() - interval '1 millisecond'
|
||||
and "asset"."updateId" <= $2
|
||||
and "asset"."updateId" >= $3
|
||||
and "album_asset"."updateId" < $2
|
||||
and "album_asset"."updateId" <= $3
|
||||
and "album_asset"."updateId" >= $4
|
||||
order by
|
||||
"asset"."updateId" asc
|
||||
"album_asset"."updateId" asc
|
||||
|
||||
-- SyncRepository.albumAsset.getUpserts
|
||||
-- SyncRepository.albumAsset.getCreates
|
||||
select
|
||||
"album_asset"."updateId",
|
||||
"asset"."id",
|
||||
"asset"."ownerId",
|
||||
"asset"."originalFileName",
|
||||
@ -96,21 +97,20 @@ select
|
||||
"asset"."duration",
|
||||
"asset"."livePhotoVideoId",
|
||||
"asset"."stackId",
|
||||
"asset"."libraryId",
|
||||
"asset"."updateId"
|
||||
"asset"."libraryId"
|
||||
from
|
||||
"asset"
|
||||
inner join "album_asset" on "album_asset"."assetsId" = "asset"."id"
|
||||
"album_asset"
|
||||
inner join "asset" on "asset"."id" = "album_asset"."assetsId"
|
||||
inner join "album" on "album"."id" = "album_asset"."albumsId"
|
||||
left join "album_user" on "album_user"."albumsId" = "album_asset"."albumsId"
|
||||
where
|
||||
"asset"."updatedAt" < now() - interval '1 millisecond'
|
||||
"album_asset"."updateId" < $1
|
||||
and (
|
||||
"album"."ownerId" = $1
|
||||
or "album_user"."usersId" = $2
|
||||
"album"."ownerId" = $2
|
||||
or "album_user"."usersId" = $3
|
||||
)
|
||||
order by
|
||||
"asset"."updateId" asc
|
||||
"album_asset"."updateId" asc
|
||||
|
||||
-- SyncRepository.albumAssetExif.getBackfill
|
||||
select
|
||||
@ -145,9 +145,9 @@ from
|
||||
inner join "album_asset" on "album_asset"."assetsId" = "asset_exif"."assetId"
|
||||
where
|
||||
"album_asset"."albumsId" = $1
|
||||
and "asset_exif"."updatedAt" < now() - interval '1 millisecond'
|
||||
and "asset_exif"."updateId" <= $2
|
||||
and "asset_exif"."updateId" >= $3
|
||||
and "asset_exif"."updateId" < $2
|
||||
and "asset_exif"."updateId" <= $3
|
||||
and "asset_exif"."updateId" >= $4
|
||||
order by
|
||||
"asset_exif"."updateId" asc
|
||||
|
||||
@ -185,10 +185,10 @@ from
|
||||
inner join "album" on "album"."id" = "album_asset"."albumsId"
|
||||
left join "album_user" on "album_user"."albumsId" = "album_asset"."albumsId"
|
||||
where
|
||||
"asset_exif"."updatedAt" < now() - interval '1 millisecond'
|
||||
"asset_exif"."updateId" < $1
|
||||
and (
|
||||
"album"."ownerId" = $1
|
||||
or "album_user"."usersId" = $2
|
||||
"album"."ownerId" = $2
|
||||
or "album_user"."usersId" = $3
|
||||
)
|
||||
order by
|
||||
"asset_exif"."updateId" asc
|
||||
@ -202,9 +202,9 @@ from
|
||||
"album_asset" as "album_assets"
|
||||
where
|
||||
"album_assets"."albumsId" = $1
|
||||
and "album_assets"."updatedAt" < now() - interval '1 millisecond'
|
||||
and "album_assets"."updateId" <= $2
|
||||
and "album_assets"."updateId" >= $3
|
||||
and "album_assets"."updateId" < $2
|
||||
and "album_assets"."updateId" <= $3
|
||||
and "album_assets"."updateId" >= $4
|
||||
order by
|
||||
"album_assets"."updateId" asc
|
||||
|
||||
@ -233,7 +233,7 @@ where
|
||||
"album_user"."usersId" = $2
|
||||
)
|
||||
)
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $3
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -247,10 +247,10 @@ from
|
||||
inner join "album" on "album"."id" = "album_asset"."albumsId"
|
||||
left join "album_user" on "album_user"."albumsId" = "album_asset"."albumsId"
|
||||
where
|
||||
"album_asset"."updatedAt" < now() - interval '1 millisecond'
|
||||
"album_asset"."updateId" < $1
|
||||
and (
|
||||
"album"."ownerId" = $1
|
||||
or "album_user"."usersId" = $2
|
||||
"album"."ownerId" = $2
|
||||
or "album_user"."usersId" = $3
|
||||
)
|
||||
order by
|
||||
"album_asset"."updateId" asc
|
||||
@ -265,9 +265,9 @@ from
|
||||
"album_user"
|
||||
where
|
||||
"albumsId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" <= $2
|
||||
and "updateId" >= $3
|
||||
and "updateId" < $2
|
||||
and "updateId" <= $3
|
||||
and "updateId" >= $4
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -296,7 +296,7 @@ where
|
||||
"album_user"."usersId" = $2
|
||||
)
|
||||
)
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $3
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -309,14 +309,14 @@ select
|
||||
from
|
||||
"album_user"
|
||||
where
|
||||
"album_user"."updatedAt" < now() - interval '1 millisecond'
|
||||
"album_user"."updateId" < $1
|
||||
and "album_user"."albumsId" in (
|
||||
select
|
||||
"id"
|
||||
from
|
||||
"album"
|
||||
where
|
||||
"ownerId" = $1
|
||||
"ownerId" = $2
|
||||
union
|
||||
(
|
||||
select
|
||||
@ -324,7 +324,7 @@ where
|
||||
from
|
||||
"album_user" as "albumUsers"
|
||||
where
|
||||
"albumUsers"."usersId" = $2
|
||||
"albumUsers"."usersId" = $3
|
||||
)
|
||||
)
|
||||
order by
|
||||
@ -338,7 +338,7 @@ from
|
||||
"asset_audit"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -365,7 +365,7 @@ from
|
||||
"asset"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -408,7 +408,7 @@ where
|
||||
where
|
||||
"ownerId" = $1
|
||||
)
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -421,7 +421,7 @@ from
|
||||
left join "asset" on "asset"."id" = "asset_face_audit"."assetId"
|
||||
where
|
||||
"asset"."ownerId" = $1
|
||||
and "asset_face_audit"."deletedAt" < now() - interval '1 millisecond'
|
||||
and "asset_face_audit"."id" < $2
|
||||
order by
|
||||
"asset_face_audit"."id" asc
|
||||
|
||||
@ -442,34 +442,11 @@ from
|
||||
"asset_face"
|
||||
left join "asset" on "asset"."id" = "asset_face"."assetId"
|
||||
where
|
||||
"asset_face"."updatedAt" < now() - interval '1 millisecond'
|
||||
and "asset"."ownerId" = $1
|
||||
"asset_face"."updateId" < $1
|
||||
and "asset"."ownerId" = $2
|
||||
order by
|
||||
"asset_face"."updateId" asc
|
||||
|
||||
-- SyncRepository.authUser.getUpserts
|
||||
select
|
||||
"id",
|
||||
"name",
|
||||
"email",
|
||||
"avatarColor",
|
||||
"deletedAt",
|
||||
"updateId",
|
||||
"profileImagePath",
|
||||
"profileChangedAt",
|
||||
"isAdmin",
|
||||
"pinCode",
|
||||
"oauthId",
|
||||
"storageLabel",
|
||||
"quotaSizeInBytes",
|
||||
"quotaUsageInBytes"
|
||||
from
|
||||
"user"
|
||||
where
|
||||
"updatedAt" < now() - interval '1 millisecond'
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
-- SyncRepository.memory.getDeletes
|
||||
select
|
||||
"id",
|
||||
@ -478,7 +455,7 @@ from
|
||||
"memory_audit"
|
||||
where
|
||||
"userId" = $1
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -501,7 +478,7 @@ from
|
||||
"memory"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -521,7 +498,7 @@ where
|
||||
where
|
||||
"ownerId" = $1
|
||||
)
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -541,7 +518,7 @@ where
|
||||
where
|
||||
"ownerId" = $1
|
||||
)
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -553,8 +530,7 @@ from
|
||||
"partner"
|
||||
where
|
||||
"sharedWithId" = $1
|
||||
and "createId" >= $2
|
||||
and "createdAt" < now() - interval '1 millisecond'
|
||||
and "createId" < $2
|
||||
order by
|
||||
"partner"."createId" asc
|
||||
|
||||
@ -570,7 +546,7 @@ where
|
||||
"sharedById" = $1
|
||||
or "sharedWithId" = $2
|
||||
)
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $3
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -587,7 +563,7 @@ where
|
||||
"sharedById" = $1
|
||||
or "sharedWithId" = $2
|
||||
)
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $3
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -614,9 +590,9 @@ from
|
||||
"asset"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" <= $2
|
||||
and "updateId" >= $3
|
||||
and "updateId" < $2
|
||||
and "updateId" <= $3
|
||||
and "updateId" >= $4
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -635,7 +611,7 @@ where
|
||||
where
|
||||
"sharedWithId" = $1
|
||||
)
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -669,7 +645,7 @@ where
|
||||
where
|
||||
"sharedWithId" = $1
|
||||
)
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -706,9 +682,9 @@ from
|
||||
inner join "asset" on "asset"."id" = "asset_exif"."assetId"
|
||||
where
|
||||
"asset"."ownerId" = $1
|
||||
and "asset_exif"."updatedAt" < now() - interval '1 millisecond'
|
||||
and "asset_exif"."updateId" <= $2
|
||||
and "asset_exif"."updateId" >= $3
|
||||
and "asset_exif"."updateId" < $2
|
||||
and "asset_exif"."updateId" <= $3
|
||||
and "asset_exif"."updateId" >= $4
|
||||
order by
|
||||
"asset_exif"."updateId" asc
|
||||
|
||||
@ -758,7 +734,7 @@ where
|
||||
"sharedWithId" = $1
|
||||
)
|
||||
)
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -777,7 +753,7 @@ where
|
||||
where
|
||||
"sharedWithId" = $1
|
||||
)
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -793,9 +769,9 @@ from
|
||||
"stack"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" <= $2
|
||||
and "updateId" >= $3
|
||||
and "updateId" < $2
|
||||
and "updateId" <= $3
|
||||
and "updateId" >= $4
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -818,7 +794,7 @@ where
|
||||
where
|
||||
"sharedWithId" = $1
|
||||
)
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -830,7 +806,7 @@ from
|
||||
"person_audit"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -851,7 +827,7 @@ from
|
||||
"person"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -863,7 +839,7 @@ from
|
||||
"stack_audit"
|
||||
where
|
||||
"userId" = $1
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -879,7 +855,7 @@ from
|
||||
"stack"
|
||||
where
|
||||
"ownerId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -890,7 +866,7 @@ select
|
||||
from
|
||||
"user_audit"
|
||||
where
|
||||
"deletedAt" < now() - interval '1 millisecond'
|
||||
"id" < $1
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -907,7 +883,7 @@ select
|
||||
from
|
||||
"user"
|
||||
where
|
||||
"updatedAt" < now() - interval '1 millisecond'
|
||||
"updateId" < $1
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -920,7 +896,7 @@ from
|
||||
"user_metadata_audit"
|
||||
where
|
||||
"userId" = $1
|
||||
and "deletedAt" < now() - interval '1 millisecond'
|
||||
and "id" < $2
|
||||
order by
|
||||
"id" asc
|
||||
|
||||
@ -934,6 +910,6 @@ from
|
||||
"user_metadata"
|
||||
where
|
||||
"userId" = $1
|
||||
and "updatedAt" < now() - interval '1 millisecond'
|
||||
and "updateId" < $2
|
||||
order by
|
||||
"updateId" asc
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { Insertable, Kysely } from 'kysely';
|
||||
import { Insertable, Kysely, sql } from 'kysely';
|
||||
import { InjectKysely } from 'nestjs-kysely';
|
||||
import { DummyValue, GenerateSql } from 'src/decorators';
|
||||
import { SyncEntityType } from 'src/enum';
|
||||
@ -39,4 +39,13 @@ export class SyncCheckpointRepository {
|
||||
.$if(!!types, (qb) => qb.where('type', 'in', types!))
|
||||
.execute();
|
||||
}
|
||||
|
||||
@GenerateSql()
|
||||
getNow() {
|
||||
return this.db
|
||||
.selectNoFrom((eb) => [
|
||||
eb.fn<string>('immich_uuid_v7', [sql.raw<Date>("now() - interval '1 millisecond'")]).as('nowId'),
|
||||
])
|
||||
.executeTakeFirstOrThrow();
|
||||
}
|
||||
}
|
||||
|
||||
@ -33,6 +33,11 @@ type UpsertTables =
|
||||
| 'user_metadata'
|
||||
| 'asset_face';
|
||||
|
||||
export type SyncQueryOptions = {
|
||||
nowId: string;
|
||||
userId: string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class SyncRepository {
|
||||
album: AlbumSync;
|
||||
@ -81,21 +86,21 @@ export class SyncRepository {
|
||||
class BaseSync {
|
||||
constructor(protected db: Kysely<DB>) {}
|
||||
|
||||
protected auditTableFilters(ack?: SyncAck) {
|
||||
protected auditTableFilters(nowId: string, ack?: SyncAck) {
|
||||
return <T extends keyof Pick<DB, AuditTables>, D>(qb: SelectQueryBuilder<DB, T, D>) => {
|
||||
const builder = qb as SelectQueryBuilder<DB, AuditTables, D>;
|
||||
return builder
|
||||
.where('deletedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('id', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('id', '>', ack!.updateId))
|
||||
.orderBy('id', 'asc') as SelectQueryBuilder<DB, T, D>;
|
||||
};
|
||||
}
|
||||
|
||||
protected upsertTableFilters(ack?: SyncAck) {
|
||||
protected upsertTableFilters(nowId: string, ack?: SyncAck) {
|
||||
return <T extends keyof Pick<DB, UpsertTables>, D>(qb: SelectQueryBuilder<DB, T, D>) => {
|
||||
const builder = qb as SelectQueryBuilder<DB, UpsertTables, D>;
|
||||
return builder
|
||||
.where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('updateId', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('updateId', '>', ack!.updateId))
|
||||
.orderBy('updateId', 'asc') as SelectQueryBuilder<DB, T, D>;
|
||||
};
|
||||
@ -103,34 +108,34 @@ class BaseSync {
|
||||
}
|
||||
|
||||
class AlbumSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
|
||||
getCreatedAfter(userId: string, afterCreateId?: string) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID] })
|
||||
getCreatedAfter({ nowId, userId }: SyncQueryOptions, afterCreateId?: string) {
|
||||
return this.db
|
||||
.selectFrom('album_user')
|
||||
.select(['albumsId as id', 'createId'])
|
||||
.where('usersId', '=', userId)
|
||||
.$if(!!afterCreateId, (qb) => qb.where('createId', '>=', afterCreateId!))
|
||||
.where('createdAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('createId', '<', nowId)
|
||||
.orderBy('createId', 'asc')
|
||||
.execute();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album_audit')
|
||||
.select(['id', 'albumId'])
|
||||
.where('userId', '=', userId)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album')
|
||||
.distinctOn(['album.id', 'album.updateId'])
|
||||
.where('album.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('album.updateId', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('album.updateId', '>', ack!.updateId))
|
||||
.orderBy('album.updateId', 'asc')
|
||||
.leftJoin('album_user as album_users', 'album.id', 'album_users.albumsId')
|
||||
@ -152,29 +157,32 @@ class AlbumSync extends BaseSync {
|
||||
}
|
||||
|
||||
class AlbumAssetSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill({ nowId }: SyncQueryOptions, albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
return this.db
|
||||
.selectFrom('album_asset')
|
||||
.innerJoin('asset', 'asset.id', 'album_asset.assetsId')
|
||||
.select(columns.syncAsset)
|
||||
.select('album_asset.updateId')
|
||||
.where('album_asset.albumsId', '=', albumId)
|
||||
.where('album_asset.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('album_asset.updateId', '<', nowId)
|
||||
.where('album_asset.updateId', '<=', beforeUpdateId)
|
||||
.$if(!!afterUpdateId, (eb) => eb.where('album_asset.updateId', '>=', afterUpdateId!))
|
||||
.orderBy('album_asset.updateId', 'asc')
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpdates(userId: string, albumToAssetAck: SyncAck, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpdates({ nowId, userId }: SyncQueryOptions, albumToAssetAck: SyncAck, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.innerJoin('album_asset', 'album_asset.assetsId', 'asset.id')
|
||||
.select(columns.syncAsset)
|
||||
.select('asset.updateId')
|
||||
.where('asset.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('asset.updateId', '<', nowId)
|
||||
.where('album_asset.updateId', '<=', albumToAssetAck.updateId) // Ensure we only send updates for assets that the client already knows about
|
||||
.$if(!!ack, (qb) => qb.where('asset.updateId', '>', ack!.updateId))
|
||||
.orderBy('asset.updateId', 'asc')
|
||||
@ -184,8 +192,8 @@ class AlbumAssetSync extends BaseSync {
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getCreates(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getCreates({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album_asset')
|
||||
.select('album_asset.updateId')
|
||||
@ -193,7 +201,7 @@ class AlbumAssetSync extends BaseSync {
|
||||
.select(columns.syncAsset)
|
||||
.innerJoin('album', 'album.id', 'album_asset.albumsId')
|
||||
.leftJoin('album_user', 'album_user.albumsId', 'album_asset.albumsId')
|
||||
.where('album_asset.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('album_asset.updateId', '<', nowId)
|
||||
.where((eb) => eb.or([eb('album.ownerId', '=', userId), eb('album_user.usersId', '=', userId)]))
|
||||
.$if(!!ack, (qb) => qb.where('album_asset.updateId', '>', ack!.updateId))
|
||||
.orderBy('album_asset.updateId', 'asc')
|
||||
@ -202,29 +210,32 @@ class AlbumAssetSync extends BaseSync {
|
||||
}
|
||||
|
||||
class AlbumAssetExifSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill({ nowId }: SyncQueryOptions, albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
return this.db
|
||||
.selectFrom('asset_exif')
|
||||
.innerJoin('album_asset', 'album_asset.assetsId', 'asset_exif.assetId')
|
||||
.select(columns.syncAssetExif)
|
||||
.select('asset_exif.updateId')
|
||||
.where('album_asset.albumsId', '=', albumId)
|
||||
.where('asset_exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('asset_exif.updateId', '<', nowId)
|
||||
.where('asset_exif.updateId', '<=', beforeUpdateId)
|
||||
.$if(!!afterUpdateId, (eb) => eb.where('asset_exif.updateId', '>=', afterUpdateId!))
|
||||
.orderBy('asset_exif.updateId', 'asc')
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_exif')
|
||||
.innerJoin('album_asset', 'album_asset.assetsId', 'asset_exif.assetId')
|
||||
.select(columns.syncAssetExif)
|
||||
.select('asset_exif.updateId')
|
||||
.where('asset_exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('asset_exif.updateId', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('asset_exif.updateId', '>', ack!.updateId))
|
||||
.orderBy('asset_exif.updateId', 'asc')
|
||||
.innerJoin('album', 'album.id', 'album_asset.albumsId')
|
||||
@ -235,21 +246,29 @@ class AlbumAssetExifSync extends BaseSync {
|
||||
}
|
||||
|
||||
class AlbumToAssetSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill(
|
||||
{ nowId, userId }: SyncQueryOptions,
|
||||
albumId: string,
|
||||
afterUpdateId: string | undefined,
|
||||
beforeUpdateId: string,
|
||||
) {
|
||||
return this.db
|
||||
.selectFrom('album_asset 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', '<', nowId)
|
||||
.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) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album_asset_audit')
|
||||
.select(['id', 'assetId', 'albumId'])
|
||||
@ -271,16 +290,16 @@ class AlbumToAssetSync extends BaseSync {
|
||||
),
|
||||
),
|
||||
)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album_asset')
|
||||
.select(['album_asset.assetsId as assetId', 'album_asset.albumsId as albumId', 'album_asset.updateId'])
|
||||
.where('album_asset.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('album_asset.updateId', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('album_asset.updateId', '>', ack!.updateId))
|
||||
.orderBy('album_asset.updateId', 'asc')
|
||||
.innerJoin('album', 'album.id', 'album_asset.albumsId')
|
||||
@ -291,22 +310,25 @@ class AlbumToAssetSync extends BaseSync {
|
||||
}
|
||||
|
||||
class AlbumUserSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill({ nowId }: SyncQueryOptions, albumId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
return this.db
|
||||
.selectFrom('album_user')
|
||||
.select(columns.syncAlbumUser)
|
||||
.select('album_user.updateId')
|
||||
.where('albumsId', '=', albumId)
|
||||
.where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('updateId', '<', nowId)
|
||||
.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) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album_user_audit')
|
||||
.select(['id', 'userId', 'albumId'])
|
||||
@ -328,17 +350,17 @@ class AlbumUserSync extends BaseSync {
|
||||
),
|
||||
),
|
||||
)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('album_user')
|
||||
.select(columns.syncAlbumUser)
|
||||
.select('album_user.updateId')
|
||||
.where('album_user.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('album_user.updateId', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('album_user.updateId', '>', ack!.updateId))
|
||||
.orderBy('album_user.updateId', 'asc')
|
||||
.where((eb) =>
|
||||
@ -364,53 +386,53 @@ class AlbumUserSync extends BaseSync {
|
||||
}
|
||||
|
||||
class AssetSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_audit')
|
||||
.select(['id', 'assetId'])
|
||||
.where('ownerId', '=', userId)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.select(columns.syncAsset)
|
||||
.select('asset.updateId')
|
||||
.where('ownerId', '=', userId)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class AuthUserSync extends BaseSync {
|
||||
@GenerateSql({ params: [], stream: true })
|
||||
getUpserts(ack?: SyncAck) {
|
||||
getUpserts({ nowId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('user')
|
||||
.select(columns.syncUser)
|
||||
.select(['isAdmin', 'pinCode', 'oauthId', 'storageLabel', 'quotaSizeInBytes', 'quotaUsageInBytes'])
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class PersonSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('person_audit')
|
||||
.select(['id', 'personId'])
|
||||
.where('ownerId', '=', userId)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('person')
|
||||
.select([
|
||||
@ -427,27 +449,27 @@ class PersonSync extends BaseSync {
|
||||
'faceAssetId',
|
||||
])
|
||||
.where('ownerId', '=', userId)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class AssetFaceSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_face_audit')
|
||||
.select(['asset_face_audit.id', 'assetFaceId'])
|
||||
.orderBy('asset_face_audit.id', 'asc')
|
||||
.leftJoin('asset', 'asset.id', 'asset_face_audit.assetId')
|
||||
.where('asset.ownerId', '=', userId)
|
||||
.where('asset_face_audit.deletedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('asset_face_audit.id', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('asset_face_audit.id', '>', ack!.updateId))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_face')
|
||||
.select([
|
||||
@ -463,7 +485,7 @@ class AssetFaceSync extends BaseSync {
|
||||
'sourceType',
|
||||
'asset_face.updateId',
|
||||
])
|
||||
.where('asset_face.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('asset_face.updateId', '<', nowId)
|
||||
.$if(!!ack, (qb) => qb.where('asset_face.updateId', '>', ack!.updateId))
|
||||
.orderBy('asset_face.updateId', 'asc')
|
||||
.leftJoin('asset', 'asset.id', 'asset_face.assetId')
|
||||
@ -473,31 +495,31 @@ class AssetFaceSync extends BaseSync {
|
||||
}
|
||||
|
||||
class AssetExifSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_exif')
|
||||
.select(columns.syncAssetExif)
|
||||
.select('asset_exif.updateId')
|
||||
.where('assetId', 'in', (eb) => eb.selectFrom('asset').select('id').where('ownerId', '=', userId))
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class MemorySync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('memory_audit')
|
||||
.select(['id', 'memoryId'])
|
||||
.where('userId', '=', userId)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('memory')
|
||||
.select([
|
||||
@ -516,97 +538,105 @@ class MemorySync extends BaseSync {
|
||||
])
|
||||
.select('updateId')
|
||||
.where('ownerId', '=', userId)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class MemoryToAssetSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('memory_asset_audit')
|
||||
.select(['id', 'memoryId', 'assetId'])
|
||||
.where('memoryId', 'in', (eb) => eb.selectFrom('memory').select('id').where('ownerId', '=', userId))
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('memory_asset')
|
||||
.select(['memoriesId as memoryId', 'assetsId as assetId'])
|
||||
.select('updateId')
|
||||
.where('memoriesId', 'in', (eb) => eb.selectFrom('memory').select('id').where('ownerId', '=', userId))
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class PartnerSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID] })
|
||||
getCreatedAfter(userId: string, afterCreateId?: string) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }] })
|
||||
getCreatedAfter({ nowId, userId }: SyncQueryOptions, afterCreateId?: string) {
|
||||
return this.db
|
||||
.selectFrom('partner')
|
||||
.select(['sharedById', 'createId'])
|
||||
.where('sharedWithId', '=', userId)
|
||||
.$if(!!afterCreateId, (qb) => qb.where('createId', '>=', afterCreateId!))
|
||||
.where('createdAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('createId', '<', nowId)
|
||||
.orderBy('partner.createId', 'asc')
|
||||
.execute();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('partner_audit')
|
||||
.select(['id', 'sharedById', 'sharedWithId'])
|
||||
.where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('partner')
|
||||
.select(['sharedById', 'sharedWithId', 'inTimeline', 'updateId'])
|
||||
.where((eb) => eb.or([eb('sharedById', '=', userId), eb('sharedWithId', '=', userId)]))
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class PartnerAssetsSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill(
|
||||
{ nowId }: SyncQueryOptions,
|
||||
partnerId: string,
|
||||
afterUpdateId: string | undefined,
|
||||
beforeUpdateId: string,
|
||||
) {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.select(columns.syncAsset)
|
||||
.select('asset.updateId')
|
||||
.where('ownerId', '=', partnerId)
|
||||
.where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('updateId', '<', nowId)
|
||||
.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) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_audit')
|
||||
.select(['id', 'assetId'])
|
||||
.where('ownerId', 'in', (eb) =>
|
||||
eb.selectFrom('partner').select(['sharedById']).where('sharedWithId', '=', userId),
|
||||
)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset')
|
||||
.select(columns.syncAsset)
|
||||
@ -614,29 +644,37 @@ class PartnerAssetsSync extends BaseSync {
|
||||
.where('ownerId', 'in', (eb) =>
|
||||
eb.selectFrom('partner').select(['sharedById']).where('sharedWithId', '=', userId),
|
||||
)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class PartnerAssetExifsSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill(
|
||||
{ nowId }: SyncQueryOptions,
|
||||
partnerId: string,
|
||||
afterUpdateId: string | undefined,
|
||||
beforeUpdateId: string,
|
||||
) {
|
||||
return this.db
|
||||
.selectFrom('asset_exif')
|
||||
.select(columns.syncAssetExif)
|
||||
.select('asset_exif.updateId')
|
||||
.innerJoin('asset', 'asset.id', 'asset_exif.assetId')
|
||||
.where('asset.ownerId', '=', partnerId)
|
||||
.where('asset_exif.updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('asset_exif.updateId', '<', nowId)
|
||||
.where('asset_exif.updateId', '<=', beforeUpdateId)
|
||||
.$if(!!afterUpdateId, (eb) => eb.where('asset_exif.updateId', '>=', afterUpdateId!))
|
||||
.orderBy('asset_exif.updateId', 'asc')
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('asset_exif')
|
||||
.select(columns.syncAssetExif)
|
||||
@ -649,61 +687,69 @@ class PartnerAssetExifsSync extends BaseSync {
|
||||
eb.selectFrom('partner').select(['sharedById']).where('sharedWithId', '=', userId),
|
||||
),
|
||||
)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class StackSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('stack_audit')
|
||||
.select(['id', 'stackId'])
|
||||
.where('userId', '=', userId)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('stack')
|
||||
.select(columns.syncStack)
|
||||
.select('updateId')
|
||||
.where('ownerId', '=', userId)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class PartnerStackSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('stack_audit')
|
||||
.select(['id', 'stackId'])
|
||||
.where('userId', 'in', (eb) => eb.selectFrom('partner').select(['sharedById']).where('sharedWithId', '=', userId))
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID, DummyValue.UUID, DummyValue.UUID], stream: true })
|
||||
getBackfill(partnerId: string, afterUpdateId: string | undefined, beforeUpdateId: string) {
|
||||
@GenerateSql({
|
||||
params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }, DummyValue.UUID, DummyValue.UUID, DummyValue.UUID],
|
||||
stream: true,
|
||||
})
|
||||
getBackfill(
|
||||
{ nowId }: SyncQueryOptions,
|
||||
partnerId: string,
|
||||
afterUpdateId: string | undefined,
|
||||
beforeUpdateId: string,
|
||||
) {
|
||||
return this.db
|
||||
.selectFrom('stack')
|
||||
.select(columns.syncStack)
|
||||
.select('updateId')
|
||||
.where('ownerId', '=', partnerId)
|
||||
.where('updatedAt', '<', sql.raw<Date>("now() - interval '1 millisecond'"))
|
||||
.where('updateId', '<', nowId)
|
||||
.where('updateId', '<=', beforeUpdateId)
|
||||
.$if(!!afterUpdateId, (eb) => eb.where('updateId', '>=', afterUpdateId!))
|
||||
.orderBy('updateId', 'asc')
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('stack')
|
||||
.select(columns.syncStack)
|
||||
@ -711,41 +757,41 @@ class PartnerStackSync extends BaseSync {
|
||||
.where('ownerId', 'in', (eb) =>
|
||||
eb.selectFrom('partner').select(['sharedById']).where('sharedWithId', '=', userId),
|
||||
)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
class UserSync extends BaseSync {
|
||||
@GenerateSql({ params: [], stream: true })
|
||||
getDeletes(ack?: SyncAck) {
|
||||
return this.db.selectFrom('user_audit').select(['id', 'userId']).$call(this.auditTableFilters(ack)).stream();
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db.selectFrom('user_audit').select(['id', 'userId']).$call(this.auditTableFilters(nowId, ack)).stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [], stream: true })
|
||||
getUpserts(ack?: SyncAck) {
|
||||
return this.db.selectFrom('user').select(columns.syncUser).$call(this.upsertTableFilters(ack)).stream();
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db.selectFrom('user').select(columns.syncUser).$call(this.upsertTableFilters(nowId, ack)).stream();
|
||||
}
|
||||
}
|
||||
|
||||
class UserMetadataSync extends BaseSync {
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getDeletes(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getDeletes({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('user_metadata_audit')
|
||||
.select(['id', 'userId', 'key'])
|
||||
.where('userId', '=', userId)
|
||||
.$call(this.auditTableFilters(ack))
|
||||
.$call(this.auditTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
|
||||
@GenerateSql({ params: [DummyValue.UUID], stream: true })
|
||||
getUpserts(userId: string, ack?: SyncAck) {
|
||||
@GenerateSql({ params: [{ nowId: DummyValue.UUID, userId: DummyValue.UUID }], stream: true })
|
||||
getUpserts({ nowId, userId }: SyncQueryOptions, ack?: SyncAck) {
|
||||
return this.db
|
||||
.selectFrom('user_metadata')
|
||||
.select(['userId', 'key', 'value', 'updateId'])
|
||||
.where('userId', '=', userId)
|
||||
.$call(this.upsertTableFilters(ack))
|
||||
.$call(this.upsertTableFilters(nowId, ack))
|
||||
.stream();
|
||||
}
|
||||
}
|
||||
|
||||
@ -16,6 +16,7 @@ import {
|
||||
SyncStreamDto,
|
||||
} from 'src/dtos/sync.dto';
|
||||
import { AssetVisibility, DatabaseAction, EntityType, Permission, SyncEntityType, SyncRequestType } from 'src/enum';
|
||||
import { SyncQueryOptions } from 'src/repositories/sync.repository';
|
||||
import { SessionSyncCheckpointTable } from 'src/schema/tables/sync-checkpoint.table';
|
||||
import { BaseService } from 'src/services/base.service';
|
||||
import { SyncAck } from 'src/types';
|
||||
@ -138,30 +139,34 @@ export class SyncService extends BaseService {
|
||||
return;
|
||||
}
|
||||
|
||||
const { nowId } = await this.syncCheckpointRepository.getNow();
|
||||
const options: SyncQueryOptions = { nowId, userId: auth.user.id };
|
||||
|
||||
const checkpoints = await this.syncCheckpointRepository.getAll(session.id);
|
||||
const checkpointMap: CheckpointMap = Object.fromEntries(checkpoints.map(({ type, ack }) => [type, fromAck(ack)]));
|
||||
|
||||
const handlers: Record<SyncRequestType, () => Promise<void>> = {
|
||||
[SyncRequestType.AuthUsersV1]: () => this.syncAuthUsersV1(response, checkpointMap),
|
||||
[SyncRequestType.UsersV1]: () => this.syncUsersV1(response, checkpointMap),
|
||||
[SyncRequestType.PartnersV1]: () => this.syncPartnersV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.AssetsV1]: () => this.syncAssetsV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.AssetExifsV1]: () => this.syncAssetExifsV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.PartnerAssetsV1]: () => this.syncPartnerAssetsV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.AuthUsersV1]: () => this.syncAuthUsersV1(options, response, checkpointMap),
|
||||
[SyncRequestType.UsersV1]: () => this.syncUsersV1(options, response, checkpointMap),
|
||||
[SyncRequestType.PartnersV1]: () => this.syncPartnersV1(options, response, checkpointMap),
|
||||
[SyncRequestType.AssetsV1]: () => this.syncAssetsV1(options, response, checkpointMap),
|
||||
[SyncRequestType.AssetExifsV1]: () => this.syncAssetExifsV1(options, response, checkpointMap),
|
||||
[SyncRequestType.PartnerAssetsV1]: () => this.syncPartnerAssetsV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.PartnerAssetExifsV1]: () =>
|
||||
this.syncPartnerAssetExifsV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.AlbumsV1]: () => this.syncAlbumsV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.AlbumUsersV1]: () => this.syncAlbumUsersV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.AlbumAssetsV1]: () => this.syncAlbumAssetsV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.AlbumToAssetsV1]: () => this.syncAlbumToAssetsV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.AlbumAssetExifsV1]: () => this.syncAlbumAssetExifsV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.MemoriesV1]: () => this.syncMemoriesV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.MemoryToAssetsV1]: () => this.syncMemoryAssetsV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.StacksV1]: () => this.syncStackV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.PartnerStacksV1]: () => this.syncPartnerStackV1(response, checkpointMap, auth, session.id),
|
||||
[SyncRequestType.PeopleV1]: () => this.syncPeopleV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.AssetFacesV1]: async () => this.syncAssetFacesV1(response, checkpointMap, auth),
|
||||
[SyncRequestType.UserMetadataV1]: () => this.syncUserMetadataV1(response, checkpointMap, auth),
|
||||
this.syncPartnerAssetExifsV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.AlbumsV1]: () => this.syncAlbumsV1(options, response, checkpointMap),
|
||||
[SyncRequestType.AlbumUsersV1]: () => this.syncAlbumUsersV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.AlbumAssetsV1]: () => this.syncAlbumAssetsV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.AlbumToAssetsV1]: () => this.syncAlbumToAssetsV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.AlbumAssetExifsV1]: () =>
|
||||
this.syncAlbumAssetExifsV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.MemoriesV1]: () => this.syncMemoriesV1(options, response, checkpointMap),
|
||||
[SyncRequestType.MemoryToAssetsV1]: () => this.syncMemoryAssetsV1(options, response, checkpointMap),
|
||||
[SyncRequestType.StacksV1]: () => this.syncStackV1(options, response, checkpointMap),
|
||||
[SyncRequestType.PartnerStacksV1]: () => this.syncPartnerStackV1(options, response, checkpointMap, session.id),
|
||||
[SyncRequestType.PeopleV1]: () => this.syncPeopleV1(options, response, checkpointMap),
|
||||
[SyncRequestType.AssetFacesV1]: async () => this.syncAssetFacesV1(options, response, checkpointMap),
|
||||
[SyncRequestType.UserMetadataV1]: () => this.syncUserMetadataV1(options, response, checkpointMap),
|
||||
};
|
||||
|
||||
for (const type of SYNC_TYPES_ORDER.filter((type) => dto.types.includes(type))) {
|
||||
@ -172,71 +177,71 @@ export class SyncService extends BaseService {
|
||||
response.end();
|
||||
}
|
||||
|
||||
private async syncAuthUsersV1(response: Writable, checkpointMap: CheckpointMap) {
|
||||
private async syncAuthUsersV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const upsertType = SyncEntityType.AuthUserV1;
|
||||
const upserts = this.syncRepository.authUser.getUpserts(checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.authUser.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, profileImagePath, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncUsersV1(response: Writable, checkpointMap: CheckpointMap) {
|
||||
private async syncUsersV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.UserDeleteV1;
|
||||
const deletes = this.syncRepository.user.getDeletes(checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.user.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.UserV1;
|
||||
const upserts = this.syncRepository.user.getUpserts(checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.user.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, profileImagePath, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncPartnersV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncPartnersV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.PartnerDeleteV1;
|
||||
const deletes = this.syncRepository.partner.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.partner.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.PartnerV1;
|
||||
const upserts = this.syncRepository.partner.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.partner.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAssetsV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncAssetsV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.AssetDeleteV1;
|
||||
const deletes = this.syncRepository.asset.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.asset.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AssetV1;
|
||||
const upserts = this.syncRepository.asset.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.asset.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV1(data) });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncPartnerAssetsV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
auth: AuthDto,
|
||||
sessionId: string,
|
||||
) {
|
||||
const deleteType = SyncEntityType.PartnerAssetDeleteV1;
|
||||
const deletes = this.syncRepository.partnerAsset.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.partnerAsset.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.PartnerAssetBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const partners = await this.syncRepository.partner.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const partners = await this.syncRepository.partner.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
const upsertType = SyncEntityType.PartnerAssetV1;
|
||||
const upsertCheckpoint = checkpointMap[upsertType];
|
||||
if (upsertCheckpoint) {
|
||||
@ -249,7 +254,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.partnerAsset.getBackfill(partner.sharedById, startId, endId);
|
||||
const backfill = this.syncRepository.partnerAsset.getBackfill(options, partner.sharedById, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, {
|
||||
@ -269,29 +274,29 @@ export class SyncService extends BaseService {
|
||||
});
|
||||
}
|
||||
|
||||
const upserts = this.syncRepository.partnerAsset.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.partnerAsset.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV1(data) });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAssetExifsV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncAssetExifsV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const upsertType = SyncEntityType.AssetExifV1;
|
||||
const upserts = this.syncRepository.assetExif.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.assetExif.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncPartnerAssetExifsV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
auth: AuthDto,
|
||||
sessionId: string,
|
||||
) {
|
||||
const backfillType = SyncEntityType.PartnerAssetExifBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const partners = await this.syncRepository.partner.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const partners = await this.syncRepository.partner.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
|
||||
const upsertType = SyncEntityType.PartnerAssetExifV1;
|
||||
const upsertCheckpoint = checkpointMap[upsertType];
|
||||
@ -305,7 +310,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.partnerAssetExif.getBackfill(partner.sharedById, startId, endId);
|
||||
const backfill = this.syncRepository.partnerAssetExif.getBackfill(options, partner.sharedById, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [partner.createId, updateId], data });
|
||||
@ -321,36 +326,41 @@ export class SyncService extends BaseService {
|
||||
});
|
||||
}
|
||||
|
||||
const upserts = this.syncRepository.partnerAssetExif.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.partnerAssetExif.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAlbumsV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncAlbumsV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.AlbumDeleteV1;
|
||||
const deletes = this.syncRepository.album.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.album.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AlbumV1;
|
||||
const upserts = this.syncRepository.album.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.album.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAlbumUsersV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto, sessionId: string) {
|
||||
private async syncAlbumUsersV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
sessionId: string,
|
||||
) {
|
||||
const deleteType = SyncEntityType.AlbumUserDeleteV1;
|
||||
const deletes = this.syncRepository.albumUser.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.albumUser.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.AlbumUserBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
const upsertType = SyncEntityType.AlbumUserV1;
|
||||
const upsertCheckpoint = checkpointMap[upsertType];
|
||||
if (upsertCheckpoint) {
|
||||
@ -363,7 +373,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.albumUser.getBackfill(album.id, startId, endId);
|
||||
const backfill = this.syncRepository.albumUser.getBackfill(options, album.id, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
@ -379,16 +389,21 @@ export class SyncService extends BaseService {
|
||||
});
|
||||
}
|
||||
|
||||
const upserts = this.syncRepository.albumUser.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.albumUser.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAlbumAssetsV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto, sessionId: string) {
|
||||
private async syncAlbumAssetsV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
sessionId: string,
|
||||
) {
|
||||
const backfillType = SyncEntityType.AlbumAssetBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
const updateType = SyncEntityType.AlbumAssetUpdateV1;
|
||||
const createType = SyncEntityType.AlbumAssetCreateV1;
|
||||
const updateCheckpoint = checkpointMap[updateType];
|
||||
@ -403,7 +418,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.albumAsset.getBackfill(album.id, startId, endId);
|
||||
const backfill = this.syncRepository.albumAsset.getBackfill(options, album.id, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data: mapSyncAssetV1(data) });
|
||||
@ -420,13 +435,13 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
if (createCheckpoint) {
|
||||
const updates = this.syncRepository.albumAsset.getUpdates(auth.user.id, createCheckpoint, updateCheckpoint);
|
||||
const updates = this.syncRepository.albumAsset.getUpdates(options, createCheckpoint, updateCheckpoint);
|
||||
for await (const { updateId, ...data } of updates) {
|
||||
send(response, { type: updateType, ids: [updateId], data: mapSyncAssetV1(data) });
|
||||
}
|
||||
}
|
||||
|
||||
const creates = this.syncRepository.albumAsset.getCreates(auth.user.id, createCheckpoint);
|
||||
const creates = this.syncRepository.albumAsset.getCreates(options, createCheckpoint);
|
||||
let first = true;
|
||||
for await (const { updateId, ...data } of creates) {
|
||||
if (first) {
|
||||
@ -443,14 +458,14 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
private async syncAlbumAssetExifsV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
auth: AuthDto,
|
||||
sessionId: string,
|
||||
) {
|
||||
const backfillType = SyncEntityType.AlbumAssetExifBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
const upsertType = SyncEntityType.AlbumAssetExifV1;
|
||||
const upsertCheckpoint = checkpointMap[upsertType];
|
||||
if (upsertCheckpoint) {
|
||||
@ -463,7 +478,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.albumAssetExif.getBackfill(album.id, startId, endId);
|
||||
const backfill = this.syncRepository.albumAssetExif.getBackfill(options, album.id, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
@ -479,27 +494,27 @@ export class SyncService extends BaseService {
|
||||
});
|
||||
}
|
||||
|
||||
const upserts = this.syncRepository.albumAssetExif.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.albumAssetExif.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAlbumToAssetsV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
auth: AuthDto,
|
||||
sessionId: string,
|
||||
) {
|
||||
const deleteType = SyncEntityType.AlbumToAssetDeleteV1;
|
||||
const deletes = this.syncRepository.albumToAsset.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.albumToAsset.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.AlbumToAssetBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const albums = await this.syncRepository.album.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
const upsertType = SyncEntityType.AlbumToAssetV1;
|
||||
const upsertCheckpoint = checkpointMap[upsertType];
|
||||
if (upsertCheckpoint) {
|
||||
@ -512,7 +527,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.albumToAsset.getBackfill(album.id, startId, endId);
|
||||
const backfill = this.syncRepository.albumToAsset.getBackfill(options, album.id, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
@ -528,64 +543,69 @@ export class SyncService extends BaseService {
|
||||
});
|
||||
}
|
||||
|
||||
const upserts = this.syncRepository.albumToAsset.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.albumToAsset.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncMemoriesV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncMemoriesV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.MemoryDeleteV1;
|
||||
const deletes = this.syncRepository.memory.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.memory.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.MemoryV1;
|
||||
const upserts = this.syncRepository.memory.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.memory.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncMemoryAssetsV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncMemoryAssetsV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.MemoryToAssetDeleteV1;
|
||||
const deletes = this.syncRepository.memoryToAsset.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.memoryToAsset.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.MemoryToAssetV1;
|
||||
const upserts = this.syncRepository.memoryToAsset.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.memoryToAsset.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncStackV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncStackV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.StackDeleteV1;
|
||||
const deletes = this.syncRepository.stack.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.stack.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.StackV1;
|
||||
const upserts = this.syncRepository.stack.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.stack.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncPartnerStackV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto, sessionId: string) {
|
||||
private async syncPartnerStackV1(
|
||||
options: SyncQueryOptions,
|
||||
response: Writable,
|
||||
checkpointMap: CheckpointMap,
|
||||
sessionId: string,
|
||||
) {
|
||||
const deleteType = SyncEntityType.PartnerStackDeleteV1;
|
||||
const deletes = this.syncRepository.partnerStack.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.partnerStack.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.PartnerStackBackfillV1;
|
||||
const backfillCheckpoint = checkpointMap[backfillType];
|
||||
const partners = await this.syncRepository.partner.getCreatedAfter(auth.user.id, backfillCheckpoint?.updateId);
|
||||
const partners = await this.syncRepository.partner.getCreatedAfter(options, backfillCheckpoint?.updateId);
|
||||
const upsertType = SyncEntityType.PartnerStackV1;
|
||||
const upsertCheckpoint = checkpointMap[upsertType];
|
||||
if (upsertCheckpoint) {
|
||||
@ -598,7 +618,7 @@ export class SyncService extends BaseService {
|
||||
}
|
||||
|
||||
const startId = getStartId(createId, backfillCheckpoint);
|
||||
const backfill = this.syncRepository.partnerStack.getBackfill(partner.sharedById, startId, endId);
|
||||
const backfill = this.syncRepository.partnerStack.getBackfill(options, partner.sharedById, startId, endId);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, {
|
||||
@ -618,50 +638,50 @@ export class SyncService extends BaseService {
|
||||
});
|
||||
}
|
||||
|
||||
const upserts = this.syncRepository.partnerStack.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.partnerStack.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncPeopleV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncPeopleV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.PersonDeleteV1;
|
||||
const deletes = this.syncRepository.people.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.people.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.PersonV1;
|
||||
const upserts = this.syncRepository.people.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.people.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncAssetFacesV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncAssetFacesV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.AssetFaceDeleteV1;
|
||||
const deletes = this.syncRepository.assetFace.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.assetFace.getDeletes(options, checkpointMap[deleteType]);
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AssetFaceV1;
|
||||
const upserts = this.syncRepository.assetFace.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.assetFace.getUpserts(options, checkpointMap[upsertType]);
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
private async syncUserMetadataV1(response: Writable, checkpointMap: CheckpointMap, auth: AuthDto) {
|
||||
private async syncUserMetadataV1(options: SyncQueryOptions, response: Writable, checkpointMap: CheckpointMap) {
|
||||
const deleteType = SyncEntityType.UserMetadataDeleteV1;
|
||||
const deletes = this.syncRepository.userMetadata.getDeletes(auth.user.id, checkpointMap[deleteType]);
|
||||
const deletes = this.syncRepository.userMetadata.getDeletes(options, checkpointMap[deleteType]);
|
||||
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.UserMetadataV1;
|
||||
const upserts = this.syncRepository.userMetadata.getUpserts(auth.user.id, checkpointMap[upsertType]);
|
||||
const upserts = this.syncRepository.userMetadata.getUpserts(options, checkpointMap[upsertType]);
|
||||
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
|
||||
@ -481,8 +481,6 @@ const personInsert = (person: Partial<Insertable<PersonTable>> & { ownerId: stri
|
||||
name: person.name || 'Test Name',
|
||||
ownerId: person.ownerId || newUuid(),
|
||||
thumbnailPath: person.thumbnailPath || '/path/to/thumbnail.jpg',
|
||||
updatedAt: person.updatedAt || newDate(),
|
||||
updateId: person.updateId || newUuid(),
|
||||
};
|
||||
return {
|
||||
...defaults,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user