//go:build crm

package crm

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/nexuscrm/backend/internal/application/services"
	"github.com/nexuscrm/backend/pkg/serverruntime"
	"github.com/nexuscrm/dialect-sf-crm/scim"
	"github.com/nexuscrm/kernel/pkg/apierr"
	kernelauth "github.com/nexuscrm/kernel/pkg/auth"
	kernelerrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

// --- fakes ----------------------------------------------------------------

type fakeRecords struct {
	contract.Records
	byID        map[string]contract.Record
	byField     map[string]map[string]contract.Record // field -> value -> record
	inserted    contract.Record
	updatedID   string
	updates     contract.Record
	insertErr   error
	updateErr   error
	retrieveErr error
	nextID      string
	// onUpdate, when set, observes each successful Update (used to simulate
	// the transactional audit head row the real engine writes).
	onUpdate func(id string, updates contract.Record)
}

func newFakeRecords() *fakeRecords {
	return &fakeRecords{
		byID:    map[string]contract.Record{},
		byField: map[string]map[string]contract.Record{},
		nextID:  "005NEWUSER",
	}
}

func (f *fakeRecords) put(rec contract.Record) {
	id, _ := rec["Id"].(string)
	if id != "" {
		f.byID[id] = rec
	}
	for _, field := range []string{scim.FieldUsername, scim.FieldEmail, scimFieldName} {
		if v, ok := rec[field].(string); ok && v != "" {
			if f.byField[field] == nil {
				f.byField[field] = map[string]contract.Record{}
			}
			f.byField[field][v] = rec
		}
	}
}

// putUnderField indexes rec so RetrieveByField(_, _, field, value, _) returns
// it — used to seed lookups (e.g. permission-set assignments) the default
// put() indexing does not cover.
func (f *fakeRecords) putUnderField(_ string, field, value string, rec contract.Record) {
	if id, _ := rec["Id"].(string); id != "" {
		f.byID[id] = rec
	}
	if f.byField[field] == nil {
		f.byField[field] = map[string]contract.Record{}
	}
	f.byField[field][value] = rec
}

func (f *fakeRecords) RetrieveByField(_ context.Context, _ string, field, value string, _ *models.UserSession) (contract.Record, error) {
	if m := f.byField[field]; m != nil {
		if rec, ok := m[value]; ok {
			return rec, nil
		}
	}
	return nil, nil
}

func (f *fakeRecords) Retrieve(_ context.Context, obj, id string, _ *models.UserSession) (contract.Record, error) {
	if f.retrieveErr != nil {
		return nil, f.retrieveErr
	}
	if rec, ok := f.byID[id]; ok {
		return rec, nil
	}
	return nil, apierr.NotFound(obj + " not found")
}

func (f *fakeRecords) Insert(_ context.Context, _ string, data contract.Record, _ *models.UserSession) (contract.Record, error) {
	if f.insertErr != nil {
		return nil, f.insertErr
	}
	data["Id"] = f.nextID
	f.inserted = data
	f.put(data)
	return data, nil
}

func (f *fakeRecords) Update(_ context.Context, obj, id string, updates contract.Record, _ *models.UserSession) error {
	if f.updateErr != nil {
		return f.updateErr
	}
	rec, ok := f.byID[id]
	if !ok {
		return apierr.NotFound(obj + " not found")
	}
	f.updatedID = id
	f.updates = updates
	for k, v := range updates {
		rec[k] = v
	}
	if f.onUpdate != nil {
		f.onUpdate(id, updates)
	}
	return nil
}

type fakeQuery struct {
	result        *contract.QueryResult
	err           error
	configRecords *fakeRecords
	configErr     error
	lastCtx       context.Context
	lastSpec      queryspec.QuerySpec
}

func (q *fakeQuery) Query(ctx context.Context, spec queryspec.QuerySpec, _ *models.UserSession) (*contract.QueryResult, error) {
	q.lastCtx = ctx
	q.lastSpec = spec
	if spec.ObjectAPIName == scimConfigObject && q.configRecords != nil {
		if q.configErr != nil {
			return nil, q.configErr
		}
		rec := q.configRecords.byField[scimFieldName][scimConfigSingletonName]
		if rec == nil {
			return &contract.QueryResult{}, nil
		}
		return &contract.QueryResult{Records: []contract.Record{rec}}, nil
	}
	return q.result, q.err
}

func (q *fakeQuery) SearchWithSpec(context.Context, queryspec.SearchSpec, *models.UserSession) ([]contract.Record, error) {
	return nil, nil
}

func (q *fakeQuery) GlobalSearchWithSpec(context.Context, queryspec.GlobalSearchSpec, *models.UserSession) ([]contract.SearchResult, error) {
	return nil, nil
}

func (q *fakeQuery) QueryHistory(context.Context, contract.HistoryQueryRequest, *models.UserSession) (*contract.HistoryQueryResult, error) {
	return nil, nil
}

// fakeAuditLog implements contract.AuditLog over an in-memory row list with
// naive offset/limit paging. fn, when set, overrides each call; reqs records
// every request seen.
type fakeAuditLog struct {
	rows []contract.AuditLogRow
	err  error
	fn   func(contract.AuditLogQueryRequest) (*contract.AuditLogQueryResult, error)
	reqs []contract.AuditLogQueryRequest
}

func (a *fakeAuditLog) QueryAuditLog(_ context.Context, req contract.AuditLogQueryRequest, _ *models.UserSession) (*contract.AuditLogQueryResult, error) {
	a.reqs = append(a.reqs, req)
	if a.fn != nil {
		return a.fn(req)
	}
	if a.err != nil {
		return nil, a.err
	}
	start := req.Offset
	if start > len(a.rows) {
		start = len(a.rows)
	}
	end := len(a.rows)
	if req.Limit > 0 && start+req.Limit < end {
		end = start + req.Limit
	}
	return &contract.AuditLogQueryResult{Rows: a.rows[start:end], Limit: req.Limit, Offset: req.Offset}, nil
}

// isActiveHeadRow builds an audit head row recording an UPDATE whose scope
// includes IsActive, as the engine's transactional emitter writes it.
func isActiveHeadRow() contract.AuditLogRow {
	action := "UPDATE"
	cj := `{"fields":["IsActive"]}`
	return contract.AuditLogRow{FieldName: "__record__", Action: &action, ChangesJSON: &cj}
}

func headRows(n int, action string, changesJSON *string) []contract.AuditLogRow {
	rows := make([]contract.AuditLogRow, 0, n)
	for i := 0; i < n; i++ {
		a := action
		rows = append(rows, contract.AuditLogRow{FieldName: "__record__", Action: &a, ChangesJSON: changesJSON})
	}
	return rows
}

func strPtr(s string) *string { return &s }

// --- harness --------------------------------------------------------------

func newTestHandler() (*scimIdentityHandler, *fakeRecords, *fakeAuditLog) {
	rec := newFakeRecords()
	rec.put(contract.Record{"Id": "PROFILE1", scimFieldName: "Standard User"})
	q := testConfigQuery(testScimConfig())
	al := &fakeAuditLog{}
	// Mirror the engine: a successful IsActive update transactionally appends
	// an audit head row, so the post-write verification sees it (tests for a
	// non-recording audit trail clear onUpdate explicitly).
	rec.onUpdate = func(_ string, updates contract.Record) {
		if _, ok := updates[scim.FieldIsActive]; ok {
			al.rows = append(al.rows, isActiveHeadRow())
		}
	}
	h := &scimIdentityHandler{
		recordsFn:  func() contract.Records { return rec },
		queryFn:    func() contract.Query { return q },
		auditLogFn: func() contract.AuditLog { return al },
		cc:         testConfigCache(testScimConfig()),
		state:      newScimIdentityState(time.Now().UTC(), 5, 300),
	}
	return h, rec, al
}

// testScimConfig mirrors a complete L2 singleton for handler tests. It is
// deliberately a record-shaped policy, not an environment helper, so tests
// cannot accidentally preserve the removed configuration read path.
func testScimConfig() scimIdentityConfig {
	return scimIdentityConfig{
		defaultProfileName:      "Standard User",
		serviceProfile:          "System Administrator",
		serviceUsername:         "scim-provisioning@svc.identity",
		exempt:                  map[string]bool{"break-glass@pingcap.com": true},
		tokenTTL:                scimTokenTTL(8760),
		breakerFailureThreshold: 5,
		breakerCooldownSeconds:  300,
	}
}

func testConfigCache(cfg scimIdentityConfig) *scimConfigCache {
	cc := newScimConfigCache()
	cc.storeForTest(cfg)
	return cc
}

func testConfigRecord(cfg scimIdentityConfig) contract.Record {
	exempt := make([]string, 0, len(cfg.exempt))
	for principal := range cfg.exempt {
		exempt = append(exempt, principal)
	}
	return contract.Record{
		"Id":                         "SCIMCFG",
		scimFieldName:                scimConfigSingletonName,
		scimCfgFieldDefaultProfile:   cfg.defaultProfileName,
		scimCfgFieldServiceProfile:   cfg.serviceProfile,
		scimCfgFieldServiceUsername:  cfg.serviceUsername,
		scimCfgFieldExemptPrincipals: strings.Join(exempt, "\n"),
		scimCfgFieldBreakerFailures:  cfg.breakerFailureThreshold,
		scimCfgFieldBreakerCooldown:  cfg.breakerCooldownSeconds,
		scimCfgFieldTokenTTLHours:    int(cfg.tokenTTL / time.Hour),
		scimCfgFieldTimeZone:         cfg.timeZoneSidKey,
		scimCfgFieldLocale:           cfg.localeSidKey,
		scimCfgFieldEmailEncoding:    cfg.emailEncodingKey,
		scimCfgFieldLanguageLocale:   cfg.languageLocaleKey,
	}
}

func testConfigQuery(cfg scimIdentityConfig) *fakeQuery {
	store := newFakeRecords()
	store.put(testConfigRecord(cfg))
	return &fakeQuery{result: &contract.QueryResult{}, configRecords: store}
}

func testRouter(h *scimIdentityHandler, session *models.UserSession) *gin.Engine {
	gin.SetMode(gin.TestMode)
	r := gin.New()
	inject := func(c *gin.Context) {
		if session != nil {
			c.Set(scimSessionKey, session)
		}
		c.Next()
	}
	g := r.Group(scimBasePath, inject)
	g.GET("/Users", h.listUsers)
	g.GET("/Users/:id", h.getUser)
	g.POST("/Users", h.createUser)
	g.PUT("/Users/:id", h.replaceUser)
	g.PATCH("/Users/:id", h.patchUser)
	g.DELETE("/Users/:id", h.deactivateUser)
	return r
}

func testSession() *models.UserSession {
	e := "svc.scim@pingcap.com"
	return &models.UserSession{ID: "005SVC", Name: "SCIM Service", Email: &e, ProfileID: "PROFILE1"}
}

func do(r *gin.Engine, method, path, body string) *httptest.ResponseRecorder {
	req := httptest.NewRequest(method, path, bytes.NewBufferString(body))
	w := httptest.NewRecorder()
	r.ServeHTTP(w, req)
	return w
}

func decode(t *testing.T, w *httptest.ResponseRecorder) map[string]any {
	t.Helper()
	var m map[string]any
	if err := json.Unmarshal(w.Body.Bytes(), &m); err != nil {
		t.Fatalf("decode body %q: %v", w.Body.String(), err)
	}
	return m
}

// --- tests ----------------------------------------------------------------

func TestScimBearer_RejectsMissingAndBad(t *testing.T) {
	mw := scimBearerMiddleware(func(_ context.Context, token string) (*models.UserSession, error) {
		if token == "good" {
			return testSession(), nil
		}
		return nil, apierr.Unauthorized("bad")
	})
	gin.SetMode(gin.TestMode)
	r := gin.New()
	r.GET("/scim/v2/Ping", mw, func(c *gin.Context) { c.Status(200) })

	cases := []struct {
		hdr  string
		want int
	}{
		{"", 401},
		{"Bearer bad", 401},
		{"Bearer good", 200},
		{"bearer good", 200}, // case-insensitive scheme
	}
	for _, tc := range cases {
		req := httptest.NewRequest("GET", "/scim/v2/Ping", nil)
		if tc.hdr != "" {
			req.Header.Set("Authorization", tc.hdr)
		}
		w := httptest.NewRecorder()
		r.ServeHTTP(w, req)
		if w.Code != tc.want {
			t.Errorf("hdr=%q code=%d want=%d", tc.hdr, w.Code, tc.want)
		}
		if tc.want == 401 && w.Header().Get("WWW-Authenticate") == "" {
			t.Errorf("401 must set WWW-Authenticate")
		}
	}
}

func TestScimCreate_SSOOnlyWithDefaults(t *testing.T) {
	h, rec, _ := newTestHandler()
	r := testRouter(h, testSession())
	body := `{"schemas":["` + scim.SchemaUser + `"],"userName":"wei.zhang@pingcap.com",
		"name":{"givenName":"Wei","familyName":"Zhang"},
		"emails":[{"value":"wei.zhang@pingcap.com","primary":true}],"title":"Engineer"}`
	w := do(r, "POST", "/scim/v2/Users", body)
	if w.Code != 201 {
		t.Fatalf("create code=%d body=%s", w.Code, w.Body.String())
	}
	if rec.inserted["ProfileId"] != "PROFILE1" {
		t.Errorf("ProfileId not resolved: %v", rec.inserted["ProfileId"])
	}
	if rec.inserted[scim.FieldAlias] == nil || rec.inserted[scim.FieldIsActive] != true {
		t.Errorf("defaults missing: %+v", rec.inserted)
	}
	if _, leaked := rec.inserted["Password"]; leaked {
		t.Error("create must never set Password (SSO-only)")
	}
	if rec.inserted[scim.FieldUsername] != "wei.zhang@pingcap.com" || rec.inserted[scim.FieldLastName] != "Zhang" {
		t.Errorf("mapped fields wrong: %+v", rec.inserted)
	}
	// Locale/timezone are OMITTED when not configured, so the platform's own
	// User metadata defaults apply (no hardcoded parallel default).
	for _, f := range []string{scim.FieldTimeZoneSidKey, scim.FieldLocaleSidKey, scim.FieldEmailEncodingKey, scim.FieldLanguageLocale} {
		if _, present := rec.inserted[f]; present {
			t.Errorf("unconfigured locale field %s must be omitted so the platform default applies: %+v", f, rec.inserted)
		}
	}
}

func TestScimCreate_ConflictOnExistingUsername(t *testing.T) {
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005OLD", scim.FieldUsername: "dup@pingcap.com"})
	r := testRouter(h, testSession())
	body := `{"schemas":["` + scim.SchemaUser + `"],"userName":"dup@pingcap.com"}`
	w := do(r, "POST", "/scim/v2/Users", body)
	if w.Code != 409 {
		t.Fatalf("expected 409, got %d", w.Code)
	}
	if decode(t, w)["scimType"] != "uniqueness" {
		t.Error("409 should carry scimType uniqueness")
	}
}

func TestScimGet_FoundAndMissing(t *testing.T) {
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession())

	w := do(r, "GET", "/scim/v2/Users/005A", "")
	if w.Code != 200 {
		t.Fatalf("get found code=%d", w.Code)
	}
	m := decode(t, w)
	if m["id"] != "005A" || m["userName"] != "a@pingcap.com" || m["active"] != true {
		t.Errorf("render wrong: %+v", m)
	}
	if w := do(r, "GET", "/scim/v2/Users/nope", ""); w.Code != 404 {
		t.Errorf("missing get should 404, got %d", w.Code)
	}
}

func TestScimList_FilterHitAndMiss(t *testing.T) {
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	rec.put(contract.Record{"Id": "005B", scim.FieldUsername: "legacy-login", scim.FieldEmail: "legacy@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession())
	query := &fakeQuery{result: &contract.QueryResult{TotalSize: 1, Records: []contract.Record{
		{"Id": "005B", scim.FieldUsername: "legacy-login", scim.FieldEmail: "legacy@pingcap.com", scim.FieldIsActive: true},
	}}}
	h.queryFn = func() contract.Query { return query }

	filterURL := func(f string) string { return "/scim/v2/Users?filter=" + url.QueryEscape(f) }

	w := do(r, "GET", filterURL(`userName eq "a@pingcap.com"`), "")
	m := decode(t, w)
	if w.Code != 200 || int(m["totalResults"].(float64)) != 1 {
		t.Fatalf("filter hit wrong: code=%d body=%s", w.Code, w.Body.String())
	}
	w = do(r, "GET", filterURL(`emails.value eq "legacy@pingcap.com"`), "")
	m = decode(t, w)
	if w.Code != 200 || int(m["totalResults"].(float64)) != 1 {
		t.Fatalf("email filter hit wrong: code=%d body=%s", w.Code, w.Body.String())
	}
	if query.lastSpec.Filter.Field != scim.FieldEmail || query.lastSpec.Filter.Op != "=" || query.lastSpec.Filter.Value != "legacy@pingcap.com" || query.lastSpec.Limit != 2 {
		t.Fatalf("email lookup query=%#v, want exact Email predicate and limit 2", query.lastSpec)
	}
	w = do(r, "GET", filterURL(`userName eq "ghost@pingcap.com"`), "")
	m = decode(t, w)
	if int(m["totalResults"].(float64)) != 0 {
		t.Errorf("filter miss should be 0, got %v", m["totalResults"])
	}
	// unsupported filter -> 400
	if w := do(r, "GET", filterURL(`email eq "x"`), ""); w.Code != 400 {
		t.Errorf("unsupported filter should 400, got %d", w.Code)
	}
}

func TestScimList_EmailFilterFailsClosedOnDuplicateOrUnknownTotal(t *testing.T) {
	filterURL := func(f string) string { return "/scim/v2/Users?filter=" + url.QueryEscape(f) }
	for name, result := range map[string]*contract.QueryResult{
		"duplicate": {
			TotalSize: 2,
			Records: []contract.Record{
				{"Id": "005A", scim.FieldEmail: "shared@pingcap.com"},
				{"Id": "005B", scim.FieldEmail: "shared@pingcap.com"},
			},
		},
		"unknown total": {
			TotalSizeUnknown: true,
			Records:          []contract.Record{{"Id": "005A", scim.FieldEmail: "shared@pingcap.com"}},
		},
		"inconsistent total": {
			TotalSize: 1,
		},
	} {
		t.Run(name, func(t *testing.T) {
			h, _, _ := newTestHandler()
			h.queryFn = func() contract.Query { return &fakeQuery{result: result} }
			w := do(testRouter(h, testSession()), "GET", filterURL(`emails.value eq "shared@pingcap.com"`), "")
			if name == "duplicate" && w.Code != http.StatusConflict {
				t.Fatalf("duplicate email filter code=%d body=%s, want 409", w.Code, w.Body.String())
			}
			if (name == "unknown total" || name == "inconsistent total") && w.Code != http.StatusServiceUnavailable {
				t.Fatalf("unverifiable email filter code=%d body=%s, want 503", w.Code, w.Body.String())
			}
		})
	}
}

func TestScimPatch_Deactivate(t *testing.T) {
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession())
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false}}]}`
	w := do(r, "PATCH", "/scim/v2/Users/005A", body)
	if w.Code != 200 {
		t.Fatalf("patch code=%d body=%s", w.Code, w.Body.String())
	}
	if rec.updates[scim.FieldIsActive] != false {
		t.Errorf("patch should set IsActive=false, got %+v", rec.updates)
	}
}

func TestScimDelete_Deactivates(t *testing.T) {
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 204 {
		t.Fatalf("delete code=%d", w.Code)
	}
	if rec.updatedID != "005A" || rec.updates[scim.FieldIsActive] != false {
		t.Errorf("delete should deactivate, got id=%q updates=%+v", rec.updatedID, rec.updates)
	}
}

func TestScimUpdate_NeverRewritesLoginIdentity(t *testing.T) {
	// A PUT/PATCH addresses the record by URL :id; it must NEVER rewrite the
	// login identity (Username/Email), even though the dialect maps
	// userName/primary-email into those fields. Other attributes still apply.
	for _, method := range []string{"PUT", "PATCH"} {
		h, rec, _ := newTestHandler()
		rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com",
			scim.FieldTitle: "Old", scim.FieldIsActive: true})
		r := testRouter(h, testSession())
		var body string
		if method == "PUT" {
			body = `{"schemas":["` + scim.SchemaUser + `"],"userName":"hijack@evil.com","emails":[{"value":"hijack@evil.com","primary":true}],"title":"New","active":true}`
		} else {
			body = `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"userName":"hijack@evil.com","title":"New"}}]}`
		}
		w := do(r, method, "/scim/v2/Users/005A", body)
		if w.Code != 200 {
			t.Fatalf("%s code=%d body=%s", method, w.Code, w.Body.String())
		}
		if _, present := rec.updates[scim.FieldUsername]; present {
			t.Errorf("%s must never write Username, got %+v", method, rec.updates)
		}
		if _, present := rec.updates[scim.FieldEmail]; present {
			t.Errorf("%s must never write Email, got %+v", method, rec.updates)
		}
		if rec.byID["005A"][scim.FieldUsername] != "a@pingcap.com" || rec.byID["005A"][scim.FieldEmail] != "a@pingcap.com" {
			t.Errorf("%s must leave the login identity intact, got %+v", method, rec.byID["005A"])
		}
	}
}

func TestScim_AdminSurfaceRejectsPatSession(t *testing.T) {
	// A personal-access-token session must not reach the admin surface: it
	// could mint a SCIM bearer that survives revocation of the PAT.
	gin.SetMode(gin.TestMode)
	run := func(grant string) int {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		c.Request = httptest.NewRequest("POST", "/services/identity-sync/scim-token", nil)
		c.Set(constants.ContextKeyUser, kernelauth.UserSession{ID: "005X", GrantType: grant})
		scimRejectApiTokenSession(c)
		if !c.IsAborted() {
			return 0 // passed through
		}
		return w.Code
	}
	if code := run("personal_access_token"); code != 403 { // == auth.ApiTokenGrantType
		t.Fatalf("PAT session must be refused with 403, got %d", code)
	}
	if code := run("oauth"); code != 0 {
		t.Fatalf("an interactive session must pass through, got abort with %d", code)
	}
}

func TestScimPolicyReadMustBeFreshForStateChanges(t *testing.T) {
	for _, tt := range []struct {
		method string
		want   bool
	}{
		{method: http.MethodGet, want: false},
		{method: http.MethodPost, want: true},
		{method: http.MethodPut, want: true},
		{method: http.MethodPatch, want: true},
		{method: http.MethodDelete, want: true},
	} {
		if got := scimPolicyReadMustBeFresh(tt.method); got != tt.want {
			t.Errorf("fresh L2 read for %s = %t, want %t", tt.method, got, tt.want)
		}
	}
}

func TestScimPut_ClearsAbsentManagedFields(t *testing.T) {
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com",
		scim.FieldTitle: "Old", scim.FieldIsActive: true})
	r := testRouter(h, testSession())
	body := `{"schemas":["` + scim.SchemaUser + `"],"userName":"a@pingcap.com","active":true}`
	w := do(r, "PUT", "/scim/v2/Users/005A", body)
	if w.Code != 200 {
		t.Fatalf("put code=%d body=%s", w.Code, w.Body.String())
	}
	// Title was absent in the PUT payload -> managed field cleared to nil.
	v, ok := rec.updates[scim.FieldTitle]
	if !ok || v != nil {
		t.Errorf("PUT should clear absent Title to nil, got ok=%v v=%v", ok, v)
	}
}

func TestScim_InProcessDisabledByDefault(t *testing.T) {
	gin.SetMode(gin.TestMode)
	// Flag unset -> the surface must not mount any /scim/v2 route.
	r := gin.New()
	if err := registerScimIdentitySurface(serverruntime.RuntimeRouteSurfaceContext{Router: r, ServiceManager: &services.ServiceManager{}}); err != nil {
		t.Fatalf("register (disabled): %v", err)
	}
	for _, rt := range r.Routes() {
		if strings.HasPrefix(rt.Path, scimBasePath) || strings.HasPrefix(rt.Path, "/services/identity-sync") {
			t.Fatalf("in-process SCIM must not mount when disabled, found %s", rt.Path)
		}
	}
	// Flag on -> /scim/v2/Users mounts.
	t.Setenv("NEXUSCRM_SCIM_INPROCESS_ENABLED", "true")
	r2 := gin.New()
	if err := registerScimIdentitySurface(serverruntime.RuntimeRouteSurfaceContext{Router: r2, ServiceManager: &services.ServiceManager{}}); err != nil {
		t.Fatalf("register (enabled): %v", err)
	}
	mounted := false
	for _, rt := range r2.Routes() {
		if rt.Path == scimBasePath+"/Users" {
			mounted = true
		}
	}
	if !mounted {
		t.Fatal("enabled in-process SCIM must mount /scim/v2/Users")
	}
}

func TestScim_VolumeBreakerTripsOnRunaway(t *testing.T) {
	s := newScimIdentityState(time.Now().UTC(), 5, 300) // maxDeactivations = 50
	now := time.Now().UTC()
	// Volume is enforced at reservation (before the write), so a burst cannot
	// exceed the cap between check and write.
	for i := 0; i < s.maxDeactivations; i++ {
		if !s.reserveDeactivation(now) {
			t.Fatalf("reservation %d within the cap should be allowed", i)
		}
	}
	if st, _ := s.breakerStatus(); st != "closed" {
		t.Fatalf("breaker should still be closed at the cap, got %s", st)
	}
	if s.reserveDeactivation(now) { // one past the cap must be refused and trip
		t.Fatal("over-cap reservation must be refused")
	}
	if st, _ := s.breakerStatus(); st != "open" {
		t.Fatalf("volume breaker must open after exceeding the per-window cap, got %s", st)
	}
}

// --- durable deactivation volume bound (design §6.1 layer 3) ----------------

func TestScim_DurableVolumeBoundRefusesAtCap(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	al.rows = headRows(h.state.volumeCap(), "UPDATE", strPtr(`{"fields":["Title","IsActive"]}`))
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 503 {
		t.Fatalf("delete at durable cap must 503, got %d body=%s", w.Code, w.Body.String())
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen when the durable bound refuses, got update on %q", rec.updatedID)
	}
}

func TestScim_DurableVolumeBoundRefusesPatchDeactivateAtCap(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	al.rows = headRows(h.state.volumeCap(), "UPDATE", strPtr(`{"fields":["IsActive"]}`))
	r := testRouter(h, testSession())
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false}}]}`
	w := do(r, "PATCH", "/scim/v2/Users/005A", body)
	if w.Code != 503 {
		t.Fatalf("patch-deactivate at durable cap must 503, got %d body=%s", w.Code, w.Body.String())
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen when the durable bound refuses, got update on %q", rec.updatedID)
	}
}

func TestScim_DurableVolumeBoundSkipsCreatesAndUnrelatedUpdates(t *testing.T) {
	// CREATE heads (the dominant legitimate SCIM volume) and UPDATE heads that
	// did not touch IsActive must not count toward the deactivation cap.
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	al.rows = append(
		headRows(h.state.volumeCap(), "CREATE", nil),
		headRows(h.state.volumeCap(), "UPDATE", strPtr(`{"fields":["Title","Department"]}`))...,
	)
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 204 {
		t.Fatalf("creates/unrelated updates must not count, got %d body=%s", w.Code, w.Body.String())
	}
	if rec.updatedID != "005A" || rec.updates[scim.FieldIsActive] != false {
		t.Errorf("expected the deactivation write, got id=%q updates=%+v", rec.updatedID, rec.updates)
	}
}

func TestScim_DurableVolumeBoundBelowCapAllows(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	al.rows = headRows(h.state.volumeCap()-1, "UPDATE", strPtr(`{"fields":["IsActive"]}`))
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 204 {
		t.Fatalf("below-cap deactivation must proceed, got %d body=%s", w.Code, w.Body.String())
	}
}

func TestScim_DurableVolumeBoundFailsClosedOnAuditError(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	al.err = fmt.Errorf("audit backend unavailable")
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 503 {
		t.Fatalf("an unverifiable durable bound must refuse (503), got %d body=%s", w.Code, w.Body.String())
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen when the bound is unverifiable, got update on %q", rec.updatedID)
	}
}

func TestScim_DurableVolumeBoundCountsUnreadableUpdateHeads(t *testing.T) {
	// UPDATE heads whose change scope cannot be read count toward the bound
	// (fail closed) — both missing and malformed ChangesJSON.
	for name, cj := range map[string]*string{
		"nil":         nil,
		"garbage":     strPtr("{not json"),
		"emptyObject": strPtr("{}"),
		"nullFields":  strPtr(`{"fields":null}`),
		"emptyFields": strPtr(`{"fields":[]}`),
	} {
		t.Run(name, func(t *testing.T) {
			h, rec, al := newTestHandler()
			rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
			al.rows = headRows(h.state.volumeCap(), "UPDATE", cj)
			r := testRouter(h, testSession())
			w := do(r, "DELETE", "/scim/v2/Users/005A", "")
			if w.Code != 503 {
				t.Fatalf("unreadable UPDATE heads must count toward the bound (fail closed), got %d", w.Code)
			}
			if rec.updatedID != "" {
				t.Errorf("no write may happen when the bound refuses, got update on %q", rec.updatedID)
			}
		})
	}
}

func TestScim_DurableVolumeBoundSingleQueryFailsClosedOnBudget(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	// A single non-paginated query is used (offset pagination over a UUID-keyed
	// audit result could skip rows). When that ONE query fills the scan budget,
	// matching rows may exist beyond it, so a below-cap count is not
	// authoritative — fail closed.
	al.fn = func(req contract.AuditLogQueryRequest) (*contract.AuditLogQueryResult, error) {
		return &contract.AuditLogQueryResult{
			Rows:   headRows(req.Limit, "CREATE", nil), // full page of non-counting noise
			Limit:  req.Limit,
			Offset: req.Offset,
		}, nil
	}
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 503 {
		t.Fatalf("budget exhaustion must refuse (fail closed), got %d body=%s", w.Code, w.Body.String())
	}
	// Exactly ONE audit query — no offset pagination.
	if len(al.reqs) != 1 {
		t.Errorf("the durable bound must issue a single query (no offset pagination), saw %d", len(al.reqs))
	}
	if al.reqs[0].Offset != 0 {
		t.Errorf("the durable bound query must not use OFFSET, got offset=%d", al.reqs[0].Offset)
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen when the bound refuses, got update on %q", rec.updatedID)
	}
}

func TestScim_BinderRejectsElevatedGrantAfterMint(t *testing.T) {
	// A permission-set assignment granted to the service user AFTER the token
	// was minted must immediately break the per-request binding — the bearer's
	// authority cannot silently expand without rotation.
	rec := newFakeRecords()
	svcName := "scim-provisioning@svc.identity"
	rec.put(contract.Record{"Id": "PROFILE_SVC", scimFieldName: "SCIM Service"})
	rec.put(contract.Record{"Id": "005SVC", scim.FieldUsername: svcName,
		scim.FieldEmail: svcName, scim.FieldProfileID: "PROFILE_SVC", scim.FieldIsActive: true})
	binder := &scimPrincipalBinder{
		serviceUsername: svcName,
		serviceProfile:  "SCIM Service",
		records:         func() contract.Records { return rec },
	}
	sess := &models.UserSession{ID: "005SVC", ProfileID: "PROFILE_SVC"}
	// Clean account verifies.
	if err := binder.verify(context.Background(), sess, svcName, "SCIM Service"); err != nil {
		t.Fatalf("clean service principal must verify: %v", err)
	}
	// A permission-set assignment appears — the next request must be refused.
	rec.putUnderField(constants.TablePermissionSetAssignment, constants.FieldPermissionSetAssignment_AssigneeID, "005SVC",
		contract.Record{"Id": "PSA1", constants.FieldPermissionSetAssignment_AssigneeID: "005SVC"})
	if err := binder.verify(context.Background(), sess, svcName, "SCIM Service"); err == nil {
		t.Fatal("an elevated grant added after mint must break the per-request binding")
	}
}

func TestScim_LatchesWhenAuditNotRecording(t *testing.T) {
	// If the audit trail observably fails to record a successful deactivation
	// (the transactional head row is missing on re-check), the surface must
	// latch further deactivations off until an operator resets the breaker —
	// an unrecorded deactivation would blind the durable volume bound.
	h, rec, al := newTestHandler()
	rec.onUpdate = nil // simulate an audit trail that records nothing
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	rec.put(contract.Record{"Id": "005B", scim.FieldUsername: "b@pingcap.com", scim.FieldEmail: "b@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession())

	// First deactivation: the write itself succeeds (204)…
	if w := do(r, "DELETE", "/scim/v2/Users/005A", ""); w.Code != 204 {
		t.Fatalf("first delete should succeed, got %d", w.Code)
	}
	// …but the missing head row latches all further deactivations off.
	if w := do(r, "DELETE", "/scim/v2/Users/005B", ""); w.Code != 503 {
		t.Fatalf("post-latch delete must 503, got %d", w.Code)
	}
	if rec.byID["005B"][scim.FieldIsActive] != true {
		t.Fatal("latched deactivation must not write")
	}
	if st, _ := h.state.breakerStatus(); st != "open" {
		t.Fatalf("latch must read open, got %s", st)
	}
	found := false
	for _, ev := range h.state.journal {
		if strings.Contains(ev.detail, "latched") {
			found = true
		}
	}
	if !found {
		t.Errorf("journal should record the latch, got %+v", h.state.journal)
	}
	// The explicit operator reset is the only way out of a latch.
	h.state.resetBreaker()
	if !h.state.reserveDeactivation(time.Now().UTC()) {
		t.Fatal("operator reset must clear the latch")
	}
	_ = al
}

func TestScim_LatchNeverHalfOpens(t *testing.T) {
	s := newScimIdentityState(time.Now().UTC(), 5, 1) // 1s cooldown
	s.latch(time.Now().UTC().Add(-time.Hour), "audit trail not recording")
	if st, _ := s.breakerStatus(); st != "open" {
		t.Fatalf("a latch must never report half_open, got %s", st)
	}
	if s.reserveDeactivation(time.Now().UTC()) {
		t.Fatal("a latch must refuse reservations even after the cooldown")
	}
}

func TestScim_DeleteAlreadyInactiveIsIdempotent(t *testing.T) {
	// A routine IdP retry of a DELETE must be an idempotent success with no
	// write and NO deactivation machinery — an unchanged IsActive=false update
	// emits no audit head row, so re-running the machinery would false-latch.
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: false})
	r := testRouter(h, testSession())
	for i := 0; i < 2; i++ {
		if w := do(r, "DELETE", "/scim/v2/Users/005A", ""); w.Code != 204 {
			t.Fatalf("already-inactive DELETE %d must 204, got %d", i, w.Code)
		}
	}
	if rec.updatedID != "" {
		t.Errorf("idempotent DELETE must not write, got update on %q", rec.updatedID)
	}
	if len(al.reqs) != 0 {
		t.Errorf("idempotent DELETE must not run the durable-bound query, saw %d", len(al.reqs))
	}
	if st, _ := h.state.breakerStatus(); st != "closed" {
		t.Fatalf("idempotent DELETE must never latch, breaker=%s", st)
	}
}

func TestScim_PatchDeactivateAlreadyInactiveIsIdempotent(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: false, scim.FieldTitle: "Old"})
	r := testRouter(h, testSession())
	// active=false retry bundled with a real attribute change: the attribute
	// still applies; the deactivation machinery must not run.
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false,"title":"New"}}]}`
	w := do(r, "PATCH", "/scim/v2/Users/005A", body)
	if w.Code != 200 {
		t.Fatalf("already-inactive PATCH must 200, got %d body=%s", w.Code, w.Body.String())
	}
	if rec.updates[scim.FieldTitle] != "New" {
		t.Errorf("other attribute changes must still apply, got %+v", rec.updates)
	}
	// The stale IsActive=false must be DROPPED from the write set: if a
	// concurrent actor reactivated the user between read and write, a
	// leftover IsActive=false would re-deactivate while bypassing every guard.
	if _, present := rec.updates[scim.FieldIsActive]; present {
		t.Errorf("the idempotent path must not write IsActive at all, got %+v", rec.updates)
	}
	if len(al.reqs) != 0 {
		t.Errorf("idempotent deactivation must not run the durable-bound query, saw %d", len(al.reqs))
	}
	if st, _ := h.state.breakerStatus(); st != "closed" {
		t.Fatalf("idempotent deactivation must never latch, breaker=%s", st)
	}
}

func TestScim_ExpiryEnforcedPerRequest(t *testing.T) {
	// A process that mounted the surface while the exception was valid must
	// self-disable at the recorded date, not at its next restart: the expiry
	// is re-checked on every request.
	gin.SetMode(gin.TestMode)
	t.Setenv("NEXUSCRM_SCIM_INPROCESS_ENABLED", "true")
	r := gin.New()
	if err := registerScimIdentitySurface(serverruntime.RuntimeRouteSurfaceContext{Router: r, ServiceManager: &services.ServiceManager{}}); err != nil {
		t.Fatalf("register: %v", err)
	}
	saved := scimInterimExceptionExpiry
	defer func() { scimInterimExceptionExpiry = saved }()
	scimInterimExceptionExpiry = time.Now().UTC().Add(-time.Hour) // lapse it mid-flight

	if w := do(r, "GET", "/scim/v2/Users", ""); w.Code != 503 {
		t.Fatalf("serving past expiry must 503, got %d body=%s", w.Code, w.Body.String())
	}
	if w := do(r, "POST", "/services/identity-sync/scim-token", ""); w.Code != 503 {
		t.Fatalf("token mint past expiry must 503, got %d", w.Code)
	} else if got := w.Header().Get("Content-Type"); got != "application/json; charset=utf-8" {
		t.Fatalf("expired admin response content type=%q, want ordinary JSON for the CRM setup client", got)
	}

	// Before the expiry the same routes proceed past the expiry middleware
	// (401 from bearer auth on the unset test wiring, NOT the 503 refusal).
	scimInterimExceptionExpiry = time.Now().UTC().Add(time.Hour)
	if w := do(r, "GET", "/scim/v2/Users", ""); w.Code == 503 {
		t.Fatalf("un-lapsed surface must not refuse with 503, got %d", w.Code)
	}
}

// fakeTokenMinter records rotation calls for scimMintServiceToken.
type fakeTokenMinter struct {
	bulkRevokes   []string // "name|keepID" of each RevokeAllNamedForOwnerExcept call
	created       int
	createErr     error
	bulkRevokeErr error
}

func (f *fakeTokenMinter) RevokeAllNamedForOwnerExcept(_ context.Context, _ string, name, keepID string) (int64, error) {
	if f.bulkRevokeErr != nil {
		return 0, f.bulkRevokeErr
	}
	f.bulkRevokes = append(f.bulkRevokes, name+"|"+keepID)
	return 1, nil
}

func (f *fakeTokenMinter) Revoke(_ context.Context, id, _ string) (bool, error) {
	return true, nil
}

func (f *fakeTokenMinter) CreateToken(context.Context, string, string, string, string, *time.Time) (string, *models.SystemApiToken, error) {
	if f.createErr != nil {
		return "", nil, f.createErr
	}
	f.created++
	return "nxt_new-token", &models.SystemApiToken{ID: "NEWTOK"}, nil
}

// fakeTxRunner runs fn and reports whether the "transaction" committed
// (fn returned nil) or rolled back (fn errored).
type fakeTxRunner struct {
	ran        int
	committed  int
	rolledBack int
}

func (f *fakeTxRunner) WithTransaction(ctx context.Context, fn func(context.Context) error) error {
	f.ran++
	if err := fn(ctx); err != nil {
		f.rolledBack++
		return err
	}
	f.committed++
	return nil
}

func TestScim_MintRefusesInsufficientScopeBeforeRotating(t *testing.T) {
	// A session whose scope cannot mint a scim token must be refused BEFORE
	// any transaction runs — no creation, no revocation.
	f, txm := &fakeTokenMinter{}, &fakeTxRunner{}
	_, err := scimMintServiceToken(context.Background(), txm, f, time.Hour, "005SVC", "api")
	if err == nil {
		t.Fatal("an 'api'-scoped session must not mint a scim token")
	}
	var se *scim.Error
	if !errors.As(err, &se) || se.Status != 403 {
		t.Fatalf("scope refusal should be a 403 scim error, got %T %v", err, err)
	}
	if txm.ran != 0 || len(f.bulkRevokes) != 0 || f.created != 0 {
		t.Fatalf("scope refusal must happen before ANY transaction: txm=%+v f=%+v", txm, f)
	}
}

func TestScim_MintRunsCreateAndRevokeInOneTransaction(t *testing.T) {
	for _, scope := range []string{"", "api scim"} {
		f, txm := &fakeTokenMinter{}, &fakeTxRunner{}
		plaintext, err := scimMintServiceToken(context.Background(), txm, f, time.Hour, "005SVC", scope)
		if err != nil {
			t.Fatalf("scope %q: mint failed: %v", scope, err)
		}
		if plaintext == "" || f.created != 1 {
			t.Fatalf("scope %q: expected one minted token", scope)
		}
		if txm.committed != 1 || txm.rolledBack != 0 {
			t.Fatalf("scope %q: create+revoke must commit as one transaction, txm=%+v", scope, txm)
		}
		// The bulk revoke is one atomic statement, scoped to the SCIM token
		// NAME (the service account's unrelated tokens are never touched) and
		// keeping only the new id.
		if len(f.bulkRevokes) != 1 || f.bulkRevokes[0] != scimTokenName+"|NEWTOK" {
			t.Fatalf("scope %q: expected one name-scoped bulk revoke keeping NEWTOK, got %v", scope, f.bulkRevokes)
		}
	}
}

func TestScim_MintCreateFailureRollsBackTransaction(t *testing.T) {
	// A failed create rolls the whole transaction back: the prior tokens stay
	// the valid credentials, nothing is revoked — the IdP is never stranded.
	f, txm := &fakeTokenMinter{createErr: fmt.Errorf("boom")}, &fakeTxRunner{}
	if _, err := scimMintServiceToken(context.Background(), txm, f, time.Hour, "005SVC", ""); err == nil {
		t.Fatal("create failure must abort the mint")
	}
	if txm.rolledBack != 1 || len(f.bulkRevokes) != 0 {
		t.Fatalf("create failure must roll back with no revocations, txm=%+v f=%+v", txm, f)
	}
}

func TestScim_MintRevokeFailureRollsBackTransaction(t *testing.T) {
	// A failed bulk revoke rolls the whole transaction back — the new token's
	// insert is undone with it, so the end state is exactly {priors}.
	f, txm := &fakeTokenMinter{bulkRevokeErr: fmt.Errorf("db down")}, &fakeTxRunner{}
	if _, err := scimMintServiceToken(context.Background(), txm, f, time.Hour, "005SVC", ""); err == nil {
		t.Fatal("bulk revoke failure must abort the mint")
	}
	if txm.rolledBack != 1 {
		t.Fatalf("revoke failure must roll back the transaction, txm=%+v", txm)
	}
}

func TestScim_MintRequiresTransactionManager(t *testing.T) {
	f := &fakeTokenMinter{}
	if _, err := scimMintServiceToken(context.Background(), nil, f, time.Hour, "005SVC", ""); err == nil {
		t.Fatal("a missing transaction manager must fail closed")
	}
	if f.created != 0 || len(f.bulkRevokes) != 0 {
		t.Fatalf("nothing may run without a transaction manager, f=%+v", f)
	}
}

func TestScim_UnreadableActivationStateFailsClosed(t *testing.T) {
	// A record whose IsActive the caller cannot read (e.g. FLS-hidden) must
	// refuse the deactivation, not sail through the guard as "already
	// inactive" — the engine's write path could still change the field.
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com"}) // no IsActive
	r := testRouter(h, testSession())
	if w := do(r, "DELETE", "/scim/v2/Users/005A", ""); w.Code != 503 {
		t.Fatalf("DELETE with unreadable IsActive must 503, got %d", w.Code)
	}
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false}}]}`
	if w := do(r, "PATCH", "/scim/v2/Users/005A", body); w.Code != 503 {
		t.Fatalf("PATCH-deactivate with unreadable IsActive must 503, got %d", w.Code)
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen when the activation state is unreadable, got update on %q", rec.updatedID)
	}
}

func TestScim_UnverifiableExemptionIdentityFailsClosed(t *testing.T) {
	// A record whose Username/Email the caller cannot read cannot be checked
	// against the exemption list — it must refuse, never pass as non-exempt.
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldIsActive: true}) // no Email
	r := testRouter(h, testSession())
	if w := do(r, "DELETE", "/scim/v2/Users/005A", ""); w.Code != 503 {
		t.Fatalf("DELETE with unverifiable identity must 503, got %d", w.Code)
	}
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false}}]}`
	if w := do(r, "PATCH", "/scim/v2/Users/005A", body); w.Code != 503 {
		t.Fatalf("PATCH-deactivate with unverifiable identity must 503, got %d", w.Code)
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen when the identity is unverifiable, got update on %q", rec.updatedID)
	}
}

func TestScim_UnverifiablePostWriteExemptionLatchesAndErrors(t *testing.T) {
	// If the post-write exemption re-check cannot run (read failure), the
	// handler must NOT report success and must latch further deactivations —
	// whether the write hit a protected principal is unknown.
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	rec.put(contract.Record{"Id": "005B", scim.FieldUsername: "b@pingcap.com", scim.FieldEmail: "b@pingcap.com", scim.FieldIsActive: true})
	rec.onUpdate = func(id string, updates contract.Record) {
		if _, ok := updates[scim.FieldIsActive]; ok {
			rec.retrieveErr = fmt.Errorf("transient read failure")
		}
	}
	r := testRouter(h, testSession())
	if w := do(r, "DELETE", "/scim/v2/Users/005A", ""); w.Code != 503 {
		t.Fatalf("unverifiable post-write exemption must NOT report success, got %d", w.Code)
	}
	if st, _ := h.state.breakerStatus(); st != "open" {
		t.Fatalf("unverifiable post-write exemption must latch, breaker=%s", st)
	}
	// The account is RESTORED too: whether the write hit a protected
	// principal is unknown, so the write is undone (a legitimate offboard is
	// redone by the IdP's retry under full verification).
	if rec.byID["005A"][scim.FieldIsActive] != true {
		t.Fatal("unverifiable post-write exemption must restore the account")
	}
	rec.retrieveErr = nil
	if w := do(r, "DELETE", "/scim/v2/Users/005B", ""); w.Code != 503 {
		t.Fatalf("post-latch deactivation must refuse, got %d", w.Code)
	}
	if rec.byID["005B"][scim.FieldIsActive] != true {
		t.Fatal("latched deactivation must not write")
	}
}

func TestScim_EmptyExemptListDisablesDeactivations(t *testing.T) {
	// Enabling SCIM without any protected principals must not allow
	// IdP-driven offboarding: deactivations refuse until the exemption list
	// is configured. Provisioning (create/update) stays available.
	rec := newFakeRecords()
	rec.put(contract.Record{"Id": "PROFILE1", scimFieldName: "Standard User"})
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com",
		scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	cfg := testScimConfig()
	cfg.exempt = map[string]bool{}
	h := &scimIdentityHandler{
		recordsFn:  func() contract.Records { return rec },
		queryFn:    func() contract.Query { return testConfigQuery(cfg) },
		auditLogFn: func() contract.AuditLog { return &fakeAuditLog{} },
		cc:         testConfigCache(cfg),
		state:      newScimIdentityState(time.Now().UTC(), 5, 300),
	}
	r := testRouter(h, testSession())
	if w := do(r, "DELETE", "/scim/v2/Users/005A", ""); w.Code != 503 {
		t.Fatalf("deactivation without an exemption list must refuse (503), got %d", w.Code)
	}
	if rec.updatedID != "" {
		t.Errorf("no write may happen without an exemption list, got update on %q", rec.updatedID)
	}
	// Non-destructive updates still work.
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","path":"title","value":"Engineer"}]}`
	if w := do(r, "PATCH", "/scim/v2/Users/005A", body); w.Code != 200 {
		t.Fatalf("non-deactivating updates must still work, got %d", w.Code)
	}
}

func TestScim_ExemptionRaceRolledBackAfterWrite(t *testing.T) {
	// The exemption guarantee survives a check-to-write race: if a concurrent
	// identity change makes the row exempt by the time the deactivation
	// lands, the post-write re-check restores the account and refuses.
	rec := newFakeRecords()
	rec.put(contract.Record{"Id": "PROFILE1", scimFieldName: "Standard User"})
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "user@pingcap.com",
		scim.FieldEmail: "user@pingcap.com", scim.FieldIsActive: true})
	al := &fakeAuditLog{}
	raced := false
	rec.onUpdate = func(id string, updates contract.Record) {
		if _, ok := updates[scim.FieldIsActive]; ok {
			al.rows = append(al.rows, isActiveHeadRow())
		}
		// Simulate the concurrent rename landing with the deactivation write:
		// the row now carries the protected identity.
		if !raced {
			raced = true
			rec.byID["005A"][scim.FieldEmail] = "admin@pingcap.com"
		}
	}
	cfg := testScimConfig()
	cfg.exempt = map[string]bool{"admin@pingcap.com": true}
	h := &scimIdentityHandler{
		recordsFn:  func() contract.Records { return rec },
		queryFn:    func() contract.Query { return testConfigQuery(cfg) },
		auditLogFn: func() contract.AuditLog { return al },
		cc:         testConfigCache(cfg),
		state:      newScimIdentityState(time.Now().UTC(), 5, 300),
	}
	r := testRouter(h, testSession())
	w := do(r, "DELETE", "/scim/v2/Users/005A", "")
	if w.Code != 403 {
		t.Fatalf("raced exempt deactivation must be refused (403), got %d body=%s", w.Code, w.Body.String())
	}
	if rec.byID["005A"][scim.FieldIsActive] != true {
		t.Fatal("the raced exempt principal must end ACTIVE (rolled back)")
	}
	found := false
	for _, ev := range h.state.journal {
		if strings.Contains(ev.detail, "rolled back") {
			found = true
		}
	}
	if !found {
		t.Errorf("journal should record the rollback, got %+v", h.state.journal)
	}
}

func TestScim_ServicePrincipalCannotDeactivateItself(t *testing.T) {
	// The SCIM bearer must never brick the integration by deactivating its
	// own service account (token validation rejects an inactive owner).
	h, rec, _ := newTestHandler()
	rec.put(contract.Record{"Id": "005SVC", scim.FieldUsername: "svc.scim@pingcap.com",
		scim.FieldEmail: "svc.scim@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession()) // session.ID == 005SVC
	if w := do(r, "DELETE", "/scim/v2/Users/005SVC", ""); w.Code != 403 {
		t.Fatalf("self-deactivation must be refused (403), got %d", w.Code)
	}
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false}}]}`
	if w := do(r, "PATCH", "/scim/v2/Users/005SVC", body); w.Code != 403 {
		t.Fatalf("self-deactivation via PATCH must be refused (403), got %d", w.Code)
	}
	if rec.byID["005SVC"][scim.FieldIsActive] != true {
		t.Fatal("the service principal must remain active")
	}
}

func TestScim_StateClampsHostileTuning(t *testing.T) {
	// A huge configured cooldown must not overflow time.Duration negative —
	// that would roll the volume window on every request and neuter the caps.
	s := newScimIdentityState(time.Now().UTC(), 1<<40, 1<<40)
	if s.cooldown <= 0 || s.cooldown > 24*time.Hour {
		t.Fatalf("cooldown must clamp to a sane positive bound, got %v", s.cooldown)
	}
	if s.maxDeactivations <= 0 {
		t.Fatalf("volume cap must stay positive, got %d", s.maxDeactivations)
	}
	if d := scimTokenTTL(1 << 50); d <= 0 || d > 87600*time.Hour {
		t.Fatalf("token TTL must clamp to a sane positive bound, got %v", d)
	}
}

func TestScim_HalfOpenTrialCountsTowardVolumeCap(t *testing.T) {
	// The half-open probe is a real deactivation and must be counted in the
	// per-window volume accounting like any other reservation.
	s := newScimIdentityState(time.Now().UTC(), 5, 1) // 1s cooldown, cap 50
	now := time.Now().UTC()
	s.onDeactivateFailure(now, "boom")
	s.onDeactivateFailure(now, "boom")
	s.onDeactivateFailure(now, "boom")
	s.onDeactivateFailure(now, "boom")
	s.onDeactivateFailure(now, "boom") // threshold 5 -> breaker opens
	if st, _ := s.breakerStatus(); st == "closed" {
		t.Fatal("breaker should be open")
	}
	probeAt := now.Add(2 * time.Second) // past cooldown -> half-open window
	if !s.reserveDeactivation(probeAt) {
		t.Fatal("the half-open probe should be admitted")
	}
	// The probe is counted in the rolling window like any other reservation.
	if got := s.rollingDeactivationCount(probeAt); got != 1 {
		t.Fatalf("the half-open probe must be counted in the volume window, got %d", got)
	}
}

func TestScim_RollingWindowNotTumbling(t *testing.T) {
	// A tumbling window would zero its counter at the boundary, so a burst
	// just before and just after the boundary could reach 2x the cap. A true
	// rolling window prunes by age, so reservations still count until a full
	// window-length has elapsed since each one.
	s := newScimIdentityState(time.Now().UTC(), 5, 100) // 100s window
	base := time.Now().UTC()
	for i := 0; i < 10; i++ {
		s.recentDeactivations = append(s.recentDeactivations, base.Add(90*time.Second))
	}
	// 10s later (t=100s, exactly one tumbling boundary), the 90s reservations
	// are only 10s old — a rolling window still counts all 10.
	if got := s.rollingDeactivationCount(base.Add(100 * time.Second)); got != 10 {
		t.Fatalf("rolling window must still count reservations within one window-length, got %d", got)
	}
	// Past a full window since the reservations, they age out.
	if got := s.rollingDeactivationCount(base.Add(191 * time.Second)); got != 0 {
		t.Fatalf("reservations older than the window must be pruned, got %d", got)
	}
}

func TestScim_NonDeactivationUpdateSkipsDurableCheck(t *testing.T) {
	h, rec, al := newTestHandler()
	rec.put(contract.Record{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldEmail: "a@pingcap.com", scim.FieldIsActive: true})
	r := testRouter(h, testSession())
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","path":"title","value":"Engineer"}]}`
	w := do(r, "PATCH", "/scim/v2/Users/005A", body)
	if w.Code != 200 {
		t.Fatalf("patch code=%d body=%s", w.Code, w.Body.String())
	}
	if len(al.reqs) != 0 {
		t.Fatalf("a non-deactivating update must not query the audit log, saw %d", len(al.reqs))
	}
}

func TestWriteScimErr_MapsEngineErrors(t *testing.T) {
	gin.SetMode(gin.TestMode)
	cases := []struct {
		err  error
		want int
	}{
		{kernelerrors.NewPermissionError("update", "User"), 403},
		{kernelerrors.NewNotFoundError("User", "005X"), 404},
		{apierr.Conflict("dup"), 409},
		{fmt.Errorf("some internal boom"), 500},
	}
	for _, tc := range cases {
		w := httptest.NewRecorder()
		c, _ := gin.CreateTestContext(w)
		writeScimErr(c, tc.err)
		if w.Code != tc.want {
			t.Errorf("err %T -> %d, want %d", tc.err, w.Code, tc.want)
		}
	}
}

func TestScimScopeGranted(t *testing.T) {
	// Only an EXACTLY-scim scope is accepted — a broader "api scim" token
	// would retain generic API authorization elsewhere (this group omits the
	// ScopeEnforcer), defeating purpose isolation. The mint always issues an
	// exactly-scim token, so nothing legitimate is rejected.
	if !scimScopeGranted("scim") || !scimScopeGranted("  scim  ") {
		t.Error("an exactly-scim scope must be accepted")
	}
	for _, s := range []string{"", "api", "api read", "scimx", "SCIM", "api scim", "scim api", "scim read"} {
		if scimScopeGranted(s) {
			t.Errorf("scope %q must not satisfy the exact SCIM scope requirement", s)
		}
	}
}

func TestScimPrincipalBinder(t *testing.T) {
	rec := newFakeRecords()
	rec.put(contract.Record{"Id": "PROF", scimFieldName: "SCIM Provisioner"})
	rec.put(contract.Record{"Id": "005SVC", scim.FieldUsername: "svc@pingcap.com", scim.FieldEmail: "svc@pingcap.com", scim.FieldProfileID: "PROF"})
	newBinder := func(profile string, r *fakeRecords) *scimPrincipalBinder {
		return &scimPrincipalBinder{serviceUsername: "svc@pingcap.com", serviceProfile: profile, records: func() contract.Records { return r }}
	}
	// Correct principal: stable id AND live session profile == configured profile.
	if err := newBinder("SCIM Provisioner", rec).verify(context.Background(), &models.UserSession{ID: "005SVC", ProfileID: "PROF"}, "svc@pingcap.com", "SCIM Provisioner"); err != nil {
		t.Fatalf("correct principal should pass: %v", err)
	}
	// A different token owner id is refused.
	if err := newBinder("SCIM Provisioner", rec).verify(context.Background(), &models.UserSession{ID: "005OTHER", ProfileID: "PROF"}, "svc@pingcap.com", "SCIM Provisioner"); err == nil {
		t.Error("a different principal id must fail closed")
	}
	// Unset profile config fails closed.
	if err := newBinder("", rec).verify(context.Background(), &models.UserSession{ID: "005SVC", ProfileID: "PROF"}, "svc@pingcap.com", ""); err == nil {
		t.Error("unset service profile must fail closed")
	}
	// KEY: the service account reassigned to a privileged profile is refused EVERY
	// request — the live session profile no longer matches the configured one.
	b := newBinder("SCIM Provisioner", rec)
	if err := b.verify(context.Background(), &models.UserSession{ID: "005SVC", ProfileID: "PROF"}, "svc@pingcap.com", "SCIM Provisioner"); err != nil {
		t.Fatalf("first request should pass: %v", err)
	}
	if err := b.verify(context.Background(), &models.UserSession{ID: "005SVC", ProfileID: "PRIVILEGED"}, "svc@pingcap.com", "SCIM Provisioner"); err == nil {
		t.Error("a reassigned (privileged) session profile must fail closed on a later request")
	}
}

func TestScimInterimExceptionLapse(t *testing.T) {
	if scimInterimExceptionLapsed(scimInterimExceptionExpiry.Add(-time.Hour)) {
		t.Error("exception must be active before its expiry")
	}
	if !scimInterimExceptionLapsed(scimInterimExceptionExpiry) {
		t.Error("exception must lapse at its expiry instant")
	}
	if !scimInterimExceptionLapsed(scimInterimExceptionExpiry.Add(time.Hour)) {
		t.Error("exception must stay lapsed after expiry")
	}
}

func TestScimTokenTTLClamp(t *testing.T) {
	if scimTokenTTL(0) != 8760*time.Hour || scimTokenTTL(-3) != 8760*time.Hour {
		t.Error("nonpositive TTL must clamp to the 1-year default (never permanent)")
	}
	if scimTokenTTL(24) != 24*time.Hour {
		t.Error("positive TTL must be honored")
	}
}

func TestScimList_CountZeroReturnsOnlyTotal(t *testing.T) {
	h, _, _ := newTestHandler()
	q := &fakeQuery{result: &contract.QueryResult{
		TotalSize: 3,
		Records:   []contract.Record{{"Id": "005A", scim.FieldUsername: "a@pingcap.com", scim.FieldIsActive: true}},
	}}
	h.queryFn = func() contract.Query { return q }
	r := testRouter(h, testSession())
	w := do(r, "GET", "/scim/v2/Users?count=0", "")
	if w.Code != 200 {
		t.Fatalf("code=%d", w.Code)
	}
	m := decode(t, w)
	if int(m["totalResults"].(float64)) != 3 {
		t.Errorf("totalResults = %v, want 3", m["totalResults"])
	}
	if res, _ := m["Resources"].([]any); len(res) != 0 {
		t.Errorf("count=0 must return no resources, got %d", len(res))
	}
}

func TestScim_ExemptPrincipalNeverDeactivated(t *testing.T) {
	rec := newFakeRecords()
	rec.put(contract.Record{"Id": "PROFILE1", scimFieldName: "Standard User"})
	rec.put(contract.Record{"Id": "005ADMIN", scim.FieldUsername: "admin@pingcap.com",
		scim.FieldEmail: "admin@pingcap.com", scim.FieldIsActive: true})
	cfg := testScimConfig()
	cfg.exempt = map[string]bool{"admin@pingcap.com": true}
	h := &scimIdentityHandler{
		recordsFn:  func() contract.Records { return rec },
		queryFn:    func() contract.Query { return testConfigQuery(cfg) },
		auditLogFn: func() contract.AuditLog { return &fakeAuditLog{} },
		cc:         testConfigCache(cfg),
		state:      newScimIdentityState(time.Now().UTC(), 5, 300),
	}
	r := testRouter(h, testSession())

	// DELETE on an exempt user must be refused and must NOT deactivate.
	w := do(r, "DELETE", "/scim/v2/Users/005ADMIN", "")
	if w.Code != 403 {
		t.Fatalf("exempt DELETE should be 403, got %d", w.Code)
	}
	if rec.updatedID == "005ADMIN" {
		t.Fatal("exempt user must never be updated/deactivated")
	}
	// PATCH active=false on an exempt user must also be refused.
	body := `{"schemas":["` + scim.SchemaPatchOp + `"],"Operations":[{"op":"replace","value":{"active":false}}]}`
	w = do(r, "PATCH", "/scim/v2/Users/005ADMIN", body)
	if w.Code != 403 {
		t.Fatalf("exempt PATCH-deactivate should be 403, got %d", w.Code)
	}
	if rec.byID["005ADMIN"][scim.FieldIsActive] != true {
		t.Fatal("exempt user stayed must remain active")
	}
}

func TestScimCreate_RejectsMalformedBody(t *testing.T) {
	h, _, _ := newTestHandler()
	r := testRouter(h, testSession())
	// not a SCIM User
	if w := do(r, "POST", "/scim/v2/Users", `{"schemas":["urn:other"]}`); w.Code != 400 {
		t.Errorf("non-User create should 400, got %d", w.Code)
	}
	// empty body
	if w := do(r, "POST", "/scim/v2/Users", ``); w.Code != 400 {
		t.Errorf("empty body should 400, got %d", w.Code)
	}
	// content-type of error is SCIM
	w := do(r, "POST", "/scim/v2/Users", `{"schemas":["urn:other"]}`)
	if ct := w.Header().Get("Content-Type"); !strings.Contains(ct, "scim+json") {
		t.Errorf("error content-type = %q, want scim+json", ct)
	}
}