plexwatch

package
v0.3.6 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Nov 9, 2026 License: GPL-3.0 Imports: 26 Imported by: 0

Documentation

Overview

Package plexwatch syncs one viewer's watched marks, resume positions and ratings with that viewer's own Plex account, and (pms*.go) with Plex Media Servers the owner runs.

Account sync

Transport: discover.provider.plex.tv, the calls Plex Web makes for "Mark as watched" on Discover, verified against a real account on 2026-11-07:

  • GET /library/metadata/{k1,k2,...}/userState answers MediaContainer. UserState (an object for one key, an array for several) with ratingKey, type, viewCount, viewOffset (ms), viewState ("progress", or "ignore" once removed from Continue Watching), lastViewedAt (unix seconds), userRating (1–10) and, for shows, viewedLeafCount. Titles with no state are omitted. Mixed movie/show/episode batches work; reads are capped at maxStateBatch keys.
  • PUT /actions/scrobble, /actions/unscrobble, /actions/rate (rating=1–10, -1 clears) and /actions/removeFromContinueWatching, each with identifier=tv.plex.provider.discover&key={key}. They are idempotent. Scrobble and unscrobble reset viewOffset; unscrobble keeps lastViewedAt, so "watched" is viewCount>0, never lastViewedAt alone. removeFromContinueWatching sets viewState "ignore" and keeps viewOffset.
  • GET /timeline?ratingKey={key}&key=/library/metadata/{key}&state=stopped &time={ms}&duration={ms} sets the resume position (PUT/POST answer 404; no duration answers 400). It moves lastViewedAt, sets viewState back to "progress", and never marks a title watched, even past 90%: Ether's own threshold decides and sends a scrobble. An episode's timeline does not move its show's userState.

{key} is the hex id inside a title's verified plex://movie|show|episode GUID (internal/plexguid). Titles without a verified GUID are never synced and no other identity matching is attempted.

Credential: the viewer's own Plex account token from the PIN sign-in (POST/GET plex.tv/api/v2/pins), stored sealed per viewer and sent only in X-Plex-Token. The PIN must be approved by the Plex account already recorded for that viewer; the owner's token is never used for a viewer. A 401 or 403 from any call erases the token and marks the connection for reconnecting.

Consent: the account's "Sync My Watch State and Ratings" setting, GET plex.tv/api/v2/user/view_state_sync ({"consent":bool}). Plex does not enforce it on the Discover routes, so Ether does: while it is off nothing is imported and nothing is sent. It is read once per viewer per pass, never per title: at the start of every scheduled pass (which also runs, at the viewer's interval, for viewers who only send while it is off) and before the first send of each delivery pass. While it is off the viewer's queued changes are parked, not dropped; the first read that finds it on wakes them, still subject to read-back and newest-wins.

Ether → Plex: watch changes write durable intents to state_sync_journal in the watch transaction, stamped with the viewer's fence generation. The sender claims a row, resolves the GUID, and admits it under the fence lock (generation unchanged, sending still on) before the request leaves. Turning sending off, disconnecting and reconnecting move the fence under the same lock, so nothing is admitted afterwards; a request admitted just before cannot be recalled. Admission sets never_sent=0. When an attempt ends without a definitive answer (5xx, timeout, crash), the next attempt reads userState first: if Plex already holds the change it is done, if Plex shows a newer change it yields, otherwise it is sent again. 429 pauses the viewer for Retry-After; other failures back off exponentially.

Resume positions are a third journal aspect. A position reported during playback is held for state.OffsetHold and later positions of the title replace it in place, so playback sends at most one position per hold; pause and stop send the latest at once. A watch change supersedes a queued position, and positions share the title's watch lock, so a stale position never reaches Plex after a completion or unwatch. Clearing a position (a rewind to the start) is read back first and sent as an unscrobble only while Plex has the title unwatched. A position Ether has no duration for is not queued (Plex refuses it).

Continue Watching removals are journaled in the same aspect, one per in-progress movie or episode the removal covers (a show covers each of its episodes with a position), carrying Ether's position. A removal replaces a queued position in place and a later position replaces a removal, as playback brings a title back. It is read back first: if Plex played the title since, it yields; if Plex lacks the position, the position is sent before removeFromContinueWatching.

Plex → Ether: a leader-only, bounded background pass per viewer reads userState in batches over Ether's verified movie and show identities, resuming from a key cursor. Shows whose watched-episode count or last view moved queue their episodes for reading. Because an episode's position does not move its show, each sweep is followed by a pass over the viewer's active shows (any watch, progress or rating activity in Ether within 180 days, plus the viewer's watchlist), reading their verified episodes in batches from a show cursor; its cost follows the viewer's activity, never the library. plexwatch_remote keeps what was last read per title, so only changes since then apply: a Plex view or resume position by its lastViewedAt against the local watch clock; an unwatch, cleared position, Continue Watching removal or rating change, which Plex does not timestamp, only when Ether has not changed that title since the previous read. A watch change read together with a position wins over it. A Plex removal of a title Ether has in progress removes it in Ether and keeps the resume point; a Plex playback newer than the viewer's removal in Ether brings the title back. Imported changes are account events, which are never journaled, and values Ether wrote itself are adopted rather than applied, so nothing echoes in either direction.

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrSendingOff   = errors.New("plexwatch: sending to Plex is off for this viewer")
	ErrReceivingOff = errors.New("plexwatch: reading from Plex is off for this viewer")
)

ErrSendingOff means the viewer's direction does not send to Plex, and ErrReceivingOff that it does not read from Plex.

View Source
var (
	// ErrPMSDuplicate means this server is already configured.
	ErrPMSDuplicate = errors.New("plexwatch: this Plex Media Server is already added")
	// ErrPMSDifferentServer means an edited address answers as another server.
	ErrPMSDifferentServer = errors.New("plexwatch: that address is a different Plex Media Server")
	// ErrPMSLimit means the server or account limit is reached.
	ErrPMSLimit = errors.New("plexwatch: limit reached")
	// ErrPMSNeedsToken means sending to Plex for this account needs its own token.
	ErrPMSNeedsToken = errors.New("plexwatch: sending to Plex for this person needs their token for this server")
	// ErrPMSInvalid means the request is malformed.
	ErrPMSInvalid = errors.New("plexwatch: invalid server settings")
)
View Source
var ErrAccountMismatch = errors.New("plexwatch: plex account does not belong to this viewer")

ErrAccountMismatch means the Plex account that approved a PIN is not the viewer's account.

View Source
var ErrNotConnected = errors.New("plexwatch: viewer has not connected a Plex account")

ErrNotConnected means the viewer has no usable Plex account connection.

View Source
var ErrPMSAddress = errors.New("plexwatch: enter the server address, like http://192.168.1.10:32400")

ErrPMSAddress means the server address is not an http(s) origin.

View Source
var ErrPMSNotOwner = errors.New("plexwatch: this token is not the server owner's")

ErrPMSNotOwner means the token is valid but is not the server owner's.

View Source
var ErrPMSToken = errors.New("plexwatch: the Plex Media Server refused this token")

ErrPMSToken means the Plex Media Server refused the token.

View Source
var ErrPMSUnreachable = errors.New("plexwatch: the Plex Media Server could not be reached")

ErrPMSUnreachable means the server could not be reached or did not answer as a PMS.

View Source
var ErrRevoked = errors.New("plexwatch: account token rejected")

ErrRevoked means Plex refused the viewer's stored account token (HTTP 401 or 403). The viewer must connect the account again.

Functions

This section is empty.

Types

type AccountGUIDs

type AccountGUIDs interface {
	GUIDMap
	VerifiedRange(ctx context.Context, after, below core.RatingKey, limit int) ([]plexguid.Verified, error)
}

AccountGUIDs is GUIDMap plus a local, paged walk of verified identities, which bounds the background import to titles Ether already matched.

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client is the allowlisted Plex HTTP API used by the account path: PIN sign-in and identity on plex.tv, watch state on discover.provider.plex.tv. Every call carries the viewer's own account token in X-Plex-Token.

type GUIDMap

type GUIDMap interface {
	Resolve(ctx context.Context, key core.RatingKey) (string, error)
	ExactKey(ctx context.Context, guid string) (core.RatingKey, error)
}

GUIDMap resolves a local title to a verified plex:// GUID and the reverse for a GUID already stored for exactly one title.

type Options

type Options struct {
	DB       *sql.DB
	State    *state.State
	Sealer   config.FieldSealer
	GUID     AccountGUIDs
	NodeID   string
	ClientID string
	BaseURL  string
	HTTP     *http.Client
	Logger   *slog.Logger
	Now      func() time.Time
	// Watchlist lists a viewer's watchlist titles (optional). Its shows
	// join the shows with recent activity whose episodes each sweep reads.
	Watchlist func(ctx context.Context, userID int64) ([]core.RatingKey, error)
	// Continue mirrors a Continue Watching removal (removed) or restore
	// read from the account into stores beyond state (optional).
	Continue func(ctx context.Context, userID int64, key core.RatingKey, removed bool) error
}

Options configure a Service. BaseURL is the test origin for every Plex host. Leave it empty in production.

type PMSAccount added in v0.3.4

type PMSAccount struct {
	ID     int64  `json:"id"`
	Name   string `json:"name"`
	Owner  bool   `json:"owner"`
	UserID int64  `json:"userId"`
}

PMSAccount is a PMS account and, when mapped, its viewer.

type PMSCatalog added in v0.3.4

type PMSCatalog interface {
	Get(ctx context.Context, key core.RatingKey) (*core.Title, error)
	ByExternalIDForType(ctx context.Context, source, id string, typ core.MediaType) (*core.Title, error)
}

PMSCatalog verifies matched titles and resolves IMDb/TVDB identifiers.

type PMSMapping added in v0.3.4

type PMSMapping struct {
	AccountID   int64  `json:"accountId"`
	AccountName string `json:"accountName"`
	Owner       bool   `json:"owner"`
	UserID      int64  `json:"userId"`
	Direction   int    `json:"direction"`
	Interval    int    `json:"interval"`
	CanSend     bool   `json:"canSend"`
	LastImport  int64  `json:"lastImport"`
	LastSend    int64  `json:"lastSend"`
	NextSync    int64  `json:"nextSync"`
	Problem     string `json:"problem"`
	Unmatched   int    `json:"unmatched"`
}

PMSMapping is one PMS account mapped to one Ether viewer.

type PMSMappingInput added in v0.3.4

type PMSMappingInput struct {
	UserID    int64
	Direction int
	Interval  int
	Token     *string
}

PMSMappingInput changes one account's mapping. UserID 0 removes it. Token nil keeps the stored per-account token; "" clears it.

type PMSServer added in v0.3.4

type PMSServer struct {
	ID        int64        `json:"id"`
	Name      string       `json:"name"`
	URL       string       `json:"url"`
	MachineID string       `json:"machineId"`
	Version   string       `json:"version"`
	Owner     string       `json:"owner"`
	CheckedAt int64        `json:"checkedAt"`
	Problem   string       `json:"problem"`
	Viewers   []PMSMapping `json:"viewers"`
}

PMSServer is one configured server as the owner console shows it. It never carries a token.

type PMSUnmatched added in v0.3.4

type PMSUnmatched struct {
	Direction string `json:"direction"`
	Ref       string `json:"ref"`
	Title     string `json:"title"`
	Reason    string `json:"reason"`
	SeenAt    int64  `json:"seenAt"`
}

PMSUnmatched is a title one direction could not match by identifier.

type Servers added in v0.3.4

type Servers struct {
	// contains filtered or unexported fields
}

Servers syncs watched state with Plex Media Servers the owner runs, over each server's own HTTP API and with the owner's token for it.

The owner adds servers and maps each PMS account to one Ether viewer. Nothing is guessed from names. Reads are incremental: PMS watch history after a stored cursor, and Ether's watch rows after a stored (updated_at, rating_key) cursor on the state_watch_updated index. No request path lists a whole PMS library or Ether's catalogue.

Echo safety rests on plexwatch_pms_sent, one row per server, viewer and title: the watched state the server is known to hold, the local state it was last reconciled with, and when Ether itself last wrote it. A write is made only when the local watch state changed since that reconciliation and the server does not already hold it, so a change imported from server A is never sent back to A, while server B (which has no such row) gets it once. History rows that are the echo of Ether's own scrobble are ignored.

This path is independent of the per-viewer Plex account sync (Service).

func NewServers added in v0.3.4

func NewServers(o ServersOptions) (*Servers, error)

NewServers starts nothing; call Run under cluster leadership.

func (*Servers) Accounts added in v0.3.4

func (s *Servers) Accounts(ctx context.Context, id int64) ([]PMSAccount, error)

Accounts lists the server's accounts live, with their current viewer.

func (*Servers) Add added in v0.3.4

func (s *Servers) Add(ctx context.Context, name, baseURL, token string) (PMSServer, error)

Add validates and stores a server. The token is sealed and never returned.

func (*Servers) Close added in v0.3.4

func (s *Servers) Close() error

Close stops Run.

func (*Servers) List added in v0.3.4

func (s *Servers) List(ctx context.Context) ([]PMSServer, error)

List returns every server with its mappings.

func (*Servers) MapAccount added in v0.3.4

func (s *Servers) MapAccount(ctx context.Context, id, accountID int64, in PMSMappingInput) (PMSServer, error)

MapAccount maps a PMS account to a viewer and sets direction and interval. Turning a direction on starts it from now: existing history is not copied either way. Sending for an account other than the owner's needs that account's own token for this server.

func (*Servers) Remove added in v0.3.4

func (s *Servers) Remove(ctx context.Context, id int64) error

Remove forgets a server, its mappings, cursors, matches and echo records.

func (*Servers) Run added in v0.3.4

func (s *Servers) Run(ctx context.Context)

Run syncs due mappings until ctx ends or Close. Start it under cluster leadership; the per-mapping lease also serializes overlapping leaders and the owner's Sync now.

func (*Servers) RunOnce added in v0.3.4

func (s *Servers) RunOnce(ctx context.Context)

func (*Servers) Server added in v0.3.4

func (s *Servers) Server(ctx context.Context, id int64) (PMSServer, error)

Server returns one server and its mappings.

func (*Servers) SyncNow added in v0.3.4

func (s *Servers) SyncNow(ctx context.Context, id int64) (PMSServer, error)

SyncNow runs every mapping of one server now. A server that asked Ether to wait (HTTP 429) is still left alone until it said.

func (*Servers) Unmatched added in v0.3.4

func (s *Servers) Unmatched(ctx context.Context, id, userID int64, after string) ([]PMSUnmatched, error)

Unmatched returns one bounded page of a viewer's unmatched titles.

func (*Servers) Update added in v0.3.4

func (s *Servers) Update(ctx context.Context, id int64, name, baseURL, token string) (PMSServer, error)

Update renames a server, moves it to a new address or replaces its token. An empty token keeps the stored one. The address must still answer as the same server, so cursors and matches stay valid.

type ServersOptions added in v0.3.4

type ServersOptions struct {
	// Enabled reads the live global switch. Nil preserves the legacy enabled behavior.
	Enabled  func() bool
	DB       *sql.DB
	State    *state.State
	Sealer   config.FieldSealer
	GUID     GUIDMap
	Catalog  PMSCatalog
	ClientID string
	HTTP     *http.Client
	Logger   *slog.Logger
	Now      func() time.Time
	// ViewerExists reports whether an Ether viewer id is real. Nil skips the check.
	ViewerExists func(context.Context, int64) (bool, error)
}

ServersOptions configure Servers. HTTP defaults to the direct egress route.

type Service

type Service struct {
	// contains filtered or unexported fields
}

Service is one server's Plex account watch-state integration.

func New

func New(opts Options) (*Service, error)

New starts nothing. Call Run to deliver and RunImports to import.

func (*Service) BeginPIN

func (s *Service) BeginPIN(ctx context.Context, userID, plexID int64) (Status, error)

BeginPIN starts a sign-in that must be approved as plexID, not as whatever account the owner happens to be signed into.

func (*Service) Close

func (s *Service) Close() error

Close stops Run and RunImports.

func (*Service) DeliverOnce

func (s *Service) DeliverOnce(ctx context.Context) (bool, error)

DeliverOnce handles at most one live intent, as a pass of its own. more is true when a row was claimed, including rows put back for later.

func (*Service) Disconnect

func (s *Service) Disconnect(ctx context.Context, userID, plexID int64) (Status, error)

Disconnect removes the stored account token, cancels pending delivery and forgets what was read from that account.

func (*Service) ExportCurrent

func (s *Service) ExportCurrent(ctx context.Context, userID, plexID int64) (Status, error)

ExportCurrent queues the viewer's watched movies and episodes, their ratings and their resume positions for sending. It runs only when the viewer asks; turning sending on does not upload history by itself. The scan reads only this viewer's own watch rows.

func (*Service) FinishPIN

func (s *Service) FinishPIN(ctx context.Context, userID, plexID int64) (Status, error)

FinishPIN stores the account only when its Plex id equals plexID. The viewer approving the PIN as themselves is the consent to sync. Connecting the same account again (after Plex refused the old token) keeps the chosen direction, the import cursor and the last-read Plex state.

func (*Service) ImportCurrent

func (s *Service) ImportCurrent(ctx context.Context, userID, plexID int64) (Status, error)

ImportCurrent runs one bounded import pass now. Scheduled and manual passes share the cursor, lease and status.

func (*Service) Retry

func (s *Service) Retry(ctx context.Context, userID, plexID int64) (Status, error)

Retry makes failed sends claimable again and clears a rate-limit or network pause so the next pass runs now.

func (*Service) Run

func (s *Service) Run(ctx context.Context)

Run delivers until Close or ctx ends. It does not decide cluster leadership: journal claims and title locks are safe on every node.

func (*Service) RunImports added in v0.3.3

func (s *Service) RunImports(ctx context.Context)

RunImports is started under cluster leadership. Each pass is bounded and the persisted lease also serializes manual imports and overlapping leaders.

func (*Service) SetDirection added in v0.3.3

func (s *Service) SetDirection(ctx context.Context, userID, plexID int64, direction, interval int) (Status, error)

SetDirection selects outbound (bit 0) and inbound (bit 1) sync. Enabling outbound does not upload existing history.

func (*Service) Status

func (s *Service) Status(ctx context.Context, userID, plexID int64) (Status, error)

type Status

type Status struct {
	Connected      bool   `json:"connected"`
	PlexID         int64  `json:"plexId,omitempty"`
	Username       string `json:"username,omitempty"`
	Direction      int    `json:"direction"`
	ImportInterval int    `json:"importInterval"`
	LastImport     int64  `json:"lastImport"`
	LastFullImport int64  `json:"lastFullImport"`
	LastSend       int64  `json:"lastSend"`
	NextImport     int64  `json:"nextImport"`
	Problem        string `json:"problem"`
	NeedsReconnect bool   `json:"needsReconnect"`
	Pending        int    `json:"pending"`
	Failed         int    `json:"failed"`
	Restricted     bool   `json:"restricted"`
	PINCode        string `json:"pinCode,omitempty"`
	PINURL         string `json:"pinUrl,omitempty"`
	PINExpires     string `json:"pinExpires,omitempty"`
	AccountMatch   bool   `json:"accountMatch"`
}

Status is the viewer-facing connection, with no token and no history. Problem is "" or one of "token" (connect again), "consent" (the account's "Sync My Watch State and Ratings" setting is off in Plex), "rate_limited" or "network".

type StatusError

type StatusError struct {
	Status     int
	RetryAfter time.Duration
	Op         string
}

StatusError is an HTTP result that is not success. Body is not retained.

func (*StatusError) Error

func (e *StatusError) Error() string

func (*StatusError) Is

func (e *StatusError) Is(target error) bool

Is maps definitive credential refusals onto ErrRevoked.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL