cynarAI/evOo-local
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat(media): expose asset externalId, filter by ingest state, push device deletions

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
This commit is contained in:
Claude 2026-10-03 18:00:02 +00:00
parent 51445e9442
commit 1aab5ab3a8
No known key found for this signature in database
7 changed files with 138 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
apps/desktop/src/media-photos-sync.test.ts
View File

@ -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", () => {

58
apps/desktop/src/media-photos-sync.ts
View File

@ -45,12 +45,7 @@ export type UploadChunk = (
signal: AbortSignal,
) => Promise<void>;
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<Photo
items: ClientMediaItem[],
syncToken: string | null,
complete: boolean,
deletedExternalIds: string[] = [],
) => {
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<Photo
const page = await helper.list({ cursor: listing.cursor, limit: batchSize }, signal);
if (listing.cursor === null) listing.token = page.changeToken;
const done = page.nextCursor === null;
// An empty final page (deletions while resuming) still has to close the listing,
// and pushItems needs at least one item: resend the last one, which is a no-op.
const items =
page.items.length > 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<Photo
return;
}
const changes = await helper.changes(target.changeToken, signal);
// pushItems cannot express deletions; a complete listing lets the server reconcile them.
if (changes.fullResyncRequired || changes.deleted.length > 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);

26
packages/contracts/src/media.ts
View File

@ -735,6 +735,8 @@ export type MediaSource = z.infer<typeof MediaSourceSchema>;
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 },

44
packages/db/src/media-library.ts
View File

@ -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,

1
packages/db/src/media-shared.ts
View File

@ -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),

27
packages/db/src/media.postgres.test.ts
View File

@ -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) =>

1
packages/db/src/media.test.ts
View File

@ -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(),