package runtimeapi

import (
	"context"
	"errors"
	"sync"

	"github.com/nexuscrm/kernel/pkg/telemetry"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

const relatedCountExecutionConcurrency = 8

// RelatedCountServer adapts one bounded count batch to the current Records and
// Query engines. The public CRM grammar remains in L1; this server sees only
// the normalized neutral operation.
type RelatedCountServer struct {
	runtimev1.UnimplementedRelatedRecordCountServiceServer
	access   RecordReadAccessEvaluator
	query    RelatedCountQueryEngine
	receipts ReceiptSource
	metrics  *recordsMetrics
}

// RelatedCountQueryEngine is the count-only L0 execution boundary. It avoids
// adapting a count back through the complete row query and projection surface.
type RelatedCountQueryEngine interface {
	CountWithSpec(context.Context, queryspec.CountSpec, *models.UserSession) (int, error)
}

func NewRelatedCountServer(access RecordReadAccessEvaluator, query RecordQueryEngine, receipts ReceiptSource, metrics *recordsMetrics) (*RelatedCountServer, error) {
	if access == nil || query == nil {
		return nil, errors.New("runtime-api: related-count access and query engines are required")
	}
	counter, ok := query.(RelatedCountQueryEngine)
	if !ok {
		return nil, errors.New("runtime-api: related-count exact count engine is required")
	}
	if metrics == nil {
		metrics = newRecordsMetrics(nil)
	}
	return &RelatedCountServer{access: access, query: counter, receipts: receipts, metrics: metrics}, nil
}

func (s *RelatedCountServer) CountRelatedRecords(ctx context.Context, wireRequest *runtimev1.CountRelatedRecordsRequest) (*runtimev1.CountRelatedRecordsResponse, error) {
	trusted, ok := trustedCallFromContext(ctx)
	if !ok {
		return nil, mapping.ToRelatedCountStatusError(&relatedsemantic.Failure{Category: relatedsemantic.FailureUnauthenticated})
	}
	request, err := mapping.FromWireCountRelatedRecordsRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToRelatedCountStatusError(err)
	}
	receipts, err := relatedCountExecutionReceipts(ctx, s.receipts)
	if err != nil {
		return nil, mapping.ToRelatedCountStatusError(err)
	}
	receipts = receiptsForCapability(receipts, relatedsemantic.CapabilityName)
	if request.MinCatalogRevision != nil && *request.MinCatalogRevision > receipts.CatalogRevision {
		return nil, mapping.ToRelatedCountStatusError(&relatedsemantic.Failure{
			Category: relatedsemantic.FailureStaleState, Receipts: &receipts,
			StaleAxis: relatedsemantic.StaleAxisCatalog, RequiredRevision: *request.MinCatalogRevision,
			ServedRevision: receipts.CatalogRevision,
		})
	}
	if _, ok := RuntimeMetadataViewFromContext(ctx); !ok {
		return nil, mapping.ToRelatedCountStatusError(&relatedsemantic.Failure{Category: relatedsemantic.FailureUnavailable})
	}
	accessView, ok := runtimeRecordAccessMetadataViewFromContext(ctx)
	if !ok {
		return nil, mapping.ToRelatedCountStatusError(&relatedsemantic.Failure{Category: relatedsemantic.FailureUnavailable})
	}
	response := relatedsemantic.Response{Results: make([]relatedsemantic.Result, len(request.Counts)), Receipts: receipts}
	if len(request.Counts) == 0 {
		wireResponse, err := mapping.ToWireCountRelatedRecordsResponse(request, response)
		if err != nil {
			return nil, mapping.ToRelatedCountStatusError(err)
		}
		return wireResponse, nil
	}

	user := trusted.UserSession()
	parentIDs := make([]string, 0, len(request.Counts))
	seenParentIDs := make(map[string]struct{}, len(request.Counts))
	for index := range request.Counts {
		parentID := request.Counts[index].ParentRecordID
		if _, duplicate := seenParentIDs[parentID]; duplicate {
			continue
		}
		seenParentIDs[parentID] = struct{}{}
		parentIDs = append(parentIDs, parentID)
	}
	accessCtx := withRuntimeRecordsMetadataView(ctx, accessView)
	accessCtx, accessSQL := telemetry.WithSQLStatementCounter(accessCtx)
	parentAccess, err := s.access.EvaluateRecordReadAccess(accessCtx, nil, parentIDs, &user)
	s.metrics.observeRepository(recordsOperationRelatedCount, accessSQL.Count())
	if err != nil {
		if relatedCountParentAccessDenied(err) {
			// Object Read and RLS denial are parent visibility facts, not
			// complete-call authority failures. Mask every unresolved parent as
			// the same counted zero as a missing parent.
			parentAccess = map[string]contract.RecordAccess{}
		} else {
			return nil, mapping.ToRelatedCountStatusError(classifyRelatedCountCompleteError(err))
		}
	}

	type countResult struct {
		index  int
		result relatedsemantic.Result
		err    error
	}
	completed := make(chan countResult, len(request.Counts))
	semaphore := make(chan struct{}, relatedCountExecutionConcurrency)
	var workers sync.WaitGroup
	for index, count := range request.Counts {
		workers.Add(1)
		go func(index int, count relatedsemantic.CountSpec) {
			defer workers.Done()
			defer func() {
				if recover() != nil {
					completed <- countResult{index: index, err: &relatedsemantic.Failure{Category: relatedsemantic.FailureInternal}}
				}
			}()
			if decision, readable := parentAccess[count.ParentRecordID]; !readable || !decision.Read {
				completed <- countResult{index: index, result: relatedsemantic.Result{Index: uint32(index), Outcome: relatedsemantic.CountOutcomeCounted}}
				return
			}
			select {
			case semaphore <- struct{}{}:
				defer func() { <-semaphore }()
			case <-ctx.Done():
				completed <- countResult{index: index, err: ctx.Err()}
				return
			}
			match := count.MatchValue
			if match == "" {
				match = count.ParentRecordID
			}
			queryCtx, querySQL := telemetry.WithSQLStatementCounter(ctx)
			countValue, queryErr := s.query.CountWithSpec(queryCtx, queryspec.CountSpec{
				ObjectAPIName: count.ChildObjectID,
				Filter:        queryspec.FieldPredicate(count.RelationshipField, "=", match),
			}, &user)
			s.metrics.observeRepository(recordsOperationRelatedCount, querySQL.Count())
			if queryErr != nil {
				if relatedCountItemNotAvailable(queryErr) {
					completed <- countResult{index: index, result: relatedsemantic.Result{Index: uint32(index), Outcome: relatedsemantic.CountOutcomeNotAvailable}}
					return
				}
				completed <- countResult{index: index, err: classifyRelatedCountCompleteError(queryErr)}
				return
			}
			if countValue == 0 {
				completed <- countResult{index: index, result: relatedsemantic.Result{Index: uint32(index), Outcome: relatedsemantic.CountOutcomeCounted}}
				return
			}
			if countValue < 0 {
				completed <- countResult{index: index, err: &relatedsemantic.Failure{Category: relatedsemantic.FailureInternal}}
				return
			}
			completed <- countResult{index: index, result: relatedsemantic.Result{Index: uint32(index), Outcome: relatedsemantic.CountOutcomeCounted, Count: uint64(countValue)}}
		}(index, count)
	}
	workers.Wait()
	close(completed)
	var failures = make([]error, len(request.Counts))
	for result := range completed {
		response.Results[result.index] = result.result
		failures[result.index] = result.err
	}
	for _, failure := range failures {
		if failure != nil {
			return nil, mapping.ToRelatedCountStatusError(failure)
		}
	}
	wireResponse, err := mapping.ToWireCountRelatedRecordsResponse(request, response)
	if err != nil {
		return nil, mapping.ToRelatedCountStatusError(err)
	}
	return wireResponse, nil
}

func relatedCountParentAccessDenied(err error) bool {
	classified := classifyRecordAccessEngineError(err)
	var failure *relatedsemantic.Failure
	return errors.As(classified, &failure) && failure.Category == relatedsemantic.FailureForbidden
}

func relatedCountExecutionReceipts(ctx context.Context, source ReceiptSource) (relatedsemantic.Receipts, error) {
	if scope, ok := runtimeExecutionScopeFromContext(ctx); ok {
		return scope.receipts, nil
	}
	if source == nil {
		return relatedsemantic.Receipts{}, &relatedsemantic.Failure{Category: relatedsemantic.FailureUnavailable}
	}
	return source.Receipts(ctx)
}

func relatedCountItemNotAvailable(err error) bool {
	classified := classifyQueryEngineError(err)
	var failure *querysemantic.Failure
	if !errors.As(classified, &failure) {
		return false
	}
	return failure.Category == querysemantic.FailureInvalidRequest || failure.Category == querysemantic.FailureForbidden || failure.Category == querysemantic.FailureMissing
}

func classifyRelatedCountCompleteError(err error) error {
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	classified := classifyQueryEngineError(err)
	var failure *querysemantic.Failure
	if !errors.As(classified, &failure) {
		return &relatedsemantic.Failure{Category: relatedsemantic.FailureInternal}
	}
	switch failure.Category {
	case querysemantic.FailureUnauthenticated:
		return &relatedsemantic.Failure{Category: relatedsemantic.FailureUnauthenticated}
	case querysemantic.FailureForbidden:
		return &relatedsemantic.Failure{Category: relatedsemantic.FailureForbidden}
	case querysemantic.FailureUnavailable:
		return &relatedsemantic.Failure{Category: relatedsemantic.FailureUnavailable}
	case querysemantic.FailureInternal:
		return &relatedsemantic.Failure{Category: relatedsemantic.FailureInternal}
	default:
		return &relatedsemantic.Failure{Category: relatedsemantic.FailureInternal}
	}
}