package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"os"
	"sync"
	"testing"
	"time"

	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/resources"
	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
	"git.yuntuo.hk/k2safe/orbit/pulse/worker"
)

func grantApplicationResources(f *executionFixture) {
	f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY_APPEND(JSON_EXTRACT(permissions,'$.actions'),'$','application:resources:read','$','application:resources:manage')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
}
func applicationAPI(f *executionFixture) *resources.Service { return resources.New(NewTaskStore(f.db)) }
func configureApplication(t *testing.T, f *executionFixture, rate, concurrency, unfinished uint32) resources.View {
	t.Helper()
	api := applicationAPI(f)
	v, err := api.Get(f.ctx, f.token)
	if err != nil {
		t.Fatal(err)
	}
	out, err := api.Configure(f.ctx, f.token, string(executionID(t)), resources.ConfigureInput{ExpectedRevision: v.Revision, Reason: "application quota regression", RatePerSecond: rate, MaxConcurrency: concurrency, MaxUnfinishedTasks: unfinished})
	if err != nil {
		t.Fatal(err)
	}
	return out.Resources
}
func assertApplicationCounts(t *testing.T, f *executionFixture, unfinished, active, pending uint64) {
	t.Helper()
	// 数据库核对不依赖当前应用是否停用，恢复仍应释放额度。
	var got [3]uint64
	if err := f.db.QueryRowContext(f.ctx, `SELECT unfinished_tasks,active_attempts,pending_starts FROM orbit_application_resources WHERE application_id=?`, f.resources.ApplicationID).Scan(&got[0], &got[1], &got[2]); err != nil {
		t.Fatal(err)
	}
	if got != [3]uint64{unfinished, active, pending} {
		t.Fatalf("application counters=%v want=%v", got, [3]uint64{unfinished, active, pending})
	}
	var actual [3]uint64
	if err := f.db.QueryRowContext(f.ctx, `SELECT COUNT(*),COALESCE(SUM(t.state IN ('LEASED','RUNNING')),0),COALESCE(SUM(t.state='LEASED'),0) FROM pulse_tasks t JOIN orbit_namespaces n ON n.id=t.namespace_id WHERE n.application_id=? AND t.state IN ('WAITING','LEASED','RUNNING','NEEDS_REVIEW')`, f.resources.ApplicationID).Scan(&actual[0], &actual[1], &actual[2]); err != nil {
		t.Fatal(err)
	}
	if actual != got {
		t.Fatalf("application counters=%v actual states=%v", got, actual)
	}
}
func TestIntegrationApplicationResources(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, 32, 8)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err := Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	fixture := func(t *testing.T, app ...core.ID) *executionFixture {
		f := newExecutionFixture(t, ctx, db, app...)
		grantApplicationResources(f)
		return f
	}
	t.Run("explicit_authority_CAS_idempotency_and_atomic_audit", func(t *testing.T) {
		f := newExecutionFixture(t, ctx, db)
		api := applicationAPI(f)
		_, err := api.Get(ctx, f.token)
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		in := resources.ConfigureInput{ExpectedRevision: 1, Reason: "configure", MaxConcurrency: 2, RatePerSecond: 10, MaxUnfinishedTasks: 10}
		_, err = api.Configure(ctx, f.token, "manage", in)
		expectWorkerCode(t, err, "PERMISSION_DENIED")
		grantApplicationResources(f)
		v, err := api.Configure(ctx, f.token, "manage", in)
		if err != nil || v.Resources.Revision != 2 {
			t.Fatal(v, err)
		}
		f.submit(1)
		replay, err := api.Configure(ctx, f.token, "manage", in)
		if err != nil || !replay.Deduplicated || replay.Resources.UnfinishedTasks != 0 {
			t.Fatal("receipt changed with usage", err)
		}
		_, err = api.Configure(ctx, f.token, "stale", in)
		expectWorkerCode(t, err, "STATE_CONFLICT")
		in.RatePerSecond = 20
		_, err = api.Configure(ctx, f.token, "manage", in)
		expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
		in.ExpectedRevision = 2
		trigger := "test_app_audit_" + string(executionID(t))
		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected audit failure'")
		_, err = api.Configure(ctx, f.token, "rollback", in)
		f.exec("DROP TRIGGER " + trigger)
		if err == nil {
			t.Fatal("audit fault ignored")
		}
		v2, err := api.Get(ctx, f.token)
		if err != nil || v2.Revision != 2 || v2.RatePerSecond != 10 {
			t.Fatal("configuration did not roll back", v2, err)
		}
		if _, err := api.Configure(ctx, f.token, "rollback", in); err != nil {
			t.Fatal(err)
		}
		if f.count("orbit_audit_events", "action='application.resources.configure'") != 2 || f.count("orbit_events", "type='application.resources.configured'") != 2 {
			t.Fatal("duplicate or missing audit/event")
		}
		g := fixture(t, f.resources.ApplicationID)
		shared, err := applicationAPI(g).Get(ctx, g.token)
		if err != nil || shared.Revision != 3 || shared.UnfinishedTasks != 1 {
			t.Fatal(shared, err)
		}
		other := fixture(t)
		if v, err := applicationAPI(other).Get(ctx, other.token); err != nil || v.Revision != 1 || v.UnfinishedTasks != 0 {
			t.Fatal("other application leaked", v, err)
		}
	})
	t.Run("concurrent_admission_across_namespaces_and_full_replay", func(t *testing.T) {
		f := fixture(t)
		g := fixture(t, f.resources.ApplicationID)
		configureApplication(t, f, 100, 100, 3)
		type answer struct {
			f   *executionFixture
			key string
			in  service.SubmitInput
			out service.SubmitResult
			err error
		}
		ch := make(chan answer, 16)
		var wg sync.WaitGroup
		for i := 0; i < 16; i++ {
			target := f
			if i%2 == 1 {
				target = g
			}
			key := string(executionID(t))
			wg.Add(1)
			go func() {
				defer wg.Done()
				in := service.SubmitInput{QueueID: target.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}
				out, err := target.tasks.Submit(ctx, target.token, key, in)
				ch <- answer{target, key, in, out, err}
			}()
		}
		wg.Wait()
		close(ch)
		var accepted []answer
		for a := range ch {
			if a.err == nil {
				accepted = append(accepted, a)
			} else {
				expectWorkerCode(t, a.err, "APPLICATION_FULL")
			}
		}
		if len(accepted) != 3 {
			t.Fatalf("accepted=%d", len(accepted))
		}
		assertApplicationCounts(t, f, 3, 0, 0)
		a := accepted[0]
		replay, err := a.f.tasks.Submit(ctx, a.f.token, a.key, a.in)
		if err != nil || !replay.Deduplicated || replay.Task.ID != a.out.Task.ID {
			t.Fatal("full app lost receipt", err)
		}
		configureApplication(t, g, 100, 100, 1)
		assertApplicationCounts(t, f, 3, 0, 0)
		for _, a := range accepted {
			_, err := a.f.tasks.CancelTask(ctx, a.f.token, a.out.Task.ID, "cancel-"+a.key, service.CancelInput{ExpectedRevision: 1, Reason: "release"})
			if err != nil {
				t.Fatal(err)
			}
		}
		assertApplicationCounts(t, g, 0, 0, 0)
		g.submit(1)
		assertApplicationCounts(t, f, 1, 0, 0)
	})
	t.Run("queue_rejection_and_failed_admission_do_not_leak_application_budget", func(t *testing.T) {
		f := fixture(t)
		configureApplication(t, f, 100, 10, 10)
		configureResources(t, f, 100, 10, 1, false)
		first := f.submit(1)
		in := service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)}
		_, err := f.tasks.Submit(ctx, f.token, "second", in)
		expectWorkerCode(t, err, "QUEUE_FULL")
		assertApplicationCounts(t, f, 1, 0, 0)
		if _, err := f.tasks.CancelTask(ctx, f.token, first.ID, "cancel-first", service.CancelInput{ExpectedRevision: first.Revision, Reason: "release"}); err != nil {
			t.Fatal(err)
		}
		trigger := "test_app_admission_" + string(executionID(t))
		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected admission failure'")
		_, err = f.tasks.Submit(ctx, f.token, "second", in)
		f.exec("DROP TRIGGER " + trigger)
		if err == nil {
			t.Fatal("admission failure ignored")
		}
		assertApplicationCounts(t, f, 0, 0, 0)
		assertQueueCount(t, f, 0)
		if _, err := f.tasks.Submit(ctx, f.token, "second", in); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
		assertQueueCount(t, f, 1)
	})
	t.Run("shared_concurrency_and_exact_release", func(t *testing.T) {
		f := fixture(t)
		g := fixture(t, f.resources.ApplicationID)
		configureApplication(t, f, 100, 1, 10)
		f.submit(1)
		g.submit(1)
		type claim struct {
			f   *executionFixture
			a   *worker.Assignment
			err error
		}
		ch := make(chan claim, 2)
		for _, target := range []*executionFixture{f, g} {
			go func() { a, err := target.a.Claim(ctx, target.token, target.binding); ch <- claim{target, a, err} }()
		}
		var winner claim
		for i := 0; i < 2; i++ {
			a := <-ch
			if a.err != nil {
				t.Fatal(a.err)
			}
			if a.a != nil {
				if winner.a != nil {
					t.Fatal("cross namespace concurrency exceeded")
				}
				winner = a
			}
		}
		if winner.a == nil {
			t.Fatal("no claim")
		}
		assertApplicationCounts(t, f, 2, 1, 1)
		task := winner.a.Task
		_, err := winner.f.tasks.CancelTask(ctx, winner.f.token, task.ID, "cancel-leased", service.CancelInput{ExpectedRevision: task.Revision, Reason: "release before start"})
		if err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
		other := f
		if winner.f == f {
			other = g
		}
		ref := retryStart(t, other)
		assertApplicationCounts(t, f, 1, 1, 0)
		message := executionID(t)
		input := worker.FinishInput{Ref: ref, Outcome: "SUCCEEDED", Result: json.RawMessage(`{}`)}
		if _, err := other.a.Finish(ctx, other.token, other.binding, message, input); err != nil {
			t.Fatal(err)
		}
		if out, err := other.a.Finish(ctx, other.token, other.binding, message, input); err != nil || !out.Deduplicated {
			t.Fatal("finish replay", err)
		}
		assertApplicationCounts(t, f, 0, 0, 0)
	})
	t.Run("shared_start_window_downgrade_and_rollback", func(t *testing.T) {
		f := fixture(t)
		g := fixture(t, f.resources.ApplicationID)
		configureApplication(t, f, 2, 10, 10)
		f.submit(1)
		g.submit(1)
		_, a := f.claim()
		_, b := g.claim()
		assertApplicationCounts(t, f, 2, 2, 2)
		configureApplication(t, g, 1, 1, 10)
		trigger := "test_app_start_" + string(executionID(t))
		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected start failure'")
		message := executionID(t)
		_, err := f.a.Start(ctx, f.token, f.binding, message, a)
		f.exec("DROP TRIGGER " + trigger)
		if err == nil {
			t.Fatal("start failure ignored")
		}
		assertApplicationCounts(t, f, 2, 2, 2)
		if _, err := f.a.Start(ctx, f.token, f.binding, message, a); err != nil {
			t.Fatal(err)
		}
		// 固定尚未过期窗口，避免慢测试主机把竞争断言跨到下一秒。
		f.exec(`UPDATE orbit_application_resources SET start_window_until=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR) WHERE application_id=?`, f.resources.ApplicationID)
		if out, err := f.a.Start(ctx, f.token, f.binding, message, a); err != nil || !out.Deduplicated {
			t.Fatal("start replay charged again", err)
		}
		_, err = g.a.Start(ctx, g.token, g.binding, executionID(t), b)
		expectWorkerCode(t, err, "RATE_LIMITED")
		assertApplicationCounts(t, f, 2, 2, 1)
		retryFinish(t, f, a, "SUCCEEDED", false)
		configureApplication(t, f, 1, 10, 10)
		g.submit(1)
		if a, err := g.a.Claim(ctx, g.token, g.binding); err != nil || a != nil {
			t.Fatal("pending start reservation bypassed", err)
		}
		g.exec(`UPDATE orbit_application_resources SET start_window_until=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE application_id=?`, g.resources.ApplicationID)
		if a, err := g.a.Claim(ctx, g.token, g.binding); err != nil || a != nil {
			t.Fatal("unstarted reservation lost across window", err)
		}
		if _, err := g.a.Start(ctx, g.token, g.binding, executionID(t), b); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 2, 1, 0)
	})
	t.Run("retry_recovery_unknown_and_disabled_application", func(t *testing.T) {
		f := fixture(t)
		configureApplication(t, f, 100, 10, 10)
		task := retryTask(t, f, 2, pulse.RecoveryReconcileOnly)
		ref := retryStart(t, f)
		retryFinish(t, f, ref, "FAILED", true)
		assertApplicationCounts(t, f, 1, 0, 0)
		makeRetryDue(f, task.ID)
		ref = retryStart(t, f)
		f.expire(ref)
		f.exec(`UPDATE orbit_applications SET enabled=FALSE WHERE id=?`, f.resources.ApplicationID)
		if _, err := f.storeB.Recover(ctx, 100); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
		if _, err := f.storeA.Recover(ctx, 100); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
		f.exec(`UPDATE orbit_applications SET enabled=TRUE WHERE id=?`, f.resources.ApplicationID)
		if f.get(task.ID).State != pulse.TaskNeedsReview {
			t.Fatal("unknown effect lost")
		}
		f.submit(1)
		_, leased := f.claim()
		f.expire(leased)
		if _, err := f.storeB.Recover(ctx, 100); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
		waiting := f.submit(1)
		f.exec(`UPDATE pulse_tasks SET available_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 2 SECOND),deadline_at=DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 1 SECOND) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, waiting.ID)
		if _, err := f.storeB.Recover(ctx, 100); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 1, 0, 0)
	})
	t.Run("schedule_partial_batch_and_global_backpressure", func(t *testing.T) {
		f := fixture(t)
		g := fixture(t, f.resources.ApplicationID)
		configureApplication(t, f, 100, 10, 2)
		g.submit(1)
		sf := newScheduleFixture(t, f)
		v := sf.create(sf.definition())
		sf.due(v, 2*time.Second)
		sf.tick()
		items := sf.occurrences(v.ID)
		if len(items) != 1 || items[0].TaskID == nil {
			t.Fatal("application schedule limit exceeded", len(items))
		}
		next := sf.current(v.ID).NextFireAt
		if next == nil || !next.Equal(items[0].ScheduledAt.Add(time.Second)) {
			t.Fatal("backpressure advanced unaccepted cursor")
		}
		assertApplicationCounts(t, f, 2, 0, 0)
		assertQueueCount(t, f, 1)
		assertQueueCount(t, g, 1)
		_, err := f.tasks.CancelTask(ctx, f.token, *items[0].TaskID, "free-schedule", service.CancelInput{ExpectedRevision: 1, Reason: "free one"})
		if err != nil {
			t.Fatal(err)
		}
		sf.due(v, 0)
		sf.tick()
		items = sf.occurrences(v.ID)
		if len(items) != 2 {
			t.Fatal("original point not resumed")
		}
		assertApplicationCounts(t, f, 2, 0, 0)
		assertQueueCount(t, f, 1)
	})
	t.Run("busy_application_is_skipped_during_recovery", func(t *testing.T) {
		f := fixture(t)
		g := fixture(t)
		for _, v := range []*executionFixture{f, g} {
			v.submit(1)
			_, ref := v.claim()
			v.expire(ref)
		}
		tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
		if err != nil {
			t.Fatal(err)
		}
		defer tx.Rollback()
		if _, err := lockApplicationResources(ctx, tx, f.resources.NamespaceID, false); err != nil {
			t.Fatal(err)
		}
		short, stop := context.WithTimeout(ctx, time.Second)
		defer stop()
		if _, err := g.storeB.Recover(short, 100); err != nil {
			t.Fatal("hot application blocked unrelated recovery", err)
		}
		assertApplicationCounts(t, f, 1, 1, 1)
		assertApplicationCounts(t, g, 0, 0, 0)
		if err := tx.Rollback(); err != nil {
			t.Fatal(err)
		}
		if _, err := f.storeB.Recover(ctx, 100); err != nil {
			t.Fatal(err)
		}
		assertApplicationCounts(t, f, 0, 0, 0)
	})
	t.Run("bootstrap_existing_application_rejects_invalid_parent", func(t *testing.T) {
		f := fixture(t)
		f.exec(`UPDATE orbit_applications SET enabled=FALSE WHERE id=?`, f.resources.ApplicationID)
		for _, in := range []BootstrapInput{
			{ExistingApplicationID: executionID(t), Namespace: "missing", Environment: "test", Queue: "q"},
			{ExistingApplicationID: f.resources.ApplicationID, Namespace: "disabled", Environment: "test", Queue: "q"},
			{Application: "new", ExistingApplicationID: f.resources.ApplicationID, Namespace: "ambiguous", Environment: "test", Queue: "q"},
		} {
			delivered := false
			err := Bootstrap(ctx, db, in, func(BootstrapResult) error { delivered = true; return nil })
			if err == nil || delivered {
				t.Fatal("invalid application accepted")
			}
		}
	})
}