Add SSE support (#50)

* Add SSE support

* Add cors

* Fix linting

Author: galt-tr

cmd/arcade/main.go

@@ -14,6 +14,7 @@ import (
 	"go.uber.org/zap/zapcore"
 
 	"github.com/bsv-blockchain/arcade/config"
+	"github.com/bsv-blockchain/arcade/events"
 	"github.com/bsv-blockchain/arcade/kafka"
 	"github.com/bsv-blockchain/arcade/merkleservice"
 	"github.com/bsv-blockchain/arcade/services"
@@ -22,6 +23,7 @@ import (
 	"github.com/bsv-blockchain/arcade/services/p2p_client"
 	"github.com/bsv-blockchain/arcade/services/propagation"
 	"github.com/bsv-blockchain/arcade/services/tx_validator"
+	"github.com/bsv-blockchain/arcade/services/webhook"
 	"github.com/bsv-blockchain/arcade/store"
 	storefactory "github.com/bsv-blockchain/arcade/store/factory"
 	"github.com/bsv-blockchain/arcade/teranode"
@@ -129,7 +131,16 @@ func run(cmd *cobra.Command, _ []string) error {
 
 	txVal := validator.NewValidator(nil, nil) // Default policy, no chain tracker yet
 
-	svcs := buildServices(cfg, logger, producer, st, leaser, txTracker, teranodeClient, merkleClient, txVal)
+	// One process-wide events.Publisher routes status updates from every
+	// service that mutates state (validator, propagation, bump-builder,
+	// api-server) onto a shared Kafka topic. The api-server SSE handler and
+	// the webhook delivery service consume from that topic, so the
+	// transaction-status fan-out works whether the deployment is monolithic
+	// (mode=all) or split across pods.
+	publisher := events.NewKafkaPublisher(producer, logger)
+	defer func() { _ = publisher.Close() }()
+
+	svcs := buildServices(cfg, logger, producer, publisher, st, leaser, txTracker, teranodeClient, merkleClient, txVal)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -188,6 +199,7 @@ func buildServices(
 	cfg *config.Config,
 	logger *zap.Logger,
 	producer *kafka.Producer,
+	publisher events.Publisher,
 	st store.Store,
 	leaser store.Leaser,
 	txTracker *store.TxTracker,
@@ -202,16 +214,22 @@ func buildServices(
 	}
 
 	if shouldRun("api-server") {
-		svcs = append(svcs, api_server.New(cfg, logger, producer, st, txTracker, teranodeClient))
+		svcs = append(svcs, api_server.New(cfg, logger, producer, publisher, st, txTracker, teranodeClient))
 	}
 	if shouldRun("bump-builder") {
-		svcs = append(svcs, bump_builder.New(cfg, logger, producer, st, teranodeClient))
+		svcs = append(svcs, bump_builder.New(cfg, logger, producer, publisher, st, teranodeClient))
 	}
 	if shouldRun("tx-validator") {
-		svcs = append(svcs, tx_validator.New(cfg, logger, producer, st, txTracker, txVal))
+		svcs = append(svcs, tx_validator.New(cfg, logger, producer, publisher, st, txTracker, txVal))
 	}
 	if shouldRun("propagation") {
-		svcs = append(svcs, propagation.New(cfg, logger, producer, st, leaser, teranodeClient, merkleClient))
+		svcs = append(svcs, propagation.New(cfg, logger, producer, publisher, st, leaser, teranodeClient, merkleClient))
+	}
+	// Webhook delivery runs alongside api-server by default (so a single-binary
+	// deployment ships callbacks without extra config) but can be split into
+	// its own pod by setting mode=webhook.
+	if shouldRun("api-server") || shouldRun("webhook") {
+		svcs = append(svcs, webhook.New(cfg.Webhook, logger, publisher, st))
 	}
 	// p2p_client is its own service; it's needed both by mode=propagation
 	// (where it feeds the local teranode.Client directly) and by mode=p2p-client

config/config.go

@@ -108,6 +108,7 @@ type Config struct {
 	Propagation   PropagationConfig   `mapstructure:"propagation"`
 	TxValidator   TxValidatorConfig   `mapstructure:"tx_validator"`
 	BumpBuilder   BumpBuilderConfig   `mapstructure:"bump_builder"`
+	Webhook       WebhookConfig       `mapstructure:"webhook"`
 	// ChaintracksServer gates whether the embedded go-chaintracks HTTP API
 	// runs alongside api-server. Default is on so the refactor is a drop-in
 	// replacement for the original single-binary arcade.
@@ -277,6 +278,29 @@ type BumpBuilderConfig struct {
 	GraceWindowMs int `mapstructure:"grace_window_ms"`
 }
 
+// WebhookConfig tunes the HTTP webhook delivery service. The service
+// subscribes to status updates and POSTs them to each submission's
+// CallbackURL; failures are retried with exponential backoff persisted via
+// the store's UpdateDeliveryStatus.
+type WebhookConfig struct {
+	// MaxRetries caps how many times a failed POST is re-attempted before
+	// the submission is given up on. Mirrors arc's default of 10.
+	MaxRetries int `mapstructure:"max_retries"`
+	// ExpirationMinutes bounds the total wall-clock lifetime of a webhook
+	// delivery. Past this point the service stops retrying even if
+	// MaxRetries hasn't been hit. Defaults to 24 hours.
+	ExpirationMinutes int `mapstructure:"expiration_minutes"`
+	// InitialBackoffMs is the first retry delay; subsequent retries double
+	// it (capped). Defaults to 5s, matching arc.
+	InitialBackoffMs int `mapstructure:"initial_backoff_ms"`
+	// MaxBackoffMs caps how long backoff can grow between retries. Default
+	// 5 minutes.
+	MaxBackoffMs int `mapstructure:"max_backoff_ms"`
+	// HTTPTimeoutMs caps how long a single POST attempt may run. Default
+	// 10s — webhook receivers should ack fast or risk being timed out.
+	HTTPTimeoutMs int `mapstructure:"http_timeout_ms"`
+}
+
 // TxValidatorConfig tunes the parallel batch validation pipeline. Parallelism
 // caps how many transactions are parsed and validated concurrently inside a
 // single flush window — bounded so a huge in-flight batch can't open more
@@ -409,6 +433,12 @@ func setDefaults() {
 	viper.SetDefault("p2p.enable_mdns", false)
 	viper.SetDefault("p2p.allow_private_urls", false)
 
+	viper.SetDefault("webhook.max_retries", 10)
+	viper.SetDefault("webhook.expiration_minutes", 60*24)
+	viper.SetDefault("webhook.initial_backoff_ms", 5000)
+	viper.SetDefault("webhook.max_backoff_ms", 300000)
+	viper.SetDefault("webhook.http_timeout_ms", 10000)
+
 	viper.SetDefault("storage_path", "~/.arcade")
 	viper.SetDefault("chaintracks_server.enabled", true)
 	// Delegate chaintracks-library defaults (mode, network, bootstrap, p2p, …)

events/events.go

@@ -0,0 +1,26 @@
+// Package events provides a publish/subscribe abstraction for transaction
+// status updates. Mutations to TransactionStatus that originate in the
+// validator, propagation, bump-builder, and api-server services flow through
+// a Publisher so the SSE handler and webhook delivery service can fan them
+// out to clients without depending on which pod the mutation came from.
+//
+// The default backend is Kafka — every Subscribe call mints a unique consumer
+// group so each subscriber receives every event (mirroring the in-memory
+// broadcast semantics of the old arcade's monolithic Publisher).
+package events
+
+import (
+	"context"
+
+	"github.com/bsv-blockchain/arcade/models"
+)
+
+// Publisher is the contract every status-update fan-out backend implements.
+// Publish must be safe for concurrent use across goroutines. Subscribe
+// returns a channel the caller drains until ctx is canceled; on cancellation,
+// the channel is closed and any backend resources released.
+type Publisher interface {
+	Publish(ctx context.Context, status *models.TransactionStatus) error
+	Subscribe(ctx context.Context) (<-chan *models.TransactionStatus, error)
+	Close() error
+}

events/kafka_publisher.go

@@ -0,0 +1,161 @@
+package events
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"go.uber.org/zap"
+
+	"github.com/bsv-blockchain/arcade/kafka"
+	"github.com/bsv-blockchain/arcade/models"
+)
+
+// KafkaPublisher fans transaction status updates through a Kafka topic so
+// services running in different processes share a view of every status
+// change. Publish JSON-encodes the TransactionStatus and sends it under
+// kafka.TopicStatusUpdate keyed by txid (so updates for the same tx land on
+// the same partition in real Kafka). Subscribe spins up a dedicated
+// consumer group with a random ID — every caller gets every message,
+// regardless of how many other subscribers are running.
+type KafkaPublisher struct {
+	producer *kafka.Producer
+	logger   *zap.Logger
+
+	mu     sync.Mutex
+	closed bool
+	subs   []*kafkaSubscription
+}
+
+// NewKafkaPublisher wraps a kafka.Producer. The producer's underlying broker
+// is also used for Subscribe, so a single Publisher serves both publishing
+// and subscribing.
+func NewKafkaPublisher(producer *kafka.Producer, logger *zap.Logger) *KafkaPublisher {
+	return &KafkaPublisher{
+		producer: producer,
+		logger:   logger.Named("events.kafka"),
+	}
+}
+
+// Publish serializes status to JSON and sends it on TopicStatusUpdate. Errors
+// are returned to the caller; the call site decides whether to log-and-continue
+// (the default for status mutations) or propagate.
+//
+// The kafka.Producer.Send signature does not take a context — the underlying
+// broker uses an internal background context for at-most-once produce; we
+// honor cancellation by short-circuiting before the call.
+func (p *KafkaPublisher) Publish(ctx context.Context, status *models.TransactionStatus) error {
+	if status == nil {
+		return fmt.Errorf("nil status")
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	return p.producer.Send(kafka.TopicStatusUpdate, status.TxID, status) //nolint:contextcheck // kafka.Producer.Send doesn't take a context; ctx already checked above
+}
+
+// Subscribe joins a fresh consumer group on TopicStatusUpdate and returns a
+// channel that yields decoded TransactionStatus values until ctx is canceled.
+// The unique groupID guarantees this subscriber sees every message — useful
+// when multiple subscribers (SSE manager + webhook service) coexist in the
+// same process or across pods.
+func (p *KafkaPublisher) Subscribe(ctx context.Context) (<-chan *models.TransactionStatus, error) {
+	p.mu.Lock()
+	if p.closed {
+		p.mu.Unlock()
+		return nil, fmt.Errorf("publisher closed")
+	}
+	p.mu.Unlock()
+
+	groupID, err := uniqueGroupID()
+	if err != nil {
+		return nil, fmt.Errorf("generating group id: %w", err)
+	}
+
+	out := make(chan *models.TransactionStatus, 256)
+
+	cg, err := kafka.NewConsumerGroup(kafka.ConsumerConfig{
+		Broker:  p.producer.Broker(),
+		GroupID: groupID,
+		Topics:  []string{kafka.TopicStatusUpdate},
+		Handler: func(ctx context.Context, msg *kafka.Message) error {
+			var status models.TransactionStatus
+			if jsonErr := json.Unmarshal(msg.Value, &status); jsonErr != nil {
+				p.logger.Warn("dropping malformed status update", zap.Error(jsonErr))
+				return nil
+			}
+			select {
+			case out <- &status:
+			case <-ctx.Done():
+				return nil
+			default:
+				// Slow consumer — drop rather than block the broker. Matches the
+				// old arcade's non-blocking fan-out semantics; SSE clients
+				// recover via Last-Event-ID catchup on reconnect.
+				p.logger.Warn("subscriber channel full, dropping update",
+					zap.String("txid", status.TxID),
+					zap.String("status", string(status.Status)),
+				)
+			}
+			return nil
+		},
+		Producer: p.producer,
+		Logger:   p.logger,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("creating consumer group: %w", err)
+	}
+
+	sub := &kafkaSubscription{cg: cg, out: out}
+
+	p.mu.Lock()
+	p.subs = append(p.subs, sub)
+	p.mu.Unlock()
+
+	go func() {
+		// Run blocks until ctx is canceled or the broker closes.
+		if err := cg.Run(ctx); err != nil {
+			p.logger.Warn("consumer group exited with error", zap.Error(err))
+		}
+		_ = cg.Close()
+		close(out)
+	}()
+
+	return out, nil
+}
+
+// Close stops the publisher. Existing subscriptions are released through
+// their context cancellation; Close does not close subscriber channels
+// directly because the consumer goroutine owns the close.
+func (p *KafkaPublisher) Close() error {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.closed {
+		return nil
+	}
+	p.closed = true
+	for _, s := range p.subs {
+		_ = s.cg.Close()
+	}
+	p.subs = nil
+	return nil
+}
+
+type kafkaSubscription struct {
+	cg  *kafka.ConsumerGroup
+	out chan *models.TransactionStatus
+}
+
+// uniqueGroupID returns a per-call group identifier. Used so each Subscribe
+// gets its own consumer group, which in turn guarantees every subscriber
+// sees every message (the broker fans out across distinct groups).
+func uniqueGroupID() (string, error) {
+	var b [12]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "", err
+	}
+	return "arcade-events-" + hex.EncodeToString(b[:]), nil
+}