feat(db): перевести хранилище с SQLite на PostgreSQL

При старте backend накатывает схему PostgreSQL 18 и, если база пустая, один раз импортирует mikrotik.db с тома. Повторный старт не копирует данные. Бэкап в UI идёт через pg_dump.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/src/services/alert-engine/confirm-stability.test.ts

@@ -4,11 +4,10 @@ import { db } from "../../db/index.js"
 import { alertEngineConfirmPending } from "../../db/schema.js"
 import type { ApiAlertRule } from "../alerts-service.js"
 import { computeStabilityReadyMap, deleteConfirmPending } from "./confirm-stability.js"
+import { withPgOrSkip } from "../../test/pg.js"
 
 const ruleId = "test-confirm-stability-recovery-bypass"
 
-deleteConfirmPending(ruleId)
-
 const baseRule: ApiAlertRule = {
   id: ruleId,
   name: "Server transition",
@@ -46,25 +45,31 @@ const recoveryHit = {
   payloadHash: "recovery-1",
 }
 
+if (!(await withPgOrSkip())) {
+  console.log("alert-engine confirm-stability tests skip")
+  process.exit(0)
+}
+
+await deleteConfirmPending(ruleId)
+
 {
-  const m = computeStabilityReadyMap([baseRule], new Map([[ruleId, problemHit]]))
+  const m = await computeStabilityReadyMap([baseRule], new Map([[ruleId, problemHit]]))
   assert.equal(m.get(ruleId), false, "проблемный переход должен ждать confirmStabilitySec")
 }
 
 {
-  const m = computeStabilityReadyMap([baseRule], new Map([[ruleId, recoveryHit]]))
+  const m = await computeStabilityReadyMap([baseRule], new Map([[ruleId, recoveryHit]]))
   assert.equal(m.get(ruleId), true, "восстановление должно отправляться без задержки")
 }
 
 {
-  const m = computeStabilityReadyMap([baseRule], new Map([[ruleId, null]]))
+  const m = await computeStabilityReadyMap([baseRule], new Map([[ruleId, null]]))
   assert.equal(m.get(ruleId), false, "без hit отправка не готова")
-  const left = db
+  const left = (await db
     .select()
     .from(alertEngineConfirmPending)
     .where(eq(alertEngineConfirmPending.ruleId, ruleId))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
   assert.equal(left, undefined, "pending должен очищаться после отсутствия hit")
 }
 

backend/src/services/alert-engine/confirm-stability.ts

@@ -4,46 +4,44 @@ import { alertEngineConfirmPending } from "../../db/schema.js"
 import type { ApiAlertRule } from "../alerts-service.js"
 import type { RuleEvalHit } from "./types.js"
 
-export function deleteConfirmPending(ruleId: string) {
-  db.delete(alertEngineConfirmPending).where(eq(alertEngineConfirmPending.ruleId, ruleId)).run()
+export async function deleteConfirmPending(ruleId: string) {
+  await db.delete(alertEngineConfirmPending).where(eq(alertEngineConfirmPending.ruleId, ruleId))
 }
 
 /**
  * Обновляет состояние ожидания стабильности и возвращает, можно ли отправлять уведомление в этом тике.
  * При отсутствии срабатывания (`hit == null`) ожидание сбрасывается.
  */
-export function stabilityReadyForSend(
+export async function stabilityReadyForSend(
   ruleId: string,
   hit: RuleEvalHit | null,
   delaySec: number | null | undefined,
-): boolean {
+): Promise<boolean> {
   if (!hit) {
-    deleteConfirmPending(ruleId)
+    await deleteConfirmPending(ruleId)
     return false
   }
   const sec = delaySec != null && delaySec > 0 ? Math.floor(delaySec) : 0
   if (sec <= 0) {
-    deleteConfirmPending(ruleId)
+    await deleteConfirmPending(ruleId)
     return true
   }
 
   const now = Date.now()
   const nowIso = new Date().toISOString()
-  const existing = db
+  const existing = (await db
     .select()
     .from(alertEngineConfirmPending)
     .where(eq(alertEngineConfirmPending.ruleId, ruleId))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
 
   if (!existing || existing.payloadHash !== hit.payloadHash) {
     if (existing) {
-      db.update(alertEngineConfirmPending)
+      await db.update(alertEngineConfirmPending)
         .set({ payloadHash: hit.payloadHash, sinceAt: nowIso })
         .where(eq(alertEngineConfirmPending.ruleId, ruleId))
-        .run()
     } else {
-      db.insert(alertEngineConfirmPending).values({ ruleId, payloadHash: hit.payloadHash, sinceAt: nowIso }).run()
+      await db.insert(alertEngineConfirmPending).values({ ruleId, payloadHash: hit.payloadHash, sinceAt: nowIso })
     }
     return false
   }
@@ -54,18 +52,18 @@ export function stabilityReadyForSend(
 }
 
 /** После успешной отправки — сбросить таймер стабильности для следующего цикла. */
-export function clearConfirmPendingAfterSend(ruleIds: string[]) {
-  for (const id of ruleIds) deleteConfirmPending(id)
+export async function clearConfirmPendingAfterSend(ruleIds: string[]) {
+  for (const id of ruleIds) await deleteConfirmPending(id)
 }
 
-export function computeStabilityReadyMap(
+export async function computeStabilityReadyMap(
   rules: ApiAlertRule[],
   hitByRule: Map<string, RuleEvalHit | null>,
-): Map<string, boolean> {
+): Promise<Map<string, boolean>> {
   const m = new Map<string, boolean>()
   for (const rule of rules) {
     if (!rule.enabled) {
-      deleteConfirmPending(rule.id)
+      await deleteConfirmPending(rule.id)
       m.set(rule.id, false)
       continue
     }
@@ -76,7 +74,7 @@ export function computeStabilityReadyMap(
           ? rule.recoveryStabilitySec
           : 0
         : rule.confirmStabilitySec
-    m.set(rule.id, stabilityReadyForSend(rule.id, hit, delaySec))
+    m.set(rule.id, await stabilityReadyForSend(rule.id, hit, delaySec))
   }
   return m
 }

backend/src/services/alert-engine/decision-engine.ts

@@ -73,20 +73,20 @@ export function buildRuleDiagnostics(
   return out
 }
 
-export function computeDecisions(args: {
+export async function computeDecisions(args: {
   rules: ApiAlertRule[]
   groups: ApiAlertGroup[]
   hitByRule: Map<string, RuleEvalHit | null>
   state: EngineStateMap
   canSend: boolean
-}): {
+}): Promise<{
   stabilityReady: Map<string, boolean>
   standalone: StandaloneDecision[]
   grouped: GroupDecision[]
   ruleDiag: AlertEngineRuleDiag[]
-} {
+}> {
   const { rules, groups, hitByRule, state, canSend } = args
-  const stabilityReady = computeStabilityReadyMap(rules, hitByRule)
+  const stabilityReady = await computeStabilityReadyMap(rules, hitByRule)
   const ruleDiag = buildRuleDiagnostics(rules, hitByRule, stabilityReady, state, canSend)
   const standalone: StandaloneDecision[] = []
   const grouped: GroupDecision[] = []

backend/src/services/alert-engine/outbox.ts

@@ -17,7 +17,9 @@ export type AlertOutboxPayload = {
   }
 }
 
-function safeParsePayload(raw: string): AlertOutboxPayload | null {
+function safeParsePayload(raw: unknown): AlertOutboxPayload | null {
+  if (raw && typeof raw === "object") return raw as AlertOutboxPayload
+  if (typeof raw !== "string") return null
   try {
     return JSON.parse(raw) as AlertOutboxPayload
   } catch {
@@ -25,21 +27,20 @@ function safeParsePayload(raw: string): AlertOutboxPayload | null {
   }
 }
 
-export function enqueueTelegramOutbox(input: {
+export async function enqueueTelegramOutbox(input: {
   id: string
   dedupeKey: string
   payload: AlertOutboxPayload
   maxRetries?: number
-}): boolean {
-  const existing = db
+}): Promise<boolean> {
+  const existing = (await db
     .select()
     .from(alertOutbox)
     .where(eq(alertOutbox.dedupeKey, input.dedupeKey))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
   if (existing && (existing.status === "pending" || existing.status === "sent")) return false
   const nowIso = new Date().toISOString()
-  db.insert(alertOutbox)
+  await db.insert(alertOutbox)
     .values({
       id: input.id,
       dedupeKey: input.dedupeKey,
@@ -48,22 +49,20 @@ export function enqueueTelegramOutbox(input: {
       retryCount: 0,
       maxRetries: Math.max(1, Math.floor(input.maxRetries ?? 3)),
       nextAttemptAt: nowIso,
-      payloadJson: JSON.stringify(input.payload),
+      payloadJson: input.payload,
     })
-    .run()
   return true
 }
 
 export function dispatchPendingOutbox(limit = 25): Promise<{ sent: number; failed: number; errors: string[] }> {
   return (async () => {
     const nowIso = new Date().toISOString()
-    const rows = db
+    const rows = await db
       .select()
       .from(alertOutbox)
       .where(and(eq(alertOutbox.status, "pending"), lte(alertOutbox.nextAttemptAt, nowIso)))
       .orderBy(asc(alertOutbox.createdAt))
       .limit(Math.max(1, limit))
-      .all()
 
     let sent = 0
     let failed = 0
@@ -72,27 +71,25 @@ export function dispatchPendingOutbox(limit = 25): Promise<{ sent: number; faile
     for (const row of rows) {
       const payload = safeParsePayload(row.payloadJson)
       if (!payload) {
-        db.update(alertOutbox)
+        await db.update(alertOutbox)
           .set({
             status: "failed",
             lastError: "invalid payload_json",
           })
           .where(eq(alertOutbox.id, row.id))
-          .run()
         failed += 1
         continue
       }
       const send = await sendTelegramAlertMessage({ text: payload.text, chatId: payload.chatId })
       if (send.ok) {
-        db.update(alertOutbox)
+        await db.update(alertOutbox)
           .set({
             status: "sent",
             sentAt: new Date().toISOString(),
             lastError: null,
           })
           .where(eq(alertOutbox.id, row.id))
-          .run()
-        appendAlertHistory({
+        await appendAlertHistory({
           id: payload.history.id,
           ruleId: payload.history.ruleId ?? null,
           groupId: payload.history.groupId ?? null,
@@ -107,7 +104,7 @@ export function dispatchPendingOutbox(limit = 25): Promise<{ sent: number; faile
       }
       const nextRetry = row.retryCount + 1
       const exhausted = nextRetry >= row.maxRetries
-      db.update(alertOutbox)
+      await db.update(alertOutbox)
         .set({
           retryCount: nextRetry,
           status: exhausted ? "failed" : "pending",
@@ -115,9 +112,8 @@ export function dispatchPendingOutbox(limit = 25): Promise<{ sent: number; faile
           lastError: send.error,
         })
         .where(eq(alertOutbox.id, row.id))
-        .run()
       if (exhausted) {
-        appendAlertHistory({
+        await appendAlertHistory({
           id: payload.history.id,
           ruleId: payload.history.ruleId ?? null,
           groupId: payload.history.groupId ?? null,

backend/src/services/alert-engine/prev-live-store.ts

@@ -1,5 +1,6 @@
 import { eq } from "drizzle-orm"
 import { db } from "../../db/index.js"
+import { parseJsonObject } from "../../db/json.js"
 import { alertEnginePrevLive } from "../../db/schema.js"
 
 export type PrevLiveKind = "gre" | "bgp"
@@ -7,42 +8,32 @@ export type PrevLiveKind = "gre" | "bgp"
 /** Предыдущие строковые состояния (GRE: up|down|degraded; BGP: state как с роутера). */
 export type PrevLiveStringMap = Record<string, string>
 
-function safeParseMap(json: string | null | undefined): PrevLiveStringMap {
-  if (!json || json.trim() === "") return {}
-  try {
-    const v = JSON.parse(json) as unknown
-    if (v == null || typeof v !== "object" || Array.isArray(v)) return {}
-    return v as PrevLiveStringMap
-  } catch {
-    return {}
-  }
+function safeParseMap(json: unknown): PrevLiveStringMap {
+  return parseJsonObject(json) as PrevLiveStringMap
 }
 
-export function loadPrevLiveMap(kind: PrevLiveKind): PrevLiveStringMap {
-  const row = db
+export async function loadPrevLiveMap(kind: PrevLiveKind): Promise<PrevLiveStringMap> {
+  const row = (await db
     .select()
     .from(alertEnginePrevLive)
     .where(eq(alertEnginePrevLive.kind, kind))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
   return safeParseMap(row?.payloadJson)
 }
 
-export function savePrevLiveMap(kind: PrevLiveKind, map: PrevLiveStringMap) {
-  const payloadJson = JSON.stringify(map)
-  const existing = db
+export async function savePrevLiveMap(kind: PrevLiveKind, map: PrevLiveStringMap) {
+  const payloadJson = map
+  const existing = (await db
     .select()
     .from(alertEnginePrevLive)
     .where(eq(alertEnginePrevLive.kind, kind))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
   if (existing) {
     if (existing.payloadJson === payloadJson) return
-    db.update(alertEnginePrevLive)
+    await db.update(alertEnginePrevLive)
       .set({ payloadJson, updatedAt: new Date().toISOString() })
       .where(eq(alertEnginePrevLive.kind, kind))
-      .run()
     return
   }
-  db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() }).run()
+  await db.insert(alertEnginePrevLive).values({ kind, payloadJson, updatedAt: new Date().toISOString() })
 }

backend/src/services/alert-engine/run-once.ts

@@ -26,7 +26,7 @@ const SNAPSHOT_SOURCE_WAIT_MS = 30_000
 const SNAPSHOT_SOURCE_POLL_MS = 20
 
 /**
- * Не строить снимок, пока коллекторы пишут в SQLite — иначе гонка с `alert_engine` по таймеру
+ * Не строить снимок, пока коллекторы пишут в PostgreSQL — иначе гонка с `alert_engine` по таймеру
  * (в т.ч. слот планировщика занят до первого `await` в коллекторе, когда внутренний `collecting` ещё false).
  */
 async function awaitSnapshotSourcesIdle(): Promise<void> {
@@ -52,8 +52,8 @@ function newOutboxId(): string {
   return `ao-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`
 }
 
-function loadEngineStateMap(): Map<string, { lastFiredAt: string; lastPayloadHash: string | null }> {
-  const rows = db.select().from(alertEngineState).all()
+async function loadEngineStateMap(): Promise<Map<string, { lastFiredAt: string; lastPayloadHash: string | null }>> {
+  const rows = await db.select().from(alertEngineState)
   const m = new Map<string, { lastFiredAt: string; lastPayloadHash: string | null }>()
   for (const r of rows) {
     m.set(r.scopeKey, { lastFiredAt: r.lastFiredAt, lastPayloadHash: r.lastPayloadHash ?? null })
@@ -61,15 +61,14 @@ function loadEngineStateMap(): Map<string, { lastFiredAt: string; lastPayloadHas
   return m
 }
 
-function upsertEngineState(scopeKey: string, lastFiredAt: string, lastPayloadHash: string | null) {
-  const existing = db.select().from(alertEngineState).where(eq(alertEngineState.scopeKey, scopeKey)).limit(1).all()[0]
+async function upsertEngineState(scopeKey: string, lastFiredAt: string, lastPayloadHash: string | null) {
+  const existing = (await db.select().from(alertEngineState).where(eq(alertEngineState.scopeKey, scopeKey)).limit(1))[0]
   if (existing) {
-    db.update(alertEngineState)
+    await db.update(alertEngineState)
       .set({ lastFiredAt, lastPayloadHash })
       .where(eq(alertEngineState.scopeKey, scopeKey))
-      .run()
   } else {
-    db.insert(alertEngineState).values({ scopeKey, lastFiredAt, lastPayloadHash }).run()
+    await db.insert(alertEngineState).values({ scopeKey, lastFiredAt, lastPayloadHash })
   }
 }
 
@@ -90,16 +89,16 @@ function ruleScopeKey(ruleId: string, transition?: RuleEvalHit["transition"]): s
 export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
   const errors: string[] = []
   const sampledAt = new Date().toISOString()
-  const latestSourceFinishedAt = getLatestSourceFinishedAt()
-  const prevSourceFinishedAt = getSourceWatermark()
+  const latestSourceFinishedAt = await getLatestSourceFinishedAt()
+  const prevSourceFinishedAt = await getSourceWatermark()
   const hasNewSources =
     latestSourceFinishedAt != null &&
     (prevSourceFinishedAt == null || latestSourceFinishedAt > prevSourceFinishedAt)
 
-  const rules = listAlertRules()
-  const groups = listAlertGroups()
-  const state = loadEngineStateMap()
-  const pub = getTelegramPublic()
+  const rules = await listAlertRules()
+  const groups = await listAlertGroups()
+  const state = await loadEngineStateMap()
+  const pub = await getTelegramPublic()
   const canSend = pub.tokenConfigured && Boolean(pub.chatId?.trim())
   let standaloneFires = 0
   let groupFires = 0
@@ -107,10 +106,10 @@ export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
   let ruleDiag: AlertEngineRunResult["ruleDiag"] = []
   if (hasNewSources) {
     await awaitSnapshotSourcesIdle()
-    const snap = ingestAlertSignals()
+    const snap = await ingestAlertSignals()
     const matched = matchRules(rules, snap)
     errors.push(...matched.errors)
-    const { standalone, grouped, ruleDiag: diag } = computeDecisions({
+    const { standalone, grouped, ruleDiag: diag } = await computeDecisions({
       rules,
       groups,
       hitByRule: matched.hitByRule,
@@ -123,7 +122,7 @@ export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
       if (!canSend) break
       const firedAt = new Date().toISOString()
       const key = ruleScopeKey(d.rule.id, d.hit.transition)
-      const queued = enqueueTelegramOutbox({
+      const queued = await enqueueTelegramOutbox({
         id: newOutboxId(),
         dedupeKey: `${key}:${d.hit.payloadHash}`,
         payload: {
@@ -142,8 +141,8 @@ export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
       })
       if (!queued) continue
       standaloneFires += 1
-      clearConfirmPendingAfterSend([d.rule.id])
-      upsertEngineState(key, firedAt, d.hit.payloadHash)
+      await clearConfirmPendingAfterSend([d.rule.id])
+      await upsertEngineState(key, firedAt, d.hit.payloadHash)
       state.set(key, { lastFiredAt: firedAt, lastPayloadHash: d.hit.payloadHash })
     }
 
@@ -155,7 +154,7 @@ export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
       const body = `Группа «${d.group.name}» (${d.group.combineMode === "any" ? "ANY" : "ALL"})\n\n${lines.join("\n")}`
       const hash = d.hits.map((h) => h.payloadHash).sort().join("|")
       const key = `group:${d.group.id}`
-      const queued = enqueueTelegramOutbox({
+      const queued = await enqueueTelegramOutbox({
         id: newOutboxId(),
         dedupeKey: `${key}:${hash}`,
         payload: {
@@ -173,18 +172,18 @@ export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
       })
       if (!queued) continue
       groupFires += 1
-      upsertEngineState(key, firedAt, hash)
+      await upsertEngineState(key, firedAt, hash)
       state.set(key, { lastFiredAt: firedAt, lastPayloadHash: hash })
-      clearConfirmPendingAfterSend(d.members.map((r) => r.id))
+      await clearConfirmPendingAfterSend(d.members.map((r) => r.id))
     }
-    updateSourceWatermark(latestSourceFinishedAt)
+    await updateSourceWatermark(latestSourceFinishedAt)
   }
 
   const outbox = await dispatchPendingOutbox()
   errors.push(...outbox.errors)
 
   if (standaloneFires > 0 || groupFires > 0) {
-    appendEvent({
+    await appendEvent({
       level: "info",
       eventType: "alerts.engine.fired",
       sourceModule: "alerts",
@@ -197,7 +196,7 @@ export async function runAlertEngineOnce(): Promise<AlertEngineRunResult> {
     })
   }
   if (errors.length > 0) {
-    appendEvent({
+    await appendEvent({
       level: "warning",
       eventType: "alerts.engine.errors",
       sourceModule: "alerts",

backend/src/services/alert-engine/signal-ingestor.ts

@@ -1,6 +1,6 @@
 import { buildSignalSnapshotFromCollectors } from "./signals-from-collectors.js"
 
 /** Единая точка входа для подготовки сигналов к rule matching. */
-export function ingestAlertSignals() {
-  return buildSignalSnapshotFromCollectors()
+export async function ingestAlertSignals() {
+  return await buildSignalSnapshotFromCollectors()
 }

backend/src/services/alert-engine/signals-from-collectors.ts

@@ -1,4 +1,4 @@
-import { sqliteDatabase } from "../../db/index.js"
+import { dbAll } from "../../db/index.js"
 import type {
   GreBgpSnapshotRunSnapshot,
   PingRunSnapshot,
@@ -20,25 +20,23 @@ function normalizeNameKey(v: string): string {
   return v.trim().toLowerCase()
 }
 
-function safeParseRunSnapshot(raw: string | null): SchedulerRunSnapshot | null {
-  if (!raw) return null
-  try {
-    return JSON.parse(raw) as SchedulerRunSnapshot
-  } catch {
-    return null
-  }
+function safeParseRunSnapshot(raw: unknown): SchedulerRunSnapshot | null {
+  const obj = typeof raw === "string" ? (() => {
+    try { return JSON.parse(raw) as unknown } catch { return null }
+  })() : raw
+  if (!obj || typeof obj !== "object") return null
+  return obj as SchedulerRunSnapshot
 }
 
-function readLatestOkRunSnapshots(jobKey: string, limit = 32): SchedulerRunSnapshot[] {
-  const rows = sqliteDatabase
-    .prepare(
-      `SELECT result_json AS resultJson
-       FROM scheduler_runs
-       WHERE job_key = ? AND status = 'ok' AND result_json IS NOT NULL
-       ORDER BY finished_at DESC
-       LIMIT ?`,
-    )
-    .all(jobKey, limit) as { resultJson: string | null }[]
+async function readLatestOkRunSnapshots(jobKey: string, limit = 32): Promise<SchedulerRunSnapshot[]> {
+  const rows = await dbAll<{ resultJson: unknown }>(
+    `SELECT result_json AS "resultJson"
+     FROM scheduler_runs
+     WHERE job_key = ? AND status = 'ok' AND result_json IS NOT NULL
+     ORDER BY finished_at DESC
+     LIMIT ?`,
+    [jobKey, limit],
+  )
   const out: SchedulerRunSnapshot[] = []
   for (const row of rows) {
     const parsed = safeParseRunSnapshot(row.resultJson)
@@ -227,13 +225,13 @@ function collectBgpSignalsFromRuns(runs: GreBgpSnapshotRunSnapshot[]): BgpPeerSi
 }
 
 /** Собирает сигналы alert_engine напрямую из результатов collector jobs (`scheduler_runs.result_json`). */
-export function buildSignalSnapshotFromCollectors(): SignalSnapshot {
+export async function buildSignalSnapshotFromCollectors(): Promise<SignalSnapshot> {
   const sampledAt = new Date().toISOString()
-  const resourceRuns = readLatestOkRunSnapshots("uptime_resources", 16) as ResourcesRunSnapshot[]
-  const restRuns = readLatestOkRunSnapshots("servers_rest_ping", 16) as ServersRestPingRunSnapshot[]
-  const pingRuns = readLatestOkRunSnapshots("uptime_ping", 24) as PingRunSnapshot[]
-  const trafficRuns = readLatestOkRunSnapshots("traffic", 8) as TrafficRunSnapshot[]
-  const greBgpRuns = readLatestOkRunSnapshots("gre_bgp", 8) as GreBgpSnapshotRunSnapshot[]
+  const resourceRuns = await readLatestOkRunSnapshots("uptime_resources", 16) as ResourcesRunSnapshot[]
+  const restRuns = await readLatestOkRunSnapshots("servers_rest_ping", 16) as ServersRestPingRunSnapshot[]
+  const pingRuns = await readLatestOkRunSnapshots("uptime_ping", 24) as PingRunSnapshot[]
+  const trafficRuns = await readLatestOkRunSnapshots("traffic", 8) as TrafficRunSnapshot[]
+  const greBgpRuns = await readLatestOkRunSnapshots("gre_bgp", 8) as GreBgpSnapshotRunSnapshot[]
 
   const resourceByKey = collectServerSignalsFromResourceRuns(resourceRuns)
   const restByKey = collectServerSignalsFromRestRuns(restRuns)

backend/src/services/alert-engine/snapshot.ts

@@ -2,6 +2,6 @@ import { buildSignalSnapshotFromCollectors } from "./signals-from-collectors.js"
 import type { SignalSnapshot } from "./types.js"
 
 /** Legacy-совместимость: единый источник сигналов — collector snapshots из scheduler_runs. */
-export function buildSignalSnapshot(): SignalSnapshot {
-  return buildSignalSnapshotFromCollectors()
+export async function buildSignalSnapshot(): Promise<SignalSnapshot> {
+  return await buildSignalSnapshotFromCollectors()
 }

backend/src/services/alert-engine/source-watermark.ts

@@ -11,44 +11,40 @@ const SOURCE_JOBS = [
   "gre_bgp",
 ] as const
 
-export function getLatestSourceFinishedAt(): string | null {
-  const row = db
+export async function getLatestSourceFinishedAt(): Promise<string | null> {
+  const row = (await db
     .select({ finishedAt: schedulerRuns.finishedAt })
     .from(schedulerRuns)
     .where(inArray(schedulerRuns.jobKey, [...SOURCE_JOBS]))
     .orderBy(desc(schedulerRuns.finishedAt))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
   return row?.finishedAt ?? null
 }
 
-export function getSourceWatermark(): string | null {
-  const row = db.select().from(alertEngineCursor).where(eq(alertEngineCursor.id, 1)).limit(1).all()[0]
+export async function getSourceWatermark(): Promise<string | null> {
+  const row = (await db.select().from(alertEngineCursor).where(eq(alertEngineCursor.id, 1)).limit(1))[0]
   return row?.lastSourceFinishedAt ?? null
 }
 
-export function updateSourceWatermark(lastSourceFinishedAt: string | null): void {
-  const existing = db
+export async function updateSourceWatermark(lastSourceFinishedAt: string | null): Promise<void> {
+  const existing = (await db
     .select()
     .from(alertEngineCursor)
     .where(eq(alertEngineCursor.id, 1))
-    .limit(1)
-    .all()[0]
+    .limit(1))[0]
   if (existing) {
-    db.update(alertEngineCursor)
+    await db.update(alertEngineCursor)
       .set({
         lastSourceFinishedAt,
         updatedAt: new Date().toISOString(),
       })
       .where(eq(alertEngineCursor.id, 1))
-      .run()
     return
   }
-  db.insert(alertEngineCursor)
+  await db.insert(alertEngineCursor)
     .values({
       id: 1,
       lastSourceFinishedAt,
       updatedAt: new Date().toISOString(),
     })
-    .run()
 }