aiaa-notification-server/internal/subscriber/dedup_test.go

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(map[string]interface{}{
		"strategyCode": "ai-crypto-signals",
		"symbol":       "CRV",
		"period":       "1h",
		"direction":    "LONG",
		"price":        0.2528,
		"eventTime":    int64(1),
	})
	b := SignalHash(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(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")
	}
}

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 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())
	}
}