feat: implement internet path functionality with backend support
Added internet path settings and snapshot management to the application. This includes new database tables for internet path settings and snapshots, API routes for fetching and managing internet path data, and integration into the dashboard and data collection pages. Enhanced the scheduler to support internet path jobs, ensuring regular data collection and updates. Updated relevant types and interfaces to accommodate the new functionality.
This commit is contained in:
Denozordec
@@ -0,0 +1,312 @@
|
||||
import { and, asc, eq, lt } from "drizzle-orm"
|
||||
import { db } from "../db/index.js"
|
||||
import {
|
||||
filterRules,
|
||||
internetPathSettings,
|
||||
internetPathSnapshots,
|
||||
servers,
|
||||
uptimeSpeedProbes,
|
||||
} from "../db/schema.js"
|
||||
import { listServersRead } from "../modules/servers/service/servers-service.js"
|
||||
import { MikrotikClient } from "./mikrotik.js"
|
||||
import type { InternetPathRunSnapshot } from "../types/scheduler-run-snapshot.js"
|
||||
import { SCHEDULER_RUN_SNAPSHOT_VERSION } from "../types/scheduler-run-snapshot.js"
|
||||
|
||||
const INTERNET_TARGET = "1.1.1.1"
|
||||
let collecting = false
|
||||
|
||||
function isTrue(v: unknown): boolean {
|
||||
const s = String(v ?? "").trim().toLowerCase()
|
||||
return s === "true" || s === "yes"
|
||||
}
|
||||
|
||||
function toIp(raw: string | null | undefined): string | null {
|
||||
const v = String(raw ?? "").trim()
|
||||
if (!v) return null
|
||||
return v.split("/")[0]?.trim() ?? null
|
||||
}
|
||||
|
||||
function norm(v: string | null | undefined): string {
|
||||
return String(v ?? "").trim().toLowerCase()
|
||||
}
|
||||
|
||||
function getSettingsRow() {
|
||||
const row = db.select().from(internetPathSettings).where(eq(internetPathSettings.id, 1)).limit(1).all()[0]
|
||||
if (row) return row
|
||||
const now = new Date().toISOString()
|
||||
db.insert(internetPathSettings).values({
|
||||
id: 1,
|
||||
enabled: true,
|
||||
intervalSec: 300,
|
||||
retentionDays: 14,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
}).run()
|
||||
return db.select().from(internetPathSettings).where(eq(internetPathSettings.id, 1)).limit(1).all()[0]
|
||||
}
|
||||
|
||||
function cleanupSnapshots(retentionDays: number) {
|
||||
const cutoff = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000).toISOString()
|
||||
db.delete(internetPathSnapshots).where(lt(internetPathSnapshots.sampledAt, cutoff)).run()
|
||||
}
|
||||
|
||||
function buildRulesets() {
|
||||
const enabled = db.select().from(servers).where(eq(servers.enabled, true)).all()
|
||||
const rules = db.select().from(filterRules).orderBy(asc(filterRules.serverId), asc(filterRules.sortOrder)).all()
|
||||
return enabled.map((s) => ({
|
||||
serverId: String(s.id),
|
||||
rules: rules
|
||||
.filter((r) => r.serverId === s.id)
|
||||
.map((r) => ({
|
||||
id: String(r.id),
|
||||
community: r.community,
|
||||
communityName: r.communityName ?? undefined,
|
||||
action: r.action,
|
||||
gateway: r.gateway,
|
||||
gatewayTunnelId: r.gatewayTunnelId,
|
||||
description: r.description,
|
||||
})),
|
||||
}))
|
||||
}
|
||||
|
||||
async function readRouteLookup(serverId: number): Promise<{ gateway: string | null; routingMark: string | null }> {
|
||||
const row = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0]
|
||||
if (!row) return { gateway: null, routingMark: null }
|
||||
const client = MikrotikClient.fromServer(row)
|
||||
const routes = await client.get<Array<Record<string, string>>>("/ip/route").catch(() => [])
|
||||
const best = routes
|
||||
.filter((r) => String(r["dst-address"] ?? "").trim() === "0.0.0.0/0")
|
||||
.filter((r) => isTrue(r.active))
|
||||
.filter((r) => {
|
||||
const rt = String(r["routing-table"] ?? "").trim().toLowerCase()
|
||||
return rt === "" || rt === "main"
|
||||
})
|
||||
.filter((r) => String(r.disabled ?? "false").toLowerCase() !== "true")
|
||||
.filter((r) => String(r.inactive ?? "false").toLowerCase() !== "true")
|
||||
.sort((a, b) => Number(a.distance ?? 255) - Number(b.distance ?? 255))[0]
|
||||
return {
|
||||
gateway: String(best?.gateway ?? "").trim() || null,
|
||||
routingMark: String(best?.["routing-mark"] ?? "").trim() || null,
|
||||
}
|
||||
}
|
||||
|
||||
async function readWanRuntime(serverId: number) {
|
||||
const server = listServersRead().find((s) => Number(s.id) === serverId)
|
||||
const row = db.select().from(servers).where(eq(servers.id, serverId)).limit(1).all()[0]
|
||||
if (!server || !row) return null
|
||||
const client = MikrotikClient.fromServer(row)
|
||||
const [dhcpRaw, ipAddrs, routes] = await Promise.all([
|
||||
client.get<Array<Record<string, string>>>("/ip/dhcp-client").catch(() => []),
|
||||
client.get<Array<Record<string, string>>>("/ip/address").catch(() => []),
|
||||
client.get<Array<Record<string, string>>>("/ip/route").catch(() => []),
|
||||
])
|
||||
const defaultRoute = routes
|
||||
.filter((r) => String(r["dst-address"] ?? "").trim() === "0.0.0.0/0")
|
||||
.filter((r) => isTrue(r.active))
|
||||
.filter((r) => {
|
||||
const rt = String(r["routing-table"] ?? "").trim().toLowerCase()
|
||||
return rt === "" || rt === "main"
|
||||
})
|
||||
.filter((r) => String(r.disabled ?? "false").toLowerCase() !== "true")
|
||||
.filter((r) => String(r.inactive ?? "false").toLowerCase() !== "true")
|
||||
.sort((a, b) => Number(a.distance ?? 255) - Number(b.distance ?? 255))[0]
|
||||
const defaultGateway = String(defaultRoute?.gateway ?? "").trim() || null
|
||||
const immediateGw = String(defaultRoute?.["immediate-gw"] ?? "").trim() || null
|
||||
const defaultInterface =
|
||||
((immediateGw?.includes("%") ? immediateGw.split("%")[1]?.trim() : ""))
|
||||
|| String(defaultRoute?.interface ?? "").trim()
|
||||
|| String(
|
||||
dhcpRaw.find((d) => toIp(d.gateway) != null && toIp(d.gateway) === toIp(defaultGateway))?.interface ?? "",
|
||||
).trim()
|
||||
|| null
|
||||
const uplinks = (server.wanUplinks ?? []).map((w) => {
|
||||
const iface = String(w.iface ?? "").trim()
|
||||
const dhcp = dhcpRaw.find((d) => norm(d.interface) === norm(iface))
|
||||
const leasedIp =
|
||||
toIp(dhcp?.address)
|
||||
?? toIp(ipAddrs.find((a) => norm(a.interface) === norm(iface))?.address)
|
||||
?? null
|
||||
return {
|
||||
id: w.id,
|
||||
iface,
|
||||
name: w.name,
|
||||
isp: w.isp,
|
||||
configuredIp: w.ip,
|
||||
leasedIp,
|
||||
dhcpStatus: String(dhcp?.status ?? "").trim() || null,
|
||||
isDefault: defaultInterface != null && norm(defaultInterface) === norm(iface),
|
||||
}
|
||||
})
|
||||
return {
|
||||
defaultGateway,
|
||||
defaultInterface,
|
||||
uplinks,
|
||||
}
|
||||
}
|
||||
|
||||
function mapSpeedProbes() {
|
||||
const rows = db.select().from(uptimeSpeedProbes).orderBy(asc(uptimeSpeedProbes.sortOrder)).all()
|
||||
return rows.map((r) => ({
|
||||
id: r.id,
|
||||
srcServerId: String(r.srcServerId),
|
||||
dstServerId: String(r.dstServerId),
|
||||
srcInterface: r.srcInterface || "",
|
||||
dstInterface: r.dstInterface || "",
|
||||
protocol: r.protocol === "udp" ? "udp" : "tcp",
|
||||
direction: r.direction === "transmit" || r.direction === "receive" ? r.direction : "both",
|
||||
durationSec: String(Math.max(3, r.durationSec || 10)),
|
||||
enabled: r.enabled !== false,
|
||||
lastRunAt: r.lastRunAt ?? null,
|
||||
lastTxAvgMbps: r.lastTxAvgMbps ?? null,
|
||||
lastRxAvgMbps: r.lastRxAvgMbps ?? null,
|
||||
lastStatus: r.lastStatus ?? null,
|
||||
lastError: r.lastError ?? null,
|
||||
lastPingRttMs: r.lastPingRttMs ?? null,
|
||||
lastPingLossPct: r.lastPingLossPct ?? null,
|
||||
lastPingAt: r.lastPingAt ?? null,
|
||||
lastPingError: r.lastPingError ?? null,
|
||||
}))
|
||||
}
|
||||
|
||||
function parseInnerIps(comment: string): { localInnerIp: string; remoteInnerIp: string } {
|
||||
const local = comment.match(/address\s*=\s*([0-9.]+\/\d+)/)?.[1] ?? ""
|
||||
const remote = comment.match(/(?:network|gateway)\s*=\s*([0-9.]+\/\d+)/)?.[1] ?? ""
|
||||
return { localInnerIp: local, remoteInnerIp: remote }
|
||||
}
|
||||
|
||||
async function collectGreTunnels() {
|
||||
const enabled = db.select().from(servers).where(eq(servers.enabled, true)).all()
|
||||
const all = await Promise.all(enabled.map(async (srv) => {
|
||||
try {
|
||||
const client = MikrotikClient.fromServer(srv)
|
||||
const rows = await client.get<Array<Record<string, string>>>("/interface/gre")
|
||||
return rows.map((g, idx) => {
|
||||
const keepalive = String(g.keepalive ?? "0,0").split(",")
|
||||
const inner = parseInnerIps(String(g.comment ?? ""))
|
||||
return {
|
||||
id: String(g.name ?? g[".id"] ?? `gre-${srv.id}-${idx}`),
|
||||
name: String(g.name ?? `gre-${idx + 1}`),
|
||||
serverId: String(srv.id),
|
||||
localAddress: String(g["local-address"] ?? ""),
|
||||
remoteAddress: String(g["remote-address"] ?? ""),
|
||||
localInnerIp: inner.localInnerIp,
|
||||
remoteInnerIp: inner.remoteInnerIp,
|
||||
poolId: "live",
|
||||
ipsec: null,
|
||||
mtu: Number.parseInt(String(g.mtu ?? "1476"), 10) || 1476,
|
||||
keepaliveInterval: Number.parseInt(String(keepalive[0] ?? "0"), 10) || 0,
|
||||
keepaliveRetries: Number.parseInt(String(keepalive[1] ?? "0"), 10) || 0,
|
||||
dscp: "inherit" as const,
|
||||
clampTcpMss: String(g["clamp-tcp-mss"] ?? "true") !== "false",
|
||||
allowFastPath: String(g["allow-fast-path"] ?? "true") !== "false",
|
||||
comment: String(g.comment ?? ""),
|
||||
enabled: String(g.disabled ?? "false") !== "true",
|
||||
status:
|
||||
String(g.disabled ?? "false") === "true"
|
||||
? "down" as const
|
||||
: (String(g.running ?? "false") === "true" ? "up" as const : "degraded" as const),
|
||||
}
|
||||
})
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}))
|
||||
return all.flat()
|
||||
}
|
||||
|
||||
export function getInternetPathSettings() {
|
||||
return getSettingsRow()
|
||||
}
|
||||
|
||||
export function updateInternetPathSettings(patch: { enabled?: boolean; intervalSec?: number; retentionDays?: number }) {
|
||||
const prev = getSettingsRow()
|
||||
db.update(internetPathSettings).set({
|
||||
enabled: patch.enabled ?? prev.enabled,
|
||||
intervalSec: patch.intervalSec ?? prev.intervalSec,
|
||||
retentionDays: patch.retentionDays ?? prev.retentionDays,
|
||||
updatedAt: new Date().toISOString(),
|
||||
}).where(eq(internetPathSettings.id, 1)).run()
|
||||
return getSettingsRow()
|
||||
}
|
||||
|
||||
export function getLatestInternetPathSnapshot() {
|
||||
return db.select().from(internetPathSnapshots).orderBy(asc(internetPathSnapshots.id)).all().at(-1) ?? null
|
||||
}
|
||||
|
||||
export async function collectInternetPathSnapshotOnce(): Promise<InternetPathRunSnapshot> {
|
||||
const sampledAt = new Date().toISOString()
|
||||
if (collecting) {
|
||||
return {
|
||||
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
||||
job: "internet_path",
|
||||
sampledAt,
|
||||
homes: 0,
|
||||
snapshotSaved: false,
|
||||
}
|
||||
}
|
||||
collecting = true
|
||||
const started = Date.now()
|
||||
const settings = getSettingsRow()
|
||||
try {
|
||||
const serversRead = listServersRead()
|
||||
const homes = serversRead.filter((s) => s.type === "home-router")
|
||||
const [greTunnels, rulesets] = await Promise.all([collectGreTunnels(), Promise.resolve(buildRulesets())])
|
||||
const speedProbes = mapSpeedProbes()
|
||||
const routeLookupByServerId: Record<string, { gateway: string | null; routingMark: string | null }> = {}
|
||||
const wanRuntimeByHomeId: Record<string, unknown> = {}
|
||||
for (const h of homes) {
|
||||
routeLookupByServerId[String(h.id)] = await readRouteLookup(Number(h.id)).catch(() => ({ gateway: null, routingMark: null }))
|
||||
wanRuntimeByHomeId[String(h.id)] = await readWanRuntime(Number(h.id)).catch(() => null)
|
||||
}
|
||||
const payload = {
|
||||
sampledAt,
|
||||
internetTarget: INTERNET_TARGET,
|
||||
servers: serversRead,
|
||||
greTunnels,
|
||||
filtersRulesets: rulesets,
|
||||
speedProbes,
|
||||
routeLookupByServerId,
|
||||
wanRuntimeByHomeId,
|
||||
}
|
||||
db.insert(internetPathSnapshots).values({
|
||||
sampledAt,
|
||||
payloadJson: JSON.stringify(payload),
|
||||
}).run()
|
||||
cleanupSnapshots(Math.max(1, settings.retentionDays))
|
||||
db.update(internetPathSettings).set({
|
||||
lastCollectedAt: sampledAt,
|
||||
lastDurationMs: Date.now() - started,
|
||||
lastError: "",
|
||||
updatedAt: sampledAt,
|
||||
}).where(eq(internetPathSettings.id, 1)).run()
|
||||
return {
|
||||
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
||||
job: "internet_path",
|
||||
sampledAt,
|
||||
homes: homes.length,
|
||||
snapshotSaved: true,
|
||||
}
|
||||
} catch (e) {
|
||||
const msg = e instanceof Error ? e.message : String(e)
|
||||
db.update(internetPathSettings).set({
|
||||
lastCollectedAt: sampledAt,
|
||||
lastDurationMs: Date.now() - started,
|
||||
lastError: msg,
|
||||
updatedAt: sampledAt,
|
||||
}).where(eq(internetPathSettings.id, 1)).run()
|
||||
return {
|
||||
v: SCHEDULER_RUN_SNAPSHOT_VERSION,
|
||||
job: "internet_path",
|
||||
sampledAt,
|
||||
homes: 0,
|
||||
snapshotSaved: false,
|
||||
fatalError: msg,
|
||||
}
|
||||
} finally {
|
||||
collecting = false
|
||||
}
|
||||
}
|
||||
|
||||
export function isInternetPathCollecting(): boolean {
|
||||
return collecting
|
||||
}
|
||||
@@ -38,6 +38,10 @@ import {
|
||||
scheduleAlertEngineAfterDataCollectors,
|
||||
wireAlertEngineRunner,
|
||||
} from "./alert-collector-hooks.js"
|
||||
import {
|
||||
collectInternetPathSnapshotOnce,
|
||||
getInternetPathSettings,
|
||||
} from "./internet-path-collector.js"
|
||||
import {
|
||||
endSchedulerJob,
|
||||
isSchedulerJobRunning,
|
||||
@@ -51,6 +55,7 @@ export const JOB_KEYS = [
|
||||
"uptime_resources",
|
||||
"uptime_ping",
|
||||
"uptime_speed",
|
||||
"internet_path",
|
||||
"gre_bgp",
|
||||
"alert_engine",
|
||||
] as const
|
||||
@@ -111,6 +116,9 @@ async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
|
||||
case "gre_bgp":
|
||||
snapshot = await collectGreBgpSnapshotOnce()
|
||||
break
|
||||
case "internet_path":
|
||||
snapshot = await collectInternetPathSnapshotOnce()
|
||||
break
|
||||
case "alert_engine": {
|
||||
const r = await runAlertEngineOnce()
|
||||
snapshot = {
|
||||
@@ -158,6 +166,7 @@ async function runSchedulerJobBody(jobKey: SchedulerJobKey): Promise<void> {
|
||||
jobKey === "uptime_resources" ||
|
||||
jobKey === "uptime_ping" ||
|
||||
jobKey === "uptime_speed" ||
|
||||
jobKey === "internet_path" ||
|
||||
jobKey === "gre_bgp"
|
||||
) {
|
||||
scheduleAlertEngineAfterDataCollectors()
|
||||
@@ -243,6 +252,7 @@ export function refreshScheduler(): void {
|
||||
}
|
||||
|
||||
const apiPing = getServersApiPingSettings()
|
||||
const internetPath = getInternetPathSettings()
|
||||
if (apiPing.enabled) {
|
||||
const apiMs = Math.max(10_000, apiPing.intervalSec * 1000)
|
||||
void executeSchedulerJob("servers_rest_ping").catch(() => {})
|
||||
@@ -292,6 +302,17 @@ export function refreshScheduler(): void {
|
||||
)
|
||||
}
|
||||
|
||||
if (internetPath.enabled) {
|
||||
const internetPathMs = Math.max(30_000, internetPath.intervalSec * 1000)
|
||||
void executeSchedulerJob("internet_path").catch(() => {})
|
||||
timers.set(
|
||||
"internet_path",
|
||||
setInterval(() => {
|
||||
void executeSchedulerJob("internet_path").catch(() => {})
|
||||
}, internetPathMs),
|
||||
)
|
||||
}
|
||||
|
||||
const greBgpMs = 30_000
|
||||
void executeSchedulerJob("gre_bgp").catch(() => {})
|
||||
timers.set(
|
||||
@@ -331,6 +352,7 @@ export function getSchedulerStatus() {
|
||||
const traffic = getTrafficSettings()
|
||||
const uptime = getUptimeSettings()
|
||||
const apiPing = getServersApiPingSettings()
|
||||
const internetPath = getInternetPathSettings()
|
||||
|
||||
const resOn = uptime.resourcesEnabled ?? uptime.enabled
|
||||
const pingOn = uptime.pingEnabled ?? uptime.enabled
|
||||
@@ -342,6 +364,7 @@ export function getSchedulerStatus() {
|
||||
uptime_resources: { enabled: resOn, intervalSec: uptime.intervalSec },
|
||||
uptime_ping: { enabled: pingOn, intervalSec: uptime.probeIntervalSec },
|
||||
uptime_speed: { enabled: spdOn, intervalSec: uptime.speedIntervalSec },
|
||||
internet_path: { enabled: internetPath.enabled, intervalSec: internetPath.intervalSec },
|
||||
gre_bgp: { enabled: true, intervalSec: 30 },
|
||||
alert_engine: { enabled: true, intervalSec: 20 },
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user