343 lines
10 KiB
Go
343 lines
10 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"gl/application/explorer"
|
|
"gl/domain/ledger"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
var ErrNotFound = ledger.ErrNotFound
|
|
|
|
type JournalFilter = ledger.JournalFilter
|
|
|
|
func (r *JournalRepository) GetByID(ctx context.Context, journalID string) (ledger.Journal, error) {
|
|
return r.get(ctx, getJournalByIDSQL, journalID)
|
|
}
|
|
|
|
func (r *JournalRepository) GetByIdempotencyKey(ctx context.Context, idempotencyKey string) (ledger.Journal, error) {
|
|
return r.get(ctx, getJournalByIdempotencySQL, idempotencyKey)
|
|
}
|
|
|
|
func (r *JournalRepository) get(ctx context.Context, query string, value string) (ledger.Journal, error) {
|
|
journal, err := scanJournal(r.database.QueryRow(ctx, query, value))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ledger.Journal{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return ledger.Journal{}, fmt.Errorf("read journal: %w", err)
|
|
}
|
|
journal.Entries, err = r.loadEntries(ctx, journal.ID)
|
|
if err != nil {
|
|
return ledger.Journal{}, err
|
|
}
|
|
return journal, nil
|
|
}
|
|
|
|
func (r *JournalRepository) List(ctx context.Context, filter JournalFilter) ([]ledger.Journal, error) {
|
|
limit := filter.Limit
|
|
if limit <= 0 || limit > 201 {
|
|
limit = 50
|
|
}
|
|
if filter.Offset < 0 {
|
|
filter.Offset = 0
|
|
}
|
|
|
|
var accountClass, ownerType, ownerID any
|
|
if filter.Account != nil {
|
|
accountClass = filter.Account.Class
|
|
ownerType = filter.Account.OwnerType
|
|
ownerID = filter.Account.OwnerID
|
|
}
|
|
return r.queryJournals(ctx, listJournalsSQL,
|
|
filter.AssetID,
|
|
accountClass,
|
|
ownerType,
|
|
ownerID,
|
|
filter.OwnerID,
|
|
filter.EffectKind,
|
|
filter.RecordedFrom,
|
|
filter.RecordedTo,
|
|
limit,
|
|
filter.Offset,
|
|
)
|
|
}
|
|
|
|
func (r *JournalRepository) GetByTransactionHash(ctx context.Context, hash string) ([]ledger.Journal, error) {
|
|
return r.queryJournals(ctx, getJournalsByTransactionHashSQL, hash)
|
|
}
|
|
|
|
func (r *JournalRepository) Stats(ctx context.Context) (explorer.Stats, error) {
|
|
var stats explorer.Stats
|
|
if err := r.database.QueryRow(ctx, getExplorerStatsSQL).Scan(
|
|
&stats.JournalCount,
|
|
&stats.EntryCount,
|
|
&stats.AccountCount,
|
|
&stats.LastRecordedAt,
|
|
); err != nil {
|
|
return explorer.Stats{}, fmt.Errorf("read explorer stats: %w", err)
|
|
}
|
|
return stats, nil
|
|
}
|
|
|
|
func (r *JournalRepository) TopHolders(ctx context.Context, assetLimit, holderLimit int) ([]explorer.Holder, error) {
|
|
if assetLimit <= 0 || assetLimit > 20 {
|
|
assetLimit = 5
|
|
}
|
|
if holderLimit <= 0 || holderLimit > 20 {
|
|
holderLimit = 5
|
|
}
|
|
rows, err := r.database.Query(ctx, topHoldersSQL, assetLimit, holderLimit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list top holders: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
holders := make([]explorer.Holder, 0, assetLimit*holderLimit)
|
|
for rows.Next() {
|
|
var holder explorer.Holder
|
|
var balance string
|
|
if err := rows.Scan(&holder.Rank, &holder.OwnerType, &holder.OwnerID, &holder.AssetID, &balance); err != nil {
|
|
return nil, fmt.Errorf("scan top holder: %w", err)
|
|
}
|
|
holder.Balance, err = ledger.ParseAmount(balance)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode top holder balance: %w", err)
|
|
}
|
|
holders = append(holders, holder)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate top holders: %w", err)
|
|
}
|
|
return holders, nil
|
|
}
|
|
|
|
func (r *JournalRepository) queryJournals(ctx context.Context, query string, args ...any) ([]ledger.Journal, error) {
|
|
rows, err := r.database.Query(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list journals: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
journals := make([]ledger.Journal, 0)
|
|
for rows.Next() {
|
|
journal, scanErr := scanJournal(rows)
|
|
if scanErr != nil {
|
|
return nil, fmt.Errorf("scan journal: %w", scanErr)
|
|
}
|
|
journals = append(journals, journal)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate journals: %w", err)
|
|
}
|
|
rows.Close()
|
|
|
|
for index := range journals {
|
|
journals[index].Entries, err = r.loadEntries(ctx, journals[index].ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return journals, nil
|
|
}
|
|
|
|
func (r *JournalRepository) Balance(ctx context.Context, account ledger.AccountReference, asOf time.Time) (ledger.Amount, error) {
|
|
if err := account.Validate(); err != nil {
|
|
return ledger.Amount{}, err
|
|
}
|
|
var asOfValue any
|
|
if !asOf.IsZero() {
|
|
asOfValue = asOf
|
|
}
|
|
var value string
|
|
if err := r.database.QueryRow(ctx, getBalanceSQL,
|
|
account.Class,
|
|
account.OwnerType,
|
|
account.OwnerID,
|
|
account.AssetID,
|
|
asOfValue,
|
|
).Scan(&value); err != nil {
|
|
return ledger.Amount{}, fmt.Errorf("read balance: %w", err)
|
|
}
|
|
amount, err := ledger.ParseAmount(value)
|
|
if err != nil {
|
|
return ledger.Amount{}, fmt.Errorf("decode balance: %w", err)
|
|
}
|
|
return amount, nil
|
|
}
|
|
|
|
type scanner interface {
|
|
Scan(...any) error
|
|
}
|
|
|
|
func scanJournal(row scanner) (ledger.Journal, error) {
|
|
var (
|
|
journal ledger.Journal
|
|
reversal *string
|
|
metadata []byte
|
|
)
|
|
err := row.Scan(
|
|
&journal.ID,
|
|
&journal.SourceService,
|
|
&journal.IdempotencyKey,
|
|
&journal.SourceTransactionID,
|
|
&journal.TrackingCode,
|
|
&journal.EffectKind,
|
|
&journal.EventVersion,
|
|
&reversal,
|
|
&journal.OccurredAt,
|
|
&journal.RecordedAt,
|
|
&journal.CorrelationID,
|
|
&journal.ActorID,
|
|
&journal.Blockchain.Network,
|
|
&journal.Blockchain.TransactionHash,
|
|
&journal.Blockchain.LedgerSequence,
|
|
&metadata,
|
|
&journal.PayloadHash,
|
|
)
|
|
if err != nil {
|
|
return ledger.Journal{}, err
|
|
}
|
|
if reversal != nil {
|
|
journal.ReversalOfJournalID = *reversal
|
|
}
|
|
if err := json.Unmarshal(metadata, &journal.Metadata); err != nil {
|
|
return ledger.Journal{}, fmt.Errorf("decode metadata: %w", err)
|
|
}
|
|
return journal, nil
|
|
}
|
|
|
|
func (r *JournalRepository) loadEntries(ctx context.Context, journalID string) ([]ledger.Entry, error) {
|
|
rows, err := r.database.Query(ctx, getEntriesSQL, journalID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("read journal entries: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
entries := make([]ledger.Entry, 0)
|
|
for rows.Next() {
|
|
var (
|
|
entry ledger.Entry
|
|
amountValue string
|
|
)
|
|
if err := rows.Scan(
|
|
&entry.LineNumber,
|
|
&entry.Account.Class,
|
|
&entry.Account.OwnerType,
|
|
&entry.Account.OwnerID,
|
|
&entry.Account.AssetID,
|
|
&amountValue,
|
|
&entry.Description,
|
|
); err != nil {
|
|
return nil, fmt.Errorf("scan journal entry: %w", err)
|
|
}
|
|
entry.Amount, err = ledger.ParseAmount(amountValue)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode journal entry amount: %w", err)
|
|
}
|
|
entries = append(entries, entry)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, fmt.Errorf("iterate journal entries: %w", err)
|
|
}
|
|
return entries, nil
|
|
}
|
|
|
|
const journalColumns = `
|
|
j.id, j.source_service, j.idempotency_key, j.source_transaction_id,
|
|
j.tracking_code, j.effect_kind, j.event_version, j.reversal_of_journal_id,
|
|
j.occurred_at, j.recorded_at, j.correlation_id, j.actor_id,
|
|
j.blockchain_network, j.blockchain_transaction_hash,
|
|
j.blockchain_ledger_sequence, j.metadata, j.payload_hash`
|
|
|
|
const getJournalByIDSQL = `SELECT ` + journalColumns + `
|
|
FROM journals j WHERE j.id = $1 AND j.sealed_at IS NOT NULL`
|
|
|
|
const getJournalByIdempotencySQL = `SELECT ` + journalColumns + `
|
|
FROM journals j WHERE j.idempotency_key = $1 AND j.sealed_at IS NOT NULL`
|
|
|
|
const getJournalsByTransactionHashSQL = `SELECT ` + journalColumns + `
|
|
FROM journals j
|
|
WHERE j.blockchain_transaction_hash = $1 AND j.sealed_at IS NOT NULL
|
|
ORDER BY j.recorded_at DESC, j.id DESC`
|
|
|
|
const topHoldersSQL = `WITH recent_assets AS (
|
|
SELECT e.asset_id, MAX(j.recorded_at) AS last_activity
|
|
FROM journal_entries e
|
|
JOIN journals j ON j.id = e.journal_id
|
|
WHERE j.sealed_at IS NOT NULL
|
|
GROUP BY e.asset_id
|
|
ORDER BY last_activity DESC, e.asset_id
|
|
LIMIT $1
|
|
), holder_balances AS (
|
|
SELECT a.owner_type, a.owner_id, e.asset_id, SUM(e.amount) AS balance
|
|
FROM journal_entries e
|
|
JOIN journals j ON j.id = e.journal_id AND j.sealed_at IS NOT NULL
|
|
JOIN ledger_accounts a ON a.id = e.account_id
|
|
JOIN recent_assets ra ON ra.asset_id = e.asset_id
|
|
WHERE a.class IN ('USER_AVAILABLE', 'USER_FROZEN')
|
|
GROUP BY a.owner_type, a.owner_id, e.asset_id
|
|
HAVING SUM(e.amount) > 0
|
|
), ranked_holders AS (
|
|
SELECT owner_type, owner_id, asset_id, balance,
|
|
ROW_NUMBER() OVER (
|
|
PARTITION BY asset_id
|
|
ORDER BY balance DESC, owner_type, owner_id
|
|
) AS holder_rank
|
|
FROM holder_balances
|
|
)
|
|
SELECT rh.holder_rank, rh.owner_type, rh.owner_id, rh.asset_id, rh.balance::text
|
|
FROM ranked_holders rh
|
|
JOIN recent_assets ra ON ra.asset_id = rh.asset_id
|
|
WHERE rh.holder_rank <= $2
|
|
ORDER BY ra.last_activity DESC, rh.asset_id, rh.holder_rank`
|
|
|
|
const listJournalsSQL = `SELECT DISTINCT ` + journalColumns + `
|
|
FROM journals j
|
|
JOIN journal_entries e ON e.journal_id = j.id
|
|
JOIN ledger_accounts a ON a.id = e.account_id
|
|
WHERE j.sealed_at IS NOT NULL
|
|
AND ($1::bigint IS NULL OR e.asset_id = $1)
|
|
AND ($2::text IS NULL OR (
|
|
a.class = $2 AND a.owner_type = $3 AND a.owner_id = $4
|
|
))
|
|
AND ($5::text IS NULL OR (
|
|
a.class IN ('USER_AVAILABLE', 'USER_FROZEN') AND a.owner_id = $5
|
|
))
|
|
AND ($6::text IS NULL OR lower(j.effect_kind) = lower($6))
|
|
AND ($7::timestamptz IS NULL OR j.recorded_at >= $7)
|
|
AND ($8::timestamptz IS NULL OR j.recorded_at <= $8)
|
|
ORDER BY j.recorded_at DESC, j.id DESC
|
|
LIMIT $9 OFFSET $10`
|
|
|
|
const getEntriesSQL = `
|
|
SELECT e.line_number, a.class, a.owner_type, a.owner_id, e.asset_id,
|
|
e.amount::text, e.description
|
|
FROM journal_entries e
|
|
JOIN ledger_accounts a ON a.id = e.account_id
|
|
WHERE e.journal_id = $1
|
|
ORDER BY e.line_number`
|
|
|
|
const getBalanceSQL = `
|
|
SELECT COALESCE(sum(e.amount), 0)::text
|
|
FROM journal_entries e
|
|
JOIN ledger_accounts a ON a.id = e.account_id
|
|
JOIN journals j ON j.id = e.journal_id
|
|
WHERE j.sealed_at IS NOT NULL
|
|
AND a.class = $1 AND a.owner_type = $2 AND a.owner_id = $3
|
|
AND a.asset_id = $4
|
|
AND ($5::timestamptz IS NULL OR j.recorded_at <= $5)`
|
|
|
|
const getExplorerStatsSQL = `
|
|
SELECT
|
|
(SELECT count(*) FROM journals WHERE sealed_at IS NOT NULL),
|
|
(SELECT count(*) FROM journal_entries e JOIN journals j ON j.id = e.journal_id WHERE j.sealed_at IS NOT NULL),
|
|
(SELECT count(*) FROM ledger_accounts),
|
|
COALESCE((SELECT max(recorded_at) FROM journals WHERE sealed_at IS NOT NULL), 'epoch'::timestamptz)`
|