#!/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, mkdirSync, writeFileSync } from "node:fs";
import { join as pathJoin } from "node:path";
import { spawn, spawnSync } from "node:child_process";
import { createWaker, type LivePane } from "../modules/botmaster/notify/waker.js";
import { openStore, type Store, type Message } from "../modules/botmaster/notify/store.js";
import { resolveParent, inboundText } from "../modules/botmaster/notify/inbox.js";
import { decryptField } from "../modules/botmaster/notify/crypto.js";
import { resolveChannel, type BotRow } from "../modules/botmaster/notify/resolve.js";
import { fetchInboundFile } from "../modules/botmaster/notify/files.js";

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";
const FULL_GIT_SHA = /^[0-9a-f]{40}$/;

export function resolveDeploymentSha(value: string | undefined): string {
  if (!value || !FULL_GIT_SHA.test(value)) {
    throw new Error("OVERDECK_DEPLOY_SHA must be a full lowercase 40-character git SHA");
  }
  return value;
}

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;

export function parseLivePanes(output: string): LivePane[] {
  return output.split("\n").flatMap((line) => {
    const match = /^(%\d+)_(.+)_(\d+)$/.exec(line);
    const panePid = Number(match?.[3]);
    if (!match || !Number.isSafeInteger(panePid) || panePid < 1) return [];
    return [{ pane: match[1]!, command: match[2]!, panePid }];
  });
}

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> {
  const deploymentSha = resolveDeploymentSha(process.env.OVERDECK_DEPLOY_SHA);
  await warmCache();
  void loopRefresh();

  // Independent from the metrics loop: steering may retry without changing the
  // cache that collector reads.
  const ownerValue = Number(process.env.OVERDECK_BOTMASTER_OWNER_USER_ID);
  const ownerUserId = Number.isSafeInteger(ownerValue) && ownerValue > 0 ? ownerValue : null;
  const store = openStore();
  // Steering must not be able to fail quietly, but a missing owner id must also not
  // take the metrics feed down with it: refuse steering loudly, keep serving metrics.
  if (!ownerUserId) {
    console.error("[botmaster-proxy] FATAL for steering: OVERDECK_BOTMASTER_OWNER_USER_ID is not set — every owner reply would be dropped silently. Metrics continue; steering is OFF until it is set.");
  }
  const poll = ownerUserId === null ? null : createInboxPoller({
    ownerUserId,
    store,
    query: async (channel, since) => d1Query<InboxMirrorRow>(
      `SELECT b.project_name AS channel, c.chat_id, c.message_id, c.date, c.text, c.user_id, c.file_id, c.file_name, c.file_size, c.mime_type
       FROM chat_messages c JOIN bots b ON b.id = c.bot_id
       WHERE lower(b.project_name) = lower(?) AND c.date > ? ORDER BY c.date, c.message_id`,
      [channel, since],
    ),
    tokenForChannel: async (channel) => { const rows=await d1Query<BotRow>("SELECT id, project_name, bot_name, telegram_token, allowed_chat_ids FROM bots"); const key=process.env.DB_ENC_KEY;if(!key)throw new Error('DB_ENC_KEY is not set');const token=await decryptField(key,resolveChannel(rows,channel).token);if(!token)throw new Error('decrypted telegram token is empty');return token; },
    notifyOwner: async (text) => { await new Promise<void>((resolve) => {
      const child = spawn("botmaster", ["--fyi", text], { stdio: "ignore" });
      child.on("exit", () => resolve());
      child.on("error", () => resolve());
    }); },
    ack: async (_channel, _chatId, id) => { await new Promise<void>((resolve) => {
      const child = spawn("botmaster", ["--raw", "--fyi", `Message #${id} received`], { stdio: "ignore" });
      child.on("exit", () => resolve());
      child.on("error", () => resolve());
    }); },
  });
  // A message the hook cannot deliver (the session is idle, so it runs no tools)
  // gets one poke on the same interval. Every guard lives in createWaker.
  const wake = createWaker({
    store,
    listPanes: (socket) => {
      const r = spawnSync("tmux", ["-S", socket, "list-panes", "-a", "-F", "#{pane_id}_#{pane_current_command}_#{pane_pid}"], { encoding: "utf8" });
      // Returning [] on a tmux failure made every failure read as "the pane is
      // gone", which is the one diagnosis that hides a broken wake path.
      if (r.error) throw new Error(`tmux could not run: ${r.error.message}`);
      if (r.status !== 0) throw new Error(`tmux -S ${socket} list-panes exited ${r.status}: ${String(r.stderr ?? "").trim()}`);
      if (!String(r.stdout ?? "").trim()) throw new Error(`tmux -S ${socket} listed no panes at all`);
      return parseLivePanes(String(r.stdout ?? ""));
    },
    paneRunsClaude: (panePid) => {
      const r = spawnSync("ps", ["-e", "-o", "pid=,ppid=,comm="], { encoding: "utf8" });
      if (r.status !== 0) throw new Error("ps failed");
      const kids = new Map<number, number[]>();
      const comm = new Map<number, string>();
      for (const line of String(r.stdout ?? "").split("\n")) {
        const m = /^\s*(\d+)\s+(\d+)\s+(.*\S)\s*$/.exec(line);
        if (!m) continue;
        const pid = Number(m[1]), ppid = Number(m[2]);
        comm.set(pid, m[3]!);
        if (!kids.has(ppid)) kids.set(ppid, []);
        kids.get(ppid)!.push(pid);
      }
      // The pane process itself counts: an unsandboxed session runs claude directly
      // as the pane process, with no child to find.
      const queue = [panePid];
      const seen = new Set<number>();
      while (queue.length) {
        const pid = queue.shift()!;
        if (seen.has(pid)) continue;
        seen.add(pid);
        if (/^claude/i.test(comm.get(pid) ?? "")) return true;
        queue.push(...(kids.get(pid) ?? []));
      }
      return false;
    },
    sendKeys: (socket, pane, text) => {
      // -l keeps the text literal; Enter is a separate call so the text can never
      // submit itself early through an embedded key sequence.
      const typed = spawnSync("tmux", ["-S", socket, "send-keys", "-t", pane, "-l", text], { encoding: "utf8" });
      if (typed.status !== 0) throw new Error(`tmux send-keys failed: ${typed.stderr ?? ""}`);
      spawnSync("tmux", ["-S", socket, "send-keys", "-t", pane, "Enter"], { encoding: "utf8" });
    },
    log: (s) => console.error(s),
    reportUndeliverable: (id, sessionId) => { void new Promise<void>((resolve) => {
      const child = spawn("botmaster", ["--fyi", `Message #${id} could not be delivered to session ${sessionId} after 2 wake attempts — it likely ended. Reply is dropped; resend if it still matters.`], { stdio: "ignore" });
      child.on("exit", () => resolve());
      child.on("error", () => resolve());
    }); },
  });
  if (poll) startInboxInterval(async () => { await poll(process.env.BOTMASTER_DEFAULT_CHANNEL ?? "overdeck"); wake() });

  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 === "/health") {
        return json({ ok: true, deployedSha: deploymentSha });
      }

      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}`);
}

// Inbox polling is deliberately separate from metrics: failures here never affect the HTTP cache.
export type InboxMirrorRow = { channel:string; chat_id:number; message_id:number; date:number; text:string; user_id:number; reply_to_message_id?:number|null; file_id?:string|null; file_name?:string|null; file_size?:number|null; mime_type?:string|null };
export function startInboxInterval(run:()=>Promise<void>, intervalMs=10_000){ let running=false; const tick=async()=>{if(running)return;running=true;try{await run()}catch(error){console.error(`[botmaster-proxy] inbox poll failed: ${(error as Error).message}`)}finally{running=false}}; const timer=setInterval(()=>void tick(),intervalMs); return { stop:()=>clearInterval(timer), tick }; }
// A main claim older than this is treated as abandoned (orchestrator died without
// a fresh SessionStart) rather than routed to blindly.
const MAIN_CLAIM_STALE_MS = 6 * 60 * 60 * 1000;
export function createInboxPoller(deps:{query:(channel:string,since:number)=>Promise<InboxMirrorRow[]>; store:Store; ownerUserId:number|null; writeMarker?:(sessionId:string,id:string,body:unknown)=>void; correct?:(channel:string,chatId:number,id:string)=>Promise<void>; ack?:(channel:string,chatId:number,id:string)=>Promise<void>; notifyOwner?:(text:string)=>Promise<void>; tokenForChannel?:(channel:string)=>Promise<string>; fetchInboundFile?:(token:string,fileId:string,fileName:string,messageId:string)=>Promise<string>; now?:()=>number; log?:(s:string)=>void}) {
  const now=deps.now??Date.now, log=deps.log??console.error, corrections=new Map<string,number>();
  // A missing owner id used to log once and then silently drop every reply the owner
  // sent — the owner cannot tell steering from silence, so this must fail loudly.
  if (!deps.ownerUserId) throw new Error('[botmaster-proxy] refusing to start inbound steering: OVERDECK_BOTMASTER_OWNER_USER_ID is not set, so every owner reply would be dropped silently');
  // The owner cannot tell "not received" from "received, still working", so every
  // registered message is acknowledged from here rather than by the session.
  const register=async(channel:string,m:Message)=>{deps.store.insertMessage(m);markerFn(m.sessionId,m.id,m);if(deps.ack){try{await deps.ack(channel,m.chatId,m.id)}catch(error){log(`[botmaster-proxy] ack for ${m.id} failed: ${(error as Error).message}`)}}};
  const markerFn=deps.writeMarker??((session,id,body)=>{const dir=pathJoin(process.env.HOME??'', '.local/state/overdeck/botmaster/inbox',session);mkdirSync(dir,{recursive:true,mode:0o700});writeFileSync(pathJoin(dir,`${id}.json`),JSON.stringify(body),{mode:0o600})});
  const withAttachment=async(channel:string,x:InboxMirrorRow,text:string)=>{if(!x.file_id)return text;const name=x.file_name??'attachment.bin';try{const token=await deps.tokenForChannel?.(channel);if(!token)throw new Error('attachment token unavailable');const path=await (deps.fetchInboundFile??fetchInboundFile)(token,x.file_id,name,String(x.message_id));return `${text}\n[attachment saved: ${path}]`}catch{log(`[botmaster-proxy] attachment fetch failed for message ${x.message_id}`);return `${text}\n[attachment could not be fetched: ${name}]`}};
  return async function poll(channel:string) { if(!deps.ownerUserId)return; const cursor=deps.store.cursor(channel); const rows=await deps.query(channel,cursor); for(const x of rows.sort((a,b)=>a.date-b.date||a.message_id-b.message_id)){ if(x.user_id!==deps.ownerUserId)continue; if(deps.store.getByTgMessageId(x.chat_id,x.message_id)){deps.store.setCursor(channel,x.date);continue} const parent=resolveParent(deps.store,x.chat_id,x.reply_to_message_id??null,x.text); const hashtag=/^\s*#([0-9a-hj-km-np-tv-z]{5})\s+/i.exec(x.text); if(!parent){
      if(hashtag&&deps.correct){const key=`${x.chat_id}:${hashtag[1].toLowerCase()}`, last=corrections.get(key)??0, chatLast=corrections.get(`chat:${x.chat_id}`)??0;if(now()-last>=3600000&&now()-chatLast>=60000){await deps.correct(channel,x.chat_id,hashtag[1]);corrections.set(key,now());corrections.set(`chat:${x.chat_id}`,now())}}
      else if(!hashtag){
        // Plain text, no reply, no #id: route to whichever top-level session most
        // recently claimed "main" — that is how the owner mid-turn-talks to the
        // coordinator instead of a specific worker session.
        const claim=deps.store.getMainClaim();
        if(claim&&now()-claim.claimedAt<=MAIN_CLAIM_STALE_MS){
          const id=deps.store.mintId();
          const m:Message={id,direction:'in',sessionId:claim.sessionId,ticketId:null,channel,chatId:x.chat_id,priority:'needs-answer',parentId:null,text:await withAttachment(channel,x,x.text),createdAt:x.date*1000,attempts:0,deliveredAt:null,escalatedAt:null,tgMessageId:x.message_id};
          await register(channel,m);
        } else if(deps.notifyOwner){
          const detail=claim?`last claim by ${claim.sessionId.slice(0,8)} went quiet ${Math.floor((now()-claim.claimedAt)/3600000)}h ago`:'no session has claimed it yet';
          await deps.notifyOwner(`No active main session — ${detail}.`);
        }
      }
      deps.store.setCursor(channel,x.date);continue} if(parent.chatId!==x.chat_id)continue; const id=deps.store.mintId(); const m:Message={id,direction:'in',sessionId:parent.sessionId,ticketId:null,channel:parent.channel,chatId:x.chat_id,priority:'needs-answer',parentId:parent.id,text:await withAttachment(channel,x,inboundText(x.text,!!hashtag)),createdAt:x.date*1000,attempts:0,deliveredAt:null,escalatedAt:null,tgMessageId:x.message_id}; await register(channel,m); deps.store.setCursor(channel,x.date) } };
}
if (import.meta.main) void main();