package orbit

import (
	"context"
	"encoding/json"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/coder/websocket"
)

// 该测试用原始WSS帧驱动交付，用于验证管理SDK；不是待交付的订阅连接SDK替代品。
type managementEventFrame struct {
	Version   string          `json:"protocol_version"`
	Type      string          `json:"type"`
	MessageID string          `json:"message_id"`
	SessionID string          `json:"session_id,omitempty"`
	ReplyTo   string          `json:"reply_to,omitempty"`
	Body      json.RawMessage `json:"body,omitempty"`
	Error     *struct {
		Code string `json:"code"`
	} `json:"error,omitempty"`
}

func TestIntegrationSubscriptionManagementSDK(t *testing.T) {
	api, _, _ := integrationWorkerClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
	defer stop()
	queue, err := api.CreateQueue(ctx, workerTestID(t), CreateQueueRequest{Name: "subscription-" + workerTestID(t), MaxConcurrency: 1, RatePerSecond: 10})
	if err != nil {
		t.Fatal(err)
	}
	key := workerTestID(t)
	in := CreateSubscriptionRequest{Name: "results-" + workerTestID(t), QueueID: queue.Queue.ID, MaxPending: 1, MaxAttempts: 1, Types: []string{"task.cancelled"}}
	created, err := api.CreateSubscription(ctx, key, in)
	if err != nil {
		t.Fatal(err)
	}
	duplicate, err := api.CreateSubscription(ctx, key, in)
	if err != nil || !duplicate.Deduplicated || duplicate.Subscription.ID != created.Subscription.ID {
		t.Fatal("SDK subscription create replay", err)
	}
	sub := created.Subscription
	current, err := api.GetSubscription(ctx, sub.ID)
	if err != nil || current.ID != sub.ID || !current.Enabled {
		t.Fatal("SDK subscription get", err)
	}
	page, err := api.ListSubscriptions(ctx, queue.Queue.ID, SubscriptionPageRequest{Limit: 1})
	if err != nil || len(page.Items) != 1 || page.Items[0].ID != sub.ID {
		t.Fatal("SDK subscription list", err)
	}
	paused, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: sub.Revision, Reason: "maintenance", Enabled: false})
	if err != nil || paused.Subscription.Enabled {
		t.Fatal("SDK pause", err)
	}
	resumed, err := api.ConfigureSubscription(ctx, sub.ID, workerTestID(t), ConfigureSubscriptionRequest{ExpectedRevision: paused.Subscription.Revision, Reason: "ready", Enabled: true})
	if err != nil || !resumed.Subscription.Enabled {
		t.Fatal("SDK resume", err)
	}
	task, err := api.SubmitTask(ctx, workerTestID(t), SubmitTaskRequest{QueueID: queue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
	if err != nil {
		t.Fatal(err)
	}
	if _, err = api.CancelTask(ctx, task.Task.ID, workerTestID(t), CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "create result"}); err != nil {
		t.Fatal(err)
	}
	send := func(conn *websocket.Conn, session, kind string, body any) string {
		t.Helper()
		id := workerTestID(t)
		raw, err := marshalWire(body)
		if err != nil {
			t.Fatal(err)
		}
		frame, err := marshalWire(managementEventFrame{Version: "1", Type: kind, MessageID: id, SessionID: session, Body: raw})
		if err != nil {
			t.Fatal(err)
		}
		if err = conn.Write(ctx, websocket.MessageText, frame); err != nil {
			t.Fatal(err)
		}
		return id
	}
	read := func(conn *websocket.Conn) managementEventFrame {
		t.Helper()
		bounded, cancel := context.WithTimeout(ctx, 5*time.Second)
		defer cancel()
		_, raw, err := conn.Read(bounded)
		if err != nil {
			t.Fatal(err)
		}
		var frame managementEventFrame
		if json.Unmarshal(raw, &frame) != nil || frame.Version != "1" || !validID(frame.MessageID) {
			t.Fatal("invalid management test frame")
		}
		return frame
	}
	connect := func() (*websocket.Conn, string) {
		t.Helper()
		headers := http.Header{}
		headers.Set("Authorization", "Bearer "+api.token)
		endpoint := "ws" + strings.TrimPrefix(api.baseURL, "http") + "/v1/events/connect"
		conn, _, err := websocket.Dial(ctx, endpoint, &websocket.DialOptions{HTTPClient: api.http, HTTPHeader: headers, Subprotocols: []string{"orbit.events.v1"}})
		if err != nil {
			t.Fatal("management test event dial failed")
		}
		t.Cleanup(func() { conn.CloseNow() })
		conn.SetReadLimit(512 << 10)
		msg := send(conn, "", "events.hello", map[string]any{"subscription_id": sub.ID, "capacity": 1})
		welcome := read(conn)
		if welcome.Type != "events.welcome" || welcome.ReplyTo != msg || !validID(welcome.SessionID) {
			t.Fatal("management test welcome missing")
		}
		return conn, welcome.SessionID
	}
	type grant struct {
		Delivery EventDelivery `json:"delivery"`
		Event    ResultEvent   `json:"event"`
		Secret   string        `json:"lease_secret"`
	}
	a, _ := connect()
	frame := read(a)
	var first grant
	if frame.Type != "event.deliver" || json.Unmarshal(frame.Body, &first) != nil || first.Event.Snapshot.TaskID != task.Task.ID {
		t.Fatal("management test delivery missing")
	}
	a.CloseNow()
	b, session := connect()
	// 使用实际30秒租约，不改数据库时钟或手工更新租约时间。
	var expired EventDelivery
	for {
		deliveries, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryExpired, Limit: 1})
		if err != nil {
			t.Fatal(err)
		}
		if len(deliveries.Items) == 1 {
			expired = deliveries.Items[0]
			break
		}
		select {
		case <-ctx.Done():
			t.Fatal("delivery did not expire through real WSS recovery")
		case <-time.After(150 * time.Millisecond):
		}
	}
	detail, err := api.GetEventDelivery(ctx, sub.ID, expired.ID)
	if err != nil || detail.Event.ID != first.Event.ID || detail.Delivery.Attempts != 1 || detail.Event.Snapshot.State != TaskCancelled {
		t.Fatal("SDK delivery detail", err)
	}
	retryKey := workerTestID(t)
	retry := RetryDeliveryRequest{ExpectedRevision: expired.Revision, Reason: "receiver fixed", AdditionalAttempts: 1}
	restored, err := api.RetryEventDelivery(ctx, sub.ID, expired.ID, retryKey, retry)
	if err != nil || restored.Delivery.Attempts != 1 || restored.Delivery.AttemptBudget != 2 {
		t.Fatal("SDK retry reset history", err)
	}
	replay, err := api.RetryEventDelivery(ctx, sub.ID, expired.ID, retryKey, retry)
	if err != nil || !replay.Deduplicated || replay.Delivery.Revision != restored.Delivery.Revision {
		t.Fatal("SDK retry replay", err)
	}
	frame = read(b)
	var second grant
	if frame.Type != "event.deliver" || json.Unmarshal(frame.Body, &second) != nil || second.Event.ID != first.Event.ID || second.Delivery.LeaseEpoch != 2 {
		t.Fatal("manual retry failed to deliver")
	}
	ackBody := map[string]any{"delivery_id": second.Delivery.ID, "epoch": "2", "lease_secret": second.Secret}
	msg := send(b, session, "event.ack", ackBody)
	ack := read(b)
	if ack.Type != "ack" || ack.ReplyTo != msg {
		t.Fatal("management test ACK failed")
	}
	delivered, err := api.ListEventDeliveries(ctx, sub.ID, DeliveryPageRequest{State: DeliveryAcked})
	if err != nil || len(delivered.Items) != 1 || delivered.Items[0].ID != expired.ID {
		t.Fatal("SDK acknowledged history", err)
	}
	current, err = api.GetSubscription(ctx, sub.ID)
	if err != nil || current.PendingCount != 0 || current.Revision != resumed.Subscription.Revision {
		t.Fatal("SDK pending count or config version", err)
	}
	retireKey := workerTestID(t)
	retireInput := RetireResourceRequest{ExpectedRevision: current.Revision, Reason: "completed SDK subscription regression"}
	retired, err := api.RetireSubscription(ctx, sub.ID, retireKey, retireInput)
	if err != nil || retired.Subscription.RetiredAt == nil {
		t.Fatal("SDK retire subscription", err)
	}
	repeated, err := api.RetireSubscription(ctx, sub.ID, retireKey, retireInput)
	if err != nil || !repeated.Deduplicated {
		t.Fatal("SDK retire replay", err)
	}
	if _, err = api.GetEventDelivery(ctx, sub.ID, expired.ID); err != nil {
		t.Fatal("retired delivery history", err)
	}
	queueRetired, err := api.RetireQueue(ctx, queue.Queue.ID, workerTestID(t), RetireResourceRequest{ExpectedRevision: queue.Queue.Revision, Reason: "completed SDK queue regression"})
	if err != nil || queueRetired.Queue.RetiredAt == nil {
		t.Fatal("SDK retire queue", err)
	}
	t.Log("management SDK exercised create/get/list/control/detail/expired retry with a real 30-second delivery lease")
}