package serviceaccount

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	serviceaccountstore "github.com/tidbcloud/AgentTeam/internal/store/serviceaccount"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
)

// ReconcileWorkspaceAudience resolves the current policy into ordinary Human
// scope memberships. The bounded member IDs form one resumable batch; Complete
// may only be set when the caller has exhausted the authoritative roster.
type ReconcileWorkspaceAudience struct {
	WorkspaceID            string
	ServiceAccountID       string
	ExpectedAccountVersion int64
	PolicyVersion          int64
	MemberIDs              []string
	Complete               bool
	Now                    time.Time
}

// ReconcileWorkspaceAudience applies one audience batch under the account and
// scope locks. New rows use the current scope tail as their read baseline.
func (service *Service) ReconcileWorkspaceAudience(ctx context.Context, input ReconcileWorkspaceAudience) (serviceaccountstore.Account, error) {
	if err := service.validate(); err != nil {
		return serviceaccountstore.Account{}, err
	}
	input.WorkspaceID = strings.TrimSpace(input.WorkspaceID)
	input.ServiceAccountID = strings.TrimSpace(input.ServiceAccountID)
	input.MemberIDs = sortedUnique(input.MemberIDs)
	input.Now = normalizedNow(input.Now)
	if input.WorkspaceID == "" || input.ServiceAccountID == "" || input.ExpectedAccountVersion <= 0 || input.PolicyVersion <= 0 {
		return serviceaccountstore.Account{}, storeerr.BadRequest("workspaceId, serviceAccountId, expectedAccountVersion, and policyVersion are required")
	}
	if len(input.MemberIDs) > 500 {
		return serviceaccountstore.Account{}, storeerr.BadRequest("audience reconcile batch cannot exceed 500 members")
	}

	var result serviceaccountstore.Account
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		accountStore := service.accounts.InTx(tx)
		account, err := accountStore.LoadAccountForUpdate(ctx, input.WorkspaceID, input.ServiceAccountID)
		if err != nil {
			return err
		}
		if account.Version != input.ExpectedAccountVersion || account.AudiencePolicyVersion != input.PolicyVersion {
			return storeerr.Conflict("service account audience version changed")
		}
		var latestSeq, latestGlobalSeq int64
		if err := tx.QueryRow(ctx, `
SELECT latest_seq, latest_global_seq
FROM scopes
WHERE id = $1 AND workspace_id = $2 AND type = 'service_account' AND status = 'active'
FOR UPDATE
`, account.MessageScopeID, account.WorkspaceID).Scan(&latestSeq, &latestGlobalSeq); err != nil {
			return fmt.Errorf("lock service account audience scope: %w", err)
		}
		if err := requireHumanMembersActive(ctx, accountStore, account.WorkspaceID, input.MemberIDs, "audience members"); err != nil {
			return err
		}
		repos := message.NewScopeRepositories(tx)
		for _, memberID := range input.MemberIDs {
			open, err := hasOpenServiceAccountAudienceMembership(ctx, tx, account.MessageScopeID, memberID)
			if err != nil {
				return err
			}
			if !open {
				if err := repos.Memberships.Upsert(ctx, message.ScopeMembershipRecord{
					ScopeID:               account.MessageScopeID,
					MemberID:              memberID,
					Visibility:            "public",
					LastReadSeq:           latestSeq,
					LastReadGlobalSeq:     latestGlobalSeq,
					LastActivityGlobalSeq: latestGlobalSeq,
					EffectiveFromSeq:      latestSeq + 1,
					CreatedAt:             input.Now,
					UpdatedAt:             input.Now,
				}); err != nil {
					return err
				}
			}
			if err := repos.Notifications.Upsert(ctx, message.ScopeNotificationSettingRecord{
				ScopeID: account.MessageScopeID, MemberID: memberID, State: "active",
				CreatedAt: input.Now, UpdatedAt: input.Now,
			}); err != nil {
				return err
			}
		}
		if !input.Complete {
			result = account
			return nil
		}
		if account.AudienceState == serviceaccountstore.AudienceStateReady &&
			account.EffectiveAudienceVersion == input.PolicyVersion {
			if account.ManagementKind == serviceaccountstore.ManagementKindWorkspace {
				if err := fulfillCreateRequestForAccount(ctx, accountStore, account, input.Now); err != nil {
					return err
				}
			}
			result = account
			return nil
		}
		account, err = accountStore.MarkAudienceReady(ctx, account.WorkspaceID, account.ID, account.Version, input.PolicyVersion, input.Now)
		if err != nil {
			return err
		}
		if account.ManagementKind == serviceaccountstore.ManagementKindWorkspace {
			if err := fulfillCreateRequestForAccount(ctx, accountStore, account, input.Now); err != nil {
				return err
			}
		}
		result = account
		return nil
	})
	return result, err
}

func (service *Service) ReconcileAllWorkspaceHumans(ctx context.Context, workspaceID, accountID string, expectedAccountVersion, policyVersion int64, now time.Time) (serviceaccountstore.Account, error) {
	workspaceID = strings.TrimSpace(workspaceID)
	accountID = strings.TrimSpace(accountID)
	now = normalizedNow(now)
	account, err := service.accounts.LoadAccount(ctx, workspaceID, accountID)
	if err != nil {
		return serviceaccountstore.Account{}, err
	}
	if account.Version != expectedAccountVersion || account.AudiencePolicyVersion != policyVersion {
		return serviceaccountstore.Account{}, storeerr.Conflict("service account audience version changed")
	}
	cursor := ""
	for {
		rows, queryErr := service.db.Query(ctx, `
SELECT id
FROM members
WHERE workspace_id = $1
  AND type = 'human'
  AND status = 'active'
  AND id > $2
ORDER BY id ASC
LIMIT 500
`, workspaceID, cursor)
		if queryErr != nil {
			return serviceaccountstore.Account{}, fmt.Errorf("list workspace service account audience: %w", queryErr)
		}
		batch := make([]string, 0, 500)
		count := 0
		for rows.Next() {
			var memberID string
			if scanErr := rows.Scan(&memberID); scanErr != nil {
				rows.Close()
				return serviceaccountstore.Account{}, fmt.Errorf("scan workspace service account audience: %w", scanErr)
			}
			count++
			cursor = strings.TrimSpace(memberID)
			batch = append(batch, cursor)
		}
		iterateErr := rows.Err()
		rows.Close()
		if iterateErr != nil {
			return serviceaccountstore.Account{}, fmt.Errorf("iterate workspace service account audience: %w", iterateErr)
		}
		if len(batch) > 0 {
			if _, err := service.ReconcileWorkspaceAudience(ctx, ReconcileWorkspaceAudience{
				WorkspaceID: workspaceID, ServiceAccountID: accountID,
				ExpectedAccountVersion: expectedAccountVersion, PolicyVersion: policyVersion,
				MemberIDs: batch, Complete: false, Now: now,
			}); err != nil {
				return serviceaccountstore.Account{}, err
			}
		}
		if count < 500 {
			break
		}
	}
	if err := service.closeStaleAudienceMemberships(ctx, account, expectedAccountVersion, policyVersion, now); err != nil {
		return serviceaccountstore.Account{}, err
	}
	return service.ReconcileWorkspaceAudience(ctx, ReconcileWorkspaceAudience{
		WorkspaceID: workspaceID, ServiceAccountID: accountID,
		ExpectedAccountVersion: expectedAccountVersion, PolicyVersion: policyVersion,
		Complete: true, Now: now,
	})
}

func (service *Service) closeStaleAudienceMemberships(
	ctx context.Context,
	account serviceaccountstore.Account,
	expectedAccountVersion, policyVersion int64,
	now time.Time,
) error {
	cursor := ""
	for {
		rows, err := service.db.Query(ctx, `
SELECT membership.member_id
FROM scope_memberships membership
JOIN members member ON member.id = membership.member_id
WHERE membership.scope_id = $1
  AND membership.member_id > $2
  AND membership.effective_to_seq IS NULL
  AND membership.removed_at IS NULL
  AND membership.archived_at IS NULL
  AND member.type = 'human'
  AND (member.workspace_id <> $3 OR member.status <> 'active')
ORDER BY membership.member_id ASC
LIMIT 500
`, account.MessageScopeID, cursor, account.WorkspaceID)
		if err != nil {
			return fmt.Errorf("list stale service account audience: %w", err)
		}
		remove := make([]string, 0, 500)
		count := 0
		for rows.Next() {
			var memberID string
			if err := rows.Scan(&memberID); err != nil {
				rows.Close()
				return fmt.Errorf("scan stale service account audience: %w", err)
			}
			count++
			cursor = strings.TrimSpace(memberID)
			remove = append(remove, cursor)
		}
		iterateErr := rows.Err()
		rows.Close()
		if iterateErr != nil {
			return fmt.Errorf("iterate stale service account audience: %w", iterateErr)
		}
		if len(remove) > 0 {
			if err := service.closeWorkspaceAudienceBatch(ctx, account, expectedAccountVersion, policyVersion, remove, now); err != nil {
				return err
			}
		}
		if count < 500 {
			return nil
		}
	}
}

func (service *Service) closeWorkspaceAudienceBatch(
	ctx context.Context,
	account serviceaccountstore.Account,
	expectedAccountVersion, policyVersion int64,
	memberIDs []string,
	now time.Time,
) error {
	memberIDs = sortedUnique(memberIDs)
	if len(memberIDs) == 0 {
		return nil
	}
	if len(memberIDs) > 500 {
		return storeerr.BadRequest("audience removal batch cannot exceed 500 members")
	}
	return coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		locked, err := service.accounts.InTx(tx).LoadAccountForUpdate(ctx, account.WorkspaceID, account.ID)
		if err != nil {
			return err
		}
		if locked.Version != expectedAccountVersion || locked.AudiencePolicyVersion != policyVersion {
			return storeerr.Conflict("service account audience version changed")
		}
		var latestSeq int64
		if err := tx.QueryRow(ctx, `
SELECT latest_seq
FROM scopes
WHERE id = $1 AND workspace_id = $2 AND type = 'service_account' AND status = 'active'
FOR UPDATE
`, locked.MessageScopeID, locked.WorkspaceID).Scan(&latestSeq); err != nil {
			return fmt.Errorf("lock service account audience scope for removal: %w", err)
		}
		args := []any{locked.MessageScopeID, latestSeq, now}
		placeholders := make([]string, 0, len(memberIDs))
		for _, memberID := range memberIDs {
			args = append(args, memberID)
			placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
		}
		if _, err := tx.Exec(ctx, `
UPDATE scope_memberships
SET effective_to_seq = $2,
    removed_at = $3,
    updated_at = $3,
    unread_generation = unread_generation + 1
WHERE scope_id = $1
  AND effective_to_seq IS NULL
  AND removed_at IS NULL
  AND member_id IN (`+strings.Join(placeholders, ",")+`)
`, args...); err != nil {
			return fmt.Errorf("close stale service account audience memberships: %w", err)
		}
		return nil
	})
}

func hasOpenServiceAccountAudienceMembership(ctx context.Context, queryer coredb.QueryRower, scopeID, memberID string) (bool, error) {
	var exists bool
	if err := queryer.QueryRow(ctx, `
SELECT EXISTS (
  SELECT 1 FROM scope_memberships
  WHERE scope_id = $1 AND member_id = $2
    AND effective_to_seq IS NULL AND removed_at IS NULL AND archived_at IS NULL
)
`, strings.TrimSpace(scopeID), strings.TrimSpace(memberID)).Scan(&exists); err != nil {
		return false, fmt.Errorf("check service account audience membership: %w", err)
	}
	return exists, nil
}

func fulfillCreateRequestForAccount(ctx context.Context, store *serviceaccountstore.TxStore, account serviceaccountstore.Account, now time.Time) error {
	request, err := store.LoadCreateRequestByCreatedAccountForUpdate(ctx, account.WorkspaceID, account.ID)
	if err != nil {
		if errors.Is(err, serviceaccountstore.ErrNotFound) {
			return nil
		}
		return err
	}
	if request.BusinessState == serviceaccountstore.RequestStateFulfilled {
		return nil
	}
	if request.BusinessState != serviceaccountstore.RequestStateProvisioning {
		return storeerr.Conflict("materialized service account create request is not provisioning")
	}
	_, err = store.TransitionCreateRequest(
		ctx, request.WorkspaceID, request.ID, request.Version,
		serviceaccountstore.RequestStateFulfilled, request.ApprovalInstanceID,
		request.ConsumedDecisionVersion, request.CreatedServiceAccountID, now,
	)
	return err
}