import { createHash } from "node:crypto"; import { mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import type { ClientMediaItem, MediaAsset, MediaUpload } from "@rakazo/contracts"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { PhotosHelper } from "./media-photos-helper.js"; import type { MediaSyncApi, PhotosSyncState, PhotosSyncStore } from "./media-photos-sync.js"; import { createChunkUploader, createPhotosSyncFileStore, parsePhotosSyncState, runPool, syncApplePhotos, } from "./media-photos-sync.js"; const sha = (bytes: string | Buffer) => createHash("sha256").update(bytes).digest("hex"); function libraryItem(index: number): ClientMediaItem { return { externalId: `ASSET-${String(index).padStart(5, "0")}/L0/001`, kind: index % 3 === 0 ? "video" : "photo", capturedAt: new Date(Date.UTC(2024, 0, 1) + index * 60_000).toISOString(), durationMs: index % 3 === 0 ? 5000 : null, width: 1920, height: 1080, fps: null, gpsLat: null, gpsLon: null, deviceModel: null, albumNames: [], isFavorite: false, fileName: `IMG_${index}.JPG`, }; } /** Pages a synthetic library the way the Swift helper does: opaque cursor = next index. */ function fakeHelper(options: { size: number; contents?: (id: string) => string }) { const library = Array.from({ length: options.size }, (_, index) => libraryItem(index)); const exports: string[] = []; let active = 0; const stats = { maxActiveExports: 0 }; const helper: PhotosHelper & { library: ClientMediaItem[] } = { library, authorize: vi.fn(async () => "authorized" as const), list: vi.fn(async ({ cursor, limit }) => { const start = cursor === null ? 0 : Number(cursor); const items = library.slice(start, start + limit); const next = start + limit < library.length ? String(start + limit) : null; return { items, nextCursor: next, changeToken: `token-${library.length}` }; }), asset: vi.fn(), changes: vi.fn(), exportOriginal: vi.fn(async (id: string, destination: string, opts) => { active += 1; stats.maxActiveExports = Math.max(stats.maxActiveExports, active); try { await new Promise((resolve) => setTimeout(resolve, 5)); opts?.signal?.throwIfAborted(); const contents = options.contents?.(id) ?? `original bytes of ${id}`; const file = path.join(destination, "IMG.MOV"); await writeFile(file, contents); exports.push(destination); return { path: file, fileName: "IMG.MOV", sizeBytes: Buffer.byteLength(contents), sha256: sha(contents), }; } finally { active -= 1; } }), }; return { helper, exports, stats }; } function memoryStore(initial?: PhotosSyncState) { let state: PhotosSyncState = initial ?? { version: 1, deviceId: "device-1", targets: {} }; const writes: PhotosSyncState[] = []; const store: PhotosSyncStore = { read: async () => JSON.parse(JSON.stringify(state)) as PhotosSyncState, write: async (next) => { state = JSON.parse(JSON.stringify(next)) as PhotosSyncState; writes.push(state); }, }; return { store, writes, current: () => state }; } function asset(id: string, externalId: string | undefined, ingestState: MediaAsset["ingestState"]) { return { id, sourceId: "source-1", kind: "video", storage: "remote", capturedAt: null, durationMs: null, width: null, height: null, fps: null, gpsLat: null, gpsLon: null, deviceModel: null, albumNames: [], isFavorite: false, ingestState, hasProxy: false, hasThumbnail: false, qualityScore: null, createdAt: "2024-01-01T00:00:00.000Z", ...(externalId === undefined ? {} : { externalId }), } as MediaAsset; } function fakeApi( options: { assets?: MediaAsset[]; chunkBytes?: number; uploads?: Record; } = {}, ) { const uploads = new Map(Object.entries(options.uploads ?? {})); const pushes: { items: number; syncToken: string | null; complete: boolean }[] = []; const chunkBytes = options.chunkBytes ?? 4; const api = { media: { sources: { registerClient: vi.fn(async (input: { deviceId: string }) => ({ id: "source-1", kind: "apple_photos" as const, label: "Apple Photos", status: "idle" as const, assetCount: 0, lastSyncedAt: null, error: null, createdAt: "2024-01-01T00:00:00.000Z", deviceId: input.deviceId, })), pushItems: vi.fn( async (input: { items: unknown[]; syncToken: string | null; complete: boolean }) => { expect(input.items.length).toBeGreaterThan(0); expect(input.items.length).toBeLessThanOrEqual(500); pushes.push({ items: input.items.length, syncToken: input.syncToken, complete: input.complete, }); return { created: input.items.length, updated: 0, unchanged: 0 }; }, ), }, assets: { search: vi.fn(async (input: { cursor?: string | null }) => { const all = options.assets ?? []; // Two pages to exercise pagination. const half = Math.ceil(all.length / 2); return input.cursor === "p2" ? { items: all.slice(half), nextCursor: null } : { items: all.slice(0, half), nextCursor: all.length > 1 ? "p2" : null }; }), }, uploads: { begin: vi.fn(async (input: { assetId: string; sizeBytes: number }) => { const upload: MediaUpload = { id: `upload-${input.assetId}`, assetId: input.assetId, sizeBytes: input.sizeBytes, chunkBytes, receivedChunks: [], status: "open", }; uploads.set(upload.id, upload); return upload; }), status: vi.fn(async (input: { uploadId: string }) => { const upload = uploads.get(input.uploadId); if (!upload) throw Object.assign(new Error("Not found"), { code: "NOT_FOUND" }); return upload; }), complete: vi.fn(async (input: { uploadId: string }) => { const upload = uploads.get(input.uploadId)!; return { ...upload, status: "complete" as const }; }), }, }, }; return { api: api as unknown as MediaSyncApi & typeof api, pushes }; } let tempRoot: string; beforeEach(async () => { tempRoot = await mkdtemp(path.join(tmpdir(), "rakazo-photos-sync-")); }); afterEach(async () => { await rm(tempRoot, { recursive: true, force: true }); }); const baseOptions = () => ({ target: "https://rakazo.example.com#space_1", label: "Apple Photos", tempRoot, signal: new AbortController().signal, }); describe("Apple Photos metadata sync", () => { it("registers once, pushes batches of at most 500 and completes the listing", async () => { const { helper } = fakeHelper({ size: 1203 }); const { api, pushes } = fakeApi(); const { store, current } = memoryStore(); const chunks = vi.fn(); const result = await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: chunks, }); expect(api.media.sources.registerClient).toHaveBeenCalledWith( { kind: "apple_photos", label: "Apple Photos", deviceId: "device-1" }, expect.anything(), ); expect(pushes).toEqual([ { items: 500, syncToken: null, complete: false }, { items: 500, syncToken: null, complete: false }, { items: 203, syncToken: "token-1203", complete: true }, ]); expect(result).toMatchObject({ indexed: 1203, created: 1203, uploaded: 0, fullListing: true }); expect(current().targets["https://rakazo.example.com#space_1"]).toEqual({ sourceId: "source-1", listing: null, changeToken: "token-1203", uploads: {}, }); // The next run reads changes only and does not register again. vi.mocked(helper.changes).mockResolvedValueOnce({ token: "token-next", changed: [libraryItem(1), libraryItem(2)], deleted: [], fullResyncRequired: false, }); const second = await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: chunks, }); expect(api.media.sources.registerClient).toHaveBeenCalledTimes(1); expect(helper.changes).toHaveBeenCalledWith("token-1203", expect.anything()); expect(pushes.at(-1)).toEqual({ items: 2, syncToken: "token-next", complete: false }); expect(second.fullListing).toBe(false); expect(current().targets["https://rakazo.example.com#space_1"]?.changeToken).toBe("token-next"); }); it("persists the cursor after every batch and resumes an interrupted listing", async () => { const { helper } = fakeHelper({ size: 1100 }); const { api, pushes } = fakeApi(); const { store, writes, current } = memoryStore(); vi.mocked(api.media.sources.pushItems).mockImplementationOnce(async (input) => { pushes.push({ items: input.items.length, syncToken: input.syncToken, complete: input.complete, }); return { created: input.items.length, updated: 0, unchanged: 0 }; }); vi.mocked(api.media.sources.pushItems).mockRejectedValueOnce(new Error("network down")); await expect( syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: vi.fn() }), ).rejects.toThrow("network down"); const interrupted = current().targets["https://rakazo.example.com#space_1"]!; expect(interrupted.listing).toMatchObject({ cursor: "500", token: "token-1100" }); expect(interrupted.listing?.lastItem?.externalId).toBe(libraryItem(499).externalId); expect(writes.length).toBeGreaterThan(1); await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: vi.fn() }); expect(vi.mocked(helper.list).mock.calls.at(-2)?.[0]).toMatchObject({ cursor: "500" }); expect(pushes.slice(1)).toEqual([ { items: 500, syncToken: null, complete: false }, { items: 100, syncToken: "token-1100", complete: true }, ]); expect(current().targets["https://rakazo.example.com#space_1"]).toMatchObject({ listing: null, changeToken: "token-1100", }); }); it("closes a resumed listing whose last page came back empty", async () => { const { helper } = fakeHelper({ size: 0 }); const { api, pushes } = fakeApi(); const last = libraryItem(7); const { store } = memoryStore({ version: 1, deviceId: "device-1", targets: { "https://rakazo.example.com#space_1": { sourceId: "source-1", listing: { cursor: "500", token: "token-a", lastItem: last }, changeToken: null, uploads: {}, }, }, }); await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: vi.fn() }); expect(pushes).toEqual([{ items: 1, syncToken: "token-a", complete: true }]); }); it("relists everything when items were deleted or the token expired", async () => { const { helper } = fakeHelper({ size: 3 }); const { api, pushes } = fakeApi(); const { store } = memoryStore({ version: 1, deviceId: "device-1", targets: { "https://rakazo.example.com#space_1": { sourceId: "source-1", listing: null, changeToken: "old", uploads: {}, }, }, }); vi.mocked(helper.changes).mockResolvedValueOnce({ token: "new", changed: [], deleted: ["gone/L0/001"], fullResyncRequired: false, }); const result = await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: vi.fn(), }); expect(result.fullListing).toBe(true); expect(pushes).toEqual([{ items: 3, syncToken: "token-3", complete: true }]); }); it("registers again when the server no longer knows the source", async () => { const { helper } = fakeHelper({ size: 2 }); const { api } = fakeApi(); const { store } = memoryStore({ version: 1, deviceId: "device-1", targets: { "https://rakazo.example.com#space_1": { sourceId: "deleted-source", listing: { cursor: null, token: null, lastItem: null }, changeToken: null, uploads: {}, }, }, }); vi.mocked(api.media.sources.pushItems).mockRejectedValueOnce( Object.assign(new Error("Not found"), { code: "NOT_FOUND" }), ); await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk: vi.fn() }); expect(api.media.sources.registerClient).toHaveBeenCalledTimes(1); expect(vi.mocked(api.media.sources.pushItems).mock.calls.at(-1)?.[0].sourceId).toBe("source-1"); }); it("stops before anything is sent when Photos access is denied", async () => { const { helper } = fakeHelper({ size: 2 }); vi.mocked(helper.authorize).mockResolvedValueOnce("denied"); const { api } = fakeApi(); await expect( syncApplePhotos({ ...baseOptions(), helper, api, store: memoryStore().store, uploadChunk: vi.fn(), }), ).rejects.toMatchObject({ code: "not_authorized" }); expect(api.media.sources.registerClient).not.toHaveBeenCalled(); }); }); describe("Apple Photos original upload", () => { const awaitingAssets = [ asset("a1", "ASSET-00001/L0/001", "awaiting_upload"), asset("a2", "ASSET-00002/L0/001", "proxied"), asset("a3", "ASSET-00003/L0/001", "awaiting_upload"), asset("a4", "ASSET-00004/L0/001", "awaiting_upload"), ]; it("uploads every awaiting original in chunks with two at a time and cleans up", async () => { const { helper, exports, stats } = fakeHelper({ size: 1 }); const { api } = fakeApi({ assets: awaitingAssets, chunkBytes: 8 }); const sent = new Map(); const uploadChunk = vi.fn(async (uploadId: string, index: number, bytes: Uint8Array) => { const parts = sent.get(uploadId) ?? []; parts[index] = Buffer.from(bytes); sent.set(uploadId, parts); }); const progress = vi.fn(); const result = await syncApplePhotos({ ...baseOptions(), helper, api, store: memoryStore().store, uploadChunk, onProgress: progress, }); expect(result.uploaded).toBe(3); expect( vi .mocked(helper.exportOriginal) .mock.calls.map((call) => call[0]) .sort(), ).toEqual(["ASSET-00001/L0/001", "ASSET-00003/L0/001", "ASSET-00004/L0/001"]); expect(stats.maxActiveExports).toBe(2); expect(Buffer.concat(sent.get("upload-a1")!).toString()).toBe( "original bytes of ASSET-00001/L0/001", ); expect(sent.get("upload-a1")).toHaveLength(Math.ceil(36 / 8)); expect(api.media.uploads.begin).toHaveBeenCalledWith( expect.objectContaining({ assetId: "a1", sizeBytes: 36, sha256: sha("original bytes of ASSET-00001/L0/001"), }), expect.anything(), ); expect(api.media.uploads.complete).toHaveBeenCalledTimes(3); expect(progress).toHaveBeenLastCalledWith({ phase: "uploading", indexed: 1, uploadsDone: 3, uploadsTotal: 3, }); for (const dir of exports) await expect(readdir(dir)).rejects.toThrow(); await expect(readdir(tempRoot)).resolves.toEqual([]); }); it("resumes an open upload and sends only the missing chunks", async () => { const { helper } = fakeHelper({ size: 1 }); const contents = "original bytes of ASSET-00001/L0/001"; const { api } = fakeApi({ assets: [awaitingAssets[0]!], uploads: { "upload-old": { id: "upload-old", assetId: "a1", sizeBytes: Buffer.byteLength(contents), chunkBytes: 10, receivedChunks: [0, 2], status: "open", }, }, }); const { store, current } = memoryStore({ version: 1, deviceId: "device-1", targets: { "https://rakazo.example.com#space_1": { sourceId: "source-1", listing: null, changeToken: "tok", uploads: { a1: "upload-old" }, }, }, }); vi.mocked(helper.changes).mockResolvedValueOnce({ token: "tok", changed: [], deleted: [], fullResyncRequired: false, }); const uploadChunk = vi.fn(async () => {}); await syncApplePhotos({ ...baseOptions(), helper, api, store, uploadChunk }); expect(api.media.uploads.begin).not.toHaveBeenCalled(); expect( uploadChunk.mock.calls.map((call) => [ call[0], call[1], Buffer.from(call[2] as Uint8Array).toString(), ]), ).toEqual([ ["upload-old", 1, contents.slice(10, 20)], ["upload-old", 3, contents.slice(30)], ]); expect(current().targets["https://rakazo.example.com#space_1"]?.uploads).toEqual({}); }); it("aborts on a checksum mismatch without starting an upload", async () => { const { helper } = fakeHelper({ size: 1 }); vi.mocked(helper.exportOriginal).mockImplementationOnce(async (_id, destination) => { const file = path.join(destination, "IMG.MOV"); await writeFile(file, "tampered"); return { path: file, fileName: "IMG.MOV", sizeBytes: 8, sha256: sha("original") }; }); const { api } = fakeApi({ assets: [awaitingAssets[0]!] }); const uploadChunk = vi.fn(); await expect( syncApplePhotos({ ...baseOptions(), helper, api, store: memoryStore().store, uploadChunk }), ).rejects.toMatchObject({ code: "integrity" }); expect(api.media.uploads.begin).not.toHaveBeenCalled(); expect(uploadChunk).not.toHaveBeenCalled(); await expect(readdir(tempRoot)).resolves.toEqual([]); }); it("cancels a running upload", async () => { const { helper } = fakeHelper({ size: 1 }); const { api } = fakeApi({ assets: [awaitingAssets[0]!], chunkBytes: 4 }); const controller = new AbortController(); const uploadChunk = vi.fn( async (_id: string, index: number, _bytes: Uint8Array, signal: AbortSignal) => { if (index === 1) controller.abort(); signal.throwIfAborted(); }, ); await expect( syncApplePhotos({ ...baseOptions(), signal: controller.signal, helper, api, store: memoryStore().store, uploadChunk, }), ).rejects.toMatchObject({ code: "cancelled" }); expect(uploadChunk).toHaveBeenCalledTimes(2); 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", () => { it("keeps a stable device id across launches", async () => { const dir = await mkdtemp(path.join(tmpdir(), "rakazo-photos-state-")); try { const store = createPhotosSyncFileStore(dir, () => "device-a"); const first = await store.read(); expect(first.deviceId).toBe("device-a"); first.targets.x = { sourceId: "s", listing: null, changeToken: "t", uploads: {} }; await store.write(first); const reopened = await createPhotosSyncFileStore(dir, () => "device-b").read(); expect(reopened).toEqual(first); expect(await readFile(path.join(dir, "media-photos-sync.json"), "utf8")).toContain( "device-a", ); } finally { await rm(dir, { recursive: true, force: true }); } }); it("drops unreadable state and malformed targets", () => { expect(parsePhotosSyncState("{")).toBeNull(); expect( parsePhotosSyncState(JSON.stringify({ version: 2, deviceId: "d", targets: {} })), ).toBeNull(); expect( parsePhotosSyncState( JSON.stringify({ version: 1, deviceId: "d", targets: { good: { sourceId: "s", listing: null, changeToken: null, uploads: {} }, bad: { sourceId: 4, listing: null, changeToken: null, uploads: {} }, badListing: { sourceId: "s", listing: { cursor: 3 }, changeToken: null, uploads: {} }, }, }), )?.targets, ).toEqual({ good: { sourceId: "s", listing: null, changeToken: null, uploads: {} } }); }); }); describe("chunk uploader", () => { it("PUTs raw bytes with the session and space", async () => { const fetch = vi.fn(async () => new Response(null, { status: 204 })); const upload = createChunkUploader({ origin: "https://rakazo.example.com", spaceId: "space_1", fetch, }); const signal = new AbortController().signal; await upload("up/1", 3, new Uint8Array([1, 2, 3]), signal); expect(fetch).toHaveBeenCalledWith( "https://rakazo.example.com/api/media/uploads/up%2F1/chunks/3", expect.objectContaining({ method: "PUT", credentials: "include", redirect: "error", signal, headers: { "content-type": "application/octet-stream", "x-rakazo-space-id": "space_1" }, }), ); }); it("rejects failed chunks and non-origin targets", async () => { const fetch = vi.fn(async () => new Response(null, { status: 409 })); const upload = createChunkUploader({ origin: "https://rakazo.example.com", spaceId: null, fetch, }); await expect( upload("u", 0, new Uint8Array(), new AbortController().signal), ).rejects.toMatchObject({ code: "upload_failed", }); expect(() => createChunkUploader({ origin: "https://rakazo.example.com/rpc", spaceId: null, fetch }), ).toThrow(); expect(() => createChunkUploader({ origin: "file:///etc", spaceId: null, fetch })).toThrow(); }); }); describe("runPool", () => { it("bounds concurrency and reports the first failure", async () => { let active = 0; let max = 0; const seen: number[] = []; await expect( runPool([1, 2, 3, 4, 5], 2, new AbortController().signal, async (item, signal) => { active += 1; max = Math.max(max, active); await new Promise((resolve) => setTimeout(resolve, 5)); active -= 1; if (item === 2) throw new Error("boom"); signal.throwIfAborted(); seen.push(item); }), ).rejects.toThrow("boom"); expect(max).toBe(2); expect(seen).not.toContain(4); }); });