package logic

import (
	"context"
	"encoding/json"
	"sync/atomic"
	"testing"
	"time"

	"spark-server/internal/module/deliver/entity"
	"spark-server/internal/module/deliver/model"
)

func TestMemoryQueueAckUsesOriginalBytes(t *testing.T) {
	q := model.NewMemoryQueue()
	ctx := t.Context()
	job := &entity.Job{MsgID: 9, FromUserID: 1, ExcludeDevice: 7, ToUserIDs: []int64{2}}
	if err := q.Push(ctx, job); err != nil {
		t.Fatal(err)
	}
	got, err := q.Take(ctx, time.Second)
	if err != nil || got == nil || got.MsgID != 9 {
		t.Fatalf("take %+v %v", got, err)
	}
	got.FromUserID = 99
	if err := q.Ack(ctx, got); err != nil {
		t.Fatal(err)
	}
	if err := q.RequeueStale(ctx); err != nil {
		t.Fatal(err)
	}
	again, err := q.Take(ctx, 10*time.Millisecond)
	if err != nil {
		t.Fatal(err)
	}
	if again != nil {
		t.Fatalf("acked job still in processing %+v", again)
	}
}

func TestDeduperSkipsSecond(t *testing.T) {
	d := model.NewMemoryDone()
	first, err := d.Mark(t.Context(), 11)
	if err != nil || !first {
		t.Fatalf("first %v %v", first, err)
	}
	second, err := d.Mark(t.Context(), 11)
	if err != nil || second {
		t.Fatalf("second %v %v", second, err)
	}
}

func TestWorkerIdempotentAck(t *testing.T) {
	q := model.NewMemoryQueue()
	d := model.NewMemoryDone()
	var n atomic.Int32
	w := NewWorker(nil, q, d, WorkerOptions{
		Workers: 1, TakeTimeout: 50 * time.Millisecond,
		Handler: func(context.Context, *entity.Job) error { n.Add(1); return nil },
	})
	ctx := t.Context()
	raw, _ := json.Marshal(&entity.Job{MsgID: 3, Cmd: 6})
	if err := q.Push(ctx, &entity.Job{MsgID: 3}); err != nil {
		t.Fatal(err)
	}
	_ = raw
	w.Start(ctx)
	defer w.Stop()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && n.Load() == 0 {
		time.Sleep(10 * time.Millisecond)
	}
	if n.Load() != 1 {
		t.Fatalf("handled %d", n.Load())
	}
	if err := q.Push(ctx, &entity.Job{MsgID: 3}); err != nil {
		t.Fatal(err)
	}
	time.Sleep(150 * time.Millisecond)
	if n.Load() != 1 {
		t.Fatalf("idempotent %d", n.Load())
	}
}

func TestWorkerRecoversPanic(t *testing.T) {
	q := model.NewMemoryQueue()
	var n atomic.Int32
	w := NewWorker(nil, q, model.NewMemoryDone(), WorkerOptions{
		Workers: 1, TakeTimeout: 50 * time.Millisecond,
		Handler: func(_ context.Context, job *entity.Job) error {
			n.Add(1)
			if job.MsgID == 1 {
				panic("boom")
			}
			return nil
		},
	})
	ctx := t.Context()
	_ = q.Push(ctx, &entity.Job{MsgID: 1})
	_ = q.Push(ctx, &entity.Job{MsgID: 2})
	w.Start(ctx)
	defer w.Stop()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) && n.Load() < 2 {
		time.Sleep(10 * time.Millisecond)
	}
	if n.Load() < 2 {
		t.Fatalf("worker died after panic, handled %d", n.Load())
	}
}