kvarta_project/apps/api/src/backup.ts

// Бэкапы базы и восстановление из копии. Копии лежат рядом с базой (/data/backups),
// поэтому временные файлы на той же файловой системе и rename при подмене атомарен.
import {
  createReadStream,
  createWriteStream,
  existsSync,
  mkdirSync,
  readdirSync,
  readFileSync,
  renameSync,
  rmSync,
  statSync,
} from 'node:fs';
import { writeFile } from 'node:fs/promises';
import { dirname, join } from 'node:path';
import { pipeline } from 'node:stream/promises';
import { createGunzip, createGzip } from 'node:zlib';
import Database from 'better-sqlite3';
import { openDb, type Db, type DbHolder } from './db/index.js';
import { MIGRATIONS } from './db/migrations.js';
import type { TelegramClient } from './telegram/client.js';
import { errorMessage, shortError, type SystemReporter } from './telegram/system.js';

export const TIME_ZONE = 'Europe/Minsk';
export const KEEP_DAILY = 14;
export const KEEP_PRE_RESTORE_DAYS = 30;
const DAY_MS = 24 * 60 * 60 * 1000;
const DAILY_HOUR = 3;

export type BackupKind = 'daily' | 'pre-restore';
export type BackupFile = { file: string; kind: BackupKind; at: Date; bytes: number };

const NAME_RE = /^(kvarta|pre-restore)-(\d{4})-(\d{2})-(\d{2})T(\d{2})(\d{2})\.sqlite\.gz$/;

export function backupDirFor(dbPath: string): string {
  return join(dirname(dbPath), 'backups');
}

type LocalParts = { year: number; month: number; day: number; hour: number; minute: number };

const MINSK_PARTS = new Intl.DateTimeFormat('en-CA', {
  timeZone: TIME_ZONE,
  year: 'numeric',
  month: '2-digit',
  day: '2-digit',
  hour: '2-digit',
  minute: '2-digit',
  hourCycle: 'h23',
});

/** Часы и дата по Минску. Через Intl, а не +3: правило могут поменять, tzdata — обновится. */
export function localParts(date: Date): LocalParts {
  const parts = MINSK_PARTS.formatToParts(date);
  const get = (type: string) => Number(parts.find((p) => p.type === type)!.value);
  return { year: get('year'), month: get('month'), day: get('day'), hour: get('hour'), minute: get('minute') };
}

const pad = (n: number) => String(n).padStart(2, '0');

/** «18.10 03:00» — для кнопок и отчётов. */
export function shortLabel(date: Date): string {
  const p = localParts(date);
  return `${pad(p.day)}.${pad(p.month)} ${pad(p.hour)}:${pad(p.minute)}`;
}

export function backupFileName(date: Date, kind: BackupKind): string {
  const p = localParts(date);
  const prefix = kind === 'daily' ? 'kvarta' : 'pre-restore';
  return `${prefix}-${p.year}-${pad(p.month)}-${pad(p.day)}T${pad(p.hour)}${pad(p.minute)}.sqlite.gz`;
}

/** Минское локальное время → UTC-момент (смещение берём на этот самый момент). */
function fromLocal(p: LocalParts): Date {
  const guess = Date.UTC(p.year, p.month - 1, p.day, p.hour, p.minute);
  const shown = localParts(new Date(guess));
  const offset = Date.UTC(shown.year, shown.month - 1, shown.day, shown.hour, shown.minute) - guess;
  return new Date(guess - offset);
}

function parseName(file: string): { kind: BackupKind; at: Date } | null {
  const m = NAME_RE.exec(file);
  if (!m) return null;
  const [, prefix, y, mo, d, h, mi] = m;
  return {
    kind: prefix === 'kvarta' ? 'daily' : 'pre-restore',
    at: fromLocal({ year: +y!, month: +mo!, day: +d!, hour: +h!, minute: +mi! }),
  };
}

/** Следующий запуск ежедневного бэкапа: ближайшие 03:00 по Минску строго после now. */
export function nextDailyRun(now: Date): Date {
  const today = localParts(now);
  let run = fromLocal({ ...today, hour: DAILY_HOUR, minute: 0 });
  if (run.getTime() <= now.getTime()) {
    const tomorrow = localParts(new Date(run.getTime() + DAY_MS));
    run = fromLocal({ ...tomorrow, hour: DAILY_HOUR, minute: 0 });
  }
  return run;
}

export function listBackups(dir: string): BackupFile[] {
  if (!existsSync(dir)) return [];
  const out: BackupFile[] = [];
  for (const file of readdirSync(dir)) {
    const parsed = parseName(file);
    if (parsed) out.push({ file, ...parsed, bytes: statSync(join(dir, file)).size });
  }
  return out.sort((a, b) => b.at.getTime() - a.at.getTime() || b.file.localeCompare(a.file));
}

export function hasDailyBackupToday(dir: string, now: Date): boolean {
  const today = localParts(now);
  return listBackups(dir).some((b) => {
    const p = localParts(b.at);
    return b.kind === 'daily' && p.year === today.year && p.month === today.month && p.day === today.day;
  });
}

/** 14 последних ежедневных; pre-restore — 30 дней. Возвращает удалённые имена. */
export function rotateBackups(dir: string, now: Date): string[] {
  const removed: string[] = [];
  const all = listBackups(dir);
  for (const b of all.filter((b) => b.kind === 'daily').slice(KEEP_DAILY)) removed.push(b.file);
  for (const b of all.filter((b) => b.kind === 'pre-restore')) {
    if (now.getTime() - b.at.getTime() > KEEP_PRE_RESTORE_DAYS * DAY_MS) removed.push(b.file);
  }
  for (const file of removed) rmSync(join(dir, file), { force: true });
  return removed;
}

export function countLeads(db: Db): number {
  return (db.prepare('SELECT count(*) AS n FROM leads').get() as { n: number }).n;
}

export type BackupResult = BackupFile & { leads: number };

/** Снимок через db.backup() (консистентен при работающей записи) → gzip → атомарный rename. */
export async function createBackup(db: Db, dir: string, now: Date, kind: BackupKind = 'daily'): Promise<BackupResult> {
  mkdirSync(dir, { recursive: true });
  const file = backupFileName(now, kind);
  const raw = join(dir, `.tmp-${process.pid}-${Date.now()}.sqlite`);
  const gz = join(dir, `.tmp-${file}`);
  try {
    await db.backup(raw);
    // Копия наследует WAL из заголовка: переводим в DELETE, иначе рядом остаются -wal/-shm,
    // а в архив попадает база, которой нужны файлы-спутники.
    const snapshot = new Database(raw);
    snapshot.pragma('journal_mode = DELETE');
    const leads = countLeads(snapshot);
    snapshot.close();
    await pipeline(createReadStream(raw), createGzip(), createWriteStream(gz));
    renameSync(gz, join(dir, file));
    return { file, kind, at: now, bytes: statSync(join(dir, file)).size, leads };
  } finally {
    rmSync(raw, { force: true });
    removeSidecars(raw);
    rmSync(gz, { force: true });
  }
}

export class RestoreRejected extends Error {}

export type InspectedBackup = {
  /** Распакованная копия во временном файле; удаляет вызывающий (discard) */
  path: string;
  userVersion: number;
  leads: { id: number; created_at: string }[];
};

/** gunzip → integrity_check → версия схемы. Кидает RestoreRejected с понятным текстом. */
export async function inspectBackup(source: string | Uint8Array, dir: string): Promise<InspectedBackup> {
  mkdirSync(dir, { recursive: true });
  const stamp = `${process.pid}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
  const path = join(dir, `.restore-${stamp}.sqlite`);
  const gzPath = typeof source === 'string' ? source : join(dir, `.upload-${stamp}.sqlite.gz`);
  try {
    if (typeof source !== 'string') await writeFile(gzPath, source);
    try {
      await pipeline(createReadStream(gzPath), createGunzip(), createWriteStream(path));
    } catch {
      throw new RestoreRejected('Файл не распаковывается: это не .sqlite.gz или он повреждён.');
    }
    // Открываем на запись: это наша временная копия, а readonly на WAL-базе оставил бы -wal/-shm.
    let copy: Database.Database;
    let integrity: unknown;
    try {
      copy = new Database(path, { fileMustExist: true });
    } catch {
      throw new RestoreRejected('Внутри архива не база SQLite.');
    }
    try {
      try {
        copy.pragma('journal_mode = DELETE');
        integrity = copy.pragma('integrity_check', { simple: true });
      } catch {
        throw new RestoreRejected('Внутри архива не база SQLite.');
      }
      if (integrity !== 'ok') throw new RestoreRejected('Копия повреждена: integrity_check не прошёл.');
      const userVersion = copy.pragma('user_version', { simple: true }) as number;
      if (userVersion > MIGRATIONS.length) {
        throw new RestoreRejected(`Копия от более новой версии api (схема v${userVersion}, у нас v${MIGRATIONS.length}).`);
      }
      const hasLeads = copy.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'leads'").get();
      if (!hasLeads) throw new RestoreRejected('В копии нет таблицы заявок.');
      return { path, userVersion, leads: leadStamps(copy) };
    } finally {
      copy.close();
    }
  } catch (err) {
    rmSync(path, { force: true });
    removeSidecars(path);
    throw err;
  } finally {
    if (typeof source !== 'string') rmSync(gzPath, { force: true });
  }
}

export function discardInspected(inspected: InspectedBackup): void {
  rmSync(inspected.path, { force: true });
}

/** Заявки текущей базы, которых нет в копии (по паре id + created_at: id после отката переиспользуются). */
export function leadsMissingFrom(db: Db, backupLeads: InspectedBackup['leads']): number[] {
  const inCopy = new Set(backupLeads.map((l) => `${l.id}|${l.created_at}`));
  return leadStamps(db).filter((l) => !inCopy.has(`${l.id}|${l.created_at}`)).map((l) => l.id);
}

function leadStamps(db: Db): InspectedBackup['leads'] {
  return db.prepare('SELECT id, created_at FROM leads ORDER BY id').all() as InspectedBackup['leads'];
}

function removeSidecars(path: string) {
  rmSync(`${path}-wal`, { force: true });
  rmSync(`${path}-shm`, { force: true });
}

export type RestoreResult = { preRestore: BackupResult; leads: number };

/**
 * Подменить базу копией. Приём заявок на паузе с самого начала: заявка, пришедшая
 * между снимком pre-restore и подменой, иначе пропала бы без следа.
 */
export async function restoreFromInspected({
  holder,
  dbPath,
  inspected,
  now,
  waitIdle,
}: {
  holder: DbHolder;
  dbPath: string;
  inspected: InspectedBackup;
  now: Date;
  /** Дождаться фоновых записей в базу (очередь Telegram), прежде чем её закрыть. */
  waitIdle?: () => Promise<void>;
}): Promise<RestoreResult> {
  const dir = backupDirFor(dbPath);
  const rollback = `${dbPath}.rollback`;
  holder.paused = true;
  try {
    const preRestore = await createBackup(holder.db, dir, now, 'pre-restore');
    await waitIdle?.();

    // От close до open — без await: никакой другой код не увидит закрытую базу.
    holder.db.close();
    removeSidecars(dbPath);
    renameSync(dbPath, rollback);
    try {
      renameSync(inspected.path, dbPath);
      holder.db = openDb(dbPath);
    } catch (err) {
      rmSync(dbPath, { force: true });
      removeSidecars(dbPath);
      renameSync(rollback, dbPath);
      holder.db = openDb(dbPath);
      throw err;
    }
    rmSync(rollback, { force: true });
    return { preRestore, leads: countLeads(holder.db) };
  } finally {
    holder.paused = false;
    discardInspected(inspected);
  }
}

export type Scheduler = { start(): void; stop(): void };

/** Ежедневно в 03:00 по Минску и сразу при старте, если сегодняшней копии ещё нет. */
export function createBackupScheduler({
  run,
  now,
  hasToday,
}: {
  run: () => Promise<void>;
  now: () => Date;
  hasToday: () => boolean;
}): Scheduler {
  let timer: NodeJS.Timeout | undefined;
  let stopped = false;

  const safeRun = () => run().catch((err: unknown) => console.error('backup job failed', err));

  function schedule() {
    if (stopped) return;
    const delay = Math.max(1000, nextDailyRun(now()).getTime() - now().getTime());
    timer = setTimeout(() => void safeRun().then(schedule), delay);
    timer.unref();
  }

  return {
    start() {
      stopped = false;
      if (!hasToday()) void safeRun();
      schedule();
    },
    stop() {
      stopped = true;
      if (timer) clearTimeout(timer);
    },
  };
}

export function pluralLeads(n: number): string {
  const mod10 = n % 10;
  const mod100 = n % 100;
  const word =
    mod10 === 1 && mod100 !== 11
      ? 'заявка'
      : mod10 >= 2 && mod10 <= 4 && (mod100 < 12 || mod100 > 14)
        ? 'заявки'
        : 'заявок';
  return `${n} ${word}`;
}

const kb = (bytes: number) => `${Math.max(1, Math.round(bytes / 1024))} КБ`;

/** Копия → ротация → файл администратору → строка в «Систему». Не бросает: только отчёт. */
export async function runBackupAndReport({
  holder,
  dir,
  now,
  client,
  adminId,
  system,
}: {
  holder: DbHolder;
  dir: string;
  now: Date;
  client?: Pick<TelegramClient, 'sendDocument'>;
  adminId?: number;
  system: SystemReporter;
}): Promise<BackupResult | undefined> {
  let result: BackupResult;
  try {
    result = await createBackup(holder.db, dir, now);
    rotateBackups(dir, now);
  } catch (err) {
    await system.report('backup failed', `Бэкап ${shortLabel(now)} не удался ❌\n${shortError(err)}`);
    return undefined;
  }
  const summary = `Бэкап ${shortLabel(result.at)}: ${pluralLeads(result.leads)}, ${kb(result.bytes)}`;
  let delivery = '';
  if (client && adminId !== undefined) {
    try {
      await client.sendDocument({
        chat_id: adminId,
        file: { name: result.file, data: readFileSync(join(dir, result.file)) },
        caption: summary,
      });
    } catch (err) {
      delivery = `, но в личку не отправлен: ${errorMessage(err)}`;
    }
  }
  await system.report('backup', `${summary} ✅${delivery}`);
  return result;
}