package serviceaccount

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	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) allocateOnePlatformBroadcast(ctx context.Context, now time.Time) (bool, error) {
	if !service.controlPlane {
		return false, nil
	}
	var allocated bool
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		var broadcastID string
		err := tx.QueryRow(ctx, `
SELECT broadcast.id
FROM platform_service_account_broadcasts broadcast
JOIN platform_service_accounts account ON account.id = broadcast.platform_service_account_id
WHERE broadcast.state = 'waiting_provisioning'
  AND broadcast.broadcast_seq IS NULL
  AND account.applied_provisioning_generation >= broadcast.required_provisioning_generation
  AND NOT EXISTS (
      SELECT 1
      FROM platform_service_account_broadcasts prior
      WHERE prior.platform_service_account_id = broadcast.platform_service_account_id
        AND prior.broadcast_seq IS NULL
        AND prior.state = 'waiting_provisioning'
        AND prior.acceptance_ordinal < broadcast.acceptance_ordinal
  )
ORDER BY broadcast.created_at ASC, broadcast.id ASC
LIMIT 1
FOR UPDATE
`).Scan(&broadcastID)
		if err != nil {
			if coredb.IsNotFound(err) {
				return nil
			}
			return fmt.Errorf("select allocatable platform broadcast: %w", err)
		}
		broadcast, found, err := loadPlatformBroadcastByIDTx(ctx, tx, broadcastID, true)
		if err != nil {
			return err
		}
		if !found || broadcast.State != serviceaccountstore.PlatformBroadcastStateWaitingProvisioning || broadcast.BroadcastSeq != nil {
			return nil
		}
		accountStore := service.accounts.InTx(tx)
		account, err := accountStore.LoadPlatformAccountForUpdate(ctx, broadcast.PlatformServiceAccountID)
		if err != nil {
			return err
		}
		if account.AppliedProvisioningGeneration < broadcast.RequiredProvisioningGeneration {
			return nil
		}
		sequence := account.LatestBroadcastSeq + 1
		sequenceResult, err := tx.Exec(ctx, `
UPDATE platform_service_accounts
SET latest_broadcast_seq = $2, version = version + 1, updated_at = $3
WHERE id = $1 AND latest_broadcast_seq = $4
`, account.ID, sequence, now, account.LatestBroadcastSeq)
		if err != nil {
			return fmt.Errorf("allocate platform broadcast source sequence: %w", err)
		}
		if sequenceResult.RowsAffected() != 1 {
			return storeerr.Conflict("platform broadcast source sequence changed")
		}
		result, err := tx.Exec(ctx, `
UPDATE platform_service_account_broadcasts
SET broadcast_seq = $2, state = 'accepted', version = version + 1,
    updated_at = $3, allocated_at = COALESCE(allocated_at, $3)
WHERE id = $1 AND state = 'waiting_provisioning' AND broadcast_seq IS NULL
`, broadcast.ID, sequence, now)
		if err != nil {
			return fmt.Errorf("assign platform broadcast sequence: %w", err)
		}
		if result.RowsAffected() != 1 {
			return storeerr.Conflict("platform broadcast allocation state changed")
		}
		metadata, err := json.Marshal(map[string]any{
			"broadcastSeq":                   sequence,
			"acceptanceOrdinal":              broadcast.AcceptanceOrdinal,
			"requiredProvisioningGeneration": broadcast.RequiredProvisioningGeneration,
		})
		if err != nil {
			return fmt.Errorf("encode platform broadcast allocation audit: %w", err)
		}
		if err := accountStore.InsertAuditEvent(ctx, serviceaccountstore.AuditEvent{
			ID:                       platformDeterministicID("saaudit", "platform-broadcast-allocated", broadcast.ID),
			PlatformServiceAccountID: account.ID, EventType: "platform_service_account.broadcast_allocated",
			SubjectType: "platform_broadcast", SubjectID: broadcast.ID, SubjectVersion: broadcast.Version + 1,
			ActorType: "internal_service", ActorID: platformBroadcastWorkerID,
			CausedByActorType: "platform_operator", CausedByActorID: broadcast.OperatorPrincipalID,
			ExecutionActorType: "internal_service", ExecutionActorID: platformBroadcastWorkerID,
			PermissionBasisKind: "platform_broadcast_acceptance", PermissionBasisRef: broadcast.ID,
			CorrelationID: broadcast.CorrelationID, Metadata: metadata, CreatedAt: now,
		}); err != nil {
			return err
		}
		allocated = true
		return nil
	})
	return allocated, err
}

func (service *PlatformService) enumerateOnePlatformBroadcastPage(ctx context.Context, now time.Time) (int, int, error) {
	if !service.controlPlane {
		return 0, 0, nil
	}
	var insertedTargets, settledExcluded int
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		var broadcastID string
		err := tx.QueryRow(ctx, `
SELECT broadcast.id
FROM platform_service_account_broadcasts broadcast
JOIN platform_service_accounts account ON account.id = broadcast.platform_service_account_id
WHERE broadcast.broadcast_seq IS NOT NULL
  AND broadcast.enumeration_complete = FALSE
  AND broadcast.state IN ('accepted','dispatching')
ORDER BY broadcast.allocated_at ASC, broadcast.id ASC
LIMIT 1
FOR UPDATE
`).Scan(&broadcastID)
		if err != nil {
			if coredb.IsNotFound(err) {
				return nil
			}
			return fmt.Errorf("select platform broadcast for target enumeration: %w", err)
		}
		broadcast, found, err := loadPlatformBroadcastByIDTx(ctx, tx, broadcastID, true)
		if err != nil {
			return err
		}
		if !found || broadcast.BroadcastSeq == nil || broadcast.EnumerationComplete {
			return nil
		}
		sequence := *broadcast.BroadcastSeq
		rows, err := tx.Query(ctx, `
SELECT id, platform_service_account_id, workspace_id, local_service_account_id,
       opened_provisioning_generation, effective_from_broadcast_seq,
       effective_to_broadcast_seq, last_applied_broadcast_seq,
       applied_profile_version, provisioning_state, version, created_at, updated_at,
       COALESCE((
         SELECT placement.placement_epoch
         FROM workspace_placements placement
         WHERE placement.workspace_id = platform_service_account_workspace_instances.workspace_id
       ), 1)
FROM platform_service_account_workspace_instances
WHERE platform_service_account_id = $1
  AND workspace_id > $2
  AND effective_from_broadcast_seq <= $3
  AND (effective_to_broadcast_seq IS NULL OR $3 < effective_to_broadcast_seq)
ORDER BY workspace_id ASC
LIMIT $4
`, broadcast.PlatformServiceAccountID, broadcast.EnumerationAfterWorkspaceID, sequence, platformProcessBatchSize+1)
		if err != nil {
			return fmt.Errorf("list platform broadcast workspace targets: %w", err)
		}
		instances := make([]platformBroadcastEnumerationInstance, 0, platformProcessBatchSize+1)
		for rows.Next() {
			instance, scanErr := scanPlatformWorkspaceInstanceRow(rows)
			if scanErr != nil {
				rows.Close()
				return fmt.Errorf("scan platform broadcast workspace target: %w", scanErr)
			}
			instances = append(instances, instance)
		}
		if err := rows.Err(); err != nil {
			rows.Close()
			return fmt.Errorf("iterate platform broadcast workspace targets: %w", err)
		}
		rows.Close()
		complete := len(instances) <= platformProcessBatchSize
		if len(instances) > platformProcessBatchSize {
			instances = instances[:platformProcessBatchSize]
		}
		cursor := broadcast.EnumerationAfterWorkspaceID
		var plan PlatformBroadcastPlan
		if err := json.Unmarshal(broadcast.PayloadSnapshot, &plan); err != nil {
			return fmt.Errorf("decode platform broadcast snapshot: %w", err)
		}
		for _, target := range instances {
			instance := target.PlatformWorkspaceInstance
			cursor = instance.WorkspaceID
			eligible := instance.OpenedProvisioningGeneration <= broadcast.RequiredProvisioningGeneration
			state := serviceaccountstore.PlatformDeliveryStateQueued
			outcome := ""
			completedAt := any(nil)
			if !eligible {
				state = serviceaccountstore.PlatformDeliveryStateClosed
				outcome = serviceaccountstore.PlatformDeliveryOutcomeGenerationExcluded
				completedAt = now
			}
			deliveryID := platformDeterministicID("sabdl", "platform-broadcast-delivery", broadcast.ID, instance.WorkspaceID)
			result, err := tx.Exec(ctx, `
INSERT INTO platform_service_account_broadcast_deliveries (
    id, broadcast_id, platform_service_account_id, workspace_id,
    local_service_account_id, broadcast_seq, required_provisioning_generation,
    recipient_eligible, state, attempt_count, lease_token, lease_expires_at,
    next_attempt_at, placement_epoch, local_message_id, outcome, last_error,
    version, created_at, updated_at, completed_at
) VALUES (
    $1, $2, $3, $4,
    $5, $6, $7,
    $8, $9, 0, NULL, NULL,
    $10, $11, NULL, $12, '',
    1, $10, $10, $13
)
ON DUPLICATE KEY UPDATE id = id
`, deliveryID, broadcast.ID, broadcast.PlatformServiceAccountID, instance.WorkspaceID,
				instance.LocalServiceAccountID, sequence, broadcast.RequiredProvisioningGeneration,
				eligible, state, now, target.PlacementEpoch, outcome, completedAt)
			if err != nil {
				return fmt.Errorf("insert platform broadcast delivery: %w", err)
			}
			inserted := result.RowsAffected() == 1
			if eligible && inserted {
				insertedTargets++
				for ordinal, asset := range plan.Assets {
					if _, err := tx.Exec(ctx, `
INSERT INTO platform_service_account_delivery_assets (
    delivery_id, asset_ordinal, source_asset_ref, content_hash,
    media_type, size_bytes, local_attachment_id, state, version,
    created_at, updated_at
) VALUES ($1, $2, $3, $4, $5, $6, NULL, 'pending', 1, $7, $7)
ON DUPLICATE KEY UPDATE delivery_id = delivery_id
`, deliveryID, ordinal, asset.SourceAssetRef, asset.ContentHash, asset.MediaType, asset.SizeBytes, now); err != nil {
						return fmt.Errorf("insert platform broadcast delivery asset: %w", err)
					}
				}
			}
			if !eligible {
				advance, err := tx.Exec(ctx, `
UPDATE platform_service_account_workspace_instances
SET last_applied_broadcast_seq = $3, version = version + 1, updated_at = $4
WHERE platform_service_account_id = $1 AND workspace_id = $2
  AND last_applied_broadcast_seq = $3 - 1
`, instance.PlatformServiceAccountID, instance.WorkspaceID, sequence, now)
				if err != nil {
					return fmt.Errorf("settle generation-excluded platform broadcast cursor: %w", err)
				}
				if advance.RowsAffected() == 1 {
					settledExcluded++
				}
			}
		}
		state := broadcast.State
		if complete {
			state = serviceaccountstore.PlatformBroadcastStateDispatching
		}
		if _, err := tx.Exec(ctx, `
UPDATE platform_service_account_broadcasts
SET enumeration_after_workspace_id = $2,
    enumeration_complete = $3,
    target_count = target_count + $4,
    state = $5,
    version = version + 1,
    updated_at = $6
WHERE id = $1
`, broadcast.ID, cursor, complete, insertedTargets, state, now); err != nil {
			return fmt.Errorf("advance platform broadcast enumeration cursor: %w", err)
		}
		return nil
	})
	return insertedTargets, settledExcluded, err
}

func (service *PlatformService) settleOneExcludedPlatformDelivery(ctx context.Context, now time.Time) (bool, error) {
	afterID := service.workerCursor(platformWorkerCursorExcluded)
	rows, err := service.db.Query(ctx, `
SELECT delivery.id, delivery.workspace_id
FROM platform_service_account_broadcast_deliveries delivery
JOIN platform_service_account_workspace_instances instance
  ON instance.platform_service_account_id = delivery.platform_service_account_id
 AND instance.workspace_id = delivery.workspace_id
WHERE delivery.recipient_eligible = FALSE
  AND delivery.state = 'closed'
  AND delivery.outcome = 'generation_excluded'
  AND instance.last_applied_broadcast_seq = delivery.broadcast_seq - 1
  AND delivery.id > $1
ORDER BY delivery.id ASC
LIMIT $2
`, afterID, platformProcessBatchSize)
	if err != nil {
		return false, fmt.Errorf("list generation-excluded platform settlements: %w", err)
	}
	type candidate struct{ id, workspaceID string }
	candidates := make([]candidate, 0, platformProcessBatchSize)
	for rows.Next() {
		var item candidate
		if err := rows.Scan(&item.id, &item.workspaceID); err != nil {
			rows.Close()
			return false, fmt.Errorf("scan generation-excluded platform settlement: %w", err)
		}
		candidates = append(candidates, item)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return false, fmt.Errorf("iterate generation-excluded platform settlements: %w", err)
	}
	rows.Close()
	for _, item := range candidates {
		if !owner.IsManaged(service.managedWorkspaces, item.workspaceID) {
			continue
		}
		if len(candidates) == platformProcessBatchSize {
			service.setWorkerCursor(platformWorkerCursorExcluded, item.id)
		} else {
			service.setWorkerCursor(platformWorkerCursorExcluded, "")
		}
		settled := false
		err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
			delivery, found, err := loadPlatformBroadcastDeliveryTx(ctx, tx, item.id, true)
			if err != nil || !found {
				return err
			}
			if !owner.IsManaged(service.managedWorkspaces, delivery.WorkspaceID) {
				return nil
			}
			var currentPlacementEpoch int64
			if err := tx.QueryRow(ctx, `
SELECT COALESCE((
  SELECT placement_epoch FROM workspace_placements WHERE workspace_id = $1
), 1)
`, delivery.WorkspaceID).Scan(&currentPlacementEpoch); err != nil {
				return fmt.Errorf("load generation-excluded workspace placement epoch: %w", err)
			}
			result, err := tx.Exec(ctx, `
UPDATE platform_service_account_workspace_instances
SET last_applied_broadcast_seq = $3, version = version + 1, updated_at = $4
WHERE platform_service_account_id = $1 AND workspace_id = $2
  AND last_applied_broadcast_seq = $3 - 1
`, delivery.PlatformServiceAccountID, delivery.WorkspaceID, delivery.BroadcastSeq, now)
			if err != nil {
				return fmt.Errorf("advance generation-excluded platform cursor: %w", err)
			}
			settled = result.RowsAffected() == 1
			if settled && currentPlacementEpoch != delivery.PlacementEpoch {
				if _, err := tx.Exec(ctx, `
UPDATE platform_service_account_broadcast_deliveries
SET placement_epoch = $2, version = version + 1, updated_at = $3
WHERE id = $1
`, delivery.ID, currentPlacementEpoch, now); err != nil {
					return fmt.Errorf("settle generation-excluded placement epoch: %w", err)
				}
			}
			return nil
		})
		return settled, err
	}
	if len(candidates) == platformProcessBatchSize {
		service.setWorkerCursor(platformWorkerCursorExcluded, candidates[len(candidates)-1].id)
	} else {
		service.setWorkerCursor(platformWorkerCursorExcluded, "")
	}
	return false, nil
}

func (service *PlatformService) deliverOnePlatformBroadcast(ctx context.Context, now time.Time) (delivered bool, retried bool, err error) {
	if service.messageWriter == nil {
		return false, false, nil
	}
	afterID := service.workerCursor(platformWorkerCursorDelivery)
	rows, err := service.db.Query(ctx, `
SELECT delivery.id, delivery.workspace_id
FROM platform_service_account_broadcast_deliveries delivery
WHERE delivery.recipient_eligible = TRUE
  AND delivery.state IN ('queued','leased','waiting_instance','waiting_audience','retry_wait')
  AND delivery.next_attempt_at <= $1
  AND (delivery.state <> 'leased' OR delivery.lease_expires_at <= $1)
  AND delivery.id > $2
ORDER BY delivery.id ASC
LIMIT $3
`, now, afterID, platformProcessBatchSize)
	if err != nil {
		return false, false, fmt.Errorf("list due platform broadcast deliveries: %w", err)
	}
	type candidate struct{ id, workspaceID string }
	candidates := make([]candidate, 0, platformProcessBatchSize)
	for rows.Next() {
		var item candidate
		if err := rows.Scan(&item.id, &item.workspaceID); err != nil {
			rows.Close()
			return false, false, fmt.Errorf("scan due platform broadcast delivery: %w", err)
		}
		candidates = append(candidates, item)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return false, false, fmt.Errorf("iterate due platform broadcast deliveries: %w", err)
	}
	rows.Close()
	for _, item := range candidates {
		if !owner.IsManaged(service.managedWorkspaces, item.workspaceID) {
			continue
		}
		if len(candidates) == platformProcessBatchSize {
			service.setWorkerCursor(platformWorkerCursorDelivery, item.id)
		} else {
			service.setWorkerCursor(platformWorkerCursorDelivery, "")
		}
		lease, leased, err := service.leasePlatformBroadcastDelivery(ctx, item.id, now)
		if err != nil {
			return false, false, err
		}
		if !leased {
			return false, false, nil
		}
		if _, err := service.messageWriter.DeliverPlatformServiceAccountBroadcast(ctx, lease.ID, lease.LeaseToken, now); err != nil {
			if settleErr := service.recordPlatformDeliveryFailure(ctx, lease, err, now); settleErr != nil {
				return false, false, settleErr
			}
			return false, true, nil
		}
		return true, false, nil
	}
	if len(candidates) == platformProcessBatchSize {
		service.setWorkerCursor(platformWorkerCursorDelivery, candidates[len(candidates)-1].id)
	} else {
		service.setWorkerCursor(platformWorkerCursorDelivery, "")
	}
	return false, false, nil
}

func (service *PlatformService) leasePlatformBroadcastDelivery(ctx context.Context, deliveryID string, now time.Time) (serviceaccountstore.PlatformBroadcastDelivery, bool, error) {
	var result serviceaccountstore.PlatformBroadcastDelivery
	var leased bool
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		delivery, found, err := loadPlatformBroadcastDeliveryTx(ctx, tx, deliveryID, true)
		if err != nil || !found {
			return err
		}
		if !delivery.RecipientEligible || delivery.State == serviceaccountstore.PlatformDeliveryStateDelivered || delivery.State == serviceaccountstore.PlatformDeliveryStateOperatorAttention {
			return nil
		}
		if delivery.State == serviceaccountstore.PlatformDeliveryStateLeased && delivery.LeaseExpiresAt != nil && delivery.LeaseExpiresAt.After(now) {
			return nil
		}
		var audienceState, instanceState string
		var audiencePolicyVersion, effectiveAudienceVersion, cursor, currentPlacementEpoch int64
		if err := tx.QueryRow(ctx, `
SELECT local.audience_state,
       local.audience_policy_version, local.effective_audience_version,
       instance.provisioning_state, instance.last_applied_broadcast_seq,
       COALESCE((
         SELECT placement.placement_epoch
         FROM workspace_placements placement
         WHERE placement.workspace_id = instance.workspace_id
       ), 1)
FROM platform_service_accounts root
JOIN platform_service_account_workspace_instances instance
  ON instance.platform_service_account_id = root.id
 AND instance.workspace_id = $2
JOIN service_accounts local
  ON local.id = instance.local_service_account_id
 AND local.workspace_id = instance.workspace_id
WHERE root.id = $1
FOR UPDATE
`, delivery.PlatformServiceAccountID, delivery.WorkspaceID).Scan(
			&audienceState,
			&audiencePolicyVersion, &effectiveAudienceVersion,
			&instanceState, &cursor, &currentPlacementEpoch,
		); err != nil {
			return fmt.Errorf("lock platform delivery authority: %w", err)
		}
		if !owner.IsManaged(service.managedWorkspaces, delivery.WorkspaceID) {
			return nil
		}
		waitState := ""
		if instanceState != "ready" || cursor != delivery.BroadcastSeq-1 {
			waitState = serviceaccountstore.PlatformDeliveryStateWaitingInstance
		} else if audienceState != serviceaccountstore.AudienceStateReady || effectiveAudienceVersion != audiencePolicyVersion {
			waitState = serviceaccountstore.PlatformDeliveryStateWaitingAudience
		}
		if waitState != "" {
			if _, err := tx.Exec(ctx, `
UPDATE platform_service_account_broadcast_deliveries
SET state = $2, lease_token = NULL, lease_expires_at = NULL,
    next_attempt_at = $3, version = version + 1, updated_at = $4
WHERE id = $1
`, delivery.ID, waitState, now.Add(5*time.Second), now); err != nil {
				return fmt.Errorf("defer platform broadcast delivery: %w", err)
			}
			return nil
		}
		leaseToken, err := identifiers.NewEntityID("sablease")
		if err != nil {
			return fmt.Errorf("generate platform broadcast delivery lease: %w", err)
		}
		leaseExpiry := now.Add(30 * time.Second)
		resultUpdate, err := tx.Exec(ctx, `
UPDATE platform_service_account_broadcast_deliveries
SET state = 'leased', attempt_count = attempt_count + 1,
    lease_token = $2, lease_expires_at = $3,
    next_attempt_at = $3, placement_epoch = $5,
    version = version + 1, updated_at = $4
WHERE id = $1
  AND state IN ('queued','leased','waiting_instance','waiting_audience','retry_wait')
`, delivery.ID, leaseToken, leaseExpiry, now, currentPlacementEpoch)
		if err != nil {
			return fmt.Errorf("lease platform broadcast delivery: %w", err)
		}
		if resultUpdate.RowsAffected() != 1 {
			return nil
		}
		result, _, err = loadPlatformBroadcastDeliveryTx(ctx, tx, delivery.ID, false)
		leased = err == nil
		return err
	})
	return result, leased, err
}

func (service *PlatformService) recordPlatformDeliveryFailure(ctx context.Context, lease serviceaccountstore.PlatformBroadcastDelivery, cause error, now time.Time) error {
	lastError := strings.TrimSpace(cause.Error())
	if len(lastError) > 4000 {
		lastError = lastError[:4000]
	}
	state := serviceaccountstore.PlatformDeliveryStateRetryWait
	outcome := ""
	nextAttempt := now.Add(time.Duration(lease.AttemptCount*lease.AttemptCount) * time.Second)
	completedAt := any(nil)
	if lease.AttemptCount >= platformDeliveryMaxAttempts {
		state = serviceaccountstore.PlatformDeliveryStateOperatorAttention
		outcome = serviceaccountstore.PlatformDeliveryOutcomeRetryExhausted
		nextAttempt = now
		completedAt = now
	}
	result, err := service.db.Exec(ctx, `
UPDATE platform_service_account_broadcast_deliveries
SET state = $3, lease_token = NULL, lease_expires_at = NULL,
    next_attempt_at = $4, outcome = $5, last_error = $6,
    version = version + 1, updated_at = $7, completed_at = $8
WHERE id = $1 AND lease_token = $2 AND state = 'leased'
`, lease.ID, lease.LeaseToken, state, nextAttempt, outcome, lastError, now, completedAt)
	if err != nil {
		return fmt.Errorf("record platform broadcast delivery failure: %w", err)
	}
	if result.RowsAffected() != 1 {
		return storeerr.Conflict("platform broadcast delivery lease changed")
	}
	return nil
}

func (service *PlatformService) aggregateOnePlatformBroadcast(ctx context.Context, now time.Time) (bool, error) {
	if !service.controlPlane {
		return false, nil
	}
	var completed bool
	err := coredb.WithTx(ctx, service.db, func(tx coredb.Tx) error {
		var broadcastID string
		err := tx.QueryRow(ctx, `
SELECT id
FROM platform_service_account_broadcasts
WHERE enumeration_complete = TRUE
  AND state IN ('accepted','dispatching')
ORDER BY broadcast_seq ASC, id ASC
LIMIT 1
FOR UPDATE
`).Scan(&broadcastID)
		if err != nil {
			if coredb.IsNotFound(err) {
				return nil
			}
			return fmt.Errorf("select platform broadcast aggregate: %w", err)
		}
		broadcast, found, err := loadPlatformBroadcastByIDTx(ctx, tx, broadcastID, true)
		if err != nil || !found {
			return err
		}
		var targetCount, deliveredCount, skippedCount, exceptionCount, pendingCount int64
		if err := tx.QueryRow(ctx, `
SELECT
  COALESCE(SUM(CASE WHEN recipient_eligible THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN recipient_eligible AND state = 'delivered' THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN recipient_eligible AND state IN ('closed','skipped') THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN recipient_eligible AND state = 'operator_attention' THEN 1 ELSE 0 END), 0),
  COALESCE(SUM(CASE WHEN recipient_eligible AND state NOT IN ('delivered','closed','skipped','operator_attention') THEN 1 ELSE 0 END), 0)
FROM platform_service_account_broadcast_deliveries
WHERE broadcast_id = $1
`, broadcast.ID).Scan(&targetCount, &deliveredCount, &skippedCount, &exceptionCount, &pendingCount); err != nil {
			return fmt.Errorf("aggregate platform broadcast deliveries: %w", err)
		}
		var unsettledExcluded int64
		if err := tx.QueryRow(ctx, `
SELECT COUNT(*)
FROM platform_service_account_broadcast_deliveries delivery
JOIN platform_service_account_workspace_instances instance
  ON instance.platform_service_account_id = delivery.platform_service_account_id
 AND instance.workspace_id = delivery.workspace_id
WHERE delivery.broadcast_id = $1
  AND delivery.recipient_eligible = FALSE
  AND delivery.outcome = 'generation_excluded'
  AND instance.last_applied_broadcast_seq < delivery.broadcast_seq
`, broadcast.ID).Scan(&unsettledExcluded); err != nil {
			return fmt.Errorf("aggregate generation-excluded platform settlements: %w", err)
		}
		state := serviceaccountstore.PlatformBroadcastStateDispatching
		completedAt := any(nil)
		if pendingCount == 0 && unsettledExcluded == 0 {
			state = serviceaccountstore.PlatformBroadcastStateCompleted
			if exceptionCount > 0 || skippedCount > 0 {
				state = serviceaccountstore.PlatformBroadcastStateCompletedExceptions
			}
			completedAt = now
			completed = true
		}
		if broadcast.State == state &&
			broadcast.TargetCount == targetCount &&
			broadcast.DeliveredCount == deliveredCount &&
			broadcast.SkippedCount == skippedCount &&
			broadcast.ExceptionCount == exceptionCount {
			return nil
		}
		if _, err := tx.Exec(ctx, `
UPDATE platform_service_account_broadcasts
SET state = $2, target_count = $3, delivered_count = $4,
    skipped_count = $5, exception_count = $6,
    version = version + 1, updated_at = $7,
    completed_at = COALESCE(completed_at, $8)
WHERE id = $1
`, broadcast.ID, state, targetCount, deliveredCount, skippedCount, exceptionCount, now, completedAt); err != nil {
			return fmt.Errorf("update platform broadcast aggregate: %w", err)
		}
		return nil
	})
	return completed, err
}

func platformBroadcastDeliverySelectSQL() string {
	return `
SELECT id, broadcast_id, platform_service_account_id, workspace_id,
       local_service_account_id, broadcast_seq, required_provisioning_generation,
       recipient_eligible, state, attempt_count, lease_token, lease_expires_at,
       next_attempt_at, placement_epoch, local_message_id, outcome, last_error,
       version, created_at, updated_at, completed_at
FROM platform_service_account_broadcast_deliveries`
}

func loadPlatformBroadcastDeliveryTx(ctx context.Context, tx coredb.Tx, deliveryID string, forUpdate bool) (serviceaccountstore.PlatformBroadcastDelivery, bool, error) {
	suffix := ""
	if forUpdate {
		suffix = " FOR UPDATE"
	}
	delivery, err := scanPlatformBroadcastDelivery(tx.QueryRow(ctx, platformBroadcastDeliverySelectSQL()+` WHERE id = $1`+suffix, strings.TrimSpace(deliveryID)))
	if err != nil {
		if coredb.IsNotFound(err) {
			return serviceaccountstore.PlatformBroadcastDelivery{}, false, nil
		}
		return serviceaccountstore.PlatformBroadcastDelivery{}, false, fmt.Errorf("load platform broadcast delivery: %w", err)
	}
	return delivery, true, nil
}

func scanPlatformBroadcastDelivery(scanner interface{ Scan(...any) error }) (serviceaccountstore.PlatformBroadcastDelivery, error) {
	var delivery serviceaccountstore.PlatformBroadcastDelivery
	var localAccountID, leaseToken, localMessageID sql.NullString
	var leaseExpiresAt, completedAt sql.NullTime
	if err := scanner.Scan(
		&delivery.ID, &delivery.BroadcastID, &delivery.PlatformServiceAccountID,
		&delivery.WorkspaceID, &localAccountID, &delivery.BroadcastSeq,
		&delivery.RequiredProvisioningGeneration, &delivery.RecipientEligible,
		&delivery.State, &delivery.AttemptCount, &leaseToken, &leaseExpiresAt,
		&delivery.NextAttemptAt, &delivery.PlacementEpoch, &localMessageID,
		&delivery.Outcome, &delivery.LastError, &delivery.Version,
		&delivery.CreatedAt, &delivery.UpdatedAt, &completedAt,
	); err != nil {
		return serviceaccountstore.PlatformBroadcastDelivery{}, err
	}
	delivery.LocalServiceAccountID = strings.TrimSpace(localAccountID.String)
	delivery.LeaseToken = strings.TrimSpace(leaseToken.String)
	delivery.LocalMessageID = strings.TrimSpace(localMessageID.String)
	delivery.LeaseExpiresAt = scanNullableTime(leaseExpiresAt)
	delivery.CompletedAt = scanNullableTime(completedAt)
	return delivery, nil
}

type platformBroadcastEnumerationInstance struct {
	serviceaccountstore.PlatformWorkspaceInstance
	PlacementEpoch int64
}

func scanPlatformWorkspaceInstanceRow(scanner interface{ Scan(...any) error }) (platformBroadcastEnumerationInstance, error) {
	var target platformBroadcastEnumerationInstance
	instance := &target.PlatformWorkspaceInstance
	var effectiveTo sql.NullInt64
	if err := scanner.Scan(
		&instance.ID, &instance.PlatformServiceAccountID, &instance.WorkspaceID,
		&instance.LocalServiceAccountID, &instance.OpenedProvisioningGeneration,
		&instance.EffectiveFromBroadcastSeq, &effectiveTo,
		&instance.LastAppliedBroadcastSeq, &instance.AppliedProfileVersion,
		&instance.ProvisioningState, &instance.Version,
		&instance.CreatedAt, &instance.UpdatedAt, &target.PlacementEpoch,
	); err != nil {
		return platformBroadcastEnumerationInstance{}, err
	}
	if effectiveTo.Valid {
		value := effectiveTo.Int64
		instance.EffectiveToBroadcastSeq = &value
	}
	return target, nil
}