package store

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"telesrv/internal/domain"
)

type blockingReadModelVersionStore struct {
	calls   atomic.Int32
	once    sync.Once
	started chan struct{}
	release chan struct{}
	hashes  map[ReadModelKey]int64
	mu      sync.Mutex
	loaded  map[ReadModelKey]int
}

func (s *blockingReadModelVersionStore) ReadModelHash(ctx context.Context, model string, ownerUserID int64, peerType domain.PeerType, peerID int64) (int64, bool, error) {
	rows, err := s.ReadModelHashes(ctx, []ReadModelKey{{Model: model, OwnerUserID: ownerUserID, PeerType: peerType, PeerID: peerID}})
	if err != nil {
		return 0, false, err
	}
	key := ReadModelKey{Model: model, OwnerUserID: ownerUserID, PeerType: peerType, PeerID: peerID}
	hash := rows[key]
	return hash, hash != 0, nil
}

func (s *blockingReadModelVersionStore) ReadModelHashes(ctx context.Context, keys []ReadModelKey) (map[ReadModelKey]int64, error) {
	s.calls.Add(1)
	s.once.Do(func() { close(s.started) })
	select {
	case <-s.release:
	case <-ctx.Done():
		return nil, ctx.Err()
	}
	out := make(map[ReadModelKey]int64, len(keys))
	s.mu.Lock()
	if s.loaded == nil {
		s.loaded = make(map[ReadModelKey]int)
	}
	for _, key := range keys {
		s.loaded[key]++
		if hash := s.hashes[key]; hash != 0 {
			out[key] = hash
		}
	}
	s.mu.Unlock()
	return out, nil
}

func TestCachedReadModelVersionStoreSingleflightsBatchMiss(t *testing.T) {
	ctx := context.Background()
	keys := []ReadModelKey{
		{Model: "channel_base", OwnerUserID: 0, PeerType: domain.PeerTypeChannel, PeerID: 10},
		{Model: "channel_member", OwnerUserID: 100, PeerType: domain.PeerTypeChannel, PeerID: 10},
	}
	base := &blockingReadModelVersionStore{
		started: make(chan struct{}),
		release: make(chan struct{}),
		hashes: map[ReadModelKey]int64{
			keys[0]: 11,
			keys[1]: 22,
		},
	}
	cache := NewCachedReadModelVersionStore(base, time.Hour, 100)

	const goroutines = 16
	errs := make(chan error, goroutines)
	var wg sync.WaitGroup
	wg.Add(goroutines)
	for i := 0; i < goroutines; i++ {
		go func() {
			defer wg.Done()
			rows, err := cache.ReadModelHashes(ctx, keys)
			if err != nil {
				errs <- err
				return
			}
			if rows[keys[0]] != 11 || rows[keys[1]] != 22 {
				errs <- fmt.Errorf("hashes = %+v, want 11/22", rows)
			}
		}()
	}
	<-base.started
	time.Sleep(20 * time.Millisecond)
	close(base.release)
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatal(err)
		}
	}
	if got := base.calls.Load(); got != 1 {
		t.Fatalf("base ReadModelHashes calls = %d, want 1", got)
	}
	if _, err := cache.ReadModelHashes(ctx, keys); err != nil {
		t.Fatalf("cached ReadModelHashes: %v", err)
	}
	if got := base.calls.Load(); got != 1 {
		t.Fatalf("cache hit called base again: calls=%d", got)
	}
}

func TestCachedReadModelVersionStoreWaitsOverlappingBatchKeys(t *testing.T) {
	ctx := context.Background()
	a := ReadModelKey{Model: "channel_base", OwnerUserID: 0, PeerType: domain.PeerTypeChannel, PeerID: 10}
	b := ReadModelKey{Model: "channel_member", OwnerUserID: 100, PeerType: domain.PeerTypeChannel, PeerID: 10}
	c := ReadModelKey{Model: "dialog_light", OwnerUserID: 100, PeerType: domain.PeerTypeChannel, PeerID: 10}
	base := &blockingReadModelVersionStore{
		started: make(chan struct{}),
		release: make(chan struct{}),
		hashes: map[ReadModelKey]int64{
			a: 11,
			b: 22,
			c: 33,
		},
	}
	cache := NewCachedReadModelVersionStore(base, time.Hour, 100)

	firstErr := make(chan error, 1)
	go func() {
		rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{a, b})
		if err == nil && (rows[a] != 11 || rows[b] != 22) {
			err = fmt.Errorf("first hashes = %+v, want A/B", rows)
		}
		firstErr <- err
	}()
	<-base.started

	secondErr := make(chan error, 1)
	go func() {
		rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{b, c})
		if err == nil && (rows[b] != 22 || rows[c] != 33) {
			err = fmt.Errorf("second hashes = %+v, want B/C", rows)
		}
		secondErr <- err
	}()
	time.Sleep(20 * time.Millisecond)
	close(base.release)

	if err := <-firstErr; err != nil {
		t.Fatal(err)
	}
	if err := <-secondErr; err != nil {
		t.Fatal(err)
	}
	base.mu.Lock()
	loadedB := base.loaded[b]
	loadedC := base.loaded[c]
	base.mu.Unlock()
	if loadedB != 1 {
		t.Fatalf("overlap key B loaded %d times, want 1", loadedB)
	}
	if loadedC != 1 {
		t.Fatalf("owned key C loaded %d times, want 1", loadedC)
	}
}

// TestCachedReadModelVersionStoreEpochGuardRejectsStaleWriteback 证明 epoch 守卫堵住了
// lost-update：一次锁外 DB load 期间到达的 NOTIFY(写入新 hash)不得被 load 返回的旧 hash 覆盖。
// 这是 4 个 hash-only 消费缓存(private/channel media counts、participants、active ids)正确性的根。
func TestCachedReadModelVersionStoreEpochGuardRejectsStaleWriteback(t *testing.T) {
	ctx := context.Background()
	key := ReadModelKey{Model: "private_media_counts", OwnerUserID: 100, PeerType: domain.PeerTypeUser, PeerID: 100}
	base := &blockingReadModelVersionStore{
		started: make(chan struct{}),
		release: make(chan struct{}),
		hashes:  map[ReadModelKey]int64{key: 11}, // DB 此刻仍是旧 hash 11
	}
	cache := NewCachedReadModelVersionStore(base, time.Hour, 100)

	resultCh := make(chan map[ReadModelKey]int64, 1)
	errCh := make(chan error, 1)
	go func() {
		rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{key})
		if err != nil {
			errCh <- err
			return
		}
		resultCh <- rows
	}()

	// 等到 load 已 claim inflight 并阻塞在 base.ReadModelHashes 内。
	<-base.started
	// 模拟 NOTIFY 在 load 期间送来权威新 hash 99（写入并自增 epoch）。
	cache.UpdateReadModelHash(key, 99)
	// 放行 load：它会返回旧 hash 11；epoch 守卫必须拒绝用 11 覆盖 99。
	close(base.release)

	select {
	case err := <-errCh:
		t.Fatalf("ReadModelHashes: %v", err)
	case rows := <-resultCh:
		if rows[key] != 99 {
			t.Fatalf("in-flight reader got hash %d, want fresh 99 (stale 11 must not win)", rows[key])
		}
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for in-flight reader")
	}

	// 后续读仍须是 99，且不再打 base —— 证明缓存没被旧值污染、也没被错误驱逐。
	callsBefore := base.calls.Load()
	rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{key})
	if err != nil {
		t.Fatalf("second ReadModelHashes: %v", err)
	}
	if rows[key] != 99 {
		t.Fatalf("cached hash = %d, want 99 (cache poisoned by stale 11)", rows[key])
	}
	if got := base.calls.Load(); got != callsBefore {
		t.Fatalf("second read hit base (calls %d -> %d): cache was poisoned or evicted", callsBefore, got)
	}
}

func TestCachedReadModelVersionStoreUpdateReadModelHashWarmsCache(t *testing.T) {
	ctx := context.Background()
	key := ReadModelKey{Model: "channel_base", PeerType: domain.PeerTypeChannel, PeerID: 10}
	base := &blockingReadModelVersionStore{
		started: make(chan struct{}),
		release: make(chan struct{}),
		hashes:  map[ReadModelKey]int64{key: 11},
	}
	cache := NewCachedReadModelVersionStore(base, time.Hour, 100)

	cache.UpdateReadModelHash(key, 99)
	rows, err := cache.ReadModelHashes(ctx, []ReadModelKey{key})
	if err != nil {
		t.Fatalf("ReadModelHashes: %v", err)
	}
	if rows[key] != 99 {
		t.Fatalf("hash = %d, want notify-provided 99", rows[key])
	}
	if got := base.calls.Load(); got != 0 {
		t.Fatalf("base calls = %d, want cache warmed by notify", got)
	}
}