From 1aab5ab3a82f0997afee7434080469a30a8f9615 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 3 Oct 2026 18:00:02 +0000 Subject: [PATCH] feat(media): expose asset externalId, filter by ingest state, push device deletions The desktop Photos sync needs to map awaiting assets back to Photos items, query only awaiting uploads, and report deletions without a full relisting. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj --- apps/desktop/src/media-photos-sync.test.ts | 47 ++++++++---------- apps/desktop/src/media-photos-sync.ts | 58 +++++++++------------- packages/contracts/src/media.ts | 26 +++++++--- packages/db/src/media-library.ts | 44 ++++++++++++++++ packages/db/src/media-shared.ts | 1 + packages/db/src/media.postgres.test.ts | 27 ++++++++++ packages/db/src/media.test.ts | 1 + 7 files changed, 138 insertions(+), 66 deletions(-) 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(),