package mysql

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"sort"
	"strings"
	"testing"
	"time"

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

func TestIntegrationTaskQuery(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("requires disposable MySQL")
	}
	ctx, stop := context.WithTimeout(context.Background(), 90*time.Second)
	defer stop()
	db, err := Open(ctx, dsn, 12, 4)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err = Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	f := newExecutionFixture(t, ctx, db)
	g := newExecutionFixture(t, ctx, db)
	from := time.Date(2026, 9, 17, 0, 0, 0, 123456000, time.UTC)
	before := from.Add(time.Hour)
	makeTask := func(owner *executionFixture, key string, at time.Time, state pulse.TaskState) core.ID {
		t.Helper()
		out, err := owner.tasks.Submit(ctx, owner.token, string(executionID(t)), service.SubmitInput{QueueID: owner.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: key})
		if err != nil || out.Task.BusinessKey != key {
			t.Fatal("business key changed on admission", err)
		}
		v := out.Task
		owner.exec(`UPDATE pulse_tasks SET created_at=? WHERE namespace_id=? AND id=?`, at, owner.resources.NamespaceID, v.ID)
		if state == pulse.TaskCancelled {
			if _, err := owner.tasks.CancelTask(ctx, owner.token, v.ID, string(executionID(t)), service.CancelInput{ExpectedRevision: v.Revision, Reason: "query regression"}); err != nil {
				t.Fatal(err)
			}
		}
		return v.ID
	}
	key := "任务 /+%? é "
	ids := []core.ID{makeTask(f, key, from, pulse.TaskWaiting), makeTask(f, key, from, pulse.TaskWaiting), makeTask(f, key, from.Add(time.Minute), pulse.TaskWaiting)}
	sort.Slice(ids[:2], func(i, j int) bool { return ids[i] > ids[j] })
	expected := []core.ID{ids[2], ids[0], ids[1]}
	makeTask(f, key, from.Add(-time.Microsecond), pulse.TaskWaiting)
	makeTask(f, key, before, pulse.TaskWaiting)
	makeTask(f, key, from, pulse.TaskCancelled)
	for _, other := range []string{strings.TrimSuffix(key, " "), strings.TrimSuffix(key, " ") + "  ", strings.Replace(key, "é", "É", 1), strings.Replace(key, "é", "e\u0301", 1), ""} {
		makeTask(f, other, from, pulse.TaskWaiting)
	}
	makeTask(g, key, from, pulse.TaskWaiting)
	spaceID := makeTask(f, " ", from, pulse.TaskWaiting)
	spaces, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: " "})
	if err != nil || len(spaces.Items) != 1 || spaces.Items[0].ID != spaceID {
		t.Fatal("space key not preserved", spaces, err)
	}

	otherQueue, err := f.tasks.CreateQueue(ctx, f.token, string(executionID(t)), service.QueueInput{Name: "other-query", MaxConcurrency: 1, RatePerSecond: 10})
	if err != nil {
		t.Fatal(err)
	}
	otherTask, err := f.tasks.Submit(ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: otherQueue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: key})
	if err != nil {
		t.Fatal(err)
	}
	f.exec(`UPDATE pulse_tasks SET created_at=? WHERE namespace_id=? AND id=?`, from, f.resources.NamespaceID, otherTask.Task.ID)
	longKey := strings.Repeat("界", 191)
	makeTask(f, longKey, from, pulse.TaskWaiting)
	makeTask(f, longKey, from, pulse.TaskWaiting)
	longPage, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: longKey, Limit: 1})
	if err != nil || longPage.NextCursor == "" || len(longPage.NextCursor) > 1024 {
		t.Fatal("maximum key overflowed cursor", err)
	}
	lastLong, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: longKey, Limit: 1, Cursor: longPage.NextCursor})
	if err != nil || len(lastLong.Items) != 1 || lastLong.Items[0].ID == longPage.Items[0].ID || lastLong.NextCursor != "" {
		t.Fatal("maximum key pagination", lastLong, err)
	}
	base := service.ListInput{QueueID: f.resources.QueueID, BusinessKey: key, State: pulse.TaskWaiting, CreatedFrom: &from, CreatedBefore: &before, Limit: 1}
	var firstCursor string
	t.Run("precise_key_time_and_same_timestamp_pages", func(t *testing.T) {
		in := base
		var got []core.ID
		for {
			page, err := f.tasks.ListTasks(ctx, f.token, in)
			if err != nil {
				t.Fatal(err)
			}
			for _, v := range page.Items {
				got = append(got, v.ID)
				if v.BusinessKey != key || v.State != pulse.TaskWaiting {
					t.Fatal("wrong query result")
				}
			}
			if firstCursor == "" {
				firstCursor = page.NextCursor
			}
			if page.NextCursor == "" {
				break
			}
			in.Cursor = page.NextCursor
		}
		if fmt.Sprint(got) != fmt.Sprint(expected) {
			t.Fatal("page boundaries", got, expected)
		}
		in = base
		in.State = ""
		in.Limit = 100
		page, err := f.tasks.ListTasks(ctx, f.token, in)
		if err != nil || len(page.Items) != 4 {
			t.Fatal(page, err)
		}
		in = base
		in.CreatedFrom = nil
		page, err = f.tasks.ListTasks(ctx, f.token, in)
		if err != nil || len(page.Items) != 1 || page.NextCursor == "" {
			t.Fatal(page, err)
		}
	})
	t.Run("cursor_binds_filters_and_current_authority", func(t *testing.T) {
		for name, change := range map[string]func(*service.ListInput){"key": func(v *service.ListInput) { v.BusinessKey = strings.TrimSuffix(key, " ") }, "remove_key": func(v *service.ListInput) { v.BusinessKey = "" }, "state": func(v *service.ListInput) { v.State = "" }, "lower": func(v *service.ListInput) { z := from.Add(-time.Second); v.CreatedFrom = &z }, "upper": func(v *service.ListInput) { z := before.Add(time.Second); v.CreatedBefore = &z }, "queue": func(v *service.ListInput) { v.QueueID = g.resources.QueueID }} {
			t.Run(name, func(t *testing.T) {
				in := base
				in.Cursor = firstCursor
				change(&in)
				_, err := f.tasks.ListTasks(ctx, f.token, in)
				expectWorkerCode(t, err, "INVALID_ARGUMENT")
			})
		}
		in := base
		in.Cursor = firstCursor
		_, err := g.tasks.ListTasks(ctx, g.token, in)
		expectWorkerCode(t, err, "INVALID_ARGUMENT")
		zone := time.FixedZone("equivalent", 8*3600)
		lo, hi := from.In(zone), before.In(zone)
		in.CreatedFrom = &lo
		in.CreatedBefore = &hi
		in.Limit = 2
		page, err := f.tasks.ListTasks(ctx, f.token, in)
		if err != nil || len(page.Items) != 2 {
			t.Fatal("equivalent UTC or page size not supported", page, err)
		}
		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(g.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID)
		_, err = f.tasks.ListTasks(ctx, f.token, in)
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY('*')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
	})
	t.Run("legacy_cursor_and_strict_cursor_validation", func(t *testing.T) {
		legacy := taskCursor{Namespace: f.resources.NamespaceID, Queue: f.resources.QueueID, State: pulse.TaskWaiting, Created: from.Add(time.Minute), ID: ids[2]}
		raw, _ := json.Marshal(legacy)
		encoded := base64.RawURLEncoding.EncodeToString(raw)
		_, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, State: pulse.TaskWaiting, Cursor: encoded})
		if err != nil {
			t.Fatal("old cursor rejected", err)
		}
		in := base
		in.Cursor = encoded
		_, err = f.tasks.ListTasks(ctx, f.token, in)
		expectWorkerCode(t, err, "INVALID_ARGUMENT")
		valid, _ := base64.RawURLEncoding.DecodeString(firstCursor)
		for _, raw := range []string{strings.TrimSuffix(string(valid), "}") + `,"v":2}`, strings.Replace(string(valid), `"v":2`, `"v":3`, 1), strings.Replace(string(valid), `"v":2`, `"v":null`, 1), strings.Replace(string(valid), `"v":2`, `"V":2`, 1), strings.TrimSuffix(string(valid), "}") + `,"secret":1}`, strings.Replace(string(valid), ".123456Z", ".123456001Z", 1)} {
			in.Cursor = base64.RawURLEncoding.EncodeToString([]byte(raw))
			_, err = f.tasks.ListTasks(ctx, f.token, in)
			expectWorkerCode(t, err, "INVALID_ARGUMENT")
		}
	})
	t.Run("http_validation_and_encoded_unicode", func(t *testing.T) {
		server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, f.tasks).Handler)
		defer server.Close()
		call := func(query string, status int) service.TaskPage {
			t.Helper()
			r, _ := http.NewRequestWithContext(ctx, "GET", server.URL+"/v1/tasks?queue_id="+string(f.resources.QueueID)+"&"+query, nil)
			r.Header.Set("Authorization", "Bearer "+f.resources.Token)
			resp, err := server.Client().Do(r)
			if err != nil {
				t.Fatal(err)
			}
			defer resp.Body.Close()
			raw, _ := io.ReadAll(resp.Body)
			if resp.StatusCode != status {
				t.Fatalf("query status=%d want=%d body=%s", resp.StatusCode, status, raw)
			}
			var out struct {
				Data service.TaskPage `json:"data"`
			}
			if err = json.Unmarshal(raw, &out); err != nil {
				t.Fatal(err)
			}
			return out.Data
		}
		q := url.Values{"business_key": {key}, "state": {"WAITING"}, "created_from": {from.Format(time.RFC3339Nano)}, "created_before": {before.Format(time.RFC3339Nano)}}
		if page := call(q.Encode(), 200); len(page.Items) != 3 {
			t.Fatal(page)
		}
		for _, q := range []string{"business_key=", "business_key=a&business_key=b", "business_key=%ff", "business_key=%00", "created_from=", "created_from=2026-02-30T00:00:00Z", "created_from=2026-09-17T00:00:00.000000001Z", "created_from=2026-09-17", "created_before=1999-01-01T00:00:00Z", "created_from=2026-09-17T00:00:00Z&created_before=2026-09-17T00:00:00Z", "CreatedFrom=1", "created_from=%zz", "created_before=a&created_before=b"} {
			call(q, 400)
		}
	})
	t.Run("all_filter_shapes_use_ordered_composite_indexes", func(t *testing.T) {
		cases := []struct {
			state      pulse.TaskState
			key, index string
		}{{"", "", "ix_task_created"}, {pulse.TaskWaiting, "", "ix_task_state_created"}, {"", key, "ix_task_queue_business_created"}, {pulse.TaskWaiting, key, "ix_task_queue_business_state_created"}}
		for _, tc := range cases {
			in := base
			in.State = tc.state
			in.BusinessKey = tc.key
			digest, err := taskFilterDigest(in)
			if err != nil {
				t.Fatal(err)
			}
			q, args, err := taskListSQL(f.resources.NamespaceID, in, digest)
			if err != nil {
				t.Fatal(err)
			}
			var raw string
			if err = db.QueryRowContext(ctx, "EXPLAIN FORMAT=JSON "+q, args...).Scan(&raw); err != nil {
				t.Fatal(err)
			}
			var plan map[string]any
			if err = json.Unmarshal([]byte(raw), &plan); err != nil {
				t.Fatal(err)
			}
			found := false
			var walk func(any)
			walk = func(v any) {
				switch x := v.(type) {
				case map[string]any:
					if x["using_filesort"] == true {
						t.Fatal("query uses filesort", raw)
					}
					if x["table_name"] == "pulse_tasks" {
						found = true
						if x["key"] != tc.index || (x["access_type"] != "range" && x["access_type"] != "ref") {
							t.Fatal("unbounded query access", raw)
						}
					}
					for _, v := range x {
						walk(v)
					}
				case []any:
					for _, v := range x {
						walk(v)
					}
				}
			}
			walk(plan)
			if !found {
				t.Fatal("task query plan missing", raw)
			}
			t.Logf("state=%q business_key=%t index=%s no filesort", tc.state, tc.key != "", tc.index)
		}
	})
}