package packs

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

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

const connectorInboundHandlerConfigOrphanSweeperGroupKey = "connector_inbound_handler_config_orphan"

type connectorInboundHandlerConfigBatch struct {
	declaredCapabilityKinds []string
}

func newConnectorInboundHandlerConfigPackSweeperComponent(packName string, declaredCapabilityKinds []string) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindConnectorInboundHandlerConfig,
		BusinessKey:   JoinComponentBusinessKey(packName, "@handler-config-sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		applyGroupKey:     connectorInboundHandlerConfigOrphanSweeperGroupKey,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyConnectorInboundHandlerConfigPackSweep(ctx, deps.DB, connectorcapabilities.Default(), packName, declaredCapabilityKinds)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyConnectorInboundHandlerConfigPackSweep(ctx, deps.DB, packName, declaredCapabilityKinds)
		},
	}
}

func applyConnectorInboundHandlerConfigPackSweep(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry, packName string, declaredCapabilityKinds []string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	declaredSet := connectorInboundHandlerConfigKindSet(declaredCapabilityKinds)
	query := `
SELECT Id, CapabilityKind
  FROM _System_ConnectorInboundHandlerConfig
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0`
	args := []any{packName}
	_, stoppedCandidate := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName)
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		query += " AND InstalledPackId = ?"
		args = append(args, candidateInstallID)
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("query connector inbound handler configs for sweep apply (pack %s): %w", packName, err)
	}
	defer rows.Close()

	type candidate struct {
		id   string
		kind string
	}
	var candidates []candidate
	for rows.Next() {
		var c candidate
		if err := rows.Scan(&c.id, &c.kind); err != nil {
			return fmt.Errorf("scan connector inbound handler config row for sweep apply: %w", err)
		}
		if _, ok := declaredSet[strings.TrimSpace(c.kind)]; ok {
			continue
		}
		candidates = append(candidates, c)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate connector inbound handler config rows for sweep apply: %w", err)
	}
	if len(candidates) == 0 {
		return nil
	}
	sort.Slice(candidates, func(i, j int) bool { return candidates[i].id < candidates[j].id })
	for _, candidate := range candidates {
		if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundHandlerConfig
   SET IsDeleted = 1, LastModifiedDate = NOW()
 WHERE Id = ?
   AND COALESCE(IsDeleted, 0) = 0`, candidate.id); err != nil {
			return fmt.Errorf("soft-delete connector inbound handler config %s for pack %s: %w", candidate.id, packName, err)
		}
		if registry != nil && !stoppedCandidate {
			registry.UnregisterCapability(packName, connectorcapabilities.CapabilityKind(candidate.kind))
		}
	}
	return nil
}

func verifyConnectorInboundHandlerConfigPackSweep(ctx context.Context, db *sql.DB, packName string, declaredCapabilityKinds []string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	declaredSet := connectorInboundHandlerConfigKindSet(declaredCapabilityKinds)
	query := `
SELECT CapabilityKind
  FROM _System_ConnectorInboundHandlerConfig
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0`
	args := []any{packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		query += " AND InstalledPackId = ?"
		args = append(args, candidateInstallID)
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("query connector inbound handler configs for sweep verify (pack %s): %w", packName, err)
	}
	defer rows.Close()

	var stale []string
	for rows.Next() {
		var kind string
		if err := rows.Scan(&kind); err != nil {
			return fmt.Errorf("scan connector inbound handler config row for sweep verify: %w", err)
		}
		if _, ok := declaredSet[strings.TrimSpace(kind)]; !ok {
			stale = append(stale, kind)
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate connector inbound handler config rows for sweep verify: %w", err)
	}
	if len(stale) > 0 {
		sort.Strings(stale)
		return fmt.Errorf("connector inbound handler config rows not declared by pack %s remain live: %s", packName, strings.Join(stale, ", "))
	}
	return nil
}

func connectorInboundHandlerConfigKindSet(kinds []string) map[string]struct{} {
	out := make(map[string]struct{}, len(kinds))
	for _, kind := range kinds {
		kind = strings.TrimSpace(kind)
		if kind != "" {
			out[kind] = struct{}{}
		}
	}
	return out
}