merge: 레거시 DB 순차 증분 이관 서비스 통합

This commit is contained in:
2026-08-18 12:36:02 +00:00
21 changed files with 1375 additions and 104 deletions
+2
View File
@@ -89,6 +89,8 @@ web_modules/
.env.local .env.local
.env.ci .env.ci
.env.ops .env.ops
tools/legacy-db-migration/migration-plan.json
tools/legacy-db-migration/secrets/
# parcel-bundler cache (https://parceljs.org/) # parcel-bundler cache (https://parceljs.org/)
.cache .cache
+2 -2
View File
@@ -38,8 +38,8 @@ describe('readReleaseManifest', () => {
await expect(readReleaseManifest(workspaceRoot)).resolves.toMatchObject({ await expect(readReleaseManifest(workspaceRoot)).resolves.toMatchObject({
controllerProtocol: RELEASE_CONTROLLER_PROTOCOL, controllerProtocol: RELEASE_CONTROLLER_PROTOCOL,
gatewaySchemaHead: '20260817002000_remove_profile_manage_capability', gatewaySchemaHead: '20260818000000_add_legacy_import_checkpoints',
gameSchemaHead: '20260817001000_add_dedicated_legacy_archive', gameSchemaHead: '20260818000000_add_legacy_import_checkpoints',
}); });
}); });
+59 -25
View File
@@ -4,17 +4,44 @@
`tools/legacy-db-migration` is the only supported importer. It has no HTTP `tools/legacy-db-migration` is the only supported importer. It has no HTTP
entrypoint, defaults to a read-only dry-run, and requires `--apply` for target entrypoint, defaults to a read-only dry-run, and requires `--apply` for target
writes. Database URLs are accepted only through environment variables. writes. The normal `run-plan` command reads a mode-0600 JSON file containing
PostgreSQL advisory locks serialize an apply per profile, and every target row structured MariaDB host/database/user settings and a password, password-file
uses a stable legacy key with `ON CONFLICT`, so an interrupted run is reference or password environment name. PostgreSQL URLs remain in environment
repeatable. variables; no URL or password is accepted as a command-line value or returned
in JSON output. PostgreSQL advisory locks serialize an apply per source/profile,
and every target row uses a stable legacy key with `ON CONFLICT`, so an
interrupted run is repeatable.
Gateway apply is one PostgreSQL transaction. A game apply records a Gateway apply is one PostgreSQL transaction and records `legacy_import_run`.
`legacy_archive.import_run`: archive and current-user projection writes commit A game apply records `legacy_archive.import_run`: archive and current-user
together with `COMPLETED`, while a rollback leaves a `FAILED` run record. A projection writes commit together with `COMPLETED`, while a rollback leaves a
repeat import updates archive-owned rows but does not replace a live Gateway `FAILED` run record. Their checkpoint updates commit in the same transaction as
account's password, reset status, login/display identity, OAuth connection, the imported rows. A repeat import updates archive-owned rows but does not
roles, sanctions, consent, icon or login timestamps. replace a live Gateway account's password, reset status, login/display identity,
OAuth connection, roles, sanctions, consent, icon or login timestamps.
### Full and incremental execution
`run-plan` first connects to every configured MariaDB and PostgreSQL target and
checks that the checkpoint migrations exist. Only after all stages pass does it
run Gateway, then `che,kwe,pwe,twe,nya,pya,hwe` in that order, skipping disabled
profiles. Dry-run and apply use the same transformations and return per-table
`progress` with strategy, start cursor, end cursor and processed count.
The initial apply uses `--mode full`. A later `--mode incremental` requires the
same `sourceSet` and a completed checkpoint for every append-only table.
`member_log`, `hall`, `ng_old_generals`, `ng_old_nations`, `emperior`,
`inheritance_result`, `user_record` and `ng_history` continue strictly after the
stored primary-key cursor. Mutable `member`, root/game `storage`, `system`, bans
and `ng_games` are rescanned and upserted. The source fingerprint includes
protocol, host, port, database, user and non-secret connection options but not
the password, so password rotation is safe while an accidental database switch
is rejected. A regressed maximum ID is also rejected.
This is an append-plus-rescan importer, not bidirectional replication. It does
not copy source deletions and cannot discover an in-place edit to a completed
append-only row behind the high-water mark. Such a source requires a reviewed
full re-import and count/hash investigation.
The source of truth for eligibility is the checked ref schema, not every table The source of truth for eligibility is the checked ref schema, not every table
that happens to exist in a dump. Tables outside that schema remain only in the that happens to exist in a dump. Tables outside that schema remain only in the
@@ -165,21 +192,28 @@ they are never merged into the current rankings or current dynasty list.
## Cutover procedure ## Cutover procedure
1. Keep the original compressed dumps immutable and restore each source to a 1. Keep the original compressed dumps immutable and restore each source to a
private MariaDB instance. private MariaDB instance. Record checksums and the snapshot boundary.
2. Deploy the gateway and game Prisma migrations to empty staging databases. 2. Copy `migration-plan.example.json` to the ignored `migration-plan.json`, give
3. Run gateway and each non-empty official profile without `--apply`; archive the JSON it and every password file mode 0600, and enter one stable `sourceSet`.
counts and excluded-table reasons. 3. Deploy the gateway and game Prisma migrations to staging, then run
4. Compare source counts, malformed JSON checks and duplicate natural-key `check-plan`. Fix every connection, source-table and target-migration failure
before continuing.
4. Run `run-plan --mode full` without `--apply`; archive the redacted JSON counts,
excluded-table reasons and source-format summary.
5. Compare source counts, malformed JSON checks and duplicate natural-key
counts. Stop on unexplained drift. counts. Stop on unexplained drift.
5. Put the affected target in maintenance mode, take a PostgreSQL backup, then 6. Put the affected target in maintenance mode, take a PostgreSQL backup, then
run the same commands with `--apply`. run the identical plan with `--mode full --apply`.
6. Repeat each apply. Counts must remain unchanged; verify the newest 7. Verify every Gateway/game run is `COMPLETED`, checkpoints equal the source
`legacy_archive.import_run` is `COMPLETED` and current Gateway credentials maxima, target counts match, and current Gateway credentials are unchanged.
are unchanged. 8. Verify valid/invalid Kakao-ID classification, password-reset-required rows,
7. Verify valid/invalid Kakao-ID classification, password-reset-required rows,
Kakao password setup, CLI fallback, archive ownership, canonical source Kakao password setup, CLI fallback, archive ownership, canonical source
format counts, opening dates, `/past-plays`, foreign-owner denial, legacy format counts, opening dates, `/past-plays`, foreign-owner denial, legacy Hall
Hall and legacy Dynasty source switches. and legacy Dynasty source switches.
8. Retain the MariaDB dumps as rollback evidence. Rollback restores the 9. For the later snapshot, retain the same `sourceSet`, run `check-plan`, then
`run-plan --mode incremental` and review every start/end cursor before adding
`--apply`. Repeat once: append-table processed counts must be zero and target
row counts must remain unchanged.
10. Retain both MariaDB snapshots and plan-output evidence. Rollback restores the
pre-cutover PostgreSQL backup; it does not reverse individual importer pre-cutover PostgreSQL backup; it does not reverse individual importer
upserts. upserts or edit checkpoint rows by hand.
@@ -0,0 +1,29 @@
CREATE TABLE "legacy_import_run" (
"id" BIGSERIAL PRIMARY KEY,
"source_key" TEXT NOT NULL,
"source_fingerprint" CHAR(64) NOT NULL,
"mode" TEXT NOT NULL,
"status" TEXT NOT NULL,
"started_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
"finished_at" TIMESTAMPTZ,
"counts" JSONB NOT NULL DEFAULT '{}'::jsonb,
"progress" JSONB NOT NULL DEFAULT '{}'::jsonb,
"error" TEXT,
CONSTRAINT "legacy_import_run_mode_check" CHECK ("mode" IN ('full', 'incremental')),
CONSTRAINT "legacy_import_run_status_check" CHECK ("status" IN ('RUNNING', 'COMPLETED', 'FAILED')),
CONSTRAINT "legacy_import_run_fingerprint_check" CHECK ("source_fingerprint" ~ '^[a-f0-9]{64}$')
);
CREATE INDEX "legacy_import_run_source_started"
ON "legacy_import_run" ("source_key", "started_at" DESC);
CREATE TABLE "legacy_import_checkpoint" (
"source_key" TEXT NOT NULL,
"source_fingerprint" CHAR(64) NOT NULL,
"source_table" TEXT NOT NULL,
"last_legacy_id" BIGINT NOT NULL,
"import_run_id" BIGINT NOT NULL REFERENCES "legacy_import_run" ("id"),
"updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY ("source_key", "source_table"),
CONSTRAINT "legacy_import_checkpoint_fingerprint_check" CHECK ("source_fingerprint" ~ '^[a-f0-9]{64}$')
);
@@ -0,0 +1,28 @@
ALTER TABLE "legacy_archive"."import_run"
ADD COLUMN "source_key" TEXT,
ADD COLUMN "source_fingerprint" CHAR(64),
ADD COLUMN "mode" TEXT NOT NULL DEFAULT 'full',
ADD COLUMN "progress" JSONB NOT NULL DEFAULT '{}'::jsonb;
ALTER TABLE "legacy_archive"."import_run"
ADD CONSTRAINT "legacy_archive_import_run_mode_check" CHECK ("mode" IN ('full', 'incremental')),
ADD CONSTRAINT "legacy_archive_import_run_fingerprint_check"
CHECK ("source_fingerprint" IS NULL OR "source_fingerprint" ~ '^[a-f0-9]{64}$');
CREATE INDEX "legacy_archive_import_run_source_started"
ON "legacy_archive"."import_run" ("source_profile", "source_key", "started_at" DESC);
CREATE TABLE "legacy_archive"."import_checkpoint" (
"source_profile" TEXT NOT NULL,
"source_key" TEXT NOT NULL,
"source_fingerprint" CHAR(64) NOT NULL,
"source_table" TEXT NOT NULL,
"last_legacy_id" BIGINT NOT NULL,
"import_run_id" BIGINT NOT NULL REFERENCES "legacy_archive"."import_run" ("id"),
"updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY ("source_profile", "source_key", "source_table"),
CONSTRAINT "legacy_archive_import_checkpoint_profile_check"
CHECK ("source_profile" IN ('che', 'kwe', 'pwe', 'twe', 'nya', 'pya', 'hwe')),
CONSTRAINT "legacy_archive_import_checkpoint_fingerprint_check"
CHECK ("source_fingerprint" ~ '^[a-f0-9]{64}$')
);
+2 -2
View File
@@ -1,7 +1,7 @@
{ {
"formatVersion": 1, "formatVersion": 1,
"controllerProtocol": 2, "controllerProtocol": 2,
"gatewaySchemaHead": "20260817002000_remove_profile_manage_capability", "gatewaySchemaHead": "20260818000000_add_legacy_import_checkpoints",
"gameSchemaHead": "20260817001000_add_dedicated_legacy_archive", "gameSchemaHead": "20260818000000_add_legacy_import_checkpoints",
"components": ["gateway-api", "gateway-frontend", "release-controller", "game-api", "game-engine", "game-frontend"] "components": ["gateway-api", "gateway-frontend", "release-controller", "game-api", "game-engine", "game-frontend"]
} }
+70 -5
View File
@@ -1,14 +1,17 @@
# Legacy DB migration CLI # Legacy DB migration CLI
This package migrates the long-lived parts of a restored ref MariaDB database This package migrates the long-lived parts of a restored or still-readable ref
into the core2026 PostgreSQL schemas. It is CLI-only; no HTTP or administrator MariaDB database into the core2026 PostgreSQL schemas. It is CLI-only; no HTTP
route invokes it. or administrator route invokes it. `run-plan` is the normal operator entrypoint:
it validates every configured connection first, then runs Gateway followed by
the enabled game profiles in the official order.
The default mode is a read-only dry-run. `--apply` is required before any target The default mode is a read-only dry-run. `--apply` is required before any target
write. PostgreSQL advisory locks prevent two applies for the same target. write. PostgreSQL advisory locks prevent two applies for the same target.
Gateway writes are transactional. Game archive writes and their completed Gateway writes are transactional. Game archive writes and their completed
`legacy_archive.import_run` record are transactional. Stable legacy keys make `legacy_archive.import_run` record are transactional. Both paths record an
completed or interrupted runs repeatable. import run and durable per-table checkpoints. Stable legacy keys make completed
or interrupted runs repeatable.
## Source restore ## Source restore
@@ -28,6 +31,64 @@ keeps the original dump as the recovery source.
Database URLs belong in a Git-ignored environment file or injected process Database URLs belong in a Git-ignored environment file or injected process
environment. They are deliberately not accepted as command-line flags. environment. They are deliberately not accepted as command-line flags.
## Ordered migration plan
Copy `migration-plan.example.json` to the Git-ignored `migration-plan.json`.
Set its mode to 0600, then enter the MariaDB host, port, database and user for
Gateway and each game profile. A password can come from a separate mode-0600
file (recommended), an environment variable, or directly from the mode-0600
plan. Target PostgreSQL URLs remain in the named environment variables.
```sh
mkdir -p tools/legacy-db-migration/secrets
chmod 700 tools/legacy-db-migration/secrets
cp tools/legacy-db-migration/migration-plan.example.json \
tools/legacy-db-migration/migration-plan.json
chmod 600 tools/legacy-db-migration/migration-plan.json
chmod 600 tools/legacy-db-migration/secrets/*
pnpm migrate:legacy -- check-plan \
--config tools/legacy-db-migration/migration-plan.json
pnpm migrate:legacy -- run-plan \
--config tools/legacy-db-migration/migration-plan.json --mode full
pnpm migrate:legacy -- run-plan \
--config tools/legacy-db-migration/migration-plan.json --mode full --apply
```
`check-plan` opens every source and target without writing. `run-plan` also
preflights every stage before the first import, is a dry-run without `--apply`,
and stops at the first failed stage. Completed earlier stages remain committed;
rerunning is safe because the Gateway and each profile have independent locks,
transactions and run records. The JSON output never includes a connection URL
or password.
For a later delta, keep the same `sourceSet`, connection identity and source
databases, restore or expose the newer snapshot, then run:
```sh
pnpm migrate:legacy -- run-plan \
--config tools/legacy-db-migration/migration-plan.json --mode incremental
pnpm migrate:legacy -- run-plan \
--config tools/legacy-db-migration/migration-plan.json --mode incremental --apply
```
Incremental mode refuses to start without checkpoints from a completed full
apply. It also refuses a changed host/database/user identity or a source table
whose maximum ID moved behind its checkpoint. Password rotation does not change
the source fingerprint.
| Source data | Incremental policy |
| --------------------------------------------- | --------------------------------------------------------------------- |
| `member_log` | Read only IDs after the committed high-water mark. |
| game archive/event-history tables | Read only IDs after the profile checkpoint. |
| `member`, root/game `storage`, `system`, bans | Rescan and idempotently upsert because old rows are mutable. |
| `ng_games` | Rescan because a season row can gain its final winner after creation. |
The append policy assumes Ref primary keys are never reused and completed
archive rows are immutable. Incremental mode does not mirror source deletions.
If either assumption is false, take a new reviewed backup and run full mode;
do not edit checkpoint rows by hand.
## Commands ## Commands
```sh ```sh
@@ -45,6 +106,10 @@ the selected current profile schema. Accepted profiles are
`che,kwe,pwe,twe,nya,pya,hwe`; run them separately against the same PostgreSQL `che,kwe,pwe,twe,nya,pya,hwe`; run them separately against the same PostgreSQL
database. database.
The individual commands also accept `--mode incremental` and `--source-key`.
Use the ordered plan for production so every configured connection is checked
before the Gateway stage starts.
### Isolated current-season comparison fixture ### Isolated current-season comparison fixture
`current-season-fixture` is separate from the long-lived archive migration. It `current-season-fixture` is separate from the long-lived archive migration. It
@@ -0,0 +1,41 @@
{
"version": 1,
"sourceSet": "ref-production-cutover",
"gateway": {
"source": {
"host": "mysql.internal.example",
"port": 3306,
"database": "sammo_root",
"user": "sammo_migration",
"passwordFile": "./secrets/mysql-root-password",
"tls": true
},
"targetUrlEnv": "GATEWAY_DATABASE_URL"
},
"profiles": [
{
"profile": "che",
"source": {
"host": "mysql.internal.example",
"port": 3306,
"database": "sammo_che",
"user": "sammo_migration",
"passwordFile": "./secrets/mysql-che-password",
"tls": true
},
"targetUrlEnv": "CHE_GAME_DATABASE_URL"
},
{
"profile": "hwe",
"enabled": false,
"source": {
"host": "mysql.internal.example",
"port": 3306,
"database": "sammo_hwe",
"user": "sammo_migration",
"passwordEnv": "LEGACY_HWE_DATABASE_PASSWORD"
},
"targetUrlEnv": "HWE_GAME_DATABASE_URL"
}
]
}
+66 -11
View File
@@ -9,8 +9,11 @@ import { isLegacyArchiveProfile, LEGACY_ARCHIVE_PROFILES, migrateGame } from './
import { migrateGateway } from './gateway.js'; import { migrateGateway } from './gateway.js';
import { hashPasswordForReset } from './password.js'; import { hashPasswordForReset } from './password.js';
import { migrateCurrentSeasonFixture } from './currentSeason.js'; import { migrateCurrentSeasonFixture } from './currentSeason.js';
import { loadMigrationPlan } from './config.js';
import { fingerprintMariaConnection, type MigrationMode } from './incremental.js';
import { checkMigrationPlan, runMigrationPlan } from './plan.js';
type Command = 'gateway' | 'game' | 'current-season-fixture' | 'reset-password'; type Command = 'gateway' | 'game' | 'current-season-fixture' | 'reset-password' | 'check-plan' | 'run-plan';
interface CliOptions { interface CliOptions {
command: Command; command: Command;
@@ -22,11 +25,17 @@ interface CliOptions {
expectedYear?: number; expectedYear?: number;
expectedMonth?: number; expectedMonth?: number;
replaceCurrentSeason: boolean; replaceCurrentSeason: boolean;
config?: string;
mode: MigrationMode;
sourceKey?: string;
} }
const usage = `Usage: const usage = `Usage:
pnpm --filter @sammo-ts/legacy-db-migration migrate gateway [--apply] pnpm --filter @sammo-ts/legacy-db-migration migrate gateway [--apply]
pnpm --filter @sammo-ts/legacy-db-migration migrate game --profile <profile> [--apply] pnpm --filter @sammo-ts/legacy-db-migration migrate game --profile <profile> [--apply]
pnpm --filter @sammo-ts/legacy-db-migration migrate check-plan --config <secure-plan.json>
pnpm --filter @sammo-ts/legacy-db-migration migrate run-plan --config <secure-plan.json> \
[--mode full|incremental] [--apply]
pnpm --filter @sammo-ts/legacy-db-migration migrate current-season-fixture --profile <profile> \ pnpm --filter @sammo-ts/legacy-db-migration migrate current-season-fixture --profile <profile> \
--expected-scenario <id> --expected-year <year> --expected-month <month> \ --expected-scenario <id> --expected-year <year> --expected-month <month> \
[--replace-current-season --apply] [--replace-current-season --apply]
@@ -47,11 +56,13 @@ const parseArguments = (argv: readonly string[]): CliOptions => {
command !== 'gateway' && command !== 'gateway' &&
command !== 'game' && command !== 'game' &&
command !== 'current-season-fixture' && command !== 'current-season-fixture' &&
command !== 'reset-password' command !== 'reset-password' &&
command !== 'check-plan' &&
command !== 'run-plan'
) { ) {
throw new Error(usage); throw new Error(usage);
} }
const options: CliOptions = { command, apply: false, replaceCurrentSeason: false }; const options: CliOptions = { command, apply: false, replaceCurrentSeason: false, mode: 'full' };
for (let index = 1; index < argv.length; index += 1) { for (let index = 1; index < argv.length; index += 1) {
const argument = argv[index]; const argument = argv[index];
if (argument === '--apply') { if (argument === '--apply') {
@@ -68,6 +79,15 @@ const parseArguments = (argv: readonly string[]): CliOptions => {
} }
if (argument === '--profile') { if (argument === '--profile') {
options.profile = next; options.profile = next;
} else if (argument === '--config') {
options.config = next;
} else if (argument === '--mode') {
if (next !== 'full' && next !== 'incremental') {
throw new Error(`--mode must be full or incremental\n\n${usage}`);
}
options.mode = next;
} else if (argument === '--source-key') {
options.sourceKey = next;
} else if (argument === '--login-id') { } else if (argument === '--login-id') {
options.loginId = next; options.loginId = next;
} else if (argument === '--password-file') { } else if (argument === '--password-file') {
@@ -94,6 +114,9 @@ const requireEnvironment = (name: string): string => {
return value; return value;
}; };
const resolveInvocationPath = (value: string): string =>
path.resolve(process.env.INIT_CWD?.trim() || process.cwd(), value);
const resetPassword = async (options: CliOptions): Promise<Record<string, unknown>> => { const resetPassword = async (options: CliOptions): Promise<Record<string, unknown>> => {
if (!options.apply) { if (!options.apply) {
throw new Error('reset-password requires --apply'); throw new Error('reset-password requires --apply');
@@ -101,7 +124,7 @@ const resetPassword = async (options: CliOptions): Promise<Record<string, unknow
if (!options.loginId || !options.passwordFile) { if (!options.loginId || !options.passwordFile) {
throw new Error(`reset-password requires --login-id and --password-file\n\n${usage}`); throw new Error(`reset-password requires --login-id and --password-file\n\n${usage}`);
} }
const passwordPath = path.resolve(options.passwordFile); const passwordPath = resolveInvocationPath(options.passwordFile);
const passwordStat = await stat(passwordPath); const passwordStat = await stat(passwordPath);
if ((passwordStat.mode & 0o077) !== 0) { if ((passwordStat.mode & 0o077) !== 0) {
throw new Error('Password file must not be readable or writable by group/other (expected mode 0600)'); throw new Error('Password file must not be readable or writable by group/other (expected mode 0600)');
@@ -136,7 +159,23 @@ const resetPassword = async (options: CliOptions): Promise<Record<string, unknow
}; };
const run = async (): Promise<void> => { const run = async (): Promise<void> => {
const options = parseArguments(process.argv.slice(2)); const argumentsAfterScript = process.argv.slice(2);
if (argumentsAfterScript[0] === '--') argumentsAfterScript.shift();
if (argumentsAfterScript[0] === '--help' || argumentsAfterScript[0] === '-h') {
console.log(usage);
return;
}
const options = parseArguments(argumentsAfterScript);
if (options.command === 'check-plan' || options.command === 'run-plan') {
if (!options.config) throw new Error(`${options.command} requires --config\n\n${usage}`);
const plan = await loadMigrationPlan(resolveInvocationPath(options.config));
const result =
options.command === 'check-plan'
? await checkMigrationPlan(plan)
: await runMigrationPlan(plan, options.mode, options.apply);
console.log(JSON.stringify(result, null, 2));
return;
}
if (options.command === 'reset-password') { if (options.command === 'reset-password') {
console.log(JSON.stringify(await resetPassword(options), null, 2)); console.log(JSON.stringify(await resetPassword(options), null, 2));
return; return;
@@ -144,14 +183,21 @@ const run = async (): Promise<void> => {
const migratedAt = new Date(); const migratedAt = new Date();
if (options.command === 'gateway') { if (options.command === 'gateway') {
const source = createMariaPool(requireEnvironment('LEGACY_ROOT_DATABASE_URL')); const sourceUrl = requireEnvironment('LEGACY_ROOT_DATABASE_URL');
const source = createMariaPool(sourceUrl);
const targetUrl = process.env.GATEWAY_DATABASE_URL?.trim(); const targetUrl = process.env.GATEWAY_DATABASE_URL?.trim();
if (options.apply && !targetUrl) { if (options.apply && !targetUrl) {
throw new Error('GATEWAY_DATABASE_URL is required with --apply'); throw new Error('GATEWAY_DATABASE_URL is required with --apply');
} }
const target = targetUrl ? createPostgresPool(targetUrl) : null; const target = targetUrl ? createPostgresPool(targetUrl) : null;
try { try {
const summary = await migrateGateway(source, target, options.apply, migratedAt); const summary = await migrateGateway(source, target, options.apply, migratedAt, {
mode: options.mode,
source: {
key: options.sourceKey ?? process.env.LEGACY_SOURCE_KEY?.trim() ?? 'legacy-root',
fingerprint: fingerprintMariaConnection(sourceUrl),
},
});
console.log(JSON.stringify(summary, null, 2)); console.log(JSON.stringify(summary, null, 2));
} finally { } finally {
await source.end(); await source.end();
@@ -166,9 +212,10 @@ const run = async (): Promise<void> => {
if (options.command === 'game' && !isLegacyArchiveProfile(options.profile)) { if (options.command === 'game' && !isLegacyArchiveProfile(options.profile)) {
throw new Error(`game requires --profile ${LEGACY_ARCHIVE_PROFILES.join('|')}\n\n${usage}`); throw new Error(`game requires --profile ${LEGACY_ARCHIVE_PROFILES.join('|')}\n\n${usage}`);
} }
const source = createMariaPool(requireEnvironment('LEGACY_GAME_DATABASE_URL')); const sourceUrl = requireEnvironment('LEGACY_GAME_DATABASE_URL');
const source = createMariaPool(sourceUrl);
const target = const target =
options.apply || options.command === 'current-season-fixture' options.apply || options.mode === 'incremental' || options.command === 'current-season-fixture'
? createPostgresPool(requireEnvironment('GAME_DATABASE_URL')) ? createPostgresPool(requireEnvironment('GAME_DATABASE_URL'))
: null; : null;
try { try {
@@ -199,7 +246,13 @@ const run = async (): Promise<void> => {
console.log(JSON.stringify(summary, null, 2)); console.log(JSON.stringify(summary, null, 2));
return; return;
} }
const summary = await migrateGame(source, target, options.apply, options.profile); const summary = await migrateGame(source, target, options.apply, options.profile, {
mode: options.mode,
source: {
key: options.sourceKey ?? process.env.LEGACY_SOURCE_KEY?.trim() ?? `legacy-game-${options.profile}`,
fingerprint: fingerprintMariaConnection(sourceUrl),
},
});
console.log(JSON.stringify(summary, null, 2)); console.log(JSON.stringify(summary, null, 2));
} finally { } finally {
await source.end(); await source.end();
@@ -208,7 +261,9 @@ const run = async (): Promise<void> => {
}; };
run().catch((error: unknown) => { run().catch((error: unknown) => {
const message = error instanceof Error ? error.message : String(error); const message = (error instanceof Error ? error.message : String(error))
.replace(/((?:mariadb|mysql|postgres(?:ql)?):\/\/[^:\s/@]+:)[^@\s/]+@/giu, '$1***@')
.replace(/([?&](?:pass(?:word)?|secret|token)=)[^&\s]+/giu, '$1***');
console.error(`[legacy-db-migration] ${message}`); console.error(`[legacy-db-migration] ${message}`);
process.exitCode = 1; process.exitCode = 1;
}); });
+219
View File
@@ -0,0 +1,219 @@
import { constants } from 'node:fs';
import { lstat, open } from 'node:fs/promises';
import { isIP } from 'node:net';
import path from 'node:path';
import { isLegacyArchiveProfile, LEGACY_ARCHIVE_PROFILES, type LegacyArchiveProfile } from './game.js';
import { fingerprintMariaConnection, type MigrationSourceIdentity } from './incremental.js';
export interface ResolvedMigrationStage {
kind: 'gateway' | 'game';
name: string;
profile?: LegacyArchiveProfile;
sourceUrl: string;
targetUrl: string;
sourceIdentity: MigrationSourceIdentity;
}
export interface ResolvedMigrationPlan {
sourceSet: string;
stages: ResolvedMigrationStage[];
}
const SAFE_KEY = /^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,95}$/u;
const ENV_NAME = /^[A-Z][A-Z0-9_]{1,127}$/u;
const assertSecureRegularFile = async (filePath: string, label: string): Promise<void> => {
const info = await lstat(filePath);
if (!info.isFile() || info.isSymbolicLink()) {
throw new Error(`${label} must be a regular file and not a symbolic link`);
}
if ((info.mode & 0o077) !== 0) {
throw new Error(`${label} must not be readable or writable by group/other (expected mode 0600)`);
}
};
const readSecureText = async (filePath: string, label: string): Promise<string> => {
await assertSecureRegularFile(filePath, label);
const handle = await open(filePath, constants.O_RDONLY | constants.O_NOFOLLOW);
try {
return await handle.readFile('utf8');
} finally {
await handle.close();
}
};
const asRecord = (value: unknown, label: string): Record<string, unknown> => {
if (value === null || Array.isArray(value) || typeof value !== 'object') {
throw new Error(`${label} must be an object`);
}
return value as Record<string, unknown>;
};
const requiredString = (record: Record<string, unknown>, key: string, label: string): string => {
const value = record[key];
if (typeof value !== 'string' || !value.trim()) {
throw new Error(`${label}.${key} must be a non-empty string`);
}
return value.trim();
};
const rejectUnknownKeys = (record: Record<string, unknown>, allowed: readonly string[], label: string): void => {
const unknown = Object.keys(record).filter((key) => !allowed.includes(key));
if (unknown.length) {
throw new Error(`${label} has unknown keys: ${unknown.join(', ')}`);
}
};
const resolvePassword = async (
source: Record<string, unknown>,
configDirectory: string,
label: string
): Promise<string> => {
const configured = ['password', 'passwordEnv', 'passwordFile'].filter(
(key) => typeof source[key] === 'string' && Boolean(String(source[key]).trim())
);
if (configured.length !== 1) {
throw new Error(`${label} must configure exactly one of password, passwordEnv, or passwordFile`);
}
if (configured[0] === 'password') {
return requiredString(source, 'password', label);
}
if (configured[0] === 'passwordEnv') {
const environmentName = requiredString(source, 'passwordEnv', label);
if (!ENV_NAME.test(environmentName)) throw new Error(`${label}.passwordEnv is not a safe environment name`);
const value = process.env[environmentName];
if (!value) throw new Error(`${environmentName} is required by ${label}`);
return value;
}
const configuredPath = requiredString(source, 'passwordFile', label);
const passwordPath = path.resolve(configDirectory, configuredPath);
const value = (await readSecureText(passwordPath, `${label}.passwordFile`)).replace(/\r?\n$/u, '');
if (!value) throw new Error(`${label}.passwordFile is empty`);
return value;
};
const resolveSource = async (value: unknown, configDirectory: string, label: string): Promise<string> => {
const source = asRecord(value, label);
rejectUnknownKeys(
source,
['host', 'port', 'database', 'user', 'password', 'passwordEnv', 'passwordFile', 'tls'],
label
);
const host = requiredString(source, 'host', label);
const database = requiredString(source, 'database', label);
const user = requiredString(source, 'user', label);
const port = source.port === undefined ? 3306 : Number(source.port);
if (!Number.isSafeInteger(port) || port < 1 || port > 65535) {
throw new Error(`${label}.port must be an integer from 1 through 65535`);
}
const dnsName =
host.length <= 253 && host.split('.').every((part) => /^(?!-)[a-zA-Z0-9-]{1,63}(?<!-)$/u.test(part));
if (!isIP(host) && !dnsName) throw new Error(`${label}.host must be an IP address or DNS name`);
if (!/^[a-zA-Z0-9_$.-]{1,64}$/u.test(database)) {
throw new Error(`${label}.database must be a safe MariaDB database name`);
}
if (source.tls !== undefined && typeof source.tls !== 'boolean') {
throw new Error(`${label}.tls must be a boolean`);
}
const password = await resolvePassword(source, configDirectory, label);
const url = new URL('mariadb://localhost');
url.hostname = host;
url.port = String(port);
url.username = user;
url.password = password;
url.pathname = `/${database}`;
if (source.tls) url.searchParams.set('ssl', 'true');
return url.toString();
};
const resolveTargetUrl = (record: Record<string, unknown>, label: string): string => {
const environmentName = requiredString(record, 'targetUrlEnv', label);
if (!ENV_NAME.test(environmentName)) throw new Error(`${label}.targetUrlEnv is not a safe environment name`);
const value = process.env[environmentName]?.trim();
if (!value) throw new Error(`${environmentName} is required by ${label}`);
return value;
};
const parseStage = (value: unknown, label: string): Record<string, unknown> => {
const record = asRecord(value, label);
rejectUnknownKeys(record, ['source', 'targetUrlEnv', 'profile', 'enabled'], label);
if (!('source' in record)) throw new Error(`${label}.source is required`);
return record;
};
export const loadMigrationPlan = async (configPathInput: string): Promise<ResolvedMigrationPlan> => {
const configPath = path.resolve(configPathInput);
const rawText = await readSecureText(configPath, 'Migration config');
let parsed: unknown;
try {
parsed = JSON.parse(rawText);
} catch (error) {
throw new Error('Migration config is not valid JSON', { cause: error });
}
const root = asRecord(parsed, 'Migration config');
rejectUnknownKeys(root, ['version', 'sourceSet', 'gateway', 'profiles'], 'Migration config');
if (root.version !== 1) throw new Error('Migration config.version must be 1');
const sourceSet = requiredString(root, 'sourceSet', 'Migration config');
if (!SAFE_KEY.test(sourceSet)) throw new Error('Migration config.sourceSet must use safe characters');
const configDirectory = path.dirname(configPath);
const stages: ResolvedMigrationStage[] = [];
if (root.gateway !== undefined) {
const gateway = parseStage(root.gateway, 'gateway');
const sourceUrl = await resolveSource(gateway.source, configDirectory, 'gateway.source');
stages.push({
kind: 'gateway',
name: 'gateway',
sourceUrl,
targetUrl: resolveTargetUrl(gateway, 'gateway'),
sourceIdentity: {
key: `${sourceSet}:gateway`,
fingerprint: fingerprintMariaConnection(sourceUrl),
},
});
}
const profiles = root.profiles === undefined ? [] : root.profiles;
if (!Array.isArray(profiles)) throw new Error('Migration config.profiles must be an array');
const seen = new Set<string>();
const profileStages = new Map<LegacyArchiveProfile, ResolvedMigrationStage>();
for (const [index, value] of profiles.entries()) {
const label = `profiles[${index}]`;
const profileConfig = parseStage(value, label);
if (profileConfig.enabled === false) continue;
if (profileConfig.enabled !== undefined && profileConfig.enabled !== true) {
throw new Error(`${label}.enabled must be a boolean`);
}
const profile = requiredString(profileConfig, 'profile', label);
if (!isLegacyArchiveProfile(profile)) {
throw new Error(`${label}.profile must be one of ${LEGACY_ARCHIVE_PROFILES.join(', ')}`);
}
if (seen.has(profile)) throw new Error(`Duplicate profile in migration config: ${profile}`);
seen.add(profile);
const sourceUrl = await resolveSource(profileConfig.source, configDirectory, `${label}.source`);
profileStages.set(profile, {
kind: 'game',
name: profile,
profile,
sourceUrl,
targetUrl: resolveTargetUrl(profileConfig, label),
sourceIdentity: {
key: `${sourceSet}:${profile}`,
fingerprint: fingerprintMariaConnection(sourceUrl),
},
});
}
for (const profile of LEGACY_ARCHIVE_PROFILES) {
const stage = profileStages.get(profile);
if (stage) stages.push(stage);
}
if (!stages.length) throw new Error('Migration config has no enabled stages');
return { sourceSet, stages };
};
export const readPasswordFileForReset = async (passwordPath: string): Promise<string> => {
const value = (await readSecureText(path.resolve(passwordPath), 'Password file')).replace(/\r?\n$/u, '');
if (!value) throw new Error('Password file is empty');
return value;
};
+12 -2
View File
@@ -68,12 +68,13 @@ export const paginateSource = async function* (
pool: MariaPool, pool: MariaPool,
table: string, table: string,
idColumn: string, idColumn: string,
batchSize: number batchSize: number,
afterId: bigint = -1n
): AsyncGenerator<SourceRow[]> { ): AsyncGenerator<SourceRow[]> {
if (!MARIA_IDENTIFIER.test(table) || !MARIA_IDENTIFIER.test(idColumn)) { if (!MARIA_IDENTIFIER.test(table) || !MARIA_IDENTIFIER.test(idColumn)) {
throw new Error('Unsafe MariaDB table or ID column'); throw new Error('Unsafe MariaDB table or ID column');
} }
let lastId = -1n; let lastId = afterId;
for (;;) { for (;;) {
const rows = await querySource( const rows = await querySource(
pool, pool,
@@ -88,6 +89,15 @@ export const paginateSource = async function* (
} }
}; };
export const sourceMaxId = async (pool: MariaPool, table: string, idColumn: string): Promise<bigint | null> => {
if (!MARIA_IDENTIFIER.test(table) || !MARIA_IDENTIFIER.test(idColumn)) {
throw new Error('Unsafe MariaDB table or ID column');
}
const rows = await querySource(pool, `SELECT MAX(\`${idColumn}\`) AS max_id FROM \`${table}\``);
const value = rows[0]?.max_id;
return value === null || value === undefined ? null : toBigInt(value, `${table}.${idColumn}`);
};
const targetValue = (value: unknown): unknown => { const targetValue = (value: unknown): unknown => {
if (value instanceof JsonParameter) { if (value instanceof JsonParameter) {
return JSON.stringify(value.value); return JSON.stringify(value.value);
+182 -38
View File
@@ -16,6 +16,7 @@ export { isLegacyArchiveProfile, LEGACY_ARCHIVE_PROFILES, type LegacyArchiveProf
import { import {
paginateSource, paginateSource,
jsonParameter, jsonParameter,
sourceMaxId,
toDate, toDate,
toFloat, toFloat,
toNullableDate, toNullableDate,
@@ -28,6 +29,15 @@ import {
type TargetRow, type TargetRow,
} from './db.js'; } from './db.js';
import type { MigrationSummary } from './gateway.js'; import type { MigrationSummary } from './gateway.js';
import {
defaultExecutionOptions,
loadCheckpoint,
requireIncrementalCheckpoint,
saveCheckpoint,
validateSourceIdentity,
type MigrationExecutionOptions,
type MigrationProgress,
} from './incremental.js';
import { legacyUserId } from './identity.js'; import { legacyUserId } from './identity.js';
import { import {
classifyGameStorage, classifyGameStorage,
@@ -45,6 +55,11 @@ interface ArchiveMigrationContext {
sourceFormats: Record<ArchivedGeneralSourceFormat, number>; sourceFormats: Record<ArchivedGeneralSourceFormat, number>;
} }
interface AppendCursor {
afterId: bigint;
endAtId: bigint;
}
const parseNullableJson = (value: unknown, fallback: JsonValue, context: string): JsonValue => const parseNullableJson = (value: unknown, fallback: JsonValue, context: string): JsonValue =>
value === null || value === undefined ? fallback : parseJson(value, context); value === null || value === undefined ? fallback : parseJson(value, context);
@@ -80,22 +95,33 @@ const migrateSimpleTable = async (
conflictColumns: readonly string[], conflictColumns: readonly string[],
mapper: (row: SourceRow) => TargetRow, mapper: (row: SourceRow) => TargetRow,
counts: Record<string, number>, counts: Record<string, number>,
size = batchSize progress: MigrationProgress,
cursor: AppendCursor,
size = batchSize,
strategy: 'append' | 'rescan' = 'append'
): Promise<void> => { ): Promise<void> => {
for await (const rows of paginateSource(source, sourceTable, sourceIdColumn, size)) { for await (const rows of paginateSource(source, sourceTable, sourceIdColumn, size, cursor.afterId)) {
const mapped = rows.map(mapper); const mapped = rows.map(mapper);
if (target) { if (target) {
await upsertRows(target, targetTable, mapped, conflictColumns); await upsertRows(target, targetTable, mapped, conflictColumns);
} }
counts[sourceTable] = (counts[sourceTable] ?? 0) + mapped.length; counts[sourceTable] = (counts[sourceTable] ?? 0) + mapped.length;
} }
progress[sourceTable] = {
strategy,
startAfterId: cursor.afterId < 0n ? null : cursor.afterId.toString(),
endAtId: cursor.endAtId < 0n ? null : cursor.endAtId.toString(),
processed: counts[sourceTable] ?? 0,
};
}; };
const migrateHall = ( const migrateHall = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number>, counts: Record<string, number>,
archive: ArchiveMigrationContext archive: ArchiveMigrationContext,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -120,14 +146,18 @@ const migrateHall = (
import_run_id: archive.importRunId, import_run_id: archive.importRunId,
}; };
}, },
counts counts,
progress,
cursor
); );
const migrateGames = ( const migrateGames = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number>, counts: Record<string, number>,
archive: ArchiveMigrationContext archive: ArchiveMigrationContext,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -159,14 +189,20 @@ const migrateGames = (
import_run_id: archive.importRunId, import_run_id: archive.importRunId,
}; };
}, },
counts counts,
progress,
cursor,
batchSize,
'rescan'
); );
const migrateOldGenerals = ( const migrateOldGenerals = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number>, counts: Record<string, number>,
archive: ArchiveMigrationContext archive: ArchiveMigrationContext,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -197,14 +233,18 @@ const migrateOldGenerals = (
import_run_id: archive.importRunId, import_run_id: archive.importRunId,
}; };
}, },
counts counts,
progress,
cursor
); );
const migrateOldNations = ( const migrateOldNations = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number>, counts: Record<string, number>,
archive: ArchiveMigrationContext archive: ArchiveMigrationContext,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -225,14 +265,18 @@ const migrateOldNations = (
import_run_id: archive.importRunId, import_run_id: archive.importRunId,
}; };
}, },
counts counts,
progress,
cursor
); );
const migrateEmperors = ( const migrateEmperors = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number>, counts: Record<string, number>,
archive: ArchiveMigrationContext archive: ArchiveMigrationContext,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -293,13 +337,17 @@ const migrateEmperors = (
import_run_id: archive.importRunId, import_run_id: archive.importRunId,
}; };
}, },
counts counts,
progress,
cursor
); );
const migrateInheritanceResults = ( const migrateInheritanceResults = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number> counts: Record<string, number>,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -321,13 +369,17 @@ const migrateInheritanceResults = (
created_at: new Date(0), created_at: new Date(0),
}; };
}, },
counts counts,
progress,
cursor
); );
const migrateUserRecords = ( const migrateUserRecords = (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number> counts: Record<string, number>,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => ): Promise<void> =>
migrateSimpleTable( migrateSimpleTable(
source, source,
@@ -349,14 +401,18 @@ const migrateUserRecords = (
created_at: toNullableDate(row.date, `user_record.${id}.date`) ?? new Date(0), created_at: toNullableDate(row.date, `user_record.${id}.date`) ?? new Date(0),
}; };
}, },
counts counts,
progress,
cursor
); );
const migrateYearbook = async ( const migrateYearbook = async (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number>, counts: Record<string, number>,
archive: ArchiveMigrationContext archive: ArchiveMigrationContext,
progress: MigrationProgress,
cursor: AppendCursor
): Promise<void> => { ): Promise<void> => {
await migrateSimpleTable( await migrateSimpleTable(
source, source,
@@ -387,6 +443,8 @@ const migrateYearbook = async (
return mapped; return mapped;
}, },
counts, counts,
progress,
cursor,
25 25
); );
}; };
@@ -451,12 +509,18 @@ export const migrateGame = async (
source: MariaPool, source: MariaPool,
targetPool: PgPool | null, targetPool: PgPool | null,
apply: boolean, apply: boolean,
profile: string profile: string,
execution: MigrationExecutionOptions = defaultExecutionOptions(`legacy-game-${profile}`)
): Promise<MigrationSummary> => { ): Promise<MigrationSummary> => {
if (!isLegacyArchiveProfile(profile)) { if (!isLegacyArchiveProfile(profile)) {
throw new Error(`Unsupported legacy archive profile: ${profile}`); throw new Error(`Unsupported legacy archive profile: ${profile}`);
} }
validateSourceIdentity(execution.source);
if (execution.mode === 'incremental' && !targetPool) {
throw new Error('Incremental game migration requires the target database to read checkpoints');
}
const counts: Record<string, number> = {}; const counts: Record<string, number> = {};
const progress: MigrationProgress = {};
const sourceFormats: Record<ArchivedGeneralSourceFormat, number> = { const sourceFormats: Record<ArchivedGeneralSourceFormat, number> = {
'legacy-flat-v0': 0, 'legacy-flat-v0': 0,
'ref-flat-v1': 0, 'ref-flat-v1': 0,
@@ -494,26 +558,88 @@ export const migrateGame = async (
vote_comment: 'Current-season vote comments.', vote_comment: 'Current-season vote comments.',
'storage:season-state': 'Only inheritance_* and user_* long-lived namespaces are archived or projected.', 'storage:season-state': 'Only inheritance_* and user_* long-lived namespaces are archived or projected.',
}; };
const client = apply && targetPool ? await targetPool.connect() : null; const client = targetPool ? await targetPool.connect() : null;
let importRunId: string | null = null; let importRunId: string | null = null;
try { try {
const run = async (archive: ArchiveMigrationContext): Promise<void> => { const run = async (archive: ArchiveMigrationContext): Promise<void> => {
await migrateGames(source, client, counts, archive); const writeClient = apply ? client : null;
await migrateHall(source, client, counts, archive); const cursorSpecs = [
await migrateOldGenerals(source, client, counts, archive); ['hall', 'id'],
await migrateOldNations(source, client, counts, archive); ['ng_old_generals', 'id'],
await migrateEmperors(source, client, counts, archive); ['ng_old_nations', 'id'],
await migrateInheritanceResults(source, client, counts); ['emperior', 'no'],
await migrateUserRecords(source, client, counts); ['inheritance_result', 'id'],
await migrateStorage(source, client, counts); ['user_record', 'id'],
await migrateYearbook(source, client, counts, archive); ['ng_history', 'no'],
] as const;
const cursors = new Map<string, AppendCursor>();
for (const [sourceTable, idColumn] of cursorSpecs) {
const endAtId = (await sourceMaxId(source, sourceTable, idColumn)) ?? -1n;
const checkpoint =
execution.mode === 'incremental' && client
? await loadCheckpoint(
client,
{
tableSql: '"legacy_archive"."import_checkpoint"',
scope: { columnSql: '"source_profile"', value: profile },
},
execution.source.key,
sourceTable
)
: null;
const afterId =
execution.mode === 'incremental'
? requireIncrementalCheckpoint(checkpoint, execution.source, sourceTable)
: -1n;
if (endAtId < afterId) {
throw new Error(`Legacy source ${sourceTable} maximum ID regressed; run a reviewed full migration`);
}
cursors.set(sourceTable, { afterId, endAtId });
}
const cursor = (table: string): AppendCursor => {
const value = cursors.get(table);
if (!value) throw new Error(`Missing migration cursor for ${table}`);
return value;
}; };
if (client) { const maxGameId = (await sourceMaxId(source, 'ng_games', 'id')) ?? -1n;
await withMigrationLock(client, `sammo-legacy-archive-v2:${profile}`, async () => { await migrateGames(source, writeClient, counts, archive, progress, { afterId: -1n, endAtId: maxGameId });
await migrateHall(source, writeClient, counts, archive, progress, cursor('hall'));
await migrateOldGenerals(source, writeClient, counts, archive, progress, cursor('ng_old_generals'));
await migrateOldNations(source, writeClient, counts, archive, progress, cursor('ng_old_nations'));
await migrateEmperors(source, writeClient, counts, archive, progress, cursor('emperior'));
await migrateInheritanceResults(source, writeClient, counts, progress, cursor('inheritance_result'));
await migrateUserRecords(source, writeClient, counts, progress, cursor('user_record'));
await migrateStorage(source, writeClient, counts);
progress.storage = {
strategy: 'rescan',
startAfterId: null,
endAtId: null,
processed: counts.storage_inspected ?? 0,
};
await migrateYearbook(source, writeClient, counts, archive, progress, cursor('ng_history'));
if (apply && client && archive.importRunId !== '0') {
for (const [sourceTable] of cursorSpecs) {
await saveCheckpoint(
client,
{
tableSql: '"legacy_archive"."import_checkpoint"',
scope: { columnSql: '"source_profile"', value: profile },
},
execution.source,
sourceTable,
cursor(sourceTable).endAtId,
archive.importRunId
);
}
}
};
if (client && apply) {
await withMigrationLock(client, `sammo-legacy-archive-v3:${profile}:${execution.source.key}`, async () => {
const created = await client.query<{ id: string }>( const created = await client.query<{ id: string }>(
`INSERT INTO "legacy_archive"."import_run" ("source_profile", "status") `INSERT INTO "legacy_archive"."import_run"
VALUES ($1, 'RUNNING') RETURNING "id"`, ("source_profile", "source_key", "source_fingerprint", "mode", "status")
[profile] VALUES ($1, $2, $3, $4, 'RUNNING') RETURNING "id"`,
[profile, execution.source.key, execution.source.fingerprint, execution.mode]
); );
importRunId = created.rows[0]?.id ?? null; importRunId = created.rows[0]?.id ?? null;
if (!importRunId) throw new Error('Failed to create legacy archive import run'); if (!importRunId) throw new Error('Failed to create legacy archive import run');
@@ -524,9 +650,10 @@ export const migrateGame = async (
await client.query( await client.query(
`UPDATE "legacy_archive"."import_run" `UPDATE "legacy_archive"."import_run"
SET "status" = 'COMPLETED', "finished_at" = CURRENT_TIMESTAMP, SET "status" = 'COMPLETED', "finished_at" = CURRENT_TIMESTAMP,
"counts" = $2::jsonb, "source_format_summary" = $3::jsonb "counts" = $2::jsonb, "source_format_summary" = $3::jsonb,
"progress" = $4::jsonb
WHERE "id" = $1`, WHERE "id" = $1`,
[importRunId, JSON.stringify(counts), JSON.stringify(sourceFormats)] [importRunId, JSON.stringify(counts), JSON.stringify(sourceFormats), JSON.stringify(progress)]
); );
await client.query('COMMIT'); await client.query('COMMIT');
} catch (error) { } catch (error) {
@@ -536,9 +663,16 @@ export const migrateGame = async (
await client.query( await client.query(
`UPDATE "legacy_archive"."import_run" `UPDATE "legacy_archive"."import_run"
SET "status" = 'FAILED', "finished_at" = CURRENT_TIMESTAMP, SET "status" = 'FAILED', "finished_at" = CURRENT_TIMESTAMP,
"counts" = $2::jsonb, "source_format_summary" = $3::jsonb, "error" = $4 "counts" = $2::jsonb, "source_format_summary" = $3::jsonb,
"progress" = $4::jsonb, "error" = $5
WHERE "id" = $1`, WHERE "id" = $1`,
[importRunId, JSON.stringify(counts), JSON.stringify(sourceFormats), message] [
importRunId,
JSON.stringify(counts),
JSON.stringify(sourceFormats),
JSON.stringify(progress),
message,
]
); );
throw error; throw error;
} }
@@ -549,5 +683,15 @@ export const migrateGame = async (
} finally { } finally {
client?.release(); client?.release();
} }
return { command: 'game', apply, counts, excluded, importRunId, sourceFormatSummary: sourceFormats }; return {
command: 'game',
apply,
counts,
excluded,
importRunId,
sourceFormatSummary: sourceFormats,
mode: execution.mode,
sourceKey: execution.source.key,
progress,
};
}; };
+111 -9
View File
@@ -6,6 +6,7 @@ import {
paginateSource, paginateSource,
jsonParameter, jsonParameter,
querySource, querySource,
sourceMaxId,
toBigInt, toBigInt,
toDate, toDate,
toNullableDate, toNullableDate,
@@ -17,6 +18,15 @@ import {
type SourceRow, type SourceRow,
type TargetRow, type TargetRow,
} from './db.js'; } from './db.js';
import {
defaultExecutionOptions,
loadCheckpoint,
requireIncrementalCheckpoint,
saveCheckpoint,
validateSourceIdentity,
type MigrationExecutionOptions,
type MigrationProgress,
} from './incremental.js';
import { mapLegacyRoles, mapLegacySanctions, parseJson, type JsonValue } from './transform.js'; import { mapLegacyRoles, mapLegacySanctions, parseJson, type JsonValue } from './transform.js';
export interface MigrationSummary { export interface MigrationSummary {
@@ -26,6 +36,9 @@ export interface MigrationSummary {
excluded: Record<string, string>; excluded: Record<string, string>;
importRunId?: string | null; importRunId?: string | null;
sourceFormatSummary?: Record<string, number>; sourceFormatSummary?: Record<string, number>;
mode?: 'full' | 'incremental';
sourceKey?: string;
progress?: MigrationProgress;
} }
const batchSize = 500; const batchSize = 500;
@@ -183,9 +196,10 @@ const processMembers = async (
const processMemberLogs = async ( const processMemberLogs = async (
source: MariaPool, source: MariaPool,
target: PoolClient | null, target: PoolClient | null,
counts: Record<string, number> counts: Record<string, number>,
afterId = -1n
): Promise<void> => { ): Promise<void> => {
for await (const rows of paginateSource(source, 'member_log', 'id', batchSize)) { for await (const rows of paginateSource(source, 'member_log', 'id', batchSize, afterId)) {
const mapped = rows.map<TargetRow>((row) => { const mapped = rows.map<TargetRow>((row) => {
const id = toBigInt(row.id, 'member_log.id'); const id = toBigInt(row.id, 'member_log.id');
const memberNo = toNumber(row.member_no, `member_log.${id}.member_no`); const memberNo = toNumber(row.member_no, `member_log.${id}.member_no`);
@@ -270,38 +284,126 @@ export const migrateGateway = async (
source: MariaPool, source: MariaPool,
targetPool: PgPool | null, targetPool: PgPool | null,
apply: boolean, apply: boolean,
migratedAt: Date migratedAt: Date,
execution: MigrationExecutionOptions = defaultExecutionOptions('legacy-root')
): Promise<MigrationSummary> => { ): Promise<MigrationSummary> => {
validateSourceIdentity(execution.source);
if (execution.mode === 'incremental' && !targetPool) {
throw new Error('Incremental gateway migration requires the target database to read checkpoints');
}
const counts: Record<string, number> = {}; const counts: Record<string, number> = {};
const progress: MigrationProgress = {};
const excluded = { const excluded = {
login_token: login_token:
'Legacy bearer tokens, IP addresses, and expired sessions are not valid in the Redis session model.', 'Legacy bearer tokens, IP addresses, and expired sessions are not valid in the Redis session model.',
}; };
const client = targetPool ? await targetPool.connect() : null; const client = targetPool ? await targetPool.connect() : null;
let importRunId: string | null = null;
try { try {
const run = async (): Promise<void> => { const run = async (runId: string | null): Promise<void> => {
await processMembers(source, client, apply, migratedAt, counts); await processMembers(source, client, apply, migratedAt, counts);
await processMemberLogs(source, apply ? client : null, counts); progress.member = {
strategy: 'rescan',
startAfterId: null,
endAtId: null,
processed: counts.member ?? 0,
};
const maxMemberLogId = (await sourceMaxId(source, 'member_log', 'id')) ?? -1n;
const checkpoint =
execution.mode === 'incremental' && client
? await loadCheckpoint(
client,
{ tableSql: '"legacy_import_checkpoint"' },
execution.source.key,
'member_log'
)
: null;
const memberLogAfter =
execution.mode === 'incremental'
? requireIncrementalCheckpoint(checkpoint, execution.source, 'member_log')
: -1n;
if (maxMemberLogId < memberLogAfter) {
throw new Error('Legacy source member_log maximum ID regressed; run a reviewed full migration');
}
await processMemberLogs(source, apply ? client : null, counts, memberLogAfter);
progress.member_log = {
strategy: 'append',
startAfterId: memberLogAfter < 0n ? null : memberLogAfter.toString(),
endAtId: maxMemberLogId < 0n ? null : maxMemberLogId.toString(),
processed: counts.member_log ?? 0,
};
await processBannedMembers(source, apply ? client : null, counts); await processBannedMembers(source, apply ? client : null, counts);
await processRootKeyValues(source, apply ? client : null, counts); await processRootKeyValues(source, apply ? client : null, counts);
await processSystem(source, apply ? client : null, counts); await processSystem(source, apply ? client : null, counts);
for (const table of ['banned_member', 'root_key_value', 'system'] as const) {
progress[table] = {
strategy: 'rescan',
startAfterId: null,
endAtId: null,
processed: counts[table] ?? 0,
};
}
if (apply && client && runId) {
await saveCheckpoint(
client,
{ tableSql: '"legacy_import_checkpoint"' },
execution.source,
'member_log',
maxMemberLogId,
runId
);
}
}; };
if (client && apply) { if (client && apply) {
await withMigrationLock(client, 'sammo-legacy-gateway-v1', async () => { await withMigrationLock(client, `sammo-legacy-gateway-v2:${execution.source.key}`, async () => {
const created = await client.query<{ id: string }>(
`INSERT INTO "legacy_import_run"
("source_key", "source_fingerprint", "mode", "status")
VALUES ($1, $2, $3, 'RUNNING') RETURNING "id"`,
[execution.source.key, execution.source.fingerprint, execution.mode]
);
importRunId = created.rows[0]?.id ?? null;
if (!importRunId) throw new Error('Failed to create legacy gateway import run');
await client.query('BEGIN'); await client.query('BEGIN');
try { try {
await run(); await run(importRunId);
await client.query(
`UPDATE "legacy_import_run"
SET "status" = 'COMPLETED', "finished_at" = CURRENT_TIMESTAMP,
"counts" = $2::jsonb, "progress" = $3::jsonb
WHERE "id" = $1`,
[importRunId, JSON.stringify(counts), JSON.stringify(progress)]
);
await client.query('COMMIT'); await client.query('COMMIT');
} catch (error) { } catch (error) {
await client.query('ROLLBACK'); await client.query('ROLLBACK');
const message =
error instanceof Error ? error.message.slice(0, 2000) : String(error).slice(0, 2000);
await client.query(
`UPDATE "legacy_import_run"
SET "status" = 'FAILED', "finished_at" = CURRENT_TIMESTAMP,
"counts" = $2::jsonb, "progress" = $3::jsonb, "error" = $4
WHERE "id" = $1`,
[importRunId, JSON.stringify(counts), JSON.stringify(progress), message]
);
throw error; throw error;
} }
}); });
} else { } else {
await run(); await run(null);
} }
} finally { } finally {
client?.release(); client?.release();
} }
return { command: 'gateway', apply, counts, excluded }; return {
command: 'gateway',
apply,
counts,
excluded,
importRunId,
mode: execution.mode,
sourceKey: execution.source.key,
progress,
};
}; };
@@ -0,0 +1,137 @@
import { createHash } from 'node:crypto';
import type { PoolClient } from 'pg';
export type MigrationMode = 'full' | 'incremental';
export interface MigrationSourceIdentity {
key: string;
fingerprint: string;
}
export interface MigrationExecutionOptions {
mode: MigrationMode;
source: MigrationSourceIdentity;
}
export interface MigrationTableProgress {
strategy: 'append' | 'rescan';
startAfterId: string | null;
endAtId: string | null;
processed: number;
}
export type MigrationProgress = Record<string, MigrationTableProgress>;
export interface StoredCheckpoint {
sourceFingerprint: string;
lastLegacyId: bigint;
}
export interface CheckpointStore {
tableSql: '"legacy_import_checkpoint"' | '"legacy_archive"."import_checkpoint"';
scope?: { columnSql: '"source_profile"'; value: string };
}
const SOURCE_KEY = /^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,127}$/;
export const validateSourceIdentity = (source: MigrationSourceIdentity): MigrationSourceIdentity => {
if (!SOURCE_KEY.test(source.key)) {
throw new Error('Legacy migration source key must use 1-128 safe characters');
}
if (!/^[a-f0-9]{64}$/u.test(source.fingerprint)) {
throw new Error('Legacy migration source fingerprint must be a SHA-256 hex value');
}
return source;
};
export const fingerprintMariaConnection = (connectionString: string): string => {
const url = new URL(connectionString);
if (url.protocol !== 'mariadb:' && url.protocol !== 'mysql:') {
throw new Error('Legacy source URL must use the mariadb or mysql protocol');
}
const query = [...url.searchParams.entries()]
.filter(([key]) => !/pass(word)?|secret|token/iu.test(key))
.sort(([leftKey, leftValue], [rightKey, rightValue]) =>
leftKey === rightKey ? leftValue.localeCompare(rightValue) : leftKey.localeCompare(rightKey)
);
const identity = {
protocol: url.protocol,
host: url.hostname.toLowerCase(),
port: url.port || '3306',
database: decodeURIComponent(url.pathname.replace(/^\//u, '')),
user: decodeURIComponent(url.username),
query,
};
return createHash('sha256').update(JSON.stringify(identity)).digest('hex');
};
export const loadCheckpoint = async (
client: PoolClient,
store: CheckpointStore,
sourceKey: string,
sourceTable: string
): Promise<StoredCheckpoint | null> => {
const scopePredicate = store.scope ? ` AND ${store.scope.columnSql} = $3` : '';
const parameters = store.scope ? [sourceKey, sourceTable, store.scope.value] : [sourceKey, sourceTable];
const result = await client.query<{ source_fingerprint: string; last_legacy_id: string }>(
`SELECT "source_fingerprint", "last_legacy_id"
FROM ${store.tableSql}
WHERE "source_key" = $1 AND "source_table" = $2${scopePredicate}
FOR UPDATE`,
parameters
);
const row = result.rows[0];
return row ? { sourceFingerprint: row.source_fingerprint, lastLegacyId: BigInt(row.last_legacy_id) } : null;
};
export const requireIncrementalCheckpoint = (
checkpoint: StoredCheckpoint | null,
source: MigrationSourceIdentity,
sourceTable: string
): bigint => {
if (!checkpoint) {
throw new Error(`Incremental migration requires a completed full checkpoint for ${sourceTable}`);
}
if (checkpoint.sourceFingerprint !== source.fingerprint) {
throw new Error(`Legacy source fingerprint changed for ${sourceTable}; run a reviewed full migration`);
}
return checkpoint.lastLegacyId;
};
export const saveCheckpoint = async (
client: PoolClient,
store: CheckpointStore,
source: MigrationSourceIdentity,
sourceTable: string,
lastLegacyId: bigint,
importRunId: string
): Promise<void> => {
const scopeColumn = store.scope ? `${store.scope.columnSql}, ` : '';
const scopeValue = store.scope ? '$1, ' : '';
const parameterOffset = store.scope ? 1 : 0;
const parameters: unknown[] = store.scope ? [store.scope.value] : [];
parameters.push(source.key, source.fingerprint, sourceTable, lastLegacyId.toString(), importRunId);
const conflictColumns = store.scope
? `${store.scope.columnSql}, "source_key", "source_table"`
: '"source_key", "source_table"';
await client.query(
`INSERT INTO ${store.tableSql}
(${scopeColumn}"source_key", "source_fingerprint", "source_table", "last_legacy_id", "import_run_id", "updated_at")
VALUES (${scopeValue}$${parameterOffset + 1}, $${parameterOffset + 2}, $${parameterOffset + 3}, $${parameterOffset + 4}, $${parameterOffset + 5}, CURRENT_TIMESTAMP)
ON CONFLICT (${conflictColumns}) DO UPDATE SET
"source_fingerprint" = EXCLUDED."source_fingerprint",
"last_legacy_id" = EXCLUDED."last_legacy_id",
"import_run_id" = EXCLUDED."import_run_id",
"updated_at" = CURRENT_TIMESTAMP`,
parameters
);
};
export const defaultExecutionOptions = (
sourceKey: string,
connectionString = `mariadb://legacy@localhost/${sourceKey}`
): MigrationExecutionOptions => ({
mode: 'full',
source: { key: sourceKey, fingerprint: fingerprintMariaConnection(connectionString) },
});
+105
View File
@@ -0,0 +1,105 @@
import { createMariaPool, createPostgresPool, querySource } from './db.js';
import { migrateGame } from './game.js';
import { migrateGateway, type MigrationSummary } from './gateway.js';
import type { MigrationMode } from './incremental.js';
import type { ResolvedMigrationPlan, ResolvedMigrationStage } from './config.js';
export interface PlanRunSummary {
command: 'run-plan';
sourceSet: string;
mode: MigrationMode;
apply: boolean;
stages: Array<{ name: string; status: 'COMPLETED'; summary: MigrationSummary }>;
}
const preflightStage = async (stage: ResolvedMigrationStage): Promise<void> => {
const source = createMariaPool(stage.sourceUrl);
const target = createPostgresPool(stage.targetUrl);
try {
const sourceDatabase = await querySource(source, 'SELECT DATABASE() AS database_name');
if (typeof sourceDatabase[0]?.database_name !== 'string') {
throw new Error(`Source preflight did not select a database for ${stage.name}`);
}
const requiredSourceTables =
stage.kind === 'gateway'
? ['member', 'member_log', 'banned_member', 'storage', 'system']
: [
'ng_games',
'hall',
'ng_old_generals',
'ng_old_nations',
'emperior',
'inheritance_result',
'user_record',
'storage',
'ng_history',
];
const sourceTables = await querySource(
source,
`SELECT table_name AS source_table_name
FROM information_schema.tables
WHERE table_schema = DATABASE() AND table_name IN (${requiredSourceTables.map(() => '?').join(', ')})`,
requiredSourceTables
);
const availableSourceTables = new Set(sourceTables.map((row) => String(row.source_table_name)));
const missingSourceTables = requiredSourceTables.filter((table) => !availableSourceTables.has(table));
if (missingSourceTables.length) {
throw new Error(`Source ${stage.name} is missing required tables: ${missingSourceTables.join(', ')}`);
}
await target.query('SELECT 1');
const checkpointTable =
stage.kind === 'gateway' ? 'public.legacy_import_checkpoint' : 'legacy_archive.import_checkpoint';
const migrationReady = await target.query<{ table_name: string | null }>(
'SELECT to_regclass($1) AS table_name',
[checkpointTable]
);
if (!migrationReady.rows[0]?.table_name) {
throw new Error(`Target migrations are not current for ${stage.name}; missing ${checkpointTable}`);
}
const targetDataTable = stage.kind === 'gateway' ? 'public.app_user' : 'inheritance_result';
const targetReady = await target.query<{ table_name: string | null }>('SELECT to_regclass($1) AS table_name', [
targetDataTable,
]);
if (!targetReady.rows[0]?.table_name) {
throw new Error(`Target migrations are not current for ${stage.name}; missing ${targetDataTable}`);
}
} finally {
await source.end();
await target.end();
}
};
export const checkMigrationPlan = async (plan: ResolvedMigrationPlan): Promise<Record<string, unknown>> => {
for (const stage of plan.stages) await preflightStage(stage);
return {
command: 'check-plan',
sourceSet: plan.sourceSet,
stages: plan.stages.map((stage) => ({ name: stage.name, kind: stage.kind, status: 'READY' })),
};
};
export const runMigrationPlan = async (
plan: ResolvedMigrationPlan,
mode: MigrationMode,
apply: boolean,
migratedAt = new Date()
): Promise<PlanRunSummary> => {
await checkMigrationPlan(plan);
const stages: PlanRunSummary['stages'] = [];
for (const stage of plan.stages) {
const source = createMariaPool(stage.sourceUrl);
const target = createPostgresPool(stage.targetUrl);
try {
const execution = { mode, source: stage.sourceIdentity } as const;
const summary =
stage.kind === 'gateway'
? await migrateGateway(source, target, apply, migratedAt, execution)
: await migrateGame(source, target, apply, stage.profile!, execution);
stages.push({ name: stage.name, status: 'COMPLETED', summary });
} finally {
await source.end();
await target.end();
}
}
return { command: 'run-plan', sourceSet: plan.sourceSet, mode, apply, stages };
};
@@ -0,0 +1,60 @@
import { chmod, mkdtemp, rm, writeFile } from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import { loadMigrationPlan } from '../src/config.js';
const workDirectories: string[] = [];
afterEach(async () => {
delete process.env.TEST_GATEWAY_DATABASE_URL;
await Promise.all(workDirectories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })));
});
const writeFixture = async (mode = 0o600): Promise<string> => {
const directory = await mkdtemp(path.join(os.tmpdir(), 'sammo-legacy-plan-'));
workDirectories.push(directory);
await writeFile(path.join(directory, 'mysql-password'), 'secret-value\n', { mode: 0o600 });
const configPath = path.join(directory, 'migration-plan.json');
await writeFile(
configPath,
JSON.stringify({
version: 1,
sourceSet: 'fixture-cutover',
gateway: {
source: {
host: '127.0.0.1',
database: 'root_dump',
user: 'migration_reader',
passwordFile: './mysql-password',
},
targetUrlEnv: 'TEST_GATEWAY_DATABASE_URL',
},
}),
{ mode }
);
await chmod(configPath, mode);
return configPath;
};
describe('legacy migration plan config', () => {
it('resolves a mode-0600 structured source without placing its password in arguments', async () => {
process.env.TEST_GATEWAY_DATABASE_URL = 'postgresql://target@127.0.0.1/gateway';
const plan = await loadMigrationPlan(await writeFixture());
const source = new URL(plan.stages[0]!.sourceUrl);
expect(plan.sourceSet).toBe('fixture-cutover');
expect(plan.stages.map((stage) => stage.name)).toEqual(['gateway']);
expect(source.hostname).toBe('127.0.0.1');
expect(source.username).toBe('migration_reader');
expect(source.password).toBe('secret-value');
expect(plan.stages[0]!.sourceIdentity.key).toBe('fixture-cutover:gateway');
});
it('rejects a config readable by group or other users', async () => {
process.env.TEST_GATEWAY_DATABASE_URL = 'postgresql://target@127.0.0.1/gateway';
await expect(loadMigrationPlan(await writeFixture(0o644))).rejects.toThrow('expected mode 0600');
});
});
@@ -0,0 +1,38 @@
USE root_legacy;
UPDATE `member` SET `REG_NUM` = 1 WHERE `NO` = 1;
UPDATE `system` SET `NOTICE` = 'incremental notice', `MDF_DATE` = '2020-02-01 00:00:00' WHERE `NO` = 1;
UPDATE `storage` SET `value` = '{"version":2}' WHERE `id` = 1;
INSERT INTO `member_log` (`id`, `member_no`, `date`, `action_type`, `action`)
VALUES (2, 1, '2020-02-01 00:00:00', 'logout', NULL);
USE che_legacy;
UPDATE `ng_games` SET `winner_nation` = 1 WHERE `id` = 1;
UPDATE `storage` SET `value` = '[20,null]' WHERE `id` = 1;
INSERT INTO `hall`
(`id`, `server_id`, `season`, `scenario`, `general_no`, `type`, `value`, `owner`, `aux`)
VALUES (2, 'che_fixture_002', 2, 2, 11, 'war', 200, 1, '{}');
INSERT INTO `ng_old_generals`
(`id`, `server_id`, `general_no`, `owner`, `name`, `last_yearmonth`, `turntime`, `data`)
VALUES
(2, 'che_fixture_002', 11, 1, 'Incremental General', 22112, '2020-02-01 00:00:00.000000',
'{"leader":81,"power":71,"intel":61,"history":"second<br>"}');
INSERT INTO `ng_old_nations` (`id`, `server_id`, `nation`, `data`, `date`)
VALUES (2, 'che_fixture_002', 2, '{}', '2020-02-01 00:00:00');
INSERT INTO `emperior` (`no`, `server_id`, `name`, `history`, `aux`)
VALUES (2, 'che_fixture_002', 'Incremental Emperor', '[]', '{}');
INSERT INTO `inheritance_result` (`id`, `server_id`, `owner`, `general_id`, `year`, `month`, `value`)
VALUES (2, 'che_fixture_002', 1, 11, 221, 12, '{}');
INSERT INTO `user_record` (`id`, `user_id`, `server_id`, `log_type`, `year`, `month`, `date`, `text`)
VALUES (2, 1, 'che_fixture_002', 'history', 221, 12, '2020-02-01 00:00:00', 'incremental history');
INSERT INTO `ng_history`
(`no`, `server_id`, `year`, `month`, `map`, `global_history`, `global_action`, `nations`)
VALUES (2, 'che_fixture_002', 221, 12, '{}', '[]', '[]', '[]');
@@ -0,0 +1,56 @@
USE root_legacy;
INSERT INTO `system` (`NO`, `REG`, `LOGIN`, `NOTICE`, `CRT_DATE`, `MDF_DATE`)
VALUES (1, 'Y', 'Y', 'initial notice', '2020-01-01 00:00:00', '2020-01-01 00:00:00');
INSERT INTO `member`
(`NO`, `oauth_id`, `ID`, `EMAIL`, `oauth_type`, `oauth_info`, `token_valid_until`, `PW`, `salt`,
`third_use`, `NAME`, `PICTURE`, `IMGSVR`, `acl`, `penalty`, `GRADE`, `REG_NUM`, `REG_DATE`,
`BLOCK_NUM`, `BLOCK_DATE`, `delete_after`)
VALUES
(1, NULL, 'fixture-user', 'fixture@example.test', 'NONE', '{}', NULL,
REPEAT('a', 128), 'fixture-salt-001', 0, 'Fixture User', 'default.jpg', 0, '{}', '{}', 1, 0,
'2020-01-01 00:00:00', 0, NULL, NULL);
INSERT INTO `member_log` (`id`, `member_no`, `date`, `action_type`, `action`)
VALUES (1, 1, '2020-01-01 00:00:00', 'login', NULL);
INSERT INTO `storage` (`id`, `namespace`, `key`, `value`)
VALUES (1, 'fixture', 'mutable', '{"version":1}');
USE che_legacy;
INSERT INTO `ng_games`
(`id`, `server_id`, `date`, `winner_nation`, `map`, `season`, `scenario`, `scenario_name`, `env`)
VALUES
(1, 'che_fixture_001', '2020-01-01 00:00:00', NULL, 'che', 1, 2, 'fixture',
'{"opentime":"2020-01-01 00:00:00"}');
INSERT INTO `hall`
(`id`, `server_id`, `season`, `scenario`, `general_no`, `type`, `value`, `owner`, `aux`)
VALUES (1, 'che_fixture_001', 1, 2, 10, 'war', 100, 1, '{}');
INSERT INTO `ng_old_generals`
(`id`, `server_id`, `general_no`, `owner`, `name`, `last_yearmonth`, `turntime`, `data`)
VALUES
(1, 'che_fixture_001', 10, 1, 'Fixture General', 22012, '2020-01-01 00:00:00.000000',
'{"leader":80,"power":70,"intel":60,"history":"first<br>"}');
INSERT INTO `ng_old_nations` (`id`, `server_id`, `nation`, `data`, `date`)
VALUES (1, 'che_fixture_001', 1, '{}', '2020-01-01 00:00:00');
INSERT INTO `emperior` (`no`, `server_id`, `name`, `history`, `aux`)
VALUES (1, 'che_fixture_001', 'Fixture Emperor', '[]', '{}');
INSERT INTO `inheritance_result` (`id`, `server_id`, `owner`, `general_id`, `year`, `month`, `value`)
VALUES (1, 'che_fixture_001', 1, 10, 220, 12, '{}');
INSERT INTO `user_record` (`id`, `user_id`, `server_id`, `log_type`, `year`, `month`, `date`, `text`)
VALUES (1, 1, 'che_fixture_001', 'history', 220, 12, '2020-01-01 00:00:00', 'fixture history');
INSERT INTO `storage` (`id`, `namespace`, `key`, `value`)
VALUES (1, 'inheritance_1', 'point', '[10,null]');
INSERT INTO `ng_history`
(`no`, `server_id`, `year`, `month`, `map`, `global_history`, `global_action`, `nations`)
VALUES (1, 'che_fixture_001', 220, 12, '{}', '[]', '[]', '[]');
+66 -3
View File
@@ -35,16 +35,30 @@ const sourceRows = {
const sourcePool = (): MariaPool => { const sourcePool = (): MariaPool => {
const seen = new Set<string>(); const seen = new Set<string>();
return { return {
query: vi.fn(async (sql: string) => { query: vi.fn(async (sql: string, values: readonly unknown[] = []) => {
const table = /FROM `([a-z_]+)`/u.exec(sql)?.[1] ?? ''; const table = /FROM `([a-z_]+)`/u.exec(sql)?.[1] ?? '';
if (sql.includes('SELECT MAX(')) {
const rows = (sourceRows[table as keyof typeof sourceRows] ?? []) as Array<Record<string, unknown>>;
const idColumn = /MAX\(`([a-z_]+)`\)/u.exec(sql)?.[1] ?? 'id';
return [{ max_id: rows.length ? rows.at(-1)?.[idColumn] : null }];
}
if (seen.has(table)) return []; if (seen.has(table)) return [];
seen.add(table); seen.add(table);
return sourceRows[table as keyof typeof sourceRows] ?? []; const afterId = BigInt(String(values[0] ?? -1));
return ((sourceRows[table as keyof typeof sourceRows] ?? []) as Array<Record<string, unknown>>).filter(
(row) => {
const idColumn = /WHERE `([a-z_]+)` >/u.exec(sql)?.[1] ?? 'id';
return BigInt(String(row[idColumn])) > afterId;
}
);
}), }),
} as unknown as MariaPool; } as unknown as MariaPool;
}; };
const targetPool = (failPattern?: string) => { const targetPool = (
failPattern?: string,
checkpoints?: Record<string, { fingerprint: string; lastLegacyId: string }>
) => {
const queries: Array<{ sql: string; values: readonly unknown[] }> = []; const queries: Array<{ sql: string; values: readonly unknown[] }> = [];
const query = vi.fn(async (sql: string, values: readonly unknown[] = []) => { const query = vi.fn(async (sql: string, values: readonly unknown[] = []) => {
queries.push({ sql, values }); queries.push({ sql, values });
@@ -55,6 +69,20 @@ const targetPool = (failPattern?: string) => {
if (sql.includes('INSERT INTO "legacy_archive"."import_run"')) { if (sql.includes('INSERT INTO "legacy_archive"."import_run"')) {
return { rows: [{ id: '77' }], rowCount: 1 } as QueryResult<{ id: string }>; return { rows: [{ id: '77' }], rowCount: 1 } as QueryResult<{ id: string }>;
} }
if (sql.includes('FROM "legacy_archive"."import_checkpoint"')) {
const checkpoint = checkpoints?.[String(values[1])];
return {
rows: checkpoint
? [
{
source_fingerprint: checkpoint.fingerprint,
last_legacy_id: checkpoint.lastLegacyId,
},
]
: [],
rowCount: checkpoint ? 1 : 0,
} as unknown as QueryResult;
}
return { rows: [], rowCount: 0 } as unknown as QueryResult; return { rows: [], rowCount: 0 } as unknown as QueryResult;
}); });
const client = { query, release: vi.fn() } as unknown as PoolClient; const client = { query, release: vi.fn() } as unknown as PoolClient;
@@ -143,4 +171,39 @@ describe('legacy archive game migration', () => {
'Unsupported legacy archive profile' 'Unsupported legacy archive profile'
); );
}); });
it('uses checkpoints for append-only tables while rescanning mutable game history', async () => {
const fingerprint = 'a'.repeat(64);
const checkpoints = Object.fromEntries(
['hall', 'ng_old_nations', 'emperior', 'inheritance_result', 'user_record', 'ng_history'].map((table) => [
table,
{ fingerprint, lastLegacyId: '-1' },
])
);
checkpoints.ng_old_generals = { fingerprint, lastLegacyId: '1' };
const target = targetPool(undefined, checkpoints);
const summary = await migrateGame(sourcePool(), target.pool, false, 'che', {
mode: 'incremental',
source: { key: 'fixture:che', fingerprint },
});
expect(summary.counts).toMatchObject({ ng_games: 1, ng_old_generals: 1 });
expect(summary.progress).toMatchObject({
ng_games: { strategy: 'rescan', startAfterId: null, processed: 1 },
ng_old_generals: { strategy: 'append', startAfterId: '1', endAtId: '2', processed: 1 },
});
expect(target.queries.some((entry) => entry.sql.includes('INSERT INTO "legacy_archive"."general"'))).toBe(
false
);
});
it('refuses incremental mode without a completed full checkpoint', async () => {
await expect(
migrateGame(sourcePool(), targetPool().pool, false, 'che', {
mode: 'incremental',
source: { key: 'fixture:che', fingerprint: 'a'.repeat(64) },
})
).rejects.toThrow('requires a completed full checkpoint');
});
}); });
+56 -2
View File
@@ -1,8 +1,9 @@
import type { PoolClient } from 'pg'; import type { Pool as MariaPool } from 'mariadb';
import type { Pool as PgPool, PoolClient, QueryResult } from 'pg';
import { describe, expect, it, vi } from 'vitest'; import { describe, expect, it, vi } from 'vitest';
import { upsertRows } from '../src/db.js'; import { upsertRows } from '../src/db.js';
import { mapMember, MEMBER_PRESERVED_COLUMNS, preflightMemberConflicts } from '../src/gateway.js'; import { mapMember, MEMBER_PRESERVED_COLUMNS, migrateGateway, preflightMemberConflicts } from '../src/gateway.js';
const memberRow = (overrides: Record<string, unknown> = {}) => ({ const memberRow = (overrides: Record<string, unknown> = {}) => ({
NO: 7, NO: 7,
@@ -110,4 +111,57 @@ describe('legacy gateway member migration', () => {
); );
expect(String(query.mock.calls[0]?.[0])).toContain('FROM "app_user"'); expect(String(query.mock.calls[0]?.[0])).toContain('FROM "app_user"');
}); });
it('reads only new immutable member logs during an incremental dry-run', async () => {
const fingerprint = 'a'.repeat(64);
const seen = new Set<string>();
const source = {
query: vi.fn(async (sql: string, values: readonly unknown[] = []) => {
if (sql.includes('MAX(`id`)') && sql.includes('member_log')) return [{ max_id: 2 }];
if (sql.includes('MAX(`date`)')) return [];
if (sql.includes('FROM `member`')) return [];
if (sql.includes('FROM `member_log`')) {
if (seen.has('member_log')) return [];
seen.add('member_log');
return Number(values[0]) < 2
? [
{
id: 2,
member_no: 7,
date: new Date('2026-08-18T00:00:00.000Z'),
action_type: 'login',
action: null,
},
]
: [];
}
return [];
}),
} as unknown as MariaPool;
const query = vi.fn(async (sql: string) => {
if (sql.includes('FROM "legacy_import_checkpoint"')) {
return {
rows: [{ source_fingerprint: fingerprint, last_legacy_id: '1' }],
rowCount: 1,
} as QueryResult;
}
return { rows: [], rowCount: 0 } as unknown as QueryResult;
});
const client = { query, release: vi.fn() } as unknown as PoolClient;
const target = { connect: vi.fn(async () => client) } as unknown as PgPool;
const summary = await migrateGateway(source, target, false, new Date('2026-08-18T00:00:00.000Z'), {
mode: 'incremental',
source: { key: 'fixture:gateway', fingerprint },
});
expect(summary.counts.member_log).toBe(1);
expect(summary.progress?.member_log).toMatchObject({
strategy: 'append',
startAfterId: '1',
endAtId: '2',
processed: 1,
});
expect(query.mock.calls.some(([sql]) => String(sql).includes('INSERT INTO "legacy_member_log"'))).toBe(false);
});
}); });
@@ -0,0 +1,29 @@
import { describe, expect, it } from 'vitest';
import {
fingerprintMariaConnection,
requireIncrementalCheckpoint,
validateSourceIdentity,
} from '../src/incremental.js';
describe('legacy incremental source identity', () => {
it('excludes passwords while binding host, database, user, and non-secret options', () => {
const first = fingerprintMariaConnection('mariadb://reader:first@db.internal:3307/root_dump?ssl=true');
const rotated = fingerprintMariaConnection('mariadb://reader:rotated@db.internal:3307/root_dump?ssl=true');
const otherDatabase = fingerprintMariaConnection('mariadb://reader:rotated@db.internal:3307/che_dump?ssl=true');
expect(first).toBe(rotated);
expect(first).not.toBe(otherDatabase);
});
it('rejects missing, mismatched, and malformed checkpoint identities', () => {
const source = { key: 'cutover:che', fingerprint: 'a'.repeat(64) };
expect(() => requireIncrementalCheckpoint(null, source, 'hall')).toThrow('completed full checkpoint');
expect(() =>
requireIncrementalCheckpoint({ sourceFingerprint: 'b'.repeat(64), lastLegacyId: 1n }, source, 'hall')
).toThrow('fingerprint changed');
expect(() => validateSourceIdentity({ key: '../unsafe', fingerprint: 'a'.repeat(64) })).toThrow(
'safe characters'
);
});
});