package namedcredential

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

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

const (
	activeConnectorPackStatus   = "active"
	connectorPackCategoryString = "connector-pack"
)

// NewControlplanePackResolver returns a PackResolver backed by
// packstate.ResolveConnectorPackByNamespace.
func NewControlplanePackResolver(db *sql.DB) PackResolver {
	return &controlplanePackResolver{db: db}
}

type controlplanePackResolver struct {
	db *sql.DB
}

func (r *controlplanePackResolver) ResolveConnectorPackByNamespace(ctx context.Context, namespace string) (bool, error) {
	_, ok, err := packstate.ResolveConnectorPackByNamespace(ctx, r.db, namespace)
	if err != nil {
		return false, err
	}
	return ok, nil
}

func (r *controlplanePackResolver) ResolveDrainOnlyConnectorPackByInstall(ctx context.Context, namespace, installID string) (bool, error) {
	namespace = strings.TrimSpace(namespace)
	installID = strings.TrimSpace(installID)
	if namespace == "" || installID == "" {
		return false, nil
	}
	allowed, err := r.resolveDrainOnlyConnectorPackByInstall(ctx, namespace, installID, true)
	if err != nil && connectorWorkItemSchemaMissing(err) {
		return r.resolveDrainOnlyConnectorPackByInstall(ctx, namespace, installID, false)
	}
	return allowed, err
}

func (r *controlplanePackResolver) resolveDrainOnlyConnectorPackByInstall(ctx context.Context, namespace, installID string, includeWorkItems bool) (bool, error) {
	drainBacklogPredicate := `
   AND EXISTS (
	       SELECT 1
	         FROM _System_ConnectorEvent e
	        WHERE e.InstalledPackId = _System_InstalledPack.PackInstallId
	          AND e.Status IN ('pending', 'failed', 'processing')
	   )`
	if includeWorkItems {
		drainBacklogPredicate = `
   AND (
       EXISTS (
           SELECT 1
             FROM _System_ConnectorEvent e
            WHERE e.InstalledPackId = _System_InstalledPack.PackInstallId
              AND e.Status IN ('pending', 'failed', 'processing')
       )
       OR EXISTS (
           SELECT 1
             FROM _System_ConnectorWorkItem w
            WHERE w.InstalledPackId = _System_InstalledPack.PackInstallId
              AND w.State IN (` + connector_workqueue.NonTerminalWorkStateSQLList + `)
       )
   )`
	}

	var exists int
	err := r.db.QueryRowContext(ctx, `
	SELECT 1
	  FROM _System_InstalledPack
 WHERE Name = ?
   AND PackInstallId = ?
   AND Category = ?
   AND Status <> ?
   AND COALESCE(IsDeleted, 0) = 0
   AND NOT EXISTS (
       SELECT 1
         FROM _System_InstalledPack active_pack
        WHERE active_pack.Name = _System_InstalledPack.Name
          AND active_pack.Category = _System_InstalledPack.Category
          AND active_pack.Status = ?
          AND active_pack.PackInstallId <> _System_InstalledPack.PackInstallId
          AND COALESCE(active_pack.IsDeleted, 0) = 0
   )
   AND NOT EXISTS (
       SELECT 1
         FROM _System_InstalledPack other_historical_pack
        WHERE other_historical_pack.Name = _System_InstalledPack.Name
          AND other_historical_pack.Category = _System_InstalledPack.Category
          AND other_historical_pack.Status <> ?
          AND other_historical_pack.PackInstallId <> _System_InstalledPack.PackInstallId
          AND COALESCE(other_historical_pack.IsDeleted, 0) = 0
   )
`+drainBacklogPredicate+`
	 LIMIT 1`, namespace, installID, connectorPackCategoryString, activeConnectorPackStatus, activeConnectorPackStatus, activeConnectorPackStatus).Scan(&exists)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("query drain-only connector pack %s (%s): %w", namespace, installID, err)
	}
	return true, nil
}

func connectorWorkItemSchemaMissing(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	for _, marker := range []string{
		"unknown column",
		"unknown table",
		"doesn't exist",
		"does not exist",
		"no such table",
		"no such column",
		"error 1054",
		"error 1146",
	} {
		if strings.Contains(msg, marker) {
			return true
		}
	}
	return false
}