package serviceaccount

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/attachment"
	serviceaccountstore "github.com/tidbcloud/AgentTeam/internal/store/serviceaccount"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	"github.com/tidbcloud/AgentTeam/internal/workspace/owner"
)

func (service *PlatformService) discoverOnePlatformProvisioningPage(ctx context.Context, now time.Time) (int, error) {
	if !service.controlPlane {
		return 0, nil
	}
	var discovered int
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		var accountID string
		err := tx.QueryRow(ctx, `
SELECT account.id
FROM platform_service_accounts account
WHERE EXISTS (
      SELECT 1
      FROM workspaces workspace
      LEFT JOIN platform_service_account_provisioning_inputs input
        ON input.platform_service_account_id = account.id
       AND input.workspace_id = workspace.id
      WHERE input.id IS NULL
  )
ORDER BY account.id ASC
LIMIT 1
FOR UPDATE
`).Scan(&accountID)
		if err != nil {
			if coredb.IsNotFound(err) {
				return nil
			}
			return fmt.Errorf("select platform account with missing provisioning inputs: %w", err)
		}
		var generation int64
		discovered, generation, err = discoverPlatformProvisioningInputsTx(ctx, tx, accountID, serviceaccountstore.PlatformProvisioningSourceWorkspaceCreate, now, platformProcessBatchSize)
		_ = generation
		return err
	})
	return discovered, err
}

func (service *PlatformService) processOnePlatformProvisioningInput(ctx context.Context, now time.Time) (bool, error) {
	afterID := service.workerCursor(platformWorkerCursorProvisioning)
	rows, err := service.db.Query(ctx, platformProvisioningInputSelectSQL()+`
JOIN platform_service_accounts account ON account.id = input.platform_service_account_id
WHERE input.state IN ('pending','identity_ready')
  AND input.id > $1
ORDER BY input.id ASC
LIMIT $2
`, afterID, platformProcessBatchSize)
	if err != nil {
		return false, fmt.Errorf("list pending platform provisioning inputs: %w", err)
	}
	candidates := make([]serviceaccountstore.PlatformProvisioningInput, 0, platformProcessBatchSize)
	for rows.Next() {
		input, scanErr := scanPlatformProvisioningInput(rows)
		if scanErr != nil {
			rows.Close()
			return false, fmt.Errorf("scan platform provisioning input: %w", scanErr)
		}
		candidates = append(candidates, input)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return false, fmt.Errorf("iterate platform provisioning inputs: %w", err)
	}
	rows.Close()

	for _, candidate := range candidates {
		if !owner.IsManaged(service.managedWorkspaces, candidate.WorkspaceID) {
			continue
		}
		if len(candidates) == platformProcessBatchSize {
			service.setWorkerCursor(platformWorkerCursorProvisioning, candidate.ID)
		} else {
			service.setWorkerCursor(platformWorkerCursorProvisioning, "")
		}
		if candidate.State == serviceaccountstore.PlatformProvisioningStatePending {
			if err := service.materializePlatformProvisioningIdentity(ctx, candidate.ID, now); err != nil {
				_ = service.recordPlatformProvisioningError(ctx, candidate.ID, err, now)
				return false, err
			}
		}
		if err := service.reconcileAndSettlePlatformProvisioning(ctx, candidate.ID, now); err != nil {
			_ = service.recordPlatformProvisioningError(ctx, candidate.ID, err, now)
			return false, err
		}
		return true, nil
	}
	if len(candidates) == platformProcessBatchSize {
		service.setWorkerCursor(platformWorkerCursorProvisioning, candidates[len(candidates)-1].ID)
	} else {
		service.setWorkerCursor(platformWorkerCursorProvisioning, "")
	}
	return false, nil
}

func (service *PlatformService) materializePlatformProvisioningIdentity(ctx context.Context, provisioningInputID string, now time.Time) error {
	return coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		input, found, err := loadPlatformProvisioningInputTx(ctx, tx, provisioningInputID, true)
		if err != nil {
			return err
		}
		if !found {
			return storeerr.NotFound("platform provisioning input not found")
		}
		if input.State != serviceaccountstore.PlatformProvisioningStatePending {
			return nil
		}
		account, err := service.accounts.InTx(tx).LoadPlatformAccountForUpdate(ctx, input.PlatformServiceAccountID)
		if err != nil {
			return err
		}
		prepared, err := preparePlatformMaterialization(PlatformInstanceMaterialization{
			Identity:    platformWorkspaceIdentity(account.ID, input.WorkspaceID),
			WorkspaceID: input.WorkspaceID, PlatformServiceAccountID: account.ID,
			AvatarURL:                    attachment.AvatarAssetURL(account.AvatarAssetID),
			OperatorUserID:               account.CreatedByOperatorUserID,
			OpenedProvisioningGeneration: input.Generation,
			CorrelationID:                input.ID, Now: now,
		})
		if err != nil {
			return err
		}
		if _, err := service.materializer.materializePlatformInTx(ctx, tx, prepared); err != nil {
			return err
		}
		result, err := tx.Exec(ctx, `
UPDATE platform_service_account_provisioning_inputs
SET state = 'identity_ready', attempt_count = attempt_count + 1,
    last_error = '', version = version + 1, updated_at = $2
WHERE id = $1 AND state = 'pending'
`, input.ID, now)
		if err != nil {
			return fmt.Errorf("mark platform provisioning identity ready: %w", err)
		}
		if result.RowsAffected() != 1 {
			return storeerr.Conflict("platform provisioning input state changed")
		}
		return nil
	})
}

func (service *PlatformService) reconcileAndSettlePlatformProvisioning(ctx context.Context, provisioningInputID string, now time.Time) error {
	input, found, err := loadPlatformProvisioningInputQuery(ctx, service.db, provisioningInputID, false)
	if err != nil {
		return err
	}
	if !found || input.State == serviceaccountstore.PlatformProvisioningStateApplied {
		return nil
	}
	if input.State != serviceaccountstore.PlatformProvisioningStateIdentityReady {
		return storeerr.Conflict("platform provisioning identity is not ready")
	}
	identity := platformWorkspaceIdentity(input.PlatformServiceAccountID, input.WorkspaceID)
	account, err := service.accounts.LoadAccount(ctx, input.WorkspaceID, identity.AccountID)
	if err != nil {
		return err
	}
	ready, err := service.materializer.ReconcileAllWorkspaceHumans(
		ctx, input.WorkspaceID, account.ID, account.Version, account.AudiencePolicyVersion, now,
	)
	if err != nil {
		// A concurrent worker may have completed the same idempotent audience
		// projection under the prior optimistic version.
		if !errors.Is(err, storeerr.ErrConflict) {
			return err
		}
		ready, err = service.accounts.LoadAccount(ctx, input.WorkspaceID, account.ID)
		if err != nil {
			return err
		}
	}
	if ready.AudienceState != serviceaccountstore.AudienceStateReady || ready.EffectiveAudienceVersion != ready.AudiencePolicyVersion {
		return storeerr.Conflict("platform service account audience is not ready")
	}

	return coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		current, found, err := loadPlatformProvisioningInputTx(ctx, tx, input.ID, true)
		if err != nil {
			return err
		}
		if !found || current.State == serviceaccountstore.PlatformProvisioningStateApplied {
			return nil
		}
		if current.State != serviceaccountstore.PlatformProvisioningStateIdentityReady {
			return storeerr.Conflict("platform provisioning input state changed")
		}
		if _, err := tx.Exec(ctx, `
UPDATE platform_service_account_workspace_instances
SET provisioning_state = 'ready', applied_profile_version = $3,
    version = version + 1, updated_at = $4
WHERE platform_service_account_id = $1 AND workspace_id = $2
`, current.PlatformServiceAccountID, current.WorkspaceID, ready.AppliedPlatformProfileVersion, now); err != nil {
			return fmt.Errorf("mark platform workspace instance ready: %w", err)
		}
		result, err := tx.Exec(ctx, `
UPDATE platform_service_account_provisioning_inputs
SET state = 'applied', last_error = '', version = version + 1,
    updated_at = $2, applied_at = COALESCE(applied_at, $2)
WHERE id = $1 AND state = 'identity_ready'
`, current.ID, now)
		if err != nil {
			return fmt.Errorf("settle platform provisioning input: %w", err)
		}
		if result.RowsAffected() != 1 {
			return storeerr.Conflict("platform provisioning input state changed")
		}
		return nil
	})
}

func (service *PlatformService) recordPlatformProvisioningError(ctx context.Context, inputID string, cause error, now time.Time) error {
	message := strings.TrimSpace(cause.Error())
	if len(message) > 4000 {
		message = message[:4000]
	}
	_, err := service.db.Exec(ctx, `
UPDATE platform_service_account_provisioning_inputs
SET attempt_count = attempt_count + 1, last_error = $2,
    version = version + 1, updated_at = $3
WHERE id = $1 AND state IN ('pending','identity_ready')
`, strings.TrimSpace(inputID), message, now)
	if err != nil {
		return fmt.Errorf("record platform provisioning failure: %w", err)
	}
	return nil
}

func (service *PlatformService) advanceOnePlatformProvisioningGeneration(ctx context.Context, now time.Time) (bool, error) {
	if !service.controlPlane {
		return false, nil
	}
	var advanced bool
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		rows, err := tx.Query(ctx, `
SELECT account.id, account.applied_provisioning_generation,
       COALESCE(MAX(input.generation), 0),
       MIN(CASE WHEN input.state <> 'applied' THEN input.generation ELSE NULL END)
FROM platform_service_accounts account
LEFT JOIN platform_service_account_provisioning_inputs input
  ON input.platform_service_account_id = account.id
GROUP BY account.id, account.applied_provisioning_generation
ORDER BY account.id ASC
`)
		if err != nil {
			return fmt.Errorf("list platform provisioning generation candidates: %w", err)
		}
		type candidate struct {
			id      string
			applied int64
			max     int64
			pending sql.NullInt64
		}
		candidates := make([]candidate, 0)
		for rows.Next() {
			var item candidate
			if err := rows.Scan(&item.id, &item.applied, &item.max, &item.pending); err != nil {
				rows.Close()
				return fmt.Errorf("scan platform provisioning generation candidate: %w", err)
			}
			candidates = append(candidates, item)
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return fmt.Errorf("iterate platform provisioning generation candidates: %w", err)
		}
		rows.Close()
		for _, item := range candidates {
			account, err := service.accounts.InTx(tx).LoadPlatformAccountForUpdate(ctx, item.id)
			if err != nil {
				return err
			}
			var maxGeneration int64
			var pendingGeneration sql.NullInt64
			if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(generation), 0),
       MIN(CASE WHEN state <> 'applied' THEN generation ELSE NULL END)
FROM platform_service_account_provisioning_inputs
WHERE platform_service_account_id = $1
`, account.ID).Scan(&maxGeneration, &pendingGeneration); err != nil {
				return fmt.Errorf("load platform provisioning generation fence: %w", err)
			}
			target := maxGeneration
			if pendingGeneration.Valid {
				target = pendingGeneration.Int64 - 1
			}
			state := serviceaccountstore.PlatformProvisioningStatePending
			if target == maxGeneration {
				state = "ready"
			}
			if target == account.AppliedProvisioningGeneration && state == account.ProvisioningState {
				continue
			}
			if _, err := tx.Exec(ctx, `
UPDATE platform_service_accounts
SET applied_provisioning_generation = $2, provisioning_state = $3,
    version = version + 1, updated_at = $4
WHERE id = $1
`, account.ID, target, state, now); err != nil {
				return fmt.Errorf("advance platform provisioning generation: %w", err)
			}
			advanced = true
			break
		}
		return nil
	})
	return advanced, err
}

func platformWorkspaceIdentity(platformAccountID, workspaceID string) IdentityRefs {
	return IdentityRefs{
		AccountID:             platformDeterministicID("sacct", "platform-local-account", platformAccountID, workspaceID),
		ServiceInstallationID: platformDeterministicID("svcinst", "platform-local-installation", platformAccountID, workspaceID),
		ServiceMemberID:       platformDeterministicID("svcmem", "platform-local-service-member", platformAccountID, workspaceID),
		AuthorMemberID:        platformDeterministicID("mem", "platform-local-author-member", platformAccountID, workspaceID),
		MessageScopeID:        platformDeterministicID("sasc", "platform-local-message-scope", platformAccountID, workspaceID),
		PlatformInstanceID:    platformDeterministicID("sapwi", "platform-workspace-instance", platformAccountID, workspaceID),
	}
}

func platformProvisioningInputSelectSQL() string {
	return `
SELECT input.id, input.platform_service_account_id, input.workspace_id,
       input.generation, input.source_kind, input.state, input.attempt_count,
       input.last_error, input.version, input.created_at, input.updated_at,
       input.applied_at
FROM platform_service_account_provisioning_inputs input
`
}

func loadPlatformProvisioningInputTx(ctx context.Context, tx coredb.Tx, inputID string, forUpdate bool) (serviceaccountstore.PlatformProvisioningInput, bool, error) {
	return loadPlatformProvisioningInputQuery(ctx, tx, inputID, forUpdate)
}

func loadPlatformProvisioningInputQuery(ctx context.Context, queryer coredb.QueryRower, inputID string, forUpdate bool) (serviceaccountstore.PlatformProvisioningInput, bool, error) {
	suffix := ""
	if forUpdate {
		suffix = " FOR UPDATE"
	}
	input, err := scanPlatformProvisioningInput(queryer.QueryRow(ctx, platformProvisioningInputSelectSQL()+` WHERE input.id = $1`+suffix, strings.TrimSpace(inputID)))
	if err != nil {
		if coredb.IsNotFound(err) {
			return serviceaccountstore.PlatformProvisioningInput{}, false, nil
		}
		return serviceaccountstore.PlatformProvisioningInput{}, false, fmt.Errorf("load platform provisioning input: %w", err)
	}
	return input, true, nil
}

func scanPlatformProvisioningInput(scanner interface{ Scan(...any) error }) (serviceaccountstore.PlatformProvisioningInput, error) {
	var input serviceaccountstore.PlatformProvisioningInput
	var appliedAt sql.NullTime
	if err := scanner.Scan(
		&input.ID, &input.PlatformServiceAccountID, &input.WorkspaceID,
		&input.Generation, &input.SourceKind, &input.State, &input.AttemptCount,
		&input.LastError, &input.Version, &input.CreatedAt, &input.UpdatedAt,
		&appliedAt,
	); err != nil {
		return serviceaccountstore.PlatformProvisioningInput{}, err
	}
	input.AppliedAt = scanNullableTime(appliedAt)
	return input, nil
}