package packs

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

	"github.com/nexuscrm/automation/triggersub"
	"github.com/nexuscrm/db"
	"github.com/nexuscrm/metadata"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

// ApplyStoppedSourceSetFlowRuntimeMaintenance publishes the trigger-subscription
// shadow rows for only the standard flows authored by the pack being applied.
// The ordinary install path keeps its full-database orphan repair and backfill;
// stopped source-set execution deliberately avoids making unrelated historical flows
// an admission gate.
func ApplyStoppedSourceSetFlowRuntimeMaintenance(
	ctx context.Context,
	authority offlinecandidateauthority.StoppedSourceSetAuthority,
	db *database.TiDBConnection,
	metaSvc *metadata.MetadataService,
	pack *SourceBackedPack,
) error {
	if !authority.Valid() {
		return fmt.Errorf("stopped source-set flow runtime authority is required")
	}
	if pack == nil {
		return fmt.Errorf("pack is required for stopped source-set flow runtime maintenance")
	}
	if !offlinecandidateauthority.StoppedSourceSetApplyOnly(ctx, pack.Name()) {
		return fmt.Errorf("stopped source-set flow runtime scope is required for pack %s", pack.Name())
	}
	if db == nil || db.DB() == nil {
		return fmt.Errorf("database connection is required for stopped source-set flow runtime maintenance")
	}
	if metaSvc == nil {
		return fmt.Errorf("metadata service is required for stopped source-set flow runtime maintenance")
	}
	for _, developerName := range pack.StandardFlowDeveloperNames() {
		if err := reconcileStoppedSourceSetFlowDefinition(ctx, db.DB(), developerName); err != nil {
			return fmt.Errorf(
				"reconcile trigger subscriptions for standard flow %s: %w",
				developerName,
				err,
			)
		}
	}
	metaSvc.InvalidateCache()
	return nil
}

func reconcileStoppedSourceSetFlowDefinition(ctx context.Context, rawDB *sql.DB, developerName string) error {
	developerName = strings.TrimSpace(developerName)
	if developerName == "" {
		return fmt.Errorf("standard flow DeveloperName is required")
	}

	var definitionID string
	err := rawDB.QueryRowContext(ctx, `
SELECT Id
FROM FlowDefinition
WHERE DeveloperName = ?
LIMIT 1`, developerName).Scan(&definitionID)
	if err == sql.ErrNoRows {
		return fmt.Errorf("FlowDefinition row is missing")
	}
	if err != nil {
		return fmt.Errorf("resolve FlowDefinition row: %w", err)
	}
	definitionID = strings.TrimSpace(definitionID)
	if definitionID == "" {
		return fmt.Errorf("FlowDefinition row has no ID")
	}

	return triggersub.ReconcileDefinition(ctx, rawDB, definitionID)
}

func reconcileFlowOrphansBeforeFlowRuntimeMaintenance(ctx context.Context, db *database.TiDBConnection, scope string) error {
	if db == nil {
		return fmt.Errorf("database connection is required for flow runtime maintenance")
	}

	touched, err := ReconcileFlowOrphans(ctx, db)
	if err != nil {
		return fmt.Errorf("reconcile flow orphans before %s flow runtime maintenance: %w", scope, err)
	}
	if touched > 0 {
		log.Printf("%s: auto-reconciled %d orphan FlowDefinition ActiveVersionId row(s) before flow runtime maintenance", scope, touched)
	}
	return nil
}

// RunFlowRuntimeMaintenanceAfterOrphanRepair normalizes broken
// FlowDefinition.ActiveVersionId pointers before rebuilding flow runtime state.
// Every entrypoint that bulk-reconciles FlowDefinition/Flow rows must use this
// helper so trigger subscription backfill never fails closed on historical
// orphan rows.
func RunFlowRuntimeMaintenanceAfterOrphanRepair(ctx context.Context, db *database.TiDBConnection, metaSvc *metadata.MetadataService, scope string) error {
	if metaSvc == nil {
		return fmt.Errorf("metadata service is required for flow runtime maintenance")
	}
	if err := reconcileFlowOrphansBeforeFlowRuntimeMaintenance(ctx, db, scope); err != nil {
		return err
	}
	if err := ApplyFlowRuntimeMaintenance(ctx, metaSvc); err != nil {
		return fmt.Errorf("flow runtime maintenance: %w", err)
	}
	return nil
}

// ReconcileFlowOrphans repairs FlowDefinition rows whose ActiveVersionId fails
// any of the integrity properties enforced by Flow verify/runtime maintenance.
func ReconcileFlowOrphans(ctx context.Context, db *database.TiDBConnection) (touched int, err error) {
	if db == nil {
		return 0, fmt.Errorf("database connection is required for reconcile flow orphans")
	}
	return ReconcileFlowOrphansRaw(ctx, db.DB())
}

// ReconcileFlowOrphansRaw is the *sql.DB variant used by tests.
func ReconcileFlowOrphansRaw(ctx context.Context, rawDB *sql.DB) (int, error) {
	const selectQuery = `
SELECT fd.Id
FROM FlowDefinition fd
LEFT JOIN Flow fv ON fv.Id = fd.ActiveVersionId
  AND fv.DefinitionId = fd.Id
  AND (fv.IsDeleted = FALSE OR fv.IsDeleted IS NULL)
  AND fv.Status = 'Active'
WHERE fd.ActiveVersionId IS NOT NULL
  AND (fd.IsDeleted = TRUE OR fv.Id IS NULL)`

	rows, err := rawDB.QueryContext(ctx, selectQuery)
	if err != nil {
		return 0, fmt.Errorf("reconcile flow orphans: query broken refs: %w", err)
	}
	defer rows.Close()

	var candidateIDs []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return 0, fmt.Errorf("reconcile flow orphans: scan row: %w", err)
		}
		candidateIDs = append(candidateIDs, id)
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("reconcile flow orphans: iterate rows: %w", err)
	}

	if len(candidateIDs) == 0 {
		log.Println("reconcile flow orphans: no broken ActiveVersionId refs found - nothing to repair")
		return 0, nil
	}

	const updateQuery = `
UPDATE FlowDefinition fd
SET    fd.ActiveVersionId = NULL
WHERE  fd.ActiveVersionId IS NOT NULL
  AND (
        fd.IsDeleted = TRUE
     OR NOT EXISTS (
          SELECT 1 FROM Flow fv
          WHERE fv.Id = fd.ActiveVersionId
            AND fv.DefinitionId = fd.Id
            AND (fv.IsDeleted = FALSE OR fv.IsDeleted IS NULL)
            AND fv.Status = 'Active'
        )
      )`

	result, err := rawDB.ExecContext(ctx, updateQuery)
	if err != nil {
		return 0, fmt.Errorf("reconcile flow orphans: update broken refs: %w", err)
	}

	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("reconcile flow orphans: rows affected: %w", err)
	}

	log.Printf("reconcile flow orphans: observed %d candidate(s), actually repaired %d row(s): candidate_ids=[%s]",
		len(candidateIDs), rowsAffected, strings.Join(candidateIDs, ", "))

	return int(rowsAffected), nil
}