#!/usr/bin/env bun
/** Read-only Botmaster API shim — queries remote D1 over the Cloudflare REST API for local collector polling. */
import { join } from "node:path";
import { readFileSync } from "node:fs";

const PORT = Number(process.env.BOTMASTER_PROXY_PORT ?? 31340);
const REFRESH_MS = Number(process.env.BOTMASTER_PROXY_REFRESH_MS ?? 30_000);
const BOTMASTER_DIR = process.env.BOTMASTER_DIR ?? join(process.env.HOME!, "Projects", "Botmaster", "bot-template");
const D1_DATABASE_NAME = process.env.BOTMASTER_D1_DATABASE_NAME ?? "botmaster";
const CF_API_BASE = process.env.CLOUDFLARE_API_BASE ?? "https://api.cloudflare.com/client/v4";

type MetricRow = Record<string, unknown>;
type LogRow = Record<string, unknown>;
type RegistryRow = {
  id: string;
  bot_name: string;
  project_name: string;
  status: string;
};

type BotMetricsAgg = {
  messages24h: number;
  errors24h: number;
  cost24h: number;
  messagesTotal: number;
  errorsTotal: number;
  costTotal: number;
  lastTs: number | null;
  chatMessages24h: number;
  chatMessagesTotal: number;
  lastChatTs: number | null;
  /** Max(metrics.ts, chat_messages.date) — Telegram/LLM mirror only. */
  lastMirrorTs: number | null;
  /** Latest github/trello poll timestamp from D1. */
  lastPollTs: number | null;
};

export type BotSummaryRow = RegistryRow & BotMetricsAgg;

const MS_24H = 24 * 60 * 60 * 1000;

let metricsCache: MetricRow[] = [];
let registryCache: RegistryRow[] = [];
let summaryCache: BotSummaryRow[] = [];
let metricsReady = false;
let metricsError: string | null = null;
const logsCache = new Map<string, LogRow[]>();

class D1Error extends Error {
  readonly permanent: boolean;
  constructor(message: string, permanent: boolean) {
    super(message);
    this.permanent = permanent;
  }
}

/** Credentials and routing are wrong, not flaky — retrying can never fix them. */
const PERMANENT_HTTP_STATUS = new Set([401, 403, 404]);

let cachedQueryUrl: string | null = null;

function d1DatabaseId(): string {
  const override = process.env.BOTMASTER_D1_DATABASE_ID;
  if (override) return override;
  const configPath = join(BOTMASTER_DIR, "wrangler.toml");
  let text: string;
  try {
    text = readFileSync(configPath, "utf8");
  } catch (error) {
    throw new D1Error(
      `cannot read ${configPath} for the ${D1_DATABASE_NAME} database id: ${(error as Error).message}`,
      true,
    );
  }
  for (const section of text.split(/^[ \t]*\[\[d1_databases\]\][ \t]*$/m).slice(1)) {
    const body = section.split(/^[ \t]*\[/m)[0]!;
    const name = body.match(/^[ \t]*database_name[ \t]*=[ \t]*"([^"]+)"/m)?.[1];
    const id = body.match(/^[ \t]*database_id[ \t]*=[ \t]*"([^"]+)"/m)?.[1];
    if (name === D1_DATABASE_NAME && id) return id;
  }
  throw new D1Error(`${configPath} declares no d1_databases entry named ${D1_DATABASE_NAME}`, true);
}

function d1QueryUrl(): string {
  if (cachedQueryUrl) return cachedQueryUrl;
  const account = process.env.CLOUDFLARE_ACCOUNT_ID;
  if (!account) throw new D1Error("CLOUDFLARE_ACCOUNT_ID is not set — the D1 REST API needs an account id", true);
  cachedQueryUrl = `${CF_API_BASE}/accounts/${account}/d1/database/${d1DatabaseId()}/query`;
  return cachedQueryUrl;
}

type D1Response<T> = {
  success?: boolean;
  errors?: Array<{ code?: number; message?: string }>;
  result?: Array<{ results?: T[] }>;
};

async function d1Query<T>(sql: string, params: unknown[] = []): Promise<T[]> {
  const url = d1QueryUrl();
  const token = process.env.CLOUDFLARE_API_TOKEN;
  if (!token) throw new D1Error("CLOUDFLARE_API_TOKEN is not set — the D1 REST API cannot authenticate", true);

  let res: Response;
  try {
    res = await fetch(url, {
      method: "POST",
      headers: { authorization: `Bearer ${token}`, "content-type": "application/json" },
      body: JSON.stringify({ sql, params }),
    });
  } catch (error) {
    throw new D1Error(`D1 request failed to reach ${CF_API_BASE}: ${(error as Error).message}`, false);
  }

  const text = await res.text();
  if (!res.ok) {
    throw new D1Error(`D1 query HTTP ${res.status}: ${text.slice(0, 500)}`, PERMANENT_HTTP_STATUS.has(res.status));
  }
  let body: D1Response<T>;
  try {
    body = JSON.parse(text) as D1Response<T>;
  } catch {
    throw new D1Error(`D1 returned non-JSON: ${text.slice(0, 200)}`, false);
  }
  if (body.success !== true) {
    const detail = (body.errors ?? []).map((e) => `${e.code ?? "?"}: ${e.message ?? "?"}`).join("; ");
    throw new D1Error(`D1 query rejected: ${detail || text.slice(0, 500)}`, false);
  }
  return body.result?.[0]?.results ?? [];
}

function num(value: unknown): number {
  const n = Number(value ?? 0);
  return Number.isFinite(n) ? n : 0;
}

function nullableTs(value: unknown): number | null {
  if (value === null || value === undefined) return null;
  const n = Number(value);
  return Number.isFinite(n) && n > 0 ? n : null;
}

async function refreshSummaries(now = Date.now()): Promise<BotSummaryRow[]> {
  const since24hMs = now - MS_24H;
  const since24hSec = Math.floor(since24hMs / 1000);
  const rows = await d1Query<Record<string, unknown>>(`
    SELECT
      b.id AS id,
      b.bot_name AS bot_name,
      b.project_name AS project_name,
      b.status AS status,
      COALESCE(m.messages24h, 0) + COALESCE(c.messages24h, 0) AS messages24h,
      COALESCE(m.errors24h, 0) AS errors24h,
      COALESCE(m.cost24h, 0) AS cost24h,
      COALESCE(m.messages_total, 0) + COALESCE(c.messages_total, 0) AS messagesTotal,
      COALESCE(m.errors_total, 0) AS errorsTotal,
      COALESCE(m.cost_total, 0) AS costTotal,
      CASE
        WHEN COALESCE(m.last_ts, 0) >= COALESCE(c.last_ts_ms, 0) THEN m.last_ts
        ELSE c.last_ts_ms
      END AS lastMirrorTs,
      p.last_poll_ts AS lastPollTs,
      CASE
        WHEN COALESCE(m.last_ts, 0) >= COALESCE(c.last_ts_ms, 0)
          AND COALESCE(m.last_ts, 0) >= COALESCE(p.last_poll_ts, 0) THEN m.last_ts
        WHEN COALESCE(c.last_ts_ms, 0) >= COALESCE(p.last_poll_ts, 0) THEN c.last_ts_ms
        ELSE p.last_poll_ts
      END AS lastTs,
      COALESCE(c.messages24h, 0) AS chatMessages24h,
      COALESCE(c.messages_total, 0) AS chatMessagesTotal,
      c.last_ts_ms AS lastChatTs
    FROM bots b
    LEFT JOIN (
      SELECT
        bot_id,
        SUM(CASE WHEN ts >= ${since24hMs} THEN 1 ELSE 0 END) AS messages24h,
        SUM(CASE WHEN ts >= ${since24hMs} AND error IS NOT NULL AND error != '' THEN 1 ELSE 0 END) AS errors24h,
        SUM(CASE WHEN ts >= ${since24hMs} THEN COALESCE(cost_usd, 0) ELSE 0 END) AS cost24h,
        COUNT(*) AS messages_total,
        SUM(CASE WHEN error IS NOT NULL AND error != '' THEN 1 ELSE 0 END) AS errors_total,
        SUM(COALESCE(cost_usd, 0)) AS cost_total,
        MAX(ts) AS last_ts
      FROM metrics
      GROUP BY bot_id
    ) m ON m.bot_id = b.id
    LEFT JOIN (
      SELECT
        bot_id,
        SUM(CASE WHEN date >= ${since24hSec} THEN 1 ELSE 0 END) AS messages24h,
        COUNT(*) AS messages_total,
        MAX(date * 1000) AS last_ts_ms
      FROM chat_messages
      GROUP BY bot_id
    ) c ON c.bot_id = b.id
    LEFT JOIN (
      SELECT bot_id, MAX(last_polled_at) AS last_poll_ts
      FROM bot_github_repos
      GROUP BY bot_id
    ) p ON p.bot_id = b.id
    ORDER BY b.bot_name
  `);

  return rows.map((row) => ({
    id: String(row.id),
    bot_name: String(row.bot_name),
    project_name: String(row.project_name),
    status: String(row.status),
    messages24h: num(row.messages24h),
    errors24h: num(row.errors24h),
    cost24h: num(row.cost24h),
    messagesTotal: num(row.messagesTotal),
    errorsTotal: num(row.errors_total),
    costTotal: num(row.cost_total),
    lastTs: nullableTs(row.lastTs),
    chatMessages24h: num(row.chatMessages24h),
    chatMessagesTotal: num(row.chatMessagesTotal),
    lastChatTs: nullableTs(row.lastChatTs),
    lastMirrorTs: nullableTs(row.lastMirrorTs),
    lastPollTs: nullableTs(row.lastPollTs),
  }));
}

async function refreshMetrics(): Promise<void> {
  try {
    const metrics = await d1Query<MetricRow>(
      "SELECT id, bot_id, model, call_type, error, tokens_in, tokens_out, cost_usd, ts, created_at FROM metrics ORDER BY ts DESC LIMIT 5000",
    );
    const [registry, summaries] = await Promise.all([
      d1Query<RegistryRow>(
        "SELECT id, bot_name, project_name, status FROM bots ORDER BY bot_name",
      ),
      refreshSummaries(),
    ]);
    metricsCache = metrics;
    registryCache = registry;
    summaryCache = summaries;
    metricsReady = true;
    metricsError = null;
  } catch (error) {
    metricsError = (error as Error).message;
    throw error;
  }
}

async function refreshLogs(botId: string, since: number): Promise<LogRow[]> {
  const rows = await d1Query<LogRow>(
    "SELECT id, ts, level, message, data FROM bot_logs WHERE bot_id = ? AND ts > ? ORDER BY ts ASC LIMIT 500",
    [botId, since],
  );
  logsCache.set(`${botId}:${since}`, rows);
  return rows;
}

function json(body: unknown, status = 200): Response {
  return new Response(JSON.stringify(body), {
    status,
    headers: { "content-type": "application/json; charset=utf-8" },
  });
}

/** sysexits.h EX_CONFIG; matches RestartPreventExitStatus in botmaster-proxy.service. */
const EXIT_CONFIG_ERROR = 78;

function exitOnPermanentError(error: D1Error): never {
  console.error(`[botmaster-proxy] permanent configuration error, not retrying: ${error.message}`);
  process.exit(EXIT_CONFIG_ERROR);
}

const BACKOFF_CEILING_MS = 30 * 60_000;

function exponentialBackoffMs(failureCount: number): number {
  return Math.min(REFRESH_MS * 2 ** (failureCount - 1), BACKOFF_CEILING_MS);
}

async function loopRefresh(): Promise<void> {
  let consecutiveFailures = 0;
  for (;;) {
    try {
      await refreshMetrics();
      consecutiveFailures = 0;
      await Bun.sleep(REFRESH_MS);
    } catch (error) {
      if (error instanceof D1Error && error.permanent) exitOnPermanentError(error);
      consecutiveFailures += 1;
      const backoff = exponentialBackoffMs(consecutiveFailures);
      console.error(
        `[botmaster-proxy] metrics refresh failed (failure ${consecutiveFailures}, retrying in ${backoff}ms): ${(error as Error).message}`,
      );
      await Bun.sleep(backoff);
    }
  }
}

/** Block startup until the first D1 snapshot succeeds so collector's first poll does not 503. */
async function warmCache(maxAttempts = 12): Promise<void> {
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      await refreshMetrics();
      return;
    } catch (error) {
      if (error instanceof D1Error && error.permanent) exitOnPermanentError(error);
      console.error(
        `[botmaster-proxy] initial refresh ${attempt}/${maxAttempts} failed: ${(error as Error).message}`,
      );
      if (attempt === maxAttempts) {
        console.error("[botmaster-proxy] starting without cache — endpoints return 503 until refresh succeeds");
        return;
      }
      await Bun.sleep(Math.min(5_000 * 2 ** (attempt - 1), 30_000));
    }
  }
}

async function main(): Promise<void> {
  await warmCache();
  void loopRefresh();

  Bun.serve({
    hostname: "127.0.0.1",
    port: PORT,
    idleTimeout: 120,
    async fetch(req) {
      const url = new URL(req.url);
      if (req.method !== "GET") return new Response("method-not-allowed", { status: 405 });

      if (url.pathname === "/api/metrics") {
        if (!metricsReady) {
          return json({ error: metricsError ?? "metrics cache warming" }, 503);
        }
        const limit = Math.min(Number(url.searchParams.get("limit") ?? 100), 5000);
        return json(metricsCache.slice(0, limit));
      }

      if (url.pathname === "/api/bots") {
        if (!metricsReady) {
          return json({ error: metricsError ?? "metrics cache warming" }, 503);
        }
        return json(registryCache);
      }

      if (url.pathname === "/api/bots/summary") {
        if (!metricsReady) {
          return json({ error: metricsError ?? "metrics cache warming" }, 503);
        }
        return json(summaryCache);
      }

      const logsMatch = url.pathname.match(/^\/api\/bots\/([^/]+)\/logs$/);
      if (logsMatch) {
        if (!metricsReady) {
          return json({ error: metricsError ?? "metrics cache warming" }, 503);
        }
        const botId = decodeURIComponent(logsMatch[1]!);
        const since = Number(url.searchParams.get("since") ?? "0");
        if (!Number.isFinite(since) || since < 0) return json({ error: "since must be a non-negative number" }, 400);
        const cacheKey = `${botId}:${since}`;
        if (!logsCache.has(cacheKey)) {
          try {
            await refreshLogs(botId, since);
          } catch (error) {
            return json({ error: (error as Error).message }, 503);
          }
        }
        const limit = Math.min(Number(url.searchParams.get("limit") ?? 100), 500);
        return json((logsCache.get(cacheKey) ?? []).slice(0, limit));
      }

      return new Response("not-found", { status: 404 });
    },
  });

  console.error(`botmaster-proxy listening on http://127.0.0.1:${PORT}`);
}

void main();