AMLO Screening — Phase 1 Design & Handoff (Data Tier)

หน้านี้เขียนให้อ่านจบแล้วทำงานได้เลย ไม่ต้องอ่านเอกสารอื่นก่อน — ถ้าอยากรู้ว่างานนี้ไปต่อยังไงค่อยอ่าน Architecture และ Implementation Plan ทีหลัง repo เดียวที่ต้องแตะ: Backend_Centralized

1. โจทย์ใน 3 บรรทัด

1. ทุกวันมีไฟล์ AMLO เข้ามาทาง SFTP แล้วระบบอ่านลงตาราง los.amlo_screening_match — ตารางนี้เป็นแค่ข้อมูลดิบ (staging) ไม่ใช่คำตอบ
2. ข้อมูลดิบมีหลายแถวต่อคนเดียวกัน ต้องจัดกลุ่มให้เหลือแถวเดียวต่อคน แล้วเก็บลงตารางใหม่ ที่เป็น source of truth จริง
3. ข้อมูลรอบเก่าต้องไม่หาย — ต้องเก็บเป็น snapshot ไว้ย้อนดูได้ เพราะระบบปลายทางต้องอ้างได้ว่า “ตัดสินใจบนข้อมูลรอบไหน”

ทำไมต้องทำ: งานปลายทาง (ทีมอื่นทำคู่ขนาน) คือการบล็อกไม่ให้ผู้ใช้ของบริษัทที่ติด AMLO ทำธุรกรรม และต้องพิสูจน์กับผู้ตรวจสอบได้ว่าตัดสินใจจากหลักฐานอะไร ตอนนี้ยังทำไม่ได้เพราะข้อมูลตอบไม่ได้ว่า “บริษัทนี้ติด ตามรอบตรวจไหน ด้วยหลักฐานกี่ชิ้น”

2. ข้อมูลจริงเป็นยังไง (วัดจาก DEV_CentralizedDb วันที่ 20/08/2026)

มีข้อมูลอยู่ 1 batch เดียว — AMLO_20260723055102187.csv, stamp 23/07, 409 แถว

การซ้ำเป็นแบบไหน — จัดกลุ่มด้วยทุกคอลัมน์ที่มีความหมาย (custCode + ตัวตน + ชื่อ + AmloType + SearchMethod + Similarity) ได้ 146 กลุ่มจาก 409 แถว แปลว่า 263 แถวเป็นแถวที่ซ้ำกันเป๊ะ

ตัวอย่างจริงของ custCode ที่ซ้ำเยอะที่สุด — คนเดียว CtlId = 12/YAKANA(N)006981 มี 8 แถว ต่างกันแค่ AmloType (EU 4 แถว / hm 4 แถว) และในแต่ละ type ก็ซ้ำกันเป๊ะอีก

3. 🔴 คำถามที่ต้องได้คำตอบก่อนเขียนโค้ด — จัดกลุ่มด้วยคีย์อะไร

นี่คือคำถามของ Compliance ไม่ใช่ของ dev เพราะเลือกต่างกันได้จำนวนแถวต่างกัน และแต่ละแบบตอบคำถาม “ใครติด” คนละความหมาย

ข้อเสนอจากฝั่ง dev (ไม่ใช่มติ): (CustCode, RecordGuid) เพราะเป็นคีย์เดียวที่ ไม่มีค่าว่าง และเป็นรหัสที่ต้นทางกำหนดเอง ไม่ใช่รหัสที่เราตีความ — แต่ต้องให้ Compliance ยืนยันว่า “1 RecordGuid = 1 คนที่ต้องรายงาน” จริงหรือไม่

4. ต้องสร้างอะไรบ้าง

ทุกตารางอยู่ใน schema los เหมือนของเดิม · คอลัมน์ทุกตัวที่เพิ่มใหม่ต้อง NULL ได้ เว้นแต่ระบุว่าบังคับ

4.1 amlo_screening_snapshot_header — 1 แถวต่อ 1 รอบตรวจ

Id                     uuid   PK
GenerationNo           bigint  NULL     เลขลำดับ กำหนดตอน T2 เท่านั้น (NULL = ยังเป็น draft)
RunId                  uuid             1 รอบ ingestion = 1 generation แม้จะอ่านหลายไฟล์
RunStartedAtUtc        timestamptz      ใช้ตรวจว่ารอบนี้ค้างเก่าหรือยัง (ดู T2 ข้อ 3)
SourceFileNames        text             ชื่อไฟล์ทุกใบที่ประกอบเป็นรอบนี้
StampDateFrom          date   NULL
StampDateTo            date   NULL
RowCountTotal          int
DistinctCustCodeCount  int              ตัวเลขที่ด่านตรวจใช้จริง
RejectCount            int
FileSemantics          varchar(20)      Snapshot หรือ Delta ตามที่ config ไว้ตอนรัน
GateResult             varchar(20) NULL Passed หรือ Held หรือ Failed (NULL = ยังไม่ได้ตรวจ)
GateReason             text   NULL
IsCurrent              boolean          default false — มีได้ทีละใบเดียว
CreatedAtUtc           timestamptz

IsCurrent บังคับให้มีใบเดียวด้วย partial unique index — เขียนใน EF config ได้ ไม่ต้องใช้ raw SQL:

builder.HasIndex(x => x.IsCurrent)
    .IsUnique()
    .HasFilter("\"IsCurrent\"");

⚠️ repo นี้ไม่เคยมี migrationBuilder.Sql(...) เลยสักที่ (grep = 0) การใช้ HasFilter จึงรักษาแนวเดิมไว้ · ผลกับ test ที่รันบน SQLite ดู §9.3

4.2 amlo_screening_snapshot_detail — สำเนาข้อมูลดิบของรอบนั้น

copy ทุกคอลัมน์ของ amlo_screening_match มาทั้งแถว รวมชื่อ/นามสกุล/CTL/SubjectId เพิ่ม SnapshotHeaderId (FK, cascade delete) และ SourceBatchId (เก็บเป็นค่า ไม่ต้องทำ FK เพราะ batch อาจถูกลบก่อน)

4.3 amlo_screened_company — ⭐ ตารางที่เป็นคำตอบจริง

ผลของการจัดกลุ่มตาม §3 · นี่คือตารางเดียวที่ระบบปลายทางจะอ่าน

Id                    uuid   PK
SnapshotHeaderId      uuid   FK -> snapshot_header
CustCodeNormalized    varchar(50)      คีย์ที่ใช้จับคู่กับบริษัทในระบบ
MatchCount            int              จัดกลุ่มแล้วเหลือกี่รายการ
EvidenceFingerprint   varchar(64)      hash ของเซ็ต (RecordGuid, AmloType) ที่เรียงแล้ว
FirstSeenGenerationId uuid   NULL      custCode นี้โผล่ครั้งแรกที่ generation ไหน
CreatedAtUtc          timestamptz
UNIQUE (SnapshotHeaderId, CustCodeNormalized)

EvidenceFingerprint สำคัญมากและห้ามคำนวณมั่ว — ปลายทางใช้ค่านี้ตอบว่า “หลักฐานยังเป็นชุดเดิมหรือมีของใหม่โผล่มา” ถ้าค่าเปลี่ยนโดยที่หลักฐานไม่ได้เปลี่ยนจริง ระบบจะไปยกเลิกการปลดล็อกของผู้ใช้ที่ Compliance อนุมัติไปแล้วโดยไม่มีเหตุ

กติกาการคำนวณ:

• ใช้เซ็ตของ (RecordGuid, AmloType) ของ custCode นั้น เรียงลำดับให้แน่นอนก่อน แล้วค่อย hash
• RecordGuid มาจากไฟล์ต้นทาง (ยืนยันจากโค้ด AmloRowParser) ไม่ใช่ค่าที่เรา generate จึงเสถียรข้ามรอบ ✅
• ห้ามใส่ Similarity, SearchMethod, Id ของแถว หรืออะไรที่เปลี่ยนได้เองตอน reprocess
• คำนวณจากเซ็ต (RecordGuid, AmloType) เสมอ ไม่ขึ้นกับคีย์จัดกลุ่มที่ §3 จะเลือก — สองอย่างนี้ตอบคนละคำถาม: คีย์จัดกลุ่มตอบว่า “นับเป็นกี่ราย” ส่วน fingerprint ตอบว่า “หลักฐานชุดเดิมหรือเปลี่ยนไปแล้ว” · เลือกคีย์จัดกลุ่มแบบไหน fingerprint ก็ยังคำนวณแบบเดิม

4.4 ไฟล์ใหม่วางที่ไหน (ตาม pattern ที่ repo ใช้อยู่)

✅ ไม่ต้องแตะ OnModelCreating — ApplyConfigurationsFromAssembly (AuthDbContext.cs:20) หา config ใหม่เจอเอง แต่ ต้องเพิ่มบรรทัด DbSet เอง

4.5 Backend_Centralized ยังไม่มี branch MIGRATION

ตรวจแล้ว 20/08 — มีแค่ Backend_UserService ที่มี origin/MIGRATION repo นี้ยังไม่มี ต้องสร้างก่อน และบอกทีมด้วยว่าสร้างแล้ว ไม่งั้นมีคนสร้างซ้ำแล้ว lineage ของ migration แตก

5. Design ของ logic

5.1 เสียบตรงไหนในโค้ดเดิม

RunOnceAsync (AmloIngestionRunService.cs:43-61) เป็นแค่เปลือกที่จับ leader lock แล้วเรียก ProcessAllFilesAsync (:63-114) — loop ไฟล์อยู่ที่ :89-108 และหลัง loop มีแค่ log กับ return summary (:110-113)

จุดเสียบ = หลังบรรทัด :108 ก่อน :110 — ยังอยู่ในขอบเขต leader lock พอดี

ProcessAllFilesAsync
  :89-108   foreach (fileName in fileNamesToProcess) { ... }     ← ของเดิม ไม่แตะ
  ★★★★      เสียบตรงนี้ — สร้าง scope ใหม่ แล้วเรียกตัวสร้าง generation
  :110-113  log + return AmloIngestionRunSummary(...)            ← ของเดิม

⚠️ ProcessAllFilesAsync ไม่มี DI scope เปิดอยู่ — แต่ละไฟล์เปิด scope ของตัวเองที่ :124 แล้วปิดไปแล้ว ดังนั้นโค้ดที่เสียบต้อง _serviceProvider.CreateScope() เอง (ใช้ pattern เดียวกับ :124)

เงื่อนไขก่อนเรียก — ถ้า processed == 0 ให้ข้ามทั้งหมด ไม่สร้าง generation ใหม่ (กติกาใน §5.4)

5.2 แบ่งเป็น 2 transaction — งานหนักอยู่นอก lock

ทำเป็นก้อนเดียวไม่ได้ เพราะการ copy detail หลายพันแถวใช้เวลานาน ถ้าถือ lock ไว้ทั้งก้อนจะบล็อกทุกอย่าง

T2 ทีละขั้น:

BeginTransaction
  1. SELECT pg_advisory_xact_lock(hashtext('amlo:generation'))
  2. อ่าน generation ปัจจุบัน (IsCurrent = true) และ generation ล่าสุดที่ GateResult = 'Passed'
  3. guard รอบค้าง — ถ้า current.CreatedAtUtc > RunStartedAtUtc ของ draft ใบนี้
     แปลว่ามีรอบอื่นแซงไปแล้ว → GateResult = 'Failed', GateReason = 'stale run'
     ไม่สลับตัวชี้ → commit → จบ
  4. GenerationNo = COALESCE(MAX(GenerationNo), 0) + 1
  5. รันด่านตรวจ (§5.3) → เขียน GateResult + GateReason
  6. UPDATE snapshot_header SET IsCurrent = false WHERE IsCurrent          ← ต้องเคลียร์ก่อน
  7. UPDATE snapshot_header SET IsCurrent = true  WHERE Id = @draftId      ← แล้วค่อยตั้ง
Commit  → advisory lock ปล่อยเองอัตโนมัติ

ทำไมไม่ต้องสร้าง sequence สำหรับ GenerationNo — MAX + 1 อยู่ใน advisory lock อยู่แล้วจึงไม่มี race และ repo นี้ยังไม่เคยใช้ sequence เลยสักที่ (grep CreateSequence = 0) การไม่เพิ่ม construct ใหม่ทำให้ migration ง่ายกว่า

ลำดับข้อ 6 → 7 ห้ามสลับ — partial unique index บังคับให้มี IsCurrent = true ได้ใบเดียว ถ้าตั้งใบใหม่ก่อนเคลียร์ใบเก่าจะชน constraint ทันที

5.3 ด่านตรวจความครบ (completeness gate)

ด่านนี้ไม่เกี่ยวกับการบล็อก — การบล็อกมีผลทันทีเสมอ ด่านนี้ตัดสินแค่ว่า “รอบนี้ครบพอจะปลดใครได้ไหม”

GateReason ต้องเขียนเป็นข้อความที่คนอ่านรู้เรื่องว่าไม่ผ่านเพราะข้อไหน — คนที่มาดูตอนเช้าต้องไม่ต้องเปิดโค้ดอ่าน

Held ยังต้องสลับตัวชี้ เพื่อให้บริษัทที่เพิ่งติดถูกบล็อกทันที แค่ตีตราว่ารอบนี้ห้ามใช้ปลดใคร — ถ้าไม่สลับเลย บริษัทใหม่ที่ติดจะไม่ถูกบล็อก ซึ่งอันตรายกว่าปัญหาที่ด่านนี้พยายามกัน

5.4 รอบที่ไม่มีไฟล์ใหม่เลย = ไม่สร้าง generation

ระบบข้ามไฟล์ที่ประมวลผลจบไปแล้ว (ดูจากชื่อไฟล์) รอบที่ตื่นมาแล้วไม่เจอไฟล์ใหม่จะได้ processed == 0 — ห้ามสร้าง snapshot_header ใบใหม่ ให้ generation ปัจจุบันอยู่ต่อเฉยๆ

ถ้าสร้างใบว่าง DistinctCustCodeCount จะเป็น 0 แล้วด่านตรวจข้อ 4 จะขึ้น Held ทุกวันที่ไม่มีไฟล์เข้า ซึ่งอ่านเหมือนระบบมีปัญหาทั้งที่ปกติดี · ให้ log ว่า “รอบนี้ไม่มีไฟล์ใหม่” แทน

5.5 กลไก lock — ใช้ pattern ที่ repo นี้มีอยู่แล้ว

🔑 ไม่ต้องคิดกลไกใหม่ — repo นี้ใช้ pg_advisory_xact_lock อยู่แล้วสำหรับปัญหาแบบเดียวกันเป๊ะ:

// DocumentNumberRunningRepository.cs:28
await _dbContext.Database
    .ExecuteSqlInterpolatedAsync($"SELECT pg_advisory_xact_lock(hashtext({configCode}))", cancellationToken)
    .ConfigureAwait(false);

พร้อมเหตุผลเขียนกำกับไว้ในโค้ด (:20) ว่าทำไมเลือก advisory lock แทน row-level FOR UPDATE · ตัวอย่างการวาง transaction boundary รอบมันดูที่ DocumentNumberResetBackgroundService.cs:109-138

ทำไมพอ ทั้งที่ leader lock เดิมยังพัง

HybridCacheService.cs:70-79 — acquire เป็น SET NX ค่า "1" ไม่มี owner token · release เป็น KeyDelete แบบไม่มีเงื่อนไข → รอบที่รันเกิน TTL 30 นาทีเสีย lock เงียบๆ รอบสองเข้ามาได้ แล้ว finally ของรอบแรก (AmloIngestionRunService.cs:59) ไปลบ lock ของรอบสอง

Phase 1 ไม่ต้องแก้ Redis lock เพราะ:

• leader lock เดิมทำหน้าที่ “อย่าเริ่มพร้อมกันโดยไม่จำเป็น” แบบ best-effort ต่อไปได้ — พังแล้วแค่เปลืองงาน ไม่ทำข้อมูลเสีย
• ส่วนที่ทำข้อมูลเสียได้จริง คือ T2 ซึ่งถูก serialize ด้วย Postgres แล้ว — ต่อให้ 2 รอบวิ่งพร้อมกัน ตัวที่สองจะรอที่ advisory lock แล้วเจอ guard “รอบค้าง” (§5.2 ข้อ 3)
• แก้ HybridCacheService = แตะ wrapper กลางที่ feature อื่นใช้ร่วม = เสี่ยงเกินความจำเป็นสำหรับเฟสนี้

⚠️ แต่ต้องบันทึกไว้ว่ายังไม่ได้แก้ — ถ้าเฟสหลังมีการเขียนแบบทำลายข้อมูลเพิ่ม ต้องกลับมาแก้ lock ก่อน

5.6 ขอบเขตของ Phase 1 จบที่ตาราง

ไม่ต้อง publish event ใดๆ ทั้งสิ้น — design หลักเป็น reconcile-by-pull ฝั่ง UserService มี timer ดึงเอง · ลูกศร “Centralized → UserService แจ้งว่ารอบจบ” ในเอกสาร architecture ไม่ใช่งานของเฟสนี้ และ ingestion ปัจจุบันก็ไม่ publish อะไรอยู่แล้ว (grep Publish/Outbox/IMediator ใน feature นี้ = 0)

6. งานย่อยเรียงตามลำดับที่ทำได้จริง

7. 🔴 บั๊ก 3 ตัวที่อยู่ในโค้ดที่จะแตะพอดี

ทั้งหมดอยู่บน origin/development วันที่ 20/08 ยังไม่มีใครแก้

7.1 ไฟล์ที่มีแต่หัวตารางได้ batch สีเขียว

IsValidHeader ตรวจแค่จำนวนคอลัมน์ (headerLine.Split('|').Length == 16, AmloRowParser.cs:144-145) ไม่ตรวจชื่อคอลัมน์

ไฟล์ที่มีแต่บรรทัดหัว → 0 match + 0 reject → Complete(0,0,0) → เพราะ rejectCount > 0 เป็นเท็จ status จึงเป็น Completed ไม่ใช่ CompletedWithErrors (AmloIngestionBatch.cs:73-75)

ผล: batch ที่ไม่มีใครอยู่ในนั้นเลย หน้าตาเหมือน batch ของวันปกติทุกประการ — นี่คือเหตุผลที่ด่านตรวจข้อ 4 ต้องนับ custCode ไม่ใช่ดูสถานะ batch

7.2 CustCode ว่างนับเป็นสำเร็จ

ดูกล่องแดงใน §5.3

7.3 leader lock ไม่มี fencing token

ดู §5.5 — เฟสนี้ไม่ต้องแก้ แต่ต้องรู้ว่ามีอยู่ และห้ามพึ่งมันเป็นตัวกันข้อมูลเสีย

8. กติกาที่ห้ามพลาด

• ห้ามกระทบการใช้งานปัจจุบัน — endpoint เดิมต้องตอบเหมือนเดิมทุกประการ · staging ไม่ต้องแก้ · ห้ามแก้ AmloIngestionRunSummary (§5.1) · คอลัมน์ใหม่ทุกตัวต้อง NULL ได้ และ NULL ต้องแปลว่า “ยังไม่ได้ตรวจ” ไม่ใช่ “ผ่าน”
• งาน schema ต้องทำบน branch MIGRATION — port เฉพาะ source ของ data tier มา แล้ว dotnet ef migrations add บน MIGRATION (ห้าม cherry-pick migration + snapshot จาก feature) → verify has-pending-model-changes = none → merge MIGRATION กลับเข้า feature ด้วย · ห้าม merge development → MIGRATION
• แก้ appsettings.{ENV}.json ในโปรเจกต์อย่างเดียวไม่มีผลบน cluster — CI เอาไฟล์จาก Backend_Iac/config/centralized-service/{env}/ ทับตอน build · key ใหม่ต้องเพิ่มที่ IaC ครบทุก env ขาด env ไหน env นั้นพังตอน deploy
• checkout ในเครื่องตามหลัง origin — ก่อนสรุปว่าโค้ดเป็นยังไง ให้ git fetch แล้วดู git show origin/development:<path> เสมอ
• verify ด้วย Release — CI รัน Release และเปิด analyzer เป็น error ผ่าน Debug ไม่ใช่หลักฐาน
• IUnitOfWork.BeginTransactionAsync throw ถ้ามี transaction เปิดค้างอยู่แล้ว (UnitOfWork.cs:43-46) — T1 กับ T2 ต้องปิดให้สนิทก่อนเปิดอันถัดไป

9. Test plan

repo มี tests/Centralized05.Tests (xUnit + Moq + EF SQLite in-memory) และมี test ของ AMLO อยู่แล้ว 13 ไฟล์

9.1 ของเดิมที่จะพังถ้าไม่แก้

9.2 ของใหม่ที่ต้องเขียน

9.3 ข้อจำกัดของ test ที่ต้องรู้ล่วงหน้า

10. เงื่อนไขรับงาน

• GET /v1/amlo-screening/{custCode} ตอบเหมือนเดิมเป๊ะ ก่อนและหลัง (มี test เทียบ)
• POST /v1/admin/amlo-ingestion/run คืน shape เดิม (Total/Processed/Skipped/Failed) ไม่เปลี่ยน
• รอบ ingestion เดิมยังจบด้วยผลเท่าเดิม — staging ไม่เปลี่ยนพฤติกรรม
• fingerprint เสถียร — ingest เนื้อหาไฟล์เดียวกันอีกครั้ง (ผ่านเส้น retry ของ batch ที่ Failed หรือใช้ไฟล์ชื่อใหม่ที่เนื้อหาเหมือนกัน) แล้ว EvidenceFingerprint ของ custCode เดียวกันต้องเท่ากันทั้งสองรอบ · ⚠️ รันไฟล์ ชื่อเดิม ซ้ำเฉยๆ ทดสอบข้อนี้ไม่ได้ เพราะระบบจะข้ามไฟล์ที่จบแล้ว
• รอบที่ไม่มีไฟล์ใหม่ → ไม่สร้าง generation ใหม่ และ generation ปัจจุบันไม่เปลี่ยน
• ไฟล์ที่มีแต่หัวตาราง → GateResult = Held ไม่ใช่ Passed
• ไฟล์ที่มี custCode น้อยลงกว่ารอบก่อน → Held + มี GateReason อ่านรู้เรื่อง
• มีมากกว่า 1 ไฟล์ในรอบเดียว → Held
• IsCurrent มีได้ทีละใบเดียวจริง (ทดสอบด้วยการยิงพร้อมกัน)
• draft ที่ T2 ไม่ทำงาน มองไม่เห็นจากฝั่งผู้อ่าน และไม่ทำให้รอบถัดไปพัง
• generation เก่ายังอ่านได้ครบหลังสลับตัวชี้แล้ว
• has-pending-model-changes = none หลัง merge MIGRATION เข้า feature และ migration รันผ่านกับ Postgres จริงใน dev
• test เดิมทั้ง 13 ไฟล์ยังผ่าน (โดยเฉพาะ harness ที่ซ้ำกัน 2 namespace)
• คอลัมน์ชื่อใน snapshot_detail ไม่โผล่ใน API หรือ log ใดๆ
• dotnet build -c Release ผ่าน (analyzer เป็น error)

11. ของที่ต้องคุยกับทีมก่อน ไม่ใช่ตัดสินเอง