package controlplane

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

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	"github.com/pingcap-inc/nexuscrm-connectors/interfaces/rest/connector_webhooks"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

type ConnectorPushLegacyMirrorBackfillResult struct {
	ContractsScanned int
	RowsScanned      int
	RowsAdopted      int
	RowsUnmatched    int
}

const connectorPushBackfillObjectMetadataQuery = `
SELECT o.ApiName, COALESCE(o.ReadSource, '')
  FROM _System_Object o
 WHERE o.ApiName = ?
   AND COALESCE(o.IsDeleted, 0) = 0
 LIMIT 1`

const connectorPushBackfillFieldMetadataQuery = `
SELECT f.ApiName, COALESCE(f.PhysicalName, '')
  FROM _System_Field f
  JOIN _System_Object o ON o.Id = f.ObjectId
 WHERE o.ApiName = ?
   AND COALESCE(o.IsDeleted, 0) = 0
   AND f.ApiName = ?
   AND COALESCE(f.IsDeleted, 0) = 0
 LIMIT 1`

const connectorPushBackfillActivePackInfoQuery = `
SELECT COALESCE(PackInstallId, ''), PackVersion
  FROM _System_InstalledPack
 WHERE Name = ?
   AND Category = ?
   AND Status = 'active'
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`

var (
	connectorPushBackfillTableIdentifierRE = regexp.MustCompile(constants.TableNamePattern)
	connectorPushBackfillFieldIdentifierRE = regexp.MustCompile(constants.FieldNamePattern)
)

func BackfillConnectorPushLegacyMirrorAdoptions(ctx context.Context, db *sql.DB) (ConnectorPushLegacyMirrorBackfillResult, error) {
	return BackfillConnectorPushLegacyMirrorAdoptionsWithComponentKindRegistrationSet(ctx, db, packs.ComponentKindRegistrationSet{})
}

func BackfillConnectorPushLegacyMirrorAdoptionsWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, registrations packs.ComponentKindRegistrationSet) (ConnectorPushLegacyMirrorBackfillResult, error) {
	if db == nil {
		return ConnectorPushLegacyMirrorBackfillResult{}, fmt.Errorf("connector push legacy mirror backfill requires a database")
	}
	store := automation.NewSQLConnectorPushDurableStore(db)
	var total ConnectorPushLegacyMirrorBackfillResult
	contracts, err := loadConnectorPushLegacyMirrorBackfillContractsWithComponentKindRegistrationSet(ctx, db, registrations)
	if err != nil {
		return total, err
	}
	for _, contract := range contracts {
		result, err := backfillConnectorPushLegacyMirrorContract(ctx, db, contract, func(ctx context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
			profile, err := loadConnectorPushBackfillProfile(ctx, db, contract)
			if err != nil {
				return nil, nil, err
			}
			if profile == nil {
				return nil, nil, nil
			}
			mappings, err := packs.ResolvePackConnectorFieldMapping(ctx, db, connector.ResolveQuery{
				Source:       contract.ConnectorOwnerKey,
				SourceObject: contract.SourceObject,
				TargetObject: contract.TargetObject,
				ActionType:   contract.ActionType,
				Direction:    connector.DirectionOutbound,
			})
			if err != nil {
				return nil, nil, err
			}
			return mappings, profile, nil
		}, func(ctx context.Context, evidence automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
			_, adopted, err := automation.AdoptConnectorPushLegacyMirrorEvidence(ctx, store, evidence)
			return adopted, err
		})
		if err != nil {
			return total, err
		}
		total.ContractsScanned += result.ContractsScanned
		total.RowsScanned += result.RowsScanned
		total.RowsAdopted += result.RowsAdopted
		total.RowsUnmatched += result.RowsUnmatched
	}
	return total, nil
}

func loadConnectorPushLegacyMirrorBackfillContracts(ctx context.Context, db *sql.DB) ([]connector.ConnectorPushDurableMetadataContract, error) {
	return loadConnectorPushLegacyMirrorBackfillContractsWithComponentKindRegistrationSet(ctx, db, packs.ComponentKindRegistrationSet{})
}

func loadConnectorPushLegacyMirrorBackfillContractsWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, registrations packs.ComponentKindRegistrationSet) ([]connector.ConnectorPushDurableMetadataContract, error) {
	rows, err := db.QueryContext(ctx, `
SELECT Name, Category, COALESCE(ManifestFormatVersion, ''), COALESCE(SourceKind, ''),
       COALESCE(SourceURI, ''), COALESCE(SourceRevision, ''), COALESCE(AppliedPayloadHash, ''),
       COALESCE(DesiredPayloadHash, '')
  FROM _System_InstalledPack
 WHERE Status = 'active'
   AND Category IN (?, ?)
   AND COALESCE(IsDeleted, 0) = 0
   AND COALESCE(AppliedPayloadHash, '') <> ''
   AND COALESCE(AppliedPayloadHash, '') = COALESCE(DesiredPayloadHash, '')
 ORDER BY Name`, string(packs.AppPack), string(packs.ConnectorPack))
	if err != nil {
		return nil, fmt.Errorf("query active installed packs for connector push mirror backfill: %w", err)
	}
	defer rows.Close()

	var contracts []connector.ConnectorPushDurableMetadataContract
	for rows.Next() {
		var name, category, manifestVersion, sourceKind, sourceURI, sourceRevision, appliedHash, desiredHash string
		if err := rows.Scan(&name, &category, &manifestVersion, &sourceKind, &sourceURI, &sourceRevision, &appliedHash, &desiredHash); err != nil {
			return nil, fmt.Errorf("scan installed pack for connector push mirror backfill: %w", err)
		}
		sourceBacked, err := resolveConnectorPushBackfillInstalledPackWithComponentKindRegistrationSet(ctx, db, name, category, manifestVersion, sourceKind, sourceURI, sourceRevision, appliedHash, registrations)
		if err != nil {
			return nil, fmt.Errorf("resolve installed pack %s for connector push mirror backfill: %w", strings.TrimSpace(name), err)
		}
		if sourceBacked == nil {
			continue
		}
		contracts = append(contracts, sourceBacked.ConnectorPushContractDeclarations()...)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate installed packs for connector push mirror backfill: %w", err)
	}
	return contracts, nil
}

func resolveConnectorPushBackfillInstalledPack(ctx context.Context, db *sql.DB, name, category, manifestVersion, sourceKind, sourceURI, sourceRevision, appliedHash string) (*packs.SourceBackedPack, error) {
	return resolveConnectorPushBackfillInstalledPackWithComponentKindRegistrationSet(
		ctx, db, name, category, manifestVersion, sourceKind, sourceURI, sourceRevision, appliedHash, packs.ComponentKindRegistrationSet{},
	)
}

func resolveConnectorPushBackfillInstalledPackWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, name, category, manifestVersion, sourceKind, sourceURI, sourceRevision, appliedHash string, registrations packs.ComponentKindRegistrationSet) (*packs.SourceBackedPack, error) {
	name = strings.TrimSpace(name)
	category = strings.TrimSpace(category)
	manifestVersion = strings.TrimSpace(manifestVersion)
	sourceRevision = strings.TrimSpace(sourceRevision)
	appliedHash = strings.TrimSpace(appliedHash)
	if name == "" || appliedHash == "" {
		return nil, nil
	}
	resolvedSourceKind, resolvedSourceURI, ok := connectorPushBackfillPackSourceIdentity(name, sourceKind, sourceURI)
	if !ok {
		return nil, fmt.Errorf("active installed pack is missing source identity")
	}
	sourceOpts := packSourceOptionsWithSchema(ctx, db)
	var (
		pack *packs.SourceBackedPack
		err  error
	)
	if sourceRevision == "" {
		pack, err = packs.ResolveInstalledSourceBackedPackForInstalledRowAtSourceRevisionWithPackSourceOptionsAndComponentKindRegistrationSet(
			name,
			packs.Category(category),
			manifestVersion,
			resolvedSourceKind,
			resolvedSourceURI,
			sourceRevision,
			sourceOpts,
			registrations,
		)
		if err != nil {
			return nil, fmt.Errorf("resolve active installed pack from current source: %w", err)
		}
	} else {
		pack, err = packs.ResolveInstalledSourceBackedPackForInstalledRowAtSourceRevisionWithPackSourceOptionsAndComponentKindRegistrationSet(
			name,
			packs.Category(category),
			manifestVersion,
			resolvedSourceKind,
			resolvedSourceURI,
			sourceRevision,
			sourceOpts,
			registrations,
		)
		if err != nil {
			return nil, fmt.Errorf("resolve active installed pack from stored source identity: %w", err)
		}
	}
	if _, err := packs.ValidateStoredPackPayloadHash(pack, appliedHash); err != nil {
		return nil, fmt.Errorf("validate stored pack payload hash: %w", err)
	}
	return pack, nil
}

func connectorPushBackfillPackSourceIdentity(name, sourceKind, sourceURI string) (string, string, bool) {
	name = strings.TrimSpace(name)
	sourceKind = strings.TrimSpace(sourceKind)
	sourceURI = strings.TrimSpace(sourceURI)
	if name == "" || sourceKind == "" || sourceURI == "" {
		return "", "", false
	}
	return sourceKind, sourceURI, true
}

func loadConnectorPushBackfillProfile(ctx context.Context, db *sql.DB, contract connector.ConnectorPushDurableMetadataContract) (*automation.ConnectorPushProfile, error) {
	requesterOwnerPack := strings.TrimSpace(contract.ConnectorOwnerKey)
	profileOwnerPack := strings.TrimSpace(contract.ProfileOwnerPack)
	if profileOwnerPack == "" {
		profileOwnerPack = requesterOwnerPack
	}
	profileName := strings.TrimSpace(contract.ProfileName)
	authorized, err := authorizeConnectorPushBackfillProfileDependency(ctx, db, requesterOwnerPack, profileOwnerPack)
	if err != nil {
		return nil, fmt.Errorf("connector push legacy mirror backfill: authorize profile dependency %s -> %s/%s: %w", requesterOwnerPack, profileOwnerPack, profileName, err)
	}
	if !authorized {
		return nil, nil
	}
	profile, err := connector_callout.LoadActiveProfile(ctx, db, profileOwnerPack, profileName)
	if err != nil {
		return nil, fmt.Errorf("connector push legacy mirror backfill: load profile %s/%s: %w", profileOwnerPack, profileName, err)
	}
	if profile == nil {
		return nil, fmt.Errorf("connector push legacy mirror backfill: profile %s/%s not found", profileOwnerPack, profileName)
	}
	return &automation.ConnectorPushProfile{
		OwnerPack:            profile.OwnerPack,
		ProfileName:          profile.ProfileName,
		CreatedPackInstallID: profile.InstalledPackID,
		CredentialRef:        profile.CredentialRef,
		EndpointClass:        profile.EndpointClass,
		RetryPolicy: automation.ConnectorPushRetryPolicy{
			MaxAttempts:          profile.RetryPolicy.MaxAttempts,
			BaseDelay:            profile.RetryPolicy.BaseDelay,
			Multiplier:           profile.RetryPolicy.Multiplier,
			MaxDelay:             profile.RetryPolicy.MaxDelay,
			RetryableStatusCodes: profile.RetryPolicy.RetryableStatusCodes,
		},
	}, nil
}

func authorizeConnectorPushBackfillProfileDependency(ctx context.Context, db *sql.DB, requesterOwnerPack, profileOwnerPack string) (bool, error) {
	requesterOwnerPack = strings.TrimSpace(requesterOwnerPack)
	profileOwnerPack = strings.TrimSpace(profileOwnerPack)
	if requesterOwnerPack == "" {
		return false, fmt.Errorf("connector push profile dependency: requester owner pack is required")
	}
	if profileOwnerPack == "" {
		return false, fmt.Errorf("connector push profile dependency: profile owner pack is required")
	}
	if strings.EqualFold(requesterOwnerPack, profileOwnerPack) {
		return true, nil
	}
	if db == nil {
		return false, fmt.Errorf("connector push profile dependency: db is required for cross-pack profile %s -> %s", requesterOwnerPack, profileOwnerPack)
	}

	requesterInstallID, _, err := connectorPushBackfillActivePackInfoByCategory(ctx, db, requesterOwnerPack, string(packs.AppPack))
	if err != nil {
		if connectorPushBackfillUnauthorizedProfileDependency(err) {
			return false, nil
		}
		return false, err
	}
	_, providerVersion, err := connectorPushBackfillActivePackInfoByCategory(ctx, db, profileOwnerPack, string(packs.ConnectorPack))
	if err != nil {
		if connectorPushBackfillUnauthorizedProfileDependency(err) {
			return false, nil
		}
		return false, err
	}
	if err := connector_webhooks.CheckCompanionConsent(ctx, db, requesterOwnerPack, requesterInstallID, profileOwnerPack, providerVersion); err != nil {
		if connectorPushBackfillUnauthorizedProfileDependency(err) {
			return false, nil
		}
		return false, fmt.Errorf("connector push profile dependency: %w", err)
	}
	return true, nil
}

func connectorPushBackfillActivePackInfoByCategory(ctx context.Context, db *sql.DB, packName, category string) (string, string, error) {
	var installID string
	var version sql.NullString
	err := db.QueryRowContext(ctx, connectorPushBackfillActivePackInfoQuery, packName, category).Scan(&installID, &version)
	if err == sql.ErrNoRows {
		return "", "", fmt.Errorf("connector push profile dependency: pack %q (%s) is not active", packName, category)
	}
	if err != nil {
		return "", "", fmt.Errorf("connector push profile dependency: query active pack %q (%s): %w", packName, category, err)
	}
	installID = strings.TrimSpace(installID)
	if installID == "" {
		return "", "", fmt.Errorf("connector push profile dependency: pack %q (%s) has blank PackInstallId", packName, category)
	}
	resolvedVersion := ""
	if version.Valid {
		resolvedVersion = strings.TrimSpace(version.String)
	}
	return installID, resolvedVersion, nil
}

func connectorPushBackfillUnauthorizedProfileDependency(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(strings.TrimSpace(err.Error()))
	if strings.Contains(msg, "query active pack") || strings.Contains(msg, "query dependency row") {
		return false
	}
	return strings.Contains(msg, " is not active") ||
		strings.Contains(msg, "has blank packinstallid") ||
		strings.HasPrefix(msg, "consent missing:") ||
		strings.HasPrefix(msg, "version unsatisfied:")
}

type connectorPushLegacyMirrorAdopter func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error)
type connectorPushLegacyMirrorResolver func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error)

func backfillConnectorPushLegacyMirrorContract(ctx context.Context, db *sql.DB, contract connector.ConnectorPushDurableMetadataContract, resolve connectorPushLegacyMirrorResolver, adopt connectorPushLegacyMirrorAdopter) (ConnectorPushLegacyMirrorBackfillResult, error) {
	var result ConnectorPushLegacyMirrorBackfillResult
	result.ContractsScanned = 1
	if strings.TrimSpace(contract.RemoteIDField) == "" || strings.TrimSpace(contract.RemoteIDProvenanceField) == "" {
		return result, nil
	}
	if adopt == nil {
		return result, fmt.Errorf("connector push legacy mirror backfill adopter is not wired")
	}
	if resolve == nil {
		return result, fmt.Errorf("connector push legacy mirror backfill resolver is not wired")
	}
	readPath, err := resolveConnectorPushBackfillReadPath(ctx, db, contract)
	if err != nil {
		return result, err
	}
	query := fmt.Sprintf(
		"SELECT `Id`, %s, %s FROM %s WHERE COALESCE(%s, '') <> '' AND COALESCE(%s, '') <> ''",
		readPath.remoteIDField, readPath.provenanceField, readPath.table, readPath.remoteIDField, readPath.provenanceField,
	)
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		return result, fmt.Errorf("query connector push legacy mirror rows for %s: %w", strings.TrimSpace(contract.SourceObject), err)
	}
	defer rows.Close()

	var (
		mappings []connector.ConnectorFieldMappingInput
		profile  *automation.ConnectorPushProfile
		loaded   bool
	)
	for rows.Next() {
		var sourceRecordID, remoteID, provenance sql.NullString
		if err := rows.Scan(&sourceRecordID, &remoteID, &provenance); err != nil {
			return result, fmt.Errorf("scan connector push legacy mirror row for %s: %w", strings.TrimSpace(contract.SourceObject), err)
		}
		if strings.TrimSpace(sourceRecordID.String) == "" {
			continue
		}
		if !loaded {
			mappings, profile, err = resolve(ctx)
			if err != nil {
				return result, fmt.Errorf("resolve connector push legacy mirror contract %s/%s: %w", strings.TrimSpace(contract.SourceObject), strings.TrimSpace(contract.ActionSlotID), err)
			}
			if profile == nil {
				return result, nil
			}
			loaded = true
		}
		result.RowsScanned++
		adopted, err := adopt(ctx, automation.ConnectorPushLegacyMirrorEvidence{
			Contract:           contract,
			SourceRecordID:     sourceRecordID.String,
			RemoteID:           remoteID.String,
			RemoteIDProvenance: provenance.String,
			Mappings:           mappings,
			Profile:            profile,
		})
		if err != nil {
			return result, fmt.Errorf("adopt connector push legacy mirror row %s/%s: %w", strings.TrimSpace(contract.SourceObject), strings.TrimSpace(sourceRecordID.String), err)
		}
		if adopted {
			result.RowsAdopted++
		} else {
			result.RowsUnmatched++
		}
	}
	if err := rows.Err(); err != nil {
		return result, fmt.Errorf("iterate connector push legacy mirror rows for %s: %w", strings.TrimSpace(contract.SourceObject), err)
	}
	return result, nil
}

type connectorPushBackfillReadPath struct {
	table           string
	remoteIDField   string
	provenanceField string
}

func resolveConnectorPushBackfillReadPath(ctx context.Context, db *sql.DB, contract connector.ConnectorPushDurableMetadataContract) (connectorPushBackfillReadPath, error) {
	objectMeta, err := loadConnectorPushBackfillObjectMetadata(ctx, db, contract.SourceObject)
	if err != nil {
		return connectorPushBackfillReadPath{}, err
	}
	tableName, err := objectMeta.ResolveReadSourceTable()
	if err != nil {
		return connectorPushBackfillReadPath{}, fmt.Errorf("connector push legacy mirror backfill resolve source table for %s: %w", strings.TrimSpace(contract.SourceObject), err)
	}
	table, err := connectorPushBackfillQuoteResolvedIdentifier(tableName, connectorPushBackfillTableIdentifierRE, "physical table")
	if err != nil {
		return connectorPushBackfillReadPath{}, err
	}
	remoteIDColumn, err := loadConnectorPushBackfillFieldColumn(ctx, db, contract.SourceObject, contract.RemoteIDField)
	if err != nil {
		return connectorPushBackfillReadPath{}, err
	}
	remoteIDField, err := connectorPushBackfillQuoteResolvedIdentifier(remoteIDColumn, connectorPushBackfillFieldIdentifierRE, "physical column")
	if err != nil {
		return connectorPushBackfillReadPath{}, err
	}
	provenanceColumn, err := loadConnectorPushBackfillFieldColumn(ctx, db, contract.SourceObject, contract.RemoteIDProvenanceField)
	if err != nil {
		return connectorPushBackfillReadPath{}, err
	}
	provenanceField, err := connectorPushBackfillQuoteResolvedIdentifier(provenanceColumn, connectorPushBackfillFieldIdentifierRE, "physical column")
	if err != nil {
		return connectorPushBackfillReadPath{}, err
	}
	return connectorPushBackfillReadPath{
		table:           table,
		remoteIDField:   remoteIDField,
		provenanceField: provenanceField,
	}, nil
}

func loadConnectorPushBackfillObjectMetadata(ctx context.Context, db *sql.DB, objectAPIName string) (*models.ObjectMetadata, error) {
	objectAPIName = strings.TrimSpace(objectAPIName)
	if objectAPIName == "" {
		return nil, fmt.Errorf("connector push legacy mirror backfill source object is blank")
	}
	var (
		apiName       string
		readSourceRaw string
	)
	if err := db.QueryRowContext(ctx, connectorPushBackfillObjectMetadataQuery, objectAPIName).Scan(&apiName, &readSourceRaw); err != nil {
		if err == sql.ErrNoRows {
			return nil, fmt.Errorf("connector push legacy mirror backfill source object %s metadata not found", objectAPIName)
		}
		return nil, fmt.Errorf("connector push legacy mirror backfill load source object %s metadata: %w", objectAPIName, err)
	}
	readSource, err := connectorPushBackfillDecodeReadSource(readSourceRaw)
	if err != nil {
		return nil, fmt.Errorf("connector push legacy mirror backfill decode source object %s readSource: %w", objectAPIName, err)
	}
	return &models.ObjectMetadata{
		APIName:    strings.TrimSpace(apiName),
		ReadSource: readSource,
	}, nil
}

func connectorPushBackfillDecodeReadSource(raw string) (*models.ObjectReadSourceConfig, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return nil, nil
	}
	var cfg models.ObjectReadSourceConfig
	if err := json.Unmarshal([]byte(raw), &cfg); err != nil {
		return nil, fmt.Errorf("parse readSource JSON: %w", err)
	}
	return &cfg, nil
}

func loadConnectorPushBackfillFieldColumn(ctx context.Context, db *sql.DB, objectAPIName, fieldAPIName string) (string, error) {
	objectAPIName = strings.TrimSpace(objectAPIName)
	fieldAPIName = strings.TrimSpace(fieldAPIName)
	if fieldAPIName == "" {
		return "", fmt.Errorf("connector push legacy mirror backfill source field is blank for object %s", objectAPIName)
	}
	var (
		apiName      string
		physicalName string
	)
	if err := db.QueryRowContext(ctx, connectorPushBackfillFieldMetadataQuery, objectAPIName, fieldAPIName).Scan(&apiName, &physicalName); err != nil {
		if err == sql.ErrNoRows {
			return "", fmt.Errorf("connector push legacy mirror backfill field %s.%s metadata not found", objectAPIName, fieldAPIName)
		}
		return "", fmt.Errorf("connector push legacy mirror backfill load field %s.%s metadata: %w", objectAPIName, fieldAPIName, err)
	}
	return connectorPushBackfillPhysicalColumnName(apiName, physicalName), nil
}

func connectorPushBackfillPhysicalColumnName(apiName, physicalName string) string {
	apiName = strings.TrimSpace(apiName)
	physicalName = strings.TrimSpace(physicalName)
	if physicalName == "" {
		return apiName
	}
	if strings.HasPrefix(physicalName, "`") && strings.HasSuffix(physicalName, "`") {
		return apiName
	}
	return physicalName
}

func connectorPushBackfillQuoteResolvedIdentifier(name string, pattern *regexp.Regexp, kind string) (string, error) {
	name = strings.TrimSpace(name)
	if !pattern.MatchString(name) {
		return "", fmt.Errorf("connector push legacy mirror backfill invalid %s %q", kind, name)
	}
	return "`" + name + "`", nil
}