package orbit

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"net/url"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

func TestIntegrationTaskBatchSDK(t *testing.T) {
	api, _, _ := integrationWorkerClient(t)
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	created, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "batch." + workerTestID(t), MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2})
	if err != nil {
		t.Fatal(err)
	}
	q := created.Queue
	if _, err := api.ConfigureQueue(ctx, q.ID, workerTestID(t), ConfigureQueueRequest{ExpectedRevision: q.Revision, Reason: "batch acceptance regression", Paused: true, MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 2}); err != nil {
		t.Fatal(err)
	}
	input := SubmitTaskRequest{QueueID: q.ID, Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}
	in := SubmitTasksRequest{Items: []SubmitTasksItem{{IdempotencyKey: workerTestID(t), Task: input}, {IdempotencyKey: workerTestID(t), Task: input}, {IdempotencyKey: workerTestID(t), Task: input}}}
	// 真实HTTP代理先完整读到数据库受理回执，再强制断开客户端连接，丢弃整批响应。
	target, _ := url.Parse(api.baseURL)
	proxy := httputil.NewSingleHostReverseProxy(target)
	var dropped atomic.Bool
	var calls atomic.Int32
	proxy.ModifyResponse = func(r *http.Response) error {
		if r.Request.URL.Path != "/v1/tasks:batch" {
			return nil
		}
		calls.Add(1)
		if dropped.CompareAndSwap(false, true) {
			raw, err := io.ReadAll(r.Body)
			r.Body.Close()
			if err != nil {
				return err
			}
			if r.StatusCode != 200 || !strings.Contains(string(raw), `"ACCEPTED"`) || !strings.Contains(string(raw), `"QUEUE_FULL"`) {
				t.Error("unexpected pre-drop server result", string(raw))
			}
			return io.ErrUnexpectedEOF
		}
		return nil
	}
	proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
		conn, _, hijackErr := w.(http.Hijacker).Hijack()
		if hijackErr == nil {
			conn.Close()
		}
	}
	server := httptest.NewServer(proxy)
	defer server.Close()
	client, err := NewClient(server.URL, Options{Token: api.token})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := client.SubmitTasks(ctx, in); err == nil || !dropped.Load() || calls.Load() != 1 {
		t.Fatal("response loss not surfaced without silent retry", err)
	}
	// 原连接已关闭；用原键和原内容核对，前两项去重，满额项保持明确拒绝。
	out, err := client.SubmitTasks(ctx, in)
	if err != nil || len(out.Items) != 3 {
		t.Fatal(out, err)
	}
	for i := 0; i < 2; i++ {
		if out.Items[i].Status != BatchAccepted || !out.Items[i].Receipt.Deduplicated {
			t.Fatal("lost response duplicated task", out)
		}
	}
	if out.Items[2].Status != BatchRejected || out.Items[2].Error.Code != "QUEUE_FULL" || out.Items[2].Error.RetryAfterMS != 1000 {
		t.Fatal(out)
	}
	page, err := api.ListTasks(ctx, ListTasksRequest{QueueID: q.ID})
	if err != nil || len(page.Items) != 2 {
		t.Fatal("duplicate durable rows", page, err)
	}
	single, err := api.SubmitTask(ctx, in.Items[0].IdempotencyKey, input)
	if err != nil || !single.Deduplicated || single.Task.ID != out.Items[0].Receipt.TaskID || string(single.Task.Payload) != `{"n":9007199254740993}` {
		t.Fatal("batch/single mismatch", err)
	}
	if _, err := api.CancelTask(ctx, single.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: single.Task.Revision, Reason: "free slot for partial retry"}); err != nil {
		t.Fatal(err)
	}
	suffix, err := api.SubmitTasks(ctx, SubmitTasksRequest{Items: in.Items[2:]})
	if err != nil || suffix.Items[0].Status != BatchAccepted || suffix.Items[0].Receipt.Deduplicated || suffix.Items[0].Index != 0 {
		t.Fatal(suffix, err)
	}
	for _, id := range []string{out.Items[1].Receipt.TaskID, suffix.Items[0].Receipt.TaskID} {
		current, err := api.GetTask(ctx, id)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := api.CancelTask(ctx, id, workerTestID(t), CancelTaskRequest{ExpectedRevision: current.Revision, Reason: "finish batch regression"}); err != nil {
			t.Fatal(err)
		}
	}
	t.Log("lost real HTTP response: two committed tasks recovered by identity, no duplicates; backpressured suffix independently accepted after capacity released")
}