/**
 * Parquet partition writer — DuckDB `COPY … (FORMAT PARQUET)`.
 *
 * Locality (local vs object storage) is decided by the files driver; this
 * module only serialises wide events into columnar files.
 */

import { mkdir, unlink, writeFile } from "node:fs/promises";
import { dirname, join } from "node:path";

import { duckIdent, duckPath, duckQuery, openDuckDB, type DuckSession } from "./duckdb.ts";
import type { WideEvent } from "./types.ts";

/**
 * Columns that must stay VARCHAR across partitions.
 *
 * `read_json_auto` infers all-null batches as JSON; a later file with
 * `"Unauthorized"` is VARCHAR. DuckDB then fails the union.
 */
const PARQUET_VARCHAR_COLUMNS = [
  "id",
  "parent_id",
  "flow",
  "unit",
  "trigger",
  "plane",
  "tenant",
  "principal",
  "subject_id",
  "gates",
  "cache",
  "replica",
  "build_version",
  "error_code",
  "error_message",
  "input",
  "output",
  "http",
  "effects",
  "logs",
  "archived",
  "dimensions",
] as const;

/** Numeric wide-event columns — all-null batches must not become JSON. */
const PARQUET_DOUBLE_COLUMNS = [
  "replica_lag_ms",
  "cost",
  "input_tokens",
  "output_tokens",
  "prompt_version",
  "duration_ms",
  "started_at",
  "ended_at",
] as const;

/** Flattened Parquet / SQL row. */
export type ParquetRow = Record<string, string | number | boolean | null>;

/**
 * Flatten a wide event into a columnar row.
 *
 * Structured fields become JSON strings; scalar dimensions are promoted to
 * top-level columns for outlier / filter queries.
 *
 * @param event - Wide event
 */
export function wideEventToRow(event: WideEvent): ParquetRow {
  const row: ParquetRow = {
    id: event.id,
    parent_id: event.parentId ?? null,
    flow: event.flow,
    unit: event.unit ?? null,
    trigger: event.trigger,
    plane: event.plane,
    tenant: event.tenant ?? null,
    principal: event.principal ?? null,
    subject_id: event.subjectId ?? null,
    gates: JSON.stringify(event.gates),
    cache: event.cache,
    replica: event.replica ?? null,
    replica_lag_ms: event.replicaLagMs ?? null,
    cost: event.cost ?? null,
    input_tokens: event.inputTokens ?? null,
    output_tokens: event.outputTokens ?? null,
    prompt_version: event.promptVersion ?? null,
    build_version: event.buildVersion ?? null,
    error_code: event.error?.code ?? null,
    error_message: event.error?.message ?? null,
    input: event.input === undefined ? null : JSON.stringify(event.input),
    output: event.output === undefined ? null : JSON.stringify(event.output),
    http: event.http === undefined ? null : JSON.stringify(event.http),
    effects: JSON.stringify(event.effects),
    logs: JSON.stringify(event.logs),
    duration_ms: event.durationMs,
    started_at: event.startedAt,
    ended_at: event.endedAt,
    archived: JSON.stringify(event.archived ?? {}),
    dimensions: JSON.stringify(event.dimensions),
  };
  for (const [k, v] of Object.entries(event.dimensions)) {
    const col = `dim_${k}`;
    if (row[col] !== undefined) continue;
    if (v === undefined || v === null) continue;
    row[col] =
      typeof v === "string" || typeof v === "number" || typeof v === "boolean"
        ? v
        : v === null
          ? null
          : String(v);
  }
  return row;
}

/**
 * Reconstruct a {@link WideEvent} from a Parquet / SQL row.
 *
 * @param row - Flat row
 */
export function rowToWideEvent(row: Record<string, unknown>): WideEvent {
  const dimensions =
    typeof row.dimensions === "string"
      ? (JSON.parse(row.dimensions) as WideEvent["dimensions"])
      : ((row.dimensions as WideEvent["dimensions"]) ?? {});
  const archivedRaw =
    typeof row.archived === "string"
      ? (JSON.parse(row.archived) as Record<string, string>)
      : ((row.archived as Record<string, string>) ?? {});
  const errorCode = row.error_code;
  return {
    id: String(row.id),
    ...(row.parent_id != null ? { parentId: String(row.parent_id) } : {}),
    flow: String(row.flow),
    ...(row.unit != null ? { unit: String(row.unit) } : {}),
    trigger: String(row.trigger),
    plane: (row.plane as WideEvent["plane"]) ?? "user",
    tenant: row.tenant != null ? String(row.tenant) : null,
    principal: row.principal != null ? String(row.principal) : null,
    subjectId: row.subject_id != null ? String(row.subject_id) : null,
    gates: parseJsonArray(row.gates) as string[],
    cache: (row.cache as WideEvent["cache"]) ?? "none",
    ...(row.replica != null ? { replica: row.replica as "primary" | "replica" } : {}),
    ...(row.replica_lag_ms != null ? { replicaLagMs: Number(row.replica_lag_ms) } : {}),
    ...(row.cost != null ? { cost: Number(row.cost) } : {}),
    ...(row.input_tokens != null ? { inputTokens: Number(row.input_tokens) } : {}),
    ...(row.output_tokens != null ? { outputTokens: Number(row.output_tokens) } : {}),
    ...(row.prompt_version != null ? { promptVersion: Number(row.prompt_version) } : {}),
    ...(row.build_version != null ? { buildVersion: String(row.build_version) } : {}),
    error:
      errorCode != null
        ? {
            code: String(errorCode),
            ...(row.error_message != null ? { message: String(row.error_message) } : {}),
          }
        : null,
    ...(row.input != null && row.input !== ""
      ? {
          input:
            typeof row.input === "string"
              ? (JSON.parse(row.input) as unknown)
              : (row.input as unknown),
        }
      : {}),
    ...(row.output != null && row.output !== ""
      ? {
          output:
            typeof row.output === "string"
              ? (JSON.parse(row.output) as unknown)
              : (row.output as unknown),
        }
      : {}),
    ...httpFromRow(row.http),
    effects: parseJsonArray(row.effects) as WideEvent["effects"],
    logs: parseJsonArray(row.logs) as WideEvent["logs"],
    durationMs: Number(row.duration_ms ?? 0),
    startedAt: Number(row.started_at ?? 0),
    endedAt: Number(row.ended_at ?? 0),
    ...(Object.keys(archivedRaw).length > 0 ? { archived: archivedRaw } : {}),
    dimensions,
  };
}

/**
 * Write rows to a Parquet file via DuckDB (JSONL → COPY).
 *
 * @param path - Absolute output path
 * @param rows - Flattened rows
 */
export async function writeParquet(path: string, rows: readonly ParquetRow[]): Promise<void> {
  if (rows.length === 0) return;
  await mkdir(dirname(path), { recursive: true });
  const jsonl = `${path}.jsonl`;
  await writeFile(jsonl, rows.map((r) => JSON.stringify(r)).join("\n"));
  const session = await openDuckDB();
  try {
    const replace = parquetWriteReplaceSql();
    await session.conn.run(
      `COPY (SELECT * REPLACE (${replace}) FROM read_json_auto('${duckPath(jsonl)}')) TO '${duckPath(path)}' (FORMAT PARQUET, COMPRESSION ZSTD)`,
    );
  } finally {
    session.close();
    await unlink(jsonl).catch(() => undefined);
  }
}

/**
 * SQL that reads one or more Parquet files as one relation.
 *
 * Per-file JSON columns (all-null `read_json_auto` inference) are cast
 * before `UNION ALL BY NAME` so VARCHAR `"Unauthorized"` and JSON null
 * can sit in the same `error_code` column.
 *
 * @param conn - Open DuckDB connection
 * @param paths - Absolute Parquet paths
 */
export async function parquetUnionSql(
  conn: DuckSession["conn"],
  paths: readonly string[],
): Promise<string> {
  if (paths.length === 0) return "SELECT * FROM (SELECT 1 WHERE FALSE)";
  const selects: string[] = [];
  for (const path of paths) {
    selects.push(await parquetFileSelectSql(conn, path));
  }
  return selects.length === 1 ? selects[0]! : selects.join(" UNION ALL BY NAME ");
}

/**
 * Read all rows from one or more Parquet files.
 *
 * @param paths - Absolute Parquet paths
 */
export async function readParquet(paths: readonly string[]): Promise<ParquetRow[]> {
  if (paths.length === 0) return [];
  const session = await openDuckDB();
  try {
    const sql = await parquetUnionSql(session.conn, paths);
    const rows = await duckQuery(session.conn, sql);
    return rows as ParquetRow[];
  } finally {
    session.close();
  }
}

/**
 * Partition key from an event timestamp (day UTC).
 *
 * @param startedAt - Epoch-ms
 */
export function partitionKey(startedAt: number): string {
  const d = new Date(startedAt);
  const y = d.getUTCFullYear();
  const m = String(d.getUTCMonth() + 1).padStart(2, "0");
  const day = String(d.getUTCDate()).padStart(2, "0");
  return `${y}-${m}-${day}`;
}

/**
 * Build a partition object key.
 *
 * @param day - Partition day
 * @param id - Unique file id
 */
export function partitionObjectKey(day: string, id: string): string {
  return join("runs", `day=${day}`, `${id}.parquet`);
}

function parquetWriteReplaceSql(): string {
  const varchar = PARQUET_VARCHAR_COLUMNS.map(
    (col) => `CAST(${duckIdent(col)} AS VARCHAR) AS ${duckIdent(col)}`,
  );
  const doubles = PARQUET_DOUBLE_COLUMNS.map(
    (col) => `CAST(${duckIdent(col)} AS DOUBLE) AS ${duckIdent(col)}`,
  );
  return [...varchar, ...doubles].join(", ");
}

async function parquetFileSelectSql(conn: DuckSession["conn"], path: string): Promise<string> {
  const from = `read_parquet('${duckPath(path)}')`;
  const described = await duckQuery(conn, `DESCRIBE SELECT * FROM ${from}`);
  const replacements: string[] = [];
  for (const row of described) {
    const name = columnNameOf(row);
    if (name === undefined) continue;
    const type = columnTypeOf(row);
    if (type === undefined) continue;
    const ident = duckIdent(name);
    if ((PARQUET_VARCHAR_COLUMNS as readonly string[]).includes(name)) {
      replacements.push(`CAST(${ident} AS VARCHAR) AS ${ident}`);
      continue;
    }
    if ((PARQUET_DOUBLE_COLUMNS as readonly string[]).includes(name)) {
      replacements.push(`CAST(${ident} AS DOUBLE) AS ${ident}`);
      continue;
    }
    if (type.toUpperCase() === "JSON") {
      replacements.push(`CAST(${ident} AS VARCHAR) AS ${ident}`);
    }
  }
  if (replacements.length === 0) return `SELECT * FROM ${from}`;
  return `SELECT * REPLACE (${replacements.join(", ")}) FROM ${from}`;
}

function columnNameOf(row: Record<string, unknown>): string | undefined {
  const name = row.column_name ?? row.Column;
  return typeof name === "string" && name.length > 0 ? name : undefined;
}

function columnTypeOf(row: Record<string, unknown>): string | undefined {
  const type = row.column_type ?? row.Type;
  return typeof type === "string" && type.length > 0 ? type : undefined;
}

function httpFromRow(value: unknown): { http: WideEvent["http"] } | undefined {
  const raw = typeof value === "string" && value.length > 0 ? safeJson(value) : value;
  if (raw === null || raw === undefined || typeof raw !== "object") return undefined;
  const frame = raw as NonNullable<WideEvent["http"]>;
  if (typeof frame.request !== "object" || frame.request === null) return undefined;
  return { http: frame };
}

function safeJson(value: string): unknown {
  try {
    return JSON.parse(value) as unknown;
  } catch {
    return undefined;
  }
}

function parseJsonArray(value: unknown): unknown[] {
  if (Array.isArray(value)) return value;
  if (typeof value === "string" && value.length > 0) {
    try {
      const parsed: unknown = JSON.parse(value);
      return Array.isArray(parsed) ? parsed : [];
    } catch {
      return [];
    }
  }
  return [];
}