# Import & Export

Bulk-load and extract records


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](https://erxes.io/docs/core-properties-tags-segments).

## Module layout

```text
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](https://github.com/erxes/erxes/blob/f91e8c6f4714c910a32c6b6e14fe085f6435c89e/backend/core-api/src/meta/import-export/index.ts)):

```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](https://github.com/erxes/erxes/blob/f91e8c6f4714c910a32c6b6e14fe085f6435c89e/backend/erxes-api-shared/src/core-modules/import-export/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](https://github.com/erxes/erxes/blob/f91e8c6f4714c910a32c6b6e14fe085f6435c89e/backend/erxes-api-shared/src/core-modules/import-export/utils/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`.

> [!WARNING]
> **Job stays queued or every row errors**
>
> A job that never leaves the queue means the plugin's API is not running: workers start inside each service, not a separate process. Look for `[ImportExport] Worker ready for queue <plugin>-import-processor` in the service logs. If every row errors, the column `key`s do not match `importFields`; rerun `importColumnPreview` and fix `columnMapping`.

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

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