From 6c4884c5406c15e1bef9a94cad0e0dca1ecaab93 Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Thu, 25 Jun 2026 20:06:08 +0530 Subject: [PATCH 1/5] feat(cli): db push --verify/--sr-tables + db backups list/prune (#16) Closes the round-3 large-DB grab-bag: - db push --verify: poll the site URL after import until it answers HTTP (large imports can briefly return 000 right after import/flush); reported as `verified` in the summary - db push --sr-tables : scope --search-replace to specific tables instead of --all-tables (faster on big DBs whose bulk has no URLs) - db backups list / db backups prune --keep/--older-than/--force: manage the ~/db-backup-*.sql.gz files db push leaves behind Extracted waitForHttp to lib/http-ready.ts (shared by sites create + db push). New pure libs (http-ready, db-backups) with unit tests; full suite 341 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 9 +- src/__tests__/db-backups.test.ts | 69 ++++++++++++++ src/__tests__/http-ready.test.ts | 22 +++++ src/commands/db.ts | 154 ++++++++++++++++++++++++++++++- src/commands/sites.ts | 25 +---- src/lib/db-backups.ts | 53 +++++++++++ src/lib/http-ready.ts | 26 ++++++ 7 files changed, 330 insertions(+), 28 deletions(-) create mode 100644 src/__tests__/db-backups.test.ts create mode 100644 src/__tests__/http-ready.test.ts create mode 100644 src/lib/db-backups.ts create mode 100644 src/lib/http-ready.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 54785cb..157a9c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,13 @@ # Changelog -## 0.0.1-beta.25 (2026-06-25) +## Unreleased + +### Added — db push readiness/scoping + backup retention (large-DB feedback #16) + +- **`db push --verify`** polls the site URL until it answers HTTP after the import (a large import can briefly return `000` right after import/flush); reported as `verified` in the summary. +- **`db push --sr-tables `** scopes `--search-replace` to specific (prefixed) tables instead of all tables — faster on big DBs whose bulk rows have no URLs. Default stays `--all-tables`. +- **`db backups list `** and **`db backups prune [--keep ] [--older-than ] [--force]`** manage the `~/db-backup-*.sql.gz` files `db push` leaves behind (they previously accumulated with no way to view/clean them). +- Internal: extracted the HTTP-readiness helper to `lib/http-ready.ts` (now shared by `sites create` and `db push --verify`). ### Improved — db push transport & progress (large-DB path) diff --git a/src/__tests__/db-backups.test.ts b/src/__tests__/db-backups.test.ts new file mode 100644 index 0000000..ad07616 --- /dev/null +++ b/src/__tests__/db-backups.test.ts @@ -0,0 +1,69 @@ +import { describe, it, expect } from 'vitest'; +import { parseBackupList, selectBackupsToPrune } from '../lib/db-backups.js'; + +describe('parseBackupList', () => { + it('parses tab-separated stat output, newest first', () => { + const out = [ + '/home/u/db-backup-2026-06-20.sql.gz\t1048576\t1718870400', + '/home/u/db-backup-2026-06-25.sql.gz\t2097152\t1719273600', + ].join('\n'); + const list = parseBackupList(out); + expect(list).toHaveLength(2); + expect(list[0].file).toBe('/home/u/db-backup-2026-06-25.sql.gz'); // newest first + expect(list[0].sizeBytes).toBe(2097152); + expect(list[1].mtime).toBe(1718870400); + }); + + it('tolerates an SSH banner/MOTD and skips non-matching lines', () => { + const out = [ + 'Welcome to Ubuntu 22.04', + 'Last login: ...', + '/home/u/db-backup-a.sql.gz\t100\t1719273600', + 'some junk line without tabs', + '/home/u/notanumber.sql.gz\tNaN\tNaN', + ].join('\n'); + const list = parseBackupList(out); + expect(list).toHaveLength(1); + expect(list[0].file).toBe('/home/u/db-backup-a.sql.gz'); + }); + + it('returns [] for empty input', () => { + expect(parseBackupList('')).toEqual([]); + }); +}); + +describe('selectBackupsToPrune', () => { + const now = 1_719_273_600; // fixed "now" + const day = 86400; + // newest → oldest + const backups = [ + { file: 'd', sizeBytes: 1, mtime: now - 1 * day }, + { file: 'c', sizeBytes: 1, mtime: now - 3 * day }, + { file: 'b', sizeBytes: 1, mtime: now - 10 * day }, + { file: 'a', sizeBytes: 1, mtime: now - 30 * day }, + ]; + + it('--keep N keeps the newest N, deletes the rest', () => { + const { toDelete, toKeep } = selectBackupsToPrune(backups, { keep: 2 }, now); + expect(toKeep.map((b) => b.file)).toEqual(['d', 'c']); + expect(toDelete.map((b) => b.file)).toEqual(['b', 'a']); + }); + + it('--older-than D deletes backups older than D days', () => { + const { toDelete, toKeep } = selectBackupsToPrune(backups, { olderThanDays: 5 }, now); + expect(toDelete.map((b) => b.file)).toEqual(['b', 'a']); // 10d, 30d old + expect(toKeep.map((b) => b.file)).toEqual(['d', 'c']); // 1d, 3d + }); + + it('combines criteria with OR (deleted if beyond keep OR too old)', () => { + const { toDelete } = selectBackupsToPrune(backups, { keep: 3, olderThanDays: 5 }, now); + // beyond keep:3 → 'a'; older than 5d → 'b','a' → union {b,a} + expect(toDelete.map((b) => b.file).sort()).toEqual(['a', 'b']); + }); + + it('deletes nothing when no criteria are given', () => { + const { toDelete, toKeep } = selectBackupsToPrune(backups, {}, now); + expect(toDelete).toHaveLength(0); + expect(toKeep).toHaveLength(4); + }); +}); diff --git a/src/__tests__/http-ready.test.ts b/src/__tests__/http-ready.test.ts new file mode 100644 index 0000000..1e22601 --- /dev/null +++ b/src/__tests__/http-ready.test.ts @@ -0,0 +1,22 @@ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { waitForHttp } from '../lib/http-ready.js'; + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe('waitForHttp', () => { + it('returns true as soon as the URL answers (any HTTP response)', async () => { + const fetchMock = vi.fn().mockResolvedValue({ status: 200 }); + vi.stubGlobal('fetch', fetchMock); + await expect(waitForHttp('https://x.test', 5000)).resolves.toBe(true); + expect(fetchMock).toHaveBeenCalledOnce(); + }); + + it('returns false without calling fetch when the budget is already exhausted', async () => { + const fetchMock = vi.fn(); + vi.stubGlobal('fetch', fetchMock); + await expect(waitForHttp('https://x.test', 0)).resolves.toBe(false); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/commands/db.ts b/src/commands/db.ts index 5c794ae..32dd808 100644 --- a/src/commands/db.ts +++ b/src/commands/db.ts @@ -11,7 +11,10 @@ import { ensureSshAccess } from '../lib/ssh-keys.js'; import { execViaSsh, execViaSshToFile, scpUpload } from '../lib/ssh-connection.js'; import { parseTablePrefix } from '../lib/local-instance.js'; import { detectDumpPrefix, readSqlHead, rewriteDumpPrefix } from '../lib/sql-dump.js'; -import { success, error, spinner, info, isJsonMode } from '../lib/output.js'; +import { waitForHttp } from '../lib/http-ready.js'; +import { parseBackupList, selectBackupsToPrune, type RemoteBackup } from '../lib/db-backups.js'; +import { success, error, spinner, info, table, isJsonMode } from '../lib/output.js'; +import type { SshConnection } from '../types.js'; /** Timestamp like `2026-05-23T12-34-56` (filename-safe — `:` is illegal on Windows). */ function isoTimestamp(): string { @@ -50,6 +53,31 @@ function isShellSafeUrl(u: string): boolean { return /^https?:\/\/[^\s'"\\$`]+$/.test(u); } +/** Resolve a site and open an SSH connection (shared by the backups subcommands). */ +async function resolveAndConnect(siteIdentifier: string): Promise<{ site: any; conn: SshConnection }> { + const rspin = spinner('Resolving site...'); + rspin.start(); + let site: any; + try { + site = await resolveSite(siteIdentifier); + rspin.succeed(`Site: ${site.name || site.sub_domain} (ID: ${site.id})`); + } catch { + rspin.fail('Site resolution failed'); + process.exit(1); + } + const conn = await ensureSshAccess(site.id); + return { site, conn }; +} + +/** List `~/db-backup-*.sql.gz` on the remote (the files `db push` writes), newest first. */ +function listRemoteBackups(conn: SshConnection): RemoteBackup[] { + // GNU stat interprets the \t escapes; the for-loop guard avoids a literal glob + // when nothing matches. parseBackupList tolerates any MOTD/banner lines. + const cmd = `for f in ~/db-backup-*.sql.gz; do [ -e "$f" ] && stat -c '%n\\t%s\\t%Y' "$f"; done`; + const res = execViaSsh(conn, cmd); + return parseBackupList(res.stdout || ''); +} + export function registerDbCommand(program: Command): void { const db = program .command('db') @@ -136,6 +164,8 @@ export function registerDbCommand(program: Command): void { .option('--no-backup', 'Skip taking a remote backup before overwrite (DANGEROUS)') .option('--rewrite-prefix', "Rewrite the dump's table prefix to match the remote site's prefix (e.g. wp_ → iwpa4c7_)") .option('--search-replace ', 'After import, run wp search-replace across all tables (serialization-safe, skips guid). Pass exactly two URLs.') + .option('--sr-tables ', 'Scope --search-replace to these (prefixed) tables instead of all tables — faster on big DBs whose bulk has no URLs') + .option('--verify', 'After import, poll the site URL until it responds (large imports can briefly return 000)') .addHelpText('after', ` Notes: - Always takes a remote backup first unless --no-backup is passed. @@ -146,10 +176,13 @@ Notes: - URLs: after a cross-domain push, remap URLs with: --search-replace (or run 'instawp wp search-replace ' yourself). + --search-replace scans all tables by default; pass --sr-tables to limit it. + - --verify confirms the site answers HTTP after the import (reported in the summary). Examples: - $ instawp db push my-site dump.sql --rewrite-prefix + $ instawp db push my-site dump.sql --rewrite-prefix --verify $ instawp db push my-site dump.sql --search-replace http://localhost:10115 https://my-site.instawp.site + $ instawp db push my-site dump.sql --search-replace http://old https://new --sr-tables iwpa4c7_options iwpa4c7_posts iwpa4c7_postmeta `) .action(async (siteIdentifier: string, file: string, opts: any) => { requireAuth(); @@ -181,6 +214,20 @@ Examples: } } + // Parse/validate --sr-tables (table names go into a shell command — keep identifier-safe) + const srTables: string[] = Array.isArray(opts.srTables) ? opts.srTables : []; + if (srTables.length) { + const bad = srTables.filter((t) => !/^[A-Za-z0-9_]+$/.test(t)); + if (bad.length) { + error(`--sr-tables: invalid table name(s): ${bad.join(', ')} (use plain prefixed table names)`); + process.exit(1); + } + if (!srFrom) { + error('--sr-tables only applies with --search-replace'); + process.exit(1); + } + } + // In JSON mode, can't prompt — require --force if (isJsonMode() && !opts.force) { error('--force is required when using --json (cannot prompt for confirmation)'); @@ -399,9 +446,11 @@ Examples: // Skip `guid` — post GUIDs are permanent identifiers, not links, and WP // best practice is to never rewrite them on a domain change. if (srFrom && srTo) { + // Scope: explicit (prefixed) tables if --sr-tables given, else all tables. + const scope = srTables.length ? srTables.join(' ') : '--all-tables'; const srSpin = spinner(`Rewriting URLs (${srFrom} -> ${srTo})...`); srSpin.start(); - const srRes = execViaSsh(conn, `cd ${wpPath} && wp search-replace '${srFrom}' '${srTo}' --all-tables --skip-columns=guid --report-changed-only`); + const srRes = execViaSsh(conn, `cd ${wpPath} && wp search-replace '${srFrom}' '${srTo}' ${scope} --skip-columns=guid --report-changed-only`); if (srRes.exitCode === 0) { srSpin.succeed('URLs rewritten'); if (!isJsonMode() && srRes.stdout.trim()) console.log(srRes.stdout.trim()); @@ -421,6 +470,18 @@ Examples: cleanupSpin.succeed('Cleanup complete'); } + // Step 6: Optional readiness check. A large import can briefly leave the + // site returning 000 right after import/flush; poll until it answers. + let verified: string | null = null; + if (opts.verify) { + const siteUrl = String(site.url || `https://${conn.domain}`).replace(/\/+$/, ''); + const vSpin = spinner(`Verifying ${siteUrl} responds...`); + vSpin.start(); + const ok = await waitForHttp(siteUrl, 90000); + if (ok) { vSpin.succeed('Site is responding'); verified = 'ok'; } + else { vSpin.fail('Site did not respond within 90s (large imports can need a moment to warm up)'); verified = 'timeout'; } + } + const elapsedSec = (Date.now() - startedAt) / 1000; const rate = elapsedSec > 0 ? `${formatBytes(localSize / elapsedSec)}/s` : 'n/a'; const elapsedStr = `${elapsedSec < 10 ? elapsedSec.toFixed(1) : Math.round(elapsedSec)}s (${rate})`; @@ -433,10 +494,97 @@ Examples: rewrote_prefix: remapFromPrefix ? `${remapFromPrefix} -> ${remotePrefix}` : null, search_replaced: srFrom && srTo ? `${srFrom} -> ${srTo}` : null, elapsed: elapsedStr, + ...(verified ? { verified } : {}), }); if (!isJsonMode() && takeBackup) { console.log(`\n ${chalk.dim('Backup:')} ~/${backupFilename} ${chalk.dim('(on remote)')}`); } }); + + // db backups list/prune — manage the ~/db-backup-*.sql.gz files db push leaves behind + const backups = db + .command('backups') + .description('List or prune db push backups (~/db-backup-*.sql.gz) on a site'); + + backups + .command('list ') + .description('List the db-backup-*.sql.gz files on the remote site (newest first)') + .action(async (siteIdentifier: string) => { + requireAuth(); + const { conn } = await resolveAndConnect(siteIdentifier); + const list = listRemoteBackups(conn); + + if (isJsonMode()) { + console.log(JSON.stringify({ + success: true, + data: { backups: list.map((b) => ({ file: b.file, size_bytes: b.sizeBytes, modified: new Date(b.mtime * 1000).toISOString() })) }, + })); + return; + } + if (!list.length) { info('No backups found (~/db-backup-*.sql.gz).'); return; } + table(['File', 'Size', 'Modified'], list.map((b) => ({ + file: b.file.replace(/^.*\//, ''), + size: formatBytes(b.sizeBytes), + modified: new Date(b.mtime * 1000).toISOString().replace('T', ' ').slice(0, 16), + }))); + const total = list.reduce((n, b) => n + b.sizeBytes, 0); + info(`${list.length} backup(s), ${formatBytes(total)} total`); + }); + + backups + .command('prune ') + .description('Delete old db-backup-*.sql.gz files (keep newest N and/or drop older than D days)') + .option('--keep ', 'Keep the newest N backups, delete the rest', (v) => parseInt(v, 10)) + .option('--older-than ', 'Delete backups older than this many days', (v) => parseInt(v, 10)) + .option('--force', 'Skip confirmation prompt') + .action(async (siteIdentifier: string, opts: any) => { + requireAuth(); + if (opts.keep === undefined && opts.olderThan === undefined) { + error('Specify --keep and/or --older-than (refusing to prune without a selector)'); + process.exit(1); + } + if ((opts.keep !== undefined && (!Number.isInteger(opts.keep) || opts.keep < 0)) || + (opts.olderThan !== undefined && (!Number.isInteger(opts.olderThan) || opts.olderThan < 0))) { + error('--keep / --older-than must be non-negative integers'); + process.exit(1); + } + if (isJsonMode() && !opts.force) { + error('--force is required when using --json (cannot prompt for confirmation)'); + process.exit(1); + } + + const { conn } = await resolveAndConnect(siteIdentifier); + const list = listRemoteBackups(conn); + const { toDelete, toKeep } = selectBackupsToPrune( + list, + { keep: opts.keep, olderThanDays: opts.olderThan }, + Math.floor(Date.now() / 1000), + ); + + if (!toDelete.length) { + if (isJsonMode()) { console.log(JSON.stringify({ success: true, data: { deleted: [], kept: toKeep.length } })); } + else { info(`Nothing to prune (${toKeep.length} backup(s) kept).`); } + return; + } + + if (!isJsonMode()) { + const freed = formatBytes(toDelete.reduce((n, b) => n + b.sizeBytes, 0)); + console.log(`Will delete ${chalk.bold.red(String(toDelete.length))} backup(s) (${freed}), keeping ${toKeep.length}:`); + for (const b of toDelete) console.log(` ${chalk.dim('-')} ${b.file.replace(/^.*\//, '')}`); + if (!opts.force) { + const ok = await promptYesNo('Delete these backups? (y/N) '); + if (!ok) { info('Cancelled.'); return; } + } + } + + const quoted = toDelete.map((b) => `'${b.file.replace(/'/g, "'\\''")}'`).join(' '); + const rm = execViaSsh(conn, `rm -f ${quoted}`); + if (rm.exitCode !== 0) { + error('Failed to delete some backups'); + if (rm.stderr) error(rm.stderr.trim()); + process.exit(1); + } + success('Backups pruned', { deleted: toDelete.length, kept: toKeep.length }); + }); } diff --git a/src/commands/sites.ts b/src/commands/sites.ts index 139fd7e..d5428ae 100644 --- a/src/commands/sites.ts +++ b/src/commands/sites.ts @@ -3,6 +3,7 @@ import chalk from 'chalk'; import { requireAuth, getClient } from '../lib/api.js'; import { getApiUrl } from '../lib/config.js'; import { resolveSite } from '../lib/site-resolver.js'; +import { waitForHttp } from '../lib/http-ready.js'; import { success, error, table, spinner, info, isJsonMode } from '../lib/output.js'; export function registerSitesCommand(program: Command): void { @@ -390,30 +391,6 @@ export function registerSitesCommand(program: Command): void { }); } -/** - * Poll a URL until it answers (any HTTP status = DNS resolved + server up), or - * the budget runs out. A fresh site's DNS/edge lags task completion by 30–120s, - * so without this "Ready" lies and callers hand-roll curl-retry gates. - */ -async function waitForHttp(url: string, maxMs: number): Promise { - const deadline = Date.now() + Math.min(Math.max(maxMs, 0), 180000); - while (Date.now() < deadline) { - try { - const ctrl = new AbortController(); - const timer = setTimeout(() => ctrl.abort(), 8000); - try { - await fetch(url, { signal: ctrl.signal, redirect: 'follow' }); - return true; // any HTTP response means it's reachable - } finally { - clearTimeout(timer); - } - } catch { - // DNS/connection not ready yet — back off and retry. - } - await new Promise((r) => setTimeout(r, 3000)); - } - return false; -} // Shared create action used by both `sites create` and top-level `create` async function createSiteAction(opts: any): Promise { diff --git a/src/lib/db-backups.ts b/src/lib/db-backups.ts new file mode 100644 index 0000000..ad03ea4 --- /dev/null +++ b/src/lib/db-backups.ts @@ -0,0 +1,53 @@ +// Helpers for `db backups list/prune` — parsing the remote backup listing and +// deciding which `~/db-backup-*.sql.gz` files to prune. Kept pure (no I/O) so +// the listing command and the prune selection are unit-testable. + +export interface RemoteBackup { + file: string; + sizeBytes: number; + mtime: number; // epoch seconds +} + +/** + * Parse the output of `stat -c '%n\t%s\t%Y' ` — tab-separated + * path / size-bytes / mtime-epoch, one per line. Tolerant of an SSH login + * banner/MOTD and any non-matching lines (silently skipped). Returns + * newest-first. + */ +export function parseBackupList(stdout: string): RemoteBackup[] { + const out: RemoteBackup[] = []; + for (const line of stdout.split('\n')) { + const parts = line.split('\t'); + if (parts.length !== 3) continue; + const file = parts[0].trim(); + const sizeBytes = Number(parts[1]); + const mtime = Number(parts[2]); + if (!file || !Number.isFinite(sizeBytes) || !Number.isFinite(mtime)) continue; + out.push({ file, sizeBytes, mtime }); + } + return out.sort((a, b) => b.mtime - a.mtime); +} + +/** + * Decide which backups to delete. A backup is selected if it matches ANY given + * criterion: it falls beyond the newest `keep`, OR it is older than + * `olderThanDays`. The caller must supply at least one criterion (selecting all + * is never implicit). Returns newest-first partitions. + */ +export function selectBackupsToPrune( + backups: RemoteBackup[], + opts: { keep?: number; olderThanDays?: number }, + nowEpoch: number, +): { toDelete: RemoteBackup[]; toKeep: RemoteBackup[] } { + const sorted = [...backups].sort((a, b) => b.mtime - a.mtime); + const cutoff = opts.olderThanDays != null ? nowEpoch - opts.olderThanDays * 86400 : null; + const toDelete: RemoteBackup[] = []; + const toKeep: RemoteBackup[] = []; + sorted.forEach((b, i) => { + const beyondKeep = opts.keep != null && i >= opts.keep; + const tooOld = cutoff != null && b.mtime < cutoff; + if (beyondKeep || tooOld) toDelete.push(b); + else toKeep.push(b); + }); + return { toDelete, toKeep }; +} diff --git a/src/lib/http-ready.ts b/src/lib/http-ready.ts new file mode 100644 index 0000000..2441ecb --- /dev/null +++ b/src/lib/http-ready.ts @@ -0,0 +1,26 @@ +/** + * Poll a URL until it answers (any HTTP status = DNS resolved + server up), or + * the budget runs out. A fresh site's DNS/edge lags task completion by 30–120s, + * and a large DB import can leave the site briefly unreachable (returns 000) + * right after import/flush — so callers use this instead of hand-rolling a + * curl-retry gate. Capped at 180s regardless of `maxMs`. + */ +export async function waitForHttp(url: string, maxMs: number): Promise { + const deadline = Date.now() + Math.min(Math.max(maxMs, 0), 180000); + while (Date.now() < deadline) { + try { + const ctrl = new AbortController(); + const timer = setTimeout(() => ctrl.abort(), 8000); + try { + await fetch(url, { signal: ctrl.signal, redirect: 'follow' }); + return true; // any HTTP response means it's reachable + } finally { + clearTimeout(timer); + } + } catch { + // DNS/connection not ready yet — back off and retry. + } + await new Promise((r) => setTimeout(r, 3000)); + } + return false; +} From e17f55a3249e904307be43f8e54c87ff57f406b1 Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Thu, 25 Jun 2026 20:20:53 +0530 Subject: [PATCH 2/5] =?UTF-8?q?feat(cli):=20incremental=20db=20push=20(bas?= =?UTF-8?q?eline=20+=20row-delta)=20=E2=80=94=20#17?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opt-in `db push --incremental` ships only what changed since the last push: diff the dump against a per-site baseline (~/.instawp/baselines//) and apply a minimal REPLACE/DELETE set (no DROP/CREATE). First run / schema change / --full do a normal full push and refresh the baseline. - Pure, unit-tested engine: lib/db-delta.ts (tuple/INSERT/CREATE parsing, schema fingerprint, PK-keyed diff, prefix remap) + lib/db-baseline.ts (per-site store). - Reuses the existing safety machinery (remote backup, role/cap remap, scoped search-replace, --verify). Requires a per-row dump (mysqldump --skip-extended-insert --order-by-primary; extended dumps rejected). - MVP: single-PK tables delta; no-PK tables or any DDL change auto-fall-back to a full push. The full `db push` path is byte-for-byte unchanged — the incremental branch is skipped entirely unless --incremental/--full is passed. Full suite 382 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 5 + src/__tests__/db-baseline.test.ts | 34 +++++ src/__tests__/db-delta.test.ts | 129 +++++++++++++++++++ src/commands/db.ts | 196 +++++++++++++++++++++++++++- src/lib/db-baseline.ts | 42 ++++++ src/lib/db-delta.ts | 205 ++++++++++++++++++++++++++++++ 6 files changed, 610 insertions(+), 1 deletion(-) create mode 100644 src/__tests__/db-baseline.test.ts create mode 100644 src/__tests__/db-delta.test.ts create mode 100644 src/lib/db-baseline.ts create mode 100644 src/lib/db-delta.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 157a9c1..934948b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +### Added — incremental `db push` (#17) + +- **`db push --incremental`** ships only the **row-level delta** since the last push instead of the whole DB: it diffs the dump against a per-site baseline (`~/.instawp/baselines//`) and applies a minimal `REPLACE`/`DELETE` set (no `DROP`/`CREATE`). First run, a schema/DDL change, or **`--full`** do a normal full push and refresh the baseline. Requires a per-row dump (`mysqldump --skip-extended-insert --order-by-primary`; extended-insert dumps are rejected). Reuses the existing safety machinery — remote backup first, prefix/role-key remap, scoped URL search-replace, `--verify`. **The full `db push` path is unchanged; incremental is purely additive.** +- MVP scope: tables with a single-column primary key (WP core + most plugin tables); anything without one — or any DDL change — auto-falls-back to a full push. The diff engine (`lib/db-delta.ts`) and baseline store (`lib/db-baseline.ts`) are pure and unit-tested. + ### Added — db push readiness/scoping + backup retention (large-DB feedback #16) - **`db push --verify`** polls the site URL until it answers HTTP after the import (a large import can briefly return `000` right after import/flush); reported as `verified` in the summary. diff --git a/src/__tests__/db-baseline.test.ts b/src/__tests__/db-baseline.test.ts new file mode 100644 index 0000000..050c0a6 --- /dev/null +++ b/src/__tests__/db-baseline.test.ts @@ -0,0 +1,34 @@ +import { describe, it, expect, afterEach } from 'vitest'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { baselineDir, loadBaseline, saveBaseline } from '../lib/db-baseline.js'; + +const dirs: string[] = []; +function tmpBase(): string { + const d = mkdtempSync(join(tmpdir(), 'iwp-baseline-')); + dirs.push(d); + return d; +} +afterEach(() => { for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }); }); + +describe('db-baseline store', () => { + it('round-trips sql + fingerprint per site', () => { + const base = tmpBase(); + expect(loadBaseline(42, base)).toBeNull(); + saveBaseline(42, 'CREATE TABLE `wp_options` ...', 'fp-abc', '2026-06-25T00:00:00Z', base); + const got = loadBaseline(42, base); + expect(got?.sql).toContain('wp_options'); + expect(got?.fingerprint).toBe('fp-abc'); + expect(got?.savedAt).toBe('2026-06-25T00:00:00Z'); + }); + + it('scopes baselines per site id', () => { + const base = tmpBase(); + saveBaseline(1, 'a', 'fp1', 't', base); + saveBaseline(2, 'b', 'fp2', 't', base); + expect(loadBaseline(1, base)?.fingerprint).toBe('fp1'); + expect(loadBaseline(2, base)?.fingerprint).toBe('fp2'); + expect(baselineDir(1, base)).not.toBe(baselineDir(2, base)); + }); +}); diff --git a/src/__tests__/db-delta.test.ts b/src/__tests__/db-delta.test.ts new file mode 100644 index 0000000..6af0dc3 --- /dev/null +++ b/src/__tests__/db-delta.test.ts @@ -0,0 +1,129 @@ +import { describe, it, expect } from 'vitest'; +import { + splitSqlTuple, + parseInsertStatement, + hasExtendedInsert, + parseCreateTables, + schemaFingerprint, + computeDelta, +} from '../lib/db-delta.js'; + +const OPTIONS_DDL = (autoInc = 120) => `CREATE TABLE \`wp_options\` ( + \`option_id\` bigint(20) unsigned NOT NULL AUTO_INCREMENT, + \`option_name\` varchar(191) NOT NULL DEFAULT '', + \`option_value\` longtext NOT NULL, + PRIMARY KEY (\`option_id\`), + UNIQUE KEY \`option_name\` (\`option_name\`) +) ENGINE=InnoDB AUTO_INCREMENT=${autoInc} DEFAULT CHARSET=utf8mb4;`; + +const dump = (ddl: string, rows: string[]) => [ddl, ...rows].join('\n') + '\n'; + +describe('splitSqlTuple', () => { + it('splits top-level fields', () => { + expect(splitSqlTuple("1,'siteurl','http://x'")).toEqual(['1', "'siteurl'", "'http://x'"]); + }); + it('respects commas and parens inside strings', () => { + expect(splitSqlTuple("5,'a, b (c)',7")).toEqual(['5', "'a, b (c)'", '7']); + }); + it('respects escaped quotes', () => { + expect(splitSqlTuple("1,'it\\'s ok',2")).toEqual(['1', "'it\\'s ok'", '2']); + }); +}); + +describe('parseInsertStatement / hasExtendedInsert', () => { + it('parses a single-row insert', () => { + const r = parseInsertStatement("INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');"); + expect(r?.table).toBe('wp_options'); + expect(r?.tuples).toEqual(["1,'siteurl','http://x'"]); + }); + it('detects an extended (multi-row) insert', () => { + expect(hasExtendedInsert("INSERT INTO `wp_options` VALUES (1,'a'),(2,'b');")).toBe(true); + }); + it('does not false-positive on a single row containing ),(', () => { + expect(hasExtendedInsert("INSERT INTO `wp_options` VALUES (1,'a),(b');")).toBe(false); + }); +}); + +describe('parseCreateTables', () => { + it('extracts ordered columns and a single-column PK', () => { + const t = parseCreateTables(OPTIONS_DDL()).get('wp_options')!; + expect(t.columns).toEqual(['option_id', 'option_name', 'option_value']); + expect(t.pkCol).toBe('option_id'); + expect(t.pkIndex).toBe(0); + }); + it('reports composite PK as not delta-eligible (pkCol null)', () => { + const ddl = `CREATE TABLE \`wp_term_relationships\` ( + \`object_id\` bigint(20) NOT NULL, + \`term_taxonomy_id\` bigint(20) NOT NULL, + PRIMARY KEY (\`object_id\`,\`term_taxonomy_id\`) +) ENGINE=InnoDB;`; + const t = parseCreateTables(ddl).get('wp_term_relationships')!; + expect(t.pkCol).toBeNull(); + expect(t.pkIndex).toBe(-1); + }); +}); + +describe('schemaFingerprint', () => { + it('is stable across AUTO_INCREMENT drift', () => { + expect(schemaFingerprint(OPTIONS_DDL(120))).toBe(schemaFingerprint(OPTIONS_DDL(999))); + }); + it('changes when a column is added', () => { + const withCol = OPTIONS_DDL().replace('`option_value` longtext NOT NULL,', '`option_value` longtext NOT NULL,\n `autoload` varchar(20) NOT NULL DEFAULT \'yes\','); + expect(schemaFingerprint(withCol)).not.toBe(schemaFingerprint(OPTIONS_DDL())); + }); +}); + +describe('computeDelta', () => { + const ddl = OPTIONS_DDL(); + const base = dump(ddl, [ + "INSERT INTO `wp_options` VALUES (1,'siteurl','http://localhost:10115');", + "INSERT INTO `wp_options` VALUES (2,'blogname','Old Name');", + "INSERT INTO `wp_options` VALUES (3,'gone','x');", + ]); + + it('emits nothing when nothing changed', () => { + const r = computeDelta({ baselineSql: base, currentSql: base, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); + expect(r.mode).toBe('delta'); + expect(r.sql).toBe(''); + expect(r.stats).toEqual({ tablesChanged: 0, replaces: 0, deletes: 0 }); + }); + + it('emits REPLACE for changed + new rows, DELETE for removed, remapping the prefix', () => { + const cur = dump(ddl, [ + "INSERT INTO `wp_options` VALUES (1,'siteurl','http://localhost:10115');", // unchanged + "INSERT INTO `wp_options` VALUES (2,'blogname','New Name');", // changed + // row 3 removed + "INSERT INTO `wp_options` VALUES (4,'new_opt','v');", // new + ]); + const r = computeDelta({ baselineSql: base, currentSql: cur, dumpPrefix: 'wp_', remotePrefix: 'iwpa4c7_' }); + expect(r.mode).toBe('delta'); + expect(r.stats).toEqual({ tablesChanged: 1, replaces: 2, deletes: 1 }); + expect(r.sql).toContain("REPLACE INTO `iwpa4c7_options` VALUES (2,'blogname','New Name');"); + expect(r.sql).toContain("REPLACE INTO `iwpa4c7_options` VALUES (4,'new_opt','v');"); + expect(r.sql).toContain('DELETE FROM `iwpa4c7_options` WHERE `option_id`=3;'); + expect(r.sql).not.toContain('siteurl'); // unchanged row not touched + expect(r.sql).not.toContain('`wp_options`'); // table identifier remapped to remote prefix + }); + + it('falls back to full when the schema changed', () => { + const cur = dump(OPTIONS_DDL().replace('`option_value` longtext NOT NULL,', '`option_value` longtext NOT NULL,\n `autoload` varchar(20) NOT NULL DEFAULT \'yes\','), [ + "INSERT INTO `wp_options` VALUES (1,'siteurl','http://x','yes');", + ]); + const r = computeDelta({ baselineSql: base, currentSql: cur, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); + expect(r.mode).toBe('full'); + expect(r.reason).toMatch(/schema/i); + }); + + it('falls back to full when a populated table lacks a single-column PK', () => { + const ddl2 = `CREATE TABLE \`wp_term_relationships\` ( + \`object_id\` bigint(20) NOT NULL, + \`term_taxonomy_id\` bigint(20) NOT NULL, + PRIMARY KEY (\`object_id\`,\`term_taxonomy_id\`) +) ENGINE=InnoDB;`; + const b = dump(ddl2, ['INSERT INTO `wp_term_relationships` VALUES (1,2);']); + const c = dump(ddl2, ['INSERT INTO `wp_term_relationships` VALUES (1,3);']); + const r = computeDelta({ baselineSql: b, currentSql: c, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); + expect(r.mode).toBe('full'); + expect(r.reason).toMatch(/primary key/i); + }); +}); diff --git a/src/commands/db.ts b/src/commands/db.ts index 32dd808..5dab0ab 100644 --- a/src/commands/db.ts +++ b/src/commands/db.ts @@ -1,6 +1,6 @@ import { Command } from 'commander'; import { join, dirname, basename } from 'node:path'; -import { existsSync, mkdirSync, statSync, createReadStream, createWriteStream, unlinkSync } from 'node:fs'; +import { existsSync, mkdirSync, statSync, createReadStream, createWriteStream, unlinkSync, readFileSync, writeFileSync } from 'node:fs'; import { createGunzip } from 'node:zlib'; import { pipeline } from 'node:stream/promises'; import { randomBytes } from 'node:crypto'; @@ -13,6 +13,8 @@ import { parseTablePrefix } from '../lib/local-instance.js'; import { detectDumpPrefix, readSqlHead, rewriteDumpPrefix } from '../lib/sql-dump.js'; import { waitForHttp } from '../lib/http-ready.js'; import { parseBackupList, selectBackupsToPrune, type RemoteBackup } from '../lib/db-backups.js'; +import { computeDelta, hasExtendedInsert, schemaFingerprint, type DeltaResult } from '../lib/db-delta.js'; +import { loadBaseline, saveBaseline } from '../lib/db-baseline.js'; import { success, error, spinner, info, table, isJsonMode } from '../lib/output.js'; import type { SshConnection } from '../types.js'; @@ -78,6 +80,174 @@ function listRemoteBackups(conn: SshConnection): RemoteBackup[] { return parseBackupList(res.stdout || ''); } +function isGzipped(file: string): boolean { + return file.endsWith('.gz') || file.endsWith('.gzip'); +} + +/** Read the passed dump as plain SQL text (decompressing a .gz to a tracked temp). */ +async function loadCanonicalDump(file: string, localTemps: Set): Promise { + if (!isGzipped(file)) return readFileSync(file, 'utf-8'); + const tmp = join(process.env.TMPDIR || '/tmp', `instawp-incr-${randomBytes(6).toString('hex')}.sql`); + localTemps.add(tmp); + await gunzipFile(file, tmp); + return readFileSync(tmp, 'utf-8'); +} + +/** + * #17 — incremental push decision. Returns done=true if it fully handled the + * push (delta applied, no-op, or cancelled); done=false to fall through to the + * full push, carrying the baseline to persist after it succeeds. Additive: never + * mutates the full-push path. + */ +async function prepareIncremental(p: { + file: string; site: any; conn: SshConnection; wpPath: string; remoteHome: string; opts: any; + srFrom?: string; srTo?: string; srTables: string[]; +}): Promise<{ done: boolean; baseline?: { sql: string; fingerprint: string } }> { + const { file, site, conn, wpPath, remoteHome, opts, srFrom, srTo, srTables } = p; + const localTemps = new Set(); + const cleanup = () => { for (const f of localTemps) { try { unlinkSync(f); } catch { /* ignore */ } } }; + try { + const sql = await loadCanonicalDump(file, localTemps); + if (hasExtendedInsert(sql)) { + error('--incremental needs a per-row dump. Re-export with: mysqldump --skip-extended-insert --order-by-primary (or `wp db export` with those flags).'); + process.exit(1); + } + const fingerprint = schemaFingerprint(sql); + const baseline = loadBaseline(site.id); + + // Full (re)base required → fall through to the full push, refresh baseline after. + if (opts.full || !baseline || baseline.fingerprint !== fingerprint) { + const why = opts.full ? '--full requested' : !baseline ? 'no baseline yet (first incremental push)' : 'schema changed since the baseline'; + info(`Full push (${why}); the incremental baseline will be refreshed afterwards.`); + return { done: false, baseline: { sql, fingerprint } }; + } + + // Compute the row delta vs the stored baseline. + const pfxRes = execViaSsh(conn, `cd ${wpPath} && wp config get table_prefix`); + const remotePrefix = parseTablePrefix(pfxRes.exitCode === 0 ? pfxRes.stdout : '', 'wp_'); + const dumpPrefix = detectDumpPrefix(sql) ?? 'wp_'; + const delta = computeDelta({ baselineSql: baseline.sql, currentSql: sql, dumpPrefix, remotePrefix }); + + if (delta.mode === 'full') { + info(`Full push (${delta.reason}); the incremental baseline will be refreshed afterwards.`); + return { done: false, baseline: { sql, fingerprint } }; + } + + if (!delta.sql) { + info('No changes since the last push.'); + saveBaseline(site.id, sql, fingerprint, new Date().toISOString()); + success('Incremental push complete', { site_id: site.id, changed: false, replaces: 0, deletes: 0, tables_changed: 0 }); + return { done: true }; + } + + if (!opts.force && !isJsonMode()) { + console.log(`\nIncremental push to ${chalk.bold(conn.domain)}: ${chalk.bold(String(delta.stats!.replaces))} change(s), ${chalk.bold(String(delta.stats!.deletes))} deletion(s) across ${delta.stats!.tablesChanged} table(s). A backup is taken first.`); + const ok = await promptYesNo('Apply this delta? (y/N) '); + if (!ok) { info('Cancelled.'); return { done: true }; } + } + + await applyDelta({ conn, wpPath, remoteHome, site, delta, takeBackup: opts.backup !== false, remapFrom: dumpPrefix, remotePrefix, srFrom, srTo, srTables, verify: !!opts.verify }); + saveBaseline(site.id, sql, fingerprint, new Date().toISOString()); + return { done: true }; + } finally { + cleanup(); + } +} + +/** Apply a computed delta to the remote (backup → upload → import → remap → search-replace → verify). */ +async function applyDelta(p: { + conn: SshConnection; wpPath: string; remoteHome: string; site: any; delta: DeltaResult; + takeBackup: boolean; remapFrom: string; remotePrefix: string; srFrom?: string; srTo?: string; srTables: string[]; verify: boolean; +}): Promise { + const { conn, wpPath, remoteHome, site, delta, takeBackup, remapFrom, remotePrefix, srFrom, srTo, srTables, verify } = p; + const startedAt = Date.now(); + const backupFilename = `db-backup-${isoTimestamp()}.sql.gz`; + const backupRemotePath = `${remoteHome}/${backupFilename}`; + + if (takeBackup) { + const s = spinner(`Backing up remote database to ~/${backupFilename}...`); + s.start(); + const r = execViaSsh(conn, `cd ${wpPath} && wp db export --single-transaction - | gzip > ${backupRemotePath}`); + if (r.exitCode !== 0) { s.fail('Backup failed — aborting delta'); if (r.stderr) error(r.stderr.trim()); process.exit(1); } + s.succeed(`Backup saved: ~/${backupFilename}`); + } else { + info('Skipping backup (--no-backup)'); + } + + const remoteTemp = `/tmp/db-delta-${randomBytes(6).toString('hex')}.sql`; + const localTemp = join(process.env.TMPDIR || '/tmp', `instawp-db-delta-${randomBytes(6).toString('hex')}.sql`); + writeFileSync(localTemp, delta.sql!, 'utf-8'); + const up = spinner('Uploading delta...'); + up.start(); + const scpExit = scpUpload(conn, localTemp, remoteTemp); + try { unlinkSync(localTemp); } catch { /* ignore */ } + if (scpExit !== 0) { up.fail(`Upload failed (scp exit ${scpExit})`); if (takeBackup) info(`Remote backup preserved: ~/${backupFilename}`); process.exit(1); } + up.succeed('Delta uploaded'); + + const imp = spinner(`Applying delta (${delta.stats!.replaces} change, ${delta.stats!.deletes} delete) on ${conn.domain}...`); + imp.start(); + const ir = execViaSsh(conn, `cd ${wpPath} && wp db import ${remoteTemp}`); + execViaSsh(conn, `rm -f ${remoteTemp}`); + if (ir.exitCode !== 0) { + imp.fail('Delta apply failed'); + if (ir.stderr) error(ir.stderr.trim()); else if (ir.stdout) error(ir.stdout.trim()); + if (takeBackup) { + console.log(''); + info(`Remote backup preserved at: ~/${backupFilename}`); + console.log(` ssh ${conn.username}@${conn.host} 'cd ${wpPath} && gunzip -c ${backupRemotePath} | wp db import -'`); + } + process.exit(1); + } + imp.succeed('Delta applied'); + + // REPLACE'd role/cap rows carry the dump prefix in their key VALUES — remap (idempotent). + if (remapFrom && remapFrom !== remotePrefix) { + const cs = spinner('Remapping user roles/capabilities to the remote prefix...'); + cs.start(); + const um = `${remotePrefix}usermeta`; + const opt = `${remotePrefix}options`; + const stmts = [ + `UPDATE ${um} SET meta_key='${remotePrefix}capabilities' WHERE meta_key='${remapFrom}capabilities'`, + `UPDATE ${um} SET meta_key='${remotePrefix}user_level' WHERE meta_key='${remapFrom}user_level'`, + `UPDATE ${opt} SET option_name='${remotePrefix}user_roles' WHERE option_name='${remapFrom}user_roles'`, + ]; + let ok = true; + for (const st of stmts) { const r = execViaSsh(conn, `cd ${wpPath} && wp db query "${st}"`); if (r.exitCode !== 0) { ok = false; if (r.stderr) error(r.stderr.trim()); } } + if (ok) cs.succeed('Roles/capabilities remapped'); else cs.fail('Could not remap roles/capabilities — wp-admin access may need a manual fix'); + } + + if (srFrom && srTo) { + const scope = srTables.length ? srTables.join(' ') : '--all-tables'; + const s = spinner(`Rewriting URLs (${srFrom} -> ${srTo})...`); + s.start(); + const r = execViaSsh(conn, `cd ${wpPath} && wp search-replace '${srFrom}' '${srTo}' ${scope} --skip-columns=guid --report-changed-only`); + if (r.exitCode === 0) { s.succeed('URLs rewritten'); if (!isJsonMode() && r.stdout.trim()) console.log(r.stdout.trim()); } + else { s.fail('URL search-replace failed (delta applied; run it manually if links are wrong)'); if (r.stderr) error(r.stderr.trim()); } + } + + let verified: string | null = null; + if (verify) { + const url = String(site.url || `https://${conn.domain}`).replace(/\/+$/, ''); + const s = spinner(`Verifying ${url} responds...`); + s.start(); + const okv = await waitForHttp(url, 90000); + if (okv) { s.succeed('Site is responding'); verified = 'ok'; } + else { s.fail('Site did not respond within 90s (large changes can need a moment)'); verified = 'timeout'; } + } + + const elapsedSec = (Date.now() - startedAt) / 1000; + success('Incremental push complete', { + site_id: site.id, + backup_path: takeBackup ? backupRemotePath : null, + changed: true, + replaces: delta.stats!.replaces, + deletes: delta.stats!.deletes, + tables_changed: delta.stats!.tablesChanged, + elapsed: `${elapsedSec < 10 ? elapsedSec.toFixed(1) : Math.round(elapsedSec)}s`, + ...(verified ? { verified } : {}), + }); +} + export function registerDbCommand(program: Command): void { const db = program .command('db') @@ -166,6 +336,8 @@ export function registerDbCommand(program: Command): void { .option('--search-replace ', 'After import, run wp search-replace across all tables (serialization-safe, skips guid). Pass exactly two URLs.') .option('--sr-tables ', 'Scope --search-replace to these (prefixed) tables instead of all tables — faster on big DBs whose bulk has no URLs') .option('--verify', 'After import, poll the site URL until it responds (large imports can briefly return 000)') + .option('--incremental', 'Push only the row-level delta since the last push (auto-fulls on first run or schema change). Needs a per-row dump: mysqldump --skip-extended-insert --order-by-primary') + .option('--full', 'Force a full push and refresh the incremental baseline') .addHelpText('after', ` Notes: - Always takes a remote backup first unless --no-backup is passed. @@ -178,11 +350,17 @@ Notes: (or run 'instawp wp search-replace ' yourself). --search-replace scans all tables by default; pass --sr-tables to limit it. - --verify confirms the site answers HTTP after the import (reported in the summary). + - --incremental ships only the row-delta since the last push (baseline stored + per site under ~/.instawp/baselines/). First run, a schema change, or --full + do a normal full push and refresh the baseline. Requires a per-row dump + (mysqldump --skip-extended-insert --order-by-primary). The full push is + untouched — incremental is purely additive. Examples: $ instawp db push my-site dump.sql --rewrite-prefix --verify $ instawp db push my-site dump.sql --search-replace http://localhost:10115 https://my-site.instawp.site $ instawp db push my-site dump.sql --search-replace http://old https://new --sr-tables iwpa4c7_options iwpa4c7_posts iwpa4c7_postmeta + $ instawp db push my-site dump.sql --incremental # delta vs baseline (full push + baseline on first run) `) .action(async (siteIdentifier: string, file: string, opts: any) => { requireAuth(); @@ -254,6 +432,16 @@ Examples: const backupRemotePath = `${remoteHome}/${backupFilename}`; const takeBackup = opts.backup !== false; + // #17 — incremental delta push (additive). May fully handle the push and + // return; otherwise falls through to the full push below (unchanged) and + // refreshes the baseline once it succeeds. + let baselineToSave: { sql: string; fingerprint: string } | null = null; + if (opts.incremental || opts.full) { + const decision = await prepareIncremental({ file, site, conn, wpPath, remoteHome, opts, srFrom, srTo, srTables }); + if (decision.done) return; + baselineToSave = decision.baseline ?? null; + } + // Confirmation if (!opts.force) { const backupLine = takeBackup @@ -500,6 +688,12 @@ Examples: if (!isJsonMode() && takeBackup) { console.log(`\n ${chalk.dim('Backup:')} ~/${backupFilename} ${chalk.dim('(on remote)')}`); } + + // #17 — record the baseline after a successful full (re)base, so the next + // --incremental push can diff against it. + if (baselineToSave) { + saveBaseline(site.id, baselineToSave.sql, baselineToSave.fingerprint, new Date().toISOString()); + } }); // db backups list/prune — manage the ~/db-backup-*.sql.gz files db push leaves behind diff --git a/src/lib/db-baseline.ts b/src/lib/db-baseline.ts new file mode 100644 index 0000000..e7245b1 --- /dev/null +++ b/src/lib/db-baseline.ts @@ -0,0 +1,42 @@ +// Per-site baseline store for `db push --incremental`. The baseline is the last +// successfully-pushed canonical dump (per-row, PK-sorted) plus its schema +// fingerprint; the next incremental push diffs the new dump against it. Stored +// under ~/.instawp/baselines// (the dump can be large — one per site). +import { homedir } from 'node:os'; +import { join } from 'node:path'; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; + +export interface Baseline { + sql: string; + fingerprint: string; + savedAt?: string; +} + +/** Root dir for baselines; `base` is injectable for tests (defaults to ~/.instawp). */ +export function baselineDir(siteId: number | string, base = join(homedir(), '.instawp')): string { + return join(base, 'baselines', String(siteId)); +} + +/** Load the stored baseline for a site, or null if none / unreadable. */ +export function loadBaseline(siteId: number | string, base?: string): Baseline | null { + const dir = baselineDir(siteId, base); + const sqlPath = join(dir, 'baseline.sql'); + const metaPath = join(dir, 'baseline.json'); + if (!existsSync(sqlPath) || !existsSync(metaPath)) return null; + try { + const sql = readFileSync(sqlPath, 'utf-8'); + const meta = JSON.parse(readFileSync(metaPath, 'utf-8')); + if (typeof meta?.fingerprint !== 'string') return null; + return { sql, fingerprint: meta.fingerprint, savedAt: meta.savedAt }; + } catch { + return null; + } +} + +/** Save (overwrite) the baseline for a site. `savedAt` is passed in (no clock here). */ +export function saveBaseline(siteId: number | string, sql: string, fingerprint: string, savedAt: string, base?: string): void { + const dir = baselineDir(siteId, base); + mkdirSync(dir, { recursive: true }); + writeFileSync(join(dir, 'baseline.sql'), sql, 'utf-8'); + writeFileSync(join(dir, 'baseline.json'), JSON.stringify({ fingerprint, savedAt }, null, 2), 'utf-8'); +} diff --git a/src/lib/db-delta.ts b/src/lib/db-delta.ts new file mode 100644 index 0000000..528c779 --- /dev/null +++ b/src/lib/db-delta.ts @@ -0,0 +1,205 @@ +// Pure engine for `db push --incremental`: parse two per-row, PK-sorted MySQL +// dumps (baseline vs current) and emit a minimal REPLACE/DELETE delta. NO I/O, +// NO DROP/CREATE — this is the data-mutating surface, so it is kept pure and +// heavily unit-tested. Falls back to a full push (mode:'full') whenever a delta +// can't be expressed safely (schema/DDL change, or a table without a single +// primary key). +// +// Input requirement: dumps produced with `mysqldump --skip-extended-insert +// --order-by-primary` (one row per INSERT, one tuple per statement). Use +// hasExtendedInsert() to reject anything else before diffing. +import { createHash } from 'node:crypto'; + +export interface TableSchema { + columns: string[]; + pkCol: string | null; // null when composite or absent → not delta-eligible + pkIndex: number; // -1 when pkCol is null +} + +export interface DeltaResult { + mode: 'delta' | 'full'; + reason?: string; + sql?: string; + stats?: { tablesChanged: number; replaces: number; deletes: number }; +} + +/** Split a VALUES tuple body (without the outer parens) into top-level fields. */ +export function splitSqlTuple(inner: string): string[] { + const out: string[] = []; + let cur = ''; + let inStr = false; + for (let i = 0; i < inner.length; i++) { + const c = inner[i]; + if (inStr) { + if (c === '\\') { cur += c + (inner[i + 1] ?? ''); i++; continue; } // backslash escape + cur += c; + if (c === "'") inStr = false; + continue; + } + if (c === "'") { inStr = true; cur += c; continue; } + if (c === ',') { out.push(cur.trim()); cur = ''; continue; } + cur += c; + } + out.push(cur.trim()); + return out; +} + +/** Parse one INSERT statement into its table + tuple bodies (>1 tuple = extended insert). */ +export function parseInsertStatement(stmt: string): { table: string; tuples: string[] } | null { + const m = stmt.match(/^INSERT INTO\s+`([^`]+)`\s+(?:\([^)]*\)\s+)?VALUES\s*/i); + if (!m) return null; + const rest = stmt.slice(m[0].length); + const tuples: string[] = []; + let i = 0; + while (i < rest.length) { + if (rest[i] !== '(') { i++; continue; } + let depth = 0; + let inStr = false; + let tuple = ''; + let j = i; + for (; j < rest.length; j++) { + const c = rest[j]; + if (inStr) { + if (c === '\\') { tuple += c + (rest[j + 1] ?? ''); j++; continue; } + tuple += c; + if (c === "'") inStr = false; + continue; + } + if (c === "'") { inStr = true; tuple += c; continue; } + if (c === '(') { depth++; if (depth === 1) continue; tuple += c; continue; } + if (c === ')') { depth--; if (depth === 0) break; tuple += c; continue; } + tuple += c; + } + tuples.push(tuple); + i = j + 1; + } + return { table: m[1], tuples }; +} + +/** True if any INSERT carries more than one tuple (i.e. NOT --skip-extended-insert). */ +export function hasExtendedInsert(sql: string): boolean { + for (const raw of sql.split('\n')) { + const line = raw.trim(); + if (!/^INSERT INTO/i.test(line)) continue; + const parsed = parseInsertStatement(line); + if (parsed && parsed.tuples.length > 1) return true; + } + return false; +} + +/** Parse CREATE TABLE blocks → ordered columns + single-column PK (if any). */ +export function parseCreateTables(sql: string): Map { + const map = new Map(); + const re = /CREATE TABLE\s+`([^`]+)`\s*\(([\s\S]*?)\n\)/gi; + let m: RegExpExecArray | null; + while ((m = re.exec(sql))) { + const table = m[1]; + const body = m[2]; + const columns: string[] = []; + for (const line of body.split('\n')) { + const cm = line.trim().match(/^`([^`]+)`/); + if (cm) columns.push(cm[1]); + } + let pkCol: string | null = null; + const pk = body.match(/PRIMARY KEY\s*\(([^)]+)\)/i); + if (pk) { + const cols = pk[1].split(',').map((s) => s.trim().replace(/`/g, '')); + if (cols.length === 1) pkCol = cols[0]; + } + map.set(table, { columns, pkCol, pkIndex: pkCol ? columns.indexOf(pkCol) : -1 }); + } + return map; +} + +/** Stable hash of the schema (normalized CREATE TABLE DDL), AUTO_INCREMENT stripped. */ +export function schemaFingerprint(sql: string): string { + const blocks: string[] = []; + const re = /CREATE TABLE[\s\S]*?\n\)[^;]*;/gi; + let m: RegExpExecArray | null; + while ((m = re.exec(sql))) { + blocks.push(m[0].replace(/AUTO_INCREMENT=\d+/gi, 'AUTO_INCREMENT=').replace(/\s+/g, ' ').trim()); + } + blocks.sort(); + return createHash('sha256').update(blocks.join('\n')).digest('hex'); +} + +/** table → (pk literal → tuple body), for single-row INSERTs of delta-eligible tables. */ +function parseInserts(sql: string, schemas: Map): Map> { + const map = new Map>(); + for (const raw of sql.split('\n')) { + const line = raw.trim(); + if (!/^INSERT INTO/i.test(line)) continue; + const stmt = parseInsertStatement(line); + if (!stmt || stmt.tuples.length !== 1) continue; + if (!map.has(stmt.table)) map.set(stmt.table, new Map()); + const sch = schemas.get(stmt.table); + if (!sch || sch.pkIndex < 0) continue; // table seen but not keyable → eligibility check flags it + const pk = splitSqlTuple(stmt.tuples[0])[sch.pkIndex]; + map.get(stmt.table)!.set(pk, stmt.tuples[0]); + } + return map; +} + +/** + * Compute a REPLACE/DELETE delta to bring the remote (== baseline) to `current`. + * Table identifiers are emitted in the REMOTE prefix (dumpPrefix → remotePrefix). + * Returns mode:'full' (with a reason) when a delta can't be safely expressed. + */ +export function computeDelta(opts: { + baselineSql: string; + currentSql: string; + dumpPrefix: string; + remotePrefix: string; +}): DeltaResult { + const { baselineSql, currentSql, dumpPrefix, remotePrefix } = opts; + + if (schemaFingerprint(baselineSql) !== schemaFingerprint(currentSql)) { + return { mode: 'full', reason: 'schema/DDL changed since the baseline' }; + } + + const schemas = parseCreateTables(currentSql); + const cur = parseInserts(currentSql, schemas); + const base = parseInserts(baselineSql, schemas); + + const allTables = new Set([...cur.keys(), ...base.keys()]); + for (const table of allTables) { + const sch = schemas.get(table); + if (!sch || sch.pkCol == null || sch.pkIndex < 0) { + return { mode: 'full', reason: `table \`${table}\` has no single-column primary key` }; + } + } + + const remap = (t: string) => (t.startsWith(dumpPrefix) ? remotePrefix + t.slice(dumpPrefix.length) : t); + const stmts: string[] = []; + const changed = new Set(); + let replaces = 0; + let deletes = 0; + + for (const table of allTables) { + const sch = schemas.get(table)!; + const remTable = remap(table); + const curRows = cur.get(table) ?? new Map(); + const baseRows = base.get(table) ?? new Map(); + for (const [pk, tuple] of curRows) { + const prev = baseRows.get(pk); + if (prev === undefined || prev !== tuple) { + stmts.push(`REPLACE INTO \`${remTable}\` VALUES (${tuple});`); + replaces++; + changed.add(table); + } + } + for (const [pk, tuple] of baseRows) { + if (!curRows.has(pk)) { + const pkLit = splitSqlTuple(tuple)[sch.pkIndex]; + stmts.push(`DELETE FROM \`${remTable}\` WHERE \`${sch.pkCol}\`=${pkLit};`); + deletes++; + changed.add(table); + } + } + } + + const stats = { tablesChanged: changed.size, replaces, deletes }; + if (!stmts.length) return { mode: 'delta', sql: '', stats }; + const sql = `SET FOREIGN_KEY_CHECKS=0;\nSET sql_mode='NO_AUTO_VALUE_ON_ZERO';\n${stmts.join('\n')}\n`; + return { mode: 'delta', sql, stats }; +} From e94c9fb154ca01c95c9d5865c5a594a1de87674d Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Thu, 25 Jun 2026 22:48:42 +0530 Subject: [PATCH 3/5] fix(cli): anchor schema-fingerprint regex so --incremental actually engages (#17) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer found --incremental was a correct no-op: schemaFingerprint()'s CREATE TABLE regex was un-anchored, so "CREATE TABLE ... );" text inside row data (posts/postmeta documenting SQL) matched too, lazily sweeping volatile content into the fingerprint → it changed on every data edit → baseline.fingerprint mismatched every run → always "Full push (schema changed)". Anchor both db-delta regexes to line-start (^ + m): real mysqldump DDL is at column 0; per-row INSERTs are single physical lines (newlines escaped), so a data occurrence can't match. Adds an adversarial regression test (row data containing "CREATE TABLE `evil` (...);"). Not AUTO_INCREMENT (already stripped). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/__tests__/db-delta.test.ts | 34 ++++++++++++++++++++++++++++++++++ src/lib/db-delta.ts | 10 ++++++++-- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/src/__tests__/db-delta.test.ts b/src/__tests__/db-delta.test.ts index 6af0dc3..6ab492f 100644 --- a/src/__tests__/db-delta.test.ts +++ b/src/__tests__/db-delta.test.ts @@ -71,6 +71,40 @@ describe('schemaFingerprint', () => { const withCol = OPTIONS_DDL().replace('`option_value` longtext NOT NULL,', '`option_value` longtext NOT NULL,\n `autoload` varchar(20) NOT NULL DEFAULT \'yes\','); expect(schemaFingerprint(withCol)).not.toBe(schemaFingerprint(OPTIONS_DDL())); }); + + // Regression: "CREATE TABLE … );" text inside row data must NOT count as schema. + // Un-anchored, it swept volatile row content into the fingerprint → it changed + // on every data edit → --incremental always fell back to a full push (no-op). + const adversarialDump = (contentVer: string) => [ + 'CREATE TABLE `wp_posts` (', + ' `ID` bigint(20) NOT NULL AUTO_INCREMENT,', + ' `post_content` longtext NOT NULL,', + ' PRIMARY KEY (`ID`)', + ') ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8mb4;', + `INSERT INTO \`wp_posts\` VALUES (1,'a doc that says CREATE TABLE \`evil\` (x int); ${contentVer}');`, + 'CREATE TABLE `wp_options` (', + ' `option_id` bigint(20) NOT NULL AUTO_INCREMENT,', + ' `option_name` varchar(191) NOT NULL DEFAULT \'\',', + ' PRIMARY KEY (`option_id`)', + ') ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;', + "INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');", + ].join('\n') + '\n'; + + it('ignores "CREATE TABLE" inside row data — stable across a data-only change', () => { + expect(schemaFingerprint(adversarialDump('v1'))).toBe(schemaFingerprint(adversarialDump('v2'))); + }); + + it('computeDelta emits a delta (not a full fallback) when only adversarial-row data changed', () => { + const r = computeDelta({ + baselineSql: adversarialDump('v1'), + currentSql: adversarialDump('v2'), + dumpPrefix: 'wp_', + remotePrefix: 'wp_', + }); + expect(r.mode).toBe('delta'); + expect(r.stats).toEqual({ tablesChanged: 1, replaces: 1, deletes: 0 }); + expect(r.sql).toContain('REPLACE INTO `wp_posts`'); + }); }); describe('computeDelta', () => { diff --git a/src/lib/db-delta.ts b/src/lib/db-delta.ts index 528c779..18123df 100644 --- a/src/lib/db-delta.ts +++ b/src/lib/db-delta.ts @@ -90,7 +90,9 @@ export function hasExtendedInsert(sql: string): boolean { /** Parse CREATE TABLE blocks → ordered columns + single-column PK (if any). */ export function parseCreateTables(sql: string): Map { const map = new Map(); - const re = /CREATE TABLE\s+`([^`]+)`\s*\(([\s\S]*?)\n\)/gi; + // Anchored to line-start (m flag): real mysqldump DDL starts at column 0, so a + // "CREATE TABLE `x` (" occurring inside a single-line INSERT's row data can't match. + const re = /^CREATE TABLE\s+`([^`]+)`\s*\(([\s\S]*?)\n\)/gim; let m: RegExpExecArray | null; while ((m = re.exec(sql))) { const table = m[1]; @@ -114,7 +116,11 @@ export function parseCreateTables(sql: string): Map { /** Stable hash of the schema (normalized CREATE TABLE DDL), AUTO_INCREMENT stripped. */ export function schemaFingerprint(sql: string): string { const blocks: string[] = []; - const re = /CREATE TABLE[\s\S]*?\n\)[^;]*;/gi; + // Anchored to line-start (m flag): without it, "CREATE TABLE … );" appearing + // inside row data (e.g. a post documenting SQL) matched too, lazily sweeping + // volatile content into the fingerprint → it changed on every data edit → + // --incremental always fell back to a full push. (AUTO_INCREMENT already stripped.) + const re = /^CREATE TABLE[\s\S]*?\n\)[^;]*;/gim; let m: RegExpExecArray | null; while ((m = re.exec(sql))) { blocks.push(m[0].replace(/AUTO_INCREMENT=\d+/gi, 'AUTO_INCREMENT=').replace(/\s+/g, ' ').trim()); From 1b3ca0d1ee578041ab5cc5d6413dba98562623fe Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Thu, 25 Jun 2026 23:29:01 +0530 Subject: [PATCH 4/5] fix(cli): scope incremental PK requirement to CHANGED tables (#17, B2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer (round 2): --incremental fell back to full on every real WP site. computeDelta required a single-column PK for EVERY data-bearing table, and wp_term_relationships (composite PK, populated everywhere) tripped the gate unconditionally → "no single-column primary key" every run. Now a table without a single-column PK is row-diffed only if it can be; if it has NO usable PK it's IGNORED when its row set is unchanged (the common case) and only forces a full fallback when it actually changed. Single-PK tables (posts/options/etc.) delta cleanly alongside an unchanged composite-PK table. parseInserts now collects all tables' rows (no PK gating); added sameRowSet for the unchanged-check + a B2 regression test. 388 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/__tests__/db-delta.test.ts | 27 ++++++++++--- src/lib/db-delta.ts | 71 ++++++++++++++++++++++------------ 2 files changed, 69 insertions(+), 29 deletions(-) diff --git a/src/__tests__/db-delta.test.ts b/src/__tests__/db-delta.test.ts index 6ab492f..378aa1f 100644 --- a/src/__tests__/db-delta.test.ts +++ b/src/__tests__/db-delta.test.ts @@ -148,16 +148,33 @@ describe('computeDelta', () => { expect(r.reason).toMatch(/schema/i); }); - it('falls back to full when a populated table lacks a single-column PK', () => { - const ddl2 = `CREATE TABLE \`wp_term_relationships\` ( + const TR_DDL = `CREATE TABLE \`wp_term_relationships\` ( \`object_id\` bigint(20) NOT NULL, \`term_taxonomy_id\` bigint(20) NOT NULL, PRIMARY KEY (\`object_id\`,\`term_taxonomy_id\`) ) ENGINE=InnoDB;`; - const b = dump(ddl2, ['INSERT INTO `wp_term_relationships` VALUES (1,2);']); - const c = dump(ddl2, ['INSERT INTO `wp_term_relationships` VALUES (1,3);']); + + it('falls back to full when a composite-PK table actually changed', () => { + const b = dump(TR_DDL, ['INSERT INTO `wp_term_relationships` VALUES (1,2);']); + const c = dump(TR_DDL, ['INSERT INTO `wp_term_relationships` VALUES (1,3);']); const r = computeDelta({ baselineSql: b, currentSql: c, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); expect(r.mode).toBe('full'); - expect(r.reason).toMatch(/primary key/i); + expect(r.reason).toMatch(/no single-column primary key/i); + }); + + // B2 regression: a populated composite-PK table (wp_term_relationships exists on + // every WP site) must NOT force a full push when it hasn't changed — only when it has. + it('ignores an UNCHANGED composite-PK table and still deltas a changed single-PK table', () => { + const mk = (blogname: string) => [ + OPTIONS_DDL(), + `INSERT INTO \`wp_options\` VALUES (1,'blogname','${blogname}');`, + TR_DDL, + 'INSERT INTO `wp_term_relationships` VALUES (5,9);', // identical in both dumps + ].join('\n') + '\n'; + const r = computeDelta({ baselineSql: mk('Old'), currentSql: mk('New'), dumpPrefix: 'wp_', remotePrefix: 'iwpa4c7_' }); + expect(r.mode).toBe('delta'); + expect(r.stats).toEqual({ tablesChanged: 1, replaces: 1, deletes: 0 }); + expect(r.sql).toContain('REPLACE INTO `iwpa4c7_options`'); + expect(r.sql).not.toContain('term_relationships'); // unchanged composite-PK table left untouched }); }); diff --git a/src/lib/db-delta.ts b/src/lib/db-delta.ts index 18123df..a2e9e20 100644 --- a/src/lib/db-delta.ts +++ b/src/lib/db-delta.ts @@ -129,27 +129,46 @@ export function schemaFingerprint(sql: string): string { return createHash('sha256').update(blocks.join('\n')).digest('hex'); } -/** table → (pk literal → tuple body), for single-row INSERTs of delta-eligible tables. */ -function parseInserts(sql: string, schemas: Map): Map> { - const map = new Map>(); +/** table → ordered list of single-row INSERT tuple bodies (ALL tables, keyable or not). */ +function parseInserts(sql: string): Map { + const map = new Map(); for (const raw of sql.split('\n')) { const line = raw.trim(); if (!/^INSERT INTO/i.test(line)) continue; const stmt = parseInsertStatement(line); if (!stmt || stmt.tuples.length !== 1) continue; - if (!map.has(stmt.table)) map.set(stmt.table, new Map()); - const sch = schemas.get(stmt.table); - if (!sch || sch.pkIndex < 0) continue; // table seen but not keyable → eligibility check flags it - const pk = splitSqlTuple(stmt.tuples[0])[sch.pkIndex]; - map.get(stmt.table)!.set(pk, stmt.tuples[0]); + if (!map.has(stmt.table)) map.set(stmt.table, []); + map.get(stmt.table)!.push(stmt.tuples[0]); } return map; } +/** Index a table's rows by its single-column PK literal (tuple → keyed map). */ +function indexByPk(rows: string[], pkIndex: number): Map { + const m = new Map(); + for (const tuple of rows) m.set(splitSqlTuple(tuple)[pkIndex], tuple); + return m; +} + +/** Order-independent equality of two row-tuple lists (rows are unique per table). */ +function sameRowSet(a: string[], b: string[]): boolean { + if (a.length !== b.length) return false; + const sa = [...a].sort(); + const sb = [...b].sort(); + for (let i = 0; i < sa.length; i++) if (sa[i] !== sb[i]) return false; + return true; +} + /** * Compute a REPLACE/DELETE delta to bring the remote (== baseline) to `current`. * Table identifiers are emitted in the REMOTE prefix (dumpPrefix → remotePrefix). * Returns mode:'full' (with a reason) when a delta can't be safely expressed. + * + * A table WITHOUT a single-column primary key (e.g. `wp_term_relationships`, + * composite PK) can't be row-diffed. Such a table is IGNORED when its row set is + * unchanged (the common case — present on every WP site), and only forces a full + * fallback when it actually changed. (Earlier this gated unconditionally, so any + * site with a populated composite-PK table fell back to full on every push.) */ export function computeDelta(opts: { baselineSql: string; @@ -164,16 +183,9 @@ export function computeDelta(opts: { } const schemas = parseCreateTables(currentSql); - const cur = parseInserts(currentSql, schemas); - const base = parseInserts(baselineSql, schemas); - + const cur = parseInserts(currentSql); + const base = parseInserts(baselineSql); const allTables = new Set([...cur.keys(), ...base.keys()]); - for (const table of allTables) { - const sch = schemas.get(table); - if (!sch || sch.pkCol == null || sch.pkIndex < 0) { - return { mode: 'full', reason: `table \`${table}\` has no single-column primary key` }; - } - } const remap = (t: string) => (t.startsWith(dumpPrefix) ? remotePrefix + t.slice(dumpPrefix.length) : t); const stmts: string[] = []; @@ -182,20 +194,31 @@ export function computeDelta(opts: { let deletes = 0; for (const table of allTables) { - const sch = schemas.get(table)!; + const sch = schemas.get(table); + const curRows = cur.get(table) ?? []; + const baseRows = base.get(table) ?? []; + + // No usable single-column PK → can't row-diff. Ignore if unchanged; else full. + if (!sch || sch.pkCol == null || sch.pkIndex < 0) { + if (!sameRowSet(baseRows, curRows)) { + return { mode: 'full', reason: `table \`${table}\` changed but has no single-column primary key` }; + } + continue; + } + const remTable = remap(table); - const curRows = cur.get(table) ?? new Map(); - const baseRows = base.get(table) ?? new Map(); - for (const [pk, tuple] of curRows) { - const prev = baseRows.get(pk); + const curMap = indexByPk(curRows, sch.pkIndex); + const baseMap = indexByPk(baseRows, sch.pkIndex); + for (const [pk, tuple] of curMap) { + const prev = baseMap.get(pk); if (prev === undefined || prev !== tuple) { stmts.push(`REPLACE INTO \`${remTable}\` VALUES (${tuple});`); replaces++; changed.add(table); } } - for (const [pk, tuple] of baseRows) { - if (!curRows.has(pk)) { + for (const [pk, tuple] of baseMap) { + if (!curMap.has(pk)) { const pkLit = splitSqlTuple(tuple)[sch.pkIndex]; stmts.push(`DELETE FROM \`${remTable}\` WHERE \`${sch.pkCol}\`=${pkLit};`); deletes++; From 83571fa0954b5a17afb52a74ef2ecbf89c995aa1 Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Fri, 26 Jun 2026 06:42:00 +0530 Subject: [PATCH 5/5] =?UTF-8?q?fix(cli):=20memory-bounded=20incremental=20?= =?UTF-8?q?db=20push=20=E2=80=94=20manifest=20+=20streaming=20(#17,=20B3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 3: --incremental engaged but OOM'd (~4 GB) on a 166 MB DB — computeDelta loaded BOTH full dumps as strings + per-row Maps for both (~20x blowup). Redesigned the engine (lib/db-delta.ts) to be streaming + manifest-based: - baseline is stored as a compact per-row HASH MANIFEST (PK -> content hash), not the full SQL (lib/db-baseline.ts now persists manifest.json); - the current dump is STREAMED line-by-line (gz-decompressed on the fly) via a line-oriented state machine — which also makes the B1 class of bug structural (a "CREATE TABLE" inside an INSERT line is never treated as DDL); - diffAgainstManifest builds the next manifest as it streams, so re-basing needs no extra pass; composite/no-PK tables are change-detected via an order- independent aggregate (count + sum of row hashes). Peak engine memory is now ~manifest-sized: measured 356 MB on an 860K-row / 72 MB dump (~3x the real DB's single-PK row count), vs the old ~4 GB OOM. db push --incremental still requires a per-row dump (extended-insert rejected). Rewrote db-delta/db-baseline tests for the streaming API (B1, B2, schema-change, composite-change, extended-insert, serialize round-trip). Full suite 380 green. The full db push path remains byte-for-byte unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 1 + src/__tests__/db-baseline.test.ts | 29 ++- src/__tests__/db-delta.test.ts | 207 +++++++--------- src/commands/db.ts | 95 ++++---- src/lib/db-baseline.ts | 36 +-- src/lib/db-delta.ts | 390 +++++++++++++++++++----------- 6 files changed, 423 insertions(+), 335 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 934948b..4fb8a8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ - **`db push --incremental`** ships only the **row-level delta** since the last push instead of the whole DB: it diffs the dump against a per-site baseline (`~/.instawp/baselines//`) and applies a minimal `REPLACE`/`DELETE` set (no `DROP`/`CREATE`). First run, a schema/DDL change, or **`--full`** do a normal full push and refresh the baseline. Requires a per-row dump (`mysqldump --skip-extended-insert --order-by-primary`; extended-insert dumps are rejected). Reuses the existing safety machinery — remote backup first, prefix/role-key remap, scoped URL search-replace, `--verify`. **The full `db push` path is unchanged; incremental is purely additive.** - MVP scope: tables with a single-column primary key (WP core + most plugin tables); anything without one — or any DDL change — auto-falls-back to a full push. The diff engine (`lib/db-delta.ts`) and baseline store (`lib/db-baseline.ts`) are pure and unit-tested. +- **Memory-bounded:** the baseline is stored as a compact per-row **hash manifest** (PK → content hash), not the full dump, and the current dump is **streamed** line-by-line (gz-decompressed on the fly) — so a large DB diffs in a few hundred MB instead of loading both dumps into the heap (a naive 2× full-dump diff OOM'd at ~4 GB on a 166 MB DB). ### Added — db push readiness/scoping + backup retention (large-DB feedback #16) diff --git a/src/__tests__/db-baseline.test.ts b/src/__tests__/db-baseline.test.ts index 050c0a6..6e7e420 100644 --- a/src/__tests__/db-baseline.test.ts +++ b/src/__tests__/db-baseline.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from 'node:os'; import { mkdtempSync, rmSync } from 'node:fs'; import { join } from 'node:path'; import { baselineDir, loadBaseline, saveBaseline } from '../lib/db-baseline.js'; +import { buildManifest } from '../lib/db-delta.js'; const dirs: string[] = []; function tmpBase(): string { @@ -12,23 +13,29 @@ function tmpBase(): string { } afterEach(() => { for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }); }); -describe('db-baseline store', () => { - it('round-trips sql + fingerprint per site', () => { +const DDL = `CREATE TABLE \`wp_options\` ( + \`option_id\` bigint(20) NOT NULL AUTO_INCREMENT, + \`option_name\` varchar(191) NOT NULL, + PRIMARY KEY (\`option_id\`) +) ENGINE=InnoDB;`; + +describe('db-baseline manifest store', () => { + it('round-trips a manifest per site', async () => { const base = tmpBase(); expect(loadBaseline(42, base)).toBeNull(); - saveBaseline(42, 'CREATE TABLE `wp_options` ...', 'fp-abc', '2026-06-25T00:00:00Z', base); + const m = await buildManifest([DDL, "INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');"].join('\n').split('\n')); + saveBaseline(42, m, base); const got = loadBaseline(42, base); - expect(got?.sql).toContain('wp_options'); - expect(got?.fingerprint).toBe('fp-abc'); - expect(got?.savedAt).toBe('2026-06-25T00:00:00Z'); + expect(got?.fingerprint).toBe(m.fingerprint); + expect(got?.single.get('wp_options')?.rows.get('1')).toBe(m.single.get('wp_options')?.rows.get('1')); }); - it('scopes baselines per site id', () => { + it('scopes baselines per site id', async () => { const base = tmpBase(); - saveBaseline(1, 'a', 'fp1', 't', base); - saveBaseline(2, 'b', 'fp2', 't', base); - expect(loadBaseline(1, base)?.fingerprint).toBe('fp1'); - expect(loadBaseline(2, base)?.fingerprint).toBe('fp2'); + const m = await buildManifest([DDL, "INSERT INTO `wp_options` VALUES (1,'a','b');"].join('\n').split('\n')); + saveBaseline(1, m, base); + expect(loadBaseline(1, base)?.fingerprint).toBe(m.fingerprint); + expect(loadBaseline(2, base)).toBeNull(); expect(baselineDir(1, base)).not.toBe(baselineDir(2, base)); }); }); diff --git a/src/__tests__/db-delta.test.ts b/src/__tests__/db-delta.test.ts index 378aa1f..1e174a1 100644 --- a/src/__tests__/db-delta.test.ts +++ b/src/__tests__/db-delta.test.ts @@ -2,179 +2,160 @@ import { describe, it, expect } from 'vitest'; import { splitSqlTuple, parseInsertStatement, - hasExtendedInsert, - parseCreateTables, - schemaFingerprint, - computeDelta, + prefixFromTableNames, + buildManifest, + diffAgainstManifest, + serializeManifest, + deserializeManifest, + ExtendedInsertError, } from '../lib/db-delta.js'; const OPTIONS_DDL = (autoInc = 120) => `CREATE TABLE \`wp_options\` ( \`option_id\` bigint(20) unsigned NOT NULL AUTO_INCREMENT, \`option_name\` varchar(191) NOT NULL DEFAULT '', \`option_value\` longtext NOT NULL, - PRIMARY KEY (\`option_id\`), - UNIQUE KEY \`option_name\` (\`option_name\`) + PRIMARY KEY (\`option_id\`) ) ENGINE=InnoDB AUTO_INCREMENT=${autoInc} DEFAULT CHARSET=utf8mb4;`; -const dump = (ddl: string, rows: string[]) => [ddl, ...rows].join('\n') + '\n'; +const TR_DDL = `CREATE TABLE \`wp_term_relationships\` ( + \`object_id\` bigint(20) NOT NULL, + \`term_taxonomy_id\` bigint(20) NOT NULL, + PRIMARY KEY (\`object_id\`,\`term_taxonomy_id\`) +) ENGINE=InnoDB;`; + +const lines = (s: string) => s.split('\n'); +const dump = (ddl: string, rows: string[]) => lines([ddl, ...rows].join('\n')); +const opts = { dumpPrefix: 'wp_', remotePrefix: 'iwpa4c7_' }; describe('splitSqlTuple', () => { - it('splits top-level fields', () => { - expect(splitSqlTuple("1,'siteurl','http://x'")).toEqual(['1', "'siteurl'", "'http://x'"]); - }); - it('respects commas and parens inside strings', () => { + it('respects commas/parens/quotes inside strings', () => { expect(splitSqlTuple("5,'a, b (c)',7")).toEqual(['5', "'a, b (c)'", '7']); - }); - it('respects escaped quotes', () => { expect(splitSqlTuple("1,'it\\'s ok',2")).toEqual(['1', "'it\\'s ok'", '2']); }); }); -describe('parseInsertStatement / hasExtendedInsert', () => { - it('parses a single-row insert', () => { - const r = parseInsertStatement("INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');"); - expect(r?.table).toBe('wp_options'); - expect(r?.tuples).toEqual(["1,'siteurl','http://x'"]); - }); - it('detects an extended (multi-row) insert', () => { - expect(hasExtendedInsert("INSERT INTO `wp_options` VALUES (1,'a'),(2,'b');")).toBe(true); - }); - it('does not false-positive on a single row containing ),(', () => { - expect(hasExtendedInsert("INSERT INTO `wp_options` VALUES (1,'a),(b');")).toBe(false); +describe('parseInsertStatement', () => { + it('parses a single-row insert and flags extended', () => { + expect(parseInsertStatement("INSERT INTO `t` VALUES (1,'a');")?.tuples).toEqual(["1,'a'"]); + expect(parseInsertStatement("INSERT INTO `t` VALUES (1,'a'),(2,'b');")?.tuples.length).toBe(2); }); }); -describe('parseCreateTables', () => { - it('extracts ordered columns and a single-column PK', () => { - const t = parseCreateTables(OPTIONS_DDL()).get('wp_options')!; - expect(t.columns).toEqual(['option_id', 'option_name', 'option_value']); - expect(t.pkCol).toBe('option_id'); - expect(t.pkIndex).toBe(0); - }); - it('reports composite PK as not delta-eligible (pkCol null)', () => { - const ddl = `CREATE TABLE \`wp_term_relationships\` ( - \`object_id\` bigint(20) NOT NULL, - \`term_taxonomy_id\` bigint(20) NOT NULL, - PRIMARY KEY (\`object_id\`,\`term_taxonomy_id\`) -) ENGINE=InnoDB;`; - const t = parseCreateTables(ddl).get('wp_term_relationships')!; - expect(t.pkCol).toBeNull(); - expect(t.pkIndex).toBe(-1); +describe('prefixFromTableNames', () => { + it('derives the table prefix from core table names', () => { + expect(prefixFromTableNames(['iwpa4c7_postmeta', 'iwpa4c7_options'])).toBe('iwpa4c7_'); + expect(prefixFromTableNames(['wp_users'])).toBe('wp_'); + expect(prefixFromTableNames(['random_table'])).toBeNull(); }); }); -describe('schemaFingerprint', () => { - it('is stable across AUTO_INCREMENT drift', () => { - expect(schemaFingerprint(OPTIONS_DDL(120))).toBe(schemaFingerprint(OPTIONS_DDL(999))); - }); - it('changes when a column is added', () => { - const withCol = OPTIONS_DDL().replace('`option_value` longtext NOT NULL,', '`option_value` longtext NOT NULL,\n `autoload` varchar(20) NOT NULL DEFAULT \'yes\','); - expect(schemaFingerprint(withCol)).not.toBe(schemaFingerprint(OPTIONS_DDL())); - }); - - // Regression: "CREATE TABLE … );" text inside row data must NOT count as schema. - // Un-anchored, it swept volatile row content into the fingerprint → it changed - // on every data edit → --incremental always fell back to a full push (no-op). - const adversarialDump = (contentVer: string) => [ - 'CREATE TABLE `wp_posts` (', - ' `ID` bigint(20) NOT NULL AUTO_INCREMENT,', - ' `post_content` longtext NOT NULL,', - ' PRIMARY KEY (`ID`)', - ') ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8mb4;', - `INSERT INTO \`wp_posts\` VALUES (1,'a doc that says CREATE TABLE \`evil\` (x int); ${contentVer}');`, - 'CREATE TABLE `wp_options` (', - ' `option_id` bigint(20) NOT NULL AUTO_INCREMENT,', - ' `option_name` varchar(191) NOT NULL DEFAULT \'\',', - ' PRIMARY KEY (`option_id`)', - ') ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;', - "INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');", - ].join('\n') + '\n'; - - it('ignores "CREATE TABLE" inside row data — stable across a data-only change', () => { - expect(schemaFingerprint(adversarialDump('v1'))).toBe(schemaFingerprint(adversarialDump('v2'))); +describe('buildManifest', () => { + it('captures single-PK row hashes, composite aggregates, and a schema fingerprint', async () => { + const m = await buildManifest(dump(OPTIONS_DDL(), [ + "INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');", + "INSERT INTO `wp_options` VALUES (2,'blogname','S');", + ]).concat(lines(TR_DDL), ['INSERT INTO `wp_term_relationships` VALUES (5,9);'])); + expect(m.single.get('wp_options')?.rows.size).toBe(2); + expect(m.single.get('wp_options')?.pkCol).toBe('option_id'); + expect(m.composite.get('wp_term_relationships')?.count).toBe(1); + expect(m.fingerprint).toMatch(/^[0-9a-f]{64}$/); + }); + + it('fingerprint is stable across AUTO_INCREMENT drift and "CREATE TABLE" in row data (B1)', async () => { + const adversarial = (v: string) => dump(OPTIONS_DDL(v === 'v1' ? 120 : 988), [ + `INSERT INTO \`wp_options\` VALUES (1,'doc','says CREATE TABLE \`evil\` (x int); ${v}');`, + ]); + const a = await buildManifest(adversarial('v1')); + const b = await buildManifest(adversarial('v2')); + expect(a.fingerprint).toBe(b.fingerprint); }); - it('computeDelta emits a delta (not a full fallback) when only adversarial-row data changed', () => { - const r = computeDelta({ - baselineSql: adversarialDump('v1'), - currentSql: adversarialDump('v2'), - dumpPrefix: 'wp_', - remotePrefix: 'wp_', - }); - expect(r.mode).toBe('delta'); - expect(r.stats).toEqual({ tablesChanged: 1, replaces: 1, deletes: 0 }); - expect(r.sql).toContain('REPLACE INTO `wp_posts`'); + it('throws on an extended-insert dump', async () => { + await expect(buildManifest(dump(OPTIONS_DDL(), ["INSERT INTO `wp_options` VALUES (1,'a','b'),(2,'c','d');"]))) + .rejects.toBeInstanceOf(ExtendedInsertError); }); }); -describe('computeDelta', () => { - const ddl = OPTIONS_DDL(); - const base = dump(ddl, [ +describe('diffAgainstManifest', () => { + const base = dump(OPTIONS_DDL(), [ "INSERT INTO `wp_options` VALUES (1,'siteurl','http://localhost:10115');", "INSERT INTO `wp_options` VALUES (2,'blogname','Old Name');", "INSERT INTO `wp_options` VALUES (3,'gone','x');", ]); - it('emits nothing when nothing changed', () => { - const r = computeDelta({ baselineSql: base, currentSql: base, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); + it('emits nothing when nothing changed', async () => { + const m = await buildManifest(base); + const r = await diffAgainstManifest(base, m, { dumpPrefix: 'wp_', remotePrefix: 'wp_' }); expect(r.mode).toBe('delta'); expect(r.sql).toBe(''); expect(r.stats).toEqual({ tablesChanged: 0, replaces: 0, deletes: 0 }); }); - it('emits REPLACE for changed + new rows, DELETE for removed, remapping the prefix', () => { - const cur = dump(ddl, [ + it('REPLACE for changed/new rows, DELETE for removed, remapping the prefix', async () => { + const m = await buildManifest(base); + const cur = dump(OPTIONS_DDL(), [ "INSERT INTO `wp_options` VALUES (1,'siteurl','http://localhost:10115');", // unchanged "INSERT INTO `wp_options` VALUES (2,'blogname','New Name');", // changed // row 3 removed "INSERT INTO `wp_options` VALUES (4,'new_opt','v');", // new ]); - const r = computeDelta({ baselineSql: base, currentSql: cur, dumpPrefix: 'wp_', remotePrefix: 'iwpa4c7_' }); + const r = await diffAgainstManifest(cur, m, opts); expect(r.mode).toBe('delta'); expect(r.stats).toEqual({ tablesChanged: 1, replaces: 2, deletes: 1 }); expect(r.sql).toContain("REPLACE INTO `iwpa4c7_options` VALUES (2,'blogname','New Name');"); expect(r.sql).toContain("REPLACE INTO `iwpa4c7_options` VALUES (4,'new_opt','v');"); expect(r.sql).toContain('DELETE FROM `iwpa4c7_options` WHERE `option_id`=3;'); - expect(r.sql).not.toContain('siteurl'); // unchanged row not touched - expect(r.sql).not.toContain('`wp_options`'); // table identifier remapped to remote prefix + expect(r.sql).not.toContain('siteurl'); + expect(r.sql).not.toContain('`wp_options`'); }); - it('falls back to full when the schema changed', () => { + it('falls back to full when the schema changed', async () => { + const m = await buildManifest(base); const cur = dump(OPTIONS_DDL().replace('`option_value` longtext NOT NULL,', '`option_value` longtext NOT NULL,\n `autoload` varchar(20) NOT NULL DEFAULT \'yes\','), [ "INSERT INTO `wp_options` VALUES (1,'siteurl','http://x','yes');", ]); - const r = computeDelta({ baselineSql: base, currentSql: cur, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); + const r = await diffAgainstManifest(cur, m, opts); expect(r.mode).toBe('full'); expect(r.reason).toMatch(/schema/i); }); - const TR_DDL = `CREATE TABLE \`wp_term_relationships\` ( - \`object_id\` bigint(20) NOT NULL, - \`term_taxonomy_id\` bigint(20) NOT NULL, - PRIMARY KEY (\`object_id\`,\`term_taxonomy_id\`) -) ENGINE=InnoDB;`; + // B2 regression: unchanged composite-PK table must not force a full push. + it('ignores an UNCHANGED composite-PK table and still deltas the changed single-PK table', async () => { + const mk = (name: string) => dump(OPTIONS_DDL(), [`INSERT INTO \`wp_options\` VALUES (1,'blogname','${name}');`]) + .concat(lines(TR_DDL), ['INSERT INTO `wp_term_relationships` VALUES (5,9);']); + const m = await buildManifest(mk('Old')); + const r = await diffAgainstManifest(mk('New'), m, opts); + expect(r.mode).toBe('delta'); + expect(r.stats).toEqual({ tablesChanged: 1, replaces: 1, deletes: 0 }); + expect(r.sql).toContain('REPLACE INTO `iwpa4c7_options`'); + expect(r.sql).not.toContain('term_relationships'); + }); - it('falls back to full when a composite-PK table actually changed', () => { - const b = dump(TR_DDL, ['INSERT INTO `wp_term_relationships` VALUES (1,2);']); - const c = dump(TR_DDL, ['INSERT INTO `wp_term_relationships` VALUES (1,3);']); - const r = computeDelta({ baselineSql: b, currentSql: c, dumpPrefix: 'wp_', remotePrefix: 'wp_' }); + it('falls back to full when a composite-PK table actually changed', async () => { + const m = await buildManifest(dump(TR_DDL, ['INSERT INTO `wp_term_relationships` VALUES (1,2);'])); + const r = await diffAgainstManifest(dump(TR_DDL, ['INSERT INTO `wp_term_relationships` VALUES (1,3);']), m, { dumpPrefix: 'wp_', remotePrefix: 'wp_' }); expect(r.mode).toBe('full'); expect(r.reason).toMatch(/no single-column primary key/i); }); - // B2 regression: a populated composite-PK table (wp_term_relationships exists on - // every WP site) must NOT force a full push when it hasn't changed — only when it has. - it('ignores an UNCHANGED composite-PK table and still deltas a changed single-PK table', () => { - const mk = (blogname: string) => [ - OPTIONS_DDL(), - `INSERT INTO \`wp_options\` VALUES (1,'blogname','${blogname}');`, - TR_DDL, - 'INSERT INTO `wp_term_relationships` VALUES (5,9);', // identical in both dumps - ].join('\n') + '\n'; - const r = computeDelta({ baselineSql: mk('Old'), currentSql: mk('New'), dumpPrefix: 'wp_', remotePrefix: 'iwpa4c7_' }); + it('builds a fresh newManifest each diff (for re-basing)', async () => { + const m = await buildManifest(base); + const cur = dump(OPTIONS_DDL(), ["INSERT INTO `wp_options` VALUES (1,'siteurl','http://x');"]); + const r = await diffAgainstManifest(cur, m, opts); + expect(r.newManifest.single.get('wp_options')?.rows.size).toBe(1); + }); +}); + +describe('serialize/deserialize manifest', () => { + it('round-trips single + composite tables', async () => { + const m = await buildManifest(dump(OPTIONS_DDL(), ["INSERT INTO `wp_options` VALUES (1,'a','b');"]).concat(lines(TR_DDL), ['INSERT INTO `wp_term_relationships` VALUES (5,9);'])); + const back = deserializeManifest(serializeManifest(m)); + expect(back.fingerprint).toBe(m.fingerprint); + expect(back.single.get('wp_options')?.rows.get('1')).toBe(m.single.get('wp_options')?.rows.get('1')); + expect(back.composite.get('wp_term_relationships')?.count).toBe(1); + // a diff using the deserialized manifest behaves identically (no false change) + const r = await diffAgainstManifest(dump(OPTIONS_DDL(), ["INSERT INTO `wp_options` VALUES (1,'a','b');"]).concat(lines(TR_DDL), ['INSERT INTO `wp_term_relationships` VALUES (5,9);']), back, { dumpPrefix: 'wp_', remotePrefix: 'wp_' }); expect(r.mode).toBe('delta'); - expect(r.stats).toEqual({ tablesChanged: 1, replaces: 1, deletes: 0 }); - expect(r.sql).toContain('REPLACE INTO `iwpa4c7_options`'); - expect(r.sql).not.toContain('term_relationships'); // unchanged composite-PK table left untouched + expect(r.sql).toBe(''); }); }); diff --git a/src/commands/db.ts b/src/commands/db.ts index 5dab0ab..e3b2a8a 100644 --- a/src/commands/db.ts +++ b/src/commands/db.ts @@ -1,7 +1,8 @@ import { Command } from 'commander'; import { join, dirname, basename } from 'node:path'; -import { existsSync, mkdirSync, statSync, createReadStream, createWriteStream, unlinkSync, readFileSync, writeFileSync } from 'node:fs'; +import { existsSync, mkdirSync, statSync, createReadStream, createWriteStream, unlinkSync, writeFileSync } from 'node:fs'; import { createGunzip } from 'node:zlib'; +import { createInterface } from 'node:readline'; import { pipeline } from 'node:stream/promises'; import { randomBytes } from 'node:crypto'; import chalk from 'chalk'; @@ -13,7 +14,7 @@ import { parseTablePrefix } from '../lib/local-instance.js'; import { detectDumpPrefix, readSqlHead, rewriteDumpPrefix } from '../lib/sql-dump.js'; import { waitForHttp } from '../lib/http-ready.js'; import { parseBackupList, selectBackupsToPrune, type RemoteBackup } from '../lib/db-backups.js'; -import { computeDelta, hasExtendedInsert, schemaFingerprint, type DeltaResult } from '../lib/db-delta.js'; +import { buildManifest, diffAgainstManifest, prefixFromTableNames, ExtendedInsertError, type Manifest } from '../lib/db-delta.js'; import { loadBaseline, saveBaseline } from '../lib/db-baseline.js'; import { success, error, spinner, info, table, isJsonMode } from '../lib/output.js'; import type { SshConnection } from '../types.js'; @@ -84,13 +85,11 @@ function isGzipped(file: string): boolean { return file.endsWith('.gz') || file.endsWith('.gzip'); } -/** Read the passed dump as plain SQL text (decompressing a .gz to a tracked temp). */ -async function loadCanonicalDump(file: string, localTemps: Set): Promise { - if (!isGzipped(file)) return readFileSync(file, 'utf-8'); - const tmp = join(process.env.TMPDIR || '/tmp', `instawp-incr-${randomBytes(6).toString('hex')}.sql`); - localTemps.add(tmp); - await gunzipFile(file, tmp); - return readFileSync(tmp, 'utf-8'); +/** Stream a dump file's lines (decompressing a .gz on the fly). Bounded memory. */ +function openDumpLines(file: string): AsyncIterable { + const raw = createReadStream(file); + const input = isGzipped(file) ? raw.pipe(createGunzip()) : raw; + return createInterface({ input, crlfDelay: Infinity }); } /** @@ -102,64 +101,70 @@ async function loadCanonicalDump(file: string, localTemps: Set): Promise async function prepareIncremental(p: { file: string; site: any; conn: SshConnection; wpPath: string; remoteHome: string; opts: any; srFrom?: string; srTo?: string; srTables: string[]; -}): Promise<{ done: boolean; baseline?: { sql: string; fingerprint: string } }> { +}): Promise<{ done: boolean; baseline?: Manifest }> { const { file, site, conn, wpPath, remoteHome, opts, srFrom, srTo, srTables } = p; - const localTemps = new Set(); - const cleanup = () => { for (const f of localTemps) { try { unlinkSync(f); } catch { /* ignore */ } } }; try { - const sql = await loadCanonicalDump(file, localTemps); - if (hasExtendedInsert(sql)) { - error('--incremental needs a per-row dump. Re-export with: mysqldump --skip-extended-insert --order-by-primary (or `wp db export` with those flags).'); - process.exit(1); - } - const fingerprint = schemaFingerprint(sql); - const baseline = loadBaseline(site.id); + const baseline = opts.full ? null : loadBaseline(site.id); - // Full (re)base required → fall through to the full push, refresh baseline after. - if (opts.full || !baseline || baseline.fingerprint !== fingerprint) { - const why = opts.full ? '--full requested' : !baseline ? 'no baseline yet (first incremental push)' : 'schema changed since the baseline'; + // First run / --full / no baseline → build a manifest from the current dump + // (streamed) and fall through to the full push, which refreshes the baseline. + if (!baseline) { + const why = opts.full ? '--full requested' : 'no baseline yet (first incremental push)'; info(`Full push (${why}); the incremental baseline will be refreshed afterwards.`); - return { done: false, baseline: { sql, fingerprint } }; + const mSpin = spinner('Indexing the dump for the baseline...'); + mSpin.start(); + const manifest = await buildManifest(openDumpLines(file)); + mSpin.succeed('Dump indexed'); + return { done: false, baseline: manifest }; } - // Compute the row delta vs the stored baseline. + // Diff the current dump (streamed) against the stored baseline manifest. const pfxRes = execViaSsh(conn, `cd ${wpPath} && wp config get table_prefix`); const remotePrefix = parseTablePrefix(pfxRes.exitCode === 0 ? pfxRes.stdout : '', 'wp_'); - const dumpPrefix = detectDumpPrefix(sql) ?? 'wp_'; - const delta = computeDelta({ baselineSql: baseline.sql, currentSql: sql, dumpPrefix, remotePrefix }); + const dumpPrefix = prefixFromTableNames([...baseline.single.keys(), ...baseline.composite.keys()]) ?? 'wp_'; + + const diffSpin = spinner('Computing the delta vs the baseline...'); + diffSpin.start(); + const result = await diffAgainstManifest(openDumpLines(file), baseline, { dumpPrefix, remotePrefix }); + diffSpin.stop(); - if (delta.mode === 'full') { - info(`Full push (${delta.reason}); the incremental baseline will be refreshed afterwards.`); - return { done: false, baseline: { sql, fingerprint } }; + if (result.mode === 'full') { + info(`Full push (${result.reason}); the incremental baseline will be refreshed afterwards.`); + return { done: false, baseline: result.newManifest }; } - if (!delta.sql) { + if (!result.sql) { info('No changes since the last push.'); - saveBaseline(site.id, sql, fingerprint, new Date().toISOString()); + saveBaseline(site.id, result.newManifest); success('Incremental push complete', { site_id: site.id, changed: false, replaces: 0, deletes: 0, tables_changed: 0 }); return { done: true }; } if (!opts.force && !isJsonMode()) { - console.log(`\nIncremental push to ${chalk.bold(conn.domain)}: ${chalk.bold(String(delta.stats!.replaces))} change(s), ${chalk.bold(String(delta.stats!.deletes))} deletion(s) across ${delta.stats!.tablesChanged} table(s). A backup is taken first.`); + console.log(`\nIncremental push to ${chalk.bold(conn.domain)}: ${chalk.bold(String(result.stats!.replaces))} change(s), ${chalk.bold(String(result.stats!.deletes))} deletion(s) across ${result.stats!.tablesChanged} table(s). A backup is taken first.`); const ok = await promptYesNo('Apply this delta? (y/N) '); if (!ok) { info('Cancelled.'); return { done: true }; } } - await applyDelta({ conn, wpPath, remoteHome, site, delta, takeBackup: opts.backup !== false, remapFrom: dumpPrefix, remotePrefix, srFrom, srTo, srTables, verify: !!opts.verify }); - saveBaseline(site.id, sql, fingerprint, new Date().toISOString()); + await applyDelta({ conn, wpPath, remoteHome, site, deltaSql: result.sql, stats: result.stats!, takeBackup: opts.backup !== false, remapFrom: dumpPrefix, remotePrefix, srFrom, srTo, srTables, verify: !!opts.verify }); + saveBaseline(site.id, result.newManifest); return { done: true }; - } finally { - cleanup(); + } catch (e) { + if (e instanceof ExtendedInsertError) { + error('--incremental needs a per-row dump. Re-export with: mysqldump --skip-extended-insert --order-by-primary (or `wp db export` with those flags).'); + process.exit(1); + } + throw e; } } /** Apply a computed delta to the remote (backup → upload → import → remap → search-replace → verify). */ async function applyDelta(p: { - conn: SshConnection; wpPath: string; remoteHome: string; site: any; delta: DeltaResult; + conn: SshConnection; wpPath: string; remoteHome: string; site: any; + deltaSql: string; stats: { tablesChanged: number; replaces: number; deletes: number }; takeBackup: boolean; remapFrom: string; remotePrefix: string; srFrom?: string; srTo?: string; srTables: string[]; verify: boolean; }): Promise { - const { conn, wpPath, remoteHome, site, delta, takeBackup, remapFrom, remotePrefix, srFrom, srTo, srTables, verify } = p; + const { conn, wpPath, remoteHome, site, deltaSql, stats, takeBackup, remapFrom, remotePrefix, srFrom, srTo, srTables, verify } = p; const startedAt = Date.now(); const backupFilename = `db-backup-${isoTimestamp()}.sql.gz`; const backupRemotePath = `${remoteHome}/${backupFilename}`; @@ -176,7 +181,7 @@ async function applyDelta(p: { const remoteTemp = `/tmp/db-delta-${randomBytes(6).toString('hex')}.sql`; const localTemp = join(process.env.TMPDIR || '/tmp', `instawp-db-delta-${randomBytes(6).toString('hex')}.sql`); - writeFileSync(localTemp, delta.sql!, 'utf-8'); + writeFileSync(localTemp, deltaSql, 'utf-8'); const up = spinner('Uploading delta...'); up.start(); const scpExit = scpUpload(conn, localTemp, remoteTemp); @@ -184,7 +189,7 @@ async function applyDelta(p: { if (scpExit !== 0) { up.fail(`Upload failed (scp exit ${scpExit})`); if (takeBackup) info(`Remote backup preserved: ~/${backupFilename}`); process.exit(1); } up.succeed('Delta uploaded'); - const imp = spinner(`Applying delta (${delta.stats!.replaces} change, ${delta.stats!.deletes} delete) on ${conn.domain}...`); + const imp = spinner(`Applying delta (${stats.replaces} change, ${stats.deletes} delete) on ${conn.domain}...`); imp.start(); const ir = execViaSsh(conn, `cd ${wpPath} && wp db import ${remoteTemp}`); execViaSsh(conn, `rm -f ${remoteTemp}`); @@ -240,9 +245,9 @@ async function applyDelta(p: { site_id: site.id, backup_path: takeBackup ? backupRemotePath : null, changed: true, - replaces: delta.stats!.replaces, - deletes: delta.stats!.deletes, - tables_changed: delta.stats!.tablesChanged, + replaces: stats.replaces, + deletes: stats.deletes, + tables_changed: stats.tablesChanged, elapsed: `${elapsedSec < 10 ? elapsedSec.toFixed(1) : Math.round(elapsedSec)}s`, ...(verified ? { verified } : {}), }); @@ -435,7 +440,7 @@ Examples: // #17 — incremental delta push (additive). May fully handle the push and // return; otherwise falls through to the full push below (unchanged) and // refreshes the baseline once it succeeds. - let baselineToSave: { sql: string; fingerprint: string } | null = null; + let baselineToSave: Manifest | null = null; if (opts.incremental || opts.full) { const decision = await prepareIncremental({ file, site, conn, wpPath, remoteHome, opts, srFrom, srTo, srTables }); if (decision.done) return; @@ -692,7 +697,7 @@ Examples: // #17 — record the baseline after a successful full (re)base, so the next // --incremental push can diff against it. if (baselineToSave) { - saveBaseline(site.id, baselineToSave.sql, baselineToSave.fingerprint, new Date().toISOString()); + saveBaseline(site.id, baselineToSave); } }); diff --git a/src/lib/db-baseline.ts b/src/lib/db-baseline.ts index e7245b1..131dfcc 100644 --- a/src/lib/db-baseline.ts +++ b/src/lib/db-baseline.ts @@ -1,42 +1,30 @@ -// Per-site baseline store for `db push --incremental`. The baseline is the last -// successfully-pushed canonical dump (per-row, PK-sorted) plus its schema -// fingerprint; the next incremental push diffs the new dump against it. Stored -// under ~/.instawp/baselines// (the dump can be large — one per site). +// Per-site baseline store for `db push --incremental`. The baseline is a compact +// per-row hash MANIFEST (not the full dump) so a large DB doesn't blow up memory. +// Stored under ~/.instawp/baselines//manifest.json. import { homedir } from 'node:os'; import { join } from 'node:path'; import { existsSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; - -export interface Baseline { - sql: string; - fingerprint: string; - savedAt?: string; -} +import { serializeManifest, deserializeManifest, type Manifest } from './db-delta.js'; /** Root dir for baselines; `base` is injectable for tests (defaults to ~/.instawp). */ export function baselineDir(siteId: number | string, base = join(homedir(), '.instawp')): string { return join(base, 'baselines', String(siteId)); } -/** Load the stored baseline for a site, or null if none / unreadable. */ -export function loadBaseline(siteId: number | string, base?: string): Baseline | null { - const dir = baselineDir(siteId, base); - const sqlPath = join(dir, 'baseline.sql'); - const metaPath = join(dir, 'baseline.json'); - if (!existsSync(sqlPath) || !existsSync(metaPath)) return null; +/** Load the stored manifest for a site, or null if none / unreadable. */ +export function loadBaseline(siteId: number | string, base?: string): Manifest | null { + const p = join(baselineDir(siteId, base), 'manifest.json'); + if (!existsSync(p)) return null; try { - const sql = readFileSync(sqlPath, 'utf-8'); - const meta = JSON.parse(readFileSync(metaPath, 'utf-8')); - if (typeof meta?.fingerprint !== 'string') return null; - return { sql, fingerprint: meta.fingerprint, savedAt: meta.savedAt }; + return deserializeManifest(readFileSync(p, 'utf-8')); } catch { return null; } } -/** Save (overwrite) the baseline for a site. `savedAt` is passed in (no clock here). */ -export function saveBaseline(siteId: number | string, sql: string, fingerprint: string, savedAt: string, base?: string): void { +/** Save (overwrite) the manifest for a site. */ +export function saveBaseline(siteId: number | string, manifest: Manifest, base?: string): void { const dir = baselineDir(siteId, base); mkdirSync(dir, { recursive: true }); - writeFileSync(join(dir, 'baseline.sql'), sql, 'utf-8'); - writeFileSync(join(dir, 'baseline.json'), JSON.stringify({ fingerprint, savedAt }, null, 2), 'utf-8'); + writeFileSync(join(dir, 'manifest.json'), serializeManifest(manifest), 'utf-8'); } diff --git a/src/lib/db-delta.ts b/src/lib/db-delta.ts index a2e9e20..d23f84b 100644 --- a/src/lib/db-delta.ts +++ b/src/lib/db-delta.ts @@ -1,29 +1,75 @@ -// Pure engine for `db push --incremental`: parse two per-row, PK-sorted MySQL -// dumps (baseline vs current) and emit a minimal REPLACE/DELETE delta. NO I/O, -// NO DROP/CREATE — this is the data-mutating surface, so it is kept pure and -// heavily unit-tested. Falls back to a full push (mode:'full') whenever a delta -// can't be expressed safely (schema/DDL change, or a table without a single -// primary key). +// Memory-bounded engine for `db push --incremental` (#17). // -// Input requirement: dumps produced with `mysqldump --skip-extended-insert -// --order-by-primary` (one row per INSERT, one tuple per statement). Use -// hasExtendedInsert() to reject anything else before diffing. +// The naive approach (load both 181 MB dumps as strings → split → per-row Maps +// for both) blew past Node's ~4 GB heap (B3). Instead: +// - the BASELINE is stored as a compact per-row hash MANIFEST (PK → content +// hash) — not the full SQL; +// - the CURRENT dump is STREAMED line-by-line (gz decompressed on the fly), +// each row hashed and compared to the manifest; only CHANGED rows are +// materialized into the delta. +// Peak memory ≈ manifest size (tens of MB), independent of dump size. +// +// A line-oriented state machine also makes the B1 class of bug structural: only +// a line *starting* with `CREATE TABLE` is schema; "CREATE TABLE …" inside a +// single-line INSERT's row data is always a row, never DDL. +// +// Input requirement: a per-row dump (`mysqldump --skip-extended-insert +// --order-by-primary`). An extended-insert dump throws ExtendedInsertError. import { createHash } from 'node:crypto'; +export class ExtendedInsertError extends Error {} + export interface TableSchema { columns: string[]; - pkCol: string | null; // null when composite or absent → not delta-eligible - pkIndex: number; // -1 when pkCol is null + pkCol: string | null; + pkIndex: number; } -export interface DeltaResult { +export interface Manifest { + version: number; + fingerprint: string; + // single-column-PK tables: PK literal → row content hash + single: Map }>; + // composite / no-PK tables: order-independent aggregate (can only detect change) + composite: Map; +} + +export interface DiffResult { mode: 'delta' | 'full'; reason?: string; sql?: string; stats?: { tablesChanged: number; replaces: number; deletes: number }; + newManifest: Manifest; } -/** Split a VALUES tuple body (without the outer parens) into top-level fields. */ +const CORE_SUFFIXES = ['options', 'users', 'posts', 'postmeta', 'usermeta', 'comments', 'commentmeta', 'term_taxonomy', 'term_relationships', 'termmeta', 'terms', 'links']; + +/** Derive the table prefix (e.g. `wp_`) from a set of table names, or null. */ +export function prefixFromTableNames(names: Iterable): string | null { + const arr = [...names]; + for (const suf of CORE_SUFFIXES) { + for (const n of arr) if (n.endsWith(suf)) return n.slice(0, n.length - suf.length); + } + return null; +} + +/** Fast 53-bit string hash (cyrb53) — for row content change-detection, not crypto. */ +export function cyrb53(str: string, seed = 0): number { + let h1 = 0xdeadbeef ^ seed; + let h2 = 0x41c6ce57 ^ seed; + for (let i = 0; i < str.length; i++) { + const ch = str.charCodeAt(i); + h1 = Math.imul(h1 ^ ch, 2654435761); + h2 = Math.imul(h2 ^ ch, 1597334677); + } + h1 = Math.imul(h1 ^ (h1 >>> 16), 2246822507); + h1 ^= Math.imul(h2 ^ (h2 >>> 13), 3266489909); + h2 = Math.imul(h2 ^ (h2 >>> 16), 2246822507); + h2 ^= Math.imul(h1 ^ (h1 >>> 13), 3266489909); + return 4294967296 * (2097151 & h2) + (h1 >>> 0); +} + +/** Split a VALUES tuple body (without outer parens) into top-level fields. */ export function splitSqlTuple(inner: string): string[] { const out: string[] = []; let cur = ''; @@ -31,7 +77,7 @@ export function splitSqlTuple(inner: string): string[] { for (let i = 0; i < inner.length; i++) { const c = inner[i]; if (inStr) { - if (c === '\\') { cur += c + (inner[i + 1] ?? ''); i++; continue; } // backslash escape + if (c === '\\') { cur += c + (inner[i + 1] ?? ''); i++; continue; } cur += c; if (c === "'") inStr = false; continue; @@ -44,7 +90,7 @@ export function splitSqlTuple(inner: string): string[] { return out; } -/** Parse one INSERT statement into its table + tuple bodies (>1 tuple = extended insert). */ +/** Parse one INSERT statement into its table + tuple bodies (>1 tuple = extended). */ export function parseInsertStatement(stmt: string): { table: string; tuples: string[] } | null { const m = stmt.match(/^INSERT INTO\s+`([^`]+)`\s+(?:\([^)]*\)\s+)?VALUES\s*/i); if (!m) return null; @@ -76,159 +122,219 @@ export function parseInsertStatement(stmt: string): { table: string; tuples: str return { table: m[1], tuples }; } -/** True if any INSERT carries more than one tuple (i.e. NOT --skip-extended-insert). */ -export function hasExtendedInsert(sql: string): boolean { - for (const raw of sql.split('\n')) { - const line = raw.trim(); - if (!/^INSERT INTO/i.test(line)) continue; - const parsed = parseInsertStatement(line); - if (parsed && parsed.tuples.length > 1) return true; +/** Parse a single CREATE TABLE block → ordered columns + single-column PK. */ +function parseCreateBlock(block: string): { table: string; schema: TableSchema } | null { + const m = block.match(/^CREATE TABLE\s+`([^`]+)`\s*\(([\s\S]*)\n\)/i); + if (!m) return null; + const body = m[2]; + const columns: string[] = []; + for (const line of body.split('\n')) { + const cm = line.trim().match(/^`([^`]+)`/); + if (cm) columns.push(cm[1]); } - return false; -} - -/** Parse CREATE TABLE blocks → ordered columns + single-column PK (if any). */ -export function parseCreateTables(sql: string): Map { - const map = new Map(); - // Anchored to line-start (m flag): real mysqldump DDL starts at column 0, so a - // "CREATE TABLE `x` (" occurring inside a single-line INSERT's row data can't match. - const re = /^CREATE TABLE\s+`([^`]+)`\s*\(([\s\S]*?)\n\)/gim; - let m: RegExpExecArray | null; - while ((m = re.exec(sql))) { - const table = m[1]; - const body = m[2]; - const columns: string[] = []; - for (const line of body.split('\n')) { - const cm = line.trim().match(/^`([^`]+)`/); - if (cm) columns.push(cm[1]); + let pkCol: string | null = null; + const pk = body.match(/PRIMARY KEY\s*\(([^)]+)\)/i); + if (pk) { + const cols = pk[1].split(',').map((s) => s.trim().replace(/`/g, '')); + if (cols.length === 1) pkCol = cols[0]; + } + return { table: m[1], schema: { columns, pkCol, pkIndex: pkCol ? columns.indexOf(pkCol) : -1 } }; +} + +function normalizeDdl(ddl: string): string { + return ddl.replace(/AUTO_INCREMENT=\d+/gi, 'AUTO_INCREMENT=').replace(/\s+/g, ' ').trim(); +} + +function fingerprintFromBlocks(blocks: string[]): string { + return createHash('sha256').update([...blocks].sort().join('\n')).digest('hex'); +} + +type DumpEvent = + | { kind: 'schema'; table: string; ddl: string; schema: TableSchema } + | { kind: 'row'; table: string; tuple: string } + | { kind: 'extended' }; + +/** Stream a dump's lines into schema/row events (line-oriented state machine). */ +async function* parseDumpStream(lines: AsyncIterable | Iterable): AsyncGenerator { + let block: string[] | null = null; + for await (const raw of lines) { + if (block !== null) { + block.push(raw); + if (/^\)/.test(raw)) { + const ddl = block.join('\n'); + block = null; + const parsed = parseCreateBlock(ddl); + if (parsed) yield { kind: 'schema', table: parsed.table, ddl, schema: parsed.schema }; + } + continue; } - let pkCol: string | null = null; - const pk = body.match(/PRIMARY KEY\s*\(([^)]+)\)/i); - if (pk) { - const cols = pk[1].split(',').map((s) => s.trim().replace(/`/g, '')); - if (cols.length === 1) pkCol = cols[0]; + if (/^CREATE TABLE/i.test(raw)) { block = [raw]; continue; } + if (/^\s*INSERT INTO/i.test(raw)) { + const stmt = parseInsertStatement(raw.trim()); + if (!stmt) continue; + if (stmt.tuples.length !== 1) { yield { kind: 'extended' }; continue; } + yield { kind: 'row', table: stmt.table, tuple: stmt.tuples[0] }; } - map.set(table, { columns, pkCol, pkIndex: pkCol ? columns.indexOf(pkCol) : -1 }); - } - return map; -} - -/** Stable hash of the schema (normalized CREATE TABLE DDL), AUTO_INCREMENT stripped. */ -export function schemaFingerprint(sql: string): string { - const blocks: string[] = []; - // Anchored to line-start (m flag): without it, "CREATE TABLE … );" appearing - // inside row data (e.g. a post documenting SQL) matched too, lazily sweeping - // volatile content into the fingerprint → it changed on every data edit → - // --incremental always fell back to a full push. (AUTO_INCREMENT already stripped.) - const re = /^CREATE TABLE[\s\S]*?\n\)[^;]*;/gim; - let m: RegExpExecArray | null; - while ((m = re.exec(sql))) { - blocks.push(m[0].replace(/AUTO_INCREMENT=\d+/gi, 'AUTO_INCREMENT=').replace(/\s+/g, ' ').trim()); - } - blocks.sort(); - return createHash('sha256').update(blocks.join('\n')).digest('hex'); -} - -/** table → ordered list of single-row INSERT tuple bodies (ALL tables, keyable or not). */ -function parseInserts(sql: string): Map { - const map = new Map(); - for (const raw of sql.split('\n')) { - const line = raw.trim(); - if (!/^INSERT INTO/i.test(line)) continue; - const stmt = parseInsertStatement(line); - if (!stmt || stmt.tuples.length !== 1) continue; - if (!map.has(stmt.table)) map.set(stmt.table, []); - map.get(stmt.table)!.push(stmt.tuples[0]); } - return map; } -/** Index a table's rows by its single-column PK literal (tuple → keyed map). */ -function indexByPk(rows: string[], pkIndex: number): Map { - const m = new Map(); - for (const tuple of rows) m.set(splitSqlTuple(tuple)[pkIndex], tuple); - return m; +function emptyManifest(): Manifest { + return { version: 1, fingerprint: '', single: new Map(), composite: new Map() }; } -/** Order-independent equality of two row-tuple lists (rows are unique per table). */ -function sameRowSet(a: string[], b: string[]): boolean { - if (a.length !== b.length) return false; - const sa = [...a].sort(); - const sb = [...b].sort(); - for (let i = 0; i < sa.length; i++) if (sa[i] !== sb[i]) return false; - return true; +/** Build a hash manifest from a (streamed) per-row dump. Throws on extended insert. */ +export async function buildManifest(lines: AsyncIterable | Iterable): Promise { + const schemas = new Map(); + const ddlBlocks: string[] = []; + const single = new Map }>(); + const composite = new Map(); + + for await (const ev of parseDumpStream(lines)) { + if (ev.kind === 'extended') throw new ExtendedInsertError(); + if (ev.kind === 'schema') { schemas.set(ev.table, ev.schema); ddlBlocks.push(normalizeDdl(ev.ddl)); continue; } + const sch = schemas.get(ev.table); + const h = cyrb53(ev.tuple); + if (sch && sch.pkCol != null && sch.pkIndex >= 0) { + const pk = splitSqlTuple(ev.tuple)[sch.pkIndex]; + let t = single.get(ev.table); + if (!t) { t = { pkCol: sch.pkCol, pkIndex: sch.pkIndex, rows: new Map() }; single.set(ev.table, t); } + t.rows.set(pk, h); + } else { + let agg = composite.get(ev.table); + if (!agg) { agg = { count: 0, sum: 0n }; composite.set(ev.table, agg); } + agg.count++; + agg.sum += BigInt(h); + } + } + + const m = emptyManifest(); + m.fingerprint = fingerprintFromBlocks(ddlBlocks); + m.single = single; + m.composite = new Map([...composite].map(([t, a]) => [t, { count: a.count, sum: a.sum.toString() }])); + return m; } /** - * Compute a REPLACE/DELETE delta to bring the remote (== baseline) to `current`. - * Table identifiers are emitted in the REMOTE prefix (dumpPrefix → remotePrefix). - * Returns mode:'full' (with a reason) when a delta can't be safely expressed. - * - * A table WITHOUT a single-column primary key (e.g. `wp_term_relationships`, - * composite PK) can't be row-diffed. Such a table is IGNORED when its row set is - * unchanged (the common case — present on every WP site), and only forces a full - * fallback when it actually changed. (Earlier this gated unconditionally, so any - * site with a populated composite-PK table fell back to full on every push.) + * Stream the current dump and diff it against the baseline manifest. Builds the + * NEW manifest as it goes (so the caller can re-base without another pass). + * Returns mode:'full' (with the new manifest) when a delta can't be expressed + * (schema/DDL change, or a composite-PK table actually changed). */ -export function computeDelta(opts: { - baselineSql: string; - currentSql: string; - dumpPrefix: string; - remotePrefix: string; -}): DeltaResult { - const { baselineSql, currentSql, dumpPrefix, remotePrefix } = opts; - - if (schemaFingerprint(baselineSql) !== schemaFingerprint(currentSql)) { - return { mode: 'full', reason: 'schema/DDL changed since the baseline' }; - } +export async function diffAgainstManifest( + lines: AsyncIterable | Iterable, + baseline: Manifest, + opts: { dumpPrefix: string; remotePrefix: string }, +): Promise { + const { dumpPrefix, remotePrefix } = opts; + const remap = (t: string) => (t.startsWith(dumpPrefix) ? remotePrefix + t.slice(dumpPrefix.length) : t); - const schemas = parseCreateTables(currentSql); - const cur = parseInserts(currentSql); - const base = parseInserts(baselineSql); - const allTables = new Set([...cur.keys(), ...base.keys()]); + const schemas = new Map(); + const ddlBlocks: string[] = []; + // working copy of baseline single-PK rows → remaining ones = DELETEs + const remaining = new Map>(); + for (const [t, info] of baseline.single) remaining.set(t, new Map(info.rows)); + // new manifest accumulators + const newSingle = new Map }>(); + const newComposite = new Map(); - const remap = (t: string) => (t.startsWith(dumpPrefix) ? remotePrefix + t.slice(dumpPrefix.length) : t); const stmts: string[] = []; const changed = new Set(); let replaces = 0; let deletes = 0; - for (const table of allTables) { - const sch = schemas.get(table); - const curRows = cur.get(table) ?? []; - const baseRows = base.get(table) ?? []; + for await (const ev of parseDumpStream(lines)) { + if (ev.kind === 'extended') throw new ExtendedInsertError(); + if (ev.kind === 'schema') { schemas.set(ev.table, ev.schema); ddlBlocks.push(normalizeDdl(ev.ddl)); continue; } + const sch = schemas.get(ev.table); + const h = cyrb53(ev.tuple); + if (sch && sch.pkCol != null && sch.pkIndex >= 0) { + const pk = splitSqlTuple(ev.tuple)[sch.pkIndex]; + let nt = newSingle.get(ev.table); + if (!nt) { nt = { pkCol: sch.pkCol, pkIndex: sch.pkIndex, rows: new Map() }; newSingle.set(ev.table, nt); } + nt.rows.set(pk, h); - // No usable single-column PK → can't row-diff. Ignore if unchanged; else full. - if (!sch || sch.pkCol == null || sch.pkIndex < 0) { - if (!sameRowSet(baseRows, curRows)) { - return { mode: 'full', reason: `table \`${table}\` changed but has no single-column primary key` }; + const baseRows = remaining.get(ev.table); + if (baseRows && baseRows.has(pk)) { + if (baseRows.get(pk) !== h) { + stmts.push(`REPLACE INTO \`${remap(ev.table)}\` VALUES (${ev.tuple});`); + replaces++; + changed.add(ev.table); + } + baseRows.delete(pk); // seen → not a DELETE + } else { + stmts.push(`REPLACE INTO \`${remap(ev.table)}\` VALUES (${ev.tuple});`); + replaces++; + changed.add(ev.table); } - continue; + } else { + let agg = newComposite.get(ev.table); + if (!agg) { agg = { count: 0, sum: 0n }; newComposite.set(ev.table, agg); } + agg.count++; + agg.sum += BigInt(h); } + } - const remTable = remap(table); - const curMap = indexByPk(curRows, sch.pkIndex); - const baseMap = indexByPk(baseRows, sch.pkIndex); - for (const [pk, tuple] of curMap) { - const prev = baseMap.get(pk); - if (prev === undefined || prev !== tuple) { - stmts.push(`REPLACE INTO \`${remTable}\` VALUES (${tuple});`); - replaces++; - changed.add(table); - } + const newManifest = emptyManifest(); + newManifest.fingerprint = fingerprintFromBlocks(ddlBlocks); + newManifest.single = newSingle; + newManifest.composite = new Map([...newComposite].map(([t, a]) => [t, { count: a.count, sum: a.sum.toString() }])); + + if (newManifest.fingerprint !== baseline.fingerprint) { + return { mode: 'full', reason: 'schema/DDL changed since the baseline', newManifest }; + } + + // composite / no-PK tables: detect change via aggregate; can't row-diff. + const compTables = new Set([...newComposite.keys(), ...baseline.composite.keys()]); + for (const t of compTables) { + const cur = newComposite.get(t); + const curCount = cur ? cur.count : 0; + const curSum = cur ? cur.sum : 0n; + const base = baseline.composite.get(t); + const baseCount = base ? base.count : 0; + const baseSum = base ? BigInt(base.sum) : 0n; + if (curCount !== baseCount || curSum !== baseSum) { + return { mode: 'full', reason: `table \`${t}\` changed but has no single-column primary key`, newManifest }; } - for (const [pk, tuple] of baseMap) { - if (!curMap.has(pk)) { - const pkLit = splitSqlTuple(tuple)[sch.pkIndex]; - stmts.push(`DELETE FROM \`${remTable}\` WHERE \`${sch.pkCol}\`=${pkLit};`); - deletes++; - changed.add(table); - } + } + + // DELETEs: baseline single-PK rows not present in current. + for (const [t, rows] of remaining) { + if (!rows.size) continue; + const pkCol = baseline.single.get(t)?.pkCol ?? schemas.get(t)?.pkCol; + if (!pkCol) continue; + const remTable = remap(t); + for (const pk of rows.keys()) { + stmts.push(`DELETE FROM \`${remTable}\` WHERE \`${pkCol}\`=${pk};`); + deletes++; + changed.add(t); } } const stats = { tablesChanged: changed.size, replaces, deletes }; - if (!stmts.length) return { mode: 'delta', sql: '', stats }; - const sql = `SET FOREIGN_KEY_CHECKS=0;\nSET sql_mode='NO_AUTO_VALUE_ON_ZERO';\n${stmts.join('\n')}\n`; - return { mode: 'delta', sql, stats }; + const sql = stmts.length + ? `SET FOREIGN_KEY_CHECKS=0;\nSET sql_mode='NO_AUTO_VALUE_ON_ZERO';\n${stmts.join('\n')}\n` + : ''; + return { mode: 'delta', sql, stats, newManifest }; +} + +/** Serialize a manifest to a compact JSON string (Maps → arrays). */ +export function serializeManifest(m: Manifest): string { + return JSON.stringify({ + version: m.version, + fingerprint: m.fingerprint, + single: [...m.single].map(([t, info]) => [t, info.pkCol, info.pkIndex, [...info.rows]]), + composite: [...m.composite].map(([t, agg]) => [t, agg.count, agg.sum]), + }); +} + +/** Inverse of serializeManifest. */ +export function deserializeManifest(s: string): Manifest { + const o = JSON.parse(s); + const single = new Map }>( + (o.single ?? []).map((e: any) => [e[0], { pkCol: e[1], pkIndex: e[2], rows: new Map(e[3]) }]), + ); + const composite = new Map( + (o.composite ?? []).map((e: any) => [e[0], { count: e[1], sum: e[2] }]), + ); + return { version: o.version ?? 1, fingerprint: o.fingerprint ?? '', single, composite }; }