package controlplane

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

	"github.com/nexuscrm/db"
)

// VerifyFlowDataIntegrity checks platform-substrate flow table consistency.
// It is called from verifyImpl and is fail-closed: any detected corruption
// returns a non-nil error so startup and `cmd/setup verify` stop until an
// operator runs setup apply or a narrower repair.
//
// Current checks:
//   - checkOrphanFlowVersions: Flow rows whose DefinitionId has no matching
//     FlowDefinition are broken state (no legitimate write path produces them).
//   - checkActiveVersionIdRefs: FlowDefinition rows whose ActiveVersionId is
//     non-null but references a missing Flow row indicate a dangling pointer
//     (MDPAI #2244 equivalent — detected at boot-time gate).
//   - checkDeletedActiveCollision: FlowDefinition rows that are soft-deleted
//     (IsDeleted = TRUE) but still carry a non-null ActiveVersionId represent
//     an impossible state (delete must clear the active pointer first).
//   - checkFlowVersionJSONValid: Flow rows whose Metadata column contains
//     non-null, non-empty content that is not valid JSON — indicates write-path
//     corruption or truncation that would cause runtime parse failures.
func VerifyFlowDataIntegrity(ctx context.Context, db *database.TiDBConnection) error {
	if err := checkOrphanFlowVersions(ctx, db.DB()); err != nil {
		return fmt.Errorf("flow integrity: %w", err)
	}
	if err := checkActiveVersionIdRefs(ctx, db.DB()); err != nil {
		return fmt.Errorf("flow integrity: %w", err)
	}
	if err := checkDeletedActiveCollision(ctx, db.DB()); err != nil {
		return fmt.Errorf("flow integrity: %w", err)
	}
	if err := checkFlowVersionJSONValid(ctx, db.DB()); err != nil {
		return fmt.Errorf("flow integrity: %w", err)
	}
	return nil
}

// checkOrphanFlowVersions detects Flow rows (FlowVersion records) whose
// DefinitionId references a FlowDefinition that no longer exists.  Such rows
// are unreachable through normal read paths and indicate a write-ordering bug
// or incomplete delete.
//
// Only non-deleted Flow rows are checked (IsDeleted = 0 or NULL); soft-deleted
// versions may legitimately outlive a hard-deleted definition row.
func checkOrphanFlowVersions(ctx context.Context, rawDB *sql.DB) error {
	const query = `
SELECT fv.Id
FROM Flow fv
LEFT JOIN FlowDefinition fd ON fv.DefinitionId = fd.Id
WHERE (fv.IsDeleted = 0 OR fv.IsDeleted IS NULL)
  AND fd.Id IS NULL
LIMIT 11`

	rows, err := rawDB.QueryContext(ctx, query)
	if err != nil {
		return fmt.Errorf("query orphan FlowVersion rows: %w", err)
	}
	defer rows.Close()

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

	if len(ids) == 0 {
		log.Println("✅ no orphan Flow (FlowVersion) rows")
		return nil
	}

	// Cap the display list at 10; the query fetches 11 to detect overflow.
	truncated := len(ids) > 10
	display := ids
	if truncated {
		display = ids[:10]
	}
	suffix := ""
	if truncated {
		suffix = " (truncated)"
	}
	return fmt.Errorf("orphan FlowVersion: %d row(s) detected%s — ids: [%s] — run 'cmd/setup apply' to reconcile",
		len(display), suffix, strings.Join(display, ", "))
}

// checkActiveVersionIdRefs detects FlowDefinition rows whose ActiveVersionId
// fails any of the four integrity properties: existence, same-definition
// ownership, not soft-deleted, Active status.
//
// A FlowDefinition's active pointer is only safe to execute when ALL of:
//  1. The referenced Flow row exists.
//  2. The Flow belongs to the same FlowDefinition: fv.DefinitionId = fd.Id.
//  3. The Flow is not soft-deleted: fv.IsDeleted = FALSE OR IS NULL.
//  4. The Flow is in Active state: fv.Status = 'Active'.
//
// The predicates are placed in the LEFT JOIN ON clause (not WHERE) so that
// fv.Id IS NULL correctly captures rows where the Flow fails any condition —
// a WHERE placement would suppress rows where fv.Id is non-null but the other
// conditions are violated.
func checkActiveVersionIdRefs(ctx context.Context, rawDB *sql.DB) error {
	const query = `
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 fv.Id IS NULL
LIMIT 11`

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

	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return fmt.Errorf("scan broken ActiveVersionId ref row: %w", err)
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate broken ActiveVersionId ref rows: %w", err)
	}

	if len(ids) == 0 {
		log.Println("✅ no broken ActiveVersionId refs")
		return nil
	}

	// Cap the display list at 10; the query fetches 11 to detect overflow.
	truncated := len(ids) > 10
	display := ids
	if truncated {
		display = ids[:10]
	}
	suffix := ""
	if truncated {
		suffix = " (truncated)"
	}
	return fmt.Errorf("broken ActiveVersionId ref: %d FlowDefinition row(s) detected%s — ids: [%s] — run 'cmd/setup verify --reconcile-flow-orphans' to repair, or re-run 'cmd/setup apply' (it auto-reconciles)",
		len(display), suffix, strings.Join(display, ", "))
}

// checkDeletedActiveCollision detects FlowDefinition rows that are both
// soft-deleted (IsDeleted = TRUE) and still carry a non-null ActiveVersionId.
// This is an impossible state: the delete path must clear ActiveVersionId first.
// Leaving such rows in place can cause scheduler and trigger code to resurrect
// a deleted flow if the IsDeleted flag is later toggled back without re-setting
// the active pointer through the normal activate path.
func checkDeletedActiveCollision(ctx context.Context, rawDB *sql.DB) error {
	const query = `
SELECT Id
FROM FlowDefinition
WHERE IsDeleted = TRUE
  AND ActiveVersionId IS NOT NULL
LIMIT 11`

	rows, err := rawDB.QueryContext(ctx, query)
	if err != nil {
		return fmt.Errorf("query deleted+active collision rows: %w", err)
	}
	defer rows.Close()

	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return fmt.Errorf("scan deleted+active collision row: %w", err)
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate deleted+active collision rows: %w", err)
	}

	if len(ids) == 0 {
		log.Println("✅ no deleted+active collision rows")
		return nil
	}

	// Cap the display list at 10; the query fetches 11 to detect overflow.
	truncated := len(ids) > 10
	display := ids
	if truncated {
		display = ids[:10]
	}
	suffix := ""
	if truncated {
		suffix = " (truncated)"
	}
	return fmt.Errorf("deleted+active collision: %d FlowDefinition row(s) detected%s — ids: [%s] — run 'cmd/setup verify --reconcile-flow-orphans' to repair, or re-run 'cmd/setup apply' (it auto-reconciles)",
		len(display), suffix, strings.Join(display, ", "))
}

// checkFlowVersionJSONValid inspects the Metadata column of every Flow row
// (FlowVersion record) that contains a non-null, non-empty value and reports
// any row whose content is not valid JSON.  Such rows indicate write-path
// corruption or truncation that would cause runtime parse failures when the
// executor calls json.Unmarshal on the stored metadata.
//
// The check is generic: it does not interpret field names or CRM-specific
// structure — it only validates JSON syntax.
func checkFlowVersionJSONValid(ctx context.Context, rawDB *sql.DB) error {
	// No LIMIT: the query is column-projected (Id + Metadata only) and
	// indexed; scanning all rows is cheap even at 10 K+ FlowVersion rows.
	// A silent LIMIT would allow invalid JSON beyond the cap to go undetected
	// on production databases.
	const query = `
SELECT Id, Metadata
FROM Flow
WHERE Metadata IS NOT NULL
  AND Metadata != ''`

	rows, err := rawDB.QueryContext(ctx, query)
	if err != nil {
		return fmt.Errorf("query Flow metadata rows: %w", err)
	}
	defer rows.Close()

	var invalidIDs []string
	for rows.Next() {
		var id string
		var metadata string
		if err := rows.Scan(&id, &metadata); err != nil {
			return fmt.Errorf("scan Flow metadata row: %w", err)
		}
		if !json.Valid([]byte(metadata)) {
			invalidIDs = append(invalidIDs, id)
			// Cap the reported list at 10 to keep error messages readable;
			// continue iterating to exhaust the cursor and free DB resources.
			if len(invalidIDs) >= 10 {
				for rows.Next() { //nolint:revive // intentional drain
				}
				break
			}
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate Flow metadata rows: %w", err)
	}

	if len(invalidIDs) == 0 {
		log.Println("✅ all Flow Metadata columns contain valid JSON")
		return nil
	}
	return fmt.Errorf("invalid JSON in Flow.Metadata: %d row(s) — ids: [%s]",
		len(invalidIDs), strings.Join(invalidIDs, ", "))
}