tech/sendico
2
0
Fork 0
Code Issues 30 Pull Requests Actions Packages Projects Releases Wiki Activity

idempotency key usage fix

Browse Source
This commit is contained in:
Stephan D
2026-01-21 15:23:50 +01:00
parent a15375f18e
commit d2e78356e6
48 changed files with 729 additions and 559 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
api/ledger/go.mod
View File

@@ -51,5 +51,5 @@ require (
golang.org/x/sync v0.19.0 // indirect
golang.org/x/sys v0.40.0 // indirect
golang.org/x/text v0.33.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260120174246-409b4a993575 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516 // indirect
)

4
api/ledger/go.sum
View File

@@ -214,8 +214,8 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260120174246-409b4a993575 h1:vzOYHDZEHIsPYYnaSYo60AqHkJronSu0rzTz/s4quL0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260120174246-409b4a993575/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516 h1:sNrWoksmOyF5bvJUcnmbeAmQi8baNhqg5IWaI3llQqU=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260120221211-b8f7ae30c516/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc=
google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=

15
api/ledger/internal/service/ledger/accounts.go
View File

@@ -11,10 +11,11 @@ import (
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
pmodel "github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mutil/mzap"
describablev1 "github.com/tech/sendico/pkg/proto/common/describable/v1"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.uber.org/zap"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
"google.golang.org/protobuf/types/known/timestamppb"
)
@@ -96,7 +97,7 @@ func (s *Service) createAccountResponder(_ context.Context, req *ledgerv1.Create
if lookupErr != nil {
s.logger.Warn("duplicate account create but failed to load existing",
zap.Error(lookupErr),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("accountCode", accountCode),
zap.String("currency", currency))
return nil, merrors.Internal("failed to load existing account after conflict")
@@ -109,7 +110,7 @@ func (s *Service) createAccountResponder(_ context.Context, req *ledgerv1.Create
recordAccountOperation("create", "error")
s.logger.Warn("failed to create account",
zap.Error(err),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("accountCode", accountCode),
zap.String("currency", currency))
return nil, merrors.Internal("failed to create account")
@@ -279,7 +280,7 @@ func (s *Service) ensureSettlementAccount(ctx context.Context, orgRef primitive.
if !errors.Is(err, storage.ErrAccountNotFound) {
s.logger.Warn("failed to resolve default settlement account",
zap.Error(err),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("currency", normalizedCurrency))
return nil, merrors.Internal("failed to resolve settlement account")
}
@@ -306,20 +307,20 @@ func (s *Service) ensureSettlementAccount(ctx context.Context, orgRef primitive.
}
s.logger.Warn("duplicate settlement account create but failed to load existing",
zap.Error(lookupErr),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("currency", normalizedCurrency))
return nil, merrors.Internal("failed to resolve settlement account after conflict")
}
s.logger.Warn("failed to create default settlement account",
zap.Error(err),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("currency", normalizedCurrency),
zap.String("accountCode", accountCode))
return nil, merrors.Internal("failed to create settlement account")
}
s.logger.Info("default settlement account created",
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("currency", normalizedCurrency),
zap.String("accountCode", accountCode))
return account, nil

3
api/ledger/internal/service/ledger/list_accounts.go
View File

@@ -6,6 +6,7 @@ import (
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.uber.org/zap"
)
@@ -31,7 +32,7 @@ func (s *Service) listAccountsResponder(_ context.Context, req *ledgerv1.ListAcc
// No pagination requested; return all accounts for the organization.
accounts, err := s.storage.Accounts().ListByOrganization(ctx, orgRef, 0, 0)
if err != nil {
s.logger.Warn("failed to list ledger accounts", zap.Error(err), zap.String("organizationRef", orgRef.Hex()))
s.logger.Warn("failed to list ledger accounts", zap.Error(err), mzap.ObjRef("organization_ref", orgRef))
return nil, err
}

26
api/ledger/internal/service/ledger/posting.go
View File

@@ -3,6 +3,7 @@ package ledger
import (
"context"
"fmt"
"strings"
"time"
"github.com/tech/sendico/ledger/storage"
@@ -10,6 +11,7 @@ import (
storageMongo "github.com/tech/sendico/ledger/storage/mongo"
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
@@ -41,12 +43,20 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
if err != nil {
return nil, err
}
logger := s.logger.With(
zap.String("idempotency_key", req.IdempotencyKey),
mzap.ObjRef("organization_ref", orgRef),
mzap.ObjRef("ledger_account_ref", accountRef),
zap.String("currency", req.Money.Currency),
)
if strings.TrimSpace(req.ContraLedgerAccountRef) != "" {
logger = logger.With(zap.String("contra_ledger_account_ref", strings.TrimSpace(req.ContraLedgerAccountRef)))
}
existingEntry, err := s.storage.JournalEntries().GetByIdempotencyKey(ctx, orgRef, req.IdempotencyKey)
if err == nil && existingEntry != nil {
recordDuplicateRequest("credit")
s.logger.Info("duplicate credit request (idempotency)",
zap.String("idempotencyKey", req.IdempotencyKey),
logger.Info("duplicate credit request (idempotency)",
zap.String("existingEntryID", existingEntry.GetID().Hex()))
return &ledgerv1.PostResponse{
JournalEntryRef: existingEntry.GetID().Hex(),
@@ -56,7 +66,7 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
}
if err != nil && err != storage.ErrJournalEntryNotFound {
recordJournalEntryError("credit", "idempotency_check_failed")
s.logger.Warn("failed to check idempotency", zap.Error(err))
logger.Warn("failed to check idempotency", zap.Error(err))
return nil, merrors.Internal("failed to check idempotency")
}
@@ -67,7 +77,7 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
return nil, merrors.NoData("account not found")
}
recordJournalEntryError("credit", "account_lookup_failed")
s.logger.Warn("failed to get account", zap.Error(err))
logger.Warn("failed to get account", zap.Error(err))
return nil, merrors.Internal("failed to get account")
}
if err := validateAccountForOrg(account, orgRef, req.Money.Currency); err != nil {
@@ -84,7 +94,7 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
charges := req.Charges
if len(charges) == 0 {
if computed, err := s.quoteFeesForCredit(ctx, req); err != nil {
s.logger.Warn("failed to quote fees", zap.Error(err))
logger.Warn("failed to quote fees", zap.Error(err))
} else if len(computed) > 0 {
charges = computed
}
@@ -118,7 +128,7 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData(fmt.Sprintf("charges[%d]: account not found", i))
}
s.logger.Warn("failed to get charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
logger.Warn("failed to get charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
return nil, merrors.Internal("failed to get charge account")
}
if err := validateAccountForOrg(chargeAccount, orgRef, charge.Money.Currency); err != nil {
@@ -189,7 +199,7 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
entry.OrganizationRef = orgRef
if err := s.storage.JournalEntries().Create(txCtx, entry); err != nil {
s.logger.Warn("failed to create journal entry", zap.Error(err))
logger.Warn("failed to create journal entry", zap.Error(err))
return nil, merrors.Internal("failed to create journal entry")
}
@@ -207,7 +217,7 @@ func (s *Service) postCreditResponder(_ context.Context, req *ledgerv1.PostCredi
}
if err := s.storage.PostingLines().CreateMany(txCtx, postingLines); err != nil {
s.logger.Warn("failed to create posting lines", zap.Error(err))
logger.Warn("failed to create posting lines", zap.Error(err))
return nil, merrors.Internal("failed to create posting lines")
}

26
api/ledger/internal/service/ledger/posting_debit.go
View File

@@ -3,6 +3,7 @@ package ledger
import (
"context"
"fmt"
"strings"
"time"
"github.com/tech/sendico/ledger/storage"
@@ -10,6 +11,7 @@ import (
storageMongo "github.com/tech/sendico/ledger/storage/mongo"
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
@@ -39,12 +41,20 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
if err != nil {
return nil, err
}
logger := s.logger.With(
zap.String("idempotency_key", req.IdempotencyKey),
mzap.ObjRef("organization_ref", orgRef),
mzap.ObjRef("ledger_account_ref", accountRef),
zap.String("currency", req.Money.Currency),
)
if strings.TrimSpace(req.ContraLedgerAccountRef) != "" {
logger = logger.With(zap.String("contra_ledger_account_ref", strings.TrimSpace(req.ContraLedgerAccountRef)))
}
existingEntry, err := s.storage.JournalEntries().GetByIdempotencyKey(ctx, orgRef, req.IdempotencyKey)
if err == nil && existingEntry != nil {
recordDuplicateRequest("debit")
s.logger.Info("duplicate debit request (idempotency)",
zap.String("idempotencyKey", req.IdempotencyKey),
logger.Info("duplicate debit request (idempotency)",
zap.String("existingEntryID", existingEntry.GetID().Hex()))
return &ledgerv1.PostResponse{
JournalEntryRef: existingEntry.GetID().Hex(),
@@ -53,7 +63,7 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
}, nil
}
if err != nil && err != storage.ErrJournalEntryNotFound {
s.logger.Warn("failed to check idempotency", zap.Error(err))
logger.Warn("failed to check idempotency", zap.Error(err))
return nil, merrors.Internal("failed to check idempotency")
}
@@ -62,7 +72,7 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("account not found")
}
s.logger.Warn("failed to get account", zap.Error(err))
logger.Warn("failed to get account", zap.Error(err))
return nil, merrors.Internal("failed to get account")
}
if err := validateAccountForOrg(account, orgRef, req.Money.Currency); err != nil {
@@ -78,7 +88,7 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
charges := req.Charges
if len(charges) == 0 {
if computed, err := s.quoteFeesForDebit(ctx, req); err != nil {
s.logger.Warn("failed to quote fees", zap.Error(err))
logger.Warn("failed to quote fees", zap.Error(err))
} else if len(computed) > 0 {
charges = computed
}
@@ -112,7 +122,7 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData(fmt.Sprintf("charges[%d]: account not found", i))
}
s.logger.Warn("failed to get charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
logger.Warn("failed to get charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
return nil, merrors.Internal("failed to get charge account")
}
if err := validateAccountForOrg(chargeAccount, orgRef, charge.Money.Currency); err != nil {
@@ -183,7 +193,7 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
entry.OrganizationRef = orgRef
if err := s.storage.JournalEntries().Create(txCtx, entry); err != nil {
s.logger.Warn("failed to create journal entry", zap.Error(err))
logger.Warn("failed to create journal entry", zap.Error(err))
return nil, merrors.Internal("failed to create journal entry")
}
@@ -201,7 +211,7 @@ func (s *Service) postDebitResponder(_ context.Context, req *ledgerv1.PostDebitR
}
if err := s.storage.PostingLines().CreateMany(txCtx, postingLines); err != nil {
s.logger.Warn("failed to create posting lines", zap.Error(err))
logger.Warn("failed to create posting lines", zap.Error(err))
return nil, merrors.Internal("failed to create posting lines")
}

25
api/ledger/internal/service/ledger/posting_fx.go
View File

@@ -10,6 +10,7 @@ import (
storageMongo "github.com/tech/sendico/ledger/storage/mongo"
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
@@ -62,13 +63,21 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
if err != nil {
return nil, err
}
logger := s.logger.With(
zap.String("idempotency_key", req.IdempotencyKey),
mzap.ObjRef("organization_ref", orgRef),
mzap.ObjRef("from_account_ref", fromAccountRef),
mzap.ObjRef("to_account_ref", toAccountRef),
zap.String("from_currency", req.FromMoney.Currency),
zap.String("to_currency", req.ToMoney.Currency),
zap.String("rate", req.Rate),
)
// Check for duplicate idempotency key
existingEntry, err := s.storage.JournalEntries().GetByIdempotencyKey(ctx, orgRef, req.IdempotencyKey)
if err == nil && existingEntry != nil {
recordDuplicateRequest("fx")
s.logger.Info("duplicate FX request (idempotency)",
zap.String("idempotencyKey", req.IdempotencyKey),
logger.Info("duplicate FX request (idempotency)",
zap.String("existingEntryID", existingEntry.GetID().Hex()))
return &ledgerv1.PostResponse{
JournalEntryRef: existingEntry.GetID().Hex(),
@@ -77,7 +86,7 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
}, nil
}
if err != nil && err != storage.ErrJournalEntryNotFound {
s.logger.Warn("failed to check idempotency", zap.Error(err))
logger.Warn("failed to check idempotency", zap.Error(err))
return nil, merrors.Internal("failed to check idempotency")
}
@@ -87,7 +96,7 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("from_account not found")
}
s.logger.Warn("failed to get from_account", zap.Error(err))
logger.Warn("failed to get from_account", zap.Error(err))
return nil, merrors.Internal("failed to get from_account")
}
if err := validateAccountForOrg(fromAccount, orgRef, req.FromMoney.Currency); err != nil {
@@ -99,7 +108,7 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("to_account not found")
}
s.logger.Warn("failed to get to_account", zap.Error(err))
logger.Warn("failed to get to_account", zap.Error(err))
return nil, merrors.Internal("failed to get to_account")
}
if err := validateAccountForOrg(toAccount, orgRef, req.ToMoney.Currency); err != nil {
@@ -153,7 +162,7 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData(fmt.Sprintf("charges[%d]: account not found", i))
}
s.logger.Warn("failed to get FX charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
logger.Warn("failed to get FX charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
return nil, merrors.Internal("failed to get charge account")
}
if err := validateAccountForOrg(chargeAccount, orgRef, charge.Money.Currency); err != nil {
@@ -206,7 +215,7 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
entry.OrganizationRef = orgRef
if err := s.storage.JournalEntries().Create(txCtx, entry); err != nil {
s.logger.Warn("failed to create journal entry", zap.Error(err))
logger.Warn("failed to create journal entry", zap.Error(err))
return nil, merrors.Internal("failed to create journal entry")
}
@@ -220,7 +229,7 @@ func (s *Service) fxResponder(_ context.Context, req *ledgerv1.FXRequest) gsresp
}
if err := s.storage.PostingLines().CreateMany(txCtx, postingLines); err != nil {
s.logger.Warn("failed to create posting lines", zap.Error(err))
logger.Warn("failed to create posting lines", zap.Error(err))
return nil, merrors.Internal("failed to create posting lines")
}

9
api/ledger/internal/service/ledger/posting_support.go
View File

@@ -11,6 +11,7 @@ import (
"github.com/tech/sendico/ledger/storage"
"github.com/tech/sendico/ledger/storage/model"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
)
@@ -90,7 +91,7 @@ func (s *Service) resolveSettlementAccount(ctx context.Context, orgRef primitive
}
s.logger.Warn("failed to resolve default settlement account",
zap.Error(err),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("currency", currency))
return nil, merrors.Internal("failed to resolve settlement account")
}
@@ -132,7 +133,7 @@ func (s *Service) upsertBalances(ctx context.Context, lines []*model.PostingLine
for accountRef, delta := range balanceDeltas {
account := accounts[accountRef]
if account == nil {
s.logger.Warn("account cache missing for balance update", zap.String("accountRef", accountRef.Hex()))
s.logger.Warn("account cache missing for balance update", mzap.ObjRef("account_ref", accountRef))
return merrors.Internal("account cache missing for balance update")
}
@@ -140,7 +141,7 @@ func (s *Service) upsertBalances(ctx context.Context, lines []*model.PostingLine
if err != nil && !errors.Is(err, storage.ErrBalanceNotFound) {
s.logger.Warn("failed to fetch account balance",
zap.Error(err),
zap.String("accountRef", accountRef.Hex()))
mzap.ObjRef("account_ref", accountRef))
return merrors.Internal("failed to update balance")
}
@@ -169,7 +170,7 @@ func (s *Service) upsertBalances(ctx context.Context, lines []*model.PostingLine
newBalance.OrganizationRef = account.OrganizationRef
if err := balancesStore.Upsert(ctx, newBalance); err != nil {
s.logger.Warn("failed to upsert account balance", zap.Error(err), zap.String("accountRef", accountRef.Hex()))
s.logger.Warn("failed to upsert account balance", zap.Error(err), mzap.ObjRef("account_ref", accountRef))
return merrors.Internal("failed to update balance")
}
}

23
api/ledger/internal/service/ledger/posting_transfer.go
View File

@@ -10,6 +10,7 @@ import (
storageMongo "github.com/tech/sendico/ledger/storage/mongo"
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
@@ -53,13 +54,19 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
if err != nil {
return nil, err
}
logger := s.logger.With(
zap.String("idempotency_key", req.IdempotencyKey),
mzap.ObjRef("organization_ref", orgRef),
mzap.ObjRef("from_account_ref", fromAccountRef),
mzap.ObjRef("to_account_ref", toAccountRef),
zap.String("currency", req.Money.Currency),
)
// Check for duplicate idempotency key
existingEntry, err := s.storage.JournalEntries().GetByIdempotencyKey(ctx, orgRef, req.IdempotencyKey)
if err == nil && existingEntry != nil {
recordDuplicateRequest("transfer")
s.logger.Info("duplicate transfer request (idempotency)",
zap.String("idempotencyKey", req.IdempotencyKey),
logger.Info("duplicate transfer request (idempotency)",
zap.String("existingEntryID", existingEntry.GetID().Hex()))
return &ledgerv1.PostResponse{
JournalEntryRef: existingEntry.GetID().Hex(),
@@ -68,7 +75,7 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
}, nil
}
if err != nil && err != storage.ErrJournalEntryNotFound {
s.logger.Warn("failed to check idempotency", zap.Error(err))
logger.Warn("failed to check idempotency", zap.Error(err))
return nil, merrors.Internal("failed to check idempotency")
}
@@ -78,7 +85,7 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("from_account not found")
}
s.logger.Warn("failed to get from_account", zap.Error(err))
logger.Warn("failed to get from_account", zap.Error(err))
return nil, merrors.Internal("failed to get from_account")
}
if err := validateAccountForOrg(fromAccount, orgRef, req.Money.Currency); err != nil {
@@ -90,7 +97,7 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("to_account not found")
}
s.logger.Warn("failed to get to_account", zap.Error(err))
logger.Warn("failed to get to_account", zap.Error(err))
return nil, merrors.Internal("failed to get to_account")
}
if err := validateAccountForOrg(toAccount, orgRef, req.Money.Currency); err != nil {
@@ -147,7 +154,7 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData(fmt.Sprintf("charges[%d]: account not found", i))
}
s.logger.Warn("failed to get charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
logger.Warn("failed to get charge account", zap.Error(err), zap.String("chargeAccountRef", chargeAccountRef.Hex()))
return nil, merrors.Internal("failed to get charge account")
}
if err := validateAccountForOrg(chargeAccount, orgRef, charge.Money.Currency); err != nil {
@@ -188,7 +195,7 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
entry.OrganizationRef = orgRef
if err := s.storage.JournalEntries().Create(txCtx, entry); err != nil {
s.logger.Warn("failed to create journal entry", zap.Error(err))
logger.Warn("failed to create journal entry", zap.Error(err))
return nil, merrors.Internal("failed to create journal entry")
}
@@ -206,7 +213,7 @@ func (s *Service) transferResponder(_ context.Context, req *ledgerv1.TransferReq
}
if err := s.storage.PostingLines().CreateMany(txCtx, postingLines); err != nil {
s.logger.Warn("failed to create posting lines", zap.Error(err))
logger.Warn("failed to create posting lines", zap.Error(err))
return nil, merrors.Internal("failed to create posting lines")
}

27
api/ledger/internal/service/ledger/queries.go
View File

@@ -10,6 +10,7 @@ import (
"github.com/tech/sendico/ledger/storage"
"github.com/tech/sendico/pkg/api/routers/gsresponse"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mutil/mzap"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"go.uber.org/zap"
@@ -27,6 +28,7 @@ func (s *Service) getBalanceResponder(_ context.Context, req *ledgerv1.GetBalanc
if err != nil {
return nil, err
}
logger := s.logger.With(mzap.ObjRef("ledger_account_ref", accountRef))
// Get account to verify it exists
account, err := s.storage.Accounts().Get(ctx, accountRef)
@@ -34,7 +36,7 @@ func (s *Service) getBalanceResponder(_ context.Context, req *ledgerv1.GetBalanc
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("account not found")
}
s.logger.Warn("failed to get account", zap.Error(err))
logger.Warn("failed to get account", zap.Error(err))
return nil, merrors.Internal("failed to get account")
}
@@ -53,7 +55,7 @@ func (s *Service) getBalanceResponder(_ context.Context, req *ledgerv1.GetBalanc
LastUpdated: timestamppb.Now(),
}, nil
}
s.logger.Warn("failed to get balance", zap.Error(err))
logger.Warn("failed to get balance", zap.Error(err))
return nil, merrors.Internal("failed to get balance")
}
@@ -82,6 +84,7 @@ func (s *Service) getJournalEntryResponder(_ context.Context, req *ledgerv1.GetE
if err != nil {
return nil, err
}
logger := s.logger.With(mzap.ObjRef("entry_ref", entryRef))
// Get journal entry
entry, err := s.storage.JournalEntries().Get(ctx, entryRef)
@@ -89,14 +92,14 @@ func (s *Service) getJournalEntryResponder(_ context.Context, req *ledgerv1.GetE
if err == storage.ErrJournalEntryNotFound {
return nil, merrors.NoData("journal entry not found")
}
s.logger.Warn("failed to get journal entry", zap.Error(err))
logger.Warn("failed to get journal entry", zap.Error(err))
return nil, merrors.Internal("failed to get journal entry")
}
// Get posting lines for this entry
lines, err := s.storage.PostingLines().ListByJournalEntry(ctx, entryRef)
if err != nil {
s.logger.Warn("failed to get posting lines", zap.Error(err))
logger.Warn("failed to get posting lines", zap.Error(err))
return nil, merrors.Internal("failed to get posting lines")
}
@@ -140,6 +143,7 @@ func (s *Service) getStatementResponder(_ context.Context, req *ledgerv1.GetStat
if err != nil {
return nil, err
}
logger := s.logger.With(mzap.ObjRef("ledger_account_ref", accountRef))
// Verify account exists
_, err = s.storage.Accounts().Get(ctx, accountRef)
@@ -147,7 +151,7 @@ func (s *Service) getStatementResponder(_ context.Context, req *ledgerv1.GetStat
if err == storage.ErrAccountNotFound {
return nil, merrors.NoData("account not found")
}
s.logger.Warn("failed to get account", zap.Error(err))
logger.Warn("failed to get account", zap.Error(err))
return nil, merrors.Internal("failed to get account")
}
@@ -167,11 +171,12 @@ func (s *Service) getStatementResponder(_ context.Context, req *ledgerv1.GetStat
return nil, merrors.InvalidArgument(fmt.Sprintf("invalid cursor: %v", err))
}
}
logger = logger.With(zap.Int("limit", limit), zap.Int("offset", offset))
// Get posting lines for account
postingLines, err := s.storage.PostingLines().ListByAccount(ctx, accountRef, limit+1, offset)
if err != nil {
s.logger.Warn("failed to get posting lines", zap.Error(err))
logger.Warn("failed to get posting lines", zap.Error(err))
return nil, merrors.Internal("failed to get posting lines")
}
@@ -189,18 +194,22 @@ func (s *Service) getStatementResponder(_ context.Context, req *ledgerv1.GetStat
entries := make([]*ledgerv1.JournalEntryResponse, 0)
for entryRefHex := range entryMap {
entryRef, _ := parseObjectID(entryRefHex)
entryRef, err := parseObjectID(entryRefHex)
if err != nil {
s.logger.Warn("invalid journal entry ref in posting lines", zap.String("entry_ref", entryRefHex), zap.Error(err))
return nil, err
}
entry, err := s.storage.JournalEntries().Get(ctx, entryRef)
if err != nil {
s.logger.Warn("failed to get journal entry for statement", zap.Error(err), zap.String("entryRef", entryRefHex))
logger.Warn("failed to get journal entry for statement", zap.Error(err), zap.String("entry_ref", entryRefHex))
continue
}
// Get all lines for this entry
lines, err := s.storage.PostingLines().ListByJournalEntry(ctx, entryRef)
if err != nil {
s.logger.Warn("failed to get posting lines for entry", zap.Error(err), zap.String("entryRef", entryRefHex))
logger.Warn("failed to get posting lines for entry", zap.Error(err), zap.String("entry_ref", entryRefHex))
continue
}

15
api/ledger/storage/mongo/store/accounts.go
View File

@@ -10,6 +10,7 @@ import (
ri "github.com/tech/sendico/pkg/db/repository/index"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/mutil/mzap"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.uber.org/zap"
@@ -87,14 +88,14 @@ func (a *accountsStore) Get(ctx context.Context, accountRef primitive.ObjectID)
result := &model.Account{}
if err := a.repo.Get(ctx, accountRef, result); err != nil {
if errors.Is(err, merrors.ErrNoData) {
a.logger.Debug("account not found", zap.String("accountRef", accountRef.Hex()))
a.logger.Debug("account not found", mzap.ObjRef("account_ref", accountRef))
return nil, storage.ErrAccountNotFound
}
a.logger.Warn("failed to get account", zap.Error(err), zap.String("accountRef", accountRef.Hex()))
a.logger.Warn("failed to get account", zap.Error(err), mzap.ObjRef("account_ref", accountRef))
return nil, err
}
a.logger.Debug("account loaded", zap.String("accountRef", accountRef.Hex()),
a.logger.Debug("account loaded", mzap.ObjRef("account_ref", accountRef),
zap.String("accountCode", result.AccountCode))
return result, nil
}
@@ -156,11 +157,11 @@ func (a *accountsStore) GetDefaultSettlement(ctx context.Context, orgRef primiti
if errors.Is(err, merrors.ErrNoData) {
a.logger.Debug("default settlement account not found",
zap.String("currency", currency),
zap.String("organizationRef", orgRef.Hex()))
mzap.ObjRef("organization_ref", orgRef))
return nil, storage.ErrAccountNotFound
}
a.logger.Warn("failed to get default settlement account", zap.Error(err),
zap.String("organizationRef", orgRef.Hex()),
mzap.ObjRef("organization_ref", orgRef),
zap.String("currency", currency))
return nil, err
}
@@ -210,11 +211,11 @@ func (a *accountsStore) UpdateStatus(ctx context.Context, accountRef primitive.O
patch := repository.Patch().Set(repository.Field("status"), status)
if err := a.repo.Patch(ctx, accountRef, patch); err != nil {
a.logger.Warn("failed to update account status", zap.Error(err), zap.String("accountRef", accountRef.Hex()))
a.logger.Warn("failed to update account status", zap.Error(err), mzap.ObjRef("account_ref", accountRef))
return err
}
a.logger.Debug("account status updated", zap.String("accountRef", accountRef.Hex()),
a.logger.Debug("account status updated", mzap.ObjRef("account_ref", accountRef),
zap.String("status", string(status)))
return nil
}

7
api/ledger/storage/mongo/store/balances.go
View File

@@ -10,6 +10,7 @@ import (
ri "github.com/tech/sendico/pkg/db/repository/index"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/mutil/mzap"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.uber.org/zap"
@@ -55,14 +56,14 @@ func (b *balancesStore) Get(ctx context.Context, accountRef primitive.ObjectID)
result := &model.AccountBalance{}
if err := b.repo.FindOneByFilter(ctx, query, result); err != nil {
if errors.Is(err, merrors.ErrNoData) {
b.logger.Debug("balance not found", zap.String("accountRef", accountRef.Hex()))
b.logger.Debug("balance not found", mzap.ObjRef("account_ref", accountRef))
return nil, storage.ErrBalanceNotFound
}
b.logger.Warn("failed to get balance", zap.Error(err), zap.String("accountRef", accountRef.Hex()))
b.logger.Warn("failed to get balance", zap.Error(err), mzap.ObjRef("account_ref", accountRef))
return nil, err
}
b.logger.Debug("balance loaded", zap.String("accountRef", accountRef.Hex()),
b.logger.Debug("balance loaded", mzap.ObjRef("account_ref", accountRef),
zap.String("balance", result.Balance))
return result, nil
}

19
api/ledger/storage/mongo/store/journal_entries.go
View File

@@ -10,6 +10,7 @@ import (
ri "github.com/tech/sendico/pkg/db/repository/index"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/mutil/mzap"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.uber.org/zap"
@@ -65,14 +66,14 @@ func (j *journalEntriesStore) Create(ctx context.Context, entry *model.JournalEn
if err := j.repo.Insert(ctx, entry, nil); err != nil {
if mongo.IsDuplicateKeyError(err) {
j.logger.Warn("duplicate idempotency key", zap.String("idempotencyKey", entry.IdempotencyKey))
j.logger.Warn("duplicate idempotency key", zap.String("idempotency_key", entry.IdempotencyKey))
return storage.ErrDuplicateIdempotency
}
j.logger.Warn("failed to create journal entry", zap.Error(err))
return err
}
j.logger.Debug("journal entry created", zap.String("idempotencyKey", entry.IdempotencyKey),
j.logger.Debug("journal entry created", zap.String("idempotency_key", entry.IdempotencyKey),
zap.String("entryType", string(entry.EntryType)))
return nil
}
@@ -86,15 +87,15 @@ func (j *journalEntriesStore) Get(ctx context.Context, entryRef primitive.Object
result := &model.JournalEntry{}
if err := j.repo.Get(ctx, entryRef, result); err != nil {
if errors.Is(err, merrors.ErrNoData) {
j.logger.Debug("journal entry not found", zap.String("entryRef", entryRef.Hex()))
j.logger.Debug("journal entry not found", mzap.ObjRef("entry_ref", entryRef))
return nil, storage.ErrJournalEntryNotFound
}
j.logger.Warn("failed to get journal entry", zap.Error(err), zap.String("entryRef", entryRef.Hex()))
j.logger.Warn("failed to get journal entry", zap.Error(err), mzap.ObjRef("entry_ref", entryRef))
return nil, err
}
j.logger.Debug("journal entry loaded", zap.String("entryRef", entryRef.Hex()),
zap.String("idempotencyKey", result.IdempotencyKey))
j.logger.Debug("journal entry loaded", mzap.ObjRef("entry_ref", entryRef),
zap.String("idempotency_key", result.IdempotencyKey))
return result, nil
}
@@ -115,15 +116,15 @@ func (j *journalEntriesStore) GetByIdempotencyKey(ctx context.Context, orgRef pr
result := &model.JournalEntry{}
if err := j.repo.FindOneByFilter(ctx, query, result); err != nil {
if errors.Is(err, merrors.ErrNoData) {
j.logger.Debug("journal entry not found by idempotency key", zap.String("idempotencyKey", idempotencyKey))
j.logger.Debug("journal entry not found by idempotency key", zap.String("idempotency_key", idempotencyKey))
return nil, storage.ErrJournalEntryNotFound
}
j.logger.Warn("failed to get journal entry by idempotency key", zap.Error(err),
zap.String("idempotencyKey", idempotencyKey))
zap.String("idempotency_key", idempotencyKey))
return nil, err
}
j.logger.Debug("journal entry loaded by idempotency key", zap.String("idempotencyKey", idempotencyKey))
j.logger.Debug("journal entry loaded by idempotency key", zap.String("idempotency_key", idempotencyKey))
return result, nil
}

9
api/ledger/storage/mongo/store/posting_lines.go
View File

@@ -10,6 +10,7 @@ import (
"github.com/tech/sendico/pkg/db/storable"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/mutil/mzap"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.uber.org/zap"
@@ -97,11 +98,11 @@ func (p *postingLinesStore) ListByJournalEntry(ctx context.Context, entryRef pri
return nil
})
if err != nil {
p.logger.Warn("failed to list posting lines by entry", zap.Error(err), zap.String("entryRef", entryRef.Hex()))
p.logger.Warn("failed to list posting lines by entry", zap.Error(err), mzap.ObjRef("entry_ref", entryRef))
return nil, err
}
p.logger.Debug("listed posting lines by entry", zap.Int("count", len(lines)), zap.String("entryRef", entryRef.Hex()))
p.logger.Debug("listed posting lines by entry", zap.Int("count", len(lines)), mzap.ObjRef("entry_ref", entryRef))
return lines, nil
}
@@ -129,10 +130,10 @@ func (p *postingLinesStore) ListByAccount(ctx context.Context, accountRef primit
return nil
})
if err != nil {
p.logger.Warn("failed to list posting lines by account", zap.Error(err), zap.String("accountRef", accountRef.Hex()))
p.logger.Warn("failed to list posting lines by account", zap.Error(err), mzap.ObjRef("account_ref", accountRef))
return nil, err
}
p.logger.Debug("listed posting lines by account", zap.Int("count", len(lines)), zap.String("accountRef", accountRef.Hex()))
p.logger.Debug("listed posting lines by account", zap.Int("count", len(lines)), mzap.ObjRef("account_ref", accountRef))
return lines, nil
}