Start listener manager, bring back eth_deposits

app/node/build.go

@@ -24,6 +24,7 @@ import (
 	blockprocessor "github.com/kwilteam/kwil-db/node/block_processor"
 	"github.com/kwilteam/kwil-db/node/consensus"
 	"github.com/kwilteam/kwil-db/node/engine/execution"
+	"github.com/kwilteam/kwil-db/node/listeners"
 	"github.com/kwilteam/kwil-db/node/mempool"
 	"github.com/kwilteam/kwil-db/node/meta"
 	"github.com/kwilteam/kwil-db/node/pg"
@@ -88,6 +89,9 @@ func buildServer(ctx context.Context, d *coreDependencies) *server {
 	// Node
 	node := buildNode(d, mp, bs, ce, ss, db)
 
+	// listeners
+	lm := buildListenerManager(d, es, txApp, node)
+
 	// RPC Services
 	rpcSvcLogger := d.logger.New("USER")
 	jsonRPCTxSvc := usersvc.NewService(db, e, node, bp, vs, rpcSvcLogger,
@@ -135,6 +139,7 @@ func buildServer(ctx context.Context, d *coreDependencies) *server {
 		closers:            closers,
 		node:               node,
 		ce:                 ce,
+		listeners:          lm,
 		jsonRPCServer:      jsonRPCServer,
 		jsonRPCAdminServer: jsonRPCAdminServer,
 		dbCtx:              db,
@@ -202,17 +207,23 @@ func buildMetaStore(ctx context.Context, db *pg.DB) {
 	}
 }
 
+// service returns a common.Service with the given logger name
+func (c *coreDependencies) service(loggerName string) *common.Service {
+	signer := auth.GetNodeSigner(c.privKey)
+
+	return &common.Service{
+		Logger:        c.logger.New(loggerName),
+		GenesisConfig: c.genesisCfg,
+		LocalConfig:   c.cfg,
+		Identity:      signer.Identity(),
+	}
+}
+
 func buildTxApp(ctx context.Context, d *coreDependencies, db *pg.DB, accounts *accounts.Accounts,
 	votestore *voting.VoteStore, engine *execution.GlobalContext) *txapp.TxApp {
 	signer := auth.GetNodeSigner(d.privKey)
-	service := &common.Service{
-		Logger:   d.logger.New("TXAPP"),
-		Identity: signer.Identity(),
-		// TODO: pass extension configs
-		// ExtensionConfigs: make(map[string]map[string]string),
-	}
 
-	txapp, err := txapp.NewTxApp(ctx, db, engine, signer, nil, service, accounts, votestore)
+	txapp, err := txapp.NewTxApp(ctx, db, engine, signer, nil, d.service("TxAPP"), accounts, votestore)
 	if err != nil {
 		failBuild(err, "failed to create txapp")
 	}
@@ -350,6 +361,10 @@ func buildSnapshotStore(d *coreDependencies) *snapshotter.SnapshotStore {
 	return ss
 }
 
+func buildListenerManager(d *coreDependencies, ev *voting.EventStore, txapp *txapp.TxApp, node *node.Node) *listeners.ListenerManager {
+	return listeners.NewListenerManager(d.service("ListenerManager"), ev, txapp, node)
+}
+
 func buildJRPCAdminServer(d *coreDependencies) *rpcserver.Server {
 	var wantTLS bool
 	addr := d.cfg.Admin.ListenAddress

app/node/node.go

@@ -15,6 +15,7 @@ import (
 	"github.com/kwilteam/kwil-db/core/log"
 	"github.com/kwilteam/kwil-db/node"
 	"github.com/kwilteam/kwil-db/node/consensus"
+	"github.com/kwilteam/kwil-db/node/listeners"
 	rpcserver "github.com/kwilteam/kwil-db/node/services/jsonrpc"
 	"github.com/kwilteam/kwil-db/version"
 
@@ -35,6 +36,7 @@ type server struct {
 	// subsystems
 	node               *node.Node
 	ce                 *consensus.ConsensusEngine
+	listeners          *listeners.ListenerManager
 	jsonRPCServer      *rpcserver.Server
 	jsonRPCAdminServer *rpcserver.Server
 }
@@ -164,6 +166,17 @@ func (s *server) Start(ctx context.Context) error {
 		return nil
 	})
 
+	// Start listener manager
+	group.Go(func() error {
+		go func() {
+			<-groupCtx.Done()
+			s.log.Info("stop listeners")
+			s.listeners.Stop()
+		}()
+		return s.listeners.Start()
+	})
+	s.log.Info("listener manager started")
+
 	// TODO: node is starting the consensus engine for ease of testing
 	// Start the consensus engine
 

core/rpc/client/admin/jsonrpc/client.go

@@ -147,7 +147,6 @@ func (cl *Client) Status(ctx context.Context) (*adminTypes.Status, error) {
 			Syncing:         res.Sync.Syncing,
 		},
 		Validator: &adminTypes.ValidatorInfo{
-			Role:   res.Validator.Role,
 			PubKey: res.Validator.PubKey,
 			Power:  res.Validator.Power,
 		},

core/rpc/json/admin/responses.go

@@ -45,7 +45,6 @@ type SyncInfo struct {
 type HealthResponse struct {
 	Version       string         `json:"version"`
 	Healthy       bool           `json:"healthy"`
-	Role          string         `json:"role"`
 	PubKey        types.HexBytes `json:"pubkey"`
 	NumValidators int            `json:"num_validators"`
 }

core/types/types.go

@@ -40,7 +40,6 @@ type JoinRequest struct {
 }
 
 type Validator struct {
-	Role   string   `json:"role"`
 	PubKey HexBytes `json:"pubkey"`
 	Power  int64    `json:"power"`
 }