diff --git a/apps/desktop/src/media-photos-sync.test.ts b/apps/desktop/src/media-photos-sync.test.ts index a1928a8..d37b569 100644 --- a/apps/desktop/src/media-photos-sync.test.ts +++ b/apps/desktop/src/media-photos-sync.test.ts @@ -88,10 +88,11 @@ function memoryStore(initial?: PhotosSyncState) { return { store, writes, current: () => state }; } -function asset(id: string, externalId: string | undefined, ingestState: MediaAsset["ingestState"]) { +function asset(id: string, externalId: string, ingestState: MediaAsset["ingestState"]) { return { id, sourceId: "source-1", + externalId, kind: "video", storage: "remote", capturedAt: null, @@ -109,7 +110,6 @@ function asset(id: string, externalId: string | undefined, ingestState: MediaAss hasThumbnail: false, qualityScore: null, createdAt: "2024-01-01T00:00:00.000Z", - ...(externalId === undefined ? {} : { externalId }), } as MediaAsset; } @@ -121,7 +121,12 @@ function fakeApi( } = {}, ) { const uploads = new Map(Object.entries(options.uploads ?? {})); - const pushes: { items: number; syncToken: string | null; complete: boolean }[] = []; + const pushes: { + items: number; + syncToken: string | null; + complete: boolean; + deleted?: number; + }[] = []; const chunkBytes = options.chunkBytes ?? 4; const api = { media: { @@ -138,15 +143,22 @@ function fakeApi( deviceId: input.deviceId, })), pushItems: vi.fn( - async (input: { items: unknown[]; syncToken: string | null; complete: boolean }) => { - expect(input.items.length).toBeGreaterThan(0); + async (input: { + items: unknown[]; + deletedExternalIds?: string[]; + syncToken: string | null; + complete: boolean; + }) => { + const deleted = input.deletedExternalIds?.length ?? 0; + expect(input.items.length > 0 || deleted > 0 || input.complete).toBe(true); expect(input.items.length).toBeLessThanOrEqual(500); pushes.push({ items: input.items.length, syncToken: input.syncToken, complete: input.complete, + ...(deleted > 0 ? { deleted } : {}), }); - return { created: input.items.length, updated: 0, unchanged: 0 }; + return { created: input.items.length, updated: 0, unchanged: 0, deleted }; }, ), }, @@ -309,10 +321,10 @@ describe("Apple Photos metadata sync", () => { }, }); await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: vi.fn() }); - expect(pushes).toEqual([{ items: 1, syncToken: "token-a", complete: true }]); + expect(pushes).toEqual([{ items: 0, syncToken: "token-a", complete: true }]); }); - it("relists everything when items were deleted or the token expired", async () => { + it("pushes device deletions without relisting", async () => { const { helper } = fakeHelper({ size: 3 }); const { api, pushes } = fakeApi(); const { store } = memoryStore({ @@ -340,8 +352,8 @@ describe("Apple Photos metadata sync", () => { store, uploadChunk: vi.fn(), }); - expect(result.fullListing).toBe(true); - expect(pushes).toEqual([{ items: 3, syncToken: "token-3", complete: true }]); + expect(result.fullListing).toBe(false); + expect(pushes).toEqual([{ items: 0, syncToken: "new", complete: false, deleted: 1 }]); }); it("registers again when the server no longer knows the source", async () => { @@ -538,21 +550,6 @@ describe("Apple Photos original upload", () => { expect(api.media.uploads.complete).not.toHaveBeenCalled(); await expect(readdir(tempRoot)).resolves.toEqual([]); }); - - it("refuses to guess which Photos item an asset is", async () => { - const { helper } = fakeHelper({ size: 1 }); - const { api } = fakeApi({ assets: [asset("a1", undefined, "awaiting_upload")] }); - await expect( - syncApplePhotos({ - ...baseOptions(), - helper, - api, - store: memoryStore().store, - uploadChunk: vi.fn(), - }), - ).rejects.toMatchObject({ code: "unsupported_server" }); - expect(helper.exportOriginal).not.toHaveBeenCalled(); - }); }); describe("Apple Photos sync state file", () => { diff --git a/apps/desktop/src/media-photos-sync.ts b/apps/desktop/src/media-photos-sync.ts index 1c1b8c4..72ea3f9 100644 --- a/apps/desktop/src/media-photos-sync.ts +++ b/apps/desktop/src/media-photos-sync.ts @@ -45,12 +45,7 @@ export type UploadChunk = ( signal: AbortSignal, ) => Promise; -export type PhotosSyncErrorCode = - | "not_authorized" - | "cancelled" - | "integrity" - | "upload_failed" - | "unsupported_server"; +export type PhotosSyncErrorCode = "not_authorized" | "cancelled" | "integrity" | "upload_failed"; export class PhotosSyncError extends Error { constructor( @@ -294,11 +289,13 @@ export async function syncApplePhotos(options: PhotosSyncOptions): Promise { const counts = await api.media.sources.pushItems( { sourceId: target.sourceId, items, + deletedExternalIds, syncToken: syncToken !== null && syncToken.length <= MAX_SYNC_TOKEN_LENGTH ? syncToken : null, complete, @@ -321,13 +318,9 @@ export async function syncApplePhotos(options: PhotosSyncOptions): Promise 0 || !done || listing.lastItem === null - ? page.items - : [listing.lastItem]; - if (items.length > 0) await push(target, items, done ? listing.token : null, done); + const items = page.items; + // An empty final page still closes the listing. + if (items.length > 0 || done) await push(target, items, done ? listing.token : null, done); if (done) { target.changeToken = listing.token; target.listing = null; @@ -347,16 +340,23 @@ export async function syncApplePhotos(options: PhotosSyncOptions): Promise 0 || changes.token === null) { + if (changes.fullResyncRequired || changes.token === null) { target.listing = null; await listAll(target); return; } - for (let start = 0; start < changes.changed.length; start += batchSize) { - const batch = changes.changed.slice(start, start + batchSize); - const last = start + batchSize >= changes.changed.length; - await push(target, batch, last ? changes.token : null, false); + const batches = Math.max( + Math.ceil(changes.changed.length / batchSize), + Math.ceil(changes.deleted.length / DELETED_BATCH_SIZE), + ); + for (let index = 0; index < batches; index++) { + const batch = changes.changed.slice(index * batchSize, (index + 1) * batchSize); + const deleted = changes.deleted.slice( + index * DELETED_BATCH_SIZE, + (index + 1) * DELETED_BATCH_SIZE, + ); + const last = index === batches - 1; + await push(target, batch, last ? changes.token : null, false, deleted); } target.changeToken = changes.token; await save(); @@ -440,12 +440,9 @@ interface AwaitingAsset { externalId: string; } -/** - * The contract has no "awaiting upload" query, so this pages through the source's - * assets and filters on `ingestState`. Mapping an asset back to a Photos item needs its - * `externalId`, which `MediaAssetSchema` does not expose yet; without it uploads fail - * loudly instead of guessing. - */ +/** Assets of this source the server is waiting for, mapped back to Photos items. */ +const DELETED_BATCH_SIZE = 5000; + export async function awaitingUploads( api: MediaSyncApi, sourceId: string, @@ -455,19 +452,12 @@ export async function awaitingUploads( let cursor: string | null = null; do { const page: { items: MediaAsset[]; nextCursor: string | null } = await api.media.assets.search( - { sourceId, cursor, limit: PHOTOS_MAX_PAGE }, + { sourceId, ingestStates: ["awaiting_upload"], cursor, limit: PHOTOS_MAX_PAGE }, { signal }, ); for (const asset of page.items) { if (asset.ingestState !== "awaiting_upload" || asset.sourceId !== sourceId) continue; - const externalId = (asset as MediaAsset & { externalId?: unknown }).externalId; - if (typeof externalId !== "string" || externalId === "") { - throw new PhotosSyncError( - "unsupported_server", - "The server does not say which Photos item an asset is.", - ); - } - awaiting.push({ id: asset.id, externalId }); + awaiting.push({ id: asset.id, externalId: asset.externalId }); } cursor = page.nextCursor; } while (cursor !== null); diff --git a/packages/contracts/src/media.ts b/packages/contracts/src/media.ts index 428fa44..b70e5b8 100644 --- a/packages/contracts/src/media.ts +++ b/packages/contracts/src/media.ts @@ -735,6 +735,8 @@ export type MediaSource = z.infer; export const MediaAssetSchema = z.object({ id: Id, sourceId: Id, + /** Identifier inside the source (relative path, Photos localIdentifier). */ + externalId: z.string(), kind: MediaAssetKindSchema, storage: MediaAssetStorageSchema, capturedAt: IsoDate.nullable(), @@ -1090,6 +1092,7 @@ export const MediaAssetQuerySchema = PageInput.extend({ personId: Id.optional(), collectionId: Id.optional(), sourceId: Id.optional(), + ingestStates: z.array(MediaIngestStateSchema).max(7).optional(), minQuality: z.number().min(0).max(1).optional(), favoritesOnly: z.boolean().optional(), }); @@ -1161,17 +1164,26 @@ export const mediaRpcShapes = { output: MediaSourceSchema, }, pushItems: { - input: z.object({ - sourceId: Id, - items: z.array(ClientMediaItemSchema).min(1).max(500), - /** Opaque change token from the device for the next incremental push. */ - syncToken: z.string().max(2000).nullable(), - complete: z.boolean(), - }), + input: z + .object({ + sourceId: Id, + items: z.array(ClientMediaItemSchema).max(500), + /** Items removed on the device since the last push. */ + deletedExternalIds: z.array(z.string().min(1).max(200)).max(5000).default([]), + /** Opaque change token from the device for the next incremental push. */ + syncToken: z.string().max(2000).nullable(), + complete: z.boolean(), + }) + .refine( + (input) => + input.items.length > 0 || input.deletedExternalIds.length > 0 || input.complete, + { message: "Push items, deletions, or completion", path: ["items"] }, + ), output: z.object({ created: z.number().int(), updated: z.number().int(), unchanged: z.number().int(), + deleted: z.number().int(), }), }, sync: { input: z.object({ sourceId: Id }), output: ok }, diff --git a/packages/db/src/media-library.ts b/packages/db/src/media-library.ts index 3bee8a5..40f669b 100644 --- a/packages/db/src/media-library.ts +++ b/packages/db/src/media-library.ts @@ -307,6 +307,46 @@ export function createMediaLibraryRepos(prisma: PrismaClient) { if (result.count !== 1) throw new IsolationError(); } + /** + * Delete assets a device reported as removed. Returns the media-root-relative files + * the caller must delete (originals, proxies, thumbnails, face keyframes, masks). + */ + async function removeSourceItems( + spaceId: string, + sourceId: string, + externalIds: string[], + ): Promise<{ deleted: number; paths: string[] }> { + if (externalIds.length === 0) return { deleted: 0, paths: [] }; + return prisma.$transaction(async (tx) => { + const source = await tx.mediaSource.findFirst({ where: { id: sourceId, spaceId } }); + if (!source) throw new IsolationError(); + const assets = await tx.mediaAsset.findMany({ + where: { spaceId, sourceId, externalId: { in: externalIds } }, + select: { + id: true, + storagePath: true, + proxyPath: true, + thumbPath: true, + spritePath: true, + faceTracks: { select: { keyframesPath: true } }, + marks: { select: { maskDir: true } }, + }, + }); + const paths = assets.flatMap((asset) => [ + ...[asset.storagePath, asset.proxyPath, asset.thumbPath, asset.spritePath], + ...asset.faceTracks.map((track) => track.keyframesPath), + ...asset.marks.map((mark) => mark.maskDir), + ]); + const result = await tx.mediaAsset.deleteMany({ + where: { spaceId, id: { in: assets.map((asset) => asset.id) } }, + }); + return { + deleted: result.count, + paths: paths.filter((path): path is string => typeof path === "string" && path !== ""), + }; + }); + } + /** * Worker entry point: jobs carry only a record id. Returns the owning space so every * further call stays scoped, or null when the record no longer exists. @@ -413,6 +453,9 @@ export function createMediaLibraryRepos(prisma: PrismaClient) { if (query.kinds && query.kinds.length > 0) { conditions.push(Prisma.sql`a.kind IN (${Prisma.join(query.kinds)})`); } + if (query.ingestStates && query.ingestStates.length > 0) { + conditions.push(Prisma.sql`a."ingestState" IN (${Prisma.join(query.ingestStates)})`); + } if (query.from) conditions.push(Prisma.sql`a."capturedAt" >= ${new Date(query.from)}`); if (query.to) conditions.push(Prisma.sql`a."capturedAt" <= ${new Date(query.to)}`); if (query.favoritesOnly) conditions.push(Prisma.sql`a."isFavorite" = TRUE`); @@ -734,6 +777,7 @@ export function createMediaLibraryRepos(prisma: PrismaClient) { getSource, updateSourceSync, removeSource, + removeSourceItems, mediaSpaceOf, upsertSourceItems, searchAssets, diff --git a/packages/db/src/media-shared.ts b/packages/db/src/media-shared.ts index 589160e..b464ca3 100644 --- a/packages/db/src/media-shared.ts +++ b/packages/db/src/media-shared.ts @@ -276,6 +276,7 @@ export function mapMediaAsset(row: MediaAssetRow): MediaAsset { return { id: row.id, sourceId: row.sourceId, + externalId: row.externalId, kind: row.kind as MediaAssetKind, storage: row.storage as MediaAssetStorage, capturedAt: iso(row.capturedAt), diff --git a/packages/db/src/media.postgres.test.ts b/packages/db/src/media.postgres.test.ts index c1e233e..b82ca83 100644 --- a/packages/db/src/media.postgres.test.ts +++ b/packages/db/src/media.postgres.test.ts @@ -34,6 +34,33 @@ describePostgres("media library and organization (PostgreSQL)", () => { return { ...user, sourceId: source.id }; } + it("removes device-deleted items and filters by ingest state", async () => { + const { spaceId, sourceId } = await newLibrary("ios_library"); + const other = await newLibrary("ios_library"); + const items = Array.from({ length: 5 }, (_, index) => sourceItem(index)); + await media.upsertSourceItems(spaceId, sourceId, items); + await media.upsertSourceItems(other.spaceId, other.sourceId, items); + + const awaiting = await media.searchAssets(spaceId, { ingestStates: ["awaiting_upload"] }); + expect(awaiting.items).toHaveLength(5); + expect(awaiting.items.map((asset) => asset.externalId).sort()).toEqual( + items.map((item) => item.externalId).sort(), + ); + expect((await media.searchAssets(spaceId, { ingestStates: ["analyzed"] })).items).toEqual([]); + + const removed = await media.removeSourceItems(spaceId, sourceId, [ + items[0]!.externalId, + items[1]!.externalId, + "unknown", + ]); + expect(removed.deleted).toBe(2); + expect((await media.searchAssets(spaceId, {})).items).toHaveLength(3); + expect((await media.searchAssets(other.spaceId, {})).items).toHaveLength(5); + await expect( + media.removeSourceItems(other.spaceId, sourceId, [items[2]!.externalId]), + ).rejects.toBeInstanceOf(IsolationError); + }); + it("indexes source items once and reports only real changes on the next sync", async () => { const { spaceId, sourceId } = await newLibrary(); const items = Array.from({ length: 1000 }, (_, index) => diff --git a/packages/db/src/media.test.ts b/packages/db/src/media.test.ts index c859369..6b3c3cd 100644 --- a/packages/db/src/media.test.ts +++ b/packages/db/src/media.test.ts @@ -122,6 +122,7 @@ describe("media DTO mappers", () => { expect(dto).toEqual({ id: "asset-1", sourceId: "source-1", + externalId: asset.externalId, kind: "video", storage: "referenced", capturedAt: at.toISOString(),