package orbit

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

func integrationEventSubscription(t *testing.T, ctx context.Context, api *Client, maxAttempts uint32) Subscription {
	t.Helper()
	queue, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "event-sdk-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10})
	if err != nil {
		t.Fatal(err)
	}
	sub, err := api.CreateSubscription(ctx, workerTestID(t), CreateSubscriptionRequest{Name: "sdk-" + workerTestID(t), QueueID: queue.Queue.ID, Types: []string{"task.cancelled"}, MaxPending: 10, MaxAttempts: maxAttempts})
	if err != nil {
		t.Fatal(err)
	}
	return sub.Subscription
}
func integrationCancelledEvent(t *testing.T, ctx context.Context, api *Client, sub Subscription) Task {
	t.Helper()
	task, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: sub.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
	if err != nil {
		t.Fatal(err)
	}
	cancelled, err := api.CancelTask(ctx, task.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "subscription SDK regression"})
	if err != nil {
		t.Fatal(err)
	}
	return cancelled.Task
}
func eventDialIntegration(t *testing.T, ctx context.Context, api *Client, sub Subscription, capacity uint32) *EventConnection {
	t.Helper()
	c, err := api.DialEvents(ctx, EventRegistration{SubscriptionID: sub.ID, Capacity: capacity})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { c.Close() })
	return c
}
func TestIntegrationEventConnectionSDK(t *testing.T) {
	api, _, _ := integrationWorkerClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 15*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, api, 3)
	task := integrationCancelledEvent(t, ctx, api, sub)
	a := eventDialIntegration(t, ctx, api, sub, 1)
	grant, err := a.Next(ctx)
	if err != nil || grant.Event.Snapshot.TaskID != task.ID || grant.Event.Snapshot.Revision != task.Revision || grant.Event.Type != "task.cancelled" {
		t.Fatal("SDK snapshot mismatch", err)
	}
	paused, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: sub.Revision, Reason: "pause with active delivery", Enabled: false})
	if err != nil {
		t.Fatal(err)
	}
	pong, err := a.Ping(ctx, workerTestID(t))
	if err != nil || pong.Enabled {
		t.Fatal("paused subscription heartbeat", err)
	}
	first, err := a.Ack(ctx, workerTestID(t), grant.Proof())
	if err != nil {
		t.Fatal(err)
	}
	a.Close()
	b := eventDialIntegration(t, ctx, api, sub, 1)
	duplicate, err := b.Ack(ctx, workerTestID(t), grant.Proof())
	if err != nil || !duplicate.Deduplicated || !duplicate.AcknowledgedAt.Equal(first.AcknowledgedAt) {
		t.Fatal("ACK proof could not cross connections", err)
	}
	_ = integrationCancelledEvent(t, ctx, api, sub)
	short, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
	_, err = b.Next(short)
	cancel()
	if !errors.Is(err, context.DeadlineExceeded) || b.Err() != nil {
		t.Fatal("paused subscription delivered or Next closed connection", err)
	}
	if _, err = api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: paused.Subscription.Revision, Reason: "resume", Enabled: true}); err != nil {
		t.Fatal(err)
	}
	second, err := b.Next(ctx)
	if err != nil || second.Event.ID == grant.Event.ID {
		t.Fatal("resume failed", err)
	}
	if _, err = b.Ack(ctx, workerTestID(t), second.Proof()); err != nil {
		t.Fatal(err)
	}
	current, err := api.GetSubscription(ctx, sub.ID)
	if err != nil || current.PendingCount != 0 {
		t.Fatal("SDK acknowledgement accounting", err)
	}
}
func TestIntegrationEventConsumerHeartbeatAndLostAck(t *testing.T) {
	api, _, token := integrationWorkerClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 25*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, api, 3)
	_ = integrationCancelledEvent(t, ctx, api, sub)
	gateway := newProtocolFaultGateway(t, api.baseURL, "/v1/events/connect", eventSubprotocol)
	gateway.drop.Store("event.ack")
	client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()})
	if err != nil {
		t.Fatal(err)
	}
	var calls atomic.Int64
	consumer, err := NewEventConsumer(client, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, event ResultEvent) error {
		calls.Add(1)
		// 慢业务期间真实等待15秒自动心跳；确认计数来自上游服务端ACK，不是客户端写入。
		ticker := time.NewTicker(20 * time.Millisecond)
		defer ticker.Stop()
		for gateway.confirmedEventPings.Load() == 0 {
			select {
			case <-work.Done():
				return work.Err()
			case <-ticker.C:
			}
		}
		return nil
	}})
	if err != nil {
		t.Fatal(err)
	}
	run, cancel := context.WithCancel(ctx)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- consumer.Run(run) }()
	for consumer.Stats().Acknowledged == 0 {
		select {
		case <-ctx.Done():
			t.Fatal("consumer did not reconcile committed ACK")
		case err := <-done:
			t.Fatal("consumer stopped", err)
		case <-time.After(20 * time.Millisecond):
		}
	}
	cancel()
	if err := awaitConsumerStop(t, done); err != nil {
		t.Fatal(err)
	}
	if calls.Load() != 1 || consumer.Stats().Reconnects < 1 || gateway.confirmedEventPings.Load() < 1 {
		t.Fatal("heartbeat or reconciliation missing", consumer.Stats())
	}
	page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 1 {
		t.Fatal("lost ACK created another delivery attempt", err)
	}
	t.Log("real 15-second heartbeat and committed ACK loss recovered; handler and delivery attempt each remained one")
}
func TestIntegrationEventConsumerSlowHandlerBackpressure(t *testing.T) {
	api, _, _ := integrationWorkerClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 25*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, api, 3)
	_ = integrationCancelledEvent(t, ctx, api, sub)
	started := make(chan struct{})
	release := make(chan struct{})
	var calls atomic.Int64
	consumer, err := NewEventConsumer(api, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, _ ResultEvent) error {
		if calls.Add(1) == 1 {
			close(started)
			select {
			case <-release:
			case <-work.Done():
				return work.Err()
			}
		}
		return nil
	}})
	if err != nil {
		t.Fatal(err)
	}
	run, cancel := context.WithCancel(ctx)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- consumer.Run(run) }()
	select {
	case <-started:
	case <-ctx.Done():
		t.Fatal("first slow handler did not start")
	case err := <-done:
		t.Fatal("consumer stopped", err)
	}
	_ = integrationCancelledEvent(t, ctx, api, sub)
	deadline := time.Now().Add(2 * time.Second)
	for {
		current, getErr := api.GetSubscription(ctx, sub.ID)
		if getErr != nil {
			t.Fatal(getErr)
		}
		if current.PendingCount == 2 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("slow handler did not retain second delivery: pending=%d", current.PendingCount)
		}
		time.Sleep(20 * time.Millisecond)
	}
	time.Sleep(150 * time.Millisecond)
	if calls.Load() != 1 || consumer.Stats().Acknowledged != 0 {
		t.Fatalf("capacity-one consumer started or acknowledged work while handler was blocked: calls=%d stats=%+v", calls.Load(), consumer.Stats())
	}
	close(release)
	for consumer.Stats().Acknowledged < 2 {
		select {
		case <-ctx.Done():
			t.Fatal("slow consumer did not drain both deliveries")
		case err := <-done:
			t.Fatal("consumer stopped", err)
		case <-time.After(20 * time.Millisecond):
		}
	}
	cancel()
	if err := awaitConsumerStop(t, done); err != nil {
		t.Fatal(err)
	}
	page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(page.Items) != 2 || page.Items[0].Attempts != 1 || page.Items[1].Attempts != 1 || calls.Load() != 2 {
		t.Fatalf("slow-consumer backpressure or delivery accounting failed: err=%v deliveries=%+v calls=%d", err, page.Items, calls.Load())
	}
	t.Log("capacity-one slow handler held the second delivery until release; both durable deliveries were acknowledged once")
}
func TestIntegrationEventConsumerFailureRedeliversNaturally(t *testing.T) {
	api, _, _ := integrationWorkerClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, api, 2)
	_ = integrationCancelledEvent(t, ctx, api, sub)
	var calls atomic.Int64
	consumer, err := NewEventConsumer(api, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error {
		if calls.Add(1) == 1 {
			return errors.New("receiver transaction rolled back")
		}
		return nil
	}})
	if err != nil {
		t.Fatal(err)
	}
	run, cancel := context.WithCancel(ctx)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- consumer.Run(run) }()
	for consumer.Stats().Acknowledged == 0 {
		select {
		case <-ctx.Done():
			t.Fatal("delivery did not retry after natural lease expiry")
		case err := <-done:
			t.Fatal("consumer stopped", err)
		case <-time.After(50 * time.Millisecond):
		}
	}
	cancel()
	if err := awaitConsumerStop(t, done); err != nil {
		t.Fatal(err)
	}
	page, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 2 || calls.Load() != 2 || consumer.Stats().HandlerFailures != 1 {
		t.Fatal("handler error or retry budget mishandled", err)
	}
	t.Log("handler error caused no ACK; same delivery reappeared after its real 30-second lease and backoff, then acknowledged at attempt two")
}