evOo-local/apps/desktop/src/media-photos-sync.ts

import { createHash, randomUUID } from "node:crypto";
import { createReadStream } from "node:fs";
import { mkdtemp, open, rm, stat } from "node:fs/promises";
import path from "node:path";
import type { ClientMediaItem, MediaAsset, MediaUpload, mediaRpcShapes } from "@rakazo/contracts";
import type { PhotosHelper } from "./media-photos-helper.js";
import { PHOTOS_MAX_PAGE } from "./media-photos-helper.js";
import { readPrivateFile, writePrivateFile } from "./setup-store.js";

/*
 * Apple Photos is a client-pushed source: the API may run on another machine, so the
 * app never sends a local path. It pushes metadata in batches and uploads the originals
 * the server asks for (`awaiting_upload`) in resumable raw-byte chunks.
 */

type Shapes = typeof mediaRpcShapes;
type Shape = { input: { _input: unknown }; output: { _output: unknown } };
/** One oRPC procedure as `ContractRouterClient<AppContract>` exposes it. */
type Procedure<S extends Shape> = (
  input: S["input"]["_input"],
  options?: { signal?: AbortSignal },
) => Promise<S["output"]["_output"]>;

/** The slice of the app RPC client the sync needs; `rpc` from a contract client satisfies it. */
export interface MediaSyncApi {
  media: {
    sources: {
      registerClient: Procedure<Shapes["sources"]["registerClient"]>;
      pushItems: Procedure<Shapes["sources"]["pushItems"]>;
    };
    assets: { search: Procedure<Shapes["assets"]["search"]> };
    uploads: {
      begin: Procedure<Shapes["uploads"]["begin"]>;
      status: Procedure<Shapes["uploads"]["status"]>;
      complete: Procedure<Shapes["uploads"]["complete"]>;
    };
  };
}

/** Sends one raw chunk to `PUT /api/media/uploads/:uploadId/chunks/:index`. */
export type UploadChunk = (
  uploadId: string,
  index: number,
  bytes: Uint8Array<ArrayBuffer>,
  signal: AbortSignal,
) => Promise<void>;

export type PhotosSyncErrorCode = "not_authorized" | "cancelled" | "integrity" | "upload_failed";

export class PhotosSyncError extends Error {
  constructor(
    readonly code: PhotosSyncErrorCode,
    message: string,
  ) {
    super(message);
    this.name = "PhotosSyncError";
  }
}

// ---------------------------------------------------------------------------
// Persistent state
// ---------------------------------------------------------------------------

export interface PhotosSyncTargetState {
  sourceId: string;
  /** A full listing in progress: where to resume, the token from its start and the last pushed item. */
  listing: { cursor: string | null; token: string | null; lastItem: ClientMediaItem | null } | null;
  /** Set after a complete listing; later syncs only read changes since it. */
  changeToken: string | null;
  /** Open uploads by asset id so an interrupted upload resumes instead of restarting. */
  uploads: Record<string, string>;
}

export interface PhotosSyncState {
  version: 1;
  /** Stable per installation; the server keys the Apple Photos source by it. */
  deviceId: string;
  /** Keyed by server origin and space, since each has its own source. */
  targets: Record<string, PhotosSyncTargetState>;
}

export interface PhotosSyncStore {
  read(): Promise<PhotosSyncState>;
  write(state: PhotosSyncState): Promise<void>;
}

export const PHOTOS_SYNC_FILE_NAME = "media-photos-sync.json";
const MAX_STATE_BYTES = 4 * 1024 * 1024;

/** Kept next to the desktop setup file, private to the user and written atomically. */
export function createPhotosSyncFileStore(
  userDataDir: string,
  newDeviceId: () => string = randomUUID,
): PhotosSyncStore {
  const file = path.join(userDataDir, PHOTOS_SYNC_FILE_NAME);
  return {
    async read() {
      const raw = await readPrivateFile(file, MAX_STATE_BYTES);
      return (raw === null ? null : parsePhotosSyncState(raw)) ?? emptyState(newDeviceId());
    },
    write: (state) => writePrivateFile(file, `${JSON.stringify(state)}\n`),
  };
}

function emptyState(deviceId: string): PhotosSyncState {
  return { version: 1, deviceId, targets: {} };
}

const isRecord = (value: unknown): value is Record<string, unknown> =>
  typeof value === "object" && value !== null && !Array.isArray(value);
const isNullableString = (value: unknown) => value === null || typeof value === "string";

/** Unreadable state starts over: the server deduplicates by device and external id. */
export function parsePhotosSyncState(raw: string): PhotosSyncState | null {
  let value: unknown;
  try {
    value = JSON.parse(raw);
  } catch {
    return null;
  }
  if (
    !isRecord(value) ||
    value.version !== 1 ||
    typeof value.deviceId !== "string" ||
    value.deviceId === "" ||
    value.deviceId.length > 200 ||
    !isRecord(value.targets)
  ) {
    return null;
  }
  const targets: Record<string, PhotosSyncTargetState> = {};
  for (const [key, target] of Object.entries(value.targets)) {
    if (
      !isRecord(target) ||
      typeof target.sourceId !== "string" ||
      !isNullableString(target.changeToken) ||
      !isRecord(target.uploads) ||
      !Object.values(target.uploads).every((id) => typeof id === "string")
    ) {
      continue;
    }
    const listing = target.listing;
    const validListing =
      listing === null ||
      (isRecord(listing) &&
        isNullableString(listing.cursor) &&
        isNullableString(listing.token) &&
        (listing.lastItem === null || isRecord(listing.lastItem)));
    if (!validListing) continue;
    targets[key] = target as unknown as PhotosSyncTargetState;
  }
  return { version: 1, deviceId: value.deviceId, targets };
}

// ---------------------------------------------------------------------------
// Chunk upload over HTTP
// ---------------------------------------------------------------------------

/**
 * Chunk PUTs ride the same signed-in session as the app window: the injected `fetch`
 * must carry that session's cookies (Electron `session.fetch` with credentials).
 */
export function createChunkUploader(options: {
  origin: string;
  spaceId: string | null;
  fetch: (url: string, init: RequestInit) => Promise<Response>;
}): UploadChunk {
  const origin = new URL(options.origin);
  if (!["http:", "https:"].includes(origin.protocol) || origin.origin !== options.origin) {
    throw new Error("Uploads need the server origin.");
  }
  return async (uploadId, index, bytes, signal) => {
    if (!Number.isSafeInteger(index) || index < 0) throw new Error("Invalid chunk index.");
    const headers: Record<string, string> = { "content-type": "application/octet-stream" };
    if (options.spaceId) headers["x-rakazo-space-id"] = options.spaceId;
    const response = await options.fetch(
      `${origin.origin}/api/media/uploads/${encodeURIComponent(uploadId)}/chunks/${index}`,
      {
        method: "PUT",
        headers,
        body: bytes,
        credentials: "include",
        redirect: "error",
        signal,
      },
    );
    if (!response.ok) {
      throw new PhotosSyncError(
        "upload_failed",
        `The server rejected an upload (${response.status}).`,
      );
    }
  };
}

// ---------------------------------------------------------------------------
// Sync
// ---------------------------------------------------------------------------

export interface PhotosSyncProgress {
  phase: "authorizing" | "indexing" | "uploading";
  /** Items pushed in this run. */
  indexed: number;
  uploadsDone: number;
  uploadsTotal: number;
}

export interface PhotosSyncResult {
  indexed: number;
  created: number;
  updated: number;
  unchanged: number;
  uploaded: number;
  /** True when this run listed the whole library instead of reading changes. */
  fullListing: boolean;
}

export interface PhotosSyncOptions {
  helper: PhotosHelper;
  api: MediaSyncApi;
  uploadChunk: UploadChunk;
  store: PhotosSyncStore;
  /** Server origin plus space, e.g. `https://rakazo.example.com#space_1`. */
  target: string;
  /** Source label shown in the library, e.g. the Mac's name or "Apple Photos". */
  label: string;
  /** Exports land in a fresh folder below this and are deleted after upload. */
  tempRoot: string;
  signal: AbortSignal;
  onProgress?: (progress: PhotosSyncProgress) => void;
  /** At most 500, the contract's batch limit. */
  batchSize?: number;
  concurrency?: number;
}

const MAX_SYNC_TOKEN_LENGTH = 2000;

export async function syncApplePhotos(options: PhotosSyncOptions): Promise<PhotosSyncResult> {
  const { helper, api, signal } = options;
  const batchSize = Math.min(Math.max(1, options.batchSize ?? PHOTOS_MAX_PAGE), PHOTOS_MAX_PAGE);
  const result: PhotosSyncResult = {
    indexed: 0,
    created: 0,
    updated: 0,
    unchanged: 0,
    uploaded: 0,
    fullListing: false,
  };
  const progress = { uploadsDone: 0, uploadsTotal: 0 };
  const report = (phase: PhotosSyncProgress["phase"]) =>
    options.onProgress?.({ phase, indexed: result.indexed, ...progress });

  const state = await options.store.read();
  // Writes are serialized so concurrent upload workers never interleave partial state.
  let saving = Promise.resolve();
  const save = () => {
    const snapshot = JSON.parse(JSON.stringify(state)) as PhotosSyncState;
    saving = saving.then(() => options.store.write(snapshot));
    return saving;
  };

  try {
    report("authorizing");
    const authorization = await helper.authorize(true, signal);
    if (authorization !== "authorized" && authorization !== "limited") {
      throw new PhotosSyncError("not_authorized", "Photos access was not granted.");
    }

    const ensureTarget = async (): Promise<PhotosSyncTargetState> => {
      const existing = state.targets[options.target];
      if (existing) return existing;
      const source = await api.media.sources.registerClient(
        { kind: "apple_photos", label: options.label, deviceId: state.deviceId },
        { signal },
      );
      const created: PhotosSyncTargetState = {
        sourceId: source.id,
        listing: null,
        changeToken: null,
        uploads: {},
      };
      state.targets[options.target] = created;
      await save();
      return created;
    };

    const push = async (
      target: PhotosSyncTargetState,
      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,
        },
        { signal },
      );
      result.indexed += items.length;
      result.created += counts.created;
      result.updated += counts.updated;
      result.unchanged += counts.unchanged;
      report("indexing");
    };

    /** Full listing, resumable page by page; the last push says `complete`. */
    const listAll = async (target: PhotosSyncTargetState) => {
      result.fullListing = true;
      target.listing ??= { cursor: null, token: null, lastItem: null };
      const listing = target.listing;
      while (true) {
        const page = await helper.list({ cursor: listing.cursor, limit: batchSize }, signal);
        if (listing.cursor === null) listing.token = page.changeToken;
        const done = page.nextCursor === null;
        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;
          await save();
          return;
        }
        listing.cursor = page.nextCursor;
        listing.lastItem = items.at(-1) ?? listing.lastItem;
        await save();
      }
    };

    const syncMetadata = async () => {
      const target = await ensureTarget();
      if (target.listing !== null || target.changeToken === null) {
        await listAll(target);
        return;
      }
      const changes = await helper.changes(target.changeToken, signal);
      if (changes.fullResyncRequired || changes.token === null) {
        target.listing = null;
        await listAll(target);
        return;
      }
      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();
    };

    report("indexing");
    try {
      await syncMetadata();
    } catch (error) {
      // The source was removed on the server: register again and list from scratch.
      if (!isNotFound(error) || signal.aborted) throw error;
      delete state.targets[options.target];
      await save();
      await syncMetadata();
    }

    const target = state.targets[options.target]!;
    const awaiting = await awaitingUploads(api, target.sourceId, signal);
    progress.uploadsTotal = awaiting.length;
    // Forget uploads for assets the server no longer waits for.
    const waiting = new Set(awaiting.map((asset) => asset.id));
    for (const assetId of Object.keys(target.uploads)) {
      if (!waiting.has(assetId)) delete target.uploads[assetId];
    }
    await save();
    report("uploading");
    await runPool(awaiting, options.concurrency ?? 2, signal, async (asset, workerSignal) => {
      await uploadAsset({ ...options, signal: workerSignal }, target, asset, save);
      result.uploaded += 1;
      progress.uploadsDone += 1;
      report("uploading");
    });
    return result;
  } catch (error) {
    if (signal.aborted) throw new PhotosSyncError("cancelled", "Sync cancelled.");
    throw error;
  } finally {
    await saving.catch(() => undefined);
  }
}

/**
 * Binds the sync to the server and space the request names. `connect` builds the RPC
 * client and chunk uploader on the signed-in session for that space.
 */
export function createPhotosSyncRunner(deps: {
  helper: PhotosHelper;
  store: PhotosSyncStore;
  tempRoot: string;
  label: string;
  connect: (
    spaceId: string | null,
  ) => Promise<{ origin: string; api: MediaSyncApi; uploadChunk: UploadChunk }>;
}) {
  return async (request: {
    spaceId: string | null;
    signal: AbortSignal;
    onProgress: (progress: PhotosSyncProgress) => void;
  }): Promise<PhotosSyncResult> => {
    const { origin, api, uploadChunk } = await deps.connect(request.spaceId);
    return syncApplePhotos({
      helper: deps.helper,
      store: deps.store,
      tempRoot: deps.tempRoot,
      label: deps.label,
      api,
      uploadChunk,
      target: `${origin}#${request.spaceId ?? "default"}`,
      signal: request.signal,
      onProgress: request.onProgress,
    });
  };
}

function isNotFound(error: unknown): boolean {
  return isRecord(error) && error.code === "NOT_FOUND";
}

interface AwaitingAsset {
  id: string;
  externalId: string;
}

/** 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,
  signal: AbortSignal,
): Promise<AwaitingAsset[]> {
  const awaiting: AwaitingAsset[] = [];
  let cursor: string | null = null;
  do {
    const page: { items: MediaAsset[]; nextCursor: string | null } = await api.media.assets.search(
      { 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;
      awaiting.push({ id: asset.id, externalId: asset.externalId });
    }
    cursor = page.nextCursor;
  } while (cursor !== null);
  return awaiting;
}

async function uploadAsset(
  options: PhotosSyncOptions,
  target: PhotosSyncTargetState,
  asset: AwaitingAsset,
  save: () => Promise<void>,
): Promise<void> {
  const { api, signal } = options;
  const directory = await mkdtemp(path.join(options.tempRoot, "rakazo-photos-"));
  try {
    const exported = await options.helper.exportOriginal(asset.externalId, directory, { signal });
    // The helper's hash is checked against the bytes on disk before anything is sent.
    const local = await hashFile(exported.path, signal);
    if (local.sha256 !== exported.sha256 || local.sizeBytes !== exported.sizeBytes) {
      throw new PhotosSyncError("integrity", "The exported file does not match its checksum.");
    }

    let upload: MediaUpload | null = null;
    const previous = target.uploads[asset.id];
    if (previous) {
      const status = await api.media.uploads
        .status({ uploadId: previous }, { signal })
        .catch((error: unknown) => {
          if (isNotFound(error)) return null;
          throw error;
        });
      if (status?.status === "complete") {
        delete target.uploads[asset.id];
        await save();
        return;
      }
      if (status?.status === "open" && status.sizeBytes === local.sizeBytes) upload = status;
    }
    if (upload === null) {
      upload = await api.media.uploads.begin(
        {
          assetId: asset.id,
          sizeBytes: local.sizeBytes,
          sha256: local.sha256,
          fileName: exported.fileName.slice(0, 255),
        },
        { signal },
      );
      target.uploads[asset.id] = upload.id;
      await save();
    }

    const received = new Set(upload.receivedChunks);
    const chunkCount = Math.ceil(local.sizeBytes / upload.chunkBytes);
    const file = await open(exported.path, "r");
    try {
      for (let index = 0; index < chunkCount; index += 1) {
        if (received.has(index)) continue;
        signal.throwIfAborted();
        const offset = index * upload.chunkBytes;
        const length = Math.min(upload.chunkBytes, local.sizeBytes - offset);
        const bytes = Buffer.alloc(length);
        const { bytesRead } = await file.read(bytes, 0, length, offset);
        if (bytesRead !== length) {
          throw new PhotosSyncError("integrity", "The exported file changed during upload.");
        }
        await options.uploadChunk(upload.id, index, bytes, signal);
      }
    } finally {
      await file.close();
    }
    const completed = await api.media.uploads.complete({ uploadId: upload.id }, { signal });
    if (completed.status !== "complete") {
      throw new PhotosSyncError("upload_failed", "The server did not accept the upload.");
    }
    delete target.uploads[asset.id];
    await save();
  } finally {
    await rm(directory, { recursive: true, force: true });
  }
}

export async function hashFile(
  file: string,
  signal?: AbortSignal,
): Promise<{ sha256: string; sizeBytes: number }> {
  const hash = createHash("sha256");
  for await (const chunk of createReadStream(file, { signal })) hash.update(chunk as Buffer);
  return { sha256: hash.digest("hex"), sizeBytes: (await stat(file)).size };
}

/**
 * Runs `work` with at most `limit` items in flight. The first failure aborts the
 * others and is the error reported.
 */
export async function runPool<T>(
  items: T[],
  limit: number,
  signal: AbortSignal,
  work: (item: T, signal: AbortSignal) => Promise<void>,
): Promise<void> {
  const controller = new AbortController();
  const forward = () => controller.abort(signal.reason);
  signal.addEventListener("abort", forward, { once: true });
  if (signal.aborted) forward();
  let next = 0;
  let failed = false;
  let failure: unknown;
  const worker = async () => {
    while (!failed && !controller.signal.aborted && next < items.length) {
      const item = items[next++]!;
      try {
        await work(item, controller.signal);
      } catch (error) {
        if (!failed) {
          failed = true;
          failure = error;
          controller.abort(error);
        }
      }
    }
  };
  try {
    await Promise.all(Array.from({ length: Math.min(Math.max(1, limit), items.length) }, worker));
  } finally {
    signal.removeEventListener("abort", forward);
  }
  if (failed) throw failure;
  signal.throwIfAborted();
}