package records

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

	connectorreaders "github.com/nexuscrm/db/pkg/connectorreaders"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

// insertInheritActionType is the _System_ObjectFieldMapping.ActionType value
// reserved for the insert-inheritance applier. Rows with this ActionType are
// consumed by applyInsertInheritance (below) and NOT by the existing
// Lead-Convert / Flow-RecordCreate applier — those filter by their own
// ActionType strings ("convert", etc.), so the two registries coexist
// without semantic overlap even though they share the table.
const insertInheritActionType = "insertInheritFromLookup"

type insertInheritanceCacheEntry struct {
	rules []connector.FieldMappingWithSource
}

type insertInheritanceRuleQueryExecutor interface {
	QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error)
}

type insertInheritanceRuleLoader func(context.Context, insertInheritanceRuleQueryExecutor, string) ([]connector.FieldMappingWithSource, error)

// insertInheritanceMetadataProvider is the narrow projection seam owned by
// CatalogSnapshot. Runtime composition supplies rules from the immutable view
// already bound to the request; legacy monolith composition keeps the live DB
// reader until its later ownership flip.
type insertInheritanceMetadataProvider interface {
	InsertInheritanceRules(context.Context, string) ([]connector.FieldMappingWithSource, error)
}

func insertInheritanceCacheKey(objectName, actionType string) string {
	return strings.ToLower(strings.TrimSpace(objectName)) + "\x00" + strings.ToLower(strings.TrimSpace(actionType))
}

func insertInheritanceLoadGroupKey(epoch uint64, cacheKey string) string {
	return fmt.Sprintf("rules:%d:%s", epoch, cacheKey)
}

func (ps *PersistenceService) clearInsertInheritanceCache() {
	ps.insertInheritanceCacheMu.Lock()
	ps.insertInheritanceCache = nil
	ps.insertInheritanceEpoch++
	ps.insertInheritanceCacheMu.Unlock()
}

// InvalidateInsertInheritanceCache clears the derived insert-inheritance rule
// cache so metadata refresh/install paths can flush stale
// _System_ObjectFieldMapping-backed inheritance state without restarting the
// process.
func (ps *PersistenceService) InvalidateInsertInheritanceCache() {
	ps.clearInsertInheritanceCache()
}

func loadInsertInheritanceRulesWithExecutor(
	ctx context.Context,
	exec insertInheritanceRuleQueryExecutor,
	objectName string,
) ([]connector.FieldMappingWithSource, error) {
	if exec == nil {
		return nil, nil
	}
	return connectorreaders.GetFieldMappingsByTargetContext(ctx, exec, objectName, insertInheritActionType)
}

func (ps *PersistenceService) insertInheritanceRuleReadExecutor(
	ctx context.Context,
) (insertInheritanceRuleQueryExecutor, bool) {
	if ps == nil || ps.rawDB == nil {
		return nil, false
	}
	if ps.txManager != nil {
		if rawTx := ps.txManager.ExtractTx(ctx); rawTx != nil {
			if tx, ok := rawTx.(*sql.Tx); ok && tx != nil {
				return tx, true
			}
		}
		if ps.txManager.IsTransactionActive(ctx) {
			// Some callers mark transaction scope via txManager even when the
			// concrete *sql.Tx is unavailable to this helper. Treat that as
			// tx-scoped for cache-bypass purposes so shared state is never
			// populated from a transaction-bound metadata snapshot.
			return ps.rawDB, true
		}
	}
	return ps.rawDB, false
}

func (ps *PersistenceService) cachedInsertInheritanceRules(ctx context.Context, objectName string) ([]connector.FieldMappingWithSource, error) {
	if ps == nil {
		return nil, nil
	}
	if ps.insertInheritanceMeta != nil {
		// The view itself is immutable and request-scoped. A process-global cache
		// here could only weaken revision binding, so resolve it on every call.
		return ps.insertInheritanceMeta.InsertInheritanceRules(ctx, objectName)
	}
	if ps.rawDB == nil {
		return nil, nil
	}
	cacheKey := insertInheritanceCacheKey(objectName, insertInheritActionType)
	exec, txScoped := ps.insertInheritanceRuleReadExecutor(ctx)
	if txScoped {
		// Reuse the active transaction's pinned connection, but never consult
		// or populate the shared cache. A tx-scoped metadata snapshot must not
		// be influenced by a process-global cache entry built from a different
		// transaction snapshot, and its own uncommitted reads must not leak
		// back into global state.
		return loadInsertInheritanceRulesWithExecutor(ctx, exec, objectName)
	}
	for {
		ps.insertInheritanceCacheMu.RLock()
		if ps.insertInheritanceCache != nil {
			if entry, ok := ps.insertInheritanceCache[cacheKey]; ok {
				ps.insertInheritanceCacheMu.RUnlock()
				return entry.rules, nil
			}
		}
		loadEpoch := ps.insertInheritanceEpoch
		ps.insertInheritanceCacheMu.RUnlock()

		loadKey := insertInheritanceLoadGroupKey(loadEpoch, cacheKey)
		loader := ps.insertInheritanceRuleLoader
		if loader == nil {
			loader = loadInsertInheritanceRulesWithExecutor
		}
		loaded, err, _ := ps.insertInheritanceLoadGroup.Do(loadKey, func() (interface{}, error) {
			return loader(ctx, exec, objectName)
		})
		if err != nil {
			return nil, err
		}
		rules, _ := loaded.([]connector.FieldMappingWithSource)

		ps.insertInheritanceCacheMu.Lock()
		if ps.insertInheritanceEpoch != loadEpoch {
			ps.insertInheritanceCacheMu.Unlock()
			continue
		}
		if ps.insertInheritanceCache == nil {
			ps.insertInheritanceCache = make(map[string]insertInheritanceCacheEntry)
		}
		if entry, ok := ps.insertInheritanceCache[cacheKey]; ok {
			ps.insertInheritanceCacheMu.Unlock()
			return entry.rules, nil
		}
		ps.insertInheritanceCache[cacheKey] = insertInheritanceCacheEntry{rules: rules}
		ps.insertInheritanceCacheMu.Unlock()
		return rules, nil
	}
}

// applyInsertInheritance runs before the caller's persistence Insert to copy
// fields from a lookup-referenced parent record onto the new record being
// inserted. Rule authoring surface: _System_ObjectFieldMapping rows with
// ActionType='insertInheritFromLookup'. Rows are grouped by
// (SourceObject, SourceLookupField); each group triggers at most one parent
// Retrieve per insert.
//
// Semantics preserve the deleted mergeQuoteDefaultsFromOpportunity
// (backend/internal/features/records/quote_create.go:41-93) behavior
// exactly — the parity test in insert_inheritance_parity_test.go is the
// falsifier. Rules:
//
//  1. Fill blank only. If data[TargetField] is already non-blank, skip.
//  2. Skip target field categories: IsSystem, Formula, !Createable. Also
//     skip target fields in the engineOwnedFieldSet (autonumber today) so
//     renderer outputs are never clobbered.
//  3. Skip the hardcoded target-field blacklist: Id, OwnerId,
//     and the rule's own SourceLookupField (the FK that identified the
//     parent — must not be overwritten by inheritance).
//  4. Rule shape:
//     - SourceField == ""  + DefaultValue != "": constant default when
//     target blank (expresses Quote.Status="Pending").
//     - SourceField != "" + DefaultValue == "": copy source[SourceField]
//     when both target blank AND source non-blank.
//     - SourceField != "" + DefaultValue != "": prefer source[SourceField];
//     fall back to DefaultValue when source blank.
//  5. Flat source read only — source[SourceField] — no dotted-path resolver.
//     Per plan §6.4: validator enforces the same constraint.
//
// No-ops (early return) when any of the following is true:
//   - neither the Runtime view provider nor legacy rawDB is wired (for example
//     in lightweight test harnesses).
//   - Zero active rules for (targetObject, insertInheritFromLookup).
//   - schema is nil.
//   - No rule group's SourceLookupField resolves to a non-blank value on data.
//
// Parent retrieval uses ps.Retrieve under the current user — so FLS-denied
// source fields are omitted from the retrieved record and silently skipped
// (matches old mergeQuoteDefaultsFromOpportunity: the Opportunity Retrieve
// at quote_create.go:237 used the same user).
// Returns a bool indicating whether data was modified — the caller uses it
// to decide whether a follow-up Normalize+prune pass is warranted. When no
// rules fire, data is untouched and the caller can skip those passes.
func (ps *PersistenceService) applyInsertInheritance(
	ctx context.Context,
	objectName string,
	data models.SObject,
	schema *models.ObjectMetadata,
	currentUser *models.UserSession,
) (bool, error) {
	if (ps.insertInheritanceMeta == nil && ps.rawDB == nil) || schema == nil || len(data) == 0 {
		return false, nil
	}

	rules, err := ps.cachedInsertInheritanceRules(ctx, objectName)
	if err != nil {
		return false, fmt.Errorf("load insert-inheritance rules for %s: %w", objectName, err)
	}
	if len(rules) == 0 {
		return false, nil
	}

	// Index target-object schema fields by lowercased API name for
	// capability checks. FieldsMap may be nil (populated lazily in
	// some code paths); fall back to Fields slice.
	targetFields := make(map[string]*models.FieldMetadata, len(schema.Fields))
	for i := range schema.Fields {
		f := &schema.Fields[i]
		targetFields[strings.ToLower(f.APIName)] = f
	}
	engineOwned, err := ps.engineOwnedFieldSet(ctx, objectName)
	if err != nil {
		return false, fmt.Errorf("resolve engine-owned fields for %s: %w", objectName, err)
	}

	// Group by (SourceObject, SourceLookupField). Each unique group gets
	// one parent Retrieve.
	type groupKey struct{ sourceObject, sourceLookupField string }
	groups := make(map[groupKey][]connector.FieldMappingWithSource)
	for _, r := range rules {
		// SourceLookupField is required for this ActionType — rows missing
		// it would be rejected by the object-level validator at bootstrap
		// time (§6 in plan). Skip defensively at runtime so a misseeded
		// row doesn't crash inserts.
		if r.SourceLookupField == "" {
			continue
		}
		k := groupKey{sourceObject: r.SourceObject, sourceLookupField: r.SourceLookupField}
		groups[k] = append(groups[k], r)
	}

	modified := false
	for k, groupRules := range groups {
		parentID := inheritLookupString(data[k.sourceLookupField])
		if parentID == "" {
			continue
		}

		// Preview-provenance short-circuit: under a no-DML insert preview
		// preparation, accept synthetic preview-* parent IDs that the
		// request-local registry knows about for THIS source object. The
		// synthetic ID was minted by an earlier RecordCreate in the same
		// flow preview; reading it through the durable ps.Retrieve path
		// would 404 because the row has never been written to the DB.
		// When the synthetic record doesn't carry the inherited source
		// fields, inheritance is a no-op for the group (no hard-fail) —
		// consistent with applyInheritanceRulesPure's "source blank →
		// skip" semantics. Stray preview-* literals without a matching
		// registry entry fall through to the normal Retrieve so the
		// preview is still gated by the durable existence/access check.
		if constants.HasInsertPreviewPreparation(ctx) {
			if synthetic, ok := constants.PreviewSyntheticRecordForObject(ctx, k.sourceObject, parentID); ok {
				parent := models.SObject(synthetic.Fields)
				if applyInheritanceRulesPure(groupRules, parent, data, targetFields, engineOwned, k.sourceLookupField) {
					modified = true
				}
				continue
			}
		}

		parent, retrieveErr := ps.Retrieve(ctx, k.sourceObject, parentID, currentUser)
		if retrieveErr != nil {
			// This hook is a pure opportunistic copier, not a validation
			// owner. Access / existence was already verified by the
			// named owner, validateNonPolymorphicLookups, which runs
			// before this hook in persistence_insert.go — if the user
			// couldn't see or the record was missing, the request has
			// already been rejected with a 400 referencing the field.
			//
			// The only error shape that should still bubble here is an
			// InternalError (DB outage between the validator's Retrieve
			// and this one). Anything else — apierr.NotFound from a
			// race-induced disappearance, a permission change mid-flight,
			// etc. — is benign-skipped: the insert proceeds without
			// inheritance for that group, other rules still fire, the
			// record lands intact.
			//
			// Ordering is load-bearing: IsInternal first so the
			// InternalError(cause=apierr.NotFound) envelope pattern
			// (tx-manager rollback wrapping a 4xx cause) surfaces as
			// 500, not silently swallowed. Mirrors the boundary error
			// projector's internal-first ordering.
			if appErrors.IsInternal(retrieveErr) {
				return modified, fmt.Errorf("insert-inheritance retrieve %s %q for %s: %w",
					k.sourceObject, parentID, objectName, retrieveErr)
			}
			continue
		}
		if len(parent) == 0 {
			continue
		}

		if applyInheritanceRulesPure(groupRules, parent, data, targetFields, engineOwned, k.sourceLookupField) {
			modified = true
		}
	}

	return modified, nil
}

// applyInheritanceRulesPure is the in-memory per-rule loop. Extracted so the
// behavior-parity test (insert_inheritance_parity_test.go) can exercise it
// against a fixture without wiring a DB. The enclosing applier wraps it
// with rule loading + parent retrieval + per-group dispatch.
//
// Returns true iff any rule wrote to data.
func applyInheritanceRulesPure(
	rules []connector.FieldMappingWithSource,
	parent models.SObject,
	data models.SObject,
	targetFields map[string]*models.FieldMetadata,
	engineOwned map[string]bool,
	sourceLookupField string,
) bool {
	modified := false
	for _, r := range rules {
		if !canInheritTarget(r.TargetField, sourceLookupField, targetFields, engineOwned) {
			continue
		}
		if !isBlankInheritValue(data[r.TargetField]) {
			continue
		}
		if r.SourceField != "" {
			if v, ok := parent[r.SourceField]; ok && !isBlankInheritValue(v) {
				data[r.TargetField] = v
				modified = true
				continue
			}
		}
		if r.DefaultValue != "" {
			data[r.TargetField] = r.DefaultValue
			modified = true
		}
	}
	return modified
}

// canInheritTarget encodes the target-field blacklist / capability checks
// from mergeQuoteDefaultsFromOpportunity. Kept pure so it can be asserted
// directly in unit tests without constructing a full PersistenceService.
func canInheritTarget(
	targetField, sourceLookupField string,
	targetFields map[string]*models.FieldMetadata,
	engineOwned map[string]bool,
) bool {
	if targetField == "" {
		return false
	}
	switch {
	case strings.EqualFold(targetField, constants.FieldID),
		strings.EqualFold(targetField, constants.FieldOwnerID),
		strings.EqualFold(targetField, sourceLookupField):
		return false
	}
	lc := strings.ToLower(targetField)
	if engineOwned != nil && engineOwned[lc] {
		return false
	}
	fm, ok := targetFields[lc]
	if !ok || fm == nil {
		// Target field not in schema — validator should have caught this
		// at bootstrap. Skip at runtime rather than silently fabricate
		// values on an unknown field.
		return false
	}
	if fm.IsSystem {
		return false
	}
	if fm.Formula != nil || fm.Type == constants.FieldTypeFormula {
		return false
	}
	if !fm.Createable {
		return false
	}
	return true
}

// isBlankInheritValue mirrors quote_create.go:95-107 isBlankValue — nil,
// empty string (ignoring whitespace), empty []byte are blank; all other
// typed values are considered non-blank. Kept separate from isBlankValue
// in quote_create.go so that file can be deleted in this PR without
// cascading the rename.
func isBlankInheritValue(v interface{}) bool {
	if v == nil {
		return true
	}
	switch val := v.(type) {
	case string:
		return strings.TrimSpace(val) == ""
	case []byte:
		return strings.TrimSpace(string(val)) == ""
	default:
		return false
	}
}

// inheritLookupString extracts the string form of a lookup-field value. SObject map
// values come through as `string` for the normal FK shape; a nil / absent
// entry is returned as "".
func inheritLookupString(v interface{}) string {
	s, _ := v.(string)
	return strings.TrimSpace(s)
}