package connector_workqueue

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

// ConnectorQueueStatsMode selects the amount of execution-state history and
// lane eligibility evaluated by QueryConnectorEventQueueStats.
type ConnectorQueueStatsMode uint8

const (
	ConnectorQueueStatsAll ConnectorQueueStatsMode = iota
	ConnectorQueueStatsLive
	ConnectorQueueStatsCheapLive
)

// QueueBacklogStats is the stable operational queue projection shared with
// the connector Event service. Mutable lifecycle values are sourced from
// WorkItem. Event is joined only for immutable namespace metadata.
type QueueBacklogStats struct {
	Total                   int64 `json:"total"`
	Pending                 int64 `json:"pending"`
	Due                     int64 `json:"due"`
	FailedDue               int64 `json:"failed_due"`
	FailedNotDue            int64 `json:"failed_not_due"`
	Processing              int64 `json:"processing"`
	StaleProcessing         int64 `json:"stale_processing"`
	Processed               int64 `json:"processed"`
	Dead                    int64 `json:"dead"`
	RetryAttempts           int64 `json:"retry_attempts"`
	OldestPendingAgeSeconds int64 `json:"oldest_pending_age_seconds"`
	OldestDueAgeSeconds     int64 `json:"oldest_due_age_seconds"`
	OldestFailedAgeSeconds  int64 `json:"oldest_failed_age_seconds"`
	OldestProcessingSeconds int64 `json:"oldest_processing_age_seconds"`
}

type QueueClassStats struct {
	Namespace       string            `json:"namespace"`
	InstalledPackID string            `json:"installed_pack_id"`
	QueueClass      string            `json:"queue_class"`
	Stats           QueueBacklogStats `json:"stats"`
}

const connectorQueueClaimabilitySQL = `CASE
               WHEN (
                   (w.State IN ('pending', 'retry', 'blocked', 'blocked_retry')
                    AND w.RunAfter <= p.NowAt)
                   OR (w.State = 'processing'
                       AND w.LeaseUntil IS NOT NULL
                       AND w.LeaseUntil <= p.NowAt)
               )
               AND (
                   NULLIF(COALESCE(w.ProjectionKey, ''), '') IS NULL
                   OR NULLIF(COALESCE(w.ProjectionActiveKey, ''), '') IS NOT NULL
               )
               AND NOT EXISTS (
                   SELECT 1
                     FROM _System_ConnectorWorkItem older USE INDEX (idx_connector_work_item_lease_due)
                    WHERE older.InstalledPackId = w.InstalledPackId
                      AND older.Generation = w.Generation
                      AND older.WorkId <> w.WorkId
                      AND (
                          NULLIF(COALESCE(older.ProjectionKey, ''), '') IS NULL
                          OR NULLIF(COALESCE(older.ProjectionActiveKey, ''), '') IS NOT NULL
                      )
                      AND (
                          older.State = 'processing'
                          OR (
                              older.State IN ('pending', 'retry', 'blocked', 'blocked_retry')
                              AND older.RunAfter <= p.NowAt
                          )
                      )
                      AND COALESCE(NULLIF(older.TargetMutationLaneKey, ''), older.SourceLaneKey, '') =
                          COALESCE(NULLIF(w.TargetMutationLaneKey, ''), w.SourceLaneKey, '')
                      AND (
                          older.LaneOrdinal < w.LaneOrdinal
                          OR (
                              older.LaneOrdinal = w.LaneOrdinal
                              AND (
                                  older.CreatedDate < w.CreatedDate
                                  OR (older.CreatedDate = w.CreatedDate AND older.WorkId < w.WorkId)
                              )
                          )
                      )
               )
               THEN 1 ELSE 0
           END`

func connectorQueueProjectionAuthorityPredicate(alias string) string {
	return fmt.Sprintf(`(
                  NULLIF(COALESCE(%[1]s.ProjectionActiveKey, ''), '') IS NOT NULL
                  OR (
                      NULLIF(COALESCE(%[1]s.ProjectionKey, ''), '') IS NULL
                      AND %[1]s.WorkId LIKE 'csp:%%'
                      AND %[1]s.State NOT IN ('processed', 'skipped', 'dead')
                  )
              )`, alias)
}

// WorkItem is the sole execution authority. Live modes intentionally exclude
// terminal rows so historical Event volume can never expand the live metrics
// read. The all-history mode includes terminal WorkItems for its Processed,
// Dead, and Total counters.
func connectorQueueWorkItemAuthorityPredicate(workAlias string, mode ConnectorQueueStatsMode) string {
	if mode == ConnectorQueueStatsAll {
		return "1 = 1"
	}
	return fmt.Sprintf(`(
                  %[1]s.State IN ('pending', 'retry', 'blocked', 'blocked_retry', 'processing')
                  AND (
                      %[1]s.WorkId NOT LIKE 'csp:%%'
                      OR %[2]s
                  )
              )`, workAlias, connectorQueueProjectionAuthorityPredicate(workAlias))
}

// connectorQueueAuthorityCTE builds the single operational read path for
// inbound connector execution state. The WorkType predicates are the same
// executor scope established by WithClaimWorkTypes at the claim boundary.
// Unsupported WorkTypes are never classified as connector Event work.
//
// Every executable row is represented by its WorkItem row. ConnectorEvent is
// joined only for namespace metadata; it is never an execution source here.
func connectorQueueAuthorityCTE(mode ConnectorQueueStatsMode) string {
	template := `
WITH params AS (
    SELECT CAST(? AS DATETIME(6)) AS NowAt,
           CAST(? AS DATETIME(6)) AS StaleCutoff
),
authoritative_queue AS (
    SELECT COALESCE(NULLIF(e.Namespace, ''), 'unknown') AS Namespace,
           COALESCE(NULLIF(w.InstalledPackId, ''), 'unknown') AS InstalledPackId,
           COALESCE(NULLIF(w.QueueClass, ''), 'default') AS QueueClass,
           w.State,
           w.RunAfter,
           w.LeaseUntil,
           w.FailureCount,
           w.CreatedDate,
           w.LastModifiedDate,
           CASE
               WHEN w.State IN ('pending', 'retry', 'blocked', 'blocked_retry')
                    AND w.RunAfter <= p.NowAt
                   THEN 1
               WHEN w.State = 'processing'
                    AND (w.LeaseUntil IS NULL OR w.LeaseUntil <= p.NowAt)
                    AND NOT (w.DependencyKind IS NOT NULL AND w.LeaseOwner IS NULL AND w.LeaseToken IS NULL AND w.LeaseUntil IS NULL)
                   THEN 1
               ELSE 0
           END AS RawDue,
           {{CLAIMABLE}} AS Claimable
      FROM _System_ConnectorWorkItem w
      JOIN _System_ConnectorEvent e
        ON e.Id = w.EventId
       AND e.InstalledPackId = w.InstalledPackId
      CROSS JOIN params p
     WHERE w.WorkType IN ('connector_event', 'workflow_action_apply')
	   AND ` + connectorQueueWorkItemAuthorityPredicate("w", mode) + `
)`
	claimability := "0"
	if mode == ConnectorQueueStatsLive {
		claimability = connectorQueueClaimabilitySQL
	}
	template = strings.Replace(template, "{{CLAIMABLE}}", claimability, 1)
	return template
}

func connectorQueueStatsQueryArgs(mode ConnectorQueueStatsMode, now, staleCutoff time.Time) []any {
	_ = mode
	return []any{now, staleCutoff}
}

func connectorQueueStatsQuery(mode ConnectorQueueStatsMode) string {
	dueColumn := "q.RawDue"
	liveWhere := ""
	if mode == ConnectorQueueStatsLive {
		dueColumn = "q.Claimable"
		liveWhere = `
 WHERE q.State IN (` + NonTerminalWorkStateSQLList + `)`
	} else if mode == ConnectorQueueStatsCheapLive {
		liveWhere = `
 WHERE q.State IN (` + NonTerminalWorkStateSQLList + `)`
	}
	return connectorQueueAuthorityCTE(mode) + `
SELECT q.Namespace,
       q.InstalledPackId,
       q.QueueClass,
       COUNT(*) AS Total,
       COALESCE(SUM(CASE
           WHEN q.State IN ('pending', 'blocked') AND ` + dueColumn + ` = 1 THEN 1
           ELSE 0 END), 0) AS Pending,
       COALESCE(SUM(CASE
           WHEN q.State IN ('retry', 'blocked_retry') AND ` + dueColumn + ` = 1 THEN 1
           ELSE 0 END), 0) AS FailedDue,
       COALESCE(SUM(CASE
           WHEN q.State IN ('retry', 'blocked_retry') AND q.RawDue = 0 THEN 1
           ELSE 0 END), 0) AS FailedNotDue,
       COALESCE(SUM(CASE WHEN q.State = 'processing' THEN 1 ELSE 0 END), 0) AS Processing,
       COALESCE(SUM(CASE
           WHEN q.State = 'processing' AND ` + dueColumn + ` = 1 THEN 1
           ELSE 0 END), 0) AS StaleProcessing,
       COALESCE(SUM(CASE
           WHEN q.State IN ('processed', 'skipped') THEN 1
           ELSE 0 END), 0) AS Processed,
       COALESCE(SUM(CASE WHEN q.State = 'dead' THEN 1 ELSE 0 END), 0) AS Dead,
       COALESCE(SUM(CASE
           WHEN q.State IN ('retry', 'blocked_retry', 'dead') THEN q.FailureCount
           ELSE 0 END), 0) AS RetryAttempts,
       MIN(CASE
           WHEN q.State IN ('pending', 'blocked') THEN q.CreatedDate
           ELSE NULL END) AS OldestPendingAt,
       MIN(CASE WHEN ` + dueColumn + ` = 1 THEN q.CreatedDate ELSE NULL END) AS OldestDueAt,
       MIN(CASE
           WHEN q.State IN ('retry', 'blocked_retry') AND ` + dueColumn + ` = 1
               THEN q.CreatedDate
           ELSE NULL END) AS OldestFailedDueAt,
       MIN(CASE
           WHEN q.State = 'processing' THEN q.LastModifiedDate
           ELSE NULL END) AS OldestProcessingAt
  FROM authoritative_queue q` + liveWhere + `
 GROUP BY q.Namespace, q.InstalledPackId, q.QueueClass
 ORDER BY q.Namespace ASC, q.InstalledPackId ASC, q.QueueClass ASC`
}

func (r *Repository) QueryConnectorEventQueueStats(
	ctx context.Context,
	now time.Time,
	staleCutoff time.Time,
	mode ConnectorQueueStatsMode,
) ([]QueueClassStats, error) {
	if r == nil || r.db == nil {
		return nil, fmt.Errorf("connector workqueue: queue stats repository is not configured")
	}
	if now.IsZero() || staleCutoff.IsZero() || !staleCutoff.Before(now) {
		return nil, fmt.Errorf("connector workqueue: invalid queue stats time")
	}
	if mode > ConnectorQueueStatsCheapLive {
		return nil, fmt.Errorf("connector workqueue: invalid queue stats mode")
	}
	rows, err := r.db.QueryContext(
		ctx,
		connectorQueueStatsQuery(mode),
		connectorQueueStatsQueryArgs(mode, now, staleCutoff)...,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var result []QueueClassStats
	for rows.Next() {
		var item QueueClassStats
		var oldestPendingAt, oldestDueAt, oldestFailedAt, oldestProcessingAt sql.NullTime
		if err := rows.Scan(
			&item.Namespace,
			&item.InstalledPackID,
			&item.QueueClass,
			&item.Stats.Total,
			&item.Stats.Pending,
			&item.Stats.FailedDue,
			&item.Stats.FailedNotDue,
			&item.Stats.Processing,
			&item.Stats.StaleProcessing,
			&item.Stats.Processed,
			&item.Stats.Dead,
			&item.Stats.RetryAttempts,
			&oldestPendingAt,
			&oldestDueAt,
			&oldestFailedAt,
			&oldestProcessingAt,
		); err != nil {
			return nil, err
		}
		item.Stats.Due = item.Stats.Pending + item.Stats.FailedDue + item.Stats.StaleProcessing
		item.Stats.OldestPendingAgeSeconds = queueAgeSeconds(now, oldestPendingAt)
		item.Stats.OldestDueAgeSeconds = queueAgeSeconds(now, oldestDueAt)
		item.Stats.OldestFailedAgeSeconds = queueAgeSeconds(now, oldestFailedAt)
		item.Stats.OldestProcessingSeconds = queueAgeSeconds(now, oldestProcessingAt)
		result = append(result, item)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return result, nil
}

func queueAgeSeconds(now time.Time, value sql.NullTime) int64 {
	if !value.Valid || now.Before(value.Time) {
		return 0
	}
	return int64(now.Sub(value.Time).Seconds())
}