DenozordecandCursor
5aef419582
Docker images / prepare-release (push) Successful in 8s
Docker images / backend-test (push) Successful in 2m10s
Docker images / frontend-image (push) Successful in 3m13s
Docker images / updater-image (push) Successful in 46s
Docker images / backend-image (push) Successful in 2m52s
Docker images / notify-webhook (push) Skipped
Docker images / publish-release (push) Successful in 10s
Co-authored-by: Cursor <[email protected]>
886 lines
29 KiB
TypeScript
886 lines
29 KiB
TypeScript
import { eq } from "drizzle-orm"
|
|
import { db, dbAll } from "../db/index.js"
|
|
import { appUsers, flowAsnMeta, servers, userInterfaceBindings } from "../db/schema.js"
|
|
import {
|
|
STATISTICS_UNBOUND_USER_ID,
|
|
type StatisticsBreakdownRow,
|
|
type StatisticsDto,
|
|
type StatisticsPivotDim,
|
|
type StatisticsPivotDto,
|
|
type StatisticsPivotQuery,
|
|
type StatisticsQuery,
|
|
} from "@mmapp/contracts/statistics"
|
|
import {
|
|
collapseServerIfaceRows,
|
|
displayFactIface,
|
|
expandBindingIfaces,
|
|
factIfaceAliases,
|
|
listCachedIfaceNames,
|
|
} from "./traffic-flow-ifindex.js"
|
|
import { refreshServerIfaces } from "./traffic-flow-ifaces.js"
|
|
import {
|
|
isDashDisplayIface,
|
|
isJunkFactIface,
|
|
isOverlayTunnelIface,
|
|
isWanFactIface,
|
|
overlayDupLabel,
|
|
wanIfaceLabel,
|
|
} from "./traffic-flow-facts-filter.js"
|
|
import { getServerCatalog, loadFlowTopology, type FlowTopology } from "./traffic-flow-topology.js"
|
|
|
|
const TOP_N = 200
|
|
const HOUR_WINDOW_MS = 48 * 3600_000
|
|
const PIVOT_ROW_CAP = 50
|
|
const PIVOT_COL_CAP = 15
|
|
const PIVOT_OTHER_ID = "__other__"
|
|
|
|
export interface ParsedPeriod {
|
|
fromIso: string
|
|
toIso: string
|
|
fromDay: string
|
|
toDayExclusive: string
|
|
grain: "hour" | "day"
|
|
windowSec: number
|
|
}
|
|
|
|
function pad2(n: number): string {
|
|
return String(n).padStart(2, "0")
|
|
}
|
|
|
|
function toUtcDay(d: Date): string {
|
|
return `${d.getUTCFullYear()}-${pad2(d.getUTCMonth() + 1)}-${pad2(d.getUTCDate())}`
|
|
}
|
|
|
|
function addUtcDays(day: string, n: number): string {
|
|
const d = new Date(`${day}T00:00:00Z`)
|
|
d.setUTCDate(d.getUTCDate() + n)
|
|
return toUtcDay(d)
|
|
}
|
|
|
|
/** Parse from/to. Date-only `to` is inclusive (end of that UTC day). */
|
|
export function parseStatisticsPeriod(fromRaw: string, toRaw: string): ParsedPeriod | null {
|
|
const from = Date.parse(fromRaw.includes("T") ? fromRaw : `${fromRaw}T00:00:00Z`)
|
|
const toHasTime = toRaw.includes("T")
|
|
const to = Date.parse(toHasTime ? toRaw : `${toRaw}T00:00:00Z`)
|
|
if (!Number.isFinite(from) || !Number.isFinite(to)) return null
|
|
const fromDate = new Date(from)
|
|
let toDate = new Date(to)
|
|
let toDayExclusive: string
|
|
if (toHasTime) {
|
|
toDayExclusive = toUtcDay(toDate)
|
|
if (toDate.getUTCHours() !== 0 || toDate.getUTCMinutes() !== 0 || toDate.getUTCSeconds() !== 0) {
|
|
toDayExclusive = addUtcDays(toDayExclusive, 1)
|
|
}
|
|
} else {
|
|
toDayExclusive = addUtcDays(toUtcDay(toDate), 1)
|
|
toDate = new Date(`${toDayExclusive}T00:00:00Z`)
|
|
}
|
|
if (toDate.getTime() <= from) return null
|
|
const windowSec = Math.max(1, Math.round((toDate.getTime() - from) / 1000))
|
|
const grain: "hour" | "day" = toDate.getTime() - from <= HOUR_WINDOW_MS ? "hour" : "day"
|
|
return {
|
|
fromIso: fromDate.toISOString(),
|
|
toIso: toDate.toISOString(),
|
|
fromDay: toUtcDay(fromDate),
|
|
toDayExclusive,
|
|
grain,
|
|
windowSec,
|
|
}
|
|
}
|
|
|
|
type FactScope = "unique" | "wan" | "overlay"
|
|
|
|
interface FilterCtx {
|
|
fromIso: string
|
|
toIso: string
|
|
fromDay: string
|
|
toDayExclusive: string
|
|
serverId?: number
|
|
iface?: string
|
|
country?: string
|
|
service?: string
|
|
asn?: number
|
|
planes: "unique" | "all"
|
|
userIfaces: Array<{ serverId: number; iface: string }> | null
|
|
unboundOnly: boolean
|
|
boundIfaces: Array<{ serverId: number; iface: string }>
|
|
overlayIfaces: Array<{ serverId: number; iface: string }>
|
|
wanIfaces: Array<{ serverId: number; iface: string }>
|
|
excludeServerIds: number[]
|
|
topo: FlowTopology | null
|
|
}
|
|
|
|
function ifaceFilterAliases(iface: string, serverId?: number): string[] {
|
|
return factIfaceAliases(iface.trim(), serverId)
|
|
}
|
|
|
|
function looksLikeIfIndex(iface: string): boolean {
|
|
const raw = iface.trim()
|
|
return /^\d+$/.test(raw) || /^#\d+$/.test(raw)
|
|
}
|
|
|
|
async function warmIfaceCache(ids: Iterable<number>): Promise<void> {
|
|
const uniq = [...new Set(ids)].filter((id) => Number.isFinite(id) && id > 0)
|
|
if (!uniq.length) return
|
|
await Promise.all(uniq.map((id) => refreshServerIfaces(id)))
|
|
}
|
|
|
|
async function warmBindingIfaceCache(): Promise<void> {
|
|
const rows = await db.select({ serverId: userInterfaceBindings.serverId }).from(userInterfaceBindings)
|
|
await warmIfaceCache(rows.map((r) => r.serverId))
|
|
}
|
|
|
|
function canonicalIfaceDimId(id: string): string {
|
|
const colon = id.indexOf(":")
|
|
if (colon < 0) return id
|
|
const sid = Number(id.slice(0, colon))
|
|
if (!Number.isFinite(sid)) return id
|
|
return `${sid}:${displayFactIface(sid, id.slice(colon + 1))}`
|
|
}
|
|
|
|
function pushIfaceTuples(
|
|
parts: string[],
|
|
params: unknown[],
|
|
alias: string,
|
|
tuples: Array<{ serverId: number; iface: string }>,
|
|
op: "IN" | "NOT IN",
|
|
): void {
|
|
if (!tuples.length) {
|
|
if (op === "IN") parts.push("FALSE")
|
|
return
|
|
}
|
|
const sql = tuples.map(() => "(?, ?)").join(", ")
|
|
parts.push(`(${alias}.server_id, ${alias}.iface) ${op} (${sql})`)
|
|
for (const t of tuples) {
|
|
params.push(t.serverId, t.iface)
|
|
}
|
|
}
|
|
|
|
function factWhere(
|
|
alias: string,
|
|
grain: "hour" | "day",
|
|
ctx: FilterCtx,
|
|
scope: FactScope = "unique",
|
|
): { sql: string; params: unknown[] } {
|
|
const params: unknown[] = []
|
|
const parts: string[] = []
|
|
if (grain === "hour") {
|
|
params.push(ctx.fromIso, ctx.toIso)
|
|
parts.push(`${alias}.bucket_at >= ? AND ${alias}.bucket_at < ?`)
|
|
} else {
|
|
params.push(ctx.fromDay, ctx.toDayExclusive)
|
|
parts.push(`${alias}.day >= ? AND ${alias}.day < ?`)
|
|
}
|
|
if (ctx.serverId != null) {
|
|
parts.push(`${alias}.server_id = ?`)
|
|
params.push(ctx.serverId)
|
|
}
|
|
if (ctx.country) {
|
|
parts.push(`${alias}.country = ?`)
|
|
params.push(ctx.country.toUpperCase())
|
|
}
|
|
if (ctx.service) {
|
|
parts.push(`${alias}.service = ?`)
|
|
params.push(ctx.service)
|
|
}
|
|
if (ctx.asn != null) {
|
|
parts.push(`${alias}.asn = ?`)
|
|
params.push(ctx.asn)
|
|
}
|
|
parts.push(`${alias}.iface NOT IN ('0', '—', '__unknown__', 'wg-flow', '')`)
|
|
|
|
if (scope === "wan") {
|
|
pushIfaceTuples(parts, params, alias, ctx.wanIfaces, "IN")
|
|
return { sql: parts.join(" AND "), params }
|
|
}
|
|
if (scope === "overlay") {
|
|
pushIfaceTuples(parts, params, alias, ctx.overlayIfaces, "IN")
|
|
return { sql: parts.join(" AND "), params }
|
|
}
|
|
|
|
if (ctx.iface) {
|
|
const aliases = ifaceFilterAliases(ctx.iface, ctx.serverId)
|
|
if (aliases.length <= 1) {
|
|
parts.push(`${alias}.iface = ?`)
|
|
params.push(aliases[0] ?? ctx.iface)
|
|
} else {
|
|
parts.push(`${alias}.iface IN (${aliases.map(() => "?").join(", ")})`)
|
|
params.push(...aliases)
|
|
}
|
|
return { sql: parts.join(" AND "), params }
|
|
}
|
|
if (ctx.userIfaces) {
|
|
pushIfaceTuples(parts, params, alias, ctx.userIfaces, "IN")
|
|
return { sql: parts.join(" AND "), params }
|
|
}
|
|
if (ctx.unboundOnly) {
|
|
parts.push("FALSE")
|
|
return { sql: parts.join(" AND "), params }
|
|
}
|
|
pushIfaceTuples(parts, params, alias, ctx.boundIfaces, "IN")
|
|
if (ctx.excludeServerIds.length) {
|
|
parts.push(`${alias}.server_id NOT IN (${ctx.excludeServerIds.map(() => "?").join(", ")})`)
|
|
params.push(...ctx.excludeServerIds)
|
|
}
|
|
return { sql: parts.join(" AND "), params }
|
|
}
|
|
|
|
function emptyDto(period: ParsedPeriod): StatisticsDto {
|
|
return {
|
|
from: period.fromIso,
|
|
to: period.toIso,
|
|
grain: period.grain,
|
|
kpis: {
|
|
bytes: 0,
|
|
packets: 0,
|
|
avgBps: 0,
|
|
users: 0,
|
|
servers: 0,
|
|
ifaces: 0,
|
|
topCountry: "—",
|
|
topService: "—",
|
|
},
|
|
series: [],
|
|
users: [],
|
|
servers: [],
|
|
interfaces: [],
|
|
countries: [],
|
|
services: [],
|
|
asns: [],
|
|
}
|
|
}
|
|
|
|
function toBreakdown(
|
|
rows: Array<{ id: string; label: string; bytes: number; packets: number }>,
|
|
totalBytes: number,
|
|
windowSec: number,
|
|
): StatisticsBreakdownRow[] {
|
|
const denom = totalBytes || 1
|
|
return rows
|
|
.sort((a, b) => b.bytes - a.bytes)
|
|
.slice(0, TOP_N)
|
|
.map((r) => ({
|
|
id: r.id,
|
|
label: r.label,
|
|
bytes: r.bytes,
|
|
packets: r.packets,
|
|
bps: (r.bytes * 8) / windowSec,
|
|
percent: (r.bytes / denom) * 100,
|
|
}))
|
|
}
|
|
|
|
interface UserBindTuple {
|
|
userId: string
|
|
serverId: number
|
|
iface: string
|
|
}
|
|
|
|
async function loadBindUserTuples(): Promise<UserBindTuple[]> {
|
|
const binds = await db.select().from(userInterfaceBindings)
|
|
const seen = new Set<string>()
|
|
const out: UserBindTuple[] = []
|
|
for (const b of binds) {
|
|
for (const iface of factIfaceAliases(b.interfaceName, b.serverId)) {
|
|
const k = `${b.userId}\0${b.serverId}\0${iface}`
|
|
if (seen.has(k)) continue
|
|
seen.add(k)
|
|
out.push({ userId: b.userId, serverId: b.serverId, iface })
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
function uniqueBoundIfaces(tuples: UserBindTuple[]): Array<{ serverId: number; iface: string }> {
|
|
const seen = new Set<string>()
|
|
const out: Array<{ serverId: number; iface: string }> = []
|
|
for (const t of tuples) {
|
|
const k = `${t.serverId}\0${t.iface}`
|
|
if (seen.has(k)) continue
|
|
seen.add(k)
|
|
out.push({ serverId: t.serverId, iface: t.iface })
|
|
}
|
|
return out
|
|
}
|
|
|
|
async function resolveUserIfaces(userId?: string): Promise<Array<{ serverId: number; iface: string }> | null> {
|
|
if (!userId || userId === STATISTICS_UNBOUND_USER_ID) return null
|
|
const binds = await db.select().from(userInterfaceBindings).where(eq(userInterfaceBindings.userId, userId))
|
|
return expandBindingIfaces(binds.map((b) => ({ serverId: b.serverId, iface: b.interfaceName })))
|
|
}
|
|
|
|
function userBindJoinSql(tuples: UserBindTuple[]): { sql: string; params: unknown[] } {
|
|
const values = tuples.map(() => "(?::text, ?::int, ?::text)").join(", ")
|
|
const params = tuples.flatMap((t) => [t.userId, t.serverId, t.iface])
|
|
return {
|
|
sql: `JOIN (VALUES ${values}) AS b(user_id, server_id, iface) ON b.server_id = f.server_id AND b.iface = f.iface`,
|
|
params,
|
|
}
|
|
}
|
|
|
|
function expandIfaceTuples(
|
|
items: Array<{ serverId: number; iface: string }>,
|
|
): Array<{ serverId: number; iface: string }> {
|
|
const seen = new Set<string>()
|
|
const out: Array<{ serverId: number; iface: string }> = []
|
|
for (const t of items) {
|
|
for (const iface of factIfaceAliases(t.iface, t.serverId)) {
|
|
const k = `${t.serverId}\0${iface}`
|
|
if (seen.has(k)) continue
|
|
seen.add(k)
|
|
out.push({ serverId: t.serverId, iface })
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
async function loadPayloadScope(serverId?: number): Promise<{
|
|
overlayIfaces: Array<{ serverId: number; iface: string }>
|
|
wanIfaces: Array<{ serverId: number; iface: string }>
|
|
excludeServerIds: number[]
|
|
topo: FlowTopology
|
|
}> {
|
|
const topo = await loadFlowTopology()
|
|
const catalog = await getServerCatalog()
|
|
await warmIfaceCache(catalog.list.map((s) => s.id))
|
|
const overlayRaw: Array<{ serverId: number; iface: string }> = []
|
|
const wanRaw: Array<{ serverId: number; iface: string }> = []
|
|
for (const s of catalog.list) {
|
|
if (serverId != null && s.id !== serverId) continue
|
|
const wanSet = topo.wanIfaces.get(s.id)
|
|
const wanNames = wanSet && wanSet.size > 0
|
|
? [...wanSet]
|
|
: s.type === "home-router" ? [] : ["ether1"]
|
|
for (const name of wanNames) wanRaw.push({ serverId: s.id, iface: name })
|
|
const names = new Set(listCachedIfaceNames(s.id))
|
|
for (const name of topo.tunnelIfaces?.get(s.id) ?? []) names.add(name)
|
|
for (const name of names) {
|
|
if (isOverlayTunnelIface(topo, s.id, name)) overlayRaw.push({ serverId: s.id, iface: name })
|
|
}
|
|
}
|
|
return {
|
|
overlayIfaces: expandIfaceTuples(overlayRaw),
|
|
wanIfaces: expandIfaceTuples(wanRaw),
|
|
excludeServerIds: serverId != null
|
|
? []
|
|
: catalog.list.filter((s) => s.type === "exit-node").map((s) => s.id),
|
|
topo,
|
|
}
|
|
}
|
|
|
|
async function buildFilterCtx(query: StatisticsQuery, period: ParsedPeriod): Promise<FilterCtx | null> {
|
|
const bindTuples = await loadBindUserTuples()
|
|
const boundIfaces = uniqueBoundIfaces(bindTuples)
|
|
const unboundOnly = query.userId === STATISTICS_UNBOUND_USER_ID
|
|
const userIfaces = unboundOnly ? null : await resolveUserIfaces(query.userId)
|
|
if (userIfaces && userIfaces.length === 0) return null
|
|
if (unboundOnly) return null
|
|
const scope = await loadPayloadScope(query.serverId)
|
|
return {
|
|
...period,
|
|
serverId: query.serverId,
|
|
iface: query.iface,
|
|
country: query.country,
|
|
service: query.service,
|
|
asn: query.asn,
|
|
planes: query.planes ?? "unique",
|
|
userIfaces,
|
|
unboundOnly,
|
|
boundIfaces,
|
|
overlayIfaces: scope.overlayIfaces,
|
|
wanIfaces: scope.wanIfaces,
|
|
excludeServerIds: scope.excludeServerIds,
|
|
topo: scope.topo,
|
|
}
|
|
}
|
|
|
|
export async function getStatistics(query: StatisticsQuery): Promise<StatisticsDto> {
|
|
const period = parseStatisticsPeriod(query.from, query.to)
|
|
if (!period) return emptyDto({
|
|
fromIso: query.from,
|
|
toIso: query.to,
|
|
fromDay: query.from.slice(0, 10),
|
|
toDayExclusive: query.to.slice(0, 10),
|
|
grain: "day",
|
|
windowSec: 1,
|
|
})
|
|
|
|
await warmBindingIfaceCache()
|
|
if (query.serverId) await warmIfaceCache([query.serverId])
|
|
const bindTuples = await loadBindUserTuples()
|
|
const ctx = await buildFilterCtx(query, period)
|
|
if (!ctx) return emptyDto(period)
|
|
|
|
const table = period.grain === "hour" ? "flow_hour_facts" : "flow_daily_facts"
|
|
const timeCol = period.grain === "hour" ? "bucket_at" : "day"
|
|
const where = factWhere("f", period.grain, ctx)
|
|
|
|
const totals = await dbAll<{ bytes: number; packets: number; servers: number }>(`
|
|
SELECT
|
|
COALESCE(SUM(f.bytes), 0) AS bytes,
|
|
COALESCE(SUM(f.packets), 0) AS packets,
|
|
COUNT(DISTINCT f.server_id)::int AS servers
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
`, where.params)
|
|
|
|
const bytes = Number(totals[0]?.bytes) || 0
|
|
const packets = Number(totals[0]?.packets) || 0
|
|
const serverCount = Number(totals[0]?.servers) || 0
|
|
|
|
const seriesRows = await dbAll<{ t: string; bytes: number }>(`
|
|
SELECT ${timeCol}::text AS t, SUM(f.bytes) AS bytes
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
GROUP BY ${timeCol}
|
|
ORDER BY ${timeCol}
|
|
`, where.params)
|
|
|
|
const countryRows = await dbAll<{ id: string; bytes: number; packets: number }>(`
|
|
SELECT f.country AS id, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
GROUP BY f.country
|
|
`, where.params)
|
|
|
|
const serviceRows = await dbAll<{ id: string; bytes: number; packets: number }>(`
|
|
SELECT f.service AS id, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
GROUP BY f.service
|
|
`, where.params)
|
|
|
|
const asnRows = await dbAll<{ id: number; bytes: number; packets: number }>(`
|
|
SELECT f.asn AS id, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
GROUP BY f.asn
|
|
`, where.params)
|
|
|
|
const serverRows = await dbAll<{ id: number; bytes: number; packets: number }>(`
|
|
SELECT f.server_id AS id, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
GROUP BY f.server_id
|
|
`, where.params)
|
|
|
|
const ifaceRowsRaw = await dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
|
|
SELECT f.server_id AS "serverId", f.iface AS iface, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${where.sql}
|
|
GROUP BY f.server_id, f.iface
|
|
`, where.params)
|
|
await warmIfaceCache(ifaceRowsRaw.filter((r) => looksLikeIfIndex(r.iface)).map((r) => r.serverId))
|
|
const ifaceRows = collapseServerIfaceRows(ifaceRowsRaw).filter((r) => {
|
|
if (isJunkFactIface(r.iface) || isDashDisplayIface(r.iface)) return false
|
|
if (ctx.iface) return true
|
|
if (ctx.topo && isOverlayTunnelIface(ctx.topo, r.serverId, r.iface)) return false
|
|
if (ctx.topo && isWanFactIface(ctx.topo, r.serverId, r.iface)) return false
|
|
return true
|
|
})
|
|
const ifaceCount = ifaceRows.length
|
|
|
|
let dupeIfaceRows: Array<{ serverId: number; iface: string; bytes: number; packets: number; kind: "wan" | "overlay" }> = []
|
|
if (ctx.planes === "all" && !ctx.iface) {
|
|
const wanWhere = factWhere("f", period.grain, ctx, "wan")
|
|
const overlayWhere = factWhere("f", period.grain, ctx, "overlay")
|
|
const [wanRaw, overlayRaw] = await Promise.all([
|
|
dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
|
|
SELECT f.server_id AS "serverId", f.iface AS iface, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${wanWhere.sql}
|
|
GROUP BY f.server_id, f.iface
|
|
`, wanWhere.params),
|
|
dbAll<{ serverId: number; iface: string; bytes: number; packets: number }>(`
|
|
SELECT f.server_id AS "serverId", f.iface AS iface, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
WHERE ${overlayWhere.sql}
|
|
GROUP BY f.server_id, f.iface
|
|
`, overlayWhere.params),
|
|
])
|
|
await warmIfaceCache([
|
|
...wanRaw.filter((r) => looksLikeIfIndex(r.iface)).map((r) => r.serverId),
|
|
...overlayRaw.filter((r) => looksLikeIfIndex(r.iface)).map((r) => r.serverId),
|
|
])
|
|
const seen = new Set(ifaceRows.map((r) => `${r.serverId}:${r.iface}`))
|
|
for (const r of collapseServerIfaceRows(wanRaw)) {
|
|
if (isJunkFactIface(r.iface) || isDashDisplayIface(r.iface)) continue
|
|
const key = `${r.serverId}:${r.iface}`
|
|
if (seen.has(key)) continue
|
|
seen.add(key)
|
|
dupeIfaceRows.push({ ...r, kind: "wan" })
|
|
}
|
|
for (const r of collapseServerIfaceRows(overlayRaw)) {
|
|
if (isJunkFactIface(r.iface) || isDashDisplayIface(r.iface)) continue
|
|
const key = `${r.serverId}:${r.iface}`
|
|
if (seen.has(key)) continue
|
|
seen.add(key)
|
|
dupeIfaceRows.push({ ...r, kind: "overlay" })
|
|
}
|
|
}
|
|
|
|
let userRows: Array<{ id: string; bytes: number; packets: number }> = []
|
|
if (bindTuples.length && !ctx.unboundOnly) {
|
|
const join = userBindJoinSql(bindTuples)
|
|
userRows = await dbAll<{ id: string; bytes: number; packets: number }>(`
|
|
SELECT b.user_id AS id, SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
${join.sql}
|
|
WHERE ${where.sql}
|
|
GROUP BY b.user_id
|
|
`, [...join.params, ...where.params])
|
|
}
|
|
|
|
const serverNames = new Map<number, string>()
|
|
const allServers = await db.select({ id: servers.id, name: servers.name, host: servers.host }).from(servers)
|
|
for (const s of allServers) serverNames.set(s.id, s.name || s.host)
|
|
|
|
const userNames = new Map<string, string>()
|
|
const allUsers = await db.select({ id: appUsers.id, name: appUsers.name, login: appUsers.login }).from(appUsers)
|
|
for (const u of allUsers) userNames.set(u.id, u.name || u.login)
|
|
|
|
const asnHolders = new Map<number, string>()
|
|
const asnMeta = await db.select({ asn: flowAsnMeta.asn, holder: flowAsnMeta.holder }).from(flowAsnMeta)
|
|
for (const a of asnMeta) asnHolders.set(a.asn, a.holder)
|
|
|
|
const countries = toBreakdown(
|
|
countryRows.map((r) => ({
|
|
id: r.id,
|
|
label: r.id === "XX" ? "Неизвестно" : r.id,
|
|
bytes: Number(r.bytes) || 0,
|
|
packets: Number(r.packets) || 0,
|
|
})),
|
|
bytes,
|
|
period.windowSec,
|
|
)
|
|
const services = toBreakdown(
|
|
serviceRows.map((r) => ({
|
|
id: r.id,
|
|
label: r.id,
|
|
bytes: Number(r.bytes) || 0,
|
|
packets: Number(r.packets) || 0,
|
|
})),
|
|
bytes,
|
|
period.windowSec,
|
|
)
|
|
const asns = toBreakdown(
|
|
asnRows.map((r) => {
|
|
const id = Number(r.id) || 0
|
|
const holder = asnHolders.get(id)
|
|
return {
|
|
id: String(id),
|
|
label: id === 0 ? "other" : holder ? `AS${id} · ${holder}` : `AS${id}`,
|
|
bytes: Number(r.bytes) || 0,
|
|
packets: Number(r.packets) || 0,
|
|
}
|
|
}),
|
|
bytes,
|
|
period.windowSec,
|
|
)
|
|
const serverBreakdown = toBreakdown(
|
|
serverRows.map((r) => ({
|
|
id: String(r.id),
|
|
label: serverNames.get(r.id) || String(r.id),
|
|
bytes: Number(r.bytes) || 0,
|
|
packets: Number(r.packets) || 0,
|
|
})),
|
|
bytes,
|
|
period.windowSec,
|
|
)
|
|
const uniqueInterfaces = toBreakdown(
|
|
ifaceRows.map((r) => {
|
|
const serverName = serverNames.get(r.serverId) || String(r.serverId)
|
|
const wan = ctx.topo ? isWanFactIface(ctx.topo, r.serverId, r.iface) : false
|
|
return {
|
|
id: `${r.serverId}:${r.iface}`,
|
|
label: wan ? wanIfaceLabel(serverName, r.iface) : `${serverName} · ${r.iface}`,
|
|
bytes: Number(r.bytes) || 0,
|
|
packets: Number(r.packets) || 0,
|
|
}
|
|
}),
|
|
bytes,
|
|
period.windowSec,
|
|
)
|
|
const dupeInterfaces: StatisticsBreakdownRow[] = dupeIfaceRows.map((r) => {
|
|
const serverName = serverNames.get(r.serverId) || String(r.serverId)
|
|
const rowBytes = Number(r.bytes) || 0
|
|
const rowPackets = Number(r.packets) || 0
|
|
return {
|
|
id: `${r.serverId}:${r.iface}`,
|
|
label: r.kind === "wan" ? wanIfaceLabel(serverName, r.iface) : overlayDupLabel(serverName, r.iface),
|
|
bytes: rowBytes,
|
|
packets: rowPackets,
|
|
bps: (rowBytes * 8) / period.windowSec,
|
|
percent: 0,
|
|
}
|
|
})
|
|
const interfaces = [...uniqueInterfaces, ...dupeInterfaces]
|
|
const matchedUsers = toBreakdown(
|
|
userRows.map((r) => ({
|
|
id: r.id,
|
|
label: userNames.get(r.id) || r.id,
|
|
bytes: Number(r.bytes) || 0,
|
|
packets: Number(r.packets) || 0,
|
|
})),
|
|
bytes,
|
|
period.windowSec,
|
|
)
|
|
|
|
const users = [...matchedUsers]
|
|
|
|
return {
|
|
from: period.fromIso,
|
|
to: period.toIso,
|
|
grain: period.grain,
|
|
kpis: {
|
|
bytes,
|
|
packets,
|
|
avgBps: (bytes * 8) / period.windowSec,
|
|
users: matchedUsers.length,
|
|
servers: serverCount,
|
|
ifaces: ifaceCount,
|
|
topCountry: countries[0]?.label || "—",
|
|
topService: services[0]?.label || "—",
|
|
},
|
|
series: seriesRows.map((r) => ({ t: r.t, bytes: Number(r.bytes) || 0 })),
|
|
users,
|
|
servers: serverBreakdown,
|
|
interfaces,
|
|
countries,
|
|
services,
|
|
asns,
|
|
}
|
|
}
|
|
|
|
function dimSql(dim: StatisticsPivotDim, factAlias: string, bindAlias: string): string {
|
|
if (dim === "country") return `${factAlias}.country`
|
|
if (dim === "service") return `${factAlias}.service`
|
|
if (dim === "asn") return `${factAlias}.asn::text`
|
|
if (dim === "server") return `${factAlias}.server_id::text`
|
|
if (dim === "iface") return `(${factAlias}.server_id::text || ':' || ${factAlias}.iface)`
|
|
return `${bindAlias}.user_id`
|
|
}
|
|
|
|
function emptyPivot(query: StatisticsPivotQuery): StatisticsPivotDto {
|
|
return {
|
|
rowDim: query.row,
|
|
colDim: query.col,
|
|
metric: query.metric,
|
|
columns: [],
|
|
rows: [],
|
|
otherBytes: 0,
|
|
}
|
|
}
|
|
|
|
export function pivotDimsConflict(row: StatisticsPivotDim, col: StatisticsPivotDim): boolean {
|
|
return row === col
|
|
}
|
|
|
|
export async function getStatisticsPivot(query: StatisticsPivotQuery): Promise<StatisticsPivotDto> {
|
|
if (pivotDimsConflict(query.row, query.col)) return emptyPivot(query)
|
|
const period = parseStatisticsPeriod(query.from, query.to)
|
|
if (!period) return emptyPivot(query)
|
|
await warmBindingIfaceCache()
|
|
if (query.serverId) await warmIfaceCache([query.serverId])
|
|
const bindTuples = await loadBindUserTuples()
|
|
const ctx = await buildFilterCtx(query, period)
|
|
if (!ctx) return emptyPivot(query)
|
|
const needsUser = query.row === "user" || query.col === "user"
|
|
if (needsUser && bindTuples.length === 0) return emptyPivot(query)
|
|
|
|
const table = period.grain === "hour" ? "flow_hour_facts" : "flow_daily_facts"
|
|
const where = factWhere("f", period.grain, ctx)
|
|
const rowExpr = dimSql(query.row, "f", "b")
|
|
const colExpr = dimSql(query.col, "f", "b")
|
|
const join = needsUser ? userBindJoinSql(bindTuples) : { sql: "", params: [] as unknown[] }
|
|
|
|
const raw = await dbAll<{ row_id: string; col_id: string; bytes: number; packets: number }>(`
|
|
SELECT ${rowExpr} AS row_id, ${colExpr} AS col_id,
|
|
SUM(f.bytes) AS bytes, SUM(f.packets) AS packets
|
|
FROM ${table} f
|
|
${join.sql}
|
|
WHERE ${where.sql}
|
|
GROUP BY 1, 2
|
|
`, [...join.params, ...where.params])
|
|
|
|
if (query.row === "iface" || query.col === "iface") {
|
|
const ifaceServerIds: number[] = []
|
|
for (const r of raw) {
|
|
for (const dim of [query.row, query.col] as const) {
|
|
if (dim !== "iface") continue
|
|
const id = dim === query.row ? String(r.row_id ?? "") : String(r.col_id ?? "")
|
|
const colon = id.indexOf(":")
|
|
if (colon < 0) continue
|
|
const sid = Number(id.slice(0, colon))
|
|
if (looksLikeIfIndex(id.slice(colon + 1)) && Number.isFinite(sid)) ifaceServerIds.push(sid)
|
|
}
|
|
}
|
|
await warmIfaceCache(ifaceServerIds)
|
|
for (const r of raw) {
|
|
if (query.row === "iface") r.row_id = canonicalIfaceDimId(String(r.row_id ?? ""))
|
|
if (query.col === "iface") r.col_id = canonicalIfaceDimId(String(r.col_id ?? ""))
|
|
}
|
|
}
|
|
|
|
const metric = query.metric
|
|
type Acc = { bytes: number; packets: number }
|
|
const cell = new Map<string, Map<string, Acc>>()
|
|
const colTotals = new Map<string, number>()
|
|
for (const r of raw) {
|
|
const rid = String(r.row_id ?? "")
|
|
const cid = String(r.col_id ?? "")
|
|
const acc: Acc = { bytes: Number(r.bytes) || 0, packets: Number(r.packets) || 0 }
|
|
const val = metric === "packets" ? acc.packets : acc.bytes
|
|
let rowMap = cell.get(rid)
|
|
if (!rowMap) {
|
|
rowMap = new Map()
|
|
cell.set(rid, rowMap)
|
|
}
|
|
const prev = rowMap.get(cid)
|
|
if (prev) {
|
|
prev.bytes += acc.bytes
|
|
prev.packets += acc.packets
|
|
} else {
|
|
rowMap.set(cid, acc)
|
|
}
|
|
colTotals.set(cid, (colTotals.get(cid) ?? 0) + val)
|
|
}
|
|
|
|
const topCols = [...colTotals.entries()]
|
|
.sort((a, b) => b[1] - a[1])
|
|
.slice(0, PIVOT_COL_CAP)
|
|
.map(([id]) => id)
|
|
const topColSet = new Set(topCols)
|
|
const folded = new Map<string, Map<string, number>>()
|
|
const foldedColTotals = new Map<string, number>()
|
|
let otherBytes = 0
|
|
for (const [rid, cols] of cell) {
|
|
const rowMap = new Map<string, number>()
|
|
for (const [cid, acc] of cols) {
|
|
const val = metric === "packets" ? acc.packets : acc.bytes
|
|
const dest = topColSet.has(cid) ? cid : PIVOT_OTHER_ID
|
|
if (dest === PIVOT_OTHER_ID) otherBytes += val
|
|
rowMap.set(dest, (rowMap.get(dest) ?? 0) + val)
|
|
foldedColTotals.set(dest, (foldedColTotals.get(dest) ?? 0) + val)
|
|
}
|
|
folded.set(rid, rowMap)
|
|
}
|
|
|
|
const rowTotals = new Map<string, number>()
|
|
for (const [rid, cols] of folded) {
|
|
let t = 0
|
|
for (const v of cols.values()) t += v
|
|
rowTotals.set(rid, t)
|
|
}
|
|
const topRows = [...rowTotals.entries()]
|
|
.sort((a, b) => b[1] - a[1])
|
|
.slice(0, PIVOT_ROW_CAP)
|
|
.map(([id]) => id)
|
|
const topRowSet = new Set(topRows)
|
|
const finalRows = new Map<string, Map<string, number>>()
|
|
const finalRowTotals = new Map<string, number>()
|
|
for (const [rid, cols] of folded) {
|
|
const dest = topRowSet.has(rid) ? rid : PIVOT_OTHER_ID
|
|
if (dest === PIVOT_OTHER_ID) {
|
|
for (const [cid, v] of cols) {
|
|
if (cid !== PIVOT_OTHER_ID) otherBytes += v
|
|
}
|
|
}
|
|
let rowMap = finalRows.get(dest)
|
|
if (!rowMap) {
|
|
rowMap = new Map()
|
|
finalRows.set(dest, rowMap)
|
|
}
|
|
for (const [cid, v] of cols) {
|
|
rowMap.set(cid, (rowMap.get(cid) ?? 0) + v)
|
|
}
|
|
}
|
|
for (const [rid, cols] of finalRows) {
|
|
let t = 0
|
|
for (const v of cols.values()) t += v
|
|
finalRowTotals.set(rid, t)
|
|
}
|
|
|
|
const colIds = [...topCols]
|
|
if (foldedColTotals.has(PIVOT_OTHER_ID)) colIds.push(PIVOT_OTHER_ID)
|
|
const rowIds = [...topRows]
|
|
if (finalRows.has(PIVOT_OTHER_ID) && !topRowSet.has(PIVOT_OTHER_ID)) rowIds.push(PIVOT_OTHER_ID)
|
|
|
|
const labels = await loadPivotLabels(query.row, query.col, rowIds, colIds)
|
|
|
|
return {
|
|
rowDim: query.row,
|
|
colDim: query.col,
|
|
metric,
|
|
columns: colIds.map((id) => ({
|
|
id,
|
|
label: labels.col.get(id) ?? (id === PIVOT_OTHER_ID ? "Прочие" : id),
|
|
total: foldedColTotals.get(id) ?? 0,
|
|
})),
|
|
rows: rowIds.map((id) => {
|
|
const cols = finalRows.get(id) ?? new Map()
|
|
const cells: Record<string, number> = {}
|
|
for (const cid of colIds) cells[cid] = cols.get(cid) ?? 0
|
|
return {
|
|
id,
|
|
label: labels.row.get(id) ?? (id === PIVOT_OTHER_ID ? "Прочие" : id),
|
|
total: finalRowTotals.get(id) ?? 0,
|
|
cells,
|
|
}
|
|
}),
|
|
otherBytes,
|
|
}
|
|
}
|
|
|
|
async function loadPivotLabels(
|
|
rowDim: StatisticsPivotDim,
|
|
colDim: StatisticsPivotDim,
|
|
rowIds: string[],
|
|
colIds: string[],
|
|
): Promise<{ row: Map<string, string>; col: Map<string, string> }> {
|
|
const topo = await loadFlowTopology()
|
|
const serverNames = new Map<string, string>()
|
|
const allServers = await db.select({ id: servers.id, name: servers.name, host: servers.host }).from(servers)
|
|
for (const s of allServers) serverNames.set(String(s.id), s.name || s.host)
|
|
const userNames = new Map<string, string>()
|
|
const allUsers = await db.select({ id: appUsers.id, name: appUsers.name, login: appUsers.login }).from(appUsers)
|
|
for (const u of allUsers) userNames.set(u.id, u.name || u.login)
|
|
const asnHolders = new Map<string, string>()
|
|
const asnMeta = await db.select({ asn: flowAsnMeta.asn, holder: flowAsnMeta.holder }).from(flowAsnMeta)
|
|
for (const a of asnMeta) asnHolders.set(String(a.asn), a.holder)
|
|
|
|
function label(dim: StatisticsPivotDim, id: string): string {
|
|
if (id === PIVOT_OTHER_ID) return "Прочие"
|
|
if (dim === "country") return id === "XX" ? "Неизвестно" : id
|
|
if (dim === "server") return serverNames.get(id) || id
|
|
if (dim === "user") return userNames.get(id) || id
|
|
if (dim === "asn") {
|
|
if (id === "0") return "other"
|
|
const holder = asnHolders.get(id)
|
|
return holder ? `AS${id} · ${holder}` : `AS${id}`
|
|
}
|
|
if (dim === "iface") {
|
|
const colon = id.indexOf(":")
|
|
if (colon < 0) return id
|
|
const sid = id.slice(0, colon)
|
|
const iface = id.slice(colon + 1)
|
|
const sidNum = Number(sid)
|
|
const name = Number.isFinite(sidNum) ? displayFactIface(sidNum, iface) : iface
|
|
const serverName = serverNames.get(sid) || sid
|
|
if (Number.isFinite(sidNum) && isWanFactIface(topo, sidNum, name)) {
|
|
return wanIfaceLabel(serverName, name)
|
|
}
|
|
if (Number.isFinite(sidNum) && isOverlayTunnelIface(topo, sidNum, name)) {
|
|
return overlayDupLabel(serverName, name)
|
|
}
|
|
return `${serverName} · ${name}`
|
|
}
|
|
return id
|
|
}
|
|
|
|
const row = new Map<string, string>()
|
|
const col = new Map<string, string>()
|
|
for (const id of rowIds) row.set(id, label(rowDim, id))
|
|
for (const id of colIds) col.set(id, label(colDim, id))
|
|
return { row, col }
|
|
}
|