Understanding the Import/Export Subsystem for Subscribers in Listmonk

Listmonk implements a dual-layered import/export subsystem that separates bulk CSV/ZIP ingestion from privacy-aware JSON exports, using goroutine-based sessions for imports and query iterators for exports.

The import/export subsystem for subscribers in the listmonk email suite relies on cleanly decoupled layers to handle high-volume data operations. Bulk imports are processed through a session-based pipeline in internal/subimporter, while exports leverage prepared SQL iterators in internal/core to stream subscriber profiles and campaign analytics. Both subsystems expose REST endpoints and maintain low memory footprints through batching and channel-based concurrency.

Import Subsystem Architecture

The import layer centers on a singleton Importer type that manages global configuration and an independent Session type for each upload operation.

The Importer and Session Types

Located in internal/subimporter/importer.go, the Importer struct holds shared resources including database handles, internationalization settings, and block/allow-list maps. It also maintains a stop channel for graceful shutdowns and atomic counters for progress tracking.

Each upload creates a new Session via Importer.NewSession(opt SessionOpt). The session owns a buffered channel subQueue chan SubReq that acts as the conduit between CSV parsing and database insertion.

type Importer struct {
    // DB, i18n, blocklists, status, stop channel
}

type Session struct {
    subQueue chan SubReq
    // session-specific options and stats
}

type SubReq struct {
    models.Subscriber
    // additional import metadata
}

CSV Processing Workflow

The HTTP handler in cmd/import.go (ImportSubscribers) validates requests and initiates the session in a background goroutine. The workflow follows four stages:

1. File Extraction: Session.ExtractZIP decompresses the first CSV from a ZIP archive, or LoadCSV reads a plain CSV directly.
2. Row Validation: Session.LoadCSV maps CSV headers to columns, sanitizes emails, validates against domain block/allow lists via ValidateFields, and constructs SubReq structs.
3. Queueing: Valid rows are pushed onto subQueue for asynchronous processing.
4. Graceful Shutdown: The Importer.Stop method signals the stop channel, causing the CSV loader to abort and the session to finalize remaining batches.

Batch Insertion and Progress Tracking

The Session.Start() method runs the database worker. It reads from subQueue and commits transactions every 10,000 rows (commitBatchSize), using prepared upsert statements from Importer.Options. This batching strategy keeps memory usage constant regardless of file size.

Real-time status is available through Importer.GetStats and Importer.GetLogs, which the admin UI polls via /api/import to display progress bars and log tails.

func (s *Session) Start() {
    // Reads subQueue, batches inserts every 10k rows
}

Export Subsystem Architecture

Export functionality splits into bulk iterators for administrative downloads and detailed profile exports for GDPR/privacy compliance.

Bulk Export Iterator

internal/core/subscribers.go defines ExportSubscribers, which returns a closure that yields batches of models.SubscriberExport. This iterator pattern allows the UI to export millions of records without loading them into memory simultaneously.

The function builds a SQL query via c.q.QuerySubscribersForExport, optionally filtered by user-supplied conditions validated through validateQueryTables. Each invocation of the returned function fetches the next page, enabling streaming HTTP responses.

func (c *Core) ExportSubscribers(...) (func() ([]models.SubscriberExport, error), error) {
    // Returns iterator closure for paged results
}

Single Subscriber Profile Export

For individual data portability requests, GetSubscriberProfileForExport (also in internal/core/subscribers.go) executes the ExportSubscriberData SQL query. This aggregates the subscriber's profile JSON, list memberships, campaign views, and link clicks into a single models.SubscriberExportProfile struct.

The endpoint ExportSubscriberData in cmd/subscribers.go enforces list permissions via hasSubPerm, then streams the indented JSON response with Content-Disposition: attachment headers. The query respects privacy settings defined in a.cfg.Privacy.Exportable.

func (c *Core) GetSubscriberProfileForExport(id int, uuid string) (models.SubscriberExportProfile, error)

Data Models

Both subsystems rely on structs defined in models/subscribers.go:

• models.Subscriber: Core entity with uuid, email, name, attribs, and status.
• models.SubscriberExport: Lightweight projection for bulk exports (CSV/JSON).
• models.SubscriberExportProfile: Aggregated view containing profile data, subscription lists, campaign views, and click analytics.

Prepared SQL statements in models/queries.go (ExportSubscriberData, QuerySubscribersForExport) support these structures.

API Usage Examples

Importing a CSV file:

curl -X POST \
  -F "file=@subscribers.csv" \
  -F 'params={ "listIDs": [1,2], "mode": "subscribe", "subStatus": "confirmed", "delim": "," }' \
  https://listmonk.example.com/api/import

This triggers the session-based import flow. Monitor progress via GET /api/import.

Exporting a single subscriber's data:

curl -O -J \
  https://listmonk.example.com/api/subscribers/123/export

Returns data.json containing the full profile assembled by GetSubscriberProfileForExport.

Summary

• Import architecture: Located in internal/subimporter/importer.go, using a singleton Importer and per-upload Session types with a subQueue channel for concurrent CSV processing.
• Batch processing: Inserts commit every 10,000 rows to maintain constant memory usage during bulk operations.
• Export architecture: Implemented in internal/core/subscribers.go via iterator functions (ExportSubscribers) for bulk data and GetSubscriberProfileForExport for individual privacy exports.
• API endpoints: cmd/import.go handles uploads, while cmd/subscribers.go handles profile exports with permission checks.
• Data models: models/subscribers.go defines Subscriber, SubscriberExport, and SubscriberExportProfile structures used across both subsystems.

Frequently Asked Questions

How does listmonk handle large CSV files without running out of memory?

The import subsystem processes files using a producer-consumer pattern. The Session.LoadCSV method parses rows and pushes them onto a buffered channel (subQueue), while Session.Start consumes from that channel in batches of 10,000 rows. Each batch is inserted inside a transaction and immediately committed, ensuring memory usage remains constant regardless of file size.

What is the difference between ExportSubscribers and GetSubscriberProfileForExport?

ExportSubscribers returns a closure function that acts as an iterator for paginated bulk exports, suitable for downloading millions of records as CSV or JSON. GetSubscriberProfileForExport retrieves a single comprehensive JSON document containing one subscriber's profile, list memberships, campaign views, and link clicks, designed for GDPR data portability requests.

Can the import process be stopped once started?

Yes. The Importer type maintains a stop channel that can be triggered via Importer.Stop. This signals the active Session to abort CSV parsing, close the subQueue channel, and finalize the current batch, allowing administrators to cancel stuck or erroneous imports through the admin UI or API.

Where are the SQL queries for exports defined?

The prepared SQL statements for both export types are defined in models/queries.go. The QuerySubscribersForExport statement supports the bulk export iterator, while ExportSubscriberData provides the single-subscriber profile aggregation, respecting privacy configuration settings stored in the application config.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}