package runtimeapi

import (
	"context"
	"errors"
	"sort"
	"strings"
	"sync"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

type RecordSearchEngine interface {
	SearchWithSpec(context.Context, queryspec.SearchSpec, *models.UserSession) ([]contract.Record, error)
}

type SearchScopeOrderEngine interface {
	ScopeOrder(
		context.Context,
		*models.UserSession,
		[]string,
		permissionstatev1.Version,
	) ([]string, error)
}

type RecordSearchServer struct {
	runtimev1.UnimplementedRecordSearchServiceServer
	engine     RecordSearchEngine
	scopeOrder SearchScopeOrderEngine
}

const maxParallelObjectSearches = 4

type objectSearchResult struct {
	records []searchsemantic.Record
	err     error
}

func NewRecordSearchServer(
	engine RecordSearchEngine,
	scopeOrder SearchScopeOrderEngine,
) (*RecordSearchServer, error) {
	if engine == nil || scopeOrder == nil {
		return nil, errors.New(
			"runtime record search: record and scope-order engines are required",
		)
	}
	return &RecordSearchServer{engine: engine, scopeOrder: scopeOrder}, nil
}

func (s *RecordSearchServer) SearchRecords(
	ctx context.Context,
	wireRequest *runtimev1.SearchRecordsRequest,
) (*runtimev1.SearchRecordsResponse, error) {
	request, err := mapping.FromWireSearchRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}
	scope, ok := runtimeExecutionScopeFromContext(ctx)
	view, viewOK := RuntimeMetadataViewFromContext(ctx)
	trusted, trustedOK := trustedCallFromContext(ctx)
	if !ok || !viewOK || !trustedOK {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureUnavailable},
			searchsemantic.CapabilityName,
		)
	}
	if err := requireCatalogRevision(request.MinCatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}

	user := trusted.UserSession()
	response := searchsemantic.Response{
		Receipts: receiptsForCapability(scope.receipts, searchsemantic.CapabilityName),
	}
	if len(request.ObjectPlans) > 0 {
		results, searchErr := s.searchObjectPlans(ctx, request, view, &user)
		if searchErr != nil {
			return nil, mapping.ToPlatformStatusError(searchErr, searchsemantic.CapabilityName)
		}
		response.Records = results
		wireResponse, wireErr := mapping.ToWireSearchResponse(request, response)
		if wireErr != nil {
			return nil, mapping.ToPlatformStatusError(wireErr, searchsemantic.CapabilityName)
		}
		return wireResponse, nil
	}
	remaining := int(request.Limit)
	for _, objectID := range request.ObjectIDs {
		if remaining == 0 {
			break
		}
		if err := ctx.Err(); err != nil {
			return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
		}
		if _, found := view.Object(objectID); !found {
			return nil, mapping.ToPlatformStatusError(
				&recordsemantic.Failure{Category: recordsemantic.FailureForbidden},
				searchsemantic.CapabilityName,
			)
		}
		records, searchErr := s.engine.SearchWithSpec(ctx, queryspec.SearchSpec{
			ObjectAPIName: objectID, Term: request.Term,
			SearchFields: append([]string(nil), request.SearchFields...),
			ReturnFields: append([]string(nil), request.SelectedFields...),
			Limit:        remaining,
		}, &user)
		if searchErr != nil {
			return nil, mapping.ToPlatformStatusError(
				classifyPlatformEngineError(
					ctx, searchErr, commonsemantic.InvalidComponentQueryPlan,
				),
				searchsemantic.CapabilityName,
			)
		}
		for _, record := range records {
			projected, projectErr := semanticQueryRecord(models.SObject(record), view, objectID, 1)
			if projectErr != nil {
				return nil, mapping.ToPlatformStatusError(projectErr, searchsemantic.CapabilityName)
			}
			searchRecord, projectErr := platformSearchRecord(
				projected, request.SelectedFields,
			)
			if projectErr != nil {
				return nil, mapping.ToPlatformStatusError(projectErr, searchsemantic.CapabilityName)
			}
			response.Records = append(response.Records, searchRecord)
			remaining--
			if remaining == 0 {
				break
			}
		}
	}
	wireResponse, err := mapping.ToWireSearchResponse(request, response)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}
	return wireResponse, nil
}

func (s *RecordSearchServer) searchObjectPlans(
	ctx context.Context,
	request searchsemantic.Request,
	view *catalogsnapshot.RuntimeMetadataView,
	user *models.UserSession,
) ([]searchsemantic.Record, error) {
	plans := searchsemantic.ObjectPlansForExecution(request)
	for _, plan := range plans {
		if _, found := view.Object(plan.ObjectID); !found {
			return nil, &recordsemantic.Failure{Category: recordsemantic.FailureForbidden}
		}
	}

	flattened := make([]searchsemantic.Record, 0, request.Limit)
	for start := 0; start < len(plans) && len(flattened) < int(request.Limit); start += maxParallelObjectSearches {
		end := start + maxParallelObjectSearches
		if end > len(plans) {
			end = len(plans)
		}
		batchCtx, cancelBatch := context.WithCancel(ctx)
		batch := make([]objectSearchResult, end-start)
		var workers sync.WaitGroup
		workers.Add(len(batch))
		for index := start; index < end; index++ {
			resultIndex := index - start
			plan := plans[index]
			go func() {
				defer workers.Done()
				limit := int(plan.Limit)
				remaining := int(request.Limit) - len(flattened)
				if limit > remaining {
					limit = remaining
				}
				records, err := s.engine.SearchWithSpec(batchCtx, queryspec.SearchSpec{
					ObjectAPIName: plan.ObjectID,
					Term:          request.Term,
					SearchFields:  append([]string(nil), plan.SearchFields...),
					ReturnFields:  append([]string(nil), plan.SelectedFields...),
					Limit:         limit,
				}, user)
				if err != nil {
					batch[resultIndex].err = classifyPlatformEngineError(
						batchCtx, err, commonsemantic.InvalidComponentQueryPlan,
					)
					cancelBatch()
					return
				}
				projected := make([]searchsemantic.Record, 0, len(records))
				for _, record := range records {
					queryRecord, projectErr := semanticQueryRecord(
						models.SObject(record), view, plan.ObjectID, 1,
					)
					if projectErr != nil {
						batch[resultIndex].err = projectErr
						cancelBatch()
						return
					}
					searchRecord, projectErr := platformSearchRecord(
						queryRecord, plan.SelectedFields,
					)
					if projectErr != nil {
						batch[resultIndex].err = projectErr
						cancelBatch()
						return
					}
					projected = append(projected, searchRecord)
				}
				batch[resultIndex].records = projected
			}()
		}
		workers.Wait()
		cancelBatch()
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		for _, result := range batch {
			if result.err != nil {
				return nil, result.err
			}
			remaining := int(request.Limit) - len(flattened)
			if remaining == 0 {
				break
			}
			if len(result.records) > remaining {
				result.records = result.records[:remaining]
			}
			flattened = append(flattened, result.records...)
		}
	}
	return flattened, nil
}

func (s *RecordSearchServer) ReadSearchScopeOrder(
	ctx context.Context,
	wireRequest *runtimev1.ReadSearchScopeOrderRequest,
) (*runtimev1.ReadSearchScopeOrderResponse, error) {
	request, err := mapping.FromWireSearchScopeOrderRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}
	scope, scopeOK := runtimeExecutionScopeFromContext(ctx)
	view, viewOK := RuntimeMetadataViewFromContext(ctx)
	trusted, trustedOK := trustedCallFromContext(ctx)
	if !scopeOK || !viewOK || !trustedOK {
		return nil, mapping.ToPlatformStatusError(
			&recordsemantic.Failure{Category: recordsemantic.FailureUnavailable},
			searchsemantic.CapabilityName,
		)
	}
	if err := requireCatalogRevision(
		request.MinCatalogRevision, scope.receipts.CatalogRevision,
	); err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}

	candidates := make([]string, 0, len(request.ObjectIDs))
	for _, objectID := range request.ObjectIDs {
		if _, included := view.Object(objectID); included {
			candidates = append(candidates, objectID)
		}
	}
	viewer := trusted.UserSession()
	ordered, err := s.scopeOrder.ScopeOrder(
		ctx, &viewer, candidates, trusted.PermissionFloor(),
	)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(
			classifyEffectivePermissionError(ctx, err),
			searchsemantic.CapabilityName,
		)
	}
	if err := ctx.Err(); err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}
	response := searchsemantic.ScopeOrderResponse{
		ObjectIDs: ordered,
		Receipts:  receiptsForCapability(scope.receipts, searchsemantic.CapabilityName),
	}
	wireResponse, err := mapping.ToWireSearchScopeOrderResponse(request, response)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, searchsemantic.CapabilityName)
	}
	return wireResponse, nil
}

func platformSearchRecord(
	record querysemantic.Record,
	selectedFields []string,
) (searchsemantic.Record, error) {
	result := searchsemantic.Record{ObjectID: record.ObjectID}
	selected := make(map[string]struct{}, len(selectedFields))
	for _, field := range selectedFields {
		selected[strings.ToLower(field)] = struct{}{}
	}
	fields := append([]querysemantic.Field(nil), record.Fields...)
	sort.Slice(fields, func(i, j int) bool { return fields[i].Name < fields[j].Name })
	for _, field := range fields {
		converted, err := platformSearchField(field)
		if err != nil {
			return searchsemantic.Record{}, err
		}
		if field.Name == "Id" && !converted.Null {
			switch {
			case converted.Reference != nil:
				result.RecordID = *converted.Reference
			case converted.String != nil:
				result.RecordID = *converted.String
			}
		}
		if _, included := selected[strings.ToLower(field.Name)]; !included {
			continue
		}
		result.Fields = append(result.Fields, converted)
	}
	if result.RecordID == "" {
		return searchsemantic.Record{}, &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	return result, nil
}

func platformSearchField(field querysemantic.Field) (recordsemantic.Field, error) {
	value := field.Value
	result := recordsemantic.Field{
		Name: field.Name, Null: value.Null, String: value.String, Boolean: value.Boolean,
		Integer: value.Integer, Date: value.Date, DateTime: value.DateTime, Reference: value.Reference,
	}
	switch value.Kind {
	case querysemantic.ValueString:
		result.Kind = recordsemantic.ValueKindString
	case querysemantic.ValueBoolean:
		result.Kind = recordsemantic.ValueKindBoolean
	case querysemantic.ValueInteger:
		result.Kind = recordsemantic.ValueKindInteger
	case querysemantic.ValueDecimal:
		result.Kind = recordsemantic.ValueKindDecimal
		if value.Decimal != nil {
			result.Decimal = &recordsemantic.Decimal{
				Coefficient: value.Decimal.Coefficient,
				Scale:       value.Decimal.Scale,
			}
		}
	case querysemantic.ValueDate:
		result.Kind = recordsemantic.ValueKindDate
	case querysemantic.ValueDateTime:
		result.Kind = recordsemantic.ValueKindDateTime
	case querysemantic.ValueReference:
		result.Kind = recordsemantic.ValueKindReference
	default:
		return recordsemantic.Field{}, &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	return result, nil
}

var _ runtimev1.RecordSearchServiceServer = (*RecordSearchServer)(nil)