cleanup, works

cmd/service/bidcollect.go

@@ -23,6 +23,8 @@ var (
 	outputTSV bool   // by default: CSV, but can be changed to TSV with this setting
 	uid       string // used in output filenames, to avoid collissions between multiple collector instances
 
+	webserverListenAddr string
+
 	runDevServerOnly    bool // used to play with file listing website
 	devServerListenAddr string
 
@@ -47,7 +49,10 @@ func init() {
 	// utils
 	bidCollectCmd.Flags().StringVar(&uid, "uid", "", "unique identifier for output files (to avoid collisions)")
 
-	// for dev purposes
+	// webserver for SSE
+	bidCollectCmd.Flags().StringVar(&webserverListenAddr, "webserver-addr", "", "listen address for SSE subscription webserver")
+
+	// devserver provides the file listing for playing with file HTML
 	bidCollectCmd.Flags().BoolVar(&runDevServerOnly, "devserver", false, "only run devserver to play with file listing website")
 	bidCollectCmd.Flags().StringVar(&devServerListenAddr, "devserver-addr", "localhost:8095", "listen address for devserver")
 
@@ -103,6 +108,7 @@ var bidCollectCmd = &cobra.Command{
 			BeaconNodeURI:           beaconNodeURI,
 			OutDir:                  outDir,
 			OutputTSV:               outputTSV,
+			WebserverAddr:           webserverListenAddr,
 		}
 
 		bidCollector := bidcollect.NewBidCollector(&opts)

go.mod

@@ -53,6 +53,7 @@ require (
 	github.com/fatih/color v1.15.0 // indirect
 	github.com/gballet/go-verkle v0.1.1-0.20231031103413-a67434b50f46 // indirect
 	github.com/getsentry/sentry-go v0.21.0 // indirect
+	github.com/go-chi/chi/v5 v5.1.0 // indirect
 	github.com/go-gorp/gorp/v3 v3.1.0 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/goccy/go-yaml v1.11.0 // indirect

go.sum

@@ -121,6 +121,8 @@ github.com/getsentry/sentry-go v0.21.0/go.mod h1:lc76E2QywIyW8WuBnwl8Lc4bkmQH4+w
 github.com/gin-contrib/sse v0.0.0-20190301062529-5545eab6dad3/go.mod h1:VJ0WA2NBN22VlZ2dKZQPAPnyWw5XTlK1KymzLKsr59s=
 github.com/gin-gonic/gin v1.4.0/go.mod h1:OW2EZn3DO8Ln9oIKOvM++LBO+5UPHJJDH72/q/3rZdM=
 github.com/go-check/check v0.0.0-20180628173108-788fd7840127/go.mod h1:9ES+weclKsC9YodN5RgxqK/VD9HM9JsCSh7rNhMZE98=
+github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw=
+github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
 github.com/go-errors/errors v1.0.1/go.mod h1:f4zRHt4oKfwPJE5k8C9vpYG+aDHdBFUsgrm6/TyX73Q=
 github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA=
 github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og=

services/bidcollect/bid-processor.go

@@ -118,6 +118,10 @@ func (c *BidProcessor) processBids(bids []*types.CommonBid) {
 		}
 
 		// Send to subscribers
+		if c.webserver != nil {
+			c.webserver.SendBid(bid)
+		}
+
 		// Write to CSV
 		c.writeBidToFile(bid, isNewBid, isTopBid)
 	}

services/bidcollect/bidcollector.go

@@ -20,6 +20,8 @@ type BidCollectorOpts struct {
 
 	OutDir    string
 	OutputTSV bool
+
+	WebserverAddr string
 }
 
 type BidCollector struct {
@@ -50,10 +52,11 @@ func NewBidCollector(opts *BidCollectorOpts) *BidCollector {
 
 	// output
 	c.processor = NewBidProcessor(&BidProcessorOpts{
-		Log:       opts.Log,
-		UID:       opts.UID,
-		OutDir:    opts.OutDir,
-		OutputTSV: opts.OutputTSV,
+		Log:           opts.Log,
+		UID:           opts.UID,
+		OutDir:        opts.OutDir,
+		OutputTSV:     opts.OutputTSV,
+		WebserverAddr: opts.WebserverAddr,
 	})
 	return c
 }

services/bidcollect/webserver/webserver.go

@@ -4,14 +4,12 @@ package webserver
 import (
 	"context"
 	"errors"
-	"fmt"
 	"net/http"
 	"sync"
 	"time"
 
-	"github.com/flashbots/go-utils/httplogger"
 	"github.com/flashbots/relayscan/services/bidcollect/types"
-	"github.com/gorilla/mux"
+	"github.com/go-chi/chi/v5"
 	"github.com/sirupsen/logrus"
 	"go.uber.org/atomic"
 )
@@ -43,23 +41,19 @@ func New(cfg *HTTPServerConfig) (srv *Server) {
 	}
 	srv.isReady.Swap(true)
 
+	router := chi.NewRouter()
+	router.Get("/v1/sse/bids", srv.handleSSESubscription)
+
 	srv.srv = &http.Server{
 		Addr:         cfg.ListenAddr,
-		Handler:      srv.getRouter(),
+		Handler:      router,
 		ReadTimeout:  cfg.ReadTimeout,
 		WriteTimeout: cfg.WriteTimeout,
 	}
 
 	return srv
 }
 
-func (srv *Server) getRouter() http.Handler {
-	r := mux.NewRouter()
-	r.HandleFunc("/", srv.handleRoot).Methods(http.MethodGet)
-	r.HandleFunc("/v1/sse/bids", srv.handleSSESubscription).Methods(http.MethodGet)
-	return httplogger.LoggingMiddlewareLogrus(srv.log, r)
-}
-
 func (srv *Server) RunInBackground() {
 	go func() {
 		srv.log.WithField("listenAddress", srv.cfg.ListenAddr).Info("Starting HTTP server")
@@ -91,16 +85,11 @@ func (srv *Server) removeSubscriber(sub *SSESubscription) {
 	srv.log.WithField("subscribers", len(srv.sseConnectionMap)).Info("removed subscriber")
 }
 
-func (srv *Server) handleRoot(w http.ResponseWriter, req *http.Request) {
-	// write hello world
-	fmt.Fprintf(w, "Hello, world!")
-}
-
-func (srv *Server) SendBid(ctx context.Context, bid *types.CommonBid) error {
+func (srv *Server) SendBid(bid *types.CommonBid) {
 	srv.sseConnectionLock.RLock()
 	defer srv.sseConnectionLock.RUnlock()
 	if len(srv.sseConnectionMap) == 0 {
-		return nil
+		return
 	}
 
 	msg := bid.ToCSVLine("\t")
@@ -112,6 +101,4 @@ func (srv *Server) SendBid(ctx context.Context, bid *types.CommonBid) error {
 		default:
 		}
 	}
-
-	return nil
 }