feat(import): stream real Arcim migration progress as NDJSON (#1485)

* feat(import): stream real Arcim migration progress as NDJSON The /migrate route ran the orchestrator to completion and answered with one JSON blob, so the wizard faked its progress bar: a hardcoded 55% anchor and a static step label for a phase that can take minutes. The orchestrator has had a real onProgress channel (eight emit points with Swedish step labels and anchors) since it was written; the route just never passed it. Now a request with Accept: application/x-ndjson gets a streamed response: one line per orchestrator progress event, then a terminal done line with the results or an error line carrying the same structured envelope the JSON path returns (the 200 status is already committed once the stream opens). Callers without the header keep the original single-JSON contract, so pre-deploy tabs and the existing error-mapping tests are untouched. The wizard opts in, drives MigratingStep from the real labels and anchors (mapped onto the 55-100 slice of the wizard bar), and treats a dropped connection as unconfirmed rather than failed, since the migration keeps running server-side and a blind retry could double-import. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs: record the opt-in NDJSON streaming decision for /migrate Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-08-10 09:26:27 +02:00
parent 4ccebd3645
commit 1d01010928
4 changed files with 322 additions and 19 deletions
@@ -847,4 +847,5 @@ One line per decision: `[YYYY-MM-DD] <decision>: <why>`. Appended by agents and
 [2026-08-08] extensions.schema.json enum also gained "stripe" while adding "shopify": the enum had drifted (stripe was enabled in extensions.config.json but missing from the schema, failing editor validation); fixed in the same touch since the file had to change anyway.
 [2026-08-08] Login panel is method-stated (BankID hero default, remembered via accounted-login-method cookie) instead of a stacked method list: matches the Swedish bank/Fortnox convention, gives exactly one primary action per view; errors moved from boxed banner to a field-adjacent single line (NN/g 3/4/10), reset link surfaces from the second failed attempt.
 [2026-08-09] Upsell FAB dismissal is session-scoped (sessionStorage), not persisted: closing the paywalled sheet or the pill's X hides all floating assistant UI for non-payers until the next browser session; a permanent dismissal would let one click silence the conversion surface forever, and payer/collapsed FAB behavior stays untouched.
+[2026-08-09] /migrate streaming is opt-in via Accept: application/x-ndjson instead of replacing the JSON contract: the wizard is the only caller today but a hard cutover would break open pre-deploy tabs and the route's locked error-status tests; mid-stream failures re-send the structured envelope as a terminal error event because the 200 is already committed once the stream opens.
 [2026-08-09] Regeluppdat + docs-freshness scans (#1417) built as local loop skills with due-date self-gating, not cloud crons: cloud routines were retired 2026-07-20, and session crons die at 7 days, so weekly/monthly cadence is achieved by loop-ignite running each loop when its run marker says it is due. loop-regeluppdat files tickets only (no auto-fix PRs): regulatory changes touch money math and compliance logic, which .claude/loops.md forbids loops from changing. Docs check diffs the live .md mirror routes against repo-built markdown (exact, canonicalised both sides) instead of diffing the gnubok-website checkout, so it also catches deployed-but-stale and route-404 states.
@@ -101,6 +101,72 @@ function displayError(err: unknown, nonErrorFallback?: string): string {
  return getUserErrorMessage(err)
 }

+/**
+ * Read the /migrate NDJSON stream: one JSON object per line. `progress`
+ * events carry the orchestrator's real step labels and anchors; the stream
+ * ends with a terminal `done` line (results) or `error` line (the same
+ * structured envelope the JSON path answers with, thrown here so the catch
+ * block shows it verbatim).
+ */
+async function consumeMigrationStream(
+  body: ReadableStream<Uint8Array>,
+  onProgress: (currentStep: string | undefined, progress: number) => void,
+): Promise<MigrationResults> {
+  const reader = body.getReader()
+  const decoder = new TextDecoder()
+  let buffer = ''
+  let results: MigrationResults | undefined
+
+  const handleLine = (line: string) => {
+    if (!line.trim()) return
+    let event: {
+      kind?: string
+      currentStep?: string
+      progress?: number
+      results?: MigrationResults
+    }
+    try {
+      event = JSON.parse(line)
+    } catch {
+      return // torn line from an intermediary flush; terminal lines are whole
+    }
+    if (event.kind === 'progress' && typeof event.progress === 'number') {
+      onProgress(
+        typeof event.currentStep === 'string' && event.currentStep ? event.currentStep : undefined,
+        event.progress,
+      )
+    } else if (event.kind === 'done') {
+      results = event.results ?? {}
+    } else if (event.kind === 'error') {
+      throw apiError(event, 'Migreringen misslyckades.')
+    }
+  }
+
+  try {
+    for (;;) {
+      const { value, done } = await reader.read()
+      if (done) break
+      buffer += decoder.decode(value, { stream: true })
+      const lines = buffer.split('\n')
+      buffer = lines.pop() ?? ''
+      for (const line of lines) handleLine(line)
+    }
+    buffer += decoder.decode()
+    if (buffer) handleLine(buffer)
+  } finally {
+    reader.releaseLock()
+  }
+
+  if (!results) {
+    // The connection dropped before the terminal line. The migration keeps
+    // running server-side, so a blind retry could double-import.
+    throw new UserFacingError(
+      'Anslutningen bröts innan migreringen bekräftades. Ladda om sidan och kontrollera om kunder och fakturor redan har importerats innan du försöker igen.'
+    )
+  }
+  return results
+}
+
 /** Pull the structured error `code` from an envelope, if present. */
 function apiErrorCode(data: unknown): string | null {
  const err = (data as { error?: unknown } | null)?.error
@@ -2381,7 +2447,7 @@ export default function ArcimMigrationWorkspace({

        const res = await fetch('/api/extensions/ext/arcim-migration/migrate', {
          method: 'POST',
-          headers: { 'Content-Type': 'application/json' },
+          headers: { 'Content-Type': 'application/json', Accept: 'application/x-ndjson' },
          body: JSON.stringify({
            consentId,
            importCompanyInfo: migrationOptions.importCompanyInfo,
@@ -2397,9 +2463,23 @@ export default function ArcimMigrationWorkspace({
          throw apiError(data, `HTTP ${res.status}`)
        }

-        const data = await res.json()
-        setMigrationResults(data.results)
-        hadStepErrors = ((data.results as MigrationResults | undefined)?.stepErrors?.length ?? 0) > 0
+        const contentType = res.headers.get('content-type') ?? ''
+        let results: MigrationResults | undefined
+        if (contentType.includes('application/x-ndjson') && res.body) {
+          results = await consumeMigrationStream(res.body, (currentStep, progress) => {
+            if (currentStep) setMigrationStep(currentStep)
+            // The orchestrator reports 0-100 on its own scale; the wizard bar
+            // reserves 55-100 for the entity phase (SIE holds 10-50).
+            setMigrationProgress(55 + Math.round(progress * 0.45))
+          })
+        } else {
+          // Pre-stream server (or a proxy that stripped the stream): the
+          // original single-JSON contract.
+          const data = await res.json()
+          results = data.results as MigrationResults | undefined
+        }
+        setMigrationResults(results ?? null)
+        hadStepErrors = (results?.stepErrors?.length ?? 0) > 0
      }

      // Mark consent as fully accepted now that import is complete
@@ -0,0 +1,161 @@
+import { describe, it, expect, beforeEach, vi, type Mock } from 'vitest'
+import { createMockSupabase } from '@/tests/helpers'
+import type { ExtensionContext } from '@/lib/extensions/types'
+
+/**
+ * Locks the /migrate NDJSON streaming contract (opt-in via
+ * `Accept: application/x-ndjson`):
+ *
+ *  - each orchestrator onProgress call becomes one `progress` line with the
+ *    real step label and anchor (the wizard's fake 55→100 jump is dead)
+ *  - success ends with a terminal `done` line carrying the results, after
+ *    acceptConsent ran
+ *  - an orchestrator failure ends with a terminal `error` line carrying the
+ *    SAME structured envelope the JSON path answers with (the 200 status is
+ *    already committed when the stream opens)
+ *  - a request WITHOUT the Accept header keeps the original JSON contract
+ */
+
+vi.mock('../lib/migration-orchestrator', () => ({
+  executeMigration: vi.fn(),
+}))
+
+vi.mock('../lib/provider-client', () => ({
+  createConsent: vi.fn(),
+  getConsent: vi.fn(),
+  listConsents: vi.fn(),
+  generateOtc: vi.fn(),
+  consumeOAuthState: vi.fn(),
+  getAuthUrl: vi.fn(),
+  exchangeAuthToken: vi.fn(),
+  submitProviderToken: vi.fn(),
+  acceptConsent: vi.fn().mockResolvedValue(undefined),
+  deleteConsent: vi.fn(),
+  resolveConsent: vi.fn(),
+  fetchCompanyInfoDirect: vi.fn(),
+  ProviderTokenInvalidError: class ProviderTokenInvalidError extends Error {},
+  ConsentNotFoundError: class ConsentNotFoundError extends Error {},
+}))
+
+import { arcimMigrationExtension } from '../index'
+import { executeMigration } from '../lib/migration-orchestrator'
+import { getConsent, acceptConsent } from '../lib/provider-client'
+
+const migrateRoute = (arcimMigrationExtension.apiRoutes ?? []).find(
+  (r) => r.method === 'POST' && r.path === '/migrate',
+)!
+
+type RouteHandler = (request: Request, ctx?: ExtensionContext) => Promise<Response>
+const handler = migrateRoute.handler as RouteHandler
+
+function buildCtx(): ExtensionContext {
+  const { supabase, mockResult } = createMockSupabase()
+  // The SIE guard awaits `from('sie_imports').select(..,{count,head}).eq().eq()`.
+  mockResult({ count: 1 })
+  ;(supabase as unknown as { auth: unknown }).auth = {
+    getUser: vi.fn().mockResolvedValue({ data: { user: { id: 'user-1' } } }),
+  }
+  return { supabase, companyId: 'company-1' } as unknown as ExtensionContext
+}
+
+function streamRequest() {
+  return new Request('http://localhost/api/extensions/ext/arcim-migration/migrate', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json', Accept: 'application/x-ndjson' },
+    body: JSON.stringify({ consentId: 'consent-1' }),
+  })
+}
+
+async function readNdjson(res: Response): Promise<Record<string, unknown>[]> {
+  const text = await res.text()
+  return text
+    .split('\n')
+    .filter((line) => line.trim())
+    .map((line) => JSON.parse(line))
+}
+
+describe('POST /migrate: NDJSON streaming', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    ;(getConsent as Mock).mockResolvedValue({ id: 'consent-1', status: 1, provider: 'visma' })
+  })
+
+  it('streams one progress line per orchestrator event, then a terminal done line with results', async () => {
+    const results = {
+      customers: { total: 5, imported: 5, updated: 0, skipped: 0, skipReasons: {} },
+      stepErrors: [],
+    }
+    ;(executeMigration as Mock).mockImplementation(
+      async (opts: { onProgress?: (p: { status: string; currentStep?: string; progress: number }) => void }) => {
+        opts.onProgress?.({ status: 'fetching', currentStep: 'Ansluter till Visma eEkonomi...', progress: 5 })
+        opts.onProgress?.({ status: 'importing', currentStep: 'Importerar kunder...', progress: 20 })
+        opts.onProgress?.({ status: 'completed', currentStep: 'Klart!', progress: 100 })
+        return results
+      },
+    )
+
+    const res = await handler(streamRequest(), buildCtx())
+
+    expect(res.status).toBe(200)
+    expect(res.headers.get('content-type')).toContain('application/x-ndjson')
+
+    const events = await readNdjson(res)
+    expect(events).toHaveLength(4)
+    expect(events[0]).toEqual({
+      kind: 'progress',
+      status: 'fetching',
+      currentStep: 'Ansluter till Visma eEkonomi...',
+      progress: 5,
+    })
+    expect(events[1]).toMatchObject({ kind: 'progress', currentStep: 'Importerar kunder...', progress: 20 })
+    expect(events[3]).toEqual({ kind: 'done', success: true, results })
+    expect(acceptConsent).toHaveBeenCalledWith('consent-1')
+  })
+
+  it('ends with a terminal error line carrying the structured envelope when the orchestrator fails', async () => {
+    const vismaError = new Error('Visma API error: 403 Forbidden') as Error & {
+      statusCode: number
+      body: string
+    }
+    vismaError.statusCode = 403
+    vismaError.body =
+      '{"ErrorCode":4002,"DeveloperErrorMessage":"ForbiddenRequestException - No access to module: api_standard","ErrorId":"x","Errors":[]}'
+    ;(executeMigration as Mock).mockImplementation(
+      async (opts: { onProgress?: (p: { status: string; progress: number }) => void }) => {
+        opts.onProgress?.({ status: 'fetching', progress: 5 })
+        throw vismaError
+      },
+    )
+
+    const res = await handler(streamRequest(), buildCtx())
+    expect(res.status).toBe(200)
+
+    const events = await readNdjson(res)
+    const terminal = events[events.length - 1] as {
+      kind: string
+      error: { code: string; message: string }
+    }
+    expect(terminal.kind).toBe('error')
+    expect(terminal.error.code).toBe('PROVIDER_API_MODULE_INACTIVE')
+    expect(terminal.error.message).toContain('Appar och tillägg')
+    expect(acceptConsent).not.toHaveBeenCalled()
+  })
+
+  it('keeps the single-JSON contract when the Accept header is absent', async () => {
+    const results = { customers: { total: 1, imported: 1, updated: 0, skipped: 0, skipReasons: {} } }
+    ;(executeMigration as Mock).mockResolvedValue(results)
+
+    const res = await handler(
+      new Request('http://localhost/api/extensions/ext/arcim-migration/migrate', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ consentId: 'consent-1' }),
+      }),
+      buildCtx(),
+    )
+
+    expect(res.status).toBe(200)
+    expect(res.headers.get('content-type')).toContain('application/json')
+    expect(await res.json()).toEqual({ success: true, results })
+  })
+})
@@ -129,6 +129,27 @@ async function buildArcimOAuthUrl(consentId: string, provider: ArcimProvider): P
  return url
 }

+/**
+ * Map a failed /migrate run to its structured error response. Shared by the
+ * JSON path (returned as-is) and the NDJSON path (body re-sent as the
+ * terminal `error` event, since the stream's 200 status is already
+ * committed). Consent missing or owned by another company: same 404 either way.
+ */
+function migrateFailureResponse(error: unknown, consentId: string): NextResponse {
+  if (error instanceof ConsentNotFoundError) {
+    return errorResponseFromCode('PROVIDER_CONSENT_NOT_FOUND', moduleLog, {
+      details: { consentId },
+    })
+  }
+  const classified = classifyProviderError(error)
+  return errorResponseFromCode(classified ?? 'PROVIDER_MIGRATE_FAILED', moduleLog, {
+    details: {
+      reason: error instanceof Error ? error.message : 'unknown',
+      classified: classified ?? 'unclassified',
+    },
+  })
+}
+
 /**
 * Provider Migration extension
 *
@@ -1093,7 +1114,7 @@ export const arcimMigrationExtension: Extension = {

          log.info(`Starting migration for user ${user.id} from ${consent.provider}`)

-          const results = await executeMigration({
+          const migrationOptions = {
            consentId,
            companyId,
            userId: user.id,
@@ -1104,7 +1125,59 @@ export const arcimMigrationExtension: Extension = {
            importSalesInvoices,
            importSupplierInvoices,
            reconcileVouchers,
-          })
+          }
+
+          // Streaming mode (the migration wizard opts in via Accept): one
+          // NDJSON line per orchestrator progress event, then a terminal
+          // `done` or `error` line. Errors after the stream opens cannot
+          // change the HTTP status, so the terminal `error` line carries the
+          // same structured envelope the JSON path answers with. Callers
+          // without the Accept header keep the original JSON contract.
+          if ((request.headers.get('accept') ?? '').includes('application/x-ndjson')) {
+            const encoder = new TextEncoder()
+            const stream = new ReadableStream<Uint8Array>({
+              async start(controller) {
+                const send = (event: Record<string, unknown>) => {
+                  try {
+                    controller.enqueue(encoder.encode(JSON.stringify(event) + '\n'))
+                  } catch {
+                    // Reader cancelled (tab closed, navigation). The migration
+                    // keeps running server-side; we just stop narrating.
+                  }
+                }
+                try {
+                  const results = await executeMigration({
+                    ...migrationOptions,
+                    onProgress: (p) => send({
+                      kind: 'progress',
+                      status: p.status,
+                      currentStep: p.currentStep,
+                      progress: p.progress,
+                    }),
+                  })
+                  log.info('Migration completed:', results)
+                  // Mark consent as fully accepted now that data has been imported
+                  await acceptConsent(consentId)
+                  send({ kind: 'done', success: true, results })
+                } catch (error) {
+                  log.error('arcim migration failed', error as Error)
+                  const envelope = await migrateFailureResponse(error, consentId).json()
+                  send({ kind: 'error', ...envelope })
+                } finally {
+                  controller.close()
+                }
+              },
+            })
+            return new Response(stream, {
+              headers: {
+                'Content-Type': 'application/x-ndjson; charset=utf-8',
+                'Cache-Control': 'no-store',
+                'X-Accel-Buffering': 'no',
+              },
+            })
+          }
+
+          const results = await executeMigration(migrationOptions)

          log.info('Migration completed:', results)

@@ -1114,19 +1187,7 @@ export const arcimMigrationExtension: Extension = {
          return NextResponse.json({ success: true, results })
        } catch (error) {
          log.error('arcim migration failed', error as Error)
-          // Consent missing or owned by another company: same 404 either way.
-          if (error instanceof ConsentNotFoundError) {
-            return errorResponseFromCode('PROVIDER_CONSENT_NOT_FOUND', moduleLog, {
-              details: { consentId },
-            })
-          }
-          const classified = classifyProviderError(error)
-          return errorResponseFromCode(classified ?? 'PROVIDER_MIGRATE_FAILED', moduleLog, {
-            details: {
-              reason: error instanceof Error ? error.message : 'unknown',
-              classified: classified ?? 'unclassified',
-            },
-          })
+          return migrateFailureResponse(error, consentId)
        }
      },
    },