import {
  AlertEventSchema,
  ClassifierHitEventSchema,
  EquityCandleSchema,
  EquityPrintSchema,
  EquityQuoteSchema,
  EquityPrintJoinSchema,
  InferredDarkEventSchema,
  FlowPacketSchema,
  NewsStorySchema,
  OptionNBBOSchema,
  OptionPrintSchema,
  SmartMoneyEventSchema
} from "@islandflow/types";
import type {
  AlertEvent,
  ClassifierHitEvent,
  EquityCandle,
  EquityPrint,
  EquityQuote,
  EquityPrintJoin,
  InferredDarkEvent,
  FlowPacket,
  NewsStory,
  SmartMoneyEvent,
  OptionNBBO,
  OptionPrint,
  OptionFlowFilters,
  OptionFlowView
} from "@islandflow/types";
import {
  normalizeOptionPrint,
  optionPrintsTableDDL,
  optionPrintsTableMigrations,
  OPTION_PRINTS_TABLE
} from "./option-prints";
import { normalizeOptionNBBO, optionNBBOTableDDL, OPTION_NBBO_TABLE } from "./option-nbbo";
import {
  equityPrintsTableDDL,
  EQUITY_PRINTS_TABLE,
  normalizeEquityPrint
} from "./equity-prints";
import {
  equityQuotesTableDDL,
  EQUITY_QUOTES_TABLE,
  normalizeEquityQuote
} from "./equity-quotes";
import {
  equityCandlesTableDDL,
  EQUITY_CANDLES_TABLE,
  normalizeEquityCandle
} from "./equity-candles";
import {
  equityPrintJoinsTableDDL,
  EQUITY_PRINT_JOINS_TABLE,
  fromEquityPrintJoinRecord,
  toEquityPrintJoinRecord,
  type EquityPrintJoinRecord
} from "./equity-print-joins";
import {
  inferredDarkTableDDL,
  INFERRED_DARK_TABLE,
  fromInferredDarkRecord,
  toInferredDarkRecord,
  type InferredDarkRecord
} from "./inferred-dark";
import {
  FLOW_PACKETS_TABLE,
  flowPacketsTableDDL,
  fromFlowPacketRecord,
  toFlowPacketRecord,
  type FlowPacketRecord
} from "./flow-packets";
import {
  CLASSIFIER_HITS_TABLE,
  classifierHitsTableDDL,
  fromClassifierHitRecord,
  toClassifierHitRecord,
  type ClassifierHitRecord
} from "./classifier-hits";
import {
  ALERTS_TABLE,
  alertsTableMigrations,
  alertsTableDDL,
  fromAlertRecord,
  toAlertRecord,
  type AlertRecord
} from "./alerts";
import {
  SMART_MONEY_EVENTS_TABLE,
  smartMoneyEventsTableDDL,
  fromSmartMoneyEventRecord,
  toSmartMoneyEventRecord,
  type SmartMoneyEventRecord
} from "./smart-money-events";
import {
  NEWS_TABLE,
  newsTableDDL,
  fromNewsRecord,
  toNewsRecord,
  type NewsRecord
} from "./news";

export type ClickHouseOptions = {
  url: string;
  database?: string;
  username?: string;
  password?: string;
};

type ClickHouseQueryFormat = "JSONEachRow";

type ClickHouseQueryResult = {
  json<T>(): Promise<T>;
};

export type ClickHouseClient = {
  exec(params: { query: string }): Promise<void>;
  insert(params: { table: string; values: unknown[]; format: ClickHouseQueryFormat }): Promise<void>;
  query(params: { query: string; format: ClickHouseQueryFormat }): Promise<ClickHouseQueryResult>;
  ping(): Promise<{ success: boolean; error?: Error }>;
  close(): Promise<void>;
};

const buildBaseUrl = (options: ClickHouseOptions): URL => {
  const url = new URL(options.url);

  if (options.database) {
    url.searchParams.set("database", options.database);
  }

  return url;
};

const buildHeaders = (options: ClickHouseOptions, hasBody: boolean): Headers => {
  const headers = new Headers();

  if (hasBody) {
    headers.set("content-type", "text/plain; charset=utf-8");
  }

  if (options.username || options.password) {
    const auth = Buffer.from(`${options.username ?? "default"}:${options.password ?? ""}`).toString("base64");
    headers.set("authorization", `Basic ${auth}`);
  }

  return headers;
};

const executeClickHouse = async (
  options: ClickHouseOptions,
  query: string,
  body?: string
): Promise<Response> => {
  const url = buildBaseUrl(options);
  url.searchParams.set("query", query);

  const response = await fetch(url, {
    method: "POST",
    headers: buildHeaders(options, body !== undefined),
    body
  });

  if (!response.ok) {
    const message = (await response.text()).trim() || `${response.status} ${response.statusText}`;
    throw new Error(message);
  }

  return response;
};

const parseJsonEachRow = <T>(text: string): T => {
  const trimmed = text.trim();

  if (!trimmed) {
    return [] as T;
  }

  const rows = trimmed
    .split("\n")
    .filter((line) => line.trim().length > 0)
    .map((line) => JSON.parse(line));

  return rows as T;
};

export const createClickHouseClient = (options: ClickHouseOptions): ClickHouseClient => {
  return {
    async exec({ query }) {
      await executeClickHouse(options, query);
    },

    async insert({ table, values, format }) {
      const rows = values.map((value) => JSON.stringify(value)).join("\n");
      const body = rows.length > 0 ? `${rows}\n` : "";
      await executeClickHouse(options, `INSERT INTO ${table} FORMAT ${format}`, body);
    },

    async query({ query, format }) {
      const response = await executeClickHouse(options, `${query} FORMAT ${format}`);
      return {
        async json<T>() {
          const text = await response.text();
          return parseJsonEachRow<T>(text);
        }
      };
    },

    async ping() {
      try {
        const url = buildBaseUrl(options);
        url.pathname = "/ping";

        const response = await fetch(url, {
          method: "GET",
          headers: buildHeaders(options, false)
        });

        if (!response.ok) {
          const message = (await response.text()).trim() || `${response.status} ${response.statusText}`;
          return { success: false, error: new Error(message) };
        }

        return { success: true };
      } catch (error) {
        if (error instanceof Error) {
          return { success: false, error };
        }

        throw error;
      }
    },

    async close() {
      return;
    }
  };
};

export const ensureOptionPrintsTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: optionPrintsTableDDL()
  });
  for (const query of optionPrintsTableMigrations()) {
    await client.exec({ query });
  }
};

export const ensureOptionNBBOTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: optionNBBOTableDDL()
  });
};

export const ensureEquityPrintsTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: equityPrintsTableDDL()
  });
};

export const ensureEquityQuotesTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: equityQuotesTableDDL()
  });
};

export const ensureEquityCandlesTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: equityCandlesTableDDL()
  });
};

export const ensureEquityPrintJoinsTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: equityPrintJoinsTableDDL()
  });
};

export const ensureInferredDarkTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: inferredDarkTableDDL()
  });
};

export const ensureFlowPacketsTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: flowPacketsTableDDL()
  });
};

export const ensureSmartMoneyEventsTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: smartMoneyEventsTableDDL()
  });
};

export const ensureClassifierHitsTable = async (
  client: ClickHouseClient
): Promise<void> => {
  await client.exec({
    query: classifierHitsTableDDL()
  });
};

export const ensureAlertsTable = async (client: ClickHouseClient): Promise<void> => {
  await client.exec({
    query: alertsTableDDL()
  });
  for (const query of alertsTableMigrations()) {
    await client.exec({ query });
  }
};

export const ensureNewsTable = async (client: ClickHouseClient): Promise<void> => {
  await client.exec({
    query: newsTableDDL()
  });
};

export const insertOptionPrint = async (
  client: ClickHouseClient,
  print: OptionPrint
): Promise<void> => {
  const record = normalizeOptionPrint(print);
  await client.insert({
    table: OPTION_PRINTS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertOptionNBBO = async (
  client: ClickHouseClient,
  nbbo: OptionNBBO
): Promise<void> => {
  const record = normalizeOptionNBBO(nbbo);
  await client.insert({
    table: OPTION_NBBO_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertEquityPrint = async (
  client: ClickHouseClient,
  print: EquityPrint
): Promise<void> => {
  const record = normalizeEquityPrint(print);
  await client.insert({
    table: EQUITY_PRINTS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertEquityQuote = async (
  client: ClickHouseClient,
  quote: EquityQuote
): Promise<void> => {
  const record = normalizeEquityQuote(quote);
  await client.insert({
    table: EQUITY_QUOTES_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertEquityCandle = async (
  client: ClickHouseClient,
  candle: EquityCandle
): Promise<void> => {
  const record = normalizeEquityCandle(candle);
  await client.insert({
    table: EQUITY_CANDLES_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertEquityPrintJoin = async (
  client: ClickHouseClient,
  join: EquityPrintJoin
): Promise<void> => {
  const record = toEquityPrintJoinRecord(join);
  await client.insert({
    table: EQUITY_PRINT_JOINS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertInferredDark = async (
  client: ClickHouseClient,
  event: InferredDarkEvent
): Promise<void> => {
  const record = toInferredDarkRecord(event);
  await client.insert({
    table: INFERRED_DARK_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertFlowPacket = async (
  client: ClickHouseClient,
  packet: FlowPacket
): Promise<void> => {
  const record = toFlowPacketRecord(packet);
  await client.insert({
    table: FLOW_PACKETS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertSmartMoneyEvent = async (
  client: ClickHouseClient,
  event: SmartMoneyEvent
): Promise<void> => {
  const record = toSmartMoneyEventRecord(event);
  await client.insert({
    table: SMART_MONEY_EVENTS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertClassifierHit = async (
  client: ClickHouseClient,
  hit: ClassifierHitEvent
): Promise<void> => {
  const record = toClassifierHitRecord(hit);
  await client.insert({
    table: CLASSIFIER_HITS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertAlert = async (client: ClickHouseClient, alert: AlertEvent): Promise<void> => {
  const record = toAlertRecord(alert);
  await client.insert({
    table: ALERTS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export const insertNewsStory = async (client: ClickHouseClient, story: NewsStory): Promise<void> => {
  const record = toNewsRecord(story);
  await client.insert({
    table: NEWS_TABLE,
    values: [record],
    format: "JSONEachRow"
  });
};

export type ClickHouseBatchWriterOptions = {
  flushIntervalMs?: number;
  maxRows?: number;
  onError?: (table: string, error: unknown, rowCount: number) => void;
};

type BatchState = {
  rows: unknown[];
  timer: ReturnType<typeof setTimeout> | null;
  flushing: Promise<void> | null;
};

const createBatchState = (): BatchState => ({
  rows: [],
  timer: null,
  flushing: null
});

export class ClickHouseBatchWriter {
  private readonly flushIntervalMs: number;
  private readonly maxRows: number;
  private readonly states = new Map<string, BatchState>();

  constructor(
    private readonly client: ClickHouseClient,
    options: ClickHouseBatchWriterOptions = {}
  ) {
    this.flushIntervalMs = Math.max(1, Math.floor(options.flushIntervalMs ?? 100));
    this.maxRows = Math.max(1, Math.floor(options.maxRows ?? 250));
    this.onError = options.onError;
  }

  private readonly onError?: (table: string, error: unknown, rowCount: number) => void;

  enqueue(table: string, row: unknown): void {
    const state = this.states.get(table) ?? createBatchState();
    if (!this.states.has(table)) {
      this.states.set(table, state);
    }

    state.rows.push(row);

    if (state.rows.length >= this.maxRows) {
      void this.flush(table);
      return;
    }

    if (!state.timer) {
      state.timer = setTimeout(() => {
        state.timer = null;
        void this.flush(table);
      }, this.flushIntervalMs);
    }
  }

  async flush(table: string): Promise<void> {
    const state = this.states.get(table);
    if (!state) {
      return;
    }

    if (state.flushing) {
      await state.flushing;
      return;
    }

    if (state.timer) {
      clearTimeout(state.timer);
      state.timer = null;
    }

    if (state.rows.length === 0) {
      return;
    }

    const rows = state.rows.splice(0, state.rows.length);
    state.flushing = this.client
      .insert({
        table,
        values: rows,
        format: "JSONEachRow"
      })
      .catch((error) => {
        this.onError?.(table, error, rows.length);
      })
      .finally(() => {
        state.flushing = null;
      });

    await state.flushing;
  }

  async flushAll(): Promise<void> {
    for (const table of this.states.keys()) {
      await this.flush(table);
    }
  }

  async close(): Promise<void> {
    for (const state of this.states.values()) {
      if (state.timer) {
        clearTimeout(state.timer);
        state.timer = null;
      }
    }
    await this.flushAll();
  }
}

export const enqueueEquityPrintJoinInsert = (
  writer: ClickHouseBatchWriter,
  join: EquityPrintJoin
): void => {
  writer.enqueue(EQUITY_PRINT_JOINS_TABLE, toEquityPrintJoinRecord(join));
};

export const enqueueInferredDarkInsert = (
  writer: ClickHouseBatchWriter,
  event: InferredDarkEvent
): void => {
  writer.enqueue(INFERRED_DARK_TABLE, toInferredDarkRecord(event));
};

export const enqueueFlowPacketInsert = (
  writer: ClickHouseBatchWriter,
  packet: FlowPacket
): void => {
  writer.enqueue(FLOW_PACKETS_TABLE, toFlowPacketRecord(packet));
};

export const enqueueSmartMoneyEventInsert = (
  writer: ClickHouseBatchWriter,
  event: SmartMoneyEvent
): void => {
  writer.enqueue(SMART_MONEY_EVENTS_TABLE, toSmartMoneyEventRecord(event));
};

export const enqueueClassifierHitInsert = (
  writer: ClickHouseBatchWriter,
  hit: ClassifierHitEvent
): void => {
  writer.enqueue(CLASSIFIER_HITS_TABLE, toClassifierHitRecord(hit));
};

export const enqueueAlertInsert = (
  writer: ClickHouseBatchWriter,
  alert: AlertEvent
): void => {
  writer.enqueue(ALERTS_TABLE, toAlertRecord(alert));
};

export const enqueueNewsStoryInsert = (
  writer: ClickHouseBatchWriter,
  story: NewsStory
): void => {
  writer.enqueue(NEWS_TABLE, toNewsRecord(story));
};

const clampLimit = (limit: number): number => {
  if (!Number.isFinite(limit)) {
    return 100;
  }

  return Math.max(1, Math.min(1000, Math.floor(limit)));
};

const clampLookupLimit = (limit: number): number => {
  if (!Number.isFinite(limit)) {
    return 100;
  }

  return Math.max(1, Math.min(5000, Math.floor(limit)));
};

const clampPositiveInt = (value: number, fallback = 1): number => {
  if (!Number.isFinite(value)) {
    return fallback;
  }

  return Math.max(1, Math.floor(value));
};

const clampCursor = (value: number): number => {
  if (!Number.isFinite(value)) {
    return 0;
  }

  return Math.max(0, Math.floor(value));
};

const coerceNumber = (value: unknown): unknown => {
  if (typeof value === "string") {
    const parsed = Number(value);
    if (Number.isFinite(parsed)) {
      return parsed;
    }
  }

  return value;
};

const quoteString = (value: string): string => {
  const escaped = value.replace(/'/g, "''");
  return `'${escaped}'`;
};

const buildStringList = (values: string[]): string => {
  return values.map((value) => quoteString(value)).join(", ");
};

const buildTracePrefixCondition = (tracePrefix: string | undefined): string | null => {
  if (!tracePrefix) {
    return null;
  }
  const normalized = tracePrefix.trim();
  if (!normalized) {
    return null;
  }
  return `startsWith(trace_id, ${quoteString(normalized)})`;
};

const buildBeforeTupleCondition = (
  tsColumn: string,
  seqColumn: string,
  beforeTs: number,
  beforeSeq: number
): string => {
  return `(${tsColumn}, ${seqColumn}) < (${clampCursor(beforeTs)}, ${clampCursor(beforeSeq)})`;
};

const normalizeNumericFields = (
  row: Record<string, unknown>,
  fields: string[]
): Record<string, unknown> => {
  const record: Record<string, unknown> = { ...row };

  for (const field of fields) {
    if (field in record) {
      record[field] = coerceNumber(record[field]);
    }
  }

  return record;
};

const normalizeOptionRow = (row: unknown): unknown => {
  if (row && typeof row === "object") {
    const record = normalizeNumericFields(row as Record<string, unknown>, [
      "source_ts",
      "ingest_ts",
      "seq",
      "ts",
      "price",
      "size",
      "notional",
      "execution_nbbo_bid",
      "execution_nbbo_ask",
      "execution_nbbo_mid",
      "execution_nbbo_spread",
      "execution_nbbo_bid_size",
      "execution_nbbo_ask_size",
      "execution_nbbo_ts",
      "execution_nbbo_age_ms",
      "execution_underlying_spot",
      "execution_underlying_bid",
      "execution_underlying_ask",
      "execution_underlying_mid",
      "execution_underlying_spread",
      "execution_underlying_ts",
      "execution_underlying_age_ms",
      "execution_iv"
    ]);

    if ("is_etf" in record) {
      record.is_etf = Boolean(record.is_etf);
    }
    if ("signal_pass" in record) {
      record.signal_pass = Boolean(record.signal_pass);
    }
    if (record.signal_reasons == null) {
      record.signal_reasons = [];
    }
    return record;
  }

  return row;
};

export type OptionPrintQueryFilters = {
  view?: OptionFlowView;
  minNotional?: number;
  security?: "stock" | "etf" | "all";
  optionTypes?: string[];
  nbboSides?: string[];
  underlyingIds?: string[];
  optionContractId?: string;
  sinceTs?: number;
};

export type EquityPrintQueryFilters = {
  underlyingIds?: string[];
  sinceTs?: number;
};

export type AlertContextBundle = {
  alert: AlertEvent | null;
  flow_packets: FlowPacket[];
  option_prints: OptionPrint[];
  missing_refs: string[];
};

const buildOptionPrintFilterConditions = (
  filters: OptionPrintQueryFilters | undefined,
  tracePrefix: string | undefined
): string[] => {
  const conditions: string[] = [];
  const traceCondition = buildTracePrefixCondition(tracePrefix);
  if (traceCondition) {
    conditions.push(traceCondition);
  }

  if (!filters) {
    return conditions;
  }

  if ((filters.view ?? "signal") === "signal") {
    conditions.push("signal_pass = 1");
  }

  if (typeof filters.minNotional === "number" && Number.isFinite(filters.minNotional)) {
    conditions.push(`notional >= ${filters.minNotional}`);
  }

  if (filters.security === "stock") {
    conditions.push("(is_etf = 0 OR is_etf IS NULL)");
  } else if (filters.security === "etf") {
    conditions.push("is_etf = 1");
  }

  if (filters.optionTypes && filters.optionTypes.length > 0) {
    conditions.push(`option_type IN (${buildStringList(filters.optionTypes)})`);
  }

  if (filters.nbboSides && filters.nbboSides.length > 0) {
    conditions.push(`nbbo_side IN (${buildStringList(filters.nbboSides)})`);
  }

  if (filters.underlyingIds && filters.underlyingIds.length > 0) {
    conditions.push(`underlying_id IN (${buildStringList(filters.underlyingIds)})`);
  }

  if (filters.optionContractId) {
    conditions.push(`option_contract_id = ${quoteString(filters.optionContractId)}`);
  }

  if (typeof filters.sinceTs === "number" && Number.isFinite(filters.sinceTs)) {
    conditions.push(`ts >= ${clampCursor(filters.sinceTs)}`);
  }

  return conditions;
};

const buildEquityPrintFilterConditions = (filters?: EquityPrintQueryFilters): string[] => {
  const conditions: string[] = [];
  if (!filters) {
    return conditions;
  }
  if (filters.underlyingIds && filters.underlyingIds.length > 0) {
    conditions.push(`underlying_id IN (${buildStringList(filters.underlyingIds)})`);
  }
  if (typeof filters.sinceTs === "number" && Number.isFinite(filters.sinceTs)) {
    conditions.push(`ts >= ${clampCursor(filters.sinceTs)}`);
  }
  return conditions;
};

const normalizeOptionNbboRow = (row: unknown): unknown => {
  if (row && typeof row === "object") {
    return normalizeNumericFields(row as Record<string, unknown>, [
      "source_ts",
      "ingest_ts",
      "seq",
      "ts",
      "bid",
      "ask",
      "bidSize",
      "askSize"
    ]);
  }

  return row;
};

const normalizeEquityQuoteRow = (row: unknown): unknown => {
  if (row && typeof row === "object") {
    return normalizeNumericFields(row as Record<string, unknown>, [
      "source_ts",
      "ingest_ts",
      "seq",
      "ts",
      "bid",
      "ask"
    ]);
  }

  return row;
};

const normalizeEquityCandleRow = (row: unknown): unknown => {
  if (row && typeof row === "object") {
    return normalizeNumericFields(row as Record<string, unknown>, [
      "source_ts",
      "ingest_ts",
      "seq",
      "ts",
      "interval_ms",
      "open",
      "high",
      "low",
      "close",
      "volume",
      "trade_count"
    ]);
  }

  return row;
};

const normalizeEquityRow = (row: unknown): unknown => {
  if (row && typeof row === "object") {
    const record = normalizeNumericFields(row as Record<string, unknown>, [
      "source_ts",
      "ingest_ts",
      "seq",
      "ts",
      "price",
      "size"
    ]);

    if ("offExchangeFlag" in record) {
      return {
        ...record,
        offExchangeFlag: Boolean(record.offExchangeFlag)
      };
    }

    return record;
  }

  return row;
};

const normalizeEquityPrintJoinRow = (row: unknown): EquityPrintJoinRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    id: String(record.id ?? ""),
    print_trace_id: String(record.print_trace_id ?? ""),
    quote_trace_id: String(record.quote_trace_id ?? ""),
    features_json: String(record.features_json ?? "{}"),
    join_quality_json: String(record.join_quality_json ?? "{}")
  };
};

const normalizeInferredDarkRow = (row: unknown): InferredDarkRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    type: String(record.type ?? ""),
    confidence: Number(coerceNumber(record.confidence) ?? 0),
    evidence_refs_json: String(record.evidence_refs_json ?? "[]")
  };
};

const normalizeFlowPacketRow = (row: unknown): FlowPacketRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    id: String(record.id ?? ""),
    members: Array.isArray(record.members)
      ? record.members.map((value) => String(value))
      : [],
    features_json: String(record.features_json ?? "{}"),
    join_quality_json: String(record.join_quality_json ?? "{}")
  };
};

const normalizeClassifierHitRow = (row: unknown): ClassifierHitRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    classifier_id: String(record.classifier_id ?? ""),
    confidence: Number(coerceNumber(record.confidence) ?? 0),
    direction: String(record.direction ?? ""),
    explanations_json: String(record.explanations_json ?? "[]")
  };
};

const normalizeSmartMoneyEventRow = (row: unknown): SmartMoneyEventRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    event_id: String(record.event_id ?? ""),
    packet_ids: Array.isArray(record.packet_ids) ? record.packet_ids.map((value) => String(value)) : [],
    member_print_ids: Array.isArray(record.member_print_ids)
      ? record.member_print_ids.map((value) => String(value))
      : [],
    underlying_id: String(record.underlying_id ?? ""),
    event_kind: String(record.event_kind ?? ""),
    event_window_ms: coerceNumber(record.event_window_ms) as number,
    features_json: String(record.features_json ?? "{}"),
    profile_scores_json: String(record.profile_scores_json ?? "[]"),
    primary_profile_id: String(record.primary_profile_id ?? ""),
    primary_direction: String(record.primary_direction ?? "unknown"),
    abstained: Boolean(record.abstained),
    suppressed_reasons_json: String(record.suppressed_reasons_json ?? "[]")
  };
};

const normalizeAlertRow = (row: unknown): AlertRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    score: Number(coerceNumber(record.score) ?? 0),
    severity: String(record.severity ?? ""),
    hits_json: String(record.hits_json ?? "[]"),
    evidence_refs_json: String(record.evidence_refs_json ?? "[]"),
    primary_profile_id: String(record.primary_profile_id ?? ""),
    profile_scores_json: String(record.profile_scores_json ?? "[]")
  };
};

const normalizeNewsRow = (row: unknown): NewsRecord | null => {
  if (!row || typeof row !== "object") {
    return null;
  }

  const record = row as Record<string, unknown>;
  return {
    source_ts: coerceNumber(record.source_ts) as number,
    ingest_ts: coerceNumber(record.ingest_ts) as number,
    seq: coerceNumber(record.seq) as number,
    trace_id: String(record.trace_id ?? ""),
    story_id: coerceNumber(record.story_id) as number,
    provider: String(record.provider ?? ""),
    source: String(record.source ?? ""),
    headline: String(record.headline ?? ""),
    summary: String(record.summary ?? ""),
    content_html: String(record.content_html ?? ""),
    url: String(record.url ?? ""),
    published_ts: coerceNumber(record.published_ts) as number,
    updated_ts: coerceNumber(record.updated_ts) as number,
    provider_symbols_json: String(record.provider_symbols_json ?? "[]"),
    resolved_symbols_json: String(record.resolved_symbols_json ?? "[]"),
    symbol_resolution: String(record.symbol_resolution ?? "none") as NewsRecord["symbol_resolution"]
  };
};

export const fetchRecentOptionPrints = async (
  client: ClickHouseClient,
  limit: number,
  tracePrefix?: string,
  filters?: OptionPrintQueryFilters
): Promise<OptionPrint[]> => {
  const safeLimit = clampLimit(limit);
  const conditions = buildOptionPrintFilterConditions(filters, tracePrefix);
  const whereClause = conditions.length > 0 ? ` WHERE ${conditions.join(" AND ")}` : "";
  const result = await client.query({
    query: `SELECT * FROM ${OPTION_PRINTS_TABLE}${whereClause} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
};

export const fetchRecentOptionNBBO = async (
  client: ClickHouseClient,
  limit: number,
  tracePrefix?: string
): Promise<OptionNBBO[]> => {
  const safeLimit = clampLimit(limit);
  const condition = buildTracePrefixCondition(tracePrefix);
  const whereClause = condition ? ` WHERE ${condition}` : "";
  const result = await client.query({
    query: `SELECT * FROM ${OPTION_NBBO_TABLE}${whereClause} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionNBBOSchema.array().parse(rows.map(normalizeOptionNbboRow));
};

export const fetchRecentEquityPrints = async (
  client: ClickHouseClient,
  limit: number,
  filters?: EquityPrintQueryFilters
): Promise<EquityPrint[]> => {
  const safeLimit = clampLimit(limit);
  const conditions = buildEquityPrintFilterConditions(filters);
  const whereClause = conditions.length > 0 ? ` WHERE ${conditions.join(" AND ")}` : "";
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINTS_TABLE}${whereClause} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
};

export const fetchRecentEquityQuotes = async (
  client: ClickHouseClient,
  limit: number
): Promise<EquityQuote[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_QUOTES_TABLE} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityQuoteSchema.array().parse(rows.map(normalizeEquityQuoteRow));
};

export const fetchRecentEquityCandles = async (
  client: ClickHouseClient,
  underlyingId: string,
  intervalMs: number,
  limit: number
): Promise<EquityCandle[]> => {
  const safeLimit = clampLimit(limit);
  const safeInterval = clampPositiveInt(intervalMs, 1000);
  const safeUnderlying = quoteString(underlyingId);
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_CANDLES_TABLE} WHERE underlying_id = ${safeUnderlying} AND interval_ms = ${safeInterval} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityCandleSchema.array().parse(rows.map(normalizeEquityCandleRow));
};

export const fetchRecentEquityPrintJoins = async (
  client: ClickHouseClient,
  limit: number
): Promise<EquityPrintJoin[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeEquityPrintJoinRow)
    .filter((record): record is EquityPrintJoinRecord => record !== null);
  const joins = records.map(fromEquityPrintJoinRecord);
  return EquityPrintJoinSchema.array().parse(joins);
};

export const fetchRecentInferredDark = async (
  client: ClickHouseClient,
  limit: number
): Promise<InferredDarkEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${INFERRED_DARK_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeInferredDarkRow)
    .filter((record): record is InferredDarkRecord => record !== null);
  const events = records.map(fromInferredDarkRecord);
  return InferredDarkEventSchema.array().parse(events);
};

export const fetchRecentFlowPackets = async (
  client: ClickHouseClient,
  limit: number
): Promise<FlowPacket[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeFlowPacketRow)
    .filter((record): record is FlowPacketRecord => record !== null);
  const packets = records.map(fromFlowPacketRecord);
  return FlowPacketSchema.array().parse(packets);
};

export const fetchRecentClassifierHits = async (
  client: ClickHouseClient,
  limit: number
): Promise<ClassifierHitEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${CLASSIFIER_HITS_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeClassifierHitRow)
    .filter((record): record is ClassifierHitRecord => record !== null);
  const hits = records.map(fromClassifierHitRecord);
  return ClassifierHitEventSchema.array().parse(hits);
};

export const fetchRecentSmartMoneyEvents = async (
  client: ClickHouseClient,
  limit: number
): Promise<SmartMoneyEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${SMART_MONEY_EVENTS_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeSmartMoneyEventRow)
    .filter((record): record is SmartMoneyEventRecord => record !== null);
  return SmartMoneyEventSchema.array().parse(records.map(fromSmartMoneyEventRecord));
};

export const fetchRecentAlerts = async (
  client: ClickHouseClient,
  limit: number
): Promise<AlertEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${ALERTS_TABLE} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeAlertRow)
    .filter((record): record is AlertRecord => record !== null);
  const alerts = records.map(fromAlertRecord);
  return AlertEventSchema.array().parse(alerts);
};

const latestNewsSelect = `
SELECT
  source_ts,
  ingest_ts,
  seq,
  trace_id,
  story_id,
  provider,
  source,
  headline,
  summary,
  content_html,
  url,
  published_ts,
  updated_ts,
  provider_symbols_json,
  resolved_symbols_json,
  symbol_resolution
FROM (
  SELECT
    *,
    row_number() OVER (PARTITION BY provider, story_id ORDER BY updated_ts DESC, ingest_ts DESC, seq DESC) AS revision_rank
  FROM ${NEWS_TABLE}
)
WHERE revision_rank = 1
`;

export const fetchRecentNews = async (
  client: ClickHouseClient,
  limit: number
): Promise<NewsStory[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `${latestNewsSelect} ORDER BY published_ts DESC, story_id DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeNewsRow)
    .filter((record): record is NewsRecord => record !== null);
  return NewsStorySchema.array().parse(records.map(fromNewsRecord));
};

const normalizeAlertEvidenceRefs = (refs: string[]): string[] => {
  return Array.from(new Set(refs.map((ref) => ref.trim()).filter(Boolean)));
};

const flowPacketCandidatesFromRef = (ref: string): string[] => {
  if (!ref) {
    return [];
  }
  if (ref.startsWith("flowpacket:")) {
    const raw = ref.slice("flowpacket:".length);
    return raw ? [ref, raw] : [ref];
  }
  return [ref, `flowpacket:${ref}`];
};

const optionPrintCandidatesFromRef = (ref: string): string[] => {
  if (!ref || ref.startsWith("flowpacket:")) {
    return [];
  }
  return [ref];
};

export const fetchAlertContextByTraceId = async (
  client: ClickHouseClient,
  traceId: string
): Promise<AlertContextBundle> => {
  const normalizedTraceId = traceId.trim();
  if (!normalizedTraceId) {
    return {
      alert: null,
      flow_packets: [],
      option_prints: [],
      missing_refs: []
    };
  }

  const alertResult = await client.query({
    query: `SELECT * FROM ${ALERTS_TABLE} WHERE trace_id = ${quoteString(normalizedTraceId)} ORDER BY source_ts DESC, seq DESC LIMIT 1`,
    format: "JSONEachRow"
  });
  const alertRows = await alertResult.json<unknown[]>();
  const alertRecord = alertRows
    .map(normalizeAlertRow)
    .find((record): record is AlertRecord => record !== null);
  const alert = alertRecord ? AlertEventSchema.parse(fromAlertRecord(alertRecord)) : null;

  if (!alert) {
    return {
      alert: null,
      flow_packets: [],
      option_prints: [],
      missing_refs: []
    };
  }

  const refs = normalizeAlertEvidenceRefs(alert.evidence_refs);
  const packetLookupIds = Array.from(new Set(refs.flatMap(flowPacketCandidatesFromRef)));
  const printLookupIds = Array.from(new Set(refs.flatMap(optionPrintCandidatesFromRef)));

  const [flowPackets, optionPrints] = await Promise.all([
    packetLookupIds.length > 0
      ? client
          .query({
            query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE id IN (${buildStringList(packetLookupIds)}) ORDER BY source_ts DESC, seq DESC LIMIT ${clampLookupLimit(packetLookupIds.length)}`,
            format: "JSONEachRow"
          })
          .then(async (result) => {
            const rows = await result.json<unknown[]>();
            const records = rows
              .map(normalizeFlowPacketRow)
              .filter((record): record is FlowPacketRecord => record !== null);
            return FlowPacketSchema.array().parse(records.map(fromFlowPacketRecord));
          })
      : Promise.resolve([]),
    printLookupIds.length > 0
      ? fetchOptionPrintsByTraceIds(client, printLookupIds)
      : Promise.resolve([])
  ]);

  const packetIds = new Set(flowPackets.flatMap((packet) => [packet.id, packet.trace_id]));
  const printIds = new Set(optionPrints.map((print) => print.trace_id));
  const missingRefs = refs.filter((ref) => {
    const packetResolved = flowPacketCandidatesFromRef(ref).some((candidate) => packetIds.has(candidate));
    const printResolved = optionPrintCandidatesFromRef(ref).some((candidate) => printIds.has(candidate));
    return !packetResolved && !printResolved;
  });

  return {
    alert,
    flow_packets: flowPackets,
    option_prints: optionPrints,
    missing_refs: missingRefs
  };
};

export const fetchOptionPrintsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number,
  tracePrefix?: string,
  filters?: OptionPrintQueryFilters
): Promise<OptionPrint[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);
  const conditions = [
    `((ts, seq) > (${safeAfterTs}, ${safeAfterSeq}))`,
    ...buildOptionPrintFilterConditions(filters, tracePrefix)
  ];

  const result = await client.query({
    query: `SELECT * FROM ${OPTION_PRINTS_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
};

export const fetchOptionNBBOAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number,
  tracePrefix?: string
): Promise<OptionNBBO[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);
  const traceCondition = buildTracePrefixCondition(tracePrefix);
  const traceClause = traceCondition ? ` AND ${traceCondition}` : "";

  const result = await client.query({
    query: `SELECT * FROM ${OPTION_NBBO_TABLE} WHERE (ts, seq) > (${safeAfterTs}, ${safeAfterSeq})${traceClause} ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionNBBOSchema.array().parse(rows.map(normalizeOptionNbboRow));
};

export const fetchEquityPrintsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number,
  filters?: EquityPrintQueryFilters
): Promise<EquityPrint[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const conditions = [
    `((ts, seq) > (${safeAfterTs}, ${safeAfterSeq}))`,
    ...buildEquityPrintFilterConditions(filters)
  ];

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINTS_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
};

export const fetchEquityPrintsRange = async (
  client: ClickHouseClient,
  underlyingId: string,
  startTs: number,
  endTs: number,
  limit: number
): Promise<EquityPrint[]> => {
  const safeLimit = clampLimit(limit);
  const safeStart = clampCursor(startTs);
  const safeEnd = clampCursor(endTs);
  const rangeStart = Math.min(safeStart, safeEnd);
  const rangeEnd = Math.max(safeStart, safeEnd);
  const safeUnderlying = quoteString(underlyingId);

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINTS_TABLE} WHERE underlying_id = ${safeUnderlying} AND ts >= ${rangeStart} AND ts <= ${rangeEnd} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const parsed = EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
  return parsed.reverse();
};

export const fetchEquityQuotesAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<EquityQuote[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_QUOTES_TABLE} WHERE (ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityQuoteSchema.array().parse(rows.map(normalizeEquityQuoteRow));
};

export const fetchEquityCandlesAfter = async (
  client: ClickHouseClient,
  underlyingId: string,
  intervalMs: number,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<EquityCandle[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);
  const safeInterval = clampPositiveInt(intervalMs, 1000);
  const safeUnderlying = quoteString(underlyingId);

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_CANDLES_TABLE} WHERE underlying_id = ${safeUnderlying} AND interval_ms = ${safeInterval} AND (ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityCandleSchema.array().parse(rows.map(normalizeEquityCandleRow));
};

export const fetchEquityCandlesRange = async (
  client: ClickHouseClient,
  underlyingId: string,
  intervalMs: number,
  startTs: number,
  endTs: number,
  limit: number
): Promise<EquityCandle[]> => {
  const safeLimit = clampLimit(limit);
  const safeStart = clampCursor(startTs);
  const safeEnd = clampCursor(endTs);
  const safeInterval = clampPositiveInt(intervalMs, 1000);
  const safeUnderlying = quoteString(underlyingId);

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_CANDLES_TABLE} WHERE underlying_id = ${safeUnderlying} AND interval_ms = ${safeInterval} AND ts >= ${safeStart} AND ts <= ${safeEnd} ORDER BY ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityCandleSchema.array().parse(rows.map(normalizeEquityCandleRow));
};

export const fetchEquityPrintJoinsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<EquityPrintJoin[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeEquityPrintJoinRow)
    .filter((record): record is EquityPrintJoinRecord => record !== null);
  const joins = records.map(fromEquityPrintJoinRecord);
  return EquityPrintJoinSchema.array().parse(joins);
};

export const fetchInferredDarkAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<InferredDarkEvent[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${INFERRED_DARK_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeInferredDarkRow)
    .filter((record): record is InferredDarkRecord => record !== null);
  const events = records.map(fromInferredDarkRecord);
  return InferredDarkEventSchema.array().parse(events);
};

export const fetchFlowPacketsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<FlowPacket[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeFlowPacketRow)
    .filter((record): record is FlowPacketRecord => record !== null);
  const packets = records.map(fromFlowPacketRecord);
  return FlowPacketSchema.array().parse(packets);
};

export const fetchClassifierHitsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<ClassifierHitEvent[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${CLASSIFIER_HITS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeClassifierHitRow)
    .filter((record): record is ClassifierHitRecord => record !== null);
  const hits = records.map(fromClassifierHitRecord);
  return ClassifierHitEventSchema.array().parse(hits);
};

export const fetchSmartMoneyEventsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<SmartMoneyEvent[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${SMART_MONEY_EVENTS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeSmartMoneyEventRow)
    .filter((record): record is SmartMoneyEventRecord => record !== null);
  return SmartMoneyEventSchema.array().parse(records.map(fromSmartMoneyEventRecord));
};

export const fetchAlertsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<AlertEvent[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);

  const result = await client.query({
    query: `SELECT * FROM ${ALERTS_TABLE} WHERE (source_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY source_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeAlertRow)
    .filter((record): record is AlertRecord => record !== null);
  const alerts = records.map(fromAlertRecord);
  return AlertEventSchema.array().parse(alerts);
};

export const fetchNewsAfter = async (
  client: ClickHouseClient,
  afterTs: number,
  afterSeq: number,
  limit: number
): Promise<NewsStory[]> => {
  const safeLimit = clampLimit(limit);
  const safeAfterTs = clampCursor(afterTs);
  const safeAfterSeq = clampCursor(afterSeq);
  const result = await client.query({
    query: `${latestNewsSelect} AND (published_ts, seq) > (${safeAfterTs}, ${safeAfterSeq}) ORDER BY published_ts ASC, seq ASC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeNewsRow)
    .filter((record): record is NewsRecord => record !== null);
  return NewsStorySchema.array().parse(records.map(fromNewsRecord));
};

export const fetchOptionPrintsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number,
  tracePrefix?: string,
  filters?: OptionPrintQueryFilters
): Promise<OptionPrint[]> => {
  const safeLimit = clampLimit(limit);
  const conditions = [
    buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq),
    ...buildOptionPrintFilterConditions(filters, tracePrefix)
  ];

  const result = await client.query({
    query: `SELECT * FROM ${OPTION_PRINTS_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
};

export const fetchOptionNBBOBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number,
  tracePrefix?: string
): Promise<OptionNBBO[]> => {
  const safeLimit = clampLimit(limit);
  const conditions = [buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq)];
  const traceCondition = buildTracePrefixCondition(tracePrefix);
  if (traceCondition) {
    conditions.push(traceCondition);
  }

  const result = await client.query({
    query: `SELECT * FROM ${OPTION_NBBO_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionNBBOSchema.array().parse(rows.map(normalizeOptionNbboRow));
};

export const fetchEquityPrintsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number,
  filters?: EquityPrintQueryFilters
): Promise<EquityPrint[]> => {
  const safeLimit = clampLimit(limit);
  const conditions = [
    buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq),
    ...buildEquityPrintFilterConditions(filters)
  ];
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINTS_TABLE} WHERE ${conditions.join(" AND ")} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityPrintSchema.array().parse(rows.map(normalizeEquityRow));
};

export const fetchEquityQuotesBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<EquityQuote[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_QUOTES_TABLE} WHERE ${buildBeforeTupleCondition("ts", "seq", beforeTs, beforeSeq)} ORDER BY ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return EquityQuoteSchema.array().parse(rows.map(normalizeEquityQuoteRow));
};

export const fetchEquityPrintJoinsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<EquityPrintJoin[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeEquityPrintJoinRow)
    .filter((record): record is EquityPrintJoinRecord => record !== null);
  return EquityPrintJoinSchema.array().parse(records.map(fromEquityPrintJoinRecord));
};

export const fetchFlowPacketsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<FlowPacket[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeFlowPacketRow)
    .filter((record): record is FlowPacketRecord => record !== null);
  return FlowPacketSchema.array().parse(records.map(fromFlowPacketRecord));
};

export const fetchClassifierHitsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<ClassifierHitEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${CLASSIFIER_HITS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeClassifierHitRow)
    .filter((record): record is ClassifierHitRecord => record !== null);
  return ClassifierHitEventSchema.array().parse(records.map(fromClassifierHitRecord));
};

export const fetchSmartMoneyEventsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<SmartMoneyEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${SMART_MONEY_EVENTS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeSmartMoneyEventRow)
    .filter((record): record is SmartMoneyEventRecord => record !== null);
  return SmartMoneyEventSchema.array().parse(records.map(fromSmartMoneyEventRecord));
};

export const fetchAlertsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<AlertEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${ALERTS_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeAlertRow)
    .filter((record): record is AlertRecord => record !== null);
  return AlertEventSchema.array().parse(records.map(fromAlertRecord));
};

export const fetchNewsBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<NewsStory[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `${latestNewsSelect} AND ${buildBeforeTupleCondition("published_ts", "seq", beforeTs, beforeSeq)} ORDER BY published_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeNewsRow)
    .filter((record): record is NewsRecord => record !== null);
  return NewsStorySchema.array().parse(records.map(fromNewsRecord));
};

export const fetchInferredDarkBefore = async (
  client: ClickHouseClient,
  beforeTs: number,
  beforeSeq: number,
  limit: number
): Promise<InferredDarkEvent[]> => {
  const safeLimit = clampLimit(limit);
  const result = await client.query({
    query: `SELECT * FROM ${INFERRED_DARK_TABLE} WHERE ${buildBeforeTupleCondition("source_ts", "seq", beforeTs, beforeSeq)} ORDER BY source_ts DESC, seq DESC LIMIT ${safeLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeInferredDarkRow)
    .filter((record): record is InferredDarkRecord => record !== null);
  return InferredDarkEventSchema.array().parse(records.map(fromInferredDarkRecord));
};

export const fetchFlowPacketById = async (
  client: ClickHouseClient,
  id: string
): Promise<FlowPacket | null> => {
  const result = await client.query({
    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE id = ${quoteString(id)} ORDER BY source_ts DESC, seq DESC LIMIT 1`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const record = rows
    .map(normalizeFlowPacketRow)
    .find((row): row is FlowPacketRecord => row !== null);
  return record ? FlowPacketSchema.parse(fromFlowPacketRecord(record)) : null;
};

export const fetchFlowPacketsByIds = async (
  client: ClickHouseClient,
  ids: string[]
): Promise<FlowPacket[]> => {
  const uniqueIds = Array.from(new Set(ids.map((id) => id.trim()).filter(Boolean)));
  if (uniqueIds.length === 0) {
    return [];
  }
  const result = await client.query({
    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE id IN (${buildStringList(uniqueIds)}) ORDER BY source_ts DESC, seq DESC LIMIT ${clampLookupLimit(uniqueIds.length)}`,
    format: "JSONEachRow"
  });
  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeFlowPacketRow)
    .filter((record): record is FlowPacketRecord => record !== null);
  return FlowPacketSchema.array().parse(records.map(fromFlowPacketRecord));
};

export const fetchFlowPacketsByMemberTraceIds = async (
  client: ClickHouseClient,
  traceIds: string[]
): Promise<FlowPacket[]> => {
  const ids = Array.from(new Set(traceIds.map((id) => id.trim()).filter(Boolean)));
  if (ids.length === 0) {
    return [];
  }

  const memberPredicates = ids.map((id) => `has(members, ${quoteString(id)})`);
  const result = await client.query({
    query: `SELECT * FROM ${FLOW_PACKETS_TABLE} WHERE ${memberPredicates.join(" OR ")} ORDER BY source_ts DESC, seq DESC LIMIT ${clampLookupLimit(ids.length * 4)}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeFlowPacketRow)
    .filter((record): record is FlowPacketRecord => record !== null);
  return FlowPacketSchema.array().parse(records.map(fromFlowPacketRecord));
};

export const fetchSmartMoneyEventsByPacketIds = async (
  client: ClickHouseClient,
  packetIds: string[]
): Promise<SmartMoneyEvent[]> => {
  const ids = Array.from(new Set(packetIds.map((id) => id.trim()).filter(Boolean)));
  if (ids.length === 0) {
    return [];
  }

  const packetPredicates = ids.map((id) => `has(packet_ids, ${quoteString(id)})`);
  const result = await client.query({
    query: `SELECT * FROM ${SMART_MONEY_EVENTS_TABLE} WHERE ${packetPredicates.join(" OR ")} ORDER BY source_ts DESC, seq DESC LIMIT ${clampLookupLimit(ids.length * 4)}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeSmartMoneyEventRow)
    .filter((record): record is SmartMoneyEventRecord => record !== null);
  return SmartMoneyEventSchema.array().parse(records.map(fromSmartMoneyEventRecord));
};

export const fetchClassifierHitsByPacketIds = async (
  client: ClickHouseClient,
  packetIds: string[]
): Promise<ClassifierHitEvent[]> => {
  const ids = Array.from(new Set(packetIds.map((id) => id.trim()).filter(Boolean)));
  if (ids.length === 0) {
    return [];
  }

  const tracePredicates = ids.map((id) => `position(trace_id, ${quoteString(id)}) > 0`);
  const result = await client.query({
    query: `SELECT * FROM ${CLASSIFIER_HITS_TABLE} WHERE ${tracePredicates.join(" OR ")} ORDER BY source_ts DESC, seq DESC LIMIT ${clampLookupLimit(ids.length * 4)}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeClassifierHitRow)
    .filter((record): record is ClassifierHitRecord => record !== null);
  return ClassifierHitEventSchema.array().parse(records.map(fromClassifierHitRecord));
};

export const fetchNearestOptionNBBOForPrints = async (
  client: ClickHouseClient,
  inputs: Array<{ trace_id: string; option_contract_id: string; ts: number }>
): Promise<Record<string, OptionNBBO | null>> => {
  const normalized = inputs
    .map((item) => ({
      trace_id: item.trace_id.trim(),
      option_contract_id: item.option_contract_id.trim(),
      ts: clampCursor(item.ts)
    }))
    .filter((item) => item.trace_id && item.option_contract_id);
  if (normalized.length === 0) {
    return {};
  }

  const byTraceId: Record<string, OptionNBBO | null> = Object.fromEntries(
    normalized.map((item) => [item.trace_id, null])
  );
  await Promise.all(
    normalized.map(async (item) => {
      const result = await client.query({
        query: `SELECT * FROM ${OPTION_NBBO_TABLE} WHERE option_contract_id = ${quoteString(item.option_contract_id)} AND ts <= ${item.ts} ORDER BY ts DESC, seq DESC LIMIT 1`,
        format: "JSONEachRow"
      });
      const rows = await result.json<unknown[]>();
      const quote = OptionNBBOSchema.array().parse(rows.map(normalizeOptionNbboRow))[0] ?? null;
      byTraceId[item.trace_id] = quote;
    })
  );
  return byTraceId;
};

export const fetchOptionPrintsByTraceIds = async (
  client: ClickHouseClient,
  traceIds: string[]
): Promise<OptionPrint[]> => {
  const ids = Array.from(new Set(traceIds.map((id) => id.trim()).filter(Boolean)));
  if (ids.length === 0) {
    return [];
  }

  const result = await client.query({
    query: `SELECT * FROM ${OPTION_PRINTS_TABLE} WHERE trace_id IN (${buildStringList(ids)}) ORDER BY ts DESC, seq DESC LIMIT ${clampLookupLimit(ids.length)}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  return OptionPrintSchema.array().parse(rows.map(normalizeOptionRow));
};

export const fetchEquityPrintJoinsByIds = async (
  client: ClickHouseClient,
  ids: string[]
): Promise<EquityPrintJoin[]> => {
  const uniqueIds = Array.from(new Set(ids.map((id) => id.trim()).filter(Boolean)));
  if (uniqueIds.length === 0) {
    return [];
  }

  const joinIds = new Set<string>();
  const printTraceIds = new Set<string>();
  for (const id of uniqueIds) {
    joinIds.add(id);
    if (id.startsWith("equityjoin:")) {
      const trace = id.slice("equityjoin:".length);
      if (trace) {
        printTraceIds.add(trace);
      }
    } else {
      joinIds.add(`equityjoin:${id}`);
      printTraceIds.add(id);
    }
  }

  const joinIdList = Array.from(joinIds);
  const printTraceList = Array.from(printTraceIds);
  const whereParts = [
    `id IN (${buildStringList(joinIdList)})`,
    `trace_id IN (${buildStringList(joinIdList)})`
  ];
  if (printTraceList.length > 0) {
    whereParts.push(`print_trace_id IN (${buildStringList(printTraceList)})`);
  }
  const lookupLimit = clampLookupLimit(joinIdList.length + printTraceList.length);

  const result = await client.query({
    query: `SELECT * FROM ${EQUITY_PRINT_JOINS_TABLE} WHERE ${whereParts.join(" OR ")} ORDER BY source_ts DESC, seq DESC LIMIT ${lookupLimit}`,
    format: "JSONEachRow"
  });

  const rows = await result.json<unknown[]>();
  const records = rows
    .map(normalizeEquityPrintJoinRow)
    .filter((record): record is EquityPrintJoinRecord => record !== null);
  return EquityPrintJoinSchema.array().parse(records.map(fromEquityPrintJoinRecord));
};