internal: add event broadcaster to subscriber

alerts/alerts_test.go

@@ -47,7 +47,7 @@ var _ webhooks.WebhookStore = (*testWebhookStore)(nil)
 
 func TestWebhooks(t *testing.T) {
 	store := &testWebhookStore{}
-	mgr, err := webhooks.NewManager(zap.NewNop().Sugar(), store)
+	mgr, err := webhooks.NewManager(store, zap.NewNop())
 	if err != nil {
 		t.Fatal(err)
 	}

chain/chain.go

@@ -22,8 +22,9 @@ var ForEachHostAnnouncement = chain.ForEachHostAnnouncement
 
 type (
 	ChainManager interface {
-		Tip() types.ChainIndex
+		Block(id types.BlockID) (types.Block, bool)
 		OnReorg(fn func(types.ChainIndex)) (cancel func())
+		Tip() types.ChainIndex
 		UpdatesSince(index types.ChainIndex, max int) (rus []chain.RevertUpdate, aus []chain.ApplyUpdate, err error)
 	}
 

internal/bus/events.go

@@ -0,0 +1,31 @@
+package bus
+
+import (
+	"context"
+	"time"
+
+	"go.sia.tech/renterd/webhooks"
+	"go.uber.org/zap"
+)
+
+type (
+	EventBroadcaster struct {
+		broadcaster webhooks.Broadcaster
+		logger      *zap.SugaredLogger
+	}
+)
+
+func NewEventBroadcaster(b webhooks.Broadcaster, l *zap.Logger) EventBroadcaster {
+	return EventBroadcaster{
+		broadcaster: b,
+		logger:      l.Named("events").Sugar(),
+	}
+}
+
+func (b EventBroadcaster) BroadcastEvent(e webhooks.WebhookEvent) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	if err := b.broadcaster.BroadcastAction(ctx, e.Event()); err != nil {
+		b.logger.Errorw("failed to broadcast event", "event", e, "error", err)
+	}
+	cancel()
+}

internal/node/chainsubscriber.go

@@ -11,6 +11,7 @@ import (
 	"go.sia.tech/coreutils/wallet"
 	"go.sia.tech/renterd/api"
 	"go.sia.tech/renterd/chain"
+	"go.sia.tech/renterd/internal/bus"
 	"go.sia.tech/renterd/internal/utils"
 	"go.uber.org/zap"
 )
@@ -29,6 +30,7 @@ type (
 	ChainSubscriber struct {
 		cm     chain.ChainManager
 		cs     chain.ChainStore
+		events bus.EventBroadcaster
 		logger *zap.SugaredLogger
 
 		announcementMaxAge time.Duration
@@ -66,7 +68,7 @@ type (
 // NewChainSubscriber creates a new chain subscriber that will sync with the
 // given chain manager and chain store. The returned subscriber is already
 // running and can be shut down by calling the Close method.
-func NewChainSubscriber(cm chain.ChainManager, cs chain.ChainStore, walletAddress types.Address, announcementMaxAge time.Duration, logger *zap.Logger) (_ *ChainSubscriber, err error) {
+func NewChainSubscriber(cm chain.ChainManager, cs chain.ChainStore, events bus.EventBroadcaster, walletAddress types.Address, announcementMaxAge time.Duration, logger *zap.Logger) (_ *ChainSubscriber, err error) {
 	if announcementMaxAge == 0 {
 		return nil, errors.New("announcementMaxAge must be non-zero")
 	}
@@ -76,6 +78,7 @@ func NewChainSubscriber(cm chain.ChainManager, cs chain.ChainStore, walletAddres
 	subscriber := &ChainSubscriber{
 		cm:     cm,
 		cs:     cs,
+		events: events,
 		logger: logger.Sugar(),
 
 		announcementMaxAge: announcementMaxAge,
@@ -246,6 +249,8 @@ func (s *ChainSubscriber) sync() error {
 		}
 		s.logger.Debugw("processed updates successfully", "new_height", index.Height, "new_block_id", index.ID, "ms", time.Since(istart).Milliseconds())
 		cnt++
+
+		// TODO: broadcast consensus update
 	}
 
 	s.logger.Debugw("sync completed", "start_height", index.Height, "block_id", index.ID, "ms", time.Since(start).Milliseconds(), "iterations", cnt)

internal/node/node.go

@@ -20,6 +20,7 @@ import (
 	"go.sia.tech/renterd/bus"
 	"go.sia.tech/renterd/chain"
 	"go.sia.tech/renterd/config"
+	ibus "go.sia.tech/renterd/internal/bus"
 	"go.sia.tech/renterd/stores"
 	"go.sia.tech/renterd/webhooks"
 	"go.sia.tech/renterd/worker"
@@ -133,14 +134,17 @@ func NewBus(cfg BusConfig, dir string, seed types.PrivateKey, logger *zap.Logger
 	}
 
 	// create webhooks manager
-	wh, err := webhooks.NewManager(logger.Named("webhooks").Sugar(), sqlStore)
+	wh, err := webhooks.NewManager(sqlStore, logger)
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}
 
 	// hookup webhooks <-> alerts
 	alertsMgr.RegisterWebhookBroadcaster(wh)
 
+	// create event broadcaster
+	events := ibus.NewEventBroadcaster(wh, logger)
+
 	// create chain manager
 	store, state, err := chain.NewDBStore(bdb, cfg.Network, cfg.Genesis)
 	if err != nil {
@@ -149,7 +153,7 @@ func NewBus(cfg BusConfig, dir string, seed types.PrivateKey, logger *zap.Logger
 	cm := chain.NewManager(store, state)
 
 	// create chain subscriber
-	cs, err := NewChainSubscriber(cm, sqlStore, types.StandardUnlockHash(seed.PublicKey()), time.Duration(cfg.AnnouncementMaxAgeHours)*time.Hour, logger.Named("chainsubscriber"))
+	cs, err := NewChainSubscriber(cm, sqlStore, events, types.StandardUnlockHash(seed.PublicKey()), time.Duration(cfg.AnnouncementMaxAgeHours)*time.Hour, logger.Named("chainsubscriber"))
 	if err != nil {
 		return nil, nil, nil, nil, err
 	}

webhooks/webhooks.go

@@ -56,6 +56,10 @@ type (
 		Event   string      `json:"event"`
 		Payload interface{} `json:"payload,omitempty"`
 	}
+
+	WebhookEvent interface {
+		Event() Event
+	}
 )
 
 type Manager struct {
@@ -213,10 +217,10 @@ func (w Webhook) String() string {
 	return fmt.Sprintf("%v.%v.%v", w.URL, w.Module, w.Event)
 }
 
-func NewManager(logger *zap.SugaredLogger, store WebhookStore) (*Manager, error) {
+func NewManager(store WebhookStore, logger *zap.Logger) (*Manager, error) {
 	shutdownCtx, shutdownCtxCancel := context.WithCancel(context.Background())
 	m := &Manager{
-		logger: logger.Named("webhooks"),
+		logger: logger.Named("webhooks").Sugar(),
 		store:  store,
 
 		shutdownCtx:       shutdownCtx,