package runtimeapi

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

type OutboundIntentEngine interface {
	ReadGeneration(
		context.Context,
		persistence.RuntimeOutboundIntentScope,
	) (uint64, error)
	ListPending(
		context.Context,
		persistence.RuntimeOutboundIntentScope,
		uint64,
		uint32,
	) ([]persistence.RuntimeOutboundIntentRow, error)
	Acknowledge(
		context.Context,
		persistence.RuntimeOutboundIntentScope,
		outboundintentsemantic.AcknowledgeRequest,
	) (persistence.RuntimeOutboundIntentRow, bool, error)
}

type VersionedOutboundIntentEngine interface {
	ListPendingForOperationSet(
		context.Context,
		persistence.RuntimeOutboundIntentScope,
		uint64,
		uint32,
		uint32,
	) ([]persistence.RuntimeOutboundIntentRow, error)
}

type OutboundMailAttachmentReferenceEngine interface {
	OpenMailAttachmentReference(
		context.Context,
		persistence.RuntimeOutboundIntentScope,
		outboundintentsemantic.OpenMailAttachmentRequest,
	) (outboundintentsemantic.MailAttachmentV1, error)
}

type FrozenOutboundMailAttachmentEngine interface {
	OpenFrozenRecordAttachment(context.Context, sharedcontract.RecordAttachmentMetadata) (*sharedcontract.FileDownload, error)
}

type OutboundResultCommitter interface {
	Commit(
		context.Context,
		*catalogsnapshot.RuntimeMetadataView,
		persistence.RuntimeOutboundIntentScope,
		outboundintentsemantic.CommitResultRequest,
	) (bool, error)
}

type OutboundIntentTransactionRunner interface {
	RunInPessimisticTransaction(context.Context, func(context.Context) error) error
}

// OutboundIntentServer exposes only Runtime's provider-neutral transactional
// handoff. Provider dispatch, credentials, retries and reconciliation are not
// reachable through this service and remain Integration-owned.
type OutboundIntentServer struct {
	runtimev1.UnimplementedOutboundIntentServiceServer
	intents        OutboundIntentEngine
	results        OutboundResultCommitter
	transactions   OutboundIntentTransactionRunner
	attachmentRefs OutboundMailAttachmentReferenceEngine
	attachments    FrozenOutboundMailAttachmentEngine
}

func NewOutboundIntentServer(
	intents OutboundIntentEngine,
	results OutboundResultCommitter,
	transactions OutboundIntentTransactionRunner,
	attachmentEngines ...FrozenOutboundMailAttachmentEngine,
) (*OutboundIntentServer, error) {
	if intents == nil || results == nil || transactions == nil || len(attachmentEngines) > 1 {
		return nil, errors.New("runtime outbound intent service: engines and transaction owner are required")
	}
	var attachments FrozenOutboundMailAttachmentEngine
	if len(attachmentEngines) == 1 {
		attachments = attachmentEngines[0]
	}
	attachmentRefs, _ := intents.(OutboundMailAttachmentReferenceEngine)
	return &OutboundIntentServer{
		intents: intents, results: results, transactions: transactions,
		attachmentRefs: attachmentRefs, attachments: attachments,
	}, nil
}

func (s *OutboundIntentServer) ReadOutboundIntentGeneration(
	ctx context.Context,
	wireRequest *runtimev1.ReadOutboundIntentGenerationRequest,
) (*runtimev1.ReadOutboundIntentGenerationResponse, error) {
	request, err := mapping.FromWireReadOutboundIntentGenerationRequest(ctx, wireRequest)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	scope, persistenceScope, err := outboundIntentExecutionScope(ctx)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	generation, err := s.intents.ReadGeneration(ctx, persistenceScope)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, classifyOutboundIntentEngineError(err))
	}
	response := outboundintentsemantic.GenerationResponse{
		Generation: generation,
		Receipts: receiptsForCapability(
			scope.receipts,
			outboundintentsemantic.CapabilityName,
		),
	}
	wireResponse, err := mapping.ToWireReadOutboundIntentGenerationResponse(request, response)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	return wireResponse, nil
}

func (s *OutboundIntentServer) ListPendingOutboundIntents(
	ctx context.Context,
	wireRequest *runtimev1.ListPendingOutboundIntentsRequest,
) (*runtimev1.ListPendingOutboundIntentsResponse, error) {
	request, err := mapping.FromWireListPendingOutboundIntentsRequest(ctx, wireRequest)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	scope, persistenceScope, err := outboundIntentExecutionScope(ctx)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	if err := requireCatalogRevision(request.MinCatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	maxOperationSetVersion := request.MaxOperationSetVersion
	if maxOperationSetVersion == 0 {
		// The published neutral contract defines omission as V1. New operations
		// require an explicit opt-in so an older consumer never receives a shape
		// it did not negotiate.
		maxOperationSetVersion = outboundintentsemantic.OperationSetVersionV1
	}
	var rows []persistence.RuntimeOutboundIntentRow
	if versioned, ok := s.intents.(VersionedOutboundIntentEngine); ok {
		rows, err = versioned.ListPendingForOperationSet(
			ctx, persistenceScope, scope.receipts.CatalogRevision, request.Limit, maxOperationSetVersion,
		)
	} else {
		rows, err = s.intents.ListPending(
			ctx, persistenceScope, scope.receipts.CatalogRevision, request.Limit,
		)
		if err == nil {
			rows = filterOutboundIntentRowsForOperationSet(rows, maxOperationSetVersion)
		}
	}
	if err != nil {
		return nil, outboundIntentStatusError(ctx, classifyOutboundIntentEngineError(err))
	}
	intents := make([]outboundintentsemantic.Intent, 0, len(rows))
	for _, row := range rows {
		intents = append(intents, row.Intent)
	}
	response := outboundintentsemantic.ListResponse{
		Intents:  intents,
		Receipts: receiptsForCapability(scope.receipts, outboundintentsemantic.CapabilityName),
	}
	wireResponse, err := mapping.ToWireListPendingOutboundIntentsResponse(request, response)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	return wireResponse, nil
}

func (s *OutboundIntentServer) AcknowledgeOutboundIntent(
	ctx context.Context,
	wireRequest *runtimev1.AcknowledgeOutboundIntentRequest,
) (*runtimev1.AcknowledgeOutboundIntentResponse, error) {
	request, err := mapping.FromWireAcknowledgeOutboundIntentRequest(ctx, wireRequest)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	scope, persistenceScope, err := outboundIntentExecutionScope(ctx)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	if err := requireExactOutboundCatalogRevision(request.CatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	var duplicate bool
	err = s.transactions.RunInPessimisticTransaction(ctx, func(txCtx context.Context) error {
		_, currentDuplicate, acknowledgeErr := s.intents.Acknowledge(txCtx, persistenceScope, request)
		duplicate = currentDuplicate
		return acknowledgeErr
	})
	if err != nil {
		return nil, outboundIntentStatusError(ctx, classifyOutboundIntentEngineError(err))
	}
	response := outboundintentsemantic.AcknowledgeResponse{
		IntentID: request.IntentID, Outcome: request.Outcome, Duplicate: duplicate,
		Receipts: receiptsForCapability(scope.receipts, outboundintentsemantic.CapabilityName),
	}
	wireResponse, err := mapping.ToWireAcknowledgeOutboundIntentResponse(request, response)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	return wireResponse, nil
}

func (s *OutboundIntentServer) OpenOutboundMailAttachment(
	wireRequest *runtimev1.OpenOutboundMailAttachmentRequest,
	stream runtimev1.OutboundIntentService_OpenOutboundMailAttachmentServer,
) error {
	if stream == nil || s == nil || s.attachments == nil || s.attachmentRefs == nil {
		return outboundIntentStatusError(context.Background(), &outboundintentsemantic.Failure{Category: commonsemantic.FailureUnavailable})
	}
	ctx := stream.Context()
	request, err := mapping.FromWireOpenOutboundMailAttachmentRequest(ctx, wireRequest)
	if err != nil {
		return outboundIntentStatusError(ctx, err)
	}
	scope, persistenceScope, err := outboundIntentExecutionScope(ctx)
	if err != nil {
		return outboundIntentStatusError(ctx, err)
	}
	if err := requireExactOutboundCatalogRevision(request.CatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return outboundIntentStatusError(ctx, err)
	}
	attachment, err := s.attachmentRefs.OpenMailAttachmentReference(ctx, persistenceScope, request)
	if err != nil {
		return outboundIntentStatusError(ctx, classifyOutboundIntentEngineError(err))
	}
	download, err := s.attachments.OpenFrozenRecordAttachment(ctx, sharedcontract.RecordAttachmentMetadata{
		FileID: attachment.FileID, FileName: attachment.FileName,
		ContentType: attachment.ContentType, ContentLength: attachment.ContentLength,
		ContentDigest: attachment.ContentDigest,
	})
	if err != nil {
		return outboundIntentStatusError(ctx, classifyOutboundIntentEngineError(err))
	}
	if download == nil || download.Reader == nil {
		return outboundIntentStatusError(ctx, &outboundintentsemantic.Failure{Category: commonsemantic.FailureInternal})
	}
	defer download.Reader.Close()
	if download.Size < 0 ||
		download.FileName != attachment.FileName || download.ContentType != attachment.ContentType ||
		uint64(download.Size) != attachment.ContentLength {
		return outboundIntentStatusError(ctx, &outboundintentsemantic.Failure{Category: commonsemantic.FailureInternal})
	}
	buffer := make([]byte, 64*1024)
	hasher := sha256.New()
	var received uint64
	first := true
	for {
		count, readErr := download.Reader.Read(buffer)
		if count > 0 || (first && readErr == io.EOF) {
			received += uint64(count)
			if received > attachment.ContentLength {
				return outboundIntentStatusError(ctx, &outboundintentsemantic.Failure{Category: commonsemantic.FailureInternal})
			}
			_, _ = hasher.Write(buffer[:count])
			chunk := outboundintentsemantic.MailAttachmentChunk{Data: append([]byte(nil), buffer[:count]...)}
			if first {
				chunk.FileName, chunk.ContentType = attachment.FileName, attachment.ContentType
				chunk.ContentLength, chunk.ContentDigest = attachment.ContentLength, attachment.ContentDigest
				receipts := receiptsForCapability(scope.receipts, outboundintentsemantic.CapabilityName)
				chunk.Receipts = &receipts
				first = false
			}
			wireChunk, mapErr := mapping.ToWireOpenOutboundMailAttachmentChunk(chunk)
			if mapErr != nil {
				return outboundIntentStatusError(ctx, mapErr)
			}
			if sendErr := stream.Send(wireChunk); sendErr != nil {
				return sendErr
			}
		}
		if readErr == io.EOF {
			break
		}
		if readErr != nil {
			return outboundIntentStatusError(ctx, &outboundintentsemantic.Failure{Category: commonsemantic.FailureInternal})
		}
	}
	if received != attachment.ContentLength || attachment.ContentDigest != "sha256:"+hex.EncodeToString(hasher.Sum(nil)) {
		return outboundIntentStatusError(ctx, &outboundintentsemantic.Failure{Category: commonsemantic.FailureInternal})
	}
	return nil
}

func (s *OutboundIntentServer) CommitOutboundResult(
	ctx context.Context,
	wireRequest *runtimev1.CommitOutboundResultRequest,
) (*runtimev1.CommitOutboundResultResponse, error) {
	request, err := mapping.FromWireCommitOutboundResultRequest(ctx, wireRequest)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	scope, persistenceScope, err := outboundIntentExecutionScope(ctx)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	if err := requireExactOutboundCatalogRevision(request.CatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	var duplicate bool
	err = s.transactions.RunInPessimisticTransaction(ctx, func(txCtx context.Context) error {
		currentDuplicate, commitErr := s.results.Commit(txCtx, scope.view, persistenceScope, request)
		duplicate = currentDuplicate
		return commitErr
	})
	if err != nil {
		return nil, outboundIntentStatusError(ctx, classifyOutboundIntentEngineError(err))
	}
	response := outboundintentsemantic.CommitResultResponse{
		IntentID: request.IntentID, Duplicate: duplicate,
		Receipts: receiptsForCapability(scope.receipts, outboundintentsemantic.CapabilityName),
	}
	wireResponse, err := mapping.ToWireCommitOutboundResultResponse(request, response)
	if err != nil {
		return nil, outboundIntentStatusError(ctx, err)
	}
	return wireResponse, nil
}

func filterOutboundIntentRowsForOperationSet(
	rows []persistence.RuntimeOutboundIntentRow,
	maximum uint32,
) []persistence.RuntimeOutboundIntentRow {
	result := make([]persistence.RuntimeOutboundIntentRow, 0, len(rows))
	for _, row := range rows {
		if row.Intent.Operation == outboundintentsemantic.OperationSendMail {
			if maximum < outboundintentsemantic.OperationSetVersionV2 ||
				row.Intent.PayloadFormatVersion == outboundintentsemantic.MailPayloadFormatV3 &&
					maximum < outboundintentsemantic.OperationSetVersionV3 {
				continue
			}
		}
		result = append(result, row)
	}
	return result
}

func outboundIntentExecutionScope(
	ctx context.Context,
) (RuntimeExecutionScope, persistence.RuntimeOutboundIntentScope, error) {
	scope, ok := runtimeExecutionScopeFromContext(ctx)
	trusted, trustedOK := trustedWorkloadCallFromContext(ctx)
	if !ok || !trustedOK || scope.view == nil {
		return RuntimeExecutionScope{}, persistence.RuntimeOutboundIntentScope{}, &outboundintentsemantic.Failure{
			Category: commonsemantic.FailureUnavailable,
		}
	}
	if trusted.orgID != scope.orgID || trusted.callerWorkloadID != scope.workloadID ||
		trusted.capability != outboundintentsemantic.CapabilityName ||
		scope.capability != outboundintentsemantic.CapabilityName {
		return RuntimeExecutionScope{}, persistence.RuntimeOutboundIntentScope{}, &outboundintentsemantic.Failure{
			Category: commonsemantic.FailureForbidden,
		}
	}
	return scope, persistence.RuntimeOutboundIntentScope{
		OrgID: scope.orgID, AppInstallationID: scope.appInstallationID,
		AuthorizedWorkloadID: scope.workloadID,
	}, nil
}

func requireExactOutboundCatalogRevision(requested, served uint64) error {
	if requested == served {
		return nil
	}
	return &outboundintentsemantic.Failure{
		Category:         commonsemantic.FailureStaleState,
		StaleAxis:        commonsemantic.StaleAxisCatalog,
		RequiredRevision: requested,
		ServedRevision:   served,
	}
}

func classifyOutboundIntentEngineError(err error) error {
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	switch {
	case errors.Is(err, persistence.ErrRuntimeOutboundIntentNotFound):
		return &outboundintentsemantic.Failure{Category: commonsemantic.FailureMissing}
	case errors.Is(err, persistence.ErrRuntimeOutboundIntentConflict):
		return &outboundintentsemantic.Failure{Category: commonsemantic.FailureInvalidRequest}
	default:
		return &outboundintentsemantic.Failure{Category: commonsemantic.FailureInternal}
	}
}

func outboundIntentStatusError(ctx context.Context, err error) error {
	if ctxErr := ctx.Err(); ctxErr != nil {
		return mapping.ToPlatformStatusError(ctxErr, outboundintentsemantic.CapabilityName)
	}
	return mapping.ToPlatformStatusError(err, outboundintentsemantic.CapabilityName)
}

var _ runtimev1.OutboundIntentServiceServer = (*OutboundIntentServer)(nil)