package redact

import (
	"encoding/json"
	"strings"
	"testing"
)

func TestTextRedactsSinkErrors(t *testing.T) {
	const reason = "[CDC:ErrKafkaNewProducer] Cluster authorization failed"
	for name, uri := range map[string]string{
		"query":                "kafka://broker/topic?protocol=debezium&sasl-password=canary-secret&partition-num=24",
		"encoded password":     "kafka://broker/topic?sasl-password=canary%2Dsecret%26more%3Dvalue",
		"duplicate parameters": "kafka://broker/topic?sasl-password=canary-secret&sasl-password=second-secret",
		"escaped separators":   `kafka://broker/topic?protocol=debezium\u0026sasl-password=canary-secret\u0026enable-tls=false`,
		"invalid whitespace":   "kafka://broker/topic?sasl-password=canary-secret\nsecond-secret",
		"encoded parameter":    "kafka://broker/topic?%73asl-password=canary-secret",
		"userinfo":             "mysql://user:canary-secret@host/db",
		"invalid escape":       "kafka://broker/topic?sasl-password=canary-secret%zz",
		"truncated URI":        "kafka://broker/topic?sasl-password=canary-secret",
		"other sink":           "s3://bucket/path?secret-access-key=canary-secret#second-secret",
	} {
		t.Run(name, func(t *testing.T) {
			input := "[CDC:ErrSinkURIInvalid]sink uri invalid '" + uri + "': " + reason
			output := Text(input)
			for _, secret := range []string{"canary-secret", "canary%2Dsecret", "second-secret"} {
				if strings.Contains(output, secret) {
					t.Fatal("credential survived redaction")
				}
			}
			if !strings.Contains(output, reason) || !strings.Contains(output, "CDC:ErrSinkURIInvalid") {
				t.Fatalf("lost actionable error: %s", output)
			}
			if Text(output) != output {
				t.Fatal("redaction is not idempotent")
			}
		})
	}
}

func TestTextRedactsJSONAndKeys(t *testing.T) {
	inner, _ := json.Marshal(map[string]string{
		"error_msg":  "sink uri invalid 'kafka://broker/topic?sasl-password=canary-secret': Cluster authorization failed",
		"error_code": "CDC:ErrSinkURIInvalid",
	})
	outer, _ := json.Marshal(map[string]interface{}{
		"message": "Failed to create changefeed: " + string(inner),
		"details": []interface{}{map[string]string{"sslKeyContent": "private-material", "password": "plain-password"}},
		"id":      json.Number("9007199254740993"),
	})
	output := Text(string(outer))
	for _, secret := range []string{"canary-secret", "private-material", "plain-password"} {
		if strings.Contains(output, secret) {
			t.Fatal("JSON credential survived redaction")
		}
	}
	if !json.Valid([]byte(output)) || !strings.Contains(output, "9007199254740993") || !strings.Contains(output, "Cluster authorization failed") {
		t.Fatal("redaction changed error envelope or numeric precision")
	}
	for _, input := range []string{
		`{"password":"plain-password`,
		"-----BEGIN RSA PRIVATE KEY-----\nprivate-material\n-----END RSA PRIVATE KEY-----",
		`-----BEGIN PRIVATE KEY-----\nprivate-material`,
		"[ ",
	} {
		output := Text(input)
		if strings.Contains(output, "plain-password") || strings.Contains(output, "private-material") {
			t.Fatal("malformed payload or private key survived redaction")
		}
	}
}

func TestTextPreservesPlainErrors(t *testing.T) {
	for _, message := range []string{"http status 500", "", "[CDC:ErrChangeFeedAlreadyExists]changefeed already exists", "[file.go:214]code=61021000,msg=failed", "[REDACTED]"} {
		if Text(message) != message {
			t.Fatalf("changed non-sensitive message %q", message)
		}
	}
}