package orbit

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

func TestIntegrationEventProcessLeaseRecovery(t *testing.T) {
	a := startOrbitProcess(t)
	_, _, token := integrationWorkerClient(t)
	api := processClient(t, a, token)
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, api, 3)
	_ = integrationCancelledEvent(t, ctx, api, sub)
	firstConn := eventDialIntegration(t, ctx, api, sub, 1)
	first, err := firstConn.Next(ctx)
	if err != nil {
		t.Fatal(err)
	}
	a.kill(t)
	select {
	case <-firstConn.Done():
	case <-time.After(3 * time.Second):
		t.Fatal("dead server connection remained live")
	}
	b := startOrbitProcess(t)
	other := processClient(t, b, token)
	secondConn := eventDialIntegration(t, ctx, other, sub, 1)
	second, err := secondConn.Next(ctx)
	if err != nil || first.Event.ID != second.Event.ID || first.Delivery.ID != second.Delivery.ID || second.Delivery.LeaseEpoch != 2 || second.Delivery.Attempts != 2 {
		t.Fatal("node restart did not recover original delivery", err)
	}
	_, err = secondConn.Ack(ctx, workerTestID(t), first.Proof())
	var remote *EventError
	if !errors.As(err, &remote) || remote.Code != "DELIVERY_LEASE_LOST" {
		t.Fatal("dead node proof acknowledged newer lease", err)
	}
	if _, err = secondConn.Ack(ctx, workerTestID(t), second.Proof()); err != nil {
		t.Fatal(err)
	}
	current, err := other.GetSubscription(ctx, sub.ID)
	if err != nil || current.PendingCount != 0 {
		t.Fatal("recovered acknowledgement not persisted", err)
	}
	t.Log("SIGKILL confirmed by Wait; new process recovered original delivery after natural lease expiry, rejected old proof and committed new ACK")
}
func TestIntegrationEventProcessAckCommitResponseLoss(t *testing.T) {
	a := startOrbitProcess(t)
	_, _, token := integrationWorkerClient(t)
	api := processClient(t, a, token)
	ctx, stop := context.WithTimeout(context.Background(), 20*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, api, 3)
	_ = integrationCancelledEvent(t, ctx, api, sub)
	gateway := newProtocolFaultGateway(t, a.url, "/v1/events/connect", eventSubprotocol)
	gateway.drop.Store("event.ack")
	gateway.dropOffline.Store(true)
	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(context.Context, ResultEvent) error { calls.Add(1); 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) }()
	awaitProcessDrop(t, ctx, gateway, "event.ack")
	before, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(before.Items) != 1 || before.Items[0].AcknowledgedAt == nil {
		t.Fatal("fault occurred before durable ACK", err)
	}
	a.kill(t)
	b := startOrbitProcess(t)
	other := processClient(t, b, token)
	gateway.target.Store(b.url)
	gateway.offline.Store(false)
	for consumer.Stats().Acknowledged == 0 {
		select {
		case <-ctx.Done():
			t.Fatal("consumer failed to reconcile ACK on new process")
		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)
	}
	after, err := other.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(after.Items) != 1 || after.Items[0].Attempts != 1 || after.Items[0].AcknowledgedAt == nil || !after.Items[0].AcknowledgedAt.Equal(*before.Items[0].AcknowledgedAt) || calls.Load() != 1 {
		t.Fatal("ACK receipt changed or committed handler reran", err)
	}
	t.Log("ACK commit verified before response loss and SIGKILL; replacement process returned original receipt without another handler or delivery attempt")
}

func TestIntegrationEventProcessSlowConsumerPeerProgress(t *testing.T) {
	a := startOrbitProcess(t)
	_, _, token := integrationWorkerClient(t)
	apiA := processClient(t, a, token)
	b := startOrbitProcess(t)
	apiB := processClient(t, b, token)
	if a.nodeID == b.nodeID {
		t.Fatal("server processes reused node identity")
	}
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, apiA, 3)
	_ = integrationCancelledEvent(t, ctx, apiA, sub)
	started := make(chan struct{})
	release := make(chan struct{})
	var callsA, callsB atomic.Int64
	consumerA, err := NewEventConsumer(apiA, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, _ ResultEvent) error {
		if callsA.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()
	doneA := make(chan error, 1)
	go func() { doneA <- consumerA.Run(run) }()
	select {
	case <-started:
	case <-ctx.Done():
		t.Fatal("node A slow handler did not start")
	case err := <-doneA:
		t.Fatal("node A consumer stopped", err)
	}
	consumerB, err := NewEventConsumer(apiB, EventConsumerOptions{SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second, Handle: func(context.Context, ResultEvent) error {
		callsB.Add(1)
		return nil
	}})
	if err != nil {
		t.Fatal(err)
	}
	doneB := make(chan error, 1)
	go func() { doneB <- consumerB.Run(run) }()
	awaitConsumerEvent(t, consumerB, "connected")
	_ = integrationCancelledEvent(t, ctx, apiA, sub)
	deadline := time.Now().Add(5 * time.Second)
	for consumerB.Stats().Acknowledged < 1 {
		select {
		case <-ctx.Done():
			t.Fatal("node B did not progress while node A was slow")
		case err := <-doneB:
			t.Fatal("node B consumer stopped", err)
		case <-time.After(20 * time.Millisecond):
		}
		if time.Now().After(deadline) {
			t.Fatal("node B did not acknowledge peer delivery before deadline")
		}
	}
	if callsA.Load() != 1 || callsB.Load() != 1 || consumerA.Stats().Acknowledged != 0 {
		t.Fatalf("peer progress violated slow consumer isolation: callsA=%d callsB=%d statsA=%+v statsB=%+v", callsA.Load(), callsB.Load(), consumerA.Stats(), consumerB.Stats())
	}
	close(release)
	for consumerA.Stats().Acknowledged < 1 {
		select {
		case <-ctx.Done():
			t.Fatal("node A did not drain its held delivery")
		case err := <-doneA:
			t.Fatal("node A consumer stopped", err)
		case <-time.After(20 * time.Millisecond):
		}
	}
	cancel()
	if err := awaitConsumerStop(t, doneA); err != nil {
		t.Fatal(err)
	}
	if err := awaitConsumerStop(t, doneB); err != nil {
		t.Fatal(err)
	}
	page, err := apiB.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(page.Items) != 2 || page.Items[0].Attempts != 1 || page.Items[1].Attempts != 1 {
		t.Fatalf("peer delivery accounting failed: err=%v deliveries=%+v", err, page.Items)
	}
	t.Log("node B acknowledged a second delivery while node A's capacity-one handler was blocked; both nodes preserved one attempt per delivery")
}

// TestIntegrationEventProcessCrossNodeReconnectStorm exercises the failure mode
// where a consumer loses several WebSocket sessions while the gateway alternates
// between two Orbit nodes. The durable delivery must be acknowledged once and
// the handler must never run again during the reconnect storm.
func TestIntegrationEventProcessCrossNodeReconnectStorm(t *testing.T) {
	a := startOrbitProcess(t)
	_, _, token := integrationWorkerClient(t)
	apiA := processClient(t, a, token)
	b := startOrbitProcess(t)
	apiB := processClient(t, b, token)
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	sub := integrationEventSubscription(t, ctx, apiA, 3)
	_ = integrationCancelledEvent(t, ctx, apiA, sub)
	gateway := newProtocolFaultGateway(t, a.url, "/v1/events/connect", eventSubprotocol)
	client, err := NewClient(gateway.server.URL, Options{Token: token, HTTPClient: gateway.server.Client()})
	if err != nil {
		t.Fatal(err)
	}
	var calls atomic.Int64
	started := make(chan struct{})
	release := make(chan struct{})
	consumer, err := NewEventConsumer(client, EventConsumerOptions{
		SubscriptionID: sub.ID, Capacity: 1, ShutdownTimeout: time.Second,
		ReconnectMinDelay: 10 * time.Millisecond, ReconnectMaxDelay: 60 * time.Millisecond,
		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) }()
	awaitConsumerEvent(t, consumer, "connected")
	awaitConsumerEvent(t, consumer, "handler_started")

	// Close whichever node currently owns the socket, then move the gateway to
	// the other node. Repeat enough times to cover a reconnect burst, while the
	// event itself remains one durable delivery.
	for i := 0; i < 4; i++ {
		gateway.disconnectActive()
		if i%2 == 0 {
			gateway.target.Store(b.url)
		} else {
			gateway.target.Store(a.url)
		}
		awaitConsumerEvent(t, consumer, "connected")
	}
	close(release)
	for consumer.Stats().Acknowledged < 1 {
		select {
		case <-ctx.Done():
			t.Fatal("cross-node consumer did not acknowledge delivery")
		case err := <-done:
			t.Fatalf("consumer stopped before acknowledgement: %v", err)
		case <-time.After(20 * time.Millisecond):
		}
	}
	cancel()
	if err := awaitConsumerStop(t, done); err != nil {
		t.Fatal(err)
	}
	stats := consumer.Stats()
	if stats.Reconnects < 4 || stats.Acknowledged != 1 || calls.Load() != 1 {
		t.Fatalf("cross-node reconnect storm duplicated work: stats=%+v calls=%d", stats, calls.Load())
	}
	page, err := apiB.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(page.Items) != 1 || page.Items[0].Attempts != 1 {
		t.Fatalf("durable delivery attempt changed during storm: err=%v deliveries=%+v", err, page.Items)
	}
	t.Logf("cross-node reconnect storm completed with %d reconnects, one handler call, and one durable delivery attempt", stats.Reconnects)
}