286 lines
9.2 KiB
TypeScript
286 lines
9.2 KiB
TypeScript
import { desc, eq } from "drizzle-orm"
|
|
import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
|
|
import { db } from "../db/index.js"
|
|
import { serverSnapshots, servers } from "../db/schema.js"
|
|
import {
|
|
collectTrafficOnce,
|
|
getTrafficCollectorState,
|
|
getTrafficSettings,
|
|
readServerSamplesInRange,
|
|
updateTrafficSettings,
|
|
} from "../services/traffic-collector.js"
|
|
|
|
type SnapshotRow = typeof serverSnapshots.$inferSelect
|
|
|
|
interface TrafficServerDto {
|
|
id: string
|
|
name: string
|
|
site: string
|
|
country: string
|
|
status: "online" | "offline" | "degraded"
|
|
rxNow: number
|
|
txNow: number
|
|
rxPeak: number
|
|
txPeak: number
|
|
rxTotal: number
|
|
txTotal: number
|
|
sessions: number
|
|
rxSeries: number[]
|
|
txSeries: number[]
|
|
}
|
|
|
|
interface TrafficInterfaceDto {
|
|
name: string
|
|
running: boolean
|
|
disabled: boolean
|
|
rxNow: number
|
|
txNow: number
|
|
}
|
|
|
|
function latestSnapshot(serverId: number): SnapshotRow | undefined {
|
|
return db
|
|
.select()
|
|
.from(serverSnapshots)
|
|
.where(eq(serverSnapshots.serverId, serverId))
|
|
.orderBy(desc(serverSnapshots.polledAt))
|
|
.limit(1)
|
|
.all()[0]
|
|
}
|
|
|
|
function rangeToMinutes(range: string | undefined): number {
|
|
switch ((range ?? "1h").toLowerCase()) {
|
|
case "5m": return 5
|
|
case "15m": return 15
|
|
case "1h": return 60
|
|
case "4h": return 240
|
|
case "24h": return 1440
|
|
default: return 60
|
|
}
|
|
}
|
|
|
|
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"],
|
|
rows: Array<{
|
|
interfaceName: string
|
|
sampledAt: string
|
|
rxBps: number
|
|
txBps: number
|
|
rxBytes: number
|
|
txBytes: number
|
|
running: boolean
|
|
disabled: boolean
|
|
}>,
|
|
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
|
|
}
|
|
}
|
|
|
|
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,
|
|
}
|
|
}
|
|
|
|
const trafficRoutes: FastifyPluginAsyncZod = async (app) => {
|
|
app.get("/traffic/settings", async (_req, reply) => {
|
|
const settings = getTrafficSettings()
|
|
const state = getTrafficCollectorState()
|
|
return reply.send({
|
|
enabled: settings.enabled,
|
|
intervalSec: settings.intervalSec,
|
|
retentionDays: settings.retentionDays,
|
|
lastCollectedAt: settings.lastCollectedAt ?? null,
|
|
lastDurationMs: settings.lastDurationMs ?? null,
|
|
lastError: settings.lastError || null,
|
|
collectorRunning: state.running,
|
|
})
|
|
})
|
|
|
|
app.put("/traffic/settings", async (req, reply) => {
|
|
const body = req.body as {
|
|
enabled?: boolean
|
|
intervalSec?: number | string
|
|
retentionDays?: number | string
|
|
}
|
|
const intervalSec = body.intervalSec == null ? undefined : Math.max(5, Number.parseInt(String(body.intervalSec), 10) || 30)
|
|
const retentionDays = body.retentionDays == null ? undefined : Math.max(1, Number.parseInt(String(body.retentionDays), 10) || 14)
|
|
const updated = updateTrafficSettings({
|
|
enabled: body.enabled,
|
|
intervalSec,
|
|
retentionDays,
|
|
})
|
|
return reply.send({
|
|
ok: true,
|
|
settings: {
|
|
enabled: updated.enabled,
|
|
intervalSec: updated.intervalSec,
|
|
retentionDays: updated.retentionDays,
|
|
lastCollectedAt: updated.lastCollectedAt ?? null,
|
|
lastDurationMs: updated.lastDurationMs ?? null,
|
|
lastError: updated.lastError || null,
|
|
},
|
|
})
|
|
})
|
|
|
|
app.post("/traffic/collect-now", async (_req, reply) => {
|
|
await collectTrafficOnce()
|
|
const updated = getTrafficSettings()
|
|
return reply.send({
|
|
ok: true,
|
|
lastCollectedAt: updated.lastCollectedAt ?? null,
|
|
lastDurationMs: updated.lastDurationMs ?? null,
|
|
lastError: updated.lastError || null,
|
|
})
|
|
})
|
|
|
|
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 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 reply.send({ servers: data })
|
|
})
|
|
|
|
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)
|
|
if (!Number.isFinite(serverId)) return reply.status(400).send({ error: "id is required" })
|
|
|
|
const allRows = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()
|
|
const server = allRows[0]
|
|
if (!server) return reply.status(404).send({ error: "Server not found" })
|
|
|
|
const sinceIso = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString()
|
|
const rows = readServerSamplesInRange(serverId, sinceIso)
|
|
const byIface = new Map<string, typeof rows>()
|
|
for (const r of rows) {
|
|
const arr = byIface.get(r.interfaceName) ?? []
|
|
arr.push(r)
|
|
byIface.set(r.interfaceName, arr)
|
|
}
|
|
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]
|
|
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),
|
|
}
|
|
}).sort((a, b) => (b.rxNow + b.txNow) - (a.rxNow + a.txNow))
|
|
|
|
return reply.send({ interfaces })
|
|
})
|
|
|
|
app.get("/traffic/servers/:id", async (req, reply) => {
|
|
const p = req.params as { id?: string | number }
|
|
const q = req.query as { range?: string; iface?: string }
|
|
const serverId = Number.parseInt(String(p.id ?? ""), 10)
|
|
if (!Number.isFinite(serverId)) return reply.status(400).send({ error: "id is required" })
|
|
const server = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0]
|
|
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 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)
|
|
return reply.send({ server: data })
|
|
})
|
|
}
|
|
|
|
export default trafficRoutes
|
|
|