package mysql

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"reflect"
	"strings"
	"testing"
	"time"

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

func TestIntegrationTaskSource(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, 16, 4)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err = Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	t.Run("direct_and_manual_retry_are_read_only_and_scoped", func(t *testing.T) {
		f := newExecutionFixture(t, ctx, db)
		grantTaskRecovery(f)
		original := f.submit(1)
		source, err := f.tasks.GetTaskSource(ctx, f.token, original.ID)
		if err != nil || source.TaskID != original.ID || source.Kind != "DIRECT" || source.Schedule != nil || source.RetryOf != nil {
			t.Fatal(source, err)
		}
		cancelled, err := f.tasks.CancelTask(ctx, f.token, original.ID, "cancel", service.CancelInput{ExpectedRevision: original.Revision, Reason: "retry source regression"})
		if err != nil {
			t.Fatal(err)
		}
		child, err := f.tasks.RetryTask(ctx, f.token, original.ID, "retry", manualRetryInput(cancelled.Task))
		if err != nil {
			t.Fatal(err)
		}
		source, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
		if err != nil || source.Kind != "MANUAL_RETRY" || source.Schedule != nil || source.RetryOf == nil || *source.RetryOf != original.ID {
			t.Fatal(source, err)
		}
		if !reflect.DeepEqual(f.get(original.ID), cancelled.Task) || !reflect.DeepEqual(f.get(child.Task.ID), child.Task) {
			t.Fatal("source read changed task")
		}
		other := newExecutionFixture(t, ctx, db)
		_, err = other.tasks.GetTaskSource(ctx, other.token, child.Task.ID)
		expectWorkerCode(t, err, "NOT_FOUND")
		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(other.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID)
		_, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
		expectWorkerCode(t, err, "NOT_FOUND")
		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY('*')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
		// 模拟无外键库的缺失与跨命名空间原任务，不能返回未验证的标识。
		for _, parent := range []core.ID{executionID(t), other.submit(1).ID} {
			f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, parent, f.resources.NamespaceID, child.Task.ID)
			_, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
			expectWorkerCode(t, err, "UNAVAILABLE")
		}
		f.exec(`UPDATE pulse_tasks SET retry_of=? WHERE namespace_id=? AND id=?`, original.ID, f.resources.NamespaceID, child.Task.ID)
		// 没有外键时，父任务即使丢失owner，反向流程关联仍能揭示其不是独立任务。
		activation := executionID(t)
		f.exec(`INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,task_id,input) VALUES (?,?,?,?,1,?,'{}')`, f.resources.NamespaceID, activation, executionID(t), executionID(t), original.ID)
		_, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
		expectWorkerCode(t, err, "UNAVAILABLE")
		f.exec(`DELETE FROM weave_activations WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, activation)
		f.exec(`UPDATE orbit_credentials SET revoked_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
		_, err = f.tasks.GetTaskSource(ctx, f.token, child.Task.ID)
		expectWorkerCode(t, err, "UNAUTHENTICATED")
	})
	t.Run("schedule_source_preserves_published_version_and_permission_boundary", func(t *testing.T) {
		f := newScheduleFixture(t, newExecutionFixture(t, ctx, db))
		in := f.definition()
		plan := f.create(in)
		f.due(plan, 2*time.Second)
		f.tick()
		o := f.occurrences(plan.ID)[0]
		if o.TaskID == nil {
			t.Fatal("missing task")
		}
		before := f.get(*o.TaskID)
		source, err := f.tasks.GetTaskSource(ctx, f.token, *o.TaskID)
		if err != nil || source.Kind != "SCHEDULE" || source.Schedule == nil || source.RetryOf != nil {
			t.Fatal(source, err)
		}
		link := source.Schedule
		if link.ScheduleID != plan.ID || link.OccurrenceID != o.ID || link.ScheduleVersion != o.ScheduleVersion || !link.ScheduledAt.Equal(o.ScheduledAt) || link.Decision != o.Decision {
			t.Fatal("incorrect schedule lineage", link, o)
		}
		in.Task.Payload = json.RawMessage(`{"version":2}`)
		now := f.current(plan.ID)
		published, err := f.api.Publish(ctx, f.token, plan.ID, "publish", schedule.PublishInput{ExpectedRevision: now.Revision, Reason: "source stays on original version", Definition: in})
		if err != nil || published.Schedule.CurrentVersion != 2 {
			t.Fatal(published, err)
		}
		after, err := f.tasks.GetTaskSource(ctx, f.token, *o.TaskID)
		if err != nil || !reflect.DeepEqual(source, after) || !reflect.DeepEqual(before, f.get(*o.TaskID)) {
			t.Fatal("latest publication changed source", after, err)
		}
		// tasks:read仅授予此任务的来源引用，不因此授予完整计划/模板查询权限。
		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY('tasks:read')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
		if _, err = f.tasks.GetTaskSource(ctx, f.token, *o.TaskID); err != nil {
			t.Fatal(err)
		}
		_, err = f.api.Get(ctx, f.token, plan.ID)
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, f.tasks).Handler)
		defer server.Close()
		call := func(path string, status int) []byte {
			t.Helper()
			r, _ := http.NewRequestWithContext(ctx, "GET", server.URL+path, 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("status=%d want=%d body=%s", resp.StatusCode, status, raw)
			}
			return raw
		}
		path := "/v1/tasks/" + string(*o.TaskID) + "/source"
		raw := call(path, 200)
		for _, secret := range []string{"task_template", "payload", "lease_secret", "secret_digest", "token"} {
			if strings.Contains(string(raw), secret) {
				t.Fatal("source leaked unrelated data")
			}
		}
		for _, suffix := range []string{"?limit=1", "?namespace_id=x", "?unknown=%zz"} {
			call(path+suffix, 400)
		}
		call("/v1/tasks/not-an-id/source", 400)
		call("/v1/tasks/"+string(executionID(t))+"/source", 404)
	})
	t.Run("coalesced_source_retains_exclusive_range", func(t *testing.T) {
		f := newScheduleFixture(t, newExecutionFixture(t, ctx, db))
		in := f.definition()
		in.Policy.Misfire = "COALESCE"
		plan := f.create(in)
		f.due(plan, 3*time.Second)
		f.tick()
		found := false
		for _, o := range f.occurrences(plan.ID) {
			if o.Decision != "COALESCED" {
				continue
			}
			found = true
			source, err := f.tasks.GetTaskSource(ctx, f.token, *o.TaskID)
			if err != nil || source.Schedule == nil || source.Schedule.CoveredUntil == nil || !source.Schedule.CoveredUntil.Equal(*o.CoveredUntil) || !source.Schedule.ScheduledAt.Equal(o.ScheduledAt) {
				t.Fatal(source, err)
			}
		}
		if !found {
			t.Fatal("coalesced trigger missing")
		}
	})
	t.Run("dangling_mismatched_and_unknown_owners_fail_closed", func(t *testing.T) {
		f := newScheduleFixture(t, newExecutionFixture(t, ctx, db))
		plan := f.create(f.definition())
		f.due(plan, time.Second)
		f.tick()
		o := f.occurrences(plan.ID)[0]
		id := *o.TaskID
		broken := func() {
			t.Helper()
			out, err := f.tasks.GetTaskSource(ctx, f.token, id)
			expectWorkerCode(t, err, "UNAVAILABLE")
			if out.TaskID != "" || out.Schedule != nil {
				t.Fatal("partial invalid source returned")
			}
		}
		f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, id)
		broken()
		f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, plan.ID, f.resources.NamespaceID, id)
		f.exec(`UPDATE pulse_schedule_occurrences SET schedule_version=999 WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, o.ID)
		broken()
		f.exec(`UPDATE pulse_schedule_occurrences SET schedule_version=? WHERE namespace_id=? AND id=?`, o.ScheduleVersion, f.resources.NamespaceID, o.ID)
		f.exec(`UPDATE pulse_schedules SET queue_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, plan.ID)
		broken()
		f.exec(`UPDATE pulse_schedules SET queue_id=? WHERE namespace_id=? AND id=?`, f.resources.QueueID, f.resources.NamespaceID, plan.ID)
		f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, id)
		broken()
		f.exec(`UPDATE pulse_tasks SET owner_kind='workflow',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, id)
		broken()
		f.exec(`UPDATE pulse_tasks SET owner_kind='schedule',owner_id=?,gate_id=? WHERE namespace_id=? AND id=?`, plan.ID, executionID(t), f.resources.NamespaceID, id)
		broken()
		f.exec(`UPDATE pulse_tasks SET gate_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, id)
		if _, err = f.tasks.GetTaskSource(ctx, f.token, id); err != nil {
			t.Fatal("restored valid source still refused", err)
		}
	})
}