import type {
  SchemaSnapshot, TableDescriptor, ColumnDescriptor, SnapshotMeta,
  IndexDescriptor, FkDescriptor, FkAction, ViewDescriptor,
} from "../types.js";
import {
  parseSqliteDefault, sqliteTypeToSqlType, sqliteRuleToAction, parseSqliteChecks,
  buildSqliteIndexDescriptor,
} from "./sqlite-shared.js";

/**
 * Runner contract: takes a SQL command string and returns wrangler's raw
 * JSON envelope stdout. The CLI wires this to a real exec; tests pass a mock.
 * The runner is responsible for ALL transport concerns (local vs remote,
 * config path, error mapping). introspectD1 only knows about SQL queries.
 */
export type D1Runner = (sql: string) => Promise<string>;

/** Private shorthand for the exec helper used throughout this module. */
type Exec = (sql: string) => Promise<Record<string, unknown>[]>;

export interface IntrospectD1Options {
  runner: D1Runner;
  /**
   * Documented passthrough — the CLI wiring uses binding/remote/configPath to
   * construct the runner; introspectD1 itself only dispatches SQL via opts.runner.
   * They live on the options so the wiring contract is self-documenting at the call site.
   */
  binding: string;
  remote: boolean;
  configPath: string | undefined;
}

export async function introspectD1(opts: IntrospectD1Options): Promise<SchemaSnapshot> {
  const exec: Exec = async (sql: string) => {
    const stdout = await opts.runner(sql);
    return parseEnvelope(stdout);
  };

  // sqlite_version() is blocked by workerd's local D1 sandbox, so we fall back
  // to a known-good static version (Cloudflare D1 ships a recent SQLite). Remote
  // wrangler executions answer the function, so we try once and only fall back
  // on failure. Keep this string ≥ any version-gated downstream feature checks
  // (see emit/sqlite.ts → parseVersion).
  let sqliteVersion = "3.44.0";
  try {
    const versionRows = await exec("SELECT sqlite_version() AS v");
    const v = versionRows[0]?.v;
    if (v !== undefined && v !== null) sqliteVersion = String(v);
  } catch {
    // Fall through to the static default — workerd local sandbox path.
  }
  const meta: SnapshotMeta = { sqliteVersion };

  // Beyond SQLite's own `sqlite_%` and our rename-shadow `__new_%` tables, D1 carries
  // two infrastructure tables that were never part of the declared schema and must
  // never reach the diff:
  //
  //   `_cf_%`         Cloudflare/miniflare reserved bookkeeping (e.g. `_cf_METADATA`).
  //                   It appears the moment ANY write touches a local D1, and D1's
  //                   authorizer then denies even a bare `pragma_table_info` against it
  //                   (SQLITE_AUTH). Since we call readTableInfo once per enumerated
  //                   table, leaving it in the list aborts introspection outright — which
  //                   breaks every second-and-later `meta migrate --dialect d1` (the first
  //                   migration is the very write that creates it, so it is invisible until
  //                   the first incremental run).
  //
  //   `d1_migrations` wrangler's own migration-tracking table. Queryable, so it doesn't
  //                   crash — it just reads as an undeclared "extra" table, and the diff
  //                   then proposes DROP TABLE on wrangler's own bookkeeping.
  //
  // Filter in the query, not after: `_cf_METADATA` must never even be fetched.
  const tableRows = await exec(
    "SELECT name, sql FROM sqlite_master WHERE type='table'"
      + " AND name NOT LIKE 'sqlite\\_%' ESCAPE '\\'"
      + " AND name NOT LIKE '\\_\\_new\\_%' ESCAPE '\\'"
      + " AND name NOT LIKE '\\_cf\\_%' ESCAPE '\\'"
      + " AND name != 'd1_migrations' ORDER BY name",
  );

  const tables: TableDescriptor[] = [];
  for (const t of tableRows) {
    const name = String(t.name);
    const createSql = String(t.sql ?? "");
    // Issue pragma_table_info ONCE per table; extractColumns + extractPrimaryKey
    // consume the same rows without re-querying (each query is a wrangler round-trip).
    const tableInfoRows = await readTableInfo(exec, name);
    const cols = extractColumns(tableInfoRows);
    const pk = extractPrimaryKey(tableInfoRows);
    const hasAutoincrement = createSql.toUpperCase().includes("AUTOINCREMENT");
    if (hasAutoincrement && pk.length === 1) {
      const pkCol = cols.find((c) => c.name === pk[0]);
      if (pkCol) pkCol.identity = "increment";
    }
    tables.push({
      name,
      columns: cols,
      indexes: await readIndexes(exec, name),
      foreignKeys: await readForeignKeys(exec, name),
      // Named CHECKs parsed from the stored CREATE TABLE DDL — required for
      // check evolution (enum @values changes) to converge on sqlite/D1.
      checks: parseSqliteChecks(createSql),
      primaryKey: pk,
    });
  }

  const views = await readViews(exec);
  return { tables, views, meta };
}

function parseEnvelope(stdout: string): Record<string, unknown>[] {
  let parsed: unknown;
  try {
    parsed = JSON.parse(stdout);
  } catch (err) {
    throw new Error(`failed to parse wrangler JSON output: ${(err as Error).message}`);
  }
  if (!Array.isArray(parsed) || parsed.length === 0) {
    throw new Error(`unexpected wrangler output shape (expected non-empty array envelope): ${stdout.slice(0, 200)}`);
  }
  const envelope = parsed[0];
  if (envelope === null || typeof envelope !== "object") {
    throw new Error(`unexpected wrangler output shape (envelope is not an object): ${stdout.slice(0, 200)}`);
  }
  const env = envelope as { success?: boolean; error?: string; results?: unknown };
  if (env.success === false) {
    throw new Error(`wrangler d1 execute failed: ${env.error ?? "(no error message)"}`);
  }
  const results = env.results;
  if (!Array.isArray(results)) return [];
  return results as Record<string, unknown>[];
}

async function readTableInfo(exec: Exec, table: string): Promise<Record<string, unknown>[]> {
  return exec(`SELECT * FROM pragma_table_info(${sqliteIdent(table)}) ORDER BY cid`);
}

function extractColumns(rows: Record<string, unknown>[]): ColumnDescriptor[] {
  return rows.map((r) => {
    const col: ColumnDescriptor = {
      name: String(r.name),
      sqlType: sqliteTypeToSqlType(String(r.type)),
      nullable: Number(r.notnull) === 0 && Number(r.pk) === 0,
    };
    const def = parseSqliteDefault(r.dflt_value === null ? null : String(r.dflt_value));
    if (def) col.default = def;
    return col;
  });
}

function extractPrimaryKey(rows: Record<string, unknown>[]): string[] {
  return rows
    .filter((r) => Number(r.pk) > 0)
    .sort((a, b) => Number(a.pk) - Number(b.pk))
    .map((r) => String(r.name));
}

async function readIndexes(exec: Exec, table: string): Promise<IndexDescriptor[]> {
  const list = await exec(`SELECT * FROM pragma_index_list(${sqliteIdent(table)})`);
  // Stored CREATE INDEX DDL per index — the only catalog for expression keys and
  // partial-index predicates (mirrors introspectSqlite's readSqliteIndexes).
  const ddlRows = await exec(
    `SELECT name, sql FROM sqlite_master WHERE type='index' AND tbl_name = ${sqliteLiteral(table)}`,
  );
  const ddlByName = new Map(ddlRows.map((r) => [String(r.name), r.sql === null ? null : String(r.sql)] as const));
  const indexes: IndexDescriptor[] = [];
  for (const ix of list) {
    if (String(ix.origin) === "pk") continue;
    const ixName = String(ix.name);
    // pragma_index_xinfo: DESC bit + key-vs-auxiliary flag; expression keys have
    // name NULL (cid -2).
    const keyRows = await exec(`SELECT * FROM pragma_index_xinfo(${sqliteIdent(ixName)}) ORDER BY seqno`);
    indexes.push(buildSqliteIndexDescriptor(
      { name: ixName, unique: Number(ix.unique) === 1, partial: Number(ix.partial) === 1 },
      keyRows
        .filter((c) => Number(c.key) === 1)
        .map((c) => ({ name: c.name === null ? null : String(c.name), desc: Number(c.desc) === 1 })),
      ddlByName.get(ixName) ?? null,
    ));
  }
  return indexes;
}

async function readForeignKeys(exec: Exec, table: string): Promise<FkDescriptor[]> {
  const rows = await exec(`SELECT * FROM pragma_foreign_key_list(${sqliteIdent(table)}) ORDER BY id, seq`);
  const byId = new Map<number, { refTable: string; cols: string[]; refCols: string[]; onDelete: FkAction; onUpdate: FkAction; }>();
  for (const r of rows) {
    const id = Number(r.id);
    let entry = byId.get(id);
    if (!entry) {
      entry = {
        refTable: String(r.table),
        cols: [],
        refCols: [],
        onDelete: sqliteRuleToAction(String(r.on_delete)),
        onUpdate: sqliteRuleToAction(String(r.on_update)),
      };
      byId.set(id, entry);
    }
    entry.cols.push(String(r.from));
    entry.refCols.push(String(r.to));
  }
  return Array.from(byId.entries()).map(([_id, v]) => {
    const fk: FkDescriptor = {
      name: `${table}_${v.cols.join("_")}_fk`,
      columns: v.cols,
      refTable: v.refTable,
      refColumns: v.refCols,
    };
    if (v.onDelete !== "no-action") fk.onDelete = v.onDelete;
    if (v.onUpdate !== "no-action") fk.onUpdate = v.onUpdate;
    return fk;
  });
}

async function readViews(exec: Exec): Promise<ViewDescriptor[]> {
  // sqlite_master.sql holds the full `CREATE VIEW <name> AS <body>` statement, so
  // D1 gets the same view-body drift detection as the kysely sqlite path — the
  // diff's comparator strips the leading CREATE VIEW before comparing bodies.
  const rows = await exec(
    "SELECT name, sql FROM sqlite_master WHERE type='view'"
      + " AND name NOT LIKE 'sqlite\\_%' ESCAPE '\\'"
      + " AND name NOT LIKE '\\_cf\\_%' ESCAPE '\\' ORDER BY name",
  );
  return rows.map((r) => {
    const view: ViewDescriptor = { name: String(r.name) };
    if (r.sql) view.sql = String(r.sql);
    return view;
  });
}

/**
 * Quote a SQLite identifier with double-quotes, escaping any embedded
 * double-quotes (SQLite identifier escape: "" → literal ").
 * Used for pragma_* calls where bind params aren't available (wrangler
 * --command takes a complete SQL string). The introspectSqlite path uses
 * Kysely tagged templates and doesn't need this.
 */
function sqliteIdent(name: string): string {
  return `"${name.replace(/"/g, '""')}"`;
}

/**
 * Quote a VALUE as a SQL string literal (single quotes, '' escaping) for the
 * same no-bind-params wrangler constraint sqliteIdent works around.
 */
function sqliteLiteral(value: string): string {
  return `'${value.replace(/'/g, "''")}'`;
}