package connectors

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/gin-gonic/gin"

	connector_admission "github.com/pingcap-inc/nexuscrm-connectors/features/connector/admission"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/interfaces/rest/internal/hostctx"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

// fakeConnectorEventsAdminService stubs ConnectorEventsAdminService for tests.
type fakeConnectorEventsAdminService struct {
	// GetDeadWorkItem
	getDeadEventID   string
	getDeadEventResp *connector_events.DeadWorkItem
	getDeadEventErr  error
	getDeadCalls     int

	// ListDeadWorkItems
	listNamespace string
	listPage      connector_events.Page
	listPermOK    bool
	listCalls     int
	listResp      []connector_events.DeadWorkItem
	listErr       error

	// ResolveActivePack
	resolvePackNS    string
	resolvePackResp  string
	resolvePackErr   error
	resolvePackCalls int

	// ReplayDeadEvent
	replayEventID       string
	replayPackInstallID string
	replayRouteVersion  int64
	replayPermOK        bool
	replayCalls         int
	replayResp          string
	replayErr           error
}

func (f *fakeConnectorEventsAdminService) GetDeadWorkItem(_ context.Context, perms connector_events.PermissionChecker, eventID string) (*connector_events.DeadWorkItem, error) {
	f.getDeadCalls++
	f.getDeadEventID = eventID
	return f.getDeadEventResp, f.getDeadEventErr
}

func (f *fakeConnectorEventsAdminService) GetDeadEvent(_ context.Context, _ connector_events.PermissionChecker, eventID string) (*connector_events.ConnectorEvent, error) {
	f.getDeadEventID = eventID
	if f.getDeadEventErr != nil || f.getDeadEventResp == nil {
		return nil, f.getDeadEventErr
	}
	item := f.getDeadEventResp
	return &connector_events.ConnectorEvent{
		ID:               item.EventID,
		Namespace:        item.Namespace,
		InstalledPackID:  item.InstalledPackID,
		EventKind:        item.EventKind,
		IdempotencyKey:   item.IdempotencyKey,
		Status:           item.State,
		RetryCount:       item.FailureCount,
		LastErrorAt:      item.LastFailureAt,
		ProcessedAt:      item.ProcessedAt,
		ReplayedAt:       item.ReplayedAt,
		ReplayOfID:       item.ReplayOfEventID,
		CreatedDate:      item.EventCreatedDate,
		LastModifiedDate: item.LastModifiedDate,
	}, nil
}

func (f *fakeConnectorEventsAdminService) ListDeadWorkItems(_ context.Context, perms connector_events.PermissionChecker, namespace string, page connector_events.Page) ([]connector_events.DeadWorkItem, error) {
	f.listCalls++
	f.listNamespace = namespace
	f.listPage = page
	if perms != nil {
		f.listPermOK = perms.Has(string(constants.PermViewConnectorEvents))
	}
	return f.listResp, f.listErr
}

func (f *fakeConnectorEventsAdminService) ListDeadEvents(_ context.Context, _ connector_events.PermissionChecker, namespace string, page connector_events.Page) ([]connector_events.ConnectorEvent, error) {
	f.listNamespace = namespace
	f.listPage = page
	if f.listErr != nil {
		return nil, f.listErr
	}
	out := make([]connector_events.ConnectorEvent, 0, len(f.listResp))
	for _, item := range f.listResp {
		out = append(out, connector_events.ConnectorEvent{
			ID:               item.EventID,
			Namespace:        item.Namespace,
			InstalledPackID:  item.InstalledPackID,
			EventKind:        item.EventKind,
			IdempotencyKey:   item.IdempotencyKey,
			Status:           item.State,
			RetryCount:       item.FailureCount,
			LastErrorAt:      item.LastFailureAt,
			ProcessedAt:      item.ProcessedAt,
			ReplayedAt:       item.ReplayedAt,
			ReplayOfID:       item.ReplayOfEventID,
			CreatedDate:      item.EventCreatedDate,
			LastModifiedDate: item.LastModifiedDate,
		})
	}
	return out, nil
}

func (f *fakeConnectorEventsAdminService) ResolveActivePack(_ context.Context, namespace string) (string, error) {
	f.resolvePackCalls++
	f.resolvePackNS = namespace
	return f.resolvePackResp, f.resolvePackErr
}

func (f *fakeConnectorEventsAdminService) ReplayDeadEvent(_ context.Context, perms connector_events.PermissionChecker, eventID string, packInstallID string, routeVersion int64) (string, error) {
	f.replayCalls++
	f.replayEventID = eventID
	f.replayPackInstallID = packInstallID
	f.replayRouteVersion = routeVersion
	if perms != nil {
		f.replayPermOK = perms.Has(string(constants.PermReplayConnectorEvents))
	}
	return f.replayResp, f.replayErr
}

// fakeRouteResolver stubs connectorEventsRouteResolver for tests.
type fakeRouteResolver struct {
	version           int64
	flowDeveloperName string
	err               error
	calls             int
}

func (f *fakeRouteResolver) ResolveFlow(_ context.Context, _, _, _ string) (int64, string, error) {
	f.calls++
	return f.version, f.flowDeveloperName, f.err
}

// setupConnectorEventsAdminRouter builds a test gin engine with both read and
// replay groups using the given systemPerms map.
func setupConnectorEventsAdminRouter(svc ConnectorEventsAdminService, resolver connectorEventsRouteResolver, systemPerms map[string]bool) *gin.Engine {
	gin.SetMode(gin.TestMode)
	r := gin.New()
	r.Use(func(c *gin.Context) {
		c.Set(hostctx.ContextKeySystemPermissions, systemPerms)
		c.Next()
	})
	h := NewConnectorEventsAdminHandler(svc, resolver)
	h.RegisterReadRoutes(r.Group("/api/connectors"))
	h.RegisterReplayRoutes(r.Group("/api/connectors"))
	return r
}

type legacyOnlyConnectorEventsAdminService struct {
	event       connector_events.ConnectorEvent
	listCalls   int
	getCalls    int
	replayCalls int
}

var _ ConnectorEventsAdminService = (*legacyOnlyConnectorEventsAdminService)(nil)

func (s *legacyOnlyConnectorEventsAdminService) GetDeadEvent(context.Context, connector_events.PermissionChecker, string) (*connector_events.ConnectorEvent, error) {
	s.getCalls++
	event := s.event
	return &event, nil
}

func (s *legacyOnlyConnectorEventsAdminService) ListDeadEvents(context.Context, connector_events.PermissionChecker, string, connector_events.Page) ([]connector_events.ConnectorEvent, error) {
	s.listCalls++
	return []connector_events.ConnectorEvent{s.event}, nil
}

func (s *legacyOnlyConnectorEventsAdminService) ResolveActivePack(context.Context, string) (string, error) {
	return "legacy-pack-current", nil
}

func (s *legacyOnlyConnectorEventsAdminService) ReplayDeadEvent(context.Context, connector_events.PermissionChecker, string, string, int64) (string, error) {
	s.replayCalls++
	return "legacy-replay-event", nil
}

// ── List tests ──────────────────────────────────────────────────────────────

func TestConnectorEventsAdminList_HappyPath_DefaultPagination(t *testing.T) {
	now := time.Date(2026, 6, 9, 12, 0, 0, 0, time.UTC)
	svc := &fakeConnectorEventsAdminService{
		listResp: []connector_events.DeadWorkItem{{
			WorkID:           "work-dead-1",
			EventID:          "evt-dead-1",
			Namespace:        "hubspot",
			InstalledPackID:  "pack-install-1",
			EventKind:        "contact.updated",
			IdempotencyKey:   "idem-1",
			State:            "dead",
			FailureCount:     5,
			FailureClass:     "inbound_execution_failed",
			FailureSummary:   "safe operator summary",
			LastFailureAt:    &now,
			TerminalAt:       &now,
			ReplayedAt:       &now,
			EventCreatedDate: now,
			LastModifiedDate: now,
		}},
	}
	resolver := &fakeRouteResolver{version: 3, flowDeveloperName: "HubSpot_Contact_Updated"}

	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
	}
	if svc.listCalls != 1 {
		t.Fatalf("expected list call count 1, got %d", svc.listCalls)
	}
	if svc.listNamespace != "hubspot" {
		t.Fatalf("namespace = %q, want %q", svc.listNamespace, "hubspot")
	}
	if svc.listPage.Limit != 50 || svc.listPage.Offset != 0 {
		t.Fatalf("page = %+v, want limit=50 offset=0", svc.listPage)
	}
	if !svc.listPermOK {
		t.Fatal("expected ViewConnectorEvents permission to be visible to service")
	}

	var body map[string]any
	if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
		t.Fatalf("invalid JSON: %v", err)
	}
	if got := body["namespace"]; got != "hubspot" {
		t.Fatalf("namespace = %v, want hubspot", got)
	}
	events, ok := body["events"].([]any)
	if !ok || len(events) != 1 {
		t.Fatalf("events payload malformed: %#v", body["events"])
	}
	event0 := events[0].(map[string]any)
	if event0["id"] != "evt-dead-1" {
		t.Fatalf("event id = %v, want evt-dead-1", event0["id"])
	}
	if event0["workId"] != "work-dead-1" || event0["executionOwner"] != "work_item" {
		t.Fatalf("WorkItem identity = workId %v owner %v", event0["workId"], event0["executionOwner"])
	}
	if event0["retryCount"] != float64(5) {
		t.Fatalf("retryCount = %v, want 5", event0["retryCount"])
	}
	if event0["state"] != "dead" {
		t.Fatalf("state = %v, want dead", event0["state"])
	}
	if event0["failureCount"] != float64(5) {
		t.Fatalf("failureCount = %v, want 5", event0["failureCount"])
	}
	if _, exposed := event0["failureClass"]; exposed {
		t.Fatalf("failureClass must not be exposed: %v", event0["failureClass"])
	}
	if _, exposed := event0["failureSummary"]; exposed {
		t.Fatalf("failureSummary must not be exposed: %v", event0["failureSummary"])
	}
	if _, present := event0["lastFailureAt"]; !present {
		t.Fatal("lastFailureAt missing from WorkItem-backed response")
	}
	if _, present := event0["terminalAt"]; !present {
		t.Fatal("terminalAt missing from WorkItem-backed response")
	}
	// Payload + LastError content must NOT appear in the list DTO (batch-leak guard).
	if _, present := event0["payload"]; present {
		t.Fatalf("list DTO must not expose raw payload; got %#v", event0["payload"])
	}
	if _, present := event0["lastError"]; present {
		t.Fatalf("list DTO must not expose lastError content; got %#v", event0["lastError"])
	}
}

func TestConnectorEventsAdminList_LabelsLegacyEventOwner(t *testing.T) {
	now := time.Date(2026, 6, 9, 12, 5, 0, 0, time.UTC)
	svc := &fakeConnectorEventsAdminService{
		listResp: []connector_events.DeadWorkItem{{
			EventID:          "evt-legacy-dead",
			Namespace:        "hubspot",
			InstalledPackID:  "legacy-pack",
			EventKind:        "contact.updated",
			IdempotencyKey:   "legacy-idem",
			State:            "dead",
			FailureCount:     5,
			EventCreatedDate: now,
			LastModifiedDate: now,
			LegacyEvent:      true,
		}},
	}
	router := setupConnectorEventsAdminRouter(svc, &fakeRouteResolver{}, map[string]bool{
		string(constants.PermViewConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
	}
	var body struct {
		Events []map[string]any `json:"events"`
	}
	if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
		t.Fatalf("invalid JSON: %v", err)
	}
	if len(body.Events) != 1 || body.Events[0]["executionOwner"] != "legacy_event" {
		t.Fatalf("legacy execution owner response = %#v", body.Events)
	}
	if _, present := body.Events[0]["workId"]; present {
		t.Fatalf("legacy Event response must not invent a WorkId: %#v", body.Events[0]["workId"])
	}
}

func TestConnectorEventsAdmin_PreservesPublishedLegacyServiceInterface(t *testing.T) {
	now := time.Date(2026, 7, 16, 16, 0, 0, 0, time.UTC)
	svc := &legacyOnlyConnectorEventsAdminService{event: connector_events.ConnectorEvent{
		ID:               "legacy-dead-event",
		Namespace:        "legacy-provider",
		InstalledPackID:  "legacy-pack-old",
		EventKind:        "contact.updated",
		IdempotencyKey:   "legacy-key",
		Status:           "dead",
		RetryCount:       5,
		CreatedDate:      now,
		LastModifiedDate: now,
	}}
	router := setupConnectorEventsAdminRouter(svc, &fakeRouteResolver{version: 3}, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})

	listRecorder := httptest.NewRecorder()
	router.ServeHTTP(listRecorder, httptest.NewRequest(http.MethodGet, "/api/connectors/legacy-provider/events/dead", nil))
	if listRecorder.Code != http.StatusOK || svc.listCalls != 1 {
		t.Fatalf("legacy list = status %d calls %d body=%s, want 200/1", listRecorder.Code, svc.listCalls, listRecorder.Body.String())
	}
	var listBody struct {
		Events []map[string]any `json:"events"`
	}
	if err := json.Unmarshal(listRecorder.Body.Bytes(), &listBody); err != nil {
		t.Fatalf("decode legacy list response: %v", err)
	}
	if len(listBody.Events) != 1 || listBody.Events[0]["executionOwner"] != "legacy_event" || listBody.Events[0]["state"] != "dead" {
		t.Fatalf("legacy list response = %#v, want Event compatibility projection", listBody.Events)
	}

	replayRecorder := httptest.NewRecorder()
	router.ServeHTTP(replayRecorder, httptest.NewRequest(http.MethodPost, "/api/connectors/events/legacy-dead-event/replay", nil))
	if replayRecorder.Code != http.StatusOK || svc.getCalls != 1 || svc.replayCalls != 1 {
		t.Fatalf("legacy replay = status %d get %d replay %d body=%s, want 200/1/1",
			replayRecorder.Code, svc.getCalls, svc.replayCalls, replayRecorder.Body.String())
	}
}

func TestConnectorEventsAdminList_BadLimitReturns400(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{}
	resolver := &fakeRouteResolver{}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead?limit=abc", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusBadRequest {
		t.Fatalf("expected 400, got %d body=%s", w.Code, w.Body.String())
	}
	if svc.listCalls != 0 {
		t.Fatalf("list should not be called on bad limit; got %d", svc.listCalls)
	}
}

func TestConnectorEventsAdminList_LimitCappedAt500(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{}
	resolver := &fakeRouteResolver{}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead?limit=10000", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
	}
	if svc.listPage.Limit != 500 {
		t.Fatalf("limit should be capped to 500, got %d", svc.listPage.Limit)
	}
}

func TestConnectorEventsAdminList_InternalErrorOpaque(t *testing.T) {
	internalErr := errors.New("sql: table _System_ConnectorEvent missing")
	svc := &fakeConnectorEventsAdminService{listErr: internalErr}
	resolver := &fakeRouteResolver{}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead?limit=10&offset=2", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusInternalServerError {
		t.Fatalf("expected 500, got %d body=%s", w.Code, w.Body.String())
	}
	if svc.listPage.Limit != 10 || svc.listPage.Offset != 2 {
		t.Fatalf("page = %+v, want limit=10 offset=2", svc.listPage)
	}
	var body map[string]any
	if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
		t.Fatalf("invalid JSON: %v", err)
	}
	if body["error"] != "internal server error" {
		t.Fatalf("error = %v, want internal server error", body["error"])
	}
	if w.Body.String() == internalErr.Error() {
		t.Fatal("raw internal error leaked in response")
	}
}

// ── Replay tests ─────────────────────────────────────────────────────────────

// deadEvent is a convenience builder for a dead WorkItem operator view.
func deadEvent(namespace, eventKind string) *connector_events.DeadWorkItem {
	return &connector_events.DeadWorkItem{
		WorkID:          "work-dead-1",
		EventID:         "evt-dead-1",
		Namespace:       namespace,
		EventKind:       eventKind,
		State:           "dead",
		InstalledPackID: "old-pack-install-id",
	}
}

func TestConnectorEventsAdminReplay_ResolvesCurrentIncarnation(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "contact.updated"),
		resolvePackResp:  "current-pack-install-id",
		replayResp:       "evt-replay-1",
	}
	resolver := &fakeRouteResolver{version: 7, flowDeveloperName: "HubSpot_Contact_Updated"}

	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusOK {
		t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
	}
	// ReplayDeadEvent must receive the packInstallID from ResolveActivePack,
	// NOT the stale InstalledPackID from the dead event.
	if svc.replayPackInstallID != "current-pack-install-id" {
		t.Fatalf("packInstallID = %q, want current-pack-install-id (NOT old stale ID)", svc.replayPackInstallID)
	}
	// routeVersion must come from ResolveFlow.
	if svc.replayRouteVersion != 7 {
		t.Fatalf("routeVersion = %d, want 7", svc.replayRouteVersion)
	}

	var body map[string]any
	if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
		t.Fatalf("invalid JSON: %v", err)
	}
	if body["eventId"] != "evt-replay-1" {
		t.Fatalf("eventId = %v, want evt-replay-1", body["eventId"])
	}
}

func TestConnectorEventsAdminReplay_RouteNotDeclared_Returns409_NoReplayCalled(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "unknown.kind"),
		resolvePackResp:  "current-pack-install-id",
	}
	resolver := &fakeRouteResolver{err: connector_inbound_routing.ErrRouteNotDeclared}

	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusConflict {
		t.Fatalf("expected 409, got %d body=%s", w.Code, w.Body.String())
	}
	// ReplayDeadEvent must NOT have been called.
	if svc.replayCalls != 0 {
		t.Fatalf("ReplayDeadEvent should not be called when route not declared; got %d calls", svc.replayCalls)
	}
}

func TestConnectorEventsAdminReplay_AdmissionRefusalsReturnRetryable503(t *testing.T) {
	testCases := []struct {
		name           string
		replayErr      error
		wantRetryAfter string
	}{
		{
			name: "quiesced",
			replayErr: &connector_admission.QuiescedError{Fence: connector_admission.Fence{
				Namespace: "hubspot", Mode: connector_admission.ModeQuiescing, RetryAfterSeconds: 45,
			}},
			wantRetryAfter: "45",
		},
		{
			name:      "schema unavailable",
			replayErr: &connector_admission.SchemaUnavailableError{Namespace: "hubspot"},
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			svc := &fakeConnectorEventsAdminService{
				getDeadEventResp: deadEvent("hubspot", "contact.updated"),
				resolvePackResp:  "current-pack-install-id",
				replayErr:        tc.replayErr,
			}
			router := setupConnectorEventsAdminRouter(svc, &fakeRouteResolver{version: 3}, map[string]bool{
				string(constants.PermViewConnectorEvents):   true,
				string(constants.PermReplayConnectorEvents): true,
			})
			w := httptest.NewRecorder()
			req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
			router.ServeHTTP(w, req)

			if w.Code != http.StatusServiceUnavailable {
				t.Fatalf("expected 503, got %d body=%s", w.Code, w.Body.String())
			}
			if got := w.Header().Get("Retry-After"); got != tc.wantRetryAfter {
				t.Fatalf("Retry-After = %q, want %q", got, tc.wantRetryAfter)
			}
		})
	}
}

// ── Error mapping tests ──────────────────────────────────────────────────────

func TestConnectorEventsAdminReplay_GetDeadEvent_NotFound_Returns404(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventErr: connector_events.ErrEventNotFound,
	}
	resolver := &fakeRouteResolver{version: 1}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/missing/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusNotFound {
		t.Fatalf("expected 404, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestConnectorEventsAdminReplay_GetDeadEvent_NotDead_Returns409(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventErr: connector_events.ErrEventNotDead,
	}
	resolver := &fakeRouteResolver{version: 1}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-processing/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusConflict {
		t.Fatalf("expected 409, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestConnectorEventsAdminReplay_ActiveProjection_Returns409(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "contact.updated"),
		resolvePackResp:  "current-pack-install-id",
		replayErr:        connector_events.ErrReplayBlocked,
	}
	resolver := &fakeRouteResolver{version: 1}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusConflict {
		t.Fatalf("expected 409, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestConnectorEventsAdminReplay_InactivePack_Returns404(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "contact.updated"),
		resolvePackErr:   connector_events.ErrInactivePack,
	}
	resolver := &fakeRouteResolver{version: 1}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusNotFound {
		t.Fatalf("expected 404, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestConnectorEventsAdminReplay_InvalidRequest_Returns400(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "contact.updated"),
		resolvePackResp:  "current-pack",
		replayErr:        connector_events.ErrInvalidReplayRequest,
	}
	resolver := &fakeRouteResolver{version: 1}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusBadRequest {
		t.Fatalf("expected 400, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestConnectorEventsAdminReplay_PackIncarnationChanged_Returns503(t *testing.T) {
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "contact.updated"),
		resolvePackResp:  "current-pack",
		replayErr:        connector_events.ErrPackIncarnationChanged,
	}
	resolver := &fakeRouteResolver{version: 1}
	router := setupConnectorEventsAdminRouter(svc, resolver, map[string]bool{
		string(constants.PermViewConnectorEvents):   true,
		string(constants.PermReplayConnectorEvents): true,
	})
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead-1/replay", nil)
	router.ServeHTTP(w, req)

	if w.Code != http.StatusServiceUnavailable {
		t.Fatalf("expected 503, got %d body=%s", w.Code, w.Body.String())
	}
}

// ── Route gate tests ─────────────────────────────────────────────────────────

func TestConnectorEventsAdminReadRouteGate_RequiresViewConnectorEvents(t *testing.T) {
	gin.SetMode(gin.TestMode)

	email := "regular@example.com"
	regularUser := models.UserSession{
		ID:        "user-regular",
		Name:      "Regular User",
		Email:     &email,
		ProfileID: "profile-regular",
	}

	newRouter := func(systemPerms map[string]bool, svc *fakeConnectorEventsAdminService) *gin.Engine {
		r := gin.New()
		r.Use(func(c *gin.Context) {
			c.Set(hostctx.ContextKeyUser, regularUser)
			c.Set(hostctx.ContextKeySystemPermissions, systemPerms)
			c.Next()
		})
		resolver := &fakeRouteResolver{}
		readGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermViewConnectorEvents))
		NewConnectorEventsAdminHandler(svc, resolver).RegisterReadRoutes(readGroup)
		return r
	}

	t.Run("denied_without_permission", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{}
		router := newRouter(map[string]bool{}, svc)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
		router.ServeHTTP(w, req)
		if w.Code != http.StatusForbidden {
			t.Fatalf("expected 403, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.listCalls != 0 {
			t.Fatalf("handler should not be reached without permission; listCalls=%d", svc.listCalls)
		}
	})

	t.Run("allowed_with_permission", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{}
		router := newRouter(map[string]bool{
			string(constants.PermViewConnectorEvents): true,
		}, svc)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
		router.ServeHTTP(w, req)
		if w.Code != http.StatusOK {
			t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.listCalls != 1 {
			t.Fatalf("handler should be reached with permission; listCalls=%d", svc.listCalls)
		}
	})
}

func TestConnectorEventsAdminReplayRouteGate_RequiresReplayConnectorEvents(t *testing.T) {
	gin.SetMode(gin.TestMode)

	email := "regular@example.com"
	regularUser := models.UserSession{
		ID:        "user-regular",
		Name:      "Regular User",
		Email:     &email,
		ProfileID: "profile-regular",
	}

	newRouter := func(systemPerms map[string]bool, svc *fakeConnectorEventsAdminService) *gin.Engine {
		r := gin.New()
		r.Use(func(c *gin.Context) {
			c.Set(hostctx.ContextKeyUser, regularUser)
			c.Set(hostctx.ContextKeySystemPermissions, systemPerms)
			c.Next()
		})
		resolver := &fakeRouteResolver{version: 1}
		replayGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermReplayConnectorEvents))
		NewConnectorEventsAdminHandler(svc, resolver).RegisterReplayRoutes(replayGroup)
		return r
	}

	t.Run("denied_without_permission", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{}
		router := newRouter(map[string]bool{}, svc)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead/replay", nil)
		router.ServeHTTP(w, req)
		if w.Code != http.StatusForbidden {
			t.Fatalf("expected 403, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.replayCalls != 0 {
			t.Fatalf("handler should not be reached without permission; replayCalls=%d", svc.replayCalls)
		}
	})

	t.Run("allowed_with_permission", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{
			getDeadEventResp: deadEvent("hubspot", "contact.updated"),
			resolvePackResp:  "pack-install",
			replayResp:       "evt-replay-2",
		}
		router := newRouter(map[string]bool{
			string(constants.PermViewConnectorEvents):   true,
			string(constants.PermReplayConnectorEvents): true,
		}, svc)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead/replay", nil)
		router.ServeHTTP(w, req)
		if w.Code != http.StatusOK {
			t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.replayCalls != 1 {
			t.Fatalf("handler should be reached with permission; replayCalls=%d", svc.replayCalls)
		}
	})
}

// ── Route mount smoke test ───────────────────────────────────────────────────

// TestConnectorEventsAdminRouteMount_NoWildcardConflict registers BOTH read and
// replay groups on a real gin engine and verifies no wildcard-conflict panic.
func TestConnectorEventsAdminRouteMount_NoWildcardConflict(t *testing.T) {
	gin.SetMode(gin.TestMode)
	r := gin.New()
	svc := &fakeConnectorEventsAdminService{}
	resolver := &fakeRouteResolver{}
	h := NewConnectorEventsAdminHandler(svc, resolver)

	readGroup := r.Group("/api/connectors")
	h.RegisterReadRoutes(readGroup)

	replayGroup := r.Group("/api/connectors")
	h.RegisterReplayRoutes(replayGroup)

	// If gin panics on wildcard conflict the test itself would panic/fail.
	// Fire a request to confirm the router is functional.
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
	r.ServeHTTP(w, req)
	// 200 OK because no permission middleware is applied and the service returns nil slice.
	if w.Code != http.StatusOK {
		t.Fatalf("expected 200 from route mount smoke test, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestConnectorEventsAdminRequiresSpecificPermissions(t *testing.T) {
	gin.SetMode(gin.TestMode)

	t.Run("list_modify_all_data_denied", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{}
		r := gin.New()
		r.Use(func(c *gin.Context) {
			c.Set(hostctx.ContextKeyUser, models.UserSession{ID: "admin", ProfileID: "jira_admin"})
			c.Set(hostctx.ContextKeySystemPermissions, map[string]bool{"ModifyAllData": true})
			c.Next()
		})
		readGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermViewConnectorEvents))
		NewConnectorEventsAdminHandler(svc, &fakeRouteResolver{}).RegisterReadRoutes(readGroup)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
		r.ServeHTTP(w, req)
		if w.Code != http.StatusForbidden {
			t.Fatalf("ModifyAllData without ViewConnectorEvents must be denied, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.listCalls != 0 {
			t.Fatalf("handler should not call service when middleware denies; calls=%d", svc.listCalls)
		}
	})

	t.Run("list_view_permission_allowed", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{}
		r := gin.New()
		r.Use(func(c *gin.Context) {
			c.Set(hostctx.ContextKeyUser, models.UserSession{ID: "observer", ProfileID: "jira_admin"})
			c.Set(hostctx.ContextKeySystemPermissions, map[string]bool{string(constants.PermViewConnectorEvents): true})
			c.Next()
		})
		readGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermViewConnectorEvents))
		NewConnectorEventsAdminHandler(svc, &fakeRouteResolver{}).RegisterReadRoutes(readGroup)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodGet, "/api/connectors/hubspot/events/dead", nil)
		r.ServeHTTP(w, req)
		if w.Code != http.StatusOK {
			t.Fatalf("ViewConnectorEvents must reach list, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.listCalls != 1 || !svc.listPermOK {
			t.Fatalf("checker must satisfy ViewConnectorEvents; calls=%d permOK=%v", svc.listCalls, svc.listPermOK)
		}
	})

	t.Run("replay_modify_all_data_denied", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{
			getDeadEventResp: deadEvent("hubspot", "contact.updated"),
			resolvePackResp:  "pack-cur",
			replayResp:       "evt-replay-su",
		}
		r := gin.New()
		r.Use(func(c *gin.Context) {
			c.Set(hostctx.ContextKeyUser, models.UserSession{ID: "admin", ProfileID: "jira_admin"})
			c.Set(hostctx.ContextKeySystemPermissions, map[string]bool{"ModifyAllData": true})
			c.Next()
		})
		replayGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermReplayConnectorEvents))
		NewConnectorEventsAdminHandler(svc, &fakeRouteResolver{version: 4}).RegisterReplayRoutes(replayGroup)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead/replay", nil)
		r.ServeHTTP(w, req)
		if w.Code != http.StatusForbidden {
			t.Fatalf("ModifyAllData without ReplayConnectorEvents must be denied, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.replayCalls != 0 {
			t.Fatalf("handler should not call service when middleware denies; calls=%d", svc.replayCalls)
		}
	})

	t.Run("replay_permission_allowed", func(t *testing.T) {
		svc := &fakeConnectorEventsAdminService{
			getDeadEventResp: deadEvent("hubspot", "contact.updated"),
			resolvePackResp:  "pack-cur",
			replayResp:       "evt-replay-su",
		}
		r := gin.New()
		r.Use(func(c *gin.Context) {
			c.Set(hostctx.ContextKeyUser, models.UserSession{ID: "operator", ProfileID: "jira_admin"})
			c.Set(hostctx.ContextKeySystemPermissions, map[string]bool{string(constants.PermReplayConnectorEvents): true})
			c.Next()
		})
		replayGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermReplayConnectorEvents))
		NewConnectorEventsAdminHandler(svc, &fakeRouteResolver{version: 4}).RegisterReplayRoutes(replayGroup)
		w := httptest.NewRecorder()
		req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead/replay", nil)
		r.ServeHTTP(w, req)
		if w.Code != http.StatusOK {
			t.Fatalf("ReplayConnectorEvents must reach replay, got %d body=%s", w.Code, w.Body.String())
		}
		if svc.replayCalls != 1 || !svc.replayPermOK {
			t.Fatalf("checker must satisfy ReplayConnectorEvents; calls=%d permOK=%v", svc.replayCalls, svc.replayPermOK)
		}
	})
}

func TestConnectorEventsAdminReplay_ReplayOnlyNoView(t *testing.T) {
	gin.SetMode(gin.TestMode)
	svc := &fakeConnectorEventsAdminService{
		getDeadEventResp: deadEvent("hubspot", "contact.updated"),
		resolvePackResp:  "pack-cur",
		replayResp:       "evt-replay-ro",
	}
	r := gin.New()
	r.Use(func(c *gin.Context) {
		c.Set(hostctx.ContextKeyUser, models.UserSession{ID: "u", ProfileID: "standard_user"})
		c.Set(hostctx.ContextKeySystemPermissions, map[string]bool{
			string(constants.PermReplayConnectorEvents): true, // NO View
		})
		c.Next()
	})
	replayGroup := r.Group("/api/connectors", hostctx.RequirePermission(constants.PermReplayConnectorEvents))
	NewConnectorEventsAdminHandler(svc, &fakeRouteResolver{version: 2}).RegisterReplayRoutes(replayGroup)
	w := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodPost, "/api/connectors/events/evt-dead/replay", nil)
	r.ServeHTTP(w, req)
	if w.Code != http.StatusOK {
		t.Fatalf("replay-only principal must replay, got %d body=%s", w.Code, w.Body.String())
	}
}