Files
ridebus-status/server/store.mjs
2026-10-13 02:24:39 +03:00

262 lines
22 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import { DatabaseSync } from 'node:sqlite';
import { mkdirSync } from 'node:fs';
import { dirname } from 'node:path';
import { affectsService } from './maintenance.mjs';
import { validateUpdate } from './incident-updates.mjs';
import { maintenanceEnd } from './maintenance.mjs';
import { reliabilitySummary } from './reliability.mjs';
export function openStore(path) {
if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true });
const db = new DatabaseSync(path);
let revision = 0;
db.exec(`
PRAGMA journal_mode = WAL;
PRAGMA busy_timeout = 5000;
CREATE TABLE IF NOT EXISTS checks (
id INTEGER PRIMARY KEY, service_id TEXT NOT NULL, signature TEXT NOT NULL,
checked_at INTEGER NOT NULL, ok INTEGER NOT NULL, status TEXT NOT NULL,
latency_ms INTEGER, http_code INTEGER, reason TEXT
);
CREATE INDEX IF NOT EXISTS checks_service_time ON checks(service_id, signature, checked_at);
CREATE TABLE IF NOT EXISTS service_state (
service_id TEXT NOT NULL, signature TEXT NOT NULL, status TEXT NOT NULL,
failures INTEGER NOT NULL, successes INTEGER NOT NULL, checked_at INTEGER NOT NULL,
PRIMARY KEY (service_id, signature)
);
CREATE TABLE IF NOT EXISTS incidents (
id INTEGER PRIMARY KEY, source TEXT NOT NULL, service_id TEXT,
signature TEXT, started_at INTEGER NOT NULL, resolved_at INTEGER,
title_ru TEXT NOT NULL, title_en TEXT NOT NULL,
message_ru TEXT NOT NULL, message_en TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS maintenance (
id INTEGER PRIMARY KEY, title_ru TEXT NOT NULL, title_en TEXT NOT NULL,
message_ru TEXT NOT NULL, message_en TEXT NOT NULL, service_ids TEXT NOT NULL,
starts_at INTEGER NOT NULL, ends_at INTEGER NOT NULL, created_at INTEGER NOT NULL,
cancelled_at INTEGER, completed_at INTEGER
);
CREATE TABLE IF NOT EXISTS scenario_state (
service_id TEXT NOT NULL, signature TEXT NOT NULL, status TEXT NOT NULL,
failures INTEGER NOT NULL, successes INTEGER NOT NULL, checked_at INTEGER NOT NULL,
ok INTEGER, latency_ms INTEGER, reason TEXT, mode TEXT NOT NULL, steps TEXT NOT NULL,
PRIMARY KEY (service_id, signature)
);
CREATE TABLE IF NOT EXISTS notification_outbox (
id INTEGER PRIMARY KEY, event_key TEXT NOT NULL UNIQUE, incident_id INTEGER NOT NULL,
service_id TEXT NOT NULL, signature TEXT NOT NULL, scope TEXT NOT NULL, kind TEXT NOT NULL,
created_at INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'pending', attempts INTEGER NOT NULL DEFAULT 0,
next_attempt_at INTEGER NOT NULL, delivered_at INTEGER, last_error TEXT
);
CREATE TABLE IF NOT EXISTS incident_updates (
id INTEGER PRIMARY KEY, incident_id INTEGER NOT NULL REFERENCES incidents(id),
source TEXT NOT NULL, stage TEXT NOT NULL, created_at INTEGER NOT NULL,
message_ru TEXT NOT NULL, message_en TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS incident_updates_time ON incident_updates(incident_id, created_at, id);
CREATE TABLE IF NOT EXISTS scenario_checks (
id INTEGER PRIMARY KEY, service_id TEXT NOT NULL, signature TEXT NOT NULL,
checked_at INTEGER NOT NULL, status TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS scenario_checks_time ON scenario_checks(service_id, signature, checked_at);
`);
// Existing MVP databases are upgraded once at startup without losing history.
if (!db.prepare('PRAGMA table_info(incidents)').all().some(column => column.name === 'scope')) db.exec("ALTER TABLE incidents ADD COLUMN scope TEXT NOT NULL DEFAULT 'availability'");
// Backfill only facts present in the old database; never invent team actions.
db.exec(`
INSERT INTO incident_updates(incident_id, source, stage, created_at, message_ru, message_en)
SELECT i.id, i.source, CASE i.source WHEN 'monitor' THEN 'detected' ELSE 'reported' END, i.started_at, i.message_ru, i.message_en
FROM incidents i WHERE NOT EXISTS (SELECT 1 FROM incident_updates u WHERE u.incident_id = i.id);
INSERT INTO incident_updates(incident_id, source, stage, created_at, message_ru, message_en)
SELECT i.id, i.source, 'resolved', i.resolved_at,
CASE i.source WHEN 'monitor' THEN 'Повторные проверки подтвердили восстановление.' ELSE 'Команда отметила инцидент как решённый.' END,
CASE i.source WHEN 'monitor' THEN 'Repeated checks confirmed recovery.' ELSE 'The team marked the incident as resolved.' END
FROM incidents i WHERE i.resolved_at IS NOT NULL AND NOT EXISTS (SELECT 1 FROM incident_updates u WHERE u.incident_id = i.id AND u.stage = 'resolved');
`);
const insertUpdate = db.prepare('INSERT INTO incident_updates(incident_id, source, stage, created_at, message_ru, message_en) VALUES (?, ?, ?, ?, ?, ?)');
function transaction(action) {
db.exec('BEGIN IMMEDIATE');
try { const value = action(); db.exec('COMMIT'); revision++; return value; } catch (error) { db.exec('ROLLBACK'); throw error; }
}
const stateStatement = db.prepare('SELECT * FROM service_state WHERE service_id = ? AND signature = ?');
const saveState = db.prepare('INSERT INTO service_state VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(service_id, signature) DO UPDATE SET status = excluded.status, failures = excluded.failures, successes = excluded.successes, checked_at = excluded.checked_at');
const insertCheck = db.prepare('INSERT INTO checks(service_id, signature, checked_at, ok, status, latency_ms, http_code, reason) VALUES (?, ?, ?, ?, ?, ?, ?, ?)');
const openIncident = db.prepare("SELECT id FROM incidents WHERE source = 'monitor' AND service_id = ? AND signature = ? AND scope = ? AND resolved_at IS NULL");
const createIncident = db.prepare('INSERT INTO incidents(source, service_id, signature, started_at, title_ru, title_en, message_ru, message_en, scope) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)');
const saveScenario = db.prepare('INSERT INTO scenario_state VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(service_id, signature) DO UPDATE SET status = excluded.status, failures = excluded.failures, successes = excluded.successes, checked_at = excluded.checked_at, ok = excluded.ok, latency_ms = excluded.latency_ms, reason = excluded.reason, mode = excluded.mode, steps = excluded.steps');
const insertScenarioCheck = db.prepare('INSERT INTO scenario_checks(service_id, signature, checked_at, status) VALUES (?, ?, ?, ?)');
function spanQuery(table) {
return db.prepare(`WITH ordered AS (
SELECT checked_at, status, LEAD(checked_at, 1, ?4) OVER (ORDER BY checked_at, id) AS next_at
FROM ${table} WHERE service_id = ?1 AND signature = ?2 AND checked_at >= ?3 - ?5 AND checked_at <= ?4
), spans AS (
SELECT kind, MAX(checked_at, ?3) AS start, MIN(next_at, checked_at + ?5, ?4) AS end
FROM ordered CROSS JOIN (SELECT 'known' AS kind UNION ALL SELECT 'down')
WHERE status IN ('operational', 'degraded', 'down') AND (kind = 'known' OR status = 'down')
), positive AS (SELECT * FROM spans WHERE end > start), previous AS (
SELECT *, MAX(end) OVER (PARTITION BY kind ORDER BY start, end ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS previous_end FROM positive
), islands AS (
SELECT *, SUM(CASE WHEN previous_end IS NULL OR start > previous_end THEN 1 ELSE 0 END) OVER (PARTITION BY kind ORDER BY start, end) AS island FROM previous
) SELECT kind, MIN(start) AS start, MAX(end) AS end FROM islands GROUP BY kind, island ORDER BY kind, start`);
}
const availabilitySpans = spanQuery('checks'), scenarioSpans = spanQuery('scenario_checks');
function readSpans(query, id, signature, since, now, staleMs) {
const result = { known: [], down: [] };
for (const row of query.all(id, signature, since, now, staleMs)) result[row.kind].push([row.start, row.end]);
return result;
}
function maintenanceRows() {
return db.prepare('SELECT * FROM maintenance ORDER BY starts_at DESC, id DESC LIMIT 100').all().map(row => ({
id: String(row.id), title: { ru: row.title_ru, en: row.title_en }, message: { ru: row.message_ru, en: row.message_en },
serviceIds: JSON.parse(row.service_ids), startsAt: row.starts_at, endsAt: row.ends_at, createdAt: row.created_at,
cancelledAt: row.cancelled_at, completedAt: row.completed_at,
}));
}
function inMaintenance(service, time) { return maintenanceRows().some(item => affectsService(item, service.id, time)); }
function enqueue(id, service, signature, scope, kind, time) {
db.prepare('INSERT OR IGNORE INTO notification_outbox(event_key, incident_id, service_id, signature, scope, kind, created_at, next_attempt_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)').run(`${id}:${kind}`, id, service.id, signature, scope, kind, time, time);
}
function updateIncident(service, signature, scope, result, state, time, notify) {
const incident = openIncident.get(service.id, signature, scope);
if (state.status === 'down' && !incident && !inMaintenance(service, time)) {
const scenario = scope === 'scenario';
const created = createIncident.run('monitor', service.id, signature, time,
`${service.name.ru}: ${scenario ? 'сценарий недоступен' : 'сервис недоступен'}`,
`${service.name.en}: ${scenario ? 'user journey unavailable' : 'service unavailable'}`,
scenario ? 'Повторные проверки подтвердили проблему с получением или отображением данных.' : 'Мониторинг подтвердил проблему после нескольких проверок. Ожидаем восстановления.',
scenario ? 'Repeated checks confirmed a problem retrieving or displaying data.' : 'Monitoring confirmed a problem after repeated checks. Waiting for recovery.', scope);
const id = Number(created.lastInsertRowid);
insertUpdate.run(id, 'monitor', 'detected', time, 'Повторные проверки подтвердили сбой. Сообщений команды пока нет.', 'Repeated checks confirmed an outage. No team update yet.');
if (notify) enqueue(id, service, signature, scope, 'outage', time);
} else if (incident && (state.status === 'operational' || state.status === 'degraded' && result.ok)) {
db.prepare('UPDATE incidents SET resolved_at = ? WHERE id = ?').run(time, incident.id);
insertUpdate.run(incident.id, 'monitor', 'resolved', time, 'Повторные проверки подтвердили восстановление.', 'Repeated checks confirmed recovery.');
if (notify) enqueue(incident.id, service, signature, scope, 'recovery', time);
}
}
return {
dataRevision() { return `${revision}:${db.prepare('PRAGMA data_version').get().data_version}`; },
state(service) { return stateStatement.get(service.id, service.signature); },
record(service, result, state, time = Date.now(), { notify = false } = {}) {
db.exec('BEGIN IMMEDIATE');
try {
insertCheck.run(service.id, service.signature, time, Number(result.ok), state.status, result.latencyMs, result.httpCode, result.reason);
saveState.run(service.id, service.signature, state.status, state.failures, state.successes, time);
updateIncident(service, service.signature, 'availability', result, state, time, notify);
db.exec('COMMIT');
revision++;
} catch (error) { db.exec('ROLLBACK'); throw error; }
},
scenario(service) {
if (!service.scenario) return null;
const row = db.prepare('SELECT * FROM scenario_state WHERE service_id = ? AND signature = ?').get(service.id, service.scenario.signature);
return row ? { ...row, steps: JSON.parse(row.steps) } : null;
},
recordScenario(service, result, state, time = Date.now(), { notify = false } = {}) {
db.exec('BEGIN IMMEDIATE');
try {
saveScenario.run(service.id, service.scenario.signature, state.status, state.failures, state.successes, time, result.ok === null ? null : Number(result.ok), result.latencyMs, result.reason, result.mode || 'http', JSON.stringify(result.steps || []));
insertScenarioCheck.run(service.id, service.scenario.signature, time, state.status);
updateIncident(service, service.scenario.signature, 'scenario', result, state, time, notify);
db.exec('COMMIT');
revision++;
} catch (error) { db.exec('ROLLBACK'); throw error; }
},
hasOpenIncident(service, scope = 'availability') { return Boolean(openIncident.get(service.id, scope === 'scenario' ? service.scenario?.signature : service.signature, scope)); },
inMaintenance,
maintenances: maintenanceRows,
createMaintenance({ titleRu, titleEn, messageRu, messageEn, serviceIds, startsAt, endsAt }, time = Date.now()) {
const result = db.prepare('INSERT INTO maintenance(title_ru, title_en, message_ru, message_en, service_ids, starts_at, ends_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)').run(titleRu, titleEn, messageRu, messageEn, JSON.stringify(serviceIds), startsAt, endsAt, time);
revision++;
return String(result.lastInsertRowid);
},
endMaintenance(id, kind, time = Date.now()) {
if (!['cancel', 'finish'].includes(kind)) throw new Error('Use cancel or finish');
const field = kind === 'cancel' ? 'cancelled_at' : 'completed_at';
const boundary = kind === 'finish' ? 'starts_at <= ?' : 'ends_at > ?';
const changed = db.prepare(`UPDATE maintenance SET ${field} = ? WHERE id = ? AND cancelled_at IS NULL AND completed_at IS NULL AND ${boundary}`).run(time, id, time).changes > 0;
if (changed) revision++;
return changed;
},
dueNotifications(time = Date.now()) { return db.prepare("SELECT n.*, i.started_at, i.resolved_at FROM notification_outbox n JOIN incidents i ON i.id = n.incident_id WHERE n.status = 'pending' AND n.next_attempt_at <= ? AND NOT EXISTS (SELECT 1 FROM notification_outbox earlier WHERE earlier.incident_id = n.incident_id AND earlier.id < n.id AND earlier.status = 'pending') ORDER BY n.id LIMIT 20").all(time); },
notificationResult(id, { status, retryAt = null, reason = null }, time = Date.now()) {
db.prepare("UPDATE notification_outbox SET status = ?, attempts = attempts + 1, next_attempt_at = COALESCE(?, next_attempt_at), delivered_at = CASE WHEN ? = 'sent' THEN ? ELSE NULL END, last_error = ? WHERE id = ?").run(status, retryAt, status, time, reason, id);
},
notificationStats() { return db.prepare('SELECT status, COUNT(*) AS count FROM notification_outbox GROUP BY status').all(); },
checks(service, since) { return db.prepare('SELECT checked_at AS time, ok, status, latency_ms AS latencyMs, http_code AS httpCode, reason FROM checks WHERE service_id = ? AND signature = ? AND checked_at >= ? ORDER BY checked_at, id').all(service.id, service.signature, since); },
reliability(service, since, now, config) {
const availability = readSpans(availabilitySpans, service.id, service.signature, since, now, config.staleAfterSeconds * 1000);
const scenario = service.scenario ? readSpans(scenarioSpans, service.id, service.scenario.signature, since, now, config.scenarioIntervalSeconds * 3000) : null;
const incidents = db.prepare("SELECT scope, started_at AS startedAt, resolved_at AS resolvedAt FROM incidents WHERE source = 'monitor' AND service_id = ? AND ((scope = 'availability' AND signature = ?) OR (scope = 'scenario' AND signature = ?)) AND started_at < ? AND (resolved_at IS NULL OR resolved_at > ?)").all(service.id, service.signature, service.scenario?.signature || '', now, since);
const lastRecovery = db.prepare("SELECT resolved_at AS time, scope FROM incidents WHERE source = 'monitor' AND service_id = ? AND ((scope = 'availability' AND signature = ?) OR (scope = 'scenario' AND signature = ?)) AND resolved_at <= ? ORDER BY resolved_at DESC, id DESC LIMIT 1").get(service.id, service.signature, service.scenario?.signature || '', now);
const maintenance = db.prepare('SELECT * FROM maintenance WHERE starts_at < ? AND ends_at > ?').all(now, since).flatMap(row => {
const ids = JSON.parse(row.service_ids);
return !ids.length || ids.includes(service.id) ? [[row.starts_at, maintenanceEnd({ endsAt: row.ends_at, cancelledAt: row.cancelled_at, completedAt: row.completed_at })]] : [];
});
return reliabilitySummary({ availability, scenario, incidents, maintenance, since, now, lastRecovery });
},
latencyDetails(service, since, now, intervalSeconds = 60) {
const summary = db.prepare('SELECT COUNT(*) AS samples, AVG(latency_ms) AS average, MIN(checked_at) AS firstAt, MAX(checked_at) AS lastAt FROM checks WHERE service_id = ? AND signature = ? AND checked_at >= ? AND checked_at <= ? AND ok = 1 AND latency_ms IS NOT NULL');
const current = summary.get(service.id, service.signature, since, now);
const baseline = summary.get(service.id, service.signature, since - (now - since), since - 1);
const p95 = current.samples ? db.prepare('SELECT latency_ms FROM checks WHERE service_id = ? AND signature = ? AND checked_at >= ? AND checked_at <= ? AND ok = 1 AND latency_ms IS NOT NULL ORDER BY latency_ms LIMIT 1 OFFSET ?').get(service.id, service.signature, since, now, Math.ceil(current.samples * .95) - 1).latency_ms : null;
// Compare populated windows, rather than a handful of samples against a month.
const minimumSamples = Math.max(3, Math.floor((now - since) / (intervalSeconds * 1000) * .5));
const comparable = [current, baseline].every(item => item.samples >= minimumSamples && item.lastAt - item.firstAt >= (now - since) * .5);
const percent = comparable && baseline.average > 0 ? Number(((current.average / baseline.average - 1) * 100).toFixed(1)) : null;
return { averageMs: current.average === null ? null : Math.round(current.average), p95Ms: p95, samples: current.samples, previousAverageMs: comparable ? Math.round(baseline.average) : null, changePercent: percent };
},
statistics(service, since, now) {
const params = [service.id, service.signature, since];
const summary = db.prepare("SELECT COUNT(*) AS checks, SUM(ok) AS successes, AVG(CASE WHEN ok = 1 THEN latency_ms END) AS latencyMs FROM checks WHERE service_id = ? AND signature = ? AND checked_at >= ? AND reason IS NOT 'unconfigured'").get(...params);
const days = db.prepare("SELECT (checked_at / 86400000) * 86400000 AS start, COUNT(*) AS checks, SUM(ok) AS successes, AVG(CASE WHEN ok = 1 THEN latency_ms END) AS latencyMs, MAX(CASE status WHEN 'down' THEN 2 WHEN 'degraded' THEN 1 ELSE 0 END) AS severity FROM checks WHERE service_id = ? AND signature = ? AND checked_at >= ? AND reason IS NOT 'unconfigured' GROUP BY start").all(...params);
const responseSince = now - 24 * 3600000;
const hours = db.prepare("SELECT CAST((checked_at - ?) / 3600000 AS INTEGER) AS bucket, COUNT(*) AS checks, SUM(ok) AS successes, AVG(CASE WHEN ok = 1 THEN latency_ms END) AS latencyMs FROM checks WHERE service_id = ? AND signature = ? AND checked_at >= ? AND checked_at < ? AND reason IS NOT 'unconfigured' GROUP BY bucket").all(responseSince, service.id, service.signature, responseSince, now);
const recent = db.prepare('SELECT checked_at AS time, ok, status, latency_ms AS latencyMs, http_code AS httpCode, reason FROM checks WHERE service_id = ? AND signature = ? ORDER BY checked_at DESC, id DESC LIMIT 12').all(service.id, service.signature);
return { summary, days, hours, recent };
},
incidents() {
return db.prepare('SELECT * FROM incidents ORDER BY started_at DESC, id DESC LIMIT 100').all().map(row => ({
id: String(row.id), source: row.source, serviceId: row.service_id,
signature: row.signature, scope: row.scope, startedAt: row.started_at, resolvedAt: row.resolved_at,
title: { ru: row.title_ru, en: row.title_en }, message: { ru: row.message_ru, en: row.message_en },
updates: db.prepare('SELECT * FROM incident_updates WHERE incident_id = ? ORDER BY created_at, id').all(row.id).map(update => ({ id: String(update.id), source: update.source, stage: update.stage, createdAt: update.created_at, message: { ru: update.message_ru, en: update.message_en } })),
}));
},
createIncident({ serviceId = null, titleRu, titleEn, messageRu, messageEn }, time = Date.now()) {
return transaction(() => {
const result = createIncident.run('manual', serviceId, null, time, titleRu, titleEn, messageRu, messageEn, 'availability');
insertUpdate.run(result.lastInsertRowid, 'manual', 'reported', time, messageRu, messageEn);
return String(result.lastInsertRowid);
});
},
updateIncident(id, input, time = Date.now()) {
const value = validateUpdate(input);
return transaction(() => {
const incident = db.prepare('SELECT * FROM incidents WHERE id = ?').get(id);
if (!incident) throw new Error('Incident not found');
if (incident.resolved_at !== null && value.stage !== 'postmortem') throw new Error('Closed incidents accept postmortem updates only');
if (incident.resolved_at === null && value.stage === 'postmortem') throw new Error('Resolve the incident before adding a postmortem');
const latest = db.prepare('SELECT MAX(created_at) AS time FROM incident_updates WHERE incident_id = ?').get(id);
if (!Number.isSafeInteger(time) || time < Math.max(incident.started_at, latest.time || 0)) throw new Error('Update must follow the incident timeline');
const result = insertUpdate.run(id, 'manual', value.stage, time, value.messageRu, value.messageEn);
return String(result.lastInsertRowid);
});
},
resolveIncident(id, time = Date.now(), input = null) {
return transaction(() => {
const row = db.prepare("SELECT * FROM incidents WHERE id = ? AND source = 'manual' AND resolved_at IS NULL").get(id);
if (!row) return false;
const latest = db.prepare('SELECT MAX(created_at) AS time FROM incident_updates WHERE incident_id = ?').get(id);
if (time < Math.max(row.started_at, latest.time || 0)) throw new Error('Resolution must follow the incident timeline');
const text = input ? validateUpdate({ ...input, stage: 'monitoring' }) : { messageRu: 'Команда отметила инцидент как решённый.', messageEn: 'The team marked the incident as resolved.' };
db.prepare('UPDATE incidents SET resolved_at = ? WHERE id = ?').run(time, id);
insertUpdate.run(id, 'manual', 'resolved', time, text.messageRu, text.messageEn);
return true;
});
},
prune(since) { db.prepare('DELETE FROM checks WHERE checked_at < ?').run(since); db.prepare('DELETE FROM scenario_checks WHERE checked_at < ?').run(since); revision++; },
close() { db.close(); },
};
}