package mysql

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

	"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
	"git.yuntuo.hk/k2safe/orbit/pulse/worker"
)

func TestIntegrationFairDispatchApplicationIsolation(t *testing.T) {
	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
	if dsn == "" {
		t.Skip("requires disposable MySQL")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
	defer cancel()
	db, err := Open(ctx, dsn, 4, 2)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err = Migrate(ctx, db); err != nil {
		t.Fatal(err)
	}
	noisy := newExecutionFixture(t, ctx, db)
	quiet := newExecutionFixture(t, ctx, db)
	if noisy.binding.ApplicationID != noisy.resources.ApplicationID || quiet.binding.ApplicationID != quiet.resources.ApplicationID {
		t.Fatal("dispatch group was not derived from authenticated namespace")
	}
	hub := httpserver.NewWorkerHub(ctx, noisy.a)
	server := httpserver.New("", func(context.Context) error { return nil })
	httpserver.RegisterWorkers(server, hub)
	tls := httptest.NewTLSServer(server.Handler)
	defer tls.Close()
	defer hub.Shutdown(ctx)
	var connections []*testWorkerSocket
	for range 12 {
		f := *noisy
		f.hello.InstanceID = executionID(t)
		c := connectTestWorker(t, ctx, tls, &f)
		id := c.send("worker.capacity", worker.CapacityInput{Sequence: 1, Capacity: 0}, "")
		if reply := c.read(); reply.ReplyTo != id || reply.Error != nil {
			t.Fatal("could not disable initial claims")
		}
		connections = append(connections, c)
	}
	quiet.hello.InstanceID = executionID(t)
	q := connectTestWorker(t, ctx, tls, quiet)
	id := q.send("worker.capacity", worker.CapacityInput{Sequence: 1, Capacity: 0}, "")
	if reply := q.read(); reply.ReplyTo != id || reply.Error != nil {
		t.Fatal("quiet worker setup failed")
	}
	noisy.submit(1)
	want := quiet.submit(1)
	lock, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer lock.Rollback()
	var ignored string
	if err = lock.QueryRowContext(ctx, `SELECT HEX(application_id) FROM orbit_application_resources WHERE application_id=? FOR UPDATE`, noisy.resources.ApplicationID).Scan(&ignored); err != nil {
		t.Fatal(err)
	}
	for _, c := range connections {
		id := c.send("worker.capacity", worker.CapacityInput{Sequence: 2, Capacity: 1}, "")
		if reply := c.read(); reply.ReplyTo != id || reply.Error != nil {
			t.Fatal("noisy worker control blocked by claims")
		}
	}
	// 等到真的有领取事务在额度行上等待，再验证另一应用；不靠固定 sleep 猜测故障已发生。
	for {
		var waiting int
		if err = db.QueryRowContext(ctx, `SELECT COUNT(*) FROM performance_schema.data_lock_waits`).Scan(&waiting); err != nil {
			t.Fatal(err)
		}
		if waiting > 0 {
			break
		}
		select {
		case <-ctx.Done():
			t.Fatal("held application did not receive a claim")
		case <-time.After(20 * time.Millisecond):
		}
	}
	id = q.send("worker.capacity", worker.CapacityInput{Sequence: 2, Capacity: 1}, "")
	if reply := q.read(); reply.ReplyTo != id || reply.Error != nil {
		t.Fatal("other application heartbeat starved")
	}
	reply := q.read()
	var assignment worker.Assignment
	if reply.Type != "task.assign" || json.Unmarshal(reply.Body, &assignment) != nil || assignment.Task.ID != want.ID {
		t.Fatal("quiet application could not claim while noisy application was blocked")
	}
	// 此时锁仍未释放；静默应用的真实领取已提交，证明没有耗尽四连接数据库池。
	if err := lock.Rollback(); err != nil {
		t.Fatal(err)
	}
	t.Log("12 noisy connections blocked on one application; another application claimed with a four-connection database pool before lock release")
}