package db import ( "context" "crypto/rand" "encoding/binary" "encoding/json" "errors" "fmt" "math/big" "sort" "strings" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" "superbanka/internal/platform" ) const signupBonusMinor int64 = 1000000 var ErrConflict = errors.New("conflict") var ErrInvalidCredentials = errors.New("invalid credentials") var ErrInsufficientFunds = errors.New("insufficient funds") var ErrNotFound = errors.New("not found") var ErrFrozen = errors.New("account frozen") type Store struct { pool *pgxpool.Pool } type User struct { ID string Email string Username string Status string AccountID string AccountNumber string BalanceMinor int64 CreatedAt time.Time } type AdminUser struct { ID string Email string Role string Status string CreatedAt time.Time } type TransactionView struct { ID string Type string Description string Direction string SenderAccount string RecipientAccount string AmountMinor int64 CreatedAt time.Time } type UserListItem struct { ID string Email string Username string Status string AccountNumber string BalanceMinor int64 CreatedAt time.Time } type LeaderboardEntry struct { Rank int Username string AccountNumber string BalanceMinor int64 } type LotteryRoundView struct { ID string PoolMinor int64 DrawAt time.Time ParticipantCount int TicketCount int UserTicketCount int UserContributionMinor int64 } type LotteryPayoutView struct { RoundID string WinnerUsername string WinnerAccount string PoolMinor int64 ParticipantCount int TicketCount int CompletedAt time.Time } type LotteryDashboard struct { ActiveRound *LotteryRoundView RecentPayouts []LotteryPayoutView } type AuditLog struct { Action string TargetType string TargetID string Metadata string CreatedAt time.Time AdminEmail string } type AdminDashboard struct { UserCount int AccountCount int TransactionCount int FrozenUserCount int RecentTransactions []TransactionView RecentAuditActivity []AuditLog } type lotteryEntryRecord struct { UserID string AccountID string } type lotterySettlementResult struct { RoundID string WinnerUsername string WinnerAccount string PoolMinor int64 ParticipantCount int TicketCount int CompletedAt time.Time } func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} } func (s *Store) BootstrapAdmin(ctx context.Context, email, passwordHash string) error { var exists bool if err := s.pool.QueryRow(ctx, `select exists(select 1 from admin_users)`).Scan(&exists); err != nil { return fmt.Errorf("check existing admins: %w", err) } if exists { return nil } id, err := platform.NewID() if err != nil { return err } _, err = s.pool.Exec(ctx, ` insert into admin_users (id, email, password_hash, role, status, created_at, updated_at) values ($1, $2, $3, 'super_admin', 'active', now(), now()) `, id, strings.ToLower(strings.TrimSpace(email)), passwordHash) if err != nil { return fmt.Errorf("insert bootstrap admin: %w", err) } return nil } func (s *Store) CreateUser(ctx context.Context, email, username, passwordHash string) (*User, error) { tx, err := s.pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("begin create user: %w", err) } defer tx.Rollback(ctx) userID, err := platform.NewID() if err != nil { return nil, err } accountID, err := platform.NewID() if err != nil { return nil, err } transactionID, err := platform.NewID() if err != nil { return nil, err } entryID, err := platform.NewID() if err != nil { return nil, err } reference, err := platform.NewID() if err != nil { return nil, err } accountNumber, err := s.generateUniqueAccountNumber(ctx, tx) if err != nil { return nil, err } _, err = tx.Exec(ctx, ` insert into users (id, email, username, password_hash, status, created_at, updated_at) values ($1, $2, $3, $4, 'active', now(), now()) `, userID, strings.ToLower(strings.TrimSpace(email)), strings.ToLower(strings.TrimSpace(username)), passwordHash) if err != nil { if isUniqueViolation(err) { return nil, ErrConflict } return nil, fmt.Errorf("insert user: %w", err) } _, err = tx.Exec(ctx, ` insert into accounts (id, user_id, account_number, currency, status, created_at) values ($1, $2, $3, 'CZK', 'open', now()) `, accountID, userID, accountNumber) if err != nil { return nil, fmt.Errorf("insert account: %w", err) } _, err = tx.Exec(ctx, ` insert into transactions (id, type, status, reference, description, created_at) values ($1, 'signup_bonus', 'completed', $2, 'Signup bonus', now()) `, transactionID, reference) if err != nil { return nil, fmt.Errorf("insert signup transaction: %w", err) } _, err = tx.Exec(ctx, ` insert into ledger_entries (id, transaction_id, account_id, entry_type, amount_minor, created_at) values ($1, $2, $3, 'credit', $4, now()) `, entryID, transactionID, accountID, signupBonusMinor) if err != nil { return nil, fmt.Errorf("insert signup ledger entry: %w", err) } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("commit create user: %w", err) } return s.GetUserDashboard(ctx, userID) } func (s *Store) GetUserByEmail(ctx context.Context, email string) (*User, string, error) { var user User var passwordHash string err := s.pool.QueryRow(ctx, ` select u.id, u.email, u.username, u.status, a.id, a.account_number, u.created_at, u.password_hash from users u join accounts a on a.user_id = u.id where u.email = $1 `, strings.ToLower(strings.TrimSpace(email))).Scan( &user.ID, &user.Email, &user.Username, &user.Status, &user.AccountID, &user.AccountNumber, &user.CreatedAt, &passwordHash, ) if errors.Is(err, pgx.ErrNoRows) { return nil, "", ErrInvalidCredentials } if err != nil { return nil, "", fmt.Errorf("get user by email: %w", err) } balance, err := s.BalanceForAccount(ctx, user.AccountID) if err != nil { return nil, "", err } user.BalanceMinor = balance return &user, passwordHash, nil } func (s *Store) GetAdminByEmail(ctx context.Context, email string) (*AdminUser, string, error) { var admin AdminUser var passwordHash string err := s.pool.QueryRow(ctx, ` select id, email, role, status, created_at, password_hash from admin_users where email = $1 `, strings.ToLower(strings.TrimSpace(email))).Scan(&admin.ID, &admin.Email, &admin.Role, &admin.Status, &admin.CreatedAt, &passwordHash) if errors.Is(err, pgx.ErrNoRows) { return nil, "", ErrInvalidCredentials } if err != nil { return nil, "", fmt.Errorf("get admin by email: %w", err) } return &admin, passwordHash, nil } func (s *Store) CreateUserSession(ctx context.Context, userID, tokenHash string, expiresAt time.Time) error { id, err := platform.NewID() if err != nil { return err } _, err = s.pool.Exec(ctx, ` insert into user_sessions (id, user_id, token_hash, expires_at, created_at) values ($1, $2, $3, $4, now()) `, id, userID, tokenHash, expiresAt) if err != nil { return fmt.Errorf("create user session: %w", err) } return nil } func (s *Store) CreateAdminSession(ctx context.Context, adminID, tokenHash string, expiresAt time.Time) error { id, err := platform.NewID() if err != nil { return err } _, err = s.pool.Exec(ctx, ` insert into admin_sessions (id, admin_user_id, token_hash, expires_at, created_at) values ($1, $2, $3, $4, now()) `, id, adminID, tokenHash, expiresAt) if err != nil { return fmt.Errorf("create admin session: %w", err) } return nil } func (s *Store) DeleteUserSession(ctx context.Context, tokenHash string) error { _, err := s.pool.Exec(ctx, `delete from user_sessions where token_hash = $1`, tokenHash) if err != nil { return fmt.Errorf("delete user session: %w", err) } return nil } func (s *Store) DeleteAdminSession(ctx context.Context, tokenHash string) error { _, err := s.pool.Exec(ctx, `delete from admin_sessions where token_hash = $1`, tokenHash) if err != nil { return fmt.Errorf("delete admin session: %w", err) } return nil } func (s *Store) UserFromSession(ctx context.Context, tokenHash string) (*User, error) { var user User err := s.pool.QueryRow(ctx, ` select u.id, u.email, u.username, u.status, a.id, a.account_number, u.created_at from user_sessions us join users u on u.id = us.user_id join accounts a on a.user_id = u.id where us.token_hash = $1 and us.expires_at > now() `, tokenHash).Scan(&user.ID, &user.Email, &user.Username, &user.Status, &user.AccountID, &user.AccountNumber, &user.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("user from session: %w", err) } balance, err := s.BalanceForAccount(ctx, user.AccountID) if err != nil { return nil, err } user.BalanceMinor = balance return &user, nil } func (s *Store) AdminFromSession(ctx context.Context, tokenHash string) (*AdminUser, error) { var admin AdminUser err := s.pool.QueryRow(ctx, ` select a.id, a.email, a.role, a.status, a.created_at from admin_sessions s join admin_users a on a.id = s.admin_user_id where s.token_hash = $1 and s.expires_at > now() `, tokenHash).Scan(&admin.ID, &admin.Email, &admin.Role, &admin.Status, &admin.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("admin from session: %w", err) } return &admin, nil } func (s *Store) GetUserDashboard(ctx context.Context, userID string) (*User, error) { var user User err := s.pool.QueryRow(ctx, ` select u.id, u.email, u.username, u.status, a.id, a.account_number, u.created_at from users u join accounts a on a.user_id = u.id where u.id = $1 `, userID).Scan(&user.ID, &user.Email, &user.Username, &user.Status, &user.AccountID, &user.AccountNumber, &user.CreatedAt) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("get user dashboard: %w", err) } balance, err := s.BalanceForAccount(ctx, user.AccountID) if err != nil { return nil, err } user.BalanceMinor = balance return &user, nil } func (s *Store) BalanceForAccount(ctx context.Context, accountID string) (int64, error) { var balance int64 err := s.pool.QueryRow(ctx, ` select coalesce(sum(case when entry_type = 'credit' then amount_minor else -amount_minor end), 0) from ledger_entries where account_id = $1 `, accountID).Scan(&balance) if err != nil { return 0, fmt.Errorf("balance for account: %w", err) } return balance, nil } func (s *Store) RecentTransactionsForAccount(ctx context.Context, accountID string, limit int) ([]TransactionView, error) { rows, err := s.pool.Query(ctx, ` select t.id, t.type, t.description, le.entry_type, coalesce( ( select a2.account_number from ledger_entries le2 join accounts a2 on a2.id = le2.account_id where le2.transaction_id = t.id and le2.entry_type = 'debit' limit 1 ), '' ) as sender_account, coalesce( ( select a2.account_number from ledger_entries le2 join accounts a2 on a2.id = le2.account_id where le2.transaction_id = t.id and le2.entry_type = 'credit' limit 1 ), '' ) as recipient_account, le.amount_minor, t.created_at from ledger_entries le join transactions t on t.id = le.transaction_id where le.account_id = $1 order by t.created_at desc limit $2 `, accountID, limit) if err != nil { return nil, fmt.Errorf("recent transactions: %w", err) } defer rows.Close() var items []TransactionView for rows.Next() { var item TransactionView if err := rows.Scan(&item.ID, &item.Type, &item.Description, &item.Direction, &item.SenderAccount, &item.RecipientAccount, &item.AmountMinor, &item.CreatedAt); err != nil { return nil, fmt.Errorf("scan transaction: %w", err) } items = append(items, item) } return items, rows.Err() } func (s *Store) CreateTransfer(ctx context.Context, senderUserID, destinationAccountNumber, description string, amountMinor int64) error { tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("begin transfer: %w", err) } defer tx.Rollback(ctx) type accountRecord struct { AccountID string UserID string Status string UserStatus string AccountNumber string } var sender accountRecord err = tx.QueryRow(ctx, ` select a.id, u.id, a.status, u.status, a.account_number from accounts a join users u on u.id = a.user_id where u.id = $1 `, senderUserID).Scan(&sender.AccountID, &sender.UserID, &sender.Status, &sender.UserStatus, &sender.AccountNumber) if err != nil { return fmt.Errorf("load sender account: %w", err) } var receiver accountRecord err = tx.QueryRow(ctx, ` select a.id, u.id, a.status, u.status, a.account_number from accounts a join users u on u.id = a.user_id where a.account_number = $1 `, strings.TrimSpace(destinationAccountNumber)).Scan(&receiver.AccountID, &receiver.UserID, &receiver.Status, &receiver.UserStatus, &receiver.AccountNumber) if errors.Is(err, pgx.ErrNoRows) { return ErrNotFound } if err != nil { return fmt.Errorf("load receiver account: %w", err) } if sender.AccountID == receiver.AccountID { return fmt.Errorf("cannot transfer to the same account") } if sender.UserStatus != "active" || sender.Status != "open" || receiver.UserStatus != "active" || receiver.Status != "open" { return ErrFrozen } ids := []string{sender.AccountID, receiver.AccountID} sort.Strings(ids) if _, err := tx.Exec(ctx, `select id from accounts where id = any($1) order by id for update`, ids); err != nil { return fmt.Errorf("lock accounts: %w", err) } balance, err := balanceForAccountTx(ctx, tx, sender.AccountID) if err != nil { return err } if balance < amountMinor { return ErrInsufficientFunds } transactionID, err := platform.NewID() if err != nil { return err } reference, err := platform.NewID() if err != nil { return err } debitID, err := platform.NewID() if err != nil { return err } creditID, err := platform.NewID() if err != nil { return err } _, err = tx.Exec(ctx, ` insert into transactions (id, type, status, reference, initiated_by_user_id, description, created_at) values ($1, 'transfer', 'completed', $2, $3, $4, now()) `, transactionID, reference, senderUserID, strings.TrimSpace(description)) if err != nil { return fmt.Errorf("insert transfer transaction: %w", err) } _, err = tx.Exec(ctx, ` insert into ledger_entries (id, transaction_id, account_id, entry_type, amount_minor, created_at) values ($1, $2, $3, 'debit', $4, now()), ($5, $2, $6, 'credit', $4, now()) `, debitID, transactionID, sender.AccountID, amountMinor, creditID, receiver.AccountID) if err != nil { return fmt.Errorf("insert ledger entries: %w", err) } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("commit transfer: %w", err) } return nil } func (s *Store) GetLotteryDashboard(ctx context.Context, userID string) (*LotteryDashboard, error) { var dashboard LotteryDashboard var active LotteryRoundView err := s.pool.QueryRow(ctx, ` select lr.id, lr.pool_minor, lr.draw_at, coalesce((select count(distinct le.user_id) from lottery_entries le where le.round_id = lr.id), 0), coalesce((select count(*) from lottery_entries le where le.round_id = lr.id), 0), coalesce((select count(*) from lottery_entries le where le.round_id = lr.id and le.user_id = $1), 0), coalesce((select sum(le.amount_minor) from lottery_entries le where le.round_id = lr.id and le.user_id = $1), 0) from lottery_rounds lr where lr.status = 'active' order by lr.created_at desc limit 1 `, userID).Scan( &active.ID, &active.PoolMinor, &active.DrawAt, &active.ParticipantCount, &active.TicketCount, &active.UserTicketCount, &active.UserContributionMinor, ) if err == nil { dashboard.ActiveRound = &active } else if !errors.Is(err, pgx.ErrNoRows) { return nil, fmt.Errorf("load active lottery round: %w", err) } rows, err := s.pool.Query(ctx, ` select lr.id, u.username, a.account_number, lr.pool_minor, coalesce((select count(distinct le.user_id) from lottery_entries le where le.round_id = lr.id), 0), coalesce((select count(*) from lottery_entries le where le.round_id = lr.id), 0), lr.completed_at from lottery_rounds lr join users u on u.id = lr.winner_user_id join accounts a on a.id = lr.winner_account_id where lr.status = 'completed' and lr.winner_user_id is not null order by lr.completed_at desc limit 8 `) if err != nil { return nil, fmt.Errorf("list lottery payouts: %w", err) } defer rows.Close() for rows.Next() { var item LotteryPayoutView if err := rows.Scan( &item.RoundID, &item.WinnerUsername, &item.WinnerAccount, &item.PoolMinor, &item.ParticipantCount, &item.TicketCount, &item.CompletedAt, ); err != nil { return nil, fmt.Errorf("scan lottery payout: %w", err) } dashboard.RecentPayouts = append(dashboard.RecentPayouts, item) } return &dashboard, rows.Err() } func (s *Store) CreateLotteryDeposit(ctx context.Context, userID string, amountMinor int64) error { tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("begin lottery deposit: %w", err) } defer tx.Rollback(ctx) if _, err := s.settleDueLotteryRoundTx(ctx, tx, time.Now()); err != nil { return err } type accountRecord struct { AccountID string UserStatus string Status string } var account accountRecord err = tx.QueryRow(ctx, ` select a.id, u.status, a.status from accounts a join users u on u.id = a.user_id where u.id = $1 for update `, userID).Scan(&account.AccountID, &account.UserStatus, &account.Status) if err != nil { return fmt.Errorf("load lottery depositor account: %w", err) } if account.UserStatus != "active" || account.Status != "open" { return ErrFrozen } balance, err := balanceForAccountTx(ctx, tx, account.AccountID) if err != nil { return err } if balance < amountMinor { return ErrInsufficientFunds } roundID, drawAt, err := s.findOrCreateActiveLotteryRoundTx(ctx, tx, time.Now()) if err != nil { return err } transactionID, err := platform.NewID() if err != nil { return err } reference, err := platform.NewID() if err != nil { return err } ledgerEntryID, err := platform.NewID() if err != nil { return err } lotteryEntryID, err := platform.NewID() if err != nil { return err } description := fmt.Sprintf("Lottery deposit for round closing at %s", drawAt.Format("15:04")) _, err = tx.Exec(ctx, ` insert into transactions (id, type, status, reference, initiated_by_user_id, description, created_at) values ($1, 'lottery_deposit', 'completed', $2, $3, $4, now()) `, transactionID, reference, userID, description) if err != nil { return fmt.Errorf("insert lottery deposit transaction: %w", err) } _, err = tx.Exec(ctx, ` insert into ledger_entries (id, transaction_id, account_id, entry_type, amount_minor, created_at) values ($1, $2, $3, 'debit', $4, now()) `, ledgerEntryID, transactionID, account.AccountID, amountMinor) if err != nil { return fmt.Errorf("insert lottery deposit ledger entry: %w", err) } _, err = tx.Exec(ctx, ` insert into lottery_entries (id, round_id, user_id, account_id, amount_minor, created_at) values ($1, $2, $3, $4, $5, now()) `, lotteryEntryID, roundID, userID, account.AccountID, amountMinor) if err != nil { return fmt.Errorf("insert lottery entry: %w", err) } _, err = tx.Exec(ctx, ` update lottery_rounds set pool_minor = pool_minor + $2 where id = $1 `, roundID, amountMinor) if err != nil { return fmt.Errorf("update lottery round pool: %w", err) } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("commit lottery deposit: %w", err) } return nil } func (s *Store) SettleDueLotteryRound(ctx context.Context) (*LotteryPayoutView, error) { tx, err := s.pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("begin lottery settlement: %w", err) } defer tx.Rollback(ctx) result, err := s.settleDueLotteryRoundTx(ctx, tx, time.Now()) if err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("commit lottery settlement: %w", err) } if result == nil { return nil, nil } return &LotteryPayoutView{ RoundID: result.RoundID, WinnerUsername: result.WinnerUsername, WinnerAccount: result.WinnerAccount, PoolMinor: result.PoolMinor, ParticipantCount: result.ParticipantCount, TicketCount: result.TicketCount, CompletedAt: result.CompletedAt, }, nil } func (s *Store) AdminDashboard(ctx context.Context) (*AdminDashboard, error) { var dashboard AdminDashboard if err := s.pool.QueryRow(ctx, `select count(*) from users`).Scan(&dashboard.UserCount); err != nil { return nil, err } if err := s.pool.QueryRow(ctx, `select count(*) from accounts`).Scan(&dashboard.AccountCount); err != nil { return nil, err } if err := s.pool.QueryRow(ctx, `select count(*) from transactions`).Scan(&dashboard.TransactionCount); err != nil { return nil, err } if err := s.pool.QueryRow(ctx, `select count(*) from users where status = 'frozen'`).Scan(&dashboard.FrozenUserCount); err != nil { return nil, err } rows, err := s.pool.Query(ctx, ` select t.id, t.type, t.description, le.entry_type, coalesce((select a.account_number from ledger_entries le2 join accounts a on a.id = le2.account_id where le2.transaction_id = t.id and le2.entry_type = 'debit' limit 1), ''), coalesce((select a.account_number from ledger_entries le2 join accounts a on a.id = le2.account_id where le2.transaction_id = t.id and le2.entry_type = 'credit' limit 1), ''), le.amount_minor, t.created_at from transactions t join ledger_entries le on le.transaction_id = t.id order by t.created_at desc limit 10 `) if err != nil { return nil, err } defer rows.Close() for rows.Next() { var item TransactionView if err := rows.Scan(&item.ID, &item.Type, &item.Description, &item.Direction, &item.SenderAccount, &item.RecipientAccount, &item.AmountMinor, &item.CreatedAt); err != nil { return nil, err } dashboard.RecentTransactions = append(dashboard.RecentTransactions, item) } auditRows, err := s.pool.Query(ctx, ` select al.action, al.target_type, al.target_id, al.metadata::text, al.created_at, au.email from audit_logs al join admin_users au on au.id = al.actor_admin_id order by al.created_at desc limit 10 `) if err != nil { return nil, err } defer auditRows.Close() for auditRows.Next() { var item AuditLog if err := auditRows.Scan(&item.Action, &item.TargetType, &item.TargetID, &item.Metadata, &item.CreatedAt, &item.AdminEmail); err != nil { return nil, err } dashboard.RecentAuditActivity = append(dashboard.RecentAuditActivity, item) } return &dashboard, nil } func (s *Store) SearchUsers(ctx context.Context, search string) ([]UserListItem, error) { search = strings.TrimSpace(search) pattern := "%" + strings.ToLower(search) + "%" rows, err := s.pool.Query(ctx, ` select u.id, u.email, u.username, u.status, a.account_number, u.created_at, coalesce(sum(case when le.entry_type = 'credit' then le.amount_minor else -le.amount_minor end), 0) as balance_minor from users u join accounts a on a.user_id = u.id left join ledger_entries le on le.account_id = a.id where ($1 = '' or lower(u.email) like $2 or lower(u.username) like $2 or a.account_number like $2) group by u.id, u.email, u.username, u.status, a.account_number, u.created_at order by u.created_at desc limit 100 `, search, pattern) if err != nil { return nil, fmt.Errorf("search users: %w", err) } defer rows.Close() var items []UserListItem for rows.Next() { var item UserListItem if err := rows.Scan(&item.ID, &item.Email, &item.Username, &item.Status, &item.AccountNumber, &item.CreatedAt, &item.BalanceMinor); err != nil { return nil, fmt.Errorf("scan user list item: %w", err) } items = append(items, item) } return items, rows.Err() } func (s *Store) Leaderboard(ctx context.Context, limit int) ([]LeaderboardEntry, error) { rows, err := s.pool.Query(ctx, ` select u.username, a.account_number, coalesce(sum(case when le.entry_type = 'credit' then le.amount_minor else -le.amount_minor end), 0) as balance_minor from users u join accounts a on a.user_id = u.id left join ledger_entries le on le.account_id = a.id where u.status = 'active' and a.status = 'open' group by u.id, u.username, a.account_number order by balance_minor desc, u.username asc limit $1 `, limit) if err != nil { return nil, fmt.Errorf("leaderboard: %w", err) } defer rows.Close() var entries []LeaderboardEntry rank := 1 for rows.Next() { var entry LeaderboardEntry if err := rows.Scan(&entry.Username, &entry.AccountNumber, &entry.BalanceMinor); err != nil { return nil, fmt.Errorf("scan leaderboard entry: %w", err) } entry.Rank = rank rank++ entries = append(entries, entry) } return entries, rows.Err() } func (s *Store) GetUserDetail(ctx context.Context, userID string) (*User, []TransactionView, error) { user, err := s.GetUserDashboard(ctx, userID) if err != nil { return nil, nil, err } transactions, err := s.RecentTransactionsForAccount(ctx, user.AccountID, 50) if err != nil { return nil, nil, err } return user, transactions, nil } func (s *Store) SetUserStatus(ctx context.Context, userID, status, adminID string) error { result, err := s.pool.Exec(ctx, `update users set status = $2, updated_at = now() where id = $1`, userID, status) if err != nil { return fmt.Errorf("update user status: %w", err) } if result.RowsAffected() == 0 { return ErrNotFound } metadata, err := json.Marshal(map[string]string{"status": status}) if err != nil { return fmt.Errorf("marshal user status audit log: %w", err) } return s.insertAuditLog(ctx, adminID, "user_status_changed", "user", userID, string(metadata)) } func (s *Store) CreateAdjustment(ctx context.Context, adminID, accountID, direction string, amountMinor int64, reason string) error { tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("begin adjustment: %w", err) } defer tx.Rollback(ctx) var userStatus string err = tx.QueryRow(ctx, ` select u.status from accounts a join users u on u.id = a.user_id where a.id = $1 for update `, accountID).Scan(&userStatus) if errors.Is(err, pgx.ErrNoRows) { return ErrNotFound } if err != nil { return fmt.Errorf("load account for adjustment: %w", err) } if userStatus != "active" { return ErrFrozen } if direction == "debit" { balance, err := balanceForAccountTx(ctx, tx, accountID) if err != nil { return err } if balance < amountMinor { return ErrInsufficientFunds } } transactionID, err := platform.NewID() if err != nil { return err } reference, err := platform.NewID() if err != nil { return err } entryID, err := platform.NewID() if err != nil { return err } _, err = tx.Exec(ctx, ` insert into transactions (id, type, status, reference, initiated_by_admin_id, description, created_at) values ($1, 'adjustment', 'completed', $2, $3, $4, now()) `, transactionID, reference, adminID, strings.TrimSpace(reason)) if err != nil { return fmt.Errorf("insert adjustment transaction: %w", err) } _, err = tx.Exec(ctx, ` insert into ledger_entries (id, transaction_id, account_id, entry_type, amount_minor, created_at) values ($1, $2, $3, $4, $5, now()) `, entryID, transactionID, accountID, direction, amountMinor) if err != nil { return fmt.Errorf("insert adjustment entry: %w", err) } metadataBytes, err := json.Marshal(map[string]any{ "direction": direction, "amount_minor": amountMinor, "reason": strings.TrimSpace(reason), }) if err != nil { return fmt.Errorf("marshal adjustment metadata: %w", err) } logID, err := platform.NewID() if err != nil { return err } _, err = tx.Exec(ctx, ` insert into audit_logs (id, actor_admin_id, action, target_type, target_id, metadata, created_at) values ($1, $2, 'balance_adjustment', 'account', $3, $4::jsonb, now()) `, logID, adminID, accountID, string(metadataBytes)) if err != nil { return fmt.Errorf("insert adjustment audit log: %w", err) } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("commit adjustment: %w", err) } return nil } func (s *Store) AuditLogs(ctx context.Context) ([]AuditLog, error) { rows, err := s.pool.Query(ctx, ` select al.action, al.target_type, al.target_id, al.metadata::text, al.created_at, au.email from audit_logs al join admin_users au on au.id = al.actor_admin_id order by al.created_at desc limit 100 `) if err != nil { return nil, fmt.Errorf("list audit logs: %w", err) } defer rows.Close() var logs []AuditLog for rows.Next() { var item AuditLog if err := rows.Scan(&item.Action, &item.TargetType, &item.TargetID, &item.Metadata, &item.CreatedAt, &item.AdminEmail); err != nil { return nil, fmt.Errorf("scan audit log: %w", err) } logs = append(logs, item) } return logs, rows.Err() } func (s *Store) insertAuditLog(ctx context.Context, adminID, action, targetType, targetID, metadata string) error { id, err := platform.NewID() if err != nil { return err } _, err = s.pool.Exec(ctx, ` insert into audit_logs (id, actor_admin_id, action, target_type, target_id, metadata, created_at) values ($1, $2, $3, $4, $5, $6::jsonb, now()) `, id, adminID, action, targetType, targetID, metadata) if err != nil { return fmt.Errorf("insert audit log: %w", err) } return nil } func (s *Store) findOrCreateActiveLotteryRoundTx(ctx context.Context, tx pgx.Tx, now time.Time) (string, time.Time, error) { var roundID string var drawAt time.Time err := tx.QueryRow(ctx, ` select id, draw_at from lottery_rounds where status = 'active' order by created_at desc limit 1 for update `).Scan(&roundID, &drawAt) if err == nil { return roundID, drawAt, nil } if !errors.Is(err, pgx.ErrNoRows) { return "", time.Time{}, fmt.Errorf("load active lottery round: %w", err) } roundID, err = platform.NewID() if err != nil { return "", time.Time{}, err } drawAt = now.Add(time.Hour) _, err = tx.Exec(ctx, ` insert into lottery_rounds (id, status, pool_minor, draw_at, created_at) values ($1, 'active', 0, $2, $3) `, roundID, drawAt, now) if err != nil { return "", time.Time{}, fmt.Errorf("create lottery round: %w", err) } return roundID, drawAt, nil } func (s *Store) settleDueLotteryRoundTx(ctx context.Context, tx pgx.Tx, now time.Time) (*lotterySettlementResult, error) { var round struct { ID string PoolMinor int64 DrawAt time.Time } err := tx.QueryRow(ctx, ` select id, pool_minor, draw_at from lottery_rounds where status = 'active' order by created_at asc limit 1 for update `).Scan(&round.ID, &round.PoolMinor, &round.DrawAt) if errors.Is(err, pgx.ErrNoRows) { return nil, nil } if err != nil { return nil, fmt.Errorf("load active lottery round for settlement: %w", err) } if round.DrawAt.After(now) { return nil, nil } rows, err := tx.Query(ctx, ` select user_id, account_id from lottery_entries where round_id = $1 order by created_at asc, id asc `, round.ID) if err != nil { return nil, fmt.Errorf("load lottery entries: %w", err) } defer rows.Close() var entries []lotteryEntryRecord participants := map[string]struct{}{} for rows.Next() { var entry lotteryEntryRecord if err := rows.Scan(&entry.UserID, &entry.AccountID); err != nil { return nil, fmt.Errorf("scan lottery entry: %w", err) } entries = append(entries, entry) participants[entry.UserID] = struct{}{} } if err := rows.Err(); err != nil { return nil, err } if len(entries) == 0 || round.PoolMinor == 0 { _, err := tx.Exec(ctx, ` update lottery_rounds set status = 'completed', completed_at = $2 where id = $1 `, round.ID, now) if err != nil { return nil, fmt.Errorf("complete empty lottery round: %w", err) } return nil, nil } winnerIndex, err := cryptoRandomIndex(len(entries)) if err != nil { return nil, err } winner := entries[winnerIndex] var winnerUsername, winnerAccount string err = tx.QueryRow(ctx, ` select u.username, a.account_number from users u join accounts a on a.id = $1 where u.id = $2 `, winner.AccountID, winner.UserID).Scan(&winnerUsername, &winnerAccount) if err != nil { return nil, fmt.Errorf("load lottery winner: %w", err) } transactionID, err := platform.NewID() if err != nil { return nil, err } reference, err := platform.NewID() if err != nil { return nil, err } ledgerEntryID, err := platform.NewID() if err != nil { return nil, err } _, err = tx.Exec(ctx, ` insert into transactions (id, type, status, reference, initiated_by_user_id, description, created_at) values ($1, 'lottery_payout', 'completed', $2, $3, $4, $5) `, transactionID, reference, winner.UserID, fmt.Sprintf("Lottery payout from %d ticket(s)", len(entries)), now) if err != nil { return nil, fmt.Errorf("insert lottery payout transaction: %w", err) } _, err = tx.Exec(ctx, ` insert into ledger_entries (id, transaction_id, account_id, entry_type, amount_minor, created_at) values ($1, $2, $3, 'credit', $4, $5) `, ledgerEntryID, transactionID, winner.AccountID, round.PoolMinor, now) if err != nil { return nil, fmt.Errorf("insert lottery payout ledger entry: %w", err) } _, err = tx.Exec(ctx, ` update lottery_rounds set status = 'completed', winner_user_id = $2, winner_account_id = $3, completed_at = $4 where id = $1 `, round.ID, winner.UserID, winner.AccountID, now) if err != nil { return nil, fmt.Errorf("complete lottery round: %w", err) } return &lotterySettlementResult{ RoundID: round.ID, WinnerUsername: winnerUsername, WinnerAccount: winnerAccount, PoolMinor: round.PoolMinor, ParticipantCount: len(participants), TicketCount: len(entries), CompletedAt: now, }, nil } func (s *Store) generateUniqueAccountNumber(ctx context.Context, tx pgx.Tx) (string, error) { for i := 0; i < 20; i++ { candidate, err := randomAccountNumber() if err != nil { return "", err } var exists bool if err := tx.QueryRow(ctx, `select exists(select 1 from accounts where account_number = $1)`, candidate).Scan(&exists); err != nil { return "", fmt.Errorf("check account number uniqueness: %w", err) } if !exists { return candidate, nil } } return "", fmt.Errorf("could not generate unique account number") } func randomAccountNumber() (string, error) { var buf [8]byte if _, err := rand.Read(buf[:]); err != nil { return "", fmt.Errorf("read random account bytes: %w", err) } n := binary.BigEndian.Uint64(buf[:]) % 10000000000 if n < 1000000000 { n += 1000000000 } return fmt.Sprintf("%010d", n), nil } func cryptoRandomIndex(limit int) (int, error) { if limit <= 0 { return 0, fmt.Errorf("limit must be positive") } n, err := rand.Int(rand.Reader, big.NewInt(int64(limit))) if err != nil { return 0, fmt.Errorf("read random index: %w", err) } return int(n.Int64()), nil } func balanceForAccountTx(ctx context.Context, tx pgx.Tx, accountID string) (int64, error) { var balance int64 err := tx.QueryRow(ctx, ` select coalesce(sum(case when entry_type = 'credit' then amount_minor else -amount_minor end), 0) from ledger_entries where account_id = $1 `, accountID).Scan(&balance) if err != nil { return 0, fmt.Errorf("balance for account tx: %w", err) } return balance, nil } func isUniqueViolation(err error) bool { var pgErr *pgconn.PgError return errors.As(err, &pgErr) && pgErr.Code == "23505" }