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

cached gateway routing

Browse Source
This commit is contained in:
Stephan D
2026-02-20 15:38:22 +01:00
parent bc2bc3770d
commit 671ccc55a0
23 changed files with 777 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
api/pkg/db/chainwalletroutes/routes.go Normal file
View File

@@ -0,0 +1,12 @@
package chainwalletroutes
import (
"context"
"github.com/tech/sendico/pkg/model"
)
type DB interface {
Get(ctx context.Context, organizationRef string, walletRef string) (*model.ChainWalletRoute, error)
Upsert(ctx context.Context, route *model.ChainWalletRoute) error
}

2
api/pkg/db/factory.go
View File

@@ -4,6 +4,7 @@ import (
"github.com/tech/sendico/pkg/auth"
"github.com/tech/sendico/pkg/db/account"
"github.com/tech/sendico/pkg/db/chainassets"
"github.com/tech/sendico/pkg/db/chainwalletroutes"
mongoimpl "github.com/tech/sendico/pkg/db/internal/mongo"
"github.com/tech/sendico/pkg/db/invitation"
"github.com/tech/sendico/pkg/db/organization"
@@ -22,6 +23,7 @@ type Factory interface {
NewRefreshTokensDB() (refreshtokens.DB, error)
NewChainAsstesDB() (chainassets.DB, error)
NewChainWalletRoutesDB() (chainwalletroutes.DB, error)
NewAccountDB() (account.DB, error)
NewOrganizationDB() (organization.DB, error)

111
api/pkg/db/internal/mongo/chainwalletroutesdb/db.go Normal file
View File

@@ -0,0 +1,111 @@
package chainwalletroutesdb
import (
"context"
"errors"
"github.com/tech/sendico/pkg/db/repository"
ri "github.com/tech/sendico/pkg/db/repository/index"
"github.com/tech/sendico/pkg/db/template"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.uber.org/zap"
)
type ChainWalletRoutesDB struct {
template.DBImp[*model.ChainWalletRoute]
}
func Create(logger mlogger.Logger, db *mongo.Database) (*ChainWalletRoutesDB, error) {
p := &ChainWalletRoutesDB{
DBImp: *template.Create[*model.ChainWalletRoute](logger, mservice.WalletRoutes, db),
}
if err := p.Repository.CreateIndex(&ri.Definition{
Name: "idx_org_wallet_unique",
Unique: true,
Keys: []ri.Key{
{Field: "organizationRef", Sort: ri.Asc},
{Field: "walletRef", Sort: ri.Asc},
},
}); err != nil {
p.Logger.Error("Failed to create unique organization/wallet route index", zap.Error(err))
return nil, err
}
if err := p.Repository.CreateIndex(&ri.Definition{
Name: "idx_wallet_ref",
Keys: []ri.Key{
{Field: "walletRef", Sort: ri.Asc},
},
}); err != nil {
p.Logger.Error("Failed to create wallet route lookup index", zap.Error(err))
return nil, err
}
return p, nil
}
func (db *ChainWalletRoutesDB) Get(ctx context.Context, organizationRef string, walletRef string) (*model.ChainWalletRoute, error) {
org := model.ChainWalletRoute{OrganizationRef: organizationRef, WalletRef: walletRef}
org.Normalize()
if org.OrganizationRef == "" || org.WalletRef == "" {
return nil, merrors.InvalidArgument("wallet route requires organizationRef and walletRef")
}
var route model.ChainWalletRoute
query := repository.Query().
Filter(repository.Field("organizationRef"), org.OrganizationRef).
Filter(repository.Field("walletRef"), org.WalletRef)
return &route, db.FindOne(ctx, query, &route)
}
func (db *ChainWalletRoutesDB) Upsert(ctx context.Context, route *model.ChainWalletRoute) error {
if route == nil {
return merrors.InvalidArgument("wallet route is nil")
}
route.Normalize()
if route.OrganizationRef == "" || route.WalletRef == "" {
return merrors.InvalidArgument("wallet route requires organizationRef and walletRef")
}
if route.Network == "" && route.GatewayID == "" {
return merrors.InvalidArgument("wallet route requires network or gatewayId")
}
existing, err := db.Get(ctx, route.OrganizationRef, route.WalletRef)
if err != nil {
if !errors.Is(err, merrors.ErrNoData) {
return err
}
if createErr := db.Create(ctx, route); createErr != nil {
if errors.Is(createErr, merrors.ErrDataConflict) {
existing, err = db.Get(ctx, route.OrganizationRef, route.WalletRef)
if err != nil {
return err
}
} else {
return createErr
}
} else {
return nil
}
}
changed := false
if route.Network != "" && existing.Network != route.Network {
existing.Network = route.Network
changed = true
}
if route.GatewayID != "" && existing.GatewayID != route.GatewayID {
existing.GatewayID = route.GatewayID
changed = true
}
if !changed {
return nil
}
return db.Update(ctx, existing)
}

6
api/pkg/db/internal/mongo/db.go
View File

@@ -11,8 +11,10 @@ import (
"github.com/tech/sendico/pkg/auth"
"github.com/tech/sendico/pkg/db/account"
"github.com/tech/sendico/pkg/db/chainassets"
"github.com/tech/sendico/pkg/db/chainwalletroutes"
"github.com/tech/sendico/pkg/db/internal/mongo/accountdb"
"github.com/tech/sendico/pkg/db/internal/mongo/chainassetsdb"
"github.com/tech/sendico/pkg/db/internal/mongo/chainwalletroutesdb"
"github.com/tech/sendico/pkg/db/internal/mongo/invitationdb"
"github.com/tech/sendico/pkg/db/internal/mongo/organizationdb"
"github.com/tech/sendico/pkg/db/internal/mongo/policiesdb"
@@ -308,6 +310,10 @@ func (db *DB) NewChainAsstesDB() (chainassets.DB, error) {
return chainassetsdb.Create(db.logger, db.db())
}
func (db *DB) NewChainWalletRoutesDB() (chainwalletroutes.DB, error) {
return chainwalletroutesdb.Create(db.logger, db.db())
}
func (db *DB) Permissions() auth.Provider {
return db
}

46
api/pkg/messaging/internal/natsb/broker.go
View File

@@ -35,6 +35,41 @@ type envConfig struct {
const defaultConsumerBufferSize = 1024
func sanitizeNATSURL(rawURL string) string {
if rawURL == "" {
return rawURL
}
parts := strings.Split(rawURL, ",")
sanitized := make([]string, 0, len(parts))
for _, part := range parts {
trimmed := strings.TrimSpace(part)
if trimmed == "" {
continue
}
if !strings.Contains(trimmed, "://") {
sanitized = append(sanitized, trimmed)
continue
}
parsed, err := url.Parse(trimmed)
if err != nil {
sanitized = append(sanitized, trimmed)
continue
}
if parsed.User == nil {
sanitized = append(sanitized, trimmed)
continue
}
sanitized = append(sanitized, parsed.Redacted())
}
if len(sanitized) == 0 {
return strings.TrimSpace(rawURL)
}
return strings.Join(sanitized, ",")
}
// loadEnv gathers and validates connection details from environment variables
// listed in the Settings struct. Invalid or missing values surface as a typed
// InvalidArgument error so callers can decide how to handle them.
@@ -109,6 +144,7 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
}
natsURL = u.String()
}
sanitizedNATSURL := sanitizeNATSURL(natsURL)
opts := []nats.Option{
nats.Name(settings.NATSName),
@@ -120,7 +156,7 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
zap.String("broker", settings.NATSName),
}
if conn != nil {
fields = append(fields, zap.String("connected_url", conn.ConnectedUrl()))
fields = append(fields, zap.String("connected_url", sanitizeNATSURL(conn.ConnectedUrl())))
}
if err != nil {
fields = append(fields, zap.Error(err))
@@ -132,7 +168,7 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
zap.String("broker", settings.NATSName),
}
if conn != nil {
fields = append(fields, zap.String("connected_url", conn.ConnectedUrl()))
fields = append(fields, zap.String("connected_url", sanitizeNATSURL(conn.ConnectedUrl())))
}
l.Info("Reconnected to NATS", fields...)
}),
@@ -142,7 +178,7 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
}
if conn != nil {
if url := conn.ConnectedUrl(); url != "" {
fields = append(fields, zap.String("connected_url", url))
fields = append(fields, zap.String("connected_url", sanitizeNATSURL(url)))
}
if err := conn.LastError(); err != nil {
fields = append(fields, zap.Error(err))
@@ -172,7 +208,7 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
}
if res.nc, err = nats.Connect(natsURL, opts...); err != nil {
l.Error("Failed to connect to NATS", zap.String("url", natsURL), zap.Error(err))
l.Error("Failed to connect to NATS", zap.String("url", sanitizedNATSURL), zap.Error(err))
return nil, err
}
if res.js, err = res.nc.JetStream(); err != nil {
@@ -180,7 +216,7 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
}
logger.Info("Connected to NATS", zap.String("broker", settings.NATSName),
zap.String("url", natsURL))
zap.String("url", sanitizedNATSURL))
return res, nil
}

58
api/pkg/messaging/internal/natsb/broker_test.go Normal file
View File

@@ -0,0 +1,58 @@
package natsb
import (
"strings"
"testing"
)
func TestSanitizeNATSURL(t *testing.T) {
t.Parallel()
t.Run("redacts single URL credentials", func(t *testing.T) {
t.Parallel()
raw := "nats://alice:supersecret@localhost:4222"
sanitized := sanitizeNATSURL(raw)
if strings.Contains(sanitized, "supersecret") {
t.Fatalf("expected password to be redacted, got %q", sanitized)
}
if !strings.Contains(sanitized, "alice:xxxxx@") {
t.Fatalf("expected redacted URL to keep username, got %q", sanitized)
}
})
t.Run("keeps URL without credentials unchanged", func(t *testing.T) {
t.Parallel()
raw := "nats://localhost:4222"
sanitized := sanitizeNATSURL(raw)
if sanitized != raw {
t.Fatalf("expected URL without credentials to remain unchanged, got %q", sanitized)
}
})
t.Run("redacts each URL in server list", func(t *testing.T) {
t.Parallel()
raw := " nats://alice:one@localhost:4222, nats://bob:two@localhost:4223 "
sanitized := sanitizeNATSURL(raw)
if strings.Contains(sanitized, "one") || strings.Contains(sanitized, "two") {
t.Fatalf("expected passwords to be redacted, got %q", sanitized)
}
if !strings.Contains(sanitized, "alice:xxxxx@") || !strings.Contains(sanitized, "bob:xxxxx@") {
t.Fatalf("expected both URLs to be redacted, got %q", sanitized)
}
})
t.Run("returns invalid URL as-is", func(t *testing.T) {
t.Parallel()
raw := "not a url"
sanitized := sanitizeNATSURL(raw)
if sanitized != raw {
t.Fatalf("expected invalid URL to remain unchanged, got %q", sanitized)
}
})
}

29
api/pkg/model/chainwalletroute.go Normal file
View File

@@ -0,0 +1,29 @@
package model
import (
"strings"
"github.com/tech/sendico/pkg/db/storable"
"github.com/tech/sendico/pkg/mservice"
)
// ChainWalletRoute stores authoritative wallet-to-gateway routing metadata.
type ChainWalletRoute struct {
storable.Base `bson:",inline" json:",inline"`
OrganizationRef string `bson:"organizationRef" json:"organizationRef"`
WalletRef string `bson:"walletRef" json:"walletRef"`
Network string `bson:"network" json:"network"`
GatewayID string `bson:"gatewayId,omitempty" json:"gatewayId,omitempty"`
}
func (*ChainWalletRoute) Collection() string {
return mservice.WalletRoutes
}
func (r *ChainWalletRoute) Normalize() {
r.OrganizationRef = strings.TrimSpace(r.OrganizationRef)
r.WalletRef = strings.TrimSpace(r.WalletRef)
r.Network = strings.ToLower(strings.TrimSpace(r.Network))
r.GatewayID = strings.TrimSpace(r.GatewayID)
}

3
api/pkg/mservice/services.go
View File

@@ -52,12 +52,13 @@ const (
Tenants Type = "tenants" // Represents tenants managed in the system
VerificationTokens Type = "verification_tokens" //Represents verification tokens managed in the system
Wallets Type = "wallets" // Represents workflows for tasks or projects
WalletRoutes Type = "wallet_routes" // Represents authoritative chain wallet gateway routing
Workflows Type = "workflows" // Represents workflows for tasks or projects
)
func StringToSType(s string) (Type, error) {
switch Type(s) {
case Accounts, Verification, Amplitude, Site, Changes, Clients, ChainGateway, ChainWallets, ChainWalletBalances,
case Accounts, Verification, Amplitude, Site, Changes, Clients, ChainGateway, ChainWallets, WalletRoutes, ChainWalletBalances,
ChainTransfers, ChainDeposits, MntxGateway, PaymentGateway, FXOracle, FeePlans, BillingDocuments, FilterProjects, Invitations, Invoices, Logo, Ledger,
LedgerAccounts, LedgerBalances, LedgerEntries, LedgerOutbox, LedgerParties, LedgerPlines, Notifications,
Organizations, Payments, PaymentRoutes, PaymentPlanTemplates, PaymentOrchestrator, PaymentMethods, Permissions, Policies, PolicyAssignements,