/* oxlint-disable executor/no-try-catch-or-throw -- boundary: DB ownership tests must tear down held DB/lock handles and child processes even when assertions fail */import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";import {  existsSync,  mkdirSync,  mkdtempSync,  realpathSync,  rmSync,  symlinkSync,  writeFileSync,} from "node:fs";import { tmpdir } from "node:os";import { join } from "node:path"; import { describe, expect, it } from "@effect/vitest";import { collectTables } from "@executor-js/api/server"; import { acquireDataDirOwnership, DataDirOwnershipHeld } from "./data-dir-ownership";import { executeSql, openLocalLibsql } from "./libsql";import { openOwnedLocalDatabase, type OwnedLocalDatabase } from "./owned-database"; const LOCK_DATABASE_FILENAME = "data.db.owner-lock";const TEST_NAMESPACE = "executor_local_owned_database_test";const appRoot = join(import.meta.dirname, "../.."); const delay = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms)); const makeOwnedTestDatabase = (dataDir: string, namespace = TEST_NAMESPACE) =>  openOwnedLocalDatabase({    dataDir,    tables: collectTables(),    namespace,    tenantId: "owned-database-test-tenant",  }); const makeWorkDir = (prefix: string): string =>  mkdtempSync(join(tmpdir(), `executor-owned-database-${prefix}-`)); // data.db exists only after the sole owner opens it, so it is the witness that a// non-owner never wrote anything: an ownership loser must fail before any DB I/O.const servingDbPath = (dataDir: string): string => join(realpathSync(dataDir), "data.db"); describe("openOwnedLocalDatabase", () => {  it("holds data-dir ownership until the serving database is closed", async () => {    const workDir = makeWorkDir("lifetime");    const dataDir = join(workDir, "data");    let owned: OwnedLocalDatabase | null = null;     try {      owned = await makeOwnedTestDatabase(dataDir);      const realDataDir = realpathSync(dataDir);      const expectedLockPath = join(realDataDir, LOCK_DATABASE_FILENAME);       expect(owned).toMatchObject({        dataDir: realDataDir,        sqlitePath: join(realDataDir, "data.db"),        lockPath: expectedLockPath,        migration: { migrated: false, warnings: [] },      });       await expect(acquireDataDirOwnership(dataDir)).rejects.toBeInstanceOf(DataDirOwnershipHeld);       await owned.close();      owned = null;       const ownership = await acquireDataDirOwnership(dataDir);      try {        expect(ownership.lockPath).toBe(expectedLockPath);      } finally {        await ownership.release();      }    } finally {      if (owned) await owned.close();      rmSync(workDir, { recursive: true, force: true });    }  });   it("refuses to open (and never creates data.db) when ownership is already held", async () => {    const workDir = makeWorkDir("gate-order");    const dataDir = join(workDir, "data");    const ownership = await acquireDataDirOwnership(dataDir);     try {      await expect(makeOwnedTestDatabase(dataDir)).rejects.toBeInstanceOf(DataDirOwnershipHeld);      // The gate runs before any migration or createSqliteFumaDb work, so the      // loser cannot have written the serving database.      expect(existsSync(servingDbPath(dataDir))).toBe(false);    } finally {      await ownership.release();    }     // Once the owner releases, a fresh open succeeds and creates data.db.    const owned = await makeOwnedTestDatabase(dataDir);    try {      expect(existsSync(owned.sqlitePath)).toBe(true);    } finally {      await owned.close();      rmSync(workDir, { recursive: true, force: true });    }  });   it("keeps the gate independent of a deleted or corrupt server.json hint", async () => {    const workDir = makeWorkDir("hint");    const dataDir = join(workDir, "data");    const owned = await makeOwnedTestDatabase(dataDir);     try {      // server.json is only a coordination hint. A garbage manifest must not let      // a second process open the DB while the real owner is alive.      const serverControlDir = join(realpathSync(dataDir), "server-control");      mkdirSync(serverControlDir, { recursive: true });      const manifestPath = join(serverControlDir, "server.json");      writeFileSync(manifestPath, "{ this is not valid json");      await expect(makeOwnedTestDatabase(dataDir)).rejects.toBeInstanceOf(DataDirOwnershipHeld);       // Deleting the hint entirely changes nothing: the kernel lock is the gate.      rmSync(manifestPath, { force: true });      await expect(makeOwnedTestDatabase(dataDir)).rejects.toBeInstanceOf(DataDirOwnershipHeld);    } finally {      await owned.close();      rmSync(workDir, { recursive: true, force: true });    }  });   it("collides on one ownership lock across symlinked data dirs", async () => {    const workDir = makeWorkDir("symlink");    const realDir = join(workDir, "real");    const linkDir = join(workDir, "link");    mkdirSync(realDir, { recursive: true });    symlinkSync(realDir, linkDir, "dir");     const owned = await makeOwnedTestDatabase(realDir);    try {      const expectedLockPath = join(realpathSync(realDir), LOCK_DATABASE_FILENAME);      expect(owned.lockPath).toBe(expectedLockPath);       // The symlinked path resolves to the same real inode, so it must lose the      // ownership race rather than open a second handle on the same data.db.      await expect(acquireDataDirOwnership(linkDir)).rejects.toMatchObject({        _tag: "DataDirOwnershipHeld",        lockPath: expectedLockPath,      });    } finally {      await owned.close();      rmSync(workDir, { recursive: true, force: true });    }  });   it("opens distinct real data dirs independently", async () => {    const workDir = makeWorkDir("distinct");    const dataDirA = join(workDir, "a");    const dataDirB = join(workDir, "b");     const [ownedA, ownedB] = await Promise.all([      makeOwnedTestDatabase(dataDirA),      makeOwnedTestDatabase(dataDirB),    ]);    try {      expect(ownedA.lockPath).not.toBe(ownedB.lockPath);      expect(existsSync(ownedA.sqlitePath)).toBe(true);      expect(existsSync(ownedB.sqlitePath)).toBe(true);    } finally {      await Promise.all([ownedA.close(), ownedB.close()]);      rmSync(workDir, { recursive: true, force: true });    }  });   it("recovers a v1->v2 migration killed mid-flip under ownership", async () => {    const workDir = makeWorkDir("migration");    const dataDir = join(workDir, "data");    mkdirSync(dataDir, { recursive: true });    const realDataDir = realpathSync(dataDir);    const sqlitePath = join(realDataDir, "data.db");    const journalPath = `${sqlitePath}.v1-v2-migration.json`;    const tenantId = "owned-migration-workspace";    const marker = join(workDir, "paused-flip");     await seedMinimalV1Db(sqlitePath, tenantId);     // Pause mid-flip: at "canonical-moved" the original data.db has already been    // moved aside to the backup and the staged v2 DB has not been swapped in    // yet, so a SIGKILL here leaves a journal the next owner must RESUME (via    // completeJournaledFlip) rather than re-run from scratch.    const child = spawnOpenOwnedChild({      dataDir,      tenantId,      pauseAt: "canonical-moved",      pauseFile: marker,    });     try {      // Wait until the flip is mid-way under the child's ownership lock.      const paused = await waitForMarker(marker, child, 30_000);      expect(        paused.markerFound,        `child never reached the flip pause:\nstdout: ${child.stdout}\nstderr: ${child.stderr}`,      ).toBe(true);      // Mid-flip witness: the canonical data.db has been moved to the backup and      // the staged copy is not yet installed.      expect(existsSync(sqlitePath)).toBe(false);       // SIGKILL the holder so the OS releases the ownership lock mid-flip.      child.child.kill("SIGKILL");      await waitForChildClose(child, 5_000);      expect(child.child.signalCode).toBe("SIGKILL");       // The journal proves the flip was interrupted; ownership must now be free      // because the holder died.      expect(existsSync(journalPath)).toBe(true);      const reclaimed = await acquireDataDirOwnership(dataDir);      await reclaimed.release();       // The next sole owner resumes the journaled flip during open (recovery      // runs completeJournaledFlip before the serving connection opens). The      // steady-state migrate then sees an already-v2 DB, so migration.migrated      // is false even though the on-disk DB is now v2 — the cleared journal and      // the v2 rows are the real proof.      const owned = await openOwnedLocalDatabase({        dataDir,        tables: collectTables(),        namespace: "executor_local",        tenantId,      });      try {        expect(existsSync(journalPath)).toBe(false);        const integrations = await owned.db.client.execute("SELECT tenant, slug FROM integration");        expect(integrations.rows).toEqual([{ tenant: tenantId, slug: "stripe_api" }]);      } finally {        await owned.close();      }    } finally {      if (child.child.exitCode === null && child.child.signalCode === null) {        child.child.kill("SIGKILL");        await waitForChildClose(child, 5_000);      }      child.child.stdout.destroy();      child.child.stderr.destroy();      rmSync(workDir, { recursive: true, force: true });    }  });}); // ---------------------------------------------------------------------------// Minimal v1 fixture + child-process harness// --------------------------------------------------------------------------- // A purpose-built v1-final database with exactly the tables readV1Snapshot// reads. plugin_storage is present so the legacy schema replay is skipped, and// a single openapi source makes isLocalV1Database return true and gives the// migration real content to carry into v2.const seedMinimalV1Db = async (dbPath: string, scopeId: string): Promise<void> => {  const client = await openLocalLibsql(dbPath);  // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: close the SQLite client even if seeding fails  try {    await client.execute("PRAGMA foreign_keys = OFF");    await client.execute(      "CREATE TABLE source (id text NOT NULL, scope_id text NOT NULL, plugin_id text NOT NULL, kind text NOT NULL, name text NOT NULL, PRIMARY KEY(scope_id, id))",    );    await client.execute(      "CREATE TABLE plugin_storage (id text NOT NULL, scope_id text NOT NULL, plugin_id text NOT NULL, collection text NOT NULL, key text NOT NULL, data text NOT NULL, created_at integer NOT NULL, updated_at integer NOT NULL, PRIMARY KEY(scope_id, id))",    );    await client.execute(      "CREATE TABLE credential_binding (id text NOT NULL, scope_id text NOT NULL, plugin_id text NOT NULL, source_id text NOT NULL, source_scope_id text NOT NULL, slot_key text NOT NULL, kind text NOT NULL, text_value text, secret_id text, connection_id text, created_at integer NOT NULL, updated_at integer NOT NULL, PRIMARY KEY(scope_id, id))",    );    await client.execute(      "CREATE TABLE secret (id text NOT NULL, scope_id text NOT NULL, name text NOT NULL, provider text NOT NULL, owned_by_connection_id text, created_at integer NOT NULL, PRIMARY KEY(scope_id, id))",    );    await client.execute(      "CREATE TABLE connection (id text NOT NULL, scope_id text NOT NULL, provider text NOT NULL, identity_label text, access_token_secret_id text, refresh_token_secret_id text, expires_at integer, provider_state text, PRIMARY KEY(scope_id, id))",    );    await client.execute(      "CREATE TABLE tool_policy (id text NOT NULL, scope_id text NOT NULL, pattern text NOT NULL, action text NOT NULL, position text NOT NULL, created_at integer NOT NULL, updated_at integer NOT NULL, PRIMARY KEY(scope_id, id))",    );    await client.execute(      "CREATE TABLE tool (id text NOT NULL, scope_id text NOT NULL, source_id text NOT NULL, plugin_id text NOT NULL, name text NOT NULL, description text NOT NULL, input_schema text, output_schema text, created_at integer NOT NULL, updated_at integer NOT NULL, PRIMARY KEY(scope_id, id))",    );     const now = Date.now();    await executeSql(      client,      "INSERT INTO source (id, scope_id, plugin_id, kind, name) VALUES (?, ?, ?, ?, ?)",      ["stripe_api", scopeId, "openapi", "openapi", "Stripe"],    );    await executeSql(      client,      "INSERT INTO plugin_storage (id, scope_id, plugin_id, collection, key, data, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",      [        "openapi-source-stripe",        scopeId,        "openapi",        "source",        "stripe_api",        JSON.stringify({          config: {            spec: "{}",            headers: {              Authorization: { kind: "binding", slot: "header:authorization", prefix: "Bearer " },            },          },        }),        now,        now,      ],    );    await executeSql(      client,      "INSERT INTO credential_binding (id, scope_id, plugin_id, source_id, source_scope_id, slot_key, kind, text_value, secret_id, connection_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",      [        "stripe-auth",        scopeId,        "openapi",        "stripe_api",        scopeId,        "header:authorization",        "secret",        null,        "stripe-key",        null,        now,        now,      ],    );    await executeSql(      client,      "INSERT INTO secret (id, scope_id, name, provider, owned_by_connection_id, created_at) VALUES (?, ?, ?, ?, ?, ?)",      ["stripe-key", scopeId, "Stripe key", "file", null, now],    );  } finally {    client.close();  }}; interface OwnedChild {  readonly child: ChildProcessWithoutNullStreams;  stdout: string;  stderr: string;} interface ChildCloseResult {  readonly markerFound: boolean;  readonly exitCode: number | null;  readonly signalCode: NodeJS.Signals | null;} const spawnOpenOwnedChild = (input: {  readonly dataDir: string;  readonly tenantId: string;  readonly pauseAt: string;  readonly pauseFile: string;}): OwnedChild => {  const code = `    import { collectTables } from "@executor-js/api/server";    import { openOwnedLocalDatabase } from "./src/db/owned-database.ts";    const owned = await openOwnedLocalDatabase({      dataDir: process.env.EXECUTOR_TEST_DATA_DIR,      tables: collectTables(),      namespace: "executor_local",      tenantId: process.env.EXECUTOR_TEST_TENANT,    });    console.log("OPENED");    await owned.close();  `;  const child = spawn(process.execPath, ["-e", code], {    cwd: appRoot,    env: {      ...process.env,      NODE_ENV: "test",      EXECUTOR_TEST_DATA_DIR: input.dataDir,      EXECUTOR_TEST_TENANT: input.tenantId,      EXECUTOR_V1_V2_MIGRATION_PAUSE_AT: input.pauseAt,      EXECUTOR_V1_V2_MIGRATION_PAUSE_FILE: input.pauseFile,    },    stdio: "pipe",  });  child.unref();   const owned: OwnedChild = { child, stdout: "", stderr: "" };  child.stdout.setEncoding("utf-8");  child.stderr.setEncoding("utf-8");  child.stdout.on("data", (chunk: string) => {    owned.stdout += chunk;  });  child.stderr.on("data", (chunk: string) => {    owned.stderr += chunk;  });  return owned;}; const waitForMarker = async (  markerPath: string,  owned: OwnedChild,  timeoutMs: number,): Promise<ChildCloseResult> => {  const deadline = Date.now() + timeoutMs;  while (Date.now() < deadline) {    if (existsSync(markerPath)) {      return {        markerFound: true,        exitCode: owned.child.exitCode,        signalCode: owned.child.signalCode,      };    }    if (owned.child.exitCode !== null || owned.child.signalCode !== null) {      return {        markerFound: false,        exitCode: owned.child.exitCode,        signalCode: owned.child.signalCode,      };    }    await delay(25);  }  return { markerFound: false, exitCode: owned.child.exitCode, signalCode: owned.child.signalCode };}; const waitForChildClose = async (owned: OwnedChild, timeoutMs: number): Promise<void> =>  new Promise((resolve) => {    if (owned.child.exitCode !== null || owned.child.signalCode !== null) {      resolve();      return;    }    const timer = setTimeout(() => {      owned.child.off("close", onClose);      resolve();    }, timeoutMs);    const onClose = () => {      clearTimeout(timer);      resolve();    };    owned.child.once("close", onClose);  });