Add Telegram cancellation import functionality with tests and documentation

- Implemented Telegram cancellation parsing logic in `telegramCancellationParser.js`.
- Created unit tests for cancellation parsing in `telegramCancellationParser.test.js`.
- Developed integration tests for Telegram import service in `telegramImport.integration.test.js`.
- Added API tests for Telegram import endpoints in `telegramImportApi.test.js`.
- Introduced service tests for Telegram import logic in `telegramImportService.test.js`.
- Documented the Telegram cancellation import process in `telegram-cancellation-import.md`.
- Built a React component for managing Telegram imports in `TelegramImportPanel.jsx`.

backend/src/config/telegram.js

@@ -0,0 +1,47 @@
+const fs = require('node:fs/promises');
+const path = require('node:path');
+
+function getTelegramConfig(env = process.env) {
+  return {
+    enabled: env.TELEGRAM_IMPORT_ENABLED === 'true',
+    autoPublish: env.TELEGRAM_AUTO_PUBLISH !== 'false',
+    apiId: Number(env.TELEGRAM_API_ID),
+    apiHash: env.TELEGRAM_API_HASH || '',
+    sessionFile: env.TELEGRAM_SESSION_FILE || '',
+    channel: (env.TELEGRAM_CHANNEL || 'avp2Polotsk').replace(/^@/, ''),
+    citySlug: env.TELEGRAM_CITY_SLUG || 'polotsk',
+    intervalMs: Math.max(30_000, Number(env.TELEGRAM_POLL_INTERVAL_MS) || 60_000)
+  };
+}
+
+function missingTelegramConfig(config) {
+  const missing = [];
+  if (!Number.isSafeInteger(config.apiId) || config.apiId <= 0) missing.push('TELEGRAM_API_ID');
+  if (!/^[a-f0-9]{32}$/i.test(config.apiHash)) missing.push('TELEGRAM_API_HASH');
+  if (!path.isAbsolute(config.sessionFile)) missing.push('TELEGRAM_SESSION_FILE');
+  if (!/^[a-zA-Z][a-zA-Z0-9_]{4,31}$/.test(config.channel)) missing.push('TELEGRAM_CHANNEL');
+  if (!/^[a-z0-9]+(?:-[a-z0-9]+)*$/.test(config.citySlug)) missing.push('TELEGRAM_CITY_SLUG');
+  return missing;
+}
+
+async function assertExternalSessionFile(file) {
+  if (!path.isAbsolute(file)) throw new Error('SESSION_FILE_INVALID');
+  const root = await fs.realpath(path.resolve(__dirname, '../../..'));
+  // Resolve the parent as well, so a symlink cannot put the credential in the checkout.
+  const parent = await fs.realpath(path.dirname(file));
+  const resolved = path.join(parent, path.basename(file));
+  if (resolved === root || resolved.startsWith(`${root}${path.sep}`)) {
+    throw new Error('SESSION_FILE_IN_CHECKOUT');
+  }
+  try {
+    const stat = await fs.lstat(resolved);
+    if (!stat.isFile() || stat.isSymbolicLink() || (stat.mode & 0o077) !== 0) {
+      throw new Error('SESSION_FILE_PERMISSIONS');
+    }
+  } catch (error) {
+    if (error.code !== 'ENOENT') throw error;
+  }
+  return resolved;
+}
+
+module.exports = { getTelegramConfig, missingTelegramConfig, assertExternalSessionFile };

backend/src/index.js

@@ -13,11 +13,13 @@ const { ensureRealtimeSchema } = require('./services/realtimeSchema');
 const { attachRealtimeServer, closeRealtimeServer } = require('./services/realtimeService');
 const { ensureBootstrapAdmin } = require('./utils/bootstrapAdmin');
 const { startAlertScheduler } = require('./services/alertSchedulerService');
+const { startTelegramImporter } = require('./services/telegramImportService');
 
 const app = express();
 const PORT = process.env.PORT || 3000;
 let server;
 let stopAlertScheduler = () => {};
+let stopTelegramImporter = () => Promise.resolve();
 
 // Security middleware
 app.use(helmet({
@@ -128,6 +130,7 @@ function shutdown(signal) {
 
   closeRealtimeServer();
   stopAlertScheduler();
+  stopTelegramImporter().catch(() => {});
 
   if (!server) {
     process.exit(0);
@@ -168,6 +171,7 @@ async function startServer() {
     await ensureRealtimeSchema();
     await ensureBootstrapAdmin();
     stopAlertScheduler = startAlertScheduler();
+    stopTelegramImporter = startTelegramImporter();
 
     server = app.listen(PORT, () => {
       logger.info(`Server is running on port ${PORT}`);

backend/src/routes/alerts.js

@@ -70,7 +70,7 @@ router.get('/active', async (req, res) => {
       WHERE a.city_id = $1
         AND a.is_active = true
         AND a.start_time <= NOW()
-        AND (a.end_time IS NULL OR a.end_time >= NOW())
+        AND (a.end_time IS NULL OR a.end_time > NOW())
     `;
     const params = [req.city.id];
 
@@ -241,7 +241,7 @@ router.post('/cleanup', async (req, res) => {
        SET is_active = false 
        WHERE city_id = $1 AND is_active = true
          AND end_time IS NOT NULL 
-         AND end_time < NOW()
+         AND end_time <= NOW()
        RETURNING id`
     , [req.city.id]);
 

backend/src/routes/api.js

@@ -24,6 +24,7 @@ router.use('/cities', require('./cities'));
 router.use(resolveCity);
 
 router.use('/schedule-import', require('./scheduleImport'));
+router.use('/telegram-import', require('./telegramImport'));
 router.get('/quality', authenticateToken, requireAdmin, qualityController.getReport.bind(qualityController));
 router.get('/schedule-center', authenticateToken, requireAdmin, scheduleCenterController.getOverview.bind(scheduleCenterController));
 

backend/src/routes/telegramImport.js

@@ -0,0 +1,33 @@
+const express = require('express');
+const { authenticateToken, requireAdmin } = require('../middleware/auth');
+const { getTelegramImportService } = require('../services/telegramImportService');
+
+function createTelegramImportRouter(service = getTelegramImportService()) {
+  const router = express.Router();
+  router.use(authenticateToken, requireAdmin);
+  const handle = action => async (req, res) => {
+    try { res.json(await action(req)); }
+    catch (error) { res.status(error.status || 500).json({ error: error.status ? error.message : 'Не удалось обработать импорт Telegram. Проверьте миграцию 013 и подключение базы.' }); }
+  };
+  router.get('/', handle(async req => ({ status: service.getStatus(), messages: await service.list(req.city.id) })));
+  router.post('/sync', handle(req => {
+    if (req.city.slug !== service.config.citySlug) {
+      const error = new Error('Источник Telegram настроен для другого города.');
+      error.status = 409;
+      throw error;
+    }
+    return service.sync();
+  }));
+  router.post('/:id/review', handle(req => {
+    if (!/^\d+$/.test(req.params.id)) {
+      const error = new Error('Некорректный идентификатор сообщения.');
+      error.status = 400;
+      throw error;
+    }
+    return service.review(Number(req.params.id), req.city.id, req.body, req.user.id);
+  }));
+  return router;
+}
+
+module.exports = createTelegramImportRouter();
+module.exports.createTelegramImportRouter = createTelegramImportRouter;

backend/src/services/alertCreationService.js

@@ -113,7 +113,7 @@ function isAlertDue(alert, now = new Date()) {
   }
 
   const endTime = new Date(alert.end_time);
-  return !Number.isNaN(endTime.getTime()) && endTime >= now;
+  return !Number.isNaN(endTime.getTime()) && endTime > now;
 }
 
 class AlertCreationService {

backend/src/services/alertNotificationService.js

@@ -90,6 +90,8 @@ function isAlertActive(alert = {}) {
 function buildAlertCreatedMessage(alert, options = {}) {
   const alertId = getAlertId(alert);
   const topic = options.topic || DEFAULT_FCM_TOPIC;
+  const endTime = toIsoString(firstDefined(alert.end_time, alert.endTime));
+  const now = options.now || new Date();
 
   return {
     topic,
@@ -100,12 +102,13 @@ function buildAlertCreatedMessage(alert, options = {}) {
       routeId: getRouteId(alert),
       routeNumber: getRouteNumber(alert),
       startTime: toIsoString(firstDefined(alert.start_time, alert.startTime)),
-      endTime: toIsoString(firstDefined(alert.end_time, alert.endTime)),
+      endTime,
       title: getNotificationTitle(alert),
       body: getNotificationBody(alert)
     },
     android: {
-      priority: 'high'
+      priority: 'high',
+      ...(endTime ? { ttl: Math.max(0, Math.min(28 * 86_400_000, Date.parse(endTime) - new Date(now).getTime())) } : {})
     }
   };
 }
@@ -124,6 +127,7 @@ class AlertNotificationService {
     this.enabled = options.enabled === undefined ? firebaseConfig.enabled : options.enabled;
     this.topic = options.topic || firebaseConfig.topic;
     this.logger = options.logger || getDefaultLogger();
+    this.now = options.now || (() => new Date());
     this.getMessaging = options.getMessaging || options.messagingFactory || (() => (
       options.messaging || getFirebaseMessaging({
         ...firebaseConfig,
@@ -144,13 +148,18 @@ class AlertNotificationService {
       return { sent: false, skipped: true, reason: 'inactive' };
     }
 
+    const endTime = toIsoString(firstDefined(alert.end_time, alert.endTime));
+    if (endTime && Date.parse(endTime) <= this.now().getTime()) {
+      return { sent: false, skipped: true, reason: 'expired' };
+    }
+
     try {
       const messaging = await this.getMessaging();
       if (!messaging || typeof messaging.send !== 'function') {
         throw new Error('Firebase Messaging is unavailable');
       }
 
-      const message = buildAlertCreatedMessage(alert, { topic: this.topic });
+      const message = buildAlertCreatedMessage(alert, { topic: this.topic, now: this.now() });
       const messageId = await messaging.send(message);
       const normalizedMessageId = toDataString(messageId);
 

backend/src/services/alertSchedulerService.js

@@ -3,9 +3,9 @@ const CLAIM_DUE_ALERTS_SQL = `
     SELECT a.id
     FROM alerts a
     WHERE a.is_active = true
-      AND a.start_time > a.created_at
+      AND (a.start_time > a.created_at OR a.notification_pending = true)
       AND a.start_time <= CURRENT_TIMESTAMP
-      AND (a.end_time IS NULL OR a.end_time >= CURRENT_TIMESTAMP)
+      AND (a.end_time IS NULL OR a.end_time > CURRENT_TIMESTAMP)
       AND a.notification_sent_at IS NULL
       AND a.notification_attempts < 5
       AND (

backend/src/services/telegramCancellationParser.js

@@ -0,0 +1,139 @@
+const { getLocalDateTime } = require('../utils/etaTime');
+
+const MONTHS = ['января', 'февраля', 'марта', 'апреля', 'мая', 'июня', 'июля', 'августа', 'сентября', 'октября', 'ноября', 'декабря'];
+const ROUTE = /маршрут(?:у|а|е|ом)?\s*(?:№|номер|no\.?)?\s*(\d{1,3}[а-яa-z]?)(?![\dа-яa-z])/giu;
+const TIME = /(?<!\d)(\d{1,2})[:.](\d{2})(?!\d|[.:]\d)/g;
+
+function validDate(value) {
+  if (!/^\d{4}-\d{2}-\d{2}$/.test(value || '')) return false;
+  const date = new Date(`${value}T12:00:00Z`);
+  return !Number.isNaN(date.getTime()) && date.toISOString().slice(0, 10) === value;
+}
+
+function addDays(value, days) {
+  const date = new Date(`${value}T12:00:00Z`);
+  date.setUTCDate(date.getUTCDate() + days);
+  return date.toISOString().slice(0, 10);
+}
+
+function dayBounds(value) {
+  if (!validDate(value)) throw new Error('INVALID_SERVICE_DATE');
+  return {
+    start: new Date(`${value}T00:00:00+03:00`).toISOString(),
+    end: new Date(`${addDays(value, 1)}T00:00:00+03:00`).toISOString()
+  };
+}
+
+function inferYear(day, month, year, publicationDate) {
+  const format = y => `${y}-${String(month).padStart(2, '0')}-${String(day).padStart(2, '0')}`;
+  if (year) return format(Number(year));
+  const base = Number(publicationDate.slice(0, 4));
+  return [base - 1, base, base + 1].map(format).filter(validDate)
+    .sort((a, b) => Math.abs(Date.parse(a) - Date.parse(publicationDate)) - Math.abs(Date.parse(b) - Date.parse(publicationDate)))[0] || format(base);
+}
+
+function parseDate(text, publishedAt) {
+  const publicationDate = getLocalDateTime(new Date(publishedAt)).date;
+  const dates = [];
+  let dateText = text;
+  const patterns = [
+    /(?<!\d)(\d{1,2})[./](\d{1,2})[./](\d{4})(?!\d)/g,
+    /(?:^|\n|(?:на|с|по|за|дата:?)\s+)(\d{1,2})[./](\d{2})(?![\d.:])/gi,
+    new RegExp(`(?<!\\d)(\\d{1,2})\\s+(${MONTHS.join('|')})(?:\\s+(\\d{4}))?`, 'gi')
+  ];
+  for (const [i, pattern] of patterns.entries()) {
+    dateText = dateText.replace(pattern, (...args) => {
+      const [full, d, m] = args;
+      const y = i === 1 ? undefined : args[3];
+      const offset = args.at(-2);
+      // In carrier posts “с остановки ... на 09.10” is a departure, not a date.
+      if (i === 1 && /с\s+остановк[иа]\s+[^\n]*$/i.test(dateText.slice(0, offset))) return full;
+      dates.push(inferYear(d, i === 2 ? MONTHS.indexOf(m.toLowerCase()) + 1 : m, y, publicationDate));
+      return ' ';
+    });
+  }
+  if (/сегодня/i.test(text)) dates.push(publicationDate);
+  if (/завтра/i.test(text)) dates.push(addDays(publicationDate, 1));
+  if (/послезавтра/i.test(text)) return { reason: 'Дата «послезавтра» требует проверки.' };
+  if (dates.some(d => !validDate(d))) return { reason: 'Некорректная дата отмены.' };
+  if (new Set(dates).size > 1) return { reason: 'Указаны разные даты отмены.' };
+  return { date: dates[0] || publicationDate, dateText, inferred: dates.length === 0 };
+}
+
+function normalizeRoute(value) { return String(value).trim().toLowerCase(); }
+function normalizeStop(value) {
+  const key = String(value).toLowerCase().replace(/ё/g, 'е').replace(/[^а-яa-z0-9]/g, '');
+  return key === 'аввокзал' ? 'автовокзал' : key;
+}
+function canonicalStop(value) { return normalizeStop(value) === 'автовокзал' ? 'Автовокзал' : value; }
+
+function parseCancellation({ text = '', publishedAt, hasMedia = false }) {
+  const normalized = text.replace(/\u00a0/g, ' ').replace(/\r/g, '').trim();
+  let serviceDate;
+  const review = reason => ({ status: 'review', reasons: [reason], groups: [], ...(serviceDate ? { serviceDate } : {}) });
+  if (!normalized) return hasMedia ? review('Публикация содержит вложение без текста. Нужна проверка.') : { status: 'ignored', reasons: [], groups: [] };
+  if (!/(?:не\s+буд(?:ут|ет)\s+выполняться|не\s+выполняются|отмен(?:ены|яется|яются|а\s+рейс)|рейс[^\n]*отмен)/i.test(normalized)) {
+    return { status: 'ignored', reasons: [], groups: [] };
+  }
+  if (hasMedia) return review('В публикации есть вложение. Проверьте все отмены по оригиналу.');
+  if (normalized.length > 20_000) return review('Слишком длинное сообщение. Нужна проверка.');
+  if (/возобнов|восстанов|кроме|только|ежеднев|еженедель|понедельник|вторник|сред[ау]|четверг|пятниц|суббот|воскресен|будн|выходн|период|дальнейш|неопредел|(?:с\s+\d[^\n]+\s+по\s+\d)|до\s+\d/i.test(normalized)) {
+    return review('Есть условия, период или восстановление движения. Нужна проверка.');
+  }
+  if (/буд(?:ут|ет)\s+выполняться/i.test(normalized.replace(/не\s+буд(?:ут|ет)\s+выполняться/gi, '')) || /отмена\s+отменена/i.test(normalized)) {
+    return review('В сообщении есть сведения о выполнении рейсов. Нужна проверка.');
+  }
+  if (Number.isNaN(new Date(publishedAt).getTime())) return review('Неизвестно время публикации.');
+  const parsedDate = parseDate(normalized, publishedAt);
+  if (parsedDate.reason) return review(parsedDate.reason);
+  serviceDate = parsedDate.date;
+  const routeMatches = [...parsedDate.dateText.matchAll(ROUTE)];
+  if (!routeMatches.length) return review('Не удалось определить номер маршрута.');
+  const groups = [];
+  for (const [i, route] of routeMatches.entries()) {
+    const section = parsedDate.dateText.slice(route.index + route[0].length, routeMatches[i + 1]?.index);
+    const stopPattern = /с\s+остановк[иа]\s+(.+?)\s+(?:на|в)\s+((?:\d{1,2}[:.]\d{2}(?:\s*[,;]\s*|\s+и\s+|\s+(?=\d))?)+)/giu;
+    const departures = [];
+    for (const stop of section.matchAll(stopPattern)) {
+      const times = [...stop[2].matchAll(TIME)].map(t => `${t[1].padStart(2, '0')}:${t[2]}`);
+      if (times.some(t => Number(t.slice(0, 2)) > 23 || Number(t.slice(3)) > 59)) return review('Некорректное время отправления.');
+      const stopName = stop[1].replace(/^[\s«"“]+|[\s»"”.,:;]+$/g, '').trim();
+      if (!stopName || !times.length) return review('Не удалось определить остановку и время.');
+      departures.push({ stop: canonicalStop(stopName), times: [...new Set(times)] });
+    }
+    if (!departures.length || [...section.matchAll(TIME)].length !== departures.reduce((n, d) => n + d.times.length, 0)) {
+      return review('Не удалось однозначно связать все времена с остановками.');
+    }
+    groups.push({ routeNumber: normalizeRoute(route[1]), departures });
+  }
+  return { status: 'ready', reasons: [], serviceDate: parsedDate.date, inferredDate: parsedDate.inferred, groups };
+}
+
+function matchCancellationRoutes(parsed, routes) {
+  const plans = new Map();
+  for (const group of parsed.groups) {
+    const candidates = routes.filter(r => normalizeRoute(r.route_number) === group.routeNumber);
+    for (const departure of group.departures) {
+      const matches = candidates.length === 1 ? candidates : candidates.filter(r => normalizeStop(r.origin_stop || '') === normalizeStop(departure.stop));
+      if (matches.length !== 1) return { reasons: [`Маршрут №${group.routeNumber}, остановка «${departure.stop}»: направление не определено однозначно.`], plans: [] };
+      const route = matches[0];
+      if (!plans.has(route.id)) plans.set(route.id, { routeId: route.id, routeNumber: route.route_number, departures: [] });
+      plans.get(route.id).departures.push(departure);
+    }
+  }
+  return { reasons: [], plans: [...plans.values()] };
+}
+
+function buildCancellationAlert(plan, serviceDate, sourceUrl, publishedAt, cityId) {
+  const bounds = dayBounds(serviceDate);
+  const dateLabel = serviceDate.split('-').reverse().join('.');
+  const start = new Date(Math.max(Date.parse(bounds.start), Date.parse(publishedAt))).toISOString();
+  return {
+    city_id: cityId, route_id: plan.routeId, alert_type: 'cancellation',
+    title: `Отмена отдельных рейсов маршрута №${plan.routeNumber}`,
+    message: `${dateLabel}: не выполняются отправления ${plan.departures.map(d => `с остановки «${d.stop}» в ${d.times.join(', ')}`).join('; ')}.\nИсточник: ${sourceUrl}`,
+    start_time: start, end_time: bounds.end, is_active: true
+  };
+}
+
+module.exports = { parseCancellation, matchCancellationRoutes, buildCancellationAlert, dayBounds, validDate, addDays };

backend/src/services/telegramImportService.js

@@ -0,0 +1,206 @@
+const crypto = require('node:crypto');
+const { getTelegramConfig, missingTelegramConfig } = require('../config/telegram');
+const { getLocalDateTime } = require('../utils/etaTime');
+const { TelegramSource } = require('./telegramSource');
+const { parseCancellation, matchCancellationRoutes, buildCancellationAlert, dayBounds, validDate, addDays } = require('./telegramCancellationParser');
+const { AlertCreationService } = require('./alertCreationService');
+
+function httpError(status, message) { return Object.assign(new Error(message), { status }); }
+function sourceUrl(row) { return `https://t.me/${row.channel_username}/${row.message_id}`; }
+function contentHash(message) {
+  return crypto.createHash('sha256').update(JSON.stringify([message.text, message.hasMedia, message.publishedAt])).digest('hex');
+}
+
+function publicError(error) {
+  const known = ['AUTH_REQUIRED', 'USER_ACCOUNT_REQUIRED', 'SESSION_MISSING', 'SESSION_FILE_INVALID', 'SESSION_FILE_IN_CHECKOUT', 'SESSION_FILE_PERMISSIONS', 'CHANNEL_INVALID', 'CHANNEL_PRIVATE', 'USERNAME_NOT_OCCUPIED', 'HISTORY_LIMIT', 'AUTH_KEY_UNREGISTERED', 'SESSION_REVOKED'];
+  const value = error.errorMessage || error.message;
+  if (known.includes(value)) return value;
+  if (error.code === 'ENOENT') return 'SESSION_MISSING';
+  if (error.seconds) return 'FLOOD_WAIT';
+  return 'IMPORT_FAILED';
+}
+
+class TelegramImportService {
+  constructor({ pool, config = getTelegramConfig(), source, logger, now = () => new Date() } = {}) {
+    this.pool = pool || require('../config/database');
+    this.config = config;
+    this.source = source || new TelegramSource(config);
+    this.logger = logger || require('../config/logger');
+    this.now = now;
+    this.running = false;
+    this.stopped = false;
+    this.state = { connection: config.enabled ? 'waiting' : 'disabled', lastSyncAt: null, lastError: null, retryAt: null };
+    this.alerts = new AlertCreationService({ pool: this.pool, logger: this.logger });
+  }
+
+  getStatus() {
+    const missing = missingTelegramConfig(this.config);
+    return {
+      ...this.state, enabled: this.config.enabled, autoPublish: this.config.autoPublish,
+      channel: this.config.channel, citySlug: this.config.citySlug,
+      connection: !this.config.enabled ? 'disabled' : missing.length ? 'configuration_required' : this.state.connection,
+      missing, syncing: this.running
+    };
+  }
+
+  async sync() {
+    if (!this.config.enabled) throw httpError(409, 'Импорт Telegram выключен в настройках сервера.');
+    if (missingTelegramConfig(this.config).length) throw httpError(409, 'Укажите Telegram API ID, API hash и файл авторизации на сервере.');
+    if (this.running || this.stopped) throw httpError(409, 'Импорт уже выполняется или остановлен.');
+    if (this.state.retryAt && Date.parse(this.state.retryAt) > this.now().getTime()) throw httpError(429, 'Telegram ограничил частоту запросов. Дождитесь следующей попытки.');
+    this.running = true;
+    this.state.connection = 'connecting';
+    try {
+      const city = (await this.pool.query('SELECT id FROM cities WHERE slug = $1 AND is_active = true', [this.config.citySlug])).rows[0];
+      if (!city) throw httpError(409, 'Город источника отсутствует или отключён.');
+      const today = getLocalDateTime(this.now()).date;
+      const messages = await this.source.fetchMessages(new Date(dayBounds(addDays(today, -2)).start));
+      const counts = { published: 0, review: 0, unchanged: 0, ignored: 0, expired: 0 };
+      for (const message of messages) {
+        if (this.stopped) break;
+        const result = await this.processMessage(message, city.id);
+        counts[result.status] = (counts[result.status] || 0) + 1;
+      }
+      this.state = { connection: 'ready', lastSyncAt: this.now().toISOString(), lastError: null, retryAt: null };
+      return counts;
+    } catch (error) {
+      const code = publicError(error);
+      const delay = error.seconds ? Math.min(86_400, Math.max(30, Number(error.seconds) || 60)) : 60;
+      this.state.connection = 'error';
+      this.state.lastError = code;
+      this.state.retryAt = new Date(this.now().getTime() + delay * 1000).toISOString();
+      this.logger.warn('Telegram channel import failed', { code });
+      await this.source.close().catch(() => {});
+      if (error.status) throw error;
+      throw httpError(503, `Не удалось прочитать канал Telegram (${code}). Проверьте подключение и авторизацию.`);
+    } finally { this.running = false; }
+  }
+
+  async getRoutes(db, cityId) {
+    return (await db.query(`SELECT r.id, r.route_number, r.name, origin.name AS origin_stop
+      FROM routes r LEFT JOIN LATERAL (
+        SELECT s.name FROM route_stops rs JOIN stops s ON s.id = rs.stop_id
+        WHERE rs.route_id = r.id ORDER BY rs.sequence LIMIT 1
+      ) origin ON true WHERE r.city_id = $1 AND r.is_active = true`, [cityId])).rows;
+  }
+
+  async processMessage(message, cityId) {
+    if (!Number.isSafeInteger(message.messageId) || message.messageId <= 0 || !/^\d+$/.test(message.channelId)
+      || message.channelUsername.toLowerCase() !== this.config.channel.toLowerCase()) throw new Error('CHANNEL_INVALID');
+    const parsed = parseCancellation(message);
+    const hash = contentHash(message);
+    const db = await this.pool.connect();
+    try {
+      await db.query('BEGIN');
+      // The unique key and row lock serialize duplicate readers and concurrent reviews.
+      if (parsed.status !== 'ignored') {
+        await db.query(`INSERT INTO telegram_import_messages
+          (city_id, channel_id, channel_username, message_id, published_at, edited_at, content_hash, raw_text, has_media, status)
+          VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,'review') ON CONFLICT DO NOTHING`,
+        [cityId, message.channelId, message.channelUsername, message.messageId, message.publishedAt, message.editedAt, hash, message.text.slice(0, 20_000), message.hasMedia]);
+      }
+      const row = (await db.query(`SELECT * FROM telegram_import_messages
+        WHERE city_id = $1 AND channel_id = $2 AND message_id = $3 FOR UPDATE`, [cityId, message.channelId, message.messageId])).rows[0];
+      if (!row) { await db.query('COMMIT'); return { status: 'ignored' }; }
+      const initialized = row.parsed && Object.keys(row.parsed).length > 0;
+      if ((initialized && row.content_hash === hash) || (row.edited_at && (!message.editedAt || Date.parse(message.editedAt) < new Date(row.edited_at).getTime()))) {
+        await db.query('COMMIT'); return { status: 'unchanged' };
+      }
+      if (initialized) await this.deactivate(db, row.id);
+      let status = row.status === 'dismissed' ? 'dismissed' : 'review';
+      let reasons = parsed.reasons;
+      let plans = [];
+      if (parsed.status === 'ignored') reasons = ['Текст исходной публикации изменился: отмена больше не распознаётся.'];
+      if (parsed.status === 'ready') {
+        const matched = matchCancellationRoutes(parsed, await this.getRoutes(db, cityId));
+        plans = matched.plans;
+        reasons = matched.reasons;
+        if (Date.parse(dayBounds(parsed.serviceDate).end) <= this.now().getTime()) status = status === 'dismissed' ? status : 'expired';
+        else if (!reasons.length && this.config.autoPublish && status !== 'dismissed') status = 'published';
+        else if (!reasons.length) reasons = ['Автоматическая публикация выключена. Подтвердите уведомление.'];
+      }
+      if (status === 'published') {
+        for (const plan of plans) await this.saveAlert(db, row.id, buildCancellationAlert(plan, parsed.serviceDate, sourceUrl(row), message.publishedAt, cityId));
+      }
+      await db.query(`UPDATE telegram_import_messages SET content_hash=$2, raw_text=$3, edited_at=$4,
+        has_media=$5, status=$6, reasons=$7::jsonb, parsed=$8::jsonb, reviewed_by=NULL, updated_at=now() WHERE id=$1`,
+      [row.id, hash, message.text.slice(0, 20_000), message.editedAt, message.hasMedia, status, JSON.stringify(reasons), JSON.stringify({ ...parsed, plans })]);
+      await db.query('COMMIT');
+      return { status, id: row.id };
+    } catch (error) { await db.query('ROLLBACK'); throw error; }
+    finally { db.release(); }
+  }
+
+  async deactivate(db, messageId) {
+    await db.query(`UPDATE alerts SET is_active=false, notification_pending=false
+      WHERE id IN (SELECT alert_id FROM telegram_import_alerts WHERE message_id=$1)`, [messageId]);
+  }
+
+  async saveAlert(db, messageId, payload) {
+    const existing = (await db.query('SELECT alert_id FROM telegram_import_alerts WHERE message_id=$1 AND route_id=$2', [messageId, payload.route_id])).rows[0];
+    if (existing) {
+      // A deleted alert stays deleted, and editing a publication doesn't resend a delivered push.
+      if (existing.alert_id) await db.query(`UPDATE alerts SET title=$2, message=$3, start_time=$4, end_time=$5,
+        is_active=true, notification_pending=true,
+        notification_attempts=CASE WHEN notification_sent_at IS NULL THEN 0 ELSE notification_attempts END
+        WHERE id=$1`, [existing.alert_id, payload.title, payload.message, payload.start_time, payload.end_time]);
+      return;
+    }
+    const result = await this.alerts.insertAlert(db, payload);
+    await db.query('UPDATE alerts SET notification_pending=true WHERE id=$1', [result.alert.id]);
+    await db.query('INSERT INTO telegram_import_alerts(message_id,route_id,alert_id) VALUES ($1,$2,$3)', [messageId, payload.route_id, result.alert.id]);
+  }
+
+  async list(cityId) {
+    return (await this.pool.query(`SELECT m.*, COALESCE((SELECT json_agg(t.alert_id) FROM telegram_import_alerts t
+      WHERE t.message_id=m.id AND t.alert_id IS NOT NULL), '[]') AS alert_ids
+      FROM telegram_import_messages m WHERE m.city_id=$1 ORDER BY (m.status='review') DESC,m.published_at DESC,m.id DESC LIMIT 100`, [cityId])).rows.map(row => ({
+        ...row, source_url: sourceUrl(row),
+        status: row.status === 'published' && validDate(row.parsed.serviceDate) && Date.parse(dayBounds(row.parsed.serviceDate).end) <= this.now().getTime() ? 'expired' : row.status
+      }));
+  }
+
+  async review(id, cityId, input, userId) {
+    const db = await this.pool.connect();
+    try {
+      await db.query('BEGIN');
+      const row = (await db.query('SELECT * FROM telegram_import_messages WHERE id=$1 AND city_id=$2 FOR UPDATE', [id, cityId])).rows[0];
+      if (!row) throw httpError(404, 'Сообщение не найдено.');
+      if (input.content_hash !== row.content_hash) throw httpError(409, 'Публикация изменилась. Обновите список и проверьте новый текст.');
+      if (!['publish', 'dismiss'].includes(input.action)) throw httpError(400, 'Неизвестное действие.');
+      if (input.action === 'publish') {
+        if (!validDate(input.service_date)) throw httpError(400, 'Укажите корректную дату отмены.');
+        if (Date.parse(dayBounds(input.service_date).end) <= this.now().getTime()) throw httpError(410, 'День отмены уже закончился.');
+        const route = (await this.getRoutes(db, cityId)).find(r => r.id === Number(input.route_id));
+        if (!route) throw httpError(400, 'Выберите действующий маршрут выбранного города.');
+        const title = String(input.title || '').trim();
+        const message = String(input.message || '').trim();
+        if (!title || title.length > 200 || !message || message.length > 5000) throw httpError(400, 'Укажите заголовок (до 200 символов) и текст (до 5000 символов).');
+        await this.deactivate(db, row.id);
+        const bounds = dayBounds(input.service_date);
+        await this.saveAlert(db, row.id, { city_id: cityId, route_id: route.id, alert_type: 'cancellation', title,
+          message: `${message}\nИсточник: ${sourceUrl(row)}`, start_time: new Date(Math.max(Date.parse(bounds.start), this.now().getTime())).toISOString(), end_time: bounds.end, is_active: true });
+      } else await this.deactivate(db, row.id);
+      const parsed = input.action === 'publish' ? { ...row.parsed, serviceDate: input.service_date, manual: true } : row.parsed;
+      await db.query(`UPDATE telegram_import_messages SET status=$2, reviewed_by=$3, parsed=$4::jsonb, updated_at=now() WHERE id=$1`, [row.id, input.action === 'publish' ? 'published' : 'dismissed', userId, JSON.stringify(parsed)]);
+      await db.query('COMMIT');
+      return { id: row.id, status: input.action === 'publish' ? 'published' : 'dismissed' };
+    } catch (error) { await db.query('ROLLBACK'); throw error; }
+    finally { db.release(); }
+  }
+
+  async stop() { this.stopped = true; await this.source.close(); }
+}
+
+let singleton;
+function getTelegramImportService() { return singleton || (singleton = new TelegramImportService()); }
+function startTelegramImporter(service = getTelegramImportService()) {
+  if (!service.config.enabled) return () => service.stop();
+  const run = () => service.sync().catch(() => {}); // sync records a sanitized error for the admin status.
+  run();
+  const timer = setInterval(run, service.config.intervalMs);
+  timer.unref?.();
+  return () => { clearInterval(timer); return service.stop(); };
+}
+
+module.exports = { TelegramImportService, getTelegramImportService, startTelegramImporter, contentHash, publicError };

backend/src/services/telegramSource.js

@@ -0,0 +1,57 @@
+const fs = require('node:fs/promises');
+const { assertExternalSessionFile } = require('../config/telegram');
+
+class TelegramSource {
+  constructor(config) { this.config = config; this.client = null; this.channel = null; }
+
+  async connect() {
+    if (this.client?.connected && this.channel) return;
+    if (this.client) await this.close();
+    const file = await assertExternalSessionFile(this.config.sessionFile);
+    const session = (await fs.readFile(file, 'utf8')).trim();
+    if (!session) throw new Error('SESSION_MISSING');
+    const { TelegramClient } = require('teleproto');
+    const { StringSession } = require('teleproto/sessions');
+    const { Logger } = require('teleproto/extensions/Logger');
+    const client = new TelegramClient(new StringSession(session), this.config.apiId, this.config.apiHash, {
+      connectionRetries: 2, requestRetries: 2, floodSleepThreshold: 0, baseLogger: new Logger('none')
+    });
+    this.client = client;
+    await client.connect();
+    if (!await client.checkAuthorization()) throw new Error('AUTH_REQUIRED');
+    if ((await client.getMe()).bot) throw new Error('USER_ACCOUNT_REQUIRED');
+    const channel = await client.getEntity(this.config.channel);
+    if (channel.className !== 'Channel' || !channel.broadcast || channel.username?.toLowerCase() !== this.config.channel.toLowerCase()) {
+      throw new Error('CHANNEL_INVALID');
+    }
+    this.channel = channel;
+  }
+
+  async fetchMessages(since) {
+    await this.connect();
+    const messages = [];
+    // Pagination revisits recent messages so edits are detected after restarts.
+    // The bound prevents an accidental full-history import of a busy channel.
+    for await (const message of this.client.iterMessages(this.channel, { limit: 1000, waitTime: 1 })) {
+      if (message.date && message.date * 1000 < since.getTime()) return messages.reverse();
+      if (message.className !== 'Message' || !message.date) continue;
+      messages.push({
+        channelId: this.channel.id.toString(), channelUsername: this.channel.username,
+        messageId: message.id, text: message.message || '', hasMedia: Boolean(message.media && message.media.className !== 'MessageMediaWebPage'),
+        publishedAt: new Date(message.date * 1000).toISOString(),
+        editedAt: message.editDate ? new Date(message.editDate * 1000).toISOString() : null
+      });
+    }
+    if (messages.length >= 1000) throw new Error('HISTORY_LIMIT');
+    return messages.reverse();
+  }
+
+  async close() {
+    const client = this.client;
+    this.client = null;
+    this.channel = null;
+    if (client) await client.destroy();
+  }
+}
+
+module.exports = { TelegramSource };