MikrotikManager/backend/src/services/traffic-flow-facts-rebuild.ts

import { dbAll, dbGet, dbQuery, withAdvisoryLock } from "../db/index.js"
import { flushEngineNow } from "./traffic-flow-engine.js"
import { resolveInternetDest } from "./traffic-flow-dest.js"
import {
  bumpFlowFact,
  discardPendingFacts,
  flushFlowFacts,
  hourBucketIso,
} from "./traffic-flow-facts.js"
import { shouldWriteFlowFact } from "./traffic-flow-facts-filter.js"
import { canonicalFactIface } from "./traffic-flow-ifindex.js"
import { getServerCatalog, loadFlowTopology } from "./traffic-flow-topology.js"

export const FACT_REBUILD_LOCK_KEY = 8_723_104
const BATCH = 4_000

export interface FlowFactsRebuildResult {
  ok: true
  buckets: number
  facts: number
  days: string[]
}

function hourFromBucket(raw: Date | string): string {
  const iso = raw instanceof Date ? raw.toISOString() : String(raw)
  const ms = new Date(iso).getTime()
  return hourBucketIso(Number.isFinite(ms) ? ms : Date.now())
}

export async function rebuildFlowFactsFromBuckets(): Promise<FlowFactsRebuildResult> {
  return await withAdvisoryLock(FACT_REBUILD_LOCK_KEY, async () => {
    await flushEngineNow({ prune: false })
    discardPendingFacts()

    const days = await dbAll<{ day: string }>(`
      SELECT DISTINCT (bucket_at AT TIME ZONE 'UTC')::date::text AS day
      FROM flow_buckets
      ORDER BY 1
    `)
    const dayList = days.map((r) => r.day).filter(Boolean)
    if (dayList.length === 0) {
      return { ok: true as const, buckets: 0, facts: 0, days: [] }
    }

    await dbQuery(
      `DELETE FROM flow_hour_facts WHERE (bucket_at AT TIME ZONE 'UTC')::date = ANY(?::date[])`,
      [dayList],
    )
    await dbQuery(
      `DELETE FROM flow_daily_facts WHERE day = ANY(?::date[])`,
      [dayList],
    )

    const topo = await loadFlowTopology()
    const catalog = await getServerCatalog()
    let offset = 0
    let buckets = 0

    for (;;) {
      const rows = await dbAll<{
        serverId: number
        bucketAt: Date | string
        src: string
        dst: string
        proto: number
        srcPort: number
        dstPort: number
        bytes: number
        packets: number
        inIface: string
        outIface: string
        natSrc: string
        natDst: string
        natSrcPort: number
        natDstPort: number
      }>(`
        SELECT server_id AS "serverId", bucket_at AS "bucketAt",
          host(src) AS src, host(dst) AS dst, proto, src_port AS "srcPort", dst_port AS "dstPort",
          bytes, packets, in_iface AS "inIface", COALESCE(out_iface, '') AS "outIface",
          COALESCE(host(nat_src), '') AS "natSrc", COALESCE(host(nat_dst), '') AS "natDst",
          COALESCE(nat_src_port, 0) AS "natSrcPort", COALESCE(nat_dst_port, 0) AS "natDstPort"
        FROM flow_buckets
        ORDER BY bucket_at, server_id
        LIMIT ? OFFSET ?
      `, [BATCH, offset])
      if (rows.length === 0) break
      for (const row of rows) {
        buckets += 1
        const serverType = catalog.byId.get(row.serverId)?.type
        const destMeta = resolveInternetDest({
          src: row.src,
          dst: row.dst,
          proto: Number(row.proto) || 0,
          srcPort: Number(row.srcPort) || 0,
          dstPort: Number(row.dstPort) || 0,
          serverId: row.serverId,
          inIface: row.inIface,
          topo,
          natSrc: row.natSrc,
          natDst: row.natDst,
          natSrcPort: Number(row.natSrcPort) || 0,
          natDstPort: Number(row.natDstPort) || 0,
        })
        if (!shouldWriteFlowFact({
          serverId: row.serverId,
          serverType,
          inIface: row.inIface,
          outIface: row.outIface,
          proto: Number(row.proto) || 0,
          srcPort: Number(row.srcPort) || 0,
          dstPort: Number(row.dstPort) || 0,
          src: row.src,
          dst: row.dst,
          topo,
          dest: destMeta.dest,
          natSrc: row.natSrc,
          natDst: row.natDst,
          natSrcPort: Number(row.natSrcPort) || 0,
          natDstPort: Number(row.natDstPort) || 0,
        })) continue
        bumpFlowFact({
          serverId: row.serverId,
          bucketAt: hourFromBucket(row.bucketAt),
          iface: canonicalFactIface(row.serverId, row.inIface),
          country: destMeta.country || "XX",
          service: destMeta.classified.service,
          asn: destMeta.asn,
          bytes: Number(row.bytes) || 0,
          packets: Number(row.packets) || 0,
        })
      }
      offset += rows.length
      if (rows.length < BATCH) break
    }

    const facts = await flushFlowFacts()
    const range = await dbGet<{ n: number }>(`SELECT COUNT(*)::int AS n FROM flow_buckets`)
    return {
      ok: true as const,
      buckets: Number(range?.n) || buckets,
      facts,
      days: dayList,
    }
  })
}