fix(traffic): проверить сбор трафика через HTTP и PostgreSQL

backend/package.json

@@ -19,7 +19,7 @@
     "test:exec": "tsx src/services/mikrotik-logs.test.ts && tsx src/routes/exec.test.ts && tsx src/plugins/auth.exec.test.ts",
     "test:auth": "tsx src/lib/permissions.test.ts && tsx src/plugins/auth.smoke.test.ts && tsx src/plugins/auth.ospf-optimizer.test.ts",
     "test:wireguard": "npx tsx src/services/wireguard-config.test.ts && npx tsx src/services/wireguard-ipv6.test.ts && npx tsx --test src/services/wireguard-import.test.ts src/services/wireguard-import-apply.test.ts src/services/wireguard-import-revision.test.ts src/services/wireguard-delete.test.ts && npx tsx src/routes/wireguard-import.test.ts",
-    "test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts",
+    "test:traffic-rate": "tsx src/services/traffic-rate.test.ts && tsx src/services/traffic-observation.test.ts && tsx src/services/traffic-observation-sql.test.ts",
     "test:traffic-flow": "tsx src/routes/traffic-flow-query.test.ts && tsx src/services/traffic-flow-month.test.ts && tsx src/services/traffic-flow-parse.test.ts && tsx src/services/traffic-flow-ipv6-pipeline.test.ts && tsx src/services/traffic-flow-map-exporter.test.ts && tsx src/services/traffic-flow-ifaces.test.ts && tsx src/services/traffic-flow-ifindex.test.ts && tsx src/services/traffic-flow-dedup.test.ts && tsx src/services/traffic-flow-planes.test.ts && tsx src/services/traffic-flow-ip.test.ts && tsx src/services/traffic-flow-dest.test.ts && tsx src/services/traffic-flow-classify.test.ts && tsx src/services/traffic-flow-ripe.test.ts && tsx src/services/traffic-flow-brands.test.ts && tsx src/services/traffic-flow-ingest.test.ts && tsx src/services/traffic-flow-analytics.test.ts && tsx src/services/traffic-flow-map-hops.test.ts && tsx src/services/traffic-flow-purge.test.ts && tsx src/services/traffic-flow-geoip.test.ts && tsx src/services/traffic-flow-facts.test.ts && tsx src/services/traffic-flow-facts-filter.test.ts && tsx src/services/traffic-flow-facts-rebuild.test.ts && tsx src/services/statistics-aggregate.test.ts",
     "test:users": "tsx src/modules/users/iface-type.test.ts && tsx src/modules/users/bindings.test.ts && tsx src/modules/users/interface-inventory.test.ts && tsx src/routes/users-catalog.test.ts",
     "test:pg": "tsx src/db/pg-pool.test.ts && tsx src/db/sql-bind.test.ts && tsx src/db/sqlite-json.test.ts && tsx src/db/traffic-flags.test.ts && tsx src/db/pg-schema.test.ts && tsx src/services/config-revisions.test.ts",

backend/src/services/traffic-observation-sql.test.ts

@@ -0,0 +1,65 @@
+import assert from "node:assert/strict"
+import { createServer } from "node:http"
+import { pool, closePool } from "../db/index.js"
+import { initDatabase } from "../db/bootstrap.js"
+import { collectTrafficOnce } from "./traffic-collector.js"
+import { buildSignalSnapshotFromCollectors } from "./alert-engine/signals-from-collectors.js"
+
+const database = process.env.MMAPP_OWN_SQL_DB
+if (!database) console.log("traffic observation SQL: skipped; explicit owned database required")
+else {
+  assert(/^mm_accept_166_\d+$/.test(database))
+  assert.equal(new URL(process.env.DATABASE_URL!).pathname, "/" + database)
+  assert.equal((await pool.query("SELECT current_database() AS name")).rows[0].name, database)
+  const prefix = "own-traffic175-" + Date.now() + "-"
+  let serverId: number | undefined, counters: unknown = "1000", second = false, cases = 0, requests = 0
+  const http = createServer((req, res) => {
+    requests++; res.setHeader("Content-Type", "application/json")
+    const rows = [{ name: "ether1", running: "true", disabled: "false", "rx-byte": counters, "tx-byte": counters }]
+    if (second) rows.push({ name: "ether2", running: "true", disabled: "false", "rx-byte": "1000", "tx-byte": "1000" })
+    res.end(JSON.stringify(req.url?.includes("/wireguard/") ? [] : rows))
+  })
+  await new Promise<void>(resolve => http.listen(0, "127.0.0.1", resolve))
+  const address = http.address(); assert(address && typeof address === "object")
+  try {
+    await initDatabase()
+    serverId = Number((await pool.query("INSERT INTO servers(name,host,port,use_ssl,enabled,username,password,updated_at) VALUES($1,'127.0.0.1',$2,false,true,'fixture','fixture',$3) RETURNING id", [prefix + "server", address.port, new Date(Date.now() - 3_600_000).toISOString()])).rows[0].id)
+    await pool.query("INSERT INTO traffic_settings(id,enabled,interval_sec) VALUES(1,true,30) ON CONFLICT(id) DO UPDATE SET enabled=true,interval_sec=30")
+    const count = async () => Number((await pool.query("SELECT count(*) AS n FROM traffic_samples WHERE server_id=$1", [serverId])).rows[0].n)
+    const collect = async (known: boolean) => {
+      const snapshot = await collectTrafficOnce(), row = snapshot.servers.find(s => s.serverId === serverId)!
+      assert.equal(typeof row.sumRxMbps === "number", known)
+      assert.equal(typeof row.sumTxMbps === "number", known)
+      await pool.query("DELETE FROM scheduler_runs WHERE job_key='traffic'")
+      await pool.query("INSERT INTO scheduler_runs(id,job_key,started_at,finished_at,status,result_json) VALUES($1,'traffic',$2,$3,'ok',$4::jsonb)", [prefix + cases, snapshot.sampledAt, new Date().toISOString(), JSON.stringify(snapshot)])
+      const source = (await buildSignalSnapshotFromCollectors()).trafficByServer
+      assert.equal(source.length, Number(known)); cases++
+      return row
+    }
+    let row = await collect(false); assert.equal(row.ok, true); assert.equal(await count(), 1)
+    // Move the fixture baseline, not the clock, to avoid slow sleeps and equal timestamps.
+    const age = async () => {
+      await pool.query("DELETE FROM traffic_samples WHERE server_id=$1 AND sampled_at<>(SELECT max(sampled_at) FROM traffic_samples WHERE server_id=$1)", [serverId])
+      await pool.query("UPDATE traffic_samples SET sampled_at=$1 WHERE server_id=$2", [new Date(Date.now() - 30_000).toISOString(), serverId])
+    }
+    await age(); row = await collect(true); assert.equal(row.sumRxMbps, 0)
+    await age(); counters = "10"; row = await collect(false); assert.equal(row.ok, true)
+    await age(); const beforeBad = await count(); counters = "123bad"; row = await collect(false)
+    assert.equal(row.ok, false); assert.match(row.error!, /некорректный счётчик/); assert.equal(await count(), beforeBad)
+    await age(); counters = undefined; row = await collect(false); assert.equal(row.ok, false); assert.equal(await count(), beforeBad)
+    await age(); counters = "10"; second = true; row = await collect(false); assert.equal(row.ok, true)
+    await age(); row = await collect(true); assert.equal(row.sumRxMbps, 0)
+    await age(); counters = "3750010"; row = await collect(true); assert(row.sumRxMbps! > 0.9)
+    await pool.query("UPDATE servers SET enabled=false WHERE id=$1", [serverId]); const beforeDisabled = requests
+    assert.equal((await collectTrafficOnce()).servers.length, 0); assert.equal(requests, beforeDisabled); cases++
+    console.log(`traffic observation owned SQL: ${cases} cases; actual HTTP/collector/persistence/scheduler/source/disabled passed`)
+  } finally {
+    await new Promise<void>(resolve => http.close(() => resolve()))
+    await pool.query("DELETE FROM scheduler_runs WHERE id LIKE $1", [prefix + "%"])
+    if (serverId !== undefined) {
+      await pool.query("DELETE FROM traffic_samples WHERE server_id=$1", [serverId])
+      await pool.query("DELETE FROM servers WHERE id=$1", [serverId])
+    }
+    await closePool()
+  }
+}