Import & Export

Bulk-load CSV files into any registered entity type and extract records to downloadable files through per-plugin BullMQ workers.

For field definitions referenced by column mapping, see Properties, Tags & Segments.

Module layout

backend/core-api/src/modules/import-export/
  graphql/schema/{import,export,common}.ts
  graphql/resolvers/{queries,mutations}
  db/definitions/{import,export}.ts, db/models/{Imports,Exports}.ts
  trpc/{import,export,templates,index}.ts
  utils/{getImportExportTypes,getRequiredPermissions,validateConfig}.ts
  workers/utils/errorFileHandler.ts
  routes.ts                          # GET /import-export/download-template
backend/core-api/src/meta/import-export/   # startImportExportWorker setup
backend/erxes-api-shared/src/core-modules/import-export/
  worker.ts                          # BullMQ workers per plugin
  utils/{importBatchProcess,exportBatchProcess,importUtils,
        headerMatcher,tempWorkspace,...}.ts

Each entity type is identified as <plugin>:<module>.<collection> (for example core:contacts.customers, core:contacts.leads, core:contacts.companies, core:product.product, core:organization.users).

How a plugin registers types

Each service calls startImportExportWorker from its meta entry (core: meta/import-export/index.ts):

// backend/core-api/src/meta/import-export/index.ts
export default async (app: Express) =>
  startImportExportWorker({
    pluginName: 'core',
    config: { import: importConfiguration, export: exportConfiguration },
    app,
  });

startImportExportWorker (worker.ts, idempotent per pluginName) starts two BullMQ workers on queues <plugin>-import-processor and <plugin>-export-processor, tunable with IMPORT_EXPORT_IMPORT_CONCURRENCY, IMPORT_EXPORT_EXPORT_CONCURRENCY, and IMPORT_EXPORT_*_LIMITER_MAX/_DURATION_MS env vars. The import/export configs map content types to module handlers; core dispatches to contacts, product, and organization handler sets (createCoreModuleProducerHandler):

• Import: importHandlers + process*Rows.ts per entity (e.g. contacts/meta/import-export/import/customers/processCustomerRows.ts).
• Export: getExportHeaders, getExportData, build*ExportRow per entity.

Each registered type also lists its required permission actions (customersImportManage, companiesImportManage, productsImportManage, teamMembersImportManage, plus the *ExportManage twins), returned by importExportTypes.

GraphQL API

Imports

The Import record tracks status, totalRows, processedRows, successRows, errorRows, importedIds, errorFileUrl, progress, elapsedSeconds, rowsPerSecond, estimatedSecondsRemaining, jobId, userId, subdomain.

Exports

The Export record tracks status, totalRows, processedRows, fileKey, filters, ids, selectedFields, progress, lastCursor (resume point), errorMessage, jobId.

Shared

Running an import

1. Upload the file and get a fileKey

   Imports read fileKey from the platform file storage (the same store /read-file?key= serves). The UI uploads the CSV first; the worker then streams it. processCSVStream (importUtils.ts) parses one row at a time through csv-parse (BOM-stripped, quotes relaxed), so memory stays O(1 row) even for large files.

2. Preview and map columns

   Call importColumnPreview with the fileKey. It returns detected columns with confidence/status and the target fields (required, type, options). Download the canonical header set anytime from GET /import-export/download-template?entityType=core:contacts.customers (templates live in trpc/templates.ts).

3. Start and watch

   Call importStart with columnMapping (index → target key). Poll importProgress for status, progress, rowsPerSecond, estimatedSecondsRemaining. Failures collect into errorRows and a downloadable errorFileUrl (workers/utils/errorFileHandler.ts). processCSVStream aborts the job on stream errors, so a partial count means the file was truncated or unreadable rather than silently skipped.

4. Export round-trip

   Call exportHeaders to pick selectedFields, then exportStart with filters or explicit ids. When status completes, download from the generated fileKey. Exports resume from lastCursor after exportRetry.

Permissions

importsManage and exportsManage (core module importExport) gate the job operations themselves; each entity type additionally requires its own action listed by importExportTypes (for example customersImportManage for core:contacts.customers).

Example

mutation StartCustomerImport {
  importStart(
    entityType: "core:contacts.customers"
    fileKey: "imports/2025/customers.csv"
    fileName: "customers.csv"
    columnMapping: [
      { index: 0, header: "First Name", key: "firstName" }
      { index: 1, header: "Email", key: "primaryEmail" }
    ]
  ) { _id status }
}

query Watch($importId: String!) {
  importProgress(importId: $importId) {
    status
    progress
    processedRows
    totalRows
    errorRows
    errorFileUrl
  }
}