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

Operation	Key arguments
importColumnPreview	entityType!, fileKey!, fileName!; returns detected columns (header, key, confidence, status, sampleValues) and target fields
importFields	entityType!; returns available target fields with required, type, options, example
importStart	entityType!, fileKey!, fileName!, columnMapping: [{ index, header, key }]
importProgress	importId!; returns status, progress, processedRows/totalRows, rowsPerSecond, estimatedSecondsRemaining
activeImports	entityType
importHistories	entityType, entityTypes, status, limit, cursor, direction, cursorMode
importCancel / importRetry / importResume	importId!

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

Exports

Operation	Key arguments
exportHeaders	entityType!, filters; returns [{ label, key, isDefault, type }]
exportStart	entityType!, filters, ids, selectedFields
exportHistories	entityTypes, status, limit, cursor, direction, cursorMode
exportCancel / exportRetry	exportId!

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

Shared

Operation	Key arguments
importExportTypes	operation: IMPORT or EXPORT; returns registered types with their required permissions

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
  }
}