package orbit

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

// A04：数据库已提交派发，但真实WS帧被网关截留，业务Worker从未收到该尝试。
func TestIntegrationWorkerProcessUndeliveredAssignment(t *testing.T) {
	a := startOrbitProcess(t)
	_, _, token := integrationWorkerClient(t)
	api := processClient(t, a, token)
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Second)
	defer cancel()
	name := "dispatch.crash." + workerTestID(t)
	queue, err := api.CreateQueue(ctx, name, CreateQueueRequest{Name: name, MaxConcurrency: 1, RatePerSecond: 20, MaxUnfinishedTasks: 2})
	if err != nil {
		t.Fatal(err)
	}
	in := SubmitTaskRequest{QueueID: queue.Queue.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`), TaskTimeoutMS: 45000, Retry: RetryPolicy{MaxAttempts: 2, AttemptTimeoutMS: 5000, InitialDelayMS: 100, MaxDelayMS: 100}, RecoverySafety: RecoveryReconcileOnly}
	key := workerTestID(t)
	accepted, err := api.SubmitTask(ctx, key, in)
	if err != nil {
		t.Fatal(err)
	}
	g := newWorkerFaultGateway(t, a.url)
	g.holdAssignment.Store(true)
	client, err := NewClient(g.server.URL, Options{Token: token, HTTPClient: g.server.Client()})
	if err != nil {
		t.Fatal(err)
	}
	var calls atomic.Int32
	w, err := NewWorker(client, WorkerOptions{QueueID: queue.Queue.ID, Capacity: 1, ReconnectMaxDelay: time.Second, Handlers: []WorkerHandler{runtimeHandler(name, func(_ context.Context, e WorkerExecution) (json.RawMessage, error) {
		calls.Add(1)
		return e.Task.Payload, nil
	})}})
	if err != nil {
		t.Fatal(err)
	}
	run, stop := context.WithCancel(ctx)
	defer stop()
	done := make(chan error, 1)
	go func() { done <- w.Run(run) }()
	var held WorkerAssignment
	select {
	case held = <-g.assignmentsHeld:
	case err := <-done:
		t.Fatal("Worker stopped before assignment fault", err)
	case <-ctx.Done():
		t.Fatal("assignment fault was not reached")
	}
	if held.Task.ID != accepted.Task.ID || calls.Load() != 0 || w.Stats().Started != 0 {
		t.Fatal("fault was not before Worker execution")
	}
	leased, err := api.GetTask(ctx, held.Task.ID)
	if err != nil || leased.State != TaskLeased || leased.AttemptCount != 1 {
		t.Fatal("dispatch was not durably committed", err)
	}
	first, err := api.GetTaskAttempt(ctx, held.Task.ID, held.AttemptID)
	if err != nil || first.State != AttemptLeased || first.AuthorizedAt != nil || first.EndedAt != nil {
		t.Fatal("intercepted assignment already authorized", err)
	}
	a.kill(t)
	close(g.releaseAssignment)
	b := startOrbitProcess(t)
	if a.nodeID == b.nodeID {
		t.Fatal("node identity reused")
	}
	api = processClient(t, b, token)
	receipt, err := api.SubmitTask(ctx, key, in)
	if err != nil || !receipt.Deduplicated || receipt.Task.ID != accepted.Task.ID || !receipt.Task.DeadlineAt.Equal(accepted.Task.DeadlineAt) {
		t.Fatal("dispatch crash lost admission receipt", err)
	}
	// 保持Worker离线，等真实原租约自然过期；不使用SQL改时钟或伪造恢复结果。
	recovered := waitRuntimeState(t, ctx, api, held.Task.ID, TaskWaiting)
	old, err := api.GetTaskAttempt(ctx, held.Task.ID, held.AttemptID)
	if err != nil || old.State != AttemptExpired || old.AuthorizedAt != nil || old.EndedAt == nil || old.EndedAt.Before(first.LeaseUntil) || recovered.EffectUncertain || recovered.AttemptCount != 1 || calls.Load() != 0 {
		t.Fatal("unreceived assignment recovery fabricated an execution", err)
	}
	g.target.Store(b.url)
	g.offline.Store(false)
	finished := waitRuntimeState(t, ctx, api, held.Task.ID, TaskSucceeded)
	for w.Stats().ReportsConfirmed != 1 {
		select {
		case err := <-done:
			t.Fatal(err)
		case <-ctx.Done():
			t.Fatal("result receipt not recovered")
		case <-time.After(20 * time.Millisecond):
		}
	}
	stop()
	select {
	case err := <-done:
		if err != nil {
			t.Fatal(err)
		}
	case <-ctx.Done():
		t.Fatal("Worker did not drain")
	}
	history, err := api.ListTaskAttempts(ctx, held.Task.ID, HistoryPageRequest{})
	if err != nil || len(history.Items) != 2 || history.NextCursor != "" || history.Items[1].ID != held.AttemptID || history.Items[1].AuthorizedAt != nil || history.Items[1].State != AttemptExpired || history.Items[0].State != AttemptSucceeded || history.Items[0].Epoch != held.Epoch+1 {
		t.Fatal("dispatch fault duplicated or replaced history", err)
	}
	if calls.Load() != 1 || w.Stats().Started != 1 || finished.AttemptCount != 2 || finished.EffectUncertain || !finished.DeadlineAt.Equal(accepted.Task.DeadlineAt) || finished.Retry != accepted.Task.Retry || string(finished.Result) != string(in.Payload) {
		t.Fatal("dispatch recovery reset budget or executed twice")
	}
	resources, err := api.GetQueue(ctx, queue.Queue.ID)
	if err != nil || resources.UnfinishedTasks != 0 {
		t.Fatal("dispatch recovery leaked unfinished task quota", err)
	}
	t.Log("A04: committed LEASED frame never forwarded to Worker; original node SIGKILL confirmed; natural lease expiry preserved unauthorized history; new node completed original task with exactly one handler call and two budgeted attempts")
}