sendico/api/ledger/internal/service/ledger/service.go

package ledger

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/shopspring/decimal"
	feesv1 "github.com/tech/sendico/pkg/proto/billing/fees/v1"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/timestamppb"

	accountingv1 "github.com/tech/sendico/pkg/proto/common/accounting/v1"
	moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
	tracev1 "github.com/tech/sendico/pkg/proto/common/trace/v1"

	"github.com/tech/sendico/ledger/storage"
	"github.com/tech/sendico/pkg/api/routers"
	"github.com/tech/sendico/pkg/merrors"
	pmessaging "github.com/tech/sendico/pkg/messaging"
	"github.com/tech/sendico/pkg/mlogger"
	ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
)

type serviceError string

func (e serviceError) Error() string {
	return string(e)
}

var (
	errStorageNotInitialized = serviceError("ledger: storage not initialized")
)

type Service struct {
	logger   mlogger.Logger
	storage  storage.Repository
	producer pmessaging.Producer
	fees     feesDependency

	outbox struct {
		once      sync.Once
		cancel    context.CancelFunc
		publisher *outboxPublisher
	}
	ledgerv1.UnimplementedLedgerServiceServer
}

type feesDependency struct {
	client  feesv1.FeeEngineClient
	timeout time.Duration
}

func (f feesDependency) available() bool {
	return f.client != nil
}

func NewService(logger mlogger.Logger, repo storage.Repository, prod pmessaging.Producer, feesClient feesv1.FeeEngineClient, feesTimeout time.Duration) *Service {
	// Initialize Prometheus metrics
	initMetrics()

	service := &Service{
		logger:   logger.Named("ledger"),
		storage:  repo,
		producer: prod,
		fees: feesDependency{
			client:  feesClient,
			timeout: feesTimeout,
		},
	}

	service.startOutboxPublisher()
	return service
}

func (s *Service) Register(router routers.GRPC) error {
	return router.Register(func(reg grpc.ServiceRegistrar) {
		ledgerv1.RegisterLedgerServiceServer(reg, s)
	})
}

// ListAccounts lists ledger accounts for an organization.
func (s *Service) ListAccounts(ctx context.Context, req *ledgerv1.ListAccountsRequest) (*ledgerv1.ListAccountsResponse, error) {
	responder := s.listAccountsResponder(ctx, req)
	return responder(ctx)
}

// CreateAccount provisions a new ledger account scoped to an organization.
func (s *Service) CreateAccount(ctx context.Context, req *ledgerv1.CreateAccountRequest) (*ledgerv1.CreateAccountResponse, error) {
	responder := s.createAccountResponder(ctx, req)
	return responder(ctx)
}

// PostCreditWithCharges handles credit posting with fees in one atomic journal entry
func (s *Service) PostCreditWithCharges(ctx context.Context, req *ledgerv1.PostCreditRequest) (*ledgerv1.PostResponse, error) {
	start := time.Now()
	defer func() {
		recordJournalEntry("credit", "attempted", time.Since(start).Seconds())
	}()

	responder := s.postCreditResponder(ctx, req)
	resp, err := responder(ctx)

	if err != nil {
		recordJournalEntryError("credit", "not_implemented")
	}

	return resp, err
}

// PostDebitWithCharges handles debit posting with fees in one atomic journal entry
func (s *Service) PostDebitWithCharges(ctx context.Context, req *ledgerv1.PostDebitRequest) (*ledgerv1.PostResponse, error) {
	start := time.Now()
	defer func() {
		recordJournalEntry("debit", "attempted", time.Since(start).Seconds())
	}()

	responder := s.postDebitResponder(ctx, req)
	resp, err := responder(ctx)

	if err != nil {
		recordJournalEntryError("debit", "failed")
	}

	return resp, err
}

// TransferInternal handles internal transfer between accounts
func (s *Service) TransferInternal(ctx context.Context, req *ledgerv1.TransferRequest) (*ledgerv1.PostResponse, error) {
	start := time.Now()
	defer func() {
		recordJournalEntry("transfer", "attempted", time.Since(start).Seconds())
	}()

	responder := s.transferResponder(ctx, req)
	resp, err := responder(ctx)

	if err != nil {
		recordJournalEntryError("transfer", "failed")
	}

	return resp, err
}

// ApplyFXWithCharges handles foreign exchange transaction with charges
func (s *Service) ApplyFXWithCharges(ctx context.Context, req *ledgerv1.FXRequest) (*ledgerv1.PostResponse, error) {
	start := time.Now()
	defer func() {
		recordJournalEntry("fx", "attempted", time.Since(start).Seconds())
	}()

	responder := s.fxResponder(ctx, req)
	resp, err := responder(ctx)

	if err != nil {
		recordJournalEntryError("fx", "failed")
	}

	return resp, err
}

// GetBalance queries current account balance
func (s *Service) GetBalance(ctx context.Context, req *ledgerv1.GetBalanceRequest) (*ledgerv1.BalanceResponse, error) {
	start := time.Now()
	defer func() {
		recordBalanceQuery("attempted", time.Since(start).Seconds())
	}()

	responder := s.getBalanceResponder(ctx, req)
	resp, err := responder(ctx)

	return resp, err
}

// GetJournalEntry gets journal entry details
func (s *Service) GetJournalEntry(ctx context.Context, req *ledgerv1.GetEntryRequest) (*ledgerv1.JournalEntryResponse, error) {
	responder := s.getJournalEntryResponder(ctx, req)
	return responder(ctx)
}

func (s *Service) Shutdown() {
	if s == nil {
		return
	}
	if s.outbox.cancel != nil {
		s.outbox.cancel()
	}
}

func (s *Service) startOutboxPublisher() {
	if s.storage == nil || s.producer == nil {
		return
	}

	s.outbox.once.Do(func() {
		outboxStore := s.storage.Outbox()
		if outboxStore == nil {
			return
		}

		ctx, cancel := context.WithCancel(context.Background())
		s.outbox.cancel = cancel
		s.outbox.publisher = newOutboxPublisher(s.logger, outboxStore, s.producer)

		go s.outbox.publisher.run(ctx)
	})
}

// GetStatement gets account statement with pagination
func (s *Service) GetStatement(ctx context.Context, req *ledgerv1.GetStatementRequest) (*ledgerv1.StatementResponse, error) {
	responder := s.getStatementResponder(ctx, req)
	return responder(ctx)
}

func (s *Service) pingStorage(ctx context.Context) error {
	if s.storage == nil {
		return errStorageNotInitialized
	}
	return s.storage.Ping(ctx)
}

func (s *Service) quoteFeesForCredit(ctx context.Context, req *ledgerv1.PostCreditRequest) ([]*ledgerv1.PostingLine, error) {
	if !s.fees.available() {
		return nil, nil
	}
	attrs := map[string]string{}
	if strings.TrimSpace(req.GetDescription()) != "" {
		attrs["description"] = req.GetDescription()
	}
	return s.quoteFees(ctx, feesv1.Trigger_TRIGGER_CAPTURE, req.GetOrganizationRef(), req.GetIdempotencyKey(), req.GetLedgerAccountRef(), "ledger.post_credit", req.GetIdempotencyKey(), req.GetEventTime(), req.Money, attrs)
}

func (s *Service) quoteFeesForDebit(ctx context.Context, req *ledgerv1.PostDebitRequest) ([]*ledgerv1.PostingLine, error) {
	if !s.fees.available() {
		return nil, nil
	}
	attrs := map[string]string{}
	if strings.TrimSpace(req.GetDescription()) != "" {
		attrs["description"] = req.GetDescription()
	}
	return s.quoteFees(ctx, feesv1.Trigger_TRIGGER_REFUND, req.GetOrganizationRef(), req.GetIdempotencyKey(), req.GetLedgerAccountRef(), "ledger.post_debit", req.GetIdempotencyKey(), req.GetEventTime(), req.Money, attrs)
}

func (s *Service) quoteFees(ctx context.Context, trigger feesv1.Trigger, organizationRef, idempotencyKey, ledgerAccountRef, originType, originRef string, eventTime *timestamppb.Timestamp, baseAmount *moneyv1.Money, attributes map[string]string) ([]*ledgerv1.PostingLine, error) {
	if !s.fees.available() {
		return nil, nil
	}
	if strings.TrimSpace(organizationRef) == "" {
		return nil, merrors.InvalidArgument("organization reference is required to quote fees")
	}
	if baseAmount == nil {
		return nil, merrors.InvalidArgument("base amount is required to quote fees")
	}

	amountCopy := &moneyv1.Money{Amount: baseAmount.GetAmount(), Currency: baseAmount.GetCurrency()}
	bookedAt := eventTime
	if bookedAt == nil {
		bookedAt = timestamppb.Now()
	}

	trace := &tracev1.TraceContext{
		RequestRef:     idempotencyKey,
		IdempotencyKey: idempotencyKey,
	}

	req := &feesv1.QuoteFeesRequest{
		Meta: &feesv1.RequestMeta{
			OrganizationRef: organizationRef,
			Trace:           trace,
		},
		Intent: &feesv1.Intent{
			Trigger:    trigger,
			BaseAmount: amountCopy,
			BookedAt:   bookedAt,
			OriginType: originType,
			OriginRef:  originRef,
			Attributes: map[string]string{},
		},
	}

	if ledgerAccountRef != "" {
		req.Intent.Attributes["ledger_account_ref"] = ledgerAccountRef
	}
	for k, v := range attributes {
		if strings.TrimSpace(k) == "" {
			continue
		}
		req.Intent.Attributes[k] = v
	}

	callCtx := ctx
	if s.fees.timeout > 0 {
		var cancel context.CancelFunc
		callCtx, cancel = context.WithTimeout(ctx, s.fees.timeout)
		defer cancel()
	}

	resp, err := s.fees.client.QuoteFees(callCtx, req)
	if err != nil {
		return nil, err
	}

	lines, err := convertFeeDerivedLines(resp.GetLines())
	if err != nil {
		return nil, err
	}
	return lines, nil
}

func convertFeeDerivedLines(lines []*feesv1.DerivedPostingLine) ([]*ledgerv1.PostingLine, error) {
	result := make([]*ledgerv1.PostingLine, 0, len(lines))
	for idx, line := range lines {
		if line == nil {
			continue
		}
		if line.GetMoney() == nil {
			return nil, merrors.Internal(fmt.Sprintf("fee line %d missing money", idx))
		}
		dec, err := decimal.NewFromString(line.GetMoney().GetAmount())
		if err != nil {
			return nil, merrors.InternalWrap(err, fmt.Sprintf("fee line %d invalid amount", idx))
		}
		dec = ensureAmountForSide(dec, line.GetSide())
		posting := &ledgerv1.PostingLine{
			LedgerAccountRef: line.GetLedgerAccountRef(),
			Money: &moneyv1.Money{
				Amount:   dec.String(),
				Currency: line.GetMoney().GetCurrency(),
			},
			LineType: mapFeeLineType(line.GetLineType()),
		}
		result = append(result, posting)
	}
	return result, nil
}

func ensureAmountForSide(amount decimal.Decimal, side accountingv1.EntrySide) decimal.Decimal {
	switch side {
	case accountingv1.EntrySide_ENTRY_SIDE_DEBIT:
		if amount.Sign() > 0 {
			return amount.Neg()
		}
	case accountingv1.EntrySide_ENTRY_SIDE_CREDIT:
		if amount.Sign() < 0 {
			return amount.Neg()
		}
	}
	return amount
}

func mapFeeLineType(lineType accountingv1.PostingLineType) ledgerv1.LineType {
	switch lineType {
	case accountingv1.PostingLineType_POSTING_LINE_FEE:
		return ledgerv1.LineType_LINE_FEE
	case accountingv1.PostingLineType_POSTING_LINE_SPREAD:
		return ledgerv1.LineType_LINE_SPREAD
	case accountingv1.PostingLineType_POSTING_LINE_REVERSAL:
		return ledgerv1.LineType_LINE_REVERSAL
	default:
		return ledgerv1.LineType_LINE_FEE
	}
}