package packstate

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"

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

type Category string

const (
	AppPack       Category = "app-pack"
	ConnectorPack Category = "connector-pack"
)

const (
	InstalledPackStatusActive     = "active"
	InstalledPackStatusRemoving   = "removing"
	InstalledPackStatusInstalling = "installing"
)

// ConnectorPackRow is the connector-pack-shaped projection of an installed-pack row.
type ConnectorPackRow struct {
	ID          string
	InstallID   string
	Namespace   string
	Category    string
	Status      string
	PayloadHash string
	PackVersion string
}

// ConnectorInboundWorkerPackRow is the startup projection used by the inbound
// worker supervisor. DrainOnly=true means the pack is not active but still has
// non-terminal queue rows that must be finalized before uninstall completes.
type ConnectorInboundWorkerPackRow struct {
	ConnectorPackRow
	DrainOnly bool
}

// ConnectorInboundHandlerConfigComponentPayload is the manifest payload for one
// connector_inbound_handler_config declaration consumed by worker startup.
type ConnectorInboundHandlerConfigComponentPayload struct {
	Namespace      string          `json:"namespace"`
	CapabilityKind string          `json:"capabilityKind"`
	Config         json.RawMessage `json:"config"`
}

// ResolveConnectorPackByNamespace returns the active connector-pack row for the
// given namespace, which is the installed-pack Name for connector packs.
func ResolveConnectorPackByNamespace(ctx context.Context, db *sql.DB, namespace string) (*ConnectorPackRow, bool, error) {
	namespace = strings.TrimSpace(namespace)
	if namespace == "" {
		return nil, false, nil
	}
	var row ConnectorPackRow
	var payloadHash, packVersion sql.NullString
	err := db.QueryRowContext(ctx, `
	SELECT Id, PackInstallId, Name, Category, Status, COALESCE(AppliedPayloadHash, ''), COALESCE(PackVersion, '')
	  FROM _System_InstalledPack
	 WHERE Name = ?
	   AND Category = ?
	   AND Status = ?
	   AND COALESCE(IsDeleted, 0) = 0
	 LIMIT 1`, namespace, string(ConnectorPack), InstalledPackStatusActive).Scan(&row.ID, &row.InstallID, &row.Namespace, &row.Category, &row.Status, &payloadHash, &packVersion)
	if err == sql.ErrNoRows {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, fmt.Errorf("query connector pack by namespace %s: %w", namespace, err)
	}
	if strings.TrimSpace(row.InstallID) == "" {
		return nil, false, fmt.Errorf("installed connector pack %s has blank PackInstallId (upgrade backfill not yet applied); refusing to surface as active connector pack", namespace)
	}
	row.PayloadHash = strings.TrimSpace(payloadHash.String)
	row.PackVersion = strings.TrimSpace(packVersion.String)
	return &row, true, nil
}

// ResolveActiveConnectorPackByInstall resolves one caller-selected active
// connector-pack generation. Unlike the namespace resolver it cannot observe a
// replacement generation and leave authorization to a read-after-filter step.
func ResolveActiveConnectorPackByInstall(ctx context.Context, db *sql.DB, namespace, packInstallID string) (*ConnectorPackRow, bool, error) {
	namespace = strings.TrimSpace(namespace)
	packInstallID = strings.TrimSpace(packInstallID)
	if namespace == "" || packInstallID == "" {
		return nil, false, nil
	}
	var row ConnectorPackRow
	var payloadHash, packVersion sql.NullString
	err := db.QueryRowContext(ctx, `
	SELECT Id, PackInstallId, Name, Category, Status, COALESCE(AppliedPayloadHash, ''), COALESCE(PackVersion, '')
	  FROM _System_InstalledPack
	 WHERE Name = ?
	   AND PackInstallId = ?
	   AND Category = ?
	   AND Status = ?
	   AND COALESCE(IsDeleted, 0) = 0
	   AND TRIM(COALESCE(AppliedPayloadHash, '')) <> ''
	   AND TRIM(COALESCE(DesiredPayloadHash, '')) <> ''
	   AND AppliedPayloadHash = DesiredPayloadHash
	   AND EXISTS (
	       SELECT 1
	         FROM _System_ConnectorCapabilityDeclaration d
	        WHERE d.InstalledPackName = _System_InstalledPack.Name
	          AND d.Namespace = _System_InstalledPack.Name
	          AND d.Kind = 'signature_verifier'
	          AND COALESCE(d.IsDeleted, 0) = 0
	   )
	   AND EXISTS (
	       SELECT 1
	         FROM _System_ConnectorCapabilityDeclaration d
	        WHERE d.InstalledPackName = _System_InstalledPack.Name
	          AND d.Namespace = _System_InstalledPack.Name
	          AND d.Kind = 'envelope_parser'
	          AND COALESCE(d.IsDeleted, 0) = 0
	   )
	 LIMIT 1`, namespace, packInstallID, string(ConnectorPack), InstalledPackStatusActive).Scan(
		&row.ID,
		&row.InstallID,
		&row.Namespace,
		&row.Category,
		&row.Status,
		&payloadHash,
		&packVersion,
	)
	if err == sql.ErrNoRows {
		return nil, false, nil
	}
	if err != nil {
		return nil, false, fmt.Errorf("query connector pack %s by install %s: %w", namespace, packInstallID, err)
	}
	row.PayloadHash = strings.TrimSpace(payloadHash.String)
	row.PackVersion = strings.TrimSpace(packVersion.String)
	return &row, true, nil
}

// ListConnectorPacksForInboundWorkers returns the connector-pack incarnations
// that must have inbound workers started at process boot.
func ListConnectorPacksForInboundWorkers(ctx context.Context, db *sql.DB) ([]ConnectorInboundWorkerPackRow, error) {
	query := strings.ReplaceAll(connectorInboundWorkerPackQuery, "__WORKITEM_BACKLOG_PREDICATE__", connectorInboundWorkerWorkItemBacklogPredicate)
	args := []any{InstalledPackStatusActive, string(ConnectorPack), InstalledPackStatusActive, InstalledPackStatusActive, InstalledPackStatusActive, InstalledPackStatusActive}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil && isMissingConnectorWorkItemSchema(err) {
		query = strings.ReplaceAll(connectorInboundWorkerPackQuery, "__WORKITEM_BACKLOG_PREDICATE__", "")
		rows, err = db.QueryContext(ctx, query, args...)
	}
	if err != nil {
		return nil, fmt.Errorf("list connector packs for inbound workers: %w", err)
	}
	defer rows.Close()

	out := make([]ConnectorInboundWorkerPackRow, 0)
	for rows.Next() {
		var (
			row       ConnectorInboundWorkerPackRow
			drainOnly int
		)
		if err := rows.Scan(&row.ID, &row.InstallID, &row.Namespace, &row.Category, &row.Status, &row.PayloadHash, &drainOnly); err != nil {
			return nil, fmt.Errorf("scan connector pack for inbound workers: %w", err)
		}
		row.DrainOnly = drainOnly != 0
		out = append(out, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate connector packs for inbound workers: %w", err)
	}
	return out, nil
}

const connectorInboundWorkerWorkItemBacklogPredicate = `
	                OR EXISTS (
	                    SELECT 1
	                      FROM _System_ConnectorWorkItem w
	                     WHERE w.InstalledPackId = p.PackInstallId
	                       AND w.State IN (` + connector_workqueue.NonTerminalWorkStateSQLList + `)
	                )`

const connectorInboundWorkerPackQuery = `
SELECT Id, PackInstallId, Name, Category, Status, COALESCE(AppliedPayloadHash, ''),
       CASE WHEN (
            Status = ?
            AND TRIM(COALESCE(AppliedPayloadHash, '')) <> ''
            AND TRIM(COALESCE(DesiredPayloadHash, '')) <> ''
            AND AppliedPayloadHash = DesiredPayloadHash
            AND EXISTS (
                SELECT 1
                  FROM _System_ConnectorCapabilityDeclaration d
                 WHERE d.InstalledPackName = p.Name
                   AND d.Namespace = p.Name
                   AND d.Kind = 'signature_verifier'
                   AND COALESCE(d.IsDeleted, 0) = 0
            )
            AND EXISTS (
                SELECT 1
                  FROM _System_ConnectorCapabilityDeclaration d
                 WHERE d.InstalledPackName = p.Name
                   AND d.Namespace = p.Name
                   AND d.Kind = 'envelope_parser'
                   AND COALESCE(d.IsDeleted, 0) = 0
            )
       ) THEN 0 ELSE 1 END AS DrainOnly
  FROM _System_InstalledPack p
 WHERE Category = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND TRIM(COALESCE(PackInstallId, '')) <> ''
   AND (
	        (
	            Status = ?
	            AND TRIM(COALESCE(AppliedPayloadHash, '')) <> ''
	            AND TRIM(COALESCE(DesiredPayloadHash, '')) <> ''
	            AND AppliedPayloadHash = DesiredPayloadHash
	            AND EXISTS (
                SELECT 1
                  FROM _System_ConnectorCapabilityDeclaration d
                 WHERE d.InstalledPackName = p.Name
                   AND d.Namespace = p.Name
                   AND d.Kind = 'signature_verifier'
                   AND COALESCE(d.IsDeleted, 0) = 0
            )
            AND EXISTS (
                SELECT 1
                  FROM _System_ConnectorCapabilityDeclaration d
                 WHERE d.InstalledPackName = p.Name
                   AND d.Namespace = p.Name
                   AND d.Kind = 'envelope_parser'
                   AND COALESCE(d.IsDeleted, 0) = 0
            )
	        )
        OR (
            Status <> ?
            AND NOT EXISTS (
                SELECT 1
                  FROM _System_InstalledPack active_pack
                 WHERE active_pack.Name = p.Name
                   AND active_pack.Category = p.Category
                   AND active_pack.Status = ?
                   AND active_pack.PackInstallId <> p.PackInstallId
                   AND COALESCE(active_pack.IsDeleted, 0) = 0
            )
            AND NOT EXISTS (
                SELECT 1
                  FROM _System_InstalledPack other_historical_pack
                 WHERE other_historical_pack.Name = p.Name
                   AND other_historical_pack.Category = p.Category
                   AND other_historical_pack.Status <> ?
                   AND other_historical_pack.PackInstallId <> p.PackInstallId
                   AND COALESCE(other_historical_pack.IsDeleted, 0) = 0
            )
	            AND (
	                EXISTS (
	                    SELECT 1
	                      FROM _System_ConnectorEvent e
	                     WHERE e.InstalledPackId = p.PackInstallId
	                       AND e.Status IN ('pending', 'failed', 'processing')
	                )
__WORKITEM_BACKLOG_PREDICATE__
	            )
	        )
	   )
	 ORDER BY Name, PackInstallId`

func isMissingConnectorWorkItemSchema(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "_system_connectorworkitem") &&
		(strings.Contains(msg, "1146") || strings.Contains(msg, "doesn't exist") || strings.Contains(msg, "no such table"))
}

// IsConnectorPackMidApply reports whether a connector pack with the given
// namespace exists but is not in a fully-applied steady state.
func IsConnectorPackMidApply(ctx context.Context, db *sql.DB, namespace string) (bool, error) {
	namespace = strings.TrimSpace(namespace)
	if namespace == "" {
		return false, nil
	}
	var (
		status    string
		installID string
		appliedH  string
		desiredH  string
	)
	err := db.QueryRowContext(ctx, `
SELECT Status,
       COALESCE(PackInstallId, ''),
       COALESCE(AppliedPayloadHash, ''),
       COALESCE(DesiredPayloadHash, '')
  FROM _System_InstalledPack
 WHERE Name = ? AND Category = ? AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`, namespace, string(ConnectorPack)).Scan(&status, &installID, &appliedH, &desiredH)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("query connector pack mid-apply state %s: %w", namespace, err)
	}
	status = strings.TrimSpace(status)
	installID = strings.TrimSpace(installID)
	appliedH = strings.TrimSpace(appliedH)
	desiredH = strings.TrimSpace(desiredH)
	steadyApplied := status == InstalledPackStatusActive &&
		installID != "" &&
		appliedH != "" &&
		desiredH != "" &&
		appliedH == desiredH
	return !steadyApplied, nil
}