fix(api): сохранять каталог пинг-проб атомарно

This commit is contained in:
Denozordec committed 2026-10-04 11:53:48 +07:00
1 parent 9c22fa8a68
commit 53c1bfbd3b
15 files changed
+243 -122

No files matched your search

@@ -0,0 +1,52 @@
import assert from "node:assert/strict"
import Fastify from "fastify"
import { SignJWT } from "jose"
import { createPingProbesWriteRoute } from "./ping-probes-write.js"
import { pingProbesWriteSchema } from "@mmapp/contracts"
import { replacePingProbes, type PingProbePatch, type PingProbeStore } from "../services/ping-probe-replacement.js"
const probe = { id: "p1", srcServerId: "1", name: " Проверка ", target: "192.0.2.2", enabled: false, showOnDashboard: true }
let transactions = 0, refreshes = 0, failWrite = false, failScheduler = false
let rows = new Map<string, PingProbePatch>([["old", { ...pingProbesWriteSchema.parse({ probes: [probe] }).probes[0], id: "old", probeFilter: "main", sortOrder: 0, updatedAt: "old", intervalSec: 60 }]])
let history = new Map([["old", [42]]])
const store: PingProbeStore = { transaction: async run => {
transactions++; const draft = new Map(rows), samples = new Map(history)
await run({ listIds: async () => [...draft.keys()], remove: async id => { draft.delete(id); samples.delete(id) }, upsert: async patch => {
draft.set(patch.id, { ...draft.get(patch.id), ...patch })
if (failWrite) throw new Error("write failed")
} })
rows = draft; history = samples
} }
const app = Fastify(); await app.register(createPingProbesWriteRoute({ replace: rows => replacePingProbes(store, rows), refreshScheduler: async () => { refreshes++; if (failScheduler) throw new Error("scheduler failed") } }), { prefix: "/api" })
const put = (payload: unknown) => app.inject({ method: "PUT", url: "/api/uptime/probes", headers: { "content-type": "application/json" }, payload: JSON.stringify(payload) })
for (const payload of [null, {}, { probes: null }, { probes: {} }, { probes: [probe, probe] }, ...[
{ id: "" }, { srcServerId: "1junk" }, { srcServerId: 0 }, { srcServerId: 9007199254740992 }, { name: " " }, { target: " " },
{ enabled: undefined }, { enabled: "true" }, { showOnDashboard: "false" }, { filter: false }, { srcInterface: null },
{ intervalSec: -1 }, { intervalSec: "10junk" }, { intervalSec: 3.5 }, { intervalSec: 2147483648 },
].map(patch => ({ probes: [probe, { ...probe, id: "p2", ...patch }] }))]) assert.equal((await put(payload)).statusCode, 400)
assert.equal(transactions, 0); assert.equal(refreshes, 0); assert.deepEqual(history.get("old"), [42])
failWrite = true; assert.equal((await put({ probes: [probe] })).statusCode, 500)
assert.ok(rows.has("old")); assert.deepEqual(history.get("old"), [42]); assert.equal(refreshes, 0)
failWrite = false; assert.equal((await put({ probes: [{ ...probe, intervalSec: 60 }] })).statusCode, 200)
assert.equal(rows.get("p1")?.name, "Проверка"); assert.equal(rows.get("p1")?.enabled, false); assert.equal(history.has("old"), false)
history.set("p1", [0, 42]); await put({ probes: [{ ...probe, showOnDashboard: undefined }] })
assert.equal(rows.get("p1")?.intervalSec, 60); assert.equal(rows.get("p1")?.showOnDashboard, true); assert.deepEqual(history.get("p1"), [0, 42])
for (const intervalSec of [3, 0, "", null]) {
await put({ probes: [{ ...probe, intervalSec }] }); assert.equal(rows.get("p1")?.intervalSec, intervalSec === 3 ? 5 : 0)
}
failScheduler = true; const saved = await put({ probes: [{ ...probe, name: "Изменена" }] })
assert.equal(saved.statusCode, 200); assert.equal(saved.json().ok, true); assert.match(saved.json().warning, /Ping-пробы сохранены/); assert.equal(rows.get("p1")?.name, "Изменена")
failScheduler = false; await put({ probes: [probe, { ...probe, id: "p2", name: "Вторая" }] })
assert.deepEqual([...rows.values()].map(row => [row.id, row.sortOrder]), [["p1", 0], ["p2", 1]])
assert.equal((await put({ probes: [] })).statusCode, 200); assert.equal(rows.size, 0); assert.equal(history.size, 0)
await app.close(); console.log("Ping probes HTTP/DI: whole request validated before mutation; modeled atomic rollback including history; omitted interval preserved; explicit delete and scheduler warning")
process.env.AUTH_REQUIRED = "true"; process.env.AUTH_JWT_SECRET = "ping-probes-test-secret"; process.env.AUTH_ISSUER = "https://auth.test.local"; process.env.NODE_ENV = "test"
const { default: authPlugin, requireAuth } = await import("../plugins/auth.js")
const secured = Fastify(); await secured.register(authPlugin); secured.addHook("preHandler", requireAuth)
let writes = 0; await secured.register(createPingProbesWriteRoute({ replace: async () => { writes++ }, refreshScheduler: async () => undefined }), { prefix: "/api" })
const send = (token?: string) => secured.inject({ method: "PUT", url: "/api/uptime/probes", payload: { probes: [] }, headers: token ? { authorization: "Bearer " + token } : {} })
assert.equal((await send()).statusCode, 401); assert.equal(writes, 0)
for (const permission of ["mm:uptime:read", "mm:servers:write", "mm:uptime:write"]) {
const token = await new SignJWT({ sub: "test", apps: ["mm"], permissions: [permission] }).setProtectedHeader({ alg: "HS256" }).setIssuer("https://auth.test.local").setExpirationTime("1h").sign(new TextEncoder().encode("ping-probes-test-secret"))
assert.equal((await send(token)).statusCode, permission === "mm:uptime:write" ? 200 : 403)
}
assert.equal(writes, 1); await secured.close(); console.log("Ping probes RBAC: unauthorized requests have no storage side effects")
+16
View File
@@ -0,0 +1,16 @@
import type { FastifyPluginAsyncZod } from "@fastify/type-provider-zod"
import { pingProbesWriteSchema, type PingProbeWrite } from "@mmapp/contracts"
export function createPingProbesWriteRoute(deps: { replace: (probes: PingProbeWrite[]) => Promise<void>; refreshScheduler: () => Promise<unknown> }): FastifyPluginAsyncZod {
return async app => {
app.put("/uptime/probes", async (req, reply) => {
const parsed = pingProbesWriteSchema.safeParse(req.body)
if (!parsed.success) return reply.code(400).send({ error: "Список Ping-проб некорректен. Проверьте устройства, имена, цели и уникальность записей. Изменения не сохранены." })
await deps.replace(parsed.data.probes)
try { await deps.refreshScheduler(); return reply.send({ ok: true }) }
catch (error) {
req.log.error({ err: error }, "Ping probes saved; scheduler refresh failed")
return reply.send({ ok: true, warning: "Ping-пробы сохранены, но расписание не обновлено. Проверьте состояние сборщика данных." })
}
})
}
}
+2 -33
View File
@@ -1,3 +1,4 @@
import { createPingProbesWriteRoute } from "./ping-probes-write.js"
import { createSpeedProbesWriteRoute } from "./speed-probes-write.js"
import { replaceSpeedProbes } from "../services/speed-probe-replacement.js"
import { mapSpeedProbeRead } from "../services/speed-probe-dto.js"
@@ -217,39 +218,7 @@ const uptimeRoutes: FastifyPluginAsyncZod = async (app) => {
refreshScheduler,
}))
app.put("/uptime/probes", async (req, reply) => {
const body = req.body as {
probes?: Array<{
id?: string
srcServerId?: string | number
srcInterface?: string
name?: string
target?: string
filter?: string
enabled?: boolean
showOnDashboard?: boolean
intervalSec?: number | string
}>
}
const normalized = (body.probes ?? [])
.map((p, i) => ({
id: p.id || `p-${Date.now()}-${i}`,
srcServerId: Number.parseInt(String(p.srcServerId ?? ""), 10),
srcInterface: String(p.srcInterface ?? "").trim(),
name: String(p.name ?? "").trim(),
target: String(p.target ?? "").trim(),
probeFilter: String(p.filter ?? "—"),
enabled: p.enabled !== false,
showOnDashboard: p.showOnDashboard === true,
intervalSec: p.intervalSec == null || p.intervalSec === ""
? undefined
: Number.parseInt(String(p.intervalSec), 10),
}))
.filter((p) => Number.isFinite(p.srcServerId) && p.name && p.target)
await replaceProbes(normalized)
await refreshScheduler()
return reply.send({ ok: true })
})
await app.register(createPingProbesWriteRoute({ replace: replaceProbes, refreshScheduler }))
app.patch("/uptime/probes/:id", async (req, reply) => {
const params = req.params as { id?: string }
@@ -0,0 +1,21 @@
import type { PingProbeWrite } from "@mmapp/contracts"
export type PingProbePatch = Omit<PingProbeWrite, "filter"> & { probeFilter: string; sortOrder: number; updatedAt: string }
export type PingProbeTransaction = {
listIds: () => Promise<string[]>
remove: (id: string) => Promise<void>
upsert: (row: PingProbePatch, createdAt: string) => Promise<void>
}
export type PingProbeStore = { transaction: (run: (tx: PingProbeTransaction) => Promise<void>) => Promise<void> }
export async function replacePingProbes(store: PingProbeStore, probes: PingProbeWrite[], now = new Date().toISOString()): Promise<void> {
await store.transaction(async tx => {
const nextIds = new Set(probes.map(row => row.id))
for (const id of await tx.listIds()) if (!nextIds.has(id)) await tx.remove(id)
for (const [sortOrder, row] of probes.entries()) {
const { filter, intervalSec, showOnDashboard, ...config } = row
await tx.upsert({ ...config, probeFilter: filter, sortOrder, updatedAt: now,
...(intervalSec === undefined ? {} : { intervalSec: intervalSec > 0 ? Math.max(5, intervalSec) : 0 }),
...(showOnDashboard === undefined ? {} : { showOnDashboard }),
}, now)
}
})
}
+15 -43
View File
@@ -1,4 +1,6 @@
import { and, asc, eq, gte, inArray, max, or } from "drizzle-orm"
import { replacePingProbes } from "./ping-probe-replacement.js"
import type { PingProbeWrite } from "@mmapp/contracts"
import { and, asc, eq, gte, inArray, max, or, sql } from "drizzle-orm"
import { db } from "../db/index.js"
import {
servers,
@@ -467,49 +469,19 @@ export async function updateProbeShowOnDashboard(probeId: string, showOnDashboar
return r.length > 0
}
export async function replaceProbes(rows: Array<{
id: string
srcServerId: number
srcInterface: string
name: string
target: string
probeFilter: string
enabled: boolean
showOnDashboard?: boolean
intervalSec?: number
}>) {
const now = new Date().toISOString()
const existing = await db.select({ id: uptimeProbes.id }).from(uptimeProbes)
const nextIds = new Set(rows.map((r) => r.id))
for (const p of existing) {
if (nextIds.has(p.id)) continue
await db.delete(uptimeProbes).where(eq(uptimeProbes.id, p.id))
}
for (let i = 0; i < rows.length; i += 1) {
const r = rows[i]
const intervalSec = r.intervalSec != null && r.intervalSec > 0 ? Math.max(5, Math.floor(r.intervalSec)) : 0
const patch = {
srcServerId: r.srcServerId,
srcInterface: r.srcInterface || "",
name: r.name,
target: r.target,
probeFilter: r.probeFilter || "—",
enabled: r.enabled,
showOnDashboard: r.showOnDashboard ?? false,
intervalSec,
sortOrder: i,
updatedAt: now,
}
const updated = await db.update(uptimeProbes).set(patch).where(eq(uptimeProbes.id, r.id)).returning({ id: uptimeProbes.id })
if (updated.length > 0) continue
await db.insert(uptimeProbes).values({
id: r.id,
...patch,
createdAt: now,
export async function replaceProbes(rows: PingProbeWrite[]) {
await replacePingProbes({ transaction: run => db.transaction(async tx => {
await tx.execute(sql`select pg_advisory_xact_lock(hashtext('uptime-ping-probes'))`)
await run({
listIds: async () => (await tx.select({ id: uptimeProbes.id }).from(uptimeProbes)).map(row => row.id),
remove: async id => { await tx.delete(uptimeProbes).where(eq(uptimeProbes.id, id)) },
upsert: async (row, createdAt) => {
const { id, ...patch } = row
await tx.insert(uptimeProbes).values({ id, ...patch, createdAt })
.onConflictDoUpdate({ target: uptimeProbes.id, set: patch })
},
})
}
}) }, rows)
}
export async function readProbeSamplesSince(sinceIso: string) {