The firm import service, v2: how it actually works

Reference layer underneath this plan: data provenance, import runs, unit model, import boundary schemas, hierarchical unit load walk, and the unit load runtime walkthrough, a step-by-step simulation of the read path this plan builds on.

What changed since the first doc

The v1 doc proposed the shape and asked one question; a week of review, three sibling docs, a simulated run and a working MVP later, every open fork has an answer. The deltas that matter, so nobody relitigates them here:

The packet this service reads

Upstream of this service, the data pipeline (dbt transforms in BigQuery, exported by Dagster) produces one packet per import: a single prefix holding everything a run needs.

• One sorted JSONL file per boundary table: seventeen of them, roots sorted by primary key, children by parent ref then primary key, UTF-8, snake_case.
• A manifest: a JSON file recording, per data file, its row count and SHA-256 checksum, plus the source dataset it describes (the source system's firm id, or an agreed label where the source has no firm concept) and, optionally, the boundary contract version it was built against. The run pins itself to this manifest, and the end-of-run reconciliation is checked against its counts.
• The firm's static file blobs, referenced from the data rows by payload URI and content hash.

How the packet is produced, scanned and stored is its own workstream and out of scope here; this plan only cares that the service can read one. In deployed environments the packet lives on S3; locally it is a directory. The DumpSource seam (openStream(entity), manifest()) hides the difference, and everything downstream of the streams is identical.

The importer trusts nothing in the packet: every row crosses the zod gate again at load time. The dump is snake_case (dbt's world) and the catalogue is camelCase (TypeScript's); the cursor converts once, in one function, which is the single place that mapping lives.

Buckets, pillars, cells and units

Four words carry the whole progress and failure model, so here they are precisely:

• A unit is the smallest set of rows that must land atomically: one person (their Identity plus FirmMember and TeamMember rows), one matter core (the Matter plus its MatterParticipant rows), one matter's notes. One unit = one database transaction.
• A pillar is one kind of content: the commit and failure granularity inside a bucket. A matter has six (core, matter_contacts, files, notes, tasks, comms); the firm level has three (teams, people, contacts).
• A bucket is everything belonging to one matter, plus the special firm bucket for the firm-level foundations.
• A cell is one bucket × pillar slot. Cells are what the run reports on: each holds a status, its row counts, and the first error if it failed.

A run's live state is literally a grid of cells:

Failure is contained by level: a failed person fails that person and the cores that reference them; a failed core fails its bucket and its pillars never run; a failed pillar is one ✗ while its siblings proceed. The run always finishes and always produces the full grid.

The order buckets and pillars are processed in is ROOT_PASS_ORDER, hand-owned domain knowledge, deliberately not derived from foreign keys:

1. teams: team
2. people: identity, draining firm_member and team_member into each person's unit
3. contacts: contact, draining contact_method and contact_address
4. matter buckets: matter with its core (matter_participant), then per matter the independent pillars: matter_contacts (matter_contact), files (matter_file_folder, matter_file), notes (matter_note), tasks (key_date, task), comms (chat_message, call)

Teams come before people because a membership needs its team; people before everything else because almost every later row references an identity; matters last. Every boundary table appears in exactly one pass, asserted at construction, so a table added to the catalogue and forgotten here fails loudly instead of being silently never read.

The service, at a glance

services/import-service: a fully internal service, no gateway route, internal service auth only. Its modules map one-to-one onto this plan's sections.

What the service records: its own database

The import service owns a small database. Two tables here plus one table per receiving service carry a run from trigger to report.

ImportRun, one row per run, is the anchor everything hangs off. Run ids stay self-describing (imprun_<sourceSystem>_<firmId>_<seq>_<startedAt>) so registry rows in other services remain meaningful without this database.

model ImportRun {
  id               String          @id                 // imprun_platform-v2_firm_9y2_3_20260815T0900Z
  kind             ImportRunKind   @default(IMPORT)    // IMPORT | TEARDOWN; teardown is a run too
  targetRunId      String?                             // for TEARDOWN: the run being undone
  sourceSystem     String
  firmId           String                              // the target v3 firm; must already exist
  sourceFirmRef    String                              // what the manifest calls the source dataset: the source system's firm id, or an agreed label where none exists
  actorIdentityId  String                              // the identity this import runs as; fills every required createdBy-style column
  label            String?                             // free-text description of what this run is
  sourceDataPrefix String                              // the packet this run reads (s3://… or a local path in dev)
  manifestSha256   String                              // pins the exact packet this run consumed
  contractVersion  String                              // from the manifest; refused on mismatch. A manifest without one is recorded as the build's own version
  dryRun           Boolean         @default(true)
  scope            Json?                               // optional: pillars / root refs to include
  batchSize        Int             @default(100)       // units per driver hop; a batch is not a transaction (each unit commits its own)
  status           ImportRunStatus @default(PENDING)   // PENDING → RUNNING → COMPLETED | FAILED | CANCELLED; TORN_DOWN after teardown
  cursorPass       String?                             // active entry of ROOT_PASS_ORDER
  cursorRootRef    String?                             // last fully committed root externalRef in that pass
  rowsRead         Json?                               // absolute per-table read counts, persisted each batch
  report           Json?                               // materialised at run end; the audit copy
  startedBy        String
  createdAt        DateTime        @default(now())
  startedAt        DateTime?
  finishedAt       DateTime?
}

Two fields worth pausing on:

• actorIdentityId is a required input, not something the service infers. Imported rows have required createdBy-style columns the boundary deliberately does not carry; every one of them is filled with this explicitly chosen identity, so there is never a question of who an import ran as.
• rowsRead earns its place at reconciliation time: it is what lets the end-of-run checks tell a staged row that failed from one no unit ever looked at.

ImportRunCell is the grid from the section above, persisted: one row per bucket and pillar, pre-created at run start so the grid exists before any work does.

model ImportRunCell {
  id             String     @id
  importRunId    String
  bucketRef      String                 // 'firm' | matter externalRef
  pillar         String                 // 'teams' | 'people' | 'contacts' | 'core' | 'matter_contacts' | 'files' | 'notes' | 'tasks' | 'comms'
  status         CellStatus             // PENDING | RUNNING | SUCCEEDED | FAILED | DISCARDED
  unitsCommitted Int        @default(0)
  rowsWritten    Int        @default(0)
  rowsDiscarded  Int        @default(0) // deliberate discards (scope, gating): counted, never silent
  firstError     Json?                  // first failure: message, entityType, externalRef, zod issues
  updatedAt      DateTime   @updatedAt

  @@unique([importRunId, bucketRef, pillar])
}

Cells live in a different database from the unit writes on purpose: when a unit transaction rolls back in matter-service, the FAILED cell still commits here.

ImportedRecord is not in this database. One table per receiving service, written in the same transaction as the rows it describes:

model ImportedRecord {
  id               String   @id                       // imp_...
  seq              BigInt   @default(autoincrement()) // teardown deletes in reverse seq
  importRunId      String
  firmId           String
  sourceSystem     String
  entityType       String                             // the Prisma model name, e.g. "MatterNote"
  entityId         String   @unique                   // the v3 row this describes
  externalRef      String?                            // staged ref, or derived:<ref>[:<ref>]
  canonicalSources Json?                              // [{ model: "matter_note", sha256: "..." }]
  payloadSha256    String?                            // hash of the exact assembled payload written
  createdAt        DateTime @default(now())

  @@unique([firmId, sourceSystem, entityType, externalRef])
}

• Every row a run inserts, imported or derived, gets one of these. No exceptions. The registry is the complete manifest of what a run wrote, or it is not a manifest.
• entityType holds the Prisma model name, because one importer writes several models (the matter core writes Matter, MatterParticipant and derived MatterTeamAccess) and teardown dispatches per model.
• Staged entities carry their staged externalRef; derived rows (MatterTeamAccess, Content, chat channels) carry a synthetic ref composed from the refs of the rows that determine them, like derived:case_71:team_3, so re-runs idempotency-check them exactly like staged rows.
• The hashes are the drift evidence, captured from run one: canonicalSources carries the source rows' hashes as the packet provides them, and payloadSha256 is the service's own hash over the exact assembled payload it wrote. Together they answer "what fed this row, what exactly landed, and did the source change since".

Controlling a run

One small control API, every operation exposed both as plain HTTP endpoints and as tRPC procedures over the same run module, so it can sit inside the admin dashboard, be driven from a script, or be hit with curl, without caring which.

Two concurrency locks, checked at creation and enforced independently at the driver, because they protect different things:

• Per firm: exactly one non-terminal run at a time. An invariant for correctness; teardown counts as a run for this lock. Enforced by a partial unique index on (firmId) WHERE status IN ('PENDING','RUNNING') plus an Inngest concurrency key on firmId.
• Global: one concurrent run, from configuration (IMPORT_GLOBAL_CONCURRENCY, default 1). This protects the downstream service databases, not correctness, and is environment-scoped, which is the right granularity: the databases it protects are per environment, so dev can run several while prod starts at 1.

A thin CLI script wraps run creation for terminal use, dry run by default.

The run driver: a self-chaining Inngest function

The driver is one Inngest function that does no import work itself: process one slice, persist the cursor, re-emit your own trigger.

export const runImportBatch = (inngest) =>
  inngest.createFunction(
    withPriority({
      id: "import-service/runImportBatch",
      priority: { run: PriorityTier.BATCH },
      concurrency: [
        { limit: env.IMPORT_GLOBAL_CONCURRENCY },        // environment-scoped global cap
        { key: "event.data.firmId", limit: 1 },          // one chain per firm
      ],
      onFailure: async ({ event }) => markRunFailed(event.data.runId), // retries exhausted → FAILED, chain ends
    }),
    { event: "import-service/run.batch.requested" },
    async ({ event, step }) => {
      const result = await step.run("process-next-batch", () =>
        processNextBatch({ runId: event.data.runId }),
      );
      if (result.outcome === "CONTINUE") {
        await step.sendEvent("ping-next", {
          name: "import-service/run.batch.requested",
          data: event.data,
        });
      }
      return result; // visible per hop in the Inngest dashboard
    },
  );

Why this shape works:

• Inngest concurrency counts executing steps, so the chain holds no capacity between hops: a long import never squats on what the rest of the platform depends on.
• Steps and step.sendEvent are memoized: a retried hop neither re-processes a batch (the endpoint is idempotent by cursor anyway) nor double-sends the next ping.
• Queue time is paid per batch, not per unit: thirty to sixty hops for a full firm, each doing a real slice of work.
• The operational controls come free: pause and resume are one status flip and one event; a stalled chain is one event away from moving; the Inngest dashboard shows one run per hop with the batch result attached.
• The import logic stays entirely in the service. Inngest contributes durability, scheduling and visibility, nothing else.

processNextBatch is the whole import, one bite at a time. A batch is a driver-side slice, nothing more: each unit inside it commits its own transaction in whichever service owns it; batchSize only bounds how much work one hop does.

One run, hop by hop

The same machinery on a concrete run: three staged matters, batch size 4, one planted scan failure, one deploy in the middle.

The failure cost one cell; the deploy cost one fast-forward. Healing case_72's files pillar afterwards is a new run over the same packet, scoped at load time.

Reading the packet: passes, cursors, one walker

The read path is the hierarchical unit load walk with the amendments the runtime walkthrough validated. Restated here only as far as the rest of the plan needs it:

• Each pass of ROOT_PASS_ORDER opens the root file and its child files; every file gets exactly one forward-only cursor with current-row semantics. No forward-skip, no pushback, so nothing is ever skipped silently. Empty files are a normal case: the cursor opens, hits EOF, and asserts clean.
• A unit is drained by equality on the parent ref: all of case_71's participants is "advance the participants cursor while matterExternalRef == case_71". Commit points sit at the core and at each pillar, never at the whole matter tree.
• A run is married to its packet. The cursor positions, the rowsRead counts, the source hashes and the reconciliation are all statements about one exact set of files, pinned by the manifest checksum at creation. Resuming a run against a different packet would make every one of those silently wrong, so new cleaned data means a new packet, and a new packet means teardown and a fresh run, never a resume. Scoping works within that rule: a scoped or healing re-run streams the same packet and deliberately discards out-of-scope rows, counted into rowsDiscarded, never silently. (Cheap scoped re-runs via per-file byte offsets are a planned upgrade; they do not change the rule.)
• End-of-pass assertion: every child cursor must be at EOF when its pass drains. Leftover rows mean a mis-sorted dump, the failure mode that would otherwise insert parents childless and finish "clean". They fail the pass loudly.

Assembling and writing units: the Importer modules

One module per unit type. The unit is the commit boundary, so it is also the code boundary: nine modules cover the seventeen boundary tables, each owning its tables end to end, with a registry-level assertion that every table has exactly one owner.

interface Importer<S extends ImportUnitShape> {
  models: PrismaModelName[];                   // the tables this module owns, staged and derived
  assemble(cursors: CursorSet): S | Discarded; // drain rows for one unit; gating decisions live here
  validate(unit: S): ValidatedUnit<S>;         // zod boundary schema + semantic refinements
  write(tx: WriteContext, unit: ValidatedUnit<S>): WrittenUnit;
                                               // derive ids · Prisma rows · ImportedRecords, one transaction
  teardown(tx: WriteContext, record: ImportedRecord): void;
                                               // delete one record's row; dispatched on the record's model name
}

• assemble owns the domain decisions the walk cannot make: what a unit is, what nests where, what gets gated as unsupported (discarded with a count, reported, never silent).
• validate is the contract gate. Every row crosses the boundary zod schema plus semantic refinements, calibrated against real data rather than ideals: at most one lead participant per side with a firm lead required (real firms have matters with no client side), and a contact needs a name, not necessarily a full one.
• write runs inside one Prisma transaction against the owning service's database: insert the domain rows (createMany in chunks) and one ImportedRecord per row, derived rows included. Row ids are derived, not random: minted deterministically from the model, the run id and the external ref. This is the idempotency mechanism, not a nicety. The cursor and the unit writes live in different databases, so any hop can be replayed; a replay mints the same ids, turning its inserts into duplicate-skips with no lookup and no race, while a fresh run after teardown mints fresh ids because the run id changes. The registry's unique key stays as the backstop underneath. Cross-service references resolve through a write-through run cache of the refs this run has written, falling back to one registry read on a cold miss.
• teardown is the same module's inverse, so create and delete knowledge never live in different places. The dispatcher matches each ImportedRecord's model name to the module that owns it.

One conversation-shaped detail: a derived chat channel's id is the matter's id, not a minted one. It is the same rule messaging-service applies to organically created matters, and the only thing that makes an imported conversation findable from its matter.

What the writers deliberately do not do, because an import must not behave like a user:

• No tRPC mutations and no service-layer calls for row creation: direct Prisma writes only.
• No Inngest event fan-out, no notifications, no per-row side effects. Search stays consistent via CDC.
• No use of the platform's organic upload or creation lifecycles.

Two knowing exceptions, both still direct database writes:

• Notes become real Content through content-service's exported markdown-conversion entry point for trusted in-process callers, with an injected Prisma client. Imported notes land Content-backed like organic ones, their content rows carry registry entries under the note's derived ref, and teardown unwinds them version → snapshot → content.
• File blobs need to move from the packet into a platform bucket, gated on a malware-scan verdict, without re-entering the upload lifecycle. The exact mechanics (where scanning runs, how the verdict is represented) are still being worked out and do not change the service's shape: the writer sees a blob store seam and a gate, and a file that fails the gate is a failed insert with that reason on its cell.

Cells, dry runs, and the report

Dry runs are the same run. Same walk, same assembly, same validation, same cells, with write stubbed: ids are minted into the run cache only, nothing touches a service database, no ImportedRecord rows are written. The report is the same shape. Dry runs en masse is a loop over run creation with dryRun: true, and it is the rehearsal every real import runs first.

The report is derived, then frozen. At run end (or cancellation), the service derives the report and materialises it into ImportRun.report: the audit copy that survives registry teardown and Inngest history expiry. It is exhaustive on purpose; nothing about a run's outcome should need a database session to answer.

What it contains:

• The full grid: every cell with its status and counts, exactly as the run left them.
• Per-table accounting: manifest expectedCount, rowsRead, written, failed, and deliberately discarded, for all seventeen tables.
• Every failure: bucket, pillar, the failing refs, and the reason (validation issues, gate refusals, transaction errors).
• The reconciliation block: dangling rows listed by table and external ref, EOF assertion results per pass, and any cross-service refs that never resolved.

{
  "runId": "imprun_platform-v2_firm_9y2_3_20260815T0900Z",
  "dryRun": false,
  "grid": {
    "buckets": 3021,
    "cells": { "SUCCEEDED": 14893, "FAILED": 12, "DISCARDED": 200 }
  },
  "entities": {
    "matter":      { "expected": 3021,  "read": 3021,  "written": 3019,  "failed": 2, "discarded": 0 },
    "matter_note": { "expected": 41200, "read": 41200, "written": 41180, "failed": 8, "discarded": 12 }
  },
  "reconciliation": {
    "danglingRows": [
      { "model": "matter_participant", "refs": ["case_88:pers_x", "case_91:pers_q"] }
    ],
    "cursorEofFailures": [],
    "unresolvedRefs": []
  },
  "failures": [
    { "bucketRef": "case_71", "pillar": "files", "refs": ["doc_991"], "error": "blob gate: scan verdict not clean" }
  ]
}

How a run is validated, end to end. Three layers, each answerable from the records above:

1. During the run: the zod gate rejects malformed rows into FAILED cells; the EOF assertion catches mis-sorted dumps at the end of every pass.
2. At run end: per-table reconciliation. For every table, the manifest's expectedCount must equal rowsRead, and read rows must equal written + failed + deliberately discarded. Any remainder is a dangling row: a staged row that never assembled into any unit, a failure that is invisible unless something counts it. The report names dangling rows by table and external ref, so "which rows" is never a research project. rowsRead separates a row that failed from a row nothing ever looked at; the cells carry the failures and discards; expected minus accounted-for has nowhere to hide.
3. After the run: the registries can be checked against the packet at any time, because the run pinned exactly which packet it read. Comparing the packet's row hashes with the canonicalSources recorded on each ImportedRecord answers, per entity: did the source change since this was imported, and which entities would a corrected packet actually touch. That is the basis for selective re-import later; the evidence is captured from run one.

Teardown

Teardown is a run: it takes the per-firm lock, gets its own ImportRun row (kind: TEARDOWN, pointing at the run it undoes), and produces a report. Per receiving service, delete in reverse seq, which is FK-safe because parents were always written before children, dispatching each ImportedRecord to the module that owns its model. Derived rows are included because they have entries like everything else; this is why the registry rule allows no exceptions. A re-import is always teardown first, then a fresh run.

Open questions

• The relaxed lead-participant rule (at most one lead per side, firm lead required) should become matter-service's own stated invariant, not just the import contract's.
• When a file's blob fails its scan gate, the failure lands on the cell with the reason. Whether the eventual clean verdict should also be recorded on the imported File row or its registry entry (so "this file was scanned before import" is queryable later) is undecided.
• INVITED and client-side participants: whether and how invited parties land on v3 matters.