diff --git a/application/ledger/service.go b/application/ledger/service.go new file mode 100644 index 0000000..22fd19e --- /dev/null +++ b/application/ledger/service.go @@ -0,0 +1,384 @@ +// Package ledger implements GL's transport-independent use cases. +package ledger + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "time" + + domain "gl/domain/ledger" +) + +var ErrInvalidArgument = errors.New("invalid ledger request") + +type Repository interface { + Append(context.Context, domain.Journal) (domain.AppendResult, error) + AppendEvent(context.Context, domain.TransactionEvent) (domain.TransactionEvent, bool, error) + GetByID(context.Context, string) (domain.Journal, error) + GetByIdempotencyKey(context.Context, string) (domain.Journal, error) + List(context.Context, domain.JournalFilter) ([]domain.Journal, error) + Balance(context.Context, domain.AccountReference, time.Time) (domain.Amount, error) +} + +type IDGenerator func() (string, error) + +type Service struct { + repository Repository + newID IDGenerator +} + +func NewService(repository Repository, newID IDGenerator) *Service { + if newID == nil { + newID = newUUID + } + return &Service{repository: repository, newID: newID} +} + +type EntryCommand struct { + LineNumber uint32 + Account domain.AccountReference + Amount string + Description string +} + +type AppendJournalCommand struct { + SourceService string + IdempotencyKey string + SourceTransactionID string + TrackingCode string + EffectKind string + EventVersion uint32 + Entries []EntryCommand + ReversalOfJournalID string + OccurredAt time.Time + CorrelationID string + ActorID string + Blockchain domain.BlockchainReference + Metadata map[string]string +} + +type AppendJournalResult struct { + Journal domain.Journal + AlreadyExisted bool +} + +func (s *Service) AppendJournal(ctx context.Context, command AppendJournalCommand) (AppendJournalResult, error) { + journalID, err := s.newID() + if err != nil { + return AppendJournalResult{}, fmt.Errorf("generate journal id: %w", err) + } + journal := domain.Journal{ + ID: journalID, + SourceService: command.SourceService, + IdempotencyKey: command.IdempotencyKey, + SourceTransactionID: command.SourceTransactionID, + TrackingCode: command.TrackingCode, + EffectKind: command.EffectKind, + EventVersion: command.EventVersion, + ReversalOfJournalID: command.ReversalOfJournalID, + OccurredAt: command.OccurredAt, + CorrelationID: command.CorrelationID, + ActorID: command.ActorID, + Blockchain: command.Blockchain, + Metadata: cloneMetadata(command.Metadata), + Entries: make([]domain.Entry, 0, len(command.Entries)), + } + for _, input := range command.Entries { + amount, parseErr := domain.ParseAmount(input.Amount) + if parseErr != nil { + return AppendJournalResult{}, invalid("entry %d amount: %v", input.LineNumber, parseErr) + } + journal.Entries = append(journal.Entries, domain.Entry{ + LineNumber: input.LineNumber, + Account: input.Account, + Amount: amount, + Description: input.Description, + }) + } + sort.Slice(journal.Entries, func(i, j int) bool { + return journal.Entries[i].LineNumber < journal.Entries[j].LineNumber + }) + + journal.PayloadHash, err = journalHash(journal) + if err != nil { + return AppendJournalResult{}, fmt.Errorf("hash journal: %w", err) + } + if err := journal.Validate(); err != nil { + return AppendJournalResult{}, invalid("%v", err) + } + if journal.ReversalOfJournalID != "" { + original, getErr := s.repository.GetByID(ctx, journal.ReversalOfJournalID) + if getErr != nil { + return AppendJournalResult{}, fmt.Errorf("read reversal target: %w", getErr) + } + if err := validateFullReversal(original, journal); err != nil { + return AppendJournalResult{}, invalid("%v", err) + } + } + + appendResult, err := s.repository.Append(ctx, journal) + if err != nil { + return AppendJournalResult{}, err + } + stored, err := s.repository.GetByID(ctx, appendResult.JournalID) + if err != nil { + return AppendJournalResult{}, fmt.Errorf("read appended journal: %w", err) + } + return AppendJournalResult{Journal: stored, AlreadyExisted: appendResult.AlreadyExists}, nil +} + +type AppendEventCommand struct { + SourceService string + IdempotencyKey string + SourceTransactionID string + TrackingCode string + EventVersion uint32 + State domain.TransactionState + ErrorCode string + ErrorMessage string + OccurredAt time.Time + CorrelationID string + ActorID string + Blockchain domain.BlockchainReference + Metadata map[string]string +} + +func (s *Service) AppendEvent(ctx context.Context, command AppendEventCommand) (domain.TransactionEvent, bool, error) { + eventID, err := s.newID() + if err != nil { + return domain.TransactionEvent{}, false, fmt.Errorf("generate event id: %w", err) + } + event := domain.TransactionEvent{ + ID: eventID, + SourceService: command.SourceService, + IdempotencyKey: command.IdempotencyKey, + SourceTransactionID: command.SourceTransactionID, + TrackingCode: command.TrackingCode, + EventVersion: command.EventVersion, + State: command.State, + ErrorCode: command.ErrorCode, + ErrorMessage: command.ErrorMessage, + OccurredAt: command.OccurredAt, + CorrelationID: command.CorrelationID, + ActorID: command.ActorID, + Blockchain: command.Blockchain, + Metadata: cloneMetadata(command.Metadata), + } + event.PayloadHash, err = eventHash(event) + if err != nil { + return domain.TransactionEvent{}, false, fmt.Errorf("hash transaction event: %w", err) + } + if err := event.Validate(); err != nil { + return domain.TransactionEvent{}, false, invalid("%v", err) + } + return s.repository.AppendEvent(ctx, event) +} + +func (s *Service) GetJournal(ctx context.Context, journalID, idempotencyKey string) (domain.Journal, error) { + if (journalID == "") == (idempotencyKey == "") { + return domain.Journal{}, invalid("exactly one journal lookup is required") + } + if journalID != "" { + return s.repository.GetByID(ctx, journalID) + } + return s.repository.GetByIdempotencyKey(ctx, idempotencyKey) +} + +type ListCommand struct { + Account *domain.AccountReference + AssetID *int64 + RecordedFrom *time.Time + RecordedTo *time.Time + PageSize uint32 + PageToken string +} + +func (s *Service) List(ctx context.Context, command ListCommand) ([]domain.Journal, string, error) { + if command.Account != nil { + if err := command.Account.Validate(); err != nil { + return nil, "", invalid("account: %v", err) + } + } + if command.AssetID != nil && *command.AssetID <= 0 { + return nil, "", invalid("asset id must be positive") + } + pageSize := int(command.PageSize) + if pageSize == 0 { + pageSize = 50 + } + if pageSize > 200 { + return nil, "", invalid("page size must not exceed 200") + } + offset, err := decodePageToken(command.PageToken) + if err != nil { + return nil, "", invalid("page token: %v", err) + } + journals, err := s.repository.List(ctx, domain.JournalFilter{ + Account: command.Account, + AssetID: command.AssetID, + RecordedFrom: command.RecordedFrom, + RecordedTo: command.RecordedTo, + Limit: pageSize + 1, + Offset: offset, + }) + if err != nil { + return nil, "", err + } + next := "" + if len(journals) > pageSize { + journals = journals[:pageSize] + next = encodePageToken(offset + pageSize) + } + return journals, next, nil +} + +func (s *Service) Balance(ctx context.Context, account domain.AccountReference, asOf time.Time) (domain.Amount, error) { + if err := account.Validate(); err != nil { + return domain.Amount{}, invalid("account: %v", err) + } + return s.repository.Balance(ctx, account, asOf) +} + +func (s *Service) Replay(ctx context.Context, commands []AppendJournalCommand) ([]AppendJournalResult, error) { + if len(commands) == 0 || len(commands) > 200 { + return nil, invalid("replay batch must contain between 1 and 200 journals") + } + results := make([]AppendJournalResult, 0, len(commands)) + for _, command := range commands { + result, err := s.AppendJournal(ctx, command) + if err != nil { + return nil, err + } + results = append(results, result) + } + return results, nil +} + +func validateFullReversal(original, reversal domain.Journal) error { + if len(original.Entries) != len(reversal.Entries) { + return fmt.Errorf("reversal must contain every original entry") + } + expected := make(map[string]int, len(original.Entries)) + for _, entry := range original.Entries { + expected[entryKey(entry.Account, entry.Amount.Negate())]++ + } + for _, entry := range reversal.Entries { + key := entryKey(entry.Account, entry.Amount) + if expected[key] == 0 { + return fmt.Errorf("reversal entries must exactly negate the original journal") + } + expected[key]-- + } + return nil +} + +func entryKey(account domain.AccountReference, amount domain.Amount) string { + return fmt.Sprintf("%s\x00%s\x00%s\x00%d\x00%s", account.Class, account.OwnerType, account.OwnerID, account.AssetID, amount.String()) +} + +func journalHash(journal domain.Journal) (string, error) { + entries := make([]map[string]any, 0, len(journal.Entries)) + for _, entry := range journal.Entries { + entries = append(entries, map[string]any{ + "line_number": entry.LineNumber, + "class": entry.Account.Class, + "owner_type": entry.Account.OwnerType, + "owner_id": entry.Account.OwnerID, + "asset_id": entry.Account.AssetID, + "amount": entry.Amount.String(), + "description": entry.Description, + }) + } + return hashPayload(map[string]any{ + "source_service": journal.SourceService, + "idempotency_key": journal.IdempotencyKey, + "source_transaction_id": journal.SourceTransactionID, + "tracking_code": journal.TrackingCode, + "effect_kind": journal.EffectKind, + "event_version": journal.EventVersion, + "entries": entries, + "reversal_of": journal.ReversalOfJournalID, + "occurred_at": journal.OccurredAt.UTC().Format(time.RFC3339Nano), + "correlation_id": journal.CorrelationID, + "actor_id": journal.ActorID, + "blockchain": journal.Blockchain, + "metadata": journal.Metadata, + }) +} + +func eventHash(event domain.TransactionEvent) (string, error) { + return hashPayload(map[string]any{ + "source_service": event.SourceService, + "idempotency_key": event.IdempotencyKey, + "source_transaction_id": event.SourceTransactionID, + "tracking_code": event.TrackingCode, + "event_version": event.EventVersion, + "state": event.State, + "error_code": event.ErrorCode, + "error_message": event.ErrorMessage, + "occurred_at": event.OccurredAt.UTC().Format(time.RFC3339Nano), + "correlation_id": event.CorrelationID, + "actor_id": event.ActorID, + "blockchain": event.Blockchain, + "metadata": event.Metadata, + }) +} + +func hashPayload(payload any) (string, error) { + encoded, err := json.Marshal(payload) + if err != nil { + return "", err + } + hash := sha256.Sum256(encoded) + return hex.EncodeToString(hash[:]), nil +} + +func newUUID() (string, error) { + value := make([]byte, 16) + if _, err := rand.Read(value); err != nil { + return "", err + } + value[6] = (value[6] & 0x0f) | 0x40 + value[8] = (value[8] & 0x3f) | 0x80 + encoded := hex.EncodeToString(value) + return encoded[0:8] + "-" + encoded[8:12] + "-" + encoded[12:16] + "-" + encoded[16:20] + "-" + encoded[20:32], nil +} + +func cloneMetadata(metadata map[string]string) map[string]string { + clone := make(map[string]string, len(metadata)) + for key, value := range metadata { + clone[key] = value + } + return clone +} + +func invalid(format string, args ...any) error { + return fmt.Errorf("%w: %s", ErrInvalidArgument, fmt.Sprintf(format, args...)) +} + +func encodePageToken(offset int) string { + return base64.RawURLEncoding.EncodeToString([]byte(strconv.Itoa(offset))) +} + +func decodePageToken(token string) (int, error) { + if token == "" { + return 0, nil + } + decoded, err := base64.RawURLEncoding.DecodeString(token) + if err != nil { + return 0, err + } + offset, err := strconv.Atoi(string(decoded)) + if err != nil || offset < 0 { + return 0, fmt.Errorf("invalid offset") + } + return offset, nil +} diff --git a/application/ledger/service_test.go b/application/ledger/service_test.go new file mode 100644 index 0000000..91a6af4 --- /dev/null +++ b/application/ledger/service_test.go @@ -0,0 +1,223 @@ +package ledger + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + domain "gl/domain/ledger" +) + +type repositoryStub struct { + journals map[string]domain.Journal + byKey map[string]string + events map[string]domain.TransactionEvent + appendCalls int + list []domain.Journal + balance domain.Amount + lastAttempt domain.Journal +} + +func newRepositoryStub() *repositoryStub { + return &repositoryStub{ + journals: make(map[string]domain.Journal), + byKey: make(map[string]string), + events: make(map[string]domain.TransactionEvent), + } +} + +func (r *repositoryStub) Append(_ context.Context, journal domain.Journal) (domain.AppendResult, error) { + r.appendCalls++ + r.lastAttempt = journal + if existingID, ok := r.byKey[journal.IdempotencyKey]; ok { + return domain.AppendResult{JournalID: existingID, AlreadyExists: true}, nil + } + journal.RecordedAt = time.Unix(2, 0).UTC() + r.journals[journal.ID] = journal + r.byKey[journal.IdempotencyKey] = journal.ID + return domain.AppendResult{JournalID: journal.ID}, nil +} + +func (r *repositoryStub) AppendEvent(_ context.Context, event domain.TransactionEvent) (domain.TransactionEvent, bool, error) { + if existing, ok := r.events[event.IdempotencyKey]; ok { + return existing, true, nil + } + event.RecordedAt = time.Unix(2, 0).UTC() + r.events[event.IdempotencyKey] = event + return event, false, nil +} + +func (r *repositoryStub) GetByID(_ context.Context, id string) (domain.Journal, error) { + journal, ok := r.journals[id] + if !ok { + return domain.Journal{}, errors.New("not found") + } + return journal, nil +} + +func (r *repositoryStub) GetByIdempotencyKey(_ context.Context, key string) (domain.Journal, error) { + id, ok := r.byKey[key] + if !ok { + return domain.Journal{}, errors.New("not found") + } + return r.journals[id], nil +} + +func (r *repositoryStub) List(context.Context, domain.JournalFilter) ([]domain.Journal, error) { + return append([]domain.Journal(nil), r.list...), nil +} + +func (r *repositoryStub) Balance(context.Context, domain.AccountReference, time.Time) (domain.Amount, error) { + return r.balance, nil +} + +func TestAppendJournalIsBalancedCanonicalAndIdempotent(t *testing.T) { + repository := newRepositoryStub() + service := NewService(repository, fixedID("11111111-1111-4111-8111-111111111111")) + command := validAppendCommand() + + first, err := service.AppendJournal(context.Background(), command) + if err != nil { + t.Fatal(err) + } + command.Entries[0], command.Entries[1] = command.Entries[1], command.Entries[0] + second, err := service.AppendJournal(context.Background(), command) + if err != nil { + t.Fatal(err) + } + if first.AlreadyExisted || !second.AlreadyExisted { + t.Fatalf("unexpected idempotency results: first=%+v second=%+v", first, second) + } + if first.Journal.PayloadHash == "" || first.Journal.PayloadHash != second.Journal.PayloadHash { + t.Fatalf("unexpected payload hashes: %q %q", first.Journal.PayloadHash, second.Journal.PayloadHash) + } + if repository.lastAttempt.PayloadHash != first.Journal.PayloadHash { + t.Fatal("entry order changed the canonical payload hash") + } + if got := first.Journal.Entries[0].Amount.String(); got != "-10.25" { + t.Fatalf("amount was not canonical: %s", got) + } +} + +func TestAppendJournalRejectsInvalidAmountBeforeRepository(t *testing.T) { + repository := newRepositoryStub() + service := NewService(repository, fixedID("11111111-1111-4111-8111-111111111111")) + command := validAppendCommand() + command.Entries[0].Amount = "NaN" + + _, err := service.AppendJournal(context.Background(), command) + if !errors.Is(err, ErrInvalidArgument) { + t.Fatalf("expected invalid argument, got %v", err) + } + if repository.appendCalls != 0 { + t.Fatal("invalid journal reached repository") + } +} + +func TestAppendJournalRequiresExactFullReversal(t *testing.T) { + repository := newRepositoryStub() + service := NewService(repository, fixedID("22222222-2222-4222-8222-222222222222")) + originalCommand := validAppendCommand() + original, err := NewService(repository, fixedID("11111111-1111-4111-8111-111111111111")).AppendJournal(context.Background(), originalCommand) + if err != nil { + t.Fatal(err) + } + + reversal := validAppendCommand() + reversal.IdempotencyKey = "wallet:1:internal-transfer-reversal:v1" + reversal.EffectKind = "internal-transfer-reversal" + reversal.ReversalOfJournalID = original.Journal.ID + reversal.Entries[0].Amount = "10.25" + reversal.Entries[1].Amount = "-10.25" + if _, err := service.AppendJournal(context.Background(), reversal); err != nil { + t.Fatalf("valid reversal failed: %v", err) + } + + reversal.IdempotencyKey = "wallet:1:bad-reversal:v1" + reversal.Entries[0].Account.OwnerID = "different-user" + if _, err := service.AppendJournal(context.Background(), reversal); !errors.Is(err, ErrInvalidArgument) { + t.Fatalf("expected invalid reversal, got %v", err) + } +} + +func TestAppendEventIsIdempotent(t *testing.T) { + repository := newRepositoryStub() + service := NewService(repository, fixedID("11111111-1111-4111-8111-111111111111")) + command := AppendEventCommand{ + SourceService: "wallet", + IdempotencyKey: "wallet:1:status:v1", + SourceTransactionID: "1", + EventVersion: 1, + State: domain.TransactionStateCreated, + OccurredAt: time.Unix(1, 0).UTC(), + } + + first, existed, err := service.AppendEvent(context.Background(), command) + if err != nil || existed { + t.Fatalf("first append: existed=%v err=%v", existed, err) + } + second, existed, err := service.AppendEvent(context.Background(), command) + if err != nil || !existed || first.PayloadHash != second.PayloadHash { + t.Fatalf("second append: existed=%v err=%v first=%+v second=%+v", existed, err, first, second) + } +} + +func TestListUsesOpaquePageToken(t *testing.T) { + repository := newRepositoryStub() + repository.list = make([]domain.Journal, 3) + service := NewService(repository, nil) + + journals, next, err := service.List(context.Background(), ListCommand{PageSize: 2}) + if err != nil { + t.Fatal(err) + } + if len(journals) != 2 || next == "" { + t.Fatalf("unexpected page: len=%d next=%q", len(journals), next) + } + if _, err := decodePageToken(next); err != nil { + t.Fatalf("invalid generated page token: %v", err) + } +} + +func TestNewUUIDProducesRFC4122Shape(t *testing.T) { + id, err := newUUID() + if err != nil { + t.Fatal(err) + } + if len(id) != 36 || id[14] != '4' || !strings.Contains("89ab", string(id[19])) { + t.Fatalf("unexpected UUID: %q", id) + } +} + +func validAppendCommand() AppendJournalCommand { + return AppendJournalCommand{ + SourceService: "wallet", + IdempotencyKey: "wallet:1:internal-transfer:v1", + SourceTransactionID: "1", + EffectKind: "internal-transfer", + EventVersion: 1, + OccurredAt: time.Unix(1, 0).UTC(), + Entries: []EntryCommand{ + { + LineNumber: 1, + Account: domain.AccountReference{ + Class: domain.AccountClassUserAvailable, OwnerType: "user", OwnerID: "1", AssetID: 5, + }, + Amount: "-10.2500", + }, + { + LineNumber: 2, + Account: domain.AccountReference{ + Class: domain.AccountClassUserAvailable, OwnerType: "user", OwnerID: "2", AssetID: 5, + }, + Amount: "10.25", + }, + }, + } +} + +func fixedID(id string) IDGenerator { + return func() (string, error) { return id, nil } +} diff --git a/cmd/gl/main.go b/cmd/gl/main.go index d883fd6..6d4c5af 100644 --- a/cmd/gl/main.go +++ b/cmd/gl/main.go @@ -9,6 +9,7 @@ import ( "syscall" "gl/application/health" + applicationledger "gl/application/ledger" "gl/infrastructure/config" "gl/infrastructure/postgres" grpcadapter "gl/interface/grpc" @@ -38,7 +39,7 @@ func main() { os.Exit(1) } - handler := grpcadapter.NewHealthHandler(health.NewService(database)) + handler := grpcadapter.NewHandler(health.NewService(database), applicationledger.NewService(postgres.NewJournalRepository(database), nil)) slog.Info("starting GL gRPC service", "host", cfg.GRPC.Host, "port", cfg.GRPC.Port) serverConfig := grpcadapter.ServerConfig{ Host: cfg.GRPC.Host, diff --git a/domain/ledger/errors.go b/domain/ledger/errors.go new file mode 100644 index 0000000..6708e0a --- /dev/null +++ b/domain/ledger/errors.go @@ -0,0 +1,10 @@ +package ledger + +import "errors" + +var ( + ErrNotFound = errors.New("ledger record not found") + ErrIdempotencyConflict = errors.New("idempotency key belongs to a different payload") + ErrIncompleteJournal = errors.New("idempotency key belongs to an unsealed journal") + ErrAlreadyReversed = errors.New("journal already has a reversal") +) diff --git a/domain/ledger/event.go b/domain/ledger/event.go new file mode 100644 index 0000000..0f9084c --- /dev/null +++ b/domain/ledger/event.go @@ -0,0 +1,74 @@ +package ledger + +import ( + "encoding/hex" + "fmt" + "strings" + "time" +) + +type TransactionState string + +const ( + TransactionStateCreated TransactionState = "CREATED" + TransactionStatePendingTransaction TransactionState = "PENDING_TRANSACTION" + TransactionStatePendingAdmin TransactionState = "PENDING_ADMIN" + TransactionStateSuccessful TransactionState = "SUCCESSFUL" + TransactionStateFailed TransactionState = "FAILED" + TransactionStateSuspended TransactionState = "SUSPENDED" +) + +func (s TransactionState) Valid() bool { + switch s { + case TransactionStateCreated, + TransactionStatePendingTransaction, + TransactionStatePendingAdmin, + TransactionStateSuccessful, + TransactionStateFailed, + TransactionStateSuspended: + return true + default: + return false + } +} + +type TransactionEvent struct { + ID string + SourceService string + IdempotencyKey string + SourceTransactionID string + TrackingCode string + EventVersion uint32 + State TransactionState + ErrorCode string + ErrorMessage string + OccurredAt time.Time + RecordedAt time.Time + CorrelationID string + ActorID string + Blockchain BlockchainReference + Metadata map[string]string + PayloadHash string +} + +func (e TransactionEvent) Validate() error { + if e.ID == "" || e.SourceService == "" || e.IdempotencyKey == "" || e.SourceTransactionID == "" { + return fmt.Errorf("transaction event identity fields are required") + } + if e.EventVersion == 0 { + return fmt.Errorf("event version must be positive") + } + if !e.State.Valid() { + return fmt.Errorf("invalid transaction state %q", e.State) + } + if e.OccurredAt.IsZero() { + return fmt.Errorf("occurred time is required") + } + if len(e.PayloadHash) != 64 || e.PayloadHash != strings.ToLower(e.PayloadHash) { + return fmt.Errorf("payload hash must be a lowercase SHA-256 hex string") + } + if _, err := hex.DecodeString(e.PayloadHash); err != nil { + return fmt.Errorf("payload hash must be a lowercase SHA-256 hex string") + } + return nil +} diff --git a/domain/ledger/journal.go b/domain/ledger/journal.go index 47c82b3..c885cef 100644 --- a/domain/ledger/journal.go +++ b/domain/ledger/journal.go @@ -80,6 +80,7 @@ type Journal struct { Entries []Entry ReversalOfJournalID string OccurredAt time.Time + RecordedAt time.Time CorrelationID string ActorID string Blockchain BlockchainReference @@ -87,6 +88,20 @@ type Journal struct { PayloadHash string } +type AppendResult struct { + JournalID string + AlreadyExists bool +} + +type JournalFilter struct { + Account *AccountReference + AssetID *int64 + RecordedFrom *time.Time + RecordedTo *time.Time + Limit int + Offset int +} + func (j Journal) Validate() error { if j.ID == "" || j.SourceService == "" || j.IdempotencyKey == "" || j.SourceTransactionID == "" || j.EffectKind == "" { return fmt.Errorf("journal identity fields are required") diff --git a/infrastructure/postgres/database.go b/infrastructure/postgres/database.go index a9f8272..f5e5a16 100644 --- a/infrastructure/postgres/database.go +++ b/infrastructure/postgres/database.go @@ -18,6 +18,13 @@ type Row interface { Scan(dest ...any) error } +type Rows interface { + Next() bool + Scan(dest ...any) error + Err() error + Close() +} + type Tx interface { Exec(context.Context, string, ...any) (pgconn.CommandTag, error) QueryRow(context.Context, string, ...any) Row @@ -27,7 +34,9 @@ type Tx interface { type Database interface { Begin(context.Context) (Tx, error) + Exec(context.Context, string, ...any) (pgconn.CommandTag, error) QueryRow(context.Context, string, ...any) Row + Query(context.Context, string, ...any) (Rows, error) Ping(context.Context) error Close() } @@ -70,6 +79,14 @@ func (p *Pool) QueryRow(ctx context.Context, sql string, args ...any) Row { return p.pool.QueryRow(ctx, sql, args...) } +func (p *Pool) Query(ctx context.Context, sql string, args ...any) (Rows, error) { + return p.pool.Query(ctx, sql, args...) +} + +func (p *Pool) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) { + return p.pool.Exec(ctx, sql, args...) +} + func (p *Pool) Close() { p.pool.Close() } diff --git a/infrastructure/postgres/event_repository.go b/infrastructure/postgres/event_repository.go new file mode 100644 index 0000000..b0fc698 --- /dev/null +++ b/infrastructure/postgres/event_repository.go @@ -0,0 +1,106 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "gl/domain/ledger" + + "github.com/jackc/pgx/v5" +) + +func (r *JournalRepository) AppendEvent(ctx context.Context, event ledger.TransactionEvent) (ledger.TransactionEvent, bool, error) { + if err := event.Validate(); err != nil { + return ledger.TransactionEvent{}, false, fmt.Errorf("validate transaction event: %w", err) + } + metadataValues := event.Metadata + if metadataValues == nil { + metadataValues = map[string]string{} + } + metadata, err := json.Marshal(metadataValues) + if err != nil { + return ledger.TransactionEvent{}, false, fmt.Errorf("encode event metadata: %w", err) + } + + err = r.database.QueryRow(ctx, insertEventSQL, + event.ID, + event.SourceService, + event.IdempotencyKey, + event.SourceTransactionID, + event.TrackingCode, + event.EventVersion, + event.State, + event.ErrorCode, + event.ErrorMessage, + event.OccurredAt, + event.CorrelationID, + event.ActorID, + event.Blockchain.Network, + event.Blockchain.TransactionHash, + event.Blockchain.LedgerSequence, + metadata, + event.PayloadHash, + ).Scan(&event.RecordedAt) + if !errors.Is(err, pgx.ErrNoRows) { + if err != nil { + return ledger.TransactionEvent{}, false, fmt.Errorf("insert transaction event: %w", err) + } + return event, false, nil + } + + var stored ledger.TransactionEvent + var storedMetadata []byte + err = r.database.QueryRow(ctx, selectEventByIdempotencySQL, event.IdempotencyKey).Scan( + &stored.ID, + &stored.SourceService, + &stored.IdempotencyKey, + &stored.SourceTransactionID, + &stored.TrackingCode, + &stored.EventVersion, + &stored.State, + &stored.ErrorCode, + &stored.ErrorMessage, + &stored.OccurredAt, + &stored.RecordedAt, + &stored.CorrelationID, + &stored.ActorID, + &stored.Blockchain.Network, + &stored.Blockchain.TransactionHash, + &stored.Blockchain.LedgerSequence, + &storedMetadata, + &stored.PayloadHash, + ) + if err != nil { + return ledger.TransactionEvent{}, false, fmt.Errorf("read idempotent transaction event: %w", err) + } + if stored.PayloadHash != event.PayloadHash { + return ledger.TransactionEvent{}, false, ErrIdempotencyConflict + } + if err := json.Unmarshal(storedMetadata, &stored.Metadata); err != nil { + return ledger.TransactionEvent{}, false, fmt.Errorf("decode event metadata: %w", err) + } + return stored, true, nil +} + +const insertEventSQL = ` +INSERT INTO transaction_events ( + id, source_service, idempotency_key, source_transaction_id, tracking_code, + event_version, state, error_code, error_message, occurred_at, correlation_id, + actor_id, blockchain_network, blockchain_transaction_hash, + blockchain_ledger_sequence, metadata, payload_hash +) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 +) +ON CONFLICT (idempotency_key) DO NOTHING +RETURNING recorded_at` + +const selectEventByIdempotencySQL = ` +SELECT id, source_service, idempotency_key, source_transaction_id, + tracking_code, event_version, state, error_code, error_message, + occurred_at, recorded_at, correlation_id, actor_id, blockchain_network, + blockchain_transaction_hash, blockchain_ledger_sequence, metadata, + payload_hash +FROM transaction_events +WHERE idempotency_key = $1` diff --git a/infrastructure/postgres/event_repository_test.go b/infrastructure/postgres/event_repository_test.go new file mode 100644 index 0000000..8149292 --- /dev/null +++ b/infrastructure/postgres/event_repository_test.go @@ -0,0 +1,73 @@ +package postgres + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "gl/domain/ledger" + + "github.com/jackc/pgx/v5" +) + +func TestAppendEventStoresAndReturnsRecordingTime(t *testing.T) { + event := repositoryEvent() + recordedAt := time.Unix(2, 0).UTC() + database := &fakeDatabase{directRows: []Row{fakeRow{values: []any{recordedAt}}}} + + stored, existed, err := NewJournalRepository(database).AppendEvent(context.Background(), event) + if err != nil { + t.Fatal(err) + } + if existed || !stored.RecordedAt.Equal(recordedAt) { + t.Fatalf("unexpected append result: existed=%v event=%+v", existed, stored) + } +} + +func TestAppendEventRejectsConflictingIdempotencyPayload(t *testing.T) { + event := repositoryEvent() + storedHash := strings.Repeat("b", 64) + database := &fakeDatabase{directRows: []Row{ + fakeRow{err: pgx.ErrNoRows}, + fakeRow{values: []any{ + event.ID, + event.SourceService, + event.IdempotencyKey, + event.SourceTransactionID, + event.TrackingCode, + event.EventVersion, + event.State, + event.ErrorCode, + event.ErrorMessage, + event.OccurredAt, + time.Unix(2, 0).UTC(), + event.CorrelationID, + event.ActorID, + event.Blockchain.Network, + event.Blockchain.TransactionHash, + event.Blockchain.LedgerSequence, + []byte(`{}`), + storedHash, + }}, + }} + + _, _, err := NewJournalRepository(database).AppendEvent(context.Background(), event) + if !errors.Is(err, ErrIdempotencyConflict) { + t.Fatalf("expected idempotency conflict, got %v", err) + } +} + +func repositoryEvent() ledger.TransactionEvent { + return ledger.TransactionEvent{ + ID: "11111111-1111-4111-8111-111111111111", + SourceService: "wallet", + IdempotencyKey: "wallet:1:status:v1", + SourceTransactionID: "1", + EventVersion: 1, + State: ledger.TransactionStateCreated, + OccurredAt: time.Unix(1, 0).UTC(), + PayloadHash: strings.Repeat("a", 64), + } +} diff --git a/infrastructure/postgres/journal_repository.go b/infrastructure/postgres/journal_repository.go index 4bc5126..72c593b 100644 --- a/infrastructure/postgres/journal_repository.go +++ b/infrastructure/postgres/journal_repository.go @@ -9,17 +9,15 @@ import ( "gl/domain/ledger" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" ) var ( - ErrIdempotencyConflict = errors.New("idempotency key belongs to a different payload") - ErrIncompleteJournal = errors.New("idempotency key belongs to an unsealed journal") + ErrIdempotencyConflict = ledger.ErrIdempotencyConflict + ErrIncompleteJournal = ledger.ErrIncompleteJournal ) -type AppendResult struct { - JournalID string - AlreadyExists bool -} +type AppendResult = ledger.AppendResult type JournalRepository struct { database Database @@ -76,6 +74,10 @@ func (r *JournalRepository) Append(ctx context.Context, journal ledger.Journal) return r.resolveDuplicate(ctx, journal) } if err != nil { + var postgresError *pgconn.PgError + if errors.As(err, &postgresError) && postgresError.ConstraintName == "journals_one_reversal_idx" { + return AppendResult{}, ledger.ErrAlreadyReversed + } return AppendResult{}, fmt.Errorf("insert journal: %w", err) } diff --git a/infrastructure/postgres/journal_repository_test.go b/infrastructure/postgres/journal_repository_test.go index 12bf4d3..a90154f 100644 --- a/infrastructure/postgres/journal_repository_test.go +++ b/infrastructure/postgres/journal_repository_test.go @@ -33,6 +33,14 @@ func (r fakeRow) Scan(dest ...any) error { *target = value.(int64) case *bool: *target = value.(bool) + case *uint32: + *target = value.(uint32) + case *time.Time: + *target = value.(time.Time) + case *ledger.TransactionState: + *target = value.(ledger.TransactionState) + case *[]byte: + *target = value.([]byte) default: return errors.New("unsupported scan target") } @@ -75,6 +83,7 @@ func (t *fakeTx) Rollback(context.Context) error { type fakeDatabase struct { tx *fakeTx directRow Row + directRows []Row beginCalled bool } @@ -83,9 +92,22 @@ func (d *fakeDatabase) Begin(context.Context) (Tx, error) { return d.tx, nil } -func (d *fakeDatabase) QueryRow(context.Context, string, ...any) Row { return d.directRow } -func (d *fakeDatabase) Ping(context.Context) error { return nil } -func (d *fakeDatabase) Close() {} +func (d *fakeDatabase) QueryRow(context.Context, string, ...any) Row { + if len(d.directRows) > 0 { + row := d.directRows[0] + d.directRows = d.directRows[1:] + return row + } + return d.directRow +} +func (d *fakeDatabase) Query(context.Context, string, ...any) (Rows, error) { + return nil, errors.New("unexpected query") +} +func (d *fakeDatabase) Exec(context.Context, string, ...any) (pgconn.CommandTag, error) { + return pgconn.CommandTag{}, errors.New("unexpected exec") +} +func (d *fakeDatabase) Ping(context.Context) error { return nil } +func (d *fakeDatabase) Close() {} func TestJournalRepositoryAppendCommitsJournalEntriesAndSeal(t *testing.T) { journal := repositoryJournal(t) diff --git a/infrastructure/postgres/migration_test.go b/infrastructure/postgres/migration_test.go index 5b612ba..ea3a977 100644 --- a/infrastructure/postgres/migration_test.go +++ b/infrastructure/postgres/migration_test.go @@ -18,6 +18,7 @@ func TestInitialMigrationContainsLedgerSafetyGuards(t *testing.T) { "journal is not balanced per asset", "cannot append to a sealed journal", "reject_ledger_mutation", + "journals_one_reversal_idx", } { if !strings.Contains(schema, required) { t.Fatalf("migration is missing %q", required) diff --git a/infrastructure/postgres/migrations/000001_init.up.sql b/infrastructure/postgres/migrations/000001_init.up.sql index b577c96..5d86fba 100644 --- a/infrastructure/postgres/migrations/000001_init.up.sql +++ b/infrastructure/postgres/migrations/000001_init.up.sql @@ -53,6 +53,10 @@ CREATE INDEX journals_source_transaction_idx CREATE INDEX journals_recorded_at_idx ON journals (recorded_at, id); +CREATE UNIQUE INDEX journals_one_reversal_idx + ON journals (reversal_of_journal_id) + WHERE reversal_of_journal_id IS NOT NULL; + CREATE TABLE journal_entries ( journal_id uuid NOT NULL REFERENCES journals (id), line_number integer NOT NULL CHECK (line_number > 0), diff --git a/infrastructure/postgres/query_repository.go b/infrastructure/postgres/query_repository.go new file mode 100644 index 0000000..50a01e6 --- /dev/null +++ b/infrastructure/postgres/query_repository.go @@ -0,0 +1,239 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "gl/domain/ledger" + + "github.com/jackc/pgx/v5" +) + +var ErrNotFound = ledger.ErrNotFound + +type JournalFilter = ledger.JournalFilter + +func (r *JournalRepository) GetByID(ctx context.Context, journalID string) (ledger.Journal, error) { + return r.get(ctx, getJournalByIDSQL, journalID) +} + +func (r *JournalRepository) GetByIdempotencyKey(ctx context.Context, idempotencyKey string) (ledger.Journal, error) { + return r.get(ctx, getJournalByIdempotencySQL, idempotencyKey) +} + +func (r *JournalRepository) get(ctx context.Context, query string, value string) (ledger.Journal, error) { + journal, err := scanJournal(r.database.QueryRow(ctx, query, value)) + if errors.Is(err, pgx.ErrNoRows) { + return ledger.Journal{}, ErrNotFound + } + if err != nil { + return ledger.Journal{}, fmt.Errorf("read journal: %w", err) + } + journal.Entries, err = r.loadEntries(ctx, journal.ID) + if err != nil { + return ledger.Journal{}, err + } + return journal, nil +} + +func (r *JournalRepository) List(ctx context.Context, filter JournalFilter) ([]ledger.Journal, error) { + limit := filter.Limit + if limit <= 0 || limit > 201 { + limit = 50 + } + if filter.Offset < 0 { + filter.Offset = 0 + } + + var accountClass, ownerType, ownerID any + if filter.Account != nil { + accountClass = filter.Account.Class + ownerType = filter.Account.OwnerType + ownerID = filter.Account.OwnerID + } + rows, err := r.database.Query(ctx, listJournalsSQL, + filter.AssetID, + accountClass, + ownerType, + ownerID, + filter.RecordedFrom, + filter.RecordedTo, + limit, + filter.Offset, + ) + if err != nil { + return nil, fmt.Errorf("list journals: %w", err) + } + defer rows.Close() + + journals := make([]ledger.Journal, 0, limit) + for rows.Next() { + journal, scanErr := scanJournal(rows) + if scanErr != nil { + return nil, fmt.Errorf("scan journal: %w", scanErr) + } + journals = append(journals, journal) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate journals: %w", err) + } + rows.Close() + + for index := range journals { + journals[index].Entries, err = r.loadEntries(ctx, journals[index].ID) + if err != nil { + return nil, err + } + } + return journals, nil +} + +func (r *JournalRepository) Balance(ctx context.Context, account ledger.AccountReference, asOf time.Time) (ledger.Amount, error) { + if err := account.Validate(); err != nil { + return ledger.Amount{}, err + } + var asOfValue any + if !asOf.IsZero() { + asOfValue = asOf + } + var value string + if err := r.database.QueryRow(ctx, getBalanceSQL, + account.Class, + account.OwnerType, + account.OwnerID, + account.AssetID, + asOfValue, + ).Scan(&value); err != nil { + return ledger.Amount{}, fmt.Errorf("read balance: %w", err) + } + amount, err := ledger.ParseAmount(value) + if err != nil { + return ledger.Amount{}, fmt.Errorf("decode balance: %w", err) + } + return amount, nil +} + +type scanner interface { + Scan(...any) error +} + +func scanJournal(row scanner) (ledger.Journal, error) { + var ( + journal ledger.Journal + reversal *string + metadata []byte + ) + err := row.Scan( + &journal.ID, + &journal.SourceService, + &journal.IdempotencyKey, + &journal.SourceTransactionID, + &journal.TrackingCode, + &journal.EffectKind, + &journal.EventVersion, + &reversal, + &journal.OccurredAt, + &journal.RecordedAt, + &journal.CorrelationID, + &journal.ActorID, + &journal.Blockchain.Network, + &journal.Blockchain.TransactionHash, + &journal.Blockchain.LedgerSequence, + &metadata, + &journal.PayloadHash, + ) + if err != nil { + return ledger.Journal{}, err + } + if reversal != nil { + journal.ReversalOfJournalID = *reversal + } + if err := json.Unmarshal(metadata, &journal.Metadata); err != nil { + return ledger.Journal{}, fmt.Errorf("decode metadata: %w", err) + } + return journal, nil +} + +func (r *JournalRepository) loadEntries(ctx context.Context, journalID string) ([]ledger.Entry, error) { + rows, err := r.database.Query(ctx, getEntriesSQL, journalID) + if err != nil { + return nil, fmt.Errorf("read journal entries: %w", err) + } + defer rows.Close() + + entries := make([]ledger.Entry, 0) + for rows.Next() { + var ( + entry ledger.Entry + amountValue string + ) + if err := rows.Scan( + &entry.LineNumber, + &entry.Account.Class, + &entry.Account.OwnerType, + &entry.Account.OwnerID, + &entry.Account.AssetID, + &amountValue, + &entry.Description, + ); err != nil { + return nil, fmt.Errorf("scan journal entry: %w", err) + } + entry.Amount, err = ledger.ParseAmount(amountValue) + if err != nil { + return nil, fmt.Errorf("decode journal entry amount: %w", err) + } + entries = append(entries, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate journal entries: %w", err) + } + return entries, nil +} + +const journalColumns = ` + j.id, j.source_service, j.idempotency_key, j.source_transaction_id, + j.tracking_code, j.effect_kind, j.event_version, j.reversal_of_journal_id, + j.occurred_at, j.recorded_at, j.correlation_id, j.actor_id, + j.blockchain_network, j.blockchain_transaction_hash, + j.blockchain_ledger_sequence, j.metadata, j.payload_hash` + +const getJournalByIDSQL = `SELECT ` + journalColumns + ` +FROM journals j WHERE j.id = $1 AND j.sealed_at IS NOT NULL` + +const getJournalByIdempotencySQL = `SELECT ` + journalColumns + ` +FROM journals j WHERE j.idempotency_key = $1 AND j.sealed_at IS NOT NULL` + +const listJournalsSQL = `SELECT DISTINCT ` + journalColumns + ` +FROM journals j +JOIN journal_entries e ON e.journal_id = j.id +JOIN ledger_accounts a ON a.id = e.account_id +WHERE j.sealed_at IS NOT NULL + AND ($1::bigint IS NULL OR e.asset_id = $1) + AND ($2::text IS NULL OR ( + a.class = $2 AND a.owner_type = $3 AND a.owner_id = $4 + )) + AND ($5::timestamptz IS NULL OR j.recorded_at >= $5) + AND ($6::timestamptz IS NULL OR j.recorded_at <= $6) +ORDER BY j.recorded_at DESC, j.id DESC +LIMIT $7 OFFSET $8` + +const getEntriesSQL = ` +SELECT e.line_number, a.class, a.owner_type, a.owner_id, e.asset_id, + e.amount::text, e.description +FROM journal_entries e +JOIN ledger_accounts a ON a.id = e.account_id +WHERE e.journal_id = $1 +ORDER BY e.line_number` + +const getBalanceSQL = ` +SELECT COALESCE(sum(e.amount), 0)::text +FROM journal_entries e +JOIN ledger_accounts a ON a.id = e.account_id +JOIN journals j ON j.id = e.journal_id +WHERE j.sealed_at IS NOT NULL + AND a.class = $1 AND a.owner_type = $2 AND a.owner_id = $3 + AND a.asset_id = $4 + AND ($5::timestamptz IS NULL OR j.recorded_at <= $5)` diff --git a/interface/grpc/health.go b/interface/grpc/health.go index f49e534..4757e5a 100644 --- a/interface/grpc/health.go +++ b/interface/grpc/health.go @@ -4,21 +4,27 @@ import ( "context" "gl/application/health" + applicationledger "gl/application/ledger" basev1 "gl/gen/base/v1" ledgerv1 "gl/gen/ledger/v1" ) -type HealthHandler struct { +type Handler struct { ledgerv1.UnimplementedGeneralLedgerServiceServer - service *health.Service + health *health.Service + ledger *applicationledger.Service } -func NewHealthHandler(service *health.Service) *HealthHandler { - return &HealthHandler{service: service} +func NewHandler(healthService *health.Service, ledgerService *applicationledger.Service) *Handler { + return &Handler{health: healthService, ledger: ledgerService} } -func (h *HealthHandler) Health(ctx context.Context, _ *basev1.Empty) (*ledgerv1.HealthResponse, error) { - result := h.service.Check(ctx) +func NewHealthHandler(service *health.Service) *Handler { + return NewHandler(service, nil) +} + +func (h *Handler) Health(ctx context.Context, _ *basev1.Empty) (*ledgerv1.HealthResponse, error) { + result := h.health.Check(ctx) return &ledgerv1.HealthResponse{ Serving: result.Serving, DatabaseReady: result.DatabaseReady, diff --git a/interface/grpc/ledger.go b/interface/grpc/ledger.go new file mode 100644 index 0000000..acff151 --- /dev/null +++ b/interface/grpc/ledger.go @@ -0,0 +1,408 @@ +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") + } +} diff --git a/interface/grpc/ledger_test.go b/interface/grpc/ledger_test.go new file mode 100644 index 0000000..6cec019 --- /dev/null +++ b/interface/grpc/ledger_test.go @@ -0,0 +1,123 @@ +package grpcadapter + +import ( + "context" + "errors" + "testing" + "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" +) + +type ledgerRepositoryStub struct { + journal domain.Journal +} + +func (r *ledgerRepositoryStub) Append(_ context.Context, journal domain.Journal) (domain.AppendResult, error) { + journal.RecordedAt = time.Unix(2, 0).UTC() + r.journal = journal + return domain.AppendResult{JournalID: journal.ID}, nil +} + +func (r *ledgerRepositoryStub) AppendEvent(_ context.Context, event domain.TransactionEvent) (domain.TransactionEvent, bool, error) { + event.RecordedAt = time.Unix(2, 0).UTC() + return event, false, nil +} + +func (r *ledgerRepositoryStub) GetByID(context.Context, string) (domain.Journal, error) { + return r.journal, nil +} + +func (r *ledgerRepositoryStub) GetByIdempotencyKey(context.Context, string) (domain.Journal, error) { + return r.journal, nil +} + +func (r *ledgerRepositoryStub) List(context.Context, domain.JournalFilter) ([]domain.Journal, error) { + return []domain.Journal{r.journal}, nil +} + +func (r *ledgerRepositoryStub) Balance(context.Context, domain.AccountReference, time.Time) (domain.Amount, error) { + return domain.ParseAmount("12.5") +} + +func TestAppendJournalMapsProtoToExactDomainAndBack(t *testing.T) { + repository := &ledgerRepositoryStub{} + service := applicationledger.NewService(repository, func() (string, error) { + return "11111111-1111-4111-8111-111111111111", nil + }) + handler := NewHandler(nil, service) + + response, err := handler.AppendJournal(context.Background(), &ledgerv1.AppendJournalRequest{ + SourceService: "wallet", + IdempotencyKey: "wallet:1:internal-transfer:v1", + SourceTransactionId: "1", + EffectKind: "internal-transfer", + EventVersion: 1, + OccurredAt: timestamppb.New(time.Unix(1, 0).UTC()), + Entries: []*ledgerv1.JournalEntry{ + { + LineNumber: 1, + Account: &ledgerv1.AccountReference{ + AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE, + OwnerType: "user", + OwnerId: "1", + AssetId: 5, + }, + Amount: "-1.2500", + }, + { + LineNumber: 2, + Account: &ledgerv1.AccountReference{ + AccountClass: ledgerv1.AccountClass_ACCOUNT_CLASS_USER_AVAILABLE, + OwnerType: "user", + OwnerId: "2", + AssetId: 5, + }, + Amount: "1.25", + }, + }, + }) + if err != nil { + t.Fatal(err) + } + if got := response.GetJournal().GetEntries()[0].GetAmount(); got != "-1.25" { + t.Fatalf("unexpected canonical amount: %q", got) + } + if response.GetJournal().GetPayloadHash() == "" { + t.Fatal("payload hash is missing") + } +} + +func TestAppendJournalRejectsMissingTimestamp(t *testing.T) { + handler := NewHandler(nil, nil) + _, err := handler.AppendJournal(context.Background(), &ledgerv1.AppendJournalRequest{}) + if status.Code(err) != codes.InvalidArgument { + t.Fatalf("expected InvalidArgument, got %v", err) + } +} + +func TestRPCErrorMapping(t *testing.T) { + for _, testCase := range []struct { + err error + code codes.Code + }{ + {err: applicationledger.ErrInvalidArgument, code: codes.InvalidArgument}, + {err: domain.ErrNotFound, code: codes.NotFound}, + {err: domain.ErrIdempotencyConflict, code: codes.AlreadyExists}, + {err: domain.ErrIncompleteJournal, code: codes.FailedPrecondition}, + {err: domain.ErrAlreadyReversed, code: codes.FailedPrecondition}, + {err: context.Canceled, code: codes.Canceled}, + {err: context.DeadlineExceeded, code: codes.DeadlineExceeded}, + {err: errors.New("database details must not escape"), code: codes.Internal}, + } { + if got := status.Code(rpcError(testCase.err)); got != testCase.code { + t.Fatalf("rpcError(%v) = %v, want %v", testCase.err, got, testCase.code) + } + } +}