package packs

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

	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

// connectorCalloutProfilePackBatch carries the set of ProfileName values declared
// by a single pack in the current manifest. It is populated during source-backed
// pack parsing and passed to the sweeper so the apply-side sweep and verify-side
// assertion share the same declared set.
type connectorCalloutProfilePackBatch struct {
	// declaredProfileNames is the set of ProfileName values declared by the pack.
	declaredProfileNames []string
}

// newConnectorCalloutProfilePackSweeperComponent returns a synthetic pack-level
// install step that garbage-collects _System_ConnectorCalloutProfile rows owned
// by packName whose ProfileName is no longer declared in the current manifest.
func newConnectorCalloutProfilePackSweeperComponent(packName string, batch *connectorCalloutProfilePackBatch) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindConnectorCalloutProfile,
		BusinessKey:   JoinComponentBusinessKey(packName, "@sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyConnectorCalloutProfilePackSweeper(ctx, deps.DB, batch, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyConnectorCalloutProfilePackSweeper(ctx, deps.DB, batch, packName)
		},
	}
}

// applyConnectorCalloutProfilePackSweeper soft-deletes _System_ConnectorCalloutProfile
// rows owned by ownerPack whose ProfileName is not in the declared set.
func applyConnectorCalloutProfilePackSweeper(ctx context.Context, db *sql.DB, batch *connectorCalloutProfilePackBatch, ownerPack string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}

	orphans, err := queryOrphanConnectorCalloutProfileRows(ctx, db, batch, ownerPack)
	if err != nil {
		return err
	}
	if len(orphans) == 0 {
		return nil
	}

	for _, orphanID := range orphans {
		if _, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorCalloutProfile
SET IsDeleted = 1, LastModifiedDate = NOW()
WHERE Id = ? AND COALESCE(IsDeleted, 0) = 0`, orphanID); err != nil {
			return fmt.Errorf("soft-delete orphan connector callout profile %s for pack %s: %w", orphanID, ownerPack, err)
		}
	}
	return nil
}

// verifyConnectorCalloutProfilePackSweeper asserts that no live
// _System_ConnectorCalloutProfile row exists for ownerPack whose ProfileName
// is not in the declared set.
func verifyConnectorCalloutProfilePackSweeper(ctx context.Context, db *sql.DB, batch *connectorCalloutProfilePackBatch, ownerPack string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}

	orphans, err := queryOrphanConnectorCalloutProfileRows(ctx, db, batch, ownerPack)
	if err != nil {
		return err
	}
	if len(orphans) == 0 {
		return nil
	}
	sort.Strings(orphans)
	return fmt.Errorf("pack %s has %d undeclared connector_callout_profile row(s) still live: %s",
		ownerPack, len(orphans), strings.Join(orphans, ", "))
}

// queryOrphanConnectorCalloutProfileRows returns the Ids of
// _System_ConnectorCalloutProfile rows owned by ownerPack that are live
// (IsDeleted=0) but whose ProfileName is NOT in the declared set.
func queryOrphanConnectorCalloutProfileRows(ctx context.Context, db *sql.DB, batch *connectorCalloutProfilePackBatch, ownerPack string) ([]string, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	declaredSet := make(map[string]struct{})
	if batch != nil {
		for _, pn := range batch.declaredProfileNames {
			declaredSet[strings.ToLower(strings.TrimSpace(pn))] = struct{}{}
		}
	}

	query := `
SELECT Id, ProfileName
FROM _System_ConnectorCalloutProfile
WHERE OwnerPack = ? AND COALESCE(IsDeleted, 0) = 0`
	args := []any{ownerPack}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, ownerPack); ok {
		query += " AND InstalledPackId = ?"
		args = append(args, candidateInstallID)
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("query live connector_callout_profile rows for pack sweep %s: %w", ownerPack, err)
	}
	defer rows.Close()

	var orphanIDs []string
	for rows.Next() {
		var id, profileName string
		if err := rows.Scan(&id, &profileName); err != nil {
			return nil, fmt.Errorf("scan connector_callout_profile row for pack sweep %s: %w", ownerPack, err)
		}
		if _, ok := declaredSet[strings.ToLower(strings.TrimSpace(profileName))]; ok {
			continue
		}
		orphanIDs = append(orphanIDs, id)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate connector_callout_profile rows for pack sweep %s: %w", ownerPack, err)
	}
	return orphanIDs, nil
}