|
| 1 | +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | + |
| 3 | +import { Command, Flags } from '@oclif/core'; |
| 4 | +import chalk from 'chalk'; |
| 5 | +import { createInterface } from 'node:readline'; |
| 6 | +import { |
| 7 | + findInterruptedRuns, |
| 8 | + readRunJournal, |
| 9 | + resumeMigrationJournal, |
| 10 | + MigrationJournalRefusal, |
| 11 | + type InterruptedRun, |
| 12 | + type MigrationPlanProvider, |
| 13 | +} from '@objectstack/core'; |
| 14 | +import { describeInterruptedRun } from '@objectstack/runtime'; |
| 15 | +import type { IObjectQLEngine } from '@objectstack/spec/contracts'; |
| 16 | +import { |
| 17 | + printHeader, |
| 18 | + printSuccess, |
| 19 | + printWarning, |
| 20 | + printError, |
| 21 | + printInfo, |
| 22 | + printStep, |
| 23 | + createTimer, |
| 24 | + emitJson, |
| 25 | +} from '../../utils/format.js'; |
| 26 | +import { bootSchemaStack } from '../../utils/schema-migrate.js'; |
| 27 | +import { buildDataMigrationPlugins } from '../../utils/data-migration-plugins.js'; |
| 28 | + |
| 29 | +async function confirm(question: string): Promise<boolean> { |
| 30 | + if (!process.stdin.isTTY) return false; // non-interactive → require --yes |
| 31 | + const rl = createInterface({ input: process.stdin, output: process.stdout }); |
| 32 | + try { |
| 33 | + const answer: string = await new Promise((resolve) => rl.question(question, resolve)); |
| 34 | + return /^y(es)?$/i.test(answer.trim()); |
| 35 | + } finally { |
| 36 | + rl.close(); |
| 37 | + } |
| 38 | +} |
| 39 | + |
| 40 | +/** |
| 41 | + * `os migrate resume` — act on a migration run the journal says was |
| 42 | + * interrupted (ADR-0119 D2, #4617 deliverable 3). |
| 43 | + * |
| 44 | + * The counterpart to `MigrationRecoveryPlugin`'s boot scan, and the division of |
| 45 | + * labour is deliberate: **boot discovers, this command acts.** Resuming is a |
| 46 | + * large, irreversible write against production data, so it happens under |
| 47 | + * explicit operator intent rather than as a side effect of a process starting. |
| 48 | + * The runner's re-entrancy is what makes deferring safe — `started ∧ ¬done` is |
| 49 | + * durable, so a run stays exactly as recoverable an hour later as it was at |
| 50 | + * boot. |
| 51 | + * |
| 52 | + * With no `--run`, this lists what the journal knows and exits without writing |
| 53 | + * anything: the read-only default the other `os migrate` commands use, for the |
| 54 | + * same reason (#2186 — a bare command must never mutate by surprise). |
| 55 | + * |
| 56 | + * ## What "resume" does is not this command's decision |
| 57 | + * |
| 58 | + * The plan's `onCrash` policy decides whether an interrupted run goes FORWARD |
| 59 | + * from the first chunk lacking `chunk_done` or UNWINDS what it committed. Only |
| 60 | + * the plan's author knows which of those is safe for their steps, so this |
| 61 | + * command carries the operator's intent to act and the plan carries what acting |
| 62 | + * means. |
| 63 | + * |
| 64 | + * ## Why a run can be unresumable here |
| 65 | + * |
| 66 | + * A journal cannot hold a plan: `forward`/`compensate` are functions and |
| 67 | + * `load()` reads the live database, so none of it crosses a process boundary. |
| 68 | + * A resume needs the plan handed back by the code that owns it, through the |
| 69 | + * `migration-plans` registry. If the package owning a run's plan is not loaded, |
| 70 | + * this command says exactly that and changes nothing — an unresumable run is |
| 71 | + * reported, never silently skipped, because "nothing to do" and "the code for |
| 72 | + * this run is not here" are different facts. |
| 73 | + */ |
| 74 | +export default class MigrateResume extends Command { |
| 75 | + static override description = |
| 76 | + 'List migration runs the journal says were interrupted, and resume or unwind one. ' + |
| 77 | + 'Read-only without --run.'; |
| 78 | + |
| 79 | + static override examples = [ |
| 80 | + '$ os migrate resume', |
| 81 | + '$ os migrate resume --json', |
| 82 | + '$ os migrate resume --run 6f1e6a3c-6a1e-4c53-9c2f-2c8a9d5b1f77', |
| 83 | + '$ os migrate resume --run 6f1e6a3c-... --yes', |
| 84 | + ]; |
| 85 | + |
| 86 | + static override flags = { |
| 87 | + 'database-url': Flags.string({ |
| 88 | + description: 'Database URL to inspect (defaults to $OS_DATABASE_URL / the project DB)', |
| 89 | + env: 'OS_DATABASE_URL', |
| 90 | + }), |
| 91 | + run: Flags.string({ |
| 92 | + description: 'Resume this run id (omit to list interrupted runs and exit without writing)', |
| 93 | + }), |
| 94 | + yes: Flags.boolean({ char: 'y', description: 'Skip the resume confirmation prompt', default: false }), |
| 95 | + json: Flags.boolean({ description: 'Machine-readable output', default: false }), |
| 96 | + }; |
| 97 | + |
| 98 | + async run(): Promise<void> { |
| 99 | + const { flags } = await this.parse(MigrateResume); |
| 100 | + const timer = createTimer(); |
| 101 | + |
| 102 | + if (!flags.json) printHeader('Migrate · resume'); |
| 103 | + if (!flags.json) printStep(flags.run ? 'Booting data stack…' : 'Booting data stack (read-only)…'); |
| 104 | + |
| 105 | + let stack; |
| 106 | + try { |
| 107 | + stack = await bootSchemaStack({ |
| 108 | + databaseUrl: flags['database-url'], |
| 109 | + extraPlugins: await buildDataMigrationPlugins(), |
| 110 | + }); |
| 111 | + } catch (error: any) { |
| 112 | + if (flags.json) { await emitJson({ error: error.message }, 0, { compact: true }); this.exit(1); } |
| 113 | + printError(error.message || String(error)); |
| 114 | + this.exit(1); |
| 115 | + return; |
| 116 | + } |
| 117 | + |
| 118 | + try { |
| 119 | + // Typed off the slot's contract, not erased to `any` (#4168/#4176/#4251): |
| 120 | + // the journal reads below are exactly the surface `IObjectQLEngine` |
| 121 | + // declares, so there is nothing here that needs the checking switched off. |
| 122 | + const engine: IObjectQLEngine = stack.kernel.getService('objectql'); |
| 123 | + if (typeof engine?.find !== 'function') { |
| 124 | + throw new Error('No ObjectQL engine on this stack — cannot read the migration journal.'); |
| 125 | + } |
| 126 | + |
| 127 | + let plans: MigrationPlanProvider | undefined; |
| 128 | + try { |
| 129 | + plans = stack.kernel.getService('migration-plans') as MigrationPlanProvider; |
| 130 | + } catch { |
| 131 | + // No registry service composed — every run reports as unresumable, |
| 132 | + // which is the truthful answer for this process. |
| 133 | + } |
| 134 | + |
| 135 | + const interrupted = await findInterruptedRuns(engine); |
| 136 | + |
| 137 | + // ── list mode (no --run): read-only ────────────────────────────── |
| 138 | + if (!flags.run) { |
| 139 | + if (flags.json) { |
| 140 | + await emitJson( |
| 141 | + { |
| 142 | + interrupted: interrupted.map((r) => ({ ...r, resumable: Boolean(plans?.get(r.planId)) })), |
| 143 | + count: interrupted.length, |
| 144 | + }, |
| 145 | + timer.elapsed(), |
| 146 | + ); |
| 147 | + return; |
| 148 | + } |
| 149 | + if (interrupted.length === 0) { |
| 150 | + printSuccess('No interrupted migration runs — every run in the journal concluded.'); |
| 151 | + return; |
| 152 | + } |
| 153 | + printWarning(`${interrupted.length} interrupted migration run(s):`); |
| 154 | + for (const run of interrupted) this.log(` ${describeInterruptedRun(run, plans)}`); |
| 155 | + printInfo('Nothing was changed. Re-run with --run <id> to act on one.'); |
| 156 | + return; |
| 157 | + } |
| 158 | + |
| 159 | + // ── act mode (--run) ───────────────────────────────────────────── |
| 160 | + const target = interrupted.find((r) => r.runId === flags.run); |
| 161 | + if (!target) { |
| 162 | + // Distinguish "no such run" from "that run already concluded" — the |
| 163 | + // second is a success the operator should not be alarmed by. |
| 164 | + const events = await readRunJournal(engine, flags.run); |
| 165 | + const msg = events.length === 0 |
| 166 | + ? `No journal rows for run '${flags.run}'.` |
| 167 | + : `Run '${flags.run}' is not interrupted — it already concluded (${ |
| 168 | + events.some((e) => e.kind === 'run_done') ? 'run_done' : 'fully compensated' |
| 169 | + }). Nothing to do.`; |
| 170 | + if (flags.json) { await emitJson({ error: msg, runId: flags.run }, timer.elapsed(), { compact: true }); this.exit(events.length === 0 ? 1 : 0); return; } |
| 171 | + if (events.length === 0) { printError(msg); this.exit(1); return; } |
| 172 | + printSuccess(msg); |
| 173 | + return; |
| 174 | + } |
| 175 | + |
| 176 | + const plan = plans?.get(target.planId); |
| 177 | + if (!plan) { |
| 178 | + const msg = |
| 179 | + `Run '${target.runId}' belongs to plan '${target.planId}', which no loaded package registers. ` + |
| 180 | + `A resume needs the plan's code — the journal stores its hash, not its callbacks. ` + |
| 181 | + `Load the package that owns this migration and re-run.`; |
| 182 | + if (flags.json) { await emitJson({ error: msg, runId: target.runId, planId: target.planId }, timer.elapsed(), { compact: true }); this.exit(1); return; } |
| 183 | + printError(msg); |
| 184 | + this.exit(1); |
| 185 | + return; |
| 186 | + } |
| 187 | + |
| 188 | + const policy = plan.onCrash ?? 'resume'; |
| 189 | + if (!flags.yes) { |
| 190 | + const summary = `${policy === 'compensate' ? 'UNWIND' : 'RESUME FORWARD'} run '${target.runId}' (plan '${plan.id}')`; |
| 191 | + if (flags.json || !process.stdin.isTTY) { |
| 192 | + const msg = `Confirmation required: ${summary}. Re-run with --yes.`; |
| 193 | + if (flags.json) { await emitJson({ error: 'confirmation_required', hint: 'pass --yes', summary }, timer.elapsed(), { compact: true }); this.exit(1); return; } |
| 194 | + printWarning(msg); |
| 195 | + this.exit(1); |
| 196 | + return; |
| 197 | + } |
| 198 | + this.log(''); |
| 199 | + this.log(` ${describeInterruptedRun(target, plans)}`); |
| 200 | + const ok = await confirm(chalk.bold(`\n${summary}? [y/N] `)); |
| 201 | + if (!ok) { printInfo('Aborted — nothing changed.'); return; } |
| 202 | + } |
| 203 | + |
| 204 | + if (!flags.json) printStep(policy === 'compensate' ? 'Unwinding…' : 'Resuming forward…'); |
| 205 | + |
| 206 | + const result = await resumeMigrationJournal(engine, plan, target.runId); |
| 207 | + |
| 208 | + if (flags.json) { |
| 209 | + await emitJson({ ...result, error: result.error ? String(result.error) : undefined }, timer.elapsed()); |
| 210 | + // A run that ended `failed` left the database in a state no clean story |
| 211 | + // covers, so the exit code has to say so — a zero here would let a |
| 212 | + // scripted recovery move on from a migration that needs a human. |
| 213 | + this.exit(result.status === 'failed' ? 1 : 0); |
| 214 | + return; |
| 215 | + } |
| 216 | + |
| 217 | + if (result.status === 'completed') { |
| 218 | + printSuccess( |
| 219 | + `Run '${result.runId}' completed — ${result.chunksCommitted}/${result.chunksTotal} chunk(s) committed.`, |
| 220 | + ); |
| 221 | + } else if (result.status === 'compensated') { |
| 222 | + printWarning( |
| 223 | + `Run '${result.runId}' was unwound — ${result.chunksCompensated} chunk(s) compensated. ` + |
| 224 | + `The database is back to its pre-run state for this plan.`, |
| 225 | + ); |
| 226 | + } else { |
| 227 | + printError( |
| 228 | + `Run '${result.runId}' FAILED and its compensation did not finish. ` + |
| 229 | + `${result.chunksCommitted} chunk(s) committed, ${result.chunksCompensated} compensated — ` + |
| 230 | + `the remainder are still applied. Inspect sys_migration_journal for run '${result.runId}'; ` + |
| 231 | + `this needs a decision, not a retry.`, |
| 232 | + ); |
| 233 | + this.exit(1); |
| 234 | + } |
| 235 | + } catch (error: any) { |
| 236 | + const msg = error instanceof MigrationJournalRefusal |
| 237 | + // A refusal is the runner working, not breaking — say what it refused. |
| 238 | + ? `Refused (${error.code}): ${error.message}` |
| 239 | + : (error?.message || String(error)); |
| 240 | + if (flags.json) { await emitJson({ error: msg }, timer.elapsed(), { compact: true }); this.exit(1); return; } |
| 241 | + printError(msg); |
| 242 | + this.exit(1); |
| 243 | + } finally { |
| 244 | + try { await stack.shutdown?.(); } catch { /* best effort */ } |
| 245 | + } |
| 246 | + } |
| 247 | +} |
0 commit comments