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 Transaction] struct {
	totalTransaction int
	totalAmount      int
	average          int
}

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

	return nil
}

func main() {
	// init
	ctx := context.Background()

	backend := state.NewMemoryStateBackend(1)

	ctx = core.InjectStateBackend(ctx, backend)

	txList := []Transaction{
		{
			User:      "Sang",
			Amount:    1,
			Timestamp: 1000,
		},
		{
			User:      "Sang",
			Amount:    4,
			Timestamp: 1200,
		},
		{
			User:      "Sang",
			Amount:    10,
			Timestamp: 1600,
		},
	}

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

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

	var wg sync.WaitGroup

	runtime.RunTask[Transaction, Transaction](ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)

	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 ch2 {
			if v.Type == core.RecordData {
				fmt.Println("data: ", v.Value)
			} else {
				fmt.Println("watermark: ", v.Timestamp)
			}
		}
	})
	wg.Wait()
}