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