From f38694da6308d8afebff3e1855358a1bd5e17464 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Fri, 6 Nov 2026 22:45:45 +0700 Subject: [PATCH] =?UTF-8?q?fix(notifications):=20=D0=B8=D0=B7=D0=BE=D0=BB?= =?UTF-8?q?=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D1=82=D1=8C=20=D1=83=D0=B2=D0=B5?= =?UTF-8?q?=D0=B4=D0=BE=D0=BC=D0=BB=D0=B5=D0=BD=D0=B8=D1=8F=20=D0=BF=D0=BE?= =?UTF-8?q?=20space=20=D0=B8=20=D1=87=D0=B8=D0=BD=D0=B8=D1=82=D1=8C=20?= =?UTF-8?q?=D0=BE=D1=82=D0=BF=D1=80=D0=B0=D0=B2=D0=BA=D1=83=20=D0=B2=20Tel?= =?UTF-8?q?egram?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - «Истекает оплата» строится только по VPS текущего space - ключ notification_state разделён по space: второй space больше не падает на PRIMARY KEY и не получает дубль каждый тик - таймаут 10 с для Telegram и webhook - текст от хостеров экранируется для parse_mode=HTML - сообщения длиннее 4096 символов обрезаются по строке - повторное падение того же VPS снова уведомляет - ошибки по отдельным чатам пишутся в журнал при частичной доставке - удалён неиспользуемый sendWebhook/notifyWebhook Co-Authored-By: Claude Opus 5.5 --- .../src/services/notifications/channels.ts | 1 + apps/api/src/services/notifications/dedup.ts | 1 + .../src/services/notifications/review.test.ts | 71 +++++++++++++++++++ apps/api/src/services/notifications/rules.ts | 50 +++++++++---- apps/api/src/services/notifications/types.ts | 2 +- apps/api/src/services/telegram.ts | 26 ++++++- apps/api/src/services/webhook.ts | 34 --------- packages/db/src/repositories/notifications.ts | 18 +++-- packages/db/src/runtime-migrate.test.ts | 15 ++++ packages/db/src/runtime-migrate.ts | 5 ++ 10 files changed, 167 insertions(+), 56 deletions(-) create mode 100644 apps/api/src/services/notifications/review.test.ts diff --git a/apps/api/src/services/notifications/channels.ts b/apps/api/src/services/notifications/channels.ts index 7bc100b..87bffae 100644 --- a/apps/api/src/services/notifications/channels.ts +++ b/apps/api/src/services/notifications/channels.ts @@ -30,6 +30,7 @@ export async function deliverWebhook( method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(payload), + signal: AbortSignal.timeout(10_000), }) if (!res.ok) return { ok: false, error: `HTTP ${res.status}` } return { ok: true } diff --git a/apps/api/src/services/notifications/dedup.ts b/apps/api/src/services/notifications/dedup.ts index 9252420..fbabeb0 100644 --- a/apps/api/src/services/notifications/dedup.ts +++ b/apps/api/src/services/notifications/dedup.ts @@ -10,6 +10,7 @@ export function shouldSkipDedup( stateKey?: string, newStatus?: string, ): boolean { + if (mode === 'none') return false const key = stateKey ?? event const state = notificationRepository.getState(key) const now = Date.now() diff --git a/apps/api/src/services/notifications/review.test.ts b/apps/api/src/services/notifications/review.test.ts new file mode 100644 index 0000000..f937d16 --- /dev/null +++ b/apps/api/src/services/notifications/review.test.ts @@ -0,0 +1,71 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { closeDb, getSqlite, MAIN_SPACE_ID, runWithSpace } from '@cfdm/db' +import { resetTestDb } from '@cfdm/db/test-setup' +import { buildPaymentExpiryNotification, buildSyncDigestNotification, buildVpsHealthNotification } from './rules.js' +import { markDedupSent, shouldSkipDedup } from './dedup.js' +import { publishNotification } from './engine.js' +import { fitTelegramText, sendTelegramMessage } from '../telegram.js' + +describe('notifications review fixes', () => { + beforeEach(() => { + resetTestDb() + getSqlite().exec(`INSERT INTO spaces (id, name, slug, kind, createdAt) VALUES ('space-x', 'X', 'x', 'personal', '2026-01-01')`) + }) + afterEach(() => { + vi.unstubAllGlobals() + closeDb() + }) + + it('payment expiry only lists VPS of the current space', () => { + const now = new Date('2026-06-01T12:00:00Z') + getSqlite() + .prepare(`INSERT INTO vps (id, spaceId, ip, dns, status, paidUntil) VALUES ('v-x', 'space-x', '1.1.1.1', 'x.example', 'active', '2026-06-03')`) + .run() + expect(runWithSpace(MAIN_SPACE_ID, () => buildPaymentExpiryNotification(now))).toBeNull() + expect(runWithSpace('space-x', () => buildPaymentExpiryNotification(now))?.messagePlain).toContain('x.example') + }) + + it('dedup state does not collide between spaces', () => { + runWithSpace(MAIN_SPACE_ID, () => markDedupSent('payment_expiry', 'fp-main', 'daily')) + runWithSpace('space-x', () => markDedupSent('payment_expiry', 'fp-x', 'daily')) + expect(runWithSpace('space-x', () => shouldSkipDedup('payment_expiry', 'fp-x', 'daily'))).toBe(true) + expect(runWithSpace(MAIN_SPACE_ID, () => shouldSkipDedup('payment_expiry', 'fp-main', 'daily'))).toBe(true) + }) + + it('escapes hoster text in HTML', () => { + const p = buildSyncDigestNotification(['✗ Acc <1>: 502 & co']) + expect(p?.messageHtml).toContain('Acc <1>: <html>502 & co') + expect(p?.messageHtml).toContain('') + }) + + it('cuts long text at a line break under the Telegram limit', () => { + const text = Array.from({ length: 500 }, (_, i) => `• host-${i} & more`).join('\n') + const out = fitTelegramText(text) + expect(out.length).toBeLessThanOrEqual(4096) + expect(out).toMatch(/… и ещё \d+ строк$/) + expect(fitTelegramText('short')).toBe('short') + }) + + it('repeat outage of the same VPS is sent again', async () => { + const fetchMock = vi.fn(async () => new Response('{}')) + vi.stubGlobal('fetch', fetchMock) + const settings = { webhookEnabled: 1, webhookUrl: 'https://hook.test' } + const down = buildVpsHealthNotification('vps_down', [{ id: 'v1', label: 'v1' }])! + expect(await publishNotification(settings, down)).toBe('sent') + expect(await publishNotification(settings, down)).toBe('sent') + expect(fetchMock).toHaveBeenCalledTimes(2) + expect(fetchMock.mock.calls[0]?.[1]?.signal).toBeInstanceOf(AbortSignal) + }) + + it('keeps errors of failed chats on partial delivery and sets a timeout', async () => { + const fetchMock = vi.fn(async (_url, init) => { + const chat = JSON.parse(String(init?.body)).chat_id + return Response.json(chat === 'a' ? { ok: true } : { ok: false, description: 'Bad Request: chat not found' }) + }) + vi.stubGlobal('fetch', fetchMock) + const r = await sendTelegramMessage('tok', 'a,b', 'hi') + expect(r.ok).toBe(true) + expect(r.error).toContain('b: Bad Request: chat not found') + expect(fetchMock.mock.calls[0]?.[1]?.signal).toBeInstanceOf(AbortSignal) + }) +}) diff --git a/apps/api/src/services/notifications/rules.ts b/apps/api/src/services/notifications/rules.ts index e7e0e35..fbddc3d 100644 --- a/apps/api/src/services/notifications/rules.ts +++ b/apps/api/src/services/notifications/rules.ts @@ -1,19 +1,40 @@ -import { desc } from 'drizzle-orm' -import { getDb, schema } from '@cfdm/db' +import { desc, eq } from 'drizzle-orm' +import { getCurrentSpaceId, getDb, schema } from '@cfdm/db' import { getPaidUntilDate } from '@cfdm/shared/utils/paid-until' import type { NotificationPayload } from './types.js' const UPCOMING_DAYS = 7 +/** Names/errors come from hosters and users; Telegram parse_mode=HTML rejects the whole message on a stray < or &. */ +export function escapeHtml(s: string): string { + return s.replace(/&/g, '&').replace(//g, '>') +} + +const htmlLines = (lines: string[]) => lines.map(escapeHtml).join('\n') + type VpsRow = typeof schema.vps.$inferSelect export function buildPaymentExpiryNotification(now = new Date()): NotificationPayload | null { const db = getDb() - const vpsList = db.select().from(schema.vps).orderBy(desc(schema.vps.createdAt)).all() - const providerAccounts = db.select().from(schema.providerAccounts).all() - const payments = db.select().from(schema.payments).all() - const balanceLedger = db.select().from(schema.balanceLedger).all() - const providers = db.select().from(schema.providers).all() + const spaceId = getCurrentSpaceId() + const vpsList = db + .select() + .from(schema.vps) + .where(eq(schema.vps.spaceId, spaceId)) + .orderBy(desc(schema.vps.createdAt)) + .all() + const providerAccounts = db + .select() + .from(schema.providerAccounts) + .where(eq(schema.providerAccounts.spaceId, spaceId)) + .all() + const payments = db.select().from(schema.payments).where(eq(schema.payments.spaceId, spaceId)).all() + const balanceLedger = db + .select() + .from(schema.balanceLedger) + .where(eq(schema.balanceLedger.spaceId, spaceId)) + .all() + const providers = db.select().from(schema.providers).where(eq(schema.providers.spaceId, spaceId)).all() const threshold = new Date(now) threshold.setDate(threshold.getDate() + UPCOMING_DAYS) @@ -36,7 +57,7 @@ export function buildPaymentExpiryNotification(now = new Date()): NotificationPa return `• ${vps.dns || vps.ip} (${provider}) — до ${dateStr}` }) const plain = `Истекает оплата (ближайшие ${UPCOMING_DAYS} дней):\n\n${lines.join('\n')}` - const html = `⚠️ Истекает оплата (ближайшие ${UPCOMING_DAYS} дней):\n\n${lines.join('\n')}` + const html = `⚠️ Истекает оплата (ближайшие ${UPCOMING_DAYS} дней):\n\n${htmlLines(lines)}` const fingerprint = upcoming .map(({ vps, paidUntil }) => `${vps.id}:${paidUntil.toISOString().slice(0, 10)}`) .sort() @@ -56,7 +77,7 @@ export function buildSyncDigestNotification(digestLines: string[]): Notification const changed = digestLines.filter((line) => !line.includes('без изменений')) if (changed.length === 0) return null const plain = `Синхронизация VPS:\n\n${changed.join('\n')}` - const html = `📋 Синхронизация VPS\n\n${changed.join('\n')}` + const html = `📋 Синхронизация VPS\n\n${htmlLines(changed)}` return { event: 'sync_digest', fingerprint: changed.sort().join('|'), @@ -70,7 +91,7 @@ export function buildSyncDigestNotification(digestLines: string[]): Notification export function buildLowBalanceNotification(lines: string[]): NotificationPayload | null { if (lines.length === 0) return null const plain = `Низкий баланс:\n\n${lines.join('\n')}` - const html = `💰 Низкий баланс\n\n${lines.join('\n')}` + const html = `💰 Низкий баланс\n\n${htmlLines(lines)}` return { event: 'low_balance', fingerprint: lines.sort().join('|'), @@ -88,7 +109,7 @@ export function buildNewTariffsNotification( if (tariffs.length === 0) return null const lines = tariffs.slice(0, 15).map((t) => `• ${t.name || '—'} — ${t.price || '—'}`) const plain = `Новые тарифы (${providerName}):\n\n${lines.join('\n')}` - const html = `🆕 Новые тарифы (${providerName}):\n\n${lines.join('\n')}` + const html = `🆕 Новые тарифы (${escapeHtml(providerName)}):\n\n${htmlLines(lines)}` const fingerprint = tariffs .map((t) => `${t.name}:${t.price}`) .sort() @@ -112,7 +133,7 @@ export function buildVpsHealthNotification( const title = event === 'vps_down' ? 'VPS недоступны' : 'VPS восстановлены' const icon = event === 'vps_down' ? '🔴' : '🟢' const plain = `${title}:\n\n${lines.join('\n')}` - const html = `${icon} ${title}\n\n${lines.join('\n')}` + const html = `${icon} ${title}\n\n${htmlLines(lines)}` const fingerprint = hosts .map((h) => h.id) .sort() @@ -123,8 +144,7 @@ export function buildVpsHealthNotification( messagePlain: plain, messageHtml: html, data: { hosts: hosts.map((h) => h.id) }, - dedup: 'state_transition', - stateKey: `vps_health:${event}`, - newStatus: fingerprint, + // newlyDown/newlyUp are already transitions (last_health_status); deduping again hid a repeat outage. + dedup: 'none', } } diff --git a/apps/api/src/services/notifications/types.ts b/apps/api/src/services/notifications/types.ts index e4ef0b7..9972204 100644 --- a/apps/api/src/services/notifications/types.ts +++ b/apps/api/src/services/notifications/types.ts @@ -11,7 +11,7 @@ export type NotificationEvent = (typeof NOTIFICATION_EVENTS)[number] export type NotificationChannel = 'telegram' | 'webhook' -export type DedupMode = 'daily' | 'fingerprint' | 'state_transition' +export type DedupMode = 'daily' | 'fingerprint' | 'state_transition' | 'none' export interface NotificationPayload { event: NotificationEvent diff --git a/apps/api/src/services/telegram.ts b/apps/api/src/services/telegram.ts index 218626f..49a76a7 100644 --- a/apps/api/src/services/telegram.ts +++ b/apps/api/src/services/telegram.ts @@ -61,6 +61,25 @@ export function formatTelegramApiError( return `${chatId}: ${fallback}` } +const TELEGRAM_TIMEOUT_MS = 10_000 +/** Telegram rejects text over 4096 chars; cut at a line break so no HTML entity/tag is split. */ +const TELEGRAM_MAX_TEXT = 4096 + +export function fitTelegramText(text: string): string { + if (text.length <= TELEGRAM_MAX_TEXT) return text + const lines = text.split('\n') + const kept: string[] = [] + let len = 0 + const reserve = 40 // room for the "and N more" tail + for (const line of lines) { + if (len + line.length + 1 > TELEGRAM_MAX_TEXT - reserve) break + kept.push(line) + len += line.length + 1 + } + if (kept.length === 0) return text.slice(0, TELEGRAM_MAX_TEXT - 1) + '…' + return `${kept.join('\n')}\n… и ещё ${lines.length - kept.length} строк` +} + export async function sendTelegramMessage( token: string, chatIds: string | string[], @@ -81,7 +100,7 @@ export async function sendTelegramMessage( const threadId = messageThreadId != null && messageThreadId !== '' ? Number(messageThreadId) : null const payload: Record = { - text, + text: fitTelegramText(text), parse_mode: 'HTML', disable_web_page_preview: true, } @@ -97,6 +116,8 @@ export async function sendTelegramMessage( method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ ...payload, chat_id: chatId }), + // A hung API/proxy would otherwise block the notify tick (and every tick skipped behind it). + signal: AbortSignal.timeout(TELEGRAM_TIMEOUT_MS), }) const rawBody = await res.text() let data: TelegramApiResponse = {} @@ -119,6 +140,7 @@ export async function sendTelegramMessage( } } - if (anyOk) return { ok: true } + // Partial delivery counts as sent, but the failed chats still reach the journal. + if (anyOk) return errors.length ? { ok: true, error: errors.join('; ') } : { ok: true } return { ok: false, error: errors.join('; ') || 'Не удалось отправить' } } diff --git a/apps/api/src/services/webhook.ts b/apps/api/src/services/webhook.ts index 4171dcc..0a4896a 100644 --- a/apps/api/src/services/webhook.ts +++ b/apps/api/src/services/webhook.ts @@ -4,37 +4,3 @@ export interface WebhookPayload { data?: Record timestamp: string } - -export async function sendWebhook(url: string, payload: WebhookPayload): Promise { - const target = url?.trim() - if (!target) return - try { - const res = await fetch(target, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(payload), - }) - if (!res.ok) { - console.warn(`Webhook failed (${res.status}): ${target}`) - } - } catch (err) { - console.warn('Webhook error:', err instanceof Error ? err.message : err) - } -} - -export async function notifyWebhook( - settings: { webhookUrl?: string | null; webhookEnabled?: number | boolean | null }, - event: string, - message: string, - data?: Record, -): Promise { - if (!settings.webhookEnabled) return - const url = settings.webhookUrl?.trim() - if (!url) return - await sendWebhook(url, { - event, - message, - data, - timestamp: new Date().toISOString(), - }) -} diff --git a/packages/db/src/repositories/notifications.ts b/packages/db/src/repositories/notifications.ts index 54f9e8d..f522655 100644 --- a/packages/db/src/repositories/notifications.ts +++ b/packages/db/src/repositories/notifications.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto' import { and, desc, eq } from 'drizzle-orm' import { getDb, schema } from '../index.js' -import { getCurrentSpaceId } from '../space-context.js' +import { getCurrentSpaceId, MAIN_SPACE_ID } from '../space-context.js' export type NotificationChannel = 'telegram' | 'webhook' export type NotificationLogStatus = 'sent' | 'failed' | 'skipped' @@ -42,6 +42,14 @@ function toLogDto(row: typeof schema.notificationLog.$inferSelect): Notification } } +/** + * notification_state.key is the table's sole PRIMARY KEY, so two spaces with the same event key collided + * (insert failed after sending → re-sent every tick). Non-main spaces get a prefixed key; main keeps old rows. + */ +function stateKeyForSpace(key: string, spaceId: string): string { + return spaceId === MAIN_SPACE_ID ? key : `${spaceId}::${key}` +} + export const notificationRepository = { listRecent(limit = 50): NotificationLogRow[] { const spaceId = getCurrentSpaceId() @@ -79,8 +87,9 @@ export const notificationRepository = { .run() }, - getState(key: string) { + getState(rawKey: string) { const spaceId = getCurrentSpaceId() + const key = stateKeyForSpace(rawKey, spaceId) return getDb() .select() .from(schema.notificationState) @@ -93,10 +102,11 @@ export const notificationRepository = { .get() }, - upsertState(key: string, patch: { lastFingerprint?: string; lastSentAt?: string; lastStatus?: string }) { + upsertState(rawKey: string, patch: { lastFingerprint?: string; lastSentAt?: string; lastStatus?: string }) { const db = getDb() const spaceId = getCurrentSpaceId() - const existing = this.getState(key) + const existing = this.getState(rawKey) + const key = stateKeyForSpace(rawKey, spaceId) const values = { key, spaceId, diff --git a/packages/db/src/runtime-migrate.test.ts b/packages/db/src/runtime-migrate.test.ts index 4db0b62..b45a132 100644 --- a/packages/db/src/runtime-migrate.test.ts +++ b/packages/db/src/runtime-migrate.test.ts @@ -65,3 +65,18 @@ describe('ensureRuntimeSchema (legacy DB without providers.apiType)', () => { expect(sqlite.prepare(`SELECT apiCredentials FROM provider_accounts`).get()).toEqual({ apiCredentials: 'u:p' }) }) }) + +describe('ensureRuntimeSchema (notification_state keys)', () => { + afterEach(() => resetRuntimeMigrate()) + + it('prefixes plain keys of non-main spaces so main can insert its own', () => { + const sqlite = new Database(':memory:') + resetRuntimeMigrate() + ensureRuntimeSchema(sqlite) + sqlite.exec(`INSERT INTO notification_state (key, spaceId) VALUES ('payment_expiry', 'space-x'), ('low_balance', 'space-main')`) + resetRuntimeMigrate() + ensureRuntimeSchema(sqlite) + const keys = (sqlite.prepare(`SELECT key FROM notification_state ORDER BY key`).all() as { key: string }[]).map((r) => r.key) + expect(keys).toEqual(['low_balance', 'space-x::payment_expiry']) + }) +}) diff --git a/packages/db/src/runtime-migrate.ts b/packages/db/src/runtime-migrate.ts index 95fd705..f60bd63 100644 --- a/packages/db/src/runtime-migrate.ts +++ b/packages/db/src/runtime-migrate.ts @@ -556,5 +556,10 @@ export function ensureRuntimeSchema(sqlite: Database.Database): void { rebuildVpsDomainsBindingUnique(sqlite) ensureMainSpace(sqlite) backfillSpaceIds(sqlite) + // Non-main spaces' notification_state rows written before key prefixing would block main's inserts. + sqlite.exec( + `UPDATE notification_state SET key = spaceId || '::' || key + WHERE spaceId != '${MAIN_SPACE_ID}' AND instr(key, '::') = 0`, + ) migrated = true }