Documentation
¶
Overview ¶
Package viewer holds the magus.viewer.v1alpha1 wire contract: the code that maps captured DOMAIN events onto the versioned protobuf tool-page contract and encodes them for a browser (a URL-fragment blob for a finished run, or a live SSE stream), plus the viewer's filter DSL. It consumes domain types straight from the repositories (e.g. the cache output store's []journal.Event) and maps them explicitly to the wire proto - no intermediate DTOs, no single-use converters. This file is the viewer encoders; query.go is the filter grammar.
Index ¶
- func ApplyEventQuery(events []journal.Event, q *viewerv1.EventQuery) []journal.Event
- func EncodeEvent(e journal.Event) (string, error)
- func EncodeJournalFragment(inv journal.Invocation, events []journal.Event) (string, error)
- func ParseEventQuery(s string) *viewerv1.EventQuery
- type LiveServer
- type Service
- func (s *Service) GetInvocation(_ context.Context, req *connect.Request[viewerv1.GetInvocationRequest]) (*connect.Response[viewerv1.Invocation], error)
- func (s *Service) GetJournal(_ context.Context, req *connect.Request[viewerv1.GetJournalRequest]) (*connect.Response[viewerv1.Journal], error)
- func (s *Service) GetOutput(_ context.Context, req *connect.Request[viewerv1.GetOutputRequest]) (*connect.Response[viewerv1.GetOutputResponse], error)
- func (s *Service) ListEvents(_ context.Context, req *connect.Request[viewerv1.ListEventsRequest]) (*connect.Response[viewerv1.ListEventsResponse], error)
- func (s *Service) ListInvocations(_ context.Context, req *connect.Request[viewerv1.ListInvocationsRequest]) (*connect.Response[viewerv1.ListInvocationsResponse], error)
- func (s *Service) ListOutputs(_ context.Context, req *connect.Request[viewerv1.ListOutputsRequest]) (*connect.Response[viewerv1.ListOutputsResponse], error)
- func (s *Service) StreamEvents(ctx context.Context, req *connect.Request[viewerv1.StreamEventsRequest], ...) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ApplyEventQuery ¶ added in v0.4.0
ApplyEventQuery returns the subset of events matching q, preserving order. A nil or empty q returns events unchanged. Set filters AND together; repeated values within a field OR; string comparison is case-insensitive; text matches are substring (a negated text match excludes events that contain it). The input slice is not mutated.
func EncodeEvent ¶
EncodeEvent marshals one event to base64(protobuf) for an SSE `data:` line (SSE payloads must be UTF-8 text, so the binary message is base64-wrapped). The JS client base64- decodes then Event.fromBinary.
func EncodeJournalFragment ¶
EncodeJournalFragment marshals a journal to protobuf, then gzip+base64url encodes it for a `#data=` URL fragment - the static delivery path. The generated JS client reverses it (base64url -> gunzip -> Journal.fromBinary). Reuses the same fragment encoder graph open uses, so the tool pages share one wire envelope.
func ParseEventQuery ¶ added in v0.4.0
func ParseEventQuery(s string) *viewerv1.EventQuery
ParseEventQuery parses a viewer filter DSL string into a typed EventQuery: whitespace separated "field:value" clauses plus free text, e.g. `project:web target:build kind:output -"cache miss"`. Repeated values on one REPEATED field OR; different fields AND; matching is case-insensitive. Field filters are include-only (they mirror the filter-menu checkboxes); negation ("-word") is supported only on free text. `status` is single-valued (a result has one status), so a second status: clause replaces the first. The time window is NOT parsed from the DSL - the filter menu sets EventQuery.Time programmatically (date pickers), and ApplyEventQuery honors it. This is the viewer's OWN grammar over the log's own fields; it deliberately does not reuse the knowledge-graph query parser (whose fields differ).
Types ¶
type LiveServer ¶
type LiveServer struct {
// contains filtered or unexported fields
}
LiveServer streams one invocation's events to a browser over SSE, on a loopback httpx.Server. Start it with StartLive before the run, hand the browser LiveServer.ViewerURL, let the run emit into the broadcaster, then LiveServer.Stop when done.
func StartLive ¶
func StartLive(origin string, bc *journal.Broadcaster) (*LiveServer, error)
StartLive starts a loopback SSE server on an ephemeral 127.0.0.1 port for bc in the background and mints a random bearer token. origin is the page allowed to read the stream. The caller owns bc and must add it to the invocation's capture logger so events flow, and call LiveServer.Stop when the run is finished.
func (*LiveServer) Addr ¶
func (ls *LiveServer) Addr() string
Addr is the loopback "127.0.0.1:PORT" the server bound - the value the viewer connects its EventSource to.
func (*LiveServer) Stop ¶
func (ls *LiveServer) Stop(ctx context.Context)
Stop shuts the server down, allowing a brief grace window first so a late or reloading browser still receives the full log. Call it once the run has finished and the broadcaster is closed.
func (*LiveServer) Token ¶
func (ls *LiveServer) Token() string
Token is the per-run bearer token the viewer must present.
func (*LiveServer) ViewerURL ¶
func (ls *LiveServer) ViewerURL(logsBase string) string
ViewerURL builds the viewer link for this live run: <logsBase>/#live=<addr>&token=<token>, where logsBase is the log viewer page URL (e.g. https://.../magus/logs/). BOTH the loopback host and the bearer token ride the URL fragment, which the browser never transmits to a server - so the connection details are handed to the page locally and nothing leaves the machine; the page reads the token and strips it from the URL.
type Service ¶ added in v0.4.0
type Service struct {
// contains filtered or unexported fields
}
Service implements viewerv1alpha1connect.ViewerServiceHandler over the same two stores the plain-JSON run-browser routes read.
func NewService ¶ added in v0.4.0
func NewService(outputs outputSource, runs runSource) *Service
NewService builds the ViewerService Connect handler reading from the run and output stores.
func (*Service) GetInvocation ¶ added in v0.4.0
func (s *Service) GetInvocation(_ context.Context, req *connect.Request[viewerv1.GetInvocationRequest]) (*connect.Response[viewerv1.Invocation], error)
GetInvocation returns one run's header.
func (*Service) GetJournal ¶ added in v0.4.0
func (s *Service) GetJournal(_ context.Context, req *connect.Request[viewerv1.GetJournalRequest]) (*connect.Response[viewerv1.Journal], error)
GetJournal returns one run whole - header plus every event.
func (*Service) GetOutput ¶ added in v0.4.0
func (s *Service) GetOutput(_ context.Context, req *connect.Request[viewerv1.GetOutputRequest]) (*connect.Response[viewerv1.GetOutputResponse], error)
GetOutput returns one stored run's captured bytes verbatim.
func (*Service) ListEvents ¶ added in v0.4.0
func (s *Service) ListEvents(_ context.Context, req *connect.Request[viewerv1.ListEventsRequest]) (*connect.Response[viewerv1.ListEventsResponse], error)
ListEvents returns a page of one run's events, narrowed by the request filter. The store returns the journal whole, so the page token is an offset into it rather than a cursor it could resume from.
The filter runs BEFORE the page is cut, so page_size counts MATCHING events. Cutting first would make a narrow filter over a long journal return an empty page while matches sat just past the boundary, and a caller cannot tell that from "no such events".
func (*Service) ListInvocations ¶ added in v0.4.0
func (s *Service) ListInvocations(_ context.Context, req *connect.Request[viewerv1.ListInvocationsRequest]) (*connect.Response[viewerv1.ListInvocationsResponse], error)
ListInvocations returns the retained run journals, newest first.
func (*Service) ListOutputs ¶ added in v0.4.0
func (s *Service) ListOutputs(_ context.Context, req *connect.Request[viewerv1.ListOutputsRequest]) (*connect.Response[viewerv1.ListOutputsResponse], error)
ListOutputs returns the stored runs' descriptors, newest first.
func (*Service) StreamEvents ¶ added in v0.4.0
func (s *Service) StreamEvents(ctx context.Context, req *connect.Request[viewerv1.StreamEventsRequest], stream *connect.ServerStream[viewerv1.StreamEventsResponse]) error
StreamEvents replays one run's stored events, then tails the journal for what the run appends next, until the run finishes or the caller goes away.
It reads the journal rather than subscribing to a broadcaster because the daemon does not run the build: every `magus` invocation is its own process appending to <cacheDir>/runs/<inv>.jsonl, and that directory is the only place the daemon can observe a run from. journal.Broadcaster (the SSE route in live.go) is the in-process path for the run that owns it, and is unreachable here.
The filter runs BEFORE each send, as ListEvents applies it before paging, so a narrow filter makes a quiet stream rather than a stream of dropped frames. filter.time.since doubles as the resume cursor: the replay skips events older than it. That boundary is inclusive, so a caller resuming from the last event it saw receives that event again - at-least-once, which is the honest guarantee when the cursor is a millisecond timestamp several events can share.
A slow client cannot slow anything but itself: each stream owns its file offset and reads on the RPC's own goroutine, so it holds no lock and no buffer the run writes into. That is why there is nothing to drop here, unlike the SSE side, which shares a Broadcaster with the running process and skips a subscriber whose buffer is full.
Source Files