การนำเข้าและ Map ข้อมูลลูกค้าเข้ากับผู้ใช้เดิม

ภาพรวม

ต่างจากการนำเข้าเพื่อสร้างผู้ใช้ใหม่ งานนี้คือการ เติมข้อมูลให้ผู้ใช้ที่มีอยู่แล้วในระบบ ลูกค้าอัปโหลดไฟล์ CSV ที่มีคอลัมน์อย่างเบอร์โทรศัพท์ อีเมล ยอดซื้อ หรือระดับสมาชิก แล้วกำหนดว่าแต่ละคอลัมน์จะ map เข้ากับฟิลด์ใดของ line_user หรือ custom attribute ตัวใด โดยจับคู่แถวเข้ากับผู้ใช้ด้วย unique key เช่นเบอร์โทรศัพท์หรือ LINE User ID

คิวนี้เป็นของใหม่ที่เกิดขึ้นในฝั่ง Go และไม่มีในระบบ NestJS เดิม จุดเด่นคือมีการตรวจความถูกต้องระดับรายเซลล์ และสร้างไฟล์ error log ให้ผู้ใช้ดาวน์โหลดไปตรวจสอบได้ว่าแถวใดไม่ผ่านและเพราะเหตุใด

Business Flow

1. รับ payload ที่มีฟิลด์ importMappingJobId หากไม่มีค่าหรือมีค่าน้อยกว่าหรือเท่ากับศูนย์ ระบบจะคืน mq.Permanent เพื่อส่งข้อความเข้า DLQ ทันทีโดยไม่ retry
2. โหลดแถว import_mapping_job พร้อม join ตาราง line_oa เพื่อเอา line_oa_hash ไว้ใช้ประกอบ path ตอนอัปโหลด error log
3. อ่านไฟล์ CSV จาก S3 แล้วอ่านแถว header เพื่อระบุตำแหน่งของแต่ละคอลัมน์
4. buildLookup สร้าง map จากค่า unique key ไปยัง line_user.id โดยรองรับ key สองรูปแบบ คือแบบอ้างฟิลด์ตรง เช่น field.mobile_no, field.email หรือ user_id และแบบอ้าง custom attribute
5. วนประมวลผลทีละแถวผ่าน processRow โดยแบ่งเป็น chunk ละ 500 แถว พร้อมรายงานความคืบหน้าระหว่างทาง
   • ค้นหา line_user จาก lookup ที่สร้างไว้ หากไม่พบจะข้ามแถวนั้นพร้อมบันทึกเหตุผล
   • ตรวจความถูกต้องของค่าตามชนิดคอลัมน์ โดย field.email ต้องตรงกับ regular expression ^[^\s@]+@[^\s@]+\.[^\s@]+$ ส่วน field.mobile_no เมื่อตัดช่องว่างและเครื่องหมายขีดออกแล้วต้องเป็นตัวเลข 9 ถึง 10 หลัก
   • เขียนค่าลงคอลัมน์โดยตรง หรือ merge เข้าไปในคอลัมน์ line_user.custom_attribute ซึ่งเป็น jsonb
6. uploadErrorLog จะทำงานเมื่อมีแถวที่ถูกข้าม โดยสร้างไฟล์ CSV รวมรายการที่ไม่ผ่านพร้อมสาเหตุ แล้วอัปโหลดขึ้น S3 ที่ private/<lineOaHash>/import-mapping/error_<jobId>.csv
7. finalizeJob เขียนสถานะสุดท้าย สถิติ (Stats) สรุปสาเหตุที่ข้ามเรียงตามจำนวนผ่าน sortedReasons และ path ของ error log กลับลงแถว job
8. เนื่องจาก service จัดการขั้นตอน finalize ด้วยตัวเอง จึงคืนค่า error เฉพาะกรณีที่การ retry ยังมีโอกาสสำเร็จเท่านั้น

ไฟล์และฟังก์ชันหลัก

• internal/importmapping/service.go
  • Service.ProcessImportMappingJob(ctx, payload) — entry point
  • run(), processRow(), buildLookup(), parseTarget(), validateFieldValue(), normalizeIdentifier(), stripPhone()
  • uploadErrorLog(), finalizeJob(), findJobByID(), sortedReasons()
  • ค่าคงที่ที่สำคัญ: chunkSize เท่ากับ 500, folderImportMappingError มีค่าเป็น import-mapping รวมถึง emailPattern และ phonePattern
• internal/importmapping/consumer.go — Consumer.HandleImportMappingJob และ Register()
• internal/importmapping/types.go — นิยามชนิด ImportMappingJobPayload, Stats, ReasonCount และ skippedRow
• คิวที่เกี่ยวข้อง: import_mapping_job (runtime profile main)

จุดเชื่อมต่อกับ Service อื่น

• ต้นทางของงาน — cms-api-go จากหน้า import mapping ของ CMS
• ฐานข้อมูล — import_mapping_job (สถานะ สถิติ และ path ของ error log), line_user (UPDATE ทั้งคอลัมน์ตรงและ custom_attribute แบบ jsonb) และ line_oa (ใช้ hash ประกอบ path บน S3)
• S3 — อ่านไฟล์ CSV ต้นทาง และเขียนไฟล์ error log
• ไม่เรียก LINE API เพราะงานนี้ทำงานกับผู้ใช้ที่มีข้อมูลอยู่แล้วเท่านั้น
• ใช้ Storage interface ตัวเดียวกับงานนำเข้ารายชื่อผู้ใช้จาก CSV ผ่าน adapter lineUserStorage ใน cmd/worker/integration.go