package connector_callout

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

const connectorPackStatusActive = "active"

// PackResolver verifies a connector pack is active before profile lookups
// are served.
type PackResolver interface {
	ResolveConnectorPackByNamespace(ctx context.Context, namespace string) (active bool, err error)
}

// ConnectorCalloutProfile holds the full runtime configuration for an outbound
// callout channel, authored by a connector pack and stored in
// _System_ConnectorCalloutProfile. Load via LoadProfile.
type ConnectorCalloutProfile struct {
	// InstalledPackID is the per-incarnation _System_InstalledPack.PackInstallId
	// whose install snapshot authored this profile row. It is populated for
	// inbound worker startup reads; generic LoadProfile keeps using the active
	// pack resolver and may leave this empty.
	InstalledPackID string

	// OwnerPack is the connector pack namespace that authored this profile.
	OwnerPack string

	// ProfileName is the unique name within the owning pack.
	ProfileName string

	// CredentialRef is the DeveloperName of the NamedCredential used for
	// endpoint and bearer-token resolution.
	CredentialRef string

	// EndpointClass is the throttle classifier forwarded to connector_throttle
	// for Engine.Do and Engine.Paginate. Required because Engine.Do rejects
	// empty endpoint class. Stored at the profile level so a single profile
	// drives one throttle class; if BatchSpec is configured, its own
	// EndpointClass MAY differ.
	EndpointClass string

	// RetryPolicy configures retry behaviour for Engine.Do calls using this
	// profile.
	RetryPolicy RetryPolicy

	// PageSpec configures cursor-based pagination for Engine.Paginate calls.
	PageSpec PageSpec

	// BatchSpec configures chunked batch reads for Engine.BatchRead calls.
	BatchSpec BatchSpec

	// ObjectPathMap is an optional vendor-neutral mapping from logical object
	// names to remote API paths (e.g. "record" → "/api/v1/objects/records").
	// Interpreted by the caller; Engine itself does not use it.
	ObjectPathMap map[string]string

	// Description is an optional admin-readable note.
	Description string

	legacyRequestOrderBatchDemux bool
}

// AllowLegacyRequestOrderBatchDemux marks a historical installed-pack profile
// whose pre-upgrade batch spec lacked IDResponseField. New authoring and live
// active rows must not set this; it exists only so drain-only workers can finish
// queue rows created under the old request-order demux contract.
func AllowLegacyRequestOrderBatchDemux(profile *ConnectorCalloutProfile) {
	if profile != nil {
		profile.legacyRequestOrderBatchDemux = true
	}
}

func UsesLegacyRequestOrderBatchDemux(profile *ConnectorCalloutProfile) bool {
	return profile != nil && profile.legacyRequestOrderBatchDemux
}

// BatchProfileReadScope controls whether profile reads require the owning
// connector pack to be active or may also read drain-only non-deleted packs.
type BatchProfileReadScope int

const (
	BatchProfileReadActiveOnly BatchProfileReadScope = iota
	BatchProfileReadAllowDrain
)

// LoadProfile reads a ConnectorCalloutProfile from _System_ConnectorCalloutProfile
// for the given (ownerPack, profileName) pair.
//
// ownerPack is trimmed before lookup to match the verbatim pack name written to
// _System_InstalledPack.Name by the install lifecycle. profileName is lowercased
// and trimmed to match the canonical storage form written by
// ConnectorCalloutProfileComponent.Apply.
//
// resolver is required. A nil resolver causes LoadProfile to return
// ErrProfileUnavailable immediately (fail-closed). An inactive owning pack
// also returns ErrProfileUnavailable. This mirrors the webhook_secret and
// namedcredential server-internal read paths.
//
// Returns ErrProfileNotFound when no row matches (or the row is soft-deleted).
func LoadProfile(ctx context.Context, db *sql.DB, resolver PackResolver, ownerPack, profileName string) (*ConnectorCalloutProfile, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	profileName = strings.ToLower(strings.TrimSpace(profileName))
	p, err := querySingleProfile(ctx, db, ownerPack, profileName)
	if err == sql.ErrNoRows {
		return nil, ErrProfileNotFound
	}
	if err != nil {
		return nil, err
	}

	// Fail closed on inactive owning pack — mirror webhook_secret and
	// namedcredential server-internal read paths. Admin-installed
	// NamedCredentials skip the pack-active gate, but the profile row
	// itself is still pack-owned and must not drive runtime behaviour
	// when its pack is inactive.
	if resolver == nil {
		return nil, ErrProfileUnavailable
	}
	active, err := resolver.ResolveConnectorPackByNamespace(ctx, p.OwnerPack)
	if err != nil {
		return nil, fmt.Errorf("resolve owner pack %s: %w", p.OwnerPack, err)
	}
	if !active {
		return nil, ErrProfileUnavailable
	}

	return p, nil
}

// LoadActiveProfile reads one active connector callout profile through the same
// exact installed-pack incarnation authority as worker startup. OwnerPack alone
// is not runtime authority for trusted outbound writes.
func LoadActiveProfile(ctx context.Context, db *sql.DB, ownerPack, profileName string) (*ConnectorCalloutProfile, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	profileName = strings.ToLower(strings.TrimSpace(profileName))
	if ownerPack == "" || profileName == "" {
		return nil, ErrProfileUnavailable
	}

	packInstallID, found, err := activeConnectorPackInstallID(ctx, db, ownerPack)
	if err != nil {
		return nil, err
	}
	if !found {
		return nil, ErrProfileUnavailable
	}
	if err := ensureNoLiveProfileRowsForDifferentInstall(ctx, db, ownerPack, packInstallID); err != nil {
		return nil, err
	}

	p, err := querySingleProfileForInstall(ctx, db, ownerPack, packInstallID, profileName)
	if err == sql.ErrNoRows {
		return nil, ErrProfileNotFound
	}
	if err != nil {
		return nil, err
	}
	return p, nil
}

// LoadProfileForPackInstall reads one profile from an exact, caller-selected
// connector-pack incarnation. It deliberately does not select an install ID
// through _System_InstalledPack: callers must first validate the selected
// generation with their own runtime authority.
//
// The profile row is still fenced to one live incarnation, so a partially
// converged profile migration fails closed rather than mixing generations.
func LoadProfileForPackInstall(ctx context.Context, db *sql.DB, ownerPack, packInstallID, profileName string) (*ConnectorCalloutProfile, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	packInstallID = strings.TrimSpace(packInstallID)
	profileName = strings.ToLower(strings.TrimSpace(profileName))
	if ownerPack == "" || packInstallID == "" || profileName == "" {
		return nil, ErrProfileUnavailable
	}
	if err := ensureNoLiveProfileRowsForDifferentInstall(ctx, db, ownerPack, packInstallID); err != nil {
		return nil, err
	}

	p, err := querySingleProfileForExactInstall(ctx, db, ownerPack, packInstallID, profileName)
	if err == sql.ErrNoRows {
		return nil, ErrProfileNotFound
	}
	if err != nil {
		return nil, err
	}
	if p == nil || strings.TrimSpace(p.InstalledPackID) != packInstallID {
		return nil, ErrProfileUnavailable
	}
	return p, nil
}

// LoadProfilesForInboundWorker returns all non-deleted profiles for the exact
// installed-pack incarnation selected by startup.
//
// BatchProfileReadActiveOnly requires the owning connector pack to be active
// with converged applied/desired payload hashes.
// BatchProfileReadAllowDrain still verifies that the pack row is non-deleted
// and connector-pack scoped, but allows removing/inactive rows for callers that
// still need visibility into the current install-scoped profile rows during
// uninstall windows.
func LoadProfilesForInboundWorker(ctx context.Context, db *sql.DB, ownerPack, packInstallID string, scope BatchProfileReadScope) ([]*ConnectorCalloutProfile, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	packInstallID = strings.TrimSpace(packInstallID)
	if ownerPack == "" {
		return nil, ErrProfileUnavailable
	}
	if packInstallID == "" {
		return nil, ErrProfileUnavailable
	}
	status, hashesConverged, found, err := loadConnectorPackStateForInstall(ctx, db, ownerPack, packInstallID)
	if err != nil {
		return nil, err
	}
	if !found {
		return nil, ErrProfileUnavailable
	}
	if scope == BatchProfileReadActiveOnly && status != connectorPackStatusActive {
		return nil, ErrProfileUnavailable
	}
	if scope == BatchProfileReadActiveOnly && !hashesConverged {
		return nil, ErrProfileUnavailable
	}
	if err := ensureNoLiveProfileRowsForDifferentInstall(ctx, db, ownerPack, packInstallID); err != nil {
		return nil, err
	}

	rows, err := db.QueryContext(ctx, `
SELECT c.InstalledPackId, c.OwnerPack, c.ProfileName, c.CredentialRef, c.EndpointClass,
       c.RetryPolicy, c.PageSpec, c.BatchSpec, c.ObjectPathMap, c.Description
FROM _System_ConnectorCalloutProfile c
JOIN _System_InstalledPack p
  ON p.PackInstallId = c.InstalledPackId
 AND p.Name = c.OwnerPack
WHERE c.OwnerPack = ?
  AND c.InstalledPackId = ?
  AND p.Category = 'connector-pack'
  AND COALESCE(p.IsDeleted, 0) = 0
  AND COALESCE(c.IsDeleted, 0) = 0
ORDER BY c.ProfileName`, ownerPack, packInstallID)
	if err != nil {
		return nil, fmt.Errorf("load connector callout profiles for %s (%s): %w", ownerPack, packInstallID, err)
	}
	defer rows.Close()

	var profiles []*ConnectorCalloutProfile
	for rows.Next() {
		p, err := scanInboundWorkerProfile(rows)
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(p.InstalledPackID) != packInstallID {
			return nil, ErrProfileUnavailable
		}
		profiles = append(profiles, p)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate connector callout profiles for %s (%s): %w", ownerPack, packInstallID, err)
	}
	return profiles, nil
}

// LoadProfilesForInboundWorkerDrainOnly returns the runtime profile rows needed
// to drain a non-active installed connector pack. Removal soft-deletes these
// rows before drain-only workers start, so this path deliberately reads the
// preserved rows without requiring COALESCE(c.IsDeleted, 0) = 0. The exact
// installed-pack incarnation is still required; OwnerPack alone is never
// authority for worker startup.
func LoadProfilesForInboundWorkerDrainOnly(ctx context.Context, db *sql.DB, ownerPack, packInstallID string) ([]*ConnectorCalloutProfile, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	packInstallID = strings.TrimSpace(packInstallID)
	if ownerPack == "" {
		return nil, ErrProfileUnavailable
	}
	if packInstallID == "" {
		return nil, ErrProfileUnavailable
	}
	status, hashesConverged, found, err := loadConnectorPackStateForInstall(ctx, db, ownerPack, packInstallID)
	if err != nil {
		return nil, err
	}
	if !found || status == connectorPackStatusActive || !hashesConverged {
		return nil, ErrProfileUnavailable
	}
	if err := ensureNoLiveProfileRowsForDifferentInstall(ctx, db, ownerPack, packInstallID); err != nil {
		return nil, err
	}

	rows, err := db.QueryContext(ctx, `
SELECT c.InstalledPackId, c.OwnerPack, c.ProfileName, c.CredentialRef, c.EndpointClass,
       c.RetryPolicy, c.PageSpec, c.BatchSpec, c.ObjectPathMap, c.Description
FROM _System_ConnectorCalloutProfile c
JOIN _System_InstalledPack p
  ON p.PackInstallId = c.InstalledPackId
 AND p.Name = c.OwnerPack
WHERE c.OwnerPack = ?
  AND c.InstalledPackId = ?
  AND p.Category = 'connector-pack'
  AND COALESCE(p.IsDeleted, 0) = 0
ORDER BY c.ProfileName`, ownerPack, packInstallID)
	if err != nil {
		return nil, fmt.Errorf("load drain-only connector callout profiles for %s (%s): %w", ownerPack, packInstallID, err)
	}
	defer rows.Close()

	var profiles []*ConnectorCalloutProfile
	for rows.Next() {
		p, err := scanInboundWorkerProfile(rows)
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(p.InstalledPackID) != packInstallID {
			return nil, ErrProfileUnavailable
		}
		profiles = append(profiles, p)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate drain-only connector callout profiles for %s (%s): %w", ownerPack, packInstallID, err)
	}
	return profiles, nil
}

func querySingleProfile(ctx context.Context, db *sql.DB, ownerPack, profileName string) (*ConnectorCalloutProfile, error) {
	row := db.QueryRowContext(ctx, `
	SELECT OwnerPack, ProfileName, CredentialRef, EndpointClass,
	       RetryPolicy, PageSpec, BatchSpec, ObjectPathMap, Description
FROM _System_ConnectorCalloutProfile
WHERE OwnerPack = ? AND ProfileName = ? AND COALESCE(IsDeleted, 0) = 0
LIMIT 1`, ownerPack, profileName)
	p, err := scanProfile(row)
	if err == sql.ErrNoRows {
		return nil, sql.ErrNoRows
	}
	if err != nil {
		return nil, fmt.Errorf("load connector callout profile %s/%s: %w", ownerPack, profileName, err)
	}
	return p, nil
}

func querySingleProfileForInstall(ctx context.Context, db *sql.DB, ownerPack, packInstallID, profileName string) (*ConnectorCalloutProfile, error) {
	row := db.QueryRowContext(ctx, `
SELECT c.InstalledPackId, c.OwnerPack, c.ProfileName, c.CredentialRef, c.EndpointClass,
       c.RetryPolicy, c.PageSpec, c.BatchSpec, c.ObjectPathMap, c.Description
FROM _System_ConnectorCalloutProfile c
JOIN _System_InstalledPack p
  ON p.PackInstallId = c.InstalledPackId
 AND p.Name = c.OwnerPack
WHERE c.OwnerPack = ?
  AND c.InstalledPackId = ?
  AND c.ProfileName = ?
  AND p.Category = 'connector-pack'
  AND p.Status = ?
  AND TRIM(COALESCE(p.AppliedPayloadHash, '')) <> ''
  AND TRIM(COALESCE(p.DesiredPayloadHash, '')) <> ''
  AND p.AppliedPayloadHash = p.DesiredPayloadHash
  AND COALESCE(p.IsDeleted, 0) = 0
  AND COALESCE(c.IsDeleted, 0) = 0
LIMIT 1`, ownerPack, packInstallID, profileName, connectorPackStatusActive)
	p, err := scanInboundWorkerProfile(row)
	if err == sql.ErrNoRows {
		return nil, sql.ErrNoRows
	}
	if err != nil {
		return nil, fmt.Errorf("load active connector callout profile %s/%s (%s): %w", ownerPack, profileName, packInstallID, err)
	}
	return p, nil
}

// querySingleProfileForExactInstall intentionally reads only the profile
// record. The caller has already selected and verified the install generation;
// joining _System_InstalledPack here would silently reintroduce a second,
// potentially different generation authority.
func querySingleProfileForExactInstall(ctx context.Context, db *sql.DB, ownerPack, packInstallID, profileName string) (*ConnectorCalloutProfile, error) {
	row := db.QueryRowContext(ctx, `
SELECT c.InstalledPackId, c.OwnerPack, c.ProfileName, c.CredentialRef, c.EndpointClass,
       c.RetryPolicy, c.PageSpec, c.BatchSpec, c.ObjectPathMap, c.Description
FROM _System_ConnectorCalloutProfile c
WHERE c.OwnerPack = ?
  AND c.InstalledPackId = ?
  AND c.ProfileName = ?
  AND COALESCE(c.IsDeleted, 0) = 0
LIMIT 1`, ownerPack, packInstallID, profileName)
	p, err := scanInboundWorkerProfile(row)
	if err == sql.ErrNoRows {
		return nil, sql.ErrNoRows
	}
	if err != nil {
		return nil, fmt.Errorf("load connector callout profile for exact install %s/%s (%s): %w", ownerPack, profileName, packInstallID, err)
	}
	return p, nil
}

type profileScanner interface {
	Scan(dest ...any) error
}

func scanProfile(scanner profileScanner) (*ConnectorCalloutProfile, error) {
	var (
		retryPolicyJSON   string
		pageSpecJSON      string
		batchSpecJSON     string
		objectPathMapJSON string
		description       sql.NullString
		p                 ConnectorCalloutProfile
	)
	if err := scanner.Scan(
		&p.OwnerPack, &p.ProfileName, &p.CredentialRef, &p.EndpointClass,
		&retryPolicyJSON, &pageSpecJSON, &batchSpecJSON, &objectPathMapJSON,
		&description,
	); err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(retryPolicyJSON), &p.RetryPolicy); err != nil {
		return nil, fmt.Errorf("decode RetryPolicy for %s/%s: %w", p.OwnerPack, p.ProfileName, err)
	}
	if err := json.Unmarshal([]byte(pageSpecJSON), &p.PageSpec); err != nil {
		return nil, fmt.Errorf("decode PageSpec for %s/%s: %w", p.OwnerPack, p.ProfileName, err)
	}
	if err := json.Unmarshal([]byte(batchSpecJSON), &p.BatchSpec); err != nil {
		return nil, fmt.Errorf("decode BatchSpec for %s/%s: %w", p.OwnerPack, p.ProfileName, err)
	}
	if objectPathMapJSON != "" && objectPathMapJSON != "null" {
		if err := json.Unmarshal([]byte(objectPathMapJSON), &p.ObjectPathMap); err != nil {
			return nil, fmt.Errorf("decode ObjectPathMap for %s/%s: %w", p.OwnerPack, p.ProfileName, err)
		}
	}
	if description.Valid {
		p.Description = description.String
	}
	return &p, nil
}

func scanInboundWorkerProfile(scanner profileScanner) (*ConnectorCalloutProfile, error) {
	var (
		retryPolicyJSON   string
		pageSpecJSON      string
		batchSpecJSON     string
		objectPathMapJSON string
		description       sql.NullString
		p                 ConnectorCalloutProfile
	)
	if err := scanner.Scan(
		&p.InstalledPackID,
		&p.OwnerPack, &p.ProfileName, &p.CredentialRef, &p.EndpointClass,
		&retryPolicyJSON, &pageSpecJSON, &batchSpecJSON, &objectPathMapJSON,
		&description,
	); err != nil {
		return nil, err
	}
	if strings.TrimSpace(p.InstalledPackID) == "" {
		return nil, ErrProfileUnavailable
	}
	if err := json.Unmarshal([]byte(retryPolicyJSON), &p.RetryPolicy); err != nil {
		return nil, fmt.Errorf("decode RetryPolicy for %s/%s (%s): %w", p.OwnerPack, p.ProfileName, p.InstalledPackID, err)
	}
	if err := json.Unmarshal([]byte(pageSpecJSON), &p.PageSpec); err != nil {
		return nil, fmt.Errorf("decode PageSpec for %s/%s (%s): %w", p.OwnerPack, p.ProfileName, p.InstalledPackID, err)
	}
	if err := json.Unmarshal([]byte(batchSpecJSON), &p.BatchSpec); err != nil {
		return nil, fmt.Errorf("decode BatchSpec for %s/%s (%s): %w", p.OwnerPack, p.ProfileName, p.InstalledPackID, err)
	}
	if objectPathMapJSON != "" && objectPathMapJSON != "null" {
		if err := json.Unmarshal([]byte(objectPathMapJSON), &p.ObjectPathMap); err != nil {
			return nil, fmt.Errorf("decode ObjectPathMap for %s/%s (%s): %w", p.OwnerPack, p.ProfileName, p.InstalledPackID, err)
		}
	}
	if description.Valid {
		p.Description = description.String
	}
	return &p, nil
}

func loadConnectorPackStateForInstall(ctx context.Context, db *sql.DB, ownerPack, packInstallID string) (string, bool, bool, error) {
	var status, appliedHash, desiredHash string
	err := db.QueryRowContext(ctx, `
	SELECT Status, COALESCE(AppliedPayloadHash, ''), COALESCE(DesiredPayloadHash, '')
	FROM _System_InstalledPack
	WHERE Name = ?
	  AND PackInstallId = ?
	  AND Category = 'connector-pack'
	  AND COALESCE(IsDeleted, 0) = 0
	LIMIT 1`, ownerPack, packInstallID).Scan(&status, &appliedHash, &desiredHash)
	if err == sql.ErrNoRows {
		return "", false, false, nil
	}
	if err != nil {
		return "", false, false, fmt.Errorf("load connector pack status %s (%s): %w", ownerPack, packInstallID, err)
	}
	appliedHash = strings.TrimSpace(appliedHash)
	desiredHash = strings.TrimSpace(desiredHash)
	hashesConverged := appliedHash != "" && desiredHash != "" && appliedHash == desiredHash
	return strings.TrimSpace(status), hashesConverged, true, nil
}

func activeConnectorPackInstallID(ctx context.Context, db *sql.DB, ownerPack string) (string, bool, error) {
	rows, err := db.QueryContext(ctx, `
SELECT PackInstallId
FROM _System_InstalledPack
WHERE Name = ?
  AND Category = 'connector-pack'
  AND Status = ?
  AND TRIM(COALESCE(AppliedPayloadHash, '')) <> ''
  AND TRIM(COALESCE(DesiredPayloadHash, '')) <> ''
  AND AppliedPayloadHash = DesiredPayloadHash
  AND COALESCE(IsDeleted, 0) = 0`, ownerPack, connectorPackStatusActive)
	if err != nil {
		return "", false, fmt.Errorf("load active connector pack install for %s: %w", ownerPack, err)
	}
	defer rows.Close()

	var installID string
	count := 0
	for rows.Next() {
		var raw sql.NullString
		if err := rows.Scan(&raw); err != nil {
			return "", false, fmt.Errorf("scan active connector pack install for %s: %w", ownerPack, err)
		}
		next := strings.TrimSpace(raw.String)
		if next == "" {
			return "", false, ErrProfileUnavailable
		}
		installID = next
		count++
	}
	if err := rows.Err(); err != nil {
		return "", false, fmt.Errorf("iterate active connector pack installs for %s: %w", ownerPack, err)
	}
	switch count {
	case 0:
		return "", false, nil
	case 1:
		return installID, true, nil
	default:
		return "", false, ErrProfileUnavailable
	}
}

func ensureNoLiveProfileRowsForDifferentInstall(ctx context.Context, db *sql.DB, ownerPack, packInstallID string) error {
	var count int
	err := db.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM _System_ConnectorCalloutProfile
WHERE OwnerPack = ?
  AND COALESCE(IsDeleted, 0) = 0
  AND COALESCE(InstalledPackId, '') <> ?`, ownerPack, packInstallID).Scan(&count)
	if err != nil {
		return fmt.Errorf("check connector callout profile install drift for %s (%s): %w", ownerPack, packInstallID, err)
	}
	if count != 0 {
		return ErrProfileUnavailable
	}
	return nil
}