package serviceactor

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/runeffect"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/model"
)

const (
	internalHandlerMaxResultBytes      = 64 * 1024
	internalHandlerMaxResultValueCount = 2000
	internalHandlerMaxResultValueDepth = 30
)

// RuntimeInternalDispatchTarget identifies one installation/handler route with
// ready AgentTeam-owned App action deliveries.
type RuntimeInternalDispatchTarget struct {
	WorkspaceID           string
	ServiceInstallationID string
	HandlerKey            string
	ServiceConnectionRef  string
}

// RuntimeInternalDispatchTargetList bounds a ready internal-handler scan.
type RuntimeInternalDispatchTargetList struct {
	WorkspaceGroupID string
	Now              time.Time
	Limit            int
}

// InternalHandlerDeliveryCompletion atomically records the effect and terminal
// ledger state produced by an in-process handler.
type InternalHandlerDeliveryCompletion struct {
	WorkspaceID string
	AttemptID   string
	HandlerKey  string
	Effect      runeffect.Effect
	ResultJSON  []byte
	Now         time.Time
}

// InternalHandlerDeliveryCompletionResult is the terminal internal-handler
// attempt, delivery, and effect projection.
type InternalHandlerDeliveryCompletionResult struct {
	Attempt  DeliveryAttempt
	Delivery EventDelivery
	Effect   DeliveryEffect
}

// ListRuntimeInternalDispatchTargets lists bounded internal-handler routes with
// ready App notification action deliveries.
func (store *Store) ListRuntimeInternalDispatchTargets(ctx context.Context, params RuntimeInternalDispatchTargetList) ([]RuntimeInternalDispatchTarget, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listRuntimeInternalDispatchTargets(ctx, queryer, params)
}

// CompleteInternalHandlerDelivery atomically appends the handler effect and
// acknowledges the current leased internal-handler attempt.
func (store *Store) CompleteInternalHandlerDelivery(ctx context.Context, params InternalHandlerDeliveryCompletion) (InternalHandlerDeliveryCompletionResult, error) {
	if store == nil || store.db == nil {
		return InternalHandlerDeliveryCompletionResult{}, fmt.Errorf("service actor store database is not configured")
	}
	var result InternalHandlerDeliveryCompletionResult
	if err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var err error
		result, err = completeInternalHandlerDelivery(ctx, tx, params)
		return err
	}); err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	return result, nil
}

// CompleteInternalHandlerDelivery atomically appends the handler effect and
// acknowledges the current leased attempt in an existing transaction. App
// domain adapters use this form to complete their action row in the same unit
// of work without introducing a serviceactor -> app dependency.
func (store *TxStore) CompleteInternalHandlerDelivery(ctx context.Context, params InternalHandlerDeliveryCompletion) (InternalHandlerDeliveryCompletionResult, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	return completeInternalHandlerDelivery(ctx, execer, params)
}

func listRuntimeInternalDispatchTargets(ctx context.Context, queryer coredb.Queryer, params RuntimeInternalDispatchTargetList) ([]RuntimeInternalDispatchTarget, error) {
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	limit := normalizeRuntimeInternalDispatchTargetLimit(params.Limit)
	rows, err := queryer.Query(ctx, `
SELECT
    delivery.workspace_id,
    delivery.service_installation_id,
    delivery.service_connection_ref,
    MIN(COALESCE(delivery.next_retry_at, delivery.created_at)) AS ready_at
FROM service_event_deliveries delivery
JOIN service_installations installation
  ON installation.id = delivery.service_installation_id
 AND installation.workspace_id = delivery.workspace_id
 AND installation.service_kind = delivery.service_kind
 AND installation.status = 'active'
JOIN service_event_subscriptions subscription
  ON subscription.id = delivery.service_event_subscription_id
 AND subscription.workspace_id = delivery.workspace_id
 AND subscription.status = 'active'
 AND subscription.removed_at IS NULL
JOIN service_target_bindings binding
  ON binding.id = delivery.service_target_binding_id
 AND binding.workspace_id = delivery.workspace_id
 AND binding.status = 'active'
 AND binding.removed_at IS NULL
WHERE delivery.service_kind = $1
  AND delivery.subscription_kind = $2
  AND delivery.delivery_policy = $3
  AND delivery.service_connection_ref LIKE $4
  AND delivery.status IN ('running', 'pending', 'retry_waiting')
  AND (delivery.next_retry_at IS NULL OR delivery.next_retry_at <= $5)
  AND ($6 = '' OR EXISTS (
      SELECT 1
      FROM workspace_placements placement
      WHERE placement.workspace_id = delivery.workspace_id
        AND placement.workspace_group_id = $6
        AND placement.state IN ('active', 'draining', 'moving')
  ))
  AND NOT EXISTS (
      SELECT 1
      FROM service_event_delivery_attempts attempt
      WHERE attempt.service_event_delivery_id = delivery.id
        AND attempt.status IN ('leased', 'delivered')
        AND (attempt.lease_expires_at IS NULL OR attempt.lease_expires_at > $5)
  )
  AND COALESCE(JSON_TYPE(JSON_EXTRACT(
      installation.metadata,
      '$.manifest.capabilities.runtime'
  )), '') = 'OBJECT'
  AND JSON_CONTAINS(
      COALESCE(JSON_EXTRACT(
          installation.metadata,
          '$.manifest.capabilities.runtime.events'
      ), JSON_ARRAY()),
      JSON_QUOTE('agentteam.app.notification.action')
  ) = 1
  AND EXISTS (
      SELECT 1
      FROM service_grants action_grant
      WHERE action_grant.service_installation_id = delivery.service_installation_id
        AND action_grant.scope = 'app.notification_actions.read'
        AND action_grant.revoked_at IS NULL
        AND (action_grant.expires_at IS NULL OR action_grant.expires_at > $5)
  )
GROUP BY delivery.workspace_id, delivery.service_installation_id, delivery.service_connection_ref
ORDER BY ready_at ASC, delivery.workspace_id ASC, delivery.service_installation_id ASC, delivery.service_connection_ref ASC
LIMIT $7
`, KindApp, SubscriptionKindAppNotificationAction, AppRuntimeDeliveryPolicyInternalHandler, AppRuntimeConnectionRefInternalPrefix+"%", now, trim(params.WorkspaceGroupID), limit)
	if err != nil {
		return nil, fmt.Errorf("list internal handler dispatch targets: %w", err)
	}
	defer rows.Close()
	targets := make([]RuntimeInternalDispatchTarget, 0, limit)
	for rows.Next() {
		var target RuntimeInternalDispatchTarget
		var readyAt time.Time
		if err := rows.Scan(&target.WorkspaceID, &target.ServiceInstallationID, &target.ServiceConnectionRef, &readyAt); err != nil {
			return nil, fmt.Errorf("scan internal handler dispatch target: %w", err)
		}
		target.WorkspaceID = trim(target.WorkspaceID)
		target.ServiceInstallationID = trim(target.ServiceInstallationID)
		target.ServiceConnectionRef = trim(target.ServiceConnectionRef)
		handlerKey, ok := InternalHandlerKeyFromConnectionRef(target.ServiceConnectionRef)
		if !ok {
			return nil, fmt.Errorf("internal handler dispatch target has an invalid connection ref")
		}
		target.HandlerKey = handlerKey
		targets = append(targets, target)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate internal handler dispatch targets: %w", err)
	}
	return targets, nil
}

func normalizeRuntimeInternalDispatchTargetLimit(limit int) int {
	if limit <= 0 {
		return 100
	}
	if limit > 500 {
		return 500
	}
	return limit
}

func completeInternalHandlerDelivery(ctx context.Context, execer coredb.QueryExecer, params InternalHandlerDeliveryCompletion) (InternalHandlerDeliveryCompletionResult, error) {
	workspaceID := trim(params.WorkspaceID)
	attemptID := trim(params.AttemptID)
	handlerRef, err := InternalHandlerConnectionRef(params.HandlerKey)
	if workspaceID == "" || attemptID == "" || err != nil {
		return InternalHandlerDeliveryCompletionResult{}, badRequest("internal handler completion workspace, attempt, and handler are required")
	}
	effect, err := params.Effect.Normalize()
	if err != nil || effect.Kind == runeffect.EffectKindNone || effect.Validate() != nil {
		return InternalHandlerDeliveryCompletionResult{}, badRequest("internal handler completion effect is invalid")
	}
	if err := validateInternalHandlerResultJSON(params.ResultJSON); err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	attempt, err := loadDeliveryAttemptForUpdate(ctx, execer, workspaceID, attemptID)
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	delivery, err := loadEventDeliveryForUpdate(ctx, execer, workspaceID, attempt.ServiceEventDeliveryID)
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	if err := validateInternalHandlerAttemptRoute(delivery, attempt, handlerRef, now); err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	if attempt.Status == DeliveryAttemptStatusAcked && delivery.Status == string(runeffect.StatusSucceeded) {
		effects, err := listDeliveryEffectsByDelivery(ctx, execer, DeliveryEffectListQuery{WorkspaceID: workspaceID, ServiceEventDeliveryID: delivery.ID, Limit: 100})
		if err != nil {
			return InternalHandlerDeliveryCompletionResult{}, err
		}
		for _, existing := range effects {
			if existing.DeliveryAttemptID == attempt.ID && existing.Status == DeliveryEffectStatusCreated {
				if existing.EffectKind != string(effect.Kind) || existing.EffectRef != effect.Ref || existing.MessageID != effect.MessageID || !internalHandlerJSONEqual(attempt.ResultJSON, params.ResultJSON) {
					return InternalHandlerDeliveryCompletionResult{}, conflict("internal handler completion replay does not match the terminal outcome")
				}
				return InternalHandlerDeliveryCompletionResult{Attempt: attempt, Delivery: delivery, Effect: existing}, nil
			}
		}
		return InternalHandlerDeliveryCompletionResult{}, conflict("internal handler delivery completed without an effect")
	}
	if attempt.Status == DeliveryAttemptStatusLeased {
		attempt, err = markDeliveryAttemptDelivered(ctx, execer, DeliveryAttemptDeliveredMark{WorkspaceID: workspaceID, AttemptID: attempt.ID, Now: now})
		if err != nil {
			return InternalHandlerDeliveryCompletionResult{}, err
		}
	}
	metadata, err := json.Marshal(map[string]any{"transport": RuntimeTransportKindInternalHandler, "handler_key": strings.TrimSpace(params.HandlerKey)})
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, fmt.Errorf("marshal internal handler effect metadata: %w", err)
	}
	appended, err := appendDeliveryEffect(ctx, execer, DeliveryEffectAppend{
		WorkspaceID: workspaceID, DeliveryAttemptID: attempt.ID,
		EffectKind: string(effect.Kind), EffectRef: effect.Ref, MessageID: effect.MessageID,
		TargetKind: delivery.TargetKind, TargetRef: delivery.TargetRef,
		Status: DeliveryEffectStatusCreated, Metadata: metadata, Now: now,
	})
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	attempt, err = setDeliveryAttemptStatus(ctx, execer, DeliveryAttemptStatusUpdate{
		WorkspaceID: workspaceID, ID: attempt.ID, Status: DeliveryAttemptStatusAcked,
		ResultJSON: params.ResultJSON, Now: now,
	})
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	if err := completeEventDelivery(ctx, execer, EventDeliveryCompletion{
		ID: delivery.ID, Status: runeffect.StatusSucceeded, Effect: effect, Now: now,
	}); err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	if err := markEventSubscriptionsDelivered(ctx, execer, []string{delivery.EventSubscriptionID}, now); err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	delivery, err = loadEventDeliveryForUpdate(ctx, execer, workspaceID, delivery.ID)
	if err != nil {
		return InternalHandlerDeliveryCompletionResult{}, err
	}
	return InternalHandlerDeliveryCompletionResult{Attempt: attempt, Delivery: delivery, Effect: appended}, nil
}

func internalHandlerJSONEqual(left []byte, right []byte) bool {
	return model.JSONDocumentsEqual(left, right)
}

func validateInternalHandlerResultJSON(payload []byte) error {
	if len(payload) == 0 {
		return nil
	}
	if len(payload) > internalHandlerMaxResultBytes {
		return badRequest("internal handler completion result is too large")
	}
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.UseNumber()
	var document any
	if err := decoder.Decode(&document); err != nil {
		return badRequest("internal handler completion result must be valid JSON")
	}
	var trailing any
	if err := decoder.Decode(&trailing); err != io.EOF {
		return badRequest("internal handler completion result must be one JSON document")
	}
	result, ok := document.(map[string]any)
	if !ok {
		return badRequest("internal handler completion result must be a JSON object")
	}
	valueCount := 0
	if err := validateInternalHandlerResultValue(result, 0, &valueCount); err != nil {
		return err
	}
	return nil
}

func validateInternalHandlerResultValue(value any, depth int, valueCount *int) error {
	if depth > internalHandlerMaxResultValueDepth {
		return badRequest("internal handler completion result is too deeply nested")
	}
	(*valueCount)++
	if *valueCount > internalHandlerMaxResultValueCount {
		return badRequest("internal handler completion result has too many values")
	}
	switch typed := value.(type) {
	case map[string]any:
		for key, child := range typed {
			if strings.TrimSpace(key) == "" || len(key) > 255 {
				return badRequest("internal handler completion result contains an invalid key")
			}
			if err := validateInternalHandlerResultValue(child, depth+1, valueCount); err != nil {
				return err
			}
		}
	case []any:
		for _, child := range typed {
			if err := validateInternalHandlerResultValue(child, depth+1, valueCount); err != nil {
				return err
			}
		}
	}
	return nil
}

func validateInternalHandlerAttemptRoute(delivery EventDelivery, attempt DeliveryAttempt, handlerRef string, now time.Time) error {
	if attempt.TransportKind != RuntimeTransportKindInternalHandler || attempt.ServiceRuntimeCredentialID != "" || attempt.ServiceRuntimeSessionID != "" || attempt.DeliveryTokenHash != "" || attempt.DeliveryTokenExpiresAt != nil || attempt.CallbackURLSnapshot != "" {
		return badRequest("delivery attempt is not an internal handler attempt")
	}
	if delivery.DeliveryPolicy != AppRuntimeDeliveryPolicyInternalHandler || delivery.ServiceConnectionRef != handlerRef || delivery.SubscriptionKind != SubscriptionKindAppNotificationAction || delivery.ServiceKind != KindApp {
		return badRequest("delivery is not routed to this internal handler")
	}
	if delivery.ID != attempt.ServiceEventDeliveryID || delivery.WorkspaceID != attempt.WorkspaceID || delivery.ServiceInstallationID != attempt.ServiceInstallationID || delivery.ServiceMemberID != attempt.ServiceMemberID {
		return badRequest("internal handler attempt does not match its delivery")
	}
	if attempt.Status == DeliveryAttemptStatusAcked && delivery.Status == string(runeffect.StatusSucceeded) {
		return nil
	}
	if attempt.Status != DeliveryAttemptStatusLeased && attempt.Status != DeliveryAttemptStatusDelivered {
		return conflict("internal handler delivery attempt is not active")
	}
	if attempt.LeaseExpiresAt == nil || !attempt.LeaseExpiresAt.After(now) || !deliveryLeaseMatchesAttempt(delivery, attempt) {
		return conflict("internal handler delivery attempt lease is no longer current")
	}
	return nil
}