package server

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"reflect"
	"sort"
	"strings"
	"testing"

	"github.com/mem9-ai/drive9/pkg/meta"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

// migrationEventWireFields mirrors the wire contract of the removed Go SDK's
// MigrationEvent so the server-side field lists stay pinned to the documented
// migration event schema.
type migrationEventWireFields struct {
	EventID              string `json:"event_id"`
	EmittedAt            string `json:"emitted_at"`
	Phase                string `json:"phase"`
	RoundID              string `json:"round_id"`
	CASAttempt           int64  `json:"cas_attempt"`
	FirstSeenAt          string `json:"first_seen_at"`
	GraceSeconds         int64  `json:"grace_seconds"`
	JobID                string `json:"job_id"`
	VolumeID             string `json:"volume_id"`
	NodeName             string `json:"node_name"`
	PodName              string `json:"pod_name"`
	SpaceID              string `json:"space_id"`
	SourcePath           string `json:"source_path"`
	TargetPath           string `json:"target_path"`
	SourceVersionToken   string `json:"source_version_token"`
	Size                 int64  `json:"size"`
	Mtime                int64  `json:"mtime"`
	SourceChecksumSHA256 string `json:"source_checksum_sha256"`
	ResourceID           string `json:"resource_id"`
	Revision             int64  `json:"revision"`
	Drive9ChecksumSHA256 string `json:"drive9_checksum_sha256"`
	ExpectedRevision     int64  `json:"expected_revision"`
	Operation            string `json:"operation"`
	Result               string `json:"result"`
	ErrorClass           string `json:"error_class"`
	LatencyMS            int64  `json:"latency_ms"`
}

func migrationEventFixture() map[string]any {
	return map[string]any{
		"event_id": "evt-1", "emitted_at": "2026-08-04T12:00:00Z", "phase": "DUAL_WRITE_REPAIRING",
		"round_id": "round-7", "cas_attempt": int64(1), "first_seen_at": "2026-08-04T11:59:00Z", "grace_seconds": int64(60),
		"job_id": "job-1", "volume_id": "vol-1", "node_name": "node-1", "pod_name": "pod-1", "space_id": "space-1",
		"source_path": "/ebs/full/path/file.bin", "target_path": "/drive9/full/path/file.bin", "source_version_token": "inode:1:2",
		"size": int64(4096), "mtime": int64(1_754_309_940_000_000_000), "source_checksum_sha256": strings.Repeat("a", 64),
		"resource_id": "resource-1", "revision": int64(3), "drive9_checksum_sha256": strings.Repeat("b", 64), "expected_revision": int64(2),
		"operation": "update", "result": "success", "error_class": "none", "latency_ms": int64(12),
	}
}

func migrationEventBody(t *testing.T, event map[string]any) []byte {
	t.Helper()
	body, err := json.Marshal(event)
	if err != nil {
		t.Fatal(err)
	}
	return body
}

func TestSupportedMigrationCapabilitiesReturnsFreshWireContract(t *testing.T) {
	first := supportedMigrationCapabilities()
	first.ChecksumRead = false
	got := supportedMigrationCapabilities()
	if !got.ChecksumRead || !got.ChecksumComplete || !got.ConditionalCreate || !got.ConditionalUpdate || !got.EventIngest {
		t.Fatalf("migration capabilities = %+v, want all prerequisites", got)
	}
	body, err := json.Marshal(got)
	if err != nil {
		t.Fatal(err)
	}
	for _, field := range []string{"checksum_read", "checksum_complete", "conditional_create", "conditional_update", "event_ingest"} {
		if !bytes.Contains(body, []byte(`"`+field+`":true`)) {
			t.Fatalf("migration capability JSON missing %q: %s", field, body)
		}
	}
}

func TestMigrationEventWireFieldsMatchClient(t *testing.T) {
	serverFields := append(strings.Fields(migrationEventStringFields), strings.Fields(migrationEventNumericFields)...)
	clientType := reflect.TypeOf(migrationEventWireFields{})
	clientFields := make([]string, 0, clientType.NumField())
	for i := range clientType.NumField() {
		name, _, _ := strings.Cut(clientType.Field(i).Tag.Get("json"), ",")
		clientFields = append(clientFields, name)
	}
	sort.Strings(serverFields)
	sort.Strings(clientFields)
	if !reflect.DeepEqual(serverFields, clientFields) {
		t.Fatalf("server migration fields = %v, client fields = %v", serverFields, clientFields)
	}
}

func TestMigrationEventOwnerLogsAndMetricsBoundedFields(t *testing.T) {
	rt, cleanup := newAuthRuntime(t)
	defer cleanup()
	core, recorded := observer.New(zap.InfoLevel)
	srv := NewWithConfig(Config{Meta: rt.meta, Pool: rt.pool, TokenSecret: rt.tokenSecret, Logger: zap.New(core)})
	defer srv.Close()
	ts := httptest.NewServer(srv)
	defer ts.Close()

	event := migrationEventFixture()
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/migration/events", bytes.NewReader(migrationEventBody(t, event)))
	req.Header.Set("Authorization", "Bearer "+rt.token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = resp.Body.Close() }()
	if resp.StatusCode != http.StatusNoContent {
		t.Fatalf("status=%d", resp.StatusCode)
	}

	entries := recorded.FilterField(zap.String("event", "migration_cas_attempt")).AllUntimed()
	if len(entries) != 1 {
		t.Fatalf("migration event logs=%d, want 1", len(entries))
	}
	fields := entries[0].ContextMap()
	loggedEvent, err := json.Marshal(fields["migration_event"])
	if err != nil {
		t.Fatal(err)
	}
	for _, want := range []string{"/ebs/full/path/file.bin", "/drive9/full/path/file.bin", "job-1", rt.tenantID, "success"} {
		if !bytes.Contains(loggedEvent, []byte(want)) {
			t.Fatalf("migration event log missing %q: %s", want, loggedEvent)
		}
	}
	for _, entry := range recorded.AllUntimed() {
		if strings.Contains(entry.Message+fmt.Sprint(entry.ContextMap()), rt.token) {
			t.Fatal("API key leaked into logs")
		}
	}

	metrics := httptest.NewRecorder()
	srv.metrics.writePrometheus(metrics)
	for _, line := range strings.Split(metrics.Body.String(), "\n") {
		if !strings.Contains(line, `event="migration_cas_attempt"`) {
			continue
		}
		for _, want := range []string{`phase="DUAL_WRITE_REPAIRING"`, `result="success"`, `error_class="none"`} {
			if !strings.Contains(line, want) {
				t.Fatalf("migration metric missing %s: %s", want, line)
			}
		}
		for _, forbidden := range []string{"/ebs/", "/drive9/", "job_id=", "volume_id=", "node_name=", "pod_name=", "tenant_id=", "space_id="} {
			if strings.Contains(line, forbidden) {
				t.Fatalf("migration metric contains high-cardinality value %q: %s", forbidden, line)
			}
		}
		return
	}
	t.Fatal("missing migration event metric")
}

func TestMigrationEventRejectsTenantMismatch(t *testing.T) {
	event := migrationEventFixture()
	event["tenant_id"] = "tenant-1"
	req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, event)))
	req = req.WithContext(withScope(req.Context(), &TenantScope{TenantID: "tenant-other"}))
	rr := httptest.NewRecorder()
	(&Server{}).handleBusiness(rr, req)
	if rr.Code != http.StatusForbidden {
		t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
	}
}

func TestMigrationEventAcceptsMatchingLegacyTenant(t *testing.T) {
	event := migrationEventFixture()
	event["tenant_id"] = "tenant-1"
	req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, event)))
	req = req.WithContext(withScope(req.Context(), &TenantScope{TenantID: "tenant-1"}))
	rr := httptest.NewRecorder()
	(&Server{}).handleBusiness(rr, req)
	if rr.Code != http.StatusNoContent {
		t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
	}
}

func TestMigrationEventRejectsMissingScope(t *testing.T) {
	req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, migrationEventFixture())))
	rr := httptest.NewRecorder()
	(&Server{}).handleBusiness(rr, req)
	if rr.Code != http.StatusUnauthorized {
		t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
	}
}

func TestMigrationEventAcceptsPreEpochMtime(t *testing.T) {
	event := migrationEventFixture()
	event["mtime"] = int64(-1)
	req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, event)))
	req = req.WithContext(withScope(req.Context(), &TenantScope{TenantID: "tenant-1"}))
	rr := httptest.NewRecorder()
	(&Server{}).handleBusiness(rr, req)
	if rr.Code != http.StatusNoContent {
		t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
	}
}

func TestMigrationEventMetricErrorClassIsBounded(t *testing.T) {
	srv := &Server{metrics: newServerMetrics()}
	for i := 0; i < 20; i++ {
		event := migrationEventFixture()
		event["event_id"] = fmt.Sprintf("evt-cardinality-%d", i)
		event["error_class"] = fmt.Sprintf("reviewclass%d", i)
		req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, event)))
		ctx := withScope(req.Context(), &TenantScope{TenantID: "tenant-1"})
		req = req.WithContext(withMetrics(ctx, srv.metrics))
		rr := httptest.NewRecorder()
		srv.handleMigrationEvent(rr, req)
		if rr.Code != http.StatusNoContent {
			t.Fatalf("event %d status=%d body=%s", i, rr.Code, rr.Body.String())
		}
	}

	metrics := httptest.NewRecorder()
	srv.metrics.writePrometheus(metrics)
	if strings.Contains(metrics.Body.String(), `error_class="reviewclass`) {
		t.Fatalf("raw error class created metric series:\n%s", metrics.Body.String())
	}
	if !strings.Contains(metrics.Body.String(), `error_class="other",event="migration_cas_attempt"`) {
		t.Fatalf("missing normalized error_class=other series:\n%s", metrics.Body.String())
	}
}

func TestMigrationEventRejectsScopedToken(t *testing.T) {
	req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, migrationEventFixture())))
	req = req.WithContext(withScope(req.Context(), &TenantScope{IsScoped: true}))
	rr := httptest.NewRecorder()
	(&Server{}).handleBusiness(rr, req)
	if rr.Code != http.StatusForbidden {
		t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
	}
}

func TestMigrationEventRejectsMalformedAndOversizedPayloads(t *testing.T) {
	tests := []struct {
		name string
		body []byte
	}{
		{name: "malformed", body: []byte(`{"event_id":`)},
		{name: "oversized", body: []byte(`{"padding":"` + strings.Repeat("x", 16<<10) + `"}`)},
		{name: "unknown field", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["api_key"] = "secret"; return e }())},
		{name: "invalid legacy tenant", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["tenant_id"] = 1; return e }())},
		{name: "empty legacy tenant", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["tenant_id"] = ""; return e }())},
		{name: "legacy source version field", body: migrationEventBody(t, func() map[string]any {
			e := migrationEventFixture()
			e["source_version"] = e["source_version_token"]
			delete(e, "source_version_token")
			return e
		}())},
		{name: "missing field", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); delete(e, "job_id"); return e }())},
		{name: "metric injection", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["error_class"] = "/customer/path"; return e }())},
		{name: "wrong type", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["latency_ms"] = "12"; return e }())},
		{name: "negative number", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["latency_ms"] = -1; return e }())},
		{name: "invalid timestamp", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["emitted_at"] = "today"; return e }())},
		{name: "invalid checksum", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["source_checksum_sha256"] = "ABC"; return e }())},
		{name: "composite operation", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["operation"] = "create|update"; return e }())},
		{name: "composite result", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["result"] = "success|conflict"; return e }())},
		{name: "invalid result", body: migrationEventBody(t, func() map[string]any { e := migrationEventFixture(); e["result"] = "retrying"; return e }())},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(tc.body))
			req = req.WithContext(withScope(req.Context(), &TenantScope{}))
			rr := httptest.NewRecorder()
			(&Server{}).handleBusiness(rr, req)
			if rr.Code != http.StatusBadRequest {
				t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
			}
		})
	}
}

func TestMigrationEventRejectsWrongMethod(t *testing.T) {
	req := httptest.NewRequest(http.MethodGet, "/v1/migration/events", nil)
	req = req.WithContext(withScope(req.Context(), &TenantScope{}))
	rr := httptest.NewRecorder()
	(&Server{}).handleBusiness(rr, req)
	if rr.Code != http.StatusMethodNotAllowed || rr.Header().Get("Allow") != http.MethodPost {
		t.Fatalf("status=%d allow=%q", rr.Code, rr.Header().Get("Allow"))
	}
}

func TestMigrationEventScopedAPIKeyRejectedThroughAuth(t *testing.T) {
	rt, cleanup := newAuthRuntime(t)
	defer cleanup()
	setAuthRuntimeScopeKind(t, rt, meta.APIKeyScopeKindFS)
	srv := NewWithConfig(Config{Meta: rt.meta, Pool: rt.pool, TokenSecret: rt.tokenSecret})
	defer srv.Close()
	req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, migrationEventFixture())))
	req.Header.Set("Authorization", "Bearer "+rt.token)
	rr := httptest.NewRecorder()
	srv.ServeHTTP(rr, req)
	if rr.Code != http.StatusForbidden {
		t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
	}
}

func TestMigrationEventRejectsMissingAndWrongAPIKeys(t *testing.T) {
	rt, cleanup := newAuthRuntime(t)
	defer cleanup()
	srv := NewWithConfig(Config{Meta: rt.meta, Pool: rt.pool, TokenSecret: rt.tokenSecret})
	defer srv.Close()
	event := migrationEventFixture()

	for _, tc := range []struct {
		name  string
		token string
	}{
		{name: "missing"},
		{name: "wrong", token: "wrong-owner-key"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			req := httptest.NewRequest(http.MethodPost, "/v1/migration/events", bytes.NewReader(migrationEventBody(t, event)))
			if tc.token != "" {
				req.Header.Set("Authorization", "Bearer "+tc.token)
			}
			rr := httptest.NewRecorder()
			srv.ServeHTTP(rr, req)
			if rr.Code != http.StatusUnauthorized {
				t.Fatalf("status=%d body=%s", rr.Code, rr.Body.String())
			}
		})
	}
}