live

Documentation

Overview

Package live provides a real-time HTTP dashboard for workflow execution.

It serves an interactive HTML dashboard that updates live via Server-Sent Events (SSE) as the workflow executes. Steps light up as they start, change color as they succeed or fail, and the full DAG structure snaps into place when the workflow completes.

Quick Start

server, auditor, err := live.New(auditlog.Config{
	WorkflowID: "my-pipeline",
}, live.Config{
	Addr: ":8080",
})
if err != nil {
	log.Fatal(err)
}

auditor.Attach(workflow)
go server.ListenAndServe()

fmt.Println("Live dashboard: http://localhost:8080")

workflow.Do(ctx)
auditor.Snapshot(workflow)
server.SignalComplete()

Architecture

The live server uses SSE (Server-Sent Events) for real-time communication:

• GET / - Interactive dashboard HTML (static, cached)
• GET /api/report - Current report as JSON (point-in-time snapshot)
• GET /api/events - SSE stream (snapshot + live events + completion)
• GET /api/health - Health check

SSE was chosen because the data flow is one-way (server to browser), SSE has native browser support via EventSource, auto-reconnects on disconnect, supports reconnection replay via Last-Event-ID, and requires no framing protocol.

Protocol

The SSE stream sends three named event types:

• snapshot: Initial state on connect (report + events + metadata + DAG)
• event: Individual events as they fire during execution
• complete: Final report with full DAG structure after Snapshot

Late clients receive the full state via the snapshot event, including all events captured so far. After completion, new clients get the final report.

Index

• Constants
• Variables
• type BroadcastEvent
• type CompleteProvider
• type Config
• type HTMLWriter
• type HealthInfo
• type HealthProvider
• type Hub
  • func NewHub() *Hub
  • func NewHubWithReplay(replayBufferSize int) *Hub
  • func (h *Hub) BufferedEventCount() int
  • func (h *Hub) ClientCount() int
  • func (h *Hub) Drain(ctx context.Context) error
  • func (h *Hub) EventStore() sse.EventStore
  • func (h *Hub) IsComplete() bool
  • func (h *Hub) IsDraining() bool
  • func (h *Hub) OnEvent(evt auditlog.Event)
  • func (h *Hub) SignalComplete()
  • func (h *Hub) Subscribe() *Subscriber
  • func (h *Hub) Unsubscribe(subscriberID uint64)
• type NDJSONWriter
• type ReportProvider
• type Server
  • func New(auditCfg auditlog.Config, serverCfg Config) (*Server, *auditlog.Auditor, error)
  • func NewServer(hub *Hub, auditor *auditlog.Auditor, cfg Config) *Server
  • func (srv *Server) Addr() string
  • func (srv *Server) ClientCount() int
  • func (srv *Server) ListenAndServe() error
  • func (srv *Server) OnEvent(evt auditlog.Event)
  • func (srv *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (srv *Server) Shutdown(ctx context.Context) error
  • func (srv *Server) SignalComplete()
• type SnapshotProvider
• type Subscriber
  • func (s *Subscriber) Done() <-chan struct{}
  • func (s *Subscriber) Events() <-chan BroadcastEvent
  • func (s *Subscriber) ID() uint64

Constants

const SchemaVersion = "0.1.0"

SchemaVersion mirrors the auditlog report version.

Variables

var ErrServerAlreadyRunning = errors.New("live server is already running")

ErrServerAlreadyRunning is returned when ListenAndServe is called on a server that is already serving.

Types

type BroadcastEvent

type BroadcastEvent struct {
	ID   sse.EventID
	Data jsontext.Value
}

BroadcastEvent carries a marshaled event payload alongside its SSE event ID. SSE clients use the ID for Last-Event-ID reconnection replay; WebSocket clients ignore it.

type CompleteProvider

type CompleteProvider func() (jsontext.Value, error)

CompleteProvider returns the final SSE complete payload as raw JSON.

type Config

type Config struct {
	// Addr is the TCP address to listen on. Default ":0" (random port).
	Addr string
	// Prefix is the URL path prefix for all dashboard routes.
	// Default "/" (root). Routes: {prefix}/, {prefix}/api/report,
	// {prefix}/api/events, {prefix}/api/health,
	// {prefix}/api/export/ndjson, {prefix}/api/export/html.
	// Set to "/workflow" to mount at /workflow/. Trailing slash is stripped.
	Prefix string
	// ReadHeaderTimeout is the maximum duration for reading the request
	// headers. Default 5 seconds. Set to 0 to disable.
	ReadHeaderTimeout time.Duration
	// HeartbeatInterval is how often to send SSE keepalive comments.
	// Default 15 seconds. Set to 0 to disable heartbeats.
	HeartbeatInterval time.Duration
	// ReplayBufferSize is the maximum number of events retained for
	// SSE reconnection replay. When a client reconnects with a Last-Event-ID
	// header, missed events are replayed from this buffer. Default 1000.
	// Set to 0 for the default.
	ReplayBufferSize int
	// CORSAllowedOrigins controls the Access-Control-Allow-Origin
	// header on API endpoints. Empty (default) disables CORS
	// (secure by default). Set to "*" to allow all origins, or a
	// specific origin like "https://dashboard.example.com".
	CORSAllowedOrigins string
}

Config controls the live dashboard server behaviour.

type HTMLWriter

type HTMLWriter func(w io.Writer) error

HTMLWriter writes the self-contained HTML report to the given writer.

type HealthInfo

type HealthInfo struct {
	Events  int   `json:"events"`
	Dropped int64 `json:"dropped"`
}

HealthInfo provides dynamic health check data beyond the built-in uptime, client count, and completion status.

type HealthProvider

type HealthProvider func() HealthInfo

HealthProvider returns additional health check information.

type Hub

type Hub struct {
	// contains filtered or unexported fields
}

Hub fans out workflow events to all connected SSE clients.

The hub is safe for concurrent use. OnEvent is called from recorder goroutines, and Subscribe/Unsubscribe are called from HTTP handler goroutines.

func NewHub

func NewHub() *Hub

NewHub creates a Hub ready for use with the default replay buffer size.

func NewHubWithReplay

func NewHubWithReplay(replayBufferSize int) *Hub

NewHubWithReplay creates a Hub with a replay ring buffer of the given capacity. Non-positive capacity uses the default (1000).

func (*Hub) BufferedEventCount

func (h *Hub) BufferedEventCount() int

BufferedEventCount returns the number of events currently stored in the replay ring buffer.

func (*Hub) ClientCount

func (h *Hub) ClientCount() int

ClientCount returns the number of currently connected subscribers.

func (*Hub) Drain

func (h *Hub) Drain(ctx context.Context) error

Drain gracefully waits for all subscriber channel buffers to empty (consumers catch up) or the context to timeout. After drain begins, the hub is marked draining. Returns nil on a clean drain, or ctx.Err() if the deadline fires before buffers empty.

func (*Hub) EventStore

func (h *Hub) EventStore() sse.EventStore

EventStore returns the replay ring buffer as an sse.EventStore for SSE reconnection replay via sse.Replay.

func (*Hub) IsComplete

func (h *Hub) IsComplete() bool

IsComplete returns whether the lifecycle has been marked as complete.

func (*Hub) IsDraining

func (h *Hub) IsDraining() bool

IsDraining returns whether the hub is currently in a drain state (Server.Shutdown is waiting for subscriber buffers to empty).

func (*Hub) OnEvent

func (h *Hub) OnEvent(evt auditlog.Event)

OnEvent marshals a workflow Event to JSON, assigns it a sequential SSE event ID, stores it in the replay ring buffer, and broadcasts it to all connected clients.

func (*Hub) SignalComplete

func (h *Hub) SignalComplete()

SignalComplete marks the lifecycle as finished. All subscribers receive a done signal so the SSE handler can send the final report.

func (*Hub) Subscribe

func (h *Hub) Subscribe() *Subscriber

Subscribe registers a new SSE client and returns a subscriber.

func (*Hub) Unsubscribe

func (h *Hub) Unsubscribe(subscriberID uint64)

Unsubscribe removes a subscriber by ID and signals its done channel.

type NDJSONWriter

type NDJSONWriter func(w io.Writer) error

NDJSONWriter writes the full NDJSON event stream to the given writer.

type ReportProvider

type ReportProvider func() ([]byte, error)

ReportProvider returns the current report as JSON bytes.

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server serves the real-time workflow dashboard over HTTP.

func New

func New(auditCfg auditlog.Config, serverCfg Config) (*Server, *auditlog.Auditor, error)

New is the convenience constructor. It creates a Hub, wires it as the auditlog OnEvent callback, creates the Auditor, and returns a ready-to-use Server.

func NewServer

func NewServer(hub *Hub, auditor *auditlog.Auditor, cfg Config) *Server

NewServer creates a Server from an existing Hub and Auditor.

func (*Server) Addr

func (srv *Server) Addr() string

Addr returns the server's listen address. After ListenAndServe succeeds, this reflects the actual address (including the OS-assigned port when ":0" was requested).

func (*Server) ClientCount

func (srv *Server) ClientCount() int

ClientCount returns the number of currently connected SSE clients.

func (*Server) ListenAndServe

func (srv *Server) ListenAndServe() error

ListenAndServe starts the HTTP server.

func (*Server) OnEvent

func (srv *Server) OnEvent(evt auditlog.Event)

OnEvent broadcasts an event to all connected SSE clients.

func (*Server) ServeHTTP

func (srv *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements http.Handler.

func (*Server) Shutdown

func (srv *Server) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the server. It first drains subscriber buffers (allowing SSE clients to consume buffered events), then shuts down the HTTP server.

func (*Server) SignalComplete

func (srv *Server) SignalComplete()

SignalComplete marks the workflow as finished.

type SnapshotProvider

type SnapshotProvider func(isComplete bool) (jsontext.Value, error)

SnapshotProvider returns the initial SSE snapshot payload as raw JSON.

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber represents a single SSE client connection.

func (*Subscriber) Done

func (s *Subscriber) Done() <-chan struct{}

Done returns a channel that is closed when the lifecycle completes or the subscriber is removed.

func (*Subscriber) Events

func (s *Subscriber) Events() <-chan BroadcastEvent

Events returns the channel that receives broadcast events.

func (*Subscriber) ID

func (s *Subscriber) ID() uint64

ID returns the subscriber's unique identifier.