Documentation
¶
Overview ¶
Package notify implements the Plex WebSocket notification listener.
The Listener dials Plex's /:/websockets/notifications endpoint, decodes the NotificationContainer JSON envelope into typed events, and delivers them to a caller-supplied Handler. Reconnect/backoff is test-injectable via Config (no global vars). Disconnect reasons are classified with typed sentinel errors so Loki alert rules can segment by cause without substring matching on error text.
Stable contracts preserved by this package:
- Plex WebSocket JSON wire format (struct tags on Notification / PlayEvent / TimelineEntry).
- WARN/ERROR slog keys and the ReasonXxx string values, which Loki alert rules match on.
- /:/websockets/notifications URL path, 1 MB read limit, the X-Plex-Token header. The read-idle backstop is Config-driven (ReadIdleTimeout, default 1 hour), not a fixed 5-minute deadline.
Index ¶
- Constants
- Variables
- func BuildStreamCacheKey(userID, ratingKey string, audioID, subID int) string
- func BuildTimelineCacheKey(itemID string) string
- func ClassifyError(err error) string
- func IsRelevantPlayEvent(ev PlayEvent) bool
- func IsRelevantTimelineEntry(entry *TimelineEntry) bool
- func TimelineAction(entry *TimelineEntry) string
- type Config
- type Handler
- type Listener
- type Notification
- type PlayEvent
- type PlexClient
- type TimelineEntry
Constants ¶
const ( ReasonUnknown = "unknown" ReasonServerClose = "server_close" ReasonReadLimit = "read_limit" ReasonDialFailed = "dial_failed" ReasonReadError = "read_error" )
Reason codes surfaced to logs (and thus to Loki alert rules). These string values are byte-for-byte frozen because Loki alert rules match on them.
Variables ¶
var ( // ErrDialFailed marks a failure to establish the WebSocket // connection (DNS, TCP, handshake, TLS). ErrDialFailed = errors.New("websocket dial") // ErrReadLimit marks a read that exceeded the 1 MB read limit. ErrReadLimit = errors.New("read limited") // ErrReadError marks a generic read failure (EOF on a truncated // frame, i/o timeout, transport error). ErrReadError = errors.New("websocket read") // ErrServerClose marks a clean or expected close from the server // side (close frames 1000/1001/1006, plain EOF). ErrServerClose = errors.New("server close") )
Sentinel errors used by the Listener to label disconnect causes. The Listener wraps the underlying error with %w so ClassifyError can use errors.Is / errors.As for typed matching instead of substring search on err.Error(). Code outside this package rarely needs these directly; ClassifyError is the consumer API.
Functions ¶
func BuildStreamCacheKey ¶
BuildStreamCacheKey builds a deduplication key from user, episode, and current stream IDs so we only process when the selection actually changes. The "streams:" prefix and colon-separated layout are part of the on-disk cache.json schema.
func BuildTimelineCacheKey ¶ added in v1.4.0
BuildTimelineCacheKey builds the per-episode timeline (library-scan) dedup key. The "timeline:" prefix is part of the on-disk cache.json schema.
func ClassifyError ¶
ClassifyError maps a WebSocket disconnect error to a stable reason code so alerts can segment by cause without matching on error text.
Classification order (first match wins):
- nil → ReasonUnknown
- context.DeadlineExceeded → ReasonReadError (read-idle backstop fired)
- typed sentinel wraps via errors.Is (ErrReadLimit, ErrServerClose, ErrDialFailed, ErrReadError)
- *websocket.CloseError via errors.As for codes 1000 / 1001 / 1006 → ReasonServerClose
- default → ReasonUnknown
func IsRelevantPlayEvent ¶
IsRelevantPlayEvent returns true if a play event should be processed (state is playing/paused and has a rating key).
func IsRelevantTimelineEntry ¶
func IsRelevantTimelineEntry(entry *TimelineEntry) bool
IsRelevantTimelineEntry returns true if a timeline entry should be processed (episode type, metadata/media created or updated, non-empty item ID).
func TimelineAction ¶
func TimelineAction(entry *TimelineEntry) string
TimelineAction returns "scan_new" if the entry represents a newly created item, or "scan_updated" otherwise. The returned strings are byte-for-byte frozen — they are emitted as log/metric values consumed by dashboards.
Types ¶
type Config ¶
type Config struct {
// MinBackoff is the initial delay between reconnect attempts and the
// floor nextBackoff clamps to. Default: 1s.
MinBackoff time.Duration
// MaxBackoff is the ceiling for exponential growth of the reconnect
// delay. Default: 30s.
MaxBackoff time.Duration
// StableThreshold is how long a connection must stay open before the
// backoff is reset on the next reconnect (so a long-lived session
// pays back accumulated backoff). Default: 1 minute.
StableThreshold time.Duration
// ReadIdleTimeout is the application-level backstop for a stuck
// websocket read. The primary dead-connection detection is TCP
// keepalive (configured on the dialer at 30s probe interval); the
// read timeout exists only as a safety net for pathological cases
// where the OS reports the connection alive but the server has
// silently stopped sending. Default: 1 hour. Plex doesn't send
// heartbeats and can legitimately be quiet for tens of minutes
// during off-peak windows; a short timeout here only churns the
// connection without improving correctness.
ReadIdleTimeout time.Duration
}
Config holds the tunables for the reconnect loop. Production code uses DefaultConfig(); tests construct a Config with shrunk durations so the reconnect-loop assertions run fast without mutating package globals.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns the production values. Preserves behaviour of the original wsMinBackoff / wsMaxBackoff / wsStableThreshold package vars from main.go.
type Handler ¶
type Handler interface {
OnPlay(ctx context.Context, ev PlayEvent)
OnTimeline(ctx context.Context, entries []TimelineEntry)
}
Handler receives decoded events from the Listener. Implementations live in the composition root (main package) and typically fan out per-event work to the sync subsystem.
OnPlay is called for each PlaySessionStateNotification entry in a "playing" Notification, unfiltered; the caller applies its own relevance policy (see IsRelevantPlayEvent). OnTimeline is called once per Notification with the full TimelineEntry slice so the caller can apply its own per-entry relevance, dedup, or batching policy.
The Handler is invoked synchronously from the read loop; a slow handler delays the next Read and can trigger the keepalive timeout. Handlers that need to perform long work should hand it off to a goroutine or worker pool.
type Listener ¶
type Listener struct {
// contains filtered or unexported fields
}
Listener dials Plex's /:/websockets/notifications endpoint, decodes the JSON envelope, and delivers events to a Handler with an outer reconnect loop and stable-connection backoff reset.
func NewListener ¶
func NewListener(client PlexClient, cfg Config) *Listener
NewListener builds a Listener from a Plex client and a Config. Production callers pass DefaultConfig(); tests pass a shrunk Config so the reconnect-loop assertions run quickly.
type Notification ¶
type Notification struct {
NotificationContainer struct {
Type string `json:"type"`
PlaySessionStateNotification []PlayEvent `json:"PlaySessionStateNotification"`
TimelineEntry []TimelineEntry `json:"TimelineEntry"`
} `json:"NotificationContainer"`
}
Notification is the top-level envelope Plex sends over the WebSocket. Field names and JSON tags mirror the Plex NotificationContainer wire format byte-for-byte.
type PlayEvent ¶
type PlayEvent struct {
SessionKey string `json:"sessionKey"`
ClientIdentifier string `json:"clientIdentifier"`
RatingKey string `json:"ratingKey"`
State string `json:"state"`
ViewOffset int64 `json:"viewOffset"`
}
PlayEvent represents a single play-session state notification from Plex.
type PlexClient ¶
PlexClient is the subset of an HTTP-based Plex client the Listener needs. Satisfied by *plex.Client; declared here to keep notify decoupled from the plex package (no import cycle, easy to fake in tests).
type TimelineEntry ¶
type TimelineEntry struct {
ItemID string `json:"itemID"`
Identifier string `json:"identifier"`
SectionID string `json:"sectionID"`
MetadataState string `json:"metadataState"`
MediaState string `json:"mediaState"`
Type int `json:"type"`
State int `json:"state"`
UpdatedAt int64 `json:"updatedAt"`
}
TimelineEntry represents a library scan timeline event from Plex.
Source Files