//go:build e2e
// +build e2e

package cases

import (
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/pingcap/log"
	"github.com/stretchr/testify/assert"

	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/test/ft/utils"
)

// integrationTaskBaseURL returns the base URL for the internal monitoring pipeline APIs.
// Route: /:bizType/v1beta1/internal/clusters/:applicationID/monitoringPipelines
func integrationTaskBaseURL() string {
	return fmt.Sprintf("%s/%s/v1beta1/internal/clusters/%s/monitoringPipelines",
		utils.GetService().ServerURL, utils.NexgenBizType, createdNextGenAppID)
}

// withInternalHeaders sets common headers required by the internal integration task API.
func withInternalHeaders(request *http.Request) {
	utils.WithDSAuthHeader(request)
	// x-org-id is required by ResolveRequestParameters (maps to tenantID)
	request.Header.Set(string(constants.OpenAPIHeaderOrgId), nextgenTenantID)
}

// createIntegrationTaskReq builds an HTTP POST request to create an integration task.
// IMPORTANT: We marshal only client-facing fields. Server-internal fields (clusterID,
// tenantID, bizType, userId) must NOT appear in the JSON body because the regional
// server extracts them from path params and headers. If included as empty strings,
// ShouldBindBodyWith will overwrite the path-param-derived values.
func createIntegrationTaskReq(req message.CreateIntegrationTaskReq) *http.Request {
	bodyMap := map[string]interface{}{
		"displayName": req.DisplayName,
		"dataType":    req.DataType,
	}
	if req.JobType != "" {
		bodyMap["jobType"] = req.JobType
	}
	if req.StartTime != "" {
		bodyMap["startTime"] = req.StartTime
	}
	if req.EndTime != "" {
		bodyMap["endTime"] = req.EndTime
	}
	if req.Frequency != "" {
		bodyMap["frequency"] = req.Frequency
	}
	if req.DataFormat != "" {
		bodyMap["dataFormat"] = req.DataFormat
	}
	if req.DataPrefix != "" {
		bodyMap["dataPrefix"] = req.DataPrefix
	}
	if req.AccessPrincipal != "" {
		bodyMap["accessPrincipal"] = req.AccessPrincipal
	}
	if len(req.RawLogComponents) > 0 {
		bodyMap["rawLogComponents"] = req.RawLogComponents
	}
	body, _ := json.Marshal(bodyMap)
	request, _ := http.NewRequest("POST", integrationTaskBaseURL(), strings.NewReader(string(body)))
	withInternalHeaders(request)
	return request
}

// queryIntegrationTasksReq builds an HTTP GET request to list integration tasks.
func queryIntegrationTasksReq(dataType string, pageSize int, pageToken string) *http.Request {
	url := integrationTaskBaseURL()
	params := []string{}
	if dataType != "" {
		params = append(params, fmt.Sprintf("dataType=%s", dataType))
	}
	if pageSize > 0 {
		params = append(params, fmt.Sprintf("pageSize=%d", pageSize))
	}
	if pageToken != "" {
		params = append(params, fmt.Sprintf("pageToken=%s", pageToken))
	}
	if len(params) > 0 {
		url = fmt.Sprintf("%s?%s", url, strings.Join(params, "&"))
	}
	request, _ := http.NewRequest("GET", url, nil)
	withInternalHeaders(request)
	return request
}

// detailIntegrationTaskReq builds an HTTP GET request to get integration task detail.
func detailIntegrationTaskReq(id string) *http.Request {
	url := fmt.Sprintf("%s/%s", integrationTaskBaseURL(), id)
	request, _ := http.NewRequest("GET", url, nil)
	withInternalHeaders(request)
	return request
}

// deleteIntegrationTaskReq builds an HTTP DELETE request to delete an integration task.
func deleteIntegrationTaskReq(id string) *http.Request {
	url := fmt.Sprintf("%s/%s", integrationTaskBaseURL(), id)
	request, _ := http.NewRequest("DELETE", url, nil)
	withInternalHeaders(request)
	return request
}

// deleteIntegrationTask is a helper that sends the DELETE and returns the response.
func deleteIntegrationTask(id string) *http.Response {
	resp, err := http.DefaultClient.Do(deleteIntegrationTaskReq(id))
	if err != nil {
		panic(err)
	}
	return resp
}

// cleanupExistingPipelines deletes all existing pipelines for the given dataType.
// This ensures the CRUD test starts from a clean state.
func cleanupExistingPipelines(t *testing.T, dataType string) {
	t.Helper()
	listReq := queryIntegrationTasksReq(dataType, 0, "")
	listResp, err := http.DefaultClient.Do(listReq)
	if err != nil {
		t.Logf("warning: failed to list existing pipelines for cleanup: %v", err)
		return
	}
	defer listResp.Body.Close()
	if listResp.StatusCode != http.StatusOK {
		return
	}
	var listResult message.QueryIntegrationTasksResp
	if err := json.NewDecoder(listResp.Body).Decode(&listResult); err != nil {
		t.Logf("warning: failed to decode list response for cleanup: %v", err)
		return
	}
	for _, p := range listResult.Pipelines {
		if p.DataType == dataType {
			delResp := deleteIntegrationTask(p.ID)
			delResp.Body.Close()
			t.Logf("cleaned up existing pipeline %s (dataType=%s)", p.ID, dataType)
		}
	}
}

// newCreateReq builds a CreateIntegrationTaskReq with all required fields populated.
func newCreateReq(displayName, dataType string) message.CreateIntegrationTaskReq {
	now := time.Now().UTC()
	req := message.CreateIntegrationTaskReq{
		DisplayName: displayName,
		DataType:    dataType,
		JobType:     "Once",
		StartTime:   now.Add(-24 * time.Hour).Format(time.RFC3339),
		EndTime:     now.Format(time.RFC3339),
	}
	if dataType == "rawlog" {
		req.RawLogComponents = []string{"tidb", "tikv", "pd"}
	}
	return req
}

// --- Tests ---

func TestIntegrationTaskAuth(t *testing.T) {
	integrationTaskURL := integrationTaskBaseURL()

	body, _ := json.Marshal(newCreateReq("ft-auth-test", "slowquery"))

	t.Run("empty token", func(t *testing.T) {
		log.Info("TestIntegrationTaskAuth-empty token start")
		defer log.Info("TestIntegrationTaskAuth-empty token end")

		request, _ := http.NewRequest("POST", integrationTaskURL, strings.NewReader(string(body)))
		request.Header.Set("Content-Type", "application/json")

		resp, err := http.DefaultClient.Do(request)
		assert.NoError(t, err)
		defer resp.Body.Close()
		// Server may return 401 (Unauthorized) or 404 (route not matched without auth context).
		assert.Contains(t, []int{http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound}, resp.StatusCode,
			"expected auth rejection status, got %d", resp.StatusCode)
	})

	t.Run("invalid token", func(t *testing.T) {
		log.Info("TestIntegrationTaskAuth-invalid token start")
		defer log.Info("TestIntegrationTaskAuth-invalid token end")

		request, _ := http.NewRequest("POST", integrationTaskURL, strings.NewReader(string(body)))
		request.Header.Set("Content-Type", "application/json")
		request.Header.Set(constants.AuthorizationHeader, fmt.Sprintf("%s %s", constants.BearerPrefix, "invalid-token"))

		resp, err := http.DefaultClient.Do(request)
		assert.NoError(t, err)
		defer resp.Body.Close()
		// Server may return 401 (Unauthorized) or 404 (route not matched without auth context).
		assert.Contains(t, []int{http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound}, resp.StatusCode,
			"expected auth rejection status, got %d", resp.StatusCode)
	})

	t.Run("valid token not 401", func(t *testing.T) {
		log.Info("TestIntegrationTaskAuth-valid token start")
		defer log.Info("TestIntegrationTaskAuth-valid token end")

		request, _ := http.NewRequest("POST", integrationTaskURL, strings.NewReader(string(body)))
		withInternalHeaders(request)

		resp, err := http.DefaultClient.Do(request)
		assert.NoError(t, err)
		defer resp.Body.Close()
		// With a valid token, the response should NOT be 401 Unauthorized.
		assert.NotEqual(t, http.StatusUnauthorized, resp.StatusCode)

		// Cleanup: if the request actually created a task, delete it.
		if resp.StatusCode == http.StatusOK {
			created := &message.CreateIntegrationTaskResp{}
			if err := json.NewDecoder(resp.Body).Decode(created); err == nil && created.ID != "" {
				defer func() {
					delResp := deleteIntegrationTask(created.ID)
					delResp.Body.Close()
				}()
				defer deleteIntegrationTask(created.ID)
			}
		}
	})
}

func TestIntegrationTaskCRUD(t *testing.T) {
	// This test covers the full lifecycle: Create → Detail → List → Delete
	// We test one dataType at a time to avoid "only one task per dataType" conflicts.

	type dataTypeCase struct {
		name             string
		dataType         string
		rawLogComponents []string
	}

	cases := []dataTypeCase{
		{name: "slowquery", dataType: "slowquery"},
		{name: "rawlog", dataType: "rawlog", rawLogComponents: []string{"tidb", "tikv", "pd"}},
		{name: "sqlstmt", dataType: "sqlstmt"},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			log.Info(fmt.Sprintf("TestIntegrationTaskCRUD-%s start", c.name))
			defer log.Info(fmt.Sprintf("TestIntegrationTaskCRUD-%s end", c.name))
			// Clean up any existing pipelines for this dataType from prior runs.
			cleanupExistingPipelines(t, c.dataType)

			// --- Create ---
			createReq := newCreateReq(fmt.Sprintf("ft-test-%s", c.dataType), c.dataType)
			request := createIntegrationTaskReq(createReq)
			resp, err := http.DefaultClient.Do(request)
			assert.NoError(t, err)
			defer resp.Body.Close()

			if !assert.Equal(t, http.StatusOK, resp.StatusCode, "create %s failed", c.dataType) {
				// Log response body for debugging
				respBody, _ := io.ReadAll(resp.Body)
				t.Logf("response body: %s", string(respBody))
				return
			}

			created := &message.CreateIntegrationTaskResp{}
			err = json.NewDecoder(resp.Body).Decode(created)
			assert.NoError(t, err, "failed to decode create response")
			assert.NotEmpty(t, created.ID, "created ID should not be empty")
			assert.Equal(t, c.dataType, created.DataType)
			assert.NotEmpty(t, created.DisplayName)

			// Ensure cleanup after all subtests
			defer func() {
				delResp := deleteIntegrationTask(created.ID)
				delResp.Body.Close()
			}()

			// --- Wait for Phase ACTIVE ---
			// The task starts in CREATING and transitions to ACTIVE asynchronously.
			// Poll the detail endpoint until Phase == ACTIVE (up to 60s).
			var detail *message.DetailIntegrationTaskResp
			t.Run("wait for active", func(t *testing.T) {
				deadline := time.Now().Add(60 * time.Second)
				ticker := time.NewTicker(2 * time.Second)
				defer ticker.Stop()
				for {
					detailReq := detailIntegrationTaskReq(created.ID)
					detailResp, err := http.DefaultClient.Do(detailReq)
					if !assert.NoError(t, err) {
						return
					}
					assert.Equal(t, http.StatusOK, detailResp.StatusCode, "detail should return 200")
					detail = &message.DetailIntegrationTaskResp{}
					err = json.NewDecoder(detailResp.Body).Decode(detail)
					detailResp.Body.Close()
					if !assert.NoError(t, err, "failed to decode detail response") {
						return
					}
					if detail.Phase == string(constants.ACTIVE) {
						t.Logf("task %s reached ACTIVE phase", created.ID)
						break
					}
					if time.Now().After(deadline) {
						t.Fatalf("task %s did not reach ACTIVE within 60s, current phase: %s", created.ID, detail.Phase)
						return
					}
					t.Logf("task %s phase is %s, waiting...", created.ID, detail.Phase)
					<-ticker.C
				}
			})

			// --- Detail (after ACTIVE) ---
			t.Run("detail", func(t *testing.T) {
				if detail == nil {
					t.Skip("skipping: task did not reach ACTIVE")
				}
				assert.Equal(t, created.ID, detail.ID)
				assert.Equal(t, c.dataType, detail.DataType)
				assert.NotEmpty(t, detail.DisplayName)
				assert.Equal(t, string(constants.ACTIVE), detail.Phase, "task phase should be ACTIVE")
			})

			// --- List ---
			t.Run("list", func(t *testing.T) {
				listReq := queryIntegrationTasksReq("", 0, "")
				listResp, err := http.DefaultClient.Do(listReq)
				assert.NoError(t, err)
				defer listResp.Body.Close()
				assert.Equal(t, http.StatusOK, listResp.StatusCode, "list should return 200")

				listResult := &message.QueryIntegrationTasksResp{}
				err = json.NewDecoder(listResp.Body).Decode(listResult)
				assert.NoError(t, err, "failed to decode list response")
				assert.NotNil(t, listResult.Pipelines)

				// Verify our created task appears in the list
				found := false
				for _, p := range listResult.Pipelines {
					if p.ID == created.ID {
						found = true
						assert.Equal(t, string(constants.ACTIVE), p.Phase, "listed task phase should be ACTIVE")
						break
					}
				}
				assert.True(t, found, "created task %s should appear in list", created.ID)
			})

			// --- List filtered by dataType ---
			t.Run("list filtered", func(t *testing.T) {
				listReq := queryIntegrationTasksReq(c.dataType, 0, "")
				listResp, err := http.DefaultClient.Do(listReq)
				assert.NoError(t, err)
				defer listResp.Body.Close()
				assert.Equal(t, http.StatusOK, listResp.StatusCode)

				listResult := &message.QueryIntegrationTasksResp{}
				err = json.NewDecoder(listResp.Body).Decode(listResult)
				assert.NoError(t, err, "failed to decode filtered list response")
				for _, p := range listResult.Pipelines {
					assert.Equal(t, c.dataType, p.DataType)
				}
			})

			// --- Delete ---
			t.Run("delete", func(t *testing.T) {
				// Create a second task specifically for deletion test.
				// Note: only one task per dataType is allowed, so we use the
				// same lifecycle — the deferred delete above is the real cleanup.
				// Here we verify the delete response on the already-created task.
				delReq := deleteIntegrationTaskReq(created.ID)
				delResp, err := http.DefaultClient.Do(delReq)
				assert.NoError(t, err)
				defer delResp.Body.Close()
				assert.Equal(t, http.StatusAccepted, delResp.StatusCode, "delete should return 202")
			})
		})
	}
}

func TestIntegrationTaskCreateValidation(t *testing.T) {
	type testCase struct {
		name string
		req  message.CreateIntegrationTaskReq
	}

	cases := []testCase{
		{
			name: "missing displayName",
			req: message.CreateIntegrationTaskReq{
				DataType:  "slowquery",
				JobType:   "Once",
				StartTime: time.Now().UTC().Add(-1 * time.Hour).Format(time.RFC3339),
			},
		},
		{
			name: "missing dataType",
			req: message.CreateIntegrationTaskReq{
				DisplayName: "ft-test-missing-datatype",
				JobType:     "Once",
				StartTime:   time.Now().UTC().Add(-1 * time.Hour).Format(time.RFC3339),
			},
		},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			log.Info("TestIntegrationTaskCreateValidation-" + c.name + " start")
			defer log.Info("TestIntegrationTaskCreateValidation-" + c.name + " end")

			request := createIntegrationTaskReq(c.req)
			resp, err := http.DefaultClient.Do(request)
			assert.NoError(t, err)
			defer resp.Body.Close()
			// The server should reject requests with missing required fields.
			// Binding errors map to MarshalError (500) and validation errors to ParameterInvalid (400).
			assert.GreaterOrEqual(t, resp.StatusCode, 400,
				"expected error status for %s, got %d", c.name, resp.StatusCode)
		})
	}
}

func TestIntegrationTaskDetailNotFound(t *testing.T) {
	log.Info("TestIntegrationTaskDetailNotFound start")
	defer log.Info("TestIntegrationTaskDetailNotFound end")

	req := detailIntegrationTaskReq("non-existent-id-12345")
	resp, err := http.DefaultClient.Do(req)
	assert.NoError(t, err)
	defer resp.Body.Close()
	assert.Equal(t, http.StatusNotFound, resp.StatusCode)
}

func TestIntegrationTaskDeleteNotFound(t *testing.T) {
	log.Info("TestIntegrationTaskDeleteNotFound start")
	defer log.Info("TestIntegrationTaskDeleteNotFound end")

	resp := deleteIntegrationTask("non-existent-id-12345")
	defer resp.Body.Close()
	assert.Equal(t, http.StatusNotFound, resp.StatusCode)
}