refactor(events): make events handler as service (#1980)

Signed-off-by: Artur Troian <[email protected]>
Co-authored-by: Artur Troian <[email protected]>

Author: troian

events/cmd/root.go

@@ -50,13 +50,20 @@ func getEvents(ctx context.Context, cmd *cobra.Command, _ []string) error {
 	group, ctx := errgroup.WithContext(ctx)
 
 	subscriber, err := bus.Subscribe()
+	if err != nil {
+		return err
+	}
 
+	evtSvc, err := events.NewEvents(ctx, cctx.Client, "akash-cli", bus)
 	if err != nil {
 		return err
 	}
 
 	group.Go(func() error {
-		return events.Publish(ctx, cctx.Client, "akash-cli", bus)
+		<-ctx.Done()
+		evtSvc.Shutdown()
+
+		return nil
 	})
 
 	group.Go(func() error {

events/publish.go

@@ -3,13 +3,13 @@ package events
 import (
 	"context"
 
-	"golang.org/x/sync/errgroup"
-
+	"github.com/boz/go-lifecycle"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	abci "github.com/tendermint/tendermint/abci/types"
 	tmclient "github.com/tendermint/tendermint/rpc/client"
 	ctypes "github.com/tendermint/tendermint/rpc/core/types"
 	tmtmtypes "github.com/tendermint/tendermint/types"
+	"golang.org/x/sync/errgroup"
 
 	atypes "github.com/akash-network/akash-api/go/node/audit/v1beta3"
 	dtypes "github.com/akash-network/akash-api/go/node/deployment/v1beta3"
@@ -20,57 +20,124 @@ import (
 	"github.com/akash-network/node/pubsub"
 )
 
-// Publish events using tm buses to clients. Waits on context
-// shutdown signals to exit.
-func Publish(ctx context.Context, client tmclient.Client, name string, bus pubsub.Bus) (err error) {
+type events struct {
+	ctx    context.Context
+	group  *errgroup.Group
+	client tmclient.Client
+	bus    pubsub.Bus
+	lc     lifecycle.Lifecycle
+}
+
+// Service represents an event monitoring service that subscribes to and processes blockchain events.
+// It monitors block headers and various transaction events, publishing them to a message bus.
+type Service interface {
+	// Shutdown gracefully stops the event monitoring service and cleans up resources.
+	// Once called, the service will unsubscribe from events and complete any pending operations.
+	Shutdown()
+}
+
+// NewEvents creates and initializes a new blockchain event monitoring service.
+//
+// Parameters:
+//   - pctx: Parent context for controlling the service lifecycle
+//   - client: Tendermint RPC client for interacting with the blockchain
+//   - name: Service name used as a prefix for subscription identifiers
+//   - bus: Message bus for publishing processed events
+//
+// Returns:
+//   - Service: A running event monitoring service interface
+//   - error: Any error encountered during service initialization
+//
+// The service subscribes to block header events and processes them to extract and publish
+// various transaction events (deployment, market, provider, audit) to the provided message bus.
+// The service starts monitoring events immediately and will continue until either the context
+// is canceled or Shutdown() is called.
+func NewEvents(pctx context.Context, client tmclient.Client, name string, bus pubsub.Bus) (Service, error) {
+	group, ctx := errgroup.WithContext(pctx)
+
+	ev := &events{
+		ctx:    ctx,
+		group:  group,
+		client: client,
+		lc:     lifecycle.New(),
+		bus:    bus,
+	}
+
 	const (
 		queuesz = 1000
 	)
-	var (
-		blkHeaderName = name + "-blk-hdr"
-	)
+
+	var blkHeaderName = name + "-blk-hdr"
 
 	tmbus := client.(tmclient.EventsClient)
 
 	blkch, err := tmbus.Subscribe(ctx, blkHeaderName, blkHeaderQuery().String(), queuesz)
 	if err != nil {
-		return err
+		return nil, err
 	}
-	defer func() {
-		err = tmbus.UnsubscribeAll(ctx, blkHeaderName)
-	}()
 
-	g, ctx := errgroup.WithContext(ctx)
+	startch := make(chan struct{}, 1)
+
+	group.Go(func() error {
+		ev.lc.WatchContext(ctx)
 
-	g.Go(func() error {
-		return publishEvents(ctx, client, blkch, bus)
+		return ev.lc.Error()
 	})
 
-	return g.Wait()
+	group.Go(func() error {
+		return ev.run(blkHeaderName, blkch, startch)
+	})
+
+	select {
+	case <-pctx.Done():
+		return nil, pctx.Err()
+	case <-startch:
+		return ev, nil
+	}
+}
+
+func (e *events) Shutdown() {
+	_, stopped := <-e.lc.Done()
+	if stopped {
+		return
+	}
+
+	e.lc.Shutdown(nil)
+
+	_ = e.group.Wait()
 }
 
-func publishEvents(ctx context.Context, client tmclient.Client, ch <-chan ctypes.ResultEvent, bus pubsub.Bus) error {
-	var err error
+func (e *events) run(subs string, ch <-chan ctypes.ResultEvent, startch chan<- struct{}) error {
+	tmbus := e.client.(tmclient.EventsClient)
+
+	defer func() {
+		_ = tmbus.UnsubscribeAll(e.ctx, subs)
+
+		e.lc.ShutdownCompleted()
+	}()
+
+	startch <- struct{}{}
 
 loop:
 	for {
 		select {
-		case <-ctx.Done():
+		case err := <-e.lc.ShutdownRequest():
+			e.lc.ShutdownInitiated(err)
 			break loop
-		case ed := <-ch:
+		case ev := <-ch:
 			// nolint: gocritic
-			switch evt := ed.Data.(type) {
+			switch evt := ev.Data.(type) {
 			case tmtmtypes.EventDataNewBlockHeader:
-				processBlock(ctx, bus, client, evt.Header.Height)
+				e.processBlock(evt.Header.Height)
 			}
 		}
 	}
 
-	return err
+	return e.ctx.Err()
 }
 
-func processBlock(ctx context.Context, bus pubsub.Bus, client tmclient.Client, height int64) {
-	blkResults, err := client.BlockResults(ctx, &height)
+func (e *events) processBlock(height int64) {
+	blkResults, err := e.client.BlockResults(e.ctx, &height)
 	if err != nil {
 		return
 	}
@@ -82,11 +149,9 @@ func processBlock(ctx context.Context, bus pubsub.Bus, client tmclient.Client, h
 
 		for _, ev := range tx.Events {
 			if mev, ok := processEvent(ev); ok {
-				if err := bus.Publish(mev); err != nil {
-					bus.Close()
+				if err := e.bus.Publish(mev); err != nil {
 					return
 				}
-				continue
 			}
 		}
 	}

events/query.go

@@ -19,6 +19,5 @@ import (
 // }
 
 func blkHeaderQuery() pubsub.Query {
-	return tmquery.MustParse(
-		fmt.Sprintf("%s='%s'", tmtypes.EventTypeKey, tmtypes.EventNewBlockHeader))
+	return tmquery.MustParse(fmt.Sprintf("%s='%s'", tmtypes.EventTypeKey, tmtypes.EventNewBlockHeader))
 }