package main

import (
	"context"
	"fmt"
	"strconv"
	"sync"

	"flux/core"
	"flux/core/operator"
	"flux/runtime"
	"flux/state"
	"flux/transport"
)

type Transaction struct {
	User      string
	Amount    int
	Timestamp int64
}

type TransactionSummary[IN Transaction, OUT string] struct {
	totalTransaction int
	average          int
}

func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out operator.Emitter[string]) error {
	totalAmount := 0
	for _, v := range in {
		totalAmount += v.Amount
	}

	msg := fmt.Sprintf("Khách: %s | Số đơn: %d | Tổng tiền: %d$", key, len(in), totalAmount)
	
	return out.Emit(ctx, core.StreamRecord[string]{
		Value:     msg,
		Type:      core.RecordData,
		Timestamp: window.MaxTimestamp(),
	})}

func main() {
	// init
	ctx := context.Background()
	// memory backend 
	// backend := state.NewMemoryStateBackend(1)
	// pebble backend 
	backend, err := state.NewPebbleStateBackend("./data_pebble", 1)

	if err != nil {
		panic(err)
	}

	defer backend.Close()
	ctx = core.InjectStateBackend(ctx, backend)

	txList := []Transaction{
		{
			User:      "Sang",
			Amount:    1,
			Timestamp: 1000,
		},
		{
			User:      "Sang",
			Amount:    4,
			Timestamp: 1500,
		},
		{
			User:      "Sang",
			Amount:    10,
			Timestamp: 2000,
		},
		{
			User:      "Sang",
			Amount:    12,
			Timestamp: 2500,
		},
		{
			User:      "Sang",
			Amount:    10,
			Timestamp: 3000,
		},
	}

	ch1 := make(chan core.StreamRecord[Transaction], len(txList))
	ch2 := make(chan core.StreamRecord[Transaction], len(txList))
	ch3 := make(chan core.StreamRecord[Transaction], len(txList))
	ch4 := make(chan core.StreamRecord[string], len(txList))

	wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 {
		return in.Timestamp
	})

	ko := operator.NewKeyByOperator[Transaction](func(in Transaction) string {
		return in.User
	})

	wop := operator.NewWindowOperator[Transaction, string](core.NewTumblingWindows[Transaction](1000), &TransactionSummary[Transaction, string]{})

	var wg sync.WaitGroup

	runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
	runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch2), transport.NewLocalEmitChannel(ch3), ko)
	runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch3), transport.NewLocalEmitChannel(ch4), wop)

	wg.Go(func() {
		for i, tx := range txList {
			ch1 <- core.NewDataRecord(strconv.Itoa(i), tx, tx.Timestamp)
		}

		close(ch1)
	})

	wg.Go(func() {
		for v := range ch4 {
			if v.Type == core.RecordData {
				fmt.Println("Total Amount after: ", v.Value)
			}
		}
	})

	wg.Wait()
}