diff --git a/.gitignore b/.gitignore index 2936def..5a6ff02 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,8 @@ build/ *.log migration/output/ migration/.venv/ +migration/ingest/ +migration/backups/ __pycache__/ *.pyc packages/database/generated/ diff --git a/PLAN.md b/PLAN.md index 1f8431c..84de7b0 100644 --- a/PLAN.md +++ b/PLAN.md @@ -128,7 +128,7 @@ Given the amount of near-duplicate/overlapping data across snapshot tables (mult 6. Shared billing/statements module (the payoff: one statement per customer spanning both utility and insurance transactions) — **DONE**. `apps/api/src/billing/` + web `/estado-cuenta` and `/estado-cuenta/[id]`. Two questions, two views: a per-customer **balances worklist** (who owes what) and a cross-customer **movement browser** (every charge and credit, filterable by line, concept, origin table and date range, with totals for the whole filtered set). The detail page is the actual statement: balance per currency, the same balance split by business line, charges broken out by concept, and the full movement list with a running balance. **Design constraint that shapes the whole module: balances are reported per currency and never collapsed into one number.** 912 of the 1,269 customers with a ledger move in both MXN and USD, the charge side is MXN-only while receipts arrive in both, and the legacy data never stored the exchange rate applied to a movement — so a single "total balance" would be a figure that never existed in the books. 7. Bank register module (`bank_transactions`/`business_line_categories` from SCOTHIA) — small, self-contained, and has no customer FK, so it can slot in independently once the core migration pipeline exists; low risk, low priority relative to the customer-facing modules. 8. VPS provisioning + Tailscale + MySQL replication setup. `utility_dbo`'s schema is now available (full dump on disk — 55 tables; see Status), so the exact replicated table/column set and inbox-table shape can be finalized against the real portal DB and the portal PHP code (`my-jorgecuadros-web`) that reads/writes it. -9. Sync worker (push replicated tables' relevant subset, poll inbox tables for payment/propane submissions) — depends on step 8. Portal write points confirmed present in `utility_dbo`: `peticion_gas` (propane requests), PayPal payment writes, `notifications_settings`, `verification_codes` — these define the VPS→internal inbox set. +9. Sync worker (push replicated tables' relevant subset, poll inbox tables for payment/propane submissions) — depends on step 8. **The separate Phase B Access additive sync is implemented:** `migration/run_all.py --sync` and the admin `SYNC` job upsert legacy-owned rows without truncating the database or touching manual rows. Portal write points confirmed present in `utility_dbo`: `peticion_gas` (propane requests), PayPal payment writes, `notifications_settings`, `verification_codes` — these define the VPS→internal inbox set. 10. Reports/email campaigns/admin — parity with old app's `reports.php`/`emailCampaigns.php` intent, rebuilt properly. ## Status @@ -146,6 +146,9 @@ Repo scaffolded at `jorgecuadros-platform/`: npm workspaces, NestJS API with a r - **i18n:** **Spanish-first** UI — matches the source data and staff usage. - **CI/CD:** **Gitea Actions** on `git.mancinas.io` — build image, push to the git.mancinas.io container registry, deploy to Portainer (mirrors the `portainer-gitea-deploy` pattern already used on this LAN). Jenkins/`git.freakma.com` dropped. - **`utility_dbo`:** resolved — full dump + portal code available (see Status). +- **Phase B Access additive sync:** implemented in `migration/run_all.py --sync` and the admin + `SYNC` job. It preserves manual rows and stable legacy-owned primary keys; end-to-end database + validation remains before production use. ## Open items (ops, not design) @@ -156,5 +159,7 @@ Repo scaffolded at `jorgecuadros-platform/`: npm workspaces, NestJS API with a r - Migration: automated row-count/sum reconciliation between `staging` and final schema per table group (see step 5 above), run as part of the migration script, not a manual spot-check. - App: standard NestJS unit/integration tests per module (auth guards, Prisma queries), Playwright/Cypress e2e for the core "look up a customer, see their unified policies + services + statement" flow — the thing the whole project exists to deliver. -- Sync: once the VPS replica and inbox tables exist, verify replication lag stays low (a few seconds to low minutes) and that a payment/propane submission on the portal reliably shows up in the internal app within one polling interval, before relying on it operationally. +- **Sync:** Phase B Access additive sync now has automated CLI/admin wiring, but must be verified + against a disposable DB with stable-PK, manual-row, update, and source-delete cases. The + separate VPS/portal sync still requires VPS provisioning and inbox-table implementation. - Before cutover: run the new app against migrated data side-by-side with the live Access files for a period, comparing balances/statements for a sample of active customers to catch migration logic errors before the Access files are retired. diff --git a/RESUME.md b/RESUME.md index 1351726..13a3f39 100644 --- a/RESUME.md +++ b/RESUME.md @@ -162,6 +162,17 @@ the reconciliation pass (done, then corrected) are all closed. See §3 and §8. 5. **`TRASPASOS PAYPAL` is a clearing account, not a customer** — carries -7.03M MXN over 309 movements and therefore tops the adeudo worklist. Deliberately not special-cased in code; needs a business decision on how to model it. +6. **DB Operations — Phase B (additive sync) — IMPLEMENTED, verification pending.** Phase A provides + the admin-only `/operaciones` page + `ops` API module (ability `db:manage`, ADMIN), ingest + folder, backup, restore, and destructive re-import. Phase B now enables `SYNC`: `OpsService` + creates a safety backup and runs `run_all.py --sync`; transforms upsert legacy-owned rows by + provenance keys while preserving existing PKs and rows whose `legacyId IS NULL` (manual). + Prisma now enforces provenance uniqueness for properties, policies, transactions, vehicles, + and bank transactions. Sync intentionally skips prune/blob steps so manual customers and + document pointers are not removed. Python compilation plus API/web production builds pass; + still required before production use: push updated Prisma schema and run an end-to-end sync + against a disposable/dev DB proving stable PKs, manual-row preservation, changed-row updates, + and legacy-delete handling. ## 7. Environment notes (current macOS machine) @@ -374,11 +385,14 @@ for what's actually next. --- -⏭ **NEXT — where to pick up:** - -- **Plan step 8–9: VPS + sync worker.** Blocked on VPS provisioning (§6.1) — the only real - external dependency left. Pure ops: provider, size, Tailscale, MySQL replica. -- **Plan step 10: reports / email campaigns / admin.** +- **Sync implementation — DONE, validation pending.** `run_all.py --sync` performs the + non-destructive legacy upsert path for customers, properties, policies, transactions, and + bank rows. It preserves manual rows and stable legacy-owned primary keys; the admin SYNC job + automatically creates a pre-sync backup. Next validation: apply schema changes, then exercise + sync against a disposable DB with added, changed, removed, and manually-created rows. +- **Plan step 9: portal sync worker** remains separate and blocked on VPS provisioning. This + Phase B feature synchronizes Access source files into the internal platform; it does not yet + poll `utility_dbo` inbox tables or replicate portal-facing data to a VPS. - **Small / open:** (a) `TRASPASOS PAYPAL` clearing account still tops the adeudo worklist (§6.4d) — a business modelling call, not code. (b) Credential rotation on the old repo's exposed MySQL password. (c) The `/estado-cuenta` browser visual pass — `/banco` was verified diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 2ecc723..7833e82 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -9,6 +9,7 @@ import { PoliciesModule } from "./policies/policies.module"; import { PropertiesModule } from "./properties/properties.module"; import { BillingModule } from "./billing/billing.module"; import { BankModule } from "./bank/bank.module"; +import { OpsModule } from "./ops/ops.module"; import { AppController } from "./app.controller"; @Module({ @@ -23,6 +24,7 @@ import { AppController } from "./app.controller"; PropertiesModule, BillingModule, BankModule, + OpsModule, ], controllers: [AppController], }) diff --git a/apps/api/src/auth/abilities.ts b/apps/api/src/auth/abilities.ts index c7395aa..9ec2460 100644 --- a/apps/api/src/auth/abilities.ts +++ b/apps/api/src/auth/abilities.ts @@ -32,7 +32,8 @@ export type Ability = | "bank:create" | "bank:void" | "lookup:manage" - | "user:manage"; + | "user:manage" + | "db:manage"; /** Minimum role required for each ability. */ export const ABILITY_MIN: Record = { @@ -51,6 +52,7 @@ export const ABILITY_MIN: Record = { "bank:void": "MANAGER", "lookup:manage": "MANAGER", "user:manage": "ADMIN", + "db:manage": "ADMIN", }; export const ALL_ABILITIES = Object.keys(ABILITY_MIN) as Ability[]; diff --git a/apps/api/src/ops/ops.controller.ts b/apps/api/src/ops/ops.controller.ts new file mode 100644 index 0000000..86b8078 --- /dev/null +++ b/apps/api/src/ops/ops.controller.ts @@ -0,0 +1,120 @@ +import { + Body, + Controller, + Delete, + Get, + Param, + Post, + Req, + Res, + StreamableFile, + UploadedFile, + UseGuards, + UseInterceptors, +} from "@nestjs/common"; +import { FileInterceptor } from "@nestjs/platform-express"; +import { Request, Response } from "express"; +import { AuthenticatedGuard } from "../auth/authenticated.guard"; +import { AbilityGuard } from "../auth/ability.guard"; +import { RequireAbility } from "../auth/require-ability.decorator"; +import { AuditService } from "../common/audit.service"; +import { OpsService } from "./ops.service"; +import { StartJobDto } from "./start-job.dto"; + +/** Every route is ADMIN-only (ability "db:manage"). */ +@UseGuards(AuthenticatedGuard, AbilityGuard) +@RequireAbility("db:manage") +@Controller("ops") +export class OpsController { + constructor( + private readonly ops: OpsService, + private readonly audit: AuditService, + ) {} + + private actingId(req: Request): string { + return (req.user as { id: string }).id; + } + + /* ------------------------------------------------------------- ingest */ + + @Get("ingest") + listIngest() { + return this.ops.listIngest(); + } + + @Post("ingest/:name") + @UseInterceptors( + FileInterceptor("file", { limits: { fileSize: 500 * 1024 * 1024 } }), + ) + async uploadIngest( + @Param("name") name: string, + @UploadedFile() file: { buffer: Buffer; size: number } | undefined, + @Req() req: Request, + ) { + if (!file) throw new Error("No se recibió ningún archivo."); + await this.ops.saveIngest(name, file.buffer); + void this.audit.log(this.actingId(req), "ops.ingest.upload", { + name, + size: file.size, + }); + return { ok: true }; + } + + @Delete("ingest/:name") + async deleteIngest(@Param("name") name: string, @Req() req: Request) { + await this.ops.deleteIngest(name); + void this.audit.log(this.actingId(req), "ops.ingest.delete", { name }); + return { ok: true }; + } + + /* ------------------------------------------------------------ backups */ + + @Get("backups") + listBackups() { + return this.ops.listBackups(); + } + + @Get("backups/:name/download") + download( + @Param("name") name: string, + @Res({ passthrough: true }) res: Response, + ): StreamableFile { + const { stream, name: safe } = this.ops.backupStream(name); + res.set({ + "Content-Type": "application/gzip", + "Content-Disposition": `attachment; filename="${safe}"`, + }); + return new StreamableFile(stream); + } + + @Delete("backups/:name") + async deleteBackup(@Param("name") name: string, @Req() req: Request) { + await this.ops.deleteBackup(name); + void this.audit.log(this.actingId(req), "ops.backup.delete", { name }); + return { ok: true }; + } + + /* --------------------------------------------------------------- jobs */ + + @Get("jobs") + listJobs() { + return this.ops.listJobs(); + } + + @Get("jobs/:id") + getJob(@Param("id") id: string) { + return this.ops.getJob(id); + } + + @Post("jobs") + async startJob(@Body() dto: StartJobDto, @Req() req: Request) { + const userId = this.actingId(req); + const job = await this.ops.startJob(dto.kind, { file: dto.file }, userId); + void this.audit.log(userId, "ops.job.start", { + jobId: job.id, + kind: dto.kind, + file: dto.file, + }); + return job; + } +} diff --git a/apps/api/src/ops/ops.module.ts b/apps/api/src/ops/ops.module.ts new file mode 100644 index 0000000..05824e1 --- /dev/null +++ b/apps/api/src/ops/ops.module.ts @@ -0,0 +1,9 @@ +import { Module } from "@nestjs/common"; +import { OpsController } from "./ops.controller"; +import { OpsService } from "./ops.service"; + +@Module({ + controllers: [OpsController], + providers: [OpsService], +}) +export class OpsModule {} diff --git a/apps/api/src/ops/ops.service.ts b/apps/api/src/ops/ops.service.ts new file mode 100644 index 0000000..891b98a --- /dev/null +++ b/apps/api/src/ops/ops.service.ts @@ -0,0 +1,361 @@ +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]; + +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); + + private readonly migrationDir = + process.env.MIGRATION_DIR ?? path.resolve(process.cwd(), "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 { + await fs.mkdir(this.ingestDir, { recursive: true }); + await fs.mkdir(this.backupDir, { recursive: true }); + } + + /* -------------------------------------------------------------- ingest */ + + private assertIngestName(name: string): IngestName { + if (!INGEST_FILES.includes(name as IngestName)) { + throw new BadRequestException( + `Archivo no permitido. Debe ser uno de: ${INGEST_FILES.join(", ")}`, + ); + } + return name as IngestName; + } + + async listIngest(): Promise< + { name: string; present: boolean; size: number | null; modifiedAt: string | null }[] + > { + return 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(), + }; + } catch { + return { name, present: false, size: null, modifiedAt: null }; + } + }), + ); + } + + async saveIngest(name: string, data: Buffer): Promise { + const safe = this.assertIngestName(name); + await fs.writeFile(path.join(this.ingestDir, safe), data); + } + + async deleteIngest(name: string): Promise { + 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 { + 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."); + return job; + } + + /** + * 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, + 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.parseDbUrl(); + 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(/^\//, ""), + }; + } + + /** 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); + } + + private async buildCommand( + kind: OpsJobKind, + params: Record, + conn: MysqlConn, + ): Promise<{ cmd: string; resolvedParams: Record }> { + 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: `mysqldump ${flags} --single-transaction --routines --triggers --no-tablespaces ${db} | gzip -c > ${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 { + cmd: `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 = + `echo '== Respaldo de seguridad previo ==' && ` + + `mysqldump ${flags} --single-transaction --routines --triggers --no-tablespaces ${db} | gzip -c > ${out} && ` + + `echo '== Sincronización aditiva desde carpeta de ingesta ==' && ` + + `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --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")); + const cmd = + `echo '== Respaldo de seguridad previo ==' && ` + + `mysqldump ${flags} --single-transaction --routines --triggers --no-tablespaces ${db} | gzip -c > ${out} && ` + + `echo '== Reimportación desde carpeta de ingesta ==' && ` + + `${shq(py)} ${runAll} --env ${shq(this.migrationEnv)} --stage`; + return { cmd, resolvedParams: { safetyBackup: file } }; + } + + throw new BadRequestException(`Operación no soportada: ${kind}`); + } + + /** Prefer the migration venv python if it exists (local dev), else system. */ + private async pythonBin(): Promise { + const venv = path.join(this.migrationDir, ".venv", "bin", "python"); + try { + await fs.access(venv); + return venv; + } catch { + return process.env.PYTHON_BIN ?? "python3"; + } + } + + 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, `'\\''`)}'`; +} diff --git a/apps/api/src/ops/start-job.dto.ts b/apps/api/src/ops/start-job.dto.ts new file mode 100644 index 0000000..80bf26d --- /dev/null +++ b/apps/api/src/ops/start-job.dto.ts @@ -0,0 +1,12 @@ +import { IsEnum, IsOptional, IsString } from "class-validator"; +import { OpsJobKind } from "@jorgecuadros/database"; + +export class StartJobDto { + @IsEnum(OpsJobKind) + kind!: OpsJobKind; + + /** Target backup filename — required for RESTORE, ignored otherwise. */ + @IsOptional() + @IsString() + file?: string; +} diff --git a/apps/web/src/app/globals.css b/apps/web/src/app/globals.css index 14f4caa..62d758a 100644 --- a/apps/web/src/app/globals.css +++ b/apps/web/src/app/globals.css @@ -247,6 +247,7 @@ button { border-radius: 7px; font-size: 14px; font-weight: 500; + white-space: nowrap; } .appbar-link:hover { color: #fff; @@ -400,13 +401,24 @@ button { } .btn-ghost { background: transparent; + color: var(--ink-soft); + border-color: var(--line-strong); +} +.btn-ghost:hover { + background: var(--surface-2); + color: var(--ink); + border-color: var(--brand-600); + text-decoration: none; +} +/* Dark appbar keeps the light-on-dark ghost button. */ +.appbar .btn-ghost { color: rgba(242, 239, 231, 0.85); border-color: rgba(255, 255, 255, 0.22); } -.btn-ghost:hover { +.appbar .btn-ghost:hover { background: rgba(255, 255, 255, 0.1); color: #fff; - text-decoration: none; + border-color: rgba(255, 255, 255, 0.35); } .btn-outline { background: var(--surface); @@ -2153,3 +2165,42 @@ button { position: relative; z-index: 1; } + +/* --- Admin DB operations (Operaciones) --- */ +.btn-danger { + background: var(--negative); + color: #fff; +} +.btn-danger:hover { + filter: brightness(0.94); + text-decoration: none; +} +.btn-danger:disabled { + opacity: 0.5; + cursor: not-allowed; +} + +.ops-log { + margin: 12px 0 0; + padding: 12px 14px; + max-height: 320px; + overflow: auto; + background: var(--surface-2); + border: 1px solid var(--line); + border-radius: var(--radius-sm); + font-family: ui-monospace, SFMono-Regular, Menlo, monospace; + font-size: 12px; + line-height: 1.5; + white-space: pre-wrap; + word-break: break-word; +} + +.modal-backdrop { + position: fixed; + inset: 0; + background: rgba(0, 0, 0, 0.4); + display: grid; + place-items: center; + padding: 20px; + z-index: 50; +} diff --git a/apps/web/src/app/operaciones/page.tsx b/apps/web/src/app/operaciones/page.tsx new file mode 100644 index 0000000..6e27dc6 --- /dev/null +++ b/apps/web/src/app/operaciones/page.tsx @@ -0,0 +1,515 @@ +"use client"; + +import { useCallback, useEffect, useRef, useState } from "react"; +import { AppShell } from "@/components/AppShell"; +import { useCan } from "@/lib/abilities"; +import { + OPS_KIND_LABELS, + OPS_STATUS_LABELS, + formatBytes, + formatDateTime, +} from "@/lib/labels"; +import { + backupDownloadUrl, + deleteBackup, + deleteIngest, + getOpsJob, + listBackups, + listIngest, + listOpsJobs, + startOpsJob, + uploadIngest, +} from "@/lib/api"; +import type { + BackupFile, + IngestFile, + OpsJob, + OpsJobKind, +} from "@/lib/types"; + +export default function OperacionesPage() { + return ( + + + + ); +} + +type ConfirmState = + | { kind: "REIMPORT" } + | { kind: "RESTORE"; file: string } + | null; + +function Operaciones() { + const allowed = useCan("db:manage"); + + const [ingest, setIngest] = useState(null); + const [backups, setBackups] = useState(null); + const [jobs, setJobs] = useState(null); + const [activeJob, setActiveJob] = useState(null); + + const [error, setError] = useState(null); + const [notice, setNotice] = useState(null); + const [confirm, setConfirm] = useState(null); + const [confirmText, setConfirmText] = useState(""); + const [uploading, setUploading] = useState(null); + const [starting, setStarting] = useState(false); + + const fileInputs = useRef>({}); + + const refreshLists = useCallback(() => { + listIngest().then(setIngest).catch(() => setIngest([])); + listBackups().then(setBackups).catch(() => setBackups([])); + listOpsJobs() + .then((rows) => { + setJobs(rows); + const running = rows.find((j) => j.status === "RUNNING"); + if (running) setActiveJob(running); + }) + .catch(() => setJobs([])); + }, []); + + useEffect(() => { + if (allowed) refreshLists(); + }, [allowed, refreshLists]); + + // Poll the active job while it runs; refresh everything when it finishes. + useEffect(() => { + if (!activeJob || activeJob.status !== "RUNNING") return; + const id = activeJob.id; + const timer = setInterval(() => { + getOpsJob(id) + .then((job) => { + setActiveJob(job); + if (job.status !== "RUNNING") { + clearInterval(timer); + refreshLists(); + setNotice( + job.status === "SUCCESS" + ? `${OPS_KIND_LABELS[job.kind]} completada.` + : `${OPS_KIND_LABELS[job.kind]} terminó con error. Revise el registro.`, + ); + } + }) + .catch(() => { + /* transient — keep polling */ + }); + }, 1500); + return () => clearInterval(timer); + }, [activeJob, refreshLists]); + + if (!allowed) { + return ( +
+

Operaciones

+
+ No tiene permisos para administrar la base de datos. +
+
+ ); + } + + const jobRunning = activeJob?.status === "RUNNING"; + + async function handleUpload(name: string, file: File | undefined) { + if (!file) return; + setError(null); + setNotice(null); + setUploading(name); + try { + await uploadIngest(name, file); + setNotice(`${name} cargado.`); + refreshLists(); + } catch (e) { + setError((e as Error)?.message ?? "No se pudo cargar el archivo."); + } finally { + setUploading(null); + const input = fileInputs.current[name]; + if (input) input.value = ""; + } + } + + async function handleDeleteIngest(name: string) { + setError(null); + try { + await deleteIngest(name); + refreshLists(); + } catch (e) { + setError((e as Error)?.message ?? "No se pudo eliminar."); + } + } + + async function handleDeleteBackup(name: string) { + setError(null); + try { + await deleteBackup(name); + refreshLists(); + } catch (e) { + setError((e as Error)?.message ?? "No se pudo eliminar el respaldo."); + } + } + + async function start(kind: OpsJobKind, file?: string) { + setError(null); + setNotice(null); + setStarting(true); + try { + const job = await startOpsJob(kind, file); + setActiveJob(job); + setJobs((prev) => (prev ? [job, ...prev] : [job])); + } catch (e) { + setError((e as Error)?.message ?? "No se pudo iniciar la operación."); + } finally { + setStarting(false); + } + } + + function askConfirm(state: ConfirmState) { + setConfirm(state); + setConfirmText(""); + setError(null); + setNotice(null); + } + + async function runConfirmed() { + if (!confirm) return; + const c = confirm; + setConfirm(null); + if (c.kind === "REIMPORT") await start("REIMPORT"); + else await start("RESTORE", c.file); + } + + const ingestReady = (ingest ?? []).every((f) => f.present); + + return ( + <> +
+

Operaciones de base de datos

+
+ + {error &&
{error}
} + {notice &&
{notice}
} + + {/* Active / running job with live log */} + {activeJob && ( +
+
+

+ {OPS_KIND_LABELS[activeJob.kind]}{" "} + + {jobRunning && } + {OPS_STATUS_LABELS[activeJob.status]} + +

+ {!jobRunning && ( + + )} +
+ 
{activeJob.log || "Iniciando…"}
+
+ )} + + {/* Ingest folder */} +
+

Carpeta de ingesta

+

+ Los cuatro archivos originales de Access. La reimportación y la + sincronización leen de aquí. +

+
+ + + + + + + + + + + + {(ingest ?? []).map((f) => ( + + + + + + + + ))} + +
ArchivoEstadoTamañoModificadoAcciones
{f.name} + + {f.present ? "Presente" : "Falta"} + + {formatBytes(f.size)}{formatDateTime(f.modifiedAt)} +
+ { + fileInputs.current[f.name] = el; + }} + type="file" + style={{ display: "none" }} + onChange={(e) => handleUpload(f.name, e.target.files?.[0])} + /> + + {f.present && ( + + )} +
+
+
+
+ + {/* Operations */} +
+

Operaciones

+
+ start("BACKUP")} + /> + askConfirm({ kind: "REIMPORT" })} + /> + {}} + /> +
+ {!ingestReady && ( +

+ La reimportación requiere que los cuatro archivos estén presentes. +

+ )} +
+ + {/* Backups */} +
+

Respaldos

+

+ Restaurar sobreescribe la base de datos completa con el respaldo elegido. +

+
+ + + + + + + + + + + {backups === null ? ( + + + + ) : backups.length === 0 ? ( + + + + ) : ( + backups.map((b) => ( + + + + + + + )) + )} + +
ArchivoTamañoCreadoAcciones
+ +
+ Sin respaldos. +
{b.name}{formatBytes(b.size)}{formatDateTime(b.createdAt)} +
+ + Descargar + + + +
+
+
+
+ + {/* Recent jobs */} +
+

Historial

+
+ + + + + + + + + + + + {(jobs ?? []).map((j) => ( + + + + + + + + ))} + {jobs && jobs.length === 0 && ( + + + + )} + +
OperaciónEstadoInicioFin
{OPS_KIND_LABELS[j.kind]} + + {OPS_STATUS_LABELS[j.status]} + + {formatDateTime(j.startedAt)}{formatDateTime(j.finishedAt)} + +
+ Sin operaciones registradas. +
+
+
+ + {/* Destructive-op confirm */} + {confirm && ( +
+
+

+ {confirm.kind === "REIMPORT" ? "Confirmar reimportación" : "Confirmar restauración"} +

+

+ {confirm.kind === "REIMPORT" + ? "Esto BORRA todos los datos actuales (incluidos los capturados a mano) y reconstruye desde los archivos de ingesta. Se creará un respaldo previo automático." + : `Esto sobreescribe la base de datos completa con “${confirm.file}”. Se recomienda crear un respaldo antes.`} +

+ +
+ + +
+
+
+ )} + + ); +} + +function OpTile({ + title, + desc, + action, + tone, + disabled, + onClick, +}: { + title: string; + desc: string; + action: string; + tone: "primary" | "danger" | "muted"; + disabled: boolean; + onClick: () => void; +}) { + const btnClass = + tone === "danger" ? "btn btn-danger" : tone === "muted" ? "btn btn-outline" : "btn btn-primary"; + return ( +
+

+ {title} +

+

+ {desc} +

+ +
+ ); +} diff --git a/apps/web/src/components/AppShell.tsx b/apps/web/src/components/AppShell.tsx index d9d721b..8824589 100644 --- a/apps/web/src/components/AppShell.tsx +++ b/apps/web/src/components/AppShell.tsx @@ -22,6 +22,7 @@ const NAV: { href: string; label: string; ability?: Ability }[] = [ { href: "/banco", label: "Chequera" }, { href: "/catalogos", label: "Catálogos", ability: "lookup:manage" }, { href: "/usuarios", label: "Usuarios", ability: "user:manage" }, + { href: "/operaciones", label: "Operaciones", ability: "db:manage" }, ]; export function AppShell({ children }: { children: ReactNode }) { diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index 4006377..6212e7f 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -34,6 +34,10 @@ import type { PolicyStats, PolicyStatus, LookupsResponse, + OpsJob, + OpsJobKind, + IngestFile, + BackupFile, PropertyDetail, PropertyFacets, PropertyInput, @@ -593,3 +597,62 @@ export function resetUserPassword(id: string, password: string): Promise { + return apiFetch("/ops/ingest"); +} + +/** Multipart upload — not JSON, so it bypasses apiFetch's Content-Type. */ +export async function uploadIngest(name: string, file: File): Promise { + const body = new FormData(); + body.append("file", file, name); + const res = await fetch(`${API_ORIGIN}/ops/ingest/${encodeURIComponent(name)}`, { + method: "POST", + credentials: "include", + body, + }); + if (!res.ok) { + let message = `Error ${res.status}`; + try { + const b = await res.json(); + if (b?.message) message = b.message; + } catch { + /* ignore */ + } + throw new ApiError(res.status, message); + } +} + +export function deleteIngest(name: string): Promise { + return apiFetch(`/ops/ingest/${encodeURIComponent(name)}`, { method: "DELETE" }); +} + +export function listBackups(): Promise { + return apiFetch("/ops/backups"); +} + +export function backupDownloadUrl(name: string): string { + return `${API_ORIGIN}/ops/backups/${encodeURIComponent(name)}/download`; +} + +export function deleteBackup(name: string): Promise { + return apiFetch(`/ops/backups/${encodeURIComponent(name)}`, { method: "DELETE" }); +} + +export function listOpsJobs(): Promise { + return apiFetch("/ops/jobs"); +} + +export function getOpsJob(id: string): Promise { + return apiFetch(`/ops/jobs/${id}`); +} + +/** Start a mutating op. `file` is required for RESTORE. 409 if one is running. */ +export function startOpsJob(kind: OpsJobKind, file?: string): Promise { + return apiFetch("/ops/jobs", { + method: "POST", + body: JSON.stringify({ kind, file }), + }); +} diff --git a/apps/web/src/lib/labels.ts b/apps/web/src/lib/labels.ts index 3134373..4dd623b 100644 --- a/apps/web/src/lib/labels.ts +++ b/apps/web/src/lib/labels.ts @@ -266,6 +266,48 @@ export function bankSourceLabel(source: string | null | undefined): string { return BANK_SOURCE_LABELS[source] ?? source; } +// ----- DB operations (admin) ----- + +export const OPS_KIND_LABELS: Record = { + BACKUP: "Respaldo", + RESTORE: "Restauración", + REIMPORT: "Reimportación", + SYNC: "Sincronización", +}; + +export const OPS_STATUS_LABELS: Record = { + RUNNING: "En curso", + SUCCESS: "Completado", + FAILED: "Con error", +}; + +/** Bytes → human size (KB/MB/GB), es-MX formatting. */ +export function formatBytes(bytes: number | null | undefined): string { + if (bytes === null || bytes === undefined) return "—"; + if (bytes < 1024) return `${bytes} B`; + const units = ["KB", "MB", "GB"]; + let n = bytes / 1024; + let i = 0; + while (n >= 1024 && i < units.length - 1) { + n /= 1024; + i++; + } + return `${n.toLocaleString("es-MX", { maximumFractionDigits: 1 })} ${units[i]}`; +} + +export function formatDateTime(iso: string | null | undefined): string { + if (!iso) return "—"; + const d = new Date(iso); + if (Number.isNaN(d.getTime())) return "—"; + return d.toLocaleString("es-MX", { + day: "2-digit", + month: "2-digit", + year: "numeric", + hour: "2-digit", + minute: "2-digit", + }); +} + export const MONTH_NAMES = [ "Enero", "Febrero", diff --git a/apps/web/src/lib/types.ts b/apps/web/src/lib/types.ts index 2cb9a18..9f37040 100644 --- a/apps/web/src/lib/types.ts +++ b/apps/web/src/lib/types.ts @@ -20,7 +20,8 @@ export type Ability = | "bank:create" | "bank:void" | "lookup:manage" - | "user:manage"; + | "user:manage" + | "db:manage"; export interface AuthUser { id: string; @@ -43,6 +44,36 @@ export interface UserRow { updatedAt: string; } +/* --------------------------------------------------- DB operations (admin) */ + +export type OpsJobKind = "BACKUP" | "RESTORE" | "REIMPORT" | "SYNC"; +export type OpsJobStatus = "RUNNING" | "SUCCESS" | "FAILED"; + +export interface OpsJob { + id: string; + kind: OpsJobKind; + status: OpsJobStatus; + log: string; + params: Record | null; + createdById: string | null; + startedAt: string; + finishedAt: string | null; +} + +/** One of the four legacy Access files expected in the ingest folder. */ +export interface IngestFile { + name: string; + present: boolean; + size: number | null; + modifiedAt: string | null; +} + +export interface BackupFile { + name: string; + size: number; + createdAt: string; +} + export interface CustomerStats { customers: number; withUtilities: number; diff --git a/docker-compose.yml b/docker-compose.yml index 3a1ac3b..3a926cc 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -31,6 +31,12 @@ services: SESSION_SECRET: ${SESSION_SECRET:?SESSION_SECRET must be set} WEB_ORIGIN: http://localhost:3000 PORT: 3001 + INGEST_DIR: /data/ingest + BACKUP_DIR: /data/backups + MIGRATION_ENV: dev + volumes: + - ingest_data:/data/ingest + - backup_data:/data/backups ports: - "3001:3001" @@ -48,3 +54,5 @@ services: volumes: mysql_data: + ingest_data: + backup_data: diff --git a/migration/blob_extract.py b/migration/blob_extract.py index e2e099d..45dc1e7 100644 --- a/migration/blob_extract.py +++ b/migration/blob_extract.py @@ -38,7 +38,9 @@ from extract import sanitize_column_name as san csv.field_size_limit(300_000_000) -SOURCE_ROOT = Path.home() / "Downloads" / "JorgeCuadros-Legacy" +# Same source folder as the rest of the pipeline (config.SOURCE_ROOT honours +# INGEST_DIR — the web "Operaciones" ingest volume). +from config import SOURCE_ROOT # (key, access_file, access_table, staged_table_name, blob_cols, model) SOURCES = [ diff --git a/migration/config.py b/migration/config.py index 1bc99b3..d960459 100644 --- a/migration/config.py +++ b/migration/config.py @@ -17,9 +17,17 @@ not work on macOS; extraction is being reworked to use mdbtools the four Access source files. """ +import os from pathlib import Path -SOURCE_ROOT = Path.home() / "Downloads" / "JorgeCuadros-Legacy" +# The folder holding the four Access source files. Overridable via INGEST_DIR so +# the web "Operaciones" ingest folder (a mounted volume in the API container) +# feeds the same pipeline. Falls back to the original macOS download location +# for a plain local run. +SOURCE_ROOT = Path( + os.environ.get("INGEST_DIR") + or (Path.home() / "Downloads" / "JorgeCuadros-Legacy") +) SOURCES = { "utilities": { diff --git a/migration/dbenv.py b/migration/dbenv.py index 691c9b1..25d34ca 100644 --- a/migration/dbenv.py +++ b/migration/dbenv.py @@ -22,6 +22,7 @@ Usage in a script: from __future__ import annotations import argparse +import os from pathlib import Path from urllib.parse import unquote, urlparse @@ -48,8 +49,16 @@ def load_env(env: str) -> dict: return out +def database_url(env: str) -> str: + """Target DB URL. A DATABASE_URL in the process environment wins over + deploy/.env. — this is how the API container (which has its own + DATABASE_URL and no deploy/.env files) drives a re-import against its own + database.""" + return os.environ.get("DATABASE_URL") or load_env(env)["DATABASE_URL"] + + def connect(env: str): - url = load_env(env)["DATABASE_URL"] + url = database_url(env) u = urlparse(url) # mysql://user:pass@host:port/db return pymysql.connect( host=u.hostname, diff --git a/migration/run_all.py b/migration/run_all.py index 00c4987..8028b1d 100644 --- a/migration/run_all.py +++ b/migration/run_all.py @@ -39,13 +39,21 @@ PY = sys.executable # the venv python running this orchestrator # service_documents / policy_documents, which would otherwise leave the # uploaded MinIO objects with no rows pointing at them. STEPS = [ - "transform_customers.py", # customers + customer_legacy_refs (everything FKs to these) - "transform_properties.py", # properties + services + trust accounts - "transform_policies.py", # policies + installments/vehicles/drivers/benef/claims/adjusters - "transform_transactions.py", # shared ledger + type_transactions + exchange_rates - "prune_empty_customers.py", # drop customers with no property/policy/transaction - "transform_bank.py", # SCOTHIA bank register (no customer FK; independent) - "blob_extract.py", # document pointers; must follow properties + policies + "transform_customers.py", + "transform_properties.py", + "transform_policies.py", + "transform_transactions.py", + "prune_empty_customers.py", + "transform_bank.py", + "blob_extract.py", +] + +SYNC_STEPS = [ + "transform_customers.py", + "transform_properties.py", + "transform_policies.py", + "transform_transactions.py", + "transform_bank.py", ] @@ -61,13 +69,18 @@ def main() -> None: ap.add_argument("--env", default="dev", help="target environment (reads deploy/.env.)") ap.add_argument("--stage", action="store_true", help="re-run the raw staging load first (needs the Access files + mdbtools)") + ap.add_argument("--sync", action="store_true", + help="upsert legacy rows and archive removed legacy rows; preserve manual rows") args = ap.parse_args() if args.stage: run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")]) - for step in STEPS: - run([PY, str(HERE / step), "--env", args.env]) + for step in SYNC_STEPS if args.sync else STEPS: + cmd = [PY, str(HERE / step), "--env", args.env] + if args.sync: + cmd.append("--sync") + run(cmd) print(f"\n✓ migration complete for env={args.env}") diff --git a/migration/sync.py b/migration/sync.py new file mode 100644 index 0000000..73346c8 --- /dev/null +++ b/migration/sync.py @@ -0,0 +1,27 @@ +"""Shared CLI and SQL helpers for migration modes.""" + +from __future__ import annotations + +import argparse + + +def parse_mode() -> tuple[str, bool]: + parser = argparse.ArgumentParser() + parser.add_argument("--env", default="dev") + parser.add_argument("--sync", action="store_true") + args = parser.parse_args() + return args.env, args.sync + + +def existing_ids(cursor, table: str, key_columns: tuple[str, ...], where: str = "") -> dict[tuple, str]: + columns = ",".join(("id", *key_columns)) + cursor.execute(f"SELECT {columns} FROM {table} {where}") + return {tuple(row[1:]): row[0] for row in cursor.fetchall()} + + +def delete_missing(cursor, table: str, key_columns: tuple[str, ...], seen: set[tuple], where: str) -> int: + rows = existing_ids(cursor, table, key_columns, where) + stale = [row_id for key, row_id in rows.items() if key not in seen] + if stale: + cursor.executemany(f"DELETE FROM {table} WHERE id=%s", [(row_id,) for row_id in stale]) + return len(stale) diff --git a/migration/transform_bank.py b/migration/transform_bank.py index d872379..bfdf957 100644 --- a/migration/transform_bank.py +++ b/migration/transform_bank.py @@ -28,6 +28,7 @@ from pathlib import Path import pandas as pd from dbenv import connect, env_arg +from sync import parse_mode STG = Path(__file__).parent / "output" / "stg_scothia" NULL = "∅" @@ -72,7 +73,7 @@ def load(name): def main(): - env = env_arg() + env, sync_mode = parse_mode() conn = connect(env) print(f"[bank] target env: {env}") c = conn.cursor() @@ -109,16 +110,17 @@ def main(): for _, r in load("datos_e").iterrows(): add(r, -(dec(r["egreso"], Decimal(0))), income=False) - c.execute("SET FOREIGN_KEY_CHECKS=0") - for t in ("bank_transactions", "business_line_categories"): - c.execute(f"TRUNCATE TABLE {t}") - c.execute("SET FOREIGN_KEY_CHECKS=1") - - c.executemany("INSERT INTO business_line_categories (id,name) VALUES (%s,%s)", cats) - c.executemany( - "INSERT INTO bank_transactions (id,transactionDate,transactionType,reference,concept," - "amount,categoryId,cleared,transferred,notes,amountInWords,legacySourceTable,legacyId) " - "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", rows) + if sync_mode: + for row in rows: + c.execute("INSERT INTO bank_transactions (id,transactionDate,transactionType,reference,concept,amount,categoryId,cleared,transferred,notes,amountInWords,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE transactionDate=VALUES(transactionDate),transactionType=VALUES(transactionType),reference=VALUES(reference),concept=VALUES(concept),amount=VALUES(amount),cleared=VALUES(cleared),transferred=VALUES(transferred),notes=VALUES(notes),amountInWords=VALUES(amountInWords),voidedAt=NULL", row) + else: + c.execute("SET FOREIGN_KEY_CHECKS=0") + for t in ("bank_transactions", "business_line_categories"): + c.execute(f"TRUNCATE TABLE {t}") + c.execute("SET FOREIGN_KEY_CHECKS=1") + c.executemany("INSERT INTO business_line_categories (id,name) VALUES (%s,%s)", cats) + c.executemany( + "INSERT INTO bank_transactions (id,transactionDate,transactionType,reference,concept,amount,categoryId,cleared,transferred,notes,amountInWords,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", rows) conn.commit() def count(t): diff --git a/migration/transform_customers.py b/migration/transform_customers.py index 804c750..84a4506 100644 --- a/migration/transform_customers.py +++ b/migration/transform_customers.py @@ -39,6 +39,7 @@ from pathlib import Path import pandas as pd from dbenv import connect, env_arg +from sync import parse_mode STG = Path(__file__).parent / "output" NULL = "∅" @@ -229,16 +230,16 @@ _CUST_COLS = [ def main() -> None: - env = env_arg() + env, sync_mode = parse_mode() conn = connect(env) print(f"[customers] target env: {env}") cur = conn.cursor() - # Fresh, idempotent rebuild. - cur.execute("SET FOREIGN_KEY_CHECKS=0") - cur.execute("TRUNCATE TABLE customer_legacy_refs") - cur.execute("TRUNCATE TABLE customers") - cur.execute("SET FOREIGN_KEY_CHECKS=1") + if not sync_mode: + cur.execute("SET FOREIGN_KEY_CHECKS=0") + cur.execute("TRUNCATE TABLE customer_legacy_refs") + cur.execute("TRUNCATE TABLE customers") + cur.execute("SET FOREIGN_KEY_CHECKS=1") util = load("stg_utilities", "datgral") ins = load("stg_seguros", "datgral") @@ -291,18 +292,28 @@ def main() -> None: new_ins += 1 refs.append((str(uuid.uuid4()), cust_id, "insurance", "DATGRAL", ins_id)) - # Insert customers. placeholders = ",".join(["%s"] * len(_CUST_COLS)) - cur.executemany( - f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders})", - [tuple(rec[c] for c in _CUST_COLS) for rec in customers], - ) - # Insert legacy refs. - cur.executemany( - "INSERT INTO customer_legacy_refs (id, customerId, sourceSystem, sourceTable, legacyId) " - "VALUES (%s,%s,%s,%s,%s)", - refs, - ) + if sync_mode: + existing = {} + cur.execute("SELECT id,sourceSystem,sourceTable,legacyId,customerId FROM customer_legacy_refs") + for rid, system, table, legacy, customer_id in cur.fetchall(): + existing[(system, table, legacy)] = (rid, customer_id) + for rec, ref in zip(customers, refs): + key = (ref[2], ref[3], ref[4]) + customer_id = existing.get(key, (None, rec["id"]))[1] + rec["id"] = customer_id + cur.execute(f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE name=VALUES(name),nameSource=VALUES(nameSource),nameMissing=VALUES(nameMissing),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),city=VALUES(city),state=VALUES(state),zipCode=VALUES(zipCode),country=VALUES(country),phone=VALUES(phone),mobile=VALUES(mobile),fax=VALUES(fax),email=VALUES(email),notes=VALUES(notes),identificationType=VALUES(identificationType),identificationNumber=VALUES(identificationNumber),identificationExpiration=VALUES(identificationExpiration),customerSince=VALUES(customerSince),status=VALUES(status),feeAmount=VALUES(feeAmount),updatedAt=VALUES(updatedAt)", tuple(rec[c] for c in _CUST_COLS)) + ref = (existing.get(key, (ref[0], customer_id))[0], customer_id, *ref[2:]) + cur.execute("INSERT INTO customer_legacy_refs (id,customerId,sourceSystem,sourceTable,legacyId) VALUES (%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId)", ref) + else: + cur.executemany( + f"INSERT INTO customers ({','.join(f'`{c}`' for c in _CUST_COLS)}) VALUES ({placeholders})", + [tuple(rec[c] for c in _CUST_COLS) for rec in customers], + ) + cur.executemany( + "INSERT INTO customer_legacy_refs (id, customerId, sourceSystem, sourceTable, legacyId) VALUES (%s,%s,%s,%s,%s)", + refs, + ) # Enrich linked customers with insurance-only ID-doc fields, and fill any # contact fields the utilities master left empty (COALESCE keeps master's). diff --git a/migration/transform_policies.py b/migration/transform_policies.py index 8d904d2..8ed9f1b 100644 --- a/migration/transform_policies.py +++ b/migration/transform_policies.py @@ -37,6 +37,7 @@ from pathlib import Path import pandas as pd from dbenv import connect, env_arg +from sync import parse_mode STG = Path(__file__).parent / "output" / "stg_seguros" LEGACY_DB = "SEGUROS 16_be" @@ -171,7 +172,7 @@ def load(name): def main(): - env = env_arg() + env, sync_mode = parse_mode() conn = connect(env) print(f"[policies] target env: {env}") c = conn.cursor() @@ -315,35 +316,44 @@ def main(): dt(r["fecha_cheque"]), s(r["num_cheque"]), 1 if truthy(r["concluido"]) else 0, s(r["resolucion"]))) - # --- write (children first on truncate) --- - c.execute("SET FOREIGN_KEY_CHECKS=0") - for t in ("policy_payment_installments", "insured_drivers", "policy_beneficiaries", - "claims", "vehicles", "policy_documents", "policies", "policy_types", - "insurance_providers", "adjusters"): - c.execute(f"TRUNCATE TABLE {t}") - c.execute("SET FOREIGN_KEY_CHECKS=1") + if sync_mode: + c.execute("SELECT id,name FROM policy_types") + ptype_ids = dict(c.fetchall()) + for n in ptypes: + ptype_ids.setdefault(n, str(uuid.uuid4())) + c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s) ON DUPLICATE KEY UPDATE name=VALUES(name)", [(i, n) for n, i in ptype_ids.items()]) + c.execute("SELECT id,name FROM insurance_providers") + prov_ids = dict(c.fetchall()) + for n in providers: + prov_ids.setdefault(n, str(uuid.uuid4())) + c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s) ON DUPLICATE KEY UPDATE name=VALUES(name)", [(i, n) for n, i in prov_ids.items()]) + for p in policies: + p = list(p); p[3] = ptype_ids[p[3]]; p[4] = prov_ids.get(p[4]) + c.execute(f"INSERT INTO policies ({pol_cols}) VALUES ({','.join(['%s'] * 23)}) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId),insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate),policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee),commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations),coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber),liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL", tuple(p)) + else: + ptype_ids = {n: str(uuid.uuid4()) for n in ptypes} + c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s)", [(i, n) for n, i in ptype_ids.items()]) + prov_ids = {n: str(uuid.uuid4()) for n in providers} + c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s)", [(i, n) for n, i in prov_ids.items()]) + c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", adj_rows) - ptype_ids = {n: str(uuid.uuid4()) for n in ptypes} - c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s)", - [(i, n) for n, i in ptype_ids.items()]) - prov_ids = {n: str(uuid.uuid4()) for n in providers} - c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s)", - [(i, n) for n, i in prov_ids.items()]) - c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", adj_rows) - - # patch policyType/provider FKs into policy tuples pol_cols = ("id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate," "policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations," "coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb," "legacySourceTable,legacyId,updatedAt") - fixed = [] - for p in policies: - p = list(p) - p[3] = ptype_ids.get(p[3]) # ptype name -> policyTypeId - p[4] = prov_ids.get(p[4]) # provider name -> insuranceProviderId - fixed.append(tuple(p)) - ph = ",".join(["%s"] * 23) - c.executemany(f"INSERT INTO policies ({pol_cols}) VALUES ({ph})", fixed) + if not sync_mode: + fixed = [] + for p in policies: + p = list(p) + p[3] = ptype_ids.get(p[3]) + p[4] = prov_ids.get(p[4]) + fixed.append(tuple(p)) + ph = ",".join(["%s"] * 23) + c.executemany(f"INSERT INTO policies ({pol_cols}) VALUES ({ph})", fixed) + else: + c.executemany("INSERT INTO policies (id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate,policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations,coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb,legacySourceTable,legacyId,updatedAt) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId),insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate),policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee),commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations),coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber),liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL", [tuple([p[0],p[1],p[2],ptype_ids.get(p[3]),prov_ids.get(p[4]),*p[5:]]) for p in policies]) + + c.executemany("INSERT INTO policy_payment_installments " "(id,policyId,sequence,amount,currency,paidDate,checkNumber,isCash) " diff --git a/migration/transform_properties.py b/migration/transform_properties.py index 46be9da..0ed9d3e 100644 --- a/migration/transform_properties.py +++ b/migration/transform_properties.py @@ -34,6 +34,7 @@ from pathlib import Path import pandas as pd from dbenv import connect, env_arg +from sync import delete_missing, existing_ids, parse_mode STG = Path(__file__).parent / "output" / "stg_utilities" NULL = "∅" @@ -100,8 +101,8 @@ def jkey(numer, casa, direc): ]) -def main() -> None: - env = env_arg() +def main(): + env, sync_mode = parse_mode() conn = connect(env) print(f"[properties] target env: {env}") cur = conn.cursor() @@ -119,6 +120,7 @@ def main() -> None: flags[jkey(r["numerid"], r["casa"], r["direccion"])] = r props, services, trusts = [], [], [] + prop_keys: set[tuple] = set() skipped_no_customer = 0 matched_profile = 0 @@ -129,6 +131,8 @@ def main() -> None: skipped_no_customer += 1 continue + legacy_id = str(int(row["_row_num"])) + prop_keys.add(("DATMEX", legacy_id)) pid = str(uuid.uuid4()) addr2_parts = [] for lbl, col in (("CASA", "casa"), ("MZ", "manzana"), ("LOTE", "lote")): @@ -137,7 +141,8 @@ def main() -> None: props.append(( pid, cust_id, s(row["direccion"]), ", ".join(addr2_parts) or None, s(row["telefono"]), s(row["telefono2"]), s(row["telefono3"]), - s(row["zona"]), "DATMEX", nid, + s(row["zona"]), "DATMEX", legacy_id, + )) pfrow = flags.get(jkey(row["numer_id"], row["casa"], row["direccion"])) @@ -199,21 +204,28 @@ def main() -> None: trusts.append((str(uuid.uuid4()), pid, bank, s(row["trust_num"]), as_dec(row["bfee"]), as_date(row["vence1"]), as_date(row["vence2"]))) - # Fresh rebuild (children first). - cur.execute("SET FOREIGN_KEY_CHECKS=0") - for t in ("property_services", "service_documents", "trust_accounts", "properties"): - cur.execute(f"TRUNCATE TABLE {t}") - cur.execute("SET FOREIGN_KEY_CHECKS=1") - - cur.executemany( - "INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2," - "phone3,zone,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", props) - cur.executemany( - "INSERT INTO property_services (id,propertyId,kind,accountNumber,meterNumber,route," - "dueDay,active,notes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", services) - cur.executemany( - "INSERT INTO trust_accounts (id,propertyId,bankName,trustNumber,bankFee,dueDate1,dueDate2)" - " VALUES (%s,%s,%s,%s,%s,%s,%s)", trusts) + # Fresh rebuild (children first), or additive upsert for legacy-owned rows. + if sync_mode: + existing = existing_ids(cur, "properties", ("legacySourceTable", "legacyId"), "WHERE legacyId IS NOT NULL") + for row in props: + key = (row[8], row[9]) + if key in existing: + row = list(row); row[0] = existing[key] + cur.execute( + "INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2,phone3,zone,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),addressLine1=VALUES(addressLine1),addressLine2=VALUES(addressLine2),phone1=VALUES(phone1),phone2=VALUES(phone2),phone3=VALUES(phone3),zone=VALUES(zone),archivedAt=NULL", tuple(row)) + delete_missing(cur, "properties", ("legacySourceTable", "legacyId"), prop_keys, "WHERE legacyId IS NOT NULL") + cur.execute("DELETE ps FROM property_services ps JOIN properties p ON p.id=ps.propertyId WHERE p.legacyId IS NOT NULL") + cur.execute("DELETE ta FROM trust_accounts ta JOIN properties p ON p.id=ta.propertyId WHERE p.legacyId IS NOT NULL") + cur.executemany("INSERT INTO property_services (id,propertyId,kind,accountNumber,meterNumber,route,dueDay,active,notes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", services) + cur.executemany("INSERT INTO trust_accounts (id,propertyId,bankName,trustNumber,bankFee,dueDate1,dueDate2) VALUES (%s,%s,%s,%s,%s,%s,%s)", trusts) + else: + cur.execute("SET FOREIGN_KEY_CHECKS=0") + for t in ("property_services", "service_documents", "trust_accounts", "properties"): + cur.execute(f"TRUNCATE TABLE {t}") + cur.execute("SET FOREIGN_KEY_CHECKS=1") + cur.executemany("INSERT INTO properties (id,customerId,addressLine1,addressLine2,phone1,phone2,phone3,zone,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", props) + cur.executemany("INSERT INTO property_services (id,propertyId,kind,accountNumber,meterNumber,route,dueDay,active,notes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)", services) + cur.executemany("INSERT INTO trust_accounts (id,propertyId,bankName,trustNumber,bankFee,dueDate1,dueDate2) VALUES (%s,%s,%s,%s,%s,%s,%s)", trusts) conn.commit() cur.execute("SELECT COUNT(*) FROM properties"); n_p = cur.fetchone()[0] diff --git a/migration/transform_transactions.py b/migration/transform_transactions.py index e7b4915..5ef2648 100644 --- a/migration/transform_transactions.py +++ b/migration/transform_transactions.py @@ -35,6 +35,7 @@ from pathlib import Path import pandas as pd from dbenv import connect, env_arg +from sync import parse_mode STG = Path(__file__).parent / "output" NULL = "∅" @@ -90,7 +91,7 @@ def load(src, name): def main(): - env = env_arg() + env, sync_mode = parse_mode() conn = connect(env) print(f"[transactions] target env: {env}") c = conn.cursor() @@ -225,18 +226,17 @@ def main(): iva() efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be", "EFECTIVO") - # --- write --- - c.execute("SET FOREIGN_KEY_CHECKS=0") - for t in ("transactions", "type_transactions", "exchange_rates"): - c.execute(f"TRUNCATE TABLE {t}") - c.execute("SET FOREIGN_KEY_CHECKS=1") - - c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", type_rows) - c.executemany("INSERT INTO exchange_rates (id,rate,effectiveDate,effectiveHour) VALUES (%s,%s,%s,%s)", xr_rows) - c.executemany( - "INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference," - "amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb," - "legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", tx) + if sync_mode: + c.executemany("INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),updatedAt=NOW(),voidedAt=NULL", tx) + else: + c.execute("SET FOREIGN_KEY_CHECKS=0") + for t in ("transactions", "type_transactions", "exchange_rates"): + c.execute(f"TRUNCATE TABLE {t}") + c.execute("SET FOREIGN_KEY_CHECKS=1") + c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", type_rows) + c.executemany("INSERT INTO exchange_rates (id,rate,effectiveDate,effectiveHour) VALUES (%s,%s,%s,%s)", xr_rows) + c.executemany( + "INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference,amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb,legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", tx) conn.commit() def count(t): diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 7dc6f6e..8cdaa68 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -224,6 +224,7 @@ model Vehicle { legacySourceTable String? legacyId String? + @@unique([legacySourceTable, legacyId]) @@map("vehicles") } @@ -331,6 +332,7 @@ model Property { documents ServiceDocument[] trustAccount TrustAccount? + @@unique([legacySourceTable, legacyId]) @@map("properties") } @@ -420,6 +422,7 @@ model Transaction { createdAt DateTime @default(now()) @@index([customerId, transactionDate]) + @@unique([legacySourceDb, legacySourceTable, legacyId]) @@map("transactions") } @@ -468,6 +471,7 @@ model BankTransaction { legacySourceTable String? legacyId String? + @@unique([legacySourceTable, legacyId]) @@map("bank_transactions") } @@ -533,3 +537,40 @@ model EmailLog { @@map("email_log") } + +// --------------------------------------------------------------------------- +// Admin database operations (Operaciones): backup / restore / re-import / sync. +// Each long-running op is one OpsJob row so the web UI can poll status + tail +// the captured log. Rows are the audit trail for who ran a destructive op. +// --------------------------------------------------------------------------- + +enum OpsJobKind { + BACKUP + RESTORE + REIMPORT + SYNC +} + +enum OpsJobStatus { + RUNNING + SUCCESS + FAILED +} + +model OpsJob { + id String @id @default(uuid()) + kind OpsJobKind + status OpsJobStatus @default(RUNNING) + // Combined stdout+stderr of the spawned process, appended as it runs. + log String @db.LongText + // Op-specific inputs (e.g. the backup filename a RESTORE targets). No FK on + // createdById — the actor id is stored flat, like activity_logs' userId use. + params Json? + createdById String? + startedAt DateTime @default(now()) + finishedAt DateTime? + + @@index([status]) + @@index([startedAt]) + @@map("ops_jobs") +}