/**
 * # adapters/lake — the data-lake accessor (chDB over object storage)
 *
 * Kestrel's durable data plane is **Parquet on S3-compatible object storage** (e.g.
 * Cloudflare R2, GCS, MinIO), queried **in-process by chDB** — embedded ClickHouse, SQL over
 * Parquet — natively under Bun. This is the read-only accessor Grade and sim read history
 * through (ARCHITECTURE §4 "Adapters — Lake"). No second language.
 *
 * ## Object storage is the ONLY source of truth
 * `readParquet` / `listParquet` / `exists` read the **canonical** store — the configured
 * `s3()` remote — by default. Local disk is never a production data source. It appears in
 * exactly two demoted roles:
 *
 * - **Engine cache** (`cacheDir`): the first read of an object fetches it from object storage
 *   and materializes a local segment; subsequent reads hit that segment with **no network**,
 *   LRU-evicted within `cacheMaxBytes`. The result is still stamped `source: "s3"` — the cache
 *   is transparent, the canonical source is object storage. The active mechanism is reported
 *   on `Lake.cache` (`"adapter" | "chdb-fs" | "none"`).
 * - **Explicit dev/fixture root** (`devLocalRoot`, env `LAKE_DEV_LOCAL_ROOT`): an *opt-in*
 *   creds-free mode for tests/CI. When set, reads resolve local-first and are stamped
 *   `source: "dev-local"`. When unset it is **never** consulted.
 *
 * ### Why an adapter cache and not chDB's filesystem cache
 * chDB 3.1.0's node binding exposes no config path to *declare* a named ClickHouse filesystem
 * cache, so `SETTINGS enable_filesystem_cache=1, filesystem_cache_name='…'` is silently
 * ignored (verified: a repeat `s3()` read still re-fetches every byte). We therefore run an
 * **adapter-level content-addressed segment cache**: on a miss we materialize the object once
 * via `INSERT INTO FUNCTION file(<segment>, Parquet) SELECT * FROM s3(…)` (one remote fetch)
 * and read all subsequent queries from the local segment. Verified hermetically: MISS = full
 * fetch, HIT = 0 network requests, ~easily >10× faster (see the cache test). Segments are
 * keyed by a hash of the object URL; a lake object at a given key is treated **immutable**
 * (versioned lakes rev the path), so a changed body needs a new key or a cache clear.
 *
 * ## Loud, typed failures (fail-closed, RUNTIME §0)
 * A missing object, an unreadable path, or a bad credential is **never** an empty result — it
 * throws {@link LakeReadError}. An empty array means "the query legitimately matched no rows".
 *
 * ## ⚠️ TIMEZONE PARITY — a known data-corruption class, read this
 * Timezone-aware timestamps stored in Parquet (ClickHouse `DateTime64(_, 'UTC')` and friends)
 * read back as **UTC instants**. That is correct and lossless — but it means **any
 * hour-of-day / minute-of-day / calendar-date derivation must localize IN SQL**, e.g.
 * `toHour(ts, 'America/New_York')`, `toStartOfDay(ts, 'America/New_York')`. Deriving a
 * wall-clock field from the raw UTC instant (`toHour(ts)`) silently drifts by the UTC offset —
 * an 18:30 New York print reads back as hour 22, and an "RTH-only" filter written naively
 * leaks overnight bars. The market session's timezone is the author's to supply, in the query.
 * (See the tz-parity test in `tests/lake.test.ts`.)
 *
 * ## Secrets
 * Access keys are held in memory only, never logged, never written to disk. The `s3()` table
 * function requires the secret in the SQL string handed to the engine (in-memory only); any
 * SQL surfaced in an error message is **redacted** first. Cache segments are Parquet data
 * keyed by a URL hash — they carry no secret.
 */

import { mkdirSync, readdirSync, renameSync, rmSync, statSync, utimesSync } from "node:fs";
import { createHash } from "node:crypto";
import { join } from "node:path";

import {
  resolveLakeConfig,
  type LakeConfig,
  type LakeConfigInput,
  type LakeS3Config,
} from "./config.ts";
import { LakeReadError, type LakeSource } from "./errors.ts";

export { LakeReadError, LakeConfigError, type LakeSource } from "./errors.ts";
export {
  resolveLakeConfig,
  parseRcloneRemote,
  defaultCacheDir,
  DEFAULT_CACHE_MAX_BYTES,
  type LakeConfig,
  type LakeConfigInput,
  type LakeS3Config,
} from "./config.ts";
/**
 * The byte-level half of the lake: raw objects (`.jsonl` tapes, `.csv.zst` vendor pulls) through the
 * SAME config, into a BOUNDED local cache. Re-exported here for discoverability — but note it is
 * deliberately **chDB-free**, so a caller that only needs bytes (the tape-corpus loaders) imports
 * `./objects.ts` DIRECTLY and never pulls the native engine into its graph.
 */
export {
  objectStore,
  sha256File,
  type ObjectStore,
  type ObjectStoreOptions,
  type LakeObject,
  type EnsureOptions,
} from "./objects.ts";

/** One row of a query result — chDB JSON objects, column name → value. */
export type Row = Record<string, unknown>;

/** Which caching mechanism the {@link Lake} instance actually runs. */
export type LakeCacheMechanism = "chdb-fs" | "adapter" | "none";

/** Options for {@link Lake.readParquet}. */
export interface ReadParquetOptions {
  /** Project to these columns (validated identifiers). Omitted ⇒ `SELECT *`. */
  columns?: readonly string[];
}

/** The result of a Parquet read: the rows plus the path that produced them. */
export interface ParquetRead {
  rows: Row[];
  /**
   * The canonical source: `"s3"` for object storage (whether or not it was served warm from
   * the engine cache) or `"dev-local"` for the explicit dev/fixture root. Never silent.
   */
  source: LakeSource;
  /** The lake-relative key that was read. */
  rel: string;
  /** For an `"s3"` read: did the engine cache serve it warm (no network)? */
  cached?: boolean;
}

/**
 * The lake accessor. All methods are async: the S3 paths do real network I/O and chDB runs
 * off the event loop, so nothing here blocks the runtime.
 */
export interface Lake {
  /** The resolved configuration (secrets included in memory; never serialize this). */
  readonly config: LakeConfig;
  /** Which cache mechanism is active on this instance. */
  readonly cache: LakeCacheMechanism;
  /** Run arbitrary SQL through chDB; returns parsed JSON rows. Throws {@link LakeReadError}. */
  query(sql: string): Promise<Row[]>;
  /** Read one Parquet object from the canonical store (cache-warmed), or the dev root. */
  readParquet(rel: string, opts?: ReadParquetOptions): Promise<ParquetRead>;
  /** Does `rel` exist? Dev root (if set), else warm cache, else a cheap S3 probe. */
  exists(rel: string): Promise<boolean>;
  /** List Parquet keys under `relPrefix` — S3 glob ∪ the dev root (when set). */
  listParquet(relPrefix: string): Promise<string[]>;
}

// ─────────────────────────────────────────────────────────────────────────────
// SQL construction — validated identifiers, escaped literals, redacted secrets
// ─────────────────────────────────────────────────────────────────────────────

const IDENT = /^[A-Za-z_][A-Za-z0-9_]*$/;

/** Validate a column identifier and backtick-quote it; reject anything injection-shaped. */
function qIdent(name: string): string {
  if (!IDENT.test(name)) {
    throw new LakeReadError(`invalid column identifier: ${JSON.stringify(name)}`);
  }
  return `\`${name}\``;
}

/** Single-quote a path/URL literal after rejecting quote/backslash/newline (no injection). */
function qLiteral(value: string, rel: string): string {
  if (/['\\\n\r\0]/.test(value)) {
    throw new LakeReadError(`unsafe characters in path: ${JSON.stringify(value)}`, { rel });
  }
  return `'${value}'`;
}

/** Reject a relative key that is empty or escapes the root; return it normalized. */
function checkRel(rel: string): string {
  const r = rel.trim();
  if (r.length === 0) throw new LakeReadError("empty relative key");
  if (r.startsWith("/") || r.split("/").includes("..")) {
    throw new LakeReadError(`relative key must stay inside the lake: ${JSON.stringify(rel)}`, {
      rel,
    });
  }
  return r;
}

const columnList = (cols: readonly string[] | undefined): string =>
  cols && cols.length > 0 ? cols.map(qIdent).join(", ") : "*";

/** `${endpoint}/${bucket}/${prefix}/${rel}` with no doubled or missing slashes. */
function s3Url(s3: LakeS3Config, rel: string): string {
  const parts = [s3.endpoint, s3.bucket, s3.prefix, rel].filter((p) => p.length > 0);
  return parts.join("/");
}

/** The S3 key (`bucket/prefix`), used to strip `_path` back to a lake-relative key when listing. */
const s3KeyBase = (s3: LakeS3Config): string =>
  [s3.bucket, s3.prefix].filter((p) => p.length > 0).join("/");

/** The `s3(url, key, secret, Parquet)` table expression + a secret-redacted twin for errors. */
function s3Table(s3: LakeS3Config, url: string, rel: string): { sql: string; redacted: string } {
  const sql = `s3(${qLiteral(url, rel)}, ${qLiteral(s3.accessKeyId, rel)}, ${qLiteral(s3.secretAccessKey, rel)}, Parquet)`;
  const redacted = `s3('${url}', '***', '***', Parquet)`;
  return { sql, redacted };
}

// ─────────────────────────────────────────────────────────────────────────────
// Engine
// ─────────────────────────────────────────────────────────────────────────────

/**
 * chDB is an **optional native dependency** (declared in `optionalDependencies`) that is materialized
 * ONLY for this lake/analytics path — it is *not* installed on the core CLI / session runtime, and
 * nothing in the CLI module graph imports the lake today. Loading it at the top of the module would make
 * this file **crash at import time** wherever chDB is absent; that violates fail-closed (RUNTIME §0),
 * which requires a loud, reasoned error at the point of USE, never a silent module-load crash. So the
 * engine is loaded LAZILY here, at the first query. A caller that only imports the module pays nothing;
 * a caller that reaches a query with chDB absent gets a typed {@link LakeReadError} that names chDB as
 * the unmet optional dependency for the lake path. The specifier is held in a variable so the type
 * checker does not require `chdb` to be resolvable once it is dropped from the install (the core build
 * ships without it).
 */
interface ChdbEngine {
  queryAsync(sql: string, opts: { format: string }): Promise<{ json<T>(): T }>;
}
const CHDB_SPECIFIER = "chdb";
let chdbEnginePromise: Promise<ChdbEngine> | undefined;
async function loadChdb(): Promise<ChdbEngine> {
  if (chdbEnginePromise === undefined) {
    chdbEnginePromise = import(CHDB_SPECIFIER).then(
      (mod: unknown): ChdbEngine => {
        const m = mod as { default?: ChdbEngine } & ChdbEngine;
        return (m.default ?? m) as ChdbEngine;
      },
      (cause: unknown): never => {
        chdbEnginePromise = undefined; // allow a later attempt (e.g. after an install)
        throw new LakeReadError(
          "chdb — the optional native ClickHouse engine for the lake/analytics path — is not installed. " +
            "Install the optional dependency to run lake queries (e.g. `npm install chdb`), or use the " +
            "chdb-free bytes accessor (`objectStore` from adapters/lake/objects) for raw objects.",
          { cause },
        );
      },
    );
  }
  return chdbEnginePromise;
}

/**
 * Run SQL through chDB with JSON output and return the `data` rows. `redactedSql` (if given)
 * is what appears in an error — the raw `sql` may carry a secret and is never surfaced.
 */
async function runRows(
  sql: string,
  ctx: { redactedSql?: string; rel?: string; source?: LakeSource } = {},
): Promise<Row[]> {
  const chdb = await loadChdb(); // throws a typed LakeReadError if the optional engine is absent
  try {
    const res = await chdb.queryAsync(sql, { format: "JSON" });
    const parsed = res.json<{ data?: Row[] }>();
    return parsed.data ?? [];
  } catch (cause) {
    const shown = ctx.redactedSql ?? sql;
    throw new LakeReadError(`lake query failed: ${shown}`, {
      cause,
      ...(ctx.rel === undefined ? {} : { rel: ctx.rel }),
      ...(ctx.source === undefined ? {} : { source: ctx.source }),
    });
  }
}

/** Run a statement (INSERT/materialize) for its side effect; redact secrets in errors. */
async function runStatement(
  sql: string,
  ctx: { redactedSql: string; rel: string; source: LakeSource },
): Promise<void> {
  const chdb = await loadChdb(); // throws a typed LakeReadError if the optional engine is absent
  try {
    await chdb.queryAsync(sql, { format: "JSON" });
  } catch (cause) {
    throw new LakeReadError(`lake fetch failed: ${ctx.redactedSql}`, {
      cause,
      rel: ctx.rel,
      source: ctx.source,
    });
  }
}

const isFile = (abs: string): boolean => {
  try {
    return statSync(abs).isFile();
  } catch {
    return false;
  }
};

// ─────────────────────────────────────────────────────────────────────────────
// Factory
// ─────────────────────────────────────────────────────────────────────────────

/** Construct a {@link Lake} from an optional override (falls back to the environment). */
export function lake(input: LakeConfigInput = {}): Lake {
  const config = resolveLakeConfig(input);

  // Determine the cache mechanism at construction. chDB's node binding cannot declare a
  // filesystem cache, so "chdb-fs" is never selected here; caching is our adapter segment
  // store, unless the cap is 0 (disabled) or the cache dir is unusable.
  let cache: LakeCacheMechanism = "none";
  if (config.cacheMaxBytes > 0) {
    try {
      mkdirSync(config.cacheDir, { recursive: true });
      cache = "adapter";
    } catch {
      cache = "none";
    }
  }

  const requireS3 = (rel: string, source: LakeSource): LakeS3Config => {
    if (config.s3 === undefined) {
      throw new LakeReadError(
        `object storage is the canonical source but no S3 remote is configured ` +
          `(need LAKE_S3_ENDPOINT/BUCKET + credentials): ${rel}`,
        { rel, source },
      );
    }
    return config.s3;
  };

  /** Stable, secret-free cache-segment path for an object URL. */
  const segmentPath = (url: string): string =>
    join(config.cacheDir, `${createHash("sha256").update(url).digest("hex").slice(0, 40)}.parquet`);

  /** LRU-evict cache segments by atime until total size is within the cap. */
  function evictToCap(): void {
    if (cache !== "adapter") return;
    let entries: { path: string; size: number; atime: number }[];
    try {
      entries = (readdirSync(config.cacheDir) as string[])
        .filter((f) => f.endsWith(".parquet"))
        .map((f) => {
          const p = join(config.cacheDir, f);
          const st = statSync(p);
          return { path: p, size: st.size, atime: st.atimeMs };
        });
    } catch {
      return;
    }
    let total = entries.reduce((a, e) => a + e.size, 0);
    if (total <= config.cacheMaxBytes) return;
    entries.sort((a, b) => a.atime - b.atime); // oldest-accessed first
    for (const e of entries) {
      if (total <= config.cacheMaxBytes) break;
      try {
        rmSync(e.path, { force: true });
        total -= e.size;
      } catch {
        /* a racing reader may hold it; skip */
      }
    }
  }

  /**
   * Ensure the object is materialized to a local cache segment (one remote fetch on miss) and
   * return `{ segment, cached }`. On a hit no network happens.
   */
  async function warmSegment(
    s3: LakeS3Config,
    url: string,
    rel: string,
  ): Promise<{ segment: string; cached: boolean }> {
    const segment = segmentPath(url);
    if (isFile(segment)) {
      try {
        const now = new Date();
        utimesSync(segment, now, now); // bump recency for LRU
      } catch {
        /* best-effort */
      }
      return { segment, cached: true };
    }
    const { sql: table, redacted } = s3Table(s3, url, rel);
    const tmp = `${segment}.${process.pid}.${Date.now()}.tmp`;
    await runStatement(
      `INSERT INTO FUNCTION file(${qLiteral(tmp, rel)}, Parquet) SELECT * FROM ${table}`,
      {
        redactedSql: `INSERT INTO FUNCTION file('<segment>', Parquet) SELECT * FROM ${redacted}`,
        rel,
        source: "s3",
      },
    );
    try {
      renameSync(tmp, segment);
    } catch (cause) {
      rmSync(tmp, { force: true });
      throw new LakeReadError(`could not commit cache segment for ${rel}`, {
        cause,
        rel,
        source: "s3",
      });
    }
    evictToCap();
    return { segment, cached: false };
  }

  async function readParquet(rel: string, opts: ReadParquetOptions = {}): Promise<ParquetRead> {
    const key = checkRel(rel);
    const cols = columnList(opts.columns);

    // Explicit dev/fixture mode: local-first, only when devLocalRoot is set.
    if (config.devLocalRoot !== undefined) {
      const abs = join(config.devLocalRoot, key);
      if (isFile(abs)) {
        const rows = await runRows(`SELECT ${cols} FROM file(${qLiteral(abs, rel)}, Parquet)`, {
          rel,
          source: "dev-local",
        });
        return { rows, source: "dev-local", rel: key };
      }
    }

    // Canonical: object storage, warmed through the engine cache when enabled.
    const s3 = requireS3(rel, "s3");
    const url = s3Url(s3, key);

    if (cache === "adapter") {
      const { segment, cached } = await warmSegment(s3, url, rel);
      const rows = await runRows(`SELECT ${cols} FROM file(${qLiteral(segment, rel)}, Parquet)`, {
        rel,
        source: "s3",
      });
      return { rows, source: "s3", rel: key, cached };
    }

    // Caching disabled: read directly from object storage every time.
    const { sql: table, redacted } = s3Table(s3, url, rel);
    const rows = await runRows(`SELECT ${cols} FROM ${table}`, {
      redactedSql: `SELECT ${cols} FROM ${redacted}`,
      rel,
      source: "s3",
    });
    return { rows, source: "s3", rel: key, cached: false };
  }

  async function exists(rel: string): Promise<boolean> {
    const key = checkRel(rel);

    if (config.devLocalRoot !== undefined && isFile(join(config.devLocalRoot, key))) {
      return true;
    }
    if (config.s3 === undefined) {
      // No S3 remote AND no dev root ⇒ no canonical source at all: that is a config fault,
      // not an absent object. Fail closed like readParquet's requireS3 — never mask a
      // misconfigured/uncredentialed lake as genuinely-absent history to a Grade/sim caller.
      // requireS3 always throws here (config.s3 is undefined); invoke it for that
      // side-effect rather than returning it — its LakeS3Config result is never a boolean.
      if (config.devLocalRoot === undefined) requireS3(rel, "s3");
      // A dev root IS configured and the key is not under it ⇒ genuinely absent.
      return false;
    }

    const url = s3Url(config.s3, key);
    if (cache === "adapter" && isFile(segmentPath(url))) return true; // warm ⇒ definitely exists

    // S3 probe: the `One` format lists the matched object WITHOUT reading its contents (one
    // row per file), a cheap HEAD-equivalent — it does not scan the Parquet. A genuinely-absent
    // key makes s3() raise a not-found error, read here as `false`; any other failure (bad
    // creds, network) stays loud as a LakeReadError.
    const sql =
      `SELECT 1 FROM s3(${qLiteral(url, rel)}, ` +
      `${qLiteral(config.s3.accessKeyId, rel)}, ${qLiteral(config.s3.secretAccessKey, rel)}, One) LIMIT 1`;
    const redactedSql = `SELECT 1 FROM s3('${url}', '***', '***', One) LIMIT 1`;
    try {
      const rows = await runRows(sql, { redactedSql, rel, source: "s3" });
      return rows.length > 0;
    } catch (err) {
      const msg =
        err instanceof Error ? (err.cause instanceof Error ? err.cause.message : err.message) : "";
      if (/not found|no such|does not exist|Cannot stat|404|NoSuchKey/i.test(msg)) return false;
      throw err;
    }
  }

  function devLocalList(relPrefix: string): string[] {
    if (config.devLocalRoot === undefined) return [];
    let entries: string[];
    try {
      entries = readdirSync(config.devLocalRoot, { recursive: true }) as string[];
    } catch {
      return [];
    }
    return entries
      .map((e) => e.split("\\").join("/"))
      .filter((e) => e.endsWith(".parquet") && e.startsWith(relPrefix));
  }

  async function s3List(relPrefix: string): Promise<string[]> {
    if (config.s3 === undefined) return [];
    // Recursive glob under prefix; `**` spans '/' in ClickHouse. `One` format lists without
    // reading file bodies. `_path` is `bucket/key`; strip the key base back to a lake key.
    const base = s3Url(config.s3, "").replace(/\/+$/, "");
    const glob = `${base}/${relPrefix}**.parquet`;
    const sql =
      `SELECT DISTINCT _path AS p FROM s3(${qLiteral(glob, relPrefix)}, ` +
      `${qLiteral(config.s3.accessKeyId, relPrefix)}, ${qLiteral(config.s3.secretAccessKey, relPrefix)}, One)`;
    const redactedSql = `SELECT DISTINCT _path AS p FROM s3('${glob}', '***', '***', One)`;
    const rows = await runRows(sql, { redactedSql, source: "s3" });
    const keyBase = `${s3KeyBase(config.s3)}/`;
    const out: string[] = [];
    for (const r of rows) {
      const p = typeof r.p === "string" ? r.p : undefined;
      if (p === undefined) continue;
      out.push(p.startsWith(keyBase) ? p.slice(keyBase.length) : p);
    }
    return out;
  }

  async function listParquet(relPrefix: string): Promise<string[]> {
    const prefix = relPrefix.trim();
    const s3keys = await s3List(prefix);
    const union = new Set<string>([...devLocalList(prefix), ...s3keys]);
    return [...union].sort();
  }

  return {
    config,
    cache,
    query: (sql: string) => runRows(sql),
    readParquet,
    exists,
    listParquet,
  };
}