package serviceactor

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

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

// RuntimeCallbackConfig is the installed-version callback transport metadata
// used by App runtime callback delivery.
type RuntimeCallbackConfig struct {
	WorkspaceID           string
	ServiceDefinitionID   string
	ServiceInstallationID string
	ServiceMemberID       string
	ServiceKind           string
	InstalledVersion      string
	CallbackURL           string
	CallbackSigningSecret string
	AppRuntimeTransports  []string
}

// RuntimeCallbackDispatchTarget identifies an installation with ready callback
// runtime deliveries. The server callback dispatcher uses these targets to
// schedule bounded callback dispatch passes without scanning every installed
// service.
type RuntimeCallbackDispatchTarget struct {
	WorkspaceID           string
	ServiceInstallationID string
}

// RuntimeCallbackDispatchTargetList selects ready callback delivery targets.
type RuntimeCallbackDispatchTargetList struct {
	WorkspaceGroupID string
	ServiceKind      string
	SubscriptionKind string
	Now              time.Time
	Limit            int
}

// LoadRuntimeCallbackConfig loads callback transport metadata for an active
// App service installation from canonical service actor ledgers.
func (store *Store) LoadRuntimeCallbackConfig(ctx context.Context, workspaceID string, serviceInstallationID string) (RuntimeCallbackConfig, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return RuntimeCallbackConfig{}, err
	}
	return loadRuntimeCallbackConfig(ctx, queryer, workspaceID, serviceInstallationID)
}

// LoadRuntimeCallbackConfig loads callback transport metadata in a transaction.
func (store *TxStore) LoadRuntimeCallbackConfig(ctx context.Context, workspaceID string, serviceInstallationID string) (RuntimeCallbackConfig, error) {
	queryer, err := execerFromTxStore(store)
	if err != nil {
		return RuntimeCallbackConfig{}, err
	}
	return loadRuntimeCallbackConfig(ctx, queryer, workspaceID, serviceInstallationID)
}

// ListRuntimeCallbackDispatchTargets lists active installations that have ready
// App callback deliveries.
func (store *Store) ListRuntimeCallbackDispatchTargets(ctx context.Context, params RuntimeCallbackDispatchTargetList) ([]RuntimeCallbackDispatchTarget, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listRuntimeCallbackDispatchTargets(ctx, queryer, params)
}

// ListRuntimeCallbackDispatchTargets lists active installations that have ready
// App callback deliveries in a transaction.
func (store *TxStore) ListRuntimeCallbackDispatchTargets(ctx context.Context, params RuntimeCallbackDispatchTargetList) ([]RuntimeCallbackDispatchTarget, error) {
	queryer, err := execerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return listRuntimeCallbackDispatchTargets(ctx, queryer, params)
}

func loadRuntimeCallbackConfig(ctx context.Context, queryer coredb.QueryRower, workspaceID string, serviceInstallationID string) (RuntimeCallbackConfig, error) {
	trimmedWorkspaceID := trim(workspaceID)
	trimmedInstallationID := trim(serviceInstallationID)
	if trimmedWorkspaceID == "" || trimmedInstallationID == "" {
		return RuntimeCallbackConfig{}, badRequest("runtime callback workspace and installation are required")
	}
	var config RuntimeCallbackConfig
	var callbackURL sql.NullString
	var callbackSigningSecret sql.NullString
	var runtimeCapabilityConfig []byte
	if err := queryer.QueryRow(ctx, `
SELECT
    installation.workspace_id,
    installation.service_definition_id,
    installation.id,
    service_member.id,
    installation.service_kind,
    installation.installed_version,
    COALESCE(JSON_UNQUOTE(JSON_EXTRACT(installation.metadata, '$.callbackURL')), ''),
    COALESCE(JSON_UNQUOTE(JSON_EXTRACT(definition.config, '$.callbackSigningSecret')), ''),
    JSON_EXTRACT(installation.metadata, '$.manifest.capabilities.runtime')
FROM service_installations installation
JOIN service_definitions definition
  ON definition.id = installation.service_definition_id
 AND definition.service_kind = installation.service_kind
JOIN service_members service_member
  ON service_member.service_installation_id = installation.id
 AND service_member.service_kind = installation.service_kind
WHERE installation.workspace_id = $1
  AND installation.id = $2
  AND installation.status = 'active'
  AND service_member.status = 'active'
  AND COALESCE(JSON_TYPE(JSON_EXTRACT(
      installation.metadata,
      '$.manifest.capabilities.runtime'
  )), '') = 'OBJECT'
LIMIT 1
`, trimmedWorkspaceID, trimmedInstallationID).Scan(
		&config.WorkspaceID,
		&config.ServiceDefinitionID,
		&config.ServiceInstallationID,
		&config.ServiceMemberID,
		&config.ServiceKind,
		&config.InstalledVersion,
		&callbackURL,
		&callbackSigningSecret,
		&runtimeCapabilityConfig,
	); err != nil {
		if coredb.IsNotFound(err) {
			return RuntimeCallbackConfig{}, notFound("app runtime callback installation not found")
		}
		return RuntimeCallbackConfig{}, fmt.Errorf("load app runtime callback config: %w", err)
	}
	config.WorkspaceID = trim(config.WorkspaceID)
	config.ServiceDefinitionID = trim(config.ServiceDefinitionID)
	config.ServiceInstallationID = trim(config.ServiceInstallationID)
	config.ServiceMemberID = trim(config.ServiceMemberID)
	config.ServiceKind = trim(config.ServiceKind)
	config.InstalledVersion = trim(config.InstalledVersion)
	if callbackURL.Valid {
		config.CallbackURL = trim(callbackURL.String)
	}
	if callbackSigningSecret.Valid {
		config.CallbackSigningSecret = trim(callbackSigningSecret.String)
	}
	transports, err := decodeAppRuntimeTransports(runtimeCapabilityConfig)
	if err != nil {
		return RuntimeCallbackConfig{}, err
	}
	if !runtimeTransportListContains(transports, RuntimeTransportKindCallback) {
		return RuntimeCallbackConfig{}, badRequest("app callback transport is not enabled")
	}
	config.AppRuntimeTransports = transports
	return config, nil
}

func listRuntimeCallbackDispatchTargets(ctx context.Context, queryer coredb.Queryer, params RuntimeCallbackDispatchTargetList) ([]RuntimeCallbackDispatchTarget, error) {
	serviceKind := trim(params.ServiceKind)
	subscriptionKind := trim(params.SubscriptionKind)
	if serviceKind == "" || subscriptionKind == "" {
		return nil, badRequest("runtime callback dispatch target service and subscription kind are required")
	}
	if serviceKind == KindApp && !IsAppRuntimeSubscriptionKind(subscriptionKind) {
		return nil, badRequest("runtime callback subscription kind is not supported")
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	limit := normalizeRuntimeCallbackDispatchTargetLimit(params.Limit)
	rows, err := queryer.Query(ctx, `
SELECT
    delivery.workspace_id,
    delivery.service_installation_id,
    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_definitions definition
  ON definition.id = delivery.service_definition_id
 AND definition.service_kind = delivery.service_kind
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 = $4
  AND delivery.service_connection_ref = $5
  AND delivery.status IN ('running', 'pending', 'retry_waiting')
  AND (delivery.next_retry_at IS NULL OR delivery.next_retry_at <= $3)
  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 > $3)
  )
  AND COALESCE(JSON_TYPE(JSON_EXTRACT(
      installation.metadata,
      '$.manifest.capabilities.runtime'
  )), '') = 'OBJECT'
  AND (
      (
          delivery.subscription_kind = 'app_mention'
          AND JSON_CONTAINS(
              COALESCE(JSON_EXTRACT(
                  installation.metadata,
                  '$.manifest.capabilities.runtime.events'
              ), JSON_ARRAY()),
              JSON_QUOTE('agentteam.message.mention')
          ) = 1
          AND EXISTS (
              SELECT 1
              FROM service_grants read_grant
              WHERE read_grant.service_installation_id = delivery.service_installation_id
                AND read_grant.scope = 'app.mentions.read'
                AND read_grant.revoked_at IS NULL
                AND (read_grant.expires_at IS NULL OR read_grant.expires_at > $3)
          )
          AND EXISTS (
              SELECT 1
              FROM service_grants write_grant
              WHERE write_grant.service_installation_id = delivery.service_installation_id
                AND write_grant.scope = 'messages.write'
                AND write_grant.revoked_at IS NULL
                AND (write_grant.expires_at IS NULL OR write_grant.expires_at > $3)
          )
      )
      OR
      (
          delivery.subscription_kind = 'app_notification_action'
          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 > $3)
          )
      )
  )
GROUP BY delivery.workspace_id, delivery.service_installation_id
ORDER BY ready_at ASC, delivery.workspace_id ASC, delivery.service_installation_id ASC
LIMIT $7
`, serviceKind, subscriptionKind, now, AppRuntimeDeliveryPolicyExternal, AppRuntimeConnectionRefExternal, trim(params.WorkspaceGroupID), limit)
	if err != nil {
		return nil, fmt.Errorf("list runtime callback dispatch targets: %w", err)
	}
	candidates := make([]RuntimeCallbackDispatchTarget, 0, limit)
	for rows.Next() {
		var target RuntimeCallbackDispatchTarget
		var readyAt time.Time
		if err := rows.Scan(&target.WorkspaceID, &target.ServiceInstallationID, &readyAt); err != nil {
			rows.Close()
			return nil, fmt.Errorf("scan runtime callback dispatch target: %w", err)
		}
		target.WorkspaceID = trim(target.WorkspaceID)
		target.ServiceInstallationID = trim(target.ServiceInstallationID)
		if target.WorkspaceID != "" && target.ServiceInstallationID != "" {
			candidates = append(candidates, target)
		}
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return nil, fmt.Errorf("iterate runtime callback dispatch targets: %w", err)
	}
	rows.Close()

	// MySQL/TiDB connections cannot execute the per-installation capability
	// lookup while the target rows are still open on the same transaction.
	targets := make([]RuntimeCallbackDispatchTarget, 0, len(candidates))
	for _, target := range candidates {
		config, err := loadRuntimeCallbackConfig(ctx, queryer, target.WorkspaceID, target.ServiceInstallationID)
		if err != nil || !runtimeTransportListContains(config.AppRuntimeTransports, RuntimeTransportKindCallback) {
			continue
		}
		targets = append(targets, target)
	}
	return targets, nil
}

func normalizeRuntimeCallbackDispatchTargetLimit(limit int) int {
	if limit <= 0 {
		return runtimeListLimitDefault
	}
	if limit > runtimeListLimitMax {
		return runtimeListLimitMax
	}
	return limit
}

func decodeAppRuntimeTransports(payload []byte) ([]string, error) {
	if len(payload) == 0 {
		return nil, badRequest("app runtime capability config is required")
	}
	var config model.AppRuntimeCapabilityConfigJSON
	if err := model.UnmarshalJSONObject("app runtime capability config", payload, &config); err != nil {
		return nil, err
	}
	return uniqueRuntimeTransports(config.Runtime.Transports), nil
}

func uniqueRuntimeTransports(values []string) []string {
	seen := map[string]struct{}{}
	out := make([]string, 0, len(values))
	for _, value := range values {
		transport := strings.TrimSpace(value)
		if transport == "" {
			continue
		}
		if _, ok := seen[transport]; ok {
			continue
		}
		seen[transport] = struct{}{}
		out = append(out, transport)
	}
	return out
}

func runtimeTransportListContains(values []string, want string) bool {
	want = strings.TrimSpace(want)
	for _, value := range values {
		if strings.TrimSpace(value) == want {
			return true
		}
	}
	return false
}