package records

import (
	"context"
	"errors"
	"net/http"
	"sort"
	"strings"
	"sync"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recorddisplay"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/kernel/pkg/telemetry"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordviewv1"
	"github.com/rs/zerolog/log"
)

// RuntimeRecordViewEngine is the existing CRUD/RLS/FLS owner behind the
// independent Runtime record-view read. It is deliberately an in-process
// adapter seam, not a new public capability plane.
type RuntimeRecordViewEngine interface {
	RetrieveWithAccess(context.Context, string, string, *models.UserSession) (models.Record, *semantic.RecordAccess, error)
	RetrieveByField(context.Context, string, string, string, *models.UserSession) (models.Record, error)
	FieldReadable(context.Context, string, string, *models.UserSession) (bool, error)
}

// RuntimeRecordViewReferenceReader resolves the bounded display projection for
// related records. It deliberately returns only Id + the authorized display
// field closure. The implementation is owned by QueryService so point and list
// reads share one query-level RLS planner instead of re-running generic
// RetrieveMany's full-record CRUD/RLS/FLS/grant pipeline for every target.
type RuntimeRecordViewReferenceReader interface {
	ReadReferenceDisplayProjection(context.Context, *models.ObjectMetadata, []string, *models.UserSession) (map[string]models.Record, error)
}

type runtimeRecordViewPersistence struct{ service *PersistenceService }

func (p runtimeRecordViewPersistence) RetrieveWithAccess(ctx context.Context, objectID, recordID string, user *models.UserSession) (models.Record, *semantic.RecordAccess, error) {
	record, access, err := p.service.RetrieveWithAccess(ctx, objectID, recordID, user)
	if err != nil || access == nil {
		return record, nil, err
	}
	return record, &semantic.RecordAccess{
		Read: access.HasReadAccess, Update: access.HasEditAccess,
		Delete: access.HasDeleteAccess, Locked: access.IsLocked,
	}, nil
}

func (p runtimeRecordViewPersistence) RetrieveByField(ctx context.Context, objectID, fieldID, value string, user *models.UserSession) (models.Record, error) {
	return p.service.RetrieveByField(ctx, objectID, fieldID, value, user)
}

func (p runtimeRecordViewPersistence) FieldReadable(ctx context.Context, objectID, fieldID string, user *models.UserSession) (bool, error) {
	if constants.GetSecurityContext(ctx).BypassesCRUDandFLS() || constants.ScopedInternalAccessBypassesFLS(ctx, objectID, constants.PermRead) {
		return true, nil
	}
	return p.service.isReadableUserField(ctx, objectID, fieldID, user)
}

// RuntimeRecordViewMetadata deliberately exposes only the immutable view
// selected by ServingAdmission. A record-view request may never fall back to a
// global metadata cache or repository lookup.
type RuntimeRecordViewMetadata interface {
	BoundRuntimeMetadataView(context.Context) (*catalogsnapshot.RuntimeMetadataView, bool)
}

// RuntimeRecordViewActivity preserves the current bounded access-decoration
// side effect without pulling an HTTP or L1 dependency into Runtime.
type RuntimeRecordViewActivity interface {
	TrackRecordAccess(context.Context, string, string, string)
}

type RuntimeRecordViewConfig struct {
	Engine          RuntimeRecordViewEngine
	ReferenceReader RuntimeRecordViewReferenceReader
	Metadata        RuntimeRecordViewMetadata
	Activity        RuntimeRecordViewActivity
}

type RuntimeRecordViewResult struct {
	RecordID               string
	Root                   models.Record
	Access                 *semantic.RecordAccess
	ReferenceDisplays      []semantic.ReferenceDisplay
	AttachmentSourceFields []semantic.AttachmentSourceField
}

// Reference targets are independent once the authorized root and immutable
// Catalog view are fixed. Keep concurrency deliberately small so remote-DB
// latency can overlap without turning one record view into unbounded fanout.
const runtimeRecordViewReferenceParallelism = 4

// RuntimeRecordViewService owns the complete bounded root/view read. Product
// relationship envelopes and attachment URL grammar remain in L1. Attachment
// resolution itself is intentionally not an operation on this service: it is
// a separate proof-bearing capability.
type RuntimeRecordViewService struct {
	engine          RuntimeRecordViewEngine
	referenceReader RuntimeRecordViewReferenceReader
	metadata        RuntimeRecordViewMetadata
	activity        RuntimeRecordViewActivity
}

func NewRuntimeRecordViewService(config RuntimeRecordViewConfig) (*RuntimeRecordViewService, error) {
	if config.Engine == nil || config.ReferenceReader == nil || config.Metadata == nil || config.Activity == nil {
		return nil, errors.New("record-view runtime: engine, reference reader, metadata and activity are required")
	}
	return &RuntimeRecordViewService{
		engine: config.Engine, referenceReader: config.ReferenceReader,
		metadata: config.Metadata, activity: config.Activity,
	}, nil
}

func NewRuntimeRecordViewServiceFromPersistence(persistence *PersistenceService, references RuntimeRecordViewReferenceReader, metadata RuntimeRecordViewMetadata, activity RuntimeRecordViewActivity) (*RuntimeRecordViewService, error) {
	if persistence == nil {
		return nil, errors.New("record-view runtime: persistence is required")
	}
	return NewRuntimeRecordViewService(RuntimeRecordViewConfig{
		Engine: runtimeRecordViewPersistence{service: persistence}, ReferenceReader: references,
		Metadata: metadata, Activity: activity,
	})
}

func (s *RuntimeRecordViewService) ReadRecordView(ctx context.Context, request semantic.ReadRequest, user *models.UserSession) (*RuntimeRecordViewResult, error) {
	totalStart := time.Now()
	var rootElapsed, compoundElapsed, referenceElapsed time.Duration
	var rootSQLTime, rootSQLMax, compoundSQLTime, compoundSQLMax, referenceSQLTime, referenceSQLMax time.Duration
	var rootSQL, compoundSQL, referenceSQL uint64
	defer func() {
		log.Debug().
			Str("event", "runtime_record_view_phase_profile").
			Str("object", request.ObjectID).
			Uint64("root_sql", rootSQL).
			Int64("root_sql_elapsed_ms", rootSQLTime.Milliseconds()).
			Int64("root_sql_max_ms", rootSQLMax.Milliseconds()).
			Int64("root_elapsed_ms", rootElapsed.Milliseconds()).
			Uint64("compound_sql", compoundSQL).
			Int64("compound_sql_elapsed_ms", compoundSQLTime.Milliseconds()).
			Int64("compound_sql_max_ms", compoundSQLMax.Milliseconds()).
			Int64("compound_elapsed_ms", compoundElapsed.Milliseconds()).
			Uint64("reference_sql", referenceSQL).
			Int64("reference_sql_elapsed_ms", referenceSQLTime.Milliseconds()).
			Int64("reference_sql_max_ms", referenceSQLMax.Milliseconds()).
			Int64("reference_elapsed_ms", referenceElapsed.Milliseconds()).
			Int64("total_elapsed_ms", time.Since(totalStart).Milliseconds()).
			Msg("Runtime record-view phase profile")
	}()
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	view, schema, err := s.boundObject(ctx, request.ObjectID)
	if err != nil {
		return nil, err
	}

	var root models.Record
	var access *semantic.RecordAccess
	rootCtx, rootCounter := telemetry.WithSQLStatementCounter(ctx)
	rootStart := time.Now()
	switch request.Selector.Kind {
	case semantic.SelectorRecordID:
		root, access, err = s.engine.RetrieveWithAccess(rootCtx, schema.APIName, request.Selector.RecordID, user)
	case semantic.SelectorFieldEqual:
		field := models.FindField(schema, request.Selector.FieldID)
		if field == nil || field.IsVirtual || field.IsCompound {
			return nil, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentField}
		}
		// A hidden selector must not become an existence oracle. Check FLS before
		// any field-equality repository operation.
		readable, readableErr := s.engine.FieldReadable(rootCtx, schema.APIName, field.APIName, user)
		if readableErr != nil {
			return nil, readableErr
		}
		if !readable {
			return nil, &semantic.Failure{Category: semantic.FailureForbidden}
		}
		root, err = s.engine.RetrieveByField(rootCtx, schema.APIName, field.APIName, request.Selector.FieldValue, user)
	default:
		return nil, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentOperation}
	}
	rootElapsed = time.Since(rootStart)
	rootSQL = rootCounter.Count()
	rootSQLTime, rootSQLMax = rootCounter.TotalDuration(), rootCounter.MaxDuration()
	if err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if root == nil {
		return nil, &semantic.Failure{Category: semantic.FailureMissing}
	}
	recordID, _ := root[constants.FieldID].(string)
	if strings.TrimSpace(recordID) == "" {
		if request.Selector.Kind == semantic.SelectorRecordID {
			recordID = request.Selector.RecordID
		} else {
			return nil, appErrors.NewInternalError("record-view row has no record identifier", nil)
		}
	}

	result := &RuntimeRecordViewResult{RecordID: recordID, Root: root, Access: access}
	if request.Projection == semantic.ProjectionEnrichedView {
		compoundCtx, compoundCounter := telemetry.WithSQLStatementCounter(ctx)
		compoundStart := time.Now()
		if err := s.synthesizeVirtualCompoundFields(compoundCtx, schema, root, user); err != nil {
			return nil, err
		}
		compoundElapsed = time.Since(compoundStart)
		compoundSQL = compoundCounter.Count()
		compoundSQLTime, compoundSQLMax = compoundCounter.TotalDuration(), compoundCounter.MaxDuration()

		referenceCtx, referenceCounter := telemetry.WithSQLStatementCounter(ctx)
		referenceStart := time.Now()
		result.ReferenceDisplays, err = s.referenceDisplays(referenceCtx, view, schema, root, user)
		referenceElapsed = time.Since(referenceStart)
		referenceSQL = referenceCounter.Count()
		referenceSQLTime, referenceSQLMax = referenceCounter.TotalDuration(), referenceCounter.MaxDuration()
		if err != nil {
			return nil, err
		}
		result.AttachmentSourceFields = recordViewAttachmentSourceFields(view, schema.APIName, root)
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if user != nil {
		s.activity.TrackRecordAccess(ctx, user.ID, schema.APIName, recordID)
	}
	return result, nil
}

func (s *RuntimeRecordViewService) boundObject(ctx context.Context, objectID string) (*catalogsnapshot.RuntimeMetadataView, *models.ObjectMetadata, error) {
	view, ok := s.metadata.BoundRuntimeMetadataView(ctx)
	if !ok || !view.SupportsRecordViewServing() {
		return nil, nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	schema, ok := view.Object(objectID)
	if !ok || schema == nil {
		return nil, nil, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentObjectID}
	}
	return view, schema, nil
}

func (s *RuntimeRecordViewService) synthesizeVirtualCompoundFields(ctx context.Context, schema *models.ObjectMetadata, root models.Record, user *models.UserSession) error {
	visible := map[string]bool{}
	for _, field := range schema.Fields {
		if !field.IsVirtual || field.VirtualKind != constants.VirtualFieldCompound {
			continue
		}
		for _, candidate := range append([]string{field.APIName}, schemautil.CompoundMemberNames(schema, field.APIName)...) {
			key := strings.ToLower(candidate)
			if _, checked := visible[key]; checked {
				continue
			}
			readable, err := s.engine.FieldReadable(ctx, schema.APIName, candidate, user)
			if err != nil {
				return err
			}
			visible[key] = readable
		}
	}
	schemautil.SynthesizeVirtualCompoundFields(schema, root, func(field string) bool { return visible[strings.ToLower(field)] })
	return nil
}

func (s *RuntimeRecordViewService) referenceDisplays(ctx context.Context, view *catalogsnapshot.RuntimeMetadataView, schema *models.ObjectMetadata, root models.Record, user *models.UserSession) ([]semantic.ReferenceDisplay, error) {
	type reference struct{ field, relationship, object, id string }
	var refs []reference
	idsByObject := map[string]map[string]struct{}{}
	for _, field := range schema.Fields {
		if !constants.IsReferenceType(field.Type) {
			continue
		}
		id, ok := root[field.APIName].(string)
		if !ok || strings.TrimSpace(id) == "" {
			continue
		}
		target := recordViewReferenceTarget(field)
		if target == "" {
			continue
		}
		if recordViewMetadataOwnedReference(field.APIName, target) {
			// RecordType is Catalog metadata, not a record-backed object. The
			// immutable record-view projection deliberately does not carry the
			// mutation profile's RecordType rows, and querying a physical
			// RecordType table would create a second, incompatible authority.
			// Preserve RecordTypeId on the authorized root without inventing an
			// optional relationship display from the Records repository.
			continue
		}
		if _, present := view.Object(target); !present {
			// A relationship that is not in the immutable closure is not a
			// record-view fact for this serving request.
			continue
		}
		relationship := strings.TrimSuffix(field.APIName, "Id")
		if field.RelationshipName != nil && strings.TrimSpace(*field.RelationshipName) != "" {
			relationship = *field.RelationshipName
		}
		refs = append(refs, reference{field: field.APIName, relationship: relationship, object: target, id: id})
		if idsByObject[target] == nil {
			idsByObject[target] = map[string]struct{}{}
		}
		idsByObject[target][id] = struct{}{}
	}

	objects := make([]string, 0, len(idsByObject))
	for object := range idsByObject {
		objects = append(objects, object)
	}
	sort.Strings(objects)

	type referenceResult struct {
		object string
		rows   map[string]models.Record
	}
	workerCtx, cancelWorkers := context.WithCancel(ctx)
	defer cancelWorkers()
	jobs := make(chan string, len(objects))
	results := make(chan referenceResult, len(objects))
	var firstErr error
	var firstErrOnce sync.Once
	workerCount := min(runtimeRecordViewReferenceParallelism, len(objects))
	var workers sync.WaitGroup
	workers.Add(workerCount)
	for range workerCount {
		go func() {
			defer workers.Done()
			for object := range jobs {
				rows, retrieveErr := s.referenceDisplayRows(workerCtx, view, schema, object, idsByObject[object], user)
				if retrieveErr != nil {
					firstErrOnce.Do(func() {
						firstErr = retrieveErr
						cancelWorkers()
					})
				}
				results <- referenceResult{object: object, rows: rows}
			}
		}()
	}
	for _, object := range objects {
		jobs <- object
	}
	close(jobs)
	workers.Wait()
	close(results)

	rowsByObject := make(map[string]map[string]models.Record, len(objects))
	for result := range results {
		rowsByObject[result.object] = result.rows
	}
	if firstErr != nil {
		return nil, firstErr
	}

	result := make([]semantic.ReferenceDisplay, 0, len(refs))
	for _, ref := range refs {
		target, ok := view.Object(ref.object)
		if !ok {
			continue
		}
		name := recordViewDisplayName(target, rowsByObject[ref.object][ref.id])
		if name == "" {
			continue
		}
		result = append(result, semantic.ReferenceDisplay{
			SourceFieldID: ref.field, RelationshipID: ref.relationship, TargetObjectID: ref.object,
			TargetRecordID: ref.id, DisplayName: name,
		})
	}
	sort.Slice(result, func(i, j int) bool { return result[i].SourceFieldID < result[j].SourceFieldID })
	return result, nil
}

func (s *RuntimeRecordViewService) referenceDisplayRows(ctx context.Context, view *catalogsnapshot.RuntimeMetadataView, source *models.ObjectMetadata, object string, idSet map[string]struct{}, user *models.UserSession) (map[string]models.Record, error) {
	ids := make([]string, 0, len(idSet))
	for id := range idSet {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	objectCtx, objectCounter := telemetry.WithSQLStatementCounter(ctx)
	objectStart := time.Now()
	target, targetPresent := view.Object(object)
	if !targetPresent {
		return nil, nil
	}
	rows, retrieveErr := s.referenceReader.ReadReferenceDisplayProjection(objectCtx, target, ids, user)
	unavailable := recordViewDisplayUnavailable(retrieveErr)
	log.Debug().
		Str("event", "runtime_record_view_reference_profile").
		Str("source_object", source.APIName).
		Str("target_object", object).
		Int("reference_count", len(ids)).
		Uint64("sql", objectCounter.Count()).
		Int64("sql_elapsed_ms", objectCounter.TotalDuration().Milliseconds()).
		Int64("sql_max_ms", objectCounter.MaxDuration().Milliseconds()).
		Int64("elapsed_ms", time.Since(objectStart).Milliseconds()).
		Bool("unavailable", unavailable).
		Msg("Runtime record-view reference profile")
	if unavailable {
		// An inaccessible related record is not an error for an already
		// authorized root: hide its display exactly as a missing/RLS-masked
		// relation. Other failures (including cancellation and database failures)
		// must not become a partial successful view.
		return nil, nil
	}
	if retrieveErr != nil {
		return nil, retrieveErr
	}
	return rows, nil
}

func recordViewDisplayUnavailable(err error) bool {
	var status interface{ HTTPStatus() int }
	return errors.As(err, &status) && (status.HTTPStatus() == http.StatusForbidden || status.HTTPStatus() == http.StatusNotFound)
}

func recordViewReferenceTarget(field models.FieldMetadata) string {
	if len(field.ReferenceTo) != 1 || strings.TrimSpace(field.ReferenceTo[0]) == "" {
		return ""
	}
	return field.ReferenceTo[0]
}

func recordViewMetadataOwnedReference(fieldAPIName, targetObjectAPIName string) bool {
	return strings.EqualFold(fieldAPIName, constants.FieldRecordTypeID) &&
		strings.EqualFold(targetObjectAPIName, constants.ObjectRecordType)
}

func recordViewDisplayName(schema *models.ObjectMetadata, record models.Record) string {
	return recorddisplay.LookupName(schema, record)
}

func recordViewAttachmentSourceFields(view *catalogsnapshot.RuntimeMetadataView, objectID string, root models.Record) []semantic.AttachmentSourceField {
	bindings := view.AttachmentSourceFieldsForObject(objectID)
	result := make([]semantic.AttachmentSourceField, 0, len(bindings))
	for _, binding := range bindings {
		value, ok := root[binding.FieldAPIName].(string)
		if !ok || strings.TrimSpace(value) == "" {
			continue
		}
		result = append(result, semantic.AttachmentSourceField{SourceFieldID: binding.FieldAPIName, ResolverID: binding.ResolverID})
	}
	sort.Slice(result, func(i, j int) bool { return result[i].SourceFieldID < result[j].SourceFieldID })
	return result
}