jorgecuadros-platform/apps/api/src/ops/ops.service.ts

import {
  BadRequestException,
  ConflictException,
  Injectable,
  Logger,
  NotFoundException,
  OnModuleInit,
} from "@nestjs/common";
import { spawn } from "node:child_process";
import { createReadStream, promises as fs } from "node:fs";
import * as path from "node:path";
import { OpsJobKind } from "@jorgecuadros/database";
import { PrismaService } from "../prisma/prisma.service";

/**
 * Admin database operations. Everything long-running (mysqldump, mysql restore,
 * the Python migration) runs as a detached child process recorded as one OpsJob
 * row whose `log` is appended as the process talks; the web polls that row.
 *
 * Only ONE mutating job runs at a time (a RUNNING row blocks a new start) — a
 * restore or re-import racing a migration would corrupt the database.
 */

/** The four legacy Access files. Uploads are allowlisted to exactly these
 *  names so an ingest write can never land at an arbitrary path. */
export const INGEST_FILES = [
  "UTILITIES.accdb",
  "SEGUROS 16.mdb",
  "SEGUROS 16_be.mdb",
  "SCOTHIA.mdb",
] as const;
export type IngestName = (typeof INGEST_FILES)[number];

/**
 * A prior-period archive: one Access snapshot per closed year, named for the
 * period it holds. `2025.accdb` is UTILITIES as it stood when 2025 was cut.
 *
 * The filename is the entire declaration of the period — nothing inside the
 * file names its year, because a snapshot's `datos2` is indistinguishable from
 * the live one — so this pattern is both the allowlist and the contract. It is
 * anchored and allows no separator, which is what keeps an upload from
 * escaping the ingest directory.
 */
const PERIOD_FILE_RE = /^(\d{4})\.accdb$/i;

/** Earliest period we will accept, so a typo'd year cannot mint a bogus one. */
const PERIOD_MIN_YEAR = 1990;

export function periodYearOf(name: string): number | null {
  const m = PERIOD_FILE_RE.exec(name);
  if (!m) return null;
  const year = Number(m[1]);
  if (year < PERIOD_MIN_YEAR || year > new Date().getUTCFullYear()) return null;
  return year;
}

/**
 * Prefix for every command containing a pipe. Without it the exit status of
 * `mysqldump | gzip` is gzip's, so a dump that failed immediately still looks
 * like a successful job. Both Alpine's busybox ash (the API image) and macOS
 * `sh` (dev) support it; POSIX does not require it, so `sh -c` is the contract.
 */
const PIPEFAIL = "set -o pipefail; ";

interface MysqlConn {
  host: string;
  port: string;
  user: string;
  password: string;
  database: string;
}

@Injectable()
export class OpsService implements OnModuleInit {
  private readonly logger = new Logger(OpsService.name);

  // Resolve from this source file so it works regardless of process.cwd()
  // (the API runs from apps/api/, but the Python ETL lives at repo-root migration/).
  private readonly migrationDir =
    process.env.MIGRATION_DIR ??
    path.resolve(__dirname, "..", "..", "..", "..", "migration");
  private readonly ingestDir =
    process.env.INGEST_DIR ?? path.join(this.migrationDir, "ingest");
  private readonly backupDir =
    process.env.BACKUP_DIR ?? path.join(this.migrationDir, "backups");
  private readonly migrationEnv = process.env.MIGRATION_ENV ?? "dev";

  constructor(private readonly prisma: PrismaService) {}

  async onModuleInit(): Promise<void> {
    await fs.mkdir(this.ingestDir, { recursive: true });
    await fs.mkdir(this.backupDir, { recursive: true });
    await this.reconcileOrphanedJobs();
  }

  /**
   * Fail any job still marked RUNNING at startup.
   *
   * Jobs run as a child of THIS process, so no job can outlive it: if a row says
   * RUNNING while we are booting, its process died with the previous instance
   * and nothing will ever finalize it. Since startJob() refuses to start while
   * any RUNNING row exists, one interrupted job wedges the panel permanently
   * with no way out from the UI — it took a manual UPDATE against production to
   * recover the first time this happened, when a deploy landed 110 seconds into
   * a REIMPORT.
   *
   * Deliberately unconditional rather than filtered on age: "started recently"
   * does not mean "still alive" here, and a fresh boot is proof enough that
   * nothing survived.
   */
  private async reconcileOrphanedJobs(): Promise<void> {
    try {
      // Read then write one by one rather than updateMany: the log needs the
      // reason APPENDED, and a job whose log just stops mid-step with no
      // explanation is what made the first occurrence hard to diagnose.
      const orphans = await this.prisma.opsJob.findMany({
        where: { status: "RUNNING" },
        select: { id: true, kind: true, log: true },
      });
      for (const job of orphans) {
        await this.prisma.opsJob.update({
          where: { id: job.id },
          data: {
            status: "FAILED",
            finishedAt: new Date(),
            log: {
              set:
                job.log +
                "\n[interrumpido: el contenedor se reinició mientras el trabajo corría; " +
                "el proceso hijo no sobrevive a un redespliegue. " +
                "Vuelva a ejecutar la operación desde el principio.]\n",
            },
          },
        });
        this.logger.warn(`trabajo ${job.kind} ${job.id} quedó huérfano; marcado FAILED`);
      }
    } catch (e) {
      // Never block startup on this. A failed reconcile leaves the panel
      // wedged, which is bad, but an API that will not boot is worse.
      this.logger.error(`no se pudieron reconciliar trabajos huérfanos: ${String(e)}`);
    }
  }

  /* -------------------------------------------------------------- ingest */

  private assertIngestName(name: string): string {
    if (INGEST_FILES.includes(name as IngestName)) return name;
    if (periodYearOf(name) !== null) return name;
    throw new BadRequestException(
      `Archivo no permitido. Debe ser uno de: ${INGEST_FILES.join(", ")}` +
        `, o un archivo de periodo anterior con nombre AAAA.accdb (por ejemplo 2025.accdb).`,
    );
  }

  async listIngest(): Promise<
    {
      name: string;
      present: boolean;
      size: number | null;
      modifiedAt: string | null;
      /** Set only on a prior-period archive; null on the four fixed sources. */
      periodYear: number | null;
    }[]
  > {
    // The four fixed sources are listed whether present or not — they are
    // required, so "missing" is the useful state to show. Period archives are
    // optional and unbounded, so they are listed only once uploaded, newest
    // year first.
    const fixed = await Promise.all(
      INGEST_FILES.map(async (name) => {
        try {
          const st = await fs.stat(path.join(this.ingestDir, name));
          return {
            name,
            present: true,
            size: st.size,
            modifiedAt: st.mtime.toISOString(),
            periodYear: null as number | null,
          };
        } catch {
          return {
            name,
            present: false,
            size: null,
            modifiedAt: null,
            periodYear: null as number | null,
          };
        }
      }),
    );

    let entries: string[] = [];
    try {
      entries = await fs.readdir(this.ingestDir);
    } catch {
      entries = [];
    }
    const periods = (
      await Promise.all(
        entries
          .map((name) => ({ name, year: periodYearOf(name) }))
          .filter((e): e is { name: string; year: number } => e.year !== null)
          .sort((a, b) => b.year - a.year)
          .map(async ({ name, year }) => {
            const st = await fs.stat(path.join(this.ingestDir, name));
            return {
              name,
              present: true,
              size: st.size,
              modifiedAt: st.mtime.toISOString(),
              periodYear: year,
            };
          }),
      )
    ).filter(Boolean);

    return [...fixed, ...periods];
  }

  async saveIngest(name: string, data: Buffer): Promise<void> {
    const safe = this.assertIngestName(name);
    await fs.writeFile(path.join(this.ingestDir, safe), data);
  }

  async deleteIngest(name: string): Promise<void> {
    const safe = this.assertIngestName(name);
    await fs.rm(path.join(this.ingestDir, safe), { force: true });
  }

  /* ------------------------------------------------------------- backups */

  private assertBackupName(name: string): string {
    // No path separators, must be a produced backup file.
    if (!/^[A-Za-z0-9._-]+\.sql\.gz$/.test(name)) {
      throw new BadRequestException("Nombre de respaldo inválido.");
    }
    return name;
  }

  async listBackups(): Promise<
    { name: string; size: number; createdAt: string }[]
  > {
    let names: string[];
    try {
      names = await fs.readdir(this.backupDir);
    } catch {
      return [];
    }
    const rows = await Promise.all(
      names
        .filter((n) => n.endsWith(".sql.gz"))
        .map(async (name) => {
          const st = await fs.stat(path.join(this.backupDir, name));
          return { name, size: st.size, createdAt: st.mtime.toISOString() };
        }),
    );
    return rows.sort((a, b) => b.createdAt.localeCompare(a.createdAt));
  }

  backupStream(name: string) {
    const safe = this.assertBackupName(name);
    const full = path.join(this.backupDir, safe);
    return { stream: createReadStream(full), name: safe };
  }

  async deleteBackup(name: string): Promise<void> {
    const safe = this.assertBackupName(name);
    await fs.rm(path.join(this.backupDir, safe), { force: true });
  }

  /* ---------------------------------------------------------------- jobs */

  listJobs(limit = 20) {
    return this.prisma.opsJob.findMany({
      orderBy: { startedAt: "desc" },
      take: limit,
    });
  }

  async getJob(id: string) {
    const job = await this.prisma.opsJob.findUnique({ where: { id } });
    if (!job) throw new NotFoundException("Trabajo no encontrado.");
    // Derived, never stored: the log is the single source of truth for how far
    // a job got, so progress cannot drift out of sync with it.
    return { ...job, progress: jobProgress(job.log, job.status) };
  }

  /**
   * Start a mutating op. Refuses if another job is already RUNNING. Returns the
   * new job row immediately; the process runs on in the background and appends
   * to `log` until it exits.
   */
  async startJob(
    kind: OpsJobKind,
    params: Record<string, unknown>,
    userId: string | undefined,
  ) {
    const running = await this.prisma.opsJob.count({ where: { status: "RUNNING" } });
    if (running > 0) {
      throw new ConflictException(
        "Ya hay una operación en curso. Espere a que termine.",
      );
    }

    const conn = this.opsConn();
    const { cmd, resolvedParams } = await this.buildCommand(kind, params, conn);

    const job = await this.prisma.opsJob.create({
      data: {
        kind,
        status: "RUNNING",
        log: "",
        params: resolvedParams as object,
        createdById: userId,
      },
    });

    this.run(job.id, cmd, conn.password);
    return job;
  }

  /* --------------------------------------------------------- internals */

  private parseDbUrl(): MysqlConn {
    const raw = process.env.DATABASE_URL;
    if (!raw) throw new BadRequestException("DATABASE_URL no está configurada.");
    const u = new URL(raw);
    return {
      host: u.hostname,
      port: u.port || "3306",
      user: decodeURIComponent(u.username),
      password: decodeURIComponent(u.password),
      database: u.pathname.replace(/^\//, ""),
    };
  }

  /**
   * The credentials mysqldump/mysql run as — deliberately NOT the application
   * user. `--single-transaction` issues FLUSH TABLES, which needs the global
   * RELOAD privilege, and the app user is granted only `ALL ON jorgecuadros.*`
   * plus `USAGE ON *.*`; `--skip-lock-tables` does not avoid it. A restore of a
   * dump taken before --set-gtid-purged=OFF likewise needs SUPER to replay its
   * SET @@GLOBAL.GTID_PURGED. So an admin credential is supplied out of band
   * rather than elevating the runtime user for the sake of one admin screen —
   * the same choice deploy/scripts/pre-migrate-backup.mjs makes.
   *
   * Host, port and database always come from DATABASE_URL: the ops user is a
   * different login on the SAME server, never a way to point at another one.
   *
   * With the vars unset this falls back to the DATABASE_URL credentials, which
   * is what local development wants — a dev MySQL grants the app user far more.
   */
  private opsConn(): MysqlConn {
    const conn = this.parseDbUrl();
    const user = process.env.OPS_DB_ADMIN_USER;
    const password = process.env.OPS_DB_ADMIN_PASSWORD;
    if (!user || !password) {
      this.logger.warn(
        "OPS_DB_ADMIN_USER/OPS_DB_ADMIN_PASSWORD no configuradas; " +
          `usando el usuario de la aplicación (${conn.user}) para mysqldump. ` +
          "En producción esto falla por falta del privilegio RELOAD.",
      );
      return conn;
    }
    return { ...conn, user, password };
  }

  /** mysql/mysqldump connection flags. The password goes through MYSQL_PWD in
   *  the child env, never on the command line (which would leak via `ps`). */
  private connFlags(c: MysqlConn): string {
    return `--host=${c.host} --port=${c.port} --user=${shq(c.user)}`;
  }

  private timestamp(): string {
    return new Date().toISOString().replace(/[:.]/g, "-").replace("T", "_").slice(0, 19);
  }

  /**
   * One hardened mysqldump, shared by BACKUP and by the safety backups SYNC and
   * REIMPORT take first. Kept byte-for-byte in spirit with the dump in
   * deploy/scripts/pre-migrate-backup.mjs — the two write into the same volume
   * and both are listed as restore points by this same screen.
   *
   * The dumper is probed at runtime rather than assumed. This command runs
   * inside the API image, whose `mysql-client` is Alpine's — i.e. MariaDB's —
   * where `mysqldump` is a deprecation-warning shim over `mariadb-dump` that
   * rejects --set-gtid-purged outright:
   *   mysqldump: unknown variable 'set-gtid-purged=OFF'
   * which failed every backup, including the safety backups SYNC and REIMPORT
   * take first. MariaDB's dumper emits no GTID state unless asked (--gtid), so
   * there is nothing to suppress there; the flag is passed only when the dumper
   * on PATH advertises it, and the real binary is called directly only in the
   * MariaDB case (calling `mariadb-dump` whenever it merely exists would pick
   * it over a MySQL `mysqldump` earlier in PATH on a host carrying both).
   *
   * The probe is a command substitution, not `--help | grep -q`: PIPEFAIL is in
   * effect and grep closing the pipe early would make a supported flag look
   * unsupported.
   *
   * --set-gtid-purged=OFF (MySQL only): the production server is the
   * replication SOURCE with GTID on, so without it every dump embeds
   * SET @@GLOBAL.GTID_PURGED and is unrestorable onto the very server it came
   * from.
   *
   * The table-count assertion is not belt-and-braces: `gzip -t` passes on the
   * ~372-byte output of a mysqldump that died on its first statement, so a
   * failed dump would otherwise be recorded as a successful backup. (`set -o
   * pipefail` is set by the caller for the same reason — without it the exit
   * status of the pipeline is gzip's, and gzip succeeded.)
   *
   * A failed attempt deletes its own output, so a truncated file never appears
   * in the restore list looking like an ordinary restore point.
   */
  private dumpCommand(flags: string, db: string, out: string): string {
    return (
      `DUMP=mysqldump; GTID=; ` +
      `case "$(mysqldump --help 2>/dev/null || true)" in ` +
      `*set-gtid-purged*) GTID=--set-gtid-purged=OFF;; ` +
      `*) command -v mariadb-dump >/dev/null 2>&1 && DUMP=mariadb-dump;; esac; ` +
      `( $DUMP ${flags} --single-transaction --routines --triggers ` +
      `--no-tablespaces $GTID ${db} | gzip -c > ${out} && ` +
      `gzip -t ${out} && ` +
      `TABLAS=$(gunzip -c ${out} | grep -c 'CREATE TABLE') && ` +
      `echo "tablas capturadas: $TABLAS" && ` +
      `[ "$TABLAS" -ge 1 ] ) || ` +
      `{ rm -f ${out}; echo 'respaldo incompleto eliminado'; exit 1; }`
    );
  }

  private async buildCommand(
    kind: OpsJobKind,
    params: Record<string, unknown>,
    conn: MysqlConn,
  ): Promise<{ cmd: string; resolvedParams: Record<string, unknown> }> {
    const flags = this.connFlags(conn);
    const db = shq(conn.database);

    if (kind === "BACKUP") {
      const file = `backup-${this.migrationEnv}-${this.timestamp()}.sql.gz`;
      const out = shq(path.join(this.backupDir, file));
      return {
        cmd: `${PIPEFAIL}${this.dumpCommand(flags, db, out)}`,
        resolvedParams: { file },
      };
    }

    if (kind === "RESTORE") {
      const name = this.assertBackupName(String(params.file ?? ""));
      const full = path.join(this.backupDir, name);
      await fs.access(full).catch(() => {
        throw new NotFoundException(`Respaldo no encontrado: ${name}`);
      });
      return {
        // pipefail matters here too: a corrupt archive makes gunzip fail while
        // mysql, fed a truncated stream, can still exit 0 — a restore that
        // reported success having replayed only part of the dump.
        cmd: `${PIPEFAIL}gunzip -c ${shq(full)} | mysql ${flags} ${db}`,
        resolvedParams: { file: name },
      };
    }

    if (kind === "SYNC") {
      const file = `pre-sync-${this.migrationEnv}-${this.timestamp()}.sql.gz`;
      const out = shq(path.join(this.backupDir, file));
      const py = await this.pythonBin();
      const runAll = shq(path.join(this.migrationDir, "run_all.py"));
      const cmd =
        `${PIPEFAIL}echo '== Respaldo de seguridad previo ==' && ` +
        `${this.dumpCommand(flags, db, out)} && ` +
        `echo '== Sincronización aditiva desde carpeta de ingesta ==' && ` +
        // --stage is not optional here. The staged Parquet lives in the image
        // at migration/output, NOT on a volume, so every redeploy wipes it and
        // a sync without --stage dies on a missing stg_*/*.parquet. Re-staging
        // is also the only thing that makes "desde carpeta de ingesta" true:
        // stale Parquet would sync the previous upload, not the current one.
        `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --stage --sync`;
      return { cmd, resolvedParams: { safetyBackup: file } };
    }

    if (kind === "REIMPORT") {
      // Safety backup first, then a full truncate+rebuild from the ingest files.
      const file = `pre-reimport-${this.migrationEnv}-${this.timestamp()}.sql.gz`;
      const out = shq(path.join(this.backupDir, file));
      const py = await this.pythonBin();
      const runAll = shq(path.join(this.migrationDir, "run_all.py"));
      // run_all.py runs native_guard.py before it truncates anything and exits
      // without touching the database when the target holds rows that only
      // exist here — allocated portal NUMids, app-created customers, OCR
      // captures. --force-full is what the operator ticks to delete them
      // anyway; without it the job fails with the list.
      const force = params.forceFull === true;
      const cmd =
        `${PIPEFAIL}echo '== Respaldo de seguridad previo ==' && ` +
        `${this.dumpCommand(flags, db, out)} && ` +
        `echo '== Reimportación desde carpeta de ingesta ==' && ` +
        `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --stage` +
        (force ? " --force-full" : "");
      return { cmd, resolvedParams: { safetyBackup: file, forceFull: force } };
    }

    throw new BadRequestException(`Operación no soportada: ${kind}`);
  }

  /** Prefer the migration venv python if it exists (local dev), else system. */
  private async pythonBin(): Promise<string> {
    const venv = path.join(this.migrationDir, ".venv", "bin", "python");
    try {
      await fs.access(venv);
      return venv;
    } catch {
      return process.env.PYTHON_BIN ?? "python3";
    }
  }

  /**
   * `password` is the ops credential from opsConn(), exported as MYSQL_PWD so it
   * never reaches argv (which `ps` exposes to every process on the host).
   *
   * It does not leak into the Python ETL that SYNC and REIMPORT go on to run:
   * migration/dbenv.py connects with pymysql using the credentials inside
   * DATABASE_URL and never consults MYSQL_PWD. The ETL keeps running as the
   * application user, which is what it should be doing.
   */
  private run(jobId: string, cmd: string, password: string): void {
    const child = spawn("sh", ["-c", cmd], {
      cwd: this.migrationDir,
      env: {
        ...process.env,
        MYSQL_PWD: password,
        INGEST_DIR: this.ingestDir,
        BACKUP_DIR: this.backupDir,
      },
    });

    let buffer = "";
    let pending = "";
    let flushing = false;
    let flushTimer: NodeJS.Timeout | null = null;

    const flush = async () => {
      if (flushing || !pending) return;
      flushing = true;
      const chunk = pending;
      pending = "";
      try {
        await this.prisma.opsJob.update({
          where: { id: jobId },
          data: { log: { set: buffer } },
        });
      } catch (e) {
        this.logger.warn(`ops job ${jobId} log flush failed: ${String(e)}`);
      } finally {
        flushing = false;
        void chunk;
      }
    };

    const onData = (d: Buffer) => {
      const text = d.toString();
      buffer += text;
      pending += text;
      if (!flushTimer) {
        flushTimer = setTimeout(() => {
          flushTimer = null;
          void flush();
        }, 1000);
      }
    };

    child.stdout.on("data", onData);
    child.stderr.on("data", onData);

    const finalize = async (status: "SUCCESS" | "FAILED", tail: string) => {
      if (flushTimer) clearTimeout(flushTimer);
      buffer += tail;
      await this.prisma.opsJob
        .update({
          where: { id: jobId },
          data: { status, log: { set: buffer }, finishedAt: new Date() },
        })
        .catch((e) => this.logger.error(`ops job ${jobId} finalize failed: ${String(e)}`));
    };

    child.on("error", (err) => {
      void finalize("FAILED", `\n[proceso no pudo iniciar] ${err.message}\n`);
    });

    child.on("close", (code) => {
      if (code === 0) void finalize("SUCCESS", `\n[completado con éxito]\n`);
      else void finalize("FAILED", `\n[terminó con código ${code}]\n`);
    });
  }
}

/** Single-quote a value for a POSIX shell command. */
function shq(v: string): string {
  return `'${v.replace(/'/g, `'\\''`)}'`;
}

/** Progress derived from a job's log. Null when the job reports no steps. */
export interface JobProgress {
  /** 1-based index of the step currently running (or last reached). */
  step: number;
  total: number;
  /** Script name, e.g. "transform_bank.py". */
  name: string;
  /** 0..100, floored. 100 only once the job is no longer RUNNING. */
  percent: number;
}

/**
 * Parse the "[paso i/N] name" markers migration/run_all.py emits.
 *
 * Progress is DERIVED from the log rather than tracked in a column: the log is
 * already the record of what happened, and a separate counter could disagree
 * with it — which is exactly the confusion a progress display is supposed to
 * remove. run_all.py owns the step count, so adding a step cannot desync this.
 *
 * BACKUP and RESTORE are a single mysqldump with no steps, so they return null
 * and the UI shows an indeterminate spinner. Reporting a fabricated percentage
 * for them would be worse than showing none.
 */
export function jobProgress(
  log: string,
  status: string,
): JobProgress | null {
  // Last marker wins: the log grows, and the newest line is the current step.
  const matches = [...log.matchAll(/^\[paso (\d+)\/(\d+)\] (\S+)/gm)];
  const last = matches[matches.length - 1];
  if (!last) return null;

  const step = Number(last[1]);
  const total = Number(last[2]);
  if (!Number.isFinite(step) || !Number.isFinite(total) || total <= 0) return null;

  // While RUNNING, step i means i is IN PROGRESS, not finished — so report
  // (i-1) completed. Claiming 100% while the last step is still working is the
  // classic progress-bar lie, and here the last step (blob_extract) is also the
  // slowest, so it would sit at "100%" for the longest stretch of the job.
  const done = status === "RUNNING" ? step - 1 : step;
  return {
    step,
    total,
    name: last[3],
    percent: Math.max(0, Math.min(100, Math.floor((done / total) * 100))),
  };
}