package http_test

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/go-resty/resty/v2"
	pkghttp "github.com/tidbcloud/dataflow-service-ng/pkg/http"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

func TestHTTPDiagnosticsDoNotExposeCredentials(t *testing.T) {
	const secret = "canary-http-secret"
	const reason = "Cluster authorization failed"
	request := map[string]string{"sink_uri": "kafka://broker/topic?sasl-password=" + secret, "key": secret}
	inner, _ := json.Marshal(map[string]string{"error_msg": request["sink_uri"] + ": " + reason, "error_code": "CDC:ErrSinkURIInvalid"})
	payload, _ := json.Marshal(map[string]string{"message": "Failed to create changefeed: " + string(inner)})
	for _, status := range []int{http.StatusOK, http.StatusInternalServerError} {
		t.Run(http.StatusText(status), func(t *testing.T) {
			var calls atomic.Int32
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				calls.Add(1)
				body, _ := io.ReadAll(r.Body)
				var received map[string]string
				if json.Unmarshal(body, &received) != nil || received["sink_uri"] != request["sink_uri"] || received["key"] != secret || r.URL.Query().Get("token") != secret {
					t.Error("redaction changed the actual request")
				}
				w.WriteHeader(status)
				_, _ = w.Write(payload)
			}))
			defer server.Close()
			core, entries := observer.New(zap.DebugLevel)
			ctx := log.WithLogger(context.Background(), zap.New(core))
			client := pkghttp.NewRestyCli(pkghttp.RestyCliWithDefaultRetry(), func(c *resty.Client) {
				c.SetRetryCount(1).SetRetryWaitTime(time.Millisecond).SetRetryMaxWaitTime(time.Millisecond)
			})
			body, err := client.Call(ctx, http.MethodPost, server.URL, nil, url.Values{"token": {secret}}, request, regionalapi.ErrorResponse2BizErr)
			if status == http.StatusOK {
				if err != nil || string(body) != string(payload) {
					t.Fatal("successful response changed")
				}
			} else {
				if err == nil || strings.Contains(err.Error(), secret) || !strings.Contains(err.ErrMsg(), reason) {
					t.Fatal("returned error leaked credentials or lost its cause")
				}
				if calls.Load() != 2 {
					t.Fatalf("retry path not exercised: calls=%d", calls.Load())
				}
			}
			if entries.Len() == 0 {
				t.Fatal("no logs captured")
			}
			for _, entry := range entries.All() {
				requestBody, ok := entry.ContextMap()["request_body"].(string)
				if !ok || !strings.Contains(requestBody, "kafka://broker/topic") {
					t.Fatal("normal and retry logs must preserve safe request details")
				}
				fields, _ := json.Marshal(entry.ContextMap())
				if strings.Contains(string(fields), secret) || strings.Contains(entry.Message, secret) {
					t.Fatal("credentials leaked through HTTP diagnostics")
				}
			}
		})
	}
}

func TestHTTPTransportErrorRedactsURL(t *testing.T) {
	const secret = "canary-transport-secret"
	// Resty's built-in logger bypasses the context logger. Capture stderr too,
	// otherwise this regression would pass while the library still leaked URLs.
	stderr, fileErr := os.CreateTemp(t.TempDir(), "resty-stderr")
	if fileErr != nil {
		t.Fatal(fileErr)
	}
	previousStderr := os.Stderr
	os.Stderr = stderr
	defer func() {
		os.Stderr = previousStderr
		_ = stderr.Close()
	}()
	server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
	server.Close()
	core, entries := observer.New(zap.DebugLevel)
	ctx := log.WithLogger(context.Background(), zap.New(core))
	client := pkghttp.NewRestyCli(pkghttp.RestyCliWithDefaultRetry(), func(c *resty.Client) {
		c.SetDebug(true).SetRetryCount(1).SetRetryWaitTime(time.Millisecond).SetRetryMaxWaitTime(time.Millisecond)
	})
	_, err := client.Call(ctx, http.MethodGet, server.URL+"?token="+secret, nil, nil, nil, regionalapi.ErrorResponse2BizErr)
	if err == nil || strings.Contains(err.Error(), secret) || !strings.Contains(err.Error(), "connection refused") {
		t.Fatal("transport error leaked a credential or lost its cause")
	}
	for _, entry := range entries.All() {
		fields, _ := json.Marshal(entry.ContextMap())
		if strings.Contains(string(fields), secret) {
			t.Fatal("transport log leaked a credential")
		}
	}
	internalLogs, readErr := os.ReadFile(stderr.Name())
	if readErr != nil {
		t.Fatal(readErr)
	}
	if strings.Contains(string(internalLogs), secret) {
		t.Fatal("Resty's internal logger leaked a credential")
	}
}