// Copyright (C) 2019 Storj Labs, Inc. // See LICENSE for copying information. package satellitedb import ( "context" "math/big" "time" pgxerrcode "github.com/jackc/pgerrcode" "github.com/shopspring/decimal" "github.com/zeebo/errs" "storj.io/common/uuid" "storj.io/private/dbutil/pgutil/pgerrcode" "storj.io/storj/satellite/payments/coinpayments" "storj.io/storj/satellite/payments/monetary" "storj.io/storj/satellite/payments/stripecoinpayments" "storj.io/storj/satellite/satellitedb/dbx" ) // ensure that coinpaymentsTransactions implements stripecoinpayments.TransactionsDB. var _ stripecoinpayments.TransactionsDB = (*coinPaymentsTransactions)(nil) // applyBalanceIntentState defines states of the apply balance intents. type applyBalanceIntentState int const ( // apply balance intent waits to be applied. applyBalanceIntentStateUnapplied applyBalanceIntentState = 0 // transaction which balance intent points to has been consumed. applyBalanceIntentStateConsumed applyBalanceIntentState = 1 ) // Int returns intent state as int. func (intent applyBalanceIntentState) Int() int { return int(intent) } // coinPaymentsTransactions is CoinPayments transactions DB. // // architecture: Database type coinPaymentsTransactions struct { db *satelliteDB } // Insert inserts new coinpayments transaction into DB. func (db *coinPaymentsTransactions) Insert(ctx context.Context, tx stripecoinpayments.Transaction) (createTime time.Time, err error) { defer mon.Task()(&ctx)(&err) amount, err := tx.Amount.AsBigFloat().GobEncode() if err != nil { return time.Time{}, errs.Wrap(err) } received, err := tx.Received.AsBigFloat().GobEncode() if err != nil { return time.Time{}, errs.Wrap(err) } dbxCPTX, err := db.db.Create_CoinpaymentsTransaction(ctx, dbx.CoinpaymentsTransaction_Id(tx.ID.String()), dbx.CoinpaymentsTransaction_UserId(tx.AccountID[:]), dbx.CoinpaymentsTransaction_Address(tx.Address), dbx.CoinpaymentsTransaction_Amount(amount), dbx.CoinpaymentsTransaction_Received(received), dbx.CoinpaymentsTransaction_Status(tx.Status.Int()), dbx.CoinpaymentsTransaction_Key(tx.Key), dbx.CoinpaymentsTransaction_Timeout(int(tx.Timeout.Seconds())), ) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { // TEMPORARY: fall back to expected new schema to facilitate transition return db.insertTransitionShim(ctx, tx) } return time.Time{}, err } return dbxCPTX.CreatedAt, nil } // insertTransitionShim inserts new coinpayments transaction into DB. // // It is to be used only during the transition from gob-encoded 'amount' and // 'received' columns to 'amount_numeric'/'received_numeric'. // // When the transition is complete, this method will go away and Insert() // will operate only on the _numeric columns. func (db *coinPaymentsTransactions) insertTransitionShim(ctx context.Context, tx stripecoinpayments.Transaction) (createTime time.Time, err error) { row := db.db.DB.QueryRowContext(ctx, db.db.Rebind(` INSERT INTO coinpayments_transactions ( id, user_id, address, amount_numeric, received_numeric, status, key, timeout, created_at ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, now() ) RETURNING created_at; `), tx.ID.String(), tx.AccountID[:], tx.Address, tx.Amount.BaseUnits(), tx.Received.BaseUnits(), tx.Status.Int(), tx.Key, int(tx.Timeout.Seconds())) if err := row.Scan(&createTime); err != nil { return time.Time{}, Error.Wrap(err) } return createTime, nil } // Update updates status and received for set of transactions. func (db *coinPaymentsTransactions) Update(ctx context.Context, updates []stripecoinpayments.TransactionUpdate, applies coinpayments.TransactionIDList) (err error) { defer mon.Task()(&ctx)(&err) if len(updates) == 0 { return nil } err = db.db.WithTx(ctx, func(ctx context.Context, tx *dbx.Tx) error { for _, update := range updates { receivedGob, err := update.Received.AsBigFloat().GobEncode() if err != nil { return errs.Wrap(err) } _, err = tx.Update_CoinpaymentsTransaction_By_Id(ctx, dbx.CoinpaymentsTransaction_Id(update.TransactionID.String()), dbx.CoinpaymentsTransaction_Update_Fields{ Received: dbx.CoinpaymentsTransaction_Received(receivedGob), Status: dbx.CoinpaymentsTransaction_Status(update.Status.Int()), }, ) if err != nil { return err } } for _, txID := range applies { query := db.db.Rebind(`INSERT INTO stripecoinpayments_apply_balance_intents ( tx_id, state, created_at ) VALUES ( ?, ?, ? ) ON CONFLICT DO NOTHING`) _, err = tx.Tx.ExecContext(ctx, query, txID.String(), applyBalanceIntentStateUnapplied.Int(), db.db.Hooks.Now().UTC()) if err != nil { return err } } return nil }) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { // TEMPORARY: fall back to expected new schema to facilitate transition return db.updateTransitionShim(ctx, updates, applies) } } return err } // updateTransitionShim updates status and received for set of transactions. // // It is to be used only during the transition from gob-encoded 'amount' and // 'received' columns to 'amount_numeric'/'received_numeric'. During the // transition, the gob-encoded columns will still exist but under a different // name ('amount_gob'/'received_gob'). If the _numeric column value for a given // row is non-null, it takes precedence over the corresponding _gob column. // // When the transition is complete, this method will go away and // Update() will operate only on the _numeric columns. func (db *coinPaymentsTransactions) updateTransitionShim(ctx context.Context, updates []stripecoinpayments.TransactionUpdate, applies coinpayments.TransactionIDList) (err error) { defer mon.Task()(&ctx)(&err) if len(updates) == 0 { return nil } return db.db.WithTx(ctx, func(ctx context.Context, tx *dbx.Tx) error { for _, update := range updates { query := db.db.Rebind(` UPDATE coinpayments_transactions SET received_gob = NULL, received_numeric = ?, status = ? WHERE id = ? `) _, err := tx.Tx.ExecContext(ctx, query, update.Received.BaseUnits(), update.Status.Int(), update.TransactionID.String()) if err != nil { return err } } for _, txID := range applies { query := db.db.Rebind(`INSERT INTO stripecoinpayments_apply_balance_intents ( tx_id, state, created_at ) VALUES ( ?, ?, ? ) ON CONFLICT DO NOTHING`) _, err = tx.Tx.ExecContext(ctx, query, txID.String(), applyBalanceIntentStateUnapplied.Int(), db.db.Hooks.Now().UTC()) if err != nil { return err } } return nil }) } // Consume marks transaction as consumed, so it won't participate in apply account balance loop. func (db *coinPaymentsTransactions) Consume(ctx context.Context, id coinpayments.TransactionID) (err error) { defer mon.Task()(&ctx)(&err) query := db.db.Rebind(` WITH intent AS ( SELECT tx_id, state FROM stripecoinpayments_apply_balance_intents WHERE tx_id = ? ), updated AS ( UPDATE stripecoinpayments_apply_balance_intents AS ints SET state = ? FROM intent WHERE intent.tx_id = ints.tx_id AND ints.state = ? RETURNING 1 ) SELECT EXISTS(SELECT 1 FROM intent) AS intent_exists, EXISTS(SELECT 1 FROM updated) AS intent_consumed; `) row := db.db.QueryRowContext(ctx, query, id, applyBalanceIntentStateConsumed, applyBalanceIntentStateUnapplied) var exists, consumed bool if err = row.Scan(&exists, &consumed); err != nil { return err } if !exists { return errs.New("can not consume transaction without apply balance intent") } if !consumed { return stripecoinpayments.ErrTransactionConsumed } return err } // LockRate locks conversion rate for transaction. func (db *coinPaymentsTransactions) LockRate(ctx context.Context, id coinpayments.TransactionID, rate decimal.Decimal) (err error) { defer mon.Task()(&ctx)(&err) buff, err := rate.BigFloat().GobEncode() if err != nil { return Error.Wrap(err) } _, err = db.db.Create_StripecoinpaymentsTxConversionRate(ctx, dbx.StripecoinpaymentsTxConversionRate_TxId(id.String()), dbx.StripecoinpaymentsTxConversionRate_Rate(buff)) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { // TEMPORARY: fall back to expected new schema to facilitate transition return db.lockRateTransitionShim(ctx, id, rate) } } return Error.Wrap(err) } // lockRateTransitionShim locks conversion rate for transaction. // // It is to be used only during the transition from the gob-encoded 'rate' // column to 'rate_numeric'. // // When the transition is complete, this method will go away and // LockRate() will operate only on the _numeric column. func (db *coinPaymentsTransactions) lockRateTransitionShim(ctx context.Context, id coinpayments.TransactionID, rate decimal.Decimal) (err error) { defer mon.Task()(&ctx)(&err) now := time.Now().UTC() query := db.db.Rebind(` INSERT INTO stripecoinpayments_tx_conversion_rates ( tx_id, rate_numeric, created_at ) VALUES ( ?, ?, ? ) `) _, err = db.db.DB.ExecContext(ctx, query, id.String(), rate.String(), now) return Error.Wrap(err) } // GetLockedRate returns locked conversion rate for transaction or error if non exists. func (db *coinPaymentsTransactions) GetLockedRate(ctx context.Context, id coinpayments.TransactionID) (_ decimal.Decimal, err error) { defer mon.Task()(&ctx)(&err) dbxRate, err := db.db.Get_StripecoinpaymentsTxConversionRate_By_TxId(ctx, dbx.StripecoinpaymentsTxConversionRate_TxId(id.String()), ) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { // TEMPORARY: fall back to expected new schema to facilitate transition return db.getLockedRateTransitionShim(ctx, id) } return decimal.Decimal{}, err } var rateF big.Float if err = rateF.GobDecode(dbxRate.Rate); err != nil { return decimal.Decimal{}, errs.Wrap(err) } rate, err := monetary.DecimalFromBigFloat(&rateF) if err != nil { return decimal.Decimal{}, errs.Wrap(err) } return rate, nil } // getLockedRateTransitionShim returns locked conversion rate for transaction // or error if none exists. // // It is to be used only during the transition from the gob-encoded 'rate' // column to 'rate_numeric'. During the transition, the gob-encoded column will // still exist but under a different name ('rate_gob'). If rate_numeric for a // given row is non-null, it takes precedence over rate_gob. // // When the transition is complete, this method will go away and // GetLockedRate() will operate only on rate_numeric. func (db *coinPaymentsTransactions) getLockedRateTransitionShim(ctx context.Context, id coinpayments.TransactionID) (_ decimal.Decimal, err error) { defer mon.Task()(&ctx)(&err) var rateGob []byte var rateNumeric *string query := db.db.Rebind(` SELECT rate_gob, rate_numeric FROM stripecoinpayments_tx_conversion_rates WHERE tx_id = ? `) row := db.db.DB.QueryRowContext(ctx, query, id.String()) err = row.Scan(&rateGob, &rateNumeric) if err != nil { return decimal.Decimal{}, Error.Wrap(err) } if rateNumeric == nil { // This row does not have a numeric rate value yet var rateF big.Float if err = rateF.GobDecode(rateGob); err != nil { return decimal.Decimal{}, Error.Wrap(err) } rate, err := monetary.DecimalFromBigFloat(&rateF) return rate, Error.Wrap(err) } rate, err := decimal.NewFromString(*rateNumeric) return rate, Error.Wrap(err) } // ListAccount returns all transaction for specific user. func (db *coinPaymentsTransactions) ListAccount(ctx context.Context, userID uuid.UUID) (_ []stripecoinpayments.Transaction, err error) { defer mon.Task()(&ctx)(&err) dbxTXs, err := db.db.All_CoinpaymentsTransaction_By_UserId_OrderBy_Desc_CreatedAt(ctx, dbx.CoinpaymentsTransaction_UserId(userID[:]), ) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { // TEMPORARY: fall back to expected new schema to facilitate transition return db.listAccountTransitionShim(ctx, userID) } return nil, err } var txs []stripecoinpayments.Transaction for _, dbxTX := range dbxTXs { tx, err := fromDBXCoinpaymentsTransaction(dbxTX) if err != nil { return nil, errs.Wrap(err) } txs = append(txs, *tx) } return txs, nil } // listAccountTransitionShim returns all transaction for specific user. // // It is to be used only during the transition from gob-encoded 'amount' and // 'received' columns to 'amount_numeric'/'received_numeric'. During the // transition, the gob-encoded columns will still exist but under a different // name ('amount_gob'/'received_gob'). If the _numeric column value for a given // row is non-null, it takes precedence over the corresponding _gob column. // // When the transition is complete, this method will go away and ListAccount() // will operate only on the _numeric columns. func (db *coinPaymentsTransactions) listAccountTransitionShim(ctx context.Context, userID uuid.UUID) (_ []stripecoinpayments.Transaction, err error) { defer mon.Task()(&ctx)(&err) query := db.db.Rebind(` SELECT id, user_id, address, amount_gob, amount_numeric, received_gob, received_numeric, status, key, timeout, created_at FROM coinpayments_transactions WHERE user_id = ? ORDER BY created_at DESC `) rows, err := db.db.DB.QueryContext(ctx, query, userID[:]) if err != nil { return nil, err } defer func() { err = errs.Combine(err, rows.Close()) }() var txs []stripecoinpayments.Transaction for rows.Next() { var tx stripecoinpayments.Transaction var amountGob, receivedGob []byte var amountNumeric, receivedNumeric *int64 var timeoutSeconds int err := rows.Scan(&tx.ID, &tx.AccountID, &tx.Address, &amountGob, &amountNumeric, &receivedGob, &receivedNumeric, &tx.Status, &tx.Key, &timeoutSeconds, &tx.CreatedAt) if err != nil { return nil, Error.Wrap(err) } tx.Timeout = time.Second * time.Duration(timeoutSeconds) if amountNumeric == nil { tx.Amount, err = monetaryAmountFromGobEncodedBigFloat(amountGob, monetary.StorjToken) if err != nil { return nil, Error.New("amount column: %v", err) } } else { tx.Amount = monetary.AmountFromBaseUnits(*amountNumeric, monetary.StorjToken) } if receivedNumeric == nil { tx.Received, err = monetaryAmountFromGobEncodedBigFloat(receivedGob, monetary.StorjToken) if err != nil { return nil, Error.New("received column: %v", err) } } else { tx.Received = monetary.AmountFromBaseUnits(*receivedNumeric, monetary.StorjToken) } txs = append(txs, tx) } if err = rows.Err(); err != nil { return nil, err } return txs, nil } // ListPending returns paginated list of pending transactions. func (db *coinPaymentsTransactions) ListPending(ctx context.Context, offset int64, limit int, before time.Time) (_ stripecoinpayments.TransactionsPage, err error) { defer mon.Task()(&ctx)(&err) query := db.db.Rebind(`SELECT id, user_id, address, amount, received, status, key, created_at FROM coinpayments_transactions WHERE status IN (?,?) AND created_at <= ? ORDER by created_at DESC LIMIT ? OFFSET ?`) rows, err := db.db.QueryContext(ctx, query, coinpayments.StatusPending, coinpayments.StatusReceived, before, limit+1, offset) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { // TEMPORARY: fall back to expected new schema to facilitate transition return db.listPendingTransitionShim(ctx, offset, limit, before) } return stripecoinpayments.TransactionsPage{}, err } defer func() { err = errs.Combine(err, rows.Close()) }() var page stripecoinpayments.TransactionsPage for rows.Next() { var id, address string var userID uuid.UUID var amountB, receivedB []byte var status int var key string var createdAt time.Time err := rows.Scan(&id, &userID, &address, &amountB, &receivedB, &status, &key, &createdAt) if err != nil { return stripecoinpayments.TransactionsPage{}, err } // TODO: the currency here should be passed in to this function or stored // in the database. currency := monetary.StorjToken amount, err := monetaryAmountFromGobEncodedBigFloat(amountB, currency) if err != nil { return stripecoinpayments.TransactionsPage{}, err } received, err := monetaryAmountFromGobEncodedBigFloat(receivedB, currency) if err != nil { return stripecoinpayments.TransactionsPage{}, err } page.Transactions = append(page.Transactions, stripecoinpayments.Transaction{ ID: coinpayments.TransactionID(id), AccountID: userID, Address: address, Amount: amount, Received: received, Status: coinpayments.Status(status), Key: key, CreatedAt: createdAt, }, ) } if err = rows.Err(); err != nil { return stripecoinpayments.TransactionsPage{}, err } if len(page.Transactions) == limit+1 { page.Next = true page.NextOffset = offset + int64(limit) page.Transactions = page.Transactions[:len(page.Transactions)-1] } return page, nil } // listPendingTransitionShim returns paginated list of pending transactions. // // It is to be used only during the transition from gob-encoded 'amount' and // 'received' columns to 'amount_numeric'/'received_numeric'. During the // transition, the gob-encoded columns will still exist but under a different // name ('amount_gob'/'received_gob'). If the _numeric column value for a given // row is non-null, it takes precedence over the corresponding _gob column. // // When the transition is complete, this method will go away and ListPending() // will operate only on the _numeric columns. func (db *coinPaymentsTransactions) listPendingTransitionShim(ctx context.Context, offset int64, limit int, before time.Time) (_ stripecoinpayments.TransactionsPage, err error) { defer mon.Task()(&ctx)(&err) query := db.db.Rebind(`SELECT id, user_id, address, amount_gob, amount_numeric, received_gob, received_numeric, status, key, created_at FROM coinpayments_transactions WHERE status IN (?,?) AND created_at <= ? ORDER BY created_at DESC LIMIT ? OFFSET ?`) rows, err := db.db.QueryContext(ctx, query, coinpayments.StatusPending, coinpayments.StatusReceived, before, limit+1, offset) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.Wrap(err) } defer func() { err = errs.Combine(err, rows.Close()) }() var page stripecoinpayments.TransactionsPage for rows.Next() { var id, address string var userID uuid.UUID var amountGob, receivedGob []byte var amountNumeric, receivedNumeric *int64 var amount, received monetary.Amount var status int var key string var createdAt time.Time err := rows.Scan(&id, &userID, &address, &amountGob, &amountNumeric, &receivedGob, &receivedNumeric, &status, &key, &createdAt) if err != nil { return stripecoinpayments.TransactionsPage{}, err } if amountNumeric == nil { // 'amount' in this row hasn't yet been updated to a numeric value amount, err = monetaryAmountFromGobEncodedBigFloat(amountGob, monetary.StorjToken) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.Wrap(err) } } else { amount = monetary.AmountFromBaseUnits(*amountNumeric, monetary.StorjToken) } if receivedNumeric == nil { // 'received' in this row hasn't yet been updated to a numeric value received, err = monetaryAmountFromGobEncodedBigFloat(receivedGob, monetary.StorjToken) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.Wrap(err) } } else { received = monetary.AmountFromBaseUnits(*receivedNumeric, monetary.StorjToken) } page.Transactions = append(page.Transactions, stripecoinpayments.Transaction{ ID: coinpayments.TransactionID(id), AccountID: userID, Address: address, Amount: amount, Received: received, Status: coinpayments.Status(status), Key: key, CreatedAt: createdAt, }, ) } if err = rows.Err(); err != nil { return stripecoinpayments.TransactionsPage{}, err } if len(page.Transactions) == limit+1 { page.Next = true page.NextOffset = offset + int64(limit) page.Transactions = page.Transactions[:len(page.Transactions)-1] } return page, nil } // ListUnapplied returns TransactionsPage with a pending or completed status, that should be applied to account balance. func (db *coinPaymentsTransactions) ListUnapplied(ctx context.Context, offset int64, limit int, before time.Time) (_ stripecoinpayments.TransactionsPage, err error) { defer mon.Task()(&ctx)(&err) query := db.db.Rebind(`SELECT txs.id, txs.user_id, txs.address, txs.amount, txs.received, txs.status, txs.key, txs.created_at FROM coinpayments_transactions as txs INNER JOIN stripecoinpayments_apply_balance_intents as ints ON txs.id = ints.tx_id WHERE txs.status >= ? AND txs.created_at <= ? AND ints.state = ? ORDER by txs.created_at DESC LIMIT ? OFFSET ?`) rows, err := db.db.QueryContext(ctx, query, coinpayments.StatusReceived, before, applyBalanceIntentStateUnapplied, limit+1, offset) if err != nil { if errCode := pgerrcode.FromError(err); errCode == pgxerrcode.UndefinedColumn { return db.listUnappliedTransitionShim(ctx, offset, limit, before) } return stripecoinpayments.TransactionsPage{}, err } defer func() { err = errs.Combine(err, rows.Close()) }() var page stripecoinpayments.TransactionsPage for rows.Next() { var id, address string var userID uuid.UUID var amountB, receivedB []byte var status int var key string var createdAt time.Time err := rows.Scan(&id, &userID, &address, &amountB, &receivedB, &status, &key, &createdAt) if err != nil { return stripecoinpayments.TransactionsPage{}, err } // TODO: the currency here should be passed in to this function or stored // in the database. currency := monetary.StorjToken amount, err := monetaryAmountFromGobEncodedBigFloat(amountB, currency) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.New("amount column: %v", err) } received, err := monetaryAmountFromGobEncodedBigFloat(receivedB, currency) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.New("received column: %v", err) } page.Transactions = append(page.Transactions, stripecoinpayments.Transaction{ ID: coinpayments.TransactionID(id), AccountID: userID, Address: address, Amount: amount, Received: received, Status: coinpayments.Status(status), Key: key, CreatedAt: createdAt, }, ) } if err = rows.Err(); err != nil { return stripecoinpayments.TransactionsPage{}, err } if len(page.Transactions) == limit+1 { page.Next = true page.NextOffset = offset + int64(limit) page.Transactions = page.Transactions[:len(page.Transactions)-1] } return page, nil } // listUnappliedTransitionShim returns TransactionsPage with a pending or // completed status, that should be applied to account balance. // // It is to be used only during the transition from gob-encoded 'amount' and // 'received' columns to 'amount_numeric'/'received_numeric'. During the // transition, the gob-encoded columns will still exist but under a different // name ('amount_gob'/'received_gob'). If the _numeric column value for a given // row is non-null, it takes precedence over the corresponding _gob column. // // When the transition is complete, this method will go away and // ListUnapplied() will operate only on the _numeric columns. func (db *coinPaymentsTransactions) listUnappliedTransitionShim(ctx context.Context, offset int64, limit int, before time.Time) (_ stripecoinpayments.TransactionsPage, err error) { defer mon.Task()(&ctx)(&err) query := db.db.Rebind(`SELECT txs.id, txs.user_id, txs.address, txs.amount_gob, txs.amount_numeric, txs.received_gob, txs.received_numeric, txs.status, txs.key, txs.created_at FROM coinpayments_transactions as txs INNER JOIN stripecoinpayments_apply_balance_intents as ints ON txs.id = ints.tx_id WHERE txs.status >= ? AND txs.created_at <= ? AND ints.state = ? ORDER by txs.created_at DESC LIMIT ? OFFSET ?`) rows, err := db.db.QueryContext(ctx, query, coinpayments.StatusReceived, before, applyBalanceIntentStateUnapplied, limit+1, offset) if err != nil { return stripecoinpayments.TransactionsPage{}, err } defer func() { err = errs.Combine(err, rows.Close()) }() var page stripecoinpayments.TransactionsPage for rows.Next() { var id, address string var userID uuid.UUID var amountGob, receivedGob []byte var amountNumeric, receivedNumeric *int64 var status int var key string var createdAt time.Time err := rows.Scan(&id, &userID, &address, &amountGob, &amountNumeric, &receivedGob, &receivedNumeric, &status, &key, &createdAt) if err != nil { return stripecoinpayments.TransactionsPage{}, err } var amount, received monetary.Amount if amountNumeric == nil { // 'amount' in this row hasn't yet been updated to a numeric value amount, err = monetaryAmountFromGobEncodedBigFloat(amountGob, monetary.StorjToken) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.Wrap(err) } } else { amount = monetary.AmountFromBaseUnits(*amountNumeric, monetary.StorjToken) } if receivedNumeric == nil { // 'received' in this row hasn't yet been updated to a numeric value received, err = monetaryAmountFromGobEncodedBigFloat(receivedGob, monetary.StorjToken) if err != nil { return stripecoinpayments.TransactionsPage{}, Error.Wrap(err) } } else { received = monetary.AmountFromBaseUnits(*receivedNumeric, monetary.StorjToken) } page.Transactions = append(page.Transactions, stripecoinpayments.Transaction{ ID: coinpayments.TransactionID(id), AccountID: userID, Address: address, Amount: amount, Received: received, Status: coinpayments.Status(status), Key: key, CreatedAt: createdAt, }, ) } if err = rows.Err(); err != nil { return stripecoinpayments.TransactionsPage{}, err } if len(page.Transactions) == limit+1 { page.Next = true page.NextOffset = offset + int64(limit) page.Transactions = page.Transactions[:len(page.Transactions)-1] } return page, nil } // fromDBXCoinpaymentsTransaction converts *dbx.CoinpaymentsTransaction to *stripecoinpayments.Transaction. func fromDBXCoinpaymentsTransaction(dbxCPTX *dbx.CoinpaymentsTransaction) (*stripecoinpayments.Transaction, error) { userID, err := uuid.FromBytes(dbxCPTX.UserId) if err != nil { return nil, errs.Wrap(err) } // TODO: the currency here should be passed in to this function or stored // in the database. currency := monetary.StorjToken amount, err := monetaryAmountFromGobEncodedBigFloat(dbxCPTX.Amount, currency) if err != nil { return nil, Error.New("amount column: %v", err) } received, err := monetaryAmountFromGobEncodedBigFloat(dbxCPTX.Received, currency) if err != nil { return nil, Error.New("received column: %v", err) } return &stripecoinpayments.Transaction{ ID: coinpayments.TransactionID(dbxCPTX.Id), AccountID: userID, Address: dbxCPTX.Address, Amount: amount, Received: received, Status: coinpayments.Status(dbxCPTX.Status), Key: dbxCPTX.Key, Timeout: time.Second * time.Duration(dbxCPTX.Timeout), CreatedAt: dbxCPTX.CreatedAt, }, nil } func monetaryAmountFromGobEncodedBigFloat(encoded []byte, currency *monetary.Currency) (_ monetary.Amount, err error) { var bf big.Float if err := bf.GobDecode(encoded); err != nil { return monetary.Amount{}, Error.Wrap(err) } return monetary.AmountFromBigFloat(&bf, currency) } // DebugPerformBigFloatTransition performs the schema changes expected as part // of Step 2 of the transition away from gob-encoded big.Float columns in the // database. // // This is for testing purposes only, to ensure that no data is lost and that // code still works after the transition. func (db *coinPaymentsTransactions) DebugPerformBigFloatTransition(ctx context.Context) error { _, err := db.db.DB.ExecContext(ctx, ` ALTER TABLE coinpayments_transactions ALTER COLUMN amount DROP NOT NULL; ALTER TABLE coinpayments_transactions ALTER COLUMN received DROP NOT NULL; ALTER TABLE coinpayments_transactions RENAME COLUMN amount TO amount_gob; ALTER TABLE coinpayments_transactions RENAME COLUMN received TO received_gob; ALTER TABLE coinpayments_transactions ADD COLUMN amount_numeric INT8; ALTER TABLE coinpayments_transactions ADD COLUMN received_numeric INT8; ALTER TABLE stripecoinpayments_tx_conversion_rates ALTER COLUMN rate DROP NOT NULL; ALTER TABLE stripecoinpayments_tx_conversion_rates RENAME COLUMN rate TO rate_gob; ALTER TABLE stripecoinpayments_tx_conversion_rates ADD COLUMN rate_numeric NUMERIC(20, 8); `) return Error.Wrap(err) }