TXM In-memory: address_state methods: step 3-01

common/txmgr/address_state.go

@@ -117,14 +117,39 @@ func NewAddressState[
 	return &as, nil
 }
 
-// CountTransactionsByState returns the number of transactions that are in the given state
+// CountTransactionsByState returns the number of transactions that are in the given state.
 func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) CountTransactionsByState(txState txmgrtypes.TxState) int {
-	return 0
+	as.RLock()
+	defer as.RUnlock()
+
+	switch txState {
+	case TxUnstarted:
+		return as.unstartedTxs.Len()
+	case TxInProgress:
+		if as.inprogressTx != nil {
+			return 1
+		}
+		return 0
+	case TxUnconfirmed:
+		return len(as.unconfirmedTxs)
+	case TxConfirmedMissingReceipt:
+		return len(as.confirmedMissingReceiptTxs)
+	case TxConfirmed:
+		return len(as.confirmedTxs)
+	case TxFatalError:
+		return len(as.fatalErroredTxs)
+	}
+
+	return -1
 }
 
-// FindTxWithIdempotencyKey returns the transaction with the given idempotency key. If no transaction is found, nil is returned.
+// FindTxWithIdempotencyKey returns the transaction with the given idempotency key.
+// If no transaction is found, nil is returned.
 func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindTxWithIdempotencyKey(key string) *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
-	return nil
+	as.RLock()
+	defer as.RUnlock()
+
+	return as.idempotencyKeyToTx[key]
 }
 
 // ApplyToTxsByState calls the given function for each transaction in the given states.
@@ -136,6 +161,31 @@ func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) ApplyT
 	fn func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]),
 	txIDs ...int64,
 ) {
+	as.Lock()
+	defer as.Unlock()
+
+	// if txStates is empty then apply the filter to only the as.allTransactions map
+	if len(txStates) == 0 {
+		as.applyToTxs(as.allTxs, fn, txIDs...)
+		return
+	}
+
+	for _, txState := range txStates {
+		switch txState {
+		case TxInProgress:
+			if as.inprogressTx != nil {
+				fn(as.inprogressTx)
+			}
+		case TxUnconfirmed:
+			as.applyToTxs(as.unconfirmedTxs, fn, txIDs...)
+		case TxConfirmedMissingReceipt:
+			as.applyToTxs(as.confirmedMissingReceiptTxs, fn, txIDs...)
+		case TxConfirmed:
+			as.applyToTxs(as.confirmedTxs, fn, txIDs...)
+		case TxFatalError:
+			as.applyToTxs(as.fatalErroredTxs, fn, txIDs...)
+		}
+	}
 }
 
 // FindTxAttempts returns all attempts for the given transactions that match the given filters.
@@ -149,7 +199,38 @@ func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindTx
 	txAttemptFilter func(*txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
 	txIDs ...int64,
 ) []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
-	return nil
+	as.RLock()
+	defer as.RUnlock()
+
+	// if txStates is empty then apply the filter to only the as.allTransactions map
+	if len(txStates) == 0 {
+		return as.findTxAttempts(as.allTxs, txFilter, txAttemptFilter, txIDs...)
+	}
+
+	var txAttempts []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	for _, txState := range txStates {
+		switch txState {
+		case TxInProgress:
+			if as.inprogressTx != nil && txFilter(as.inprogressTx) {
+				for i := 0; i < len(as.inprogressTx.TxAttempts); i++ {
+					txAttempt := as.inprogressTx.TxAttempts[i]
+					if txAttemptFilter(&txAttempt) {
+						txAttempts = append(txAttempts, txAttempt)
+					}
+				}
+			}
+		case TxUnconfirmed:
+			txAttempts = append(txAttempts, as.findTxAttempts(as.unconfirmedTxs, txFilter, txAttemptFilter, txIDs...)...)
+		case TxConfirmedMissingReceipt:
+			txAttempts = append(txAttempts, as.findTxAttempts(as.confirmedMissingReceiptTxs, txFilter, txAttemptFilter, txIDs...)...)
+		case TxConfirmed:
+			txAttempts = append(txAttempts, as.findTxAttempts(as.confirmedTxs, txFilter, txAttemptFilter, txIDs...)...)
+		case TxFatalError:
+			txAttempts = append(txAttempts, as.findTxAttempts(as.fatalErroredTxs, txFilter, txAttemptFilter, txIDs...)...)
+		}
+	}
+
+	return txAttempts
 }
 
 // FindTxs returns all transactions that match the given filters.
@@ -161,7 +242,33 @@ func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindTx
 	filter func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
 	txIDs ...int64,
 ) []txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
-	return nil
+	as.RLock()
+	defer as.RUnlock()
+
+	// if txStates is empty then apply the filter to only the as.allTransactions map
+	if len(txStates) == 0 {
+		return as.findTxs(as.allTxs, filter, txIDs...)
+	}
+
+	var txs []txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	for _, txState := range txStates {
+		switch txState {
+		case TxInProgress:
+			if as.inprogressTx != nil && filter(as.inprogressTx) {
+				txs = append(txs, *as.inprogressTx)
+			}
+		case TxUnconfirmed:
+			txs = append(txs, as.findTxs(as.unconfirmedTxs, filter, txIDs...)...)
+		case TxConfirmedMissingReceipt:
+			txs = append(txs, as.findTxs(as.confirmedMissingReceiptTxs, filter, txIDs...)...)
+		case TxConfirmed:
+			txs = append(txs, as.findTxs(as.confirmedTxs, filter, txIDs...)...)
+		case TxFatalError:
+			txs = append(txs, as.findTxs(as.fatalErroredTxs, filter, txIDs...)...)
+		}
+	}
+
+	return txs
 }
 
 // PruneUnstartedTxQueue removes the transactions with the given IDs from the unstarted transaction queue.
@@ -238,3 +345,151 @@ func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) MoveIn
 func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) MoveConfirmedToUnconfirmed(attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error {
 	return nil
 }
+
+// Close releases all resources held by the address state.
+func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Close() {
+	clear(as.idempotencyKeyToTx)
+
+	as.unstartedTxs.Close()
+	as.unstartedTxs = nil
+	as.inprogressTx = nil
+
+	clear(as.unconfirmedTxs)
+	clear(as.confirmedMissingReceiptTxs)
+	clear(as.confirmedTxs)
+	clear(as.allTxs)
+	clear(as.fatalErroredTxs)
+
+	as.idempotencyKeyToTx = nil
+	as.unconfirmedTxs = nil
+	as.confirmedMissingReceiptTxs = nil
+	as.confirmedTxs = nil
+	as.allTxs = nil
+	as.fatalErroredTxs = nil
+}
+
+func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Abandon() {
+	as.Lock()
+	defer as.Unlock()
+
+	for as.unstartedTxs.Len() > 0 {
+		tx := as.unstartedTxs.RemoveNextTx()
+		as.abandonTx(tx)
+	}
+
+	if as.inprogressTx != nil {
+		tx := as.inprogressTx
+		as.abandonTx(tx)
+		as.inprogressTx = nil
+	}
+	for _, tx := range as.unconfirmedTxs {
+		as.abandonTx(tx)
+	}
+
+	clear(as.unconfirmedTxs)
+}
+
+func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) abandonTx(tx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) {
+	if tx == nil {
+		return
+	}
+
+	tx.State = TxFatalError
+	tx.Sequence = nil
+	tx.Error = null.NewString("abandoned", true)
+
+	as.fatalErroredTxs[tx.ID] = tx
+}
+
+func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) applyToTxs(
+	txIDsToTx map[int64]*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
+	fn func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]),
+	txIDs ...int64,
+) {
+	// if txIDs is not empty then only apply the filter to those transactions
+	if len(txIDs) > 0 {
+		for _, txID := range txIDs {
+			tx := txIDsToTx[txID]
+			if tx != nil {
+				fn(tx)
+			}
+		}
+		return
+	}
+
+	// if txIDs is empty then apply the filter to all transactions
+	for _, tx := range txIDsToTx {
+		fn(tx)
+	}
+}
+
+func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) findTxAttempts(
+	txIDsToTx map[int64]*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
+	txFilter func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
+	txAttemptFilter func(*txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
+	txIDs ...int64,
+) []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	as.RLock()
+	defer as.RUnlock()
+
+	var txAttempts []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	// if txIDs is not empty then only apply the filter to those transactions
+	if len(txIDs) > 0 {
+		for _, txID := range txIDs {
+			tx := txIDsToTx[txID]
+			if tx != nil && txFilter(tx) {
+				for i := 0; i < len(tx.TxAttempts); i++ {
+					txAttempt := tx.TxAttempts[i]
+					if txAttemptFilter(&txAttempt) {
+						txAttempts = append(txAttempts, txAttempt)
+					}
+				}
+			}
+		}
+		return txAttempts
+	}
+
+	// if txIDs is empty then apply the filter to all transactions
+	for _, tx := range txIDsToTx {
+		if txFilter(tx) {
+			for i := 0; i < len(tx.TxAttempts); i++ {
+				txAttempt := tx.TxAttempts[i]
+				if txAttemptFilter(&txAttempt) {
+					txAttempts = append(txAttempts, txAttempt)
+				}
+			}
+		}
+	}
+
+	return txAttempts
+}
+
+func (as *AddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) findTxs(
+	txIDsToTx map[int64]*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
+	filter func(*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool,
+	txIDs ...int64,
+) []txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE] {
+	as.RLock()
+	defer as.RUnlock()
+
+	var txs []txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]
+	// if txIDs is not empty then only apply the filter to those transactions
+	if len(txIDs) > 0 {
+		for _, txID := range txIDs {
+			tx := txIDsToTx[txID]
+			if tx != nil && filter(tx) {
+				txs = append(txs, *tx)
+			}
+		}
+		return txs
+	}
+
+	// if txIDs is empty then apply the filter to all transactions
+	for _, tx := range txIDsToTx {
+		if filter(tx) {
+			txs = append(txs, *tx)
+		}
+	}
+
+	return txs
+}