Feature: Web UI

.gitignore

@@ -12,3 +12,4 @@ ZENO-*
 *.log
 .old/
 *.warc.*
+static/

cmd/get.go

@@ -52,6 +52,8 @@ func addBasicFlags(getCmd *cobra.Command) {
 	getCmd.PersistentFlags().Bool("disable-seencheck", false, "Disable the (remote or local) seencheck that avoid re-crawling of URIs.")
 	getCmd.PersistentFlags().Bool("api", false, "Enable API")
 	getCmd.PersistentFlags().Int("api-port", 9090, "Port to listen on for the API.")
+	getCmd.PersistentFlags().String("api-static-dir", "static", "Directory which shall be served as static files at /")
+	getCmd.PersistentFlags().Duration("api-frontier-poll-interval", time.Second, "Interval at which the frontier WebSocket polls the reactor for item changes. (Default: 1s)")
 	getCmd.PersistentFlags().Int("max-redirect", 20, "Specifies the maximum number of redirections to follow for a resource.")
 	getCmd.PersistentFlags().Int("max-css-jump", 10, "Specifies the maximum number of CSS @import jumps to follow for a resource.")
 	getCmd.PersistentFlags().Int("max-retry", 5, "Number of retry if error happen when executing HTTP request.")

internal/pkg/api/api.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/internetarchive/Zeno/internal/pkg/config"
-	"github.com/internetarchive/Zeno/internal/pkg/stats"
 )
 
 var (
@@ -25,10 +24,8 @@ var (
 func Start() error {
 	once.Do(func() {
 		mux := http.NewServeMux()
-
-		if config.Get().Prometheus {
-			mux.Handle("/metrics", stats.PrometheusHandler())
-		}
+		// Register routes
+		registerRoutes(mux)
 
 		server = &http.Server{
 			Addr:    ":" + strconv.Itoa(config.Get().APIPort),

internal/pkg/api/handlers/frontier/handler.go

@@ -0,0 +1,59 @@
+package frontier
+
+import (
+	"net/http"
+
+	"github.com/gobwas/ws"
+	"github.com/gobwas/ws/wsutil"
+)
+
+// Handles WebSocket connections for streaming the frontier
+func Handler(w http.ResponseWriter, r *http.Request) {
+	conn, _, _, err := ws.UpgradeHTTP(r, w)
+	if err != nil {
+		http.Error(w, "WebSocket upgrade failed", http.StatusBadRequest)
+		return
+	}
+	defer conn.Close()
+
+	hub := GetHub()
+	poller := GetPoller()
+
+	// Register client and ensure poller is running.
+	ch := hub.Register()
+	if hub.ClientCount() == 1 {
+		poller.Start()
+	}
+
+	defer func() {
+		hub.Unregister(ch)
+		if hub.ClientCount() == 0 {
+			poller.Stop()
+		}
+	}()
+
+	// Writer goroutine: reads from client channel and writes to WebSocket.
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for data := range ch {
+			err := wsutil.WriteServerMessage(conn, ws.OpText, data)
+			if err != nil {
+				return
+			}
+		}
+	}()
+
+	// Reader goroutine: detect client close.
+	go func() {
+		for {
+			_, _, err := wsutil.ReadClientData(conn)
+			if err != nil {
+				conn.Close()
+				return
+			}
+		}
+	}()
+
+	<-done
+}

internal/pkg/api/handlers/frontier/hub.go

@@ -0,0 +1,112 @@
+// Package frontier provides WebSocket streaming of reactor item deltas.
+package frontier
+
+import (
+	"context"
+	"sync"
+)
+
+// Hub manages WebSocket client channels for broadcasting.
+type Hub struct {
+	mu      sync.RWMutex
+	clients map[chan []byte]struct{}
+
+	// Broadcaster channel
+	broadcastCh chan []byte
+
+	// Lifecycle management
+	ctx    context.Context
+	cancel context.CancelFunc
+	wg     sync.WaitGroup
+}
+
+// Creates a new Hub instance and starts the broadcaster goroutine.
+func NewHub() *Hub {
+	ctx, cancel := context.WithCancel(context.Background())
+	h := &Hub{
+		clients:     make(map[chan []byte]struct{}),
+		broadcastCh: make(chan []byte, 64),
+		ctx:         ctx,
+		cancel:      cancel,
+	}
+	h.wg.Add(1)
+	go h.runBroadcaster()
+	return h
+}
+
+// Adds a new client channel to the hub
+func (h *Hub) Register() chan []byte {
+	ch := make(chan []byte, 64)
+	h.mu.Lock()
+	h.clients[ch] = struct{}{}
+	h.mu.Unlock()
+	return ch
+}
+
+// Removes a client channel from the hub and closes it.
+func (h *Hub) Unregister(ch chan []byte) {
+	h.mu.Lock()
+	delete(h.clients, ch)
+	h.mu.Unlock()
+	close(ch)
+}
+
+// Returns the number of connected clients.
+func (h *Hub) ClientCount() int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return len(h.clients)
+}
+
+// Send queues data to be broadcast to all clients (non-blocking).
+func (h *Hub) Send(data []byte) {
+	select {
+	case h.broadcastCh <- data:
+	default:
+		// Channel full; drop message to avoid blocking.
+	}
+}
+
+// Reads from broadcastCh and fans out to all client channels
+func (h *Hub) runBroadcaster() {
+	defer h.wg.Done()
+	for {
+		select {
+		case <-h.ctx.Done():
+			return
+		case data, ok := <-h.broadcastCh:
+			if !ok {
+				return
+			}
+			h.broadcast(data)
+		}
+	}
+}
+
+// Sends data to all connected clients (non-blocking).
+func (h *Hub) broadcast(data []byte) {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for ch := range h.clients {
+		select {
+		case ch <- data:
+		default:
+			// Client channel full; skip to avoid blocking.
+		}
+	}
+}
+
+// Shuts down the hub and its broadcaster(s)
+func (h *Hub) Close() {
+	h.cancel()
+	close(h.broadcastCh)
+	h.wg.Wait()
+}
+
+// Singleton instance of the hub
+var globalHub = NewHub()
+
+// Returns the global hub instance
+func GetHub() *Hub {
+	return globalHub
+}

internal/pkg/api/handlers/frontier/poller.go

@@ -0,0 +1,183 @@
+package frontier
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/internetarchive/Zeno/internal/pkg/config"
+	"github.com/internetarchive/Zeno/internal/pkg/reactor"
+	"github.com/internetarchive/Zeno/pkg/models"
+)
+
+// Representation of models.Item sent over WebSocket.
+type SerializedItem struct {
+	ID      string `json:"id"`
+	URL     string `json:"url"`
+	SeedVia string `json:"seedVia"`
+	Status  string `json:"status"`
+	Source  string `json:"source"`
+	Parent  string `json:"parent"` // children are ommited, because they would trigger lots of updates, parent is enough to track the tree
+	Err     string `json:"err,omitempty"`
+}
+
+// Converts a models.Item to a SerializedItem.
+func serializedFromItem(item *models.Item) SerializedItem {
+	s := SerializedItem{
+		ID:      item.GetID(),
+		SeedVia: item.GetSeedVia(),
+		Status:  item.GetStatus().String(),
+		Source:  item.GetSource().String(),
+	}
+	if item.GetURL() != nil {
+		s.URL = item.GetURL().Raw
+	}
+	if item.GetParent() != nil {
+		s.Parent = item.GetParent().GetID()
+	}
+	if item.GetError() != nil {
+		s.Err = item.GetError().Error()
+	}
+	return s
+}
+
+// Computes a SHA-256 hash of a SerializedItem
+func hashItem(s SerializedItem) string {
+	data, _ := json.Marshal(s)
+	sum := sha256.Sum256(data)
+	return hex.EncodeToString(sum[:])
+}
+
+// Recursively collect all items from a seed's tree
+// O(n) but required.
+func flattenItem(root *models.Item) []*models.Item {
+	var items []*models.Item
+	var traverse func(node *models.Item)
+	traverse = func(node *models.Item) {
+		if node == nil {
+			return
+		}
+		items = append(items, node)
+		for _, child := range node.GetChildren() {
+			traverse(child)
+		}
+	}
+	traverse(root)
+	return items
+}
+
+// Poller struct
+type Poller struct {
+	hub        *Hub
+	interval   time.Duration
+	prevHashes map[string]string
+
+	mu     sync.Mutex
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+// Creates a new Poller instance.
+func NewPoller(hub *Hub, interval time.Duration) *Poller {
+	return &Poller{
+		hub:        hub,
+		interval:   interval,
+		prevHashes: make(map[string]string),
+	}
+}
+
+// Begins the polling loop.
+func (p *Poller) Start() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.cancel != nil {
+		return // Already running
+	}
+
+	p.ctx, p.cancel = context.WithCancel(context.Background())
+	go p.run()
+}
+
+// Stops the polling loop.
+func (p *Poller) Stop() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.cancel != nil {
+		p.cancel()
+		p.cancel = nil
+	}
+}
+
+// main polling loop.
+func (p *Poller) run() {
+	ticker := time.NewTicker(p.interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-p.ctx.Done():
+			return
+		case <-ticker.C:
+			if p.hub.ClientCount() == 0 {
+				continue
+			}
+			p.poll()
+		}
+	}
+}
+
+// fetches reactor state, computes deltas, and sends changed items to the hub for WebSocket egress
+func (p *Poller) poll() {
+	// Flatten all seeds and their children.
+	seeds := reactor.GetStateTableItems()
+	currentItems := make(map[string]SerializedItem)
+	for _, seed := range seeds {
+		for _, item := range flattenItem(seed) {
+			s := serializedFromItem(item)
+			currentItems[s.ID] = s
+		}
+	}
+
+	// Compute hashes and find deltas.
+	var changed []SerializedItem
+	currentHashes := make(map[string]string)
+	for id, s := range currentItems {
+		hash := hashItem(s)
+		currentHashes[id] = hash
+		if p.prevHashes[id] != hash {
+			changed = append(changed, s)
+		}
+	}
+
+	p.prevHashes = currentHashes
+
+	if len(changed) > 0 {
+		data, err := json.Marshal(changed)
+		if err == nil {
+			p.hub.Send(data)
+		}
+	}
+}
+
+// Singleton poller instance (lazy-initialized).
+var (
+	globalPoller     *Poller
+	globalPollerOnce sync.Once
+)
+
+// Returns the singleton poller instance.
+func GetPoller() *Poller {
+	globalPollerOnce.Do(func() {
+		interval := config.Get().APIFrontierPollInterval
+		if interval <= 0 {
+			interval = time.Second // fallback default
+		}
+		globalPoller = NewPoller(globalHub, interval)
+	})
+	return globalPoller
+}