const test = require('node:test'); const assert = require('node:assert/strict'); const fs = require('node:fs/promises'); const path = require('node:path'); const { TelegramImportService, contentHash } = require('../src/services/telegramImportService'); const { CLAIM_DUE_ALERTS_SQL } = require('../src/services/alertSchedulerService'); // Run against an isolated in-memory PostgreSQL engine, never the working database. // TELEGRAM_TEST_PGLITE_MODULE can point to a temporary npm installation. const modulePath = process.env.TELEGRAM_TEST_PGLITE_MODULE; const carrierText = 'Не будут выполняться рейсы:\nПО МАРШРУТУ № 27 С ОСТАНОВКИ АВ/ВОКЗАЛ НА 15.43,16.43.\nС ОСТАНОВКИ ЗОДЧЕГО НА 15.58,16.58.'; const config = { enabled: true, autoPublish: true, apiId: 123, apiHash: 'a'.repeat(32), sessionFile: '/tmp/private/account.session', channel: 'avp2Polotsk', citySlug: 'polotsk', intervalMs: 60_000 }; const message = (id = 1166, text = carrierText) => ({ channelId: '1234567', channelUsername: 'avp2Polotsk', messageId: id, publishedAt: '2026-10-02T12:52:00Z', editedAt: null, hasMedia: false, text }); test('fixed parser reprocesses a stored invalid-date review without a Telegram text edit or duplicate alerts', { skip: !modulePath }, async () => { const { db, pool, service, clock } = await setup(); try { clock('2026-10-05T04:20:00Z'); await pool.query("UPDATE routes SET route_number='2' WHERE city_id=1"); await pool.query("UPDATE stops SET name=CASE id WHEN 1 THEN 'Больница (Ксты)' ELSE 'Больница (Новка)' END WHERE city_id=1"); await service.stopAliases.save(1, { alias: 'больница ксты', stop_id: 1 }); await service.stopAliases.save(1, { alias: 'больница(новка)', stop_id: 2 }); const text = await fs.readFile(path.join(__dirname, 'fixtures/telegram-hospital-cancellation.txt'), 'utf8'); const input = { ...message(1400, text), publishedAt: '2026-10-05T04:16:53Z' }; const oldParsed = { status: 'review', reasons: ['Некорректная дата отмены.'], groups: [], plans: [] }; await pool.query(`INSERT INTO telegram_import_messages (city_id,channel_id,channel_username,message_id,published_at,content_hash,raw_text,status,reasons,parsed) VALUES (1,$1,$2,$3,$4,$5,$6,'review',$7::jsonb,$8::jsonb)`, [input.channelId,input.channelUsername,input.messageId,input.publishedAt,contentHash(input),input.text,JSON.stringify(oldParsed.reasons),JSON.stringify(oldParsed)]); assert.equal((await service.processMessage(input, 1)).status, 'published'); const row = (await service.list(1))[0]; assert.equal(row.parsed.serviceDate, '2026-10-05'); assert.deepEqual(row.reasons, []); const alerts = (await pool.query('SELECT a.* FROM alerts a JOIN telegram_message_alerts t ON t.alert_id=a.id WHERE t.message_id=$1 ORDER BY id', [row.id])).rows; assert.equal(alerts.length, 1); assert.match(alerts[0].message, /Больница \(Ксты\) — 07:15, 08:45/); assert.match(alerts[0].message, /Больница \(Новка\) — 08:00, 09:30/); assert.ok(alerts.every(alert => alert.end_time.toISOString() === '2026-10-05T21:00:00.000Z')); assert.equal((await service.processMessage(input, 1)).status, 'unchanged'); assert.equal((await pool.query('SELECT count(*)::int AS n FROM telegram_message_alerts WHERE message_id=$1', [row.id])).rows[0].n, 1); } finally { await db.close(); } }); async function setup() { const { PGlite } = require(modulePath); const db = new PGlite(); await db.exec(`CREATE TABLE cities(id serial PRIMARY KEY,slug text,is_active boolean DEFAULT true); CREATE TABLE users(id serial PRIMARY KEY); CREATE TABLE routes(id serial PRIMARY KEY,route_number text,name text,color text,city_id int REFERENCES cities(id),is_active boolean DEFAULT true); CREATE TABLE stops(id serial PRIMARY KEY,name text,address text,city_id int REFERENCES cities(id),is_active boolean DEFAULT true); CREATE TABLE route_stops(route_id int,stop_id int,sequence int); CREATE TABLE alerts(id serial PRIMARY KEY,route_id int REFERENCES routes(id),alert_type text,title text,message text, city_id int,start_time timestamp,end_time timestamp,is_active boolean DEFAULT true,created_at timestamp DEFAULT now(), notification_sent_at timestamptz,notification_claimed_at timestamptz,notification_attempts int DEFAULT 0); INSERT INTO cities(slug) VALUES ('polotsk'),('novopolotsk'); INSERT INTO users DEFAULT VALUES; INSERT INTO routes(route_number,name,city_id) VALUES ('27','Автовокзал → Зодчего',1),('27','Зодчего → Автовокзал',1),('27','Другой город',2); INSERT INTO stops(name,city_id) VALUES ('Автовокзал',1),('Зодчего Иоанна',1),('Остановка другого города',2); INSERT INTO route_stops VALUES (1,1,1),(2,2,1); INSERT INTO alerts(route_id,city_id,start_time,end_time,title,message,alert_type) VALUES (1,1,'2026-10-02 12:00','2026-10-02 21:00','Old','Old','info'); SET TIME ZONE 'Europe/Minsk';`); const migration = await fs.readFile(path.join(__dirname, '../migrations/013_telegram_alert_import.sql'), 'utf8'); await db.exec(migration); await db.exec(migration); // idempotent upgrades const aliasesMigration = await fs.readFile(path.join(__dirname, '../migrations/014_telegram_stop_aliases.sql'), 'utf8'); await db.exec(aliasesMigration); await db.exec(aliasesMigration); const targetsMigration = await fs.readFile(path.join(__dirname, '../migrations/015_alert_route_numbers.sql'), 'utf8'); await db.exec(targetsMigration); await db.exec(targetsMigration); await db.exec(await fs.readFile(path.join(__dirname, '../migrations/007_alert_creation_idempotency.sql'), 'utf8')); // A session is leased for the whole transaction, just like a pg Pool client. let tail = Promise.resolve(); async function lease() { const previous = tail; let release; tail = new Promise(resolve => { release = resolve; }); await previous; return { query: (sql, args) => db.query(sql, args), release }; } const pool = { connect: lease, query: async (sql, args) => { const client = await lease(); try { return await client.query(sql, args); } finally { client.release(); } } }; let now = new Date('2026-10-02T13:00:00Z'); const service = new TelegramImportService({ pool, config, source: { close: async () => {} }, logger: { warn() {}, error() {} }, now: () => now }); return { db, pool, service, clock: value => { now = new Date(value); } }; } test('Telegram SQL lifecycle: atomic import, duplicates, edits, expiry, deletion and city isolation', { skip: !modulePath }, async () => { const { db, pool, service, clock } = await setup(); try { assert.equal((await pool.query('SELECT end_time FROM alerts WHERE id=1')).rows[0].end_time.toISOString(), '2026-10-02T21:00:00.000Z'); const results = await Promise.all([service.processMessage(message(), 1), service.processMessage(message(), 1)]); assert.deepEqual(results.map(r => r.status), ['published', 'unchanged']); let alerts = (await pool.query('SELECT * FROM alerts WHERE notification_pending=true ORDER BY id')).rows; assert.equal(alerts.length, 1); assert.deepEqual(alerts[0].route_numbers, ['27']); assert.equal(alerts[0].end_time.toISOString(), '2026-10-02T21:00:00.000Z'); assert.match(alerts[0].message, /Автовокзал.*15:43, 16:43/); assert.ok(alerts.every(a => a.notification_sent_at === null)); // Delivered pushes aren't repeated on text edits, while the stored alert changes. await pool.query('UPDATE alerts SET notification_sent_at=now() WHERE notification_pending=true'); const edited = { ...message(), text: carrierText.replace('15.43', '15.45'), editedAt: '2026-10-02T13:01:00Z' }; assert.equal((await service.processMessage(edited, 1)).status, 'published'); alerts = (await pool.query('SELECT * FROM alerts WHERE notification_pending=true ORDER BY id')).rows; assert.equal(alerts.length, 1); assert.ok(alerts.every(a => a.notification_sent_at !== null)); assert.match(alerts[0].message, /15:45/); assert.equal((await service.processMessage(message(), 1)).status, 'unchanged', 'stale Telegram snapshot cannot undo an edit'); // Missing text after an edit withdraws the old alerts and requires a review. assert.equal((await service.processMessage({ ...edited, text: 'Движение восстановлено.', editedAt: '2026-10-02T13:02:00Z' }, 1)).status, 'review'); assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts WHERE notification_pending=true')).rows[0].n, 0); const row = (await service.list(1))[0]; const input = { action: 'publish', content_hash: row.content_hash, service_date: '2026-10-02', route_id: 1, title: 'Отмена рейса', message: 'Автовокзал, 15:45' }; await assert.rejects(service.review(row.id, 2, input, 1), /не найдено/); await assert.rejects(service.review(row.id, 1, { ...input, route_id: 3 }, 1), /выбранного города/); await assert.rejects(service.review(row.id, 1, { ...input, content_hash: 'stale' }, 1), /изменилась/); await service.review(row.id, 1, input, 1); await service.review(row.id, 1, { action: 'dismiss', content_hash: row.content_hash }, 1); assert.equal((await service.processMessage({ ...edited, editedAt: '2026-10-02T14:00:00Z' }, 1)).status, 'dismissed'); assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts WHERE notification_pending=true')).rows[0].n, 0); // Administrator deletion is respected on a subsequent source edit. await service.processMessage(message(1167), 1); const deletedId = (await pool.query('SELECT alert_id FROM telegram_message_alerts WHERE message_id=(SELECT id FROM telegram_import_messages WHERE message_id=1167)')).rows[0].alert_id; await pool.query('DELETE FROM alerts WHERE id=$1', [deletedId]); await service.processMessage({ ...message(1167), text: carrierText.replace('15.43', '15.46'), editedAt: '2026-10-02T14:01:00Z' }, 1); assert.equal((await pool.query('SELECT alert_id FROM telegram_message_alerts WHERE message_id=(SELECT id FROM telegram_import_messages WHERE message_id=1167)')).rows[0].alert_id, null); // At midnight, late imports and manual publication cannot reactivate yesterday's cancellation. clock('2026-10-02T21:00:00Z'); assert.equal((await service.processMessage(message(1168), 1)).status, 'expired'); const expired = (await service.list(1)).find(m => m.message_id === 1168); await assert.rejects(service.review(expired.id, 1, { ...input, content_hash: expired.content_hash }, 1), /уже закончился/); assert.equal((await service.list(2)).length, 0); // Pending imported alerts are claimed even when start_time is before created_at. await pool.query(`UPDATE alerts SET is_active=true, notification_pending=true, start_time=now()-interval '1 minute', end_time=now()+interval '1 hour', notification_sent_at=NULL WHERE id<>1`); const claimed = await pool.query(CLAIM_DUE_ALERTS_SQL, [50]); assert.ok(claimed.rows.length > 0); assert.ok(claimed.rows.every(a => a.notification_attempts === 1)); assert.equal((await pool.query(CLAIM_DUE_ALERTS_SQL, [50])).rows.length, 0); } finally { await db.close(); } }); test('Telegram SQL import rolls back the alert if its source link insert fails', { skip: !modulePath }, async () => { const { db, pool, service } = await setup(); try { const connect = pool.connect; pool.connect = async () => { const client = await connect(); const query = client.query; client.query = (sql, args) => { if (/INSERT INTO telegram_message_alerts/.test(sql)) throw new Error('Injected failure'); return query(sql, args); }; return client; }; await assert.rejects(service.processMessage(message(), 1), /Injected failure/); assert.equal((await pool.query('SELECT count(*)::int AS n FROM telegram_import_messages')).rows[0].n, 0); assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts')).rows[0].n, 1); } finally { await db.close(); } }); test('stop dictionary validates city and uniqueness, resolves waiting messages and preserves published text', { skip: !modulePath }, async () => { const { db, pool, service, clock } = await setup(); try { const dictionary = service.stopAliases; let aliases = await dictionary.list(1); assert.equal(aliases.length, 2, 'migration seeds known abbreviations once'); const zodchego = aliases.find(a => a.alias === 'ЗОДЧЕГО'); assert.equal(zodchego.stop_name, 'Зодчего Иоанна'); await assert.rejects(dictionary.save(1, { alias: ' зод-чего ', stop_id: 2 }), error => error.status === 409); await assert.rejects(dictionary.save(1, { alias: 'Другой город', stop_id: 3 }), error => error.status === 400); await assert.rejects(dictionary.save(1, { alias: '?!', stop_id: 2 }), error => error.status === 400); await assert.rejects(dictionary.remove(zodchego.id, 2), error => error.status === 404); await dictionary.remove(zodchego.id, 1); const original = message(1200); assert.equal((await service.processMessage(original, 1)).status, 'review'); assert.equal((await service.processMessage(original, 1)).status, 'unchanged'); let row = (await service.list(1)).find(m => m.message_id === 1200); assert.match(row.drafts.find(d => d.routeId === null).message, /Автовокзал — 15:43, 16:43/); assert.doesNotMatch(row.drafts[0].message, /Уважаемые|С УВАЖЕНИЕМ/); const saved = await dictionary.save(1, { alias: 'ЗОДЧЕГО', stop_id: 2 }); assert.equal((await service.processMessage(original, 1)).status, 'published', 'unchanged Telegram post can be resolved after adding an alias'); const alerts = (await pool.query('SELECT a.* FROM alerts a JOIN telegram_message_alerts t ON t.alert_id=a.id WHERE t.message_id=$1 ORDER BY id', [row.id])).rows; assert.equal(alerts.length, 1); assert.match(alerts[0].message, /Зодчего Иоанна — 15:58, 16:58/); row = (await service.list(1)).find(m => m.message_id === 1200); assert.ok(row.drafts.some(d => d.routeNumber === '27' && d.message.includes('Зодчего Иоанна'))); await dictionary.save(1, { alias: 'ЗОДЧЕГО', stop_id: 1 }, saved.id); assert.equal((await service.processMessage(original, 1)).status, 'unchanged'); assert.equal((await pool.query('SELECT message FROM alerts WHERE id=$1', [alerts[0].id])).rows[0].message, alerts[0].message); await assert.rejects(dictionary.save(2, { alias: 'Чужое', stop_id: 3 }, saved.id), error => error.status === 404); // Inactive mappings never auto-publish, even for a route with one direction. await dictionary.save(1, { alias: 'ЗОДЧЕГО', stop_id: 2 }, saved.id); await pool.query('UPDATE stops SET is_active=false WHERE id=2'); assert.equal((await service.processMessage(message(1201), 1)).status, 'review'); const inactive = (await service.list(1)).find(m => m.message_id === 1201); assert.match(inactive.reasons[0], /отключена/); await assert.rejects(dictionary.save(1, { alias: 'Новое', stop_id: 2 }), error => error.status === 400); await pool.query('UPDATE stops SET is_active=true WHERE id=2'); clock('2026-10-02T21:00:00Z'); assert.equal((await service.processMessage(message(1201), 1)).status, 'expired', 'dictionary changes cannot publish yesterday after midnight'); assert.equal((await dictionary.list(2)).length, 0); // Re-running the migration never picks one of two identically named platforms. await dictionary.remove(saved.id, 1); await pool.query("INSERT INTO stops(name,city_id) VALUES ('Зодчего Иоанна',1)"); await db.exec(await fs.readFile(path.join(__dirname, '../migrations/014_telegram_stop_aliases.sql'), 'utf8')); assert.equal((await dictionary.list(1)).some(a => a.alias === 'ЗОДЧЕГО'), false); } finally { await db.close(); } }); test('reviewed snapshots respect auto-publish configuration changes and midnight without requiring a source edit', { skip: !modulePath }, async () => { const { db, pool, service, clock } = await setup(); // This test uses its own configuration so it cannot change other tests. service.config = { ...config, autoPublish: false }; try { assert.equal((await service.processMessage(message(1300), 1)).status, 'review'); assert.equal((await service.processMessage(message(1300), 1)).status, 'unchanged'); service.config.autoPublish = true; assert.equal((await service.processMessage(message(1300), 1)).status, 'published'); service.config.autoPublish = false; assert.equal((await service.processMessage(message(1301), 1)).status, 'review'); await service.processMessage(message(1302, carrierText.replace('\nС ОСТАНОВКИ ЗОДЧЕГО', '\nПО МАРШРУТУ № 27 С ОСТАНОВКИ ЗОДЧЕГО')), 1); const draft = (await service.list(1)).find(m => m.message_id === 1302).drafts.filter(d => d.routeId === null); assert.equal(draft.length, 1); assert.match(draft[0].message, /Автовокзал — 15:43, 16:43\nЗодчего Иоанна — 15:58, 16:58/); clock('2026-10-02T21:00:00Z'); assert.equal((await service.processMessage(message(1301), 1)).status, 'expired'); assert.equal((await pool.query('SELECT count(*)::int AS n FROM telegram_message_alerts t JOIN telegram_import_messages m ON m.id=t.message_id WHERE m.message_id=1301')).rows[0].n, 0); } finally { await db.close(); } }); test('several Telegram route numbers share one alert through import, review and edits', { skip: !modulePath }, async () => { const { db, pool, service } = await setup(); try { await pool.query("INSERT INTO routes(route_number,name,city_id) VALUES ('2','Туда',1),('2','Обратно',1),('99','Другой город',2)"); const multi = message(1500, carrierText + '\nПО МАРШРУТУ №2 ОТ ОСТАНОВКИ АВ/ВОКЗАЛ НА 18.00.'); assert.equal((await service.processMessage(multi, 1)).status, 'published'); let row = (await service.list(1))[0]; assert.equal(row.alert_ids.length, 1); const id = row.alert_ids[0]; let alert = (await pool.query('SELECT * FROM alerts WHERE id=$1', [id])).rows[0]; assert.deepEqual(alert.route_numbers, ['27', '2']); assert.match(alert.message, /Маршрут №27:/); assert.match(alert.message, /Маршрут №2:\nАвтовокзал — 18:00/); await pool.query('UPDATE alerts SET notification_sent_at=now() WHERE id=$1', [id]); const edited = { ...multi, text: multi.text.replace('18.00', '18.05'), editedAt: '2026-10-02T14:00:00Z' }; await service.processMessage(edited, 1); row = (await service.list(1))[0]; assert.deepEqual(row.alert_ids, [id]); assert.match((await pool.query('SELECT * FROM alerts WHERE id=$1', [id])).rows[0].message, /18:05/); const review = { action: 'publish', content_hash: row.content_hash, service_date: '2026-10-02', route_numbers: ['2', '27'], title: 'Отмены', message: 'Проверенный текст' }; await assert.rejects(service.review(row.id, 1, { ...review, route_numbers: [] }, 1), /хотя бы один/); await assert.rejects(service.review(row.id, 1, { ...review, route_numbers: ['27', '99'] }, 1), /выбранном городе/); await service.review(row.id, 1, review, 1); alert = (await pool.query('SELECT * FROM alerts WHERE id=$1', [id])).rows[0]; assert.deepEqual(alert.route_numbers, ['2', '27']); assert.ok(alert.notification_sent_at); assert.equal(alert.end_time.toISOString(), '2026-10-02T21:00:00.000Z'); // Removing a route section changes the targets on the same alert. await service.processMessage({ ...edited, text: carrierText, editedAt: '2026-10-02T15:00:00Z' }, 1); assert.deepEqual((await pool.query('SELECT route_numbers FROM alerts WHERE id=$1', [id])).rows[0].route_numbers, ['27']); const unknown = message(1501, multi.text.replace('№2 ', '№99 ')); assert.equal((await service.processMessage(unknown, 1)).status, 'review'); assert.equal((await service.list(1)).find(m => m.message_id === 1501).alert_ids.length, 0); // The scheduler includes both directions of every selected number. await pool.query("UPDATE alerts SET start_time=now()-interval '1 minute',end_time=now()+interval '1 hour',notification_sent_at=NULL WHERE id=$1", [id]); const claimed = (await pool.query(CLAIM_DUE_ALERTS_SQL, [50])).rows; assert.deepEqual(claimed.find(a => a.id === id).route_ids, [1, 2]); } finally { await db.close(); } }); test('manual alert API validates numbers, returns all directions, filters, edits and replays one push', { skip: !modulePath }, async () => { const { db, pool } = await setup(); const express = require('express'); const databasePath = require.resolve('../src/config/database'); const previous = require.cache[databasePath]; require.cache[databasePath] = { exports: pool }; delete require.cache[require.resolve('../src/routes/alerts')]; const { AlertCreationService, setAlertCreationServiceForTesting } = require('../src/services/alertCreationService'); const pushes = []; setAlertCreationServiceForTesting(new AlertCreationService({ pool, logger: { error() {} }, notificationService: { sendAlertCreated: async alert => { pushes.push(alert); } } })); const app = express(); app.use(express.json()); app.use((req, res, next) => { req.city = { id: Number(req.get('X-City-Id') || 1) }; next(); }); app.use('/alerts', require('../src/routes/alerts')); const server = await new Promise(resolve => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); }); const url = `http://127.0.0.1:${server.address().port}/alerts`; const request = async (path = '', method = 'GET', body, extra = {}) => { const response = await fetch(url + path, { method, headers: { 'Content-Type': 'application/json', ...extra }, ...(body ? { body: JSON.stringify(body) } : {}) }); return { status: response.status, data: await response.json() }; }; try { await pool.query("INSERT INTO routes(route_number,name,city_id) VALUES ('2','Туда',1),('2','Обратно',1),('99','Чужой',2),('5','Отключён',1)"); await pool.query('UPDATE routes SET is_active=false WHERE route_number=$1', ['5']); const input = { alert_type: 'detour', title: 'Переезд закрыт', message: 'Объезд для №2 и №27', start_time: new Date(Date.now() - 60000).toISOString(), end_time: new Date(Date.now() + 3600000).toISOString(), route_numbers: ['27', '2'] }; for (const numbers of [['99'], ['27', '5'], [2], null, ['']]) { assert.equal((await request('', 'POST', { ...input, route_numbers: numbers })).status, 400); } assert.equal(pushes.length, 0); const first = await request('', 'POST', input, { 'Idempotency-Key': 'multi-route' }); assert.equal(first.status, 201); assert.deepEqual(first.data.route_numbers, ['2', '27']); assert.deepEqual(first.data.route_ids, [1, 2, 4, 5]); assert.equal(pushes.length, 1); const replay = await request('', 'POST', { ...input, route_numbers: [' №2 ', '27', '2'] }, { 'Idempotency-Key': 'multi-route' }); assert.equal(replay.data.id, first.data.id); assert.equal(pushes.length, 1); assert.equal((await request('', 'POST', { ...input, route_numbers: ['27'] }, { 'Idempotency-Key': 'multi-route' })).status, 409); for (const routeId of [1, 2, 4, 5]) { const filtered = await request(`/active?route_id=${routeId}`); assert.equal(filtered.status, 200); assert.ok(filtered.data.some(a => a.id === first.data.id)); } assert.equal((await request('/active?route_id=3')).data.length, 0, 'another city is excluded'); assert.equal((await request('/' + first.data.id, 'GET', null, { 'X-City-Id': '2' })).status, 404); assert.equal((await request('/' + first.data.id, 'PUT', { route_numbers: ['99'] })).status, 400); assert.deepEqual((await request('/' + first.data.id)).data.route_numbers, ['2', '27']); const updated = await request('/' + first.data.id, 'PUT', { route_numbers: ['27'] }); assert.deepEqual(updated.data.route_ids, [1, 2]); assert.equal(updated.data.end_time, input.end_time, 'partial target edits preserve expiry'); assert.equal((await request(`/active?route_id=4`)).data.length, 0); const general = await request('/' + first.data.id, 'PUT', { route_numbers: [] }); assert.deepEqual(general.data.route_ids, []); assert.equal(general.data.route_id, null); const legacy = await request('', 'POST', { ...input, route_numbers: undefined, route_id: 2 }); assert.deepEqual(legacy.data.route_numbers, ['27']); assert.deepEqual(legacy.data.route_ids, [1, 2]); await pool.query("UPDATE alerts SET end_time=now()-interval '1 minute' WHERE id=1"); const cleanup = await request('/cleanup', 'POST', {}); assert.equal(cleanup.status, 200); assert.ok(cleanup.data.deactivated_count >= 1); const single = await request('', 'POST', { ...input, route_numbers: ['27'] }, { 'Idempotency-Key': 'single-number' }); assert.equal(single.status, 201); // Backfill and FK changes keep notifications when one direction is removed. assert.deepEqual((await pool.query('SELECT route_numbers FROM alerts WHERE id=1')).rows[0].route_numbers, ['27']); await pool.query('DELETE FROM routes WHERE id=1'); assert.equal((await pool.query('SELECT route_id FROM alerts WHERE id=1')).rows[0].route_id, null); assert.deepEqual((await request('/' + legacy.data.id)).data.route_ids, [2]); const singleReplay = await request('', 'POST', { ...input, route_numbers: ['27'] }, { 'Idempotency-Key': 'single-number' }); assert.equal(singleReplay.data.id, single.data.id, 'changing the compatibility anchor cannot change the request hash'); const newDirection = (await pool.query("INSERT INTO routes(route_number,name,city_id) VALUES ('27','Новое направление',1) RETURNING id")).rows[0].id; assert.deepEqual((await request('/' + legacy.data.id)).data.route_ids, [2, newDirection]); assert.ok((await request('/active?route_id=' + newDirection)).data.some(a => a.id === legacy.data.id)); } finally { await new Promise(resolve => server.close(resolve)); setAlertCreationServiceForTesting(null); if (previous) require.cache[databasePath] = previous; else delete require.cache[databasePath]; await db.close(); } }); test('legacy per-direction Telegram alerts consolidate on edit without repeating a delivered push', { skip: !modulePath }, async () => { const { db, pool, service } = await setup(); try { await service.processMessage(message(1600), 1); const row = (await service.list(1))[0]; const id = row.alert_ids[0]; await pool.query('DELETE FROM telegram_message_alerts WHERE message_id=$1', [row.id]); await pool.query('INSERT INTO telegram_import_alerts(message_id,route_id,alert_id) VALUES ($1,1,$2)', [row.id, id]); const second = (await pool.query(`INSERT INTO alerts(route_id,city_id,alert_type,title,message,start_time,end_time,notification_sent_at) VALUES (2,1,'cancellation','Old second','Old second',now(),now()+interval '1 hour',now()) RETURNING id`)).rows[0].id; await pool.query('INSERT INTO telegram_import_alerts(message_id,route_id,alert_id) VALUES ($1,2,$2)', [row.id, second]); await service.processMessage({ ...message(1600), text: carrierText.replace('15.43','15.44'), editedAt:'2026-10-02T14:00Z' }, 1); assert.deepEqual((await service.list(1))[0].alert_ids, [id]); const alerts = (await pool.query('SELECT * FROM alerts WHERE id=ANY($1::int[]) ORDER BY id', [[id, second]])).rows; assert.equal(alerts[0].is_active, true); assert.ok(alerts[0].notification_sent_at); assert.equal(alerts[1].is_active, false); assert.match(alerts[0].message, /Автовокзал.*15:44/); assert.match(alerts[0].message, /Зодчего Иоанна/); } finally { await db.close(); } }); async function setupRoute4() { const context = await setup(); await context.pool.query("UPDATE routes SET route_number='4' WHERE city_id=1"); await context.pool.query("UPDATE stops SET name=CASE id WHEN 1 THEN 'Мариненко' ELSE 'Боровуха' END WHERE city_id=1"); context.clock('2026-10-08T16:20:00Z'); return context; } async function route4Message(id, fixture = 'telegram-delay-route-4.txt') { const text = await fs.readFile(path.join(__dirname, 'fixtures', fixture), 'utf8'); return { ...message(id, text), publishedAt: fixture.includes('delay') ? '2026-10-08T16:14:01Z' : '2026-10-08T02:03:13Z' }; } async function seedLegacyReview(pool, input, parsed, reasons) { return (await pool.query(`INSERT INTO telegram_import_messages (city_id,channel_id,channel_username,message_id,published_at,content_hash,raw_text,status,reasons,parsed) VALUES (1,$1,$2,$3,$4,$5,$6,'review',$7::jsonb,$8::jsonb) RETURNING id`, [input.channelId,input.channelUsername,input.messageId,input.publishedAt,contentHash(input),input.text,JSON.stringify(reasons),JSON.stringify(parsed)])).rows[0].id; } test('screenshot reviews reprocess unchanged text into correctly typed alerts with no duplicates', { skip: !modulePath }, async () => { const { db, pool, service } = await setupRoute4(); try { const delayed = await route4Message(1800); const cancelled = await route4Message(1801, 'telegram-stop-point-cancellation.txt'); await seedLegacyReview(pool, delayed, { status: 'ignored', groups: [], plans: [] }, ['Текст исходной публикации изменился: отмена больше не распознаётся.']); await seedLegacyReview(pool, cancelled, { status: 'review', groups: [], plans: [] }, ['Некорректная дата отмены.']); for (const input of [delayed, cancelled]) { assert.equal((await service.processMessage(input, 1)).status, 'published'); assert.equal((await service.processMessage(input, 1)).status, 'unchanged'); } const rows = await service.list(1); const delay = rows.find(row => row.message_id === delayed.messageId); const cancellation = rows.find(row => row.message_id === cancelled.messageId); assert.deepEqual(delay.reasons, []); assert.deepEqual(cancellation.reasons, []); assert.equal(delay.parsed.alertType, 'delay'); assert.equal(cancellation.parsed.alertType, 'cancellation'); assert.equal(delay.alert_ids.length, 1); assert.equal(cancellation.alert_ids.length, 1); assert.match(delay.drafts[0].title, /Задержка/); assert.match(delay.drafts[0].message, /Мариненко — 19:05\nБоровуха — 20:00/); assert.match(cancellation.drafts[0].message, /Мариненко — 06:00, 07:30\nБоровуха — 06:40, 08:15/); const alerts = (await pool.query('SELECT * FROM alerts WHERE notification_pending=true ORDER BY id')).rows; assert.deepEqual(alerts.map(alert => alert.alert_type), ['delay', 'cancellation']); assert.ok(alerts.every(alert => alert.end_time.toISOString() === '2026-10-08T21:00:00.000Z')); assert.equal((await pool.query('SELECT count(*)::int AS n FROM telegram_message_alerts')).rows[0].n, 2); } finally { await db.close(); } }); test('editing a cancellation into a delay updates its type and content on the same delivered alert', { skip: !modulePath }, async () => { const { db, pool, service } = await setupRoute4(); try { const input = await route4Message(1802, 'telegram-stop-point-cancellation.txt'); await service.processMessage(input, 1); const row = (await service.list(1))[0]; const id = row.alert_ids[0]; await pool.query('UPDATE alerts SET notification_sent_at=now() WHERE id=$1', [id]); const delay = await route4Message(1802); await service.processMessage({ ...input, text: delay.text, editedAt: '2026-10-08T16:15:00Z' }, 1); const updated = (await pool.query('SELECT * FROM alerts WHERE id=$1', [id])).rows[0]; assert.equal(updated.alert_type, 'delay'); assert.match(updated.title, /Задержка/); assert.match(updated.message, /Мариненко — 19:05\nБоровуха — 20:00/); assert.ok(updated.notification_sent_at, 'the delivered push is not repeated'); assert.deepEqual((await service.list(1))[0].alert_ids, [id]); assert.equal((await service.processMessage(input, 1)).status, 'unchanged', 'older source cannot restore the cancellation'); } finally { await db.close(); } }); test('manual Telegram review defaults to delay, permits an explicit type choice, and rejects unrelated types', { skip: !modulePath }, async () => { const { db, pool, service } = await setupRoute4(); service.config = { ...config, autoPublish: false }; try { const input = await route4Message(1803); assert.equal((await service.processMessage(input, 1)).status, 'review'); const row = (await service.list(1))[0]; const values = { action: 'publish', content_hash: row.content_hash, service_date: '2026-10-08', route_numbers: ['4'], ...row.drafts[0] }; await assert.rejects(service.review(row.id, 1, { ...values, alert_type: 'info' }, 1), error => error.status === 400); await service.review(row.id, 1, values, 1); const saved = (await pool.query('SELECT * FROM alerts WHERE notification_pending=true')).rows[0]; assert.equal(saved.alert_type, 'delay'); assert.equal(saved.end_time.toISOString(), '2026-10-08T21:00:00.000Z'); await service.review(row.id, 1, { ...values, alert_type: 'cancellation', title: 'Уточнённая отмена', message: 'Уточнённый текст' }, 1); assert.equal((await pool.query('SELECT alert_type FROM alerts WHERE id=$1', [saved.id])).rows[0].alert_type, 'cancellation'); assert.equal((await service.list(1))[0].parsed.alertType, 'cancellation'); } finally { await db.close(); } }); test('past screenshot messages become expired on parser upgrade and cannot send a new push after midnight', { skip: !modulePath }, async () => { const { db, pool, service, clock } = await setupRoute4(); clock('2026-10-08T21:00:00Z'); try { for (const [id, fixture] of [[1804, 'telegram-delay-route-4.txt'], [1805, 'telegram-stop-point-cancellation.txt']]) { const input = await route4Message(id, fixture); await seedLegacyReview(pool, input, { status: 'review', groups: [] }, ['Старая ошибка']); assert.equal((await service.processMessage(input, 1)).status, 'expired'); assert.equal((await service.processMessage(input, 1)).status, 'unchanged'); } const rows = await service.list(1); assert.ok(rows.every(row => row.status === 'expired' && row.alert_ids.length === 0)); assert.ok(rows.every(row => row.parsed.serviceDate === '2026-10-08')); assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts WHERE notification_pending=true')).rows[0].n, 0); const delay = rows.find(row => row.parsed.alertType === 'delay'); await assert.rejects(service.review(delay.id, 1, { action: 'publish', content_hash: delay.content_hash, service_date: '2026-10-08', route_numbers: ['4'], title: 'Опоздание', message: 'Текст' }, 1), error => error.status === 410); } finally { await db.close(); } }); test('parser upgrade refreshes ambiguous delay review once and does not auto-publish attachments', { skip: !modulePath }, async () => { const { db, pool, service } = await setupRoute4(); try { const input = { ...await route4Message(1806), hasMedia: true }; await seedLegacyReview(pool, input, { status: 'ignored', groups: [] }, ['Отмена не распознана']); assert.equal((await service.processMessage(input, 1)).status, 'review'); const row = (await service.list(1))[0]; assert.equal(row.parsed.alertType, 'delay'); assert.match(row.reasons[0], /вложение/); assert.equal(row.alert_ids.length, 0); assert.equal((await service.processMessage(input, 1)).status, 'unchanged'); await service.review(row.id, 1, { action: 'dismiss', content_hash: row.content_hash }, 1); assert.equal((await service.processMessage({ ...input, hasMedia: false, editedAt: '2026-10-08T16:16:00Z' }, 1)).status, 'dismissed'); assert.equal((await pool.query('SELECT count(*)::int AS n FROM alerts WHERE notification_pending=true')).rows[0].n, 0); } finally { await db.close(); } });