package serviceaccount

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strings"
	"unicode"
	"unicode/utf8"

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"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"
)

// SubmitPlatformBroadcast durably accepts one immutable MessagePlan under the
// platform-account source lock. Sequence allocation remains fenced on the
// captured provisioning generation and acceptance ordinal.
func (service *PlatformService) SubmitPlatformBroadcast(ctx context.Context, input PlatformBroadcastSubmit) (serviceaccountstore.PlatformBroadcast, error) {
	if err := service.validateControlPlane(); err != nil {
		return serviceaccountstore.PlatformBroadcast{}, err
	}
	input.PlatformServiceAccountID = strings.TrimSpace(input.PlatformServiceAccountID)
	input.OperatorUserID = strings.TrimSpace(input.OperatorUserID)
	input.ClientBroadcastID = strings.TrimSpace(input.ClientBroadcastID)
	input.CorrelationID = strings.TrimSpace(input.CorrelationID)
	input.Now = service.normalizedNow(input.Now)
	if input.PlatformServiceAccountID == "" || input.OperatorUserID == "" || input.ClientBroadcastID == "" {
		return serviceaccountstore.PlatformBroadcast{}, storeerr.BadRequest("platformServiceAccountId, operatorUserId, and clientBroadcastId are required")
	}
	if len(input.ClientBroadcastID) > 128 {
		return serviceaccountstore.PlatformBroadcast{}, storeerr.BadRequest("clientBroadcastId must be at most 128 bytes")
	}
	plan, snapshot, digest, err := normalizePlatformBroadcastPlan(input.Plan)
	if err != nil {
		return serviceaccountstore.PlatformBroadcast{}, err
	}
	input.Plan = plan

	var result serviceaccountstore.PlatformBroadcast
	err = coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		if err := requirePlatformOperatorTx(ctx, tx, input.OperatorUserID); err != nil {
			return err
		}
		accounts := service.accounts.InTx(tx)
		account, err := accounts.LoadPlatformAccountForUpdate(ctx, input.PlatformServiceAccountID)
		if err != nil {
			return err
		}
		existing, found, err := loadPlatformBroadcastByClientTx(ctx, tx, account.ID, input.OperatorUserID, input.ClientBroadcastID, true)
		if err != nil {
			return err
		}
		if found {
			if existing.PayloadDigest != digest {
				return storeerr.Conflict("clientBroadcastId was already used with a different payload")
			}
			result = existing
			return nil
		}
		for {
			discovered, _, discoverErr := discoverPlatformProvisioningInputsTx(ctx, tx, account.ID, serviceaccountstore.PlatformProvisioningSourceWorkspaceCreate, input.Now, platformProcessBatchSize)
			if discoverErr != nil {
				return discoverErr
			}
			if discovered < platformProcessBatchSize {
				break
			}
		}

		var requiredGeneration, acceptanceOrdinal int64
		if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(generation), 0)
FROM platform_service_account_provisioning_inputs
WHERE platform_service_account_id = $1
`, account.ID).Scan(&requiredGeneration); err != nil {
			return fmt.Errorf("capture platform broadcast provisioning generation: %w", err)
		}
		if err := tx.QueryRow(ctx, `
SELECT COALESCE(MAX(acceptance_ordinal), 0) + 1
FROM platform_service_account_broadcasts
WHERE platform_service_account_id = $1
`, account.ID).Scan(&acceptanceOrdinal); err != nil {
			return fmt.Errorf("allocate platform broadcast acceptance ordinal: %w", err)
		}
		broadcastID, err := identifiers.NewEntityID("sabrd")
		if err != nil {
			return fmt.Errorf("generate platform broadcast id: %w", err)
		}
		if _, err := tx.Exec(ctx, `
INSERT INTO platform_service_account_broadcasts (
    id, platform_service_account_id, operator_principal_id, client_broadcast_id,
    acceptance_ordinal, required_provisioning_generation, broadcast_seq,
    payload_snapshot, payload_digest, state, enumeration_after_workspace_id,
    enumeration_complete, target_count, delivered_count, skipped_count,
    exception_count, version, correlation_id, created_at, updated_at
) VALUES (
    $1, $2, $3, $4,
    $5, $6, NULL,
    $7, $8, 'waiting_provisioning', '',
    FALSE, 0, 0, 0,
    0, 1, $9, $10, $10
)
`, broadcastID, account.ID, input.OperatorUserID, input.ClientBroadcastID,
			acceptanceOrdinal, requiredGeneration, snapshot, digest, input.CorrelationID, input.Now); err != nil {
			if coredb.IsUniqueViolation(err) {
				replayed, replayFound, loadErr := loadPlatformBroadcastByClientTx(ctx, tx, account.ID, input.OperatorUserID, input.ClientBroadcastID, true)
				if loadErr != nil {
					return loadErr
				}
				if replayFound && replayed.PayloadDigest == digest {
					result = replayed
					return nil
				}
				return storeerr.Conflict("platform broadcast idempotency key or source ordinal changed")
			}
			return fmt.Errorf("insert platform service account broadcast: %w", err)
		}
		metadata, err := json.Marshal(map[string]any{
			"clientBroadcastId":              input.ClientBroadcastID,
			"acceptanceOrdinal":              acceptanceOrdinal,
			"requiredProvisioningGeneration": requiredGeneration,
			"payloadDigest":                  digest,
		})
		if err != nil {
			return fmt.Errorf("encode platform broadcast acceptance audit: %w", err)
		}
		if err := accounts.InsertAuditEvent(ctx, serviceaccountstore.AuditEvent{
			ID:                       platformDeterministicID("saaudit", "platform-broadcast-accepted", broadcastID),
			PlatformServiceAccountID: account.ID, EventType: "platform_service_account.broadcast_accepted",
			SubjectType: "platform_broadcast", SubjectID: broadcastID, SubjectVersion: 1,
			ActorType: "platform_operator", ActorID: input.OperatorUserID,
			CausedByActorType: "platform_operator", CausedByActorID: input.OperatorUserID,
			ExecutionActorType: "platform_operator", ExecutionActorID: input.OperatorUserID,
			PermissionBasisKind: "platform_operator", PermissionBasisRef: input.OperatorUserID,
			CorrelationID: input.CorrelationID, Metadata: metadata, CreatedAt: input.Now,
		}); err != nil {
			return err
		}
		result, _, err = loadPlatformBroadcastByIDTx(ctx, tx, broadcastID, false)
		return err
	})
	return result, err
}

func (service *PlatformService) GetPlatformBroadcast(ctx context.Context, platformAccountID, broadcastID string) (serviceaccountstore.PlatformBroadcast, error) {
	if service == nil || service.db == nil {
		return serviceaccountstore.PlatformBroadcast{}, fmt.Errorf("platform service account service is not configured")
	}
	var result serviceaccountstore.PlatformBroadcast
	result, found, err := loadPlatformBroadcastByIDQuery(ctx, service.db, strings.TrimSpace(broadcastID), false)
	if err != nil {
		return serviceaccountstore.PlatformBroadcast{}, err
	}
	if !found || result.PlatformServiceAccountID != strings.TrimSpace(platformAccountID) {
		return serviceaccountstore.PlatformBroadcast{}, storeerr.NotFound("platform broadcast not found")
	}
	return result, nil
}

func (service *PlatformService) ListPlatformBroadcasts(ctx context.Context, platformAccountID, afterID string, limit int) ([]serviceaccountstore.PlatformBroadcast, error) {
	if service == nil || service.db == nil {
		return nil, fmt.Errorf("platform service account service is not configured")
	}
	if limit <= 0 || limit > 100 {
		return nil, storeerr.BadRequest("limit must be between 1 and 100")
	}
	rows, err := service.db.Query(ctx, platformBroadcastSelectSQL()+`
WHERE platform_service_account_id = $1 AND id > $2
ORDER BY id ASC
LIMIT $3
`, strings.TrimSpace(platformAccountID), strings.TrimSpace(afterID), limit)
	if err != nil {
		return nil, fmt.Errorf("list platform broadcasts: %w", err)
	}
	defer rows.Close()
	broadcasts := make([]serviceaccountstore.PlatformBroadcast, 0, limit)
	for rows.Next() {
		broadcast, scanErr := scanPlatformBroadcast(rows)
		if scanErr != nil {
			return nil, fmt.Errorf("scan platform broadcast: %w", scanErr)
		}
		broadcasts = append(broadcasts, broadcast)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate platform broadcasts: %w", err)
	}
	return broadcasts, nil
}

// ListPlatformBroadcastItems returns the broadcast ledger together with the
// Human-facing publisher and the current Workspace's delivered Inbox message.
// The workspace projection is optional: an empty workspace id still returns
// every broadcast, but without a local message locator.
func (service *PlatformService) ListPlatformBroadcastItems(ctx context.Context, platformAccountID, workspaceID, afterID string, limit int) ([]PlatformBroadcastListItem, error) {
	broadcasts, err := service.ListPlatformBroadcasts(ctx, platformAccountID, afterID, limit)
	if err != nil {
		return nil, err
	}
	if len(broadcasts) == 0 {
		return []PlatformBroadcastListItem{}, nil
	}
	rows, err := service.db.Query(ctx, `
SELECT broadcast.id,
       COALESCE(NULLIF(TRIM(operator.display_name), ''), NULLIF(TRIM(operator.name), ''), broadcast.operator_principal_id),
       COALESCE(operator.avatar_url, ''),
       COALESCE(delivery.local_service_account_id, ''),
       COALESCE(delivery.local_message_id, '')
FROM platform_service_account_broadcasts broadcast
JOIN users operator ON operator.id = broadcast.operator_principal_id
LEFT JOIN platform_service_account_broadcast_deliveries delivery
  ON delivery.broadcast_id = broadcast.id
 AND delivery.workspace_id = $2
WHERE broadcast.platform_service_account_id = $1
  AND broadcast.id > $3
ORDER BY broadcast.id ASC
LIMIT $4
`, strings.TrimSpace(platformAccountID), strings.TrimSpace(workspaceID), strings.TrimSpace(afterID), limit)
	if err != nil {
		return nil, fmt.Errorf("list platform broadcast presentations: %w", err)
	}
	defer rows.Close()

	type presentation struct {
		operatorDisplayName       string
		operatorAvatarURL         string
		workspaceServiceAccountID string
		workspaceMessageID        string
	}
	presentations := make(map[string]presentation, len(broadcasts))
	for rows.Next() {
		var broadcastID string
		var item presentation
		if err := rows.Scan(
			&broadcastID,
			&item.operatorDisplayName,
			&item.operatorAvatarURL,
			&item.workspaceServiceAccountID,
			&item.workspaceMessageID,
		); err != nil {
			return nil, fmt.Errorf("scan platform broadcast presentation: %w", err)
		}
		presentations[broadcastID] = item
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate platform broadcast presentations: %w", err)
	}

	items := make([]PlatformBroadcastListItem, 0, len(broadcasts))
	for _, broadcast := range broadcasts {
		presentation := presentations[broadcast.ID]
		items = append(items, PlatformBroadcastListItem{
			Broadcast:                 broadcast,
			OperatorDisplayName:       strings.TrimSpace(presentation.operatorDisplayName),
			OperatorAvatarURL:         strings.TrimSpace(presentation.operatorAvatarURL),
			WorkspaceServiceAccountID: strings.TrimSpace(presentation.workspaceServiceAccountID),
			WorkspaceMessageID:        strings.TrimSpace(presentation.workspaceMessageID),
		})
	}
	return items, nil
}

func normalizePlatformBroadcastPlan(input PlatformBroadcastPlan) (PlatformBroadcastPlan, []byte, string, error) {
	input.Title = strings.TrimSpace(input.Title)
	input.Content = strings.TrimSpace(input.Content)
	if input.Title == "" || utf8.RuneCountInString(input.Title) > 120 {
		return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("broadcast title must be between 1 and 120 characters")
	}
	for _, r := range input.Title {
		if unicode.IsControl(r) {
			return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("broadcast title cannot contain control characters")
		}
	}
	if input.Content == "" {
		return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("broadcast content is required")
	}
	input.MessageType = message.MessageTypeServiceAccountPost
	if !message.ClassifyMessageSemantics(input.MessageType, input.Metadata).CurrentWriteAllowed() {
		return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("broadcast message type is not allowed for active writes")
	}
	metadata := message.CopyMessageMetadata(input.Metadata)
	if _, reserved := metadata[platformBroadcastMetadataKey]; reserved {
		return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest(platformBroadcastMetadataKey + " metadata is server-owned")
	}
	if _, reserved := metadata["serviceAccountPost"]; reserved {
		return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("serviceAccountPost metadata is server-owned")
	}
	if metadata == nil {
		metadata = make(map[string]any, 1)
	}
	metadata["serviceAccountPost"] = map[string]any{"title": input.Title}
	input.Metadata = metadata
	for index := range input.Assets {
		asset := &input.Assets[index]
		asset.SourceAssetRef = strings.TrimSpace(asset.SourceAssetRef)
		asset.ContentHash = strings.ToLower(strings.TrimSpace(asset.ContentHash))
		asset.MediaType = strings.TrimSpace(asset.MediaType)
		asset.ClientLabel = strings.TrimSpace(asset.ClientLabel)
		if asset.SourceAssetRef == "" || len(asset.ContentHash) != 64 || asset.SizeBytes < 0 {
			return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("broadcast assets require sourceAssetRef, SHA-256 contentHash, and non-negative sizeBytes")
		}
		decoded, err := hex.DecodeString(asset.ContentHash)
		if err != nil || len(decoded) != sha256.Size {
			return PlatformBroadcastPlan{}, nil, "", storeerr.BadRequest("broadcast asset contentHash must be a SHA-256 hex digest")
		}
	}
	snapshot, err := json.Marshal(input)
	if err != nil {
		return PlatformBroadcastPlan{}, nil, "", fmt.Errorf("encode platform broadcast plan: %w", err)
	}
	sum := sha256.Sum256(snapshot)
	return input, snapshot, hex.EncodeToString(sum[:]), nil
}

func platformBroadcastSelectSQL() string {
	return `
SELECT id, platform_service_account_id, operator_principal_id, client_broadcast_id,
       acceptance_ordinal, required_provisioning_generation, broadcast_seq,
       payload_snapshot, payload_digest, state, enumeration_after_workspace_id,
       enumeration_complete, target_count, delivered_count, skipped_count,
       exception_count, version, correlation_id, created_at, updated_at,
       allocated_at, completed_at, closed_at
FROM platform_service_account_broadcasts`
}

func loadPlatformBroadcastByClientTx(ctx context.Context, tx coredb.Tx, platformAccountID, operatorID, clientID string, forUpdate bool) (serviceaccountstore.PlatformBroadcast, bool, error) {
	suffix := ""
	if forUpdate {
		suffix = " FOR UPDATE"
	}
	return scanPlatformBroadcastMaybe(tx.QueryRow(ctx, platformBroadcastSelectSQL()+`
WHERE platform_service_account_id = $1
  AND operator_principal_id = $2
  AND client_broadcast_id = $3`+suffix, platformAccountID, operatorID, clientID))
}

func loadPlatformBroadcastByIDTx(ctx context.Context, tx coredb.Tx, broadcastID string, forUpdate bool) (serviceaccountstore.PlatformBroadcast, bool, error) {
	return loadPlatformBroadcastByIDQuery(ctx, tx, broadcastID, forUpdate)
}

func loadPlatformBroadcastByIDQuery(ctx context.Context, queryer coredb.QueryRower, broadcastID string, forUpdate bool) (serviceaccountstore.PlatformBroadcast, bool, error) {
	suffix := ""
	if forUpdate {
		suffix = " FOR UPDATE"
	}
	return scanPlatformBroadcastMaybe(queryer.QueryRow(ctx, platformBroadcastSelectSQL()+` WHERE id = $1`+suffix, strings.TrimSpace(broadcastID)))
}

func scanPlatformBroadcastMaybe(scanner interface{ Scan(...any) error }) (serviceaccountstore.PlatformBroadcast, bool, error) {
	broadcast, err := scanPlatformBroadcast(scanner)
	if err != nil {
		if coredb.IsNotFound(err) {
			return serviceaccountstore.PlatformBroadcast{}, false, nil
		}
		return serviceaccountstore.PlatformBroadcast{}, false, fmt.Errorf("load platform broadcast: %w", err)
	}
	return broadcast, true, nil
}

func scanPlatformBroadcast(scanner interface{ Scan(...any) error }) (serviceaccountstore.PlatformBroadcast, error) {
	var broadcast serviceaccountstore.PlatformBroadcast
	var sequence sql.NullInt64
	var allocatedAt, completedAt, closedAt sql.NullTime
	if err := scanner.Scan(
		&broadcast.ID, &broadcast.PlatformServiceAccountID, &broadcast.OperatorPrincipalID,
		&broadcast.ClientBroadcastID, &broadcast.AcceptanceOrdinal,
		&broadcast.RequiredProvisioningGeneration, &sequence,
		&broadcast.PayloadSnapshot, &broadcast.PayloadDigest, &broadcast.State,
		&broadcast.EnumerationAfterWorkspaceID, &broadcast.EnumerationComplete,
		&broadcast.TargetCount, &broadcast.DeliveredCount, &broadcast.SkippedCount,
		&broadcast.ExceptionCount, &broadcast.Version, &broadcast.CorrelationID,
		&broadcast.CreatedAt, &broadcast.UpdatedAt, &allocatedAt, &completedAt, &closedAt,
	); err != nil {
		return serviceaccountstore.PlatformBroadcast{}, err
	}
	if sequence.Valid {
		value := sequence.Int64
		broadcast.BroadcastSeq = &value
	}
	broadcast.AllocatedAt = scanNullableTime(allocatedAt)
	broadcast.CompletedAt = scanNullableTime(completedAt)
	broadcast.ClosedAt = scanNullableTime(closedAt)
	return broadcast, nil
}