bff dev upgrde #373

Merged
tech merged 1 commits from bff-372 into main 2026-01-30 16:02:58 +00:00
22 changed files with 969 additions and 185 deletions

View File

@@ -0,0 +1,101 @@
syntax = "proto3";
package billing.documents.v1;
option go_package = "github.com/tech/sendico/pkg/proto/billing/documents/v1;documentsv1";
// ---------------------------
// ENUMS
// ---------------------------
// DocumentType defines supported accounting document kinds.
enum DocumentType {
DOCUMENT_TYPE_UNSPECIFIED = 0;
// Invoice issued for the payment
DOCUMENT_TYPE_INVOICE = 1;
// Service acceptance act (common in EU/RU accounting)
DOCUMENT_TYPE_ACT = 2;
// Simple receipt confirmation
DOCUMENT_TYPE_RECEIPT = 3;
}
// ---------------------------
// SERVICE
// ---------------------------
// DocumentService provides document metadata for payment lists
// and lazy document generation on demand.
service DocumentService {
// BatchResolveDocuments is used by BFF when rendering
// a page of payments. This prevents N+1 calls by resolving
// document metadata for many payments in a single request.
rpc BatchResolveDocuments(BatchResolveDocumentsRequest)
returns (BatchResolveDocumentsResponse);
// GetDocument returns the actual PDF file.
// If the document was not generated before, the service
// generates it lazily, stores it, and returns it.
rpc GetDocument(GetDocumentRequest)
returns (GetDocumentResponse);
}
// ---------------------------
// BATCH RESOLVE (for payment tables)
// ---------------------------
// BatchResolveDocumentsRequest contains a list of payment references
// for which document availability should be resolved.
message BatchResolveDocumentsRequest {
repeated string payment_refs = 1;
}
// DocumentMeta describes document availability for a single payment.
message DocumentMeta {
// Payment reference
string payment_ref = 1;
// Document types that are applicable for this payment
// based on business rules and payment snapshot.
repeated DocumentType available_types = 2;
// Document types that were already generated and stored.
// Other available types will be generated lazily when requested.
repeated DocumentType ready_types = 3;
}
// BatchResolveDocumentsResponse returns metadata for all requested payments.
message BatchResolveDocumentsResponse {
repeated DocumentMeta items = 1;
}
// ---------------------------
// GET DOCUMENT (lazy generation)
// ---------------------------
// GetDocumentRequest requests a specific document for a payment.
message GetDocumentRequest {
string payment_ref = 1;
// Type of document to retrieve (invoice, act, receipt, etc.)
DocumentType type = 2;
}
// GetDocumentResponse returns the generated PDF content.
message GetDocumentResponse {
// Raw PDF bytes
bytes content = 1;
// Suggested filename for download (e.g. invoice_123.pdf)
string filename = 2;
// MIME type, typically "application/pdf"
string mime_type = 3;
}

View File

@@ -0,0 +1,19 @@
syntax = "proto3";
package common.account_role.v1;
option go_package = "github.com/tech/sendico/pkg/proto/common/account_role/v1;accountrolev1";
enum AccountRole {
ACCOUNT_ROLE_UNSPECIFIED = 0;
OPERATING = 1;
HOLD = 2;
TRANSIT = 3;
SETTLEMENT = 4;
CLEARING = 5;
PENDING = 6;
RESERVE = 7;
LIQUIDITY = 8;
FEE = 9;
CHARGEBACK = 10;
ADJUSTMENT = 11;
}

View File

@@ -47,6 +47,7 @@ enum RailOperation {
RAIL_OPERATION_FX_CONVERT = 6;
RAIL_OPERATION_BLOCK = 7;
RAIL_OPERATION_RELEASE = 8;
RAIL_OPERATION_MOVE = 9;
}
// Limits in minor units, e.g. cents

View File

@@ -6,6 +6,8 @@ option go_package = "github.com/tech/sendico/pkg/proto/connector/v1;connectorv1"
import "google/protobuf/struct.proto";
import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";
import "common/account_role/v1/account_role.proto";
import "common/describable/v1/describable.proto";
import "common/money/v1/money.proto";
import "common/pagination/v1/cursor.proto";
@@ -18,6 +20,7 @@ service ConnectorService {
rpc GetAccount(GetAccountRequest) returns (GetAccountResponse);
rpc ListAccounts(ListAccountsRequest) returns (ListAccountsResponse);
rpc GetBalance(GetBalanceRequest) returns (GetBalanceResponse);
rpc UpdateAccountState(UpdateAccountStateRequest) returns (UpdateAccountStateResponse);
rpc SubmitOperation(SubmitOperationRequest) returns (SubmitOperationResponse);
rpc GetOperation(GetOperationRequest) returns (GetOperationResponse);
@@ -133,6 +136,7 @@ message Account {
google.protobuf.Timestamp created_at = 8;
google.protobuf.Timestamp updated_at = 9;
common.describable.v1.Describable describable = 10;
common.account_role.v1.AccountRole role = 11; // functional role within the organization (ledger-only; unset for non-ledger connectors)
}
message Balance {
@@ -167,6 +171,8 @@ message Operation {
string provider_ref = 11;
google.protobuf.Timestamp created_at = 12;
google.protobuf.Timestamp updated_at = 13;
common.account_role.v1.AccountRole from_role = 14;
common.account_role.v1.AccountRole to_role = 15;
}
message OperationReceipt {
@@ -192,6 +198,7 @@ message OpenAccountRequest {
google.protobuf.Struct params = 6;
string correlation_id = 7;
string parent_intent_id = 8;
common.account_role.v1.AccountRole role = 9; // functional role (ledger-only; ignored by non-ledger connectors)
}
message OpenAccountResponse {
@@ -208,11 +215,17 @@ message GetAccountResponse {
}
message ListAccountsRequest {
string owner_ref = 1;
reserved 1;
reserved "owner_ref";
AccountKind kind = 2;
string asset = 3; // canonical asset string (USD, ETH, USDT-TRC20)
common.pagination.v1.CursorPageRequest page = 4;
string organization_ref = 5; // optional org scope (preferred over owner_ref)
string organization_ref = 5;
// Optional owner filter with 3-state semantics:
// - not set: return all accounts within organization
// - set to empty string: return accounts where owner_ref is null/empty
// - set to a value: return accounts where owner_ref matches
google.protobuf.StringValue owner_ref_filter = 6;
}
message ListAccountsResponse {
@@ -220,6 +233,17 @@ message ListAccountsResponse {
common.pagination.v1.CursorPageResponse page = 2;
}
message UpdateAccountStateRequest {
AccountRef account_ref = 1;
AccountState target_state = 2;
common.account_role.v1.AccountRole source_role = 3; // optional: assert account has this role before mutation
}
message UpdateAccountStateResponse {
Account account = 1;
ConnectorError error = 2;
}
message GetBalanceRequest {
AccountRef account_ref = 1;
}

View File

@@ -5,6 +5,7 @@ package chain.gateway.v1;
option go_package = "github.com/tech/sendico/pkg/proto/gateway/chain/v1;chainv1";
import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";
import "common/money/v1/money.proto";
import "common/pagination/v1/cursor.proto";
import "common/describable/v1/describable.proto";
@@ -85,9 +86,15 @@ message GetManagedWalletResponse {
message ListManagedWalletsRequest {
string organization_ref = 1;
string owner_ref = 2;
reserved 2;
reserved "owner_ref";
Asset asset = 3;
common.pagination.v1.CursorPageRequest page = 4;
// Optional owner filter with 3-state semantics:
// - not set: return all wallets within organization
// - set to empty string: return wallets where owner_ref is null/empty
// - set to a value: return wallets where owner_ref matches
google.protobuf.StringValue owner_ref_filter = 5;
}
message ListManagedWalletsResponse {

View File

@@ -5,6 +5,7 @@ package ledger.v1;
option go_package = "github.com/tech/sendico/pkg/proto/ledger/v1;ledgerv1";
import "google/protobuf/timestamp.proto";
import "google/protobuf/wrappers.proto";
import "common/describable/v1/describable.proto";
import "common/money/v1/money.proto";
@@ -43,6 +44,21 @@ enum AccountStatus {
ACCOUNT_STATUS_FROZEN = 2;
}
enum AccountRole {
ACCOUNT_ROLE_UNSPECIFIED = 0;
ACCOUNT_ROLE_OPERATING = 1;
ACCOUNT_ROLE_HOLD = 2;
ACCOUNT_ROLE_TRANSIT = 3;
ACCOUNT_ROLE_SETTLEMENT = 4;
ACCOUNT_ROLE_CLEARING = 5;
ACCOUNT_ROLE_PENDING = 6;
ACCOUNT_ROLE_RESERVE = 7;
ACCOUNT_ROLE_LIQUIDITY = 8;
ACCOUNT_ROLE_FEE = 9;
ACCOUNT_ROLE_CHARGEBACK = 10;
ACCOUNT_ROLE_ADJUSTMENT = 11;
}
// LedgerAccount captures the canonical representation of an account resource.
message LedgerAccount {
string ledger_account_ref = 1;
@@ -52,12 +68,14 @@ message LedgerAccount {
string currency = 5;
AccountStatus status = 6;
bool allow_negative = 7;
bool is_settlement = 8;
reserved 8;
reserved "is_settlement";
map<string, string> metadata = 9;
google.protobuf.Timestamp created_at = 10;
google.protobuf.Timestamp updated_at = 11;
common.describable.v1.Describable describable = 12;
string owner_ref = 13;
AccountRole role = 14;
}
// A single posting line (mirrors your PostingLine model)
@@ -78,9 +96,11 @@ message CreateAccountRequest {
string currency = 5;
AccountStatus status = 6;
bool allow_negative = 7;
bool is_settlement = 8;
reserved 8;
reserved "is_settlement";
map<string, string> metadata = 9;
common.describable.v1.Describable describable = 10;
AccountRole role = 11;
}
message CreateAccountResponse {
@@ -98,6 +118,7 @@ message PostCreditRequest {
map<string, string> metadata = 7;
google.protobuf.Timestamp event_time = 8;
string contra_ledger_account_ref = 9; // optional override for settlement/contra account
AccountRole role = 10; // optional: assert target account has this role
}
message PostDebitRequest {
@@ -110,6 +131,7 @@ message PostDebitRequest {
map<string, string> metadata = 7;
google.protobuf.Timestamp event_time = 8;
string contra_ledger_account_ref = 9; // optional override for settlement/contra account
AccountRole role = 10; // optional: assert target account has this role
}
message TransferRequest {
@@ -122,6 +144,8 @@ message TransferRequest {
repeated PostingLine charges = 7; // optional FEE/SPREAD lines
map<string, string> metadata = 8;
google.protobuf.Timestamp event_time = 9;
AccountRole from_role = 10;
AccountRole to_role = 11;
}
message FXRequest {
@@ -188,8 +212,35 @@ message StatementResponse {
message ListAccountsRequest {
string organization_ref = 1;
// Optional owner filter with 3-state semantics:
// - not set: return all accounts within organization
// - set to empty string: return accounts where owner_ref is null/empty
// - set to a value: return accounts where owner_ref matches
google.protobuf.StringValue owner_ref_filter = 2;
}
message ListAccountsResponse {
repeated LedgerAccount accounts = 1;
}
// ---- Account status mutations ----
message BlockAccountRequest {
string ledger_account_ref = 1;
string organization_ref = 2;
AccountRole role = 3; // optional: assert account has this role before blocking
}
message BlockAccountResponse {
LedgerAccount account = 1;
}
message UnblockAccountRequest {
string ledger_account_ref = 1;
string organization_ref = 2;
AccountRole role = 3; // optional: assert account has this role before unblocking
}
message UnblockAccountResponse {
LedgerAccount account = 1;
}

View File

@@ -1,57 +1,46 @@
# Config file for [Air](https://github.com/air-verse/air) in TOML format
# Working directory
# . or absolute path, please note that the directories following must be under root.
root = "./.."
root = "."
testdata_dir = "testdata"
tmp_dir = "tmp"
[build]
# Just plain old shell command. You could use `make` as well.
cmd = "go build -o app -ldflags \"-X 'github.com/tech/sendico/server/internal/appversion.BuildUser=$(whoami)' -X 'github.com/tech/sendico/server/internal/appversion.Version=$APP_V' -X 'github.com/tech/sendico/server/internal/appversion.Branch=$BUILD_BRANCH' -X 'github.com/tech/sendico/server/internal/appversion.Revision=$GIT_REV' -X 'github.com/tech/sendico/server/internal/appversion.BuildDate=$(date)' -X 'github.com/tech/sendico/server/internal/mutil/ampli.Version=$APP_V'\""
# Binary file yields from `cmd`.
bin = "./app"
# Customize binary, can setup environment variables when run your app.
full_bin = "./app --debug"
# Watch these filename extensions.
include_ext = ["go"]
# Ignore these filename extensions or directories.
exclude_dir = ["server/.git", "pkg/.git", "server/tmp", "server/storage", "server/resources", "server/env"]
# Watch these directories if you specified.
include_dir = []
# Watch these files.
include_file = []
# Exclude files.
exclude_file = []
# Exclude specific regular expressions.
exclude_regex = ["_test\\.go"]
# Exclude unchanged files.
exclude_unchanged = true
# Follow symlink for directories
follow_symlink = true
# This log file places in your tmp_dir.
log = "air.log"
# It's not necessary to trigger build each time file changes if it's too frequent.
delay = 0 # ms
# Stop running old binary when build errors occur.
stop_on_error = true
# Send Interrupt signal before killing process (windows does not support this feature)
send_interrupt = true
# Delay after sending Interrupt signal
kill_delay = 500 # ms
# Add additional arguments when running binary (bin/full_bin). Will run './tmp/main hello world'.
args_bin = []
[log]
# Show log time
time = false
args_bin = []
entrypoint = "./tmp/main"
cmd = "go build -o ./tmp/main ."
delay = 1000
exclude_dir = ["assets", "tmp", "vendor", "testdata"]
exclude_file = []
exclude_regex = ["_test.go", "_templ.go"]
exclude_unchanged = false
follow_symlink = false
full_bin = ""
include_dir = []
include_ext = ["go", "tpl", "tmpl", "html"]
include_file = []
kill_delay = "0s"
log = "build-errors.log"
poll = false
poll_interval = 0
post_cmd = []
pre_cmd = []
rerun = false
rerun_delay = 500
send_interrupt = false
stop_on_error = false
[color]
# Customize each part's color. If no color found, use the raw app log.
main = "magenta"
watcher = "cyan"
build = "yellow"
runner = "green"
app = ""
build = "yellow"
main = "magenta"
runner = "green"
watcher = "cyan"
[log]
main_only = false
time = false
[misc]
# Delete tmp directory on exit
clean_on_exit = true
clean_on_exit = false
[screen]
clear_on_rebuild = false
keep_scroll = true

View File

@@ -2,3 +2,4 @@
/server
/storage
.gocache
tmp

Binary file not shown.

Before

Width:  |  Height:  |  Size: 13 KiB

After

Width:  |  Height:  |  Size: 2.2 KiB

121
api/server/config.dev.yml Executable file
View File

@@ -0,0 +1,121 @@
http_server:
listen_address: :8080
read_header_timeout: 60
shutdown_timeout: 5
api:
amplitude:
ampli_environment_env: AMPLI_ENVIRONMENT
middleware:
api_protocol_env: API_PROTOCOL
domain_env: SERVICE_HOST
api_endpoint_env: API_ENDPOINT
signature:
secret_key_env: API_ENDPOINT_SECRET
algorithm: HS256
CORS:
max_age: 300
allowed_origins:
- "*"
allowed_methods:
- "GET"
- "POST"
- "PUT"
- "PATCH"
- "DELETE"
- "OPTIONS"
allowed_headers:
- "Accept"
- "Authorization"
- "Content-Type"
- "X-Requested-With"
exposed_headers:
allow_credentials: false
websocket:
endpoint_env: WS_ENDPOINT
timeout: 60
message_broker:
driver: NATS
settings:
url_env: NATS_URL
host_env: NATS_HOST
port_env: NATS_PORT
username_env: NATS_USER
password_env: NATS_PASSWORD
broker_name: Sendico Backend server
max_reconnects: 10
reconnect_wait: 5
buffer_size: 1024
# type: in-process
# settings:
# buffer_size: 10
token:
expiration_hours:
account: 24
refresh: 720
length: 32
password:
token_length: 32
check:
min_length: 8
digit: true
upper: true
lower: true
special: true
storage:
# driver: aws_s3
# settings:
# access_key_id_env: S3_ACCESS_KEY_ID
# secret_access_key_env: S3_ACCESS_KEY_SECRET
# region_env: S3_REGION
# bucket_name_env: S3_BUCKET_NAME
driver: local_fs
settings:
root_path: ./storage
chain_gateway:
address: dev-chain-gateway:50070
address_env: CHAIN_GATEWAY_ADDRESS
dial_timeout_seconds: 5
call_timeout_seconds: 5
insecure: true
default_asset:
chain: TRON_NILE
token_symbol: USDT
contract_address: ""
ledger:
address: dev-ledger:50052
address_env: LEDGER_ADDRESS
dial_timeout_seconds: 5
call_timeout_seconds: 5
insecure: true
payment_orchestrator:
address: dev-payments-orchestrator:50062
address_env: PAYMENTS_ADDRESS
dial_timeout_seconds: 5
call_timeout_seconds: 5
insecure: true
app:
database:
driver: mongodb
settings:
host_env: MONGO_HOST
port_env: MONGO_PORT
database_env: MONGO_DATABASE
user_env: MONGO_USER
password_env: MONGO_PASSWORD
auth_source_env: MONGO_AUTH_SOURCE
replica_set_env: MONGO_REPLICA_SET
enforcer:
driver: native
settings:
model_path_env: PERMISSION_MODEL
adapter:
collection_name_env: PERMISSION_COLLECTION
database_name_env: MONGO_DATABASE
timeout_seconds_env: PERMISSION_TIMEOUT
is_filtered_env: PERMISSION_IS_FILTERED

View File

@@ -1,6 +1,6 @@
module github.com/tech/sendico/server
go 1.25.3
go 1.25.6
replace github.com/tech/sendico/pkg => ../pkg
@@ -14,7 +14,7 @@ require (
github.com/aws/aws-sdk-go-v2 v1.41.1
github.com/aws/aws-sdk-go-v2/config v1.32.7
github.com/aws/aws-sdk-go-v2/credentials v1.19.7
github.com/aws/aws-sdk-go-v2/service/s3 v1.95.1
github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0
github.com/go-chi/chi/v5 v5.2.4
github.com/go-chi/cors v1.2.2
github.com/go-chi/jwtauth/v5 v5.3.3
@@ -32,6 +32,7 @@ require (
go.mongodb.org/mongo-driver v1.17.7
go.uber.org/zap v1.27.1
golang.org/x/net v0.49.0
google.golang.org/grpc v1.78.0
google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v3 v3.0.1
moul.io/chizap v1.0.3
@@ -139,6 +140,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-20260122232226-8e98ce8d340d // indirect
google.golang.org/grpc v1.78.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 // indirect
)

View File

@@ -32,8 +32,8 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17 h1:RuNSMooz
github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.17/go.mod h1:F2xxQ9TZz5gDWsclCtPQscGpP0VUOc8RqgFM3vDENmU=
github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17 h1:bGeHBsGZx0Dvu/eJC0Lh9adJa3M1xREcndxLNZlve2U=
github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.17/go.mod h1:dcW24lbU0CzHusTE8LLHhRLI42ejmINN8Lcr22bwh/g=
github.com/aws/aws-sdk-go-v2/service/s3 v1.95.1 h1:C2dUPSnEpy4voWFIq3JNd8gN0Y5vYGDo44eUE58a/p8=
github.com/aws/aws-sdk-go-v2/service/s3 v1.95.1/go.mod h1:5jggDlZ2CLQhwJBiZJb4vfk4f0GxWdEDruWKEJ1xOdo=
github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0 h1:oeu8VPlOre74lBA/PMhxa5vewaMIMmILM+RraSyB8KA=
github.com/aws/aws-sdk-go-v2/service/s3 v1.96.0/go.mod h1:5jggDlZ2CLQhwJBiZJb4vfk4f0GxWdEDruWKEJ1xOdo=
github.com/aws/aws-sdk-go-v2/service/signin v1.0.5 h1:VrhDvQib/i0lxvr3zqlUwLwJP4fpmpyD9wYG1vfSu+Y=
github.com/aws/aws-sdk-go-v2/service/signin v1.0.5/go.mod h1:k029+U8SY30/3/ras4G/Fnv/b88N4mAfliNn08Dem4M=
github.com/aws/aws-sdk-go-v2/service/sso v1.30.9 h1:v6EiMvhEYBoHABfbGB4alOYmCIrcgyPPiBE1wZAEbqk=
@@ -361,8 +361,8 @@ 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/api v0.0.0-20251029180050-ab9386a59fda h1:+2XxjfsAu6vqFxwGBRcHiMaDCuZiqXGDUDVWVtrFAnE=
google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda/go.mod h1:fDMmzKV90WSg1NbozdqrE64fkuTv6mlq2zxo9ad+3yo=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260122232226-8e98ce8d340d h1:xXzuihhT3gL/ntduUZwHECzAn57E8dA6l8SOtYWdD8Q=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260122232226-8e98ce8d340d/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 h1:H86B94AW+VfJWDqFeEbBPhEtHzJwJfTbgE2lZa54ZAQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409/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=

View File

@@ -3,6 +3,7 @@ package srequest
import (
"strings"
"github.com/tech/sendico/pkg/ledgerconv"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/model"
"go.mongodb.org/mongo-driver/bson/primitive"
@@ -30,7 +31,7 @@ type CreateLedgerAccount struct {
AccountType LedgerAccountType `json:"accountType"`
Currency string `json:"currency"`
AllowNegative bool `json:"allowNegative,omitempty"`
IsSettlement bool `json:"isSettlement,omitempty"`
Role model.AccountRole `json:"role"`
Describable model.Describable `json:"describable"`
OwnerRef *primitive.ObjectID `json:"ownerRef,omitempty"`
Metadata map[string]string `json:"metadata,omitempty"`
@@ -43,5 +44,10 @@ func (r *CreateLedgerAccount) Validate() error {
if strings.TrimSpace(string(r.AccountType)) == "" || strings.EqualFold(string(r.AccountType), string(LedgerAccountTypeUnspecified)) {
return merrors.InvalidArgument("accountType is required", "accountType")
}
if role := strings.TrimSpace(string(r.Role)); role != "" {
if _, ok := ledgerconv.ParseAccountRole(role); !ok || ledgerconv.IsAccountRoleUnspecified(role) {
return merrors.InvalidArgument("role is invalid", "role")
}
}
return nil
}

View File

@@ -21,7 +21,7 @@ type ledgerAccount struct {
Currency string `json:"currency"`
Status string `json:"status"`
AllowNegative bool `json:"allowNegative"`
IsSettlement bool `json:"isSettlement"`
Role string `json:"role"`
Metadata map[string]string `json:"metadata,omitempty"`
CreatedAt time.Time `json:"createdAt,omitempty"`
UpdatedAt time.Time `json:"updatedAt,omitempty"`
@@ -96,7 +96,7 @@ func toLedgerAccount(acc *ledgerv1.LedgerAccount) ledgerAccount {
Currency: acc.GetCurrency(),
Status: acc.GetStatus().String(),
AllowNegative: acc.GetAllowNegative(),
IsSettlement: acc.GetIsSettlement(),
Role: acc.GetRole().String(),
Metadata: acc.GetMetadata(),
CreatedAt: acc.GetCreatedAt().AsTime(),
UpdatedAt: acc.GetUpdatedAt().AsTime(),

View File

@@ -1,6 +1,7 @@
package sresponse
import (
"fmt"
"net/http"
"strings"
"time"
@@ -8,7 +9,9 @@ import (
"github.com/tech/sendico/pkg/api/http/response"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/model"
moneyv1 "github.com/tech/sendico/pkg/proto/common/money/v1"
paginationv1 "github.com/tech/sendico/pkg/proto/common/pagination/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
"google.golang.org/protobuf/types/known/timestamppb"
)
@@ -153,3 +156,137 @@ func chainNetworkValue(chain chainv1.ChainNetwork) string {
}
return strings.ToLower(trimmed)
}
// WalletsFromAccounts converts connector accounts to wallet response format.
// Used when querying multiple gateways via discovery.
func WalletsFromAccounts(logger mlogger.Logger, accounts []*connectorv1.Account, accessToken *TokenData) http.HandlerFunc {
dto := walletsResponse{
authResponse: authResponse{AccessToken: *accessToken},
}
dto.Wallets = make([]wallet, 0, len(accounts))
for _, acc := range accounts {
if acc == nil {
continue
}
dto.Wallets = append(dto.Wallets, accountToWallet(acc))
}
return response.Ok(logger, dto)
}
func accountToWallet(acc *connectorv1.Account) wallet {
if acc == nil {
return wallet{}
}
// Extract wallet details from provider details
details := map[string]interface{}{}
if acc.GetProviderDetails() != nil {
details = acc.GetProviderDetails().AsMap()
}
walletRef := ""
if ref := acc.GetRef(); ref != nil {
walletRef = strings.TrimSpace(ref.GetAccountId())
}
if v := stringFromDetails(details, "wallet_ref"); v != "" {
walletRef = v
}
organizationRef := stringFromDetails(details, "organization_ref")
ownerRef := strings.TrimSpace(acc.GetOwnerRef())
if v := stringFromDetails(details, "owner_ref"); v != "" {
ownerRef = v
}
chain := stringFromDetails(details, "network")
tokenSymbol := stringFromDetails(details, "token_symbol")
contractAddress := stringFromDetails(details, "contract_address")
depositAddress := stringFromDetails(details, "deposit_address")
name := ""
if d := acc.GetDescribable(); d != nil {
name = strings.TrimSpace(d.GetName())
}
if name == "" {
name = strings.TrimSpace(acc.GetLabel())
}
if name == "" {
name = walletRef
}
var description *string
if d := acc.GetDescribable(); d != nil && d.Description != nil {
if trimmed := strings.TrimSpace(d.GetDescription()); trimmed != "" {
description = &trimmed
}
}
status := acc.GetState().String()
// Convert connector state to wallet status format
switch acc.GetState() {
case connectorv1.AccountState_ACCOUNT_ACTIVE:
status = "MANAGED_WALLET_ACTIVE"
case connectorv1.AccountState_ACCOUNT_SUSPENDED:
status = "MANAGED_WALLET_SUSPENDED"
case connectorv1.AccountState_ACCOUNT_CLOSED:
status = "MANAGED_WALLET_CLOSED"
}
return wallet{
WalletRef: walletRef,
OrganizationRef: organizationRef,
OwnerRef: ownerRef,
Asset: walletAsset{
Chain: chain,
TokenSymbol: tokenSymbol,
ContractAddress: contractAddress,
},
DepositAddress: depositAddress,
Status: status,
Name: name,
Description: description,
CreatedAt: tsToString(acc.GetCreatedAt()),
UpdatedAt: tsToString(acc.GetUpdatedAt()),
}
}
func stringFromDetails(details map[string]interface{}, key string) string {
if details == nil {
return ""
}
if value, ok := details[key]; ok {
return strings.TrimSpace(fmt.Sprint(value))
}
return ""
}
// WalletBalanceFromConnector converts connector balance to wallet balance response format.
// Used when querying gateways via discovery.
func WalletBalanceFromConnector(logger mlogger.Logger, bal *connectorv1.Balance, accessToken *TokenData) http.HandlerFunc {
return response.Ok(logger, walletBalanceResponse{
Balance: connectorBalanceToWalletBalance(bal),
authResponse: authResponse{AccessToken: *accessToken},
})
}
func connectorBalanceToWalletBalance(b *connectorv1.Balance) walletBalance {
if b == nil {
return walletBalance{}
}
return walletBalance{
Available: connectorMoneyToModel(b.GetAvailable()),
PendingInbound: connectorMoneyToModel(b.GetPendingInbound()),
PendingOutbound: connectorMoneyToModel(b.GetPendingOutbound()),
CalculatedAt: tsToString(b.GetCalculatedAt()),
}
}
func connectorMoneyToModel(m *moneyv1.Money) *model.Money {
if m == nil {
return nil
}
return &model.Money{
Amount: m.GetAmount(),
Currency: m.GetCurrency(),
}
}

View File

@@ -10,18 +10,18 @@ import (
func Network2Proto(network model.ChainNetwork) (chainv1.ChainNetwork, error) {
switch network {
case model.ChainNetworkARB:
case model.ChainNetworkArbitrumOne:
return chainv1.ChainNetwork_CHAIN_NETWORK_ARBITRUM_ONE, nil
case model.ChainNetworkEthMain:
case model.ChainNetworkEthereumMainnet:
return chainv1.ChainNetwork_CHAIN_NETWORK_ETHEREUM_MAINNET, nil
case model.ChainNetworkTronMain:
case model.ChainNetworkTronMainnet:
return chainv1.ChainNetwork_CHAIN_NETWORK_TRON_MAINNET, nil
case model.ChainNetworkTronNile:
return chainv1.ChainNetwork_CHAIN_NETWORK_TRON_NILE, nil
case model.ChainNetworkUnspecified:
return chainv1.ChainNetwork_CHAIN_NETWORK_UNSPECIFIED, nil
default:
return chainv1.ChainNetwork_CHAIN_NETWORK_UNSPECIFIED, merrors.InvalidArgument(fmt.Sprintf("Unkwnown chain network value '%s'", network), "network")
return chainv1.ChainNetwork_CHAIN_NETWORK_UNSPECIFIED, merrors.InvalidArgument(fmt.Sprintf("unknown chain network value '%s'", network), "network")
}
}

View File

@@ -47,6 +47,10 @@ func (a *LedgerAPI) createAccount(r *http.Request, account *model.Account, token
if err != nil {
return response.BadPayload(a.logger, a.Name(), err)
}
accountRole, err := mapLedgerAccountRole(payload.Role)
if err != nil {
return response.BadPayload(a.logger, a.Name(), err)
}
if a.client == nil {
return response.Internal(a.logger, mservice.Ledger, merrors.Internal("ledger client is not configured"))
}
@@ -78,7 +82,7 @@ func (a *LedgerAPI) createAccount(r *http.Request, account *model.Account, token
Currency: payload.Currency,
Status: ledgerv1.AccountStatus_ACCOUNT_STATUS_ACTIVE,
AllowNegative: payload.AllowNegative,
IsSettlement: payload.IsSettlement,
Role: accountRole,
Metadata: payload.Metadata,
Describable: describable,
})
@@ -128,14 +132,14 @@ func mapLedgerAccountType(accountType srequest.LedgerAccountType) (ledgerv1.Acco
return parsed, nil
}
func mapLedgerAccountStatus(status srequest.LedgerAccountStatus) (ledgerv1.AccountStatus, error) {
raw := string(status)
if ledgerconv.IsAccountStatusUnspecified(raw) {
return ledgerv1.AccountStatus_ACCOUNT_STATUS_UNSPECIFIED, nil
func mapLedgerAccountRole(role model.AccountRole) (ledgerv1.AccountRole, error) {
raw := strings.TrimSpace(string(role))
if ledgerconv.IsAccountRoleUnspecified(raw) {
return ledgerv1.AccountRole_ACCOUNT_ROLE_OPERATING, nil
}
parsed, ok := ledgerconv.ParseAccountStatus(raw)
parsed, ok := ledgerconv.ParseAccountRole(raw)
if !ok {
return ledgerv1.AccountStatus_ACCOUNT_STATUS_UNSPECIFIED, merrors.InvalidArgument("unsupported status: "+string(status), "status")
return ledgerv1.AccountRole_ACCOUNT_ROLE_UNSPECIFIED, merrors.InvalidArgument("unsupported role: "+raw, "role")
}
return parsed, nil
}

View File

@@ -7,11 +7,13 @@ import (
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
"github.com/tech/sendico/pkg/mutil/mzap"
ledgerv1 "github.com/tech/sendico/pkg/proto/ledger/v1"
"github.com/tech/sendico/server/interface/api/sresponse"
mutil "github.com/tech/sendico/server/internal/mutil/param"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
"google.golang.org/protobuf/types/known/wrapperspb"
)
func (a *LedgerAPI) listAccounts(r *http.Request, account *model.Account, token *sresponse.TokenData) http.HandlerFunc {
@@ -22,22 +24,28 @@ func (a *LedgerAPI) listAccounts(r *http.Request, account *model.Account, token
}
ctx := r.Context()
res, err := a.enf.Enforce(ctx, a.permissionRef, account.ID, orgRef, primitive.NilObjectID, model.ActionRead)
hasReadPermission, err := a.enf.Enforce(ctx, a.permissionRef, account.ID, orgRef, primitive.NilObjectID, model.ActionRead)
if err != nil {
a.logger.Warn("Failed to check ledger accounts access permissions", zap.Error(err), mutil.PLog(a.oph, r))
return response.Auto(a.logger, a.Name(), err)
}
if !res {
a.logger.Debug("Access denied when listing ledger accounts", mutil.PLog(a.oph, r))
return response.AccessDenied(a.logger, a.Name(), "ledger accounts read permission denied")
}
if a.client == nil {
return response.Internal(a.logger, mservice.Ledger, merrors.Internal("ledger client is not configured"))
}
resp, err := a.client.ListAccounts(ctx, &ledgerv1.ListAccountsRequest{
req := &ledgerv1.ListAccountsRequest{
OrganizationRef: orgRef.Hex(),
})
}
// If user has read permission, return all accounts in organization.
// Otherwise, filter to only accounts owned by the requesting account.
if !hasReadPermission {
req.OwnerRefFilter = wrapperspb.String(account.ID.Hex())
a.logger.Debug("Filtering ledger accounts by owner due to limited permissions",
mzap.ObjRef("owner_ref", account.ID), mutil.PLog(a.oph, r))
}
resp, err := a.client.ListAccounts(ctx, req)
if err != nil {
a.logger.Warn("Failed to list ledger accounts", zap.Error(err), zap.String("organization_ref", orgRef.Hex()))
return response.Auto(a.logger, mservice.Ledger, err)

View File

@@ -1,18 +1,25 @@
package walletapiimp
import (
"context"
"crypto/tls"
"net/http"
"strings"
"sync"
"github.com/tech/sendico/pkg/api/http/response"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
"github.com/tech/sendico/server/interface/api/sresponse"
mutil "github.com/tech/sendico/server/internal/mutil/param"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
)
func (a *WalletAPI) getWalletBalance(r *http.Request, account *model.Account, token *sresponse.TokenData) http.HandlerFunc {
@@ -36,21 +43,126 @@ func (a *WalletAPI) getWalletBalance(r *http.Request, account *model.Account, to
a.logger.Debug("Access denied when reading wallet balance", mutil.PLog(a.oph, r), zap.String("wallet_ref", walletRef))
return response.AccessDenied(a.logger, a.Name(), "wallet balance read permission denied")
}
if a.chainGateway == nil {
return response.Internal(a.logger, mservice.ChainGateway, merrors.Internal("chain gateway client is not configured"))
if a.discovery == nil {
return response.Internal(a.logger, mservice.ChainGateway, merrors.Internal("discovery client is not configured"))
}
resp, err := a.chainGateway.GetWalletBalance(ctx, &chainv1.GetWalletBalanceRequest{WalletRef: walletRef})
// Discover CRYPTO rail gateways
lookupCtx, cancel := context.WithTimeout(ctx, discoveryLookupTimeout)
defer cancel()
lookupResp, err := a.discovery.Lookup(lookupCtx)
if err != nil {
a.logger.Warn("Failed to fetch wallet balance", zap.Error(err), zap.String("wallet_ref", walletRef))
a.logger.Warn("Failed to lookup discovery registry", zap.Error(err))
return response.Auto(a.logger, a.Name(), err)
}
// Filter gateways by CRYPTO rail
cryptoGateways := filterCryptoGateways(lookupResp.Gateways)
if len(cryptoGateways) == 0 {
a.logger.Debug("No CRYPTO rail gateways found in discovery")
return response.Auto(a.logger, a.Name(), merrors.NoData("no crypto gateways available"))
}
// Query all gateways in parallel to find the wallet balance
bal, err := a.queryBalanceFromGateways(ctx, cryptoGateways, walletRef)
if err != nil {
a.logger.Warn("Failed to fetch wallet balance from gateways", zap.Error(err), zap.String("wallet_ref", walletRef))
return response.Auto(a.logger, mservice.ChainGateway, err)
}
bal := resp.GetBalance()
if bal == nil {
a.logger.Warn("Wallet balance missing in response", zap.String("wallet_ref", walletRef))
return response.Auto(a.logger, mservice.ChainGateway, merrors.Internal("wallet balance not available"))
a.logger.Warn("Wallet balance not found on any gateway", zap.String("wallet_ref", walletRef))
return response.Auto(a.logger, mservice.ChainGateway, merrors.NoData("wallet not found"))
}
return sresponse.WalletBalance(a.logger, bal, token)
return sresponse.WalletBalanceFromConnector(a.logger, bal, token)
}
func (a *WalletAPI) queryBalanceFromGateways(ctx context.Context, gateways []discovery.GatewaySummary, walletRef string) (*connectorv1.Balance, error) {
var mu sync.Mutex
var wg sync.WaitGroup
var result *connectorv1.Balance
var lastErr error
for _, gw := range gateways {
wg.Add(1)
go func(gateway discovery.GatewaySummary) {
defer wg.Done()
bal, err := a.queryGatewayBalance(ctx, gateway, walletRef)
if err != nil {
a.logger.Debug("Failed to query gateway for balance",
zap.String("gateway_id", gateway.ID),
zap.String("invoke_uri", gateway.InvokeURI),
zap.String("wallet_ref", walletRef),
zap.Error(err))
mu.Lock()
lastErr = err
mu.Unlock()
return
}
if bal != nil {
mu.Lock()
if result == nil {
result = bal
a.logger.Debug("Found wallet balance on gateway",
zap.String("gateway_id", gateway.ID),
zap.String("network", gateway.Network),
zap.String("wallet_ref", walletRef))
}
mu.Unlock()
}
}(gw)
}
wg.Wait()
if result != nil {
return result, nil
}
if lastErr != nil {
return nil, lastErr
}
return nil, nil
}
func (a *WalletAPI) queryGatewayBalance(ctx context.Context, gateway discovery.GatewaySummary, walletRef string) (*connectorv1.Balance, error) {
// Create connection with timeout
dialCtx, cancel := context.WithTimeout(ctx, a.dialTimeout)
defer cancel()
var dialOpts []grpc.DialOption
if a.insecure {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
} else {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})))
}
conn, err := grpc.DialContext(dialCtx, gateway.InvokeURI, dialOpts...)
if err != nil {
return nil, merrors.InternalWrap(err, "dial gateway")
}
defer conn.Close()
client := connectorv1.NewConnectorServiceClient(conn)
// Call with timeout
callCtx, callCancel := context.WithTimeout(ctx, a.callTimeout)
defer callCancel()
req := &connectorv1.GetBalanceRequest{
AccountRef: &connectorv1.AccountRef{
AccountId: walletRef,
},
}
resp, err := client.GetBalance(callCtx, req)
if err != nil {
return nil, err
}
return resp.GetBalance(), nil
}

View File

@@ -1,24 +1,29 @@
package walletapiimp
import (
"context"
"crypto/tls"
"encoding/json"
"net/http"
"strings"
"github.com/google/uuid"
"github.com/tech/sendico/pkg/api/http/response"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
"github.com/tech/sendico/pkg/mutil/mzap"
describablev1 "github.com/tech/sendico/pkg/proto/common/describable/v1"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
"github.com/tech/sendico/server/interface/api/srequest"
"github.com/tech/sendico/server/interface/api/sresponse"
mutil "github.com/tech/sendico/server/internal/mutil/param"
ast "github.com/tech/sendico/server/internal/mutil/proto"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/types/known/structpb"
)
func (a *WalletAPI) create(r *http.Request, account *model.Account, token *sresponse.TokenData) http.HandlerFunc {
@@ -52,47 +57,134 @@ func (a *WalletAPI) create(r *http.Request, account *model.Account, token *sresp
return response.Auto(a.logger, a.Name(), err)
}
if a.chainGateway == nil {
return response.Internal(a.logger, mservice.ChainGateway, merrors.Internal("chain gateway client is not configured"))
if a.discovery == nil {
return response.Internal(a.logger, mservice.ChainGateway, merrors.Internal("discovery client is not configured"))
}
// Find gateway for this network
lookupCtx, cancel := context.WithTimeout(ctx, discoveryLookupTimeout)
defer cancel()
lookupResp, err := a.discovery.Lookup(lookupCtx)
if err != nil {
a.logger.Warn("Failed to lookup discovery registry", zap.Error(err))
return response.Auto(a.logger, a.Name(), err)
}
// Find gateway that handles this network
networkName := strings.ToLower(string(asset.Asset.Chain))
gateway := findGatewayForNetwork(lookupResp.Gateways, networkName)
if gateway == nil {
a.logger.Warn("No gateway found for network",
zap.String("network", networkName),
zap.String("chain", string(sr.Asset.Chain)))
return response.Auto(a.logger, a.Name(), merrors.InvalidArgument("no gateway available for network: "+networkName))
}
var ownerRef string
if sr.OwnerRef != nil && !sr.OwnerRef.IsZero() {
ownerRef = sr.OwnerRef.Hex()
}
passet, err := ast.Asset2Proto(&asset.Asset)
if err != nil {
a.logger.Warn("Failed to convert asset to proto asset", zap.Error(err),
mzap.StorableRef(asset), mzap.StorableRef(account))
return response.Auto(a.logger, a.Name(), err)
// Build params for connector OpenAccount
params := map[string]interface{}{
"organization_ref": orgRef.Hex(),
"network": networkName,
"token_symbol": asset.Asset.TokenSymbol,
"contract_address": asset.Asset.ContractAddress,
}
if sr.Description.Description != nil {
params["description"] = *sr.Description.Description
}
params["metadata"] = map[string]interface{}{
"source": "create",
"login": account.Login,
}
req := &chainv1.CreateManagedWalletRequest{
IdempotencyKey: uuid.NewString(),
OrganizationRef: orgRef.Hex(),
OwnerRef: ownerRef,
Describable: &describablev1.Describable{
Name: sr.Description.Name,
Description: sr.Description.Description,
},
Asset: passet,
Metadata: map[string]string{
"source": "create",
"login": account.Login,
},
paramsStruct, _ := structpb.NewStruct(params)
assetString := networkName + "-" + asset.Asset.TokenSymbol
req := &connectorv1.OpenAccountRequest{
IdempotencyKey: uuid.NewString(),
Kind: connectorv1.AccountKind_CHAIN_MANAGED_WALLET,
Asset: assetString,
OwnerRef: ownerRef,
Label: sr.Description.Name,
Params: paramsStruct,
}
resp, err := a.chainGateway.CreateManagedWallet(ctx, req)
// Connect to gateway and create wallet
walletRef, err := a.createWalletOnGateway(ctx, *gateway, req)
if err != nil {
a.logger.Warn("Failed to create managed wallet", zap.Error(err), mzap.ObjRef("organization_ref", orgRef), mzap.StorableRef(account))
a.logger.Warn("Failed to create managed wallet", zap.Error(err),
mzap.ObjRef("organization_ref", orgRef), mzap.StorableRef(account),
zap.String("gateway_id", gateway.ID), zap.String("network", gateway.Network))
return response.Auto(a.logger, a.Name(), err)
}
if resp == nil || resp.Wallet == nil || strings.TrimSpace(resp.Wallet.WalletRef) == "" {
return response.Auto(a.logger, a.Name(), merrors.Internal("chain gateway returned empty wallet reference"))
}
a.logger.Info("Managed wallet created for organization", mzap.ObjRef("organization_ref", orgRef),
zap.String("wallet_ref", resp.Wallet.WalletRef), mzap.StorableRef(account))
zap.String("wallet_ref", walletRef), mzap.StorableRef(account),
zap.String("gateway_id", gateway.ID), zap.String("network", gateway.Network))
return sresponse.Success(a.logger, token)
}
func findGatewayForNetwork(gateways []discovery.GatewaySummary, network string) *discovery.GatewaySummary {
network = strings.ToLower(strings.TrimSpace(network))
for _, gw := range gateways {
if !strings.EqualFold(gw.Rail, cryptoRail) || !gw.Healthy || strings.TrimSpace(gw.InvokeURI) == "" {
continue
}
// Check if gateway network matches
gwNetwork := strings.ToLower(strings.TrimSpace(gw.Network))
if gwNetwork == network {
return &gw
}
// Also check if network starts with gateway network prefix (e.g., "tron" matches "tron_mainnet")
if strings.HasPrefix(network, gwNetwork) || strings.HasPrefix(gwNetwork, network) {
return &gw
}
}
return nil
}
func (a *WalletAPI) createWalletOnGateway(ctx context.Context, gateway discovery.GatewaySummary, req *connectorv1.OpenAccountRequest) (string, error) {
// Create connection with timeout
dialCtx, cancel := context.WithTimeout(ctx, a.dialTimeout)
defer cancel()
var dialOpts []grpc.DialOption
if a.insecure {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
} else {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})))
}
conn, err := grpc.DialContext(dialCtx, gateway.InvokeURI, dialOpts...)
if err != nil {
return "", merrors.InternalWrap(err, "dial gateway")
}
defer conn.Close()
client := connectorv1.NewConnectorServiceClient(conn)
// Call with timeout
callCtx, callCancel := context.WithTimeout(ctx, a.callTimeout)
defer callCancel()
resp, err := client.OpenAccount(callCtx, req)
if err != nil {
return "", err
}
if resp.GetError() != nil {
return "", merrors.Internal(resp.GetError().GetMessage())
}
account := resp.GetAccount()
if account == nil || account.GetRef() == nil {
return "", merrors.Internal("gateway returned empty account")
}
return strings.TrimSpace(account.GetRef().GetAccountId()), nil
}

View File

@@ -1,18 +1,27 @@
package walletapiimp
import (
"context"
"crypto/tls"
"net/http"
"strings"
"sync"
"github.com/tech/sendico/pkg/api/http/response"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
"github.com/tech/sendico/pkg/model"
"github.com/tech/sendico/pkg/mservice"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
"github.com/tech/sendico/pkg/mutil/mzap"
connectorv1 "github.com/tech/sendico/pkg/proto/connector/v1"
"github.com/tech/sendico/server/interface/api/sresponse"
mutil "github.com/tech/sendico/server/internal/mutil/param"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/protobuf/types/known/wrapperspb"
)
func (a *WalletAPI) listWallets(r *http.Request, account *model.Account, token *sresponse.TokenData) http.HandlerFunc {
@@ -23,31 +32,126 @@ func (a *WalletAPI) listWallets(r *http.Request, account *model.Account, token *
}
ctx := r.Context()
res, err := a.enf.Enforce(ctx, a.walletsPermissionRef, account.ID, orgRef, primitive.NilObjectID, model.ActionRead)
hasReadPermission, err := a.enf.Enforce(ctx, a.walletsPermissionRef, account.ID, orgRef, primitive.NilObjectID, model.ActionRead)
if err != nil {
a.logger.Warn("Failed to check chain wallet access permissions", zap.Error(err), mutil.PLog(a.oph, r))
return response.Auto(a.logger, a.Name(), err)
}
if !res {
a.logger.Debug("Access denied when listing organization wallets", mutil.PLog(a.oph, r))
return response.AccessDenied(a.logger, a.Name(), "wallets read permission denied")
}
if a.chainGateway == nil {
return response.Internal(a.logger, mservice.ChainGateway, merrors.Internal("chain gateway client is not configured"))
if a.discovery == nil {
return response.Internal(a.logger, mservice.ChainGateway, merrors.Internal("discovery client is not configured"))
}
req := &chainv1.ListManagedWalletsRequest{
OrganizationRef: orgRef.Hex(),
}
if owner := strings.TrimSpace(r.URL.Query().Get("owner_ref")); owner != "" {
req.OwnerRef = owner
}
// Discover CRYPTO rail gateways
lookupCtx, cancel := context.WithTimeout(ctx, discoveryLookupTimeout)
defer cancel()
resp, err := a.chainGateway.ListManagedWallets(ctx, req)
lookupResp, err := a.discovery.Lookup(lookupCtx)
if err != nil {
a.logger.Warn("Failed to list managed wallets", zap.Error(err), zap.String("organization_ref", orgRef.Hex()))
return response.Auto(a.logger, mservice.ChainGateway, err)
a.logger.Warn("Failed to lookup discovery registry", zap.Error(err))
return response.Auto(a.logger, a.Name(), err)
}
return sresponse.Wallets(a.logger, resp, token)
// Filter gateways by CRYPTO rail
cryptoGateways := filterCryptoGateways(lookupResp.Gateways)
if len(cryptoGateways) == 0 {
a.logger.Debug("No CRYPTO rail gateways found in discovery")
return sresponse.Wallets(a.logger, nil, token)
}
// Build request
req := &connectorv1.ListAccountsRequest{
OrganizationRef: orgRef.Hex(),
Kind: connectorv1.AccountKind_CHAIN_MANAGED_WALLET,
}
// If user has read permission, return all wallets in organization.
// Otherwise, filter to only wallets owned by the requesting account.
if !hasReadPermission {
req.OwnerRefFilter = wrapperspb.String(account.ID.Hex())
a.logger.Debug("Filtering wallets by owner due to limited permissions",
mzap.ObjRef("owner_ref", account.ID), mutil.PLog(a.oph, r))
}
// Query all gateways in parallel
allAccounts := a.queryAllGateways(ctx, cryptoGateways, req)
return sresponse.WalletsFromAccounts(a.logger, allAccounts, token)
}
func filterCryptoGateways(gateways []discovery.GatewaySummary) []discovery.GatewaySummary {
result := make([]discovery.GatewaySummary, 0)
for _, gw := range gateways {
if strings.EqualFold(gw.Rail, cryptoRail) && gw.Healthy && strings.TrimSpace(gw.InvokeURI) != "" {
result = append(result, gw)
}
}
return result
}
func (a *WalletAPI) queryAllGateways(ctx context.Context, gateways []discovery.GatewaySummary, req *connectorv1.ListAccountsRequest) []*connectorv1.Account {
var mu sync.Mutex
var wg sync.WaitGroup
allAccounts := make([]*connectorv1.Account, 0)
for _, gw := range gateways {
wg.Add(1)
go func(gateway discovery.GatewaySummary) {
defer wg.Done()
accounts, err := a.queryGateway(ctx, gateway, req)
if err != nil {
a.logger.Warn("Failed to query gateway",
zap.String("gateway_id", gateway.ID),
zap.String("invoke_uri", gateway.InvokeURI),
zap.String("network", gateway.Network),
zap.Error(err))
return
}
mu.Lock()
allAccounts = append(allAccounts, accounts...)
mu.Unlock()
a.logger.Debug("Queried gateway successfully",
zap.String("gateway_id", gateway.ID),
zap.String("network", gateway.Network),
zap.Int("accounts_count", len(accounts)))
}(gw)
}
wg.Wait()
return allAccounts
}
func (a *WalletAPI) queryGateway(ctx context.Context, gateway discovery.GatewaySummary, req *connectorv1.ListAccountsRequest) ([]*connectorv1.Account, error) {
// Create connection with timeout
dialCtx, cancel := context.WithTimeout(ctx, a.dialTimeout)
defer cancel()
var dialOpts []grpc.DialOption
if a.insecure {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials()))
} else {
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{})))
}
conn, err := grpc.DialContext(dialCtx, gateway.InvokeURI, dialOpts...)
if err != nil {
return nil, merrors.InternalWrap(err, "dial gateway")
}
defer conn.Close()
client := connectorv1.NewConnectorServiceClient(conn)
// Call with timeout
callCtx, callCancel := context.WithTimeout(ctx, a.callTimeout)
defer callCancel()
resp, err := client.ListAccounts(callCtx, req)
if err != nil {
return nil, err
}
return resp.GetAccounts(), nil
}

View File

@@ -2,60 +2,63 @@ package walletapiimp
import (
"context"
"fmt"
"os"
"strings"
"time"
chaingatewayclient "github.com/tech/sendico/gateway/chain/client"
api "github.com/tech/sendico/pkg/api/http"
"github.com/tech/sendico/pkg/auth"
"github.com/tech/sendico/pkg/db/chainassets"
"github.com/tech/sendico/pkg/discovery"
"github.com/tech/sendico/pkg/merrors"
msg "github.com/tech/sendico/pkg/messaging"
"github.com/tech/sendico/pkg/mlogger"
"github.com/tech/sendico/pkg/mservice"
chainv1 "github.com/tech/sendico/pkg/proto/gateway/chain/v1"
eapi "github.com/tech/sendico/server/interface/api"
mutil "github.com/tech/sendico/server/internal/mutil/param"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.uber.org/zap"
)
const (
cryptoRail = "CRYPTO"
defaultDialTimeout = 5 * time.Second
defaultCallTimeout = 10 * time.Second
discoveryLookupTimeout = 3 * time.Second
)
type WalletAPI struct {
logger mlogger.Logger
chainGateway chainWalletClient
discovery *discovery.Client
enf auth.Enforcer
oph mutil.ParamHelper
wph mutil.ParamHelper
walletsPermissionRef primitive.ObjectID
balancesPermissionRef primitive.ObjectID
assets chainassets.DB
}
type chainWalletClient interface {
CreateManagedWallet(ctx context.Context, req *chainv1.CreateManagedWalletRequest) (*chainv1.CreateManagedWalletResponse, error)
ListManagedWallets(ctx context.Context, req *chainv1.ListManagedWalletsRequest) (*chainv1.ListManagedWalletsResponse, error)
GetWalletBalance(ctx context.Context, req *chainv1.GetWalletBalanceRequest) (*chainv1.GetWalletBalanceResponse, error)
Close() error
// Gateway connection settings
dialTimeout time.Duration
callTimeout time.Duration
insecure bool
}
func (a *WalletAPI) Name() mservice.Type { return mservice.ChainWallets }
func (a *WalletAPI) Finish(ctx context.Context) error {
if a.chainGateway != nil {
if err := a.chainGateway.Close(); err != nil {
a.logger.Warn("Failed to close chain gateway client", zap.Error(err))
}
if a.discovery != nil {
a.discovery.Close()
}
return nil
}
func CreateAPI(apiCtx eapi.API) (*WalletAPI, error) {
p := &WalletAPI{
logger: apiCtx.Logger().Named(mservice.Wallets),
enf: apiCtx.Permissions().Enforcer(),
oph: mutil.CreatePH(mservice.Organizations),
wph: mutil.CreatePH(mservice.Wallets),
logger: apiCtx.Logger().Named(mservice.Wallets),
enf: apiCtx.Permissions().Enforcer(),
oph: mutil.CreatePH(mservice.Organizations),
wph: mutil.CreatePH(mservice.Wallets),
dialTimeout: defaultDialTimeout,
callTimeout: defaultCallTimeout,
insecure: true,
}
var err error
@@ -83,9 +86,22 @@ func CreateAPI(apiCtx eapi.API) (*WalletAPI, error) {
p.logger.Error("Failed to fetch service configuration")
return nil, merrors.InvalidArgument("No configuration provided")
}
if err := p.initChainGateway(cfg.ChainGateway); err != nil {
p.logger.Error("Failed to initialize chain gateway client", zap.Error(err))
return nil, err
// Apply gateway connection settings from config
if gatewayCfg := cfg.ChainGateway; gatewayCfg != nil {
if gatewayCfg.DialTimeoutSeconds > 0 {
p.dialTimeout = time.Duration(gatewayCfg.DialTimeoutSeconds) * time.Second
}
if gatewayCfg.CallTimeoutSeconds > 0 {
p.callTimeout = time.Duration(gatewayCfg.CallTimeoutSeconds) * time.Second
}
p.insecure = gatewayCfg.Insecure
}
// Initialize discovery client
if err := p.initDiscoveryClient(cfg); err != nil {
p.logger.Warn("Failed to initialize discovery client", zap.Error(err))
// Not fatal - we can still work without discovery
}
apiCtx.Register().AccountHandler(p.Name(), p.oph.AddRef("/"), api.Get, p.listWallets)
@@ -95,31 +111,22 @@ func CreateAPI(apiCtx eapi.API) (*WalletAPI, error) {
return p, nil
}
func (a *WalletAPI) initChainGateway(cfg *eapi.ChainGatewayConfig) error {
if cfg == nil {
return merrors.InvalidArgument("chain gateway configuration is not provided")
func (a *WalletAPI) initDiscoveryClient(cfg *eapi.Config) error {
if cfg == nil || cfg.Mw == nil {
return nil
}
cfg.Address = strings.TrimSpace(cfg.Address)
if cfg.Address == "" {
cfg.Address = strings.TrimSpace(os.Getenv(cfg.AddressEnv))
msgCfg := cfg.Mw.Messaging
if msgCfg.Driver == "" {
return nil
}
if cfg.Address == "" {
return merrors.InvalidArgument(fmt.Sprintf("chain gateway address is not specified and address env %s is empty", cfg.AddressEnv))
}
clientCfg := chaingatewayclient.Config{
Address: cfg.Address,
DialTimeout: time.Duration(cfg.DialTimeoutSeconds) * time.Second,
CallTimeout: time.Duration(cfg.CallTimeoutSeconds) * time.Second,
Insecure: cfg.Insecure,
}
client, err := chaingatewayclient.New(context.Background(), clientCfg)
broker, err := msg.CreateMessagingBroker(a.logger.Named("discovery_bus"), &msgCfg)
if err != nil {
return err
}
a.chainGateway = client
client, err := discovery.NewClient(a.logger, broker, nil, string(a.Name()))
if err != nil {
return err
}
a.discovery = client
return nil
}