Function insertImportBatch

packages/db/src/services/import.service.ts:108–136

(
  events: IClickhouseEvent[],
  importId: string
)

 * Insert a batch of events into the imports staging table
 */
export async function insertImportBatch(
  events: IClickhouseEvent[],
  importId: string
): Promise<ImportStageResult> {
  if (events.length === 0) {
    return { importId, totalEvents: 0, insertedEvents: 0 };
  }

  const now = formatClickhouseDate(new Date());
  const rows = events.map((event) => ({
    ...event,
    import_id: importId,
    import_status: 'pending',
    imported_at: event.imported_at || now,
    imported_at_meta: now,
  }));

  await ch.insert({
    table: TABLE_NAMES.events_imports,
    values: rows,
    format: 'JSONEachRow',
  });

  return {
    importId,
    totalEvents: events.length,
    insertedEvents: events.length,
  };
}

/**
 * Insert a batch of profiles into the production profiles table.