feat: initial Handshake indexer support

Signed-off-by: Aurora Gaffney <aurora@blinklabs.io>

Author: agaffney

internal/config/config.go

@@ -50,13 +50,14 @@ type MetricsConfig struct {
 }
 
 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"`
-	InterceptHash string `yaml:"interceptHash" envconfig:"INDEXER_INTERCEPT_HASH"`
-	InterceptSlot uint64 `yaml:"interceptSlot" envconfig:"INDEXER_INTERCEPT_SLOT"`
-	Verify        bool   `yaml:"verify"        envconfig:"INDEXER_VERIFY"`
+	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"`
+	InterceptHash    string `yaml:"interceptHash"    envconfig:"INDEXER_INTERCEPT_HASH"`
+	InterceptSlot    uint64 `yaml:"interceptSlot"    envconfig:"INDEXER_INTERCEPT_SLOT"`
+	Verify           bool   `yaml:"verify"           envconfig:"INDEXER_VERIFY"`
+	HandshakeAddress string `yaml:"handshakeAddress" envconfig:"INDEXER_HANDSHAKE_ADDRESS"`
 }
 
 type StateConfig struct {

internal/handshake/peer.go

@@ -99,7 +99,10 @@ func (p *Peer) Close() error {
 		return err
 	}
 	p.conn = nil
+	// Close done channel to signify shutdown internally
 	close(p.doneCh)
+	// Close error channel to signify shutdown to consumer
+	close(p.errorCh)
 	return nil
 }
 
@@ -187,6 +190,11 @@ func (p *Peer) recvLoop() {
 		}
 	}()
 	if err != nil {
+		// Don't return an async error if we're already shutting down
+		select {
+		case <-p.doneCh:
+			return
+		}
 		p.errorCh <- err
 		_ = p.Close()
 	}
@@ -473,6 +481,11 @@ func (p *Peer) Sync(locator [][32]byte, syncFunc SyncFunc) error {
 			}
 		}()
 		if err != nil {
+			// Don't return an async error if we're already shutting down
+			select {
+			case <-p.doneCh:
+				return
+			}
 			p.errorCh <- err
 			_ = p.Close()
 		}

internal/indexer/handshake.go

@@ -0,0 +1,100 @@
+// Copyright 2025 Blink Labs Software
+//
+// Use of this source code is governed by an MIT-style
+// license that can be found in the LICENSE file or at
+// https://opensource.org/licenses/MIT.
+
+package indexer
+
+import (
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/blinklabs-io/cdnsd/internal/config"
+	"github.com/blinklabs-io/cdnsd/internal/handshake"
+)
+
+type handshakeState struct {
+	peer             *handshake.Peer
+	peerAddress      string
+	peerBackoffDelay time.Duration
+	blockHeight      int
+}
+
+func (i *Indexer) startHandshake() error {
+	cfg := config.GetConfig()
+	if cfg.Indexer.HandshakeAddress == "" {
+		return nil
+	}
+	i.handshakeState.peerAddress = cfg.Indexer.HandshakeAddress
+	if err := i.handshakeConnectPeer(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (i *Indexer) handshakeConnectPeer() error {
+	slog.Info("connecting to Handshake peer", "address", i.handshakeState.peerAddress)
+	p, err := handshake.NewPeer(nil, handshake.NetworkMainnet)
+	if err != nil {
+		return err
+	}
+	i.handshakeState.peer = p
+	if err := i.handshakeState.peer.Connect(i.handshakeState.peerAddress); err != nil {
+		return err
+	}
+	// Async error handler
+	go func() {
+		err := <-i.handshakeState.peer.ErrorChan()
+		if err != nil {
+			slog.Error(
+				"peer disconnected",
+				"error",
+				err,
+			)
+		}
+		// Try reconnecting to peer until we are successful
+		for {
+			if err := i.handshakeConnectPeer(); err == nil {
+				i.handshakeState.peerBackoffDelay = 0
+				return
+			}
+			if i.handshakeState.peerBackoffDelay == 0 {
+				// Set initial backoff delay
+				i.handshakeState.peerBackoffDelay = 1 * time.Second
+			} else {
+				// Double backoff delay
+				i.handshakeState.peerBackoffDelay *= 2
+			}
+			// Don't delay longer than 2m
+			if i.handshakeState.peerBackoffDelay > 120*time.Second {
+				i.handshakeState.peerBackoffDelay = 120 * time.Second
+			}
+			slog.Error(
+				"connection to Handshake peer failed",
+				"error",
+				err,
+				"delay",
+				i.handshakeState.peerBackoffDelay.String(),
+			)
+			time.Sleep(i.handshakeState.peerBackoffDelay)
+		}
+	}()
+	// Start sync
+	if err := i.handshakeState.peer.Sync(nil, i.handshakeHandleSync); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (i *Indexer) handshakeHandleSync(block *handshake.Block) error {
+	i.handshakeState.blockHeight++
+	slog.Debug(
+		"synced Handshake block",
+		"height", i.handshakeState.blockHeight,
+		"hash", fmt.Sprintf("%x", block.Hash()),
+		"prevHash", fmt.Sprintf("%x", block.Header.PrevBlock),
+	)
+	return nil
+}

internal/indexer/indexer.go

@@ -55,12 +55,13 @@ type Domain struct {
 }
 
 type Indexer struct {
-	pipeline     *pipeline.Pipeline
-	domains      map[string]Domain
-	tipReached   bool
-	syncLogTimer *time.Timer
-	syncStatus   input_chainsync.ChainSyncStatus
-	watched      []watchedAddr
+	pipeline       *pipeline.Pipeline
+	domains        map[string]Domain
+	tipReached     bool
+	syncLogTimer   *time.Timer
+	syncStatus     input_chainsync.ChainSyncStatus
+	watched        []watchedAddr
+	handshakeState handshakeState
 }
 
 type watchedAddr struct {
@@ -76,6 +77,16 @@ var globalIndexer = &Indexer{
 }
 
 func (i *Indexer) Start() error {
+	if err := i.startCardano(); err != nil {
+		return err
+	}
+	if err := i.startHandshake(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (i *Indexer) startCardano() error {
 	// Build watched addresses from enabled profiles
 	cfg := config.GetConfig()
 	for _, profile := range config.GetProfiles() {