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

import assert from "node:assert/strict"
import { dbQuery } from "../db/index.js"
import { withPgOrSkip } from "../test/pg.js"
import { ensurePartitionFor } from "../db/partitions.js"
import { pool } from "../db/index.js"
import { applySqlMigrations } from "../db/migrate.js"
import { invalidateFlowCatalogCache } from "./traffic-flow-topology.js"
import {
  disableRipeEnqueueForTests,
  disableRipePersistForTests,
  resetRipeCacheForTests,
  seedRipeCacheForTests,
} from "./traffic-flow-ripe.js"
import { rememberServerIfaces, resetIfaceCacheForTests } from "./traffic-flow-ifindex.js"
import { resetEngineForTests } from "./traffic-flow-engine.js"
import { disableCatalogFetchForTests, resetFlowCatalogForTests } from "./traffic-flow-classify.js"

if (!(await withPgOrSkip())) {
  console.log("traffic-flow-facts-rebuild.test.ts: skip")
  process.exit(0)
}

await applySqlMigrations(pool)

const nServers = (await dbQuery<{ n: number }>(`SELECT COUNT(*)::int AS n FROM servers`)).rows[0]?.n ?? 0
if (nServers > 10) {
  console.warn("traffic-flow-facts-rebuild.test.ts: skip (не пустая БД)")
  process.exit(0)
}

disableCatalogFetchForTests()
resetFlowCatalogForTests()
disableRipePersistForTests()
disableRipeEnqueueForTests()
resetRipeCacheForTests()
resetEngineForTests()
resetIfaceCacheForTests()

seedRipeCacheForTests({
  prefix: "8.8.8.0/24",
  asn: 15169,
  country: "US",
  lat: null,
  lng: null,
  holder: "GOOGLE",
  ok: true,
  fetchedAt: Date.now(),
})
seedRipeCacheForTests({
  prefix: "95.167.0.0/16",
  asn: 12389,
  country: "RU",
  lat: null,
  lng: null,
  holder: "ROSTELECOM-AS",
  ok: true,
  fetchedAt: Date.now(),
})

const inserted = await dbQuery<{ id: number }>(`
  INSERT INTO servers (name, host, type, wan_uplinks)
  VALUES ('rebuild-facts-jh', '203.0.113.10', 'jump-host', '[{"iface":"ether1"}]'::jsonb)
  RETURNING id
`)
const serverId = inserted.rows[0]?.id
if (serverId == null) throw new Error("no server")

const ts = new Date()
await ensurePartitionFor(pool, "flow_buckets", "day", ts)
await ensurePartitionFor(pool, "flow_hour_facts", "day", ts)
await ensurePartitionFor(pool, "flow_daily_facts", "month", ts)

await dbQuery(`DELETE FROM flow_buckets WHERE server_id = $1`, [serverId])
await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId])
await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId])
await dbQuery(`DELETE FROM user_interface_bindings WHERE server_id = $1`, [serverId])
await dbQuery(`DELETE FROM app_users WHERE id = 'u-rebuild-1'`)

await dbQuery(`
  INSERT INTO app_users (id, name, login, role, active)
  VALUES ('u-rebuild-1', 'Клиент', 'rebuild-user', 'viewer', TRUE)
  ON CONFLICT (id) DO NOTHING
`)
await dbQuery(`
  INSERT INTO user_interface_bindings (id, user_id, server_id, interface_name, interface_type)
  VALUES ('bind-rebuild-1', 'u-rebuild-1', $1, 'gre-client', 'gre')
`, [serverId])
await dbQuery(`
  INSERT INTO server_snapshots (server_id, polled_at, status, raw_interfaces)
  VALUES ($1, now(), 'online', $2::jsonb)
`, [serverId, JSON.stringify([{ name: "gre-client", type: "gre-tunnel" }, { name: "ether1", type: "ether" }])])

rememberServerIfaces(serverId, [{ name: "gre-client", ifindex: "2" }])
invalidateFlowCatalogCache()

const bucketAt = new Date(Date.UTC(
  ts.getUTCFullYear(),
  ts.getUTCMonth(),
  ts.getUTCDate(),
  ts.getUTCHours(),
  0, 0, 0,
)).toISOString()

await dbQuery(`
  INSERT INTO flow_buckets (server_id, bucket_at, src, dst, proto, src_port, dst_port, bytes, packets, in_iface, out_iface)
  VALUES
    ($1, $2, '95.167.1.10', '10.200.100.53', 6, 51234, 443, 100, 2, 'gre-client', 'ether1'),
    ($1, $2, '95.167.1.10', '8.8.8.8', 6, 51234, 443, 50, 1, 'gre-client', 'ether1')
`, [serverId, bucketAt])

try {
  const { rebuildFlowFactsFromBuckets } = await import("./traffic-flow-facts-rebuild.js")
  const result = await rebuildFlowFactsFromBuckets()
  assert.equal(result.ok, true)
  assert.ok(result.buckets >= 2)

  const rows = await dbQuery<{ asn: number; bytes: number }>(`
    SELECT asn, SUM(bytes)::bigint AS bytes
    FROM flow_hour_facts
    WHERE server_id = $1
    GROUP BY asn
  `, [serverId])
  const byAsn = new Map(rows.rows.map((r) => [Number(r.asn), Number(r.bytes)]))
  const total = [...byAsn.values()].reduce((s, n) => s + n, 0)
  assert.equal(total, 50)
  assert.equal(byAsn.get(12389), undefined, "ASN клиента не в hour facts")
  assert.equal(byAsn.get(15169), 50)
  assert.equal(byAsn.get(0), undefined)
} finally {
  await dbQuery(`DELETE FROM flow_hour_facts WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM flow_daily_facts WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM flow_buckets WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM user_interface_bindings WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM server_snapshots WHERE server_id = $1`, [serverId])
  await dbQuery(`DELETE FROM app_users WHERE id = 'u-rebuild-1'`)
  await dbQuery(`DELETE FROM servers WHERE id = $1`, [serverId])
  resetEngineForTests()
  invalidateFlowCatalogCache()
}

console.log("traffic-flow-facts-rebuild.test.ts: ok")