package controlplane

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

	wq "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

// VerifyConnectorExecutionProtocols is the release gate used by every apply
// and recovery path. A range upgrade must not invoke the older destructive
// Event-backlog retirement to manufacture its own prerequisite.
func VerifyConnectorExecutionProtocols(ctx context.Context, db *sql.DB) error {
	// A global receipt is not evidence that every later row has a live owner.
	// Check residual rows before writing either receipt or selecting packs:
	// blank/deleted IDs would otherwise disappear from that selection.
	if err := requireOwnedConnectorExecutionHistory(ctx, db); err != nil {
		return err
	}
	applied, err := workItemCutoverMigrationApplied(ctx, db)
	if err != nil {
		return err
	}
	if !applied {
		if err = verifyConnectorWorkItemOnlyExecutionCutover(ctx, db, false); err != nil {
			return fmt.Errorf("connector range upgrade requires the prior WorkItem-only handoff; legacy backlog cannot be automatically retired by this upgrade: %w", err)
		}
	}
	rows, err := db.QueryContext(ctx, `SELECT p.PackInstallId FROM _System_InstalledPack p
 WHERE COALESCE(p.IsDeleted,0)=0 AND (
 EXISTS(SELECT 1 FROM _System_ConnectorInboundRoute r WHERE r.InstalledPackId=p.PackInstallId AND COALESCE(r.IsDeleted,0)=0)
 OR EXISTS(SELECT 1 FROM _System_ConnectorEvent e WHERE e.InstalledPackId=p.PackInstallId)
 OR EXISTS(SELECT 1 FROM _System_ConnectorWorkItem w WHERE w.InstalledPackId=p.PackInstallId))
 ORDER BY p.PackInstallId`)
	if err != nil {
		return err
	}
	var packs []string
	for rows.Next() {
		var id string
		if err = rows.Scan(&id); err != nil {
			rows.Close()
			return err
		}
		packs = append(packs, id)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	for _, packID := range packs {
		if _, err := EnsureConnectorProjectionRangeProtocol(ctx, db, packID); err != nil {
			return err
		}
	}
	return nil
}

func requireOwnedConnectorExecutionHistory(ctx context.Context, db *sql.DB) error {
	var unowned bool
	err := db.QueryRowContext(ctx, `SELECT EXISTS (
 SELECT 1 FROM _System_ConnectorEvent e
 LEFT JOIN _System_InstalledPack p ON p.PackInstallId=e.InstalledPackId AND COALESCE(p.IsDeleted,0)=0
 LEFT JOIN _System_ConnectorProjectionProtocol protocol ON protocol.InstalledPackId=e.InstalledPackId
 WHERE `+connectorNonterminalEventPredicate("e")+` AND
 (NULLIF(TRIM(e.InstalledPackId),'') IS NULL OR p.PackInstallId IS NULL)
 AND NOT (NULLIF(TRIM(e.InstalledPackId),'') IS NOT NULL AND protocol.InstalledPackId IS NOT NULL
 AND protocol.ProtocolVersion=? AND protocol.LegacyMaxOrdinal>=0 AND e.EnqueueOrdinal>=0
 AND (e.EnqueueOrdinal<=protocol.LegacyMaxOrdinal OR `+connectorTerminalNativeProjectionCoveragePredicate()+`))) OR EXISTS (
 SELECT 1 FROM _System_ConnectorWorkItem w
 LEFT JOIN _System_InstalledPack p ON p.PackInstallId=w.InstalledPackId AND COALESCE(p.IsDeleted,0)=0
 LEFT JOIN _System_ConnectorEvent e ON e.Id=w.EventId AND e.InstalledPackId=w.InstalledPackId
 WHERE w.WorkType IN ('connector_event','workflow_action_apply') AND `+connectorNonterminalWorkItemPredicate("w")+` AND
 (NULLIF(TRIM(w.InstalledPackId),'') IS NULL OR p.PackInstallId IS NULL OR e.Id IS NULL))`, wq.ProjectionRangeProtocolVersion).Scan(&unowned)
	if err != nil {
		return err
	}
	if unowned {
		return fmt.Errorf("connector projection protocol: unowned nonterminal Event or WorkItem history requires explicit recovery")
	}
	return verifyRetiredNativeProjectionKeys(ctx, db)
}

// A revision replaces the live InstalledPack ID, but keeps accounted Event
// receipts. Legacy input is covered by its immutable cutoff; later input must
// have a terminal native range with the exact identity and head. Merely having
// a protocol row never proves later input complete. Live orphan WorkItems are
// checked independently above, including rows below a legacy cutoff.
func connectorTerminalNativeProjectionCoveragePredicate() string {
	return `EXISTS (
 SELECT 1 FROM _System_ConnectorWorkItem projection
 JOIN _System_ConnectorEvent selected ON selected.Id=projection.EventId
 AND selected.InstalledPackId=projection.InstalledPackId
 WHERE ` + connectorTerminalNativeProjectionMatchPredicate() + `)`
}

func connectorTerminalNativeProjectionMatchPredicate() string {
	return `projection.InstalledPackId=e.InstalledPackId
 AND projection.WorkId LIKE 'csp:%' AND NULLIF(projection.ProjectionKey,'') IS NOT NULL
 AND projection.WorkType=CASE WHEN LOWER(TRIM(e.EventKind))='workflow.action.execute' THEN 'workflow_action_apply' ELSE 'connector_event' END
 AND projection.State IN ('processed','skipped','dead')
 AND ` + connectorWorkItemExecutionAuthorityPredicate("projection", "selected") + `
 AND projection.InputStartOrdinal>0 AND projection.InputStartOrdinal<=e.EnqueueOrdinal
 AND e.EnqueueOrdinal<=projection.LaneOrdinal AND projection.LaneOrdinal=selected.EnqueueOrdinal
 AND NULLIF(projection.SourceLaneKey,'') IS NOT NULL AND projection.SourceLaneKey=e.SourceLaneKey
 AND selected.SourceLaneKey=projection.SourceLaneKey
 AND LOWER(TRIM(selected.EventKind))=LOWER(TRIM(e.EventKind))
 AND COALESCE(selected.RouteVersion,0)=COALESCE(e.RouteVersion,0)
 AND COALESCE(selected.CompanionPackInstallId,'')=COALESCE(e.CompanionPackInstallId,'')
 AND COALESCE(selected.CompanionFlowVersionId,'')=COALESCE(e.CompanionFlowVersionId,'')`
}

// Validate distinct native identities in one batched read, using Workqueue's
// canonical key builder. Do not duplicate its hash in SQL or issue a query for
// each historical input. The preceding match already checks head and bounds.
func verifyRetiredNativeProjectionKeys(ctx context.Context, db *sql.DB) error {
	rows, err := db.QueryContext(ctx, `SELECT DISTINCT projection.ProjectionKey,projection.WorkType,
 selected.InstalledPackId,selected.EventKind,COALESCE(selected.RouteVersion,0),
 COALESCE(selected.CompanionPackInstallId,''),COALESCE(selected.CompanionFlowVersionId,''),selected.SourceLaneKey
 FROM _System_ConnectorEvent e
 JOIN _System_ConnectorProjectionProtocol protocol ON protocol.InstalledPackId=e.InstalledPackId
 LEFT JOIN _System_InstalledPack p ON p.PackInstallId=e.InstalledPackId AND COALESCE(p.IsDeleted,0)=0
 JOIN _System_ConnectorWorkItem projection ON projection.InstalledPackId=e.InstalledPackId
 JOIN _System_ConnectorEvent selected ON selected.Id=projection.EventId AND selected.InstalledPackId=projection.InstalledPackId
 WHERE p.PackInstallId IS NULL AND `+connectorNonterminalEventPredicate("e")+`
 AND protocol.ProtocolVersion=? AND protocol.LegacyMaxOrdinal>=0 AND e.EnqueueOrdinal>protocol.LegacyMaxOrdinal
 AND `+connectorTerminalNativeProjectionMatchPredicate(), wq.ProjectionRangeProtocolVersion)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var key string
		var identity wq.SourceProjectionIdentity
		if err := rows.Scan(&key, &identity.WorkType, &identity.InstalledPackID, &identity.EventKind, &identity.RouteVersion, &identity.CompanionPackInstallID, &identity.CompanionFlowVersionID, &identity.SourceLaneKey); err != nil {
			return err
		}
		if key != wq.SourceProjectionKey(identity) {
			return fmt.Errorf("connector projection protocol: retired native range has invalid identity key for pack %q", identity.InstalledPackID)
		}
	}
	return rows.Err()
}