package records

import (
	"context"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"github.com/nexuscrm/kernel/pkg/apierr"
	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// SelectedRootProjectionReader is the narrow in-process seam implemented by
// analytics.QueryService. It is not a transport or a contract.Deps capability:
// records calls it after object CRUD and FLS normalization so query-level RLS
// and repository projection retain their existing owner.
type SelectedRootProjectionReader interface {
	ReadSelectedRootProjection(ctx context.Context, schema *models.ObjectMetadata, recordID string, selectedFields []string, user *models.UserSession) (models.Record, error)
}

type SelectedRootValueKind string

const (
	SelectedRootString    SelectedRootValueKind = "string"
	SelectedRootBoolean   SelectedRootValueKind = "boolean"
	SelectedRootInteger   SelectedRootValueKind = "integer"
	SelectedRootDecimal   SelectedRootValueKind = "decimal"
	SelectedRootDate      SelectedRootValueKind = "date"
	SelectedRootDateTime  SelectedRootValueKind = "datetime"
	SelectedRootReference SelectedRootValueKind = "reference"
)

// SelectedRootDecimalValue is an exact base-10 value. Coefficient contains an
// optional leading '-' and digits only; Scale is the stored fractional width.
type SelectedRootDecimalValue struct {
	Coefficient string
	Scale       uint32
}

// SelectedRootField is a semantic, transport-independent tagged union. Pointer
// arms preserve false, zero, and empty string as present values. Null has no
// value arm. An absent or FLS-hidden field has no entry in Fields.
type SelectedRootField struct {
	Name      string
	Kind      SelectedRootValueKind
	Null      bool
	String    *string
	Boolean   *bool
	Integer   *int64
	Decimal   *SelectedRootDecimalValue
	Date      *string
	DateTime  *string
	Reference *string
}

type SelectedRootRecord struct {
	RecordID string
	Fields   []SelectedRootField
}

// SetSelectedRootProjectionReader wires the existing query-level RLS and
// projected repository path. A missing seam is a startup/configuration error,
// never permission to fall back to SELECT *.
func (ps *PersistenceService) SetSelectedRootProjectionReader(reader SelectedRootProjectionReader) {
	ps.selectedRootProjection = reader
}

// ReadSelectedSystemRecord performs one field-capped internal point read.
// Records owns schema validation, metadata-declared hard read protection, and
// the existing query/RLS projection seam. Callers own the justification for
// bypassing ordinary CRUD/FLS, but System context never bypasses
// ReadProtectionPolicy=blocked. This is deliberately not a cache and never
// falls back to GetRaw/SELECT *.
func (ps *PersistenceService) ReadSelectedSystemRecord(ctx context.Context, objectName, recordID string, selectedFields []string) (models.Record, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if strings.TrimSpace(recordID) == "" {
		return nil, apierr.Invalid("record identifier is required")
	}
	if len(selectedFields) == 0 {
		return nil, apierr.Invalid("at least one selected root field is required")
	}
	if ps == nil || ps.metadata == nil || ps.selectedRootProjection == nil {
		return nil, appErrors.NewInternalError("selected system-record reader is not configured", nil)
	}

	schema := ps.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil, apierr.Invalid(fmt.Sprintf("unknown object %q", objectName))
	}
	canonical, metadataByName, err := normalizeSelectedRootProjectionFields(schema, selectedFields)
	if err != nil {
		return nil, err
	}
	for _, name := range canonical {
		field := metadataByName[strings.ToLower(name)]
		if permissions.FieldReadBlockedByMetadata(&field) {
			return nil, apierr.Invalid(fmt.Sprintf("selected field %q is read-protected", field.APIName))
		}
	}
	systemCtx := constants.WithSecurityContext(ctx, constants.SystemContext,
		"Records: field-capped internal record read")
	record, err := ps.selectedRootProjection.ReadSelectedRootProjection(systemCtx, schema, recordID, canonical, nil)
	if err != nil {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return nil, ctxErr
		}
		return nil, appErrors.NewInternalError("selected system-record projection read failed", err)
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if record == nil {
		return nil, nil
	}
	return cloneSelectedSystemRecord(record, canonical), nil
}

func cloneSelectedSystemRecord(record models.Record, fields []string) models.Record {
	out := make(models.Record, len(fields)+1)
	if value, ok := record[models.ObjectTypeMetadataKey]; ok {
		out[models.ObjectTypeMetadataKey] = value
	}
	if value, ok := record[constants.FieldID]; ok {
		out[constants.FieldID] = value
	}
	for _, field := range fields {
		if value, ok := record[field]; ok {
			out[field] = value
		}
	}
	schemautil.CopyExactDecimalsForPresentFields(record, out)
	return out
}

// ReadSelectedRecord is the current in-process data/projection implementation
// for the bounded selected-root semantic operation. Trusted delegation,
// consistency receipts, numeric admission and wire errors remain outside this
// slice. It deliberately stays off contract.Deps and has no wire DTO; P2 owns
// the future public contract and transport.
func (ps *PersistenceService) ReadSelectedRecord(ctx context.Context, objectName, recordID string, selectedFields []string, user *models.UserSession) (*SelectedRootRecord, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if strings.TrimSpace(recordID) == "" {
		return nil, apierr.Invalid("record identifier is required")
	}
	if len(selectedFields) == 0 {
		return nil, apierr.Invalid("at least one selected root field is required")
	}

	stageStarted := perftiming.StartObservation(ctx)
	schema, err := ps.prepareOperation(ctx, objectName, constants.PermRead, user)
	perftiming.Observe(ctx, "selected_root.metadata_security_preparation", stageStarted)
	if err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	stageStarted = perftiming.StartObservation(ctx)
	canonical, metadataByName, err := normalizeSelectedRootFields(schema, selectedFields)
	perftiming.Observe(ctx, "selected_root.selection_normalization", stageStarted)
	if err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}

	stageStarted = perftiming.StartObservation(ctx)
	readable := make([]string, 0, len(canonical))
	for _, name := range canonical {
		if ps.permissions != nil && ps.permissions.IsFieldReadBlocked(ctx, objectName, name) {
			continue
		}
		fieldReadable := constants.GetSecurityContext(ctx).BypassesCRUDandFLS() ||
			constants.ScopedInternalAccessBypassesFLS(ctx, objectName, constants.PermRead)
		if !fieldReadable {
			var err error
			fieldReadable, err = ps.isReadableUserField(ctx, objectName, name, user)
			if err != nil {
				return nil, err
			}
		}
		if fieldReadable {
			readable = append(readable, name)
		}
	}
	perftiming.Observe(ctx, "selected_root.fls_preparation", stageStarted)
	if err := ctx.Err(); err != nil {
		return nil, err
	}

	if ps.selectedRootProjection == nil {
		return nil, appErrors.NewInternalError("selected-root projection reader is not configured", nil)
	}
	stageStarted = perftiming.StartObservation(ctx)
	record, err := ps.selectedRootProjection.ReadSelectedRootProjection(ctx, schema, recordID, readable, user)
	perftiming.Observe(ctx, "selected_root.projection_read", stageStarted)
	if err != nil {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return nil, ctxErr
		}
		var internal interface{ InternalCondition() bool }
		if errors.As(err, &internal) && internal.InternalCondition() {
			return nil, err
		}
		return nil, appErrors.NewInternalError("selected-root projection read failed", err)
	}
	// Cancellation wins at the projection boundary even when the reader
	// concurrently reports an empty result. Otherwise a canceled in-flight read
	// can be misclassified as a durable missing/RLS-masked outcome.
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if record == nil {
		// Query-level RLS is applied in the same SQL statement as Id lookup, so
		// absent and denied rows are intentionally indistinguishable here.
		return nil, apierr.NotFound("record not found")
	}

	stageStarted = perftiming.StartObservation(ctx)
	fields := make([]SelectedRootField, 0, len(readable))
	for _, name := range readable {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		value, present := record[name]
		if !present {
			continue
		}
		field, err := selectedRootFieldFromValue(metadataByName[strings.ToLower(name)], value, record)
		if err != nil {
			return nil, appErrors.NewInternalError("selected-root field projection disagrees with bound metadata", err)
		}
		fields = append(fields, field)
	}
	perftiming.Observe(ctx, "selected_root.typed_projection", stageStarted)
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return &SelectedRootRecord{RecordID: recordID, Fields: fields}, nil
}

func normalizeSelectedRootFields(schema *models.ObjectMetadata, selected []string) ([]string, map[string]models.FieldMetadata, error) {
	canonical, metadataByName, err := normalizeSelectedRootProjectionFields(schema, selected)
	if err != nil {
		return nil, nil, err
	}
	for _, name := range canonical {
		field := metadataByName[strings.ToLower(name)]
		if _, err := selectedRootKind(field); err != nil {
			return nil, nil, apierr.Invalid(fmt.Sprintf("selected field %q has no v1 scalar representation", field.APIName))
		}
	}
	return canonical, metadataByName, nil
}

// normalizeSelectedRootProjectionFields owns the selected-root query shape:
// canonical root fields that the query repository can project exactly from its
// resolved base-table or logical-view read target. Public typed contracts may
// impose a narrower representation after this step, but internal engine reads
// must not confuse that wire-type choice with query projectability (for example
// MultiPicklist and JSON are valid directly stored fields).
func normalizeSelectedRootProjectionFields(schema *models.ObjectMetadata, selected []string) ([]string, map[string]models.FieldMetadata, error) {
	if schema == nil {
		return nil, nil, apierr.Invalid("object metadata is unavailable")
	}
	metadataByName := make(map[string]models.FieldMetadata, len(schema.Fields))
	for _, field := range schema.Fields {
		key := strings.ToLower(strings.TrimSpace(field.APIName))
		if key == "" {
			continue
		}
		if _, exists := metadataByName[key]; exists {
			return nil, nil, appErrors.NewInternalError("ambiguous field metadata", nil)
		}
		metadataByName[key] = field
	}

	seen := make(map[string]struct{}, len(selected))
	canonical := make([]string, 0, len(selected))
	for _, raw := range selected {
		name := strings.TrimSpace(raw)
		if name == "" || strings.Contains(name, ".") {
			return nil, nil, apierr.Invalid("selected fields must be non-blank root field identifiers")
		}
		field, ok := metadataByName[strings.ToLower(name)]
		if !ok {
			return nil, nil, apierr.Invalid(fmt.Sprintf("unknown selected field %q", name))
		}
		if field.IsVirtual || field.IsCompound {
			return nil, nil, apierr.Invalid(fmt.Sprintf("selected field %q is not a scalar root field", field.APIName))
		}
		key := strings.ToLower(field.APIName)
		if _, exists := seen[key]; exists {
			continue
		}
		seen[key] = struct{}{}
		canonical = append(canonical, field.APIName)
	}
	sort.Strings(canonical)
	return canonical, metadataByName, nil
}

func selectedRootKind(field models.FieldMetadata) (SelectedRootValueKind, error) {
	switch schemautil.EffectiveFieldType(field) {
	case constants.FieldTypeString, constants.FieldTypePicklist, constants.FieldTypeEmail,
		constants.FieldTypePhone, constants.FieldTypeTextarea, constants.FieldTypeUrl,
		constants.FieldTypeRichtextarea, constants.FieldTypeAutonumber,
		constants.FieldTypeEncryptedString, constants.FieldTypePassword:
		return SelectedRootString, nil
	case constants.FieldTypeBoolean:
		return SelectedRootBoolean, nil
	case constants.SchemaFieldType("int"), constants.SchemaFieldType("integer"), constants.SchemaFieldType("long"):
		return SelectedRootInteger, nil
	case constants.FieldTypeDouble, constants.FieldTypeCurrency, constants.FieldTypePercent:
		return SelectedRootDecimal, nil
	case constants.FieldTypeDate:
		return SelectedRootDate, nil
	case constants.FieldTypeDatetime:
		return SelectedRootDateTime, nil
	case constants.FieldTypeReference, constants.FieldTypeMasterDetail:
		return SelectedRootReference, nil
	default:
		return "", fmt.Errorf("unsupported field type %q", field.Type)
	}
}

func selectedRootFieldFromValue(metadata models.FieldMetadata, value any, record models.Record) (SelectedRootField, error) {
	kind, err := selectedRootKind(metadata)
	if err != nil {
		return SelectedRootField{}, err
	}
	field := SelectedRootField{Name: metadata.APIName, Kind: kind}
	if value == nil {
		field.Null = true
		return field, nil
	}

	switch kind {
	case SelectedRootString:
		v, ok := value.(string)
		if !ok {
			return field, fmt.Errorf("%s: want string, got %T", metadata.APIName, value)
		}
		field.String = &v
	case SelectedRootBoolean:
		v, ok := value.(bool)
		if !ok {
			return field, fmt.Errorf("%s: want boolean, got %T", metadata.APIName, value)
		}
		field.Boolean = &v
	case SelectedRootInteger:
		v, ok := selectedRootInt64(value)
		if !ok {
			return field, fmt.Errorf("%s: want signed integer, got %T", metadata.APIName, value)
		}
		field.Integer = &v
	case SelectedRootDecimal:
		exact, ok := schemautil.ExactDecimalForField(record, metadata.APIName)
		if !ok {
			return field, fmt.Errorf("%s: DECIMAL-backed value has no exact carrier", metadata.APIName)
		}
		coefficient, scale, ok := exact.CoefficientAndScale()
		if !ok {
			return field, fmt.Errorf("%s: invalid exact decimal carrier", metadata.APIName)
		}
		field.Decimal = &SelectedRootDecimalValue{Coefficient: coefficient, Scale: scale}
	case SelectedRootDate:
		v, ok := value.(string)
		if !ok {
			return field, fmt.Errorf("%s: want date string, got %T", metadata.APIName, value)
		}
		parsed, err := time.Parse("2006-01-02", v)
		if err != nil || parsed.Format("2006-01-02") != v {
			return field, fmt.Errorf("%s: invalid Gregorian date", metadata.APIName)
		}
		field.Date = &v
	case SelectedRootDateTime:
		v, err := selectedRootDateTime(value)
		if err != nil {
			return field, fmt.Errorf("%s: %w", metadata.APIName, err)
		}
		field.DateTime = &v
	case SelectedRootReference:
		v, ok := value.(string)
		if !ok {
			return field, fmt.Errorf("%s: want reference identifier, got %T", metadata.APIName, value)
		}
		field.Reference = &v
	}
	return field, nil
}

func selectedRootInt64(value any) (int64, bool) {
	switch v := value.(type) {
	case int:
		return int64(v), true
	case int8:
		return int64(v), true
	case int16:
		return int64(v), true
	case int32:
		return int64(v), true
	case int64:
		return v, true
	case uint:
		if uint64(v) <= math.MaxInt64 {
			return int64(v), true
		}
	case uint8:
		return int64(v), true
	case uint16:
		return int64(v), true
	case uint32:
		return int64(v), true
	case uint64:
		if v <= math.MaxInt64 {
			return int64(v), true
		}
	}
	return 0, false
}

func selectedRootDateTime(value any) (string, error) {
	var parsed time.Time
	switch v := value.(type) {
	case time.Time:
		parsed = v
	case string:
		var err error
		parsed, err = time.Parse(time.RFC3339Nano, v)
		if err != nil {
			return "", fmt.Errorf("invalid RFC3339 datetime")
		}
	default:
		return "", fmt.Errorf("want datetime, got %T", value)
	}
	return parsed.UTC().Format(time.RFC3339Nano), nil
}