/**
 * The weekly funnel reading: `marketing/metrics/weekly.md`'s data half.
 *
 * The funnel the marketing cluster agreed on is
 *
 *   visit → download click → DMG served → trial started → checkout started
 *         → purchase → activation
 *
 * and this module reads it from the two live databases, plus the campaign
 * dimension the /analytics page already exposes through
 * `getAnalyticsSummary.topCampaigns`.
 *
 * Two properties matter more than the numbers:
 *
 *  - It takes both connections as arguments. `~/server/analytics` imports the
 *    full server env (Stripe keys, Apple secret, licence key …); the weekly
 *    job only needs the two databases, and must be runnable from a checkout
 *    whose `.env` carries nothing else. The SQL here is the same shape as the
 *    analytics summaries, so the weekly read and the dashboard cannot drift
 *    into different label sets.
 *  - Every stage is measured by its own label. FRE-366 gave the funnel that:
 *    the download door and the DMG file route record distinct `stage` values
 *    (`click` / `served`), a created Stripe checkout session records its own
 *    row, and every row carries the real client address and the host it
 *    addressed, so the visit stage scopes to the Nook. Rows that predate
 *    that instrumentation are stated next to the numbers, never silently
 *    counted (see `renderWeeklyMarkdown`).
 *
 * No `~/env`, no `~/server/database`: the caller owns the connections, the
 * tests own an in-memory SQLite, and the script owns the libsql clients.
 */

import { DOWNLOAD_PATH_PREFIX } from "~/lib/download-asset";
import {
  VISITOR_ANALYTICS_ADDED_COLUMNS,
  VISITOR_ANALYTICS_DDL
} from "~/server/analytics-schema";

/** The one asset key the Nook lands under — `downloadPath("thenook")`. */
export const NOOK_DOWNLOAD_PATH = `${DOWNLOAD_PATH_PREFIX}thenook`;

/** The host the visit stage scopes to. Recorded per row since FRE-366. */
export const NOOK_HOST = "nook.freno.me";

/** The path a created Stripe checkout session records (FRE-366). */
export const NOOK_CHECKOUT_PATH = "/api/the-nook/checkout";

/** An `execute`-shaped connection: the libsql client and the test adapter both fit. */
export interface MetricsConn {
  execute(query: {
    sql: string;
    args?: any[];
  }): Promise<{ rows: unknown[]; rowsAffected?: number }>;
}

export interface Connections {
  /** freno.me Turso DB — VisitorAnalytics. */
  analytics: MetricsConn;
  /** the-nook Turso DB — trials, licenses, activations. */
  nook: MetricsConn;
}

export type StageKey =
  | "visit"
  | "downloadClick"
  | "dmgServed"
  | "trial"
  | "checkout"
  | "purchase"
  | "activation";

export interface StageDef {
  key: StageKey;
  label: string;
}

/**
 * The seven agreed stages, each with its own label in the data:
 * `downloadClick` and `dmgServed` split what used to be one merged download
 * label (the `stage` column, FRE-366), and `checkout` reads created Stripe
 * sessions instead of the /checkout page-view proxy.
 */
export const STAGES: StageDef[] = [
  { key: "visit", label: `visit (${NOOK_HOST} page views)` },
  { key: "downloadClick", label: "download click" },
  { key: "dmgServed", label: "DMG served" },
  { key: "trial", label: "trial started" },
  { key: "checkout", label: "checkout started (Stripe session)" },
  { key: "purchase", label: "purchase (licence issued)" },
  { key: "activation", label: "activation" }
];

export interface FunnelCounts {
  visit: number;
  downloadClick: number;
  dmgServed: number;
  trial: number;
  checkout: number;
  purchase: number;
  activation: number;
}

export interface CampaignRow {
  source: string | null;
  medium: string | null;
  campaign: string | null;
  count: number;
}

/** Traffic-shape facts the report has to state next to the funnel. */
export interface WindowTraffic {
  analyticsRows: number;
  /** Rows carrying a host — written since the FRE-366 instrumentation. */
  hostRows: number;
  /** Rows with no host — pre-FRE-366; their address is the proxy's loopback. */
  legacyRows: number;
  /** Post-instrumentation rows still recording loopback: the fix is not live. */
  hostLoopbackRows: number;
  downloadRows: number;
  downloadClickRows: number;
  downloadServedRows: number;
  /** Download rows with no stage label — they predate FRE-366 and cannot be split. */
  downloadUnstagedRows: number;
  downloadNonBrowserRows: number;
}

export interface Mover {
  key: StageKey;
  label: string;
  now: number;
  before: number;
  delta: number;
}

export interface WeeklyReport {
  window: { start: string; end: string; days: number };
  previous: { start: string; end: string };
  counts: FunnelCounts;
  previousCounts: FunnelCounts;
  campaigns: CampaignRow[];
  taggedRows: number;
  traffic: WindowTraffic;
  mover: Mover | null;
  generatedAt: string;
}

/**
 * A recorded address that is not a visitor, in every spelling the column can
 * hold: the proxy the request came through (`127.0.0.1`, `::1`, an IPv4-mapped
 * form) *and* the prefixes the FRE-369 privacy floor writes instead of them
 * (`127.0.0.0`, all-zero IPv6). Whole `127/8` because truncation keeps the first
 * three octets, so `127.0.53.1` and `127.0.53.0` are the same non-visitor.
 */
const LOOPBACK_SQL = `(ip_address LIKE '127.%'
     OR ip_address LIKE '::ffff:127.%'
     OR ip_address IN ('::1', '0:0:0:0:0:0:0:1', '0:0:0:0:0:0:0:0'))`;

/** SQLite's `datetime()` string, UTC: `YYYY-MM-DD HH:MM:SS`. */
export function sqlTime(date: Date): string {
  return date.toISOString().slice(0, 19).replace("T", " ");
}

export interface Windows {
  current: { start: string; end: string };
  previous: { start: string; end: string };
}

/**
 * The current window is `[now - days, now)` and the previous one sits directly
 * before it, so a weekly read always compares like with like. Boundaries are
 * passed as SQL parameters rather than `datetime('now', …)` so the file a run
 * writes is reproducible and the two queries cannot straddle a second.
 */
export function windowsFor(now: Date, days: number): Windows {
  const span = days * 24 * 60 * 60 * 1000;
  const end = now;
  const start = new Date(end.getTime() - span);
  const previousEnd = start;
  const previousStart = new Date(previousEnd.getTime() - span);

  return {
    current: { start: sqlTime(start), end: sqlTime(end) },
    previous: { start: sqlTime(previousStart), end: sqlTime(previousEnd) }
  };
}

function toNumber(value: unknown): number {
  if (typeof value === "number") return value;
  if (typeof value === "bigint") return Number(value);
  if (typeof value === "string") return Number(value) || 0;
  return 0;
}

async function scalar(
  conn: MetricsConn,
  sql: string,
  args: any[]
): Promise<number> {
  const result = await conn.execute({ sql, args });
  const row = result.rows[0] as Record<string, unknown> | undefined;
  if (!row) return 0;
  return toNumber(Object.values(row)[0]);
}

/**
 * Bring the analytics table up to the schema the readers expect, through the
 * same idempotent bootstrap path the app server uses (`analytics.ts`): the
 * DDL is a no-op against the live table and each post-hoc column is ALTERed
 * in only when absent. The weekly job runs against databases the app may not
 * have migrated yet — a fresh checkout, a restored backup, or production
 * before the next deploy — and must render its seven stages regardless.
 */
export async function ensureAnalyticsSchema(conn: MetricsConn): Promise<void> {
  await conn.execute({ sql: VISITOR_ANALYTICS_DDL });

  const info = await conn.execute({
    sql: `PRAGMA table_info(VisitorAnalytics)`
  });
  const present = new Set(info.rows.map((row) => String((row as any).name)));
  for (const column of VISITOR_ANALYTICS_ADDED_COLUMNS) {
    if (!present.has(column.name)) {
      await conn.execute({ sql: column.ddl });
    }
  }
}

async function countInWindow(
  conn: MetricsConn,
  sql: string,
  start: string,
  end: string
): Promise<number> {
  return scalar(conn, sql, [start, end]);
}

/** One window's funnel. Two databases, one call. */
export async function readFunnel(
  conns: Connections,
  window: { start: string; end: string }
): Promise<FunnelCounts> {
  const { analytics, nook } = conns;
  const { start, end } = window;

  // Scoped to the Nook host (FRE-366): before the host column existed, a
  // page view of `/` on nook.freno.me was the same row as freno.me's root.
  // Pre-instrumentation rows carry no host and are excluded — the traffic
  // section states how many.
  const visit = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND path NOT LIKE '/api/%'
     AND host = ?`,
    [start, end, NOOK_HOST]
  );

  const downloadClick = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND path = ? AND stage = 'click'`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  const dmgServed = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND path = ? AND stage = 'served'`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  // Created Stripe checkout sessions, recorded by the checkout route itself
  // (FRE-366) — not the /checkout page view that used to proxy for them.
  const checkout = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND path = ?`,
    [start, end, NOOK_CHECKOUT_PATH]
  );

  // The licence DB stores ISO-8601 (`2026-10-28T20:28:46.806Z`); the analytics
  // DB stores `YYYY-MM-DD HH:MM:SS`. `datetime()` normalises both sides so a
  // boundary seconds apart cannot include or drop a row on string comparison.
  const trial = await scalar(
    nook,
    `SELECT COUNT(*) AS n FROM trials
     WHERE datetime(started_at) >= datetime(?) AND datetime(started_at) < datetime(?)`,
    [start, end]
  );

  const purchase = await scalar(
    nook,
    `SELECT COUNT(*) AS n FROM licenses
     WHERE datetime(created_at) >= datetime(?) AND datetime(created_at) < datetime(?)`,
    [start, end]
  );

  const activation = await scalar(
    nook,
    `SELECT COUNT(*) AS n FROM activations
     WHERE datetime(activated_at) >= datetime(?) AND datetime(activated_at) < datetime(?)`,
    [start, end]
  );

  return {
    visit,
    downloadClick,
    dmgServed,
    trial,
    checkout,
    purchase,
    activation
  };
}

/**
 * Traffic shape for one window: how much of it predates the FRE-366
 * instrumentation (no host, proxy's address), whether rows recorded since
 * still see loopback (which would mean the forwarded-header fix is not
 * live), and how the download label splits. Without these facts the numbers
 * read as demand, which they are not.
 */
export async function readTraffic(
  conns: Connections,
  window: { start: string; end: string }
): Promise<WindowTraffic> {
  const { analytics } = conns;
  const { start, end } = window;

  const analyticsRows = await countInWindow(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?`,
    start,
    end
  );

  const hostRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND host IS NOT NULL`,
    [start, end]
  );

  const legacyRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND host IS NULL`,
    [start, end]
  );

  const hostLoopbackRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ?
     AND host IS NOT NULL
     AND ${LOOPBACK_SQL}`,
    [start, end]
  );

  const downloadRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ? AND path = ?`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  const downloadClickRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ? AND path = ? AND stage = 'click'`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  const downloadServedRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ? AND path = ? AND stage = 'served'`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  const downloadUnstagedRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ? AND path = ? AND stage IS NULL`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  // Anything whose UA is not a browser engine: curl, Sparkle's updater,
  // link-preview crawlers. A real install click arrives as `Mozilla/5.0 …`.
  const downloadNonBrowserRows = await scalar(
    analytics,
    `SELECT COUNT(*) AS n FROM VisitorAnalytics
     WHERE created_at >= ? AND created_at < ? AND path = ?
     AND (user_agent IS NULL OR user_agent NOT LIKE 'Mozilla/%')`,
    [start, end, NOOK_DOWNLOAD_PATH]
  );

  return {
    analyticsRows,
    hostRows,
    legacyRows,
    hostLoopbackRows,
    downloadRows,
    downloadClickRows,
    downloadServedRows,
    downloadUnstagedRows,
    downloadNonBrowserRows
  };
}

/**
 * The campaign dimension, grouped exactly as `getAnalyticsSummary.topCampaigns`
 * groups it: top 10 by volume, untagged rows excluded (the untagged bucket is
 * a different question and would otherwise swallow the list).
 */
export async function readCampaigns(
  analytics: MetricsConn,
  window: { start: string; end: string }
): Promise<CampaignRow[]> {
  const result = await analytics.execute({
    sql: `SELECT utm_source, utm_medium, utm_campaign, COUNT(*) AS count
          FROM VisitorAnalytics
          WHERE created_at >= ? AND created_at < ?
          AND (utm_source IS NOT NULL OR utm_medium IS NOT NULL OR utm_campaign IS NOT NULL)
          GROUP BY utm_source, utm_medium, utm_campaign
          ORDER BY count DESC
          LIMIT 10`,
    args: [window.start, window.end]
  });

  return result.rows.map((raw) => {
    const row = raw as Record<string, unknown>;
    return {
      source: (row.utm_source as string | null) ?? null,
      medium: (row.utm_medium as string | null) ?? null,
      campaign: (row.utm_campaign as string | null) ?? null,
      count: toNumber(row.count)
    };
  });
}

/**
 * "The one thing that moved": the stage with the largest absolute change
 * against the previous window. Ties go to the earlier stage, so a week when
 * nothing moved returns null rather than a made-up winner.
 */
export function pickMover(
  current: FunnelCounts,
  previous: FunnelCounts
): Mover | null {
  let best: Mover | null = null;

  for (const stage of STAGES) {
    const delta = current[stage.key] - previous[stage.key];
    if (delta === 0) continue;
    if (!best || Math.abs(delta) > Math.abs(best.delta)) {
      best = {
        key: stage.key,
        label: stage.label,
        now: current[stage.key],
        before: previous[stage.key],
        delta
      };
    }
  }

  return best;
}

export interface ReportOptions {
  /** End of the current window; defaults to the wall clock. */
  now?: Date;
  /** Window length in days; defaults to 7. */
  days?: number;
}

/** Read everything `.md` needs, in one call. */
export async function getWeeklyReport(
  conns: Connections,
  options: ReportOptions = {}
): Promise<WeeklyReport> {
  const days = options.days ?? 7;
  if (!Number.isFinite(days) || days <= 0) {
    throw new Error(`days must be a positive number, got ${days}`);
  }

  const now = options.now ?? new Date();
  const windows = windowsFor(now, days);

  const [counts, previousCounts, traffic, campaigns] = await Promise.all([
    readFunnel(conns, windows.current),
    readFunnel(conns, windows.previous),
    readTraffic(conns, windows.current),
    readCampaigns(conns.analytics, windows.current)
  ]);

  return {
    window: { ...windows.current, days },
    previous: windows.previous,
    counts,
    previousCounts,
    campaigns,
    taggedRows: campaigns.reduce((sum, row) => sum + row.count, 0),
    traffic,
    mover: pickMover(counts, previousCounts),
    generatedAt: now.toISOString().replace(/\.\d{3}Z$/, "Z")
  };
}

function cell(value: string | null): string {
  return value === null || value === "" ? "—" : value;
}

function signed(value: number): string {
  if (value === 0) return "0";
  return value > 0 ? `+${value}` : `−${Math.abs(value)}`;
}

/**
 * Render the weekly file. Interpretation lives in the CMO's read; this renderer
 * only states what the data is, what moved, and what the numbers cannot say.
 * The Gaps section is not boilerplate: each bullet is computed from the window
 * it describes, so it empties itself if the underlying hole is ever fixed.
 */
export function renderWeeklyMarkdown(report: WeeklyReport): string {
  const { counts, previousCounts, traffic } = report;
  const lines: string[] = [];

  lines.push("# Weekly funnel snapshot");
  lines.push("");
  lines.push(
    `Window: **${report.window.start} → ${report.window.end} UTC** (${report.window.days} days). ` +
      `Previous: ${report.previous.start} → ${report.previous.end} UTC.`
  );
  lines.push("");
  lines.push(
    "Generated by `bun --env-file=.env scripts/weekly-metrics.ts`; regenerated " +
      "weekly, never hand-edited. The analysis is the CMO's."
  );
  lines.push("");

  lines.push("## Funnel");
  lines.push("");
  lines.push("| Stage | This week | Last week | Change |");
  lines.push("|---|---:|---:|---:|");
  for (const stage of STAGES) {
    const now = counts[stage.key];
    const before = previousCounts[stage.key];
    lines.push(`| ${stage.label} | ${now} | ${before} | ${signed(now - before)} |`);
  }
  lines.push("");

  lines.push("## Campaigns");
  lines.push("");
  if (report.campaigns.length === 0) {
    lines.push("_No tagged traffic in the window._");
  } else {
    lines.push(
      "Top 10 by volume, untagged excluded — the same grouping " +
        "`getAnalyticsSummary.topCampaigns` returns."
    );
    lines.push("");
    lines.push("| Source | Medium | Campaign | Rows |");
    lines.push("|---|---|---|---:|");
    for (const row of report.campaigns) {
      lines.push(
        `| ${cell(row.source)} | ${cell(row.medium)} | ${cell(row.campaign)} | ${row.count} |`
      );
    }
  }
  lines.push("");
  lines.push(
    `${report.taggedRows} of ${traffic.analyticsRows} rows in the window carry ` +
      `campaign tags; the rest are untagged and therefore unattributed.`
  );
  lines.push("");

  lines.push("## The one thing that moved");
  lines.push("");
  if (!report.mover) {
    lines.push("_Nothing moved: every stage held its previous-week value._");
  } else {
    const mover = report.mover;
    lines.push(
      `**${mover.label}: ${mover.before} → ${mover.now} (${signed(mover.delta)}).**`
    );
  }
  lines.push("");

  lines.push("## Before acting on these numbers");
  lines.push("");

  if (traffic.legacyRows > 0) {
    lines.push(
      `- **${traffic.legacyRows} of ${traffic.analyticsRows} rows in the window predate the FRE-366 instrumentation.** ` +
        `They carry no host, and their recorded address is the production proxy's loopback ` +
        `(127.0.0.1 / ::1) — the writers then recorded the direct peer, not the visitor. ` +
        `They are excluded from every host-scoped stage above (visits, and the download ` +
        `click/served split), and "unique visitors" over them still collapses to one host. ` +
        `Rows recorded since carry the site they addressed and the visitor's address ` +
        `truncated to a prefix (/24, or /56 for IPv6 — the FRE-369 privacy floor), so ` +
        `"unique visitors" counts networks, not machines.`
    );
  }

  if (traffic.hostLoopbackRows > 0) {
    lines.push(
      `- **${traffic.hostLoopbackRows} rows recorded since the instrumentation still carry a ` +
        `loopback address** — the forwarded-header fix is not taking effect in production ` +
        `(no \`cf-connecting-ip\`, and no usable \`x-forwarded-for\` hop). Do not trust ` +
        `per-channel reach until this is zero.`
    );
  }

  if (traffic.downloadRows > 0) {
    lines.push(
      `- **Download rows are automation-heavy.** Of this window's ${traffic.downloadRows} ` +
        `download rows, ${traffic.downloadClickRows} are clicks and ` +
        `${traffic.downloadServedRows} are served files` +
        (traffic.downloadUnstagedRows > 0
          ? `; the remaining ${traffic.downloadUnstagedRows} predate the click/served stage ` +
            `label and cannot be split, so they count in neither stage`
          : "") +
        `. ${traffic.downloadNonBrowserRows} of all ${traffic.downloadRows} are not browser ` +
        `user agents (curl, Sparkle, link-preview crawlers).`
    );
  }

  lines.push(
    `- **No conversion rate before the cohort matures.** Trial→paid belongs to the mature ` +
      `Day 28–35 read (FRE-197); a ratio from a partial cohort is not a metric.`
  );
  lines.push("");
  lines.push("## Sources");
  lines.push("");
  lines.push("- `VisitorAnalytics` (freno.me Turso DB) — visits, downloads, campaigns.");
  lines.push("- `trials`, `licenses`, `activations` (the-nook Turso DB) — the paid funnel.");
  lines.push(`- Generated ${report.generatedAt}.`);
  lines.push("");

  return lines.join("\n");
}