package database

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

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/api/meshxapi"
	"github.com/tidbcloud/AgentTeam/internal/api/roles"
	"github.com/tidbcloud/AgentTeam/internal/identity/trusted"
	roleservice "github.com/tidbcloud/AgentTeam/internal/service/roles"
	rolesstore "github.com/tidbcloud/AgentTeam/internal/store/resource/roles"
	"gorm.io/gorm"
)

func TestTiDBRoleOperationConcurrentCreateReplay(t *testing.T) {
	svc, content, db := openTiDBRoleService(t)
	content.arm()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	operation := roleservice.MutationOperation{OperationKey: "create-replay", RequestDigest: tidbRoleDigest("same")}
	input := roleservice.Mutation{Name: "Reviewer", Definition: "review production changes"}

	first := make(chan roleOperationResult[roleservice.Role], 1)
	go func() {
		role, err := svc.CreateRoleOperation(ctx, "wsp_1", "usr_1", operation, input)
		first <- roleOperationResult[roleservice.Role]{value: role, err: err}
	}()
	content.waitUntilBlocked(t)

	second := make(chan roleOperationResult[roleservice.Role], 1)
	go func() {
		role, err := svc.CreateRoleOperation(ctx, "wsp_1", "usr_1", operation, input)
		second <- roleOperationResult[roleservice.Role]{value: role, err: err}
	}()
	requireOperationStillWaiting(t, second)
	content.release()

	firstResult := <-first
	secondResult := <-second
	require.NoError(t, firstResult.err)
	require.NoError(t, secondResult.err)
	require.Equal(t, firstResult.value.ID, secondResult.value.ID)
	require.Equal(t, firstResult.value.RoleVersionID, secondResult.value.RoleVersionID)
	require.Equal(t, 1, content.storeCalls())

	var roleCount, versionCount int64
	require.NoError(t, db.Model(&rolesstore.Role{}).Count(&roleCount).Error)
	require.NoError(t, db.Model(&rolesstore.RoleVersion{}).Count(&versionCount).Error)
	require.EqualValues(t, 1, roleCount)
	require.EqualValues(t, 1, versionCount)
}

func TestTiDBRoleOperationConcurrentCreateDigestConflict(t *testing.T) {
	svc, content, db := openTiDBRoleService(t)
	content.arm()
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	input := roleservice.Mutation{Name: "Reviewer", Definition: "review production changes"}

	first := make(chan roleOperationResult[roleservice.Role], 1)
	go func() {
		role, err := svc.CreateRoleOperation(ctx, "wsp_1", "usr_1", roleservice.MutationOperation{
			OperationKey: "create-conflict", RequestDigest: tidbRoleDigest("winner"),
		}, input)
		first <- roleOperationResult[roleservice.Role]{value: role, err: err}
	}()
	content.waitUntilBlocked(t)

	second := make(chan roleOperationResult[roleservice.Role], 1)
	go func() {
		role, err := svc.CreateRoleOperation(ctx, "wsp_1", "usr_1", roleservice.MutationOperation{
			OperationKey: "create-conflict", RequestDigest: tidbRoleDigest("loser"),
		}, input)
		second <- roleOperationResult[roleservice.Role]{value: role, err: err}
	}()
	requireOperationStillWaiting(t, second)
	content.release()

	firstResult := <-first
	secondResult := <-second
	require.NoError(t, firstResult.err)
	require.ErrorIs(t, secondResult.err, roleservice.ErrRoleMutationConflict)
	require.Equal(t, 1, content.storeCalls())

	var roleCount, versionCount int64
	require.NoError(t, db.Model(&rolesstore.Role{}).Count(&roleCount).Error)
	require.NoError(t, db.Model(&rolesstore.RoleVersion{}).Count(&versionCount).Error)
	require.EqualValues(t, 1, roleCount)
	require.EqualValues(t, 1, versionCount)
}

func TestTiDBRoleOperationConcurrentDefinitionUpdateVersionFence(t *testing.T) {
	svc, content, db := openTiDBRoleService(t)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	role, err := svc.CreateRole(ctx, "wsp_1", "usr_1", roleservice.Mutation{
		Name: "Reviewer", Definition: "review v1",
	})
	require.NoError(t, err)
	content.arm()

	winnerDefinition := "review winner"
	winner := make(chan roleOperationResult[roleservice.Role], 1)
	go func() {
		updated, updateErr := svc.UpdateRoleOperation(
			ctx, "wsp_1", "usr_1", role.ID,
			roleservice.MutationOperation{OperationKey: "update-winner", RequestDigest: tidbRoleDigest("winner")},
			role.RoleVersionID, roleservice.RoleUpdatePatch{Definition: &winnerDefinition},
		)
		winner <- roleOperationResult[roleservice.Role]{value: updated, err: updateErr}
	}()
	content.waitUntilBlocked(t)

	loserDefinition := "review loser"
	loser := make(chan roleOperationResult[roleservice.Role], 1)
	go func() {
		updated, updateErr := svc.UpdateRoleOperation(
			ctx, "wsp_1", "usr_1", role.ID,
			roleservice.MutationOperation{OperationKey: "update-loser", RequestDigest: tidbRoleDigest("loser")},
			role.RoleVersionID, roleservice.RoleUpdatePatch{Definition: &loserDefinition},
		)
		loser <- roleOperationResult[roleservice.Role]{value: updated, err: updateErr}
	}()
	requireOperationStillWaiting(t, loser)
	content.release()

	winnerResult := <-winner
	loserResult := <-loser
	require.NoError(t, winnerResult.err)
	require.ErrorIs(t, loserResult.err, roleservice.ErrRoleVersionConflict)
	require.Equal(t, 1, content.storeCalls(), "loser must not write role-package content")

	current, err := svc.GetRole(ctx, "wsp_1", "usr_1", role.ID)
	require.NoError(t, err)
	require.Equal(t, winnerResult.value.RoleVersionID, current.RoleVersionID)
	require.Equal(t, winnerDefinition, current.Definition)
	var versionCount, manifestCount, receiptCount int64
	require.NoError(t, db.Model(&rolesstore.RoleVersion{}).Where("role_id = ?", role.ID).Count(&versionCount).Error)
	require.NoError(t, db.Model(&rolesstore.RolePackageManifestItem{}).Where("role_id = ?", role.ID).Count(&manifestCount).Error)
	require.NoError(t, db.Model(&rolesstore.RoleOperationReceipt{}).Count(&receiptCount).Error)
	require.EqualValues(t, 2, versionCount, "only one new immutable version may commit")
	require.EqualValues(t, 2, manifestCount, "only one new manifest may commit")
	require.EqualValues(t, 1, receiptCount, "loser receipt reservation must roll back")
}

func TestTiDBRoleOperationPreflightsLeaveEveryRoleFactUnchanged(t *testing.T) {
	svc, content, db := openTiDBRoleService(t)
	ctx := trusted.WithIdentity(context.Background(), trusted.Identity{
		Subject: "usr_1",
		RawClaims: map[string]any{
			"agent_ownership_workspace_id": "wsp_1",
			"owned_agent_ids":              []any{"agt_1"},
		},
	})
	role, err := svc.CreateRole(ctx, "wsp_1", "usr_1", roleservice.Mutation{
		Name: "Reviewer", Definition: "review production changes",
	})
	require.NoError(t, err)
	before := tidbRoleFactSnapshotForTest(t, db, content)

	require.NoError(t, svc.PreflightCreateRoleOperation(
		ctx, "wsp_1", "usr_1",
		roleservice.Mutation{Name: "Writer", Definition: "write release notes"},
	))
	name := "Senior reviewer"
	require.NoError(t, svc.PreflightUpdateRoleOperation(
		ctx, "wsp_1", "usr_1", role.ID, role.RoleVersionID,
		roleservice.RoleUpdatePatch{Name: &name},
	))
	require.NoError(t, svc.PreflightDeleteRoleOperation(
		ctx, "wsp_1", "usr_1", role.ID, role.RoleVersionID,
	))
	_, err = svc.PreflightApplyRoleOperation(
		ctx, "wsp_1", "usr_1", role.ID,
		roleservice.RealizationRequest{AgentID: "agt_1", RoleVersionID: role.RoleVersionID, Apply: true},
	)
	require.NoError(t, err)

	require.Equal(t, before, tidbRoleFactSnapshotForTest(t, db, content))
}

func TestTiDBRoleOperationHTTPPreflightAndExecuteReceipt(t *testing.T) {
	svc, content, db := openTiDBRoleService(t)
	handler := roles.NewRoleOperationTransportHandler(svc)
	ctx := trusted.WithIdentity(context.Background(), trusted.Identity{
		Subject: "usr_1",
		RawClaims: map[string]any{
			"agentteam_internal_purpose":        roles.RoleOperationInternalPurpose,
			"agent_ownership_workspace_id": "wsp_1",
			"owned_agent_ids":              []any{"agt_1"},
		},
	})

	before := tidbRoleFactSnapshotForTest(t, db, content)
	preflight := httptest.NewRequest(
		http.MethodPost,
		roles.RoleOperationPreflightPath,
		bytes.NewBufferString(`{"create":{"name":"Reviewer","definition":"Review production changes"}}`),
	).WithContext(ctx)
	preflight.Header.Set(meshxapi.HeaderWorkspaceID, "wsp_1")
	preflightResponse := httptest.NewRecorder()
	handler.ServeHTTP(preflightResponse, preflight)
	require.Equal(t, http.StatusNoContent, preflightResponse.Code)
	require.Empty(t, preflightResponse.Body.String())
	require.Equal(t, before, tidbRoleFactSnapshotForTest(t, db, content))

	body := fmt.Sprintf(
		`{"operationKey":"op_http1","requestDigest":%q,"create":{"name":"Reviewer","definition":"Review production changes"}}`,
		tidbRoleDigest("http-create"),
	)
	first := performTiDBRoleOperationHTTP(t, handler, ctx, roles.RoleOperationExecutePath, body)
	replay := performTiDBRoleOperationHTTP(t, handler, ctx, roles.RoleOperationExecutePath, body)
	require.Equal(t, http.StatusOK, first.Code)
	require.Equal(t, http.StatusOK, replay.Code)
	var firstResult, replayResult struct {
		RoleID        string `json:"roleId"`
		RoleVersionID string `json:"roleVersionId"`
	}
	require.NoError(t, json.Unmarshal(first.Body.Bytes(), &firstResult))
	require.NoError(t, json.Unmarshal(replay.Body.Bytes(), &replayResult))
	require.NotEmpty(t, firstResult.RoleID)
	require.NotEmpty(t, firstResult.RoleVersionID)
	require.Equal(t, firstResult, replayResult)
	var roleCount, versionCount, receiptCount int64
	require.NoError(t, db.Model(&rolesstore.Role{}).Count(&roleCount).Error)
	require.NoError(t, db.Model(&rolesstore.RoleVersion{}).Count(&versionCount).Error)
	require.NoError(t, db.Model(&rolesstore.RoleOperationReceipt{}).Count(&receiptCount).Error)
	require.EqualValues(t, 1, roleCount)
	require.EqualValues(t, 1, versionCount)
	require.EqualValues(t, 1, receiptCount)
}

func performTiDBRoleOperationHTTP(
	t *testing.T,
	handler http.Handler,
	ctx context.Context,
	path, body string,
) *httptest.ResponseRecorder {
	t.Helper()
	request := httptest.NewRequest(http.MethodPost, path, bytes.NewBufferString(body)).WithContext(ctx)
	request.Header.Set(meshxapi.HeaderWorkspaceID, "wsp_1")
	response := httptest.NewRecorder()
	handler.ServeHTTP(response, request)
	return response
}

type roleOperationResult[T any] struct {
	value T
	err   error
}

type tidbRoleFactSnapshot struct {
	Roles            string
	Versions         string
	ManifestItems    string
	Grants           string
	Receipts         string
	Realizations     string
	RealizationItems string
	Bindings         string
	Content          string
}

func tidbRoleFactSnapshotForTest(
	t *testing.T,
	db *gorm.DB,
	content *blockingRolePackageContentStore,
) tidbRoleFactSnapshot {
	t.Helper()
	content.mu.Lock()
	contentJSON, err := json.Marshal(content.contents)
	content.mu.Unlock()
	require.NoError(t, err)
	return tidbRoleFactSnapshot{
		Roles:            tidbRoleTableJSONForTest[rolesstore.Role](t, db, "id"),
		Versions:         tidbRoleTableJSONForTest[rolesstore.RoleVersion](t, db, "id"),
		ManifestItems:    tidbRoleTableJSONForTest[rolesstore.RolePackageManifestItem](t, db, "id"),
		Grants:           tidbRoleTableJSONForTest[rolesstore.RoleGrant](t, db, "id"),
		Receipts:         tidbRoleTableJSONForTest[rolesstore.RoleOperationReceipt](t, db, "operation_key"),
		Realizations:     tidbRoleTableJSONForTest[rolesstore.RoleRealization](t, db, "id"),
		RealizationItems: tidbRoleTableJSONForTest[rolesstore.RoleRealizationItem](t, db, "id"),
		Bindings:         tidbRoleTableJSONForTest[rolesstore.AgentRoleBinding](t, db, "id"),
		Content:          string(contentJSON),
	}
}

func tidbRoleTableJSONForTest[T any](t *testing.T, db *gorm.DB, order string) string {
	t.Helper()
	var rows []T
	require.NoError(t, db.Order(order).Find(&rows).Error)
	data, err := json.Marshal(rows)
	require.NoError(t, err)
	return string(data)
}

func requireOperationStillWaiting[T any](t *testing.T, result <-chan roleOperationResult[T]) {
	t.Helper()
	select {
	case completed := <-result:
		t.Fatalf("concurrent operation returned before the blocked transaction completed: %v", completed.err)
	case <-time.After(100 * time.Millisecond):
	}
}

func openTiDBRoleService(t *testing.T) (*roleservice.Service, *blockingRolePackageContentStore, *gorm.DB) {
	t.Helper()
	db := openTiDBTestDB(t, func(db *gorm.DB) error {
		return db.AutoMigrate(
			&rolesstore.Role{},
			&rolesstore.RoleVersion{},
			&rolesstore.RolePackageManifestItem{},
			&rolesstore.RoleGrant{},
			&rolesstore.RoleRealization{},
			&rolesstore.RoleRealizationItem{},
			&rolesstore.AgentRoleBinding{},
			&rolesstore.RoleOperationReceipt{},
		)
	})
	sqlDB, err := db.DB()
	require.NoError(t, err)
	sqlDB.SetMaxOpenConns(8)
	sqlDB.SetMaxIdleConns(8)
	require.GreaterOrEqual(t, sqlDB.Stats().MaxOpenConnections, 2, "concurrency evidence requires multiple database connections")
	content := newBlockingRolePackageContentStore()
	svc := roleservice.New(rolesstore.New(db), roleservice.Options{
		ContentStore: content,
		Now:          func() time.Time { return time.Date(2026, 8, 3, 12, 0, 0, 0, time.UTC) },
	})
	return svc, content, db
}

type blockingRolePackageContentStore struct {
	mu       sync.Mutex
	contents map[string]string
	calls    int
	block    bool
	entered  chan struct{}
	released chan struct{}
}

func newBlockingRolePackageContentStore() *blockingRolePackageContentStore {
	return &blockingRolePackageContentStore{contents: map[string]string{}}
}

func (s *blockingRolePackageContentStore) arm() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.calls = 0
	s.block = true
	s.entered = make(chan struct{})
	s.released = make(chan struct{})
}

func (s *blockingRolePackageContentStore) waitUntilBlocked(t *testing.T) {
	t.Helper()
	s.mu.Lock()
	entered := s.entered
	s.mu.Unlock()
	select {
	case <-entered:
	case <-time.After(5 * time.Second):
		t.Fatal("role-package content write did not reach blocking point")
	}
}

func (s *blockingRolePackageContentStore) release() {
	s.mu.Lock()
	released := s.released
	s.mu.Unlock()
	close(released)
}

func (s *blockingRolePackageContentStore) storeCalls() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.calls
}

func (s *blockingRolePackageContentStore) StoreRolePackage(_ context.Context, input roleservice.StoreRolePackageInput) (roleservice.StoreRolePackageOutput, error) {
	s.mu.Lock()
	s.calls++
	shouldBlock := s.block
	released := s.released
	if shouldBlock {
		s.block = false
		close(s.entered)
	}
	s.mu.Unlock()
	if shouldBlock {
		<-released
	}

	assets := make(map[string]roleservice.RolePackageStoredAsset, len(input.Assets))
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, asset := range input.Assets {
		path := rolePackageAssetPathForTiDBTest(asset)
		ref := fmt.Sprintf("ags://tidb-test/%s/%s/%s/%s", input.WorkspaceID, input.RoleID, input.VersionID, path)
		content := strings.TrimSpace(asset.Content)
		s.contents[ref] = content
		assets[path] = roleservice.RolePackageStoredAsset{
			Path: path, ContentRef: ref, ContentHash: tidbRoleDigest(content),
		}
	}
	return roleservice.StoreRolePackageOutput{Assets: assets}, nil
}

func (s *blockingRolePackageContentStore) ResolveRolePackage(_ context.Context, input roleservice.ResolveRolePackageInput) (string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	content, ok := s.contents[strings.TrimSpace(input.ContentRef)]
	if !ok {
		return "", errors.New("role package content not found")
	}
	return content, nil
}

func rolePackageAssetPathForTiDBTest(asset roleservice.RolePackageAsset) string {
	if asset.ItemType == roleservice.ManifestItemRoleDefinitionAsset {
		return "IDENTITY.md"
	}
	return strings.TrimSpace(asset.Name)
}

func tidbRoleDigest(value string) string {
	sum := sha256.Sum256([]byte(value))
	return fmt.Sprintf("%x", sum)
}