#!/usr/bin/env node
/**
 * paas-traffic-rollup — aggregate Caddy JSON access logs into paas.db.
 *
 * Runs every 5 min via systemd timer (User=deploy, Nice=19, io idle). Reads
 * only bytes appended since the last run (offset+inode state per file), so a
 * run typically touches a few KB. Zero impact on the request path: Caddy logs
 * asynchronously; this process is fully offline.
 *
 * Log files: /srv/paas/logs/caddy/<app>.access.log  (paas regen writes the
 * Caddy `log` directive per app). On rotation (inode change) parsing restarts
 * at offset 0 of the new file; the tail of the rolled file inside the last
 * 5-min window is deliberately not chased (negligible, documented).
 *
 * Same pass also aggregates per-article click counts into
 * article_clicks_daily, for apps that ship a route-manifest.json (currently
 * dotnews-an-tier-a based apps only — see loadManifest). Apps without one are
 * untouched: this is opt-in per app, not a guess at what counts as an
 * "article" for a codebase this script knows nothing about.
 */
'use strict'
const fs = require('node:fs')
const path = require('node:path')
const { execFileSync } = require('node:child_process')

const LOG_DIR = process.env.CADDY_LOG_DIR || '/srv/paas/logs/caddy'
const DB_PATH = process.env.PAAS_DB || '/srv/paas/paas.db'
const STATE_PATH = path.join(LOG_DIR, '.rollup-state.json')
const PANEL_SQLITE = process.env.PANEL_SQLITE || '/srv/paas/panel/node_modules/better-sqlite3'
const APPS_ROOT = process.env.APPS_ROOT || '/srv/apps'

function loadState() {
  try { return JSON.parse(fs.readFileSync(STATE_PATH, 'utf8')) } catch { return {} }
}

// Bounded work per run: at most 64 MB per file per invocation — a traffic
// burst never makes a single run large; the remainder is picked up by the
// next timer tick. Keeps memory well under the unit's MemoryMax.
const MAX_READ_PER_RUN = 64 * 1024 * 1024

function readNewLines(file, st) {
  const stat = fs.statSync(file)
  let offset = 0
  if (st && st.ino === stat.ino && st.offset <= stat.size) offset = st.offset
  if (offset === stat.size) return { lines: [], state: { ino: stat.ino, offset } }

  const fd = fs.openSync(file, 'r')
  let buf
  try {
    const len = Math.min(stat.size - offset, MAX_READ_PER_RUN)
    buf = Buffer.alloc(len)
    fs.readSync(fd, buf, 0, len, offset)
  } finally {
    fs.closeSync(fd)
  }
  const text = buf.toString('utf8')
  const lastNl = text.lastIndexOf('\n')
  if (lastNl === -1) return { lines: [], state: { ino: stat.ino, offset } } // partial line only
  const complete = text.slice(0, lastNl)
  return {
    lines: complete.length ? complete.split('\n') : [],
    state: { ino: stat.ino, offset: offset + Buffer.byteLength(complete, 'utf8') + 1 },
  }
}

/* ------------------------------------------------------------
   Per-article click counting (opt-in per app via route-manifest.json,
   generated at build time by dotnews-an-tier-a's scripts/generate-route-
   manifest.js and shipped in .output/). Not a guess: any path that is
   neither a known static/dynamic non-article route nor multi-segment nor
   asset-shaped is treated as an article slug.
   ------------------------------------------------------------ */
const manifestCache = new Map() // app -> manifest object | null (checked, not found)

function loadManifest(app) {
  if (manifestCache.has(app)) return manifestCache.get(app)
  let manifest = null
  try {
    // paas deploy flattens .output/* into the release root (see paas
    // deploy_app) — so this lives at current/route-manifest.json, not
    // nested under a surviving current/.output/ directory.
    const file = path.join(APPS_ROOT, app, 'current', 'route-manifest.json')
    manifest = JSON.parse(fs.readFileSync(file, 'utf8'))
  } catch { manifest = null }
  manifestCache.set(app, manifest)
  return manifest
}

function classifyArticlePath(manifest, uri) {
  const p = (uri || '').split('?')[0]
  if (!p.startsWith('/') || p === '/') return null
  const rest = p.slice(1)
  if (!rest || rest.includes('/') || rest.includes('.')) return null
  if (manifest.exact.includes(p)) return null
  if ((manifest.prefix || []).some((pre) => p.startsWith(pre))) return null
  if ((manifest.suffix || []).some((suf) => p.endsWith(suf))) return null
  if ((manifest.both || []).some((b) => p.startsWith(b.prefix) && p.endsWith(b.suffix))) return null
  return rest
}

/* Static pages (landing pages, guide articles, /broker, /podcast, …) — the
   mirror-image selection: a single-segment, dot-free path that IS in the
   manifest's exact list. Multi-segment routes (/categories/…, /autor/…) and
   dynamic prefix/suffix routes stay out of scope for click counting. */
function classifyPagePath(manifest, uri) {
  const p = (uri || '').split('?')[0]
  if (!p.startsWith('/') || p === '/') return null
  const rest = p.slice(1)
  if (!rest || rest.includes('/') || rest.includes('.')) return null
  return manifest.exact.includes(p) ? p : null
}

function headerValue(headers, name) {
  if (!headers) return ''
  const v = headers[name]
  if (Array.isArray(v)) return v[0] || ''
  return typeof v === 'string' ? v : ''
}

// Not exhaustive by design — a bot UA missed here just means one crawler
// hit briefly inflates a count until noticed; a legitimate reader wrongly
// filtered would be the worse failure mode, so this stays a plain substring
// match on well-known crawler/tooling signatures rather than an allowlist.
// Tuned 2026-08-20 against a real sample of gen-a.younex.de access logs
// (312 distinct UAs) — catches, among others: search/AI crawlers, SEO/perf
// scanners (Lighthouse, GTmetrix, MarketGoo, Chrome's prefetch proxy,
// GoogleOther), generic scripted HTTP clients (python/aiohttp/okhttp/
// go-http-client/curl/wget/httpx), any UA embedding a URL (a near-universal
// bot self-identification convention real browsers don't follow), and —
// critically — this platform's own synthetic traffic: paas-warm (Varnish
// cache warming) and the Checkmk uptime probe (check_http /
// monitoring-plugins), both of which hit real article paths and would
// otherwise inflate "reads" with zero human involvement.
const BOT_UA_RE = /bot|crawl|spider|slurp|facebookexternalhit|whatsapp|telegram|preview|pingdom|uptimerobot|curl\/|wget\/|scrapy|headlesschrome|phantomjs|ia_archiver|googleother|lighthouse|prefetch|checker|auditor|marketgoo|gtmetrix|scrape|trafilatura|sniffer|networkingextension|paas-warm|check_http|monitoring-plugins|go-http-client|python|urllib|aiohttp|okhttp|httpx|https?:\/\/|\bnode\b/i

function aggregate(lines, app, agg, manifest, clickAgg, pageAgg) {
  for (const line of lines) {
    if (!line) continue
    let e
    try { e = JSON.parse(line) } catch { continue }
    const ts = typeof e.ts === 'number' ? e.ts : null
    if (ts === null) continue
    const day = new Date(ts * 1000).toISOString().slice(0, 10)
    const key = `${app} ${day}`
    const a = agg.get(key) || { app, day, requests: 0, bytes: 0, hit: 0, miss: 0, s2xx: 0, s4xx: 0, s5xx: 0, bytes_cache: 0, bytes_origin: 0 }
    a.requests += 1
    const size = typeof e.size === 'number' ? e.size : 0
    a.bytes += size
    const status = e.status | 0
    if (status >= 200 && status < 300) a.s2xx += 1
    else if (status >= 400 && status < 500) a.s4xx += 1
    else if (status >= 500) a.s5xx += 1
    const xc = e.resp_headers && e.resp_headers['X-Cache']
    const xcv = Array.isArray(xc) ? xc[0] : xc
    if (xcv === 'HIT') { a.hit += 1; a.bytes_cache += size }
    else if (xcv === 'MISS') { a.miss += 1; a.bytes_origin += size }
    else a.bytes_origin += size // no Varnish header: app direct or Caddy static
    agg.set(key, a)

    if (manifest && status >= 200 && status < 300 && e.request?.method === 'GET') {
      const ua = headerValue(e.request?.headers, 'User-Agent')
      if (ua && !BOT_UA_RE.test(ua)) {
        const slug = classifyArticlePath(manifest, e.request?.uri)
        if (slug) {
          const ckey = `${app} ${slug} ${day}`
          const c = clickAgg.get(ckey) || { app, slug, day, clicks: 0 }
          c.clicks += 1
          clickAgg.set(ckey, c)
        } else {
          /* not an article — a static page from the manifest counts too
             (landing/guide pages have no CMS row; their clicks feed the
             admin Klick Report via page_clicks_daily/clicks_monthly) */
          const page = classifyPagePath(manifest, e.request?.uri)
          if (page) {
            const pkey = `${app} ${page} ${day}`
            const pRow = pageAgg.get(pkey) || { app, path: page, day, clicks: 0 }
            pRow.clicks += 1
            pageAgg.set(pkey, pRow)
          }
        }
      }
    }
  }
}

/* Archive fully completed months exactly once (never updated afterwards):
   month must be over AND >=1 day of the next month passed (grace for the
   last rollup ticks), and not already archived. */
const ARCHIVE_MONTHS = `INSERT OR IGNORE INTO traffic_monthly
  (app, month, requests, bytes, hit, miss, s2xx, s4xx, s5xx, bytes_cache, bytes_origin, finalized_at)
  SELECT app, substr(day, 1, 7), SUM(requests), SUM(bytes), SUM(hit), SUM(miss),
         SUM(s2xx), SUM(s4xx), SUM(s5xx), SUM(bytes_cache), SUM(bytes_origin), datetime('now')
  FROM traffic_daily
  WHERE substr(day, 1, 7) < strftime('%Y-%m', 'now')
    AND date(substr(day, 1, 7) || '-01', '+1 month', '+1 day') <= date('now')
  GROUP BY app, substr(day, 1, 7)`

const MONTHLY_TABLE = `CREATE TABLE IF NOT EXISTS traffic_monthly (
  app TEXT NOT NULL, month TEXT NOT NULL, requests INTEGER NOT NULL DEFAULT 0,
  bytes INTEGER NOT NULL DEFAULT 0, hit INTEGER NOT NULL DEFAULT 0,
  miss INTEGER NOT NULL DEFAULT 0, s2xx INTEGER NOT NULL DEFAULT 0,
  s4xx INTEGER NOT NULL DEFAULT 0, s5xx INTEGER NOT NULL DEFAULT 0,
  finalized_at TEXT NOT NULL, PRIMARY KEY (app, month))`

const UPSERT = `INSERT INTO traffic_daily (app, day, requests, bytes, hit, miss, s2xx, s4xx, s5xx, bytes_cache, bytes_origin)
  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT(app, day) DO UPDATE SET
    requests = requests + excluded.requests, bytes = bytes + excluded.bytes,
    hit = hit + excluded.hit, miss = miss + excluded.miss,
    s2xx = s2xx + excluded.s2xx, s4xx = s4xx + excluded.s4xx, s5xx = s5xx + excluded.s5xx,
    bytes_cache = bytes_cache + excluded.bytes_cache, bytes_origin = bytes_origin + excluded.bytes_origin`

const ALTERS = ['traffic_daily', 'traffic_monthly'].flatMap((t) =>
  ['bytes_cache', 'bytes_origin'].map((c) => `ALTER TABLE ${t} ADD COLUMN ${c} INTEGER NOT NULL DEFAULT 0`))

const CLICK_TABLE = `CREATE TABLE IF NOT EXISTS article_clicks_daily (
  app TEXT NOT NULL, slug TEXT NOT NULL, day TEXT NOT NULL,
  clicks INTEGER NOT NULL DEFAULT 0,
  PRIMARY KEY (app, slug, day))`

const CLICK_UPSERT = `INSERT INTO article_clicks_daily (app, slug, day, clicks)
  VALUES (?, ?, ?, ?)
  ON CONFLICT(app, slug, day) DO UPDATE SET clicks = clicks + excluded.clicks`

// bin/paas-clicks-writeback queries trailing 7/30-day windows —
// prune well past that so this table stays small regardless of catalog size
const CLICK_PRUNE = `DELETE FROM article_clicks_daily WHERE day < date('now', '-35 days')`

/* Static-page clicks (paths from the route manifest's exact list) — same
   shape and lifecycle as article_clicks_daily, keyed by path. */
const PAGE_TABLE = `CREATE TABLE IF NOT EXISTS page_clicks_daily (
  app TEXT NOT NULL, path TEXT NOT NULL, day TEXT NOT NULL,
  clicks INTEGER NOT NULL DEFAULT 0,
  PRIMARY KEY (app, path, day))`

const PAGE_UPSERT = `INSERT INTO page_clicks_daily (app, path, day, clicks)
  VALUES (?, ?, ?, ?)
  ON CONFLICT(app, path, day) DO UPDATE SET clicks = clicks + excluded.clicks`

const PAGE_PRUNE = `DELETE FROM page_clicks_daily WHERE day < date('now', '-35 days')`

/* Unified 12-month click history for articles AND static pages (articles
   stored as path = '/' + slug), incremented in the same pass as the daily
   rows and kept 13 months — this is what survives the 35-day daily prune
   and feeds the Klick Report's monthly totals. */
const CLICKS_MONTHLY_TABLE = `CREATE TABLE IF NOT EXISTS clicks_monthly (
  app TEXT NOT NULL, path TEXT NOT NULL, month TEXT NOT NULL,
  clicks INTEGER NOT NULL DEFAULT 0,
  PRIMARY KEY (app, path, month))`

const CLICKS_MONTHLY_UPSERT = `INSERT INTO clicks_monthly (app, path, month, clicks)
  VALUES (?, ?, ?, ?)
  ON CONFLICT(app, path, month) DO UPDATE SET clicks = clicks + excluded.clicks`

const CLICKS_MONTHLY_PRUNE = `DELETE FROM clicks_monthly WHERE month < strftime('%Y-%m', date('now', '-13 months'))`

function writeRows(rows, clickRows, pageRows) {
  /* clicks_monthly rows derived from the same per-run increments —
     articles keyed as '/' + slug so the table holds one path namespace */
  const monthlyRows = [
    ...clickRows.map((r) => ({ app: r.app, path: '/' + r.slug, month: r.day.slice(0, 7), clicks: r.clicks })),
    ...pageRows.map((r) => ({ app: r.app, path: r.path, month: r.day.slice(0, 7), clicks: r.clicks })),
  ]
  try {
    const Database = require(PANEL_SQLITE)
    const db = new Database(DB_PATH)
    db.pragma('busy_timeout = 5000')
    db.exec(`CREATE TABLE IF NOT EXISTS traffic_daily (
      app TEXT NOT NULL, day TEXT NOT NULL, requests INTEGER NOT NULL DEFAULT 0,
      bytes INTEGER NOT NULL DEFAULT 0, hit INTEGER NOT NULL DEFAULT 0,
      miss INTEGER NOT NULL DEFAULT 0, s2xx INTEGER NOT NULL DEFAULT 0,
      s4xx INTEGER NOT NULL DEFAULT 0, s5xx INTEGER NOT NULL DEFAULT 0,
      PRIMARY KEY (app, day))`)
    db.exec(MONTHLY_TABLE)
    db.exec(CLICK_TABLE)
    db.exec(PAGE_TABLE)
    db.exec(CLICKS_MONTHLY_TABLE)
    for (const a of ALTERS) { try { db.exec(a) } catch { /* column exists */ } }
    if (rows.length) {
      const stmt = db.prepare(UPSERT)
      const tx = db.transaction((rs) => { for (const r of rs) stmt.run(r.app, r.day, r.requests, r.bytes, r.hit, r.miss, r.s2xx, r.s4xx, r.s5xx, r.bytes_cache, r.bytes_origin) })
      tx(rows)
    }
    if (clickRows.length) {
      const stmt = db.prepare(CLICK_UPSERT)
      const tx = db.transaction((rs) => { for (const r of rs) stmt.run(r.app, r.slug, r.day, r.clicks) })
      tx(clickRows)
    }
    if (pageRows.length) {
      const stmt = db.prepare(PAGE_UPSERT)
      const tx = db.transaction((rs) => { for (const r of rs) stmt.run(r.app, r.path, r.day, r.clicks) })
      tx(pageRows)
    }
    if (monthlyRows.length) {
      const stmt = db.prepare(CLICKS_MONTHLY_UPSERT)
      const tx = db.transaction((rs) => { for (const r of rs) stmt.run(r.app, r.path, r.month, r.clicks) })
      tx(monthlyRows)
    }
    db.exec(ARCHIVE_MONTHS) // close the ledger for any fully completed month
    db.exec(CLICK_PRUNE)
    db.exec(PAGE_PRUNE)
    db.exec(CLICKS_MONTHLY_PRUNE)
    db.close()
  } catch (err) {
    // fallback: sqlite3 CLI (all values are integers / controlled strings);
    // table creates first so a fresh DB accepts the inserts
    const sql = [MONTHLY_TABLE + ';', CLICK_TABLE + ';', PAGE_TABLE + ';', CLICKS_MONTHLY_TABLE + ';']
      .concat(ALTERS.map((a) => a + ';')).concat(rows.map((r) =>
      `INSERT INTO traffic_daily (app, day, requests, bytes, hit, miss, s2xx, s4xx, s5xx, bytes_cache, bytes_origin) VALUES ('${r.app.replace(/'/g, "''")}', '${r.day}', ${r.requests}, ${r.bytes}, ${r.hit}, ${r.miss}, ${r.s2xx}, ${r.s4xx}, ${r.s5xx}, ${r.bytes_cache}, ${r.bytes_origin}) ON CONFLICT(app, day) DO UPDATE SET requests = requests + excluded.requests, bytes = bytes + excluded.bytes, hit = hit + excluded.hit, miss = miss + excluded.miss, s2xx = s2xx + excluded.s2xx, s4xx = s4xx + excluded.s4xx, s5xx = s5xx + excluded.s5xx, bytes_cache = bytes_cache + excluded.bytes_cache, bytes_origin = bytes_origin + excluded.bytes_origin;`
    )).concat(clickRows.map((r) =>
      `INSERT INTO article_clicks_daily (app, slug, day, clicks) VALUES ('${r.app.replace(/'/g, "''")}', '${r.slug.replace(/'/g, "''")}', '${r.day}', ${r.clicks}) ON CONFLICT(app, slug, day) DO UPDATE SET clicks = clicks + excluded.clicks;`
    )).concat(pageRows.map((r) =>
      `INSERT INTO page_clicks_daily (app, path, day, clicks) VALUES ('${r.app.replace(/'/g, "''")}', '${r.path.replace(/'/g, "''")}', '${r.day}', ${r.clicks}) ON CONFLICT(app, path, day) DO UPDATE SET clicks = clicks + excluded.clicks;`
    )).concat(monthlyRows.map((r) =>
      `INSERT INTO clicks_monthly (app, path, month, clicks) VALUES ('${r.app.replace(/'/g, "''")}', '${r.path.replace(/'/g, "''")}', '${r.month}', ${r.clicks}) ON CONFLICT(app, path, month) DO UPDATE SET clicks = clicks + excluded.clicks;`
    )).concat([ARCHIVE_MONTHS + ';', CLICK_PRUNE + ';', PAGE_PRUNE + ';', CLICKS_MONTHLY_PRUNE + ';']).join('\n')
    execFileSync('sqlite3', [DB_PATH], { input: sql })
  }
}

function main() {
  if (!fs.existsSync(LOG_DIR)) return
  const state = loadState()
  const agg = new Map()
  const clickAgg = new Map()
  const pageAgg = new Map()
  const files = fs.readdirSync(LOG_DIR).filter((f) => f.endsWith('.access.log'))
  for (const f of files) {
    const app = f.replace(/\.access\.log$/, '')
    const full = path.join(LOG_DIR, f)
    try {
      const { lines, state: st } = readNewLines(full, state[f])
      aggregate(lines, app, agg, loadManifest(app), clickAgg, pageAgg)
      state[f] = st
    } catch (err) {
      console.error(`[rollup] ${f}: ${err.message}`)
    }
  }
  writeRows([...agg.values()], [...clickAgg.values()], [...pageAgg.values()])
  fs.writeFileSync(STATE_PATH, JSON.stringify(state))
  const total = [...agg.values()].reduce((n, a) => n + a.requests, 0)
  const clickTotal = [...clickAgg.values()].reduce((n, c) => n + c.clicks, 0)
  const pageTotal = [...pageAgg.values()].reduce((n, c) => n + c.clicks, 0)
  if (total) console.log(`[rollup] aggregated ${total} requests across ${agg.size} app-days${clickTotal ? `, ${clickTotal} article clicks across ${clickAgg.size} app-slug-days` : ''}${pageTotal ? `, ${pageTotal} page clicks across ${pageAgg.size} app-path-days` : ''}`)
}

main()