fix(notifications): изолировать уведомления по space и чинить отправку в Telegram

- «Истекает оплата» строится только по 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 <noreply@anthropic.com>

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 }

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()

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>: <html>502 & co'])
+    expect(p?.messageHtml).toContain('Acc &lt;1&gt;: &lt;html&gt;502 &amp; co')
+    expect(p?.messageHtml).toContain('<b>')
+  })
+
+  it('cuts long text at a line break under the Telegram limit', () => {
+    const text = Array.from({ length: 500 }, (_, i) => `• host-${i} &amp; 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<typeof fetch>(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<typeof fetch>(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)
+  })
+})

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, '&amp;').replace(/</g, '&lt;').replace(/>/g, '&gt;')
+}
+
+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 = `⚠️ <b>Истекает оплата</b> (ближайшие ${UPCOMING_DAYS} дней):\n\n${lines.join('\n')}`
+  const html = `⚠️ <b>Истекает оплата</b> (ближайшие ${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 = `📋 <b>Синхронизация VPS</b>\n\n${changed.join('\n')}`
+  const html = `📋 <b>Синхронизация VPS</b>\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 = `💰 <b>Низкий баланс</b>\n\n${lines.join('\n')}`
+  const html = `💰 <b>Низкий баланс</b>\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 = `🆕 <b>Новые тарифы</b> (${providerName}):\n\n${lines.join('\n')}`
+  const html = `🆕 <b>Новые тарифы</b> (${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} <b>${title}</b>\n\n${lines.join('\n')}`
+  const html = `${icon} <b>${title}</b>\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',
   }
 }

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

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<string, unknown> = {
-    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('; ') || 'Не удалось отправить' }
 }

apps/api/src/services/webhook.ts

@@ -4,37 +4,3 @@ export interface WebhookPayload {
   data?: Record<string, unknown>
   timestamp: string
 }
-
-export async function sendWebhook(url: string, payload: WebhookPayload): Promise<void> {
-  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<string, unknown>,
-): Promise<void> {
-  if (!settings.webhookEnabled) return
-  const url = settings.webhookUrl?.trim()
-  if (!url) return
-  await sendWebhook(url, {
-    event,
-    message,
-    data,
-    timestamp: new Date().toISOString(),
-  })
-}

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,

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'])
+  })
+})

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
 }