mirror of
https://github.com/immich-app/immich
synced 2026-08-22 13:13:05 +00:00
fix(server): respect backpressure in the sync stream (#30764)
Co-authored-by: enol5423 <your-email@example.com> Co-authored-by: Daniel Dietzler <mail@ddietzler.dev>
This commit is contained in:
enol5423
GitHub
parent
b0a9468da7
commit
618dc0d397
2 changed files with 162 additions and 66 deletions
78
server/src/services/sync.service.spec.ts
Normal file
78
server/src/services/sync.service.spec.ts
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
import { Writable } from 'node:stream';
|
||||
import { SyncEntityType } from 'src/enum';
|
||||
import { send } from 'src/services/sync.service';
|
||||
import { serialize } from 'src/utils/sync';
|
||||
|
||||
type TestStream = {
|
||||
stream: Writable;
|
||||
chunks: string[];
|
||||
flushNext: () => void;
|
||||
pendingCount: () => number;
|
||||
};
|
||||
|
||||
const createTestStream = (highWaterMark: number): TestStream => {
|
||||
const chunks: string[] = [];
|
||||
const pendingCallbacks: Array<() => void> = [];
|
||||
|
||||
const stream = new Writable({
|
||||
highWaterMark,
|
||||
write(chunk, _encoding, callback) {
|
||||
chunks.push(chunk.toString());
|
||||
pendingCallbacks.push(callback);
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
stream,
|
||||
chunks,
|
||||
flushNext: () => pendingCallbacks.shift()?.(),
|
||||
pendingCount: () => pendingCallbacks.length,
|
||||
};
|
||||
};
|
||||
|
||||
describe('send', () => {
|
||||
const item = {
|
||||
type: SyncEntityType.SyncCompleteV1 as const,
|
||||
data: {},
|
||||
ids: ['now-id'] as [string],
|
||||
};
|
||||
|
||||
it('resolves immediately when the stream has capacity', async () => {
|
||||
// A large highWaterMark means write() never signals backpressure for a
|
||||
// single small item.
|
||||
const { stream, chunks, flushNext } = createTestStream(1024 * 1024);
|
||||
|
||||
const sendPromise = send(stream, item);
|
||||
flushNext();
|
||||
await sendPromise;
|
||||
|
||||
expect(chunks).toEqual([serialize(item)]);
|
||||
});
|
||||
|
||||
it('waits for the drain event before resolving when the stream signals backpressure', async () => {
|
||||
// A tiny highWaterMark means the very first write already exceeds
|
||||
// capacity, so write() returns false and send() must wait for 'drain'.
|
||||
const { stream, chunks, flushNext, pendingCount } = createTestStream(1);
|
||||
|
||||
let resolved = false;
|
||||
const sendPromise = send(stream, item).then(() => {
|
||||
resolved = true;
|
||||
});
|
||||
|
||||
// Let any pending microtasks run; send() should still be waiting on the
|
||||
// underlying write to complete and 'drain' to fire — it must not resolve
|
||||
// just because write() was called.
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
expect(resolved).toBe(false);
|
||||
expect(pendingCount()).toBe(1);
|
||||
|
||||
// Completing the write lets the stream's internal buffer drop back below
|
||||
// highWaterMark, which is what triggers the 'drain' event.
|
||||
flushNext();
|
||||
await sendPromise;
|
||||
|
||||
expect(resolved).toBe(true);
|
||||
expect(chunks).toEqual([serialize(item)]);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,6 +1,7 @@
|
|||
import { BadRequestException, ForbiddenException, Injectable } from '@nestjs/common';
|
||||
import { Insertable } from 'kysely';
|
||||
import { DateTime, Duration } from 'luxon';
|
||||
import { once } from 'node:events';
|
||||
import { Writable } from 'node:stream';
|
||||
import { OnJob } from 'src/decorators';
|
||||
import { AuthDto } from 'src/dtos/auth.dto';
|
||||
|
|
@ -42,12 +43,17 @@ const isEntityBackfillComplete = (createId: string, checkpoint: SyncAck | undefi
|
|||
const getStartId = (createId: string, checkpoint: SyncAck | undefined): string | undefined =>
|
||||
createId === checkpoint?.updateId ? checkpoint?.extraId : undefined;
|
||||
|
||||
const send = <T extends keyof SyncItem, D extends SyncItem[T]>(response: Writable, item: SerializeOptions<T, D>) => {
|
||||
response.write(serialize(item));
|
||||
export const send = async <T extends keyof SyncItem, D extends SyncItem[T]>(
|
||||
response: Writable,
|
||||
item: SerializeOptions<T, D>,
|
||||
) => {
|
||||
if (!response.write(serialize(item))) {
|
||||
await once(response, 'drain');
|
||||
}
|
||||
};
|
||||
|
||||
const sendEntityBackfillCompleteAck = (response: Writable, ackType: SyncEntityType, id: string) => {
|
||||
send(response, { type: SyncEntityType.SyncAckV1, data: {}, ackType, ids: [id, COMPLETE_ID] });
|
||||
const sendEntityBackfillCompleteAck = async (response: Writable, ackType: SyncEntityType, id: string) => {
|
||||
await send(response, { type: SyncEntityType.SyncAckV1, data: {}, ackType, ids: [id, COMPLETE_ID] });
|
||||
};
|
||||
|
||||
export const SYNC_TYPES_ORDER = [
|
||||
|
|
@ -141,7 +147,7 @@ export class SyncService extends BaseService {
|
|||
|
||||
const isPendingSyncReset = await this.sessionRepository.isPendingSyncReset(session.id);
|
||||
if (isPendingSyncReset) {
|
||||
send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
|
||||
await send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
|
||||
response.end();
|
||||
return;
|
||||
}
|
||||
|
|
@ -150,7 +156,7 @@ export class SyncService extends BaseService {
|
|||
const checkpointMap: CheckpointMap = Object.fromEntries(checkpoints.map(({ type, ack }) => [type, fromAck(ack)]));
|
||||
|
||||
if (this.needsFullSync(checkpointMap)) {
|
||||
send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
|
||||
await send(response, { type: SyncEntityType.SyncResetV1, ids: ['reset'], data: {} });
|
||||
response.end();
|
||||
return;
|
||||
}
|
||||
|
|
@ -201,7 +207,7 @@ export class SyncService extends BaseService {
|
|||
await handler();
|
||||
}
|
||||
|
||||
send(response, { type: SyncEntityType.SyncCompleteV1, ids: [nowId], data: {} });
|
||||
await send(response, { type: SyncEntityType.SyncCompleteV1, ids: [nowId], data: {} });
|
||||
|
||||
response.end();
|
||||
}
|
||||
|
|
@ -242,7 +248,11 @@ export class SyncService extends BaseService {
|
|||
const upsertType = SyncEntityType.AuthUserV1;
|
||||
const upserts = this.syncRepository.authUser.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, profileImagePath, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } });
|
||||
await send(response, {
|
||||
type: upsertType,
|
||||
ids: [updateId],
|
||||
data: { ...data, hasProfileImage: !!profileImagePath },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -250,13 +260,17 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.UserDeleteV1;
|
||||
const deletes = this.syncRepository.user.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.UserV1;
|
||||
const upserts = this.syncRepository.user.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, profileImagePath, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: { ...data, hasProfileImage: !!profileImagePath } });
|
||||
await send(response, {
|
||||
type: upsertType,
|
||||
ids: [updateId],
|
||||
data: { ...data, hasProfileImage: !!profileImagePath },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -264,13 +278,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.PartnerDeleteV1;
|
||||
const deletes = this.syncRepository.partner.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.PartnerV1;
|
||||
const upserts = this.syncRepository.partner.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -282,13 +296,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.AssetDeleteV1;
|
||||
const deletes = this.syncRepository.asset.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AssetV2;
|
||||
const upserts = this.syncRepository.asset.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
await send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -307,7 +321,7 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.PartnerAssetDeleteV1;
|
||||
const deletes = this.syncRepository.partnerAsset.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.PartnerAssetBackfillV2;
|
||||
|
|
@ -334,14 +348,14 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, {
|
||||
await send(response, {
|
||||
type: backfillType,
|
||||
ids: [createId, updateId],
|
||||
data: mapSyncAssetV2(data),
|
||||
});
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
}
|
||||
} else if (partners.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -353,7 +367,7 @@ export class SyncService extends BaseService {
|
|||
|
||||
const upserts = this.syncRepository.partnerAsset.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
await send(response, { type: upsertType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -361,7 +375,7 @@ export class SyncService extends BaseService {
|
|||
const upsertType = SyncEntityType.AssetExifV1;
|
||||
const upserts = this.syncRepository.assetExif.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -370,13 +384,13 @@ export class SyncService extends BaseService {
|
|||
const deletes = this.syncRepository.assetEdit.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
const upsertType = SyncEntityType.AssetEditV1;
|
||||
const upserts = this.syncRepository.assetEdit.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -411,10 +425,10 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [partner.createId, updateId], data });
|
||||
await send(response, { type: backfillType, ids: [partner.createId, updateId], data });
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, partner.createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, partner.createId);
|
||||
}
|
||||
} else if (partners.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -426,7 +440,7 @@ export class SyncService extends BaseService {
|
|||
|
||||
const upserts = this.syncRepository.partnerAssetExif.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -434,14 +448,14 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.AlbumDeleteV1;
|
||||
const deletes = this.syncRepository.album.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AlbumV1;
|
||||
const upserts = this.syncRepository.album.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
const albumUsers = await this.syncRepository.album.getAlbumUsers(data.id);
|
||||
send(response, {
|
||||
await send(response, {
|
||||
type: upsertType,
|
||||
ids: [updateId],
|
||||
// TODO: return null instead of '' in v4
|
||||
|
|
@ -454,14 +468,18 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.AlbumDeleteV1;
|
||||
const deletes = this.syncRepository.album.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AlbumV2;
|
||||
const upserts = this.syncRepository.album.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
// TODO: return null instead of '' in v4
|
||||
send(response, { type: upsertType, ids: [updateId], data: { ...data, description: data.description ?? '' } });
|
||||
await send(response, {
|
||||
type: upsertType,
|
||||
ids: [updateId],
|
||||
data: { ...data, description: data.description ?? '' },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -474,7 +492,7 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.AlbumUserDeleteV1;
|
||||
const deletes = this.syncRepository.albumUser.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.AlbumUserBackfillV1;
|
||||
|
|
@ -501,10 +519,10 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
await send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
}
|
||||
} else if (albums.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -516,7 +534,7 @@ export class SyncService extends BaseService {
|
|||
|
||||
const upserts = this.syncRepository.albumUser.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -559,10 +577,10 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data: mapSyncAssetV2(data) });
|
||||
await send(response, { type: backfillType, ids: [createId, updateId], data: mapSyncAssetV2(data) });
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
}
|
||||
} else if (albums.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -578,7 +596,7 @@ export class SyncService extends BaseService {
|
|||
createCheckpoint,
|
||||
);
|
||||
for await (const { updateId, ...data } of updates) {
|
||||
send(response, { type: updateType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
await send(response, { type: updateType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -586,7 +604,7 @@ export class SyncService extends BaseService {
|
|||
let isFirst = true;
|
||||
for await (const { updateId, ...data } of creates) {
|
||||
if (isFirst) {
|
||||
send(response, {
|
||||
await send(response, {
|
||||
type: SyncEntityType.SyncAckV1,
|
||||
data: {},
|
||||
ackType: SyncEntityType.AlbumAssetUpdateV2,
|
||||
|
|
@ -594,7 +612,7 @@ export class SyncService extends BaseService {
|
|||
});
|
||||
isFirst = false;
|
||||
}
|
||||
send(response, { type: createType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
await send(response, { type: createType, ids: [updateId], data: mapSyncAssetV2(data) });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -630,10 +648,10 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
await send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
}
|
||||
} else if (albums.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -649,7 +667,7 @@ export class SyncService extends BaseService {
|
|||
createCheckpoint,
|
||||
);
|
||||
for await (const { updateId, ...data } of updates) {
|
||||
send(response, { type: updateType, ids: [updateId], data });
|
||||
await send(response, { type: updateType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -657,7 +675,7 @@ export class SyncService extends BaseService {
|
|||
let isFirst = true;
|
||||
for await (const { updateId, ...data } of creates) {
|
||||
if (isFirst) {
|
||||
send(response, {
|
||||
await send(response, {
|
||||
type: SyncEntityType.SyncAckV1,
|
||||
data: {},
|
||||
ackType: SyncEntityType.AlbumAssetExifUpdateV1,
|
||||
|
|
@ -665,7 +683,7 @@ export class SyncService extends BaseService {
|
|||
});
|
||||
isFirst = false;
|
||||
}
|
||||
send(response, { type: createType, ids: [updateId], data });
|
||||
await send(response, { type: createType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -678,7 +696,7 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.AlbumToAssetDeleteV1;
|
||||
const deletes = this.syncRepository.albumToAsset.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.AlbumToAssetBackfillV1;
|
||||
|
|
@ -705,10 +723,10 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
await send(response, { type: backfillType, ids: [createId, updateId], data });
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
}
|
||||
} else if (albums.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -720,7 +738,7 @@ export class SyncService extends BaseService {
|
|||
|
||||
const upserts = this.syncRepository.albumToAsset.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -728,13 +746,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.MemoryDeleteV1;
|
||||
const deletes = this.syncRepository.memory.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.MemoryV1;
|
||||
const upserts = this.syncRepository.memory.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -742,13 +760,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.MemoryToAssetDeleteV1;
|
||||
const deletes = this.syncRepository.memoryToAsset.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.MemoryToAssetV1;
|
||||
const upserts = this.syncRepository.memoryToAsset.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -756,13 +774,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.StackDeleteV1;
|
||||
const deletes = this.syncRepository.stack.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.StackV1;
|
||||
const upserts = this.syncRepository.stack.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -775,7 +793,7 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.PartnerStackDeleteV1;
|
||||
const deletes = this.syncRepository.partnerStack.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const backfillType = SyncEntityType.PartnerStackBackfillV1;
|
||||
|
|
@ -802,14 +820,14 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of backfill) {
|
||||
send(response, {
|
||||
await send(response, {
|
||||
type: backfillType,
|
||||
ids: [createId, updateId],
|
||||
data,
|
||||
});
|
||||
}
|
||||
|
||||
sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
await sendEntityBackfillCompleteAck(response, backfillType, createId);
|
||||
}
|
||||
} else if (partners.length > 0) {
|
||||
await this.upsertBackfillCheckpoint({
|
||||
|
|
@ -821,7 +839,7 @@ export class SyncService extends BaseService {
|
|||
|
||||
const upserts = this.syncRepository.partnerStack.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -829,13 +847,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.PersonDeleteV1;
|
||||
const deletes = this.syncRepository.person.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.PersonV1;
|
||||
const upserts = this.syncRepository.person.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -849,13 +867,13 @@ export class SyncService extends BaseService {
|
|||
const deleteType = SyncEntityType.AssetFaceDeleteV1;
|
||||
const deletes = this.syncRepository.assetFace.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AssetFaceV2;
|
||||
const upserts = this.syncRepository.assetFace.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -864,14 +882,14 @@ export class SyncService extends BaseService {
|
|||
const deletes = this.syncRepository.userMetadata.getDeletes({ ...options, ack: checkpointMap[deleteType] });
|
||||
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.UserMetadataV1;
|
||||
const upserts = this.syncRepository.userMetadata.getUpserts({ ...options, ack: checkpointMap[upsertType] });
|
||||
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -888,7 +906,7 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { id, ...data } of deletes) {
|
||||
send(response, { type: deleteType, ids: [id], data });
|
||||
await send(response, { type: deleteType, ids: [id], data });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AssetMetadataV1;
|
||||
|
|
@ -898,7 +916,7 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -915,7 +933,7 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const row of deletes) {
|
||||
send(response, { type: deleteType, ids: [row.id], data: row });
|
||||
await send(response, { type: deleteType, ids: [row.id], data: row });
|
||||
}
|
||||
|
||||
const upsertType = SyncEntityType.AssetOcrV1;
|
||||
|
|
@ -925,7 +943,7 @@ export class SyncService extends BaseService {
|
|||
);
|
||||
|
||||
for await (const { updateId, ...data } of upserts) {
|
||||
send(response, { type: upsertType, ids: [updateId], data });
|
||||
await send(response, { type: upsertType, ids: [updateId], data });
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue