package controlplane

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

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

// ProjectionRangeCutoverResult reports the immutable last old-protocol input.
// It does not attest business success or rewrite historical task membership.
type ProjectionRangeCutoverResult struct {
	InstalledPackID  string
	LegacyMaxOrdinal int64
	AlreadyApplied   bool
}

type projectionCutoverEvent struct {
	id         string
	identity   wq.SourceProjectionIdentity
	ordinal    int64
	closed     bool
	projection bool
}

type projectionCutoverWork struct {
	id, eventID, key, state string
	identity                wq.SourceProjectionIdentity
	ordinal                 int64
	terminal                bool
	authoritative           bool
}

// EnsureConnectorProjectionRangeProtocol records the protocol boundary after
// old projection execution has drained. The caller must stop old writers and
// workers while retaining the namespace admission fence. Only the checkpoint
// is written; historical Events, WorkItems, leases and payloads stay unchanged.
func EnsureConnectorProjectionRangeProtocol(ctx context.Context, db *sql.DB, packID string) (result ProjectionRangeCutoverResult, err error) {
	packID = strings.TrimSpace(packID)
	result.InstalledPackID = packID
	if db == nil || packID == "" {
		return result, fmt.Errorf("projection cutover requires database and pack incarnation")
	}
	if result.LegacyMaxOrdinal, result.AlreadyApplied, err = readProjectionProtocol(ctx, db, packID); err != nil || result.AlreadyApplied {
		return result, err
	}
	applied, err := workItemCutoverMigrationApplied(ctx, db)
	if err != nil {
		return result, err
	}
	if !applied {
		return result, fmt.Errorf("projection cutover requires the existing WorkItem-only handoff; it does not retire legacy backlog")
	}
	var namespace string
	if err = db.QueryRowContext(ctx, `SELECT Name FROM _System_InstalledPack WHERE PackInstallId=? AND COALESCE(IsDeleted,0)=0`, packID).Scan(&namespace); err != nil {
		return result, err
	}
	tx, cleanup, err := persistence.BeginPessimisticTx(persistence.WithReadCommittedIsolation(ctx), db)
	if err != nil {
		return result, err
	}
	defer func() {
		_ = tx.Rollback()
		if cleanupErr := cleanup(); err == nil && cleanupErr != nil {
			err = cleanupErr
		}
	}()
	// Match ingress lock order: namespace fence, then pack, then WorkItems.
	if err = requireProjectionCutoverFence(ctx, tx, namespace, packID); err != nil {
		return result, err
	}
	var lockedName string
	if err = tx.QueryRowContext(ctx, `SELECT Name FROM _System_InstalledPack WHERE PackInstallId=? AND COALESCE(IsDeleted,0)=0 FOR UPDATE`, packID).Scan(&lockedName); err != nil {
		return result, err
	}
	if lockedName != namespace {
		return result, fmt.Errorf("projection cutover incarnation changed")
	}
	if result.LegacyMaxOrdinal, result.AlreadyApplied, err = readProjectionProtocol(ctx, tx, packID); err != nil {
		return result, err
	}
	if result.AlreadyApplied {
		err = tx.Commit()
		return result, err
	}
	events, works, err := loadProjectionCutoverFacts(ctx, tx, packID)
	if err != nil {
		return result, err
	}
	cutoff, err := auditDrainedProjectionHistory(events, works)
	if err != nil {
		return result, err
	}
	if _, err = tx.ExecContext(ctx, `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,?,?)`, packID, wq.ProjectionRangeProtocolVersion, cutoff); err != nil {
		return result, err
	}
	if err = tx.Commit(); err != nil {
		return result, err
	}
	result.LegacyMaxOrdinal = cutoff
	return result, nil
}

// The caller holds the old InstalledPack row lock shared with admission. This
// check cannot perform cutover: that would acquire the namespace fence after
// the pack lock. An occupied old incarnation must be explicitly checkpointed
// before any entrypoint can replace its identity. Empty failed attempts may retry.
func requireConnectorProjectionCheckpointBeforeRevision(ctx context.Context, tx *sql.Tx, packID string) error {
	if _, found, err := readProjectionProtocol(ctx, tx, packID); err != nil || found {
		return err
	}
	var occupied bool
	if err := tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM _System_ConnectorEvent WHERE InstalledPackId=?)
 OR EXISTS(SELECT 1 FROM _System_ConnectorWorkItem WHERE InstalledPackId=?)`, packID, packID).Scan(&occupied); err != nil {
		return err
	}
	if occupied {
		return fmt.Errorf("connector incarnation %q requires explicit projection protocol cutover before source-revision advance", packID)
	}
	return nil
}

func readProjectionProtocol(ctx context.Context, q workItemCutoverReceiptQuerier, packID string) (cutoff int64, found bool, err error) {
	var version int
	err = q.QueryRowContext(ctx, `SELECT ProtocolVersion,LegacyMaxOrdinal FROM _System_ConnectorProjectionProtocol WHERE InstalledPackId=?`, packID).Scan(&version, &cutoff)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, err
	}
	if version != wq.ProjectionRangeProtocolVersion || cutoff < 0 {
		return 0, false, fmt.Errorf("invalid projection protocol for pack %q", packID)
	}
	return cutoff, true, nil
}

func requireProjectionCutoverFence(ctx context.Context, tx *sql.Tx, namespace, packID string) error {
	rows, err := tx.QueryContext(ctx, `SELECT COALESCE(PackInstallId,''),Channel,Mode,(ExpiresAt IS NULL),
 (ExpiresAt IS NULL OR ExpiresAt>UTC_TIMESTAMP()) FROM _System_ConnectorIngressControl
 WHERE LOWER(TRIM(Namespace))=LOWER(TRIM(?)) AND (PackInstallId=? OR COALESCE(PackInstallId,'')='')
 AND COALESCE(IsDeleted,0)=0 FOR UPDATE`, namespace, packID)
	if err != nil {
		return err
	}
	defer rows.Close()
	fenced := false
	for rows.Next() {
		var pack, channel, mode string
		var nonExpiring bool
		var live bool
		if err := rows.Scan(&pack, &channel, &mode, &nonExpiring, &live); err != nil {
			return err
		}
		if !live {
			continue
		}
		mode = strings.ToLower(strings.TrimSpace(mode))
		if mode == "active" {
			return fmt.Errorf("projection cutover: active ingress override remains for %q", namespace)
		}
		if pack == "" && channel == "*" && nonExpiring && (mode == "paused" || mode == "quiescing") {
			fenced = true
		}
	}
	if err := rows.Err(); err != nil {
		return err
	}
	if !fenced {
		return fmt.Errorf("projection cutover: %q requires a non-expiring namespace ingress fence", namespace)
	}
	return nil
}

func loadProjectionCutoverFacts(ctx context.Context, tx *sql.Tx, packID string) (map[string]*projectionCutoverEvent, []*projectionCutoverWork, error) {
	events := map[string]*projectionCutoverEvent{}
	rows, err := tx.QueryContext(ctx, `SELECT e.Id,e.EventKind,COALESCE(e.RouteVersion,0),COALESCE(e.CompanionPackInstallId,''),
 COALESCE(e.CompanionFlowVersionId,''),COALESCE(e.SourceLaneKey,''),e.EnqueueOrdinal,
 (e.ExecutionOwner='event' AND (e.Status IN ('processed','dead') OR
 (e.Status='skipped' AND COALESCE(e.ClaimToken,'')<>'' AND e.LastError IS NULL))),COALESCE(r.InvocationMode,'')
 FROM _System_ConnectorEvent e LEFT JOIN _System_ConnectorInboundRoute r
 ON r.InstalledPackId=e.InstalledPackId AND r.EventKind=e.EventKind AND r.Version=e.RouteVersion
 WHERE e.InstalledPackId=?`, packID)
	if err != nil {
		return nil, nil, err
	}
	for rows.Next() {
		e := &projectionCutoverEvent{}
		var mode string
		e.identity.InstalledPackID = packID
		if err := rows.Scan(&e.id, &e.identity.EventKind, &e.identity.RouteVersion, &e.identity.CompanionPackInstallID, &e.identity.CompanionFlowVersionID,
			&e.identity.SourceLaneKey, &e.ordinal, &e.closed, &mode); err != nil {
			rows.Close()
			return nil, nil, err
		}
		e.identity.WorkType = wq.WorkTypeConnectorEvent
		if strings.EqualFold(strings.TrimSpace(e.identity.EventKind), "workflow.action.execute") {
			e.identity.WorkType = wq.WorkTypeWorkflowActionApply
		}
		e.projection = mode == "latest_by_source_full_record_preserve_payload_settled"
		events[e.id] = e
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return nil, nil, err
	}
	rows, err = tx.QueryContext(ctx, `SELECT w.WorkId,w.EventId,COALESCE(w.ProjectionKey,''),w.State,
 w.WorkType,COALESCE(w.SourceLaneKey,''),w.LaneOrdinal,
 COALESCE(`+connectorWorkItemExecutionAuthorityPredicate("w", "e")+`,FALSE)
 FROM _System_ConnectorWorkItem w LEFT JOIN _System_ConnectorEvent e
 ON e.Id=w.EventId AND e.InstalledPackId=w.InstalledPackId
 WHERE w.InstalledPackId=? ORDER BY w.WorkId FOR UPDATE`, packID)
	if err != nil {
		return nil, nil, err
	}
	defer rows.Close()
	var works []*projectionCutoverWork
	for rows.Next() {
		w := &projectionCutoverWork{}
		w.identity.InstalledPackID = packID
		if err := rows.Scan(&w.id, &w.eventID, &w.key, &w.state, &w.identity.WorkType, &w.identity.SourceLaneKey, &w.ordinal, &w.authoritative); err != nil {
			return nil, nil, err
		}
		w.terminal = w.state == "processed" || w.state == "skipped" || w.state == "dead"
		works = append(works, w)
	}
	return events, works, rows.Err()
}

// auditDrainedProjectionHistory checks retained execution responsibility, not
// historical business success. Positive ordinals can identify completed
// projection boundaries. Mixed zero/positive history cannot establish the
// membership of an indirect input, so it requires explicit recovery. No ranges
// are reconstructed and no old input is made executable by this check.
func auditDrainedProjectionHistory(events map[string]*projectionCutoverEvent, works []*projectionCutoverWork) (int64, error) {
	type history struct {
		hasZero      bool
		lastTerminal int64
		ordinals     map[int64]string
		inputs       []*projectionCutoverEvent
	}
	groups := map[string]*history{}
	direct := map[string]*projectionCutoverWork{}
	// Discover every projection before validating its owners. A later row must
	// not change whether an earlier ordinary/type-split WorkItem is relevant.
	counts := map[string]int{}
	for _, w := range works {
		counts[w.eventID]++
		if e := events[w.eventID]; e != nil && (wq.IsSourceProjectionWorkID(w.id) || w.key != "") {
			groups[wq.SourceProjectionKey(e.identity)] = &history{ordinals: map[int64]string{}}
		}
	}
	for _, e := range events {
		e.projection = e.projection || groups[wq.SourceProjectionKey(e.identity)] != nil
		if e.projection && counts[e.id] > 1 {
			return 0, fmt.Errorf("projection cutover: Event %q has ambiguous WorkItem owners", e.id)
		}
	}
	var cutoff int64
	for _, w := range works {
		e := events[w.eventID]
		projection := wq.IsSourceProjectionWorkID(w.id) || w.key != ""
		if e == nil {
			if !w.terminal || projection {
				return 0, fmt.Errorf("projection cutover: WorkItem %q has no Event", w.id)
			}
			continue
		}
		if !projection {
			if e.projection && !w.terminal {
				return 0, fmt.Errorf("projection cutover: old WorkItem %q is %s; drain it with the old runtime", w.id, w.state)
			}
			if w.authoritative {
				direct[e.id] = w
			}
			continue
		}
		if !w.terminal {
			return 0, fmt.Errorf("projection cutover: old WorkItem %q is %s; drain it with the old runtime", w.id, w.state)
		}
		key := wq.SourceProjectionKey(e.identity)
		if w.ordinal != e.ordinal || w.identity.SourceLaneKey != e.identity.SourceLaneKey || w.identity.WorkType != e.identity.WorkType || (w.key != "" && w.key != key) {
			return 0, fmt.Errorf("projection cutover: WorkItem %q identity or ordinal mismatch", w.id)
		}
		e.projection = true
		if w.authoritative {
			direct[e.id] = w
		}
		if groups[key] == nil {
			groups[key] = &history{ordinals: map[int64]string{}}
		}
	}
	for _, e := range events {
		if e.ordinal < 0 {
			return 0, fmt.Errorf("projection cutover: Event %q has a negative ordinal", e.id)
		}
		if e.ordinal > cutoff {
			cutoff = e.ordinal
		}
		key := wq.SourceProjectionKey(e.identity)
		group := groups[key]
		if !e.projection && group == nil {
			if direct[e.id] == nil && !e.closed {
				return 0, fmt.Errorf("projection cutover: Event %q has no represented execution owner", e.id)
			}
			continue
		}
		if group == nil {
			group = &history{ordinals: map[int64]string{}}
			groups[key] = group
		}
		if e.ordinal == 0 {
			group.hasZero = true
		} else if other := group.ordinals[e.ordinal]; other != "" {
			return 0, fmt.Errorf("projection cutover: duplicate Event ordinal for %q and %q", other, e.id)
		} else {
			group.ordinals[e.ordinal] = e.id
		}
		group.inputs = append(group.inputs, e)
		if w := direct[e.id]; e.closed || (w != nil && w.terminal && (wq.IsSourceProjectionWorkID(w.id) || w.key != "")) {
			if e.ordinal > group.lastTerminal {
				group.lastTerminal = e.ordinal
			}
		}
	}
	for _, group := range groups {
		for _, e := range group.inputs {
			if w := direct[e.id]; e.closed || (w != nil && w.terminal) {
				continue
			}
			// A missing source is unknown identity, not one shared source. Even
			// positive ordinals cannot let another terminal input cover it.
			if strings.TrimSpace(e.identity.SourceLaneKey) == "" || group.hasZero || e.ordinal > group.lastTerminal {
				return 0, fmt.Errorf("projection cutover: Event %q has unresolved historical responsibility; explicit recovery is required", e.id)
			}
		}
	}
	return cutoff, nil
}

// ensureEmptyConnectorProjectionProtocolTx is part of pack activation, so a
// later direct connector installation cannot miss the new execution protocol.
func ensureEmptyConnectorProjectionProtocolTx(ctx context.Context, tx *sql.Tx, namespace string) error {
	var packID string
	err := tx.QueryRowContext(ctx, `SELECT p.PackInstallId FROM _System_InstalledPack p WHERE p.Name=? AND COALESCE(p.IsDeleted,0)=0
	 AND (p.Category='connector-pack'
	 OR 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))`, namespace).Scan(&packID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	if _, found, err := readProjectionProtocol(ctx, tx, packID); err != nil || found {
		return err
	}
	var occupied bool
	if err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM _System_ConnectorEvent WHERE InstalledPackId=?)
 OR EXISTS(SELECT 1 FROM _System_ConnectorWorkItem WHERE InstalledPackId=?)`, packID, packID).Scan(&occupied); err != nil {
		return err
	}
	if occupied {
		return fmt.Errorf("connector pack %q requires an explicit fenced projection range cutover before activation", namespace)
	}
	_, err = tx.ExecContext(ctx, `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,?,0)`, packID, wq.ProjectionRangeProtocolVersion)
	return err
}