feat(import): job-level undo (logical rollback)

Add POST /data/import/jobs/:jobId/undo — logically reverses a finished
import: deletes the records it created and restores the fields it updated
to their pre-import values.

- import-runner: capture an ImportUndoLog (created ids + per-updated
  before-snapshots of only the written fields) when captureUndo is set.
- sys_import_job: persist undo_log (json) + reverted_at (datetime).
- rest-server: worker captures undo for non-dry-run jobs <= 5000 rows
  (IMPORT_JOB_UNDO_MAX_ROWS); undo route deletes/restores with automations
  skipped, then stamps reverted_at. undoable/revertedAt surfaced on
  progress + summary DTOs.
- spec: undoable/revertedAt on progress+summary; UndoImportJobResponse;
  undoImportJob contract.
- client SDK: data.undoImportJob(jobId) on both namespaces + test.
- tests: end-to-end undo (delete created + restore updated), double-undo
  409, unknown-job 404.

Author: baozhoutao

packages/client/src/client.test.ts

@@ -1098,4 +1098,15 @@ describe('Import-job namespace', () => {
         expect(init.method).toBe('POST');
         expect(res.success).toBe(true);
     });
+
+    it('undoImportJob POSTs the /undo sub-route', async () => {
+        const { client, fetchMock } = createMockClient({ success: true, jobId: 'imp_x', object: 'task', deleted: 3, restored: 2, failed: 0 });
+        const res = await client.data.undoImportJob('imp_x');
+        const [url, init] = fetchMock.mock.calls[0];
+        expect(url).toBe('http://localhost:3000/api/v1/data/import/jobs/imp_x/undo');
+        expect(init.method).toBe('POST');
+        expect(res.success).toBe(true);
+        expect(res.deleted).toBe(3);
+        expect(res.restored).toBe(2);
+    });
 });

packages/client/src/index.ts

@@ -91,6 +91,7 @@ import {
   ImportJobSummary,
   ListImportJobsRequest,
   ListImportJobsResponse,
+  UndoImportJobResponse,
 } from '@objectstack/spec/api';
 import type {
   ApprovalRequestRow,
@@ -3130,6 +3131,20 @@ export class ObjectStackClient {
         return this.unwrapResponse<{ success: boolean }>(res);
     },
 
+    /**
+     * Logically roll back a finished import: delete the records it created and
+     * restore the fields it updated to their pre-import values. Only jobs that
+     * captured an undo log (small, non-dry-run, not yet reverted) are undoable —
+     * others return 422. See {@link ImportJobProgress.undoable}.
+     */
+    undoImportJob: async (jobId: string): Promise<UndoImportJobResponse> => {
+        const route = this.getRoute('data');
+        const res = await this.fetch(`${this.baseUrl}${route}/import/jobs/${encodeURIComponent(jobId)}/undo`, {
+            method: 'POST',
+        });
+        return this.unwrapResponse<UndoImportJobResponse>(res);
+    },
+
     update: async <T = any>(
         object: string,
         id: string,
@@ -3592,6 +3607,12 @@ export class ScopedProjectClient {
       });
       return this.parent._unwrap<{ success: boolean }>(res);
     },
+    undoImportJob: async (jobId: string): Promise<UndoImportJobResponse> => {
+      const res = await this.parent._fetch(this.url(`/data/import/jobs/${encodeURIComponent(jobId)}/undo`), {
+        method: 'POST',
+      });
+      return this.parent._unwrap<UndoImportJobResponse>(res);
+    },
     update: async <T = any>(object: string, id: string, data: Partial<T>): Promise<UpdateDataResult<T>> => {
       const res = await this.parent._fetch(this.url(`/data/${object}/${id}`), {
         method: 'PATCH',
@@ -3781,6 +3802,7 @@ export type {
   ImportJobSummary,
   ListImportJobsRequest,
   ListImportJobsResponse,
+  UndoImportJobResponse,
 } from '@objectstack/spec/api';
 
 // Approval runtime types (ADR-0019) — surfaced so SDK consumers can type the

packages/platform-objects/src/audit/sys-import-job.object.ts

@@ -74,6 +74,20 @@ export const SysImportJob = ObjectSchema.create({
       group: 'Outcome',
     }),
 
+    // ── undo / logical rollback ──
+    undo_log: Field.json({
+      label: 'Undo Log',
+      required: false,
+      description: 'Reversal instructions ({created:[ids], updated:[{id,before}]}) captured for small non-dry-run jobs so the import can be undone',
+      group: 'Outcome',
+    }),
+    reverted_at: Field.datetime({
+      label: 'Reverted At',
+      required: false,
+      description: 'Set when the import was undone (created records deleted, updated records restored)',
+      group: 'Outcome',
+    }),
+
     // ── lifecycle timestamps ──
     started_at: Field.datetime({ label: 'Started At', required: false, group: 'State' }),
     completed_at: Field.datetime({ label: 'Completed At', required: false, group: 'State' }),

packages/rest/src/import-job-integration.test.ts

@@ -140,6 +140,7 @@ async function boot() {
     results: find('GET', '/api/v1/data/import/jobs/:jobId/results'),
     list: find('GET', '/api/v1/data/import/jobs'),
     cancel: find('POST', '/api/v1/data/import/jobs/:jobId/cancel'),
+    undo: find('POST', '/api/v1/data/import/jobs/:jobId/undo'),
   };
 }
 
@@ -168,7 +169,7 @@ describe('async import job — real engine + protocol integration', () => {
   beforeEach(async () => {
     ctx = await boot();
     expect(ctx.create).toBeDefined();
-    expect(ctx.progress && ctx.results && ctx.list && ctx.cancel).toBeTruthy();
+    expect(ctx.progress && ctx.results && ctx.list && ctx.cancel && ctx.undo).toBeTruthy();
   });
 
   it('creates a job, processes it in the background, and persists rows', async () => {
@@ -243,4 +244,56 @@ describe('async import job — real engine + protocol integration', () => {
     const after = await callJob(ctx.progress, jobId);
     expect(after._json.status).toBe('succeeded');
   });
+
+  it('undoes a job: deletes created records and restores updated ones', async () => {
+    // Seed one existing record so an upsert both creates and updates.
+    await ctx.protocol.createData({ object: 'task', data: { id: 'u_existing', title: 'old title', score: 1 } });
+
+    const created = await callCreate(ctx.create, {
+      format: 'json', writeMode: 'upsert', matchFields: ['id'],
+      rows: [
+        { id: 'u_existing', title: 'new title', score: 99 }, // update
+        { id: 'u_new1', title: 'fresh one' },                // create
+        { id: 'u_new2', title: 'fresh two' },                // create
+      ],
+    });
+    const jobId = created._json.jobId;
+    const done = await waitForTerminal(ctx.progress, jobId);
+    expect(done).toMatchObject({ status: 'succeeded', created: 2, updated: 1 });
+    expect(done.undoable).toBe(true);
+
+    // Writes really landed.
+    expect(await ctx.engine.findOne('task', { where: { id: 'u_new1' } })).toBeTruthy();
+    expect(await ctx.engine.findOne('task', { where: { id: 'u_existing' } })).toMatchObject({ title: 'new title', score: 99 });
+
+    // Undo.
+    const u = await callJob(ctx.undo, jobId);
+    expect(u._json).toMatchObject({ success: true, deleted: 2, restored: 1, failed: 0 });
+
+    // Created records are gone; the updated record is back to its pre-import values.
+    expect(await ctx.engine.findOne('task', { where: { id: 'u_new1' } })).toBeFalsy();
+    expect(await ctx.engine.findOne('task', { where: { id: 'u_new2' } })).toBeFalsy();
+    expect(await ctx.engine.findOne('task', { where: { id: 'u_existing' } })).toMatchObject({ title: 'old title', score: 1 });
+
+    // Job now flags as reverted + no longer undoable.
+    const after = await callJob(ctx.progress, jobId);
+    expect(after._json.undoable).toBe(false);
+    expect(after._json.revertedAt).toBeTruthy();
+  });
+
+  it('undoing twice is rejected (409 already reverted)', async () => {
+    const created = await callCreate(ctx.create, { format: 'json', rows: [{ id: 'uu1', title: 'x' }] });
+    const jobId = created._json.jobId;
+    await waitForTerminal(ctx.progress, jobId);
+    const first = await callJob(ctx.undo, jobId);
+    expect(first._json.success).toBe(true);
+    const second = await callJob(ctx.undo, jobId);
+    expect(second._status).toBe(409);
+    expect(second._json.code).toBe('ALREADY_REVERTED');
+  });
+
+  it('404s undo for an unknown job id', async () => {
+    const res = await callJob(ctx.undo, 'imp_nope');
+    expect(res._status).toBe(404);
+  });
 });

packages/rest/src/import-runner.ts

@@ -35,10 +35,25 @@ export interface ImportProgress {
   errors: number;
 }
 
+/**
+ * Records exactly what a non-dry-run import changed, so the job can be undone:
+ * created records are deleted, and updated records have the touched fields
+ * restored to their pre-import values. Only the fields the import wrote are
+ * captured (keyed to `before`), keeping the log precise and bounded.
+ */
+export interface ImportUndoLog {
+  /** Ids of records this import created (delete to undo). */
+  created: string[];
+  /** Per updated record: the touched fields' values *before* the import. */
+  updated: Array<{ id: string; before: Record<string, any> }>;
+}
+
 export interface ImportRunSummary extends ImportProgress {
   ok: number;
   results: ImportRowResult[];
   cancelled: boolean;
+  /** Present only when `captureUndo` was set — the reversal instructions. */
+  undoLog?: ImportUndoLog;
 }
 
 /** Minimal protocol surface the runner needs (find / create / update). */
@@ -80,15 +95,31 @@ export interface RunImportOptions {
    * truthy the runner stops and returns `cancelled: true` with partial results.
    */
   shouldCancel?: () => boolean | Promise<boolean>;
+  /**
+   * When true (and not a dry run), accumulate an {@link ImportUndoLog} so the
+   * import can be reverted later. Callers gate this on row count to bound the
+   * stored snapshot size.
+   */
+  captureUndo?: boolean;
 }
 
 export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
   const {
     p, objectName, environmentId, context, rows, metaMap,
     writeMode, matchFields, dryRun, runAutomations,
     trimWhitespace, nullValues, createMissingOptions, skipBlankMatchKey,
-    onProgress, shouldCancel,
+    onProgress, shouldCancel, captureUndo,
   } = opts;
+  const collectUndo = !!captureUndo && !dryRun;
+  const undoLog: ImportUndoLog = { created: [], updated: [] };
+  // Snapshot only the fields the import touched, so undo restores exactly what
+  // changed. A field absent before the import is recorded as null → undo clears
+  // it. Never captured on dry runs (nothing was written).
+  const captureBefore = (before: Record<string, any>, written: Record<string, any>): Record<string, any> => {
+    const snap: Record<string, any> = {};
+    for (const k of Object.keys(written)) snap[k] = before[k] ?? null;
+    return snap;
+  };
   const progressEvery = Math.max(1, opts.progressEvery ?? 200);
 
   const findRows = (r: any): any[] =>
@@ -213,11 +244,15 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
               const res2 = await p.updateData({ object: objectName, id: target.id, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) });
               const id = (res2 as any)?.id ?? (res2 as any)?.record?.id ?? target.id;
               okCount++; updated++;
+              if (collectUndo && target.id != null) {
+                undoLog.updated.push({ id: String(target.id), before: captureBefore(target, data) });
+              }
               results.push({ row: rowNo, ok: true, action: 'updated', id: id != null ? String(id) : undefined });
             } else {
               const res2 = await p.createData({ object: objectName, data, context: writeCtx, ...(environmentId ? { environmentId } : {}) });
               const id = (res2 as any)?.id ?? (res2 as any)?.record?.id;
               okCount++; created++;
+              if (collectUndo && id != null) undoLog.created.push(String(id));
               results.push({ row: rowNo, ok: true, action: 'created', id: id != null ? String(id) : undefined });
             }
           }
@@ -238,6 +273,9 @@ export function runImport(opts: RunImportOptions): Promise<ImportRunSummary> {
       }
     }
 
-    return { ...snapshot(results.length), ok: okCount, results, cancelled };
+    return {
+      ...snapshot(results.length), ok: okCount, results, cancelled,
+      ...(collectUndo ? { undoLog } : {}),
+    };
   })();
 }

packages/rest/src/rest-server.ts

@@ -582,6 +582,9 @@ const IMPORT_JOB_OBJECT = 'sys_import_job';
 const IMPORT_JOB_MAX_ROWS = 50_000;
 /** Cap on per-row results persisted on the job (failures first). */
 const IMPORT_JOB_RESULTS_CAP = 500;
+/** Undo (logical rollback) is only recorded for jobs at or under this row
+ *  count — larger jobs skip the undo log to bound the stored before-snapshots. */
+const IMPORT_JOB_UNDO_MAX_ROWS = 5_000;
 
 /** Generate a sortable-ish, collision-resistant import job id. */
 function newImportJobId(): string {
@@ -597,11 +600,36 @@ function capImportResults(results: Array<{ ok: boolean }>): { items: any[]; trun
     return { items, truncated: true };
 }
 
+/** Parse the persisted undo log (json column may arrive as object or string). */
+function parseUndoLog(raw: any): { created: string[]; updated: Array<{ id: string; before: Record<string, any> }> } | undefined {
+    if (!raw) return undefined;
+    let v = raw;
+    if (typeof v === 'string') { try { v = JSON.parse(v); } catch { return undefined; } }
+    if (!v || typeof v !== 'object') return undefined;
+    const created = Array.isArray(v.created) ? v.created.map(String) : [];
+    const updated = Array.isArray(v.updated)
+        ? v.updated.filter((u: any) => u && u.id != null).map((u: any) => ({ id: String(u.id), before: u.before ?? {} }))
+        : [];
+    return { created, updated };
+}
+
+/** True when a job can still be undone: it wrote data (undo log present with
+ *  entries), hasn't already been reverted, and finished in a terminal state. */
+function importJobUndoable(row: any): boolean {
+    if (row?.reverted_at) return false;
+    const status = String(row?.status ?? '');
+    if (status !== 'succeeded' && status !== 'cancelled') return false;
+    const log = parseUndoLog(row?.undo_log);
+    return !!log && (log.created.length > 0 || log.updated.length > 0);
+}
+
 /** Map a persisted `sys_import_job` row to the ImportJobProgress DTO. */
 function importJobToProgress(row: any): Record<string, any> {
     const total = Number(row?.total_rows ?? 0);
     const processed = Number(row?.processed_rows ?? 0);
     return {
+        undoable: importJobUndoable(row),
+        ...(row?.reverted_at ? { revertedAt: String(row.reverted_at) } : {}),
         jobId: String(row?.id ?? ''),
         object: String(row?.object_name ?? ''),
         status: String(row?.status ?? 'pending'),
@@ -629,7 +657,9 @@ function importJobToSummary(row: any): Record<string, any> {
         total: p.total, processed: p.processed,
         created: p.created, updated: p.updated, skipped: p.skipped, errors: p.errors,
         createdAt: p.createdAt,
+        undoable: p.undoable,
         ...(p.completedAt ? { completedAt: p.completedAt } : {}),
+        ...(p.revertedAt ? { revertedAt: p.revertedAt } : {}),
     };
 }
 
@@ -3583,11 +3613,15 @@ export class RestServer {
                             logError('[REST] import job progress write failed:', err);
                         }
                     };
+                    // Record undo instructions for small non-dry-run jobs so the
+                    // import can be logically rolled back later.
+                    const captureUndo = !prepared.dryRun && prepared.rows.length <= IMPORT_JOB_UNDO_MAX_ROWS;
                     void (async () => {
                         await patch({ status: 'running', started_at: new Date().toISOString() });
                         try {
                             const summary = await runImport({
                                 p, objectName, environmentId, context, ...prepared,
+                                captureUndo,
                                 progressEvery: 200,
                                 onProgress: (pr) => patch({
                                     processed_rows: pr.processed,
@@ -3607,6 +3641,7 @@ export class RestServer {
                                 error_count: summary.errors,
                                 results: capImportResults(summary.results),
                                 completed_at: new Date().toISOString(),
+                                ...(summary.undoLog ? { undo_log: summary.undoLog } : {}),
                             });
                         } catch (err: any) {
                             await patch({
@@ -3677,6 +3712,67 @@ export class RestServer {
             metadata: { summary: 'Cancel an in-flight import job', tags: ['data', 'import'] },
         });
 
+        // POST /data/import/jobs/:jobId/undo — logical rollback of a finished
+        // job: delete the records it created and restore the fields it updated
+        // to their pre-import values (from the captured undo log).
+        this.routeManager.register({
+            method: 'POST',
+            path: `${dataPath}/import/jobs/:jobId/undo`,
+            handler: async (req: any, res: any) => {
+                try {
+                    const environmentId = isScoped ? req.params?.environmentId : undefined;
+                    const p = await this.resolveProtocol(environmentId, req);
+                    const context = await this.resolveExecCtx(environmentId, req);
+                    if (this.enforceAuth(req, res, context)) return;
+                    const jobId = String(req.params.jobId || '');
+                    const row = await loadImportJob(p, jobId, environmentId, context);
+                    if (!row) {
+                        res.status(404).json({ code: 'NOT_FOUND', error: `No import job ${jobId}` });
+                        return;
+                    }
+                    if (row.reverted_at) {
+                        res.status(409).json({ code: 'ALREADY_REVERTED', error: 'This import has already been undone' });
+                        return;
+                    }
+                    if (!importJobUndoable(row)) {
+                        res.status(422).json({ code: 'NOT_UNDOABLE', error: 'This import cannot be undone (too large, still running, or nothing was written)' });
+                        return;
+                    }
+                    const objectName = String(row.object_name ?? '');
+                    const log = parseUndoLog(row.undo_log)!;
+                    // Undo automations too: reversing writes shouldn't re-fire triggers.
+                    const writeCtx = { ...(context ?? {}), skipAutomations: true };
+                    let deleted = 0, restored = 0, failed = 0;
+
+                    // Delete created records first (they didn't exist before).
+                    for (const id of log.created) {
+                        try {
+                            await (p as any).deleteData({ object: objectName, id, context: writeCtx, ...(environmentId ? { environmentId } : {}) });
+                            deleted++;
+                        } catch { failed++; }
+                    }
+                    // Restore the touched fields on updated records.
+                    for (const u of log.updated) {
+                        try {
+                            await (p as any).updateData({ object: objectName, id: u.id, data: u.before, context: writeCtx, ...(environmentId ? { environmentId } : {}) });
+                            restored++;
+                        } catch { failed++; }
+                    }
+
+                    await (p as any).updateData({
+                        object: IMPORT_JOB_OBJECT, id: jobId,
+                        data: { reverted_at: new Date().toISOString() },
+                        context, ...(environmentId ? { environmentId } : {}),
+                    });
+                    res.json({ success: true, jobId, object: objectName, deleted, restored, failed });
+                } catch (error: any) {
+                    logError('[REST] Unhandled error:', error);
+                    sendError(res, error, '');
+                }
+            },
+            metadata: { summary: 'Undo (logically roll back) a finished import job', tags: ['data', 'import'] },
+        });
+
         // GET /data/import/jobs/:jobId/results — progress + capped per-row report.
         this.routeManager.register({
             method: 'GET',