clean-architecture-frontend-template/tests/unit/indexeddb-opfs-journal.test.ts

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

import type {
  BeginOpfsJournalTransaction,
  OpfsPreparedObject,
  OpfsStorageScope,
} from "../../src/application/ports/browser-file-storage/opfs-ports.ts";
import type {
  BrowserStoragePolicy,
} from "../../src/application/ports/browser-file-storage/shared.ts";
import {
  createIndexedDbOpfsJournal,
  opfsJournalDatabaseName,
} from "../../src/adapters/storage/opfs/indexeddb-opfs-journal.ts";
import { MemoryIndexedDbFactory } from "../helpers/memory-indexeddb.ts";

const authorityToken = "authority_12345678";
const scope: OpfsStorageScope = Object.freeze({
  namespace: "durable-objects",
  authorityToken,
  namespaceToken: "namespace_12345678",
  partitionToken: "partition_12345678",
});
const otherPartition: OpfsStorageScope = Object.freeze({
  ...scope,
  partitionToken: "partition_87654321",
});
const policy: BrowserStoragePolicy = Object.freeze({
  owner: "test-owner",
  namespace: scope.namespace,
  classification: "PERSONAL",
  authority: "LOCAL_FIRST",
  accountScope: "OPAQUE_PARTITION",
  retention: Object.freeze({ kind: "EXPLICIT_DELETE" }),
  softBudgetBytes: 8,
  hardBudgetBytes: 10,
  evictionPriority: "USER_AUTHORED",
  logoutAction: "EXPORT_THEN_PURGE",
  accountDeletionAction: "PURGE_PARTITION",
  pressureAction: "RETAIN",
  unavailableFallback: "EXPORT_REQUIRED",
});

function beginInput(
  transactionId: string,
  objectId: string,
  byteLength: number,
  targetScope = scope,
): BeginOpfsJournalTransaction {
  return {
    transactionId,
    mutation: "PUT",
    scope: targetScope,
    objectId,
    expectedGeneration: null,
    targetGeneration: 1,
    targetByteLength: byteLength,
    targetStoragePolicy: {
      ...policy,
      namespace: targetScope.namespace,
    },
    startedAtEpochMs: 100,
  };
}

function prepared(
  input: BeginOpfsJournalTransaction,
): OpfsPreparedObject {
  return {
    physicalSchemaVersion: 1,
    descriptor: {
      objectId: input.objectId,
      scope: input.scope,
      generation: input.targetGeneration,
      byteLength: input.targetByteLength,
      mediaType: "application/octet-stream",
      createdAtEpochMs: input.startedAtEpochMs,
      integrity: {
        algorithm: "SHA-256-TREE-V1",
        rootDigestHex: "a".repeat(64),
        chunkSizeBytes: 64 * 1024,
      },
      storagePolicy: input.targetStoragePolicy,
    },
    chunks:
      input.targetByteLength === 0
        ? []
        : [
            {
              sequence: 0,
              byteLength: input.targetByteLength,
              digestHex: "b".repeat(64),
            },
          ],
  };
}

/**
 * Records every `transaction.abort()` the journal itself calls, with the mode
 * of the transaction it aborted.
 *
 * A failure code alone cannot tell `fail` (record and abort) from
 * `requestFailed` (record only): the fake aborts a transaction on a request
 * error by itself, so both land on the same code while only one of them stops
 * the writes queued behind it.
 */
function trackAborts(memory: MemoryIndexedDbFactory): Readonly<{
  factory: IDBFactory;
  aborts: readonly string[];
}> {
  const aborts: string[] = [];
  const wrapped = new WeakSet<object>();
  const wrapDatabase = (database: IDBDatabase): IDBDatabase => {
    if (wrapped.has(database)) return database;
    wrapped.add(database);
    const openTransaction = database.transaction.bind(database);
    Object.defineProperty(database, "transaction", {
      configurable: true,
      value: (...args: Parameters<IDBDatabase["transaction"]>) => {
        const transaction = openTransaction(...args);
        const abort = transaction.abort.bind(transaction);
        Object.defineProperty(transaction, "abort", {
          configurable: true,
          value: () => {
            aborts.push(transaction.mode);
            abort();
          },
        });
        return transaction;
      },
    });
    return database;
  };
  const factory = {
    cmp: (first: IDBValidKey, second: IDBValidKey) =>
      memory.factory.cmp(first, second),
    open: (name: string, version?: number) => {
      const request = memory.factory.open(name, version);
      let stored: IDBDatabase | undefined = request.result;
      if (stored) wrapDatabase(stored);
      Object.defineProperty(request, "result", {
        configurable: true,
        get: () => stored,
        set: (value: IDBDatabase | undefined) => {
          stored = value ? wrapDatabase(value) : value;
        },
      });
      return request;
    },
  } as unknown as IDBFactory;
  return Object.freeze({ factory, aborts });
}

function createHarness(factory = new MemoryIndexedDbFactory()) {
  let tokenSequence = 0;
  const journal = createIndexedDbOpfsJournal({
    authorityToken,
    factory: factory.factory,
    // `listCommittedObjects` is the only caller that needs a key range, and
    // this environment has no global one.
    keyRange: factory.keyRange,
    createFencingToken: () =>
      `fencing_${String(++tokenSequence).padStart(8, "0")}`,
  });
  return { factory, journal };
}

async function commitObject(
  journal: ReturnType<typeof createHarness>["journal"],
  transactionId: string,
  objectId: string,
  byteLength: number,
): Promise<void> {
  const input = beginInput(transactionId, objectId, byteLength);
  const begun = await journal.begin(input);
  if (!begun.ok) throw new Error(`begin failed for ${objectId}`);
  const token = begun.value.fencingToken;
  const ready = await journal.markFilesReady(
    transactionId,
    token,
    prepared(input),
  );
  if (!ready.ok) throw new Error(`markFilesReady failed for ${objectId}`);
  const committed = await journal.commitPut(transactionId, token);
  if (!committed.ok) throw new Error(`commitPut failed for ${objectId}`);
  const completed = await journal.complete(transactionId, token);
  if (!completed.ok) throw new Error(`complete failed for ${objectId}`);
}

describe("IndexedDB OPFS journal", () => {
  it("fences stale writers and atomically publishes object, budget and chunk refs", async () => {
    const { factory, journal } = createHarness();
    const input = beginInput(
      "transaction_12345678",
      "object_12345678",
      3,
    );
    const begun = await journal.begin(input);
    expect(begun).toMatchObject({ ok: true });
    if (!begun.ok) throw new Error("begin failed");

    expect(
      await journal.markFilesReady(
        input.transactionId,
        "fencing_stale_1234",
        prepared(input),
      ),
    ).toMatchObject({ ok: false, error: { code: "CONFLICT" } });
    expect(
      await journal.markFilesReady(
        input.transactionId,
        begun.value.fencingToken,
        prepared(input),
      ),
    ).toMatchObject({ ok: true, value: { phase: "FILES_READY" } });

    factory.failNextWriteCommit(
      new DOMException("quota", "QuotaExceededError"),
    );
    expect(
      await journal.commitPut(
        input.transactionId,
        begun.value.fencingToken,
      ),
    ).toMatchObject({ ok: false, error: { code: "QUOTA_EXCEEDED" } });
    expect(
      await journal.getCommittedObject(scope, input.objectId),
    ).toEqual({ ok: true, value: null });
    expect(
      await journal.isChunkReferenced(scope, "b".repeat(64)),
    ).toEqual({ ok: true, value: false });

    expect(
      await journal.commitPut(
        input.transactionId,
        begun.value.fencingToken,
      ),
    ).toMatchObject({ ok: true, value: { phase: "COMMITTED" } });
    expect(
      await journal.getCommittedObject(scope, input.objectId),
    ).toMatchObject({
      ok: true,
      value: { descriptor: { byteLength: 3, generation: 1 } },
    });
    expect(
      await journal.isChunkReferenced(scope, "b".repeat(64)),
    ).toEqual({ ok: true, value: true });
    expect(
      await journal.isChunkReferenced(otherPartition, "b".repeat(64)),
    ).toEqual({ ok: true, value: false });
  });

  it("reserves the hard budget in the same transaction and releases it on rollback", async () => {
    const { journal } = createHarness();
    const first = await journal.begin(
      beginInput("transaction_11111111", "object_11111111", 6),
    );
    expect(first).toMatchObject({ ok: true });
    if (!first.ok) throw new Error("begin failed");

    expect(
      await journal.begin(
        beginInput("transaction_22222222", "object_22222222", 5),
      ),
    ).toMatchObject({
      ok: false,
      error: { code: "LIMIT_EXCEEDED" },
    });
    expect(
      await journal.begin(
        beginInput(
          "transaction_33333333",
          "object_33333333",
          10,
          otherPartition,
        ),
      ),
    ).toMatchObject({ ok: true });

    expect(
      await journal.rollback(
        "transaction_11111111",
        first.value.fencingToken,
      ),
    ).toEqual({ ok: true, value: undefined });
    expect(
      await journal.begin(
        beginInput("transaction_44444444", "object_44444444", 5),
      ),
    ).toMatchObject({ ok: true });
  });

  it("commits and completes a DELETE row without requiring a prepared object", async () => {
    const { journal } = createHarness();
    const putInput = beginInput(
      "transaction_55555555",
      "object_55555555",
      2,
    );
    const put = await journal.begin(putInput);
    if (!put.ok) throw new Error("put begin failed");
    await journal.markFilesReady(
      putInput.transactionId,
      put.value.fencingToken,
      prepared(putInput),
    );
    await journal.commitPut(
      putInput.transactionId,
      put.value.fencingToken,
    );
    await journal.complete(
      putInput.transactionId,
      put.value.fencingToken,
    );

    const deletion = await journal.begin({
      transactionId: "transaction_66666666",
      mutation: "DELETE",
      scope,
      objectId: putInput.objectId,
      expectedGeneration: 1,
      targetGeneration: 2,
      targetByteLength: 0,
      targetStoragePolicy: policy,
      startedAtEpochMs: 200,
    });
    if (!deletion.ok) throw new Error("delete begin failed");
    expect(
      await journal.commitDelete(
        "transaction_66666666",
        deletion.value.fencingToken,
      ),
    ).toMatchObject({
      ok: true,
      value: { mutation: "DELETE", phase: "COMMITTED" },
    });
    expect(
      await journal.complete(
        "transaction_66666666",
        deletion.value.fencingToken,
      ),
    ).toEqual({ ok: true, value: undefined });
    expect(
      await journal.getCommittedObject(scope, putInput.objectId),
    ).toEqual({ ok: true, value: null });
  });

  it("binds one opaque physical authority to one deterministic database", async () => {
    expect(opfsJournalDatabaseName(authorityToken)).toBe(
      `ca-frontend-opfs-metadata-v1:${authorityToken}`,
    );
    expect(() =>
      createIndexedDbOpfsJournal({
        authorityToken,
        databaseName: "unrelated-database",
        factory: new MemoryIndexedDbFactory().factory,
      }),
    ).toThrow();

    const { journal } = createHarness();
    const binding = await journal.begin(
      beginInput(
        "transaction_binding_1234",
        "object_binding_12345678",
        1,
      ),
    );
    if (!binding.ok) throw new Error("scope binding failed");
    await journal.rollback(
      "transaction_binding_1234",
      binding.value.fencingToken,
    );

    const remappedScope: OpfsStorageScope = {
      ...scope,
      namespace: "different-namespace",
    };
    expect(
      await journal.begin(
        beginInput(
          "transaction_remap_1234",
          "object_remap_12345678",
          1,
          remappedScope,
        ),
      ),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    const retokenedScope: OpfsStorageScope = {
      ...scope,
      namespaceToken: "namespace_87654321",
    };
    expect(
      await journal.begin(
        beginInput(
          "transaction_retoken_1234",
          "object_retoken_12345678",
          1,
          retokenedScope,
        ),
      ),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });
    expect(
      await journal.begin({
        ...beginInput(
          "transaction_policy_1234",
          "object_policy_12345678",
          1,
        ),
        targetStoragePolicy: {
          ...policy,
          owner: "different-owner",
        },
      }),
    ).toMatchObject({
      ok: false,
      error: { code: "POLICY_REJECTED" },
    });

    const foreignScope: OpfsStorageScope = {
      ...scope,
      authorityToken: "authority_87654321",
    };
    expect(
      await journal.getCommittedObject(
        foreignScope,
        "object_12345678",
      ),
    ).toMatchObject({ ok: false, error: { code: "INVALID_INPUT" } });
    expect(
      await journal.begin(
        beginInput(
          "transaction_77777777",
          "object_77777777",
          1,
          foreignScope,
        ),
      ),
    ).toMatchObject({ ok: false, error: { code: "INVALID_INPUT" } });
  });

  it("maps synchronous open failures and blocked upgrades to Result failures", async () => {
    const throwingFactory = {
      open() {
        throw new DOMException("denied", "SecurityError");
      },
    } as unknown as IDBFactory;
    const throwingJournal = createIndexedDbOpfsJournal({
      authorityToken,
      factory: throwingFactory,
    });
    await expect(
      throwingJournal.getCommittedObject(scope, "object_12345678"),
    ).resolves.toMatchObject({
      ok: false,
      error: { code: "PERMISSION_DENIED" },
    });

    const factory = new MemoryIndexedDbFactory();
    factory.blockNextOpen();
    const callbacks: Array<() => void> = [];
    const blockedJournal = createIndexedDbOpfsJournal({
      authorityToken,
      factory: factory.factory,
      blockedTimeoutMs: 1,
      scheduler: {
        setTimeout(callback) {
          callbacks.push(callback);
          return callback;
        },
        clearTimeout() {},
      },
    });
    const opening = blockedJournal.getCommittedObject(
      scope,
      "object_12345678",
    );
    await Promise.resolve();
    callbacks.forEach((callback) => callback());
    await expect(opening).resolves.toMatchObject({
      ok: false,
      error: { code: "BLOCKED" },
    });
  });

  it("reopens after versionchange and rejects corrupt persisted journal rows", async () => {
    const { factory, journal } = createHarness();
    expect(
      await journal.getCommittedObject(scope, "object_12345678"),
    ).toEqual({ ok: true, value: null });
    factory.triggerVersionChange(2);
    expect(factory.isConnectionClosed()).toBe(true);
    expect(
      await journal.getCommittedObject(scope, "object_12345678"),
    ).toEqual({ ok: true, value: null });

    factory.seed("opfs-journal", {
      transactionId: "transaction_88888888",
      startedAtEpochMs: 1,
      corrupt: true,
    });
    expect(await journal.listIncomplete(10)).toMatchObject({
      ok: false,
      error: { code: "CORRUPT_DATA" },
    });
  });

  it("reports a unique logical-key violation as the request's own ConstraintError", async () => {
    // OP-1. Two open journal rows for one object collide on the `by-logical-key`
    // index, which is the only `unique: true` index in `src/adapters`. The
    // violation belongs to the `add` request, not to the transaction: the
    // journal never asks for the row's uniqueness, it asks the index to enforce
    // it, so the answer has to be the request's `ConstraintError` rather than
    // whatever the transaction happened to abort with.
    const memory = new MemoryIndexedDbFactory();
    const tracker = trackAborts(memory);
    let tokenSequence = 0;
    const journal = createIndexedDbOpfsJournal({
      authorityToken,
      factory: tracker.factory,
      createFencingToken: () =>
        `fencing_${String(++tokenSequence).padStart(8, "0")}`,
    });

    const first = await journal.begin(
      beginInput("transaction_dupe0001", "object_dupe_1234", 1),
    );
    expect(first).toMatchObject({ ok: true });

    expect(
      await journal.begin(
        beginInput("transaction_dupe0002", "object_dupe_1234", 1),
      ),
    ).toEqual({
      ok: false,
      error: {
        code: "CONFLICT",
        operation: "INDEXEDDB_WRITE",
        retryable: false,
        recovery: "NONE",
      },
    });
    // `mapIndexedDbException`'s ConstraintError arm carries no recovery, while
    // the journal's own `conflict()` helper carries REOPEN. The NONE above is
    // what proves the answer came from the native error rather than from a
    // predicate the journal evaluated itself.
    expect(tracker.aborts).toEqual([]);
    // The losing transaction rolled back whole: no budget reservation from the
    // second `begin` survives, so the partition still admits its full budget.
    expect(memory.readRaw("opfs-budgets", `${scope.authorityToken}|${scope.namespaceToken}|${scope.partitionToken}`)).toMatchObject({
      reservedBytes: 1,
    });
    expect(
      (await journal.listIncomplete(10)) as unknown,
    ).toMatchObject({
      ok: true,
      value: { transactions: [{ transactionId: "transaction_dupe0001" }] },
    });
  });

  it("lets the first explicit failure win over a later success in the same transaction", async () => {
    // OP-2. `begin` runs two independent request chains: the scope-binding
    // chain can fail while the object/budget chain is still queued to succeed.
    // The binding chain's `get` is issued first, so its verdict always lands
    // first and the later `succeed` never reaches the caller. The abort count
    // is what separates "the journal rejected this" from "the store did".
    const memory = new MemoryIndexedDbFactory();
    const tracker = trackAborts(memory);
    let tokenSequence = 0;
    const journal = createIndexedDbOpfsJournal({
      authorityToken,
      factory: tracker.factory,
      createFencingToken: () =>
        `fencing_${String(++tokenSequence).padStart(8, "0")}`,
    });

    const bound = await journal.begin(
      beginInput("transaction_order001", "object_order_1234", 1),
    );
    if (!bound.ok) throw new Error("binding begin failed");
    await journal.rollback(
      "transaction_order001",
      bound.value.fencingToken,
    );
    const abortsBefore = tracker.aborts.length;

    expect(
      await journal.begin({
        ...beginInput("transaction_order002", "object_order_5678", 1),
        targetStoragePolicy: { ...policy, owner: "rebound-owner" },
      }),
    ).toEqual({
      ok: false,
      error: {
        code: "POLICY_REJECTED",
        operation: "INDEXEDDB_WRITE",
        retryable: false,
        recovery: "READ_ONLY",
      },
    });
    // Exactly one abort, from the binding chain's `fail`. A second would mean
    // the budget chain also reached a verdict.
    expect(tracker.aborts.length - abortsBefore).toBe(1);
    expect(memory.readRaw("opfs-journal", "transaction_order002")).toBeUndefined();
  });

  it("keeps a commit failure attributable to the transaction rather than a request", async () => {
    // The commit itself fails, so no request ever errors. The journal must not
    // claim an abort it did not perform, and the quota code has to come from
    // the transaction's own error.
    const memory = new MemoryIndexedDbFactory();
    const tracker = trackAborts(memory);
    let tokenSequence = 0;
    const journal = createIndexedDbOpfsJournal({
      authorityToken,
      factory: tracker.factory,
      createFencingToken: () =>
        `fencing_${String(++tokenSequence).padStart(8, "0")}`,
    });

    memory.failNextWriteCommit(
      new DOMException("quota", "QuotaExceededError"),
    );
    expect(
      await journal.begin(
        beginInput("transaction_quota001", "object_quota_1234", 1),
      ),
    ).toEqual({
      ok: false,
      error: {
        code: "QUOTA_EXCEEDED",
        operation: "INDEXEDDB_WRITE",
        retryable: false,
        recovery: "READ_ONLY",
      },
    });
    expect(tracker.aborts).toEqual([]);
  });

  it("pages incomplete transactions and validates a row before the page limit", async () => {
    const { factory, journal } = createHarness();
    for (const index of [1, 2, 3]) {
      const id = `transaction_page000${index}`;
      expect(
        await journal.begin(
          beginInput(id, `object_page_000${index}`, 1),
        ),
      ).toMatchObject({ ok: true });
    }

    expect(await journal.listIncomplete(2)).toMatchObject({
      ok: true,
      value: { moreAvailable: true },
    });
    const page = await journal.listIncomplete(2);
    if (!page.ok) throw new Error("listIncomplete failed");
    expect(page.value.transactions).toHaveLength(2);

    const whole = await journal.listIncomplete(10);
    if (!whole.ok) throw new Error("listIncomplete failed");
    expect(whole.value.transactions).toHaveLength(3);
    expect(whole.value.moreAvailable).toBe(false);

    // A corrupt row one past the requested page still fails the read. The
    // journal validates a row before it checks the limit, so a page that is
    // already full does not hide the next row's damage behind `moreAvailable`.
    factory.seed("opfs-journal", {
      transactionId: "transaction_page9999",
      startedAtEpochMs: 9_000,
      corrupt: true,
    });
    expect(await journal.listIncomplete(3)).toMatchObject({
      ok: false,
      error: { code: "CORRUPT_DATA", recovery: "READ_ONLY" },
    });
  });

  it("pages committed objects and reports where the next page resumes", async () => {
    const { journal } = createHarness();
    await commitObject(journal, "transaction_obj00001", "object_aaaa0001", 1);
    await commitObject(journal, "transaction_obj00002", "object_bbbb0002", 1);
    await commitObject(journal, "transaction_obj00003", "object_cccc0003", 1);

    const first = await journal.listCommittedObjects({ scope, limit: 2 });
    if (!first.ok) throw new Error("listCommittedObjects failed");
    expect(first.value.objects.map((o) => o.descriptor.objectId)).toEqual([
      "object_aaaa0001",
      "object_bbbb0002",
    ]);
    // A page cut short by the limit names the cursor the caller resumes from.
    expect(first.value.moreAvailable).toBe(true);
    expect(first.value.nextObjectId).toBe("object_bbbb0002");

    const second = await journal.listCommittedObjects({
      scope,
      limit: 2,
      afterObjectId: first.value.nextObjectId ?? undefined,
    });
    if (!second.ok) throw new Error("listCommittedObjects failed");
    expect(second.value.objects.map((o) => o.descriptor.objectId)).toEqual([
      "object_cccc0003",
    ]);
    // An exhausted scan has nothing to resume from, so it reports neither.
    expect(second.value.moreAvailable).toBe(false);
    expect(second.value.nextObjectId).toBeNull();
  });

  it("answers UNSUPPORTED once the journal is closed", async () => {
    const { factory, journal } = createHarness();
    expect(
      await journal.getCommittedObject(scope, "object_12345678"),
    ).toEqual({ ok: true, value: null });

    journal.close();
    expect(factory.isConnectionClosed()).toBe(true);
    // A closed journal is not a retryable outage and not an abort: the store
    // is gone for this realm, so the caller is told to work online. The
    // failure names INDEXEDDB_OPEN rather than the operation that asked,
    // because it belongs to the open that never happened.
    expect(
      await journal.getCommittedObject(scope, "object_12345678"),
    ).toEqual({
      ok: false,
      error: {
        code: "UNSUPPORTED",
        operation: "INDEXEDDB_OPEN",
        retryable: false,
        recovery: "ONLINE_ONLY",
      },
    });
    expect(
      await journal.begin(
        beginInput("transaction_closed001", "object_closed_001", 1),
      ),
    ).toMatchObject({ ok: false, error: { code: "UNSUPPORTED" } });
  });
});