30 KiB
Implementation
This is the code-level walkthrough. It is organized by concern, not by file. We follow the path of one event from external API to a pixel on the globe, then loop back to cover the cross-cutting code.
All snippets are excerpts from the actual code. Snippets are intentionally not annotated with line numbers — formatting and small refactors invalidate line numbers in days, and the package + function name is enough to find any block again with grep.
The Event Type
Everything in the system funnels through events.Event. It is deliberately tiny.
// internal/events/event.go
package events
import "time"
type Event struct {
Topic Topic `json:"topic"`
Timestamp time.Time `json:"timestamp"`
Source string `json:"source"`
Payload any `json:"payload"`
}
Payload any is the interesting choice. We could have made it a sealed union of typed payloads, one per topic. We chose not to. The reasons:
- Heterogeneous payload shapes. A KEV add is a single CVE ID with a date. A DShield snapshot is a fused blob from three endpoints. A Coinbase tick is a price plus volume. Forcing them into one Go type tree adds zero runtime safety because the consumers (snapshot store, ws hub, alerts engine) all immediately re-marshal to JSON.
- The bus is the only typed boundary on the producer side. Every collector packs its own payload as
json.RawMessagebefore emitting, so the JSON shape is committed at the collector. Subscribers pay onejson.Unmarshalif they need fields.
The trade is real: a typo in a payload field name in a collector will not be caught at compile time. We mitigate with golden-file tests under each collector's testdata/ directory.
The Topic Enum
// internal/events/topic.go
type Topic string
const (
TopicHeartbeat Topic = "heartbeat"
TopicScanFirehose Topic = "scan_firehose"
TopicInternetOutage Topic = "internet_outage"
TopicBGPHijack Topic = "bgp_hijack"
TopicCVENew Topic = "cve_new"
TopicCVEUpdated Topic = "cve_updated"
TopicEPSS Topic = "epss"
TopicKEVAdded Topic = "kev_added"
TopicRansomwareVictim Topic = "ransomware_victim"
TopicCoinbasePrice Topic = "coinbase_price"
TopicEarthquake Topic = "earthquake"
TopicSpaceWeather Topic = "space_weather"
TopicWikipediaITN Topic = "wiki_itn"
TopicGDELTSpike Topic = "gdelt_spike"
TopicISSPosition Topic = "iss_position"
TopicCollectorState Topic = "collector_state"
)
func (t Topic) IsValid() bool { ... }
func AllTopics() []Topic { ... }
IsValid is used by the WebSocket handler to whitelist ?topics= query parameters. AllTopics is the default fallback when a client connects without specifying topics. Adding a feed means adding one constant here, one case in IsValid, and one entry in AllTopics. Then a collector emits with the new topic and the WS layer picks it up automatically.
The Bus
internal/bus/bus.go is the spine.
type Bus struct {
ch chan events.Event
persister Persister
broadcaster Broadcaster
logger *slog.Logger
dropped atomic.Uint64
subsMu sync.RWMutex
subscribers []chan events.Event
subDropped atomic.Uint64
}
func (b *Bus) Emit(ev events.Event) {
select {
case b.ch <- ev:
default:
b.dropped.Add(1)
b.logger.Warn("event bus full, dropped", "topic", ev.Topic, "source", ev.Source)
}
}
Emit is the producer-side API. The select with a default makes it non-blocking — if the buffer is full, we drop and increment the counter rather than block the calling collector goroutine.
The drain side, in Run, is the consumer:
func (b *Bus) Run(ctx context.Context) error {
defer b.closeSubscribers()
for {
select {
case <-ctx.Done():
return ctx.Err()
case ev := <-b.ch:
if b.persister != nil {
if err := b.persister.Save(ctx, ev); err != nil {
b.logger.Error("persist event failed", "err", err, "topic", ev.Topic)
}
}
if b.broadcaster != nil {
payload, err := json.Marshal(ev.Payload)
if err != nil {
b.logger.Error("marshal payload failed", "err", err, "topic", ev.Topic)
continue
}
b.broadcaster.Broadcast(string(ev.Topic), payload)
}
b.fanout(ev)
}
}
}
The order matters. Persist first so the snapshot is correct for any browser that connects between this event and the next. Broadcast second so live consumers see the event right after persistence. Fan out to channel subscribers (alerts engine) last because they may evaluate slow CEL predicates.
fanout is the second drop site:
func (b *Bus) fanout(ev events.Event) {
b.subsMu.RLock()
subs := b.subscribers
b.subsMu.RUnlock()
for _, ch := range subs {
select {
case ch <- ev:
default:
b.subDropped.Add(1)
}
}
}
A slow alerts engine cannot back up the bus. It just loses alert evaluations, which is logged and visible via SubscriberDroppedCount() for ops monitoring.
A Walking Tour of One Collector
Every collector has the same skeleton: poll, fetch, persist row(s), emit event(s), record state. We will use CVE because it is the most representative — pollable HTTP, dependent on a second API (EPSS), with both upsert and per-row event emission.
CVE Collector — The Tick
// internal/collectors/cve/collector.go
func (c *Collector) tick(ctx context.Context) {
end := time.Now().UTC()
start := end.Add(-c.cfg.Window)
resp, err := c.cfg.NVD.Fetch(ctx, start, end)
if err != nil {
c.logger.Warn("nvd fetch", "err", err)
c.cfg.State.RecordError(ctx, Name, err.Error())
return
}
Note the structure: a window of [end - Window, end], not "everything". CVE feeds are append-mostly but can re-publish modifications, so a 2-hour window with a 2-hour poll interval gives a 1-window overlap that catches late modifications without re-processing the whole history.
ids := make([]string, 0, len(resp.Vulnerabilities))
rows := make([]Row, 0, len(resp.Vulnerabilities))
for _, v := range resp.Vulnerabilities {
score, severity := v.PrimarySeverity()
raw, _ := json.Marshal(v)
rows = append(rows, Row{
CveID: v.CVE.ID,
Published: v.CVE.Published.Time,
LastModified: v.CVE.LastModified.Time,
Severity: severity,
CVSS: score,
Payload: raw,
})
ids = append(ids, v.CVE.ID)
}
scores, err := c.cfg.EPSS.LookupBatch(ctx, ids)
Two API calls per tick: NVD for the CVE bodies, EPSS for the scores. We do them sequentially because EPSS is keyed on the CVE IDs the NVD call just returned. The EPSS error is logged but does not abort the tick — a CVE without an EPSS score is still useful, just not as well prioritized.
for _, row := range rows {
if err := c.cfg.Repo.Upsert(ctx, row); err != nil { ... }
if s, ok := scores[row.CveID]; ok {
_ = c.cfg.Repo.UpdateEPSS(ctx, row.CveID, s.Score, s.Percentile)
row.EPSSScore = &s.Score
row.EPSSPercentile = &s.Percentile
}
body, _ := json.Marshal(row)
c.cfg.Emitter.Emit(events.Event{
Topic: events.TopicCVENew,
Timestamp: end,
Source: Name,
Payload: json.RawMessage(body),
})
}
c.cfg.State.RecordSuccess(ctx, Name, emitted)
}
Persist-then-emit, per row. We emit TopicCVENew for every row in the window, even ones that were already in the table. The reason: a CVE's EPSS score can change without the CVE itself changing, and the frontend should reflect that. We could split into cve_new vs cve_updated based on insert vs update, but the consumers do not need that distinction yet.
RecordSuccess writes one row to collector_state so /v1/admin/health and the freshness ribbon on the frontend can both see "CVE collector ran 12 minutes ago, processed 47 rows".
DShield — Multi-Endpoint Fusion
DShield is the cleanest example of the "fuse multiple upstream endpoints into one event" pattern.
// internal/collectors/dshield/collector.go
snaps, err := c.cfg.Fetcher.FetchAll(ctx) // hits 3 endpoints concurrently
merged := map[string]json.RawMessage{"ts": tsRaw}
for _, s := range snaps {
payload := s.Payload
if s.Kind == KindTopIPs {
payload = c.enrichSources(ctx, payload)
}
_ = c.cfg.Persister.PutSnapshot(ctx, now, s.Kind, payload)
merged[s.Kind] = payload
}
body, _ := json.Marshal(merged)
c.cfg.Emitter.Emit(events.Event{
Topic: events.TopicScanFirehose,
Timestamp: now,
Source: Name,
Payload: json.RawMessage(body),
})
The merged JSON has shape {ts, top_ips, top_ports, country_dist}. The frontend's DShieldPanel consumes the merged blob in one render pass. The Postgres persistence is per-kind so we can query "show me top IPs from 2026-04-01" without parsing a fused blob.
The enrichSources step is optional and runs only if an Enricher is configured (AbuseIPDB + GreyNoise). It iterates the top-IPs list, looks up each IP, and returns enriched rows with country code, classification, and threat actor name. If the lookup fails or the enricher is nil, it falls through to the original payload.
Coinbase — Persistent WebSocket
Coinbase is the only collector that does not poll. The decomposition into multiple files reflects how messy real WebSocket clients get:
client.go— raw connection,ReadFrame,WriteFrame.readloop.go— frame loop with sequencer reset on snapshot frames.aggregator.go— tick-to-minute-bar OHLC.reconnect.go— exponential backoff on disconnect.sequencer.go— gap detection onsequence_num.collector.go— orchestration.
The aggregator is the most interesting:
// internal/collectors/coinbase/aggregator.go
func (a *Aggregator) Push(t Tick) (*MinuteBar, MinuteBar) {
minute := t.TS.UTC().Truncate(time.Minute)
cur, exists := a.open[t.Symbol]
if !exists {
cur = MinuteBar{Symbol: t.Symbol, Minute: minute, Open: t.Price, High: t.Price, Low: t.Price, Close: t.Price, Volume24hAtClose: t.Volume24h}
a.open[t.Symbol] = cur
return nil, cur
}
if minute.After(cur.Minute) {
closed := cur
cur = MinuteBar{...} // start new minute
a.open[t.Symbol] = cur
return &closed, cur
}
// same minute, update OHLC
if t.Price.GreaterThan(cur.High) { cur.High = t.Price }
if t.Price.LessThan(cur.Low) { cur.Low = t.Price }
cur.Close = t.Price
cur.Volume24hAtClose = t.Volume24h
a.open[t.Symbol] = cur
return nil, cur
}
Returns (closed, current). When a tick crosses a minute boundary, the previous minute is returned as closed and the collector persists it to btc_eth_minute. The current minute is always returned for live display.
This is also why the minute volume column was renamed in migration 0005: the original column was called volume and we documented "per-minute volume", but the Coinbase ticker channel only gives 24-hour rolling volume, not per-trade size. The rename to volume_24h_at_close makes the actual semantic visible in the schema.
ISS — Two Cadences, Client-Side Propagation
// cmd/api/main.go
coll := iss.NewCollector(iss.CollectorConfig{
PositionInterval: cfg.Collectors.ISS.PositionInterval, // 10s
TLEInterval: cfg.Collectors.ISS.TLEInterval, // 24h
Fetcher: iss.NewClient(iss.ClientConfig{}),
TLEStore: iss.NewTLEStore(redis.Client),
...
})
The ISS collector polls wheretheiss.at every 10s for the live position (this is the source the dashboard's "current ISS pin" uses). It also polls CelesTrak every 24 hours for the TLE (Two-Line Element set), caches it in Redis, and emits it.
The frontend has frontend/src/lib/sgp4.ts (or equivalent) that propagates the cached TLE on each animation frame to draw the orbital track. So the visible orbit is computed client-side from a 24-hour-old TLE; only the "current pin" is server-pushed every 10s.
This is a good example of moving compute to the client when the algorithm is well-defined and the data input is small. SGP4 is ~200 lines of math; pushing the propagated position over WebSocket every animation frame would saturate the network for no benefit.
The WebSocket Hub
internal/ws/hub.go:
const (
defaultSubscriberBuf = 256
defaultPingInterval = 30 * time.Second
defaultPingTimeout = 10 * time.Second
defaultWriteTimeout = 5 * time.Second
defaultMaxSubs = 5000
)
func (h *Hub) Broadcast(topic events.Topic, payload []byte) {
env, err := EncodeEnvelope(string(topic), payload)
if err != nil {
h.logger.Error("encode envelope", "err", err, "topic", topic)
return
}
h.mu.Lock()
defer h.mu.Unlock()
for sub := range h.subs {
if _, ok := sub.topics[topic]; !ok {
continue
}
select {
case sub.msgs <- env:
default:
go sub.closeSlow()
}
}
}
The lock spans the whole iteration. This is fine because the iteration is fast (just a non-blocking channel send per sub) and the lock serializes against add/remove, which are the only other writers to h.subs. The connection close happens in a separate goroutine to avoid holding the lock through any I/O.
The slow-consumer comment in the constant block is worth quoting in full because it is the kind of thing you only learn by getting it wrong:
Per-connection outbound buffer. Must absorb peak event rate × browser stutter window: heartbeat (5s) + coinbase (~8/s peak) + ISS (10s) + occasional bursts of CVE/KEV/ransomware. 16 is too small — a single browser GC pause fills it and triggers slow-consumer close, which the user sees as "NO SIGNAL".
Serve Loop
// internal/ws/hub.go
func (h *Hub) Serve(ctx context.Context, c *cdrws.Conn, topics []events.Topic) error {
sub := newSubscriber(topics, h.bufSize, func() {
_ = c.Close(cdrws.StatusPolicyViolation, "slow consumer")
})
if !h.add(sub) {
_ = c.Close(cdrws.StatusTryAgainLater, "server at capacity")
return ErrAtCapacity
}
defer h.remove(sub)
connCtx, cancelRead := context.WithCancel(ctx)
defer cancelRead()
go func() {
for {
if _, _, err := c.Read(connCtx); err != nil {
cancelRead()
return
}
}
}()
pingT := time.NewTicker(h.pingInterval)
defer pingT.Stop()
for {
select {
case msg := <-sub.msgs:
wctx, cancel := context.WithTimeout(connCtx, h.writeTimeout)
err := c.Write(wctx, cdrws.MessageText, msg)
cancel()
if err != nil { return err }
case <-pingT.C:
pctx, cancel := context.WithTimeout(connCtx, h.pingTimeout)
err := c.Ping(pctx)
cancel()
if err != nil { return err }
case <-connCtx.Done():
_ = c.Close(cdrws.StatusNormalClosure, "")
return connCtx.Err()
}
}
}
The dedicated read goroutine is the post-mortem of a real bug. The coder/websocket library has a helper c.CloseRead(ctx) that auto-closes the connection on any received frame. We used it. The frontend's setReady() sends {"op":"init"} after snapshot hydration. That message tripped CloseRead's "received unexpected data" path, killing the connection with StatusPolicyViolation. The user saw a reconnect loop.
The fix: drain client→server messages explicitly with c.Read(connCtx) and discard them. Connection stays alive; ping/pong frames still get handled by the library.
Envelope
// internal/ws/envelope.go
func EncodeEnvelope(channel string, payload []byte) ([]byte, error) {
var buf bytes.Buffer
buf.WriteString(`{"ch":`)
chRaw, _ := json.Marshal(channel)
buf.Write(chRaw)
buf.WriteString(`,"data":`)
if len(payload) == 0 {
buf.WriteString("null")
} else {
buf.Write(payload)
}
buf.WriteString(`,"ts":`)
tsRaw, _ := json.Marshal(time.Now().UTC().Format(time.RFC3339Nano))
buf.Write(tsRaw)
buf.WriteByte('}')
return buf.Bytes(), nil
}
Built by hand, not via json.Marshal of a struct. Reason: the data field is already-encoded JSON. Marshalling a struct with Data json.RawMessage works but goes through a generic encoder path. Concatenation is faster and avoids re-allocating for the largest field. At ~50k events/min peak, this matters.
The Snapshot Store
// internal/snapshot/store.go
const (
keyPrefix = "state:"
keyHashPrice = "state:coinbase_price"
scanCount = 100
wrongTypeMarker = "WRONGTYPE"
)
func (s *Store) PutLatest(ctx context.Context, topic events.Topic, payload json.RawMessage) error {
return s.rdb.Set(ctx, keyPrefix+string(topic), []byte(payload), 0).Err()
}
func (s *Store) MergeSymbolMap(ctx context.Context, topic events.Topic, payload json.RawMessage) error {
var tick struct{ Symbol string `json:"symbol"` }
if err := json.Unmarshal(payload, &tick); err != nil || tick.Symbol == "" {
return fmt.Errorf("coinbase merge: no symbol in payload")
}
key := keyPrefix + string(topic)
err := s.rdb.HSet(ctx, key, tick.Symbol, []byte(payload)).Err()
if err == nil { return nil }
if !isWrongType(err) { return fmt.Errorf("redis hset %s/%s: %w", topic, tick.Symbol, err) }
if delErr := s.rdb.Del(ctx, key).Err(); delErr != nil { ... }
// retry as hash
return s.rdb.HSet(ctx, key, tick.Symbol, []byte(payload)).Err()
}
MergeSymbolMap has a self-healing migration path: if the key exists as a string (legacy format) and HSet fails with WRONGTYPE, we delete and retry as a hash. This means a developer who upgraded mid-tick does not have to manually redis-cli del.
GetAll uses SCAN rather than KEYS. KEYS * is O(n) blocking on the Redis main thread; SCAN is incremental. For a snapshot store with ~16 keys this is overkill, but the habit is the right one — KEYS in any production code path is a foot-gun.
func (s *Store) readKey(ctx context.Context, key string) (json.RawMessage, error) {
t, _ := s.rdb.Type(ctx, key).Result()
switch t {
case "string":
v, err := s.rdb.Get(ctx, key).Bytes()
return json.RawMessage(v), err
case "hash":
fields, _ := s.rdb.HGetAll(ctx, key).Result()
m := make(map[string]json.RawMessage, len(fields))
for k, v := range fields {
m[k] = json.RawMessage(v)
}
return json.Marshal(m)
}
return nil, fmt.Errorf("unsupported redis type %q for key %s", t, key)
}
Type-aware deserialization. The snapshot HTTP handler is dumb — core.OK(w, all) and a no-cache header — because all the work happens in the store.
The Alert Engine
internal/alerts/engine.go is the most complex part of the system because it has to compile predicates, refresh them, and dispatch with cooldowns and parallel delivery.
Compile and Refresh
func (e *Engine) reload(ctx context.Context) error {
all, _ := e.repo.ListAll(ctx)
indexed := make(map[string][]compiledRule, len(all))
for _, r := range all {
prog, err := e.compile(r.Predicate)
if err != nil {
e.logger.Warn("alerts: skipping rule with bad predicate",
"rule_id", r.ID, "topic", r.Topic, "err", err)
continue
}
indexed[r.Topic] = append(indexed[r.Topic], compiledRule{rule: r, program: prog})
}
e.rulesByTopic.Store(&indexed)
return nil
}
atomic.Pointer[map[string][]compiledRule] gives a lock-free read path. The map itself is never mutated after Store; reload always builds a fresh map and swaps the pointer. Readers grab the pointer once and operate on a stable map.
Compile errors do not fail the whole reload — a single bad rule predicate is logged and skipped. This matters: the rule UI lets users author predicates, and one user typing event.severity = "high" (assignment instead of comparison) should not silently break alerts for everyone.
Evaluate
func (e *Engine) Evaluate(ctx context.Context, ev events.Event) {
idx := e.rulesByTopic.Load()
if idx == nil { return }
rules, ok := (*idx)[string(ev.Topic)]
if !ok || len(rules) == 0 { return }
payload, err := normalizePayload(ev.Payload)
if err != nil {
e.logger.Warn("alerts: payload normalize failed", ...)
return
}
for _, cr := range rules {
if !cr.rule.Enabled { continue }
match, err := evalPredicate(cr.program, payload)
if err != nil { ... continue }
if !match { continue }
e.fire(ctx, cr.rule, ev, payload)
}
}
normalizePayload handles the any type discipline. The bus carries Payload any; CEL needs a map[string]any for field access. The function tries the cheap path (already a map), then json.RawMessage (unmarshal once), then full round-trip via json.Marshal+Unmarshal. The bulk of events come in as json.RawMessage because that is what the collectors produce.
Fire
func (e *Engine) fire(ctx context.Context, rule Rule, ev events.Event, payload map[string]any) {
channels, _ := e.loader.LoadChannels(ctx, rule.UserID)
cooldown := time.Duration(rule.CooldownSec) * time.Second
message := formatMessage(rule, ev, payload)
var wg sync.WaitGroup
var mu sync.Mutex
delivered := []string{}
deliveryErrs := map[string]string{}
for _, ch := range channels {
ch := ch
key := fmt.Sprintf("alert_cooldown:%s:%s:%s", rule.ID, ch.Type, ch.ID)
ok, _ := e.cooldowns.TryAcquire(ctx, key, cooldown)
if !ok { continue }
wg.Add(1)
go func() {
defer wg.Done()
err := e.notifier.SendAlert(ctx, ch, message)
mu.Lock(); defer mu.Unlock()
if err != nil {
deliveryErrs[ch.ID] = err.Error()
return
}
delivered = append(delivered, ch.ID)
}()
}
wg.Wait()
body, _ := json.Marshal(payload)
errBody, _ := json.Marshal(deliveryErrs)
_ = e.repo.RecordFire(ctx, HistoryRow{
RuleID: rule.ID, UserID: rule.UserID,
FiredAt: time.Now().UTC(),
Payload: body,
DeliveredTo: delivered,
DeliveryErrors: errBody,
})
}
Three things worth pointing at:
- Per-channel cooldown via Redis
SET NX EX. TheRedisCooldown.TryAcquireis a single round-trip and the TTL ensures the key auto-cleans. - Parallel delivery. Telegram is slow (2-5s for a botless rate-limited send), Slack is fast (~200ms). Sequential delivery would pace everything to the slowest. We dispatch all channels concurrently, then
wg.Waitfor all to return. - Fire history is recorded once per rule firing, not per channel.
deliveredlists which channels succeeded;deliveryErrsmaps channel ID to error text. That makes the UI's "alert history" page useful for debugging integration failures.
Webhook Encryption
// internal/notifications/crypto.go
func (e *Encryptor) Encrypt(plaintext []byte) (ciphertext, nonce []byte, err error) {
block, _ := aes.NewCipher(e.key)
gcm, _ := cipher.NewGCM(block)
nonce = make([]byte, gcm.NonceSize())
if _, err := io.ReadFull(rand.Reader, nonce); err != nil { ... }
ciphertext = gcm.Seal(nil, nonce, plaintext, nil)
return ciphertext, nonce, nil
}
Standard library AES-256-GCM. The key is base64-decoded from NOTIFICATION_ENCRYPTION_KEY and validated to be exactly 32 bytes at startup:
func NewEncryptor(b64Key string) (*Encryptor, error) {
key, _ := base64.StdEncoding.DecodeString(b64Key)
if len(key) != 32 {
return nil, fmt.Errorf("encryption key must be 32 bytes, got %d", len(key))
}
return &Encryptor{key: key}, nil
}
If NewEncryptor returns an error, the entire alert engine is disabled for the lifetime of the process — bridge and notifBridge stay nil, so notifBridge != nil checks in cmd/api/main.go skip the engine setup. This is fail-closed behaviour: misconfiguration disables the feature rather than running it with broken/no encryption.
Frontend Wiring
The WebSocket Driver
frontend/src/api/ws.ts:
export function createDashboardWS(opts: CreateDashboardWSOpts): DashboardWS {
const backoff = opts.backoff ?? DEFAULT_BACKOFF
const onEvent = opts.onEvent ?? (() => undefined)
let driver: WSDriver | null = null
let ready = false
let opened = false
let closed = false
let buffer: WSEvent[] = []
let nextDelay = backoff.initialMs
let reconnectTimer: ReturnType<typeof setTimeout> | null = null
function attach() {
if (closed) return
const d = opts.driver()
driver = d
opened = false
d.onOpen = () => {
nextDelay = backoff.initialMs
opened = true
sendInitIfReady()
}
d.onMessage = (data) => {
let parsed: WSEvent
try { parsed = JSON.parse(data) as WSEvent } catch { return }
if (ready) onEvent(parsed)
else buffer.push(parsed)
}
d.onClose = () => {
driver = null
opened = false
if (closed) return
const delay = nextDelay
nextDelay = Math.min(nextDelay * 2, backoff.maxMs)
reconnectTimer = setTimeout(attach, delay)
}
}
return {
connect() { attach() },
setReady() {
ready = true
sendInitIfReady()
const flush = buffer
buffer = []
for (const ev of flush) onEvent(ev)
},
disconnect() { ... },
}
}
The factory takes a driver thunk so tests can inject a fake. browserDriver is the production implementation that wraps the platform WebSocket. The buffer-flush in setReady is the snapshot+stream handshake on the client side.
Reconnect uses exponential backoff with nextDelay * 2 capped at maxMs (default 30s). On a clean open, nextDelay resets so a reconnect after long uptime starts at 1s again.
Store Routing
The App boot sequence (paraphrased):
- Mount providers (TanStack Query client, Zustand stores already global).
- Open the WebSocket via
createDashboardWS. Hand it anonEventthat dispatches byev.chinto the right store. - Fetch
/v1/snapshot. On success, hydrate every store from the response. - Call
ws.setReady(). Buffered events flush throughonEventinto the now-hydrated stores.
Stores are flat by topic. For example, stores/cve.ts holds the recent CVE list and exposes a setter (upsertCVE) that the WS dispatcher calls. Components subscribe via useCVEStore(state => state.list) so a Coinbase tick does not re-render the CVE panel.
The Globe
frontend/src/pages/globe/Globe.tsx is the big consumer. It subscribes to the globeEvents store, which is a fan-in of the topic-specific stores: outages, BGP hijacks, earthquakes, ransomware victims, mass-scan ASN dots, ISS position. The globe component listens, computes the layer mutations, and applies them in a requestAnimationFrame batch.
The TLE-driven ISS orbit track is propagated client-side. Every animation frame, the SGP4 propagator advances the satellite's mean anomaly and emits a (lat, lon) the globe layer renders as a polyline. The TLE itself comes from the snapshot or a iss_position event with a tle field; the position pin (separate from the orbit track) comes from the wheretheiss.at 10-second poll.
Common Pitfalls
A short list of mistakes we made and you might too.
Forgetting to register a topic in IsValid/AllTopics. Adding a topic constant is not enough. The WebSocket handler validates ?topics= against IsValid, and the default fallback uses AllTopics. Miss either and your new topic silently never reaches the browser.
Putting expensive work in Bus.Run. The drain loop is single-goroutine. If you add a synchronous DB write before the broadcast, you cap the entire system's broadcast rate at the DB write rate. Move expensive work into a subscriber channel or the persister implementation.
Using c.CloseRead on coder/websocket. It looks helpful. It closes the connection on any received frame, including the client init op the dashboard uses. Either drain reads explicitly or design a strict server-push protocol with close-on-receive semantics — pick one.
Not validating the encryption key length at startup. AES-256 silently accepts a 16-byte key as AES-128 if you do not check. The NewEncryptor constructor must enforce len(key) == 32.
Returning *MinuteBar from the aggregator and forgetting to nil-check. The Push return signature (*MinuteBar, MinuteBar) puts the closed bar behind a pointer. Persisting *closed without a nil check is the most common bug we caught in code review on the Coinbase package.
B-tree on time columns at scale. Tempting because it's the default. For tens of millions of append-mostly rows, BRIN is dozens of times smaller and just as fast for range scans. If you use B-tree on occurred_at, your index is bigger than the table within months.
Keying alert cooldowns globally instead of per-channel. A user with Telegram and Slack rules on the same predicate wants an independent cooldown on each. Keying on rule_id alone causes the channel that fires first to mute the other.
Where To Go Next
Continue to 04-CHALLENGES.md for extension ideas, harder feature work, and deployment hardening.