Zakhar Timoshenko
666a0b268f
- Added TelegramStopAliasService for handling stop aliases in the database. - Created API endpoints for managing stop aliases (list, create, update, delete). - Introduced TelegramStopAliases component in the frontend for user interaction with stop aliases. - Updated TelegramImportService to utilize stop aliases for message processing. - Enhanced telegramCancellationParser to resolve stop names using aliases. - Added migration script to create telegram_stop_aliases table and seed initial data. - Updated documentation to include details about stop alias functionality.
207 lines
15 KiB
JavaScript
207 lines
15 KiB
JavaScript
const test = require('node:test');
|
||
const assert = require('node:assert/strict');
|
||
const fs = require('node:fs/promises');
|
||
const path = require('node:path');
|
||
const { TelegramImportService } = 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 });
|
||
|
||
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,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);
|
||
// 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 route_id')).rows;
|
||
assert.equal(alerts.length, 2);
|
||
assert.deepEqual(alerts.map(a => a.route_id), [1, 2]);
|
||
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 route_id')).rows;
|
||
assert.equal(alerts.length, 2);
|
||
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_import_alerts WHERE message_id=(SELECT id FROM telegram_import_messages WHERE message_id=1167) AND route_id=1')).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_import_alerts WHERE message_id=(SELECT id FROM telegram_import_messages WHERE message_id=1167) AND route_id=1')).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 all alerts if a later route 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 alerts/.test(sql) && args[0] === 2) 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_import_alerts t ON t.alert_id=a.id WHERE t.message_id=$1 ORDER BY route_id', [row.id])).rows;
|
||
assert.equal(alerts.length, 2);
|
||
assert.match(alerts[1].message, /Зодчего Иоанна — 15:58, 16:58/);
|
||
row = (await service.list(1)).find(m => m.message_id === 1200);
|
||
assert.ok(row.drafts.some(d => d.routeId === 2 && 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[1].id])).rows[0].message, alerts[1].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_import_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(); }
|
||
});
|