tech-log-frontend/tests/unit/opfs-byte-store.test.ts

import { describe, expect, it, vi } from "vitest";

import type {
  OpfsJournalPage,
  OpfsJournalPort,
  OpfsJournalTransaction,
  OpfsPreparedObject,
  OpfsStorageScope,
} from "../../src/application/ports/browser-file-storage/opfs-ports.ts";
import type {
  BrowserDataResult,
  BrowserStoragePolicy,
  ByteSource,
} from "../../src/application/ports/browser-file-storage/shared.ts";
import {
  createOpfsByteStoreAdapter,
  type OpfsMaintenanceAuthorityConsumer,
  type OpfsMaintenanceAuthorityProvider,
} from "../../src/adapters/storage/opfs/opfs-byte-store-adapter.ts";
import {
  resolveOpfsRuntimePolicy,
} from "../../src/adapters/storage/opfs/opfs-policy.ts";
import type {
  OpfsWorkerGateway,
  PreparePhysicalObjectRequest,
} from "../../src/adapters/storage/opfs/opfs-worker-protocol.ts";
import {
  writeWithSyncAccessHandle,
} from "../../src/adapters/storage/opfs/opfs-worker-runtime.ts";
import {
  browserDataFailure,
  browserDataSuccess,
} from "../../src/adapters/browser-file-storage/result.ts";

const storagePolicy: BrowserStoragePolicy = Object.freeze({
  owner: "test-owner",
  namespace: "durable-objects",
  classification: "PERSONAL",
  authority: "LOCAL_FIRST",
  accountScope: "OPAQUE_PARTITION",
  retention: Object.freeze({ kind: "EXPLICIT_DELETE" }),
  softBudgetBytes: 1024 * 1024,
  hardBudgetBytes: 2 * 1024 * 1024,
  evictionPriority: "USER_AUTHORED",
  logoutAction: "EXPORT_THEN_PURGE",
  accountDeletionAction: "PURGE_PARTITION",
  pressureAction: "RETAIN",
  unavailableFallback: "EXPORT_REQUIRED",
});

const scope: OpfsStorageScope = Object.freeze({
  namespace: "durable-objects",
  authorityToken: "authority_12345678",
  namespaceToken: "namespace_12345678",
  partitionToken: "partition_12345678",
});

function sourceFrom(bytes: Uint8Array): ByteSource {
  return {
    byteLength: bytes.byteLength,
    async *stream(signal: AbortSignal) {
      if (signal.aborted) {
        yield browserDataFailure("ABORTED", "OBJECT_READ");
        return;
      }
      yield browserDataSuccess(Uint8Array.from(bytes));
    },
  };
}

function preparedFrom(
  request: PreparePhysicalObjectRequest,
): OpfsPreparedObject {
  return Object.freeze({
    physicalSchemaVersion: 1,
    descriptor: Object.freeze({
      ...request.descriptor,
      integrity: Object.freeze({
        algorithm: "SHA-256-TREE-V1",
        rootDigestHex: "a".repeat(64),
        chunkSizeBytes: 64 * 1024,
      }),
    }),
    chunks: Object.freeze([
      Object.freeze({
        sequence: 0,
        byteLength: request.descriptor.byteLength,
        digestHex: "b".repeat(64),
      }),
    ]),
  });
}

function committedObject(
  objectId: string,
  objectPolicy: BrowserStoragePolicy,
  createdAtEpochMs: number,
  byteLength = 1,
): OpfsPreparedObject {
  return {
    physicalSchemaVersion: 1,
    descriptor: {
      objectId,
      scope,
      generation: 1,
      byteLength,
      mediaType: "application/octet-stream",
      createdAtEpochMs,
      integrity: {
        algorithm: "SHA-256-TREE-V1",
        rootDigestHex: "a".repeat(64),
        chunkSizeBytes: 64 * 1024,
      },
      storagePolicy: objectPolicy,
    },
    chunks: [
      {
        sequence: 0,
        byteLength,
        digestHex: "b".repeat(64),
      },
    ],
  };
}

function createJournal(): OpfsJournalPort & {
  readonly transactions: Map<string, OpfsJournalTransaction>;
  readonly objects: Map<string, OpfsPreparedObject>;
} {
  const transactions = new Map<string, OpfsJournalTransaction>();
  const objects = new Map<string, OpfsPreparedObject>();
  return {
    transactions,
    objects,
    async getCommittedObject(_scope, objectId) {
      return browserDataSuccess(objects.get(objectId) ?? null);
    },
    async begin(input) {
      const current = objects.get(input.objectId);
      if (
        (input.expectedGeneration === null && current) ||
        (input.expectedGeneration !== null &&
          current?.descriptor.generation !== input.expectedGeneration)
      ) {
        return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
      }
      const transaction: OpfsJournalTransaction = {
        ...input,
        fencingToken: "fencing_12345678",
        phase: "PREPARING",
        budgetReservation: {
          namespace: input.scope.namespace,
          reservedBytes: input.targetByteLength,
          hardBudgetBytes:
            input.targetStoragePolicy.hardBudgetBytes,
        },
      };
      transactions.set(input.transactionId, transaction);
      return browserDataSuccess(transaction);
    },
    async markFilesReady(transactionId, fencingToken, preparedObject) {
      const current = transactions.get(transactionId);
      if (!current || current.fencingToken !== fencingToken) {
        return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
      }
      const updated: OpfsJournalTransaction = {
        ...current,
        phase: "FILES_READY",
        preparedObject,
      };
      transactions.set(transactionId, updated);
      return browserDataSuccess(updated);
    },
    async commitPut(transactionId, fencingToken) {
      const current = transactions.get(transactionId);
      if (
        !current ||
        current.fencingToken !== fencingToken ||
        !current.preparedObject
      ) {
        return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
      }
      objects.set(current.objectId, current.preparedObject);
      const updated: OpfsJournalTransaction = {
        ...current,
        phase: "COMMITTED",
      };
      transactions.set(transactionId, updated);
      return browserDataSuccess(updated);
    },
    async commitDelete(transactionId, fencingToken) {
      const current = transactions.get(transactionId);
      if (!current || current.fencingToken !== fencingToken) {
        return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
      }
      objects.delete(current.objectId);
      const updated: OpfsJournalTransaction = {
        ...current,
        phase: "COMMITTED",
      };
      transactions.set(transactionId, updated);
      return browserDataSuccess(updated);
    },
    async complete(transactionId, fencingToken) {
      const current = transactions.get(transactionId);
      if (!current || current.fencingToken !== fencingToken) {
        return browserDataFailure("CONFLICT", "INDEXEDDB_WRITE");
      }
      transactions.delete(transactionId);
      return browserDataSuccess(undefined);
    },
    async rollback(transactionId) {
      transactions.delete(transactionId);
      return browserDataSuccess(undefined);
    },
    async listIncomplete(limit) {
      const all = [...transactions.values()];
      const page: OpfsJournalPage = {
        transactions: all.slice(0, limit),
        moreAvailable: all.length > limit,
      };
      return browserDataSuccess(page);
    },
    async listCommittedObjects(request) {
      const values = [...objects.values()].filter(
        (object) =>
          object.descriptor.scope.namespaceToken ===
            request.scope.namespaceToken &&
          object.descriptor.scope.partitionToken ===
            request.scope.partitionToken,
      );
      return browserDataSuccess({
        objects: values.slice(0, request.limit),
        nextObjectId: null,
        moreAvailable: values.length > request.limit,
      });
    },
    async isChunkReferenced(_scope, digestHex) {
      return browserDataSuccess(
        [...objects.values()].some((object) =>
          object.chunks.some((chunk) => chunk.digestHex === digestHex),
        ),
      );
    },
  };
}

function createWorker(
  overrides: Partial<OpfsWorkerGateway> = {},
): OpfsWorkerGateway {
  const physical = new Map<string, OpfsPreparedObject>();
  const worker: OpfsWorkerGateway = {
    async capabilities() {
      return browserDataSuccess({
        available: true,
        dedicatedWorkerRequired: true,
        crossContextMutationLockAvailable: true,
        synchronousAccessHandleAvailable: true,
      });
    },
    async preparePut(request) {
      let count = 0;
      for await (const chunk of request.source.stream(
        request.signal ?? new AbortController().signal,
      )) {
        if (!chunk.ok) return chunk;
        count += chunk.value.byteLength;
      }
      if (count !== request.descriptor.byteLength) {
        return browserDataFailure("INTEGRITY_FAILED", "OBJECT_WRITE");
      }
      const prepared = preparedFrom(request);
      physical.set(request.descriptor.objectId, prepared);
      return browserDataSuccess(prepared);
    },
    async verifyObject(preparedObject) {
      return browserDataSuccess(
        physical.has(preparedObject.descriptor.objectId),
      );
    },
    async openObject(preparedObject) {
      return browserDataSuccess(
        sourceFrom(
          new Uint8Array(preparedObject.descriptor.byteLength).fill(7),
        ),
      );
    },
    async removeObject(_scope, objectId) {
      physical.delete(objectId);
      return browserDataSuccess(undefined);
    },
    async cleanupTransaction() {
      return browserDataSuccess({ kind: "CLEANED" as const });
    },
    async abortPreparedPut() {
      return browserDataSuccess({ kind: "CLEANED" as const });
    },
    async finalizePut() {
      return browserDataSuccess(undefined);
    },
    async listOrphanCandidates() {
      return browserDataSuccess({
        safeToSweep: true,
        digests: [],
        moreAvailable: false,
      });
    },
    async deleteOrphanChunk() {
      return browserDataSuccess({
        deleted: false,
        skippedUnsafe: false,
      });
    },
    close() {},
  };
  return Object.assign(worker, overrides);
}

describe("OPFS byte-store coordinator", () => {
  it("publishes bytes only after FILES_READY and an atomic journal commit", async () => {
    const journal = createJournal();
    const observations: unknown[] = [];
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker: createWorker(),
      scope,
      storagePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      createTransactionId: () => "transaction_12345678",
      now: () => 100,
      observer: (event) => observations.push(event),
    });

    const result = await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1, 2, 3])),
    });

    expect(result).toMatchObject({
      ok: true,
      value: { objectId: "object_12345678", generation: 1 },
    });
    expect(journal.objects.get("object_12345678")).toBeDefined();
    expect(journal.transactions.size).toBe(0);
    expect(JSON.stringify(observations)).not.toContain("object_12345678");
  });

  /**
   * NS-04. Awaiting `journal.complete` without checking it reported a settled
   * transaction that nobody settled: the caller saw plain success and success
   * telemetry while the durable row stayed `COMMITTED`, so the reconcile
   * backlog grew invisibly.
   */
  it("does not report a settled write when the journal cannot complete it", async () => {
    const journal = createJournal();
    journal.complete = async () =>
      browserDataFailure("UNAVAILABLE", "INDEXEDDB_WRITE", {
        retryable: true,
        recovery: "RECONCILE",
      });
    const observations: { operation: string; outcome: string }[] = [];
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker: createWorker(),
      scope,
      storagePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      createTransactionId: () => "transaction_12345678",
      now: () => 100,
      observer: (event) =>
        observations.push(event as { operation: string; outcome: string }),
    });

    const result = await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1, 2, 3])),
    });

    expect(result.ok).toBe(false);
    if (!result.ok) {
      // The payload is durable, so recovery is reconciliation, not a rewrite.
      expect(result.error.recovery).toBe("RECONCILE");
      expect(result.error.operation).toBe("OBJECT_WRITE");
    }
    expect(
      observations.some(
        (event) =>
          event.operation === "OBJECT_WRITE" && event.outcome === "SUCCEEDED",
      ),
    ).toBe(false);
    // The committed row stays the reconciler's authority.
    expect(journal.transactions.get("transaction_12345678")?.phase).toBe(
      "COMMITTED",
    );
    expect(journal.objects.get("object_12345678")).toBeDefined();
  });

  it("records maintenance debt when a committed delete cannot be completed", async () => {
    const journal = createJournal();
    const observations: { operation: string; outcome: string }[] = [];
    const policy = {
      ...resolveOpfsRuntimePolicy(),
      chunkSizeBytes: 64 * 1024,
      maxObjectBytes: 64 * 1024,
      maxChunkCount: 1,
    };
    const writer = createOpfsByteStoreAdapter({
      journal,
      worker: createWorker(),
      scope,
      storagePolicy,
      policy,
      createTransactionId: () => "transaction_12345678",
      now: () => 100,
    });
    await writer.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1, 2, 3])),
    });

    // The journal can commit the deletion but cannot settle the transaction.
    const remover = createOpfsByteStoreAdapter({
      journal: {
        ...journal,
        complete: async () =>
          browserDataFailure("UNAVAILABLE", "INDEXEDDB_WRITE", {
            retryable: true,
            recovery: "RECONCILE",
          }),
      },
      worker: createWorker(),
      scope,
      storagePolicy,
      policy,
      createTransactionId: () => "transaction_87654321",
      now: () => 200,
      observer: (event) =>
        observations.push(event as { operation: string; outcome: string }),
    });

    // Logical deletion is already committed, so the caller must not be asked to
    // repeat a non-idempotent delete — but it is not a settled success either.
    const result = await remover.objects.remove({
      objectId: "object_12345678",
      expectedGeneration: 1,
    });

    expect(result.ok).toBe(true);
    const deleteOutcomes = observations
      .filter((event) => event.operation === "OBJECT_DELETE")
      .map((event) => event.outcome);
    expect(deleteOutcomes.at(-1)).toBe("DEGRADED");
    expect(deleteOutcomes).not.toContain("SUCCEEDED");
  });

  it("deep-snapshots scope and policy at composition against caller mutation", async () => {
    const journal = createJournal();
    const mutableScope = { ...scope };
    const mutableRetention = {
      kind: "TTL" as const,
      maxAgeMs: 60_000,
    };
    const mutablePolicy = {
      ...storagePolicy,
      retention: mutableRetention,
    };
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker: createWorker(),
      scope: mutableScope,
      storagePolicy: mutablePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      createTransactionId: () => "transaction_snapshot_1234",
      now: () => 100,
    });

    mutableScope.namespace = "mutated-namespace";
    mutableScope.partitionToken = "partition_mutated_1234";
    mutablePolicy.namespace = "mutated-namespace";
    mutableRetention.maxAgeMs = 1;

    const result = await adapter.objects.put({
      objectId: "object_snapshot_1234",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1, 2, 3])),
    });

    expect(result).toMatchObject({
      ok: true,
      value: {
        scope,
        storagePolicy: {
          namespace: scope.namespace,
          retention: { kind: "TTL", maxAgeMs: 60_000 },
        },
      },
    });
    expect(result.ok && Object.isFrozen(result.value.scope)).toBe(true);
    expect(
      result.ok && Object.isFrozen(result.value.storagePolicy.retention),
    ).toBe(true);
  });

  it("snapshots a put request at entry and dependency methods at composition", async () => {
    const journal = createJournal();
    const originalGetCommittedObject =
      journal.getCommittedObject.bind(journal);
    let markJournalEntered!: () => void;
    let releaseJournal!: () => void;
    const journalEntered = new Promise<void>((resolve) => {
      markJournalEntered = resolve;
    });
    const journalRelease = new Promise<void>((resolve) => {
      releaseJournal = resolve;
    });
    journal.getCommittedObject = async (...args) => {
      markJournalEntered();
      await journalRelease;
      return await originalGetCommittedObject(...args);
    };

    const worker = createWorker();
    const originalPreparePut = worker.preparePut.bind(worker);
    const preparePut = vi.fn(originalPreparePut);
    worker.preparePut = preparePut;
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker,
      scope,
      storagePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      createTransactionId: () => "transaction_request_snapshot",
      now: () => 100,
    });

    const mutableSource = {
      byteLength: 3,
      async *stream(_signal: AbortSignal) {
        yield browserDataSuccess(new Uint8Array([1, 2, 3]));
      },
    };
    const mutableRequest: {
      objectId: string;
      expectedGeneration: number | null;
      mediaType: string;
      source: typeof mutableSource;
    } = {
      objectId: "object_request_snapshot",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: mutableSource,
    };

    const pending = adapter.objects.put(mutableRequest);
    await journalEntered;

    mutableRequest.objectId = "mutated_object_request";
    mutableRequest.expectedGeneration = 99;
    mutableRequest.mediaType = "text/html";
    mutableSource.byteLength = 999;
    mutableSource.stream = async function* (_signal: AbortSignal) {
      yield browserDataSuccess(new Uint8Array([9]));
    };
    const replacementJournalMethod = vi.fn(async () =>
      browserDataFailure("UNAVAILABLE", "INDEXEDDB_READ"),
    );
    const replacementWorkerMethod = vi.fn(async () =>
      browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"),
    );
    journal.getCommittedObject = replacementJournalMethod;
    worker.preparePut = replacementWorkerMethod;
    releaseJournal();

    const result = await pending;

    expect(result).toMatchObject({
      ok: true,
      value: {
        objectId: "object_request_snapshot",
        byteLength: 3,
        mediaType: "application/octet-stream",
      },
    });
    expect(preparePut).toHaveBeenCalledWith(
      expect.objectContaining({
        descriptor: expect.objectContaining({
          objectId: "object_request_snapshot",
          byteLength: 3,
          mediaType: "application/octet-stream",
        }),
      }),
    );
    expect(replacementJournalMethod).not.toHaveBeenCalled();
    expect(replacementWorkerMethod).not.toHaveBeenCalled();
  });

  it("keeps PREPARING journal when compensating cleanup is aborted or unavailable", async () => {
    for (const failing of [
      {
        async abortPreparedPut() {
          return browserDataFailure("ABORTED", "OBJECT_RECONCILE");
        },
      },
      {
        async abortPreparedPut() {
          return browserDataSuccess({ kind: "EFFECT_UNKNOWN" as const });
        },
      },
    ]) {
      const journal = createJournal();
      const worker = createWorker({
        async preparePut() {
          return browserDataFailure("QUOTA_EXCEEDED", "OBJECT_WRITE");
        },
        ...failing,
      });
      const adapter = createOpfsByteStoreAdapter({
        journal,
        worker,
        scope,
        storagePolicy,
        createTransactionId: () => "transaction_12345678",
        now: () => 100,
      });

      const result = await adapter.objects.put({
        objectId: "object_12345678",
        expectedGeneration: null,
        mediaType: "application/octet-stream",
        source: sourceFrom(new Uint8Array([1])),
      });

      expect(result.ok).toBe(false);
      // The journal row is the only durable evidence that staging bytes may
      // still exist, so it survives an unconfirmed compensation.
      expect(
        journal.transactions.get("transaction_12345678")?.phase,
      ).toBe("PREPARING");
    }
  });

  it("does not roll back journal after an unknown worker mutation effect", async () => {
    const journal = createJournal();
    const worker = createWorker({
      async abortPreparedPut() {
        return browserDataSuccess({ kind: "EFFECT_UNKNOWN" as const });
      },
    });
    const rollback = vi.spyOn(journal, "rollback");
    vi.spyOn(journal, "markFilesReady").mockResolvedValueOnce(
      browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"),
    );
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker,
      scope,
      storagePolicy,
      createTransactionId: () => "transaction_12345678",
      now: () => 100,
    });

    const result = await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1])),
    });

    expect(result.ok).toBe(false);
    expect(rollback).not.toHaveBeenCalled();
  });

  it("holds the OPFS mutation lease until exact physical cleanup completes", async () => {
    const journal = createJournal();
    const observed: Array<Readonly<Record<string, unknown>>> = [];
    const worker = createWorker({
      async preparePut() {
        return browserDataFailure("QUOTA_EXCEEDED", "OBJECT_WRITE");
      },
      async abortPreparedPut(request) {
        observed.push({ ...request });
        return browserDataSuccess({ kind: "CLEANED" as const });
      },
    });
    const compensation = new AbortController();
    const caller = new AbortController();
    caller.abort();
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker,
      scope,
      storagePolicy,
      createTransactionId: () => "transaction_12345678",
      createPhysicalGenerationId: () => "f".repeat(32) as never,
      compensationSignal: compensation.signal,
      now: () => 100,
    });

    await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1])),
    });

    expect(observed).toHaveLength(1);
    expect(observed[0]).toMatchObject({
      transactionId: "transaction_12345678",
      physicalGenerationId: "f".repeat(32),
    });
    // Compensation never inherits the caller signal.
    expect(observed[0]?.signal).toBe(compensation.signal);
    expect(
      journal.transactions.has("transaction_12345678"),
    ).toBe(false);
  });

  it("delayed stale cleanup cannot delete a reused logical generation", async () => {
    const journal = createJournal();
    const issued: string[] = [];
    let nextToken = 0;
    const worker = createWorker({
      async abortPreparedPut(request) {
        issued.push(request.physicalGenerationId);
        return browserDataSuccess({ kind: "CLEANED" as const });
      },
    });
    let transaction = 0;
    // T1 abandons its prepared put at the same logical generation 1.
    vi.spyOn(journal, "markFilesReady").mockResolvedValueOnce(
      browserDataFailure("UNAVAILABLE", "OBJECT_WRITE"),
    );
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker,
      scope,
      storagePolicy,
      createTransactionId: () => `transaction_1234567${(transaction += 1)}`,
      createPhysicalGenerationId: () =>
        String(nextToken += 1).padStart(32, "0") as never,
      now: () => 100,
    });
    const first = await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1])),
    });
    expect(first.ok).toBe(false);

    // T2 legitimately reuses logical generation 1 with a different token.
    const second = await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1])),
    });
    expect(second.ok).toBe(true);
    if (second.ok) expect(second.value.generation).toBe(1);

    // The stale compensation targeted only T1's physical token.
    expect(issued).toEqual([String(1).padStart(32, "0")]);
    const stored = journal.objects.get("object_12345678");
    expect(stored?.physicalSchemaVersion).toBe(1);
  });

  /**
   * STO-RR-01. Finalization runs after the commit fence, so the payload is
   * durable and the journal row must survive for reconciliation. What the
   * caller must not be told is that the write settled: the previous generation
   * and the staging directory are still there.
   */
  it("reports a failed finalization instead of a plain success", async () => {
    const journal = createJournal();
    const worker = createWorker({
      async finalizePut() {
        return browserDataFailure("UNAVAILABLE", "OBJECT_RECONCILE", {
          retryable: true,
          recovery: "RETRY",
        });
      },
    });
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker,
      scope,
      storagePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      createTransactionId: () => "transaction_12345678",
      now: () => 100,
    });

    const result = await adapter.objects.put({
      objectId: "object_12345678",
      expectedGeneration: null,
      mediaType: "application/octet-stream",
      source: sourceFrom(new Uint8Array([1])),
    });

    expect(result.ok).toBe(false);
    expect(result.ok ? null : result.error.operation).toBe("OBJECT_WRITE");
    expect(
      journal.transactions.get("transaction_12345678")?.phase,
    ).toBe("COMMITTED");
  });

  it("closes a sync access handle even when flush fails", async () => {
    const calls: string[] = [];
    const fakeFileHandle = {
      async createSyncAccessHandle() {
        return {
          truncate: (size: number) => calls.push(`truncate:${size}`),
          write: (part: ArrayBufferView) => {
            calls.push(`write:${part.byteLength}`);
            return part.byteLength;
          },
          flush: () => {
            calls.push("flush");
            throw new Error("disk failure");
          },
          close: () => calls.push("close"),
        };
      },
    };

    await expect(
      writeWithSyncAccessHandle(
        fakeFileHandle as never,
        new Uint8Array([1, 2, 3]),
      ),
    ).rejects.toThrow("disk failure");
    expect(calls.at(-1)).toBe("close");
  });

  it("enforces TTL and logout policy within the runtime-bound partition", async () => {
    const journal = createJournal();
    journal.objects.set(
      "expired_12345678",
      committedObject(
        "expired_12345678",
        {
          ...storagePolicy,
          retention: { kind: "TTL", maxAgeMs: 10 },
        },
        1,
        7,
      ),
    );
    journal.objects.set(
      "logout_12345678",
      committedObject(
        "logout_12345678",
        {
          ...storagePolicy,
          retention: { kind: "TTL", maxAgeMs: 10 },
        },
        95,
        9,
      ),
    );
    const removed: string[] = [];
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker: createWorker({
        async removeObject(_scope, objectId) {
          removed.push(objectId);
          return browserDataSuccess(undefined);
        },
      }),
      scope,
      storagePolicy: {
        ...storagePolicy,
        retention: { kind: "TTL", maxAgeMs: 10 },
      },
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      createTransactionId: (() => {
        let sequence = 0;
        return () => `transaction_policy_${++sequence}_1234`;
      })(),
      now: () => 100,
      requestMaintenanceAuthority: async ({ reason }) =>
        reason === "LOGOUT"
          ? {
              authorized: true,
              proofToken: "receipt_1234567890",
              expiresAtEpochMs: 200,
            }
          : { authorized: false },
      consumeMaintenanceAuthority: async ({
        reason,
        proofToken,
        expiresAtEpochMs,
      }) =>
        reason === "LOGOUT" &&
        proofToken === "receipt_1234567890" &&
        expiresAtEpochMs === 200,
    });

    expect(
      await adapter.maintenance.enforcePolicies({
        reason: "TTL",
      }),
    ).toMatchObject({
      ok: true,
      value: { removedObjects: 1, releasedBytes: 7 },
    });
    expect(removed).toEqual(["expired_12345678"]);
    expect(
      await adapter.maintenance.enforcePolicies({
        reason: "LOGOUT",
      }),
    ).toMatchObject({
      ok: true,
      value: { removedObjects: 1, releasedBytes: 9 },
    });
    expect(removed).toEqual([
      "expired_12345678",
      "logout_12345678",
    ]);
  });

  it("pressure eviction skips user-authored data and rejects invalid maintenance budgets", async () => {
    const journal = createJournal();
    const reconstructablePolicy: BrowserStoragePolicy = {
      owner: "test-owner",
      namespace: scope.namespace,
      classification: "PUBLIC",
      authority: "RECONSTRUCTABLE",
      accountScope: "ORIGIN_SHARED",
      retention: { kind: "TTL", maxAgeMs: 60_000 },
      softBudgetBytes: 100,
      hardBudgetBytes: 200,
      evictionPriority: "RECONSTRUCTABLE",
      logoutAction: "KEEP_ORIGIN_SHARED",
      accountDeletionAction: "KEEP_ORIGIN_SHARED",
      pressureAction: "EVICT_RECONSTRUCTABLE",
      unavailableFallback: "ONLINE_ONLY",
    };
    journal.objects.set(
      "cacheable_12345678",
      committedObject(
        "cacheable_12345678",
        reconstructablePolicy,
        1,
        12,
      ),
    );
    journal.objects.set(
      "authored_12345678",
      committedObject(
        "authored_12345678",
        storagePolicy,
        2,
        20,
      ),
    );
    const removed: string[] = [];
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker: createWorker({
        async removeObject(_scope, objectId) {
          removed.push(objectId);
          return browserDataSuccess(undefined);
        },
      }),
      scope,
      storagePolicy: reconstructablePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
        reconciliationBudgetMs: 10,
        reconciliationBatchSize: 1,
      },
      createTransactionId: () => "transaction_pressure_1234",
      now: () => 100,
    });

    expect(
      await adapter.maintenance.enforcePolicies({
        reason: "PRESSURE",
        targetBytesToRelease: 1,
      }),
    ).toMatchObject({
      ok: true,
      value: { removedObjects: 1, releasedBytes: 12 },
    });
    expect(removed).toEqual(["cacheable_12345678"]);
    expect(journal.objects.has("authored_12345678")).toBe(true);

    for (const invalid of [0, -1, Number.NaN]) {
      expect(
        await adapter.maintenance.reconcile({ budgetMs: invalid }),
      ).toMatchObject({
        ok: false,
        error: { code: "INVALID_INPUT" },
      });
      expect(
        await adapter.maintenance.enforcePolicies({
          reason: "TTL",
          maxObjects: invalid,
        }),
      ).toMatchObject({
        ok: false,
        error: { code: "INVALID_INPUT" },
      });
    }
    expect(
      await adapter.maintenance.reconcile({ budgetMs: 11 }),
    ).toMatchObject({
      ok: false,
      error: { code: "LIMIT_EXCEEDED" },
    });
    expect(
      await adapter.maintenance.enforcePolicies({
        reason: "PRESSURE",
        targetBytesToRelease: 1,
        maxObjects: 2,
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "LIMIT_EXCEEDED" },
    });
  });

  it("runs bounded orphan GC and fails closed when staging is unreadable", async () => {
    const digestHex = "c".repeat(64);
    const listOrphans = vi.fn(async () =>
      browserDataSuccess({
        safeToSweep: true,
        digests: [digestHex],
        moreAvailable: false,
      }),
    );
    const deleteOrphan = vi.fn(async () =>
      browserDataSuccess({
        deleted: true,
        skippedUnsafe: false,
      }),
    );
    const adapter = createOpfsByteStoreAdapter({
      journal: createJournal(),
      worker: createWorker({
        listOrphanCandidates: listOrphans,
        deleteOrphanChunk: deleteOrphan,
      }),
      scope,
      storagePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      now: () => 100_000,
    });
    expect(await adapter.maintenance.reconcile()).toMatchObject({
      ok: true,
      value: {
        inspectedOrphanChunks: 1,
        deletedOrphanChunks: 1,
        orphanGcStatus: "COMPLETED",
      },
    });
    expect(listOrphans).toHaveBeenCalledWith(
      scope,
      expect.any(Number),
      expect.any(Number),
      undefined,
    );
    expect(deleteOrphan).toHaveBeenCalledWith(
      scope,
      digestHex,
      expect.any(Number),
      undefined,
    );

    const failClosed = createOpfsByteStoreAdapter({
      journal: createJournal(),
      worker: createWorker({
        async listOrphanCandidates() {
          return browserDataSuccess({
            safeToSweep: false,
            digests: [],
            moreAvailable: false,
          });
        },
      }),
      scope,
      storagePolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
      },
      now: () => 100_000,
    });
    expect(await failClosed.maintenance.reconcile()).toMatchObject({
      ok: true,
      value: {
        orphanGcStatus: "STAGING_STATE_UNREADABLE",
        deletedOrphanChunks: 0,
      },
    });
  });

  it("does not scan and reports remaining work when authority consumes the maintenance budget", async () => {
    const syncedPolicy: BrowserStoragePolicy = {
      ...storagePolicy,
      authority: "SERVER",
      retention: { kind: "UNTIL_SYNCED" },
      evictionPriority: "SYNCED_COPY",
      unavailableFallback: "READ_ONLY",
    };
    const journal = createJournal();
    journal.objects.set(
      "synced_budget_1234",
      committedObject("synced_budget_1234", syncedPolicy, 1, 11),
    );
    const listCommittedObjects = vi.spyOn(
      journal,
      "listCommittedObjects",
    );
    const worker = createWorker();
    const removeObject = vi.spyOn(worker, "removeObject");
    let epochMs = 100;
    const adapter = createOpfsByteStoreAdapter({
      journal,
      worker,
      scope,
      storagePolicy: syncedPolicy,
      policy: {
        ...resolveOpfsRuntimePolicy(),
        chunkSizeBytes: 64 * 1024,
        maxObjectBytes: 64 * 1024,
        maxChunkCount: 1,
        reconciliationBudgetMs: 10,
        reconciliationBatchSize: 1,
      },
      now: () => epochMs,
      requestMaintenanceAuthority: async () => {
        epochMs = 110;
        return {
          authorized: true,
          proofToken: "budget_1234567890",
          expiresAtEpochMs: 200,
        };
      },
      consumeMaintenanceAuthority: async () => true,
    });

    expect(
      await adapter.maintenance.enforcePolicies({
        reason: "UNTIL_SYNCED",
      }),
    ).toEqual({
      ok: true,
      value: {
        inspectedObjects: 0,
        removedObjects: 0,
        releasedBytes: 0,
        moreObjectsAvailable: true,
        deadlineReached: true,
      },
    });
    expect(listCommittedObjects).not.toHaveBeenCalled();
    expect(removeObject).not.toHaveBeenCalled();
    expect(journal.objects.has("synced_budget_1234")).toBe(true);
  });

  it("obtains and atomically consumes composition authority for sensitive maintenance", async () => {
    const syncedPolicy: BrowserStoragePolicy = {
      ...storagePolicy,
      authority: "SERVER",
      retention: { kind: "UNTIL_SYNCED" },
      evictionPriority: "SYNCED_COPY",
      unavailableFallback: "READ_ONLY",
    };
    const createAdapter = (
      authority: Readonly<{
        requestMaintenanceAuthority?: OpfsMaintenanceAuthorityProvider;
        consumeMaintenanceAuthority?: OpfsMaintenanceAuthorityConsumer;
      }> = {},
    ) => {
      const journal = createJournal();
      journal.objects.set(
        "synced_12345678",
        committedObject(
          "synced_12345678",
          syncedPolicy,
          1,
          11,
        ),
      );
      const removed: string[] = [];
      return {
        journal,
        removed,
        adapter: createOpfsByteStoreAdapter({
          journal,
          worker: createWorker({
            async removeObject(_scope, objectId) {
              removed.push(objectId);
              return browserDataSuccess(undefined);
            },
          }),
          scope,
          storagePolicy: syncedPolicy,
          policy: {
            ...resolveOpfsRuntimePolicy(),
            chunkSizeBytes: 64 * 1024,
            maxObjectBytes: 64 * 1024,
            maxChunkCount: 1,
          },
          createTransactionId: () => "transaction_proof_1234",
          now: () => 100,
          ...authority,
        }),
      };
    };

    const noAuthority = createAdapter();
    expect(
      await noAuthority.adapter.maintenance.enforcePolicies({
        reason: "UNTIL_SYNCED",
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(noAuthority.removed).toEqual([]);
    expect(
      await noAuthority.adapter.maintenance.enforcePolicies({
        reason: "LOGOUT",
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(noAuthority.removed).toEqual([]);

    const deniedConsumer = vi.fn(async () => true);
    const denied = createAdapter({
      requestMaintenanceAuthority: async () => ({ authorized: false }),
      consumeMaintenanceAuthority: deniedConsumer,
    });
    expect(
      await denied.adapter.maintenance.enforcePolicies({
        reason: "UNTIL_SYNCED",
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(denied.removed).toEqual([]);
    expect(deniedConsumer).not.toHaveBeenCalled();

    let markProviderEntered!: () => void;
    let releaseProvider!: () => void;
    const providerEntered = new Promise<void>((resolve) => {
      markProviderEntered = resolve;
    });
    const providerRelease = new Promise<void>((resolve) => {
      releaseProvider = resolve;
    });
    const providerImplementation: OpfsMaintenanceAuthorityProvider =
      async ({ reason }) => {
        markProviderEntered();
        await providerRelease;
        return reason === "UNTIL_SYNCED"
          ? {
              authorized: true,
              proofToken: "receipt_1234567890",
              expiresAtEpochMs: 200,
            }
          : { authorized: false };
      };
    const provider = vi.fn(providerImplementation);
    const consumedProofs = new Set<string>();
    const consumerImplementation: OpfsMaintenanceAuthorityConsumer =
      async (request) => {
        if (
          request.reason !== "UNTIL_SYNCED" ||
          request.scope.partitionToken !== scope.partitionToken ||
          request.storagePolicy.retention.kind !== "UNTIL_SYNCED" ||
          request.expiresAtEpochMs !== 200 ||
          consumedProofs.has(request.proofToken)
        ) {
          return false;
        }
        consumedProofs.add(request.proofToken);
        return true;
      };
    const consumer = vi.fn(consumerImplementation);
    const authorized = createAdapter({
      requestMaintenanceAuthority: provider,
      consumeMaintenanceAuthority: consumer,
    });
    const mutableMaintenanceRequest = {
      reason: "UNTIL_SYNCED" as const,
      maxObjects: 1,
    };
    const pendingMaintenance =
      authorized.adapter.maintenance.enforcePolicies(
        mutableMaintenanceRequest,
      );
    await providerEntered;
    const mutationTarget = mutableMaintenanceRequest as {
      reason: string;
      maxObjects: number;
    };
    mutationTarget.reason = "ACCOUNT_DELETION";
    mutationTarget.maxObjects = 999;
    releaseProvider();

    expect(await pendingMaintenance).toMatchObject({
      ok: true,
      value: { removedObjects: 1, releasedBytes: 11 },
    });
    expect(provider).toHaveBeenCalledWith(
      expect.objectContaining({
        reason: "UNTIL_SYNCED",
        scope,
        storagePolicy: syncedPolicy,
      }),
    );
    expect(consumer).toHaveBeenCalledWith(
      expect.objectContaining({
        reason: "UNTIL_SYNCED",
        proofToken: "receipt_1234567890",
        expiresAtEpochMs: 200,
        scope,
        storagePolicy: syncedPolicy,
      }),
    );
    expect(authorized.removed).toEqual(["synced_12345678"]);

    authorized.journal.objects.set(
      "synced_12345678",
      committedObject("synced_12345678", syncedPolicy, 1, 11),
    );
    expect(
      await authorized.adapter.maintenance.enforcePolicies({
        reason: "UNTIL_SYNCED",
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(authorized.removed).toEqual(["synced_12345678"]);

    const expiredConsumer = vi.fn(async () => true);
    const expired = createAdapter({
      requestMaintenanceAuthority: async () => ({
        authorized: true,
        proofToken: "expired_1234567890",
        expiresAtEpochMs: 100,
      }),
      consumeMaintenanceAuthority: expiredConsumer,
    });
    expect(
      await expired.adapter.maintenance.enforcePolicies({
        reason: "UNTIL_SYNCED",
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(expiredConsumer).not.toHaveBeenCalled();
    expect(expired.removed).toEqual([]);

    const accountDeletionDenied = createAdapter({
      requestMaintenanceAuthority: async () => ({ authorized: false }),
      consumeMaintenanceAuthority: async () => true,
    });
    expect(
      await accountDeletionDenied.adapter.maintenance.enforcePolicies({
        reason: "ACCOUNT_DELETION",
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(accountDeletionDenied.removed).toEqual([]);
  });
});