package redis_test

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"log/slog"
	"sync"
	"testing"
	"time"

	miniredis "github.com/alicebob/miniredis/v2"
	goredis "github.com/redis/go-redis/v9"

	"bit.admedia.com/scm/ad/adpilot-indexing-commit-intel.com/internal/config"
	contractevents "bit.admedia.com/scm/ad/adpilot-indexing-commit-intel.com/internal/contracts/events"
	redisclient "bit.admedia.com/scm/ad/adpilot-indexing-commit-intel.com/internal/redis"
)

type mockDeltaHandler struct {
	mu    sync.Mutex
	calls []contractevents.GraphDeltaReadyEvent
}

func (m *mockDeltaHandler) ProcessGraphDeltaReady(_ context.Context, event contractevents.GraphDeltaReadyEvent) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.calls = append(m.calls, event)
	return nil
}

func (m *mockDeltaHandler) RecordGraphDeltaFailure(context.Context, contractevents.GraphDeltaReadyEvent, string) error {
	return nil
}

func (m *mockDeltaHandler) callCount() int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return len(m.calls)
}

type failingDeltaHandler struct{}

func (f *failingDeltaHandler) ProcessGraphDeltaReady(context.Context, contractevents.GraphDeltaReadyEvent) error {
	return errors.New("transient analysis failure")
}

func (f *failingDeltaHandler) RecordGraphDeltaFailure(context.Context, contractevents.GraphDeltaReadyEvent, string) error {
	return nil
}

func setupConsumer(t *testing.T, handler redisclient.GraphDeltaProcessor) (*miniredis.Miniredis, *redisclient.Consumer, config.Config) {
	t.Helper()

	server, err := miniredis.Run()
	if err != nil {
		t.Fatalf("start miniredis: %v", err)
	}

	redisCfg := config.RedisConfig{
		URL:           "redis://" + server.Addr(),
		ConsumerGroup: "commit-intelligence-service",
	}
	appCfg := config.Config{
		Redis: redisCfg,
		Stream: config.StreamConsumerConfig{
			Group:      "commit-intelligence-service",
			MaxRetries: 5,
			BlockMs:    50,
		},
		Analysis: config.AnalysisConfig{Workers: 2},
	}
	client, err := redisclient.New(redisCfg)
	if err != nil {
		server.Close()
		t.Fatalf("new client: %v", err)
	}

	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	consumer := redisclient.NewConsumer(client, appCfg, handler, logger)

	t.Cleanup(func() {
		_ = client.Close()
		server.Close()
	})

	return server, consumer, appCfg
}

func xaddEvent(t *testing.T, server *miniredis.Miniredis, stream string, event any) {
	t.Helper()

	payload, err := json.Marshal(event)
	if err != nil {
		t.Fatalf("marshal event: %v", err)
	}

	if _, err := server.XAdd(stream, "*", []string{"payload", string(payload)}); err != nil {
		t.Fatalf("xadd: %v", err)
	}
}

func pendingCount(t *testing.T, server *miniredis.Miniredis, stream, group string) int {
	t.Helper()

	client := goredis.NewClient(&goredis.Options{Addr: server.Addr()})
	defer client.Close()

	pending, err := client.XPending(context.Background(), stream, group).Result()
	if err != nil {
		t.Fatalf("xpending: %v", err)
	}
	return int(pending.Count)
}

func startDeltaConsumer(t *testing.T, consumer *redisclient.Consumer) context.CancelFunc {
	t.Helper()

	ctx, cancel := context.WithCancel(context.Background())
	go func() { _ = consumer.ConsumeGraphDeltaReady(ctx) }()
	time.Sleep(50 * time.Millisecond)
	return cancel
}

func startArtifactConsumer(t *testing.T, consumer *redisclient.Consumer) context.CancelFunc {
	t.Helper()

	ctx, cancel := context.WithCancel(context.Background())
	go func() { _ = consumer.ConsumeGraphArtifactReady(ctx) }()
	time.Sleep(50 * time.Millisecond)
	return cancel
}

func waitForDeltaCalls(t *testing.T, handler *mockDeltaHandler, want int) {
	t.Helper()

	deadline := time.After(3 * time.Second)
	for handler.callCount() < want {
		select {
		case <-deadline:
			t.Fatalf("timeout waiting for %d ProcessGraphDeltaReady calls, got %d", want, handler.callCount())
		default:
			time.Sleep(10 * time.Millisecond)
		}
	}
}

func TestConsumer_GraphDeltaReadyDelegates(t *testing.T) {
	handler := &mockDeltaHandler{}
	server, consumer, cfg := setupConsumer(t, handler)

	cancel := startDeltaConsumer(t, consumer)
	defer cancel()

	event := contractevents.NewGraphDeltaReadyEvent(
		"evt_delta_01",
		"ad/example",
		"base1234567890",
		"target4567890ab",
		"delta_target456_a1b2",
		"mongo://graph_deltas/delta_target456_a1b2",
		"v1",
	)
	xaddEvent(t, server, cfg.Redis.StreamName(contractevents.StreamGraphDeltaReady), event)

	waitForDeltaCalls(t, handler, 1)

	if handler.calls[0].DeltaArtifactID != "delta_target456_a1b2" {
		t.Fatalf("unexpected delta id: %q", handler.calls[0].DeltaArtifactID)
	}
}

func TestConsumer_GraphArtifactReadyAckOnly(t *testing.T) {
	handler := &mockDeltaHandler{}
	server, consumer, cfg := setupConsumer(t, handler)

	cancel := startArtifactConsumer(t, consumer)
	defer cancel()

	event := contractevents.NewGraphArtifactReadyEvent(
		"evt_artifact_01",
		"ad/example",
		"snap_01",
		"abc123",
		"artifact_01",
		"mongo://snapshot_graphs/artifact_01",
		"v0.1.0",
		"v1",
	)
	xaddEvent(t, server, cfg.Redis.StreamName(contractevents.StreamGraphArtifactReady), event)

	time.Sleep(300 * time.Millisecond)

	if handler.callCount() != 0 {
		t.Fatalf("expected 0 delta calls, got %d", handler.callCount())
	}

	stream := cfg.Redis.StreamName(contractevents.StreamGraphArtifactReady)
	if pending := pendingCount(t, server, stream, "commit-intelligence-service"); pending != 0 {
		t.Fatalf("expected 0 pending messages, got %d", pending)
	}
}

func TestConsumer_GraphDeltaReadySkipsInvalid(t *testing.T) {
	handler := &mockDeltaHandler{}
	server, consumer, cfg := setupConsumer(t, handler)

	cancel := startDeltaConsumer(t, consumer)
	defer cancel()

	xaddEvent(t, server, cfg.Redis.StreamName(contractevents.StreamGraphDeltaReady), map[string]string{"bad": "event"})

	time.Sleep(300 * time.Millisecond)

	if handler.callCount() != 0 {
		t.Fatalf("expected 0 calls for invalid event, got %d", handler.callCount())
	}
}

func TestConsumer_TransientFailureLeavesPending(t *testing.T) {
	handler := &failingDeltaHandler{}
	server, consumer, cfg := setupConsumer(t, handler)

	cancel := startDeltaConsumer(t, consumer)
	defer cancel()

	stream := cfg.Redis.StreamName(contractevents.StreamGraphDeltaReady)
	event := contractevents.NewGraphDeltaReadyEvent(
		"evt_fail_01",
		"ad/example",
		"base1234567890",
		"target4567890ab",
		"delta_target456_a1b2",
		"mongo://graph_deltas/delta_target456_a1b2",
		"v1",
	)
	xaddEvent(t, server, stream, event)

	time.Sleep(500 * time.Millisecond)

	if pending := pendingCount(t, server, stream, "commit-intelligence-service"); pending == 0 {
		t.Fatal("expected transient failure to leave message pending")
	}
}