feat(ui): integrate KpiStatGrid for enhanced statistics display

Replaced existing KPI display implementations across multiple pages with the new KpiStatGrid component for a more consistent and visually appealing presentation of statistics. Updated the Backups, BGP, Certificates, Communities, Containers, Dashboard, and Data Collection pages to utilize the KpiStatGrid, improving the overall user experience and maintainability of the codebase. Additionally, added new dependencies in package.json for required libraries.

backend/src/routes/firewall.ts

@@ -0,0 +1,294 @@
+import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
+import { z } from "zod"
+import { MikrotikClient, MikrotikError, encodeRosId, firewallRestPath } from "../services/mikrotik.js"
+import { getEnabledServerById } from "../services/wireguard-live.js"
+import { listFirewallAll } from "../services/firewall-live.js"
+import type { FirewallFamily, FirewallTable } from "../types/server.js"
+
+const FamilySchema = z.enum(["ip", "ip6"])
+const TableSchema = z.enum(["filter", "nat", "mangle", "raw"])
+
+const RuleKeySchema = z.object({
+  serverId: z.string().min(1),
+  family: FamilySchema,
+  table: TableSchema,
+  rosId: z.string().min(1),
+})
+
+const RuleWriteSchema = z.object({
+  serverId: z.string().min(1),
+  family: FamilySchema,
+  table: TableSchema,
+  rosId: z.string().min(1).optional(),
+  chain: z.string().min(1),
+  action: z.string().min(1),
+  protocol: z.string().optional(),
+  srcAddress: z.string().optional(),
+  dstAddress: z.string().optional(),
+  srcAddressList: z.string().optional(),
+  dstAddressList: z.string().optional(),
+  srcPort: z.string().optional(),
+  dstPort: z.string().optional(),
+  inInterface: z.string().optional(),
+  outInterface: z.string().optional(),
+  connectionState: z.string().optional(),
+  comment: z.string().optional(),
+  disabled: z.boolean().optional(),
+  log: z.boolean().optional(),
+  logPrefix: z.string().optional(),
+  tlsHost: z.string().optional(),
+  layer7Proto: z.string().optional(),
+})
+
+const RulePatchSchema = RuleKeySchema.extend({
+  disabled: z.boolean(),
+})
+
+const RuleMoveSchema = RuleKeySchema.extend({
+  destinationRosId: z.string().min(1).optional(),
+})
+
+const AddressKeySchema = z.object({
+  serverId: z.string().min(1),
+  family: FamilySchema,
+  rosId: z.string().min(1),
+})
+
+const AddressWriteSchema = z.object({
+  serverId: z.string().min(1),
+  family: FamilySchema,
+  rosId: z.string().min(1).optional(),
+  list: z.string().min(1),
+  address: z.string().min(1),
+  comment: z.string().optional(),
+  timeout: z.string().optional(),
+  disabled: z.boolean().optional(),
+})
+
+const AddressPatchSchema = AddressKeySchema.extend({
+  disabled: z.boolean(),
+})
+
+function toRosBody(obj: Record<string, string | undefined>): Record<string, string> {
+  const out: Record<string, string> = {}
+  for (const [k, v] of Object.entries(obj)) {
+    if (v !== undefined && v !== "") out[k] = v
+  }
+  return out
+}
+
+function ruleToRos(d: z.infer<typeof RuleWriteSchema>): Record<string, string> {
+  return toRosBody({
+    chain: d.chain,
+    action: d.action,
+    protocol: d.protocol && d.protocol !== "all" ? d.protocol : undefined,
+    "src-address": d.srcAddress,
+    "dst-address": d.dstAddress,
+    "src-address-list": d.srcAddressList,
+    "dst-address-list": d.dstAddressList,
+    "src-port": d.srcPort,
+    "dst-port": d.dstPort,
+    "in-interface": d.inInterface,
+    "out-interface": d.outInterface,
+    "connection-state": d.connectionState,
+    comment: d.comment,
+    disabled: d.disabled === true ? "yes" : d.disabled === false ? "no" : undefined,
+    log: d.log === true ? "yes" : d.log === false ? "no" : undefined,
+    "log-prefix": d.logPrefix,
+    "tls-host": d.tlsHost,
+    "layer7-protocol": d.layer7Proto,
+  })
+}
+
+function addressToRos(d: z.infer<typeof AddressWriteSchema>): Record<string, string> {
+  return toRosBody({
+    list: d.list,
+    address: d.address,
+    comment: d.comment,
+    timeout: d.timeout,
+    disabled: d.disabled === true ? "yes" : d.disabled === false ? "no" : undefined,
+  })
+}
+
+function rosErr(e: unknown): string {
+  if (e instanceof MikrotikError) return e.message
+  if (e instanceof Error) return e.message
+  return String(e)
+}
+
+function requireServer(serverId: string) {
+  return getEnabledServerById(serverId)
+}
+
+const firewallRoutes: FastifyPluginAsyncZod = async (app) => {
+  app.get("/firewall/all", async (_req, reply) => {
+    const data = await listFirewallAll()
+    return reply.send(data)
+  })
+
+  app.post("/firewall/rules", async (req, reply) => {
+    const parsed = RuleWriteSchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = firewallRestPath(body.family as FirewallFamily, body.table as FirewallTable)
+    try {
+      await client.put(path, ruleToRos(body))
+      return reply.status(201).send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.put("/firewall/rules", async (req, reply) => {
+    const parsed = RuleWriteSchema.extend({ rosId: z.string().min(1) }).safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family as FirewallFamily, body.table as FirewallTable)}/${encodeRosId(body.rosId)}`
+    try {
+      await client.patch(path, ruleToRos(body))
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.patch("/firewall/rules", async (req, reply) => {
+    const parsed = RulePatchSchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family, body.table)}/${encodeRosId(body.rosId)}`
+    try {
+      await client.patch(path, { disabled: body.disabled ? "yes" : "no" })
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.delete("/firewall/rules", async (req, reply) => {
+    const parsed = RuleKeySchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family, body.table)}/${encodeRosId(body.rosId)}`
+    try {
+      await client.delete(path)
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.post("/firewall/rules/move", async (req, reply) => {
+    const parsed = RuleMoveSchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family, body.table)}/move`
+    try {
+      await client.post(path, {
+        numbers: body.rosId,
+        ...(body.destinationRosId ? { destination: body.destinationRosId } : {}),
+      })
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.post("/firewall/address-lists", async (req, reply) => {
+    const parsed = AddressWriteSchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    try {
+      await client.put(firewallRestPath(body.family, "address-list"), addressToRos(body))
+      return reply.status(201).send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.put("/firewall/address-lists", async (req, reply) => {
+    const parsed = AddressWriteSchema.extend({ rosId: z.string().min(1) }).safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family, "address-list")}/${encodeRosId(body.rosId)}`
+    try {
+      await client.patch(path, addressToRos(body))
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.patch("/firewall/address-lists", async (req, reply) => {
+    const parsed = AddressPatchSchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family, "address-list")}/${encodeRosId(body.rosId)}`
+    try {
+      await client.patch(path, { disabled: body.disabled ? "yes" : "no" })
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+
+  app.delete("/firewall/address-lists", async (req, reply) => {
+    const parsed = AddressKeySchema.safeParse(req.body ?? {})
+    if (!parsed.success) {
+      return reply.status(400).send({ error: "Некорректное тело запроса", details: parsed.error.flatten() })
+    }
+    const body = parsed.data
+    const server = requireServer(body.serverId)
+    if (!server) return reply.status(404).send({ error: "Сервер не найден" })
+    const client = MikrotikClient.fromServer(server)
+    const path = `${firewallRestPath(body.family, "address-list")}/${encodeRosId(body.rosId)}`
+    try {
+      await client.delete(path)
+      return reply.send({ ok: true })
+    } catch (e) {
+      return reply.status(502).send({ error: `RouterOS: ${rosErr(e)}` })
+    }
+  })
+}
+
+export default firewallRoutes

backend/src/routes/traffic.ts

@@ -1,3 +1,4 @@
+import { env } from "../config.js"
 import { desc, eq } from "drizzle-orm"
 import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
 import { db } from "../db/index.js"
@@ -12,6 +13,15 @@ import {
 import { scheduleAlertEngineAfterDataCollectors } from "../services/alert-collector-hooks.js"
 import { refreshScheduler } from "../services/scheduler.js"
 import { appendEvent } from "../modules/events/service/events-service.js"
+import { MikrotikClient } from "../services/mikrotik.js"
+import { getEnabledServerById } from "../services/wireguard-live.js"
+import {
+  bpsToMbps,
+  buildTrafficFromSamples,
+  isLoopbackName,
+  parseMonitorTraffic,
+  rateBpsFromDelta,
+} from "../services/traffic-rate.js"
 
 type SnapshotRow = typeof serverSnapshots.$inferSelect
 
@@ -40,6 +50,9 @@ interface TrafficInterfaceDto {
   txNow: number
 }
 
+const LIVE_TICK_MS = 1500
+const LIVE_ROS_TIMEOUT_MS = 4000
+
 function latestSnapshot(serverId: number): SnapshotRow | undefined {
   return db
     .select()
@@ -61,14 +74,6 @@ function rangeToMinutes(range: string | undefined): number {
   }
 }
 
-function toSeries(values: number[], target = 60): number[] {
-  if (values.length === 0) return Array(target).fill(0)
-  if (values.length === target) return values
-  if (values.length > target) return values.slice(values.length - target)
-  const head = Array(target - values.length).fill(values[0] ?? 0)
-  return [...head, ...values]
-}
-
 function buildServerTraffic(
   s: typeof servers.$inferSelect,
   status: TrafficServerDto["status"],
@@ -82,88 +87,135 @@ function buildServerTraffic(
     running: boolean
     disabled: boolean
   }>,
+  rangeStartMs: number,
+  rangeEndMs: number,
   onlyInterface?: string,
 ): TrafficServerDto {
-  const filteredRows = onlyInterface
-    ? rows.filter((r) => r.interfaceName === onlyInterface)
-    : rows
-
-  if (filteredRows.length === 0) {
-    return {
-      id: String(s.id),
-      name: s.name || s.host,
-      site: s.site || "—",
-      country: s.country || "UN",
-      status,
-      rxNow: 0,
-      txNow: 0,
-      rxPeak: 0,
-      txPeak: 0,
-      rxTotal: 0,
-      txTotal: 0,
-      sessions: 0,
-      rxSeries: Array(60).fill(0),
-      txSeries: Array(60).fill(0),
-    }
-  }
-
-  const bySampleTs = new Map<string, { rx: number; tx: number }>()
-  const byIface = new Map<string, typeof filteredRows>()
-  for (const r of filteredRows) {
-    const ts = r.sampledAt
-    const cur = bySampleTs.get(ts) ?? { rx: 0, tx: 0 }
-    cur.rx += Math.max(0, r.rxBps) / 1_000_000
-    cur.tx += Math.max(0, r.txBps) / 1_000_000
-    bySampleTs.set(ts, cur)
-    const arr = byIface.get(r.interfaceName) ?? []
-    arr.push(r)
-    byIface.set(r.interfaceName, arr)
-  }
-
-  const seriesPoints = [...bySampleTs.entries()]
-    .sort((a, b) => a[0].localeCompare(b[0]))
-    .map(([, v]) => ({ rx: Math.round(v.rx), tx: Math.round(v.tx) }))
-  const rxSeries = toSeries(seriesPoints.map((p) => p.rx))
-  const txSeries = toSeries(seriesPoints.map((p) => p.tx))
-  const rxNow = rxSeries[rxSeries.length - 1] ?? 0
-  const txNow = txSeries[txSeries.length - 1] ?? 0
-  const rxPeak = rxSeries.reduce((m, v) => Math.max(m, v), 0)
-  const txPeak = txSeries.reduce((m, v) => Math.max(m, v), 0)
-
-  let rxBytesDelta = 0
-  let txBytesDelta = 0
-  let sessions = 0
-  for (const arr of byIface.values()) {
-    const sorted = [...arr].sort((a, b) => a.sampledAt.localeCompare(b.sampledAt))
-    const first = sorted[0]
-    const last = sorted[sorted.length - 1]
-    if (first && last) {
-      const dRx = last.rxBytes - first.rxBytes
-      const dTx = last.txBytes - first.txBytes
-      rxBytesDelta += dRx >= 0 ? dRx : last.rxBytes
-      txBytesDelta += dTx >= 0 ? dTx : last.txBytes
-      if (last.running && !last.disabled) sessions += 1
-    }
-  }
-
+  const built = buildTrafficFromSamples(rows, rangeStartMs, rangeEndMs, onlyInterface)
   return {
     id: String(s.id),
     name: s.name || s.host,
     site: s.site || "—",
     country: s.country || "UN",
     status,
-    rxNow,
-    txNow,
-    rxPeak,
-    txPeak,
-    rxTotal: Number((rxBytesDelta / (1024 ** 3)).toFixed(1)),
-    txTotal: Number((txBytesDelta / (1024 ** 3)).toFixed(1)),
-    sessions,
-    rxSeries,
-    txSeries,
+    rxNow: built.rxNow,
+    txNow: built.txNow,
+    rxPeak: built.rxPeak,
+    txPeak: built.txPeak,
+    rxTotal: built.rxTotalGiB,
+    txTotal: built.txTotalGiB,
+    sessions: built.sessions,
+    rxSeries: built.rxSeries,
+    txSeries: built.txSeries,
   }
 }
 
+function snapshotStatus(serverId: number): TrafficServerDto["status"] {
+  const snap = latestSnapshot(serverId)
+  return snap?.status === "offline" ? "offline" : (snap?.status === "online" ? "online" : "degraded")
+}
+
+function ifaceNowMbps(
+  prev: { rxBytes: number; txBytes: number; sampledAt: string } | undefined,
+  last: { rxBytes: number; txBytes: number; sampledAt: string; rxBps: number; txBps: number },
+): { rxNow: number; txNow: number } {
+  if (!prev) {
+    return { rxNow: bpsToMbps(last.rxBps), txNow: bpsToMbps(last.txBps) }
+  }
+  const t0 = Date.parse(prev.sampledAt)
+  const t1 = Date.parse(last.sampledAt)
+  const rxBps = rateBpsFromDelta(prev.rxBytes, last.rxBytes, t0, t1)
+  const txBps = rateBpsFromDelta(prev.txBytes, last.txBytes, t0, t1)
+  return {
+    rxNow: bpsToMbps(rxBps ?? last.rxBps),
+    txNow: bpsToMbps(txBps ?? last.txBps),
+  }
+}
+
+function writeSse(raw: NodeJS.WritableStream, event: string, data: unknown) {
+  raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
+}
+
+function sleep(ms: number, signal: AbortSignal): Promise<void> {
+  return new Promise((resolve, reject) => {
+    if (signal.aborted) {
+      reject(new Error("aborted"))
+      return
+    }
+    const timer = setTimeout(() => {
+      signal.removeEventListener("abort", onAbort)
+      resolve()
+    }, ms)
+    const onAbort = () => {
+      clearTimeout(timer)
+      reject(new Error("aborted"))
+    }
+    signal.addEventListener("abort", onAbort, { once: true })
+  })
+}
+
+function flattenMonitor(raw: unknown): unknown[] {
+  if (Array.isArray(raw)) return raw
+  if (raw != null) return [raw]
+  return []
+}
+
+async function listRunningIfaceNames(client: MikrotikClient): Promise<string[]> {
+  const ifaces = await client.get<Array<{ name?: string; running?: string; disabled?: string }>>(
+    "/interface",
+    LIVE_ROS_TIMEOUT_MS,
+  )
+  return ifaces
+    .filter((i) => (i.running ?? "false") === "true"
+      && (i.disabled ?? "false") !== "true"
+      && !isLoopbackName(i.name ?? ""))
+    .map((i) => i.name ?? "")
+    .filter(Boolean)
+}
+
+async function monitorTrafficOnce(
+  client: MikrotikClient,
+  onlyInterface: string | undefined,
+  cache: { names: string[]; joinedFailed: boolean },
+  signal: AbortSignal,
+): Promise<unknown> {
+  if (onlyInterface) {
+    return client.post(
+      "/interface/monitor-traffic",
+      { interface: onlyInterface, once: "" },
+      LIVE_ROS_TIMEOUT_MS,
+      signal,
+    )
+  }
+  if (cache.names.length === 0) {
+    cache.names = await listRunningIfaceNames(client)
+  }
+  if (cache.names.length === 0) return []
+  if (!cache.joinedFailed) {
+    try {
+      return await client.post(
+        "/interface/monitor-traffic",
+        { interface: cache.names.join(","), once: "" },
+        LIVE_ROS_TIMEOUT_MS,
+        signal,
+      )
+    } catch {
+      cache.joinedFailed = true
+    }
+  }
+  const chunks = await Promise.all(
+    cache.names.map((name) =>
+      client.post(
+        "/interface/monitor-traffic",
+        { interface: name, once: "" },
+        LIVE_ROS_TIMEOUT_MS,
+        signal,
+      ).then(flattenMonitor).catch(() => [] as unknown[]),
+    ),
+  )
+  return chunks.flat()
+}
+
 const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
   app.get("/traffic/settings", async (_req, reply) => {
     const settings = getTrafficSettings()
@@ -243,20 +295,77 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
   app.get("/traffic/servers", async (req, reply) => {
     const q = req.query as { range?: string }
     const minutes = rangeToMinutes(q.range)
-    const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
+    const rangeEndMs = Date.now()
+    const rangeStartMs = rangeEndMs - minutes * 60_000
+    const sinceIso = new Date(rangeStartMs).toISOString()
     const allServers = db.select().from(servers).where(eq(servers.enabled, true)).all()
     const data = allServers.map((s): TrafficServerDto => {
-      const snap = latestSnapshot(s.id)
-      const status: TrafficServerDto["status"] =
-        snap?.status === "offline" ? "offline" : (snap?.status === "online" ? "online" : "degraded")
-
       const rows = readServerSamplesInRange(s.id, sinceIso)
-      return buildServerTraffic(s, status, rows)
+      return buildServerTraffic(s, snapshotStatus(s.id), rows, rangeStartMs, rangeEndMs)
     })
 
     return reply.send({ servers: data })
   })
 
+  app.get("/traffic/servers/:id/live", async (req, reply) => {
+    const p = req.params as { id?: string | number }
+    const q = req.query as { iface?: string }
+    const server = getEnabledServerById(p.id ?? "")
+    if (!server || !server.enabled) return reply.status(404).send({ error: "Server not found" })
+
+    const onlyInterface = q.iface && q.iface !== "__all__" ? q.iface : undefined
+    const abort = new AbortController()
+    const onClose = () => abort.abort()
+    req.raw.on("close", onClose)
+
+    reply.hijack()
+    req.raw.setTimeout(0)
+    reply.raw.setTimeout(0)
+    const origin = typeof req.headers.origin === "string" ? req.headers.origin : ""
+    const allowed = env.CORS_ORIGIN
+    const sseHeaders: Record<string, string> = {
+      "Content-Type": "text/event-stream; charset=utf-8",
+      "Cache-Control": "no-cache, no-transform",
+      Connection: "keep-alive",
+      "X-Accel-Buffering": "no",
+    }
+    if (origin && (allowed === "*" || allowed === origin)) {
+      sseHeaders["Access-Control-Allow-Origin"] = origin
+      sseHeaders["Access-Control-Allow-Credentials"] = "true"
+      sseHeaders["Access-Control-Allow-Headers"] = "Authorization, Accept"
+      sseHeaders.Vary = "Origin"
+    }
+    reply.raw.writeHead(200, sseHeaders)
+    reply.raw.write(":\n\n")
+
+    const client = MikrotikClient.fromServer(server)
+    const cache = { names: onlyInterface ? [onlyInterface] : [] as string[], joinedFailed: false }
+
+    try {
+      while (!abort.signal.aborted) {
+        try {
+          const raw = await monitorTrafficOnce(client, onlyInterface, cache, abort.signal)
+          const sample = parseMonitorTraffic(raw, { onlyInterface })
+          writeSse(reply.raw, "sample", sample)
+        } catch (error) {
+          if (abort.signal.aborted) break
+          const msg = error instanceof Error ? error.message : String(error)
+          writeSse(reply.raw, "error", { error: msg })
+        }
+        await sleep(LIVE_TICK_MS, abort.signal)
+      }
+    } catch {
+      /* abort / disconnect */
+    } finally {
+      req.raw.off("close", onClose)
+      try {
+        reply.raw.end()
+      } catch {
+        /* already closed */
+      }
+    }
+  })
+
   app.get("/traffic/servers/:id/interfaces", async (req, reply) => {
     const p = req.params as { id?: string | number }
     const serverId = Number.parseInt(String(p.id ?? ""), 10)
@@ -270,6 +379,7 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     const rows = readServerSamplesInRange(serverId, sinceIso)
     const byIface = new Map<string, typeof rows>()
     for (const r of rows) {
+      if (isLoopbackName(r.interfaceName)) continue
       const arr = byIface.get(r.interfaceName) ?? []
       arr.push(r)
       byIface.set(r.interfaceName, arr)
@@ -277,12 +387,17 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     const interfaces: TrafficInterfaceDto[] = [...byIface.entries()].map(([name, arr]) => {
       const sorted = arr.sort((a, b) => a.sampledAt.localeCompare(b.sampledAt))
       const last = sorted[sorted.length - 1]
+      const prev = sorted[sorted.length - 2]
+      if (!last) {
+        return { name, running: false, disabled: true, rxNow: 0, txNow: 0 }
+      }
+      const now = ifaceNowMbps(prev, last)
       return {
         name,
-        running: Boolean(last?.running),
-        disabled: Boolean(last?.disabled),
-        rxNow: Math.round((last?.rxBps ?? 0) / 1_000_000),
-        txNow: Math.round((last?.txBps ?? 0) / 1_000_000),
+        running: Boolean(last.running),
+        disabled: Boolean(last.disabled),
+        rxNow: now.rxNow,
+        txNow: now.txNow,
       }
     }).sort((a, b) => (b.rxNow + b.txNow) - (a.rxNow + a.txNow))
 
@@ -298,15 +413,20 @@ const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
     if (!server) return reply.status(404).send({ error: "Server not found" })
 
     const minutes = rangeToMinutes(q.range)
-    const sinceIso = new Date(Date.now() - minutes * 60_000).toISOString()
+    const rangeEndMs = Date.now()
+    const rangeStartMs = rangeEndMs - minutes * 60_000
+    const sinceIso = new Date(rangeStartMs).toISOString()
     const rows = readServerSamplesInRange(server.id, sinceIso)
-    const snap = latestSnapshot(server.id)
-    const status: TrafficServerDto["status"] =
-      snap?.status === "offline" ? "offline" : (snap?.status === "online" ? "online" : "degraded")
-    const data = buildServerTraffic(server, status, rows, q.iface && q.iface !== "__all__" ? q.iface : undefined)
+    const data = buildServerTraffic(
+      server,
+      snapshotStatus(server.id),
+      rows,
+      rangeStartMs,
+      rangeEndMs,
+      q.iface && q.iface !== "__all__" ? q.iface : undefined,
+    )
     return reply.send({ server: data })
   })
 }
 
 export default trafficRoutes
-