การประมวลผล Audience (สร้าง/แก้/ลบ/Mookept)

ภาพรวม

Audience คือกลุ่มเป้าหมายที่ใช้ยิงแคมเปญแบบ multicast ฝั่ง CMS ทำหน้าที่เพียงบันทึกคำสั่งลงฐานข้อมูลแล้วส่งงานเข้าคิว ส่วนงานหนักทั้งหมดอยู่ที่ worker ตัวนี้ ได้แก่ การอ่านรายชื่อ LINE User ID จากไฟล์ที่ผู้ใช้อัปโหลด ตรวจความถูกต้องของรูปแบบ ดึงโปรไฟล์จาก LINE มา upsert ลงตาราง line_user สร้างไฟล์ CSV สรุปผล และอัปเดตสถิติกลับลงตาราง audience

ระบบแยกคิวออกเป็น 4 คิวตามชนิดคำสั่ง คือ สร้าง แก้ไข ลบ และ mookept (เพิ่มหรือถอนสมาชิกทีละชุด)

Business Flow

สร้าง Audience (create_audience)

1. รับ payload ชนิด AudienceQueueType ซึ่งประกอบด้วย audienceId, lineOaId, organizationId และ path ของไฟล์ที่อัปโหลด
2. processValidationPayload โหลดแถว audience พร้อมข้อมูล OA ที่เกี่ยวข้อง ทั้ง lineOaHash, access token และ login channel
3. อ่านไฟล์ที่ผู้ใช้อัปโหลดจาก S3 แล้วแยกออกมาเป็นรายการ LINE User ID
4. กรองรายการด้วย regular expression ^U[a-f0-9]{32}$ ผ่านฟังก์ชัน isValidLineUserID โดยตัดรายการที่รูปแบบไม่ถูกต้องและรายการที่ซ้ำกันออก
5. สำหรับ ID ที่ผ่านการตรวจ ระบบดึงโปรไฟล์จาก LINE ด้วย GetProfile แล้วทำ find-or-create แถว line_user คือสร้างใหม่หากยังไม่มี หรืออัปเดตชื่อที่แสดง รูปโปรไฟล์ และภาษาหากมีอยู่แล้ว
6. สร้างไฟล์ CSV คอลัมน์เดียวชื่อคอลัมน์ Line UserID ตั้งชื่อไฟล์ตามรูปแบบ audience-<id>-<timestamp>.csv แล้วอัปโหลดขึ้น S3 ที่ path private/<lineOaHash>/audience/
7. รวมสถิติ (จำนวนทั้งหมด จำนวนที่ผ่าน และจำนวนที่ไม่ผ่าน) พร้อม path ของไฟล์ผลลัพธ์เข้าไปในคอลัมน์ audience.info ซึ่งเป็นชนิด jsonb
8. เรียก trigger evaluator เพื่อประเมินกฎประเภท audience_membership สำหรับสมาชิกที่เพิ่มเข้ามาหรือถูกถอนออก

แก้ไข Audience (update_audience)

ใช้ลำดับการทำงานเดียวกับการสร้าง แต่เพิ่มขั้นตอนเปรียบเทียบรายชื่อเดิมกับรายชื่อใหม่ เพื่อแยกให้ได้ว่าใครคือสมาชิกที่ถูกเพิ่ม (added) และใครคือสมาชิกที่ถูกถอน (removed) จากนั้นจึงส่งทั้งสองชุดเข้า trigger evaluator

ลบ Audience (delete_audience)

เคลียร์การเป็นสมาชิกทั้งหมด โดยถอน audience id ออกจาก line_user.audience_ids แล้วทำเครื่องหมายลบให้กับแถว audience

Mookept (mookept_audience)

payload มีฟิลด์ actionType เป็นตัวกำหนดทิศทาง หากเป็น add จะเรียก AppendMembersByAudienceID กรณีอื่นจะเรียก RemoveMembersByAudienceID ทั้งสองเมธอดจัดการข้อผิดพลาดภายในตัวเอง คือบันทึก log แล้วจบงาน ดังนั้น handler จึงตอบ ack ทุกข้อความเสมอ ซึ่งตรงกับพฤติกรรมของระบบเดิม

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

• internal/audience/consumer.go — Consumer.Register() ผูก handler เข้ากับทั้ง 4 คิว ได้แก่ OnAudienceCreateProcess, OnAudienceUpdateProcess, OnAudienceDeleteProcess และ OnMookeptAudienceUpdateProcess
• internal/audience/service.go — ตรรกะหลักทั้งหมดของโดเมนนี้
  • CreateAudience(), UpdateAudience(), DeleteAudience()
  • AppendMembersByAudienceID(), RemoveMembersByAudienceID()
  • processCreateFileData(), processUpsertLineUserByLineUserID(), processValidationPayload()
  • mergeStatsIntoInfo(), diff(), isValidLineUserID()
• internal/audience/interfaces.go — interface ขนาดเล็กที่แยกความรับผิดชอบออกจากกัน ได้แก่ CSVReader, Storage, LineOaLookup, LineProfileFetcher, LineUserUpserter และ AudienceTriggerEvaluator
• internal/audience/repository.go และ internal/audience/payloads.go
• cmd/worker/integration.go — adapter ที่ implement interface ข้างต้นจริง ได้แก่ audienceStorage, lineOaForAudience, lineProfileForAudience, lineUserRepo และ audienceTriggerAdapter
• คิวที่เกี่ยวข้อง: create_audience, update_audience, delete_audience และ mookept_audience (runtime profile main)

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

• ต้นทางของงาน — cms-api-go โดเมน audience ส่งงานเข้าคิวเมื่อผู้ใช้สร้าง แก้ไข หรือลบกลุ่มเป้าหมาย
• ฐานข้อมูล — audience (คอลัมน์ info ชนิด jsonb), line_user (upsert และฟิลด์ audience_ids) และ line_oa
• S3 — อ่านไฟล์ที่ผู้ใช้อัปโหลด และเขียนไฟล์ CSV ผลลัพธ์ที่ private/<lineOaHash>/audience/
• LINE API — GET /v2/bot/profile/{userId} สำหรับดึงโปรไฟล์ผู้ใช้
• Trigger engine — เรียก EvaluateAudienceTriggers ต่อทุกครั้งที่สมาชิกเปลี่ยนแปลง
• ปลายทางของไฟล์ CSV — ไฟล์ที่สร้างจากงานนี้คือไฟล์ที่งานส่งแคมเปญแบบ multicast อ่านตอนส่งข้อความจริง