feat: initial indexer

cmd/chnsd/main.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/blinklabs-io/chnsd/internal/config"
 	"github.com/blinklabs-io/chnsd/internal/dns"
+	"github.com/blinklabs-io/chnsd/internal/indexer"
 	"github.com/blinklabs-io/chnsd/internal/logging"
 )
 
@@ -65,6 +66,11 @@ func main() {
 		}()
 	}
 
+	// Start indexer
+	if err := indexer.GetIndexer().Start(); err != nil {
+		logger.Fatalf("failed to start indexer: %s", err)
+	}
+
 	// Start DNS listener
 	logger.Infof("starting DNS listener on %s:%d", cfg.Dns.ListenAddress, cfg.Dns.ListenPort)
 	if err := dns.Start(); err != nil {

go.mod

@@ -3,7 +3,8 @@ module github.com/blinklabs-io/chnsd
 go 1.19
 
 require (
-	github.com/blinklabs-io/gouroboros v0.47.0
+	github.com/blinklabs-io/gouroboros v0.48.0
+	github.com/blinklabs-io/snek v0.4.0
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/miekg/dns v1.1.55
 	go.uber.org/zap v1.24.0
@@ -17,9 +18,9 @@ require (
 	github.com/x448/float16 v0.8.4 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
-	golang.org/x/crypto v0.10.0 // indirect
+	golang.org/x/crypto v0.11.0 // indirect
 	golang.org/x/mod v0.9.0 // indirect
 	golang.org/x/net v0.10.0 // indirect
-	golang.org/x/sys v0.9.0 // indirect
+	golang.org/x/sys v0.10.0 // indirect
 	golang.org/x/tools v0.7.0 // indirect
 )

go.sum

@@ -1,6 +1,8 @@
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
-github.com/blinklabs-io/gouroboros v0.47.0 h1:N8SoWoT6h6m+w60OU5qNGNyAHj/MxJzSZaw6cnR1Dc8=
-github.com/blinklabs-io/gouroboros v0.47.0/go.mod h1:J3Zf9Fx65LoLldIZB4+dZZpsUsGfwXMQQKQOhNBNlwY=
+github.com/blinklabs-io/gouroboros v0.48.0 h1:A3mqNX1C8QtW/CoIoF3iyXfKPrKyfyWfbjzdnkwj6Jw=
+github.com/blinklabs-io/gouroboros v0.48.0/go.mod h1:WBghLJF3Im3hXjMd8BFk929tfuSjcUZdn2zE2xtZJEo=
+github.com/blinklabs-io/snek v0.4.0 h1:IAuA4HBg2agK2XmOtvs9siJJJWwNCoI5fAl3UuMdIRg=
+github.com/blinklabs-io/snek v0.4.0/go.mod h1:3jwdMN/IsrpjsdZT4yqxtZVMt/2kXq/NaWvcS/lD42A=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -28,15 +30,15 @@ go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
 go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
-golang.org/x/crypto v0.10.0 h1:LKqV2xt9+kDzSTfOhx4FrkEBcMrAgHSYgzywV9zcGmM=
-golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I=
+golang.org/x/crypto v0.11.0 h1:6Ewdq3tDic1mg5xRO4milcWCfMVQhI4NkqWWvqejpuA=
+golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
 golang.org/x/mod v0.9.0 h1:KENHtAZL2y3NLMYZeHY9DW8HW8V+kQyJsY/V9JlKvCs=
 golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
-golang.org/x/sys v0.9.0 h1:KS/R3tvhPqvJvwcKfnBHJwwthS11LRhmM5D59eEXa0s=
-golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA=
+golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/tools v0.7.0 h1:W4OVu8VVOaIO0yzWMNdepAulS7YfoS3Zabrm8DOXXU4=
 golang.org/x/tools v0.7.0/go.mod h1:4pg6aUX35JBAogB10C9AtvVL+qowtN4pT3CGSQex14s=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=

internal/config/config.go

@@ -4,17 +4,34 @@ import (
 	"fmt"
 	"os"
 
-	ouroboros "github.com/blinklabs-io/gouroboros"
 	"github.com/kelseyhightower/envconfig"
 	"gopkg.in/yaml.v2"
 )
 
+// Per-network script address for Handshake
+var networkScriptAddresses = map[string]string{
+	"preprod": "addr_test1wqhlsl9dsny9d2hdc9uyx4ktj0ty0s8kxev4y9futq4qt4s5anczn",
+}
+
+// Per-network intercept points for starting the chain-sync
+// We start the sync somewhere near where we expect the first data to appear to save time
+// during the initial sync
+var networkInterceptPoints = map[string]struct {
+	Hash string
+	Slot uint64
+}{
+	"preprod": {
+		Hash: "f5366caf6cc87383a33fece0968c3c8c3b25ec496829ab3ba324f7dce5a89c5d",
+		Slot: 29852950,
+	},
+}
+
 type Config struct {
 	Logging LoggingConfig `yaml:"logging"`
 	Metrics MetricsConfig `yaml:"metrics"`
 	Dns     DnsConfig     `yaml:"dns"`
 	Debug   DebugConfig   `yaml:"debug"`
-	Node    NodeConfig    `yaml:"node"`
+	Indexer IndexerConfig `yaml:"indexer"`
 }
 
 type LoggingConfig struct {
@@ -37,12 +54,14 @@ type MetricsConfig struct {
 	ListenPort    uint   `yaml:"port" envconfig:"METRICS_LISTEN_PORT"`
 }
 
-type NodeConfig struct {
-	Network      string `yaml:"network" envconfig:"CARDANO_NETWORK"`
-	NetworkMagic uint32 `yaml:"networkMagic" envconfig:"CARDANO_NODE_NETWORK_MAGIC"`
-	Address      string `yaml:"address" envconfig:"CARDANO_NODE_SOCKET_TCP_HOST"`
-	Port         uint   `yaml:"port" envconfig:"CARDANO_NODE_SOCKET_TCP_PORT"`
-	SocketPath   string `yaml:"socketPath" envconfig:"CARDANO_NODE_SOCKET_PATH"`
+type IndexerConfig struct {
+	Network       string `yaml:"network" envconfig:"INDEXER_NETWORK"`
+	NetworkMagic  uint32 `yaml:"networkMagic" envconfig:"INDEXER_NETWORK_MAGIC"`
+	Address       string `yaml:"address" envconfig:"INDEXER_TCP_ADDRESS"`
+	SocketPath    string `yaml:"socketPath" envconfig:"INDEXER_SOCKET_PATH"`
+	ScriptAddress string `yaml:"scriptAddress" envconfig:"INDEXER_SCRIPT_ADDRESS"`
+	InterceptHash string `yaml:"interceptHash" envconfig:"INDEXER_INTERCEPT_HASH"`
+	InterceptSlot uint64 `yaml:"interceptSlot" envconfig:"INDEXER_INTERCEPT_SLOT"`
 }
 
 // Singleton config instance with default values
@@ -63,9 +82,8 @@ var globalConfig = &Config{
 		ListenAddress: "",
 		ListenPort:    8081,
 	},
-	Node: NodeConfig{
-		Network:    "mainnet",
-		SocketPath: "/node-ipc/node.socket",
+	Indexer: IndexerConfig{
+		Network: "preprod",
 	},
 }
 
@@ -88,18 +106,21 @@ func Load(configFile string) (*Config, error) {
 	if err != nil {
 		return nil, fmt.Errorf("error processing environment: %s", err)
 	}
-	// Populate network magic value from network name
-	if globalConfig.Node.Network != "" {
-		network := ouroboros.NetworkByName(globalConfig.Node.Network)
-		if network == ouroboros.NetworkInvalid {
-			return nil, fmt.Errorf("unknown network: %s", globalConfig.Node.Network)
-		}
-		globalConfig.Node.NetworkMagic = network.NetworkMagic
+	// Provide default script address for named network
+	if scriptAddress, ok := networkScriptAddresses[globalConfig.Indexer.Network]; ok {
+		globalConfig.Indexer.ScriptAddress = scriptAddress
+	} else {
+		return nil, fmt.Errorf("no built-in script address for specified network, please provide one")
+	}
+	// Provide default intercept point for named network
+	if interceptPoint, ok := networkInterceptPoints[globalConfig.Indexer.Network]; ok {
+		globalConfig.Indexer.InterceptHash = interceptPoint.Hash
+		globalConfig.Indexer.InterceptSlot = interceptPoint.Slot
 	}
 	return globalConfig, nil
 }
 
-// Config returns the global config instance
+// GetConfig returns the global config instance
 func GetConfig() *Config {
 	return globalConfig
 }

internal/indexer/indexer.go

@@ -0,0 +1,125 @@
+package indexer
+
+import (
+	"encoding/hex"
+
+	"github.com/blinklabs-io/chnsd/internal/config"
+	"github.com/blinklabs-io/chnsd/internal/logging"
+
+	"github.com/blinklabs-io/gouroboros/cbor"
+	ocommon "github.com/blinklabs-io/gouroboros/protocol/common"
+	"github.com/blinklabs-io/snek/event"
+	filter_chainsync "github.com/blinklabs-io/snek/filter/chainsync"
+	filter_event "github.com/blinklabs-io/snek/filter/event"
+	input_chainsync "github.com/blinklabs-io/snek/input/chainsync"
+	output_embedded "github.com/blinklabs-io/snek/output/embedded"
+	"github.com/blinklabs-io/snek/pipeline"
+)
+
+type Indexer struct {
+	pipeline *pipeline.Pipeline
+}
+
+// Singleton indexer instance
+var globalIndexer = &Indexer{}
+
+func (i *Indexer) Start() error {
+	cfg := config.GetConfig()
+	logger := logging.GetLogger()
+	// Create pipeline
+	i.pipeline = pipeline.New()
+	// Configure pipeline input
+	inputOpts := []input_chainsync.ChainSyncOptionFunc{
+		//input_chainsync.WithIntersectTip(true),
+	}
+	if cfg.Indexer.NetworkMagic > 0 {
+		inputOpts = append(
+			inputOpts,
+			input_chainsync.WithNetworkMagic(cfg.Indexer.NetworkMagic),
+		)
+	} else {
+		inputOpts = append(
+			inputOpts,
+			input_chainsync.WithNetwork(cfg.Indexer.Network),
+		)
+	}
+	if cfg.Indexer.InterceptHash != "" && cfg.Indexer.InterceptSlot > 0 {
+		hashBytes, err := hex.DecodeString(cfg.Indexer.InterceptHash)
+		if err != nil {
+			return err
+		}
+		inputOpts = append(
+			inputOpts,
+			input_chainsync.WithIntersectPoints(
+				[]ocommon.Point{
+					{
+						Hash: hashBytes,
+						Slot: cfg.Indexer.InterceptSlot,
+					},
+				},
+			),
+		)
+	}
+	input := input_chainsync.New(
+		inputOpts...,
+	)
+	i.pipeline.AddInput(input)
+	// Configure pipeline filters
+	// We only care about transaction events
+	filterEvent := filter_event.New(
+		filter_event.WithType("chainsync.transaction"),
+	)
+	i.pipeline.AddFilter(filterEvent)
+	// We only care about transactions on a certain address
+	filterChainsync := filter_chainsync.New(
+		filter_chainsync.WithAddress(cfg.Indexer.ScriptAddress),
+	)
+	i.pipeline.AddFilter(filterChainsync)
+	// Configure pipeline output
+	output := output_embedded.New(
+		output_embedded.WithCallbackFunc(i.handleEvent),
+	)
+	i.pipeline.AddOutput(output)
+	// Start pipeline
+	if err := i.pipeline.Start(); err != nil {
+		logger.Fatalf("failed to start pipeline: %s\n", err)
+	}
+	// Start error handler
+	go func() {
+		err, ok := <-i.pipeline.ErrorChan()
+		if ok {
+			logger.Fatalf("pipeline failed: %s\n", err)
+		}
+	}()
+	return nil
+}
+
+func (i *Indexer) handleEvent(evt event.Event) error {
+	logger := logging.GetLogger()
+	eventTx := evt.Payload.(input_chainsync.TransactionEvent)
+	for _, txOutput := range eventTx.Outputs {
+		datum := txOutput.Datum()
+		if datum != nil {
+			if _, err := datum.Decode(); err != nil {
+				logger.Warnf("error decoding TX (%s) output datum: %s", eventTx.TransactionHash, err)
+				return err
+			}
+			datumFields := datum.Value().(cbor.Constructor).Fields()
+			domainName := string(datumFields[0].(cbor.ByteString).Bytes())
+			nsRecords := []string{}
+			for _, record := range datumFields[1].([]any) {
+				nsRecords = append(
+					nsRecords,
+					string(record.(cbor.ByteString).Bytes()),
+				)
+			}
+			logger.Infof("found domain %s with NS records: %v", domainName, nsRecords)
+		}
+	}
+	return nil
+}
+
+// GetIndexer returns the global indexer instance
+func GetIndexer() *Indexer {
+	return globalIndexer
+}