package runtimeapi

import (
	"context"
	"errors"
	"io"
	"time"

	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

type BulkIngestEngine interface {
	CreateIngestJob(context.Context, sharedcontract.BulkIngestCreateRequest, *models.UserSession) (*models.BulkJob, error)
	GetIngestJob(context.Context, string, *models.UserSession) (*models.BulkJob, error)
	CloseIngestJob(context.Context, string, models.BulkJobState, *models.UserSession) (*models.BulkJob, error)
	UploadIngestData(context.Context, sharedcontract.BulkIngestUploadRequest, *models.UserSession) error
}

type BulkIngestServer struct {
	runtimev1.UnimplementedBulkIngestServiceServer
	engine BulkIngestEngine
}

func NewBulkIngestServer(engine BulkIngestEngine) (*BulkIngestServer, error) {
	if engine == nil {
		return nil, errors.New("runtime bulk ingest: engine is required")
	}
	return &BulkIngestServer{engine: engine}, nil
}

func (s *BulkIngestServer) CreateBulkIngestJob(ctx context.Context, wireRequest *runtimev1.CreateBulkIngestJobRequest) (*runtimev1.CreateBulkIngestJobResponse, error) {
	request, err := mapping.FromWireCreateBulkIngestJobRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	scope, trusted, err := bulkIngestExecution(ctx)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	user := trusted.UserSession()
	job, err := s.engine.CreateIngestJob(ctx, sharedcontract.BulkIngestCreateRequest{
		Object: request.ObjectID, Operation: fromSemanticBulkOperation(request.Operation),
		ExternalIDFieldName: request.ExternalIDFieldName, APIVersion: request.ClientVersion,
	}, &user)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(classifyBulkIngestEngineError(ctx, err, commonsemantic.InvalidComponentOperation), bulkingestsemantic.CapabilityName)
	}
	response := bulkingestsemantic.CreateResponse{Job: toSemanticBulkJob(job), Receipts: receiptsForCapability(scope.receipts, bulkingestsemantic.CapabilityName)}
	wireResponse, err := mapping.ToWireCreateBulkIngestJobResponse(request, response)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	return wireResponse, nil
}

func (s *BulkIngestServer) ReadBulkIngestJob(ctx context.Context, wireRequest *runtimev1.ReadBulkIngestJobRequest) (*runtimev1.ReadBulkIngestJobResponse, error) {
	request, err := mapping.FromWireReadBulkIngestJobRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	scope, trusted, err := bulkIngestExecution(ctx)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	user := trusted.UserSession()
	job, err := s.engine.GetIngestJob(ctx, request.JobID, &user)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(classifyBulkIngestEngineError(ctx, err, commonsemantic.InvalidComponentRecordID), bulkingestsemantic.CapabilityName)
	}
	response := bulkingestsemantic.ReadResponse{Job: toSemanticBulkJob(job), Receipts: receiptsForCapability(scope.receipts, bulkingestsemantic.CapabilityName)}
	wireResponse, err := mapping.ToWireReadBulkIngestJobResponse(request, response)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	return wireResponse, nil
}

func (s *BulkIngestServer) CloseBulkIngestJob(ctx context.Context, wireRequest *runtimev1.CloseBulkIngestJobRequest) (*runtimev1.CloseBulkIngestJobResponse, error) {
	request, err := mapping.FromWireCloseBulkIngestJobRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	scope, trusted, err := bulkIngestExecution(ctx)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	user := trusted.UserSession()
	job, err := s.engine.CloseIngestJob(ctx, request.JobID, fromSemanticBulkState(request.State), &user)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(classifyBulkIngestEngineError(ctx, err, commonsemantic.InvalidComponentOperation), bulkingestsemantic.CapabilityName)
	}
	response := bulkingestsemantic.CloseResponse{Job: toSemanticBulkJob(job), Receipts: receiptsForCapability(scope.receipts, bulkingestsemantic.CapabilityName)}
	wireResponse, err := mapping.ToWireCloseBulkIngestJobResponse(request, response)
	if err != nil {
		return nil, mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	return wireResponse, nil
}

func (s *BulkIngestServer) UploadBulkIngestData(stream runtimev1.BulkIngestService_UploadBulkIngestDataServer) error {
	if stream == nil {
		return mapping.ToPlatformStatusError(&commonsemantic.Failure{Category: commonsemantic.FailureInternal}, bulkingestsemantic.CapabilityName)
	}
	ctx := stream.Context()
	scope, trusted, err := bulkIngestExecution(ctx)
	if err != nil {
		return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	firstWire, err := stream.Recv()
	if err != nil {
		return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	first, err := mapping.FromWireUploadBulkIngestChunk(firstWire)
	if err != nil {
		return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	if first.Header == nil {
		return mapping.ToPlatformStatusError(&commonsemantic.Failure{Category: commonsemantic.FailureInvalidRequest, InvalidComponent: commonsemantic.InvalidComponentCommandData}, bulkingestsemantic.CapabilityName)
	}
	state, err := bulkingestsemantic.NewUploadStreamState(ctx)
	if err != nil {
		return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	if err := state.Observe(first); err != nil {
		return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}

	reader, writer := io.Pipe()
	stopCancellation := context.AfterFunc(ctx, func() {
		_ = writer.CloseWithError(ctx.Err())
		_ = reader.CloseWithError(ctx.Err())
	})
	defer stopCancellation()
	type uploadOutcome struct{ err error }
	outcome := make(chan uploadOutcome, 1)
	user := trusted.UserSession()
	contentLength := int64(-1)
	if first.Header.ContentLengthKnown {
		contentLength = int64(first.Header.ContentLength)
	}
	go func() {
		defer reader.Close()
		uploadErr := s.engine.UploadIngestData(ctx, sharedcontract.BulkIngestUploadRequest{
			JobID: first.Header.JobID, Body: reader, ContentLength: contentLength,
		}, &user)
		outcome <- uploadOutcome{err: uploadErr}
	}()
	abort := func(cause error, preferEngineFailure bool) error {
		_ = writer.CloseWithError(cause)
		completed := <-outcome
		var bulkErr *sharedcontract.BulkIngestError
		if preferEngineFailure && errors.As(completed.err, &bulkErr) {
			cause = completed.err
		}
		var protocolFailure *commonsemantic.Failure
		if errors.As(cause, &protocolFailure) {
			return mapping.ToPlatformStatusError(protocolFailure, bulkingestsemantic.CapabilityName)
		}
		return mapping.ToPlatformStatusError(classifyBulkIngestEngineError(ctx, cause, commonsemantic.InvalidComponentCommandData), bulkingestsemantic.CapabilityName)
	}
	earlyEngineCompletion := func(completed uploadOutcome) error {
		cause := completed.err
		if cause == nil {
			cause = errors.New("bulk ingest engine completed before the client upload stream")
		}
		_ = writer.CloseWithError(cause)
		return mapping.ToPlatformStatusError(classifyBulkIngestEngineError(ctx, cause, commonsemantic.InvalidComponentCommandData), bulkingestsemantic.CapabilityName)
	}

	received := clientStreamReceivePump(ctx, stream.Recv)
	for {
		var wireChunk *runtimev1.UploadBulkIngestDataRequest
		var recvErr error
		select {
		case completed := <-outcome:
			return earlyEngineCompletion(completed)
		case next := <-received:
			wireChunk, recvErr = next.message, next.err
		}
		if recvErr == io.EOF {
			break
		}
		if recvErr != nil {
			return abort(recvErr, false)
		}
		chunk, mapErr := mapping.FromWireUploadBulkIngestChunk(wireChunk)
		if mapErr != nil {
			return abort(mapErr, false)
		}
		if observeErr := state.Observe(chunk); observeErr != nil {
			return abort(observeErr, false)
		}
		if _, writeErr := writer.Write(chunk.Data); writeErr != nil {
			return abort(writeErr, true)
		}
	}
	if err := state.Complete(); err != nil {
		return abort(err, false)
	}
	_ = writer.Close()
	completed := <-outcome
	if completed.err != nil {
		return mapping.ToPlatformStatusError(classifyBulkIngestEngineError(ctx, completed.err, commonsemantic.InvalidComponentCommandData), bulkingestsemantic.CapabilityName)
	}
	wireResponse, err := mapping.ToWireUploadBulkIngestResponse(*first.Header, bulkingestsemantic.UploadResponse{
		JobID: first.Header.JobID, Receipts: receiptsForCapability(scope.receipts, bulkingestsemantic.CapabilityName),
	})
	if err != nil {
		return mapping.ToPlatformStatusError(err, bulkingestsemantic.CapabilityName)
	}
	return stream.SendAndClose(wireResponse)
}

func bulkIngestExecution(ctx context.Context) (RuntimeExecutionScope, TrustedCall, error) {
	scope, scopeOK := runtimeExecutionScopeFromContext(ctx)
	trusted, trustedOK := trustedCallFromContext(ctx)
	if !scopeOK || !trustedOK {
		return RuntimeExecutionScope{}, TrustedCall{}, &commonsemantic.Failure{Category: commonsemantic.FailureUnavailable}
	}
	if _, ok := RuntimeMetadataViewFromContext(ctx); !ok {
		return RuntimeExecutionScope{}, TrustedCall{}, &commonsemantic.Failure{Category: commonsemantic.FailureUnavailable}
	}
	return scope, trusted, nil
}

func classifyBulkIngestEngineError(ctx context.Context, err error, invalidComponent commonsemantic.InvalidComponent) error {
	if err == nil {
		return &commonsemantic.Failure{Category: commonsemantic.FailureInternal}
	}
	if contextFailure, owned := classifyEngineContextOwnership(ctx, err); owned {
		return contextFailure
	}
	var bulkErr *sharedcontract.BulkIngestError
	if !errors.As(err, &bulkErr) {
		return &commonsemantic.Failure{Category: commonsemantic.FailureInternal}
	}
	switch bulkErr.Kind {
	case sharedcontract.BulkIngestErrorInvalidRequest, sharedcontract.BulkIngestErrorInvalidOperation:
		return &commonsemantic.Failure{Category: commonsemantic.FailureInvalidRequest, InvalidComponent: invalidComponent}
	case sharedcontract.BulkIngestErrorInvalidJobState:
		return &commonsemantic.Failure{Category: commonsemantic.FailureInvalidRequest, InvalidComponent: commonsemantic.InvalidComponentOperation}
	case sharedcontract.BulkIngestErrorInvalidObject:
		return &commonsemantic.Failure{Category: commonsemantic.FailureInvalidRequest, InvalidComponent: commonsemantic.InvalidComponentObjectID}
	case sharedcontract.BulkIngestErrorInsufficientPermission, sharedcontract.BulkIngestErrorAccessDenied:
		return &commonsemantic.Failure{Category: commonsemantic.FailureForbidden}
	case sharedcontract.BulkIngestErrorNotFound:
		return &commonsemantic.Failure{Category: commonsemantic.FailureMissing}
	case sharedcontract.BulkIngestErrorRequestTooLarge:
		return &commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitUploadBytes, Maximum: bulkingestsemantic.MaxUploadBytes}
	default:
		return &commonsemantic.Failure{Category: commonsemantic.FailureInternal}
	}
}

func toSemanticBulkJob(job *models.BulkJob) bulkingestsemantic.Job {
	if job == nil {
		return bulkingestsemantic.Job{}
	}
	return bulkingestsemantic.Job{
		ID: job.ID, ObjectID: job.Object, Operation: toSemanticBulkOperation(job.Operation), State: toSemanticBulkState(job.State),
		CreatedByID: job.CreatedById, CreatedAt: bulkIngestTime(job.CreatedDate), ModifiedAt: bulkIngestTime(job.SystemModstamp),
		FailureCode: bulkIngestFailureCode(job), ExternalIDFieldName: job.ExternalIdFieldName,
		NumberRecordsProcessed: nonnegativeBulkCount(job.NumberRecordsProcessed), NumberRecordsFailed: nonnegativeBulkCount(job.NumberRecordsFailed),
		TotalProcessingTime: uint64(max(job.TotalProcessingTime, 0)), ActiveProcessingTime: uint64(max(job.ApiActiveProcessingTime, 0)),
		ClientVersion: job.ApiVersion,
	}
}

func bulkIngestFailureCode(job *models.BulkJob) bulkingestsemantic.FailureCode {
	if job != nil && job.State == models.JobStateFailed {
		return bulkingestsemantic.FailureProcessingFailed
	}
	return ""
}

func bulkIngestTime(value time.Time) string { return value.UTC().Format(time.RFC3339Nano) }
func nonnegativeBulkCount(value int) uint64 { return uint64(max(value, 0)) }

func toSemanticBulkOperation(operation models.BulkOperation) bulkingestsemantic.Operation {
	switch operation {
	case models.OperationInsert:
		return bulkingestsemantic.OperationInsert
	case models.OperationUpdate:
		return bulkingestsemantic.OperationUpdate
	case models.OperationUpsert:
		return bulkingestsemantic.OperationUpsert
	case models.OperationDelete:
		return bulkingestsemantic.OperationDelete
	default:
		return ""
	}
}

func fromSemanticBulkOperation(operation bulkingestsemantic.Operation) models.BulkOperation {
	switch operation {
	case bulkingestsemantic.OperationInsert:
		return models.OperationInsert
	case bulkingestsemantic.OperationUpdate:
		return models.OperationUpdate
	case bulkingestsemantic.OperationUpsert:
		return models.OperationUpsert
	case bulkingestsemantic.OperationDelete:
		return models.OperationDelete
	default:
		return ""
	}
}

func toSemanticBulkState(state models.BulkJobState) bulkingestsemantic.State {
	switch state {
	case models.JobStateOpen:
		return bulkingestsemantic.StateOpen
	case models.JobStateUploadComplete:
		return bulkingestsemantic.StateUploadComplete
	case models.JobStateInProgress:
		return bulkingestsemantic.StateInProgress
	case models.JobStateJobComplete:
		return bulkingestsemantic.StateComplete
	case models.JobStateFailed:
		return bulkingestsemantic.StateFailed
	case models.JobStateAborted:
		return bulkingestsemantic.StateAborted
	default:
		return ""
	}
}

func fromSemanticBulkState(state bulkingestsemantic.State) models.BulkJobState {
	switch state {
	case bulkingestsemantic.StateUploadComplete:
		return models.JobStateUploadComplete
	case bulkingestsemantic.StateAborted:
		return models.JobStateAborted
	default:
		return ""
	}
}

var _ runtimev1.BulkIngestServiceServer = (*BulkIngestServer)(nil)