core

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 10, 2026 License: Apache-2.0 Imports: 47 Imported by: 0

Documentation

Overview

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at 

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

Constants

View Source 
const (
	AllocationFIFO = "FIFO"
	AllocationLIFO = "LIFO"
	AllocationProp = "PROPORTIONAL"
)

Allocation strategies for fund lineage debit processing.

View Source 
const (
	InflightActionCommit = "commit"
	InflightActionVoid   = "void"
)

Inflight action names carried in an InflightActionPayload.

View Source 
const (
	StatusStarted    = reconciliation.StatusStarted
	StatusInProgress = reconciliation.StatusInProgress
	StatusCompleted  = reconciliation.StatusCompleted
	StatusFailed     = reconciliation.StatusFailed
)
View Source 
const (
	StatusQueued    = "QUEUED"
	StatusApplied   = "APPLIED"
	StatusScheduled = "SCHEDULED"
	StatusRejected  = "REJECTED"
)
View Source 
const (
	StatusInflight = "INFLIGHT"
	StatusVoid     = "VOID"
	StatusCommit   = "COMMIT"
)
View Source 
const (
	GeneralLedgerID = "general_ledger_id"
)
View Source 
const IndexModeRefresh = "refresh"

IndexModeRefresh marks an index task that carries only an entity reference. The worker re-reads Postgres before upserting so queue ordering cannot leave Typesense behind a newer metadata commit.

View Source 
const LineageFundAllocation = "LEDGERFORGE_FUND_ALLOCATION"

LineageFundAllocation is the metadata key used to store fund allocation details in a transaction.

View Source 
const LineageProviderKey = "LEDGERFORGE_LINEAGE_PROVIDER"

LineageProviderKey is the metadata key used to identify the provider of funds in a transaction.

Variables

View Source 
var ErrEntityNotFound = errors.New("entity not found")

ErrEntityNotFound is returned by UpdateMetadata when the target entity does not exist. API handlers match it with errors.Is to return 404.

View Source 
var ErrInflightActionQueued = errors.New("a commit or void is already queued for this transaction")

ErrInflightActionQueued is returned by EnqueueInflightAction when a commit or void for the same transaction is already queued (asynq TaskID conflict). The API maps it to HTTP 409.

View Source 
var ErrQueueBackpressure = errors.New("queue backpressure: redis memory or pending task limit reached")

ErrQueueBackpressure is returned when enqueue is rejected to keep Redis below OOM.

View Source 
var SQLFiles embed.FS

Functions

func ClassifyInflightError 

func ClassifyInflightError(err error) string

classifyInflightError maps the unstructured errors emitted by CommitInflightTransaction / VoidInflightTransaction to stable codes so bulk callers can branch on them programmatically instead of regex-matching human-readable messages.

Codes are intentionally action-agnostic where possible. ALREADY_VOIDED and ALREADY_COMMITTED are reported as observed — a void on an already-committed txn surfaces ALREADY_COMMITTED, and vice versa. ClassifyInflightError maps an inflight commit/void error to a stable code (see classifyInflightError) so callers outside this package — e.g. the queued bulk API path — can report per-item codes without regex-matching messages.

func DocumentToIndexMap 

func DocumentToIndexMap(document interface{}) (map[string]interface{}, error)

DocumentToIndexMap converts a struct or map into the object Typesense upsert expects.

func IsDuplicateReferenceError 

func IsDuplicateReferenceError(err error) bool

func IsInflightTransaction 

func IsInflightTransaction(transaction *model.Transaction) bool

IsInflightTransaction checks whether a transaction is considered "inflight" based on its status and metadata. A transaction is considered inflight if: 1. It has a status of StatusInflight, OR 2. It has a status of StatusQueued AND has the inflight flag set to true in its metadata

Parameters: - transaction *model.Transaction: The transaction to check

Returns: - bool: true if the transaction is considered inflight, false otherwise

func NewBalanceTracker 

func NewBalanceTracker() *model.BalanceTracker

NewBalanceTracker creates a new BalanceTracker instance. It initializes the Balances and Frequencies maps.

Returns: - *model.BalanceTracker: A pointer to the newly created BalanceTracker instance.

Types

type Allocation 

type Allocation struct {
	BalanceID string
	Amount    *big.Int
}

Allocation represents the amount allocated from a specific shadow balance during debit processing.

Fields: - BalanceID string: The ID of the shadow balance from which funds are allocated. - Amount *big.Int: The amount allocated from this balance.

type BalanceLineage 

type BalanceLineage struct {
	BalanceID          string              `json:"balance_id"`
	TotalWithLineage   *big.Int            `json:"total_with_lineage"`
	AggregateBalanceID string              `json:"aggregate_balance_id"`
	Providers          []ProviderBreakdown `json:"providers"`
}

BalanceLineage represents the complete fund lineage for a balance.

Fields: - BalanceID string: The ID of the balance being queried. - TotalWithLineage *big.Int: The total funds tracked across all providers. - AggregateBalanceID string: The ID of the aggregate shadow balance. - Providers []ProviderBreakdown: The breakdown of funds by provider.

type BatchJobResult 

type BatchJobResult struct {
	Txn   *model.Transaction
	Error error
}

BatchJobResult represents the result of processing a transaction in a batch job.

Fields: - Txn *model.Transaction: A pointer to the processed Transaction model. - Error error: An error if the transaction could not be processed.

type BulkInflightAction 

type BulkInflightAction int

BulkInflightAction discriminates between the two operations supported by BulkInflightUpdate. 

const (
	BulkInflightVoid BulkInflightAction = iota
	BulkInflightCommit
)

type BulkInflightItem 

type BulkInflightItem struct {
	TransactionID string
	Amount        *big.Int
}

BulkInflightItem is the internal per-item shape consumed by BulkInflightUpdate. For void calls only TransactionID is used.

type BulkInflightOutcome 

type BulkInflightOutcome struct {
	TransactionID string
	Txn           *model.Transaction
	Err           error
	Code          string
}

BulkInflightOutcome is the per-item outcome from BulkInflightUpdate. On success Err is nil; on failure Code carries a stable classification (see classifyInflightError) and Err preserves the original message.

type BulkTransactionError 

type BulkTransactionError struct {
	Message string
	// contains filtered or unexported fields
}

BulkTransactionError reports the failure of a synchronous bulk batch.

Message is the client-facing sentence: the failing item's error plus what happened to the rest of the batch (rolled back, or left in place). cause is the error the item itself raised, kept so callers can unwrap to it — the API layer resolves the response's error code from there, which is how a missing balance inside a batch reports BAL_NOT_FOUND rather than being text-matched into TXN_NOT_FOUND.

func (*BulkTransactionError) Error 

func (e *BulkTransactionError) Error() string

func (*BulkTransactionError) Unwrap 

func (e *BulkTransactionError) Unwrap() error

type ChainProcessor 

type ChainProcessor struct {
	// contains filtered or unexported fields
}

ChainProcessor seals transactions into the tamper-evident hash chain off the hot path. It polls for unchained rows and links them in a single DB transaction per batch — modeled on LineageOutboxProcessor.

func NewChainProcessor 

func NewChainProcessor(ledgerforge *LedgerForge) *ChainProcessor

NewChainProcessor builds a ChainProcessor from the HashChain config (with safe fallbacks).

func (*ChainProcessor) IsRunning 

func (p *ChainProcessor) IsRunning() bool

IsRunning reports whether the loop is active.

func (*ChainProcessor) Start 

func (p *ChainProcessor) Start(ctx context.Context)

Start launches the background chainer loop.

func (*ChainProcessor) Stop 

func (p *ChainProcessor) Stop()

Stop signals the loop to exit and waits for it.

type ChainVerifyResult 

type ChainVerifyResult = model.ChainVerifyResult

ChainVerifyResult is the outcome of replaying the chain from genesis.

type IndexTask 

type IndexTask struct {
	Collection string                 `json:"collection"`
	Payload    map[string]interface{} `json:"payload,omitempty"`
	DocumentID string                 `json:"document_id,omitempty"`
	ScopeID    string                 `json:"scope_id,omitempty"`
	Mode       string                 `json:"mode,omitempty"`
}

IndexTask is the payload for the Typesense index queue. Legacy tasks embed a snapshot in Payload; metadata updates enqueue Mode=refresh with DocumentID or ScopeID instead.

type IndexUpserter 

type IndexUpserter func(ctx context.Context, collection string, document interface{}) error

IndexUpserter indexes one document into a Typesense collection.

type InflightActionPayload 

type InflightActionPayload struct {
	TransactionID string `json:"transaction_id"`
	Action        string `json:"action"`         // "commit" | "void"
	PreciseAmount string `json:"precise_amount"` // big.Int string; "0" = full remaining
	ActionID      string `json:"action_id"`
}

InflightActionPayload is the task payload for a queued inflight commit/void. ActionID is generated once at enqueue time and seeds the per-leg commit/void references so asynq retries are idempotent.

type LedgerForge 

type LedgerForge struct {
	Hooks hooks.HookManager
	// contains filtered or unexported fields
}

LedgerForge represents the main struct for the LedgerForge application.

func NewLedgerForge 

func NewLedgerForge(db database.IDataSource) (*LedgerForge, error)

NewLedgerForge initializes a new instance of LedgerForge with the provided database datasource. It fetches the configuration, initializes Redis clients, queue, reconciliation, and search services.

Parameters: - db database.IDataSource: The datasource for database operations.

Returns: - *LedgerForge: A pointer to the newly created LedgerForge instance. - error: An error if any of the initialization steps fail.

func (*LedgerForge) BulkInflightUpdate 

func (l *LedgerForge) BulkInflightUpdate(ctx context.Context, action BulkInflightAction, items []BulkInflightItem, maxWorkers int) ([]BulkInflightOutcome, error)

BulkInflightUpdate voids or commits a list of independently-created inflight transactions in parallel.

Each id is processed in its own goroutine via a worker pool of size `maxWorkers` (clamped to [1, 16]). Per-item failures are reported in the returned result slice and do not abort the rest of the batch; the returned error is non-nil only for input validation failures (empty list, too many items).

Idempotency: bulk has no batch-wide idempotency key. Retries that include already-processed ids will see those ids surface as ALREADY_VOIDED / ALREADY_COMMITTED, which callers should treat as success-equivalent for retry semantics.

Results are returned in the same order as the input.

func (*LedgerForge) Close 

func (b *LedgerForge) Close() error

Close properly closes all connections and resources used by the LedgerForge instance.

func (*LedgerForge) CommitInflightTransaction 

func (l *LedgerForge) CommitInflightTransaction(ctx context.Context, transactionID string, amount *big.Int) (*model.Transaction, error)

CommitInflightTransaction commits an inflight transaction by validating and updating its amount, and finalizing the commitment. It starts a tracing span, fetches and validates the inflight transaction, updates the amount, and finalizes the commitment.

Parameters: - ctx context.Context: The context for the operation. - transactionID string: The ID of the inflight transaction to be committed. - amount float64: The amount to be validated and updated in the transaction.

Returns: - *model.Transaction: A pointer to the committed Transaction model. - error: An error if the transaction could not be committed.

func (*LedgerForge) CommitInflightTransactionWithQueue 

func (l *LedgerForge) CommitInflightTransactionWithQueue(ctx context.Context, transactionID string, amount *big.Int) (*model.Transaction, error)

func (*LedgerForge) CommitInflightTransactionWithRef 

func (l *LedgerForge) CommitInflightTransactionWithRef(ctx context.Context, transactionID string, amount *big.Int, reference string) (*model.Transaction, error)

CommitInflightTransactionWithRef commits an inflight transaction using a caller-supplied reference for the committed child. A non-empty reference makes asynq retries idempotent: a retry that re-commits the same amount collides on the unique reference index. An empty reference preserves the legacy behavior of generating a random reference.

func (*LedgerForge) CommitWorker 

func (l *LedgerForge) CommitWorker(ctx context.Context, jobs <-chan *model.Transaction, results chan<- BatchJobResult, wg *sync.WaitGroup, amount *big.Int)

CommitWorker processes commit transactions from the jobs channel and sends the results to the results channel. It starts a tracing span, processes each transaction, and records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - jobs <-chan *model.Transaction: A channel from which transactions are received for processing. - results chan<- BatchJobResult: A channel to which the results of the processing are sent. - wg *sync.WaitGroup: A wait group to synchronize the completion of the worker. - amount *big.Int: The amount to be processed in the transaction.

func (*LedgerForge) Config 

func (b *LedgerForge) Config() *config.Configuration

Config returns the cached configuration for the LedgerForge instance. Falls back to config.Fetch() if not initialized (for backward compatibility with tests).

func (*LedgerForge) CreateAPIKey 

func (l *LedgerForge) CreateAPIKey(ctx context.Context, name, ownerID string, scopes []string, expiresAt time.Time) (*model.APIKey, error)

CreateAPIKey creates a new API key for the specified owner

Parameters: - ctx: The context for the operation - name: Name of the API key - ownerID: ID of the key owner - scopes: List of permission scopes - expiresAt: Expiration time for the key

Returns: - *model.APIKey: The created API key - error: An error if the operation fails

func (*LedgerForge) CreateAccount 

func (l *LedgerForge) CreateAccount(account model.Account) (model.Account, error)

CreateAccount creates a new account in the database. It overrides the ledger and identity details, applies the account name, and fetches external account details.

Parameters: - account model.Account: The Account model to be created.

Returns: - model.Account: The created Account model. - error: An error if the account could not be created.

func (*LedgerForge) CreateBalance 

func (l *LedgerForge) CreateBalance(ctx context.Context, balance model.Balance) (model.Balance, error)

CreateBalance creates a new balance. It starts a tracing span, creates the balance, and performs post-creation actions.

Parameters: - ctx context.Context: The context for the operation. - balance model.Balance: The Balance model to be created.

Returns: - model.Balance: The created Balance model. - error: An error if the balance could not be created.

func (*LedgerForge) CreateBulkTransactions 

func (l *LedgerForge) CreateBulkTransactions(ctx context.Context, req *model.BulkTransactionRequest) (*model.BulkTransactionResult, error)

CreateBulkTransactions handles the creation of multiple transactions in a batch. If atomic is true: Any failure will cause all transactions to be rolled back (or voided if inflight). If atomic is false: Failures will stop processing but previous transactions remain unaffected. If run_async is true: Processing happens in background with webhook notifications.

func (*LedgerForge) CreateIdentity 

func (l *LedgerForge) CreateIdentity(identity model.Identity) (model.Identity, error)

CreateIdentity creates a new identity in the database.

Parameters: - identity model.Identity: The Identity model to be created.

Returns: - model.Identity: The created Identity model. - error: An error if the identity could not be created.

func (*LedgerForge) CreateLedger 

func (l *LedgerForge) CreateLedger(ledger model.Ledger) (model.Ledger, error)

CreateLedger creates a new ledger. It calls postLedgerActions after a successful creation.

Parameters: - ledger: A Ledger model representing the ledger to be created.

Returns: - model.Ledger: The created Ledger model. - error: An error if the ledger could not be created.

func (*LedgerForge) CreateMatchingRule 

func (b *LedgerForge) CreateMatchingRule(ctx context.Context, rule model.MatchingRule) (*model.MatchingRule, error)

func (*LedgerForge) CreateMonitor 

func (l *LedgerForge) CreateMonitor(ctx context.Context, monitor model.BalanceMonitor) (model.BalanceMonitor, error)

CreateMonitor creates a new balance monitor. It starts a tracing span, applies precision to the monitor's condition value, and creates the monitor. It records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - monitor model.BalanceMonitor: The BalanceMonitor model to be created.

Returns: - model.BalanceMonitor: The created BalanceMonitor model. - error: An error if the monitor could not be created.

func (*LedgerForge) DecidePolicyApproval 

func (l *LedgerForge) DecidePolicyApproval(ctx context.Context, id string, approve bool) (*model.PolicyAdmission, error)

func (*LedgerForge) DeleteIdentity 

func (l *LedgerForge) DeleteIdentity(id string) error

DeleteIdentity deletes an identity by its ID.

Parameters: - id string: The ID of the identity to delete.

Returns: - error: An error if the identity could not be deleted.

func (*LedgerForge) DeleteKeyPolicy 

func (l *LedgerForge) DeleteKeyPolicy(ctx context.Context, key string) error

func (*LedgerForge) DeleteMatchingRule 

func (b *LedgerForge) DeleteMatchingRule(ctx context.Context, id string) error

func (*LedgerForge) DeleteMonitor 

func (l *LedgerForge) DeleteMonitor(ctx context.Context, id string) error

DeleteMonitor deletes a balance monitor by its ID. It starts a tracing span, deletes the monitor, and records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - id string: The ID of the monitor to delete.

Returns: - error: An error if the monitor could not be deleted.

func (*LedgerForge) DetokenizeIdentity 

func (l *LedgerForge) DetokenizeIdentity(identityID string) (map[string]string, error)

DetokenizeIdentity detokenizes and returns all tokenized fields in an identity.

Parameters: - identityID string: The ID of the identity.

Returns: - map[string]string: A map of field names to their detokenized values. - error: An error if any field could not be detokenized.

func (*LedgerForge) DetokenizeIdentityField 

func (l *LedgerForge) DetokenizeIdentityField(identityID, fieldName string) (string, error)

DetokenizeIdentityField detokenizes a specific field in an identity.

Parameters: - identityID string: The ID of the identity. - fieldName string: The name of the field to detokenize.

Returns: - string: The detokenized field value. - error: An error if the field could not be detokenized.

func (*LedgerForge) GetAPIKeyByKey 

func (l *LedgerForge) GetAPIKeyByKey(ctx context.Context, key string) (*model.APIKey, error)

GetAPIKeyByKey retrieves an API key by its key string

Parameters: - ctx: The context for the operation - key: The API key string

Returns: - *model.APIKey: The API key if found - error: An error if the operation fails

func (*LedgerForge) GetAccount 

func (l *LedgerForge) GetAccount(id string, include []string) (*model.Account, error)

GetAccount retrieves an account by its ID. It fetches the account from the datasource and includes additional data as specified.

Parameters: - id string: The ID of the account to retrieve. - include []string: A slice of strings specifying additional data to include.

Returns: - *model.Account: A pointer to the Account model if found. - error: An error if the account could not be retrieved.

func (*LedgerForge) GetAccountByNumber 

func (l *LedgerForge) GetAccountByNumber(id string) (*model.Account, error)

GetAccountByNumber retrieves an account from the database by its account number.

Parameters: - id string: The account number of the account to retrieve.

Returns: - *model.Account: A pointer to the Account model if found. - error: An error if the account could not be retrieved.

func (*LedgerForge) GetAllAccounts 

func (l *LedgerForge) GetAllAccounts() ([]model.Account, error)

GetAllAccounts retrieves all accounts from the database.

Returns: - []model.Account: A slice of Account models. - error: An error if the accounts could not be retrieved.

func (*LedgerForge) GetAllAccountsWithFilter 

func (l *LedgerForge) GetAllAccountsWithFilter(ctx context.Context, filters *filter.QueryFilterSet, limit, offset int) ([]model.Account, error)

GetAllAccountsWithFilter retrieves accounts using advanced filters.

Parameters: - ctx context.Context: The context for the operation. - filters *filter.QueryFilterSet: Filter conditions to apply. - limit int: Maximum number of accounts to return. - offset int: Offset for pagination.

Returns: - []model.Account: A slice of Account models matching the filter criteria. - error: An error if the accounts could not be retrieved.

func (*LedgerForge) GetAllAccountsWithFilterAndOptions 

func (l *LedgerForge) GetAllAccountsWithFilterAndOptions(ctx context.Context, filters *filter.QueryFilterSet, opts *filter.QueryOptions, limit, offset int) ([]model.Account, *int64, error)

GetAllAccountsWithFilterAndOptions retrieves accounts with filters, sorting, and optional count.

Parameters: - ctx context.Context: The context for the operation. - filters *filter.QueryFilterSet: Filter conditions to apply. - opts *filter.QueryOptions: Query options including sorting and count settings. - limit int: Maximum number of accounts to return. - offset int: Offset for pagination.

Returns: - []model.Account: A slice of Account models matching the filter criteria. - *int64: Optional total count of matching records (if opts.IncludeCount is true). - error: An error if the accounts could not be retrieved.

func (*LedgerForge) GetAllBalances 

func (l *LedgerForge) GetAllBalances(ctx context.Context, limit, offset int) ([]model.Balance, error)

GetAllBalances retrieves all balances. It starts a tracing span, fetches all balances, and records relevant events.

Parameters: - ctx context.Context: The context for the operation.

Returns: - []model.Balance: A slice of Balance models. - error: An error if the balances could not be retrieved.

func (*LedgerForge) GetAllBalancesWithFilter 

func (l *LedgerForge) GetAllBalancesWithFilter(ctx context.Context, filters *filter.QueryFilterSet, limit, offset int) ([]model.Balance, error)

GetAllBalancesWithFilter retrieves balances using advanced filters. It starts a tracing span, fetches balances matching the filter criteria, and records relevant events.

Parameters: - ctx context.Context: The context for the operation. - filters *filter.QueryFilterSet: Filter conditions to apply. - limit int: Maximum number of balances to return. - offset int: Offset for pagination.

Returns: - []model.Balance: A slice of Balance models matching the filter criteria. - error: An error if the balances could not be retrieved.

func (*LedgerForge) GetAllBalancesWithFilterAndOptions 

func (l *LedgerForge) GetAllBalancesWithFilterAndOptions(ctx context.Context, filters *filter.QueryFilterSet, opts *filter.QueryOptions, limit, offset int) ([]model.Balance, *int64, error)

GetAllBalancesWithFilterAndOptions retrieves balances with advanced filters, sorting, and optional count.

func (*LedgerForge) GetAllIdentities 

func (l *LedgerForge) GetAllIdentities() ([]model.Identity, error)

GetAllIdentities retrieves all identities from the database.

Returns: - []model.Identity: A slice of Identity models. - error: An error if the identities could not be retrieved.

func (*LedgerForge) GetAllIdentitiesWithFilter 

func (l *LedgerForge) GetAllIdentitiesWithFilter(ctx context.Context, filters *filter.QueryFilterSet, limit, offset int) ([]model.Identity, error)

GetAllIdentitiesWithFilter retrieves identities using advanced filters.

Parameters: - ctx context.Context: The context for the operation. - filters *filter.QueryFilterSet: Filter conditions to apply. - limit int: Maximum number of identities to return. - offset int: Offset for pagination.

Returns: - []model.Identity: A slice of Identity models matching the filter criteria. - error: An error if the identities could not be retrieved.

func (*LedgerForge) GetAllIdentitiesWithFilterAndOptions 

func (l *LedgerForge) GetAllIdentitiesWithFilterAndOptions(ctx context.Context, filters *filter.QueryFilterSet, opts *filter.QueryOptions, limit, offset int) ([]model.Identity, *int64, error)

GetAllIdentitiesWithFilterAndOptions retrieves identities with filters, sorting, and optional count.

Parameters: - ctx context.Context: The context for the operation. - filters *filter.QueryFilterSet: Filter conditions to apply. - opts *filter.QueryOptions: Query options including sorting and count settings. - limit int: Maximum number of identities to return. - offset int: Offset for pagination.

Returns: - []model.Identity: A slice of Identity models matching the filter criteria. - *int64: Optional total count of matching records (if opts.IncludeCount is true). - error: An error if the identities could not be retrieved.

func (*LedgerForge) GetAllLedgers 

func (l *LedgerForge) GetAllLedgers(limit, offset int) ([]model.Ledger, error)

GetAllLedgers retrieves all ledgers from the datasource. It returns a slice of Ledger models and an error if the operation fails.

Returns: - []model.Ledger: A slice of Ledger models. - error: An error if the ledgers could not be retrieved.

func (*LedgerForge) GetAllLedgersWithFilter 

func (l *LedgerForge) GetAllLedgersWithFilter(ctx context.Context, filters *filter.QueryFilterSet, limit, offset int) ([]model.Ledger, error)

GetAllLedgersWithFilter retrieves ledgers from the datasource using advanced filters. It returns a slice of Ledger models and an error if the operation fails.

Parameters: - ctx: Context for the operation. - filters: A QueryFilterSet containing filter conditions. - limit: Maximum number of ledgers to return. - offset: Offset for pagination.

Returns: - []model.Ledger: A slice of Ledger models matching the filter criteria. - error: An error if the ledgers could not be retrieved.

func (*LedgerForge) GetAllLedgersWithFilterAndOptions 

func (l *LedgerForge) GetAllLedgersWithFilterAndOptions(ctx context.Context, filters *filter.QueryFilterSet, opts *filter.QueryOptions, limit, offset int) ([]model.Ledger, *int64, error)

GetAllLedgersWithFilterAndOptions retrieves ledgers with filters, sorting, and optional count.

Parameters: - ctx: Context for the operation. - filters: A QueryFilterSet containing filter conditions. - opts: Query options including sorting and count settings. - limit: Maximum number of ledgers to return. - offset: Offset for pagination.

Returns: - []model.Ledger: A slice of Ledger models matching the filter criteria. - *int64: Optional total count of matching records (if opts.IncludeCount is true). - error: An error if the ledgers could not be retrieved.

func (*LedgerForge) GetAllMonitors 

func (l *LedgerForge) GetAllMonitors(ctx context.Context) ([]model.BalanceMonitor, error)

GetAllMonitors retrieves all balance monitors. It starts a tracing span, fetches all monitors, and records relevant events.

Parameters: - ctx context.Context: The context for the operation.

Returns: - []model.BalanceMonitor: A slice of BalanceMonitor models. - error: An error if the monitors could not be retrieved.

func (*LedgerForge) GetAllTransactions 

func (l *LedgerForge) GetAllTransactions(limit, offset int) ([]model.Transaction, error)

GetAllTransactions retrieves all transactions from the datasource. It starts a tracing span, fetches all transactions, and records relevant events and errors.

Returns: - []model.Transaction: A slice of all retrieved Transaction models. - error: An error if the transactions could not be retrieved.

func (*LedgerForge) GetAllTransactionsWithFilter 

func (l *LedgerForge) GetAllTransactionsWithFilter(ctx context.Context, filters *filter.QueryFilterSet, limit, offset int) ([]model.Transaction, error)

GetAllTransactionsWithFilter retrieves transactions using advanced filters. It starts a tracing span, fetches transactions matching the filter criteria, and records relevant events.

Parameters: - ctx context.Context: The context for the operation. - filters *filter.QueryFilterSet: Filter conditions to apply. - limit int: Maximum number of transactions to return. - offset int: Offset for pagination.

Returns: - []model.Transaction: A slice of Transaction models matching the filter criteria. - error: An error if the transactions could not be retrieved.

func (*LedgerForge) GetAllTransactionsWithFilterAndOptions 

func (l *LedgerForge) GetAllTransactionsWithFilterAndOptions(ctx context.Context, filters *filter.QueryFilterSet, opts *filter.QueryOptions, limit, offset int) ([]model.Transaction, *int64, error)

GetAllTransactionsWithFilterAndOptions retrieves transactions with advanced filters, sorting, and optional count.

func (*LedgerForge) GetBalanceAtTime 

func (l *LedgerForge) GetBalanceAtTime(ctx context.Context, balanceID string, targetTime time.Time, fromSource bool) (*model.Balance, error)

GetBalanceAtTime retrieves a balance's state at a specific point in time. It can either use balance snapshots for efficiency or calculate from all source transactions based on the fromSource parameter.

Parameters: - ctx context.Context: The context for the operation. - balanceID string: The ID of the balance to retrieve. - targetTime time.Time: The point in time for which to retrieve the balance state. - fromSource bool: If true, calculates balance from all transactions instead of using snapshots.

Returns: - *model.Balance: A pointer to the Balance model representing the state at the given time. - error: An error if the historical balance state could not be retrieved.

func (*LedgerForge) GetBalanceByID 

func (l *LedgerForge) GetBalanceByID(ctx context.Context, id string, include []string, withQueued bool) (*model.Balance, error)

GetBalanceByID retrieves a balance by its ID. It starts a tracing span, fetches the balance, and records relevant events.

Parameters: - ctx context.Context: The context for the operation. - id string: The ID of the balance to retrieve. - include []string: A slice of strings specifying additional data to include.

Returns: - *model.Balance: A pointer to the Balance model if found. - error: An error if the balance could not be retrieved.

func (*LedgerForge) GetBalanceByIndicator 

func (l *LedgerForge) GetBalanceByIndicator(ctx context.Context, indicator, currency string) (*model.Balance, error)

GetBalanceByIndicator retrieves a balance by its indicator and currency. It starts a tracing span, fetches the balance, and records relevant events.

Parameters: - ctx context.Context: The context for the operation. - indicator string: The indicator of the balance to retrieve. - currency string: The currency of the balance to retrieve.

Returns: - *model.Balance: A pointer to the Balance model if found. - error: An error if the balance could not be retrieved.

func (*LedgerForge) GetBalanceLineage 

func (l *LedgerForge) GetBalanceLineage(ctx context.Context, balanceID string) (*BalanceLineage, error)

func (*LedgerForge) GetBalanceMonitors 

func (l *LedgerForge) GetBalanceMonitors(ctx context.Context, balanceID string) ([]model.BalanceMonitor, error)

GetBalanceMonitors retrieves all monitors for a given balance ID. It starts a tracing span, fetches the monitors, and records relevant events.

Parameters: - ctx context.Context: The context for the operation. - balanceID string: The ID of the balance for which to retrieve monitors.

Returns: - []model.BalanceMonitor: A slice of BalanceMonitor models. - error: An error if the monitors could not be retrieved.

func (*LedgerForge) GetDataSource 

func (b *LedgerForge) GetDataSource() database.IDataSource

func (*LedgerForge) GetDetokenizedIdentity 

func (l *LedgerForge) GetDetokenizedIdentity(identityID string) (*model.Identity, error)

GetDetokenizedIdentity returns a copy of the identity with all fields detokenized. Note: This does not modify the stored identity.

Parameters: - identityID string: The ID of the identity.

Returns: - *model.Identity: A pointer to the detokenized Identity model. - error: An error if the identity could not be detokenized.

func (*LedgerForge) GetIdentity 

func (l *LedgerForge) GetIdentity(id string) (*model.Identity, error)

GetIdentity retrieves an identity by its ID.

Parameters: - id string: The ID of the identity to retrieve.

Returns: - *model.Identity: A pointer to the Identity model if found. - error: An error if the identity could not be retrieved.

func (*LedgerForge) GetInflightTransactionsByParentID 

func (l *LedgerForge) GetInflightTransactionsByParentID(ctx context.Context, parentTransactionID string, batchSize int, offset int64) ([]*model.Transaction, error)

GetInflightTransactionsByParentID retrieves inflight transactions by their parent transaction ID. It starts a tracing span, fetches the transactions from the datasource, and records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - parentTransactionID string: The ID of the parent transaction. - batchSize int: The number of transactions to retrieve in a batch. - offset int64: The offset for pagination.

Returns: - []*model.Transaction: A slice of pointers to the retrieved Transaction models. - error: An error if the transactions could not be retrieved.

func (*LedgerForge) GetKeyPolicy 

func (l *LedgerForge) GetKeyPolicy(ctx context.Context, key string) (*model.KeyPolicy, error)

func (*LedgerForge) GetLedgerByID 

func (l *LedgerForge) GetLedgerByID(id string) (*model.Ledger, error)

GetLedgerByID retrieves a ledger by its ID from the datasource. It returns a pointer to the Ledger model and an error if the operation fails.

Parameters: - id: A string representing the ID of the ledger to retrieve.

Returns: - *model.Ledger: A pointer to the Ledger model if found. - error: An error if the ledger could not be retrieved.

func (*LedgerForge) GetMatchingRule 

func (b *LedgerForge) GetMatchingRule(ctx context.Context, id string) (*model.MatchingRule, error)

func (*LedgerForge) GetMonitorByID 

func (l *LedgerForge) GetMonitorByID(ctx context.Context, id string) (*model.BalanceMonitor, error)

GetMonitorByID retrieves a balance monitor by its ID. It starts a tracing span, fetches the monitor, and records relevant events.

Parameters: - ctx context.Context: The context for the operation. - id string: The ID of the monitor to retrieve.

Returns: - *model.BalanceMonitor: A pointer to the BalanceMonitor model if found. - error: An error if the monitor could not be retrieved.

func (*LedgerForge) GetReconciliation 

func (b *LedgerForge) GetReconciliation(ctx context.Context, id string) (*model.Reconciliation, error)

func (*LedgerForge) GetRefundableTransactionsByParentID 

func (l *LedgerForge) GetRefundableTransactionsByParentID(ctx context.Context, parentTransactionID string, batchSize int, offset int64) ([]*model.Transaction, error)

func (*LedgerForge) GetReindexProgress 

func (b *LedgerForge) GetReindexProgress() (search.ReindexProgress, error)

func (*LedgerForge) GetSearchClient 

func (b *LedgerForge) GetSearchClient() *search.TypesenseClient

func (*LedgerForge) GetTransaction 

func (l *LedgerForge) GetTransaction(ctx context.Context, TransactionID string) (*model.Transaction, error)

func (*LedgerForge) GetTransactionByRef 

func (l *LedgerForge) GetTransactionByRef(ctx context.Context, reference string) (model.Transaction, error)

GetTransactionByRef retrieves a transaction by its reference from the datasource. It starts a tracing span, fetches the transaction by reference, and records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - reference string: The reference of the transaction to be retrieved.

Returns: - model.Transaction: The retrieved Transaction model. - error: An error if the transaction could not be retrieved.

func (*LedgerForge) GetTransactionLineage 

func (l *LedgerForge) GetTransactionLineage(ctx context.Context, transactionID string) (*TransactionLineage, error)

GetTransactionLineage retrieves the lineage information for a transaction. It returns the fund allocation details and any shadow transactions created for the transaction.

Parameters: - ctx context.Context: The context for the operation. - transactionID string: The ID of the transaction to get lineage for.

Returns: - *TransactionLineage: The lineage information for the transaction. - error: An error if the lineage could not be retrieved.

func (*LedgerForge) ListAPIKeys 

func (l *LedgerForge) ListAPIKeys(ctx context.Context, ownerID string) ([]*model.APIKey, error)

ListAPIKeys retrieves all API keys for a specific owner

Parameters: - ctx: The context for the operation - ownerID: ID of the key owner

Returns: - []*model.APIKey: List of API keys - error: An error if the operation fails

func (*LedgerForge) ListMatchingRules 

func (b *LedgerForge) ListMatchingRules(ctx context.Context) ([]*model.MatchingRule, error)

func (*LedgerForge) ListPolicyApprovals 

func (l *LedgerForge) ListPolicyApprovals(ctx context.Context) ([]model.PolicyAdmission, error)

func (*LedgerForge) MultiSearch 

func (l *LedgerForge) MultiSearch(searchParams *api.MultiSearchSearchesParameter) (*api.MultiSearchResult, error)

MultiSearch performs a multi-search operation across collections.

func (*LedgerForge) PolicyCapacity 

func (l *LedgerForge) PolicyCapacity(ctx context.Context, key string) ([]model.PolicyCapacity, error)

func (*LedgerForge) PrepareLineageOutbox 

func (l *LedgerForge) PrepareLineageOutbox(ctx context.Context, txn *model.Transaction, sourceBalance, destinationBalance *model.Balance) *model.LineageOutbox

func (*LedgerForge) PreviewBulkTransactions 

func (l *LedgerForge) PreviewBulkTransactions(ctx context.Context, request *model.BulkTransactionRequest) (*model.BulkTransactionPreview, error)

PreviewBulkTransactions projects a batch without applying any of it.

How the batch would really execute decides how it is projected: 

skip_queue=true   items are applied inline, one after another, so a later
                  item sees what earlier ones did. The projection carries
                  balances forward across items, which is what catches a
                  batch that spends within itself more than it has.

skip_queue=false  each item is handed to its own goroutine and the batch
                  loop moves on, so the items race and no order is
                  guaranteed. Each item is projected independently against
                  the balances as they stand.

The mode is reported as `cumulative` so a caller can tell which answer they were given.

func (*LedgerForge) PreviewInflightAction 

func (l *LedgerForge) PreviewInflightAction(ctx context.Context, txID, action string, amount *big.Int) (*model.TransactionPreview, error)

PreviewInflightAction projects a commit or void of an inflight transaction without settling it.

txID may name a single inflight transaction or the parent of several legs. A parent is expanded and every leg is projected, so the caller sees the whole balance effect rather than one representative leg's.

Nothing is written: no settlement transaction is recorded, no hold is released, and nothing is queued.

func (*LedgerForge) PreviewRefund 

func (l *LedgerForge) PreviewRefund(ctx context.Context, transactionID string) (*model.TransactionPreview, error)

PreviewRefund projects the reversal of an existing transaction without creating it.

It runs the same lookup a real refund runs — GetRefundableTransactionsByParentID, batch size 1 — rather than a separate hand-rolled eligibility check. That query is a single WHERE clause covering both "does this id exist" and "is it in a refundable status"; a real refund can't and doesn't distinguish the two; either one comes back as an empty result set, reported as "no transaction to refund" (404). A preview with its own, more detailed eligibility check would answer a different question than the one the real endpoint actually answers, and its rejection code would stop matching what a real post returns — the one thing a preview promises. Nothing is written: the refund is never queued and the parent is not marked refunded.

func (*LedgerForge) PreviewTransaction 

func (l *LedgerForge) PreviewTransaction(ctx context.Context, transaction *model.Transaction) (*model.TransactionPreview, error)

PreviewTransaction projects the effect of a transaction on its source and destination balances and returns the result without applying it.

The projection runs the same arithmetic a real post runs — processBalances, and through it UpdateBalances and canProcessTransaction — against deep copies of the balances, so a preview cannot drift from enforcement as the posting path evolves. Nothing on this path changes ledger state: no transaction row, no balance update, no queue entry, no webhook, no hook, and the reference is not consumed.

The answer is advisory. It describes the balances as they stand now, and another transaction may change them before the caller posts for real. Callers that need the funds held should use inflight instead.

func (*LedgerForge) ProcessIndexTask 

func (l *LedgerForge) ProcessIndexTask(ctx context.Context, task IndexTask, upsert IndexUpserter) error

ProcessIndexTask handles a single index-queue message: either upsert an embedded snapshot (legacy) or refresh from the database (metadata paths).

func (*LedgerForge) ProcessLineageFromOutbox 

func (l *LedgerForge) ProcessLineageFromOutbox(ctx context.Context, entry model.LineageOutbox) error

ProcessLineageFromOutbox processes a lineage outbox entry. This is called by the outbox worker to perform deferred lineage processing.

Parameters: - ctx context.Context: The context for the operation. - entry model.LineageOutbox: The outbox entry to process.

Returns: - error: An error if processing fails.

func (*LedgerForge) ProcessQueuedTransaction 

func (l *LedgerForge) ProcessQueuedTransaction(ctx context.Context, transaction *model.Transaction, hotLane bool) (*model.Transaction, error)

func (*LedgerForge) ProcessRefundsInBatches 

func (l *LedgerForge) ProcessRefundsInBatches(ctx context.Context, parentTransactionID string, opts RefundOptions) ([]*model.Transaction, error)

ProcessRefundsInBatches fans out a refund across every refundable leg under parentTransactionID. Legs that were already reversed are skipped; the call succeeds when at least one leg is refunded and fails with conflict only when nothing was left to refund.

func (*LedgerForge) ProcessTransactionInBatches 

func (l *LedgerForge) ProcessTransactionInBatches(ctx context.Context, parentTransactionID string, amount *big.Int, maxWorkers int, streamMode bool, gt getTxns, tw transactionWorker) ([]*model.Transaction, error)

ProcessTransactionInBatches processes transactions in batches or streams them based on the provided mode. It starts a tracing span, initializes worker pools, fetches transactions, and processes them concurrently.

Parameters: - ctx context.Context: The context for the operation. - parentTransactionID string: The ID of the parent transaction. - amount *big.Int: The amount to be processed in the transaction. - maxWorkers int: The maximum number of workers to process transactions concurrently. - streamMode bool: A flag indicating whether to process transactions in streaming mode. - gt getTxns: A function to retrieve transactions in batches. - tw transactionWorker: A function to process transactions.

Returns: - []*model.Transaction: A slice of pointers to the processed Transaction models. - error: An error if the transactions could not be processed.

func (*LedgerForge) ProcessWebhook 

func (b *LedgerForge) ProcessWebhook(ctx context.Context, task *asynq.Task) error

ProcessWebhook processes a webhook notification task from the queue.

Parameters: - _ context.Context: The context for the operation. - task *asynq.Task: The task containing the webhook notification data.

Returns: - error: An error if the webhook processing fails.

func (*LedgerForge) PutKeyPolicy 

func (l *LedgerForge) PutKeyPolicy(ctx context.Context, p model.KeyPolicy) (*model.KeyPolicy, error)

func (*LedgerForge) QueueInflightAction 

func (l *LedgerForge) QueueInflightAction(ctx context.Context, transactionID string, amount *big.Int, action string) (*model.Transaction, error)

QueueInflightAction pre-validates a commit/void and enqueues it for the worker. It returns the still-inflight parent transaction (for the API response) and ErrInflightActionQueued if an action for the same transaction is already queued.

func (*LedgerForge) QueueTransaction 

func (l *LedgerForge) QueueTransaction(ctx context.Context, transaction *model.Transaction) (*model.Transaction, error)

QueueTransaction processes and queues a transaction for execution. It handles both single transactions and split transactions, preparing them for processing by setting metadata, status, and managing their persistence and queueing.

Parameters: - ctx context.Context: The context for the operation. - transaction *model.Transaction: The transaction to be queued.

Returns: - *model.Transaction: A pointer to the queued Transaction model. - error: An error if the transaction could not be queued.

func (*LedgerForge) RecordTransaction 

func (l *LedgerForge) RecordTransaction(ctx context.Context, transaction *model.Transaction) (*model.Transaction, error)

RecordTransaction records a transaction by validating, processing balances, and finalizing the transaction. It starts a tracing span, acquires a lock, and performs the necessary steps to record the transaction.

Parameters: - ctx context.Context: The context for the operation. - transaction *model.Transaction: The transaction to be recorded.

Returns: - *model.Transaction: A pointer to the recorded Transaction model. - error: An error if the transaction could not be recorded.

func (*LedgerForge) RecoverQueuedTransactions 

func (b *LedgerForge) RecoverQueuedTransactions(ctx context.Context, threshold time.Duration) (int, error)

RecoverQueuedTransactions triggers an immediate recovery of stuck queued transactions using the provided threshold. This is exposed for the manual trigger API endpoint. recovered is the number of stuck rows successfully resolved; err is non-nil when any row in the batch could not be recovered.

func (*LedgerForge) RefundTransaction 

func (l *LedgerForge) RefundTransaction(ctx context.Context, transactionID string, skipQueue bool) (*model.Transaction, error)

RefundTransaction processes a refund for a given transaction by its ID. It starts a tracing span, retrieves the original transaction, validates its status, creates a new refund transaction, and queues it.

Parameters: - ctx context.Context: The context for the operation. - transactionID string: The ID of the transaction to be refunded.

Returns: - *model.Transaction: A pointer to the refunded Transaction model. - error: An error if the transaction could not be refunded.

func (*LedgerForge) RefundTransactionWithOptions 

func (l *LedgerForge) RefundTransactionWithOptions(ctx context.Context, transactionID string, opts RefundOptions) (*model.Transaction, error)

RefundTransactionWithOptions refunds a transaction, letting the caller set the reversal's description and metadata rather than inheriting the original's.

RefundTransaction is the same call with only SkipQueue set.

func (*LedgerForge) RefundWorker 

func (l *LedgerForge) RefundWorker(ctx context.Context, jobs <-chan *model.Transaction, results chan<- BatchJobResult, wg *sync.WaitGroup, amount *big.Int)

func (*LedgerForge) RefundWorkerWithOptions 

func (l *LedgerForge) RefundWorkerWithOptions(skipQueue bool) transactionWorker

func (*LedgerForge) RefundWorkerWithRefundOptions 

func (l *LedgerForge) RefundWorkerWithRefundOptions(opts RefundOptions) transactionWorker

RefundWorkerWithOptions returns a transactionWorker that refunds each job with an explicit skipQueue decision, rather than inheriting it from the original transaction.

The caller can choose synchronous processing for the reversal independently of the immutable routing decision on the original transaction.

RefundWorker is left unchanged; this is purely additive. RefundWorkerWithRefundOptions returns a transactionWorker that refunds each job using the supplied options, so a caller can set the reversal's description and metadata as well as its queueing.

RefundWorkerWithOptions is the same worker with only SkipQueue set.

func (*LedgerForge) RejectTransaction 

func (l *LedgerForge) RejectTransaction(ctx context.Context, transaction *model.Transaction, reason string) (*model.Transaction, error)

func (*LedgerForge) RevokeAPIKey 

func (l *LedgerForge) RevokeAPIKey(ctx context.Context, id, ownerID string) error

RevokeAPIKey revokes an API key if it belongs to the specified owner

Parameters: - ctx: The context for the operation - id: ID of the API key to revoke - ownerID: ID of the key owner

Returns: - error: An error if the operation fails

func (*LedgerForge) RunInflightActionByParent 

func (l *LedgerForge) RunInflightActionByParent(ctx context.Context, parentTransactionID, action string, amount *big.Int, actionID string) (retryable bool, err error)

RunInflightActionByParent commits or voids every inflight leg under a parent transaction — the same expansion the synchronous endpoint performs — and is the queued worker's entrypoint.

retryable is true when at least one leg failed with a transient error (lock contention, fetch failure, not-yet-visible), so the asynq task should retry; already-finalized legs are terminal, skipped, and never block the ack.

func (*LedgerForge) Search 

func (l *LedgerForge) Search(collection string, query *api.SearchCollectionParams) (interface{}, error)

Search performs a search on the specified collection using the provided query parameters.

Parameters: - collection string: The name of the collection to search. - query *api.SearchCollectionParams: The search query parameters.

Returns: - interface{}: The search results. - error: An error if the search operation fails.

func (*LedgerForge) SendWebhook 

func (b *LedgerForge) SendWebhook(newWebhook NewWebhook) error

SendWebhook enqueues a webhook notification task using the LedgerForge instance's asynq client.

Parameters: - newWebhook NewWebhook: The webhook notification data to enqueue.

Returns: - error: An error if the task could not be enqueued.

func (*LedgerForge) StartInstantReconciliation 

func (b *LedgerForge) StartInstantReconciliation(ctx context.Context, transactions []model.ExternalTransaction, strategy, groupCriteria string, matchingRuleIDs []string, isDryRun bool) (string, error)

func (*LedgerForge) StartReconciliation 

func (b *LedgerForge) StartReconciliation(ctx context.Context, uploadID, strategy, groupCriteria string, matchingRuleIDs []string, isDryRun bool) (string, error)

func (*LedgerForge) StartReindex 

func (b *LedgerForge) StartReindex(batchSize int) (search.ReindexProgress, error)

func (*LedgerForge) TakeBalanceSnapshots 

func (l *LedgerForge) TakeBalanceSnapshots(ctx context.Context, batchSize int)

TakeBalanceSnapshots creates daily snapshots of balances in batches. It accepts a batch size parameter to control the number of balances processed at once, helping to manage memory usage for large datasets.

Parameters: - ctx context.Context: The context for managing the operation's lifecycle and cancellation - batchSize int: The number of balances to process in each batch

Returns: - int: The total number of snapshots created - error: An error if the snapshot creation fails

func (*LedgerForge) TokenizeAllPII 

func (l *LedgerForge) TokenizeAllPII(identityID string) error

TokenizeAllPII tokenizes all eligible PII fields in an identity.

Parameters: - identityID string: The ID of the identity.

Returns: - error: An error if any field could not be tokenized.

func (*LedgerForge) TokenizeIdentity 

func (l *LedgerForge) TokenizeIdentity(identityID string, fields []string) error

TokenizeIdentity tokenizes all specified fields in an identity.

Parameters: - identityID string: The ID of the identity. - fields []string: The names of the fields to tokenize.

Returns: - error: An error if any field could not be tokenized.

func (*LedgerForge) TokenizeIdentityField 

func (l *LedgerForge) TokenizeIdentityField(identityID, fieldName string) error

TokenizeIdentityField tokenizes a specific field in an identity.

Parameters: - identityID string: The ID of the identity. - fieldName string: The name of the field to tokenize.

Returns: - error: An error if the field could not be tokenized.

func (*LedgerForge) TryRecordQueuedTransactionBatch 

func (l *LedgerForge) TryRecordQueuedTransactionBatch(ctx context.Context, transaction *model.Transaction) (handled bool, err error)

func (*LedgerForge) TryRecordQueuedTransactionBatchForHotLane 

func (l *LedgerForge) TryRecordQueuedTransactionBatchForHotLane(ctx context.Context, transaction *model.Transaction) (handled bool, err error)

func (*LedgerForge) TryReindexIfNeeded 

func (b *LedgerForge) TryReindexIfNeeded(ctx context.Context)

TryReindexIfNeeded starts a rebuild on an empty search index using this instance's job coordinator.

func (*LedgerForge) UpdateBalanceIdentity 

func (l *LedgerForge) UpdateBalanceIdentity(balanceID, identityID string) error

UpdateBalanceIdentity updates only the identity_id associated with a balance. It validates that both the balance and the identity exist before applying the change.

Parameters: - balanceID string: The ID of the balance whose identity reference should be modified. - identityID string: The new identity ID to associate with the balance.

Returns: - error: An error is returned if either the balance or identity records are not found or the update fails.

func (*LedgerForge) UpdateIdentity 

func (l *LedgerForge) UpdateIdentity(identity *model.Identity) error

UpdateIdentity updates an existing identity in the database. When the update writes meta_data, it reloads the stored identity and enqueues identity.metadata.updated with that full record and the meta_data this request committed. Field-only updates with no meta_data do not emit that event. A failed reload or enqueue does not fail the update; the write is already committed.

Parameters: - identity *model.Identity: A pointer to the Identity model to be updated.

Returns: - error: An error if the identity could not be updated.

func (*LedgerForge) UpdateLastUsed 

func (l *LedgerForge) UpdateLastUsed(ctx context.Context, id string) error

UpdateLastUsed updates the last used timestamp of an API key

Parameters: - ctx: The context for the operation - id: ID of the API key to update

Returns: - error: An error if the operation fails

func (*LedgerForge) UpdateLedger 

func (l *LedgerForge) UpdateLedger(id, name string) (*model.Ledger, error)

UpdateLedger updates an existing ledger's name. It calls postLedgerActions after a successful update to handle indexing and webhooks.

Parameters: - id: A string representing the ID of the ledger to update. - name: A string representing the new name for the ledger.

Returns: - *model.Ledger: A pointer to the updated Ledger model. - error: An error if the ledger could not be updated.

func (*LedgerForge) UpdateMatchingRule 

func (b *LedgerForge) UpdateMatchingRule(ctx context.Context, rule model.MatchingRule) (*model.MatchingRule, error)

func (*LedgerForge) UpdateMetadata 

func (l *LedgerForge) UpdateMetadata(ctx context.Context, entityID string, newMetadata map[string]interface{}) (map[string]interface{}, error)

UpdateMetadata updates the metadata for a given entity ID.

Ordering per branch is: load current state, persist the merge, then emit side effects. Once the merge is persisted the call succeeds; the returned error only ever describes a failure to look up or persist, never a failure to publish. Both side effects are best-effort:

  • the *.metadata.updated webhook is enqueued to Redis (see enqueueMetadataUpdatedWebhook for the delivery guarantee), and
  • the Typesense reindex is queued asynchronously.

Parameters: - ctx: The context for the operation. - entityID: A string representing the ID of the entity to update. - newMetadata: A map containing the new metadata to merge.

Returns: - map[string]interface{}: The merged metadata after the update. - error: An error if the entity cannot be found or the merge cannot be persisted.

func (*LedgerForge) UpdateMonitor 

func (l *LedgerForge) UpdateMonitor(ctx context.Context, monitor *model.BalanceMonitor) error

UpdateMonitor updates an existing balance monitor. It starts a tracing span, updates the monitor, and records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - monitor *model.BalanceMonitor: A pointer to the BalanceMonitor model to be updated.

Returns: - error: An error if the monitor could not be updated.

func (*LedgerForge) UploadExternalData 

func (b *LedgerForge) UploadExternalData(ctx context.Context, source string, reader io.Reader, filename string) (string, int, error)

func (*LedgerForge) VerifyChain 

func (l *LedgerForge) VerifyChain(ctx context.Context, progress func(seq int64)) (ChainVerifyResult, error)

VerifyChain replays the whole chain from genesis, checking that each row's chain_seq is contiguous, its chain_prev_hash matches the running head, and its recomputed hash matches the stored chain_hash. It reports the first broken link, or success with the final head. progress (optional) is called with each verified seq.

func (*LedgerForge) VoidInflightTransaction 

func (l *LedgerForge) VoidInflightTransaction(ctx context.Context, transactionID string) (*model.Transaction, error)

VoidInflightTransaction voids an inflight transaction by validating it, calculating the remaining amount, and finalizing the void. It starts a tracing span, fetches and validates the inflight transaction, calculates the remaining amount, and finalizes the void.

Parameters: - ctx context.Context: The context for the operation. - transactionID string: The ID of the inflight transaction to be voided.

Returns: - *model.Transaction: A pointer to the voided Transaction model. - error: An error if the transaction could not be voided.

func (*LedgerForge) VoidInflightTransactionWithRef 

func (l *LedgerForge) VoidInflightTransactionWithRef(ctx context.Context, transactionID string, reference string) (*model.Transaction, error)

VoidInflightTransactionWithRef voids an inflight transaction using a caller-supplied reference for the voided child, making asynq retries idempotent. An empty reference preserves the legacy random-reference behavior.

func (*LedgerForge) VoidWorker 

func (l *LedgerForge) VoidWorker(ctx context.Context, jobs <-chan *model.Transaction, results chan<- BatchJobResult, wg *sync.WaitGroup, amount *big.Int)

VoidWorker processes void transactions from the jobs channel and sends the results to the results channel. It starts a tracing span, processes each transaction, and records relevant events and errors.

Parameters: - ctx context.Context: The context for the operation. - jobs <-chan *model.Transaction: A channel from which transactions are received for processing. - results chan<- BatchJobResult: A channel to which the results of the processing are sent. - wg *sync.WaitGroup: A wait group to synchronize the completion of the worker. - amount *big.Int: The amount to be processed in the transaction.

type LineageOutboxPayload 

type LineageOutboxPayload struct {
	Amount        float64 `json:"amount"`
	PreciseAmount string  `json:"precise_amount"`
	Currency      string  `json:"currency"`
	Precision     float64 `json:"precision"`
	Reference     string  `json:"reference"`
	SkipQueue     bool    `json:"skip_queue"`
	Inflight      bool    `json:"inflight"`
}

LineageOutboxPayload contains the transaction data needed for deferred lineage processing.

type LineageOutboxProcessor 

type LineageOutboxProcessor struct {
	// contains filtered or unexported fields
}

LineageOutboxProcessor processes pending lineage outbox entries. It polls the database for pending entries and processes them asynchronously, ensuring that lineage work is never lost even if Redis/queue operations fail.

func NewLineageOutboxProcessor 

func NewLineageOutboxProcessor(ledgerforge *LedgerForge) *LineageOutboxProcessor

NewLineageOutboxProcessor creates a new outbox processor.

Parameters: - ledgerforge *LedgerForge: The LedgerForge instance containing the datasource and processing logic.

Returns: - *LineageOutboxProcessor: The configured processor.

func (*LineageOutboxProcessor) IsRunning 

func (p *LineageOutboxProcessor) IsRunning() bool

IsRunning returns whether the processor is currently running.

Returns: - bool: True if the processor is running.

func (*LineageOutboxProcessor) Start 

func (p *LineageOutboxProcessor) Start(ctx context.Context)

Start begins processing outbox entries in the background. The processor will poll for pending entries at the configured interval.

Parameters: - ctx context.Context: The context for the operation. When cancelled, processing stops.

func (*LineageOutboxProcessor) Stop 

func (p *LineageOutboxProcessor) Stop()

Stop gracefully stops the outbox processor. It signals the processor to stop and waits for pending work to complete.

func (*LineageOutboxProcessor) WithBatchSize 

func (p *LineageOutboxProcessor) WithBatchSize(size int) *LineageOutboxProcessor

WithBatchSize sets the batch size for processing outbox entries.

Parameters: - size int: The number of entries to process per batch.

Returns: - *LineageOutboxProcessor: The processor for chaining.

func (*LineageOutboxProcessor) WithLockDuration 

func (p *LineageOutboxProcessor) WithLockDuration(duration time.Duration) *LineageOutboxProcessor

WithLockDuration sets the lock duration for claimed entries.

Parameters: - duration time.Duration: The lock duration.

Returns: - *LineageOutboxProcessor: The processor for chaining.

func (*LineageOutboxProcessor) WithPollInterval 

func (p *LineageOutboxProcessor) WithPollInterval(interval time.Duration) *LineageOutboxProcessor

WithPollInterval sets the interval between polls.

Parameters: - interval time.Duration: The poll interval.

Returns: - *LineageOutboxProcessor: The processor for chaining.

type LineageSource 

type LineageSource struct {
	BalanceID string
	Balance   *big.Int
	CreatedAt time.Time
}

LineageSource represents a source of funds available for allocation during lineage debit processing.

Fields: - BalanceID string: The ID of the shadow balance holding the funds. - Balance *big.Int: The available balance amount. - CreatedAt time.Time: The creation time of the lineage mapping, used for FIFO/LIFO ordering.

type NewWebhook 

type NewWebhook struct {
	Event   string      `json:"event"` // The event type that triggered the webhook.
	Payload interface{} `json:"data"`  // The data associated with the event.
}

NewWebhook represents the structure of a webhook notification. It includes an event type and associated payload data.

type ProviderBreakdown 

type ProviderBreakdown struct {
	Provider  string   `json:"provider"`
	Amount    *big.Int `json:"amount"`
	Available *big.Int `json:"available"`
	Spent     *big.Int `json:"spent"`
	BalanceID string   `json:"shadow_balance_id"`
}

ProviderBreakdown represents the fund breakdown for a specific provider in a balance's lineage.

Fields: - Provider string: The name/identifier of the fund provider. - Amount *big.Int: The total amount received from this provider. - Available *big.Int: The amount still available (not yet spent). - Spent *big.Int: The amount that has been debited. - BalanceID string: The ID of the shadow balance tracking this provider's funds.

type Queue 

type Queue struct {
	Client    *asynq.Client
	Inspector *asynq.Inspector
	// contains filtered or unexported fields
}

Queue represents a queue for handling various tasks.

func NewQueue 

func NewQueue(conf *config.Configuration, client *asynq.Client, redisClients ...redis.UniversalClient) *Queue

NewQueue initializes a new Queue instance with the provided configuration.

Parameters: - conf *config.Configuration: The configuration for the queue.

Returns: - *Queue: A pointer to the newly created Queue instance.

func (*Queue) Enqueue 

func (q *Queue) Enqueue(ctx context.Context, transaction *model.Transaction) error

Enqueue enqueues a transaction to the Redis queue.

Parameters: - ctx context.Context: The context for the operation. - transaction *model.Transaction: The transaction to be enqueued.

Returns: - error: An error if the transaction could not be enqueued.

func (*Queue) EnqueueInflightAction 

func (q *Queue) EnqueueInflightAction(ctx context.Context, p InflightActionPayload) error

EnqueueInflightAction enqueues a commit or void to the inflight-commit queue. The asynq TaskID dedups concurrent actions for the same transaction: a second enqueue while one is pending/active returns ErrInflightActionQueued. The "inflight-action:" prefix is intentionally distinct from the scheduled auto-commit TaskID (the bare transaction ID) so a pre-scheduled inflight_commit_date task does not block a manual commit/void.

func (*Queue) GetTransactionFromQueue 

func (q *Queue) GetTransactionFromQueue(transactionID string) (*model.Transaction, error)

GetTransactionFromQueue retrieves a transaction from the queue by its ID.

Parameters: - transactionID string: The ID of the transaction to retrieve.

Returns: - *model.Transaction: A pointer to the Transaction model if found. - error: An error if the transaction could not be retrieved.

func (*Queue) QueueInflightCommit 

func (q *Queue) QueueInflightCommit(ctx context.Context, transaction *model.Transaction) error

QueueInflightCommit schedules an automatic commit for an inflight transaction at the specified date.

Parameters: - ctx context.Context: The context for the operation. - transaction *model.Transaction: The transaction to be committed automatically.

Returns: - error: An error if the task could not be enqueued.

func (*Queue) QueueInflightExpiry 

func (q *Queue) QueueInflightExpiry(ctx context.Context, transaction *model.Transaction) error

QueueInflightExpiry handles queuing a transaction for inflight expiration. This method is separate from the main Enqueue to ensure expiration is handled regardless of whether the transaction is queued or processed immediately.

Parameters: - ctx context.Context: The context for the operation. - transaction *model.Transaction: The transaction to queue for expiration.

Returns: - error: An error if the expiration could not be queued.

type QueuedTransactionRecoveryProcessor 

type QueuedTransactionRecoveryProcessor struct {
	// contains filtered or unexported fields
}

func NewQueuedTransactionRecoveryProcessor 

func NewQueuedTransactionRecoveryProcessor(ledgerforge *LedgerForge) *QueuedTransactionRecoveryProcessor

NewQueuedTransactionRecoveryProcessor creates the stuck queued-transaction recovery loop with conservative single-worker defaults to avoid recovery-induced lock storms.

func (*QueuedTransactionRecoveryProcessor) IsRunning 

func (p *QueuedTransactionRecoveryProcessor) IsRunning() bool

IsRunning reports whether the recovery processor is currently active.

func (*QueuedTransactionRecoveryProcessor) Start 

func (p *QueuedTransactionRecoveryProcessor) Start(ctx context.Context)

Start begins the background recovery loop for stuck queued transactions.

func (*QueuedTransactionRecoveryProcessor) Stop 

func (p *QueuedTransactionRecoveryProcessor) Stop()

Stop shuts down the background recovery loop and waits for the worker goroutine to exit.

type RefundOptions 

type RefundOptions struct {
	// SkipQueue processes the refund synchronously instead of placing it on
	// the transaction queue.
	SkipQueue bool

	// Description replaces the reversal's description, which otherwise copies
	// the original transaction's.
	Description string

	// MetaData is merged onto the metadata inherited from the original
	// transaction, with these keys winning on conflict. Merging rather than
	// replacing preserves caller metadata. Reservation flags are cleared
	// because a refund always reverses committed funds.
	MetaData map[string]interface{}
}

RefundOptions carries the caller-supplied settings for a refund.

The zero value reproduces the historical behaviour: queued asynchronously, with the reversal inheriting the original transaction's description and metadata.

type TransactionLineage 

type TransactionLineage struct {
	TransactionID      string                   `json:"transaction_id"`
	FundAllocation     []map[string]interface{} `json:"fund_allocation,omitempty"`
	ShadowTransactions []model.Transaction      `json:"shadow_transactions"`
}

TransactionLineage represents the lineage information for a specific transaction.

Fields: - TransactionID string: The ID of the transaction being queried. - FundAllocation []map[string]interface{}: The allocation of funds by provider for debit transactions. - ShadowTransactions []model.Transaction: The shadow transactions created for this transaction.

type TransactionTypePayload 

type TransactionTypePayload struct {
	Data model.Transaction
}

TransactionTypePayload represents the payload for a transaction type.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.