package httpapi

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
	"github.com/jackc/pgx/v5/pgxpool"
)

func TestAsyncSubmissionReservesConcurrencyOnlyAfterWorkerClaim(t *testing.T) {
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run async admission timing tests")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	applyMigration(t, ctx, databaseURL)

	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect store: %v", err)
	}
	defer db.Close()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect acceptance pool: %v", err)
	}
	defer pool.Close()
	restoreAcceptanceState := isolateHTTPAcceptanceState(t, ctx, pool, false)
	defer restoreAcceptanceState()

	serverCtx, cancelServer := context.WithCancel(ctx)
	defer cancelServer()
	server := httptest.NewServer(NewServerWithContext(serverCtx, config.Config{
		AppEnv:                            "test",
		HTTPAddr:                          ":0",
		DatabaseURL:                       databaseURL,
		IdentityMode:                      "hybrid",
		JWTSecret:                         "test-secret",
		BillingEngineMode:                 "observe",
		CORSAllowedOrigin:                 "*",
		AsyncQueueWorkerEnabled:           true,
		AsyncWorkerHardLimit:              1,
		AsyncWorkerRefreshIntervalSeconds: 1,
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
	defer server.Close()

	adminToken := createAsyncAcceptanceAdmin(t, ctx, pool, server.URL)
	if _, err := pool.Exec(ctx, `
UPDATE gateway_user_groups
SET rate_limit_policy = '{"rules":[{"metric":"concurrent","limit":256,"leaseTtlSeconds":120}]}'::jsonb
WHERE status = 'active'`); err != nil {
		t.Fatalf("raise acceptance user-group concurrency: %v", err)
	}
	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
	model := "worker-claim-admission-" + suffix
	platform := createAsyncAcceptancePlatform(t, server.URL, adminToken, "claim-admission-"+suffix, "Claim Admission Simulation", 3)
	defer updateAsyncAcceptancePlatform(t, server.URL, adminToken, platform, 3, "disabled")
	modelID := createAsyncAcceptanceModel(t, server.URL, adminToken, platform.ID, model, "video_generate", "inherit", 0, 120)

	waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_capacity", 1, 15*time.Second)
	taskIDs := submitAsyncSimulationTasks(t, server.URL, adminToken, "/api/v1/videos/generations", model, 3, 5*time.Second)
	waitForTaskCount(t, ctx, pool, taskIDs, "running", 1, 10*time.Second)
	waitForActiveLeaseCount(t, ctx, pool, modelID, 1, 10*time.Second)

	var admitted, attempts int
	if err := pool.QueryRow(ctx, `
SELECT
  (SELECT COUNT(*)
   FROM gateway_task_admissions
   WHERE task_id = ANY($1::uuid[]) AND status = 'admitted'),
  (SELECT COUNT(*)
   FROM gateway_task_attempts
   WHERE task_id = ANY($1::uuid[]))`, taskIDs).Scan(&admitted, &attempts); err != nil {
		t.Fatalf("read queued admission state: %v", err)
	}
	if admitted != 1 || attempts != 1 {
		t.Fatalf("claimed admission state admitted=%d attempts=%d, want 1/1", admitted, attempts)
	}

	waitForTaskCount(t, ctx, pool, taskIDs, "succeeded", len(taskIDs), 30*time.Second)
	assertAcceptanceAttempts(t, ctx, pool, taskIDs, len(taskIDs))
}

func TestAsyncAdmissionPoisonTaskDoesNotBlockHealthyFollower(t *testing.T) {
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run async admission resilience tests")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	applyMigration(t, ctx, databaseURL)

	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect store: %v", err)
	}
	defer db.Close()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect acceptance pool: %v", err)
	}
	defer pool.Close()
	restoreAcceptanceState := isolateHTTPAcceptanceState(t, ctx, pool, false)
	defer restoreAcceptanceState()

	serverCtx, cancelServer := context.WithCancel(ctx)
	defer cancelServer()
	server := httptest.NewServer(NewServerWithContext(serverCtx, config.Config{
		AppEnv:                             "test",
		HTTPAddr:                           ":0",
		DatabaseURL:                        databaseURL,
		IdentityMode:                       "hybrid",
		JWTSecret:                          "test-secret",
		BillingEngineMode:                  "observe",
		CORSAllowedOrigin:                  "*",
		AsyncQueueWorkerEnabled:            true,
		AsyncWorkerHardLimit:               1,
		AsyncWorkerInstanceHardLimit:       1,
		AsyncWorkerRefreshIntervalSeconds:  1,
		AsyncAdmissionMicrobatchSize:       3,
		AsyncAdmissionDispatcherEnabled:    true,
		AsyncAdmissionDispatcherConfigured: true,
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
	defer server.Close()

	adminToken := createAsyncAcceptanceAdmin(t, ctx, pool, server.URL)
	if _, err := pool.Exec(ctx, `
UPDATE gateway_user_groups
SET rate_limit_policy = '{"rules":[{"metric":"concurrent","limit":256,"leaseTtlSeconds":120}]}'::jsonb
WHERE status = 'active'`); err != nil {
		t.Fatalf("raise resilience user-group concurrency: %v", err)
	}
	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
	model := "admission-poison-" + suffix
	platform := createAsyncAcceptancePlatform(t, server.URL, adminToken, "admission-poison-"+suffix, "Admission Poison Simulation", 3)
	defer updateAsyncAcceptancePlatform(t, server.URL, adminToken, platform, 3, "disabled")
	createAsyncAcceptanceModel(t, server.URL, adminToken, platform.ID, model, "video_generate", "inherit", 0, 120)

	waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_capacity", 1, 15*time.Second)
	taskIDs := submitAsyncSimulationTasks(t, server.URL, adminToken, "/api/v1/videos/generations", model, 3, 10*time.Second)
	waitForTaskCount(t, ctx, pool, taskIDs, "running", 1, 10*time.Second)

	var poisonTaskID, followerTaskID string
	if err := pool.QueryRow(ctx, `
SELECT (array_agg(admission.task_id::text ORDER BY admission.priority, admission.enqueued_at, admission.task_id))[1],
       (array_agg(admission.task_id::text ORDER BY admission.priority, admission.enqueued_at, admission.task_id))[2]
FROM gateway_task_admissions admission
JOIN gateway_tasks task ON task.id = admission.task_id
WHERE task.id = ANY($1::uuid[])
  AND task.status = 'queued'
  AND admission.status = 'waiting'`, taskIDs).Scan(&poisonTaskID, &followerTaskID); err != nil {
		t.Fatalf("select waiting poison and follower tasks: %v", err)
	}
	var prematureReselect int
	if err := pool.QueryRow(ctx, `
SELECT count(*)
FROM gateway_task_admissions
WHERE task_id = ANY($1::uuid[])
  AND status = 'waiting'
  AND reselect_requested_at IS NOT NULL`, taskIDs).Scan(&prematureReselect); err != nil {
		t.Fatalf("count global-capacity reselections: %v", err)
	}
	if prematureReselect != 0 {
		t.Fatalf("global worker saturation marked %d tasks for reselection", prematureReselect)
	}
	if _, err := pool.Exec(ctx, `
UPDATE gateway_tasks
SET model = $2,
    requested_model = $2,
    updated_at = now()
WHERE id = $1::uuid`, poisonTaskID, "missing-model-"+suffix); err != nil {
		t.Fatalf("poison queued task model: %v", err)
	}
	if _, err := pool.Exec(ctx, `
UPDATE gateway_task_admissions
SET reselect_requested_at = now(),
    dispatch_next_at = now(),
    updated_at = now()
WHERE task_id = $1::uuid`, poisonTaskID); err != nil {
		t.Fatalf("poison queued admission binding: %v", err)
	}

	waitForTaskCount(t, ctx, pool, []string{poisonTaskID}, "failed", 1, 20*time.Second)
	waitForTaskCount(t, ctx, pool, []string{followerTaskID}, "succeeded", 1, 30*time.Second)
	var poisonCode string
	var poisonAttempts, poisonAdmissions, poisonLeases, poisonFailedEvents int
	var poisonRiverJobID int64
	if err := pool.QueryRow(ctx, `
SELECT COALESCE(task.error_code, ''), COALESCE(task.river_job_id, 0),
       (SELECT count(*) FROM gateway_task_attempts attempt WHERE attempt.task_id = task.id),
       (SELECT count(*) FROM gateway_task_admissions admission WHERE admission.task_id = task.id),
       (SELECT count(*) FROM gateway_concurrency_leases lease WHERE lease.task_id = task.id AND lease.released_at IS NULL),
       (SELECT count(*) FROM gateway_task_events event WHERE event.task_id = task.id AND event.event_type = 'task.failed')
FROM gateway_tasks task
WHERE task.id = $1::uuid`, poisonTaskID).Scan(
		&poisonCode,
		&poisonRiverJobID,
		&poisonAttempts,
		&poisonAdmissions,
		&poisonLeases,
		&poisonFailedEvents,
	); err != nil {
		t.Fatalf("read poison task terminal state: %v", err)
	}
	if poisonCode != "no_model_candidate" || poisonRiverJobID != 0 || poisonAttempts != 0 ||
		poisonAdmissions != 0 || poisonLeases != 0 || poisonFailedEvents != 1 {
		t.Fatalf(
			"poison state code=%s river=%v attempts=%d admissions=%d leases=%d events=%d",
			poisonCode,
			poisonRiverJobID,
			poisonAttempts,
			poisonAdmissions,
			poisonLeases,
			poisonFailedEvents,
		)
	}
}

func TestAsyncWorkerThousandConcurrentSchedulingAcceptance(t *testing.T) {
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run thousand-task worker acceptance tests")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()
	applyMigration(t, ctx, databaseURL)

	apiDB, err := store.ConnectWithMaxConns(ctx, databaseURL, 16)
	if err != nil {
		t.Fatalf("connect API store: %v", err)
	}
	defer apiDB.Close()
	workerDB, err := store.ConnectWithMaxConns(ctx, databaseURL, 24)
	if err != nil {
		t.Fatalf("connect Worker store: %v", err)
	}
	defer workerDB.Close()
	apiCoordinationDB, err := store.ConnectWithMaxConns(ctx, databaseURL, 4)
	if err != nil {
		t.Fatalf("connect API critical store: %v", err)
	}
	defer apiCoordinationDB.Close()
	workerCoordinationDB, err := store.ConnectWithMaxConns(ctx, databaseURL, 4)
	if err != nil {
		t.Fatalf("connect Worker critical store: %v", err)
	}
	defer workerCoordinationDB.Close()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect acceptance pool: %v", err)
	}
	defer pool.Close()
	restoreAcceptanceState := isolateHTTPAcceptanceState(t, ctx, pool, false)
	defer restoreAcceptanceState()

	serverCtx, cancelServer := context.WithCancel(ctx)
	defer cancelServer()
	server := httptest.NewServer(NewServerWithStores(serverCtx, config.Config{
		AppEnv:                             "test",
		HTTPAddr:                           ":0",
		DatabaseURL:                        databaseURL,
		IdentityMode:                       "hybrid",
		JWTSecret:                          "test-secret",
		BillingEngineMode:                  "observe",
		CORSAllowedOrigin:                  "*",
		ProcessRole:                        "api",
		AsyncQueueWorkerEnabled:            true,
		AsyncAdmissionDispatcherEnabled:    true,
		AsyncAdmissionDispatcherConfigured: true,
		AsyncWorkerHardLimit:               1000,
		AsyncWorkerInstanceHardLimit:       1000,
		AsyncWorkerRefreshIntervalSeconds:  1,
	}, apiDB, apiCoordinationDB, apiDB, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))))
	defer server.Close()
	workerServer := httptest.NewServer(NewServerWithStores(serverCtx, config.Config{
		AppEnv:                             "test",
		HTTPAddr:                           ":0",
		DatabaseURL:                        databaseURL,
		IdentityMode:                       "hybrid",
		JWTSecret:                          "test-secret",
		BillingEngineMode:                  "observe",
		CORSAllowedOrigin:                  "*",
		ProcessRole:                        "worker",
		AsyncQueueWorkerEnabled:            true,
		AsyncAdmissionDispatcherEnabled:    false,
		AsyncAdmissionDispatcherConfigured: true,
		AsyncWorkerHardLimit:               1000,
		AsyncWorkerInstanceHardLimit:       1000,
		AsyncWorkerRefreshIntervalSeconds:  1,
	}, workerDB, workerCoordinationDB, workerDB, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))))
	defer workerServer.Close()

	adminToken := createAsyncAcceptanceAdmin(t, ctx, pool, server.URL)
	if _, err := pool.Exec(ctx, `
UPDATE gateway_user_groups
SET rate_limit_policy = '{"rules":[{"metric":"concurrent","limit":1024,"leaseTtlSeconds":120}]}'::jsonb
WHERE status = 'active'`); err != nil {
		t.Fatalf("raise acceptance user-group concurrency: %v", err)
	}
	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
	model := "worker-thousand-" + suffix
	platform := createAsyncAcceptancePlatform(t, server.URL, adminToken, "thousand-"+suffix, "Thousand Task Simulation", 1000)
	defer updateAsyncAcceptancePlatform(t, server.URL, adminToken, platform, 1000, "disabled")
	modelID := createAsyncAcceptanceModel(t, server.URL, adminToken, platform.ID, model, "video_generate", "override", 1000, 120)

	waitForAsyncWorkerMetric(t, workerServer.URL, "easyai_gateway_async_worker_capacity", 1000, 15*time.Second)
	taskIDs := submitAsyncSimulationTasks(t, server.URL, adminToken, "/api/v1/videos/generations", model, 1000, 45*time.Second)
	peakRunning, peakLeases := waitForAcceptanceTasks(t, ctx, pool, taskIDs, modelID, 2*time.Minute)
	if peakRunning < 800 || peakLeases < 800 {
		t.Fatalf("thousand-task peak running=%d leases=%d, want both >=800", peakRunning, peakLeases)
	}
	if peakLeases > 1000 {
		t.Fatalf("thousand-task lease peak=%d exceeded model concurrent limit 1000", peakLeases)
	}
	assertAcceptanceAttempts(t, ctx, pool, taskIDs, len(taskIDs))
	t.Logf(
		"千级并发证据: tasks=%d duration=%s running_peak=%d lease_peak=%d api_database_max_connections=16 api_critical_max_connections=4 worker_database_max_connections=24 worker_critical_max_connections=4",
		len(taskIDs),
		45*time.Second,
		peakRunning,
		peakLeases,
	)
}

func TestAsyncWorkerDynamicConcurrencyAcceptance(t *testing.T) {
	databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run dynamic worker acceptance tests")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 7*time.Minute)
	defer cancel()
	applyMigration(t, ctx, databaseURL)

	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect store: %v", err)
	}
	defer db.Close()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect acceptance pool: %v", err)
	}
	defer pool.Close()
	restoreAcceptanceState := isolateHTTPAcceptanceState(t, ctx, pool, false)
	defer restoreAcceptanceState()

	serverCtx, cancelServer := context.WithCancel(ctx)
	defer cancelServer()
	server := httptest.NewServer(NewServerWithContext(serverCtx, config.Config{
		AppEnv:                            "test",
		HTTPAddr:                          ":0",
		DatabaseURL:                       databaseURL,
		IdentityMode:                      "hybrid",
		JWTSecret:                         "test-secret",
		BillingEngineMode:                 "observe",
		CORSAllowedOrigin:                 "*",
		AsyncQueueWorkerEnabled:           true,
		AsyncWorkerHardLimit:              256,
		AsyncWorkerInstanceHardLimit:      256,
		AsyncWorkerRefreshIntervalSeconds: 1,
	}, db, slog.New(slog.NewTextHandler(io.Discard, nil))))
	defer server.Close()

	adminToken := createAsyncAcceptanceAdmin(t, ctx, pool, server.URL)
	if _, err := pool.Exec(ctx, `
UPDATE gateway_user_groups
SET rate_limit_policy = '{"rules":[{"metric":"concurrent","limit":256,"leaseTtlSeconds":120}]}'::jsonb
WHERE status = 'active'`); err != nil {
		t.Fatalf("raise acceptance user-group concurrency: %v", err)
	}
	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)

	t.Run("超过旧64上限的高并发", func(t *testing.T) {
		model := "worker-burst-" + suffix
		platform := createAsyncAcceptancePlatform(t, server.URL, adminToken, "burst-"+suffix, "Burst Simulation", 128)
		defer updateAsyncAcceptancePlatform(t, server.URL, adminToken, platform, 128, "disabled")
		modelID := createAsyncAcceptanceModel(t, server.URL, adminToken, platform.ID, model, "video_generate", "override", 96, 120)

		waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_capacity", 96, 15*time.Second)
		startedAt := time.Now()
		taskIDs := submitAsyncSimulationTasks(t, server.URL, adminToken, "/api/v1/videos/generations", model, 128, 15*time.Second)
		peakRunning, peakLeases := waitForAcceptanceTasks(t, ctx, pool, taskIDs, modelID, 60*time.Second)
		elapsed := time.Since(startedAt)
		if peakRunning < 80 || peakLeases < 80 {
			t.Fatalf("running peak=%d lease peak=%d, want both >=80", peakRunning, peakLeases)
		}
		if peakLeases > 96 {
			t.Fatalf("lease peak=%d exceeded model concurrent limit 96", peakLeases)
		}
		assertAcceptanceAttempts(t, ctx, pool, taskIDs, len(taskIDs))
		if elapsed >= 60*time.Second {
			t.Fatalf("128 tasks completed in %s, want <60s", elapsed)
		}
		t.Logf("高并发证据: tasks=128 duration=%s running_peak=%d lease_peak=%d capacity=96", elapsed.Round(time.Millisecond), peakRunning, peakLeases)
	})

	t.Run("三分钟视频任务在线扩缩容与续租", func(t *testing.T) {
		model := "worker-video-long-" + suffix
		platform := createAsyncAcceptancePlatform(t, server.URL, adminToken, "video-long-"+suffix, "Long Video Simulation", 1)
		modelID := createAsyncAcceptanceModel(t, server.URL, adminToken, platform.ID, model, "video_generate", "inherit", 0, 120)
		waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_capacity", 1, 15*time.Second)

		longTaskIDs := submitAsyncSimulationTasks(t, server.URL, adminToken, "/api/v1/videos/generations", model, 3, 180*time.Second)
		waitForTaskCount(t, ctx, pool, longTaskIDs, "running", 1, 10*time.Second)
		updateStartedAt := time.Now()
		updateAsyncAcceptancePlatform(t, server.URL, adminToken, platform, 3, "enabled")
		waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_desired_capacity", 3, 15*time.Second)
		waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_capacity", 3, 15*time.Second)
		waitForActiveLeaseCount(t, ctx, pool, modelID, 3, 15*time.Second)
		waitForTaskCount(t, ctx, pool, longTaskIDs, "running", 3, 5*time.Second)
		if time.Since(updateStartedAt) > 15*time.Second {
			t.Fatalf("capacity expansion took %s, want <=15s", time.Since(updateStartedAt))
		}

		initialExpiry := activeLeaseExpiry(t, ctx, pool, modelID, 3)
		updateAsyncAcceptancePlatform(t, server.URL, adminToken, platform, 1, "enabled")
		waitForAsyncWorkerMetric(t, server.URL, "easyai_gateway_async_worker_capacity", 1, 15*time.Second)
		shortTaskIDs := submitAsyncSimulationTasks(t, server.URL, adminToken, "/api/v1/videos/generations", model, 1, time.Second)
		time.Sleep(3 * time.Second)
		if attempts := taskAttemptCount(t, ctx, pool, shortTaskIDs); attempts != 0 {
			t.Fatalf("short task created %d upstream attempts while three long leases exceeded reduced limit", attempts)
		}
		waitForTaskCount(t, ctx, pool, longTaskIDs, "running", 3, 5*time.Second)

		time.Sleep(130 * time.Second)
		renewedExpiry := activeLeaseExpiry(t, ctx, pool, modelID, 3)
		if !renewedExpiry.After(initialExpiry.Add(30*time.Second)) || !renewedExpiry.After(time.Now()) {
			t.Fatalf("lease was not renewed: initial=%s renewed=%s now=%s", initialExpiry, renewedExpiry, time.Now())
		}

		waitForTaskCount(t, ctx, pool, longTaskIDs, "succeeded", 3, 75*time.Second)
		shortStartedAt := time.Now()
		waitForTaskAttempts(t, ctx, pool, shortTaskIDs, 1, 5*time.Second)
		waitForTaskCount(t, ctx, pool, shortTaskIDs, "succeeded", 1, 10*time.Second)
		if time.Since(shortStartedAt) > 5*time.Second {
			t.Fatalf("short task started after %s once long tasks completed, want <=5s", time.Since(shortStartedAt))
		}
		allTaskIDs := append(append([]string{}, longTaskIDs...), shortTaskIDs...)
		assertAcceptanceAttempts(t, ctx, pool, allTaskIDs, 4)
		assertSimulationOutputsAndEvents(t, ctx, pool, allTaskIDs)
		t.Logf("长任务证据: initial_capacity=1 expanded_capacity=3 reduced_capacity=1 initial_expiry=%s renewed_expiry=%s tasks=4", initialExpiry.UTC().Format(time.RFC3339Nano), renewedExpiry.UTC().Format(time.RFC3339Nano))
	})
}

type asyncAcceptancePlatform struct {
	ID          string `json:"id"`
	Provider    string `json:"provider"`
	PlatformKey string `json:"platformKey"`
	Name        string `json:"name"`
	BaseURL     string `json:"baseUrl"`
	AuthType    string `json:"authType"`
	Priority    int    `json:"priority"`
}

func createAsyncAcceptanceAdmin(t *testing.T, ctx context.Context, pool *pgxpool.Pool, baseURL string) string {
	t.Helper()
	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
	username := "async_acceptance_" + suffix
	password := "password123"
	doJSON(t, baseURL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username": username,
		"email":    username + "@example.com",
		"password": password,
	}, http.StatusCreated, &map[string]any{})
	if _, err := pool.Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil {
		t.Fatalf("promote acceptance user: %v", err)
	}
	var login struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, baseURL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
		"account": username, "password": password,
	}, http.StatusOK, &login)
	if login.AccessToken == "" {
		t.Fatal("acceptance login returned no access token")
	}
	return login.AccessToken
}

func createAsyncAcceptancePlatform(t *testing.T, baseURL, token, key, name string, concurrent int) asyncAcceptancePlatform {
	t.Helper()
	var platform asyncAcceptancePlatform
	doJSON(t, baseURL, http.MethodPost, "/api/admin/platforms", token, asyncAcceptancePlatformPayload(key, name, concurrent, "enabled"), http.StatusCreated, &platform)
	if platform.ID == "" {
		t.Fatal("acceptance platform returned no id")
	}
	return platform
}

func updateAsyncAcceptancePlatform(t *testing.T, baseURL, token string, platform asyncAcceptancePlatform, concurrent int, status string) {
	t.Helper()
	doJSON(t, baseURL, http.MethodPatch, "/api/admin/platforms/"+platform.ID, token,
		asyncAcceptancePlatformPayload(platform.PlatformKey, platform.Name, concurrent, status), http.StatusOK, &asyncAcceptancePlatform{})
}

func asyncAcceptancePlatformPayload(key, name string, concurrent int, status string) map[string]any {
	return map[string]any{
		"provider":    "openai",
		"platformKey": key,
		"name":        name,
		"baseUrl":     "https://api.openai.com/v1",
		"authType":    "bearer",
		"credentials": map[string]any{"mode": "simulation"},
		"config":      map[string]any{"testMode": true},
		"priority":    1,
		"status":      status,
		"rateLimitPolicy": map[string]any{"rules": []any{
			map[string]any{"metric": "concurrent", "limit": concurrent, "leaseTtlSeconds": 120},
			map[string]any{"metric": "queue_size", "limit": 512, "maxWaitSeconds": 600},
		}},
	}
}

func createAsyncAcceptanceModel(t *testing.T, baseURL, token, platformID, model, modelType, mode string, concurrent, ttl int) string {
	t.Helper()
	payload := map[string]any{
		"modelName":           model,
		"providerModelName":   model,
		"modelAlias":          model,
		"modelType":           []string{modelType},
		"displayName":         model,
		"rateLimitPolicyMode": mode,
	}
	if mode == "override" {
		payload["rateLimitPolicy"] = map[string]any{"rules": []any{
			map[string]any{"metric": "concurrent", "limit": concurrent, "leaseTtlSeconds": ttl},
			map[string]any{"metric": "queue_size", "limit": 512, "maxWaitSeconds": 600},
		}}
	}
	var response struct {
		ID string `json:"id"`
	}
	doJSON(t, baseURL, http.MethodPost, "/api/admin/platforms/"+platformID+"/models", token, payload, http.StatusCreated, &response)
	if response.ID == "" {
		t.Fatal("acceptance model returned no id")
	}
	return response.ID
}

func submitAsyncSimulationTasks(t *testing.T, baseURL, token, path, model string, count int, duration time.Duration) []string {
	t.Helper()
	ids := make([]string, count)
	errs := make(chan error, count)
	var wg sync.WaitGroup
	for index := 0; index < count; index++ {
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			payload := map[string]any{
				"model":                model,
				"runMode":              "simulation",
				"simulation":           true,
				"simulationDurationMs": duration.Milliseconds(),
				"prompt":               fmt.Sprintf("async acceptance %d", index),
				"input":                fmt.Sprintf("async acceptance %d", index),
			}
			raw, _ := json.Marshal(payload)
			req, err := http.NewRequest(http.MethodPost, baseURL+path, bytes.NewReader(raw))
			if err != nil {
				errs <- err
				return
			}
			req.Header.Set("Content-Type", "application/json")
			req.Header.Set("Authorization", "Bearer "+token)
			req.Header.Set("X-Async", "true")
			resp, err := http.DefaultClient.Do(req)
			if err != nil {
				errs <- err
				return
			}
			defer resp.Body.Close()
			responseBody, _ := io.ReadAll(resp.Body)
			if resp.StatusCode != http.StatusAccepted {
				errs <- fmt.Errorf("submit task %d status=%d body=%s", index, resp.StatusCode, responseBody)
				return
			}
			var response struct {
				TaskID string `json:"taskId"`
			}
			if err := json.Unmarshal(responseBody, &response); err != nil || response.TaskID == "" {
				errs <- fmt.Errorf("decode task %d: %w body=%s", index, err, responseBody)
				return
			}
			ids[index] = response.TaskID
		}(index)
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		t.Fatal(err)
	}
	return ids
}

func waitForAcceptanceTasks(t *testing.T, ctx context.Context, pool *pgxpool.Pool, taskIDs []string, modelID string, timeout time.Duration) (int64, int64) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	var peakRunning, peakLeases int64
	var lastRunning, lastSucceeded, lastFailed, lastLeases int64
	for time.Now().Before(deadline) {
		var running, succeeded, failed, leases int64
		if err := pool.QueryRow(ctx, `
SELECT COUNT(*) FILTER (WHERE status = 'running'),
       COUNT(*) FILTER (WHERE status = 'succeeded'),
       COUNT(*) FILTER (WHERE status IN ('failed', 'cancelled', 'manual_review'))
FROM gateway_tasks
WHERE id = ANY($1::uuid[])`, taskIDs).Scan(&running, &succeeded, &failed); err != nil {
			t.Fatalf("read acceptance tasks: %v", err)
		}
		if err := pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_concurrency_leases
WHERE scope_type = 'platform_model'
  AND scope_key = $1
  AND released_at IS NULL
  AND expires_at > now()`, modelID).Scan(&leases); err != nil {
			t.Fatalf("read acceptance leases: %v", err)
		}
		peakRunning = maxInt64(peakRunning, running)
		peakLeases = maxInt64(peakLeases, leases)
		lastRunning = running
		lastSucceeded = succeeded
		lastFailed = failed
		lastLeases = leases
		if failed > 0 {
			t.Fatalf("acceptance tasks entered failed/manual status: %d", failed)
		}
		if succeeded == int64(len(taskIDs)) {
			return peakRunning, peakLeases
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf(
		"tasks did not finish within %s: running=%d succeeded=%d failed=%d leases=%d peak_running=%d peak_leases=%d total=%d",
		timeout,
		lastRunning,
		lastSucceeded,
		lastFailed,
		lastLeases,
		peakRunning,
		peakLeases,
		len(taskIDs),
	)
	return 0, 0
}

func waitForTaskCount(t *testing.T, ctx context.Context, pool *pgxpool.Pool, taskIDs []string, status string, count int, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		var got int
		if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM gateway_tasks WHERE id = ANY($1::uuid[]) AND status = $2`, taskIDs, status).Scan(&got); err != nil {
			t.Fatalf("read task count: %v", err)
		}
		if got == count {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	rows, err := pool.Query(ctx, `
SELECT task.status, COALESCE(task.error_code, ''), count(*)
FROM gateway_tasks task
WHERE task.id = ANY($1::uuid[])
GROUP BY task.status, COALESCE(task.error_code, '')
ORDER BY task.status, COALESCE(task.error_code, '')`, taskIDs)
	diagnostics := make([]string, 0)
	if err == nil {
		defer rows.Close()
		for rows.Next() {
			var taskStatus, code string
			var taskCount int
			if scanErr := rows.Scan(&taskStatus, &code, &taskCount); scanErr == nil {
				diagnostics = append(diagnostics, fmt.Sprintf("%s/%s=%d", taskStatus, code, taskCount))
			}
		}
	}
	t.Fatalf("task status %s count did not reach %d within %s; observed=%v", status, count, timeout, diagnostics)
}

func activeLeaseExpiry(t *testing.T, ctx context.Context, pool *pgxpool.Pool, modelID string, wantCount int) time.Time {
	t.Helper()
	var count int
	var earliest time.Time
	if err := pool.QueryRow(ctx, `
SELECT COUNT(*), MIN(expires_at)
FROM gateway_concurrency_leases
WHERE scope_type = 'platform_model'
  AND scope_key = $1
  AND released_at IS NULL
  AND expires_at > now()`, modelID).Scan(&count, &earliest); err != nil {
		t.Fatalf("read active lease expiry: %v", err)
	}
	if count != wantCount {
		t.Fatalf("active lease count=%d, want=%d", count, wantCount)
	}
	return earliest
}

func waitForActiveLeaseCount(t *testing.T, ctx context.Context, pool *pgxpool.Pool, modelID string, count int, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		var got int
		if err := pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_concurrency_leases
WHERE scope_type = 'platform_model'
  AND scope_key = $1
  AND released_at IS NULL
  AND expires_at > now()`, modelID).Scan(&got); err != nil {
			t.Fatalf("read active lease count: %v", err)
		}
		if got == count {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatalf("active lease count did not reach %d within %s", count, timeout)
}

func taskAttemptCount(t *testing.T, ctx context.Context, pool *pgxpool.Pool, taskIDs []string) int {
	t.Helper()
	var count int
	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM gateway_task_attempts WHERE task_id = ANY($1::uuid[])`, taskIDs).Scan(&count); err != nil {
		t.Fatalf("count task attempts: %v", err)
	}
	return count
}

func waitForTaskAttempts(t *testing.T, ctx context.Context, pool *pgxpool.Pool, taskIDs []string, count int, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if taskAttemptCount(t, ctx, pool, taskIDs) == count {
			return
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf("attempt count did not reach %d within %s", count, timeout)
}

func assertAcceptanceAttempts(t *testing.T, ctx context.Context, pool *pgxpool.Pool, taskIDs []string, want int) {
	t.Helper()
	var attempts, duplicateTasks, manualReview int
	if err := pool.QueryRow(ctx, `
SELECT COUNT(*),
       COUNT(DISTINCT task_id) FILTER (WHERE per_task.attempts > 1),
       COUNT(*) FILTER (WHERE status = 'manual_review')
FROM (
  SELECT task_id, COUNT(*) AS attempts, MAX(status) AS status
  FROM gateway_task_attempts
  WHERE task_id = ANY($1::uuid[])
  GROUP BY task_id
) per_task`, taskIDs).Scan(&attempts, &duplicateTasks, &manualReview); err != nil {
		t.Fatalf("read attempt acceptance evidence: %v", err)
	}
	if attempts != want || duplicateTasks != 0 || manualReview != 0 {
		t.Fatalf("attempts=%d duplicate_tasks=%d manual_review=%d, want %d/0/0", attempts, duplicateTasks, manualReview, want)
	}
}

func assertSimulationOutputsAndEvents(t *testing.T, ctx context.Context, pool *pgxpool.Pool, taskIDs []string) {
	t.Helper()
	var complete int
	if err := pool.QueryRow(ctx, `
SELECT COUNT(*)
FROM gateway_tasks task
WHERE task.id = ANY($1::uuid[])
  AND task.status = 'succeeded'
  AND task.finished_at IS NOT NULL
  AND task.result IS NOT NULL
  AND task.result <> '{}'::jsonb
  AND task.metrics IS NOT NULL
  AND jsonb_typeof(task.billings) = 'array'
  AND jsonb_array_length(task.billings) > 0
  AND EXISTS (SELECT 1 FROM gateway_task_events event WHERE event.task_id = task.id AND event.event_type = 'task.completed')`,
		taskIDs).Scan(&complete); err != nil {
		t.Fatalf("read simulation output evidence: %v", err)
	}
	if complete != len(taskIDs) {
		t.Fatalf("complete simulation outputs/events=%d, want=%d", complete, len(taskIDs))
	}
}

func waitForAsyncWorkerMetric(t *testing.T, baseURL, metric string, want int64, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		resp, err := http.Get(baseURL + "/metrics")
		if err == nil {
			raw, _ := io.ReadAll(resp.Body)
			resp.Body.Close()
			if resp.StatusCode == http.StatusOK {
				for _, line := range strings.Split(string(raw), "\n") {
					fields := strings.Fields(line)
					if len(fields) == 2 && fields[0] == metric {
						value, _ := strconv.ParseInt(fields[1], 10, 64)
						if value == want {
							return
						}
					}
				}
			}
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatalf("metric %s did not reach %d within %s", metric, want, timeout)
}

func maxInt64(left, right int64) int64 {
	if right > left {
		return right
	}
	return left
}
