package packs

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

	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
)

const connectorInboundHandlerConfigIncarnationSweeperGroupKey = "connector_inbound_handler_config_incarnation"

func newConnectorInboundHandlerConfigIncarnationSweeper(packName string) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindConnectorInboundHandlerConfig,
		BusinessKey:   JoinComponentBusinessKey(packName, "@handler-config-incarnation-sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		applyGroupKey:     connectorInboundHandlerConfigIncarnationSweeperGroupKey,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyConnectorInboundHandlerConfigIncarnationSweep(ctx, deps.DB, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyConnectorInboundHandlerConfigIncarnationSweep(ctx, deps.DB, packName)
		},
	}
}

func applyConnectorInboundHandlerConfigIncarnationSweep(ctx context.Context, db *sql.DB, packName string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	candidateGuard := ""
	args := []any{packName, packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		candidateGuard = " AND InstalledPackId <> ?"
		args = append(args, candidateInstallID)
	}
	_, err := db.ExecContext(ctx, `
UPDATE _System_ConnectorInboundHandlerConfig
   SET IsDeleted = 1, LastModifiedDate = NOW()
 WHERE Namespace = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND InstalledPackId NOT IN (
       SELECT PackInstallId
         FROM _System_InstalledPack
        WHERE Name = ?
          AND COALESCE(IsDeleted, 0) = 0
          AND Status IN ('active', 'installing', 'removing')
          AND COALESCE(PackInstallId, '') <> ''
	   )`+candidateGuard,
		args...,
	)
	if err != nil {
		return fmt.Errorf("connector inbound handler config incarnation sweep for pack %s: %w", packName, err)
	}
	return nil
}

func verifyConnectorInboundHandlerConfigIncarnationSweep(ctx context.Context, db *sql.DB, packName string) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	candidateGuard := ""
	args := []any{packName, packName}
	if candidateInstallID, ok := offlinecandidateauthority.StoppedSourceSetPackInstallID(ctx, packName); ok {
		candidateGuard = " AND c.InstalledPackId <> ?"
		args = append(args, candidateInstallID)
	}
	var count int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorInboundHandlerConfig c
 WHERE c.Namespace = ?
   AND COALESCE(c.IsDeleted, 0) = 0
   AND c.InstalledPackId NOT IN (
       SELECT p.PackInstallId
         FROM _System_InstalledPack p
        WHERE p.Name = ?
          AND COALESCE(p.IsDeleted, 0) = 0
          AND p.Status IN ('active', 'installing', 'removing')
          AND COALESCE(p.PackInstallId, '') <> ''
	   )`+candidateGuard,
		args...,
	).Scan(&count); err != nil {
		return fmt.Errorf("verify connector inbound handler config incarnation sweep for pack %s: %w", packName, err)
	}
	if count > 0 {
		return fmt.Errorf("connector inbound handler config incarnation sweep for pack %s has %d stale live row(s)", packName, count)
	}
	return nil
}