orchard

  1package main
  2
  3import (
  4	"context"
  5	"encoding/json"
  6	"fmt"
  7	"log/slog"
  8	"sync"
  9	"time"
 10)
 11
 12// State is the whole page as data. Every panel is rebuilt into this and the
 13// same struct is what the template renders on a cold load and what goes down
 14// the SSE connection on every update, so there is one shape to get right rather
 15// than a server one and a browser one that drift.
 16type State struct {
 17	Market    Market       `json:"market"`
 18	Rates     Rates        `json:"rates"`
 19	Sectors   []SectorCell `json:"sectors"`
 20	Signal    Signal       `json:"signal"`
 21	Earnings  Earnings     `json:"earnings"`
 22	Wire      []Headline   `json:"wire"`
 23	HN        []Story      `json:"hn"`
 24	Lobsters  []Story      `json:"lobsters"`
 25	Weather   Weather      `json:"weather"`
 26	Air       Air          `json:"air"`
 27	Alerts    []Alert      `json:"alerts"`
 28	Outlook   Outlook      `json:"outlook"`
 29	Steam     []Game       `json:"steam"`
 30	Streaming []Title      `json:"streaming"`
 31	Systems   Systems      `json:"systems"`
 32	Feeds     []Feed       `json:"feeds"`
 33	Updated   string       `json:"updated"`
 34	Guarded   []string     `json:"guarded"`
 35}
 36
 37// Store holds the latest of everything and hands out snapshots. Each poller
 38// writes its own field, so the lock is only ever held for a struct copy.
 39type Store struct {
 40	mu    sync.RWMutex
 41	state State
 42
 43	// The daily series the conditions panel reads, kept here because the market
 44	// poll rebuilds that panel every thirty seconds and this only refreshes
 45	// hourly.
 46	history *history
 47
 48	hub *Hub
 49}
 50
 51func NewStore(hub *Hub) *Store { return &Store{hub: hub} }
 52
 53func (s *Store) Snapshot() State {
 54	s.mu.RLock()
 55	defer s.mu.RUnlock()
 56	return s.state
 57}
 58
 59// update applies one change and pushes the whole state out. Taking a mutator
 60// rather than a field keeps every write on one path, which is what makes the
 61// broadcast unconditional and impossible to forget.
 62func (s *Store) update(f func(*State)) {
 63	s.mu.Lock()
 64	f(&s.state)
 65	s.state.Updated = time.Now().UTC().Format(time.RFC3339)
 66	snapshot := s.state
 67	s.mu.Unlock()
 68
 69	b, err := json.Marshal(snapshot)
 70	if err != nil {
 71		slog.Error("state unmarshalable", slog.Any("err", err))
 72		return
 73	}
 74	s.hub.Broadcast(b)
 75}
 76
 77// Hub is the SSE fan-out. One poller writes and every open browser reads, so
 78// ten tabs on this page still cost one request upstream.
 79type Hub struct {
 80	mu      sync.Mutex
 81	clients map[chan []byte]struct{}
 82	last    []byte
 83
 84	// Closed on nothing; sent to when the first browser of a quiet spell
 85	// connects. The markets loop waits on it alongside its timer, so somebody
 86	// opening the page does not have to sit out the rest of an idle interval.
 87	// Buffered by one and sent without blocking, so an arrival while the loop
 88	// is mid-fetch is remembered rather than lost.
 89	wake chan struct{}
 90}
 91
 92func NewHub() *Hub {
 93	return &Hub{
 94		clients: map[chan []byte]struct{}{},
 95		wake:    make(chan struct{}, 1),
 96	}
 97}
 98
 99// Subscribe returns a channel of frames and the function that closes it. The
100// buffer is small: a browser that cannot keep up with a state this size is
101// gone, and dropping the frame is better than holding the broadcast for it.
102func (h *Hub) Subscribe() (<-chan []byte, func()) {
103	ch := make(chan []byte, 4)
104
105	h.mu.Lock()
106	first := len(h.clients) == 0
107	h.clients[ch] = struct{}{}
108	last := h.last
109	h.mu.Unlock()
110
111	if first {
112		select {
113		case h.wake <- struct{}{}:
114		default:
115		}
116	}
117
118	// A new connection gets the current state at once rather than waiting out
119	// the poll interval, which is what makes a reconnect invisible.
120	if last != nil {
121		ch <- last
122	}
123
124	return ch, func() {
125		h.mu.Lock()
126		if _, ok := h.clients[ch]; ok {
127			delete(h.clients, ch)
128			close(ch)
129		}
130		h.mu.Unlock()
131	}
132}
133
134func (h *Hub) Broadcast(b []byte) {
135	h.mu.Lock()
136	defer h.mu.Unlock()
137
138	h.last = b
139	for ch := range h.clients {
140		select {
141		case ch <- b:
142		default:
143		}
144	}
145}
146
147func (h *Hub) Watching() int {
148	h.mu.Lock()
149	defer h.mu.Unlock()
150	return len(h.clients)
151}
152
153// Poll intervals. Markets move and the rest do not, so only markets gets a fast
154// one, and it only gets it while somebody is looking.
155const (
156	marketWatched = 30 * time.Second
157
158	// Yahoo's edge sends cache-control: max-age=10 on these responses, so
159	// anything under about ten seconds is asking for bytes it already has.
160	// Thirty is comfortably past that and still live enough to watch.
161
162	marketIdle = 5 * time.Minute
163	newsEvery  = 5 * time.Minute
164)
165
166// Prime fetches what the first page render needs, synchronously, before the
167// server starts listening.
168//
169// The daily history comes first because the conditions panel is built by the
170// market poll out of it, and priming them the other way round renders a panel
171// with one of its four rows for the first thirty seconds after every deploy.
172func (s *Store) Prime(ctx context.Context, g *Guard) {
173	s.refreshSignal(ctx, g)
174	s.refreshMarket(ctx, g)
175}
176
177// Run starts one goroutine per source and blocks until ctx is done. It must not
178// start before the listener is up: the health strip probes this process over
179// loopback, and a first round against a socket nobody is listening on reports
180// the site the dashboard is running on as unknown until the next round a minute
181// later, which is every deploy.
182func (s *Store) Run(ctx context.Context, g *Guard) {
183	go s.loop(ctx, "market", func() time.Duration {
184		if s.hub.Watching() > 0 {
185			return marketWatched
186		}
187		// Nobody is on the page, so this drops to a keep-warm poll. It does not
188		// stop outright because the first visitor after a quiet night should
189		// open onto real numbers, not a spinner.
190		return marketIdle
191	}, func() { s.refreshMarket(ctx, g) })
192
193	go s.loop(ctx, "news", nil, func() { s.refreshNews(ctx, g) })
194	go s.loop(ctx, "wire", nil, func() { s.refreshWire(ctx, g) })
195	go s.loop(ctx, "signal", nil, func() { s.refreshSignal(ctx, g) })
196	go s.loop(ctx, "board", nil, func() { s.refreshBoard(ctx, g) })
197	go s.loop(ctx, "earnings", nil, func() { s.refreshEarnings(ctx, g) })
198	go s.loop(ctx, "alerts", nil, func() { s.refreshAlerts(ctx, g) })
199	go s.loop(ctx, "air", nil, func() { s.refreshAir(ctx, g) })
200	go s.loop(ctx, "outdoors", nil, func() { s.refreshOutlook(ctx, g) })
201	go s.loop(ctx, "steam", nil, func() { s.refreshSteam(ctx, g) })
202	go s.loop(ctx, "streaming", nil, func() { s.refreshStreaming(ctx, g) })
203	go s.loop(ctx, "weather", nil, func() { s.refreshWeather(ctx, g) })
204	go s.loop(ctx, "systems", nil, func() { s.refreshSystems(ctx, g) })
205
206	// The guard writes itself out on a timer rather than on every call, so a
207	// thirty second poll does not mean a file write per tick.
208	go func() {
209		t := time.NewTicker(time.Minute)
210		defer t.Stop()
211		for {
212			select {
213			case <-ctx.Done():
214				g.Flush()
215				return
216			case <-t.C:
217				g.Flush()
218				s.update(func(st *State) {
219					st.Guarded = g.Status()
220					st.Feeds = g.Feeds(time.Now())
221				})
222			}
223		}
224	}()
225
226	<-ctx.Done()
227}
228
229// loop runs work on an interval. every is a function so the markets loop can
230// change its mind about the cadence between ticks; a nil one means use the
231// fixed interval the source was registered with.
232//
233// The markets loop also wakes when the first browser of a quiet spell connects.
234// Without it a visitor arriving a second after an idle timer was set would wait
235// out the whole five minutes watching a page that says it is live.
236func (s *Store) loop(ctx context.Context, name string, every func() time.Duration, work func()) {
237	fixed := map[string]time.Duration{
238		"news":      newsEvery,
239		"wire":      wireEvery,
240		"signal":    signalEvery,
241		"board":     boardEvery,
242		"earnings":  earningsEvery,
243		"alerts":    alertsEvery,
244		"air":       airEvery,
245		"outdoors":  outdoorsEvery,
246		"steam":     steamEvery,
247		"streaming": streamingEvery,
248		"weather":   weatherEvery,
249		"systems":   probeEvery,
250	}[name]
251
252	if every == nil {
253		every = func() time.Duration { return fixed }
254	}
255
256	var wake <-chan struct{}
257	if name == "market" {
258		wake = s.hub.wake
259	}
260
261	// The first run is immediate for everything Prime did not already do.
262	last := time.Now()
263	if name != "market" && name != "signal" {
264		work()
265	}
266
267	for {
268		t := time.NewTimer(every())
269		select {
270		case <-ctx.Done():
271			t.Stop()
272			return
273
274		case <-t.C:
275			work()
276			last = time.Now()
277
278		case <-wake:
279			t.Stop()
280			// A viewer arriving right after a fetch does not need another one,
281			// and the hub has already replayed the current state to them.
282			if time.Since(last) >= marketWatched {
283				work()
284				last = time.Now()
285			}
286		}
287	}
288}
289
290func (s *Store) refreshMarket(ctx context.Context, g *Guard) {
291	quotes, err := fetchStrip(ctx, g)
292	if err != nil {
293		slog.Warn("market poll failed", slog.String("component", "market"), slog.Any("err", err))
294		// The previous cards stay up rather than being blanked, and the panel
295		// marks itself stale so the page says so instead of quietly lying.
296		s.update(func(st *State) { st.Market.Stale = true })
297		return
298	}
299	m := buildMarket(quotes, time.Now())
300	m.Updated = time.Now().UTC().Format("15:04:05")
301	m = carrySparks(m, s.Snapshot().Market)
302
303	s.mu.RLock()
304	h := s.history
305	s.mu.RUnlock()
306
307	sig := buildSignal(h, quotes)
308
309	s.update(func(st *State) {
310		st.Market = m
311		st.Signal = sig
312	})
313}
314
315// refreshBoard is the rates and sectors poll. Neither moves on the scale the
316// market strip does, and riding the fast poll cost a third of every Yahoo
317// request this site makes for numbers that had not changed.
318func (s *Store) refreshBoard(ctx context.Context, g *Guard) {
319	quotes, err := fetchQuotes(ctx, g, rateAndSectorSymbols(), sessionRange)
320	if err != nil {
321		slog.Warn("board poll failed", slog.String("component", "board"), slog.Any("err", err))
322		return
323	}
324
325	rates := buildRates(quotes)
326	sectors := buildSectors(quotes)
327	s.update(func(st *State) {
328		st.Rates = rates
329		st.Sectors = sectors
330	})
331}
332
333func (s *Store) refreshSignal(ctx context.Context, g *Guard) {
334	h, err := fetchHistory(ctx, g, signalSymbol)
335	if err != nil {
336		slog.Warn("history poll failed", slog.String("component", "signal"), slog.Any("err", err))
337		return
338	}
339	s.mu.Lock()
340	s.history = h
341	s.mu.Unlock()
342}
343
344func (s *Store) refreshEarnings(ctx context.Context, g *Guard) {
345	rows, err := fetchEarnings(ctx, g, time.Now())
346	if err != nil {
347		slog.Warn("earnings poll failed", slog.String("component", "earnings"), slog.Any("err", err))
348		return
349	}
350	s.update(func(st *State) { st.Earnings = rows })
351}
352
353// An empty alert list is a result, not a failure, so this writes it: the panel
354// has to be able to say all clear rather than showing yesterday's warning.
355func (s *Store) refreshAlerts(ctx context.Context, g *Guard) {
356	alerts, err := fetchAlerts(ctx, g)
357	if err != nil {
358		slog.Warn("alerts poll failed", slog.String("component", "local"), slog.Any("err", err))
359		return
360	}
361	s.update(func(st *State) { st.Alerts = alerts })
362}
363
364func (s *Store) refreshAir(ctx context.Context, g *Guard) {
365	air, err := fetchAir(ctx, g)
366	if err != nil {
367		slog.Warn("air poll failed", slog.String("component", "local"), slog.Any("err", err))
368		return
369	}
370
371	// Pollen is the unofficial source here, so it losing its footing costs its
372	// own row and not the air quality beside it.
373	if index, top, err := fetchPollen(ctx, g); err != nil {
374		slog.Warn("pollen poll failed", slog.String("component", "local"), slog.Any("err", err))
375	} else {
376		air.Pollen = fmt.Sprintf("%.1f", index)
377		air.PollenState = pollenBand(index)
378		air.PollenTop = top
379		air.PollenKnown = true
380		air.PollenFill, air.PollenLevel = gauge(index, 9.7, 2.5, 4.9, 7.3)
381	}
382
383	s.update(func(st *State) { st.Air = air })
384}
385
386func (s *Store) refreshStreaming(ctx context.Context, g *Guard) {
387	titles, err := fetchStreaming(ctx, g)
388	if err != nil {
389		slog.Warn("streaming poll failed", slog.String("component", "streaming"), slog.Any("err", err))
390		return
391	}
392	s.update(func(st *State) { st.Streaming = titles })
393}
394
395func (s *Store) refreshOutlook(ctx context.Context, g *Guard) {
396	outlook, err := fetchOutlook(ctx, g, time.Now())
397	if err != nil {
398		slog.Warn("outlook poll failed", slog.String("component", "local"), slog.Any("err", err))
399		return
400	}
401	s.update(func(st *State) { st.Outlook = outlook })
402}
403
404func (s *Store) refreshSteam(ctx context.Context, g *Guard) {
405	games, err := fetchSteam(ctx, g)
406	if err != nil {
407		slog.Warn("steam poll failed", slog.String("component", "steam"), slog.Any("err", err))
408		return
409	}
410	s.update(func(st *State) {
411		if keepSteam(games, st.Steam) {
412			st.Steam = games
413		}
414	})
415}
416
417func (s *Store) refreshWire(ctx context.Context, g *Guard) {
418	wire, err := fetchWire(ctx, g, time.Now())
419	if err != nil {
420		slog.Warn("wire poll failed", slog.String("component", "wire"), slog.Any("err", err))
421		return
422	}
423	s.update(func(st *State) { st.Wire = wire })
424}
425
426func (s *Store) refreshNews(ctx context.Context, g *Guard) {
427	now := time.Now()
428
429	if stories, err := fetchHackerNews(ctx, g, now); err != nil {
430		slog.Warn("hacker news poll failed", slog.String("component", "news"), slog.Any("err", err))
431	} else {
432		s.update(func(st *State) { st.HN = stories })
433	}
434
435	if stories, err := fetchLobsters(ctx, g, now); err != nil {
436		slog.Warn("lobsters poll failed", slog.String("component", "news"), slog.Any("err", err))
437	} else {
438		s.update(func(st *State) { st.Lobsters = stories })
439	}
440}
441
442func (s *Store) refreshWeather(ctx context.Context, g *Guard) {
443	w, err := fetchWeather(ctx, g)
444	if err != nil {
445		slog.Warn("weather poll failed", slog.String("component", "weather"), slog.Any("err", err))
446		return
447	}
448	s.update(func(st *State) { st.Weather = w })
449}
450
451func (s *Store) refreshSystems(ctx context.Context, g *Guard) {
452	sys := buildSystems(ctx, g, time.Now())
453	// Guarded and Feeds are the same fact told two ways, in the status line and
454	// in the SIGNAL panel, so they are written together or the page contradicts
455	// itself for up to a minute.
456	s.update(func(st *State) {
457		st.Systems = sys
458		st.Feeds = g.Feeds(time.Now())
459		st.Guarded = g.Status()
460	})
461}