package controlplane

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

type connectorInboundReapplyScope int

const (
	connectorPackInboundReapplyScope connectorInboundReapplyScope = iota
	companionAppPackInboundReapplyScope
)

var connectorInboundWorkItemRequiredColumns = []string{
	"WorkId",
	"EventId",
	"WorkType",
	"InstalledPackId",
	"State",
	"Attempt",
	"ProjectionKey",
	"SourceLaneKey",
}

var connectorWorkItemCutoverRequiredColumns = append(
	append([]string{}, connectorInboundWorkItemRequiredColumns...),
	"LeaseOwner",
	"LeaseToken",
	"LeaseUntil",
	"TerminalAt",
	"LastModifiedDate",
)

func connectorWorkItemSchemaReady(ctx context.Context, db *sql.DB, requiredColumns []string) (bool, error) {
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(requiredColumns)), ",")
	args := make([]interface{}, 0, len(requiredColumns)+1)
	args = append(args, "_System_ConnectorWorkItem")
	for _, column := range requiredColumns {
		args = append(args, column)
	}
	var count int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(DISTINCT COLUMN_NAME)
  FROM INFORMATION_SCHEMA.COLUMNS
 WHERE TABLE_SCHEMA = DATABASE()
	   AND TABLE_NAME = ?
	   AND COLUMN_NAME IN (`+placeholders+`)`, args...).Scan(&count); err != nil {
		return false, fmt.Errorf("query connector WorkItem schema readiness: %w", err)
	}
	return count == len(requiredColumns), nil
}

func connectorInboundWorkItemSchemaReady(ctx context.Context, db *sql.DB) (bool, error) {
	return connectorWorkItemSchemaReady(ctx, db, connectorInboundWorkItemRequiredColumns)
}

func connectorWorkItemCutoverSchemaReady(ctx context.Context, db *sql.DB) (bool, error) {
	return connectorWorkItemSchemaReady(ctx, db, connectorWorkItemCutoverRequiredColumns)
}

func connectorNonterminalEventPredicate(alias string) string {
	return fmt.Sprintf("COALESCE(%s.Status, '') NOT IN ('processed', 'skipped', 'dead')", alias)
}

func connectorNonterminalWorkItemPredicate(alias string) string {
	return fmt.Sprintf("COALESCE(%s.State, '') NOT IN ('processed', 'skipped', 'dead')", alias)
}

func connectorWorkItemExecutionAuthorityPredicate(workAlias, eventAlias string) string {
	return fmt.Sprintf("(%s.Attempt > 0 OR %s.ExecutionOwner = 'work_item')", workAlias, eventAlias)
}

func connectorInboundPackInstallIDMatch(left, right string, exact bool) string {
	if exact {
		return "CAST(" + left + " AS BINARY) <=> CAST(" + right + " AS BINARY)"
	}
	return left + " = " + right
}

// connectorInboundReapplyBlockerPredicate keeps pack lifecycle changes atomic
// with inbound admission. Nonterminal projection work blocks from creation,
// including before its first claim. Ordinary compatibility mirrors become
// authoritative only after their durable Attempt becomes positive or the Event
// CAS fence names the WorkItem executor. An Event row blocks only when it has
// neither directly authoritative work nor a covering source-projection successor
// that acquired WorkItem authority. Older or incomplete schemas retain the
// legacy Event.Status guard.
func connectorInboundReapplyBlockerPredicate(scope connectorInboundReapplyScope, workItemAuthority bool) string {
	return connectorInboundReapplyBlockerPredicateWithIdentity(
		scope, workItemAuthority, false,
	)
}

// connectorInboundExactIncarnationBlockerPredicate preserves the canonical
// WorkItem/Event/projection lifecycle authority while comparing every
// pack-install identity byte-for-byte. Exact source-revision transitions use
// this form after locking the frozen InstalledPack row.
func connectorInboundExactIncarnationBlockerPredicate(scope connectorInboundReapplyScope, workItemAuthority bool) string {
	return connectorInboundReapplyBlockerPredicateWithIdentity(
		scope, workItemAuthority, true,
	)
}

func connectorInboundReapplyBlockerPredicateWithIdentity(
	scope connectorInboundReapplyScope,
	workItemAuthority bool,
	exactIdentity bool,
) string {
	eventPackMatch := connectorInboundPackInstallIDMatch(
		"e.InstalledPackId", "p.PackInstallId", exactIdentity,
	)
	workItemPackMatch := connectorInboundPackInstallIDMatch(
		"w.InstalledPackId", "p.PackInstallId", exactIdentity,
	)
	if scope == companionAppPackInboundReapplyScope {
		eventPackMatch = connectorInboundPackInstallIDMatch(
			"COALESCE(e.CompanionPackInstallId, '')", "p.PackInstallId", exactIdentity,
		)
		workItemPackMatch = connectorInboundPackInstallIDMatch(
			"COALESCE(queued_event.CompanionPackInstallId, '')", "p.PackInstallId", exactIdentity,
		)
	}
	if !workItemAuthority {
		return `EXISTS (
       SELECT 1
         FROM _System_ConnectorEvent e
        WHERE ` + eventPackMatch + `
          AND e.Status IN ('pending', 'failed', 'processing')
   )`
	}

	workItemEventMatch := connectorInboundPackInstallIDMatch(
		"queued_event.InstalledPackId", "w.InstalledPackId", exactIdentity,
	)
	workItemFrom := `_System_ConnectorWorkItem w
         LEFT JOIN _System_ConnectorEvent queued_event
           ON queued_event.Id = w.EventId
          AND ` + workItemEventMatch
	if scope == companionAppPackInboundReapplyScope {
		workItemFrom = `_System_ConnectorWorkItem w
         JOIN _System_ConnectorEvent queued_event
           ON queued_event.Id = w.EventId
          AND ` + workItemEventMatch
	}
	directWorkInstallMatch := ""
	if exactIdentity {
		directWorkInstallMatch = `
          AND ` + connectorInboundPackInstallIDMatch(
			"direct_work.InstalledPackId", "e.InstalledPackId", true,
		)
	}
	projectionInstallMatch := connectorInboundPackInstallIDMatch(
		"projection.InstalledPackId", "e.InstalledPackId", exactIdentity,
	)
	selectedInstallMatch := connectorInboundPackInstallIDMatch(
		"selected.InstalledPackId", "e.InstalledPackId", exactIdentity,
	)
	selectedCompanionMatch := connectorInboundPackInstallIDMatch(
		"COALESCE(selected.CompanionPackInstallId, '')",
		"COALESCE(e.CompanionPackInstallId, '')",
		exactIdentity,
	)
	return `(EXISTS (
       SELECT 1
         FROM ` + workItemFrom + `
        WHERE ` + workItemPackMatch + `
          AND w.WorkType IN ('connector_event', 'workflow_action_apply')
          AND (` + connectorWorkItemExecutionAuthorityPredicate("w", "queued_event") + `
               OR w.WorkId LIKE 'csp:%'
               OR NULLIF(w.ProjectionKey, '') IS NOT NULL)
          AND ` + connectorNonterminalWorkItemPredicate("w") + `
   )
    OR EXISTS (
       SELECT 1
         FROM _System_ConnectorEvent e
         LEFT JOIN _System_ConnectorWorkItem direct_work
           ON direct_work.EventId = e.Id
          AND direct_work.WorkType IN ('connector_event', 'workflow_action_apply')
		  AND ` + connectorWorkItemExecutionAuthorityPredicate("direct_work", "e") + directWorkInstallMatch + `
        WHERE ` + eventPackMatch + `
          AND ` + connectorNonterminalEventPredicate("e") + `
          AND direct_work.WorkId IS NULL
          AND NOT EXISTS (
              SELECT 1
                FROM _System_ConnectorWorkItem projection
                JOIN _System_ConnectorEvent selected ON selected.Id = projection.EventId
               WHERE ` + projectionInstallMatch + `
                 AND projection.WorkType IN ('connector_event', 'workflow_action_apply')
				 AND ` + connectorWorkItemExecutionAuthorityPredicate("projection", "selected") + `
                 AND (
                     NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NOT NULL
                     OR (
                         NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NULL
                         AND projection.WorkId LIKE 'csp:%'
                         AND projection.State NOT IN ('processed', 'skipped', 'dead')
                     )
                 )
                 AND COALESCE(projection.SourceLaneKey, '') = COALESCE(e.SourceLaneKey, '')
                 AND ` + selectedInstallMatch + `
                 AND LOWER(TRIM(selected.EventKind)) = LOWER(TRIM(e.EventKind))
                 AND COALESCE(selected.RouteVersion, 0) = COALESCE(e.RouteVersion, 0)
                 AND ` + selectedCompanionMatch + `
                 AND COALESCE(selected.CompanionFlowVersionId, '') = COALESCE(e.CompanionFlowVersionId, '')
                 AND (
                     (
                         e.EnqueueOrdinal <> 0
                         AND selected.EnqueueOrdinal <> 0
                         AND (
                             e.EnqueueOrdinal < selected.EnqueueOrdinal
                             OR (e.EnqueueOrdinal = selected.EnqueueOrdinal AND e.Id <= selected.Id)
                         )
                     )
                     OR (
                         (e.EnqueueOrdinal = 0 OR selected.EnqueueOrdinal = 0)
                         AND (
                             e.CreatedDate < selected.CreatedDate
                             OR (
                                 e.CreatedDate = selected.CreatedDate
                                 AND (
                                     e.DeliverySequence < selected.DeliverySequence
                                     OR (e.DeliverySequence = selected.DeliverySequence AND e.Id <= selected.Id)
                                 )
                             )
                         )
                     )
                 )
          )
   ))`
}