Zakhar Timoshenko
360b684019
- Add boardDeviceService for managing device pairing, status, claiming, and revocation. - Introduce boardDeviceSettings for default settings and validation logic. - Create integration tests for device lifecycle, concurrent claims, and updates. - Add unit tests for device pairing, claiming, and settings validation. - Implement API tests for device endpoints with JWT authentication. - Develop in-memory repository for testing purposes. - Create frontend components for managing devices, including a form for device settings. - Style DevicesPage with CSS for better user experience.
261 lines
17 KiB
JavaScript
261 lines
17 KiB
JavaScript
const crypto = require('node:crypto');
|
||
const { getTelegramConfig, missingTelegramConfig } = require('../config/telegram');
|
||
const { getLocalDateTime } = require('../utils/etaTime');
|
||
const { TelegramSource } = require('./telegramSource');
|
||
const { parseTransitNotice, resolveCancellationStops, matchCancellationRoutes, buildTransitNoticeDraft, buildTransitNoticeAlert, dayBounds, validDate, addDays } = require('./telegramCancellationParser');
|
||
const { AlertCreationService } = require('./alertCreationService');
|
||
const { TelegramStopAliasService } = require('./telegramStopAliasService');
|
||
const { resolveAlertTargets } = require('./alertTargets');
|
||
|
||
const PARSER_VERSION = 'transit-notices-v2';
|
||
|
||
function noticeOptions(parsed) { return { alertType: parsed.alertType || 'cancellation', delayPossible: parsed.delayPossible }; }
|
||
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 messageDrafts(parsed) {
|
||
if (!validDate(parsed.serviceDate)) return [];
|
||
const groups = new Map();
|
||
for (const group of parsed.groups || []) {
|
||
if (!groups.has(group.routeNumber)) groups.set(group.routeNumber, { routeNumber: group.routeNumber, departures: [] });
|
||
groups.get(group.routeNumber).departures.push(...group.departures);
|
||
}
|
||
return [...groups.values()].map(group => ({ routeId: null, routeNumber: group.routeNumber, ...buildTransitNoticeDraft(group, parsed.serviceDate, noticeOptions(parsed)) }));
|
||
}
|
||
|
||
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 });
|
||
this.stopAliases = new TelegramStopAliasService(this.pool);
|
||
}
|
||
|
||
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 id,route_number FROM routes WHERE city_id=$1 AND is_active=true ORDER BY id', [cityId])).rows;
|
||
}
|
||
|
||
async getStopContext(db, cityId) {
|
||
const stops = (await db.query('SELECT id,name FROM stops WHERE city_id=$1 AND is_active=true ORDER BY id', [cityId])).rows;
|
||
const aliases = await this.stopAliases.list(cityId, db);
|
||
return { stops, aliases };
|
||
}
|
||
|
||
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');
|
||
let parsed = { ...parseTransitNotice(message, { preserveSourceStops: true }), parserVersion: PARSER_VERSION };
|
||
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;
|
||
const sameContent = initialized && row.content_hash === hash;
|
||
if ((sameContent && (row.status !== 'review' || (parsed.status !== 'ready' && row.parsed.parserVersion === PARSER_VERSION))) || (row.edited_at && (!message.editedAt || Date.parse(message.editedAt) < new Date(row.edited_at).getTime()))) {
|
||
await db.query('COMMIT'); return { status: 'unchanged' };
|
||
}
|
||
let routes = [];
|
||
if (parsed.status === 'ready') {
|
||
const context = await this.getStopContext(db, cityId);
|
||
routes = await this.getRoutes(db, cityId);
|
||
const resolutionHash = crypto.createHash('sha256').update(JSON.stringify([
|
||
PARSER_VERSION, context.stops, context.aliases.map(a => [a.normalized_alias, a.stop_id]), routes, this.config.autoPublish
|
||
])).digest('hex');
|
||
if (sameContent && row.parsed.resolutionHash === resolutionHash && Date.parse(dayBounds(parsed.serviceDate).end) > this.now().getTime()) {
|
||
await db.query('COMMIT'); return { status: 'unchanged' };
|
||
}
|
||
parsed = { ...resolveCancellationStops(parsed, context.stops, context.aliases), resolutionHash };
|
||
}
|
||
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, routes);
|
||
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') {
|
||
const draft = buildTransitNoticeDraft(plans, parsed.serviceDate, noticeOptions(parsed));
|
||
if (draft.title.length > 200 || draft.message.length > 5000 || plans.length > 100) {
|
||
status = 'review';
|
||
reasons = ['Слишком длинное уведомление. Сократите заголовок или текст перед публикацией.'];
|
||
}
|
||
}
|
||
if (status === 'published') {
|
||
await this.saveAlert(db, row.id, buildTransitNoticeAlert(plans, parsed.serviceDate, sourceUrl(row), message.publishedAt, cityId, noticeOptions(parsed)));
|
||
}
|
||
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
|
||
UNION SELECT alert_id FROM telegram_message_alerts WHERE message_id=$1)`, [messageId]);
|
||
}
|
||
|
||
async saveAlert(db, messageId, payload) {
|
||
let existing = (await db.query('SELECT alert_id FROM telegram_message_alerts WHERE message_id=$1', [messageId])).rows[0];
|
||
if (!existing) {
|
||
const legacy = (await db.query('SELECT alert_id FROM telegram_import_alerts WHERE message_id=$1 ORDER BY route_id', [messageId])).rows;
|
||
if (legacy.length) {
|
||
// Consolidate only on an edit/review; keep tombstones and delivered pushes.
|
||
existing = { alert_id: legacy.some(link => link.alert_id === null) ? null : legacy[0].alert_id };
|
||
if (existing.alert_id && legacy.length > 1) await db.query(`UPDATE alerts SET
|
||
notification_sent_at=(SELECT max(notification_sent_at) FROM alerts WHERE id=ANY($2::int[])) WHERE id=$1`,
|
||
[existing.alert_id, legacy.map(link => link.alert_id)]);
|
||
await db.query('INSERT INTO telegram_message_alerts(message_id,alert_id) VALUES ($1,$2)', [messageId, existing.alert_id]);
|
||
}
|
||
}
|
||
if (existing) {
|
||
// Deletion stays deleted. Editing a delivered alert doesn't resend its push.
|
||
if (existing.alert_id) await db.query(`UPDATE alerts SET title=$2, message=$3, start_time=$4, end_time=$5,
|
||
route_id=$6, route_numbers=$7::text[], alert_type=$8, 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, payload.route_id, payload.route_numbers, payload.alert_type]);
|
||
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_message_alerts(message_id,alert_id) VALUES ($1,$2)', [messageId, result.alert.id]);
|
||
}
|
||
|
||
async list(cityId) {
|
||
return (await this.pool.query(`SELECT m.*, CASE WHEN EXISTS(SELECT 1 FROM telegram_message_alerts t WHERE t.message_id=m.id)
|
||
THEN COALESCE((SELECT json_agg(t.alert_id) FROM telegram_message_alerts t WHERE t.message_id=m.id AND t.alert_id IS NOT NULL), '[]')
|
||
ELSE 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), '[]') END 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),
|
||
drafts: messageDrafts(row.parsed),
|
||
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, 'Неизвестное действие.');
|
||
let alertType = row.parsed.alertType || 'cancellation';
|
||
if (input.action === 'publish') {
|
||
alertType = input.alert_type === undefined ? alertType : input.alert_type;
|
||
if (!['cancellation', 'delay'].includes(alertType)) throw httpError(400, 'Выберите отмену или опоздание рейсов.');
|
||
if (!validDate(input.service_date)) throw httpError(400, 'Укажите корректную дату уведомления.');
|
||
if (Date.parse(dayBounds(input.service_date).end) <= this.now().getTime()) throw httpError(410, 'День уведомления уже закончился.');
|
||
const targets = await resolveAlertTargets(db, cityId, input, { required: true });
|
||
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, ...targets, alert_type: alertType, title,
|
||
message: `${message}`, 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, alertType, manual: true, selectedRouteNumbers: input.route_numbers } : 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 };
|