ridebus-admin/backend/test/telegramImport.integration.test.js

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