Phase 2: Channel Ownership & Streaming Pipelines

package main

import (
    "fmt"
    "sync"
    "time"
)

type PaymentTx struct {
    TxID   string
    Amount int64 // สตางค์
    Score  int   // Fraud Risk Score
}

// 1. Stage 1: Generator/Producer (Owner of txStream)
// คืนค่า <-chan PaymentTx (Receive-Only) ให้ผู้บริโภค ป้องกันไม่ให้ใครแอบส่งข้อมูลเข้าหรือปิด Channel ผิดจังหวะ
func GenerateTransactions(txs []PaymentTx) <-chan PaymentTx {
    out := make(chan PaymentTx, 100) // Buffer 100 รายการสำหรับ Smooth Ingestion

    go func() {
        defer close(out) // กฎเหล็ก: ผู้สร้างคือผู้ปิด Channel เสมอเมื่อผลิตของหมด!
        for _, tx := range txs {
            out <- tx
        }
    }()

    return out
}

// 2. Stage 2: Fan-Out Worker
// รับ Receive-Only เข้ามาคำนวณ และส่งผลลัพธ์ออกทาง Receive-Only ใหม่ที่ตนเป็นเจ้าของ
func RunFraudScorer(in <-chan PaymentTx) <-chan PaymentTx {
    out := make(chan PaymentTx)

    go func() {
        defer close(out)
        for tx := range in {
            // จำลองการคำนวณโมเดลความเสี่ยงทางการเงิน
            if tx.Amount > 10000000 { // เกิน 100,000 บาท
                tx.Score = 95
            } else {
                tx.Score = 10
            }
            out <- tx
        }
    }()

    return out
}

// 3. Stage 3: Fan-In Multiplexer
// รวบรวมหลาย Channels (จาก Worker หลายตัว) ให้ไหลกลับมารวมเป็น Channel เดียว
func FanIn(channels ...<-chan PaymentTx) <-chan PaymentTx {
    out := make(chan PaymentTx)
    var wg sync.WaitGroup
    wg.Add(len(channels))

    // เปิด Goroutine แยกฟังแต่ละ Channel แล้วส่งข้อมูลเข้า out
    for _, ch := range channels {
        go func(c <-chan PaymentTx) {
            defer wg.Done()
            for tx := range c {
                out <- tx
            }
        }(ch)
    }

    // เปิด Goroutine เฝ้ารอให้ทุก worker ปิดช่องทาง จึงทำการปิด out
    go func() {
        wg.Wait()
        close(out) // ปิด Fan-In Channel เมื่อทุกเส้นย่อยจบงาน
    }()

    return out
}