# How the OpenCTI Ingestion Manager Handles CSV Feeds: Architecture and Implementation

> Learn how the OpenCTI ingestion manager processes CSV feeds by parsing rows, converting them to STIX bundles, and queuing for import.

- Repository: [OpenCTI Platform/opencti](https://github.com/opencti-platform/opencti)
- Tags: architecture
- Published: 2026-02-19

---

**The OpenCTI ingestion manager processes CSV feeds by polling eligible ingestions every 30 seconds, downloading the source file, parsing rows through configurable mappers, converting them into STIX bundles, and queuing them for import while tracking state via SHA-256 hashes and execution timestamps.**

The OpenCTI ingestion manager is a periodic background service that drives every built-in data feed type across the platform, including RSS, TAXII, JSON, and CSV sources. For CSV feeds specifically, the manager orchestrates a robust six-step pipeline that transforms raw delimited data into structured STIX objects through configurable mapping definitions. This article examines the implementation details in the `OpenCTI-Platform/opencti` repository, focusing on how [`src/manager/ingestionManager.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/manager/ingestionManager.ts) and its supporting modules handle scheduling, throttling, parsing, and state persistence for CSV ingestion workflows.

## Architecture Overview

The ingestion manager treats CSV feeds as scheduled background jobs that execute asynchronously. According to the source code in [`ingestionManager.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/ingestionManager.ts), the pipeline follows six distinct phases:

1. **Selection**: Query CSV ingestions where `ingestion_running` is true, checking scheduling periods and queue capacity.
2. **Download**: Fetch the raw CSV file using the configured URI and authentication via `fetchCsvFromUrl`.
3. **Parsing**: Process the CSV with an inline or external mapper definition, handling headers and separators.
4. **Bundling**: Transform each row into STIX objects using the CSV bundler and create work records.
5. **Queueing**: Push generated bundles to the connector worker queue for asynchronous import.
6. **State Update**: Record the file's SHA-256 hash, execution timestamp, and cursor position to enable resumable, idempotent processing.

## Core Implementation in ingestionManager.ts

### Scheduler and Entry Point

The manager initializes via `initIngestionManager`, which creates a `setIntervalAsync` timer that invokes `ingestionHandler` every `SCHEDULE_TIME` milliseconds (defaulting to 30 seconds). At lines 699–708, this scheduler ensures the service runs continuously as a background worker.

When the interval triggers, `csvExecutor` (starting at lines 77–84) builds a GraphQL filter to fetch all CSV ingestions where the `ingestion_running` flag is true:

```typescript
const csvExecutor = async (context: AuthContext) => {
  const filters = {
    mode: 'and',
    filters: [{ key: 'ingestion_running', values: [true] }],
    filterGroups: [],
  };
  const opts = { filters, connectionFormat: false, noLimits: true };
  const ingestions = await findAllCsvIngestion(context, SYSTEM_USER, opts);
  // ...
};

```

### Eligibility and Throttling

Before processing, the manager validates execution eligibility through two guards. First, `isMustExecuteIteration` (lines 72–78) checks whether the current time satisfies the ingestion's `scheduling_period` (e.g., "auto", "1h", "24h"). Second, `shouldExecuteIngestion` enforces the `CSV_FEED_MIN_INTERVAL_MINUTES` constraint to prevent overwhelming external sources.

The code also verifies queue pressure via `queueDetails`, ensuring `messages_number` is zero before fetching new data:

```typescript
for (const ingestion of ingestions) {
  if (isMustExecuteIteration(ingestion)) {
    const { messages_number } = await queueDetails(connectorIdFromIngestId(ingestion.id));
    if (messages_number === 0 && shouldExecuteIngestion(ingestion, CSV_FEED_MIN_INTERVAL_MINUTES)) {
      await csvDataHandler(context, ingestion);
    }
  }
}

```

### Data Retrieval and Parsing

The `csvDataHandler` function (lines 63–68) resolves the CSV mapper configuration—either an inline JSON definition or a reference to an external `csvMapper` entity—then downloads the file via `fetchCsvFromUrl` from [`src/utils/http-client.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/utils/http-client.ts). 

Subsequently, `processCsvLines` (lines 19–53) handles the raw CSV stream. It strips headers if configured, initializes a work record for tracking, and constructs a `CsvBundlerIngestionOpts` object containing the mapper, applicant user, and connector ID.

### Bundling and Queue Distribution

The core transformation occurs in `generateAndSendBundleProcess`, implemented in [`src/parser/csv-bundler.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/parser/csv-bundler.ts). The bundler iterates over CSV rows and converts them into STIX objects (typically Indicators) based on the mapper's column-to-field mappings:

```typescript
const { bundleCount, objectCount } = await generateAndSendBundleProcess(
  ctx,
  csvLines,
  {
    workId,
    applicantUser,
    csvMapper,
    connectorId,
    entity: undefined,
  }
);

```

After bundling, the manager pushes the resulting STIX bundle to the connector worker queue:

```typescript
await pushBundleToConnectorQueue(context, ingestion, {
  type: 'bundle',
  spec_version: '2.1',
  id: `bundle--${uuidv4()}`,
  objects,
});

```

### State Management and Error Handling

Upon successful completion, `processCsvLines` updates the ingestion record (lines 55–58) with:
- The SHA-256 hash of the downloaded file to detect changes
- The `added_after_start` cursor for incremental processing
- The `last_execution_date` timestamp

Error handling occurs in the `csvExecutor` catch block (lines 4–7). Any failure during download or processing triggers `patchCsvIngestion` to update the `last_execution_date`, ensuring the minimum interval is respected before the next retry, while logging the failure for observability.

## Supporting Modules and File Structure

The ingestion manager delegates specialized tasks to dedicated modules:

- **[`src/modules/ingestion/ingestion-csv-domain.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/modules/ingestion/ingestion-csv-domain.ts)**: Provides CRUD helpers including `findAllCsvIngestion` and `patchCsvIngestion` for persistence operations.
- **[`src/parser/csv-bundler.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/parser/csv-bundler.ts)**: Contains `generateAndSendBundleProcess`, the core logic for transforming CSV rows into STIX 2.1 bundles.
- **[`src/modules/internal/csvMapper/csvMapper-utils.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/modules/internal/csvMapper/csvMapper-utils.ts)**: Parses mapper definitions, handling header detection, separator configuration, and column mapping.
- **[`src/modules/internal/csvMapper/csvMapper-domain.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/modules/internal/csvMapper/csvMapper-domain.ts)**: Manages external CSV mapper entities stored in the database.
- **[`src/utils/http-client.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/utils/http-client.ts)**: Implements `fetchCsvFromUrl` with proper timeout, header management, and authentication handling.

## Practical Configuration Example

To create a CSV ingestion via the GraphQL API:

```graphql
mutation CreateCsvIngestion($input: IngestionCsvAddInput!) {
  ingestionCsvAdd(input: $input) {
    id
    name
    uri
    csv_mapper_type
    csv_mapper
  }
}

```

With variables:

```json
{
  "input": {
    "name": "Blocklist Feed",
    "uri": "https://lists.blocklist.de/lists/all.txt",
    "authentication_type": "none",
    "csv_mapper_type": "inline",
    "csv_mapper": {
      "has_header": false,
      "separator": ",",
      "skip_line_char": "",
      "attributes": [
        { "column_name": "value", "stix_field": "pattern", "default_value": "value: {value}" }
      ]
    },
    "scheduling_period": "auto",
    "ingestion_running": true
  }
}

```

Once created, the ingestion appears in the manager's next 30-second cycle, or can be triggered immediately via the `/ingestion/manager/start` endpoint.

## Summary

- The OpenCTI ingestion manager polls for active CSV feeds every 30 seconds using `setIntervalAsync` in [`ingestionManager.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/ingestionManager.ts).
- **Eligibility checks** include scheduling periods, queue capacity (`messages_number`), and minimum intervals (`CSV_FEED_MIN_INTERVAL_MINUTES`).
- **Data retrieval** uses `fetchCsvFromUrl` with support for authentication headers and configurable mappers.
- **Transformation** occurs via `generateAndSendBundleProcess` in the CSV bundler, converting rows to STIX objects.
- **State tracking** relies on SHA-256 hashes and timestamps to prevent duplicate processing and enable failure recovery.
- **Error handling** updates execution dates to maintain throttling constraints while logging failures.

## Frequently Asked Questions

### How often does the OpenCTI ingestion manager check for new CSV data?

The manager creates a `setIntervalAsync` timer in `initIngestionManager` that executes every 30 seconds (`SCHEDULE_TIME`). However, individual CSV ingestions only run if they satisfy their `scheduling_period` and respect the `CSV_FEED_MIN_INTERVAL_MINUTES` threshold since their last execution.

### What prevents the same CSV file from being processed multiple times?

The manager calculates a SHA-256 hash of each downloaded file and stores it in the ingestion record. Before processing, it compares the current file hash against the stored value. Additionally, the `added_after_start` cursor tracks incremental feeds, ensuring only new rows are processed during subsequent runs.

### Can OpenCTI handle CSV feeds without column headers?

Yes. The mapper configuration includes a `has_header` boolean flag. When set to `false`, the parser treats every line as data rows, mapping columns by index according to the mapper's `column_name` definitions. This is configured via [`csvMapper-utils.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/csvMapper-utils.ts) during the parsing phase.

### Where are CSV ingestion errors logged in OpenCTI?

Errors occurring within `csvExecutor` or `csvDataHandler` are caught in the executor's catch block, which invokes `patchCsvIngestion` to update the `last_execution_date` and prevent immediate retries. Detailed error messages are logged through the standard OpenCTI logging infrastructure, accessible via the platform's logging configuration or the connector status dashboard updated by `updateBuiltInConnectorInfo`.