fix(alerts): читать согласованный снимок и реальные цели проб

This commit is contained in:
Denozordec committed 2026-10-03 18:27:22 +07:00
1 parent dc5594ca7c
commit c5c1c2812d
12 files changed
+114 -88

No files matched your search

+1 -1
View File
@@ -6,7 +6,7 @@ const {default:authPlugin,requireAuth}=await import("./auth.js")
const {createAlertsRoutes,defaultAlertsDependencies}=await import("../routes/alerts.js")
const app=Fastify();await app.register(authPlugin);app.addHook("preHandler",requireAuth)
let writes=0,sends=0,reads=0
await app.register(createAlertsRoutes({...defaultAlertsDependencies,telegram:async()=>({chatId:"123",tokenConfigured:true,messageThreadId:null}),groups:async()=>[],rules:async()=>[],history:async()=>[],meta:async()=>({servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}),writeTelegram:async()=>{writes++;return {chatId:"123",tokenConfigured:true,messageThreadId:null}},readTelegram:async()=>{reads++;return {botToken:"fixture-only",chatId:"123",messageThreadId:null}},send:async()=>{sends++;return {ok:true}},replace:async()=>{writes++;return {ok:true as const,groups:[],rules:[]}}}),{prefix:"/api"})
await app.register(createAlertsRoutes({...defaultAlertsDependencies,overview:async()=>({telegram:{chatId:"123",tokenConfigured:true,connected:true,messageThreadId:null},groups:[],rules:[],history:[],meta:{servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}}),writeTelegram:async()=>{writes++;return {chatId:"123",tokenConfigured:true,messageThreadId:null}},readTelegram:async()=>{reads++;return {botToken:"fixture-only",chatId:"123",messageThreadId:null}},send:async()=>{sends++;return {ok:true}},replace:async()=>{writes++;return {ok:true as const,groups:[],rules:[]}}}),{prefix:"/api"})
for(const permission of ["mm:alerts:read","mm:alerts:write","mm:alerts:admin","mm:settings:admin"]){
const token=await new SignJWT({sub:"fixture",apps:["mm"],permissions:[permission]}).setProtectedHeader({alg:"HS256"}).setIssuer("https://auth.test.local").setExpirationTime("1h").sign(new TextEncoder().encode("evolution-fixture-secret"))
for(const [method,url] of [["GET","/api/alerts"],["PUT","/api/alerts/telegram"],["POST","/api/alerts/telegram/test"],["PUT","/api/alerts/rules"]] as const){const permitted=permission==="mm:alerts:write"||permission==="mm:alerts:admin"||method==="GET"&&permission==="mm:alerts:read";const response=await app.inject({method,url,headers:{authorization:"Bearer "+token},...(method==="PUT"?{payload:url.endsWith("rules")?{rules:[],groups:[]}:{chatId:"123"}}:{})});assert.equal(response.statusCode,permitted?200:403);assert.equal(response.body.includes("fixture-only"),false)}
+3 -3
View File
@@ -2,11 +2,11 @@ import assert from "node:assert/strict"
import Fastify from "fastify"
import {createAlertsRoutes,defaultAlertsDependencies} from "./alerts.js"
import {executeTelegramSettingsUpdate} from "../services/alerts-telegram-settings.js"
let replacements=0,replaceConflict=false
let replacements=0,replaceConflict=false,overviewMalformed=false
let stored={botToken:"fixture-secret",chatId:"123",messageThreadId:10 as number|null},reads=0,writes=0,sends=0,failSend=false,failWrite=false
const app=Fastify();await app.register(createAlertsRoutes({...defaultAlertsDependencies,meta:async()=>({servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}),telegram:async()=>({chatId:stored.chatId,tokenConfigured:!!stored.botToken,messageThreadId:stored.messageThreadId}),groups:async()=>[],rules:async()=>[],history:async()=>[],replace:async()=>{replacements++;if(replaceConflict)throw Error("Сохранение правил уже выполняется");return {ok:true as const,groups:[],rules:[]}},writeTelegram:async patch=>executeTelegramSettingsUpdate(patch,{ensure:async()=>{},update:async values=>{writes++;if(failWrite)return undefined;stored={...stored,...values};return {...stored}}}),readTelegram:async()=>{reads++;return {...stored}},send:async input=>{sends++;assert.equal(input.token,"fixture-secret");assert.equal(input.chatId,"123");assert.equal(input.messageThreadId,10);return failSend?{ok:false,error:"send unknown"}:{ok:true}}}),{prefix:"/api"})
const app=Fastify();await app.register(createAlertsRoutes({...defaultAlertsDependencies,overview:async()=>overviewMalformed?{}:({telegram:{chatId:stored.chatId,tokenConfigured:!!stored.botToken,connected:!!stored.botToken&&!!stored.chatId.trim(),messageThreadId:stored.messageThreadId},groups:[],rules:[],history:[],meta:{servers:[],greClients:[],probeTargets:[],rttLossTargets:[],trafficServers:[],serversDetail:[]}}),replace:async()=>{replacements++;if(replaceConflict)throw Error("Сохранение правил уже выполняется");return {ok:true as const,groups:[],rules:[]}},writeTelegram:async patch=>executeTelegramSettingsUpdate(patch,{ensure:async()=>{},update:async values=>{writes++;if(failWrite)return undefined;stored={...stored,...values};return {...stored}}}),readTelegram:async()=>{reads++;return {...stored}},send:async input=>{sends++;assert.equal(input.token,"fixture-secret");assert.equal(input.chatId,"123");assert.equal(input.messageThreadId,10);return failSend?{ok:false,error:"send unknown"}:{ok:true}}}),{prefix:"/api"})
const test=()=>app.inject({method:"POST",url:"/api/alerts/telegram/test",payload:{}})
let response=await app.inject({method:"GET",url:"/api/alerts"});assert.equal(response.statusCode,200);assert.equal(response.body.includes("fixture-secret"),false);assert.deepEqual(response.json().rules,[])
let response=await app.inject({method:"GET",url:"/api/alerts"});assert.equal(response.statusCode,200);assert.equal(response.body.includes("fixture-secret"),false);assert.deepEqual(response.json().rules,[]);assert.equal(response.headers["cache-control"],"no-store");overviewMalformed=true;response=await app.inject({method:"GET",url:"/api/alerts"});assert.equal(response.statusCode,500);assert.equal(response.json().ok,undefined);overviewMalformed=false
response=await test();assert.equal(response.statusCode,200);assert.deepEqual(response.json(),{ok:true});assert.equal(reads,1);assert.equal(sends,1)
failSend=true;response=await test();assert.equal(response.statusCode,502);assert.equal(response.json().ok,false);assert.equal(reads,2);assert.equal(sends,2)
for(const payload of [{messageThreadId:0},{messageThreadId:1.2},{messageThreadId:true},{messageThreadId:"1"},{messageThreadId:2147483648},{unknown:true}]){response=await app.inject({method:"POST",url:"/api/alerts/telegram/test",payload});assert.equal(response.statusCode,400)}assert.equal(sends,2);assert.equal(reads,2)
+7 -16
View File
@@ -1,17 +1,14 @@
import {sendTelegramRequest} from "../services/alerts-telegram-send.js"
import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
import {
getAlertsMeta,
getTelegramPublic,
getAlertsOverview,
getTelegramSettingsRow,
listAlertGroups,
listAlertRules,
listMergedHistory,
replaceAlertsConfig,
formatAlertRuleTestTelegramText,
updateTelegramSettings,
} from "../services/alerts-service.js"
import {
alertsResponseSchema,
putRulesSchema,
normalizeAlertsConfig,
putTelegramSchema,
@@ -20,21 +17,15 @@ import {
} from "@mmapp/contracts/alerts"
interface Dependencies {
meta:typeof getAlertsMeta;telegram:typeof getTelegramPublic;groups:typeof listAlertGroups;rules:typeof listAlertRules;history:typeof listMergedHistory;replace:typeof replaceAlertsConfig;writeTelegram:typeof updateTelegramSettings
overview:()=>Promise<unknown>;replace:typeof replaceAlertsConfig;writeTelegram:typeof updateTelegramSettings
readTelegram:()=>Promise<{botToken:string;chatId:string;messageThreadId:number|null}>
send:(input:{token:string;chatId:string;text:string;messageThreadId?:number})=>Promise<{ok:true}|{ok:false;error:string}>
}
export function createAlertsRoutes(deps:Dependencies):FastifyPluginAsyncZod {return async(app)=>{
app.get("/alerts", async (_req, reply) => {
const tg = await deps.telegram()
const connected = tg.tokenConfigured && Boolean(tg.chatId?.trim())
return reply.send({
telegram: { ...tg, connected },
groups: await deps.groups(),
rules: await deps.rules(),
history: await deps.history(),
meta: await deps.meta(),
})
const result=alertsResponseSchema.safeParse(await deps.overview())
if(!result.success)return reply.status(500).send({error:"Снимок оповещений не подтверждён"})
return reply.header("Cache-Control","no-store").send(result.data)
})
app.put("/alerts/rules", async (req, reply) => {
@@ -116,5 +107,5 @@ export function createAlertsRoutes(deps:Dependencies):FastifyPluginAsyncZod {ret
}
}
export const defaultAlertsDependencies:Dependencies={meta:getAlertsMeta,telegram:getTelegramPublic,groups:listAlertGroups,rules:listAlertRules,history:listMergedHistory,replace:replaceAlertsConfig,writeTelegram:updateTelegramSettings,readTelegram:getTelegramSettingsRow,send:input=>sendTelegramRequest(input.token,{chat_id:input.chatId,text:input.text,...(input.messageThreadId!==undefined?{message_thread_id:input.messageThreadId}:{})})}
export const defaultAlertsDependencies:Dependencies={overview:getAlertsOverview,replace:replaceAlertsConfig,writeTelegram:updateTelegramSettings,readTelegram:getTelegramSettingsRow,send:input=>sendTelegramRequest(input.token,{chat_id:input.chatId,text:input.text,...(input.messageThreadId!==undefined?{message_thread_id:input.messageThreadId}:{})})}
export default createAlertsRoutes(defaultAlertsDependencies)
@@ -0,0 +1,18 @@
import assert from "node:assert/strict"
import {test} from "node:test"
import type {AlertsResponse} from "@mmapp/contracts/alerts"
import {readAlertsOverview,alertsOverviewTransaction,buildAlertsMeta} from "./alerts-overview.js"
const empty:AlertsResponse={telegram:{chatId:"",tokenConfigured:false,connected:false,messageThreadId:null},groups:[],rules:[],history:[],meta:buildAlertsMeta([],[])}
test("Alerts overview uses one context with repeatable read/read only and rejects partial snapshots",async()=>{
const context={snapshot:structuredClone(empty)},calls:string[]=[];let transactions=0
const read=<K extends keyof typeof empty>(key:K)=>async(value:typeof context)=>{assert.equal(value,context);calls.push(key);return value.snapshot[key]}
const deps={transaction:async(work:(value:typeof context)=>Promise<typeof empty>,options:typeof alertsOverviewTransaction)=>{transactions++;assert.deepEqual(options,{isolationLevel:"repeatable read",accessMode:"read only"});return work(context)},telegram:read("telegram"),groups:read("groups"),rules:read("rules"),history:read("history"),meta:read("meta")}
assert.deepEqual(await readAlertsOverview(deps),empty);assert.equal(transactions,1);assert.deepEqual(calls,["telegram","groups","rules","history","meta"])
await assert.rejects(readAlertsOverview({...deps,rules:async()=>null}),/Снимок оповещений не подтверждён/)
await assert.rejects(readAlertsOverview({...deps,history:async()=>{throw Error("DB read failed")}}),/DB read failed/)
})
test("Alerts metadata exposes existing probe keys without synthesizing per-server targets",()=>{
const servers=[{name:"router-fixture",host:"192.0.2.1",site:null,country:null,type:null}]
const withoutProbes=buildAlertsMeta(servers,[]);assert.deepEqual(withoutProbes.probeTargets,[]);assert.deepEqual(withoutProbes.rttLossTargets,[]);assert.equal(withoutProbes.serversDetail[0]?.type,"")
const actual=buildAlertsMeta(servers,[{name:"probe-fixture",target:"1.1.1.1"}]);assert.deepEqual(actual.probeTargets,["probe-fixture → 1.1.1.1"]);assert.deepEqual(actual.rttLossTargets,actual.probeTargets);assert.equal(JSON.stringify(actual).includes("8.8.8.8"),false)
})
+17
View File
@@ -0,0 +1,17 @@
import {alertsResponseSchema,type AlertsResponse} from "@mmapp/contracts/alerts"
export const alertsOverviewTransaction={isolationLevel:"repeatable read",accessMode:"read only"} as const
interface Dependencies<C>{transaction:(work:(context:C)=>Promise<AlertsResponse>,options:typeof alertsOverviewTransaction)=>Promise<AlertsResponse>;telegram:(context:C)=>Promise<unknown>;groups:(context:C)=>Promise<unknown>;rules:(context:C)=>Promise<unknown>;history:(context:C)=>Promise<unknown>;meta:(context:C)=>Promise<unknown>}
export async function readAlertsOverview<C>(deps:Dependencies<C>):Promise<AlertsResponse>{
return deps.transaction(async context=>{
const telegram=await deps.telegram(context)
const parsed=alertsResponseSchema.safeParse({telegram,groups:await deps.groups(context),rules:await deps.rules(context),history:await deps.history(context),meta:await deps.meta(context)})
if(!parsed.success)throw Error("Снимок оповещений не подтверждён")
return parsed.data
},alertsOverviewTransaction)
}
export function buildAlertsMeta(servers:{name:string|null;host:string;site:string|null;country:string|null;type:string|null}[],probes:{name:string;target:string}[]){
const labels=servers.map(server=>server.name?.trim()||server.host)
const probeTargets=probes.map(probe=>probe.name+" → "+probe.target)
return {servers:labels,greClients:labels,probeTargets,rttLossTargets:probeTargets,trafficServers:labels,serversDetail:servers.map(server=>({name:server.name?.trim()||"",host:server.host.trim(),site:server.site?.trim()||"",country:server.country?.trim()||"",type:server.type??""}))}
}
+38 -53
View File
@@ -1,8 +1,9 @@
import {readAlertsOverview,buildAlertsMeta} from "./alerts-overview.js"
import {executeAlertsConfigReplacement} from "./alerts-config-execution.js"
import {sendTelegramRequest} from "./alerts-telegram-send.js"
import {executeTelegramSettingsUpdate} from "./alerts-telegram-settings.js"
import { asc, desc, eq, sql } from "drizzle-orm"
import { db, dbAll } from "../db/index.js"
import { db } from "../db/index.js"
import {
alertGroups,
alertHistory,
@@ -297,14 +298,12 @@ export async function getTelegramSettingsRow() {
return await ensureTelegramRow()
}
export async function getTelegramPublic() {
const row = await ensureTelegramRow()
const tid = row.messageThreadId
const messageThreadId =
typeof tid === "number" && Number.isFinite(tid) && Math.floor(tid) >= 1 ? Math.floor(tid) : null
export async function getTelegramPublic(context:AlertsReadContext=db) {
const row = (await context.select().from(alertTelegramSettings).where(eq(alertTelegramSettings.id,1)).limit(1))[0]
const messageThreadId = row?.messageThreadId ?? null
return {
chatId: row.chatId ?? "",
tokenConfigured: Boolean(row.botToken?.trim()),
chatId: row?.chatId ?? "",
tokenConfigured: Boolean(row?.botToken?.trim()),
messageThreadId,
}
}
@@ -423,9 +422,8 @@ interface LiveProbeRow {
lossPct: number | null
}
async function loadLiveResourceIssues(): Promise<LiveResRow[]> {
return dbAll<LiveResRow>(
`SELECT s.name AS "serverName", urs.sampled_at AS "sampledAt"
async function loadLiveResourceIssues(context:AlertsReadContext): Promise<LiveResRow[]> {
const result=await context.execute(sql.raw(`SELECT s.name AS "serverName", urs.sampled_at AS "sampledAt"
FROM uptime_resource_samples urs
INNER JOIN (
SELECT DISTINCT ON (server_id) server_id, sampled_at, status
@@ -433,13 +431,12 @@ async function loadLiveResourceIssues(): Promise<LiveResRow[]> {
ORDER BY server_id, sampled_at DESC
) latest ON latest.server_id = urs.server_id AND latest.sampled_at = urs.sampled_at
INNER JOIN servers s ON s.id = urs.server_id
WHERE urs.status = 'offline'`,
)
WHERE urs.status = 'offline'`))
return result.rows as unknown as LiveResRow[]
}
async function loadLiveProbeIssues(): Promise<LiveProbeRow[]> {
return dbAll<LiveProbeRow>(
`SELECT p.name AS "probeName", p.target AS target, ups.sampled_at AS "sampledAt",
async function loadLiveProbeIssues(context:AlertsReadContext): Promise<LiveProbeRow[]> {
const result=await context.execute(sql.raw(`SELECT p.name AS "probeName", p.target AS target, ups.sampled_at AS "sampledAt",
ups.status AS status, ups.rtt_ms AS "rttMs", ups.loss_pct AS "lossPct"
FROM uptime_probe_samples ups
INNER JOIN (
@@ -448,25 +445,25 @@ async function loadLiveProbeIssues(): Promise<LiveProbeRow[]> {
ORDER BY probe_id, sampled_at DESC
) latest ON latest.probe_id = ups.probe_id AND latest.sampled_at = ups.sampled_at
INNER JOIN uptime_probes p ON p.id = ups.probe_id
WHERE ups.status IN ('down', 'warn')`,
)
WHERE ups.status IN ('down', 'warn')`))
return result.rows as unknown as LiveProbeRow[]
}
async function toLiveHistoryEntries(): Promise<ApiAlertHistoryEntry[]> {
async function toLiveHistoryEntries(context:AlertsReadContext): Promise<ApiAlertHistoryEntry[]> {
const out: ApiAlertHistoryEntry[] = []
for (const r of await loadLiveResourceIssues()) {
for (const r of await loadLiveResourceIssues(context)) {
const safeId = `live-res-${r.serverName}-${r.sampledAt}`.replace(/[^a-zA-Z0-9._-]+/g, "_")
out.push({
id: safeId,
ruleName: "Сервер (ресурсы)",
severity: "critical",
message: `${r.serverName}: статус offline по данным мониторинга`,
sent: true,
sent: false,
firedAt: r.sampledAt,
source: "live",
})
}
for (const p of await loadLiveProbeIssues()) {
for (const p of await loadLiveProbeIssues(context)) {
const sev = p.status === "down" ? "critical" : "warning"
const loss = p.lossPct != null ? `${p.lossPct}%` : "—"
const rtt = p.rttMs != null ? `${p.rttMs} мс` : "—"
@@ -476,7 +473,7 @@ async function toLiveHistoryEntries(): Promise<ApiAlertHistoryEntry[]> {
ruleName: "Ping-проба",
severity: sev,
message: `${p.probeName} → ${p.target}: ${p.status} (RTT ${rtt}, потери ${loss})`,
sent: true,
sent: false,
firedAt: p.sampledAt,
source: "live",
})
@@ -484,8 +481,8 @@ async function toLiveHistoryEntries(): Promise<ApiAlertHistoryEntry[]> {
return out
}
export async function listPersistedHistory(limit: number): Promise<ApiAlertHistoryEntry[]> {
const rows = await db
export async function listPersistedHistory(limit: number,context:AlertsReadContext=db): Promise<ApiAlertHistoryEntry[]> {
const rows = await context
.select()
.from(alertHistory)
.orderBy(desc(alertHistory.firedAt))
@@ -501,9 +498,9 @@ export async function listPersistedHistory(limit: number): Promise<ApiAlertHisto
}))
}
export async function listMergedHistory(persistLimit = 40, liveCap = 25): Promise<ApiAlertHistoryEntry[]> {
const persisted = await listPersistedHistory(persistLimit)
const live = (await toLiveHistoryEntries()).slice(0, liveCap)
export async function listMergedHistory(persistLimit = 40, liveCap = 25,context:AlertsReadContext=db): Promise<ApiAlertHistoryEntry[]> {
const persisted = await listPersistedHistory(persistLimit,context)
const live = (await toLiveHistoryEntries(context)).slice(0, liveCap)
const merged = [...persisted, ...live].sort((a, b) => (a.firedAt < b.firedAt ? 1 : a.firedAt > b.firedAt ? -1 : 0))
const seen = new Set<string>()
const dedup: ApiAlertHistoryEntry[] = []
@@ -516,8 +513,8 @@ export async function listMergedHistory(persistLimit = 40, liveCap = 25): Promis
return dedup
}
export async function getAlertsMeta() {
const srvRows = await db
export async function getAlertsMeta(context:AlertsReadContext=db) {
const srvRows = await context
.select({
name: servers.name,
host: servers.host,
@@ -526,27 +523,15 @@ export async function getAlertsMeta() {
type: servers.type,
})
.from(servers)
const labels = srvRows.map((s) => (s.name?.trim() ? s.name.trim() : s.host))
const serversDetail = srvRows.map((r) => ({
name: r.name?.trim() || "",
host: r.host?.trim() || "",
site: r.site?.trim() || "",
country: r.country?.trim() || "",
type: (r.type ?? "home-router") as string,
}))
const probes = await db.select({ name: uptimeProbes.name, target: uptimeProbes.target }).from(uptimeProbes)
const probeTargets = probes.map((p) => `${p.name} → ${p.target}`)
const defaults = ["8.8.8.8", "1.1.1.1"]
const rttLossTargets: string[] = []
for (const L of labels.slice(0, 12)) {
for (const d of defaults) rttLossTargets.push(`${L} → ${d}`)
}
return {
servers: labels,
greClients: labels,
probeTargets: probeTargets.length ? probeTargets : rttLossTargets,
rttLossTargets: rttLossTargets.length ? rttLossTargets : labels.map((L) => `${L} → 8.8.8.8`),
trafficServers: labels,
serversDetail,
}
const probes=await context.select({name:uptimeProbes.name,target:uptimeProbes.target}).from(uptimeProbes)
return buildAlertsMeta(srvRows,probes)
}
export async function getAlertsOverview(){
return readAlertsOverview<AlertsWriteContext>({
transaction:(work,options)=>db.transaction(work,options),
telegram:async context=>{const value=await getTelegramPublic(context);return {...value,connected:value.tokenConfigured&&Boolean(value.chatId.trim())}},
groups:context=>listAlertGroups(context),rules:context=>listAlertRules(context),
history:context=>listMergedHistory(40,25,context),meta:context=>getAlertsMeta(context),
})
}