package mysql

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/credentialvault"
	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/pulse/schedule"
	"git.yuntuo.hk/k2safe/orbit/pulse/service"
)

func TestCredentialPermissionOperationValidation(t *testing.T) {
	base := CredentialChangeInput{NamespaceID: testID(t), CredentialID: testID(t), ExpectedRevision: 1, Action: credentialPermissionsAction, Reason: "replace explicit actions", Actions: []string{"schedules:manage", "tasks:read"}}
	if err := validateCredentialChangeInput(base); err != nil {
		t.Fatal(err)
	}
	for _, actions := range [][]string{nil, {}, {""}, {"*"}, {"schedules:*"}, {"unknown"}, {"tasks:read", "tasks:read"}, {" tasks:read"}} {
		bad := base
		bad.Actions = actions
		if err := validateCredentialChangeInput(bad); fault.Code(err) != "INVALID_ARGUMENT" {
			t.Fatalf("actions %q accepted: %v", actions, err)
		}
	}
	for _, field := range []string{"validity", "grace"} {
		bad := base
		if field == "validity" {
			bad.ValidFor = time.Hour
		} else {
			bad.GraceFor = time.Hour
		}
		if err := validateCredentialChangeInput(bad); fault.Code(err) != "INVALID_ARGUMENT" {
			t.Fatalf("permission change accepted %s: %v", field, err)
		}
	}
	op := CredentialOperation{Version: 3, OperationID: testID(t), NamespaceID: base.NamespaceID, CredentialID: base.CredentialID, ExpectedRevision: 1, Action: credentialPermissionsAction, Reason: base.Reason, Actions: base.Actions}
	if err := ValidateCredentialOperation(op); err != nil {
		t.Fatal(err)
	}
	for _, mutate := range []func(*CredentialOperation){
		func(v *CredentialOperation) { v.Version = 1 },
		func(v *CredentialOperation) { v.Version = 2 },
		func(v *CredentialOperation) { v.Actions = []string{"tasks:read", "schedules:manage"} },
		func(v *CredentialOperation) { v.Actions = []string{"tasks:read", "tasks:read"} },
		func(v *CredentialOperation) { v.Actions = nil },
		func(v *CredentialOperation) { v.Token = "a token is forbidden" },
		func(v *CredentialOperation) { v.ExpiresAt = timePtr(time.Now().Add(time.Hour)) },
		func(v *CredentialOperation) { v.PreviousValidUntil = timePtr(time.Now().Add(time.Hour)) },
		func(v *CredentialOperation) { v.Action = credentialExpiryAction },
	} {
		bad := op
		mutate(&bad)
		if err := ValidateCredentialOperation(bad); fault.Code(err) != "INVALID_ARGUMENT" {
			t.Fatalf("invalid permission operation accepted: %v", err)
		}
	}
	// 旧固定文件的精确字段、序列化原文及摘要保持不变；包括旧轮换的秘密字段。
	type legacyOperation struct {
		Version            int        `json:"version"`
		OperationID        core.ID    `json:"operation_id"`
		NamespaceID        core.ID    `json:"namespace_id"`
		CredentialID       core.ID    `json:"credential_id"`
		ExpectedRevision   uint64     `json:"expected_revision"`
		Action             string     `json:"action"`
		Reason             string     `json:"reason"`
		Token              string     `json:"token,omitempty"`
		PreviousValidUntil *time.Time `json:"previous_valid_until,omitempty"`
		ExpiresAt          *time.Time `json:"expires_at,omitempty"`
	}
	for _, version := range []int{1, 2} {
		for _, action := range []string{credentialRotateAction, credentialRevokeAction, credentialExpiryAction, credentialFinalizeAction} {
			if version == 1 && action != credentialRotateAction && action != credentialRevokeAction {
				continue
			}
			old := legacyOperation{Version: version, OperationID: testID(t), NamespaceID: base.NamespaceID, CredentialID: base.CredentialID, ExpectedRevision: 1, Action: action, Reason: "legacy frozen shape"}
			if action == credentialRotateAction {
				old.Token, _, _ = auth.Issue(base.NamespaceID, base.CredentialID)
				old.ExpiresAt = timePtr(time.Now().Add(time.Hour))
			}
			raw, _ := json.Marshal(old)
			var current CredentialOperation
			if err := json.Unmarshal(raw, &current); err != nil || ValidateCredentialOperation(current) != nil {
				t.Fatal("legacy operation no longer valid", err)
			}
			currentRaw, _ := json.Marshal(current)
			oldDigest, _ := requestDigest(old)
			currentDigest, _ := requestDigest(current)
			if !bytes.Equal(raw, currentRaw) || oldDigest != currentDigest {
				t.Fatal("legacy JSON or digest changed")
			}
			for _, actions := range [][]string{{}, {"tasks:read"}} {
				current.Actions = actions
				if err := ValidateCredentialOperation(current); fault.Code(err) != "INVALID_ARGUMENT" {
					t.Fatal("legacy operation accepted actions", err)
				}
				bad := base
				bad.Action, bad.Actions = action, actions
				if err := validateCredentialChangeInput(bad); fault.Code(err) != "INVALID_ARGUMENT" {
					t.Fatal("legacy preparation accepted actions", err)
				}
			}
		}
	}
}

func TestIntegrationCredentialPermissions(t *testing.T) {
	ctx, db := overlapDB(t)
	ring, err := credentialvault.Read(strings.NewReader(`{"active":"test","keys":{"test":"` + base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{43}, 32)) + `"}}`))
	if err != nil {
		t.Fatal(err)
	}
	var initial BootstrapResult
	if err := Bootstrap(ctx, db, BootstrapInput{Application: "permissions-" + string(testID(t)), Namespace: "default", Environment: "test", Queue: "work", ValidFor: 48 * time.Hour}, func(v BootstrapResult) error { initial = v; return nil }, ring); err != nil {
		t.Fatal(err)
	}
	oldToken, _ := auth.Parse(initial.Token)
	tasks := NewTaskStore(db)
	otherQueue, err := service.New(tasks).CreateQueue(ctx, oldToken, string(testID(t)), service.QueueInput{Name: "outside-scope", MaxConcurrency: 1, RatePerSecond: 1})
	if err != nil {
		t.Fatal(err)
	}
	// 隔离测试夹具先收窄队列，证明添加动作不能隐式恢复命名空间通配范围。
	queues, _ := json.Marshal([]string{string(initial.QueueID)})
	if _, err := db.ExecContext(ctx, `UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',CAST(? AS JSON)) WHERE namespace_id=? AND id=?`, queues, initial.NamespaceID, initial.CredentialID); err != nil {
		t.Fatal(err)
	}
	schedules, err := NewScheduleStore(db, testID(t))
	if err != nil {
		t.Fatal(err)
	}
	scheduleAPI := schedule.New(schedules)
	plan := schedule.CreateInput{Name: "permission-check", Paused: true, Definition: schedule.Definition{
		Rule:   schedule.Rule{Kind: "FIXED_RATE", Timezone: "UTC", StartAt: time.Now().UTC().Add(time.Hour), IntervalMS: 60000},
		Policy: schedule.Policy{Misfire: "SKIP", Overlap: "PARALLEL", CatchupLimit: 1, CatchupWindowMS: 30000, GraceMS: 1000},
		Task:   service.SubmitInput{QueueID: initial.QueueID, Handler: "permission.check", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)},
	}}
	if _, err := scheduleAPI.Create(ctx, oldToken, string(testID(t)), plan); fault.Code(err) != "PERMISSION_DENIED" {
		t.Fatal("schedule permission unexpectedly present before change", err)
	}
	prepare := func(revision uint64, actions []string) CredentialOperation {
		t.Helper()
		op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: revision, Action: credentialPermissionsAction, Reason: "explicit complete action set", Actions: actions})
		if err != nil {
			t.Fatal(err)
		}
		return op
	}
	rotate, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, ValidFor: 24 * time.Hour, GraceFor: time.Hour, Reason: "test two retained generations"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err = ApplyCredentialOperation(ctx, db, rotate, ring); err != nil {
		t.Fatal(err)
	}
	before, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
	if err != nil {
		t.Fatal(err)
	}
	// 快照只在测试内比较，绝不把摘要、密文或 Token 打到失败日志。
	protected := func() string {
		t.Helper()
		var out string
		if err := db.QueryRowContext(ctx, `SELECT CAST(JSON_OBJECT('namespace',HEX(namespace_id),'id',HEX(id),'name',name,'current',HEX(secret_digest),'previous',HEX(previous_secret_digest),'key',secret_key_id,'cipher',HEX(secret_ciphertext),'expiry',expires_at,'grace',previous_valid_until,'revoked',revoked_at,'queues',JSON_EXTRACT(permissions,'$.queues'),'created',created_at) AS CHAR) FROM orbit_credentials WHERE namespace_id=? AND id=?`, initial.NamespaceID, initial.CredentialID).Scan(&out); err != nil {
			t.Fatal(err)
		}
		return out
	}
	frozen := protected()
	actions := []string{"tasks:submit", "schedules:read", "queues:read", "schedules:manage"}
	op := prepare(2, actions)
	actions[0] = "tasks:cancel" // Prepare 必须复制输入，不共享调用方数组。
	if op.Version != 3 || op.Token != "" || !sort.StringsAreSorted(op.Actions) || op.Actions[len(op.Actions)-1] != "tasks:submit" {
		t.Fatal("permission operation did not freeze a canonical secret-free set")
	}
	trigger := "permissions_audit_" + string(testID(t))
	if _, err := db.ExecContext(ctx, "CREATE TRIGGER "+trigger+" BEFORE INSERT ON orbit_audit_events FOR EACH ROW BEGIN IF NEW.namespace_id=UNHEX('"+string(initial.NamespaceID)+"') AND NEW.action='credential.set_permissions' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='permission audit failure test'; END IF; END"); err != nil {
		t.Fatal(err)
	}
	defer db.ExecContext(ctx, "DROP TRIGGER IF EXISTS "+trigger)
	if _, err := ApplyCredentialOperation(ctx, db, op); err == nil {
		t.Fatal("audit failure accepted")
	}
	failed, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
	if err != nil || !reflect.DeepEqual(failed, before) || protected() != frozen {
		t.Fatal("failed audit did not roll back credential state", err)
	}
	var rows int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_idempotency_keys WHERE namespace_id=? AND id=?`, initial.NamespaceID, op.OperationID).Scan(&rows); err != nil || rows != 0 {
		t.Fatal("failed audit kept operation identity", err)
	}
	if _, err := db.ExecContext(ctx, "DROP TRIGGER "+trigger); err != nil {
		t.Fatal(err)
	}
	grant, err := ApplyCredentialOperation(ctx, db, op) // 无 vault 也不能清除既有密文。
	if err != nil || grant.Revision != 3 || grant.Deduplicated || protected() != frozen {
		t.Fatal("permission grant changed protected credential state", err)
	}
	newToken, _ := auth.Parse(rotate.Token)
	assertActions := func(want auth.Permissions) {
		t.Helper()
		for _, token := range []auth.Token{oldToken, newToken} {
			err := WithTx(ctx, db, func(tx *sql.Tx) error {
				p, err := authenticate(ctx, tx, token)
				if err == nil && !reflect.DeepEqual(p, want) {
					t.Error("token authentication did not reload exact current permissions")
				}
				return err
			})
			if err != nil {
				t.Fatal("preserved token no longer authenticates", err)
			}
			if _, err := tasks.GetQueue(ctx, token, initial.QueueID); err != nil {
				t.Fatal("original queue unavailable", err)
			}
			if _, err := tasks.GetQueue(ctx, token, otherQueue.Queue.ID); fault.Code(err) != "NOT_FOUND" {
				t.Fatal("permission change expanded queue scope", err)
			}
		}
	}
	after, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
	if err != nil || !reflect.DeepEqual(after.Permissions.Actions, op.Actions) || !reflect.DeepEqual(after.Permissions.Queues, before.Permissions.Queues) {
		t.Fatal("exact permissions status incorrect", err)
	}
	assertActions(after.Permissions)
	created, err := scheduleAPI.Create(ctx, oldToken, string(testID(t)), plan)
	if err != nil {
		t.Fatal("new permission did not immediately allow real schedule creation", err)
	}
	if _, err := scheduleAPI.Get(ctx, newToken, created.Schedule.ID); err != nil {
		t.Fatal("current token did not receive new schedule read permission", err)
	}
	forbiddenPlan := plan
	forbiddenPlan.Name = "outside-queue"
	forbiddenPlan.Definition.Task.QueueID = otherQueue.Queue.ID
	if _, err := scheduleAPI.Create(ctx, newToken, string(testID(t)), forbiddenPlan); fault.Code(err) != "PERMISSION_DENIED" {
		t.Fatal("new schedule action bypassed preserved queue scope", err)
	}
	var auditBefore, auditAfter []byte
	if err := db.QueryRowContext(ctx, "SELECT `before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND operation_id=?", initial.NamespaceID, op.OperationID).Scan(&auditBefore, &auditAfter); err != nil {
		t.Fatal(err)
	}
	for i, raw := range [][]byte{auditBefore, auditAfter} {
		var audit struct {
			Permissions auth.Permissions `json:"permissions"`
		}
		if err := json.Unmarshal(raw, &audit); err != nil {
			t.Fatal(err)
		}
		want := before.Permissions
		if i == 1 {
			want = after.Permissions
		}
		if !reflect.DeepEqual(audit.Permissions, want) || bytes.Contains(raw, []byte("secret_")) || bytes.Contains(raw, []byte(rotate.Token)) {
			t.Fatal("permissions audit missing exact scope or leaked secret")
		}
	}
	if replay, err := ApplyCredentialOperation(ctx, db, op); err != nil || !replay.Deduplicated || replay.Revision != grant.Revision {
		t.Fatal("permission replay changed receipt", err)
	}
	tampered := op
	tampered.Actions = []string{"queues:read"}
	if _, err := ApplyCredentialOperation(ctx, db, tampered); fault.Code(err) != "IDEMPOTENCY_CONFLICT" {
		t.Fatal("changed frozen actions accepted", err)
	}
	wrongScope := op
	wrongScope.NamespaceID = testID(t)
	if _, err := ApplyCredentialOperation(ctx, db, wrongScope); fault.Code(err) != "NOT_FOUND" {
		t.Fatal("cross namespace credential accepted", err)
	}
	remove := prepare(3, []string{"queues:read"})
	if _, err := ApplyCredentialOperation(ctx, db, remove); err != nil || protected() != frozen {
		t.Fatal("permission removal changed protected state", err)
	}
	assertActions(auth.Permissions{Actions: remove.Actions, Queues: before.Permissions.Queues})
	for _, token := range []auth.Token{oldToken, newToken} {
		if err := tasks.AuthorizeSubmit(ctx, token); fault.Code(err) != "PERMISSION_DENIED" {
			t.Fatal("removed permission still usable", err)
		}
		if _, err := scheduleAPI.Get(ctx, token, created.Schedule.ID); fault.Code(err) != "PERMISSION_DENIED" {
			t.Fatal("removed schedule permission still usable", err)
		}
	}
	if _, err := ApplyCredentialOperation(ctx, db, op); err != nil {
		t.Fatal(err)
	}
	assertActions(auth.Permissions{Actions: remove.Actions, Queues: before.Permissions.Queues})
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND resource_id=? AND action='credential.set_permissions'`, initial.NamespaceID, initial.CredentialID).Scan(&rows); err != nil || rows != 2 {
		t.Fatal("replay duplicated permission audit", err)
	}
	var foreignKeys int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM information_schema.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE()`).Scan(&foreignKeys); err != nil || foreignKeys != 0 {
		t.Fatal("schema acquired a foreign key", err)
	}
}

func TestIntegrationCredentialPermissionsLegacyReplay(t *testing.T) {
	ctx, db := overlapDB(t)
	var initial BootstrapResult
	if err := Bootstrap(ctx, db, BootstrapInput{Application: "permission-legacy-" + string(testID(t)), Namespace: "default", Environment: "test", Queue: "work"}, func(v BootstrapResult) error { initial = v; return nil }); err != nil {
		t.Fatal(err)
	}
	old, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 1, Action: credentialRotateAction, ValidFor: time.Hour, Reason: "legacy v1 replay"})
	if err != nil {
		t.Fatal(err)
	}
	old.Version = 1
	if _, err := ApplyCredentialOperation(ctx, db, old); err != nil {
		t.Fatal("legacy v1 no longer applies", err)
	}
	permissions, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 2, Action: credentialPermissionsAction, Actions: []string{"queues:read"}, Reason: "new permission revision"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := ApplyCredentialOperation(ctx, db, permissions); err != nil {
		t.Fatal(err)
	}
	if replay, err := ApplyCredentialOperation(ctx, db, old); err != nil || !replay.Deduplicated || replay.Revision != 2 {
		t.Fatal("legacy receipt no longer replays after v3", err)
	}
	expiry, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: 3, Action: credentialExpiryAction, Reason: "legacy v2 still generated"})
	if err != nil || expiry.Version != 2 {
		t.Fatal("old action changed fixed format", err)
	}
	if _, err := ApplyCredentialOperation(ctx, db, expiry); err != nil {
		t.Fatal("legacy v2 no longer applies", err)
	}
	status, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
	if err != nil || status.Revision != 4 || status.ExpiresAt != nil || !reflect.DeepEqual(status.Permissions.Actions, permissions.Actions) {
		t.Fatal("legacy changes overwrote new permissions", err)
	}
}

// 竞争由专用事务持有真实凭据行锁，确认每个请求实际到达 InnoDB 等待后再放行。
func competeCredentialPermissions(t *testing.T, ctx context.Context, db *sql.DB, namespace, credential core.ID, operations ...func() error) []error {
	t.Helper()
	guard, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer guard.Rollback()
	var connection uint64
	if err := guard.QueryRowContext(ctx, `SELECT CONNECTION_ID()`).Scan(&connection); err != nil {
		t.Fatal(err)
	}
	var revision uint64
	if err := guard.QueryRowContext(ctx, `SELECT revision FROM orbit_credentials WHERE namespace_id=? AND id=? FOR UPDATE`, namespace, credential).Scan(&revision); err != nil {
		t.Fatal(err)
	}
	results := make([]chan error, len(operations))
	for i, operation := range operations {
		results[i] = make(chan error, 1)
		go func(ch chan error, run func() error) { ch <- run() }(results[i], operation)
		deadline := time.Now().Add(5 * time.Second)
		for {
			select {
			case early := <-results[i]:
				t.Fatalf("permission competitor returned before DB lock: %v", early)
			default:
			}
			var waiting int
			if err := db.QueryRowContext(ctx, `SELECT COUNT(DISTINCT w.REQUESTING_THREAD_ID) FROM performance_schema.data_lock_waits w JOIN performance_schema.threads b ON b.THREAD_ID=w.BLOCKING_THREAD_ID WHERE b.PROCESSLIST_ID=?`, connection).Scan(&waiting); err != nil {
				t.Fatal(err)
			}
			if waiting == i+1 {
				break
			}
			if time.Now().After(deadline) {
				t.Fatalf("only %d/%d requests reached credential lock", waiting, i+1)
			}
			time.Sleep(5 * time.Millisecond)
		}
	}
	if err := guard.Commit(); err != nil {
		t.Fatal(err)
	}
	errs := make([]error, len(operations))
	for i, ch := range results {
		select {
		case errs[i] = <-ch:
		case <-ctx.Done():
			t.Fatal(ctx.Err())
		}
	}
	return errs
}

func TestIntegrationCredentialPermissionsConcurrency(t *testing.T) {
	ctx, db := overlapDB(t)
	var initial BootstrapResult
	if err := Bootstrap(ctx, db, BootstrapInput{Application: "permission-race-" + string(testID(t)), Namespace: "default", Environment: "test", Queue: "work"}, func(v BootstrapResult) error { initial = v; return nil }); err != nil {
		t.Fatal(err)
	}
	prepare := func(revision uint64, action string) CredentialOperation {
		t.Helper()
		op, err := PrepareCredentialOperation(ctx, db, CredentialChangeInput{NamespaceID: initial.NamespaceID, CredentialID: initial.CredentialID, ExpectedRevision: revision, Action: credentialPermissionsAction, Reason: "permission CAS race", Actions: []string{action}})
		if err != nil {
			t.Fatal(err)
		}
		return op
	}
	op := prepare(1, "queues:read")
	var receipts [2]CredentialChangeReceipt
	errs := competeCredentialPermissions(t, ctx, db, initial.NamespaceID, initial.CredentialID,
		func() (err error) { receipts[0], err = ApplyCredentialOperation(ctx, db, op); return },
		func() (err error) { receipts[1], err = ApplyCredentialOperation(ctx, db, op); return })
	if errs[0] != nil || errs[1] != nil || receipts[0].Deduplicated == receipts[1].Deduplicated || receipts[0].Revision != 2 || receipts[1].Revision != 2 {
		t.Fatalf("concurrent replay: %v", errs)
	}
	left, right := prepare(2, "tasks:read"), prepare(2, "schedules:read")
	errs = competeCredentialPermissions(t, ctx, db, initial.NamespaceID, initial.CredentialID,
		func() error { _, err := ApplyCredentialOperation(ctx, db, left); return err },
		func() error { _, err := ApplyCredentialOperation(ctx, db, right); return err })
	if !((errs[0] == nil && fault.Code(errs[1]) == "STATE_CONFLICT") || (errs[1] == nil && fault.Code(errs[0]) == "STATE_CONFLICT")) {
		t.Fatalf("permission CAS did not select exactly one winner: %v", errs)
	}
	winner := left
	if errs[1] == nil {
		winner = right
	}
	status, err := GetCredentialStatus(ctx, db, initial.NamespaceID, initial.CredentialID)
	if err != nil || status.Revision != 3 || !reflect.DeepEqual(status.Permissions.Actions, winner.Actions) {
		t.Fatal("CAS did not persist winner", err)
	}
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_audit_events WHERE namespace_id=? AND action='credential.set_permissions'`, initial.NamespaceID).Scan(&count); err != nil || count != 2 {
		t.Fatal("losing CAS or replay created audit", err)
	}
}