package connector_events

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

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

// WorkItemOnlyExecutionCutoverMigrationID is stamped by the controlplane
// migration that atomically retires the abandoned Event-owned backlog behind a
// durable ingress fence. It is the versioned release receipt for the
// intentionally one-way WorkItem executor cutover.
//
// The receipt is not an Event backlog migration: pre-cutover Event-owned rows
// are deliberately outside the WorkItem executor's authority and are neither
// scanned nor replayed by this runtime.
const (
	WorkItemOnlyExecutionCutoverMigrationID          = "20260724_verify_connector_workitem_only_execution_cutover_v2"
	WorkItemOnlyExecutionCutoverMigrationVersion     = 2
	WorkItemOnlyExecutionCutoverMigrationDescription = "Atomically hand off connector execution to WorkItems after terminalizing Event-owned backlog while ingress is durably quiesced (NexusCRM #3980)."
)

var ErrWorkItemOnlyExecutionCutoverRequired = errors.New("connector events: WorkItem-only execution cutover receipt is required")

// RequireWorkItemOnlyExecutionCutover proves that this binary is running with
// the matching controlplane migration. Call it once at worker activation, not
// on enqueue or queue-stat hot paths.
func RequireWorkItemOnlyExecutionCutover(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return fmt.Errorf("%w: database is required", ErrWorkItemOnlyExecutionCutoverRequired)
	}
	var version int
	var description string
	err := db.QueryRowContext(ctx, `
SELECT Version, Description
  FROM _System_Migration
 WHERE Id = ?
 LIMIT 1`, WorkItemOnlyExecutionCutoverMigrationID).Scan(&version, &description)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("%w: apply the matching controlplane L0 migration before starting WorkItem workers", ErrWorkItemOnlyExecutionCutoverRequired)
	}
	if err != nil {
		return fmt.Errorf("verify WorkItem-only execution cutover receipt: %w", err)
	}
	if version != WorkItemOnlyExecutionCutoverMigrationVersion || description != WorkItemOnlyExecutionCutoverMigrationDescription {
		return fmt.Errorf("%w: matching controlplane handoff attestation is missing", ErrWorkItemOnlyExecutionCutoverRequired)
	}
	return nil
}

// RequireWorkItemOnlyExecutionCutover is retained as a convenience method for
// callers that already own the Event service. Supervisor admission is owned by
// the host and deliberately does not add a requirement to WorkItemService.
func (s *Service) RequireWorkItemOnlyExecutionCutover(ctx context.Context) error {
	if s == nil {
		return fmt.Errorf("%w: database is required", ErrWorkItemOnlyExecutionCutoverRequired)
	}
	return RequireWorkItemOnlyExecutionCutover(ctx, s.db)
}

// rejectLegacyEventReplayAfterWorkItemOnlyCutover closes the legacy Event-owned
// replay branch once the one-way handoff is attested. It intentionally fails
// closed for a malformed receipt as well: replaying Event history is not a
// recovery mechanism after this release mode is selected. The ownership
// preflight is read before Event or WorkItem replay locks; an attested
// WorkItem-owned generation keeps its native replay path. The returned
// attestation flag makes the later locked reservation fail closed if its owner
// disappeared after this read.
func rejectLegacyEventReplayAfterWorkItemOnlyCutover(ctx context.Context, tx *sql.Tx, eventID string) (bool, error) {
	var version int
	var description string
	err := tx.QueryRowContext(ctx, `
SELECT Version, Description
  FROM _System_Migration
 WHERE Id = ?
 LIMIT 1`, WorkItemOnlyExecutionCutoverMigrationID).Scan(&version, &description)
	if errors.Is(err, sql.ErrNoRows) || isMissingMigrationReceiptTable(err) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("verify WorkItem-only cutover before Event replay: %w", err)
	}
	if version != WorkItemOnlyExecutionCutoverMigrationVersion || description != WorkItemOnlyExecutionCutoverMigrationDescription {
		return false, fmt.Errorf("%w: cutover receipt attestation is invalid", ErrLegacyEventReplayRetired)
	}
	var ownedCount int
	err = tx.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorWorkItem
 WHERE EventId = ?
   AND WorkType IN (?, ?)
   AND Attempt > 0`, eventID, connector_workqueue.WorkTypeConnectorEvent, connector_workqueue.WorkTypeWorkflowActionApply).Scan(&ownedCount)
	if err != nil {
		return false, fmt.Errorf("verify WorkItem replay authority after cutover: %w", err)
	}
	if ownedCount > 0 {
		return true, nil
	}
	return false, ErrLegacyEventReplayRetired
}

func isMissingMigrationReceiptTable(err error) bool {
	if err == nil {
		return false
	}
	message := strings.ToLower(err.Error())
	return strings.Contains(message, "_system_migration") &&
		(strings.Contains(message, "1146") || strings.Contains(message, "doesn't exist") ||
			strings.Contains(message, "does not exist") || strings.Contains(message, "no such table"))
}