ridebus-admin/backend/src/services/scheduleImportService.js

const { randomUUID } = require('node:crypto');
const fs = require('node:fs/promises');
const path = require('node:path');
const { getLocalDateTime } = require('../utils/etaTime');
const { HOME_URL, CATALOG_URL, hash, normalizeName, normalizeNumber, sourceError,
  safeUrl, fetchHtml, parseCatalog, parseSchedule, parseNews } = require('./scheduleSource');

const DAY_LABELS = { weekday: 'Будние дни', saturday: 'Суббота', sunday: 'Воскресенье' };
const tripKey = trip => JSON.stringify({
  departureTime: trip.departureTime?.length === 5 ? `${trip.departureTime}:00` : trip.departureTime,
  startStopId: trip.startStopId || null, endStopId: trip.endStopId || null,
  viaStopIds: trip.viaStopIds || [], daysOfWeek: trip.daysOfWeek || [],
  exceptDaysOfWeek: trip.exceptDaysOfWeek || [], note: trip.note || ''
});
const blankTrip = (time, startStopId = null, endStopId = null) => ({
  departureTime: time, startStopId, endStopId, viaStopIds: [], daysOfWeek: [], exceptDaysOfWeek: [], note: ''
});

function compareTrips(before, after) {
  // Multisets preserve duplicate departures in legacy data in the removal preview.
  const remaining = before.map(trip => ({ trip, key: tripKey(trip) }));
  const added = [];
  let unchanged = 0;
  for (const trip of after) {
    const index = remaining.findIndex(item => item.key === tripKey(trip));
    if (index === -1) added.push(trip);
    else { remaining.splice(index, 1); unchanged += 1; }
  }
  return { added, removed: remaining.map(item => item.trip), unchanged };
}

function buildPlan(source, mappings, snapshot, validFrom, today = getLocalDateTime().date) {
  if (!/^\d{4}-\d{2}-\d{2}$/.test(validFrom || '')
    || Number.isNaN(Date.parse(validFrom)) || new Date(validFrom).toISOString().slice(0, 10) !== validFrom
    || validFrom < today) throw sourceError('Выберите дату начала не раньше сегодняшнего дня.');
  if (source.warnings.length) throw sourceError('В источнике есть нераспознанные условия. Автоматический импорт заблокирован.');
  if (normalizeNumber(source.number) !== normalizeNumber(snapshot.route.route_number)) {
    throw sourceError('Номер маршрута в базе не совпадает с номером на сайте.');
  }
  const stops = new Set(snapshot.stops.map(stop => Number(stop.stop_id)));
  const groups = {};
  const notices = new Set();
  for (const row of source.rows) {
    if (!mappings || !Object.hasOwn(mappings, row.stopName)) throw sourceError(`Укажите назначение строки «${row.stopName}».`);
    const mapping = mappings[row.stopName];
    for (const day of row.days) groups[day] ||= [];
    if (mapping?.mode === 'checkpoint') continue;
    if (mapping?.mode !== 'departure') throw sourceError(`Укажите назначение строки «${row.stopName}».`);
    const start = Number(mapping.startStopId);
    const end = Number(mapping.endStopId);
    if (!stops.has(start) || !stops.has(end) || start === end) {
      throw sourceError(`Для «${row.stopName}» выберите разные начальную и конечную остановки этого маршрута.`);
    }
    const departures = row.departures || row.times.map(departureTime => ({ departureTime }));
    for (const departure of departures) {
      const trip = { ...blankTrip(departure.departureTime, start, end),
        daysOfWeek: departure.daysOfWeek || [], exceptDaysOfWeek: departure.exceptDaysOfWeek || [] };
      const unresolved = [];
      for (const name of departure.viaStopNames || []) {
        const matches = snapshot.stops.filter(stop => normalizeName(stop.name) === normalizeName(name));
        if (matches.length === 1) trip.viaStopIds.push(Number(matches[0].stop_id));
        else {
          unresolved.push(name);
          notices.add(`«${name}»: ${matches.length ? 'найдено несколько остановок с этим именем' : 'нет в остановках направления'}. Условие «через» сохранится текстовым примечанием без привязки к остановке.`);
        }
      }
      trip.note = unresolved.map(name => `через «${name}»`).join('; ');
      let included = false;
      for (const day of row.days) {
        const activeDays = { weekday: [1, 2, 3, 4, 5], saturday: [6], sunday: [7] }[day];
        if (!activeDays.some(value => (!trip.daysOfWeek.length || trip.daysOfWeek.includes(value)) && !trip.exceptDaysOfWeek.includes(value))) continue;
        groups[day].push(trip);
        included = true;
      }
      if (!included) throw sourceError(`${departure.departureTime.slice(0, 5)}: условие по дням противоречит разделу «${row.days.map(day => DAY_LABELS[day]).join(', ')}». Проверьте источник.`);
    }
  }
  const changes = [];
  for (const [dayType, trips] of Object.entries(groups)) {
    if (!trips.length) throw sourceError(`${DAY_LABELS[dayType]}: выберите хотя бы одну строку отправлений.`);
    if (trips.length > 2000 || new Set(trips.map(tripKey)).size !== trips.length) {
      throw sourceError('После сопоставления появились повторяющиеся рейсы или слишком много отправлений.');
    }
    trips.sort((a, b) => a.departureTime.localeCompare(b.departureTime) || a.startStopId - b.startStopId);
    const candidates = snapshot.schedules.filter(s => s.day_type === dayType && (!s.valid_until || s.valid_until >= validFrom));
    if (candidates.length > 1 || candidates.some(s => s.valid_from && s.valid_from > validFrom)) {
      throw sourceError(`${DAY_LABELS[dayType]}: есть пересекающиеся или будущие расписания. Сначала проверьте их в разделе «Расписания».`);
    }
    const current = candidates[0];
    const before = current?.trips?.length ? current.trips : (current?.departure_times || []).map(time => blankTrip(time));
    changes.push({ dayType, label: DAY_LABELS[dayType], existingId: current?.id || null,
      previousValidFrom: current?.valid_from || null, validFrom, validUntil: current?.valid_until || null,
      before, after: trips, ...compareTrips(before, trips) });
  }
  return { changes, mappings, validFrom, notices: [...notices], before: snapshot, beforeHash: hash(snapshot),
    hasChanges: changes.some(change => change.added.length || change.removed.length) };
}

function departureStops(source) {
  const names = [...new Set(source.rows.map(row => row.stopName))];
  const origins = names.filter(name => source.directions.some(direction => normalizeName(direction.stops[0]) === normalizeName(name)));
  return origins.length ? origins : names;
}

function buildDirectionPlan(source, sourceStopName, snapshot, validFrom, today) {
  if (!departureStops(source).includes(sourceStopName)) throw sourceError('Выберите строку отправлений для каждого направления.');
  const first = snapshot.stops[0];
  const last = snapshot.stops.at(-1);
  if (!first || !last || first.stop_id === last.stop_id) {
    throw sourceError(`В маршруте «${snapshot.route.name}» должны быть заданы разные первая и последняя остановки.`);
  }
  const rows = source.rows.filter(row => row.stopName === sourceStopName);
  const mappings = { [sourceStopName]: { mode: 'departure', startStopId: first.stop_id, endStopId: last.stop_id } };
  return { ...buildPlan({ ...source, rows }, mappings, snapshot, validFrom, today),
    routeId: snapshot.route.id, route: snapshot.route, stops: snapshot.stops, sourceStopName };
}

class ScheduleImportService {
  constructor({ pool, load = fetchHtml, invalidate = async () => {} }) {
    this.pool = pool;
    this.load = load;
    this.invalidate = invalidate;
    this.schemaPromise = null;
    this.catalogCache = null;
  }

  ensureSchema() {
    this.schemaPromise ||= fs.readFile(path.join(__dirname, '../../migrations/008_schedule_imports.sql'), 'utf8')
      .then(sql => this.pool.query(sql)).catch(error => { this.schemaPromise = null; throw error; });
    return this.schemaPromise;
  }

  async catalog() {
    if (this.catalogCache && this.catalogCache.expires > Date.now()) return this.catalogCache.value;
    const page = await this.load(CATALOG_URL);
    const value = parseCatalog(page.html);
    this.catalogCache = { value, expires: Date.now() + 5 * 60000 };
    return value;
  }

  async snapshot(routeId, db = this.pool) {
    const route = (await db.query('SELECT id, route_number, name FROM routes WHERE id = $1', [routeId])).rows[0];
    if (!route) throw sourceError('Маршрут не найден.', 404);
    const stops = (await db.query(`SELECT rs.stop_id, rs.sequence, s.name FROM route_stops rs
      JOIN stops s ON s.id = rs.stop_id WHERE rs.route_id = $1 ORDER BY rs.sequence, rs.id`, [routeId])).rows;
    const schedules = (await db.query(`SELECT id, day_type, departure_times, trips,
      to_char(valid_from, 'YYYY-MM-DD') AS valid_from, to_char(valid_until, 'YYYY-MM-DD') AS valid_until,
      updated_at::text FROM schedules WHERE route_id = $1 ORDER BY id`, [routeId])).rows;
    return { route, stops, schedules };
  }

  async store({ routeId, source, status, payload = null, userId }) {
    const id = randomUUID();
    await this.pool.query(`INSERT INTO schedule_imports
      (id, route_id, source_url, source_hash, source, status, payload, created_by)
      VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7::jsonb,$8)`,
    [id, routeId, source.url, hash(source), JSON.stringify(source), status, JSON.stringify(payload), userId]);
    return id;
  }

  async getDraft(id, db = this.pool, lock = false) {
    if (!/^[0-9a-f-]{36}$/i.test(id || '')) throw sourceError('Некорректный идентификатор импорта.', 400);
    const draft = (await db.query(`SELECT * FROM schedule_imports WHERE id = $1${lock ? ' FOR UPDATE' : ''}`, [id])).rows[0];
    if (!draft) throw sourceError('Предпросмотр не найден. Загрузите расписание заново.', 404);
    if (draft.status !== 'applied' && Date.now() - new Date(draft.created_at).getTime() > 30 * 60000) {
      throw sourceError('Предпросмотр устарел. Загрузите расписание заново.', 409);
    }
    return draft;
  }

  async inspect({ routeIds, sourceUrl, userId }) {
    await this.ensureSchema();
    if (!Array.isArray(routeIds) || !routeIds.length || routeIds.length > 30
      || routeIds.some(id => !Number.isInteger(Number(id)) || Number(id) <= 0)
      || new Set(routeIds.map(Number)).size !== routeIds.length) throw sourceError('Выберите направления в админке.', 400);
    routeIds = routeIds.map(Number).sort((a, b) => a - b);
    const url = safeUrl(sourceUrl);
    if (!(await this.catalog()).some(route => route.url === url)) throw sourceError('Выберите маршрут из списка сайта.');
    const page = await this.load(url);
    const source = parseSchedule(page.html, page.url);
    const snapshots = await Promise.all(routeIds.map(id => this.snapshot(id)));
    if (snapshots.some(snapshot => normalizeNumber(source.number) !== normalizeNumber(snapshot.route.route_number))) {
      throw sourceError('Все выбранные направления должны иметь тот же номер, что и маршрут на сайте.');
    }
    const mappings = {};
    const origins = departureStops(source);
    for (const snapshot of snapshots) {
      const matches = origins.filter(name => normalizeName(name) === normalizeName(snapshot.stops[0]?.name));
      mappings[snapshot.route.id] = matches.length === 1 ? matches[0] : '';
    }
    await this.pool.query("DELETE FROM schedule_imports WHERE status <> 'applied' AND created_at < now() - interval '7 days'");
    const id = await this.store({ routeId: null, source, status: 'inspection', payload: { version: 2, routeIds }, userId });
    return { id, source, mappings, directions: snapshots.map(({ route, stops }) => ({ route, stops })),
      departureStops: origins, today: getLocalDateTime().date };
  }

  async preview({ inspectionId, mappings, validFrom, userId }) {
    await this.ensureSchema();
    const inspection = await this.getDraft(inspectionId);
    if (inspection.status !== 'inspection') throw sourceError('Загрузите исходную таблицу заново.', 409);
    if (inspection.payload?.version !== 2) throw sourceError('Формат импорта обновлён. Загрузите таблицы заново.', 409);
    const routeIds = inspection.payload.routeIds;
    if (!mappings || Object.keys(mappings).length !== routeIds.length
      || routeIds.some(id => !Object.hasOwn(mappings, id))) throw sourceError('Сопоставьте каждое выбранное направление.');
    const snapshots = await Promise.all(routeIds.map(id => this.snapshot(id)));
    const plans = snapshots.map(snapshot => buildDirectionPlan(inspection.source, mappings[snapshot.route.id], snapshot, validFrom));
    const payload = { version: 2, plans, validFrom, hasChanges: plans.some(plan => plan.hasChanges) };
    const id = await this.store({ routeId: null, source: inspection.source, status: 'ready', payload, userId });
    return { id, ...payload, sourceUrl: inspection.source_url };
  }

  async apply(id) {
    await this.ensureSchema();
    const draft = await this.getDraft(id);
    if (draft.status === 'applied') return { id, applied: true, alreadyApplied: true };
    if (draft.payload?.version !== 2) throw sourceError('Формат импорта обновлён. Загрузите таблицы заново.', 409);
    if (draft.status !== 'ready' || !draft.payload?.hasChanges) throw sourceError('Нет подготовленных изменений.', 409);
    // A changed website invalidates the preview rather than silently importing a different table.
    const page = await this.load(draft.source_url);
    if (hash(parseSchedule(page.html, page.url)) !== draft.source_hash) throw sourceError('Расписание на сайте изменилось. Загрузите новый предпросмотр.', 409);
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
      const locked = await this.getDraft(id, client, true);
      if (locked.status === 'applied') {
        await client.query('COMMIT');
        return { id, applied: true, alreadyApplied: true };
      }
      // Existing CRUD endpoints do not share an advisory lock; table locks also cover inserts.
      await client.query('LOCK TABLE schedules, route_stops IN SHARE ROW EXCLUSIVE MODE');
      for (const plan of locked.payload.plans) {
        const snapshot = await this.snapshot(plan.routeId, client);
        if (hash(snapshot) !== plan.beforeHash) throw sourceError(`Направление «${plan.route.name}» изменилось после предпросмотра. Постройте его заново.`, 409);
      }
      if (locked.payload.validFrom < getLocalDateTime().date) throw sourceError('Дата начала уже прошла. Постройте новый предпросмотр.', 409);
      for (const plan of locked.payload.plans) {
        for (const change of plan.changes) {
          if (!change.added.length && !change.removed.length) continue;
          if (change.existingId && change.previousValidFrom === change.validFrom) {
            await client.query(`UPDATE schedules SET departure_times = $1::time[], trips = $2::jsonb,
              updated_at = CURRENT_TIMESTAMP WHERE id = $3`,
            [change.after.map(trip => trip.departureTime), JSON.stringify(change.after), change.existingId]);
          } else {
            if (change.existingId) await client.query(`UPDATE schedules SET valid_until = $1::date - 1,
              updated_at = CURRENT_TIMESTAMP WHERE id = $2`, [change.validFrom, change.existingId]);
            await client.query(`INSERT INTO schedules (route_id, day_type, departure_times, trips, valid_from, valid_until)
              VALUES ($1,$2,$3::time[],$4::jsonb,$5,$6)`,
            [plan.routeId, change.dayType, change.after.map(trip => trip.departureTime), JSON.stringify(change.after), change.validFrom, change.validUntil]);
          }
        }
      }
      await client.query("UPDATE schedule_imports SET status = 'applied', applied_at = now() WHERE id = $1", [id]);
      await client.query('COMMIT');
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally { client.release(); }
    let cacheWarning = null;
    const cacheResults = await Promise.allSettled(draft.payload.plans.map(plan => this.invalidate(plan.routeId)));
    if (cacheResults.some(result => result.status === 'rejected')) {
      cacheWarning = 'Расписание сохранено, но кеш не удалось обновить. Данные в кеше обновятся после истечения его срока.';
    }
    return { id, applied: true, cacheWarning };
  }

  async history() {
    await this.ensureSchema();
    return (await this.pool.query(`SELECT i.id, i.source_url, i.applied_at, i.payload->>'validFrom' AS valid_from,
      (SELECT jsonb_agg(p->'route') FROM jsonb_array_elements(COALESCE(i.payload->'plans', '[]'::jsonb)) p) AS directions,
      r.route_number, r.name AS route_name FROM schedule_imports i LEFT JOIN routes r ON r.id = i.route_id
      WHERE i.status = 'applied' ORDER BY i.applied_at DESC LIMIT 30`)).rows;
  }

  async news(refresh = false) {
    await this.ensureSchema();
    if (refresh) {
      const page = await this.load(HOME_URL);
      const items = parseNews(page.html);
      for (const item of items) {
        await this.pool.query(`INSERT INTO schedule_import_news (url,title,published_date,content,content_hash)
          VALUES ($1,$2,$3,$4,$5) ON CONFLICT (url) DO UPDATE SET title = EXCLUDED.title,
          published_date = EXCLUDED.published_date, content = EXCLUDED.content,
          content_hash = EXCLUDED.content_hash, checked_at = now()`,
        [item.url, item.title, item.publishedDate, item.content, item.contentHash]);
      }
    }
    return (await this.pool.query(`SELECT *, (reviewed_hash = content_hash) IS TRUE AS reviewed
      FROM schedule_import_news ORDER BY published_date DESC NULLS LAST, url DESC LIMIT 50`)).rows;
  }

  async reviewNews({ url, contentHash }) {
    await this.ensureSchema();
    const result = await this.pool.query(`UPDATE schedule_import_news SET reviewed_hash = content_hash
      WHERE url = $1 AND content_hash = $2 RETURNING url`, [safeUrl(url), contentHash]);
    if (!result.rows.length) throw sourceError('Новость изменилась. Обновите список перед проверкой.', 409);
    return { reviewed: true };
  }
}

module.exports = { ScheduleImportService, buildPlan, compareTrips, buildDirectionPlan, departureStops };