/* oxlint-disable executor/no-json-parse, executor/no-raw-fetch, executor/no-try-catch-or-throw -- boundary: one-shot local SQLite/auth-file migration normalizes legacy on-disk state */ import type { Client } from "@libsql/client";import { drizzle } from "drizzle-orm/libsql";import { migrate } from "drizzle-orm/libsql/migrator";import { Effect } from "effect";import { createId } from "@executor-js/fumadb/cuid";import { createHash, randomBytes } from "node:crypto";import * as fs from "node:fs";import { homedir, tmpdir } from "node:os";import { dirname, join } from "node:path";import { TextDecoder } from "node:util"; import type { FumaTables } from "@executor-js/sdk";import {  buildV1RuntimeMetadataIndex,  migrateGraphqlSourceConfig,  migrateMcpSourceConfig,  migrateOpenApiSourceConfig,  migrateV1PluginStorageRuntimeRow,  migrateV1ToolAnnotations,  migrationOAuthAuthorizationUrlFor as authorizationUrlFor,  migrationOAuthClientPlanKey as oauthClientPlanKey,  migrationSourceKey,  parseScope,  planMigration,  resolveMigrationOAuthAuthorizationUrls,  type MigratedSourceConfig,  type MigrationInput,  type MigrationOAuthMetadataFetch,  type MigrationOwner,  type MigrationPlan,  type OwnerKeys,  type V1SourceRow,} from "@executor-js/sdk/migration";import { makeKeychainProvider } from "@executor-js/plugin-keychain"; import { createSqliteFumaDb } from "./sqlite-fumadb";import embeddedLegacyMigrations from "./embedded-migrations.gen";import { executeSql, openLocalLibsql, queryFirst, queryRows } from "./libsql"; type Row = Record<string, unknown>; interface V1ToolRow {  readonly scopeId: string;  readonly sourceId: string;  readonly pluginId: string;  readonly name: string;  readonly description: string;  readonly inputSchema: unknown;  readonly outputSchema: unknown;  readonly annotations: unknown;  readonly createdAt: number;  readonly updatedAt: number;} interface V1DefinitionRow {  readonly scopeId: string;  readonly sourceId: string;  readonly pluginId: string;  readonly name: string;  readonly schema: unknown;  readonly createdAt: number;} interface V1PluginStorageRow {  readonly scopeId: string;  readonly pluginId: string;  readonly collection: string;  readonly key: string;  readonly data: unknown;  readonly createdAt: number;  readonly updatedAt: number;} interface V1BlobRow {  readonly namespace: string;  readonly key: string;  readonly value: string;} interface LocalV1Snapshot {  readonly input: MigrationInput;  readonly tools: readonly V1ToolRow[];  readonly definitions: readonly V1DefinitionRow[];  readonly pluginStorage: readonly V1PluginStorageRow[];  readonly blobs: readonly V1BlobRow[];} export interface LocalV1V2MigrationResult {  readonly migrated: boolean;  readonly backupPath?: string;  readonly report?: MigrationPlan["report"];  readonly warnings: readonly string[];} export interface LocalV1V2MigrationOptions {  readonly sqlitePath: string;  readonly tables: FumaTables;  readonly namespace: string;  readonly tenantId: string;  readonly oauthMetadataFetch?: MigrationOAuthMetadataFetch;  readonly oauthMetadataTimeoutMs?: number;} const FILE_PROVIDER = "file";const KEYCHAIN_PROVIDER = "keychain"; const fileSetSuffixes = ["", "-wal", "-shm"] as const; const quoteIdent = (value: string): string => `"${value.replaceAll('"', '""')}"`; const tableExists = async (client: Client, table: string): Promise<boolean> =>  (await queryFirst(client, "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?", [    table,  ])) != null; const columnNames = async (client: Client, table: string): Promise<ReadonlySet<string>> =>  new Set(    (await queryRows<{ name: string }>(client, `PRAGMA table_info(${quoteIdent(table)})`)).map(      (row) => row.name,    ),  ); const optionalColumn = (columns: ReadonlySet<string>, table: string, column: string): string =>  columns.has(column) ? `${quoteIdent(table)}.${quoteIdent(column)}` : "NULL"; const isLocalV1Database = async (client: Client): Promise<boolean> => {  if (!(await tableExists(client, "source"))) return false;  const sourceColumns = await columnNames(client, "source");  if (!sourceColumns.has("scope_id")) return false;  if (!(await tableExists(client, "integration"))) return true;  const connectionColumns = await columnNames(client, "connection");  return !connectionColumns.has("tenant") || connectionColumns.has("scope_id");}; // ---------------------------------------------------------------------------// Data-migration-ledger gate.//// Once a database has passed the v1 gate (either it was migrated, or it was// inspected and found to be v2-native), the boot ledger stamps// LOCAL_V1_V2_LEDGER_NAME (see apps/local/src/db/data-migrations.ts). On// every later boot the stamp short-circuits this module before any schema// probing — the stamp row, not the data shape, is the source of truth.// --------------------------------------------------------------------------- export const LOCAL_V1_V2_LEDGER_NAME = "2026-06-11-local-v1-to-v2"; const hasV1GateStamp = async (client: Client): Promise<boolean> => {  if (!(await tableExists(client, "data_migration"))) return false;  return (    (await queryFirst(client, "SELECT name FROM data_migration WHERE name = ?", [      LOCAL_V1_V2_LEDGER_NAME,    ])) != null  );}; // ---------------------------------------------------------------------------// Legacy v1 schema replay.//// The v1→v2 data migration below reads the v1-FINAL schema (it queries// `plugin_storage`, which only exists after v1 migration 0011). A database// last touched by an older release is still mid-chain, so replay the bundled// legacy drizzle migrations (`apps/local/drizzle-legacy-v1`, embedded into// the binary by apps/cli/src/build.ts) to bring it to v1-final first — the// same step every pre-v1.5 release performed at startup.// --------------------------------------------------------------------------- const resolveLegacyMigrationsFolder = (): string => {  if (!embeddedLegacyMigrations) {    return join(import.meta.dirname, "../../drizzle-legacy-v1");  }  // drizzle's migrate() reads a folder from disk; materialize the embedded  // contents into a tmpdir.  const dir = fs.mkdtempSync(join(tmpdir(), "executor-legacy-migrations-"));  for (const [rel, content] of Object.entries(embeddedLegacyMigrations)) {    const target = join(dir, rel);    fs.mkdirSync(dirname(target), { recursive: true });    fs.writeFileSync(target, content);  }  return dir;}; const readBundledLegacyMigrationHashes = (migrationsFolder: string): readonly string[] => {  const journal = JSON.parse(    fs.readFileSync(join(migrationsFolder, "meta", "_journal.json")).toString(),  ) as { entries: ReadonlyArray<{ idx: number; tag: string }> };  return [...journal.entries]    .sort((left, right) => left.idx - right.idx)    .map((entry) =>      createHash("sha256")        .update(fs.readFileSync(join(migrationsFolder, `${entry.tag}.sql`)).toString())        .digest("hex"),    );}; const readAppliedLegacyMigrationHashes = async (client: Client): Promise<readonly string[]> =>  (    await queryRows<{ hash: string }>(      client,      "SELECT hash FROM __drizzle_migrations ORDER BY id ASC",    )  ).map((row) => row.hash); // Errors a statement raises when its work is already done — or when it// references schema from an earlier era that a later (already-applied)// migration removed ("no such table/column", "has no column named"). The// legacy chain's data statements are idempotent by construction (INSERT OR// IGNORE / OR REPLACE, conditional UPDATEs), so a chain re-executed over a// database in an unknown mid-chain state converges by skipping exactly these.const TOLERANT_REPLAY_SKIPPABLE =  /already exists|duplicate column name|no such table|no such column|no such index|has no column/i; const readLegacyJournalTags = (migrationsFolder: string): readonly string[] => {  const journal = JSON.parse(    fs.readFileSync(join(migrationsFolder, "meta", "_journal.json")).toString(),  ) as { entries: ReadonlyArray<{ idx: number; tag: string }> };  return [...journal.entries].sort((left, right) => left.idx - right.idx).map((entry) => entry.tag);}; /** Recovery path for a v1 database whose drizzle journal can't be trusted *  (wiped, rewritten by another tool, or from an unknown build): re-execute *  the entire frozen legacy chain statement-by-statement, skipping the *  errors that mean "already applied". Statements the database has already *  seen fail with `already exists`/`duplicate column`/`no such ...` and are *  skipped; statements it never reached execute for real. The chain's own *  0011 backfill then populates `plugin_storage` from the per-plugin source *  tables, which is exactly the shape `readV1Snapshot` needs. */const tolerantReplayLegacyChain = async (client: Client, warnings: string[]): Promise<void> => {  const migrationsFolder = resolveLegacyMigrationsFolder();  let executed = 0;  let skipped = 0;  for (const tag of readLegacyJournalTags(migrationsFolder)) {    const sql = fs.readFileSync(join(migrationsFolder, `${tag}.sql`)).toString();    for (const chunk of sql.split("--> statement-breakpoint")) {      const statement = chunk.trim();      if (statement.length === 0) continue;      try {        await client.execute(statement);        executed++;      } catch (cause) {        // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: classifying raw SqliteError text from the libSQL driver        const message = cause instanceof Error ? cause.message : String(cause);        if (!TOLERANT_REPLAY_SKIPPABLE.test(message)) throw cause;        skipped++;      }    }  }  warnings.push(    `v1 database had unusable drizzle migration history; recovered the v1-final schema by tolerant replay (${executed} statements applied, ${skipped} already done).`,  );}; /** Bring a v1 database up to the v1-final schema `readV1Snapshot` requires. * *  The replay exists for ONE reader: `readV1Snapshot` queries *  `plugin_storage`, which appears in legacy migration 0011. Real-world v1 *  databases show up in franken-states the journal can't describe — the *  observed crash case was a v1-final schema (plugin_storage and all) with *  an EMPTY `__drizzle_migrations` journal, which the old prefix check read *  as "nothing applied" and replayed from 0000 straight into *  `table blob already exists`. So the gates, in order: * *  1. Schema marker: `plugin_storage` exists → the schema is already *     sufficient; never consult the journal, never replay. *  2. Journal is a non-empty strict prefix of the bundled chain → the exact *     drizzle replay every pre-v1.5 release performed (fast, hash-checked). *  3. Anything else (missing/empty/foreign journal) → tolerant replay; the *     old behavior of "skip and read as-is" just crashed later in *     `readV1Snapshot` on the missing `plugin_storage` table. */const replayLegacyV1Migrations = async (client: Client, warnings: string[]): Promise<void> => {  if (await tableExists(client, "plugin_storage")) return;   if (await tableExists(client, "__drizzle_migrations")) {    const applied = await readAppliedLegacyMigrationHashes(client);    if (applied.length > 0) {      const migrationsFolder = resolveLegacyMigrationsFolder();      const bundled = readBundledLegacyMigrationHashes(migrationsFolder);      const isPrefix =        applied.length < bundled.length && applied.every((hash, index) => hash === bundled[index]);      // A full-length journal can't be trusted here: plugin_storage is      // missing (checked above), so a journal claiming v1-final is lying —      // fall through to the tolerant replay.      if (isPrefix) {        await migrate(drizzle({ client }), { migrationsFolder });        return;      }    }  }   await tolerantReplayLegacyChain(client, warnings);}; const textDecoder = new TextDecoder(); const decodeBytes = (value: ArrayBuffer | ArrayBufferView): string => {  const bytes =    value instanceof ArrayBuffer      ? new Uint8Array(value)      : new Uint8Array(value.buffer, value.byteOffset, value.byteLength);  return textDecoder.decode(bytes);}; const parseJson = (value: unknown): unknown => {  if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) {    return parseJson(decodeBytes(value));  }  if (typeof value !== "string") return value;  if (value.trim() === "") return null;  return JSON.parse(value);}; const stringOrNull = (value: unknown): string | null => (value == null ? null : String(value)); const numberOrNull = (value: unknown): number | null => {  if (value == null) return null;  const n = Number(value);  return Number.isFinite(n) ? n : null;}; const numberOrDefault = (value: unknown, fallback: number): number =>  numberOrNull(value) ?? fallback; const normalizePluginId = (pluginId: string, kind: string): string =>  pluginId === "graphql-greenfield" ? "graphql" : pluginId || kind; const buildConfig = (kind: string, data: Record<string, unknown>): MigratedSourceConfig => {  const cfg = (data.config as Record<string, unknown> | undefined) ?? data;  if (kind === "mcp") return migrateMcpSourceConfig(cfg as never);  if (kind === "graphql") return migrateGraphqlSourceConfig(cfg as never);  return migrateOpenApiSourceConfig(cfg as never);}; const isObjectRecord = (value: unknown): value is Record<string, unknown> =>  typeof value === "object" && value !== null && !Array.isArray(value); const sourceKeyForBinding = (binding: Row): string =>  migrationSourceKey(    binding.source_scope_id == null ? String(binding.scope_id) : String(binding.source_scope_id),    String(binding.source_id),  ); const mcpOAuthEndpoint = (config: MigratedSourceConfig | undefined): string | null => {  const value = config?.config;  if (!isObjectRecord(value)) return null;  if (typeof value.endpoint !== "string" || value.endpoint.length === 0) return null;  if (!isObjectRecord(value.auth) || value.auth.kind !== "oauth2") return null;  return value.endpoint;}; const canonicalResource = (value: string): string | null => {  try {    const url = new URL(value);    return `${url.protocol.toLowerCase()}//${url.host.toLowerCase()}${url.pathname.replace(/\/+$/, "")}`;  } catch {    return null;  }}; const resourceMatchesEndpoint = (resource: string, endpoint: string): boolean => {  const actual = canonicalResource(resource);  const expected = canonicalResource(endpoint);  return (    actual != null && expected != null && (actual === expected || expected.startsWith(`${actual}/`))  );}; const protectedResourceMetadataUrls = (endpoint: string): readonly string[] => {  try {    const url = new URL(endpoint);    const origin = url.origin;    const path = url.pathname.replace(/\/+$/, "");    const urls: string[] = [];    if (path && path !== "/") urls.push(`${origin}/.well-known/oauth-protected-resource${path}`);    urls.push(`${origin}/.well-known/oauth-protected-resource`);    return [...new Set(urls)];  } catch {    return [];  }}; const discoverProtectedResource = async (endpoint: string): Promise<string | null> => {  for (const url of protectedResourceMetadataUrls(endpoint)) {    try {      const response = await fetch(url, {        headers: { accept: "application/json" },        signal: AbortSignal.timeout(5_000),      });      if (!response.ok) continue;      const json = (await response.json()) as unknown;      if (!isObjectRecord(json) || typeof json.resource !== "string") continue;      if (resourceMatchesEndpoint(json.resource, endpoint)) return json.resource;    } catch {      continue;    }  }  return null;}; const discoverMcpOAuthResourceOverrides = async (  bindings: readonly Row[],  migratedConfigs: ReadonlyMap<string, MigratedSourceConfig>,): Promise<ReadonlyMap<string, string>> => {  const endpointByKey = new Map<string, string>();  for (const binding of bindings) {    if (binding.kind !== "connection") continue;    const key = sourceKeyForBinding(binding);    const endpoint = mcpOAuthEndpoint(migratedConfigs.get(key));    if (endpoint) endpointByKey.set(key, endpoint);  }  const resourceByEndpoint = new Map<string, string | null>();  await Promise.all(    [...new Set(endpointByKey.values())].map(async (endpoint) => {      resourceByEndpoint.set(endpoint, await discoverProtectedResource(endpoint));    }),  );  const overrides = new Map<string, string>();  for (const [key, endpoint] of endpointByKey) {    const resource = resourceByEndpoint.get(endpoint);    if (resource) overrides.set(key, resource);  }  return overrides;}; const localOwnerForScope =  (tenantId: string) =>  (scopeId: string): OwnerKeys | null => {    const cloud = parseScope(scopeId);    if (cloud) return cloud;    return { owner: "org", subject: "", tenant: tenantId };  }; const readV1Snapshot = async (client: Client, tenantId: string): Promise<LocalV1Snapshot> => {  const hasDefinition = await tableExists(client, "definition");  const hasBlob = await tableExists(client, "blob");  const bindingColumns = await columnNames(client, "credential_binding");  const toolColumns = await columnNames(client, "tool");  const definitionColumns = hasDefinition    ? await columnNames(client, "definition")    : new Set<string>();   const [    sources,    secrets,    bindings,    connections,    policies,    sourceStorage,    allPluginStorage,    toolSources,    tools,    definitions,    blobs,  ] = await Promise.all([    queryRows<Row>(client, "SELECT scope_id, id, plugin_id, kind, name FROM source"),    queryRows<Row>(      client,      "SELECT id, scope_id, name, provider, owned_by_connection_id FROM secret",    ),    queryRows<Row>(      client,      `SELECT scope_id, ${optionalColumn(bindingColumns, "credential_binding", "source_scope_id")} AS source_scope_id, source_id, slot_key, kind, secret_id, ${optionalColumn(bindingColumns, "credential_binding", "secret_scope_id")} AS secret_scope_id, connection_id, text_value FROM credential_binding`,    ),    queryRows<Row>(      client,      "SELECT id, scope_id, provider, identity_label, access_token_secret_id, refresh_token_secret_id, expires_at, provider_state FROM connection",    ),    queryRows<Row>(client, "SELECT id, scope_id, pattern, action, position FROM tool_policy"),    queryRows<Row>(      client,      "SELECT ps.scope_id, ps.key AS source_id, ps.data, s.kind FROM plugin_storage ps JOIN source s ON ps.key = s.id AND ps.scope_id = s.scope_id WHERE ps.collection = 'source'",    ),    queryRows<Row>(      client,      "SELECT scope_id, plugin_id, collection, key, data, created_at, updated_at FROM plugin_storage",    ),    queryRows<Row>(client, "SELECT DISTINCT source_id FROM tool"),    queryRows<Row>(      client,      `SELECT scope_id, source_id, plugin_id, name, description, ${optionalColumn(toolColumns, "tool", "input_schema")} AS input_schema, ${optionalColumn(toolColumns, "tool", "output_schema")} AS output_schema, ${optionalColumn(toolColumns, "tool", "annotations")} AS annotations, created_at, updated_at FROM tool`,    ),    hasDefinition      ? queryRows<Row>(          client,          `SELECT scope_id, source_id, plugin_id, name, ${optionalColumn(definitionColumns, "definition", "schema")} AS schema, created_at FROM definition`,        )      : Promise.resolve([]),    hasBlob      ? queryRows<Row>(client, "SELECT namespace, key, value FROM blob")      : Promise.resolve([]),  ]);   const migratedConfigs = new Map<string, MigratedSourceConfig>();  for (const row of sourceStorage) {    const data = parseJson(row.data) as Record<string, unknown>;    migratedConfigs.set(      migrationSourceKey(String(row.scope_id), String(row.source_id)),      buildConfig(String(row.kind), data),    );  }

Lines 1–500 of 1078. Load more