package monthlyusagehistory

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/gogf/gf/v2/container/gvar"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/testutil"
)

// datasetScanner applies the SQL eligibility predicate before LIMIT; metadata
// eligibility is applied by the producer. Optional hooks simulate SQL latency,
// live additions and failures without accessing an application database.
type datasetScanner struct {
	cards    []Card
	channels []channelMetadata
	upper    int64
	hook     func(context.Context, string, int64) error
	queries  atomic.Int32
	lists    atomic.Int32
	inflight atomic.Int32
	peak     atomic.Int32
}

func (s *datasetScanner) SnapshotMaxID(context.Context) (int64, error) { return s.upper, nil }
func (s *datasetScanner) ListChannels(context.Context) ([]channelMetadata, error) {
	s.lists.Add(1)
	return s.channels, nil
}
func (s *datasetScanner) Scan(ctx context.Context, last, upper int64, size int) ([]Card, error) {
	var rows []Card
	for _, card := range s.cards {
		if card.ID > last && card.ID <= upper {
			for _, channel := range s.channels {
				if channel.ChanKey == card.ChanKey {
					card.TempKey = channel.TempKey
					card.SyncEnabled = channel.SyncEnabled
					card.Operator = channel.Operator
				}
			}
			rows = append(rows, card)
			if len(rows) == size {
				break
			}
		}
	}
	return rows, nil
}
func (s *datasetScanner) ScanChannel(ctx context.Context, key string, last int64, size int) ([]Card, error) {
	s.queries.Add(1)
	running := s.inflight.Add(1)
	defer s.inflight.Add(-1)
	for {
		old := s.peak.Load()
		if running <= old || s.peak.CompareAndSwap(old, running) {
			break
		}
	}
	if s.hook != nil {
		if err := s.hook(ctx, key, last); err != nil {
			return nil, err
		}
	}
	var rows []Card
	for _, card := range s.cards {
		if card.ChanKey == key && card.ID > last {
			rows = append(rows, card)
			if len(rows) == size {
				break
			}
		}
	}
	return rows, nil
}
func scannerFor(cards []Card, upper int64) *datasetScanner {
	keys := map[string]bool{}
	s := &datasetScanner{cards: cards, upper: upper}
	for _, card := range cards {
		if card.ChanKey != "" && !keys[card.ChanKey] {
			keys[card.ChanKey] = true
			s.channels = append(s.channels, channelMetadata{ChanKey: card.ChanKey, TempKey: "sim_yd_v5", Operator: 1})
		}
	}
	return s
}
func testStore(t *testing.T) *zset.Store {
	t.Helper()
	store, err := zset.New(testutil.Redis(t), "flow:")
	if err != nil {
		t.Fatal(err)
	}
	return store
}
func testProducer(t *testing.T, scanner CardScanner, store *zset.Store, config Config) *Producer {
	t.Helper()
	config.RoundRetryEnabled = true
	if config.ScanMode == "" {
		config.ScanMode = "channel"
	}
	p, err := New(scanner, store, config)
	if err != nil {
		t.Fatal(err)
	}
	return p
}
func contextForTest(t *testing.T) context.Context {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	t.Cleanup(cancel)
	return ctx
}
func taskIDs(t *testing.T, store *zset.Store, run string) []string {
	t.Helper()
	key, _ := store.TaskStateKey(run)
	result, err := store.RedisClient().Do(context.Background(), "HKEYS", key)
	if err != nil {
		t.Fatal(err)
	}
	ids := result.Strings()
	sort.Strings(ids)
	return ids
}
func requireMeta(t *testing.T, store *zset.Store, run string, done bool, active int64) {
	t.Helper()
	meta, err := store.GetRunMeta(context.Background(), run)
	if err != nil {
		t.Fatal(err)
	}
	if meta.ProducerDone != done || meta.ActiveCount != active {
		t.Fatalf("meta=%+v, want done=%t active=%d", meta, done, active)
	}
}
func waitUntil(t *testing.T, ctx context.Context, condition func() bool) {
	t.Helper()
	for !condition() {
		select {
		case <-ctx.Done():
			t.Fatal(ctx.Err())
		case <-time.After(time.Millisecond):
		}
	}
}

func TestChannelModeSettings(t *testing.T) {
	store, _ := zset.New(&producerRedis{}, "flow:")
	p, err := New(&fakeScanner{}, store, Config{})
	if err != nil || p.config.ScanMode != "serial" || p.config.ScanConcurrency != 4 {
		t.Fatalf("defaults: p=%+v err=%v", p, err)
	}
	for _, config := range []Config{{ScanMode: "parallel"}, {ScanConcurrency: -1}, {ScanMode: "channel"}} {
		if _, err = New(&fakeScanner{}, store, config); err == nil {
			t.Fatalf("accepted invalid config/capability %+v", config)
		}
	}
}

func TestChannelScanFiniteMemoryBoundary(t *testing.T) {
	for _, test := range []struct {
		name    string
		upper   int64
		cards   []Card
		ids     []int64
		queries int32
	}{
		{"zero upper", 0, []Card{{ID: 1, ChanKey: "a"}}, nil, 0},
		{"exact page", 4, []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "a"}}, []int64{1, 2, 3, 4}, 2},
		{"tail then empty", 9, []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}}, []int64{1, 2, 3}, 3},
		{"first outside", 3, []Card{{ID: 4, ChanKey: "a"}, {ID: 5, ChanKey: "a"}}, nil, 1},
		{"mixed final page", 3, []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "a"}, {ID: 5, ChanKey: "a"}}, []int64{1, 2, 3}, 2},
		{"empty channel", 9, nil, nil, 1},
	} {
		t.Run(test.name, func(t *testing.T) {
			store := testStore(t)
			scanner := scannerFor(test.cards, test.upper)
			if len(scanner.channels) == 0 {
				scanner.channels = []channelMetadata{{ChanKey: "a", TempKey: "sim_yd_v5"}}
			}
			p := testProducer(t, scanner, store, Config{PageSize: 2})
			if err := p.Produce(contextForTest(t), scopedMessage("boundary")); err != nil {
				t.Fatal(err)
			}
			var want []string
			for _, id := range test.ids {
				want = append(want, zset.StableTaskID("boundary", operation, id))
			}
			sort.Strings(want)
			got := taskIDs(t, store, "boundary")
			if len(got) != len(want) || (len(want) > 0 && !reflect.DeepEqual(got, want)) {
				t.Fatalf("got=%v want=%v", got, want)
			}
			if scanner.queries.Load() != test.queries {
				t.Fatalf("queries=%d want=%d", scanner.queries.Load(), test.queries)
			}
			requireMeta(t, store, "boundary", true, int64(len(want)))
		})
	}
}

func TestChannelScanLiveHigherIDsDoNotExtendRun(t *testing.T) {
	store := testStore(t)
	scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}}, 3)
	scanner.hook = func(_ context.Context, _ string, last int64) error {
		if last > 0 {
			scanner.cards = append(scanner.cards, Card{ID: 100 + last, ChanKey: "a"})
		}
		return nil
	}
	p := testProducer(t, scanner, store, Config{PageSize: 2})
	if err := p.Produce(contextForTest(t), scopedMessage("live")); err != nil {
		t.Fatal(err)
	}
	if len(taskIDs(t, store, "live")) != 3 || scanner.queries.Load() != 2 {
		t.Fatal("higher IDs extended finite range")
	}
}

func TestChannelScanSerialTaskIDSetParity(t *testing.T) {
	for _, scope := range [][]string{nil, {"a", "b"}, {"disabled", "unknown"}} {
		t.Run(fmt.Sprint(scope), func(t *testing.T) {
			store := testStore(t)
			cards := []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "b"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "disabled"}, {ID: 5, ChanKey: "no-template"}, {ID: 6, ChanKey: "unknown"}, {ID: 7, ChanKey: "b"}, {ID: 8, ChanKey: "a"}}
			var sets [][]string
			for _, mode := range []string{"serial", "channel"} {
				scanner := scannerFor(cards, 7)
				scanner.channels = []channelMetadata{{ChanKey: "a", TempKey: "sim_yd_v5", Operator: 1}, {ChanKey: "b", TempKey: "sim_dx_v1", Operator: 2}, {ChanKey: "disabled", TempKey: "sim_yd_v5", SyncEnabled: true}, {ChanKey: "no-template"}}
				p := testProducer(t, scanner, store, Config{ScanMode: mode, PageSize: 2})
				run := "parity-" + mode
				if err := p.Produce(contextForTest(t), scopedMessage(run, scope...)); err != nil {
					t.Fatal(err)
				}
				ids := taskIDs(t, store, run)
				for i, id := range ids {
					ids[i] = strings.TrimPrefix(id, run+":")
				}
				sets = append(sets, ids)
				if mode == "channel" && scanner.lists.Load() != 1 {
					t.Fatal("metadata not loaded once")
				}
				_, _ = store.RedisClient().Do(context.Background(), "SREM", store.ActiveRunsKey(), run)
			}
			if !reflect.DeepEqual(sets[0], sets[1]) {
				t.Fatalf("serial=%v channel=%v", sets[0], sets[1])
			}
		})
	}
}

func TestChannelSQLSlotsReleasedWhileChannelsWait(t *testing.T) {
	store := testStore(t)
	ctx := contextForTest(t)
	var cards []Card
	for i := 0; i < 9; i++ {
		cards = append(cards, Card{ID: int64(i + 1), ChanKey: fmt.Sprint(i)}, Card{ID: int64(i + 10), ChanKey: fmt.Sprint(i)})
	}
	sort.Slice(cards, func(i, j int) bool { return cards[i].ID < cards[j].ID })
	scanner := scannerFor(cards, 100)
	scanner.hook = func(ctx context.Context, _ string, _ int64) error { return sleepContext(ctx, 5*time.Millisecond) }
	p := testProducer(t, scanner, store, Config{PageSize: 2, ScanConcurrency: 4, ChannelHighWaterMark: 1, GlobalHighWaterMark: 100, PollInterval: time.Millisecond})
	result := make(chan error, 1)
	go func() { result <- p.Produce(ctx, scopedMessage("isolation")) }()
	waitUntil(t, ctx, func() bool { return len(taskIDs(t, store, "isolation")) == 9 })
	if peak := scanner.peak.Load(); peak > 4 || peak < 2 {
		t.Fatalf("SQL peak=%d", peak)
	}
	// All nine channels got a first task despite the first four waiting at
	// channel capacity. SQL slots are free while every channel is paused.
	if scanner.inflight.Load() != 0 {
		t.Fatal("SQL tokens held during channel backpressure")
	}
	for _, channel := range scanner.channels {
		if _, err := store.BlockRunScoped(ctx, "isolation", channel.ChanKey); err != nil {
			t.Fatal(err)
		}
	}
	if err := <-result; err != nil {
		t.Fatal(err)
	}
	meta, err := store.GetRunMeta(ctx, "isolation")
	if err != nil || !meta.ProducerDone {
		t.Fatalf("meta=%+v err=%v", meta, err)
	}
}

func TestChannelRemainingCapacityAndDuplicateAdmission(t *testing.T) {
	store := testStore(t)
	ctx := contextForTest(t)
	scanner := scannerFor(nil, 0)
	p := testProducer(t, scanner, store, Config{GlobalHighWaterMark: 3, ChannelHighWaterMark: 2, PollInterval: time.Millisecond})
	if err := store.CreateRun(ctx, zset.RunMeta{RunID: "capacity", Status: "producing"}); err != nil {
		t.Fatal(err)
	}
	// Existing success and waiting states still deduplicate and consume no new
	// capacity. Active includes leases: inflight must not be added again.
	stateKey, _ := store.TaskStateKey("capacity")
	metaKey, _ := store.RunMetaKey("capacity")
	activeKey, _ := store.RoundActiveKey("capacity", 1, "a")
	_, _ = store.RedisClient().Do(ctx, "HSET", stateKey, "one", `{"status":"success"}`, "two", `{"status":"waiting"}`)
	_, _ = store.RedisClient().Do(ctx, "HSET", metaKey, "activeCount", 2, "inflightCount", 2)
	stats := &channelScanStats{chanKey: "a", lastSummary: time.Now()}
	tasks := []zset.Task{{TaskID: "one", RunID: "capacity", ChanKey: "a"}, {TaskID: "two", RunID: "capacity", ChanKey: "a"}, {TaskID: "three", RunID: "capacity", ChanKey: "a"}, {TaskID: "four", RunID: "capacity", ChanKey: "a"}}
	gate := make(chan struct{}, 1)
	result := make(chan error, 1)
	go func() { _, err := p.admitChannelPage(ctx, "capacity", tasks, time.Now(), gate, stats); result <- err }()
	waitUntil(t, ctx, func() bool { meta, e := store.GetRunMeta(ctx, "capacity"); return e == nil && meta.ActiveCount == 3 })
	if size, err := store.SizeKey(ctx, activeKey); err != nil || size != 1 {
		t.Fatalf("initial slice size=%d err=%v", size, err)
	}
	_, _ = store.RedisClient().Do(ctx, "HSET", metaKey, "activeCount", 0, "inflightCount", 0)
	if err := <-result; err != nil {
		t.Fatal(err)
	}
	if stats.duplicates != 2 || stats.added != 2 || stats.enqueueBatches != 2 || stats.globalWait == 0 {
		t.Fatalf("stats=%+v", stats)
	}
}

// faultRedis forwards to a real isolated Redis, optionally returning a lost
// response after Lua succeeded. Never mocks the Lua semantics under test.
type faultRedis struct {
	zset.RedisClient
	failEnqueue  atomic.Int32
	failRead     atomic.Bool
	failRenew    atomic.Bool
	lostResponse bool
	reject       bool
}

func (r *faultRedis) Do(ctx context.Context, command string, args ...any) (*gvar.Var, error) {
	if r.failRead.Load() && command == "HMGET" {
		return nil, errors.New("Redis read failure")
	}
	return r.RedisClient.Do(ctx, command, args...)
}
func (r *faultRedis) Eval(ctx context.Context, script string, n int64, keys []string, args []any) (*gvar.Var, error) {
	if r.failRenew.Load() && strings.Contains(script, "PEXPIRE") && strings.Contains(script, "owner ~=") {
		return gvar.New(0), nil
	}
	if n == 4 && strings.Contains(script, "HEXISTS") {
		if r.reject {
			return gvar.New(0), nil
		}
		if r.failEnqueue.CompareAndSwap(1, 0) {
			if r.lostResponse {
				if _, err := r.RedisClient.Eval(ctx, script, n, keys, args); err != nil {
					return nil, err
				}
			}
			return nil, errors.New("enqueue response unavailable")
		}
	}
	return r.RedisClient.Eval(ctx, script, n, keys, args)
}

func TestChannelRescanPartialAndUncertainWrites(t *testing.T) {
	for _, uncertain := range []bool{false, true} {
		t.Run(fmt.Sprint(uncertain), func(t *testing.T) {
			client := &faultRedis{RedisClient: testutil.Redis(t), lostResponse: uncertain}
			store, _ := zset.New(client, "flow:")
			scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "a"}}, 4)
			// Fail the second page: first page must survive; uncertain success is
			// confirmed only on a same-Run rescan, with stable IDs and no double count.
			scanner.hook = func(_ context.Context, _ string, last int64) error {
				if last == 2 {
					client.failEnqueue.Store(1)
				}
				return nil
			}
			p := testProducer(t, scanner, store, Config{PageSize: 2})
			if err := p.Produce(contextForTest(t), scopedMessage("rescan")); err == nil {
				t.Fatal("expected enqueue failure")
			}
			active := int64(2)
			if uncertain {
				active = 4
			}
			requireMeta(t, store, "rescan", false, active)
			scanner.hook = nil
			if err := p.Produce(contextForTest(t), scopedMessage("rescan")); err != nil {
				t.Fatal(err)
			}
			requireMeta(t, store, "rescan", true, 4)
			if len(taskIDs(t, store, "rescan")) != 4 {
				t.Fatal("wrong stable task set")
			}
			before := scanner.queries.Load()
			if err := p.Produce(contextForTest(t), scopedMessage("rescan")); err != nil {
				t.Fatal(err)
			}
			if before != scanner.queries.Load() {
				t.Fatal("producerDone redelivery rescanned cards")
			}
		})
	}
}

func TestChannelFailureCancelsAllWorkersWithoutProducerDone(t *testing.T) {
	for _, kind := range []string{"sql", "redis", "unexplained_rejection", "lease"} {
		t.Run(kind, func(t *testing.T) {
			client := &faultRedis{RedisClient: testutil.Redis(t)}
			store, _ := zset.New(client, "flow:")
			cards := []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "b"}, {ID: 3, ChanKey: "c"}}
			scanner := scannerFor(cards, 10)
			allStarted := make(chan struct{})
			var started atomic.Int32
			var once sync.Once
			scanner.hook = func(ctx context.Context, key string, _ int64) error {
				if started.Add(1) >= 3 {
					once.Do(func() { close(allStarted) })
				}
				select {
				case <-allStarted:
				case <-ctx.Done():
					return ctx.Err()
				}
				if kind == "lease" {
					client.failRenew.Store(true)
					<-ctx.Done()
					return ctx.Err()
				}
				if key == "a" {
					switch kind {
					case "sql":
						return errors.New("SQL failed")
					case "redis":
						client.failRead.Store(true)
					case "unexplained_rejection":
						client.reject = true
					}
					return nil
				}
				<-ctx.Done()
				return ctx.Err()
			}
			p := testProducer(t, scanner, store, Config{PageSize: 2, RunLockTTL: 90 * time.Millisecond})
			if err := p.Produce(contextForTest(t), scopedMessage("failure")); err == nil {
				t.Fatal("expected fatal failure")
			}
			if scanner.inflight.Load() != 0 {
				t.Fatal("scanner goroutines leaked")
			}
			meta, err := store.GetRunMeta(context.Background(), "failure")
			if err != nil || meta.ProducerDone {
				t.Fatalf("partial scan marked done: %+v err=%v", meta, err)
			}
		})
	}
}

func TestChannelScanStateStopsAndNoTarget(t *testing.T) {
	for _, kind := range []string{"terminal", "producerDone", "blocked", "missing", "no-target", "no-cards"} {
		t.Run(kind, func(t *testing.T) {
			store := testStore(t)
			scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}}, 2)
			if kind == "no-target" {
				scanner.channels[0].SyncEnabled = true
			}
			if kind == "no-cards" {
				scanner.cards = nil
			}
			p := testProducer(t, scanner, store, Config{})
			if kind != "no-target" && kind != "no-cards" {
				if err := store.CreateRun(context.Background(), zset.RunMeta{RunID: "stop", Status: "producing"}); err != nil {
					t.Fatal(err)
				}
				key, _ := store.RunMetaKey("stop")
				switch kind {
				case "terminal":
					_, _ = store.RedisClient().Do(context.Background(), "HSET", key, "status", "completed")
				case "producerDone":
					if err := store.MarkProducerDone(context.Background(), "stop"); err != nil {
						t.Fatal(err)
					}
				case "blocked":
					_, _ = store.BlockRunScoped(context.Background(), "stop", "a")
				case "missing":
					scanner.hook = func(ctx context.Context, _ string, _ int64) error {
						_, err := store.RedisClient().Do(ctx, "DEL", key)
						return err
					}
				}
			}
			err := p.Produce(contextForTest(t), scopedMessage("stop"))
			if kind == "terminal" {
				if !errors.Is(err, ErrRunFinished) {
					t.Fatalf("err=%v", err)
				}
			} else if kind == "missing" {
				if !errors.Is(err, zset.ErrRunNotFound) {
					t.Fatalf("err=%v", err)
				}
			} else if err != nil {
				t.Fatal(err)
			}
			if len(taskIDs(t, store, "stop")) != 0 {
				t.Fatal("stopped/empty run enqueued tasks")
			}
		})
	}
}

func TestChannelOldNonRunPathRemainsSerial(t *testing.T) {
	redis := &producerRedis{batches: make(map[string]int)}
	store, _ := zset.New(redis, "flow:")
	scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}}, 1)
	p, err := New(scanner, store, Config{ScanMode: "channel", RoundRetryEnabled: false})
	if err != nil {
		t.Fatal(err)
	}
	if err = p.Produce(context.Background(), scopedMessage("legacy")); err != nil {
		t.Fatal(err)
	}
	if scanner.queries.Load() != 0 || scanner.lists.Load() != 0 {
		t.Fatal("non-Run path invoked channel scanner")
	}
}

const operation = FullSyncOperation

type fakeScanner struct{}

func (*fakeScanner) SnapshotMaxID(context.Context) (int64, error)            { return 0, nil }
func (*fakeScanner) Scan(context.Context, int64, int64, int) ([]Card, error) { return nil, nil }
func scopedMessage(run string, channels ...string) model.TrafficUsageMonthlyHistoryStartMessage {
	m := model.TrafficUsageMonthlyHistoryStartMessage{RunID: run, QueryMonth: "202609", Operation: FullSyncOperation}
	if len(channels) > 0 {
		m.Scope = &model.SyncScope{Channels: channels}
	}
	return m
}