409 lines
14 KiB
Go
409 lines
14 KiB
Go
package grpcadapter
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
applicationledger "gl/application/ledger"
|
|
domain "gl/domain/ledger"
|
|
ledgerv1 "gl/gen/ledger/v1"
|
|
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
func (h *Handler) AppendJournal(ctx context.Context, request *ledgerv1.AppendJournalRequest) (*ledgerv1.AppendJournalResponse, error) {
|
|
command, err := appendJournalCommand(request)
|
|
if err != nil {
|
|
return nil, status.Error(codes.InvalidArgument, err.Error())
|
|
}
|
|
result, err := h.ledger.AppendJournal(ctx, command)
|
|
if err != nil {
|
|
return nil, rpcError(err)
|
|
}
|
|
return &ledgerv1.AppendJournalResponse{
|
|
Journal: journalMessage(result.Journal),
|
|
AlreadyExisted: result.AlreadyExisted,
|
|
}, nil
|
|
}
|
|
|
|
func (h *Handler) AppendTransactionEvent(ctx context.Context, request *ledgerv1.AppendTransactionEventRequest) (*ledgerv1.AppendTransactionEventResponse, error) {
|
|
occurredAt, err := requiredTime(request.GetOccurredAt(), "occurred_at")
|
|
if err != nil {
|
|
return nil, status.Error(codes.InvalidArgument, err.Error())
|
|
}
|
|
event, alreadyExisted, err := h.ledger.AppendEvent(ctx, applicationledger.AppendEventCommand{
|
|
SourceService: request.GetSourceService(),
|
|
IdempotencyKey: request.GetIdempotencyKey(),
|
|
SourceTransactionID: request.GetSourceTransactionId(),
|
|
TrackingCode: request.GetTrackingCode(),
|
|
EventVersion: request.GetEventVersion(),
|
|
State: transactionState(request.GetState()),
|
|
ErrorCode: request.GetErrorCode(),
|
|
ErrorMessage: request.GetErrorMessage(),
|
|
OccurredAt: occurredAt,
|
|
CorrelationID: request.GetCorrelationId(),
|
|
ActorID: request.GetActorId(),
|
|
Blockchain: blockchainReference(request.GetBlockchain()),
|
|
Metadata: request.GetMetadata(),
|
|
})
|
|
if err != nil {
|
|
return nil, rpcError(err)
|
|
}
|
|
return &ledgerv1.AppendTransactionEventResponse{
|
|
Event: eventMessage(event),
|
|
AlreadyExisted: alreadyExisted,
|
|
}, nil
|
|
}
|
|
|
|
func (h *Handler) GetJournal(ctx context.Context, request *ledgerv1.GetJournalRequest) (*ledgerv1.Journal, error) {
|
|
var journalID, idempotencyKey string
|
|
switch lookup := request.GetLookup().(type) {
|
|
case *ledgerv1.GetJournalRequest_JournalId:
|
|
journalID = lookup.JournalId
|
|
case *ledgerv1.GetJournalRequest_IdempotencyKey:
|
|
idempotencyKey = lookup.IdempotencyKey
|
|
}
|
|
journal, err := h.ledger.GetJournal(ctx, journalID, idempotencyKey)
|
|
if err != nil {
|
|
return nil, rpcError(err)
|
|
}
|
|
return journalMessage(journal), nil
|
|
}
|
|
|
|
func (h *Handler) ListEntries(ctx context.Context, request *ledgerv1.ListEntriesRequest) (*ledgerv1.ListEntriesResponse, error) {
|
|
command := applicationledger.ListCommand{
|
|
PageSize: request.GetPageSize(),
|
|
PageToken: request.GetPageToken(),
|
|
}
|
|
if request.Account != nil {
|
|
account := accountReference(request.Account)
|
|
command.Account = &account
|
|
}
|
|
if request.AssetId != nil {
|
|
assetID := request.GetAssetId()
|
|
command.AssetID = &assetID
|
|
}
|
|
var err error
|
|
if request.RecordedFrom != nil {
|
|
value, parseErr := requiredTime(request.RecordedFrom, "recorded_from")
|
|
if parseErr != nil {
|
|
return nil, status.Error(codes.InvalidArgument, parseErr.Error())
|
|
}
|
|
command.RecordedFrom = &value
|
|
}
|
|
if request.RecordedTo != nil {
|
|
value, parseErr := requiredTime(request.RecordedTo, "recorded_to")
|
|
if parseErr != nil {
|
|
return nil, status.Error(codes.InvalidArgument, parseErr.Error())
|
|
}
|
|
command.RecordedTo = &value
|
|
}
|
|
journals, nextPageToken, err := h.ledger.List(ctx, command)
|
|
if err != nil {
|
|
return nil, rpcError(err)
|
|
}
|
|
response := &ledgerv1.ListEntriesResponse{
|
|
Journals: make([]*ledgerv1.Journal, 0, len(journals)),
|
|
NextPageToken: nextPageToken,
|
|
}
|
|
for _, journal := range journals {
|
|
response.Journals = append(response.Journals, journalMessage(journal))
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func (h *Handler) GetBalance(ctx context.Context, request *ledgerv1.GetBalanceRequest) (*ledgerv1.GetBalanceResponse, error) {
|
|
if request.Account == nil {
|
|
return nil, status.Error(codes.InvalidArgument, "account is required")
|
|
}
|
|
asOf := time.Now().UTC()
|
|
if request.AsOf != nil {
|
|
value, err := requiredTime(request.AsOf, "as_of")
|
|
if err != nil {
|
|
return nil, status.Error(codes.InvalidArgument, err.Error())
|
|
}
|
|
asOf = value
|
|
}
|
|
account := accountReference(request.Account)
|
|
balance, err := h.ledger.Balance(ctx, account, asOf)
|
|
if err != nil {
|
|
return nil, rpcError(err)
|
|
}
|
|
return &ledgerv1.GetBalanceResponse{
|
|
Account: accountMessage(account),
|
|
Balance: balance.String(),
|
|
AsOf: timestamppb.New(asOf),
|
|
}, nil
|
|
}
|
|
|
|
func (h *Handler) ReplayJournals(ctx context.Context, request *ledgerv1.ReplayJournalsRequest) (*ledgerv1.ReplayJournalsResponse, error) {
|
|
commands := make([]applicationledger.AppendJournalCommand, 0, len(request.GetJournals()))
|
|
for _, journal := range request.GetJournals() {
|
|
command, err := appendJournalCommand(journal)
|
|
if err != nil {
|
|
return nil, status.Error(codes.InvalidArgument, err.Error())
|
|
}
|
|
commands = append(commands, command)
|
|
}
|
|
results, err := h.ledger.Replay(ctx, commands)
|
|
if err != nil {
|
|
return nil, rpcError(err)
|
|
}
|
|
response := &ledgerv1.ReplayJournalsResponse{Results: make([]*ledgerv1.ReplayJournalResult, 0, len(results))}
|
|
for _, result := range results {
|
|
response.Results = append(response.Results, &ledgerv1.ReplayJournalResult{
|
|
IdempotencyKey: result.Journal.IdempotencyKey,
|
|
Journal: journalMessage(result.Journal),
|
|
AlreadyExisted: result.AlreadyExisted,
|
|
})
|
|
}
|
|
return response, nil
|
|
}
|
|
|
|
func appendJournalCommand(request *ledgerv1.AppendJournalRequest) (applicationledger.AppendJournalCommand, error) {
|
|
occurredAt, err := requiredTime(request.GetOccurredAt(), "occurred_at")
|
|
if err != nil {
|
|
return applicationledger.AppendJournalCommand{}, err
|
|
}
|
|
command := applicationledger.AppendJournalCommand{
|
|
SourceService: request.GetSourceService(),
|
|
IdempotencyKey: request.GetIdempotencyKey(),
|
|
SourceTransactionID: request.GetSourceTransactionId(),
|
|
TrackingCode: request.GetTrackingCode(),
|
|
EffectKind: request.GetEffectKind(),
|
|
EventVersion: request.GetEventVersion(),
|
|
ReversalOfJournalID: request.GetReversalOfJournalId(),
|
|
OccurredAt: occurredAt,
|
|
CorrelationID: request.GetCorrelationId(),
|
|
ActorID: request.GetActorId(),
|
|
Blockchain: blockchainReference(request.GetBlockchain()),
|
|
Metadata: request.GetMetadata(),
|
|
Entries: make([]applicationledger.EntryCommand, 0, len(request.GetEntries())),
|
|
}
|
|
for _, entry := range request.GetEntries() {
|
|
if entry.Account == nil {
|
|
return applicationledger.AppendJournalCommand{}, errors.New("entry account is required")
|
|
}
|
|
command.Entries = append(command.Entries, applicationledger.EntryCommand{
|
|
LineNumber: entry.GetLineNumber(),
|
|
Account: accountReference(entry.Account),
|
|
Amount: entry.GetAmount(),
|
|
Description: entry.GetDescription(),
|
|
})
|
|
}
|
|
return command, nil
|
|
}
|
|
|
|
func requiredTime(value *timestamppb.Timestamp, field string) (time.Time, error) {
|
|
if value == nil {
|
|
return time.Time{}, errors.New(field + " is required")
|
|
}
|
|
if err := value.CheckValid(); err != nil {
|
|
return time.Time{}, errors.New(field + " is invalid: " + err.Error())
|
|
}
|
|
return value.AsTime().UTC(), nil
|
|
}
|
|
|
|
func accountReference(value *ledgerv1.AccountReference) domain.AccountReference {
|
|
return domain.AccountReference{
|
|
Class: accountClass(value.GetAccountClass()),
|
|
OwnerType: value.GetOwnerType(),
|
|
OwnerID: value.GetOwnerId(),
|
|
AssetID: value.GetAssetId(),
|
|
}
|
|
}
|
|
|
|
func accountClass(value ledgerv1.AccountClass) domain.AccountClass {
|
|
switch value {
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE:
|
|
return domain.AccountClassUserAvailable
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_USER_FROZEN:
|
|
return domain.AccountClassUserFrozen
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_EXTERNAL_BLOCKCHAIN:
|
|
return domain.AccountClassExternalBlockchain
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_TREASURY:
|
|
return domain.AccountClassTreasury
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_MARKET_CLEARING:
|
|
return domain.AccountClassMarketClearing
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_IPG_CLEARING:
|
|
return domain.AccountClassIPGClearing
|
|
case ledgerv1.AccountClass_ACCOUNT_CLASS_COMMISSION_REVENUE:
|
|
return domain.AccountClassCommissionRevenue
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func accountClassMessage(value domain.AccountClass) ledgerv1.AccountClass {
|
|
switch value {
|
|
case domain.AccountClassUserAvailable:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE
|
|
case domain.AccountClassUserFrozen:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_USER_FROZEN
|
|
case domain.AccountClassExternalBlockchain:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_EXTERNAL_BLOCKCHAIN
|
|
case domain.AccountClassTreasury:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_TREASURY
|
|
case domain.AccountClassMarketClearing:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_MARKET_CLEARING
|
|
case domain.AccountClassIPGClearing:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_IPG_CLEARING
|
|
case domain.AccountClassCommissionRevenue:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_COMMISSION_REVENUE
|
|
default:
|
|
return ledgerv1.AccountClass_ACCOUNT_CLASS_UNSPECIFIED
|
|
}
|
|
}
|
|
|
|
func transactionState(value ledgerv1.TransactionState) domain.TransactionState {
|
|
switch value {
|
|
case ledgerv1.TransactionState_TRANSACTION_STATE_CREATED:
|
|
return domain.TransactionStateCreated
|
|
case ledgerv1.TransactionState_TRANSACTION_STATE_PENDING_TRANSACTION:
|
|
return domain.TransactionStatePendingTransaction
|
|
case ledgerv1.TransactionState_TRANSACTION_STATE_PENDING_ADMIN:
|
|
return domain.TransactionStatePendingAdmin
|
|
case ledgerv1.TransactionState_TRANSACTION_STATE_SUCCESSFUL:
|
|
return domain.TransactionStateSuccessful
|
|
case ledgerv1.TransactionState_TRANSACTION_STATE_FAILED:
|
|
return domain.TransactionStateFailed
|
|
case ledgerv1.TransactionState_TRANSACTION_STATE_SUSPENDED:
|
|
return domain.TransactionStateSuspended
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func transactionStateMessage(value domain.TransactionState) ledgerv1.TransactionState {
|
|
switch value {
|
|
case domain.TransactionStateCreated:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_CREATED
|
|
case domain.TransactionStatePendingTransaction:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_PENDING_TRANSACTION
|
|
case domain.TransactionStatePendingAdmin:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_PENDING_ADMIN
|
|
case domain.TransactionStateSuccessful:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_SUCCESSFUL
|
|
case domain.TransactionStateFailed:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_FAILED
|
|
case domain.TransactionStateSuspended:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_SUSPENDED
|
|
default:
|
|
return ledgerv1.TransactionState_TRANSACTION_STATE_UNSPECIFIED
|
|
}
|
|
}
|
|
|
|
func blockchainReference(value *ledgerv1.BlockchainReference) domain.BlockchainReference {
|
|
if value == nil {
|
|
return domain.BlockchainReference{}
|
|
}
|
|
return domain.BlockchainReference{
|
|
Network: value.GetNetwork(),
|
|
TransactionHash: value.GetTransactionHash(),
|
|
LedgerSequence: value.GetLedgerSequence(),
|
|
}
|
|
}
|
|
|
|
func journalMessage(value domain.Journal) *ledgerv1.Journal {
|
|
message := &ledgerv1.Journal{
|
|
JournalId: value.ID,
|
|
SourceService: value.SourceService,
|
|
IdempotencyKey: value.IdempotencyKey,
|
|
SourceTransactionId: value.SourceTransactionID,
|
|
TrackingCode: value.TrackingCode,
|
|
EffectKind: value.EffectKind,
|
|
EventVersion: value.EventVersion,
|
|
OccurredAt: timestampMessage(value.OccurredAt),
|
|
RecordedAt: timestampMessage(value.RecordedAt),
|
|
CorrelationId: value.CorrelationID,
|
|
ActorId: value.ActorID,
|
|
Blockchain: blockchainMessage(value.Blockchain),
|
|
Metadata: value.Metadata,
|
|
PayloadHash: value.PayloadHash,
|
|
Entries: make([]*ledgerv1.JournalEntry, 0, len(value.Entries)),
|
|
}
|
|
if value.ReversalOfJournalID != "" {
|
|
message.ReversalOfJournalId = &value.ReversalOfJournalID
|
|
}
|
|
for _, entry := range value.Entries {
|
|
message.Entries = append(message.Entries, &ledgerv1.JournalEntry{
|
|
LineNumber: entry.LineNumber,
|
|
Account: accountMessage(entry.Account),
|
|
Amount: entry.Amount.String(),
|
|
Description: entry.Description,
|
|
})
|
|
}
|
|
return message
|
|
}
|
|
|
|
func eventMessage(value domain.TransactionEvent) *ledgerv1.TransactionEvent {
|
|
return &ledgerv1.TransactionEvent{
|
|
EventId: value.ID,
|
|
Event: &ledgerv1.AppendTransactionEventRequest{
|
|
SourceService: value.SourceService,
|
|
IdempotencyKey: value.IdempotencyKey,
|
|
SourceTransactionId: value.SourceTransactionID,
|
|
TrackingCode: value.TrackingCode,
|
|
EventVersion: value.EventVersion,
|
|
State: transactionStateMessage(value.State),
|
|
ErrorCode: value.ErrorCode,
|
|
ErrorMessage: value.ErrorMessage,
|
|
OccurredAt: timestampMessage(value.OccurredAt),
|
|
CorrelationId: value.CorrelationID,
|
|
ActorId: value.ActorID,
|
|
Blockchain: blockchainMessage(value.Blockchain),
|
|
Metadata: value.Metadata,
|
|
},
|
|
RecordedAt: timestampMessage(value.RecordedAt),
|
|
PayloadHash: value.PayloadHash,
|
|
}
|
|
}
|
|
|
|
func accountMessage(value domain.AccountReference) *ledgerv1.AccountReference {
|
|
return &ledgerv1.AccountReference{
|
|
AccountClass: accountClassMessage(value.Class),
|
|
OwnerType: value.OwnerType,
|
|
OwnerId: value.OwnerID,
|
|
AssetId: value.AssetID,
|
|
}
|
|
}
|
|
|
|
func blockchainMessage(value domain.BlockchainReference) *ledgerv1.BlockchainReference {
|
|
return &ledgerv1.BlockchainReference{
|
|
Network: value.Network,
|
|
TransactionHash: value.TransactionHash,
|
|
LedgerSequence: value.LedgerSequence,
|
|
}
|
|
}
|
|
|
|
func timestampMessage(value time.Time) *timestamppb.Timestamp {
|
|
if value.IsZero() {
|
|
return nil
|
|
}
|
|
return timestamppb.New(value)
|
|
}
|
|
|
|
func rpcError(err error) error {
|
|
switch {
|
|
case errors.Is(err, applicationledger.ErrInvalidArgument):
|
|
return status.Error(codes.InvalidArgument, err.Error())
|
|
case errors.Is(err, domain.ErrNotFound):
|
|
return status.Error(codes.NotFound, err.Error())
|
|
case errors.Is(err, domain.ErrIdempotencyConflict):
|
|
return status.Error(codes.AlreadyExists, err.Error())
|
|
case errors.Is(err, domain.ErrIncompleteJournal):
|
|
return status.Error(codes.FailedPrecondition, err.Error())
|
|
case errors.Is(err, domain.ErrAlreadyReversed):
|
|
return status.Error(codes.FailedPrecondition, err.Error())
|
|
case errors.Is(err, context.Canceled):
|
|
return status.Error(codes.Canceled, err.Error())
|
|
case errors.Is(err, context.DeadlineExceeded):
|
|
return status.Error(codes.DeadlineExceeded, err.Error())
|
|
default:
|
|
return status.Error(codes.Internal, "internal ledger error")
|
|
}
|
|
}
|