package mysql

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
)

type apiReply struct {
	status int
	body   []byte
	data   json.RawMessage
	code   string
}

func TestIntegrationTaskAPI(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("requires disposable MySQL")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	db, err := Open(ctx, dsn, 24, 4)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err := Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	id := func() core.ID {
		t.Helper()
		id, err := core.NewID()
		if err != nil {
			t.Fatal(err)
		}
		return id
	}
	bootstrap := func() BootstrapResult {
		t.Helper()
		var out BootstrapResult
		err := Bootstrap(ctx, db, BootstrapInput{Application: "app-" + string(id()), Namespace: "default", Environment: "test", Queue: "files"}, func(v BootstrapResult) error { out = v; return nil })
		if err != nil {
			t.Fatal(err)
		}
		return out
	}
	a, b := bootstrap(), bootstrap()
	svc := service.New(NewTaskStore(db))
	server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, svc).Handler)
	defer server.Close()
	call := func(method, path, token, key string, body []byte) (apiReply, error) {
		r, err := http.NewRequestWithContext(ctx, method, server.URL+path, bytes.NewReader(body))
		if err != nil {
			return apiReply{}, err
		}
		r.Header.Set("Content-Type", "application/json")
		if token != "" {
			r.Header.Set("Authorization", "Bearer "+token)
		}
		if key != "" {
			r.Header.Set("Idempotency-Key", key)
		}
		response, err := server.Client().Do(r)
		if err != nil {
			return apiReply{}, err
		}
		defer response.Body.Close()
		raw, err := io.ReadAll(response.Body)
		if err != nil {
			return apiReply{}, err
		}
		var envelope struct {
			RequestID string          `json:"request_id"`
			Data      json.RawMessage `json:"data"`
			Error     struct {
				Code string `json:"code"`
			} `json:"error"`
		}
		if err := json.Unmarshal(raw, &envelope); err != nil {
			return apiReply{}, fmt.Errorf("invalid response JSON: %s", raw)
		}
		if core.ID(envelope.RequestID).Validate() != nil || response.Header.Get("X-Request-ID") != envelope.RequestID {
			return apiReply{}, errors.New("missing request correlation")
		}
		return apiReply{response.StatusCode, raw, envelope.Data, envelope.Error.Code}, nil
	}
	must := func(method, path, token, key string, body []byte, status int) apiReply {
		t.Helper()
		out, err := call(method, path, token, key, body)
		if err != nil {
			t.Fatal(err)
		}
		if out.status != status {
			t.Fatalf("%s %s: wanted %d got %d: %s", method, path, status, out.status, out.body)
		}
		return out
	}
	marshal := func(v any) []byte {
		t.Helper()
		raw, err := json.Marshal(v)
		if err != nil {
			t.Fatal(err)
		}
		return raw
	}
	input := service.SubmitInput{QueueID: a.QueueID, Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"sequence":9007199254740993,"decimal":12345678901234567890.1234567890123456789,"one":1}`)}
	raw := marshal(input)
	var accepted service.SubmitResult
	out := must("POST", "/v1/tasks", a.Token, "stable-submit", raw, 201)
	if err := json.Unmarshal(out.data, &accepted); err != nil {
		t.Fatal(err)
	}
	if !bytes.Contains(out.body, []byte("12345678901234567890.1234567890123456789")) || !bytes.Contains(out.body, []byte("9007199254740993")) {
		t.Fatal("payload precision lost")
	}
	taskID := accepted.Task.ID
	count := func(table, where string, args ...any) int {
		t.Helper()
		var n int
		if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table+" WHERE "+where, args...).Scan(&n); err != nil {
			t.Fatal(err)
		}
		return n
	}
	execute := func(q string, args ...any) {
		t.Helper()
		if _, err := db.ExecContext(ctx, q, args...); err != nil {
			t.Fatal(err)
		}
	}
	t.Run("concurrent_duplicate_and_semantic_request_identity", func(t *testing.T) {
		// 模拟首次响应未被业务确认：重复原始请求，必须读到相同提交回执。
		again := must("POST", "/v1/tasks", a.Token, "stable-submit", raw, 200)
		var dupe service.SubmitResult
		if err := json.Unmarshal(again.data, &dupe); err != nil || dupe.Task.ID != taskID || !dupe.Deduplicated {
			t.Fatalf("duplicate: %s %v", again.body, err)
		}
		equivalent := input
		equivalent.Payload = json.RawMessage(`{"one":1.00,"decimal":12345678901234567890.123456789012345678900,"sequence":9007199254740993}`)
		must("POST", "/v1/tasks", a.Token, "stable-submit", marshal(equivalent), 200)
		equivalent.Payload = json.RawMessage(`{"one":2}`)
		reply := must("POST", "/v1/tasks", a.Token, "stable-submit", marshal(equivalent), 409)
		if reply.code != "IDEMPOTENCY_CONFLICT" {
			t.Fatal(reply.code)
		}
		const n = 16
		start := make(chan struct{})
		results := make(chan apiReply, n)
		errs := make(chan error, n)
		var wg sync.WaitGroup
		for i := 0; i < n; i++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				<-start
				out, err := call("POST", "/v1/tasks", a.Token, "concurrent-submit", raw)
				results <- out
				errs <- err
			}()
		}
		close(start)
		wg.Wait()
		close(results)
		close(errs)
		for err := range errs {
			if err != nil {
				t.Fatal(err)
			}
		}
		var shared core.ID
		created := 0
		for out := range results {
			if out.status != 200 && out.status != 201 {
				t.Fatalf("concurrent response %d: %s", out.status, out.body)
			}
			if out.status == 201 {
				created++
			}
			var v service.SubmitResult
			if err := json.Unmarshal(out.data, &v); err != nil {
				t.Fatal(err)
			}
			if shared == "" {
				shared = v.Task.ID
			}
			if v.Task.ID != shared {
				t.Fatal("duplicate task created")
			}
		}
		if created != 1 || count("orbit_events", "namespace_id=? AND aggregate_id=?", a.NamespaceID, shared) != 1 || count("orbit_audit_events", "namespace_id=? AND resource_id=? AND action='task.submit'", a.NamespaceID, shared) != 1 {
			t.Fatalf("creation/event count: %d", created)
		}
	})
	t.Run("authorization_scope_and_revocation", func(t *testing.T) {
		must("GET", "/v1/tasks/"+string(taskID), "", "", nil, 401)
		wrong := a.Token[:len(a.Token)-43] + strings.Repeat("A", 43)
		must("GET", "/v1/tasks/"+string(taskID), wrong, "", nil, 401)
		must("GET", "/v1/tasks/"+string(taskID), b.Token, "", nil, 404)
		must("POST", "/v1/tasks", b.Token, "cross-submit", raw, 404)
		credential := id()
		secret, digest, err := auth.Issue(a.NamespaceID, credential)
		if err != nil {
			t.Fatal(err)
		}
		permissions := auth.Permissions{Actions: []string{"tasks:read"}, Queues: []string{string(a.QueueID)}}
		execute("INSERT INTO orbit_credentials(namespace_id,id,name,secret_digest,permissions) VALUES (?,?,?,?,?)", a.NamespaceID, credential, string(credential), digest, string(marshal(permissions)))
		must("GET", "/v1/tasks/"+string(taskID), secret, "", nil, 200)
		must("POST", "/v1/tasks", secret, "read-only", raw, 403)
		execute("UPDATE orbit_credentials SET expires_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?", a.NamespaceID, credential)
		must("GET", "/v1/tasks/"+string(taskID), secret, "", nil, 401)
		execute("UPDATE orbit_credentials SET expires_at=NULL,revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?", a.NamespaceID, credential)
		must("GET", "/v1/tasks/"+string(taskID), secret, "", nil, 401)
		execute("UPDATE orbit_credentials SET revoked_at=NULL,permissions=? WHERE namespace_id=? AND id=?", string(marshal(auth.Permissions{Actions: []string{"tasks:read"}, Queues: []string{string(b.QueueID)}})), a.NamespaceID, credential)
		must("GET", "/v1/tasks/"+string(taskID), secret, "", nil, 404)
		execute("UPDATE orbit_namespaces SET enabled=FALSE WHERE id=?", b.NamespaceID)
		must("GET", "/v1/tasks/"+string(taskID), b.Token, "", nil, 401)
		execute("UPDATE orbit_namespaces SET enabled=TRUE WHERE id=?", b.NamespaceID)
		execute("UPDATE orbit_applications SET enabled=FALSE WHERE id=?", b.ApplicationID)
		must("GET", "/v1/tasks/"+string(taskID), b.Token, "", nil, 401)
		execute("UPDATE orbit_applications SET enabled=TRUE WHERE id=?", b.ApplicationID)
	})
	t.Run("strict_input_and_failed_submission_rollback", func(t *testing.T) {
		for _, bad := range []string{`null`, `{"namespace_id":"override"}`, `{"handler":"one","handler":"two"}`, `{"Handler":"case-alias"}`, `{"payload":"\ud800"}`, `{"retry":{"max_attempts":null}}`} {
			must("POST", "/v1/tasks", a.Token, "invalid-body", []byte(bad), 400)
		}
		must("POST", "/v1/tasks", a.Token, "oversize", []byte(`{"payload":"`+strings.Repeat("x", 512<<10)+`"}`), 413)
		must("POST", "/v1/tasks", a.Token, "", raw, 400)
		invalid := input
		past := time.Now().UTC().Add(-time.Hour).Truncate(time.Microsecond)
		invalid.AvailableAt = &past
		must("POST", "/v1/tasks", a.Token, "past", marshal(invalid), 400)
		hash := core.Digest(sha256.Sum256([]byte("past")))
		if count("orbit_idempotency_keys", "namespace_id=? AND key_digest=?", a.NamespaceID, hash) != 0 {
			t.Fatal("failed submission committed a reservation")
		}
		must("POST", "/v1/tasks", a.Token, "past", raw, 201)
	})
	t.Run("keyset_pagination_and_opaque_payload", func(t *testing.T) {
		seen := map[core.ID]bool{}
		path := "/v1/tasks?queue_id=" + string(a.QueueID) + "&limit=1"
		for i := 0; i < 10; i++ {
			reply := must("GET", path, a.Token, "", nil, 200)
			var page service.TaskPage
			if err := json.Unmarshal(reply.data, &page); err != nil {
				t.Fatal(err)
			}
			if bytes.Contains(reply.data, []byte(`"payload"`)) {
				t.Fatal("list included payload")
			}
			for _, item := range page.Items {
				if seen[item.ID] {
					t.Fatal("repeated list item")
				}
				seen[item.ID] = true
			}
			if page.NextCursor == "" {
				break
			}
			path = "/v1/tasks?queue_id=" + string(a.QueueID) + "&limit=1&cursor=" + url.QueryEscape(page.NextCursor)
			must("GET", path+"&state=CANCELLED", a.Token, "", nil, 400)
		}
		if len(seen) != 3 {
			t.Fatalf("list count %d", len(seen))
		}
		reply := must("GET", "/v1/tasks/"+string(taskID), a.Token, "", nil, 200)
		if !bytes.Contains(reply.data, []byte("12345678901234567890.1234567890123456789")) {
			t.Fatal("get changed number")
		}
		must("GET", "/v1/tasks?queue_id="+string(a.QueueID)+"&limit=101", a.Token, "", nil, 400)
	})
	t.Run("cancel_transaction_rolls_back_when_audit_fails", func(t *testing.T) {
		execute("CREATE TRIGGER orbit_test_reject_audit BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='integration injected audit failure'")
		defer execute("DROP TRIGGER orbit_test_reject_audit")
		in := marshal(service.CancelInput{Reason: "test rollback", ExpectedRevision: 1})
		reply := must("POST", "/v1/tasks/"+string(taskID)+":cancel", a.Token, "cancel-rollback", in, 503)
		if strings.Contains(string(reply.body), "injected") {
			t.Fatal("leaked internal failure")
		}
		var task service.TaskView
		reply = must("GET", "/v1/tasks/"+string(taskID), a.Token, "", nil, 200)
		if err := json.Unmarshal(reply.data, &task); err != nil || task.State != pulse.TaskWaiting || task.Revision != 1 {
			t.Fatalf("rollback state: %+v %v", task, err)
		}
		if count("orbit_events", "namespace_id=? AND aggregate_id=?", a.NamespaceID, taskID) != 1 {
			t.Fatal("event escaped rollback")
		}
		hash := core.Digest(sha256.Sum256([]byte("cancel-rollback")))
		if count("orbit_idempotency_keys", "namespace_id=? AND key_digest=?", a.NamespaceID, hash) != 0 {
			t.Fatal("reservation escaped rollback")
		}
	})
	t.Run("cancel_compare_and_swap_and_duplicate_receipt", func(t *testing.T) {
		in := marshal(service.CancelInput{Reason: "finish task", ExpectedRevision: 1})
		replies := make(chan apiReply, 2)
		errs := make(chan error, 2)
		for _, key := range []string{"cancel-a", "cancel-b"} {
			go func() {
				out, err := call("POST", "/v1/tasks/"+string(taskID)+":cancel", a.Token, key, in)
				replies <- out
				errs <- err
			}()
		}
		successes, conflicts := 0, 0
		for i := 0; i < 2; i++ {
			if err := <-errs; err != nil {
				t.Fatal(err)
			}
			out := <-replies
			switch out.status {
			case 200:
				successes++
			case 409:
				conflicts++
			default:
				t.Fatalf("cancel: %d %s", out.status, out.body)
			}
		}
		if successes != 1 || conflicts != 1 {
			t.Fatalf("cancel winners=%d conflicts=%d", successes, conflicts)
		}
		duplicate := 0
		for _, key := range []string{"cancel-a", "cancel-b"} {
			out, err := call("POST", "/v1/tasks/"+string(taskID)+":cancel", a.Token, key, in)
			if err != nil {
				t.Fatal(err)
			}
			if out.status == 200 {
				var v service.CancelResult
				if err := json.Unmarshal(out.data, &v); err != nil || !v.Deduplicated || v.Task.State != pulse.TaskCancelled || v.Task.Revision != 2 {
					t.Fatalf("receipt: %s %v", out.body, err)
				}
				duplicate++
			}
		}
		if duplicate != 1 || count("orbit_audit_events", "namespace_id=? AND resource_id=? AND action='task.cancel'", a.NamespaceID, taskID) != 1 || count("orbit_events", "namespace_id=? AND aggregate_id=?", a.NamespaceID, taskID) != 2 {
			t.Fatal("cancel duplicated effects")
		}
		reply := must("POST", "/v1/tasks", a.Token, "stable-submit", raw, 200)
		var original service.SubmitResult
		if err := json.Unmarshal(reply.data, &original); err != nil || original.Task.State != pulse.TaskWaiting {
			t.Fatal("original receipt should remain immutable")
		}
	})
	t.Run("queue_management_and_bootstrap_rollback", func(t *testing.T) {
		body := marshal(service.QueueInput{Name: "second", MaxConcurrency: 8, RatePerSecond: 100})
		reply := must("POST", "/v1/queues", a.Token, "queue-second", body, 201)
		var v service.QueueResult
		if err := json.Unmarshal(reply.data, &v); err != nil {
			t.Fatal(err)
		}
		must("POST", "/v1/queues", a.Token, "queue-second", body, 200)
		must("POST", "/v1/queues", a.Token, "queue-name-conflict", body, 409)
		must("GET", "/v1/queues/"+string(v.Queue.ID), a.Token, "", nil, 200)
		must("GET", "/v1/queues/"+string(v.Queue.ID), b.Token, "", nil, 404)
		name := "failed-" + string(id())
		sentinel := errors.New("failed private output")
		err := Bootstrap(ctx, db, BootstrapInput{Application: name, Namespace: "n", Environment: "test", Queue: "q"}, func(BootstrapResult) error { return sentinel })
		if !errors.Is(err, sentinel) {
			t.Fatal(err)
		}
		if count("orbit_applications", "name=?", name) != 0 {
			t.Fatal("bootstrap escaped rollback")
		}
	})
	t.Run("cancelled_context_cannot_commit", func(t *testing.T) {
		token, err := auth.Parse(a.Token)
		if err != nil {
			t.Fatal(err)
		}
		cancelled, stop := context.WithCancel(ctx)
		stop()
		_, err = svc.Submit(cancelled, token, "cancelled-context", input)
		if !errors.Is(err, context.Canceled) {
			t.Fatal(err)
		}
	})
	t.Run("relative_delay_uses_database_acceptance_and_original_identity", func(t *testing.T) {
		delayed := input
		delayed.DelayMS = 5000
		key := "delayed-submit"
		first := must("POST", "/v1/tasks", a.Token, key, marshal(delayed), 201)
		var v service.SubmitResult
		if err := json.Unmarshal(first.data, &v); err != nil || !v.Task.AvailableAt.Equal(v.Task.CreatedAt.Add(5*time.Second)) {
			t.Fatal("delay not anchored to persisted database acceptance", err)
		}
		const n = 8
		delayedBody := marshal(delayed)
		results := make(chan apiReply, n)
		errs := make(chan error, n)
		for range n {
			go func() {
				out, err := call("POST", "/v1/tasks", a.Token, key, delayedBody)
				results <- out
				errs <- err
			}()
		}
		for range n {
			out, err := <-results, <-errs
			var duplicate service.SubmitResult
			if err != nil || out.status != 200 || json.Unmarshal(out.data, &duplicate) != nil || !duplicate.Deduplicated || duplicate.Task.ID != v.Task.ID || !duplicate.Task.AvailableAt.Equal(v.Task.AvailableAt) || !duplicate.Task.DeadlineAt.Equal(v.Task.DeadlineAt) {
				t.Fatal("repeated delay reset execution time or identity", err)
			}
		}
		changed := delayed
		changed.DelayMS++
		must("POST", "/v1/tasks", a.Token, key, marshal(changed), 409)
		changed = delayed
		changed.DelayMS = 0
		changed.AvailableAt = &v.Task.AvailableAt
		must("POST", "/v1/tasks", a.Token, key, marshal(changed), 409)
		changed.DelayMS = delayed.DelayMS
		must("POST", "/v1/tasks", a.Token, "both-timing-modes", marshal(changed), 400)
		changed = delayed
		tooEarly := v.Task.CreatedAt.Add(time.Second)
		changed.StartDeadlineAt = &tooEarly
		must("POST", "/v1/tasks", a.Token, "delay-after-start-deadline", marshal(changed), 400)
		changed = delayed
		changed.DelayMS = 86400000
		must("POST", "/v1/tasks", a.Token, "delay-at-total-deadline", marshal(changed), 400)
		if count("pulse_tasks", "namespace_id=? AND id=?", a.NamespaceID, v.Task.ID) != 1 || count("orbit_events", "namespace_id=? AND aggregate_id=?", a.NamespaceID, v.Task.ID) != 1 {
			t.Fatal("delayed duplicate emitted another task or event")
		}
		// 省略与显式零保持原有即刻/绝对时间请求语义，不破坏升级前的摘要。
		zero := bytes.Clone(bytes.TrimSuffix(raw, []byte("}")))
		zero = append(zero, []byte(`,"delay_ms":0}`)...)
		must("POST", "/v1/tasks", a.Token, "stable-submit", zero, 200)
		batch := service.BatchSubmitInput{Items: []service.BatchSubmitItem{{IdempotencyKey: key, Task: delayed}, {IdempotencyKey: "delay-batch-new", Task: delayed}}}
		batchReply := must("POST", "/v1/tasks:batch", a.Token, "", marshal(batch), 200)
		var batchOut service.BatchSubmitResult
		if json.Unmarshal(batchReply.data, &batchOut) != nil || len(batchOut.Items) != 2 || batchOut.Items[0].Receipt == nil || batchOut.Items[1].Receipt == nil || !batchOut.Items[0].Receipt.Deduplicated || batchOut.Items[0].Receipt.TaskID != v.Task.ID {
			t.Fatal("batch did not reuse relative-delay identity")
		}
		fresh := must("GET", "/v1/tasks/"+string(batchOut.Items[1].Receipt.TaskID), a.Token, "", nil, 200)
		var freshTask service.TaskView
		if json.Unmarshal(fresh.data, &freshTask) != nil || !freshTask.AvailableAt.Equal(freshTask.CreatedAt.Add(5*time.Second)) {
			t.Fatal("batch delay did not use its own acceptance time")
		}
	})
}