refactor(database): remove direct utxo manipulation (#975)

Rather than directly manipulating UTxOs from the ledger, we process transactions. This moves the specifics of UTxO processing to the database package.

Signed-off-by: Chris Gianelloni <wolf31o2@blinklabs.io>

Author: wolf31o2

database/block.go

@@ -157,10 +157,13 @@ func BlockByPointTxn(txn *Txn, point ocommon.Point) (models.Block, error) {
 	return blockByKey(txn, key)
 }
 
-func (d *Database) BlockByIndex(blockIndex uint64, txn *Txn) (models.Block, error) {
+func (d *Database) BlockByIndex(
+	blockIndex uint64,
+	txn *Txn,
+) (models.Block, error) {
 	if txn == nil {
 		txn = d.BlobTxn(false)
-		defer txn.Commit() //nolint:errcheck
+		defer txn.Rollback() //nolint:errcheck
 	}
 	indexKey := BlockBlobIndexKey(blockIndex)
 	item, err := txn.Blob().Get(indexKey)

database/models/transaction.go

@@ -16,12 +16,12 @@ package models
 
 // Transaction represents a transaction record
 type Transaction struct {
-	Type       string
 	Hash       []byte `gorm:"uniqueIndex"`
 	BlockHash  []byte `gorm:"index"`
-	Inputs     []byte
-	Outputs    []byte
-	ID         uint `gorm:"primaryKey"`
+	Inputs     []Utxo `gorm:"foreignKey:SpentAtTxId;references:Hash"`
+	Outputs    []Utxo `gorm:"foreignKey:TransactionID;references:ID"`
+	ID         uint   `gorm:"primaryKey"`
+	Type       int
 	BlockIndex uint32
 }
 

database/models/utxo.go

@@ -16,6 +16,7 @@ package models
 
 import (
 	"github.com/blinklabs-io/gouroboros/ledger"
+	lcommon "github.com/blinklabs-io/gouroboros/ledger/common"
 )
 
 // Utxo represents an unspent transaction output
@@ -26,6 +27,7 @@ type Utxo struct {
 	StakingKey    []byte `gorm:"index"`
 	Assets        []Asset
 	Cbor          []byte `gorm:"-"` // This is here for convenience but not represented in the metadata DB
+	SpentAtTxId   []byte `gorm:"index"`
 	ID            uint   `gorm:"primarykey"`
 	AddedSlot     uint64 `gorm:"index"`
 	DeletedSlot   uint64 `gorm:"index"`
@@ -41,6 +43,37 @@ func (u *Utxo) Decode() (ledger.TransactionOutput, error) {
 	return ledger.NewTransactionOutputFromCbor(u.Cbor)
 }
 
+func UtxoLedgerToModel(
+	utxo ledger.Utxo,
+	slot uint64,
+) Utxo {
+	outAddr := utxo.Output.Address()
+	ret := Utxo{
+		TxId:      utxo.Id.Id().Bytes(),
+		Cbor:      utxo.Output.Cbor(),
+		AddedSlot: slot,
+		Amount:    utxo.Output.Amount(),
+		OutputIdx: utxo.Id.Index(),
+	}
+	pkh := outAddr.PaymentKeyHash()
+	if pkh != ledger.NewBlake2b224(nil) {
+		ret.PaymentKey = pkh.Bytes()
+	}
+	skh := outAddr.StakeKeyHash()
+	if skh != ledger.NewBlake2b224(nil) {
+		ret.StakingKey = skh.Bytes()
+	}
+	if multiAssetOutput, ok := utxo.Output.(interface {
+		MultiAsset() *lcommon.MultiAsset[lcommon.MultiAssetTypeOutput]
+	}); ok {
+		if multiAsset := multiAssetOutput.MultiAsset(); multiAsset != nil {
+			ret.Assets = ConvertMultiAssetToModels(multiAsset)
+		}
+	}
+
+	return ret
+}
+
 // UtxoSlot allows providing a slot number with a ledger.Utxo object
 type UtxoSlot struct {
 	Utxo ledger.Utxo

database/plugin/metadata/sqlite/transaction.go

@@ -16,9 +16,13 @@ package sqlite
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/blinklabs-io/dingo/database/models"
+	lcommon "github.com/blinklabs-io/gouroboros/ledger/common"
+	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 // GetTransactionByHash returns a transaction by its hash
@@ -42,26 +46,78 @@ func (d *MetadataStoreSqlite) GetTransactionByHash(
 
 // SetTransaction adds a new transaction to the database
 func (d *MetadataStoreSqlite) SetTransaction(
-	hash []byte,
-	txType string,
-	blockHash []byte,
-	blockIndex uint32,
-	inputs []byte,
-	outputs []byte,
+	tx lcommon.Transaction,
+	point ocommon.Point,
+	idx uint32,
 	txn *gorm.DB,
 ) error {
 	if txn == nil {
 		txn = d.DB()
 	}
-	tx := &models.Transaction{
-		Hash:       hash,
-		Type:       txType,
-		BlockHash:  blockHash,
-		BlockIndex: blockIndex,
-		Inputs:     inputs,
-		Outputs:    outputs,
+	tmpTx := &models.Transaction{
+		Hash:       tx.Hash().Bytes(),
+		Type:       tx.Type(),
+		BlockHash:  point.Hash,
+		BlockIndex: idx,
+	}
+	if tx.IsValid() {
+		for _, utxo := range tx.Produced() {
+			tmpTx.Outputs = append(
+				tmpTx.Outputs,
+				models.UtxoLedgerToModel(utxo, point.Slot),
+			)
+		}
+	}
+	result := txn.Create(&tmpTx)
+	if result.Error != nil {
+		return fmt.Errorf("create transaction: %w", result.Error)
+	}
+	// Explicitly create produced outputs with TransactionID set
+	if tx.IsValid() && len(tmpTx.Outputs) > 0 {
+		for o := range tmpTx.Outputs {
+			tmpTx.Outputs[o].TransactionID = &tmpTx.ID
+		}
+		result = txn.Clauses(clause.OnConflict{
+			UpdateAll: true,
+		}).Create(&tmpTx.Outputs)
+		if result.Error != nil {
+			return fmt.Errorf("create outputs: %w", result.Error)
+		}
+	}
+	for i, input := range tx.Consumed() {
+		inTxId := input.Id().Bytes()
+		inIdx := input.Index()
+		utxo, err := d.GetUtxo(inTxId, inIdx, txn)
+		if err != nil {
+			return fmt.Errorf(
+				"failed to fetch input %x#%d: %w",
+				inTxId,
+				inIdx,
+				err,
+			)
+		}
+		if utxo == nil {
+			return fmt.Errorf("input UTxO not found: %x#%d", inTxId, inIdx)
+		}
+		// Update existing UTxOs
+		result = txn.Model(&models.Utxo{}).
+			Where("tx_id = ? AND output_idx = ?", inTxId, inIdx).
+			Updates(map[string]any{
+				"deleted_slot":   point.Slot,
+				"spent_at_tx_id": tx.Hash().Bytes(),
+			})
+		if result.Error != nil {
+			return result.Error
+		}
+		// Explicitly set consumed inputs with TransactionID set
+		tmpTx.Inputs = append(
+			tmpTx.Inputs,
+			*utxo,
+		)
+		tmpTx.Inputs[i].TransactionID = &utxo.ID
 	}
-	result := txn.Create(&tx)
+	// Avoid updating associations
+	result = txn.Omit(clause.Associations).Save(&tmpTx.Inputs)
 	if result.Error != nil {
 		return result.Error
 	}

database/plugin/metadata/sqlite/utxo.go

@@ -20,7 +20,6 @@ import (
 
 	"github.com/blinklabs-io/dingo/database/models"
 	"github.com/blinklabs-io/gouroboros/ledger"
-	lcommon "github.com/blinklabs-io/gouroboros/ledger/common"
 	"gorm.io/gorm"
 )
 
@@ -102,6 +101,9 @@ func (d *MetadataStoreSqlite) GetUtxosByAddress(
 	var ret []models.Utxo
 	// Build sub-query for address
 	var addrQuery *gorm.DB
+	if txn == nil {
+		txn = d.DB()
+	}
 	if addr.PaymentKeyHash() != ledger.NewBlake2b224(nil) {
 		addrQuery = txn.Where("payment_key = ?", addr.PaymentKeyHash().Bytes())
 	}
@@ -115,6 +117,9 @@ func (d *MetadataStoreSqlite) GetUtxosByAddress(
 			addrQuery = txn.Where("staking_key = ?", addr.StakeKeyHash().Bytes())
 		}
 	}
+	if addrQuery == nil {
+		return ret, nil
+	}
 	result := txn.
 		Where("deleted_slot = 0").
 		Where(addrQuery).
@@ -193,43 +198,6 @@ func (d *MetadataStoreSqlite) DeleteUtxosAfterSlot(
 	return nil
 }
 
-// SetUtxo saves a UTxO
-func (d *MetadataStoreSqlite) SetUtxo(
-	txId []byte, // hash
-	idx uint32, // idx
-	slot uint64, // slot
-	payment []byte, // payment
-	stake []byte, // stake
-	amount uint64, // amount
-	assets *lcommon.MultiAsset[lcommon.MultiAssetTypeOutput], // assets (multi-asset)
-	txn *gorm.DB,
-) error {
-	tmpUtxo := models.Utxo{
-		TxId:       txId,
-		OutputIdx:  idx,
-		AddedSlot:  slot,
-		PaymentKey: payment,
-		StakingKey: stake,
-	}
-
-	// Handle assets if present
-	if assets != nil {
-		tmpUtxo.Assets = models.ConvertMultiAssetToModels(assets)
-	}
-
-	var result *gorm.DB
-	if txn != nil {
-		result = txn.Create(&tmpUtxo)
-	} else {
-		result = d.DB().Create(&tmpUtxo)
-	}
-
-	if result.Error != nil {
-		return result.Error
-	}
-	return nil
-}
-
 // AddUtxos saves a batch of UTxOs
 func (d *MetadataStoreSqlite) AddUtxos(
 	utxos []models.UtxoSlot,
@@ -239,19 +207,16 @@ func (d *MetadataStoreSqlite) AddUtxos(
 	for _, utxo := range utxos {
 		items = append(
 			items,
-			utxoLedgerToModel(utxo.Utxo, utxo.Slot),
+			models.UtxoLedgerToModel(utxo.Utxo, utxo.Slot),
 		)
 	}
-	if txn != nil {
-		result := txn.Create(items)
-		if result.Error != nil {
-			return result.Error
-		}
-	} else {
-		result := d.DB().Create(items)
-		if result.Error != nil {
-			return result.Error
-		}
+	if txn == nil {
+		txn = d.DB()
+	}
+	result := txn.Session(&gorm.Session{FullSaveAssociations: true}).
+		CreateInBatches(items, 1000)
+	if result.Error != nil {
+		return result.Error
 	}
 	return nil
 }
@@ -302,28 +267,3 @@ func (d *MetadataStoreSqlite) SetUtxosNotDeletedAfterSlot(
 	}
 	return nil
 }
-
-func utxoLedgerToModel(
-	utxo ledger.Utxo,
-	slot uint64,
-) models.Utxo {
-	outAddr := utxo.Output.Address()
-	ret := models.Utxo{
-		TxId:       utxo.Id.Id().Bytes(),
-		OutputIdx:  utxo.Id.Index(),
-		AddedSlot:  slot,
-		PaymentKey: outAddr.PaymentKeyHash().Bytes(),
-		StakingKey: outAddr.StakeKeyHash().Bytes(),
-		Cbor:       utxo.Output.Cbor(),
-	}
-
-	if multiAssetOutput, ok := utxo.Output.(interface {
-		MultiAsset() *lcommon.MultiAsset[lcommon.MultiAssetTypeOutput]
-	}); ok {
-		if multiAsset := multiAssetOutput.MultiAsset(); multiAsset != nil {
-			ret.Assets = models.ConvertMultiAssetToModels(multiAsset)
-		}
-	}
-
-	return ret
-}

database/plugin/metadata/store.go

@@ -22,6 +22,7 @@ import (
 	"github.com/blinklabs-io/gouroboros/ledger"
 	lcommon "github.com/blinklabs-io/gouroboros/ledger/common"
 	ochainsync "github.com/blinklabs-io/gouroboros/protocol/chainsync"
+	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
 	"github.com/prometheus/client_golang/prometheus"
 	"gorm.io/gorm"
 )
@@ -187,32 +188,19 @@ type MetadataStore interface {
 		uint64, // deposit
 		*gorm.DB,
 	) error
-	SetTransaction(
-		[]byte, // hash
-		string, // type
-		[]byte, // blockHash
-		uint32, // blockIndex
-		[]byte, // inputs
-		[]byte, // outputs
-		*gorm.DB,
-	) error
 	SetTip(
 		ochainsync.Tip,
 		*gorm.DB,
 	) error
-	SetUpdateDrep(
-		*lcommon.UpdateDrepCertificate,
-		uint64, // slot
+	SetTransaction(
+		lcommon.Transaction,
+		ocommon.Point,
+		uint32, // idx
 		*gorm.DB,
 	) error
-	SetUtxo(
-		[]byte, // hash
-		uint32, // idx
+	SetUpdateDrep(
+		*lcommon.UpdateDrepCertificate,
 		uint64, // slot
-		[]byte, // payment
-		[]byte, // stake
-		uint64, // amount
-		*lcommon.MultiAsset[lcommon.MultiAssetTypeOutput], // asset
 		*gorm.DB,
 	) error
 	SetVoteDelegation(