Skip to main content

Customer Data Import & Mapping to Existing Users

Overview​

Unlike the import that creates new users, this job enriches users who already exist. The customer uploads a CSV containing columns such as phone number, email, purchase total, or membership tier, then maps each column onto a line_user field or a custom attribute. Rows are matched to users by a unique key — typically a phone number or a LINE User ID.

This queue is new to the Go implementation and has no counterpart in the legacy NestJS system. Its distinguishing features are per-cell validation and a downloadable error log that tells the user exactly which rows failed and why.

Business Flow​

  1. Receive a payload carrying importMappingJobId. If it is missing or not greater than zero, the handler returns mq.Permanent so the message goes straight to the DLQ without retrying.
  2. Load the import_mapping_job row, joining line_oa to obtain line_oa_hash for the error-log upload path.
  3. Read the CSV from S3 and parse the header row to locate each column.
  4. buildLookup builds a map from unique-key value to line_user.id. Two key forms are supported: direct field references such as field.mobile_no, field.email, or user_id, and custom attribute references.
  5. Iterate row by row through processRow, working in chunks of 500 and reporting progress as it goes.
    • Resolve the line_user from the lookup; unmatched rows are skipped with a recorded reason.
    • Validate the value against the column type: field.email must match ^[^\s@]+@[^\s@]+\.[^\s@]+$, and field.mobile_no, once spaces and dashes are stripped, must be 9 to 10 digits.
    • Write the value directly to its column, or merge it into the line_user.custom_attribute jsonb column.
  6. uploadErrorLog runs whenever rows were skipped: it builds a CSV of the failures with their reasons and uploads it to S3 at private/<lineOaHash>/import-mapping/error_<jobId>.csv.
  7. finalizeJob writes the final status, the statistics (Stats), the skip reasons ranked by frequency via sortedReasons, and the error-log path back to the job row.
  8. Because the service finalises the job itself, it only returns an error in cases where a retry still has a realistic chance of succeeding.

Key Files & Functions​

  • internal/importmapping/service.go
    • Service.ProcessImportMappingJob(ctx, payload) — entry point
    • run(), processRow(), buildLookup(), parseTarget(), validateFieldValue(), normalizeIdentifier(), stripPhone()
    • uploadErrorLog(), finalizeJob(), findJobByID(), sortedReasons()
    • Key constants: chunkSize is 500, folderImportMappingError is import-mapping, plus emailPattern and phonePattern
  • internal/importmapping/consumer.go — Consumer.HandleImportMappingJob and Register()
  • internal/importmapping/types.go — defines ImportMappingJobPayload, Stats, ReasonCount, and skippedRow
  • Queue: import_mapping_job (runtime profile main)

Connections to Other Services​

  • Job source — cms-api-go, from the CMS import-mapping screen.
  • Database — import_mapping_job (status, statistics, error-log path), line_user (UPDATE of both direct columns and the custom_attribute jsonb), and line_oa (hash used to build the S3 path).
  • S3 — reads the source CSV and writes the error log.
  • No LINE API calls — the job only touches users who are already known to the system.
  • Shares the Storage interface with the CSV user-list import, through the lineUserStorage adapter in cmd/worker/integration.go.