การนำเข้ารายชื่อผู้ใช้จากไฟล์ CSV

ภาพรวม

ลูกค้าที่มีรายชื่อ LINE User ID อยู่แล้ว เช่นกรณีย้ายข้อมูลมาจากระบบเดิม สามารถอัปโหลดไฟล์ CSV เข้ามาเพื่อสร้างแถว line_user ในระบบได้

งานนี้เป็น long-running job ที่ตรวจความถูกต้องทีละรายการ ดึงโปรไฟล์จาก LINE บันทึกผลลัพธ์รายแถว สรุปสถิติ และเก็บถาวรรายละเอียดขึ้น S3 เมื่อทำงานเสร็จ

เนื่องจากเป็นงานที่ใช้เวลานานและมีโอกาสล้มเหลวกลางทาง ระบบจึงมีกลไกรองรับสามชั้น คือ retry ได้สูงสุด 2 ครั้ง มี Redis lock กันงานซ้อน และมีคำสั่ง re-process สำหรับ import ที่ค้างอยู่

Business Flow

1. รับ payload ชนิด LineUserImportCSVUserPayload ซึ่งประกอบด้วย importId, lineOaId และ organizationId
2. ตั้ง Redis flag ที่ key LINE_USER:<REDIS_KEY_IMPORT_CSV_USER_PROCESSING>:<importId> เพื่อกันไม่ให้งานเดียวกันถูกประมวลผลซ้อนกัน
3. โหลดแถว import จากฐานข้อมูล แล้วอ่านไฟล์ CSV จาก S3 ด้วย GetFileContent
4. validateDelimiter ตรวจว่าไฟล์ใช้ตัวคั่นถูกต้อง จากนั้น parseLineUserIDs แยกรายชื่อออกมา
5. ประมวลผลทีละรายการผ่าน processOneImportUser
   • ตรวจรูปแบบด้วย regular expression ^U[a-f0-9]{32}$ หากไม่ผ่านจะบันทึกเป็น invalid พร้อมเหตุผล
   • ตรวจว่ามีแถว line_user อยู่แล้วหรือไม่ผ่าน checkLineUserExists หากมีจะนับเป็น duplicate
   • หากเป็นผู้ใช้ใหม่ จะดึงโปรไฟล์จาก LINE ด้วย getLineUserProfile แล้วบันทึกลงตาราง import_detail
6. ประมวลผลเป็น batch ละ 100 รายการ พร้อมอัปเดตความคืบหน้ากลับลงแถว import ผ่าน updateImport
7. saveLineUserAccount สร้างแถว line_user จริงจากข้อมูลใน import_detail ที่ผ่านการตรวจแล้ว
8. updateImportSummary สรุปตัวเลขรวม ได้แก่ total, success, failed และ duplicate
9. emitAttributeChangeBatch publish รายชื่อผู้ใช้ใหม่เข้าคิว attribute_change เพื่อให้ trigger engine ประเมินกฎกับผู้ใช้ที่เพิ่งเข้าระบบ
10. archiveImportDetail ดึงข้อมูล import_detail ทั้งหมดออกมาเป็นไฟล์แล้วอัปโหลดขึ้น S3 ที่ private/<lineOaHash>/line-user-archive/ เพื่อควบคุมขนาดตารางในระยะยาว
11. หากงานล้มเหลว ระบบจะ retry สูงสุด 2 ครั้งตามค่าคงที่ maxRetries จากนั้นจึงทำเครื่องหมายว่า import ล้มเหลว ส่วนฟังก์ชัน ReProcessImportCSVUser มีไว้กวาด import ที่ค้างอยู่ในสถานะ pending

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

• internal/lineuser/import_service.go — ไฟล์หลักของงานนี้ ขนาดประมาณ 1,000 บรรทัด
  • ImportService.ProcessImportCSVUser(ctx, payload) — entry point
  • processOneImportUser(), getLineUserProfile(), parseLineUserIDs(), validateDelimiter()
  • checkLineUserExists(), createImportDetails(), saveLineUserAccount()
  • updateImport(), updateImportSummary(), emitAttributeChangeBatch()
  • archiveImportDetail(), getImportDetails(), ReProcessImportCSVUser()
  • ค่าคงที่ที่สำคัญ: maxRetries เท่ากับ 2, batchSize เท่ากับ 100 และ folderLineUserArchive
• internal/lineuser/consumer.go — Consumer.HandleImportCSVUser
• cmd/worker/integration.go — lineUserStorage ซึ่งเป็น adapter ของ s3x
• คิวที่เกี่ยวข้อง: consume จาก line_import_csv_user และ publish ไปยัง attribute_change (runtime profile main)

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

• ต้นทางของงาน — cms-api-go โดเมน import เมื่อผู้ใช้อัปโหลดไฟล์แล้วกดเริ่มนำเข้า
• ฐานข้อมูล — import (สถานะและสถิติ), import_detail (ผลลัพธ์รายแถว), line_user (สร้างแถวจริง) และ line_oa
• S3 — อ่านไฟล์ CSV ต้นทาง และเขียนไฟล์เก็บถาวรที่ private/<lineOaHash>/line-user-archive/
• Redis — key LINE_USER:IMPORT_CSV_USER_PROCESSING:<importId> โดยส่วน prefix ปรับได้ผ่านตัวแปร REDIS_KEY_IMPORT_CSV_USER_PROCESSING
• LINE API — GET /v2/bot/profile/{userId}
• RabbitMQ — publish ต่อไปยังคิว attribute_change ซึ่ง trigger engine เป็นผู้บริโภค
• ความต่างจากงาน import mapping — งานนี้เป็นการ สร้างผู้ใช้ใหม่ ส่วนงาน import mapping เป็นการ เติมข้อมูลให้ผู้ใช้ที่มีอยู่แล้ว