package controlplane

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

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

func connectorInboundProjectionSnapshotID(ownerPack, ownerPackInstallID, companionPack, companionPackInstallID, objectKind, sourceField string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(ownerPack) + "\x00" +
		strings.TrimSpace(ownerPackInstallID) + "\x00" +
		strings.TrimSpace(companionPack) + "\x00" +
		strings.TrimSpace(companionPackInstallID) + "\x00" +
		strings.TrimSpace(objectKind) + "\x00" +
		strings.TrimSpace(sourceField)))
	return "cips_" + hex.EncodeToString(sum[:])
}

func refreshConnectorInboundProjectionSnapshotsAfterPackActivated(ctx context.Context, db execQuerier, pack packs.Pack, now time.Time) error {
	switch pack.Category() {
	case packs.ConnectorPack:
		return refreshConnectorInboundProjectionSnapshotsForOwnerPack(ctx, db, pack.Name(), now)
	case packs.AppPack:
		owners, err := listActiveConnectorProjectionOwnersForCompanion(ctx, db, pack.Name())
		if err != nil {
			return err
		}
		for _, owner := range owners {
			if err := refreshConnectorInboundProjectionSnapshotsForOwnerPack(ctx, db, owner, now); err != nil {
				return err
			}
		}
	}
	return nil
}

func listActiveConnectorProjectionOwnersForCompanion(ctx context.Context, db execQuerier, companionPack string) ([]string, error) {
	companionPack = strings.TrimSpace(companionPack)
	if companionPack == "" {
		return nil, nil
	}
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT d.DependsOnPackName
  FROM _System_InstalledPackDependency d
  JOIN _System_InstalledPack cp
    ON cp.Name = d.DependsOnPackName
   AND cp.Category = 'connector-pack'
   AND cp.Status = 'active'
   AND COALESCE(cp.IsDeleted, 0) = 0
 WHERE d.DependentPackName = ?
   AND COALESCE(d.IsDeleted, 0) = 0
 ORDER BY d.DependsOnPackName`, companionPack)
	if err != nil {
		return nil, fmt.Errorf("list active connector projection owners for companion %s: %w", companionPack, err)
	}
	defer rows.Close()

	var owners []string
	for rows.Next() {
		var owner string
		if err := rows.Scan(&owner); err != nil {
			return nil, fmt.Errorf("scan active connector projection owner for companion %s: %w", companionPack, err)
		}
		owner = strings.TrimSpace(owner)
		if owner != "" {
			owners = append(owners, owner)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate active connector projection owners for companion %s: %w", companionPack, err)
	}
	return owners, nil
}

func activeConnectorProjectionOwnerState(ctx context.Context, db execQuerier, ownerPack string) (string, bool, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	if ownerPack == "" {
		return "", false, nil
	}
	var ownerInstallID string
	err := db.QueryRowContext(ctx, `
SELECT COALESCE(PackInstallId, '')
  FROM _System_InstalledPack
 WHERE Name = ?
   AND Category = 'connector-pack'
   AND Status = 'active'
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`, ownerPack).Scan(&ownerInstallID)
	if err == sql.ErrNoRows {
		return "", false, nil
	}
	if err != nil {
		return "", false, fmt.Errorf("read active connector projection owner %s: %w", ownerPack, err)
	}
	ownerInstallID = strings.TrimSpace(ownerInstallID)
	if ownerInstallID == "" {
		return "", false, fmt.Errorf("active connector projection owner %s has blank PackInstallId", ownerPack)
	}
	var projectionProfileCount int
	if err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
  FROM _System_ConnectorCalloutProfile
 WHERE OwnerPack = ?
   AND InstalledPackId = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND TRIM(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(BatchSpec, '$.projectionBodyField')), '')) <> ''`,
		ownerPack,
		ownerInstallID,
	).Scan(&projectionProfileCount); err != nil {
		return "", false, fmt.Errorf("read projection-enabled profile count for %s (%s): %w", ownerPack, ownerInstallID, err)
	}
	return ownerInstallID, projectionProfileCount > 0, nil
}

// RefreshConnectorInboundProjectionSnapshotsForActiveConnectorPacks repairs the
// runtime projection inventory used by connector inbound batch enrichment. Pack
// apply refreshes these rows at lifecycle boundaries; startup/reconcile calls
// this as a convergence guard for active source-backed packs that predate the
// snapshot table or missed a prior lifecycle refresh.
func RefreshConnectorInboundProjectionSnapshotsForActiveConnectorPacks(ctx context.Context, db *sql.DB, now time.Time) error {
	if db == nil {
		return nil
	}
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT cp.Name
  FROM _System_InstalledPack cp
 WHERE cp.Category = 'connector-pack'
   AND cp.Status = 'active'
   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_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')), '')) <> ''
   )
 ORDER BY cp.Name`)
	if err != nil {
		return fmt.Errorf("refresh connector inbound projection snapshots for active connector packs: list owners: %w", err)
	}
	defer rows.Close()

	var owners []string
	for rows.Next() {
		var owner string
		if err := rows.Scan(&owner); err != nil {
			return fmt.Errorf("refresh connector inbound projection snapshots for active connector packs: scan owner: %w", err)
		}
		owner = strings.TrimSpace(owner)
		if owner != "" {
			owners = append(owners, owner)
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("refresh connector inbound projection snapshots for active connector packs: iterate owners: %w", err)
	}
	for _, owner := range owners {
		if err := refreshConnectorInboundProjectionSnapshotsForOwnerPack(ctx, db, owner, now); err != nil {
			return err
		}
	}
	return nil
}

func listConnectorProjectionBacklogCompanionInstallIDs(ctx context.Context, db execQuerier, ownerInstallID string) (map[string]struct{}, error) {
	ownerInstallID = strings.TrimSpace(ownerInstallID)
	if ownerInstallID == "" {
		return nil, nil
	}
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT COALESCE(CompanionPackInstallId, '')
  FROM _System_ConnectorEvent
 WHERE InstalledPackId = ?
   AND Status IN ('pending', 'failed', 'processing')
   AND TRIM(COALESCE(CompanionPackInstallId, '')) <> ''
 ORDER BY COALESCE(CompanionPackInstallId, '')`, ownerInstallID)
	if err != nil {
		return nil, fmt.Errorf("list connector projection backlog companion installs for %s: %w", ownerInstallID, err)
	}
	defer rows.Close()

	out := make(map[string]struct{})
	for rows.Next() {
		var companionInstallID string
		if err := rows.Scan(&companionInstallID); err != nil {
			return nil, fmt.Errorf("scan connector projection backlog companion install for %s: %w", ownerInstallID, err)
		}
		companionInstallID = strings.TrimSpace(companionInstallID)
		if companionInstallID != "" {
			out[companionInstallID] = struct{}{}
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate connector projection backlog companion installs for %s: %w", ownerInstallID, err)
	}
	return out, nil
}

type connectorInboundProjectionSnapshotCandidate struct {
	rowID              string
	companionPack      string
	companionInstallID string
	objectKind         string
	sourceField        string
}

func listLiveConnectorInboundProjectionSnapshotCandidates(ctx context.Context, db execQuerier, ownerPack string, ownerInstallID string) ([]connectorInboundProjectionSnapshotCandidate, map[string]struct{}, error) {
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT d.DependentPackName, COALESCE(d.DependentPackInstallId, ''), m.SourceObject, m.SourceField
  FROM _System_InstalledPackDependency d
  JOIN _System_InstalledPack p
    ON p.Name = d.DependentPackName
   AND p.PackInstallId = d.DependentPackInstallId
   AND p.Category = 'app-pack'
   AND p.Status = 'active'
   AND TRIM(COALESCE(p.AppliedPayloadHash, '')) <> ''
   AND TRIM(COALESCE(p.DesiredPayloadHash, '')) <> ''
   AND p.AppliedPayloadHash = p.DesiredPayloadHash
   AND COALESCE(p.IsDeleted, 0) = 0
  JOIN _System_ObjectFieldMapping m
    ON m.Source = d.DependentPackName
 WHERE d.DependsOnPackName = ?
   AND COALESCE(d.IsDeleted, 0) = 0
   AND COALESCE(m.IsActive, 0) = 1
   AND COALESCE(m.IsDeleted, 0) = 0
   AND m.Direction IN ('inbound', 'bidirectional')
   AND TRIM(COALESCE(m.SourceObject, '')) <> ''
   AND TRIM(COALESCE(m.SourceField, '')) <> ''
 ORDER BY d.DependentPackName, COALESCE(d.DependentPackInstallId, ''), m.SourceObject, m.SourceField`, ownerPack)
	if err != nil {
		return nil, nil, fmt.Errorf("refresh connector inbound projection snapshots for %s (%s): query live companion mappings: %w", ownerPack, ownerInstallID, err)
	}
	defer rows.Close()

	keepIDs := make(map[string]struct{})
	var candidates []connectorInboundProjectionSnapshotCandidate
	for rows.Next() {
		var companionPack, companionInstallID, objectKind, sourceField string
		if err := rows.Scan(&companionPack, &companionInstallID, &objectKind, &sourceField); err != nil {
			return nil, nil, fmt.Errorf("refresh connector inbound projection snapshots for %s (%s): scan live companion mappings: %w", ownerPack, ownerInstallID, err)
		}
		companionPack = strings.TrimSpace(companionPack)
		companionInstallID = strings.TrimSpace(companionInstallID)
		objectKind = strings.TrimSpace(objectKind)
		sourceField = strings.TrimSpace(sourceField)
		if companionPack == "" || companionInstallID == "" || objectKind == "" || sourceField == "" {
			continue
		}
		rowID := connectorInboundProjectionSnapshotID(ownerPack, ownerInstallID, companionPack, companionInstallID, objectKind, sourceField)
		keepIDs[rowID] = struct{}{}
		candidates = append(candidates, connectorInboundProjectionSnapshotCandidate{
			rowID:              rowID,
			companionPack:      companionPack,
			companionInstallID: companionInstallID,
			objectKind:         objectKind,
			sourceField:        sourceField,
		})
	}
	if err := rows.Err(); err != nil {
		return nil, nil, fmt.Errorf("refresh connector inbound projection snapshots for %s (%s): iterate live companion mappings: %w", ownerPack, ownerInstallID, err)
	}
	return candidates, keepIDs, nil
}

func refreshConnectorInboundProjectionSnapshotsForOwnerPack(ctx context.Context, db execQuerier, ownerPack string, now time.Time) error {
	ownerPack = strings.TrimSpace(ownerPack)
	if ownerPack == "" {
		return nil
	}
	ownerInstallID, projectionEnabled, err := activeConnectorProjectionOwnerState(ctx, db, ownerPack)
	if err != nil {
		return err
	}
	if !projectionEnabled {
		return nil
	}
	backlogCompanionInstallIDs, err := listConnectorProjectionBacklogCompanionInstallIDs(ctx, db, ownerInstallID)
	if err != nil {
		return err
	}

	candidates, keepIDs, err := listLiveConnectorInboundProjectionSnapshotCandidates(ctx, db, ownerPack, ownerInstallID)
	if err != nil {
		return err
	}
	for _, candidate := range candidates {
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_ConnectorInboundProjectionSnapshot
  (Id, OwnerPackName, OwnerPackInstallId, CompanionPackName, CompanionPackInstallId, ObjectKind, SourceField, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?)
	ON DUPLICATE KEY UPDATE
	  LastModifiedDate = CASE
	    WHEN COALESCE(IsDeleted, 0) <> 0
	      OR CompanionPackName <> VALUES(CompanionPackName)
	      OR ObjectKind <> VALUES(ObjectKind)
	      OR SourceField <> VALUES(SourceField)
	    THEN VALUES(LastModifiedDate)
	    ELSE LastModifiedDate
	  END,
	  CompanionPackName = VALUES(CompanionPackName),
	  ObjectKind = VALUES(ObjectKind),
	  SourceField = VALUES(SourceField),
	  IsDeleted = 0`,
			candidate.rowID,
			ownerPack,
			ownerInstallID,
			candidate.companionPack,
			candidate.companionInstallID,
			candidate.objectKind,
			candidate.sourceField,
			now,
			now,
		); err != nil {
			return fmt.Errorf("refresh connector inbound projection snapshots for %s (%s): upsert %s/%s %s.%s: %w", ownerPack, ownerInstallID, candidate.companionPack, candidate.companionInstallID, candidate.objectKind, candidate.sourceField, err)
		}
	}
	if err := softDeleteStaleConnectorInboundProjectionSnapshots(ctx, db, ownerInstallID, backlogCompanionInstallIDs, keepIDs, now); err != nil {
		return err
	}
	return nil
}

func softDeleteStaleConnectorInboundProjectionSnapshots(ctx context.Context, db execQuerier, ownerInstallID string, backlogCompanionInstallIDs map[string]struct{}, keepIDs map[string]struct{}, now time.Time) error {
	ownerInstallID = strings.TrimSpace(ownerInstallID)
	if ownerInstallID == "" {
		return nil
	}
	args := []any{now, ownerInstallID}
	query := `
UPDATE _System_ConnectorInboundProjectionSnapshot
   SET IsDeleted = 1, LastModifiedDate = ?
 WHERE OwnerPackInstallId = ?
   AND COALESCE(IsDeleted, 0) = 0`
	if len(keepIDs) > 0 {
		keepPlaceholders := make([]string, 0, len(keepIDs))
		for rowID := range keepIDs {
			keepPlaceholders = append(keepPlaceholders, "?")
			args = append(args, rowID)
		}
		query += `
   AND Id NOT IN (` + strings.Join(keepPlaceholders, ", ") + `)`
	}
	if len(backlogCompanionInstallIDs) > 0 {
		retainedPlaceholders := make([]string, 0, len(backlogCompanionInstallIDs))
		for companionInstallID := range backlogCompanionInstallIDs {
			retainedPlaceholders = append(retainedPlaceholders, "?")
			args = append(args, companionInstallID)
		}
		query += `
   AND CompanionPackInstallId NOT IN (` + strings.Join(retainedPlaceholders, ", ") + `)`
	}
	if _, err := db.ExecContext(ctx, query, args...); err != nil {
		return fmt.Errorf("soft-delete stale connector inbound projection snapshots for install %s: %w", ownerInstallID, err)
	}
	return nil
}