package connector_delivery

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"strings"
	"time"

	connector_oauth "github.com/pingcap-inc/nexuscrm-connectors/features/connector/oauth"
)

const (
	maxPayloadBytes          = 12*1024*1024 - 3
	maxCompletionSnapshot    = 64 * 1024
	defaultCompletionTimeout = 5 * time.Second
)

var errDispatchLeaseBudget = errors.New("connected delivery dispatch lease budget is exhausted")

type repository interface {
	Reserve(context.Context, Request, connector_oauth.ConnectedPrincipal, payloadRecord, time.Time) (Job, error)
	Lookup(context.Context, string, string) (Job, error)
	BeginDispatch(context.Context, string, string, string, time.Time, time.Time) (Job, *dispatchClaim, error)
	ReleaseDispatch(context.Context, dispatchClaim, time.Time) (Job, error)
	FinishDispatch(context.Context, dispatchClaim, terminalDispatch, time.Time) (Job, error)
	Complete(context.Context, string, string, time.Time) (Job, error)
}

type payloadRecord struct {
	Base64      string
	SHA256      string
	EncodedSize int64
}

type terminalDispatch struct {
	State                string
	RemoteReceiptID      string
	RemoteConversationID string
	ResponseStatus       int
	FailureCode          string
}

type Service struct {
	repository        repository
	oauth             OAuthResolver
	providers         ProviderResolver
	now               func() time.Time
	random            func() (string, error)
	lease             time.Duration
	completionTimeout time.Duration
}

func NewSQLService(db *sql.DB, oauth OAuthResolver, providers ProviderResolver) (*Service, error) {
	if db == nil {
		return nil, ErrDependencyMissing
	}
	return newService(newSQLRepository(db), oauth, providers)
}

func newService(repository repository, oauth OAuthResolver, providers ProviderResolver) (*Service, error) {
	if repository == nil || oauth == nil || providers == nil {
		return nil, ErrDependencyMissing
	}
	return &Service{
		repository:        repository,
		oauth:             oauth,
		providers:         providers,
		now:               time.Now,
		random:            randomToken,
		lease:             DefaultDispatchLease,
		completionTimeout: defaultCompletionTimeout,
	}, nil
}

// Lookup returns the frozen result for a request owned by the user. Missing
// and foreign-user requests are intentionally indistinguishable.
func (s *Service) Lookup(ctx context.Context, requestID, userID string) (Result, error) {
	if s == nil || s.repository == nil {
		return Result{}, ErrDependencyMissing
	}
	requestID, userID = strings.TrimSpace(requestID), strings.TrimSpace(userID)
	if requestID == "" || userID == "" {
		return Result{}, ErrInvalidInput
	}
	job, err := s.repository.Lookup(ctx, requestID, userID)
	if err != nil {
		return Result{}, err
	}
	return resultFromJob(job), nil
}

// Dispatch reserves one immutable request and performs at most one provider
// delivery call for that request ID. A dispatching lease is never reclaimed;
// expiry converges to uncertain because the remote effect cannot be proven.
func (s *Service) Dispatch(ctx context.Context, request Request) (Result, error) {
	if s == nil || s.repository == nil || s.oauth == nil || s.providers == nil {
		return Result{}, ErrDependencyMissing
	}
	request, payload, err := normalizeRequest(request)
	if err != nil {
		return Result{}, err
	}
	principal, err := s.oauth.Principal(ctx, request.UserID, request.ConnectorKey)
	if err != nil {
		return Result{}, err
	}
	job, err := s.repository.Reserve(ctx, request, *principal, payload, s.now().UTC())
	if err != nil {
		return Result{}, err
	}
	if result, terminalErr, done := existingResult(job); done {
		return result, terminalErr
	}

	claimToken, err := s.random()
	if err != nil || strings.TrimSpace(claimToken) == "" {
		return Result{}, ErrDependencyMissing
	}
	now := s.now().UTC()
	job, claim, err := s.repository.BeginDispatch(ctx, request.RequestID, request.UserID, claimToken, now, now.Add(s.lease))
	if err != nil {
		return resultFromJob(job), err
	}
	if claim == nil {
		result, terminalErr, _ := existingResult(job)
		return result, terminalErr
	}

	provider, ok := s.providers.Provider(request.ConnectorKey)
	if !ok || provider == nil {
		return s.finishFailed(ctx, *claim, "provider_missing")
	}
	accessToken, err := s.resolveAccessToken(ctx, request, *claim)
	if err != nil {
		if failureCode, terminal := terminalOAuthFailureCode(err); terminal {
			return s.finishFailed(ctx, *claim, failureCode)
		}
		return s.releaseForCredentialRetry(ctx, *claim)
	}
	defer clearString(&accessToken)
	if strings.TrimSpace(accessToken) == "" {
		return s.finishFailed(ctx, *claim, "credential_unavailable")
	}
	providerCtx, cancelProvider, err := s.providerEffectContext(ctx, provider, claim.Job.LeaseExpiresAt)
	if err != nil {
		failureCode := "provider_contract_invalid"
		if errors.Is(err, errDispatchLeaseBudget) {
			failureCode = "dispatch_lease_budget_exhausted"
		}
		return s.finishFailed(ctx, *claim, failureCode)
	}
	defer cancelProvider()
	if err := providerCtx.Err(); err != nil {
		return s.finishFailed(ctx, *claim, "request_canceled_before_dispatch")
	}

	dispatchPayload := append([]byte(nil), request.Payload...)
	defer clearBytes(dispatchPayload)
	dispatchResult, dispatchErr := provider.Dispatch(providerCtx, DispatchRequest{
		AccessToken: accessToken,
		Payload:     dispatchPayload,
	})
	terminal := classifyDispatch(dispatchResult, dispatchErr)
	completionCtx, cancelCompletion := s.localCompletionContext(ctx)
	defer cancelCompletion()
	job, err = s.repository.FinishDispatch(completionCtx, *claim, terminal, s.now().UTC())
	if err != nil {
		return resultFromJob(claim.Job), errors.Join(err, ErrDeliveryUncertain)
	}
	result := resultFromJob(job)
	if terminal.State == StateUncertain {
		return result, ErrDeliveryUncertain
	}
	if terminal.State == StateFailed {
		return result, ErrDeliveryFailed
	}
	return result, nil
}

// Complete records that the consumer idempotently materialized its frozen
// local completion snapshot. It never makes a provider call.
func (s *Service) Complete(ctx context.Context, requestID, userID string) (Result, error) {
	if s == nil || s.repository == nil {
		return Result{}, ErrDependencyMissing
	}
	requestID, userID = strings.TrimSpace(requestID), strings.TrimSpace(userID)
	if requestID == "" || userID == "" {
		return Result{}, ErrInvalidInput
	}
	job, err := s.repository.Complete(ctx, requestID, userID, s.now().UTC())
	if err != nil {
		return Result{}, err
	}
	return resultFromJob(job), nil
}

func normalizeRequest(request Request) (Request, payloadRecord, error) {
	request.RequestID = strings.TrimSpace(request.RequestID)
	request.UserID = strings.TrimSpace(request.UserID)
	request.ConnectorKey = strings.TrimSpace(request.ConnectorKey)
	request.SourceObject = strings.TrimSpace(request.SourceObject)
	request.SourceRecordID = strings.TrimSpace(request.SourceRecordID)
	if request.RequestID == "" || request.UserID == "" || request.ConnectorKey == "" ||
		request.SourceObject == "" || request.SourceRecordID == "" || len(request.Payload) == 0 ||
		len(request.Payload) > maxPayloadBytes || len(request.CompletionSnapshot) == 0 ||
		len(request.CompletionSnapshot) > maxCompletionSnapshot || !validJSONWithUniqueObjectKeys(request.CompletionSnapshot) ||
		len(request.RequestID) > 255 || len(request.UserID) > 255 || len(request.ConnectorKey) > 128 ||
		len(request.SourceObject) > 255 || len(request.SourceRecordID) > 255 || request.ExpectedGeneration <= 0 {
		return Request{}, payloadRecord{}, ErrInvalidInput
	}
	var compact bytes.Buffer
	if err := json.Compact(&compact, request.CompletionSnapshot); err != nil {
		return Request{}, payloadRecord{}, ErrInvalidInput
	}
	request.CompletionSnapshot = json.RawMessage(compact.String())
	request.Payload = append([]byte(nil), request.Payload...)
	digest := sha256.Sum256(request.Payload)
	encoded := base64.StdEncoding.EncodeToString(request.Payload)
	return request, payloadRecord{
		Base64: encoded, SHA256: hex.EncodeToString(digest[:]), EncodedSize: int64(len(encoded)),
	}, nil
}

func classifyDispatch(result DispatchResult, err error) terminalDispatch {
	terminal := terminalDispatch{
		ResponseStatus: safeResponseStatus(result.ResponseStatus),
	}
	receipt, receiptOK := safeRemoteID(result.RemoteReceiptID)
	conversation, conversationOK := safeRemoteID(result.RemoteConversationID)
	if err != nil {
		var providerErr *ProviderError
		if errors.As(err, &providerErr) && providerErr != nil &&
			(providerErr.Kind == ProviderErrorRejected || providerErr.Kind == ProviderErrorAuthorization) {
			terminal.State = StateFailed
			terminal.FailureCode = safeCode(providerErr.Code, "provider_rejected")
			return terminal
		}
		terminal.State = StateUncertain
		terminal.FailureCode = "provider_effect_uncertain"
		if receiptOK {
			terminal.RemoteReceiptID = receipt
		}
		if conversationOK {
			terminal.RemoteConversationID = conversation
		}
		return terminal
	}
	if result.ResponseStatus == 429 || result.ResponseStatus >= 500 {
		terminal.State = StateUncertain
		terminal.FailureCode = "provider_effect_uncertain"
		if receiptOK {
			terminal.RemoteReceiptID = receipt
		}
		if conversationOK {
			terminal.RemoteConversationID = conversation
		}
		return terminal
	}
	if result.Outcome == DispatchAccepted && result.ResponseStatus >= 200 &&
		result.ResponseStatus < 300 && receiptOK && receipt != "" && conversationOK {
		terminal.State = StateAccepted
		terminal.RemoteReceiptID = receipt
		terminal.RemoteConversationID = conversation
		return terminal
	}
	if result.Outcome == DispatchRejected {
		terminal.State = StateFailed
		terminal.FailureCode = "provider_rejected"
		return terminal
	}
	terminal.State = StateUncertain
	terminal.FailureCode = "provider_effect_uncertain"
	if receiptOK {
		terminal.RemoteReceiptID = receipt
	}
	if conversationOK {
		terminal.RemoteConversationID = conversation
	}
	return terminal
}

func (s *Service) resolveAccessToken(ctx context.Context, request Request, claim dispatchClaim) (string, error) {
	resolveRequest := connector_oauth.ResolveAccessTokenRequest{
		UserID: request.UserID, ConnectorKey: request.ConnectorKey,
		ExpectedGeneration: claim.Job.PrincipalGeneration,
	}
	accessToken, err := s.oauth.ResolveAccessToken(ctx, resolveRequest)
	if errors.Is(err, connector_oauth.ErrEffectResultStaged) {
		return s.oauth.ResolveAccessToken(ctx, resolveRequest)
	}
	return accessToken, err
}

func (s *Service) finishFailed(ctx context.Context, claim dispatchClaim, failureCode string) (Result, error) {
	completionCtx, cancel := s.localCompletionContext(ctx)
	defer cancel()
	job, err := s.repository.FinishDispatch(completionCtx, claim, terminalDispatch{
		State: StateFailed, FailureCode: failureCode,
	}, s.now().UTC())
	if err != nil {
		return resultFromJob(claim.Job), err
	}
	return resultFromJob(job), ErrDeliveryFailed
}

func (s *Service) releaseForCredentialRetry(ctx context.Context, claim dispatchClaim) (Result, error) {
	completionCtx, cancel := s.localCompletionContext(ctx)
	defer cancel()
	job, err := s.repository.ReleaseDispatch(completionCtx, claim, s.now().UTC())
	if err != nil {
		return resultFromJob(claim.Job), ErrCredentialUnavailable
	}
	return resultFromJob(job), ErrCredentialUnavailable
}

func (s *Service) localCompletionContext(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), s.completionTimeout)
}

func (s *Service) providerEffectContext(ctx context.Context, provider Provider, leaseExpiresAt *time.Time) (context.Context, context.CancelFunc, error) {
	if leaseExpiresAt == nil || s.lease <= 0 || s.completionTimeout <= 0 {
		return nil, nil, ErrStateInvalid
	}
	remaining := s.lease - s.completionTimeout
	if remaining <= s.completionTimeout {
		return nil, nil, ErrStateInvalid
	}
	providerBudget := remaining - s.completionTimeout
	terminationBound := provider.EffectTerminationBound()
	if terminationBound <= 0 || terminationBound >= providerBudget {
		return nil, nil, ErrStateInvalid
	}
	deadline := leaseExpiresAt.Add(-terminationBound).Add(-s.completionTimeout).Add(-s.completionTimeout)
	if !deadline.After(s.now().UTC()) {
		return nil, nil, errDispatchLeaseBudget
	}
	providerCtx, cancel := context.WithDeadline(ctx, deadline)
	return providerCtx, cancel, nil
}

func safeRemoteID(value string) (string, bool) {
	value = strings.TrimSpace(value)
	return value, len(value) <= 255
}

func safeResponseStatus(value int) int {
	if value < 100 || value > 599 {
		return 0
	}
	return value
}

func existingResult(job Job) (Result, error, bool) {
	result := resultFromJob(job)
	switch job.State {
	case StateAccepted, StateCompleted:
		return result, nil, true
	case StateFailed:
		return result, ErrDeliveryFailed, true
	case StateUncertain:
		return result, ErrDeliveryUncertain, true
	case StateDispatching:
		return result, nil, false
	case StateReserved:
		return result, nil, false
	default:
		return result, ErrRequestConflict, true
	}
}

func resultFromJob(job Job) Result {
	status := 0
	if job.ResponseStatus != nil {
		status = *job.ResponseStatus
	}
	return Result{
		RequestID: job.RequestID, State: job.State,
		CompletionSnapshot: append(json.RawMessage(nil), job.CompletionSnapshot...),
		RemoteReceiptID:    job.RemoteReceiptID, RemoteConversationID: job.RemoteConversationID,
		ResponseStatus: status, FailureCode: job.FailureCode,
		AcceptedAt: cloneTime(job.AcceptedAt),
	}
}

func cloneTime(value *time.Time) *time.Time {
	if value == nil {
		return nil
	}
	copy := *value
	return &copy
}

func terminalOAuthFailureCode(err error) (string, bool) {
	if errors.Is(err, connector_oauth.ErrGenerationMismatch) {
		return "connection_generation_changed", true
	}
	if errors.Is(err, connector_oauth.ErrReconnectRequired) ||
		errors.Is(err, connector_oauth.ErrPrincipalNotFound) ||
		errors.Is(err, connector_oauth.ErrManualActionRequired) ||
		errors.Is(err, connector_oauth.ErrCiphertextInvalid) {
		return "reconnect_required", true
	}
	return "", false
}

func safeCode(value, fallback string) string {
	value = strings.TrimSpace(strings.ToLower(value))
	if value == "" || len(value) > 64 {
		return fallback
	}
	for _, char := range value {
		if (char < 'a' || char > 'z') && (char < '0' || char > '9') && char != '_' && char != '-' {
			return fallback
		}
	}
	return value
}

func randomToken() (string, error) {
	value := make([]byte, 24)
	if _, err := rand.Read(value); err != nil {
		return "", err
	}
	return base64.RawURLEncoding.EncodeToString(value), nil
}

func clearString(value *string) {
	if value != nil {
		*value = ""
	}
}

func clearBytes(value []byte) {
	for i := range value {
		value[i] = 0
	}
}