Skip to main content

CSV User List Import

Overview​

Customers who already hold a list of LINE User IDs — for example when migrating off a legacy platform — can upload a CSV file to create the corresponding line_user rows.

This is a long-running job that validates each entry, fetches its profile from LINE, records a per-row result, summarises the statistics, and archives the detail records to S3 once finished.

Because the job runs for a long time and can fail partway through, it is protected on three fronts: up to two retries, a Redis processing lock, and a re-process path for imports left in limbo.

Business Flow​

  1. Receive a LineUserImportCSVUserPayload containing importId, lineOaId, and organizationId.
  2. Set the Redis flag LINE_USER:<REDIS_KEY_IMPORT_CSV_USER_PROCESSING>:<importId> to prevent concurrent processing of the same import.
  3. Load the import row from the database and read the CSV file from S3 via GetFileContent.
  4. validateDelimiter confirms the file uses the expected separator, then parseLineUserIDs extracts the ID list.
  5. Process one entry at a time in processOneImportUser:
    • Validate the format against ^U[a-f0-9]{32}$; failures are recorded as invalid with a reason.
    • Check for an existing line_user via checkLineUserExists; a match counts as a duplicate.
    • For genuinely new users, fetch the profile from LINE with getLineUserProfile and write the result into import_detail.
  6. Work through the list in batches of 100, writing progress back to the import row via updateImport.
  7. saveLineUserAccount creates the real line_user rows from the validated import_detail records.
  8. updateImportSummary totals up the counts: total, success, failed, and duplicate.
  9. emitAttributeChangeBatch publishes the newly created users onto the attribute_change queue so the trigger engine can evaluate rules against them.
  10. archiveImportDetail exports all import_detail rows to a file and uploads it to S3 under private/<lineOaHash>/line-user-archive/, keeping the table size manageable over time.
  11. On failure the job retries up to twice (maxRetries) before marking the import as failed. ReProcessImportCSVUser exists to sweep up imports stuck in the pending state.

Key Files & Functions​

  • internal/lineuser/import_service.go — the core of this feature, roughly 1,000 lines
    • ImportService.ProcessImportCSVUser(ctx, payload) — entry point
    • processOneImportUser(), getLineUserProfile(), parseLineUserIDs(), validateDelimiter()
    • checkLineUserExists(), createImportDetails(), saveLineUserAccount()
    • updateImport(), updateImportSummary(), emitAttributeChangeBatch()
    • archiveImportDetail(), getImportDetails(), ReProcessImportCSVUser()
    • Key constants: maxRetries is 2, batchSize is 100, plus folderLineUserArchive
  • internal/lineuser/consumer.go — Consumer.HandleImportCSVUser
  • cmd/worker/integration.go — lineUserStorage, the s3x adapter
  • Queues: consumes line_import_csv_user, publishes to attribute_change (runtime profile main)

Connections to Other Services​

  • Job source — cms-api-go (import domain), when a user uploads a file and starts the import.
  • Database — import (status and statistics), import_detail (per-row results), line_user (the rows actually created), and line_oa.
  • S3 — reads the source CSV and writes the archive file to private/<lineOaHash>/line-user-archive/.
  • Redis — the key LINE_USER:IMPORT_CSV_USER_PROCESSING:<importId>, whose prefix is configurable through REDIS_KEY_IMPORT_CSV_USER_PROCESSING.
  • LINE API — GET /v2/bot/profile/{userId}.
  • RabbitMQ — publishes onward to attribute_change, consumed by the trigger engine.
  • Contrast with the import mapping job — this job creates new users, whereas import mapping enriches users that already exist.