refactor(event): move block, rollback, and transaction from input/chainsync (#525)

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

Author: wolf31o2

input/chainsync/block.go renamed to event/block.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2025 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,25 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package chainsync
+package event
 
 import (
 	"github.com/blinklabs-io/gouroboros/ledger"
 )
 
 type BlockContext struct {
+	Era          string `json:"era"`
 	BlockNumber  uint64 `json:"blockNumber"`
 	SlotNumber   uint64 `json:"slotNumber"`
 	NetworkMagic uint32 `json:"networkMagic"`
-	Era          string `json:"era"`
 }
 
 type BlockEvent struct {
 	Block            ledger.Block     `json:"-"`
-	BlockBodySize    uint64           `json:"blockBodySize"`
 	IssuerVkey       string           `json:"issuerVkey"`
 	BlockHash        string           `json:"blockHash"`
 	BlockCbor        byteSliceJsonHex `json:"blockCbor,omitempty"`
+	BlockBodySize    uint64           `json:"blockBodySize"`
 	TransactionCount uint64           `json:"transactionCount"`
 }
 

event/event.go

@@ -19,10 +19,10 @@ import (
 )
 
 type Event struct {
-	Type      string    `json:"type"`
 	Timestamp time.Time `json:"timestamp"`
 	Context   any       `json:"context,omitempty"`
 	Payload   any       `json:"payload"`
+	Type      string    `json:"type"`
 }
 
 func New(

input/chainsync/json.go renamed to event/json.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2025 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package chainsync
+package event
 
 import (
 	"encoding/hex"

input/chainsync/rollback.go renamed to event/rollback.go

@@ -1,4 +1,4 @@
-// Copyright 2023 Blink Labs Software
+// Copyright 2025 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package chainsync
+package event
 
 import (
 	"encoding/hex"

input/chainsync/tx.go renamed to event/tx.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package chainsync
+package event
 
 import (
 	"github.com/blinklabs-io/gouroboros/cbor"
@@ -21,27 +21,27 @@ import (
 )
 
 type TransactionContext struct {
+	TransactionHash string `json:"transactionHash"`
 	BlockNumber     uint64 `json:"blockNumber"`
 	SlotNumber      uint64 `json:"slotNumber"`
-	TransactionHash string `json:"transactionHash"`
 	TransactionIdx  uint32 `json:"transactionIdx"`
 	NetworkMagic    uint32 `json:"networkMagic"`
 }
 
 type TransactionEvent struct {
 	Transaction     ledger.Transaction            `json:"-"`
+	Witnesses       lcommon.TransactionWitnessSet `json:"witnesses,omitempty"`
+	Withdrawals     map[string]uint64             `json:"withdrawals,omitempty"`
+	Metadata        *cbor.LazyValue               `json:"metadata,omitempty"`
 	BlockHash       string                        `json:"blockHash"`
-	TransactionCbor byteSliceJsonHex              `json:"transactionCbor,omitempty"`
-	Inputs          []ledger.TransactionInput     `json:"inputs"`
-	Outputs         []ledger.TransactionOutput    `json:"outputs"`
-	Certificates    []ledger.Certificate          `json:"certificates,omitempty"`
 	ReferenceInputs []ledger.TransactionInput     `json:"referenceInputs,omitempty"`
-	Metadata        *cbor.LazyValue               `json:"metadata,omitempty"`
+	Certificates    []ledger.Certificate          `json:"certificates,omitempty"`
+	Outputs         []ledger.TransactionOutput    `json:"outputs"`
+	ResolvedInputs  []ledger.TransactionOutput    `json:"resolvedInputs,omitempty"`
+	Inputs          []ledger.TransactionInput     `json:"inputs"`
+	TransactionCbor byteSliceJsonHex              `json:"transactionCbor,omitempty"`
 	Fee             uint64                        `json:"fee"`
 	TTL             uint64                        `json:"ttl,omitempty"`
-	ResolvedInputs  []ledger.TransactionOutput    `json:"resolvedInputs,omitempty"`
-	Withdrawals     map[string]uint64             `json:"withdrawals,omitempty"`
-	Witnesses       lcommon.TransactionWitnessSet `json:"witnesses,omitempty"`
 }
 
 func NewTransactionContext(

filter/chainsync/chainsync.go

@@ -1,4 +1,4 @@
-// Copyright 2024 Blink Labs Software
+// Copyright 2025 Blink Labs Software
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -19,7 +19,6 @@ import (
 	"strings"
 
 	"github.com/blinklabs-io/adder/event"
-	"github.com/blinklabs-io/adder/input/chainsync"
 	"github.com/blinklabs-io/adder/plugin"
 	"github.com/blinklabs-io/gouroboros/ledger"
 	"github.com/blinklabs-io/gouroboros/ledger/common"
@@ -61,14 +60,14 @@ func (c *ChainSync) Start() error {
 				return
 			}
 			switch v := evt.Payload.(type) {
-			case chainsync.BlockEvent:
+			case event.BlockEvent:
 				// Check pool filter
 				if len(c.filterPoolIds) > 0 {
 					filterMatched := false
 					for _, filterPoolId := range c.filterPoolIds {
 						isPoolBech32 := strings.HasPrefix(filterPoolId, "pool")
 						foundMatch := false
-						be := evt.Payload.(chainsync.BlockEvent)
+						be := evt.Payload.(event.BlockEvent)
 						if be.IssuerVkey == filterPoolId {
 							foundMatch = true
 						} else if isPoolBech32 {
@@ -100,7 +99,7 @@ func (c *ChainSync) Start() error {
 						continue
 					}
 				}
-			case chainsync.TransactionEvent:
+			case event.TransactionEvent:
 				// Check address filter
 				if len(c.filterAddresses) > 0 {
 					filterMatched := false

filter/chainsync/chainsync_test.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	"github.com/blinklabs-io/adder/event"
-	"github.com/blinklabs-io/adder/input/chainsync"
 	"github.com/blinklabs-io/gouroboros/cbor"
 	"github.com/blinklabs-io/gouroboros/ledger"
 	"github.com/blinklabs-io/gouroboros/ledger/common"
@@ -303,7 +302,7 @@ func TestFilterByAddress(t *testing.T) {
 				datum:   nil,
 			}
 
-			txEvent := chainsync.TransactionEvent{
+			txEvent := event.TransactionEvent{
 				Outputs:        []ledger.TransactionOutput{output},
 				ResolvedInputs: []ledger.TransactionOutput{output},
 			}
@@ -365,7 +364,7 @@ func TestFilterByPolicyId(t *testing.T) {
 		datum:   nil,
 	}
 	evt := event.Event{
-		Payload: chainsync.TransactionEvent{
+		Payload: event.TransactionEvent{
 			Outputs:        []ledger.TransactionOutput{output},
 			ResolvedInputs: []ledger.TransactionOutput{output},
 		},
@@ -418,7 +417,7 @@ func TestFilterByAssetFingerprint(t *testing.T) {
 		datum:   nil,
 	}
 	evt := event.Event{
-		Payload: chainsync.TransactionEvent{
+		Payload: event.TransactionEvent{
 			Outputs:        []ledger.TransactionOutput{output},
 			ResolvedInputs: []ledger.TransactionOutput{output},
 		},
@@ -452,7 +451,7 @@ func TestFilterByPoolId(t *testing.T) {
 
 	// Mock block event
 	evt := event.Event{
-		Payload: chainsync.BlockEvent{
+		Payload: event.BlockEvent{
 			IssuerVkey: "pool1", // Match the filterPoolIds
 		},
 	}

filter/chainsync/plugin_test.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/blinklabs-io/adder/event"
-	"github.com/blinklabs-io/adder/input/chainsync"
 	"github.com/blinklabs-io/adder/plugin"
 	"github.com/stretchr/testify/assert"
 )
@@ -73,7 +72,7 @@ func TestPluginEventProcessing(t *testing.T) {
 	testEvent := event.Event{
 		Type:      "transaction",
 		Timestamp: time.Now(),
-		Payload:   chainsync.TransactionEvent{},
+		Payload:   event.TransactionEvent{},
 	}
 
 	// Send the event to the input channel

input/chainsync/block_test.go

@@ -3,6 +3,7 @@ package chainsync
 import (
 	"testing"
 
+	"github.com/blinklabs-io/adder/event"
 	"github.com/blinklabs-io/gouroboros/ledger/common"
 	"github.com/stretchr/testify/assert"
 	utxorpc "github.com/utxorpc/go-codegen/utxorpc/v1alpha/cardano"
@@ -197,7 +198,7 @@ func TestNewBlockContext(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			blockContext := NewBlockContext(tc.block, tc.networkMagic)
+			blockContext := event.NewBlockContext(tc.block, tc.networkMagic)
 			assert.Equal(
 				t,
 				tc.expectedEra,
@@ -279,7 +280,7 @@ func TestNewBlockContextEdgeCases(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			blockContext := NewBlockContext(tc.block, tc.networkMagic)
+			blockContext := event.NewBlockContext(tc.block, tc.networkMagic)
 			assert.Equal(
 				t,
 				tc.expectedEra,

input/chainsync/chainsync.go

@@ -51,40 +51,40 @@ const (
 )
 
 type ChainSync struct {
-	oConn              *ouroboros.Connection
 	logger             plugin.Logger
-	network            string
-	networkMagic       uint32
-	address            string
-	socketPath         string
-	ntcTcp             bool
-	intersectTip       bool
-	intersectPoints    []ocommon.Point
-	includeCbor        bool
-	autoReconnect      bool
-	autoReconnectDelay time.Duration
 	statusUpdateFunc   StatusUpdateFunc
-	status             *ChainSyncStatus
-	errorChan          chan error
+	blockfetchDoneChan chan struct{}
+	kupoClient         *kugo.Client
+	oConn              *ouroboros.Connection
 	eventChan          chan event.Event
-	cursorCache        []ocommon.Point
-	dialAddress        string
+	errorChan          chan error
+	status             *ChainSyncStatus
 	dialFamily         string
 	kupoUrl            string
-	kupoClient         *kugo.Client
-	delayConfirmations uint
-	delayBuffer        [][]event.Event
+	network            string
+	socketPath         string
+	dialAddress        string
+	address            string
+	intersectPoints    []ocommon.Point
 	pendingBlockPoints []ocommon.Point
-	blockfetchDoneChan chan struct{}
+	delayBuffer        [][]event.Event
+	cursorCache        []ocommon.Point
 	lastTip            ochainsync.Tip
+	delayConfirmations uint
+	autoReconnectDelay time.Duration
+	networkMagic       uint32
+	includeCbor        bool
+	ntcTcp             bool
+	intersectTip       bool
+	autoReconnect      bool
 }
 
 type ChainSyncStatus struct {
+	BlockHash     string
+	TipBlockHash  string
 	SlotNumber    uint64
 	BlockNumber   uint64
-	BlockHash     string
 	TipSlotNumber uint64
-	TipBlockHash  string
 	TipReached    bool
 }
 
@@ -297,15 +297,15 @@ func (c *ChainSync) handleRollBackward(
 		"chainsync.rollback",
 		time.Now(),
 		nil,
-		NewRollbackEvent(point),
+		event.NewRollbackEvent(point),
 	)
 	// Remove rolled-back events from buffer
 	if len(c.delayBuffer) > 0 {
 		// We iterate backwards to avoid the issues with deleting from a list while iterating over it
 		for i := len(c.delayBuffer) - 1; i >= 0; i-- {
 			for _, evt := range c.delayBuffer[i] {
 				// Look for block event
-				if blockEvtCtx, ok := evt.Context.(BlockContext); ok {
+				if blockEvtCtx, ok := evt.Context.(event.BlockContext); ok {
 					// Delete event batch if slot is after rollback point
 					if blockEvtCtx.SlotNumber > point.Slot {
 						c.delayBuffer = slices.Delete(c.delayBuffer, i, i+1)
@@ -367,8 +367,8 @@ func (c *ChainSync) handleRollForward(
 	blockEvt := event.New(
 		"chainsync.block",
 		time.Now(),
-		NewBlockHeaderContext(block.Header()),
-		NewBlockEvent(block, c.includeCbor),
+		event.NewBlockHeaderContext(block.Header()),
+		event.NewBlockEvent(block, c.includeCbor),
 	)
 	tmpEvents = append(tmpEvents, blockEvt)
 	for t, transaction := range block.Transactions() {
@@ -382,13 +382,13 @@ func (c *ChainSync) handleRollForward(
 		txEvt := event.New(
 			"chainsync.transaction",
 			time.Now(),
-			NewTransactionContext(
+			event.NewTransactionContext(
 				block,
 				transaction,
 				uint32(t),
 				c.networkMagic,
 			),
-			NewTransactionEvent(
+			event.NewTransactionEvent(
 				block,
 				transaction,
 				c.includeCbor,
@@ -416,7 +416,7 @@ func (c *ChainSync) handleRollForward(
 		if uint(len(c.delayBuffer)) > c.delayConfirmations {
 			for _, evt := range c.delayBuffer[0] {
 				// Look for block event
-				if blockEvt, ok := evt.Payload.(BlockEvent); ok {
+				if blockEvt, ok := evt.Payload.(event.BlockEvent); ok {
 					// Populate current point for update status based on most recently sent events
 					updateTip = ochainsync.Tip{
 						Point: ocommon.Point{
@@ -449,8 +449,8 @@ func (c *ChainSync) handleBlockFetchBlock(
 	blockEvt := event.New(
 		"chainsync.block",
 		time.Now(),
-		NewBlockContext(block, c.networkMagic),
-		NewBlockEvent(block, c.includeCbor),
+		event.NewBlockContext(block, c.networkMagic),
+		event.NewBlockEvent(block, c.includeCbor),
 	)
 	c.eventChan <- blockEvt
 	for t, transaction := range block.Transactions() {
@@ -464,13 +464,13 @@ func (c *ChainSync) handleBlockFetchBlock(
 		txEvt := event.New(
 			"chainsync.transaction",
 			time.Now(),
-			NewTransactionContext(
+			event.NewTransactionContext(
 				block,
 				transaction,
 				uint32(t),
 				c.networkMagic,
 			),
-			NewTransactionEvent(
+			event.NewTransactionEvent(
 				block,
 				transaction,
 				c.includeCbor,