package controlplane

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"fmt"
	"strings"
	"sync"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

type connectorInboundWorkerReconciler func(context.Context, *sql.DB) error

var connectorInboundWorkerReconcilerRegistry struct {
	mu        sync.RWMutex
	reconcile connectorInboundWorkerReconciler
}

// SetConnectorInboundWorkerReconciler registers the live runtime hook used by
// control-plane pack lifecycle mutations to refresh connector inbound workers.
// Headless setup processes leave it unset, making lifecycle reconciliation a
// no-op because there is no in-process supervisor to update.
func SetConnectorInboundWorkerReconciler(reconcile func(context.Context, *sql.DB) error) func() {
	connectorInboundWorkerReconcilerRegistry.mu.Lock()
	previous := connectorInboundWorkerReconcilerRegistry.reconcile
	connectorInboundWorkerReconcilerRegistry.reconcile = reconcile
	connectorInboundWorkerReconcilerRegistry.mu.Unlock()

	return func() {
		connectorInboundWorkerReconcilerRegistry.mu.Lock()
		connectorInboundWorkerReconcilerRegistry.reconcile = previous
		connectorInboundWorkerReconcilerRegistry.mu.Unlock()
	}
}

func reconcileRegisteredConnectorInboundWorkers(ctx context.Context, db *sql.DB) error {
	connectorInboundWorkerReconcilerRegistry.mu.RLock()
	reconcile := connectorInboundWorkerReconcilerRegistry.reconcile
	connectorInboundWorkerReconcilerRegistry.mu.RUnlock()
	if reconcile == nil {
		return nil
	}
	return reconcile(ctx, db)
}

// ConnectorInboundWorkerLifecycleVersion returns a durable cross-process
// version of connector inbound worker eligibility. A running server polls this
// projection so external cmd/setup mutations and drain-only non-terminal queue
// changes wake worker reconciliation without rebuilding workers on every tick.
func ConnectorInboundWorkerLifecycleVersion(ctx context.Context, db *sql.DB) (string, error) {
	if db == nil {
		return "", nil
	}
	eligibleRows, err := packs.ListConnectorPacksForInboundWorkers(ctx, db)
	if err != nil {
		return "", fmt.Errorf("connector inbound worker eligibility version query: %w", err)
	}
	hash := sha256.New()
	count := 0
	for _, row := range eligibleRows {
		count++
		fmt.Fprintf(hash, "worker\x00%s\x00%s\x00%s\x00%s\x00%s\x00%t\n", row.ID, row.InstallID, row.Namespace, row.Status, row.PayloadHash, row.DrainOnly)
	}
	projectionCount, err := appendConnectorInboundProjectionLifecycleRows(ctx, db, hash)
	if err != nil {
		return "", fmt.Errorf("connector inbound worker projection version query: %w", err)
	}
	count += projectionCount
	return fmt.Sprintf("%d:%s", count, hex.EncodeToString(hash.Sum(nil))), nil
}

func appendConnectorInboundProjectionLifecycleRows(ctx context.Context, db *sql.DB, hash interface {
	Write([]byte) (int, error)
}) (int, error) {
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT cp.Name, cp.PackInstallId, s.CompanionPackName, s.CompanionPackInstallId, s.ObjectKind, s.SourceField
  FROM _System_InstalledPack cp
  JOIN _System_ConnectorInboundProjectionSnapshot s
    ON s.OwnerPackName = cp.Name
   AND s.OwnerPackInstallId = cp.PackInstallId
 WHERE cp.Category = ?
   AND cp.Status = ?
   AND COALESCE(cp.IsDeleted, 0) = 0
   AND TRIM(COALESCE(cp.PackInstallId, '')) <> ''
   AND TRIM(COALESCE(cp.AppliedPayloadHash, '')) <> ''
   AND TRIM(COALESCE(cp.DesiredPayloadHash, '')) <> ''
   AND cp.AppliedPayloadHash = cp.DesiredPayloadHash
   AND EXISTS (
        SELECT 1
          FROM _System_ConnectorCapabilityDeclaration c
         WHERE c.InstalledPackName = cp.Name
           AND c.Namespace = cp.Name
           AND c.Kind = 'signature_verifier'
           AND COALESCE(c.IsDeleted, 0) = 0
   )
   AND EXISTS (
        SELECT 1
          FROM _System_ConnectorCapabilityDeclaration c
         WHERE c.InstalledPackName = cp.Name
           AND c.Namespace = cp.Name
           AND c.Kind = 'envelope_parser'
           AND COALESCE(c.IsDeleted, 0) = 0
   )
   AND EXISTS (
        SELECT 1
          FROM _System_ConnectorCalloutProfile profile
         WHERE profile.OwnerPack = cp.Name
           AND profile.InstalledPackId = cp.PackInstallId
           AND COALESCE(profile.IsDeleted, 0) = 0
           AND TRIM(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(profile.BatchSpec, '$.projectionBodyField')), '')) <> ''
   )
   AND COALESCE(s.IsDeleted, 0) = 0
   AND (
        EXISTS (
            SELECT 1
              FROM _System_InstalledPackDependency d
             WHERE d.DependsOnPackName = cp.Name
               AND COALESCE(d.IsDeleted, 0) = 0
               AND d.DependentPackName = s.CompanionPackName
               AND d.DependentPackInstallId = s.CompanionPackInstallId
               AND (
                    EXISTS (
                        SELECT 1
                          FROM _System_InstalledPack p
                         WHERE p.Name = d.DependentPackName
                           AND p.PackInstallId = d.DependentPackInstallId
                           AND p.Category = ?
                           AND p.Status = ?
                           AND TRIM(COALESCE(p.AppliedPayloadHash, '')) <> ''
                           AND TRIM(COALESCE(p.DesiredPayloadHash, '')) <> ''
                           AND p.AppliedPayloadHash = p.DesiredPayloadHash
                           AND COALESCE(p.IsDeleted, 0) = 0
                    )
                    OR EXISTS (
                        SELECT 1
                          FROM _System_InstalledPack p
                         WHERE p.Name = d.DependentPackName
                           AND p.Category = ?
                           AND p.Status = 'installing'
                           AND COALESCE(p.IsDeleted, 0) = 0
                    )
               )
        )
        OR EXISTS (
            SELECT 1
              FROM _System_ConnectorEvent e
             WHERE e.InstalledPackId = cp.PackInstallId
               AND COALESCE(e.CompanionPackInstallId, '') = s.CompanionPackInstallId
               AND e.Status IN ('pending', 'failed', 'processing')
        )
   )
 ORDER BY cp.Name, cp.PackInstallId, s.CompanionPackName, s.CompanionPackInstallId, s.ObjectKind, s.SourceField`,
		string(packs.ConnectorPack), installedPackStatusActive, string(packs.AppPack), installedPackStatusActive, string(packs.AppPack))
	if err != nil {
		return 0, err
	}
	defer rows.Close()

	count := 0
	for rows.Next() {
		var connectorPack, connectorInstallID, companionPack, companionInstallID, sourceObject, sourceField string
		if err := rows.Scan(&connectorPack, &connectorInstallID, &companionPack, &companionInstallID, &sourceObject, &sourceField); err != nil {
			return 0, err
		}
		count++
		fmt.Fprintf(hash, "projection\x00%s\x00%s\x00%s\x00%s\x00%s\x00%s\n",
			strings.TrimSpace(connectorPack),
			strings.TrimSpace(connectorInstallID),
			strings.TrimSpace(companionPack),
			strings.TrimSpace(companionInstallID),
			strings.TrimSpace(sourceObject),
			strings.TrimSpace(sourceField),
		)
	}
	if err := rows.Err(); err != nil {
		return 0, err
	}
	return count, nil
}