transaction.go

package history

import (
	"context"
	"fmt"

	sq "github.com/Masterminds/squirrel"

	"github.com/shantanu-hashcash/go/services/aurora/internal/db2"
	"github.com/shantanu-hashcash/go/support/errors"
	"github.com/shantanu-hashcash/go/toid"
	"github.com/shantanu-hashcash/go/xdr"
)

// TransactionByHash is a query that loads a single row from the
// `history_transactions` table based upon the provided hash.
func (q *Q) TransactionByHash(ctx context.Context, dest interface{}, hash string) error {
	innerOrOuter := sq.Or{sq.Eq{"ht.transaction_hash": hash}, sq.Eq{"ht.inner_transaction_hash": hash}}
	byHashOrInnerHashHistory := selectTransactionHistory.Where(innerOrOuter)

	return q.Get(ctx, dest, byHashOrInnerHashHistory)
}

func (q *Q) PreFilteredTransactionByHash(ctx context.Context, dest interface{}, hash string) error {
	innerOrOuter := sq.Or{sq.Eq{"ht.transaction_hash": hash}, sq.Eq{"ht.inner_transaction_hash": hash}}
	byHashOrInnerHashPreFilter := selectTransactionPreFilteredTmp.Where(innerOrOuter)

	return q.Get(ctx, dest, byHashOrInnerHashPreFilter)
}

// TransactionsByHashesSinceLedger fetches transactions from `history_transactions_filtered_tmp`
// table which match the given hash since the given ledger sequence (for perf reasons).
func (q *Q) AllTransactionsByHashesSinceLedger(ctx context.Context, hashes []string, sinceLedgerSeq uint32) ([]Transaction, error) {
	var dest []Transaction
	innerOrOuterAndSeqGtEq :=
		sq.And{sq.GtOrEq{"ht.ledger_sequence": sinceLedgerSeq}, sq.Or{sq.Eq{"ht.transaction_hash": hashes}, sq.Eq{"ht.inner_transaction_hash": hashes}}}

	preFilteredTxs := selectTransactionPreFilteredTmp.Where(innerOrOuterAndSeqGtEq)
	historyTxs := selectTransactionHistory.Where(innerOrOuterAndSeqGtEq)

	preFilteredTxsString, args, err := preFilteredTxs.ToSql()
	if err != nil {
		return nil, errors.Wrap(err, "could not get string for un filtered sql query")
	}

	union := historyTxs.Suffix("UNION ALL "+preFilteredTxsString, args...)
	if err := q.Select(ctx, &dest, union); err != nil {
		return nil, err
	}
	return dest, nil
}

// TransactionsByIDs fetches transactions from the `history_transactions` table
// which match the given ids
func (q *Q) TransactionsByIDs(ctx context.Context, ids ...int64) (map[int64]Transaction, error) {
	if len(ids) == 0 {
		return nil, errors.New("no id arguments provided")
	}

	sql := selectTransactionHistory.Where(map[string]interface{}{
		"ht.id": ids,
	})

	var transactions []Transaction
	if err := q.Select(ctx, &transactions, sql); err != nil {
		return nil, err
	}

	byID := map[int64]Transaction{}
	for _, transaction := range transactions {
		byID[transaction.TotalOrderID.ID] = transaction
	}

	return byID, nil
}

// DeleteTransactionsFilteredTmpOlderThan deletes entries older than certain duration
func (q *Q) DeleteTransactionsFilteredTmpOlderThan(ctx context.Context, howOldInSeconds uint64) (int64, error) {
	sql := sq.Delete("history_transactions_filtered_tmp").
		Where(sq.Expr("now() >= (created_at + interval '1 second' * ?)", howOldInSeconds))
	result, err := q.Exec(ctx, sql)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected()
}

// Transactions provides a helper to filter rows from the `history_transactions`
// table with pre-defined filters.  See `TransactionsQ` methods for the
// available filters.
func (q *Q) Transactions() *TransactionsQ {
	return &TransactionsQ{
		parent:        q,
		sql:           selectTransactionHistory,
		includeFailed: false,
		txIdCol:       "ht.id",
	}
}

// ForAccount filters the transactions collection to a specific account
func (q *TransactionsQ) ForAccount(ctx context.Context, aid string) *TransactionsQ {
	var account Account
	q.Err = q.parent.AccountByAddress(ctx, &account, aid)
	if q.Err != nil {
		return q
	}

	q.sql = q.sql.
		Join("history_transaction_participants htp ON htp.history_transaction_id = ht.id").
		Where("htp.history_account_id = ?", account.ID)
	q.txIdCol = "htp.history_transaction_id"

	return q
}

// ForClaimableBalance filters the transactions collection to a specific claimable balance
func (q *TransactionsQ) ForClaimableBalance(ctx context.Context, cbID string) *TransactionsQ {

	var hCB HistoryClaimableBalance
	hCB, q.Err = q.parent.ClaimableBalanceByID(ctx, cbID)
	if q.Err != nil {
		return q
	}

	q.sql = q.sql.
		Join("history_transaction_claimable_balances htcb ON htcb.history_transaction_id = ht.id").
		Where("htcb.history_claimable_balance_id = ?", hCB.InternalID)
	q.txIdCol = "htcb.history_transaction_id"

	return q
}

// ForLiquidityPool filters the transactions collection to a specific liquidity pool
func (q *TransactionsQ) ForLiquidityPool(ctx context.Context, poolID string) *TransactionsQ {

	var hLP HistoryLiquidityPool
	hLP, q.Err = q.parent.LiquidityPoolByID(ctx, poolID)
	if q.Err != nil {
		return q
	}

	q.sql = q.sql.
		Join("history_transaction_liquidity_pools htlp ON htlp.history_transaction_id = ht.id").
		Where("htlp.history_liquidity_pool_id = ?", hLP.InternalID)
	q.txIdCol = "htlp.history_transaction_id"

	return q
}

// ForLedger filters the query to a only transactions in a specific ledger,
// specified by its sequence.
func (q *TransactionsQ) ForLedger(ctx context.Context, seq int32) *TransactionsQ {
	var ledger Ledger
	q.Err = q.parent.LedgerBySequence(ctx, &ledger, seq)
	if q.Err != nil {
		return q
	}

	start := toid.ID{LedgerSequence: seq}
	end := toid.ID{LedgerSequence: seq + 1}
	q.sql = q.sql.Where(
		"ht.id >= ? AND ht.id < ?",
		start.ToInt64(),
		end.ToInt64(),
	)

	return q
}

// IncludeFailed changes the query to include failed transactions.
func (q *TransactionsQ) IncludeFailed() *TransactionsQ {
	q.includeFailed = true
	return q
}

// Page specifies the paging constraints for the query being built by `q`.
func (q *TransactionsQ) Page(page db2.PageQuery) *TransactionsQ {
	if q.Err != nil {
		return q
	}

	q.sql, q.Err = page.ApplyTo(q.sql, q.txIdCol)
	return q
}

// Select loads the results of the query specified by `q` into `dest`.
func (q *TransactionsQ) Select(ctx context.Context, dest interface{}) error {
	if q.Err != nil {
		return q.Err
	}

	if !q.includeFailed {
		q.sql = q.sql.
			Where("(ht.successful = true OR ht.successful IS NULL)")
	}

	q.Err = q.parent.Select(ctx, dest, q.sql)
	if q.Err != nil {
		return q.Err
	}

	transactions, ok := dest.(*[]Transaction)
	if !ok {
		return errors.New("dest is not *[]Transaction")
	}

	for _, t := range *transactions {
		var resultXDR xdr.TransactionResult
		err := xdr.SafeUnmarshalBase64(t.TxResult, &resultXDR)
		if err != nil {
			return err
		}

		if !q.includeFailed {
			if !t.Successful {
				return errors.Errorf("Corrupted data! `include_failed=false` but returned transaction is failed: %s", t.TransactionHash)
			}

			if !resultXDR.Successful() {
				return errors.Errorf("Corrupted data! `include_failed=false` but returned transaction is failed: %s %s", t.TransactionHash, t.TxResult)
			}
		}

		// Check if `successful` equals resultXDR
		if t.Successful && !resultXDR.Successful() {
			return errors.Errorf("Corrupted data! `successful=true` but returned transaction is not success: %s %s", t.TransactionHash, t.TxResult)
		}

		if !t.Successful && resultXDR.Successful() {
			return errors.Errorf("Corrupted data! `successful=false` but returned transaction is success: %s %s", t.TransactionHash, t.TxResult)
		}
	}

	return nil
}

// QTransactions defines transaction related queries.
type QTransactions interface {
	NewTransactionBatchInsertBuilder() TransactionBatchInsertBuilder
	NewTransactionFilteredTmpBatchInsertBuilder() TransactionBatchInsertBuilder
}

func selectTransaction(table string) sq.SelectBuilder {
	return sq.Select(
		"ht.id, " +
			"ht.transaction_hash, " +
			"ht.ledger_sequence, " +
			"ht.application_order, " +
			"ht.account, " +
			"ht.account_muxed, " +
			"ht.account_sequence, " +
			"ht.max_fee, " +
			// `fee_charged` is NULL by default, DB needs to be reingested
			// to populate the value. If value is not present display `max_fee`.
			"COALESCE(ht.fee_charged, ht.max_fee) as fee_charged, " +
			"ht.operation_count, " +
			"ht.tx_envelope, " +
			"ht.tx_result, " +
			"ht.tx_meta, " +
			"ht.tx_fee_meta, " +
			"ht.created_at, " +
			"ht.updated_at, " +
			"COALESCE(ht.successful, true) as successful, " +
			"ht.signatures, " +
			"ht.memo_type, " +
			"ht.memo, " +
			"ht.time_bounds, " +
			"ht.ledger_bounds, " +
			"ht.min_account_sequence, " +
			"ht.min_account_sequence_age, " +
			"ht.min_account_sequence_ledger_gap, " +
			"ht.extra_signers, " +
			"hl.closed_at AS ledger_close_time, " +
			"ht.inner_transaction_hash, " +
			"ht.fee_account, " +
			"ht.fee_account_muxed, " +
			"ht.new_max_fee, " +
			"ht.inner_signatures").
		From(fmt.Sprintf("%s ht", table)).
		LeftJoin("history_ledgers hl ON ht.ledger_sequence = hl.sequence")
}

var selectTransactionHistory = selectTransaction("history_transactions")
var selectTransactionPreFilteredTmp = selectTransaction("history_transactions_filtered_tmp")