package httpapi

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

	"github.com/easyai/easyai-ai-gateway/apps/api/internal/config"
	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)

func TestKelingOmniCompatibleHTTPFlow(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 the Kling-compatible HTTP integration flow")
	}

	var upstreamTaskSequence atomic.Int64
	var upstreamPayloadMu sync.Mutex
	var upstreamPayloads []map[string]any
	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("Authorization") != "Bearer upstream-keling-key" {
			t.Fatalf("unexpected upstream Authorization: %q", r.Header.Get("Authorization"))
		}
		switch {
		case r.Method == http.MethodPost && r.URL.Path == "/videos/omni-video":
			var payload map[string]any
			if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
				t.Fatalf("decode upstream request: %v", err)
			}
			upstreamPayloadMu.Lock()
			upstreamPayloads = append(upstreamPayloads, payload)
			upstreamPayloadMu.Unlock()
			id := "upstream-" + strconv.FormatInt(upstreamTaskSequence.Add(1), 10)
			_ = json.NewEncoder(w).Encode(map[string]any{
				"code":       0,
				"request_id": "submit-" + id,
				"data":       map[string]any{"task_id": id, "task_status": "submitted"},
			})
		case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/videos/omni-video/upstream-"):
			id := strings.TrimPrefix(r.URL.Path, "/videos/omni-video/")
			_ = json.NewEncoder(w).Encode(map[string]any{
				"code":       0,
				"request_id": "poll-" + id,
				"data": map[string]any{
					"task_id":     id,
					"task_status": "succeed",
					"created_at":  time.Now().UnixMilli(),
					"task_result": map[string]any{"videos": []any{map[string]any{
						"id":            "video-" + id,
						"url":           "https://example.com/" + id + ".mp4",
						"watermark_url": "https://example.com/" + id + "-watermark.mp4",
						"duration":      "3",
					}}},
				},
			})
		default:
			t.Fatalf("unexpected upstream request %s %s", r.Method, r.URL.Path)
		}
	}))
	defer upstream.Close()

	ctx := context.Background()
	applyMigration(t, ctx, databaseURL)
	db, err := store.Connect(ctx, databaseURL)
	if err != nil {
		t.Fatalf("connect store: %v", err)
	}
	defer db.Close()

	suffix := strconv.FormatInt(time.Now().UnixNano(), 10)
	platform, err := db.CreatePlatform(ctx, store.CreatePlatformInput{
		Provider:    "keling",
		PlatformKey: "keling-compatible-test-" + suffix,
		Name:        "Kling Compatible Test",
		BaseURL:     upstream.URL,
		AuthType:    "APIKey",
		Credentials: map[string]any{"apiKey": "upstream-keling-key"},
		Config: map[string]any{
			"kelingPollIntervalMs":     100,
			"kelingPollTimeoutSeconds": 5,
		},
		Priority: 1,
		Status:   "enabled",
	})
	if err != nil {
		t.Fatalf("create test platform: %v", err)
	}
	_, err = db.CreatePlatformModel(ctx, store.CreatePlatformModelInput{
		PlatformID:        platform.ID,
		CanonicalModelKey: "keling:kling-video-o1",
		ModelName:         "kling-o1",
		ProviderModelName: "kling-video-o1",
		ModelAlias:        "kling-o1",
		ModelType:         store.StringList{"omni_video", "video_generate"},
		DisplayName:       "Kling O1 Compatible Test",
		Capabilities: map[string]any{
			"omni_video": map[string]any{
				"supported_modes":      []any{"text_to_video", "image_reference"},
				"output_resolutions":   []any{"720p", "1080p"},
				"aspect_ratio_allowed": []any{"16:9", "9:16", "1:1"},
				"duration_options":     []any{3, 4, 5, 6, 7, 8, 9, 10},
				"output_audio":         false,
				"max_images":           7,
			},
			"video_generate": map[string]any{
				"supported_modes":      []any{"text_to_video"},
				"output_resolutions":   []any{"720p", "1080p"},
				"aspect_ratio_allowed": []any{"16:9", "9:16", "1:1"},
				"duration_options":     []any{3, 4, 5, 6, 7, 8, 9, 10},
				"output_audio":         true,
			},
		},
		Enabled: true,
	})
	if err != nil {
		t.Fatalf("create test platform model: %v", err)
	}

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

	firstUserToken, firstAPIKey := createKelingCompatIntegrationUser(t, ctx, db, gateway.URL, "first", suffix, true)
	_ = firstUserToken
	_, secondAPIKey := createKelingCompatIntegrationUser(t, ctx, db, gateway.URL, "second", suffix, false)

	var created KelingCompatibleEnvelope
	doJSON(t, gateway.URL, http.MethodPost, "/v1/videos/omni-video", firstAPIKey, map[string]any{
		"model_name":       "kling-video-o1",
		"prompt":           "A clean product reveal",
		"mode":             "std",
		"aspect_ratio":     "16:9",
		"duration":         "3",
		"sound":            "off",
		"image_list":       []any{map[string]any{"image_url": "https://example.com/reference.png"}},
		"external_task_id": "compat-http-1",
	}, http.StatusOK, &created)
	createdData, _ := created.Data.(map[string]any)
	if created.Code != 0 || created.RequestID == "" || strings.TrimSpace(stringFromKelingCompat(createdData["task_id"])) == "" || createdData["task_status"] != "submitted" {
		t.Fatalf("unexpected compatible create response: %+v", created)
	}
	taskID := stringFromKelingCompat(createdData["task_id"])

	var hidden KelingCompatibleEnvelope
	doJSON(t, gateway.URL, http.MethodGet, "/v1/videos/omni-video/"+taskID, secondAPIKey, nil, http.StatusNotFound, &hidden)
	if hidden.Code != 1203 {
		t.Fatalf("cross-user task must be hidden: %+v", hidden)
	}

	completed := waitForKelingCompatTask(t, gateway.URL, firstAPIKey, taskID, 5*time.Second)
	if completed.Code != 0 {
		t.Fatalf("compatible task failed: %+v", completed)
	}
	completedData, _ := completed.Data.(map[string]any)
	if completedData["task_status"] != "succeed" {
		t.Fatalf("compatible task did not succeed: %+v", completedData)
	}
	taskResult, _ := completedData["task_result"].(map[string]any)
	videos, _ := taskResult["videos"].([]any)
	video, _ := videos[0].(map[string]any)
	if video["id"] == "" || video["watermark_url"] == "" || video["duration"] != "3" {
		t.Fatalf("compatible result lost video metadata: %+v", video)
	}

	var standard struct {
		TaskID string `json:"taskId"`
	}
	doJSONWithHeaders(t, gateway.URL, http.MethodPost, "/api/v1/videos/generations", firstAPIKey, map[string]any{
		"model":        "kling-o1",
		"prompt":       "A second product reveal",
		"resolution":   "720p",
		"aspect_ratio": "16:9",
		"duration":     3,
		"audio":        false,
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &standard)
	if standard.TaskID == "" {
		t.Fatal("standard video generation did not return taskId")
	}
	waitForTaskStatus(t, gateway.URL, firstAPIKey, standard.TaskID, []string{"succeeded"}, 5*time.Second)

	var unsupportedAudio struct {
		TaskID string `json:"taskId"`
	}
	doJSONWithHeaders(t, gateway.URL, http.MethodPost, "/api/v1/videos/generations", firstAPIKey, map[string]any{
		"model":        "kling-o1",
		"prompt":       "An O1 request that must not silently ignore audio",
		"resolution":   "1080p",
		"aspect_ratio": "9:16",
		"duration":     5,
		"audio":        true,
	}, map[string]string{"X-Async": "true"}, http.StatusAccepted, &unsupportedAudio)
	if unsupportedAudio.TaskID == "" {
		t.Fatal("unsupported O1 audio request did not return taskId")
	}
	waitForTaskStatus(t, gateway.URL, firstAPIKey, unsupportedAudio.TaskID, []string{"failed"}, 5*time.Second)
	var failedAudioTask store.GatewayTask
	doJSON(t, gateway.URL, http.MethodGet, "/api/v1/tasks/"+unsupportedAudio.TaskID, firstAPIKey, nil, http.StatusOK, &failedAudioTask)
	if failedAudioTask.ErrorCode != "invalid_parameter" || !strings.Contains(failedAudioTask.ErrorMessage, "does not support generated audio") {
		t.Fatalf("O1 audio request must fail visibly before upstream submission: %+v", failedAudioTask)
	}

	upstreamPayloadMu.Lock()
	defer upstreamPayloadMu.Unlock()
	if len(upstreamPayloads) != 2 {
		t.Fatalf("expected two upstream submissions, got %d", len(upstreamPayloads))
	}
	compatiblePayload := upstreamPayloads[0]
	if compatiblePayload["model_name"] != "kling-video-o1" || compatiblePayload["mode"] != "std" || compatiblePayload["sound"] != "off" || compatiblePayload["duration"] != "3" || compatiblePayload["aspect_ratio"] != "16:9" || compatiblePayload["external_task_id"] != "compat-http-1" {
		t.Fatalf("unexpected compatible upstream payload: %+v", compatiblePayload)
	}
}

func createKelingCompatIntegrationUser(t *testing.T, ctx context.Context, db *store.Store, baseURL string, prefix string, suffix string, fund bool) (string, string) {
	t.Helper()
	username := fmt.Sprintf("kling_compat_%s_%s", prefix, suffix)
	password := "password123"
	var registered struct {
		AccessToken string `json:"accessToken"`
	}
	doJSON(t, baseURL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{
		"username": username,
		"email":    username + "@example.com",
		"password": password,
	}, http.StatusCreated, &registered)
	var apiKey struct {
		Secret string `json:"secret"`
	}
	doJSON(t, baseURL, http.MethodPost, "/api/v1/api-keys", registered.AccessToken, map[string]any{
		"name":   "Kling compatible integration key",
		"scopes": []string{"video"},
	}, http.StatusCreated, &apiKey)
	if fund {
		if _, err := db.Pool().Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil {
			t.Fatalf("promote integration user: %v", err)
		}
		var loggedIn struct {
			AccessToken string `json:"accessToken"`
		}
		doJSON(t, baseURL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{
			"account":  username,
			"password": password,
		}, http.StatusOK, &loggedIn)
		var gatewayUserID string
		if err := db.Pool().QueryRow(ctx, `SELECT id::text FROM gateway_users WHERE username = $1`, username).Scan(&gatewayUserID); err != nil {
			t.Fatalf("read integration user id: %v", err)
		}
		doJSON(t, baseURL, http.MethodPatch, "/api/admin/users/"+gatewayUserID+"/wallet", loggedIn.AccessToken, map[string]any{
			"currency": "resource",
			"balance":  1000,
			"reason":   "seed Kling compatible integration wallet",
		}, http.StatusOK, nil)
	}
	return registered.AccessToken, apiKey.Secret
}

func waitForKelingCompatTask(t *testing.T, baseURL string, apiKey string, taskID string, timeout time.Duration) KelingCompatibleEnvelope {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		var response KelingCompatibleEnvelope
		doJSON(t, baseURL, http.MethodGet, "/v1/videos/omni-video/"+taskID, apiKey, nil, http.StatusOK, &response)
		data, _ := response.Data.(map[string]any)
		switch data["task_status"] {
		case "succeed":
			return response
		case "failed":
			t.Fatalf("Kling-compatible task failed: %+v", response)
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for Kling-compatible task %s", taskID)
	return KelingCompatibleEnvelope{}
}