package subscriber

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

	"aiaa-notification-service/internal/model"
	"aiaa-notification-service/internal/notify"
	"aiaa-notification-service/internal/subscriber/cryptostrategy"
	"aiaa-notification-service/internal/subscriber/tradesignal"
)

func TestMessageHashStable(t *testing.T) {
	a := MessageHash([]byte(`{"action":"OPEN","signalId":"s1"}`))
	b := MessageHash([]byte(`{"action":"OPEN","signalId":"s1"}`))
	c := MessageHash([]byte(`{"action":"CLOSE","signalId":"s1"}`))
	if a == "" || a != b {
		t.Fatalf("hash should be stable, a=%q b=%q", a, b)
	}
	if a == c {
		t.Fatal("different bodies should hash differently")
	}
}

func TestSignalHashIgnoresUnrelatedFields(t *testing.T) {
	a := SignalHash("crypto-strategy", map[string]interface{}{
		"strategyCode": "ai-crypto-signals",
		"symbol":       "CRV",
		"period":       "1h",
		"direction":    "LONG",
		"price":        0.2528,
		"eventTime":    int64(1),
	})
	b := SignalHash("crypto-strategy", map[string]interface{}{
		"strategyCode": "ai-crypto-signals",
		"currency":     "CRV",
		"period":       "1h",
		"side":         "long",
		"price":        0.2528,
		"eventTime":    int64(2),
	})
	if a == "" || a != b {
		t.Fatalf("same signal fields should hash equal, a=%q b=%q", a, b)
	}
	c := SignalHash("crypto-strategy", map[string]interface{}{
		"strategyCode": "ai-crypto-signals",
		"symbol":       "CRV",
		"period":       "1h",
		"direction":    "LONG",
		"price":        0.26,
	})
	if a == c {
		t.Fatal("different price should hash differently")
	}
	otherSrc := SignalHash("trade-signal", map[string]interface{}{
		"strategyCode": "ai-crypto-signals",
		"symbol":       "CRV",
		"period":       "1h",
		"direction":    "LONG",
		"price":        0.2528,
	})
	if a == otherSrc {
		t.Fatal("different sources should hash differently")
	}
}

func TestSignalHashDistinguishesAction(t *testing.T) {
	base := map[string]interface{}{
		"strategyCode": "BLONG",
		"symbol":       "SOLUSDT",
		"period":       "30m",
		"side":         "SHORT",
		"price":        101.32,
	}
	reduce := cloneFields(base)
	reduce["action"] = "REDUCE"
	closeMsg := cloneFields(base)
	closeMsg["action"] = "CLOSE"
	if SignalHash("trade-signal", reduce) == SignalHash("trade-signal", closeMsg) {
		t.Fatal("REDUCE and CLOSE at the same price should hash differently")
	}
}

func cloneFields(in map[string]interface{}) map[string]interface{} {
	out := make(map[string]interface{}, len(in)+1)
	for k, v := range in {
		out[k] = v
	}
	return out
}

func TestMemoryDeduperClaimOnce(t *testing.T) {
	d := NewMemoryDeduper()
	ok, err := d.Claim(context.Background(), "abc")
	if err != nil || !ok {
		t.Fatalf("first claim ok=%v err=%v", ok, err)
	}
	ok, err = d.Claim(context.Background(), "abc")
	if err != nil || ok {
		t.Fatalf("second claim should miss, ok=%v err=%v", ok, err)
	}
	if err := d.Release(context.Background(), "abc"); err != nil {
		t.Fatal(err)
	}
	ok, err = d.Claim(context.Background(), "abc")
	if err != nil || !ok {
		t.Fatalf("after release should claim, ok=%v err=%v", ok, err)
	}
}

func TestHandleDedupByStrategySymbolPeriodDirectionPrice(t *testing.T) {
	dedup := NewMemoryDeduper()
	var n atomic.Int32
	process := func(context.Context, notify.Request) (notify.Result, error) {
		n.Add(1)
		return notify.Result{Matched: true}, nil
	}
	lookup := func(context.Context, string) (*model.Source, error) {
		return &model.Source{ID: 1, Name: "crypto-strategy", Status: 1}, nil
	}
	conv := cryptostrategy.NewConverter()
	in := func(eventTime int64) HandleInput {
		body := []byte(fmt.Sprintf(`{
			"eventType":"SIGNAL_RECEIVED","symbol":"CRV","direction":"LONG",
			"payload":"{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"CRV\",\"isClose\":true,\"isGain\":true,\"price\":0.2528}",
			"eventTime":%d
		}`, eventTime))
		return HandleInput{Body: body, SourceName: "crypto-strategy", MaxRetry: 3, Deduper: dedup}
	}

	if d := HandleMessage(context.Background(), in(1786899538978), conv, lookup, process); d != DispositionAck {
		t.Fatalf("first=%v", d)
	}
	if d := HandleMessage(context.Background(), in(1786899539730), conv, lookup, process); d != DispositionAck {
		t.Fatalf("dup=%v", d)
	}
	if n.Load() != 1 {
		t.Fatalf("same strategy/symbol/period/direction/price should process once, got %d", n.Load())
	}

	bodyDiffPrice := []byte(`{
		"eventType":"SIGNAL_RECEIVED","symbol":"CRV","direction":"LONG",
		"payload":"{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"CRV\",\"isClose\":true,\"isGain\":true,\"price\":0.26}",
		"eventTime":1786899539731
	}`)
	if d := HandleMessage(context.Background(), HandleInput{
		Body: bodyDiffPrice, SourceName: "crypto-strategy", MaxRetry: 3, Deduper: dedup,
	}, conv, lookup, process); d != DispositionAck {
		t.Fatalf("diff price=%v", d)
	}
	if n.Load() != 2 {
		t.Fatalf("different price should process again, got %d", n.Load())
	}
}

func TestHandleDedupKeepsReduceThenClose(t *testing.T) {
	dedup := NewMemoryDeduper()
	var events []string
	process := func(_ context.Context, req notify.Request) (notify.Result, error) {
		events = append(events, req.Event)
		return notify.Result{Matched: true}, nil
	}
	lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }
	conv := tradesignal.NewConverter(nil)
	reduce := []byte(`{
		"signalId":"local-0f6f4b00e143f8ee:update:1787629790828:0.12",
		"sourcePosId":"local-0f6f4b00e143f8ee","sourcePosIds":["3340443882114850823"],
		"strategyCode":"BLONG","symbol":"SOLUSDT","side":"SHORT","action":"REDUCE",
		"quantity":0.11,"price":101.32,"leverage":20,"period":"30m",
		"eventTime":"2026-08-25T03:49:51.155Z","addCount":0,"totalPos":0.12,
		"totalAvgPx":102.64,"posMarginRatio":0.478261,"oldQuantity":0.23,"deltaQuantity":-0.11
	}`)
	closeBody := []byte(`{
		"signalId":"local-0f6f4b00e143f8ee:close:1787629790936:0",
		"sourcePosId":"local-0f6f4b00e143f8ee","sourcePosIds":["3340443882114850823"],
		"strategyCode":"BLONG","symbol":"SOLUSDT","side":"SHORT","action":"CLOSE",
		"quantity":0.12,"price":101.32,"leverage":1,"period":"30m",
		"eventTime":"2026-08-25T03:49:51.389Z","addCount":0,"totalPos":0,
		"totalAvgPx":0,"posMarginRatio":1,"oldQuantity":0.12
	}`)
	if d := HandleMessage(context.Background(), HandleInput{
		Body: reduce, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup,
	}, conv, lookup, process); d != DispositionAck {
		t.Fatalf("reduce=%v", d)
	}
	if d := HandleMessage(context.Background(), HandleInput{
		Body: closeBody, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup,
	}, conv, lookup, process); d != DispositionAck {
		t.Fatalf("close=%v", d)
	}
	if len(events) != 2 {
		t.Fatalf("reduce then close should both notify, got %v", events)
	}
	if events[0] != "trade.reduce" || events[1] != "trade.close" {
		t.Fatalf("events=%v", events)
	}
}

func TestHandleDedupKeepsDifferentSources(t *testing.T) {
	dedup := NewMemoryDeduper()
	var n atomic.Int32
	process := func(context.Context, notify.Request) (notify.Result, error) {
		n.Add(1)
		return notify.Result{Matched: true}, nil
	}
	lookup := func(_ context.Context, name string) (*model.Source, error) {
		return &model.Source{ID: 1, Name: name, Status: 1}, nil
	}
	conv := cryptostrategy.NewConverter()
	body := []byte(`{
		"eventType":"SIGNAL_RECEIVED","symbol":"CRV","direction":"LONG",
		"payload":"{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"CRV\",\"isClose\":true,\"isGain\":true,\"price\":0.2528}",
		"eventTime":1786899538978
	}`)
	if d := HandleMessage(context.Background(), HandleInput{
		Body: body, SourceName: "crypto-strategy", MaxRetry: 3, Deduper: dedup,
	}, conv, lookup, process); d != DispositionAck {
		t.Fatalf("first=%v", d)
	}
	if d := HandleMessage(context.Background(), HandleInput{
		Body: body, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup,
	}, conv, lookup, process); d != DispositionAck {
		t.Fatalf("other source=%v", d)
	}
	if n.Load() != 2 {
		t.Fatalf("different sources should both process, got %d", n.Load())
	}
}

func TestHandleDuplicateAckSkipsProcess(t *testing.T) {
	dedup := NewMemoryDeduper()
	var n atomic.Int32
	process := func(context.Context, notify.Request) (notify.Result, error) {
		n.Add(1)
		return notify.Result{Matched: true}, nil
	}
	lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }
	conv := tradesignal.NewConverter(nil)
	body := []byte(`{"action":"OPEN","signalId":"dup-1"}`)
	in := HandleInput{Body: body, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup}

	if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionAck {
		t.Fatalf("first=%v", d)
	}
	if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionAck {
		t.Fatalf("dup=%v", d)
	}
	if n.Load() != 1 {
		t.Fatalf("process called %d times, want 1", n.Load())
	}
}

func TestHandleProcessErrorReleasesDedup(t *testing.T) {
	dedup := NewMemoryDeduper()
	var n atomic.Int32
	process := func(context.Context, notify.Request) (notify.Result, error) {
		n.Add(1)
		return notify.Result{}, errors.New("db down")
	}
	lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }
	conv := tradesignal.NewConverter(nil)
	body := []byte(`{"action":"OPEN","signalId":"retry-1"}`)
	in := HandleInput{Body: body, SourceName: "s", MaxRetry: 3, Deduper: dedup}

	if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionRetry {
		t.Fatalf("first=%v", d)
	}
	if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionRetry {
		t.Fatalf("retry should process again, got %v", d)
	}
	if n.Load() != 2 {
		t.Fatalf("process called %d times, want 2", n.Load())
	}
}