bulk-inserter-pattern

name

bulk-inserter-pattern

description

Use when implementing a new bulk inserter in the AudioReach Creator Backend (packages/infrastructure/persistence). Covers the full TDD pattern: write integration tests first, then implement StepResult skip-sets, groupRawFailures, BatchInserter, BinaryUtils.toHexString, and Promise.all parallelism. Trigger when the user asks to "add an inserter", "implement insertXxx", "wire up a bulk insert", or wants to persist a new entity type through BulkImportRepository.

Implementing a Bulk Inserter

This skill guides implementation of a new inserter following the established pattern in packages/infrastructure/persistence/src/persistence-typeorm-sqllite/repositories/bulk-import/.

The order matters: tests first, implementation second.

Before you start — read these files

Always read these to understand the exact types before writing any code:

1. Shared framework (already built):

   • …/bulk-import/common/step-result.ts — StepResult interface
   • …/bulk-import/common/group-raw-failures.ts — groupRawFailures function
   • …/bulk-import/batch-inserter.ts — BatchInserter, InsertRow, RawFailure

2. Domain class(es) for the aggregate you're inserting. Check whether the domain type is actually exported from @arc/core — look in packages/core/src/index.ts. If it isn't exported yet, ask the caller to confirm the shape before writing the inserter. Do NOT assume a domain type is in @arc/core without verifying.

3. Schema file(s) for the aggregate and every child table — the exact TypeScript property names matter. Always verify by reading the file, not from memory.

4. id-generation.port.ts if you need idGeneration — check the exact signature of getNextId. Do NOT call reserveBlock inside an inserter; the upload orchestrator does a blanket reservation upfront that covers all IDs.

5. An existing inserter as a reference. key-definition.inserter.ts is a good two-step example. tag-definition.inserter.ts shows the getNextId pattern for value objects. vcpm-module-definition.inserter.ts shows Promise.all for parallel child steps.

6. An existing test as a reference: tests/integration/bulk-import/spf-module-ckv.spec.ts — shows the full test structure: FK seeding, domain entity builders, and the four test case types.

---

Step 1: Write the integration test first

File: tests/integration/bulk-import/<entity-name>.spec.ts

The test runs against an in-memory SQLite DB (no mocks, no migrations needed — the DB is created from schemas via synchronize: true). Write all four test cases before touching the inserter implementation. They will all fail with "Cannot find module './my-entity.inserter.js'" — that's the expected starting state.

Test file structure

/*
 * Copyright (c) Qualcomm Technologies, Inc. and/or its subsidiaries.
 * SPDX-License-Identifier: BSD-3-Clause
 */

import type {DataSource, EntityManager} from 'typeorm';
import {
  setupIntegrationTest,
  teardownIntegrationTest,
  setupEachTest,
  getTestDataSource,
} from '../helpers/test-database-setup.js';
import {MyEntityInserter} from '../../../src/persistence-typeorm-sqllite/repositories/bulk-import/my-entity/my-entity.inserter.js';

// ─── Constants ────────────────────────────────────────────────────────────────

const FILE_ID = 100;
// Add other FK dependency IDs as needed

// ─── Helpers ──────────────────────────────────────────────────────────────────

async function seedFkDependencies(manager: EntityManager): Promise<void> {
  await manager.insert('Project', {
    systemId: 1, name: 'Test', description: '', type: 'Offline', version: 1,
  });
  await manager.insert('ArcDbFile', {
    systemId: FILE_ID, : , : ,
    : , : , : , : , : ,
  });
  
}



(,  {
   : ;
   : ;
   : ;

  ( () => {
     ();
    dataSource = ();
  });

  ( () => {
     ();
  });

  ( () => {
     ();
    manager = dataSource.;
     (manager);
    inserter =  (manager);  
  });

  

  (,  () => {
     result =  inserter.([]);
    (result.).();
     rows =  dataSource.();
    (rows).();
  });

  

  (,  () => {
     entity = ();  

     result =  inserter.([entity]);

     (!result.) {
        (
        ,
      );
    }

     rows =  dataSource.(
      ,
    );
    (rows).();
    (rows[].).(expectedValue);
    
  });

  

  (,  () => {
    
     manager.(, {: ,  : });

     entity = ();  
     result =  inserter.([entity]);

    (result.).();
     (result.)   ();
    (result.).();
    (result.[].).();
    (result.[].).();  
  });

  

  (,  () => {
    
     manager.(, {: ,  : });
     bad = ();   

     good = ();  

     result =  inserter.([bad, good]);

    (result.).();
    (result.).();   

     goodRows =  dataSource.(
      ,
    );
    (goodRows).();   
  });
});

For two-level inserters: add a fifth test case

  // ── Test 5: parent failure skips children ─────────────────────────────────

  it('skips child rows when their parent fails', async () => {
    // Pre-seed a conflict so the root entity fails
    await manager.insert('MyEntityTable', {systemId: 1005, version: 1});

    const entity = buildMyEntityWithChildren(1005, [
      {systemId: 2001, /* child fields */ },
    ]);

    await inserter.insert([entity]);

    // No child rows should exist — they were skipped, not attempted
    const childRows = await dataSource.query(
      `SELECT * FROM my_child_table WHERE parent_system_id = 1005`,
    );
    expect(childRows).toHaveLength(0);
  });

Run the tests — confirm they fail

pnpm --filter @arc/persistence run test:integration -- --testPathPattern="my-entity.spec.ts"

Expected: FAIL — "Cannot find module './my-entity.inserter.js'"

Only after all four (or five) tests are written and confirmed to fail, proceed to Step 2.

---

Step 2: Implement the inserter

File: repositories/bulk-import/<entity-name>/<entity-name>.inserter.ts

Class skeleton

/*
 * Copyright (c) Qualcomm Technologies, Inc. and/or its subsidiaries.
 * SPDX-License-Identifier: BSD-3-Clause
 */

import type {EntityManager} from 'typeorm';
import type {BulkInsertResult, MyEntity} from '@arc/core';
import {okBulkInsert, BinaryUtils} from '@arc/core';
import {BatchInserter, type InsertRow, type RawFailure} from '../batch-inserter.js';
import {groupRawFailures} from '../common/group-raw-failures.js';
import type {StepResult} from '../common/step-result.js';
import {MyEntitySchema, type MyEntityRow} from '../../../entity-schema/…/my-entity.schema.js';

export class MyEntityInserter {
  // Only include idGeneration if this inserter actually calls getNextId.
  // If no generated IDs are needed, omit it entirely.
  constructor(
    private readonly manager: EntityManager,
      : ,  
  ) {}

   (: []): <> {
     (items. === )  ();

     bySystemId =  (items.( [i., i]));

     rootStep =  .(items);

    
     activeItems = items.(
       !rootStep..(i.),
    );

     childStep =  .(activeItems);

     : [] = [
      ...rootStep.,
      ...childStep.,
    ];

     (
      allRawFailures,
      bySystemId,
      
        ,
    );
  }

Run the tests — confirm they pass

pnpm --filter @arc/persistence run test:integration -- --testPathPattern="my-entity.spec.ts"

Expected: PASS for all cases.

---

Constructor rule

Only include idGeneration: IdGenerationPort when the inserter calls getNextId. If no generated IDs are needed (all systemIds are pre-assigned), omit it entirely. Do NOT include it with an underscore prefix — a parameter that is never used has no place in the constructor.

// Inserter that generates IDs — include idGeneration
constructor(
  private readonly manager: EntityManager,
  private readonly idGeneration: IdGenerationPort,
) {}

// Flat inserter or one with only pre-assigned systemIds — omit idGeneration
constructor(
  private readonly manager: EntityManager,
) {}

The repository call site must match:

// With idGeneration:
return new TagDefinitionInserter(this.manager, this.idGeneration).insert([...items]);

// Without:
return new ProcessorDefinitionInserter(this.manager).insert([...items]);

For inserters that use idGeneration, the test needs a mock:

function makeIdGenerator(): IdGenerationPort {
  let counter = 9000;
  return {
    getNextId: async () => ++counter,
    reserveBlock: async () => counter,
    persistLastUsedId: async () => undefined,
  };
}
// Used as: inserter = new MyEntityInserter(manager, makeIdGenerator());

---

StepResult contract

Every private step method must return StepResult:

interface StepResult {
  rawFailures: RawFailure[];
  failedEntityIds: Set<number>; // systemIds of entities that failed AT THIS STEP
}

failedEntityIds contains the entity-level systemIds, not the aggregate root's. The aggregate root's systemId goes into rawFailure.systemId for error grouping. These two are the same for flat inserters; they differ for multi-level hierarchies.

---

Anatomy of a step method

private async insertMyChildren(
  items: MyEntity[],
  skipParentIds?: Set<number>,  // omit if no parent skip needed
): Promise<StepResult> {
  const contextBySystemId = new Map<number, {parent: MyEntity; childId: number}>();

  const rows: InsertRow<MyChildRow>[] = items.flatMap(parent => {
    if (skipParentIds?.has(parent.someGeneratedId)) return [];
    return parent.children.map(child => {
      const row: InsertRow<MyChildRow> = {
        systemId: child.systemId,       // pre-assigned
        childNaturalId: child.childNaturalId,
        parentSystemId: parent.systemId,
      };
      contextBySystemId.set(row.systemId, {parent, childId: child.childNaturalId});
       row;
    });
  });

   (rows. === )  {: [], :  ()};

   {failedEntities} =  .(
    .,
    ,
    rows,
  );

   : [] = failedEntities.( {
     ctx = contextBySystemId.(error.)!;
     row = rows.( r. === error.)!;
     {
      : ctx..,    
      : ,
      : ,
      : error.,
    };
  });

   {
    rawFailures,
    :  (failedEntities.( e.)),
  };
}

---

idGeneration pattern (for value objects without domain systemId)

The upload orchestrator performs a blanket reserveBlock before the entire pipeline runs. Do NOT call reserveBlock inside an inserter. Just call getNextId directly when you need a generated ID.

Since getNextId is async, use a for loop rather than flatMap:

const rows: InsertRow<ChildRow>[] = [];
for (const parent of items) {
  for (const child of parent.children) {
    const systemId = await this.idGeneration.getNextId(parent.fileSystemId);
    contextBySystemId.set(systemId, {parent, childNaturalId: child.naturalId});
    rows.push({systemId, ...});
  }
}

---

Promise.all pattern for independent child steps

const [stepA, stepB, stepC] = await Promise.all([
  this.insertParamsStep(activeItems),
  this.insertAttributesStep(activeItems),
  this.insertLinksStep(activeItems),
]);

// Steps that depend on stepA run after Promise.all resolves:
const stepD = await this.insertGrandchildren(
  activeItems,
  stepA.failedEntityIds,
  parentToGeneratedIdMap,
);

When a step must return both a StepResult and a context map:

interface StepWithContextResult {
  stepResult: StepResult;
  parentToGeneratedId: Map<ParentType, number>;
}

---

failedRowJson format

// Root entity:
`(naturalId=${BinaryUtils.toHexString(entity.naturalId)}) Row: ${JSON.stringify(row)}`

// Child:
`(rootId=${BinaryUtils.toHexString(root.naturalId)}, childId=${BinaryUtils.toHexString(child.naturalId)}) Row: ${JSON.stringify(row)}`

Aggregate label (for groupRawFailures)

item => `SpfModuleDefinition (moduleDefinitionId=${BinaryUtils.toHexString(item.moduleDefinitionId)}, name='${item.name}')`
item => `KeyDefinition (keyId=${BinaryUtils.toHexString(item.keyId)}, name='${item.name}')`
item => `ProcessorDefinition (processorDefinitionId=${BinaryUtils.toHexString(item.processorDefinitionId)})`

---

Flat inserters (no children, no skip-sets)

async insert(items: SimpleEntity[]): Promise<BulkInsertResult> {
  if (items.length === 0) return okBulkInsert();
  const bySystemId = new Map(items.map(i => [i.systemId, i]));
  const step = await this.insertSimpleEntities(items);
  return groupRawFailures(step.rawFailures, bySystemId,
    i => `SimpleEntity (naturalId=${BinaryUtils.toHexString(i.naturalId)})`);
}

Flat inserters need only 4 test cases (no Test 5 — there are no children to skip).

---

Step 3: Wire into the repository

Update typeorm-bulk-import.repository.ts:

// Add import:
import {MyEntityInserter} from './my-entity/my-entity.inserter.js';

// Replace stub:
insertMyEntities(items: readonly MyEntity[]): Promise<BulkInsertResult> {
  return new MyEntityInserter(this.manager).insert([...items]);
  // or: new MyEntityInserter(this.manager, this.idGeneration).insert([...items]);
}

---

Code rules

• ESM imports: all intra-package imports use .js extension
• Copyright header: /* Copyright (c) Qualcomm Technologies... BSD-3-Clause */
• No comments explaining what the code does
• Use BinaryUtils.toHexString() for all IDs — import BinaryUtils from @arc/core
• Only include idGeneration in constructor when the inserter calls getNextId
• Do NOT call reserveBlock — the orchestrator handles the blanket reservation
• Always return {rawFailures: [], failedEntityIds: new Set()} for empty collections

---

Checklist before submitting

• Integration test file written before any implementation
• All 4 test cases present (5 for two-level inserters)
• Tests confirmed failing before writing the inserter
• Tests passing after implementing the inserter
• Step methods return StepResult (not RawFailure[])
• activeItems filter is present after every root step
• idGeneration only in constructor if getNextId is called
• No reserveBlock calls anywhere in the inserter
• rawFailure.systemId is always the aggregate root systemId
• failedEntityIds contains entity-level systemIds at that step's level
• failedRowJson includes natural IDs + full JSON.stringify(row)
• BinaryUtils.toHexString() used for all IDs (imported from @arc/core)
• All imports use .js extension
• Repository stub replaced with live call