> ## Documentation Index
> Fetch the complete documentation index at: https://docs.westyx.cloud/llms.txt
> Use this file to discover all available pages before exploring further.
# Go SDK - Stream Observer
> Opt-in structured callbacks for SSE lifecycle events in the Go SDK.
Pass a `nexus.StreamObserver` via `Config.Observer` to receive structured callbacks for every SSE lifecycle transition. The observer is the **canonical** way to wire SDK state into ops dashboards, custom telemetry, or load-test tooling - far more reliable than scraping `Logger` output by substring match.
The observer is **opt-in**. When `Config.Observer == nil` the SDK uses `NoopStreamObserver` at zero overhead.
## The interface
```go theme={null}
type StreamObserver interface {
OnConnected()
OnDisconnected(err error)
OnEvent(name string)
OnReconnectAttempt(attempt int)
OnFallback(reason string)
OnQuarantined(reason string, expiresAt time.Time)
}
```
Embed `nexus.NoopStreamObserver` to override only the hooks you care about:
```go theme={null}
type myObserver struct {
nexus.NoopStreamObserver
events atomic.Int64
}
func (o *myObserver) OnEvent(string) { o.events.Add(1) }
```
## Hook semantics
| Hook | Fires when | Argument |
| ---------------------------------- | -------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------- |
| `OnConnected()` | `GET /v1/stream` returns `200 OK` and the SDK is ready to read events | none |
| `OnDisconnected(err)` | An active stream connection ends - clean cancel, transport error, or server-initiated close | `nil` for clean shutdown, non-nil otherwise |
| `OnEvent(name)` | A whitelisted SSE event arrives, **before** the SDK runs the event-triggered sync | event name string (e.g. `"config.updated"`) |
| `OnReconnectAttempt(n)` | The SDK is about to back off and retry after a transport error | 1-based attempt counter (max 3 with current threshold) |
| `OnFallback(reason)` | The SDK switches to TTL polling **or** signals an active rate-limit episode it will recover from automatically | `nexus.FallbackReasonMaxErrors` or `nexus.FallbackReasonRateLimited` |
| `OnQuarantined(reason, expiresAt)` | The backend returned `429` with a quarantine body - the SDK will not retry sync until `expiresAt` | `reason`: a short code describing why the service was quarantined; `expiresAt`: the absolute `time.Time` after which normal sync resumes |
### What counts as a "whitelisted event"
`OnEvent` fires only for events on the SDK's documented event-type list:
* `resync`
* `config.created` / `config.updated` / `config.deleted`
* `flag.created` / `flag.updated` / `flag.toggled` / `flag.deleted`
* `secret.created` / `secret.updated` / `secret.deleted`
Unknown events (or the WIF `auth_expiring` control event) do NOT trigger `OnEvent`.
### `OnFallback` reasons
```go theme={null}
const (
FallbackReasonMaxErrors = "max-errors"
FallbackReasonRateLimited = "rate-limited"
)
```
| Constant | When |
| --------------------------- | ---------------------------------------------------------------------------------------------------- |
| `FallbackReasonMaxErrors` | 3 consecutive transport errors - backoff exhausted, the SDK gives up and `RunStream` returns |
| `FallbackReasonRateLimited` | server returned `429 Too Many Requests` (tier connection-limit hit) - see active-recovery note below |
#### `FallbackReasonRateLimited` - active-recovery semantics (v0.4.0+)
`FallbackReasonRateLimited` is dispatched in **both** of the following situations:
* **Terminate case (no `Retry-After`)** - the server returned 429 without a parseable `Retry-After` header. The SDK closes the stream, `RunStream` returns, and TTL polling takes over.
* **Active-recovery case (`Retry-After` present)** - the server returned 429 with a parseable `Retry-After` header. The SDK fires `OnFallback("rate-limited")` for visibility, then sleeps the indicated delay (clamped `[5 s, 5 min]`) and **reconnects automatically**. `RunStream` does NOT return, the transport-failure counter is NOT incremented, and the stream resumes without intervention.
Observers that previously treated `FallbackReasonRateLimited` as a hard "stream is gone" signal should be aware that in v0.4.0+ the stream may recover on its own. A subsequent `OnConnected()` callback will fire when the reconnect succeeds.
### `OnQuarantined` - quarantine semantics (v0.4.0+)
`OnQuarantined(reason string, expiresAt time.Time)` fires when the backend returns `429 Too Many Requests` with a structured quarantine body:
```json theme={null}
{"error":"quarantined","reason":"abuse-detected","expires_at":"2026-05-28T14:00:00Z"}
```
**SDK behaviour while quarantined:**
* Background sync (`ensureFresh`) is skipped entirely - no requests are sent to the backend.
* The in-memory cache continues to be served for all reads (`GetConfig`, `GetSecret`, `GetFlag`).
* The quarantine state clears automatically when `expiresAt` passes.
## Threading contract
Observer methods are called **synchronously from the SDK's own goroutine**. They MUST return promptly. The SDK does not protect against blocking observers.
If your observer does anything substantial (network I/O, mutex contention, disk writes), copy the argument and dispatch the work to your own goroutine before returning:
```go theme={null}
func (o *slowObserver) OnEvent(name string) {
select {
case o.queue <- name:
default:
// Drop on overflow - applying backpressure to the SDK goroutine
// would block sync() and stall reads.
}
}
```
## Counter pattern
The most common use - counting events for a dashboard or load-test report:
```go theme={null}
type counterObserver struct {
nexus.NoopStreamObserver
events atomic.Int64
reconnects atomic.Int64
fallbacks atomic.Int64
}
func (c *counterObserver) OnEvent(string) { c.events.Add(1) }
func (c *counterObserver) OnReconnectAttempt(int) { c.reconnects.Add(1) }
func (c *counterObserver) OnFallback(string) { c.fallbacks.Add(1) }
obs := &counterObserver{}
client, _ := nexus.NewClient(ctx, nexus.Config{
BaseURL: "...",
APIKey: "...",
Observer: obs,
})
// later ...
fmt.Printf("SSE: events=%d reconnects=%d fallback=%d\n",
obs.events.Load(), obs.reconnects.Load(), obs.fallbacks.Load())
```
## Active-connection gauge pattern
Maintain a live "is the SDK connected" gauge:
```go theme={null}
type gaugeObserver struct {
nexus.NoopStreamObserver
connected atomic.Bool
}
func (g *gaugeObserver) OnConnected() { g.connected.Store(true) }
func (g *gaugeObserver) OnDisconnected(error) { g.connected.Store(false) }
func (g *gaugeObserver) OnFallback(string) { g.connected.Store(false) }
```
`OnFallback` is included because after the 3-strike fallback the SDK will not reconnect - the gauge should stay false until you spin up a new client or the next manual `RunStream(ctx)` call. Note: with `FallbackReasonRateLimited` (v0.4.0+) the SDK may auto-recover via `Retry-After`; the subsequent `OnConnected` will flip the gauge back to true.
## Per-event-type counter
```go theme={null}
type perEventObserver struct {
nexus.NoopStreamObserver
mu sync.Mutex
counts map[string]int
}
func (p *perEventObserver) OnEvent(name string) {
p.mu.Lock()
defer p.mu.Unlock()
if p.counts == nil {
p.counts = make(map[string]int)
}
p.counts[name]++
}
```
## Independence from `Logger`
The observer is independent of `Logger`. The two are complementary:
* `Logger` carries human-readable text for operators (`"nexus: SSE stream connected"`).
* `StreamObserver` carries structured metrics for tooling.
A consumer can set both, neither, or just one. Setting only the observer is the right choice when you don't want any text output from the SDK.
## Compatibility note
If you previously relied on substring-matching `Logger.Errorf` lines to count reconnects/fallbacks, switch to the observer - log line wording is not a stable contract and may change between SDK releases.