package orbit

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

func TestIntegrationQueueResourcesSDK(t *testing.T) {
	api, _, _ := integrationWorkerClient(t)
	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
	defer cancel()
	name := "resources." + workerTestID(t)
	created, err := api.CreateQueue(ctx, "create-"+name, CreateQueueRequest{Name: name, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2})
	if err != nil {
		t.Fatal(err)
	}
	q := created.Queue
	configure := func(paused bool) {
		t.Helper()
		out, err := api.ConfigureQueue(ctx, q.ID, workerTestID(t), ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "SDK resource regression", Paused: paused, MaxConcurrency: 2, RatePerSecond: 1, MaxUnfinishedTasks: 2})
		if err != nil {
			t.Fatal(err)
		}
		q = out.Queue
	}
	configure(true)
	// 丢失 paused 不能被当成解除暂停；这是原始 HTTP 对象，SDK 本身总会发送 bool。
	var ignored CreateQueueResult
	err = api.call(ctx, "POST", "/v1/queues/"+q.ID+":configure", workerTestID(t), map[string]any{"expected_revision": "2", "reason": "missing pause", "max_concurrency": 2, "rate_per_second": 1, "max_unfinished_tasks": 2}, &ignored)
	var apiErr *APIError
	if !errors.As(err, &apiErr) || apiErr.StatusCode != 400 {
		t.Fatalf("missing paused accepted: %v", err)
	}
	input := SubmitTaskRequest{QueueID: q.ID, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}
	first, err := api.SubmitTask(ctx, name+"-one", input)
	if err != nil {
		t.Fatal(err)
	}
	second, err := api.SubmitTask(ctx, name+"-two", input)
	if err != nil {
		t.Fatal(err)
	}
	_, err = api.SubmitTask(ctx, name+"-three", input)
	if !errors.As(err, &apiErr) || apiErr.StatusCode != 429 || apiErr.Code != "QUEUE_FULL" || apiErr.RetryAfter != time.Second {
		t.Fatalf("queue did not backpressure: %v", err)
	}
	if duplicate, err := api.SubmitTask(ctx, name+"-one", input); err != nil || !duplicate.Deduplicated || duplicate.Task.ID != first.Task.ID {
		t.Fatal("full queue lost original identity")
	}
	if _, err := api.CancelTask(ctx, first.Task.ID, name+"-cancel", CancelTaskRequest{ExpectedRevision: first.Task.Revision, Reason: "free backlog slot"}); err != nil {
		t.Fatal(err)
	}
	third, err := api.SubmitTask(ctx, name+"-three", input)
	if err != nil {
		t.Fatal(err)
	}
	w, err := NewWorker(api, WorkerOptions{QueueID: q.ID, Capacity: 2, Handlers: []WorkerHandler{runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) { return e.Task.Payload, nil })}})
	if err != nil {
		t.Fatal(err)
	}
	run, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- w.Run(run) }()
	waited := false
	defer func() {
		stop()
		if !waited {
			<-done
		}
	}()
	select {
	case <-time.After(350 * time.Millisecond):
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	}
	current, err := api.GetTask(ctx, second.Task.ID)
	if err != nil || current.AttemptCount != 0 {
		t.Fatal("paused queue consumed attempt")
	}
	configure(false)
	for _, id := range []string{second.Task.ID, third.Task.ID} {
		finished := waitRuntimeState(t, ctx, api, id, TaskSucceeded)
		if finished.AttemptCount != 1 || string(finished.Result) != `{"n":9007199254740993}` {
			t.Fatal("resource controls altered execution")
		}
	}
	var authorized []time.Time
	for _, id := range []string{second.Task.ID, third.Task.ID} {
		attempts, err := api.ListTaskAttempts(ctx, id, HistoryPageRequest{})
		if err != nil || len(attempts.Items) != 1 || attempts.Items[0].AuthorizedAt == nil {
			t.Fatal("missing start history")
		}
		authorized = append(authorized, *attempts.Items[0].AuthorizedAt)
	}
	gap := authorized[1].Sub(authorized[0])
	if gap < 0 {
		gap = -gap
	}
	if gap < time.Second {
		t.Fatalf("start rate exceeded: %s", gap)
	}
	q, err = api.GetQueue(ctx, q.ID)
	if err != nil || q.UnfinishedTasks != 0 || q.Revision != 3 {
		t.Fatal("quota count or configuration revision drifted")
	}
	stop()
	err = <-done
	waited = true
	if err != nil {
		t.Fatal(err)
	}
}