/**
 * Lake configuration — **entirely env-driven, zero hardcoding**. No endpoint, bucket, or
 * prefix is ever baked into source; all of it arrives from the environment (or an explicit
 * override object) so the accessor is publication-safe and portable across any
 * S3-compatible object store (e.g. Cloudflare R2, GCS, MinIO).
 *
 * **Object storage is the ONLY source of truth.** Local disk is either an engine-managed
 * cache (`cacheDir`) or an explicit dev/fixture root (`devLocalRoot`) — never a production
 * data source.
 *
 * ## Resolution precedence (first non-empty wins)
 * 1. explicit `LakeConfigInput` passed to {@link resolveLakeConfig} / `lake(config)`
 * 2. process environment
 * 3. an rclone remote (only for endpoint + keys), when `LAKE_RCLONE_REMOTE` names one
 *
 * ## Environment
 * - `LAKE_S3_ENDPOINT`       — full scheme+host of the object store, e.g. `https://<host>`  (canonical source)
 * - `LAKE_S3_BUCKET`         — bucket name
 * - `LAKE_S3_PREFIX`         — key prefix under the bucket (may be empty)
 * - `LAKE_ACCESS_KEY_ID`     — access key (falls back to `AWS_ACCESS_KEY_ID`)
 * - `LAKE_SECRET_ACCESS_KEY` — secret key (falls back to `AWS_SECRET_ACCESS_KEY`)
 * - `LAKE_RCLONE_REMOTE`     — name of a remote in the rclone config to source endpoint+keys from
 * - `RCLONE_CONFIG`          — override path to the rclone config (defaults to `~/.config/rclone/rclone.conf`)
 * - `LAKE_CACHE_DIR`         — on-disk engine cache dir (default under the OS cache dir)
 * - `LAKE_CACHE_MAX_BYTES`   — cache size cap, LRU within it (default 25 GiB; `0` disables caching)
 * - `LAKE_DEV_LOCAL_ROOT`    — **explicit dev/fixture** root; when set, reads resolve local-first
 *                              (creds-free tests/CI) and are labelled `source: "dev-local"`.
 *                              Never consulted when unset.
 *
 * ## Secret handling
 * Keys resolved here are held in memory only. They are never logged and never written to
 * disk. When an S3 table function must be built, the secret is interpolated into the
 * in-memory SQL handed to the engine — never into anything persisted or emitted.
 */

import { readFileSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";

import { LakeConfigError } from "./errors.ts";

/** Default cache cap: 25 GiB. */
export const DEFAULT_CACHE_MAX_BYTES = 25 * 1024 ** 3;

/** Fully-resolved S3 remote — present only when every field is available. */
export interface LakeS3Config {
  /** Full scheme+host, no trailing slash (e.g. `https://<host>`). */
  endpoint: string;
  bucket: string;
  /** Key prefix under the bucket, no leading/trailing slash (may be empty). */
  prefix: string;
  accessKeyId: string;
  secretAccessKey: string;
}

export interface LakeConfig {
  /** Resolved S3 remote (the canonical source), or `undefined` when not fully described. */
  s3: LakeS3Config | undefined;
  /** On-disk engine cache dir (always set; reads materialize segments here). */
  cacheDir: string;
  /** Cache size cap in bytes; `0` disables caching entirely. */
  cacheMaxBytes: number;
  /**
   * **Explicit** dev/fixture root, absolute, or `undefined`. When set, reads resolve
   * local-first from here (labelled `source: "dev-local"`); when unset it is never consulted.
   * This is NOT a production data source — object storage is the only source of truth.
   */
  devLocalRoot: string | undefined;
}

/** Caller override: any subset of the config; provided fields win over the environment. */
export interface LakeConfigInput {
  s3?: Partial<LakeS3Config>;
  /** Explicit dev/fixture root (else `LAKE_DEV_LOCAL_ROOT`). */
  devLocalRoot?: string;
  /** Engine cache dir (else `LAKE_CACHE_DIR`, else a default under the OS cache dir). */
  cacheDir?: string;
  /** Cache cap in bytes (else `LAKE_CACHE_MAX_BYTES`, else 25 GiB; `0` disables). */
  cacheMaxBytes?: number;
  /** Override the rclone config path (else `RCLONE_CONFIG` env, else `~/.config/rclone/rclone.conf`). */
  rcloneConfigPath?: string;
  /** Override the rclone remote name (else `LAKE_RCLONE_REMOTE` env). */
  rcloneRemote?: string;
  /** Environment to read from; defaults to `process.env` (injected for tests). */
  env?: Record<string, string | undefined>;
}

/** Default engine-cache dir under the OS cache location (XDG on Linux, Caches on macOS). */
export function defaultCacheDir(env: Record<string, string | undefined>): string {
  const base =
    process.platform === "darwin"
      ? join(homedir(), "Library", "Caches")
      : (nonEmpty(env.XDG_CACHE_HOME) ?? join(homedir(), ".cache"));
  return join(base, "kestrel", "lake");
}

const nonEmpty = (v: string | undefined): string | undefined => {
  if (v === undefined) return undefined;
  const t = v.trim();
  return t.length > 0 ? t : undefined;
};

const stripTrailingSlash = (s: string): string => s.replace(/\/+$/, "");
const stripSlashes = (s: string): string => s.replace(/^\/+/, "").replace(/\/+$/, "");

/**
 * Parse an rclone config (INI) and return one remote's key/value map, or `undefined` if the
 * named remote is absent. Comments (`#`/`;`) and blank lines are ignored. This is the only
 * place a secret is read off disk; the caller keeps it in memory and never re-serializes it.
 */
export function parseRcloneRemote(
  configText: string,
  remote: string,
): Record<string, string> | undefined {
  let current: string | undefined;
  let found: Record<string, string> | undefined;
  for (const raw of configText.split(/\r?\n/)) {
    const line = raw.trim();
    if (line.length === 0 || line.startsWith("#") || line.startsWith(";")) continue;
    const section = /^\[(.+)\]$/.exec(line);
    if (section) {
      current = section[1]?.trim();
      if (current === remote) found = {};
      continue;
    }
    if (current !== remote || found === undefined) continue;
    const eq = line.indexOf("=");
    if (eq < 0) continue;
    const key = line.slice(0, eq).trim();
    const value = line.slice(eq + 1).trim();
    if (key.length > 0) found[key] = value;
  }
  return found;
}

/** endpoint+keys sourced from an rclone remote; bucket/prefix still come from env/override. */
function rcloneS3(
  input: LakeConfigInput,
  env: Record<string, string | undefined>,
): Partial<LakeS3Config> {
  const fromCaller = nonEmpty(input.rcloneRemote);
  const fromEnv = nonEmpty(env.LAKE_RCLONE_REMOTE);
  const remote = fromCaller ?? fromEnv;
  if (remote === undefined) return {};
  // Name the remote the way it was actually chosen. A caller-supplied remote is usually a DEFAULT
  // (`CORPUS_DEFAULTS.rcloneRemote` is a bare `"r2"`), and reporting that as `LAKE_RCLONE_REMOTE=r2
  // set` sends the reader hunting for an env var nothing ever set.
  const named =
    fromEnv !== undefined
      ? `LAKE_RCLONE_REMOTE=${remote}`
      : `rclone remote '${remote}' (a caller default — nothing set LAKE_RCLONE_REMOTE)`;
  const path =
    nonEmpty(input.rcloneConfigPath) ??
    nonEmpty(env.RCLONE_CONFIG) ??
    join(homedir(), ".config", "rclone", "rclone.conf");
  let text: string;
  try {
    text = readFileSync(path, "utf8");
  } catch (cause) {
    throw new LakeConfigError(`${named}, but the rclone config is unreadable at ${path}`, { cause });
  }
  const r = parseRcloneRemote(text, remote);
  if (r === undefined) {
    throw new LakeConfigError(`${named}, but that remote is not in ${path}`);
  }
  const out: Partial<LakeS3Config> = {};
  const endpoint = nonEmpty(r.endpoint);
  const accessKeyId = nonEmpty(r.access_key_id);
  const secretAccessKey = nonEmpty(r.secret_access_key);
  if (endpoint !== undefined) out.endpoint = stripTrailingSlash(endpoint);
  if (accessKeyId !== undefined) out.accessKeyId = accessKeyId;
  if (secretAccessKey !== undefined) out.secretAccessKey = secretAccessKey;
  return out;
}

/** Parse a byte-count from env; non-numeric/absent → fallback. `0`/negative → 0 (disabled). */
function parseBytes(raw: string | undefined, fallback: number): number {
  const s = nonEmpty(raw);
  if (s === undefined) return fallback;
  const n = Number(s);
  if (!Number.isFinite(n)) return fallback;
  return n <= 0 ? 0 : Math.floor(n);
}

/**
 * Resolve the effective {@link LakeConfig}. Never throws for a *missing* S3 remote (a
 * legitimate dev-only setup); it throws only when an rclone remote is explicitly named but
 * unreadable. An incomplete S3 remote resolves to `s3: undefined` — reads that need it then
 * fail loudly at read time with the field that was missing.
 */
export function resolveLakeConfig(input: LakeConfigInput = {}): LakeConfig {
  const env = input.env ?? process.env;
  const rc = rcloneS3(input, env);
  const ov = input.s3 ?? {};

  const devLocalRoot = nonEmpty(input.devLocalRoot) ?? nonEmpty(env.LAKE_DEV_LOCAL_ROOT);
  const cacheDir =
    nonEmpty(input.cacheDir) ?? nonEmpty(env.LAKE_CACHE_DIR) ?? defaultCacheDir(env);
  const cacheMaxBytes =
    input.cacheMaxBytes !== undefined
      ? input.cacheMaxBytes <= 0
        ? 0
        : Math.floor(input.cacheMaxBytes)
      : parseBytes(env.LAKE_CACHE_MAX_BYTES, DEFAULT_CACHE_MAX_BYTES);

  const endpoint =
    nonEmpty(ov.endpoint) ?? nonEmpty(env.LAKE_S3_ENDPOINT) ?? nonEmpty(rc.endpoint);
  const bucket = nonEmpty(ov.bucket) ?? nonEmpty(env.LAKE_S3_BUCKET);
  const prefixRaw = ov.prefix ?? env.LAKE_S3_PREFIX ?? "";
  const accessKeyId =
    nonEmpty(ov.accessKeyId) ??
    nonEmpty(env.LAKE_ACCESS_KEY_ID) ??
    nonEmpty(env.AWS_ACCESS_KEY_ID) ??
    nonEmpty(rc.accessKeyId);
  const secretAccessKey =
    nonEmpty(ov.secretAccessKey) ??
    nonEmpty(env.LAKE_SECRET_ACCESS_KEY) ??
    nonEmpty(env.AWS_SECRET_ACCESS_KEY) ??
    nonEmpty(rc.secretAccessKey);

  let s3: LakeS3Config | undefined;
  if (
    endpoint !== undefined &&
    bucket !== undefined &&
    accessKeyId !== undefined &&
    secretAccessKey !== undefined
  ) {
    s3 = {
      endpoint: stripTrailingSlash(endpoint),
      bucket,
      prefix: stripSlashes(prefixRaw),
      accessKeyId,
      secretAccessKey,
    };
  }

  return { s3, cacheDir, cacheMaxBytes, devLocalRoot };
}