state

package
v0.3.3 Latest Latest
Warning

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

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

Documentation

Overview

Package state implements per-user watch state and playlists over SQLite: core.State and core.Playlists.

Semantics follow Plex:

  • Progress at >= 90% of duration marks the item watched (view count increments once per viewing, the resume offset clears).
  • Progress below 5% is treated as "not really started": no resume point is kept. Between 5% and 90% the item is in progress.
  • Scrobble marks watched unconditionally; Unscrobble clears the view count and resume point (the user's rating is kept).

Index

Constants

View Source 
const (
	SyncComplete   = "complete"
	SyncUnplayed   = "unplayed"
	SyncRatingKind = "rating"
)

SyncKind is what a journal row asks a subscriber to deliver. Watch kinds are the Plex viewState values "complete" and "unplayed". "rating" carries the raw local rating; the subscriber decides which values it can send.

View Source 
const (
	WatchedThreshold = 0.90
	StartedThreshold = 0.05
)

WatchedThreshold and StartedThreshold are the fractions of duration at which an item counts as watched / meaningfully started.

Variables

View Source 
var ErrLocalNewer = errors.New("state: local watch state is newer than the remote record")

ErrLocalNewer means a remote import is older than the stored row.

Functions

func Fraction 

func Fraction(ws core.WatchStatus) float64

Fraction returns offset/duration, or 0 when the duration is unknown.

func ImportPlaylistTx 

func ImportPlaylistTx(ctx context.Context, tx *sql.Tx, id, userID int64, title string, keys []core.RatingKey, at time.Time) (int64, error)

ImportPlaylistTx preserves imported playlist IDs while enforcing ownership.

func IsInProgress 

func IsInProgress(ws core.WatchStatus) bool

IsInProgress reports whether a status has a resume point.

func IsWatched 

func IsWatched(ws core.WatchStatus) bool

IsWatched reports whether a status counts as watched (watched at least once).

func WithOrigin added in v0.3.0 

func WithOrigin(ctx context.Context, origin Origin) context.Context

WithOrigin attaches the authenticated request origin to committed events.

Types

type Event 

type Event struct {
	Kind      EventKind
	UserID    int64
	Key       core.RatingKey
	Offset    time.Duration
	Duration  time.Duration
	Fraction  float64
	Rating    float64 // EventRated only
	State     string  // raw timeline state for EventProgress: playing, paused, stopped
	Completed bool
	At        time.Time
	Imported  bool // reviewed history import; never an active playback timeline
	Origin    Origin
	// Account is a manual Plex-account import. Local clients refresh from it.
	// It is not journaled and is not a Trakt or Simkl export.
	Account bool
}

Event is a watch-state change. Fraction is offset/duration (0 when the duration is unknown). Completed is true for EventCompleted/EventScrobble.

type EventKind 

type EventKind string

EventKind identifies a state change subscribers may react to. 

const (
	EventProgress   EventKind = "progress"   // a timeline update (incl. start/stop)
	EventCompleted  EventKind = "completed"  // crossed the watched threshold
	EventScrobble   EventKind = "scrobble"   // explicitly marked watched
	EventUnscrobble EventKind = "unscrobble" // explicitly marked unwatched
	EventRated      EventKind = "rated"
)

type ExportEntry 

type ExportEntry struct {
	Key    core.RatingKey
	Rating float64
	At     time.Time
}

ExportEntry is the state-owned, source-neutral view used by CSV exports.

type ImportBatch 

type ImportBatch struct{ Records, History []ImportRecord }

type ImportRecord 

type ImportRecord struct {
	UserID           int64
	Key              core.RatingKey
	Views            int
	Offset, Duration time.Duration
	Rating           float64
	ViewedAt         time.Time
	RatingsOnly      bool
}

ImportRecord uses native time units; source-specific parsing belongs to the importer. RatingsOnly preserves existing progress and completion counts.

type Options 

type Options struct {
	Logger *slog.Logger
	Now    func() time.Time
	// Availability, when set (a clustered node on a shared PostgreSQL
	// database), makes timeline progress survive a database outage: a
	// Progress the database cannot take is held in memory (the latest per
	// user and title) and written when the database answers again, so
	// players are never blocked or failed by the outage. Nil (standalone)
	// writes progress synchronously as before.
	Availability *store.Availability
}

Options tune a State. Now is injectable for tests.

type Origin added in v0.3.0 

type Origin struct {
	Surface             core.Protocol
	SessionID, ClientID string
}

Origin identifies the surface and client that made a watch-state change.

type Rollup 

type Rollup struct {
	Total      int
	Watched    int  // viewCount > 0
	InProgress int  // has a resume point
	AllWatched bool // Total > 0 && Watched == Total
	AnyWatched bool
}

Rollup summarises the watched state of a set of children (the episodes of a season/show) for Plex's viewedLeafCount-style fields.

func Summarize 

func Summarize(children []core.RatingKey, statuses map[core.RatingKey]core.WatchStatus) Rollup

Summarize rolls up children using an already-fetched status map. Missing entries count as untouched.

type ShowActivity added in v0.3.0 

type ShowActivity struct {
	UserID     int64
	Show       core.RatingKey
	LastViewed time.Time
}

ShowActivity is a viewer's most recent watch activity on one show.

type State 

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

State implements core.State and core.Playlists.

func New 

func New(db *sql.DB, opts Options) *State

New builds a State over an already-migrated database (see store.Open).

func (*State) AckSync added in v0.2.57 

func (s *State) AckSync(ctx context.Context, seq int64, holder string, at time.Time) (bool, error)

AckSync marks a send done only while the row is still the live intent. ok is false when a newer intent superseded it during the send.

func (*State) Add 

func (s *State) Add(ctx context.Context, userID int64, key core.RatingKey, items []core.RatingKey) error

Add appends items to a playlist.

func (*State) ApplyRemote added in v0.2.57 

func (s *State) ApplyRemote(ctx context.Context, userID int64, key core.RatingKey, watched *bool, watchAt time.Time, rating *float64, ratingAt time.Time) error

ApplyRemote writes a remote watched flag and/or rating without journaling an upload. Each aspect is compared with its own clock: a newer local rating does not block an older watch mark, and a newer watch mark does not block an older rating. ErrLocalNewer means every requested aspect was older than the local clock.

func (*State) BeginSourceSession added in v0.3.0 

func (s *State) BeginSourceSession(ctx context.Context) (int64, error)

BeginSourceSession obtains a shared monotonic epoch before any accepted samples. A late report from an older session cannot replace a newer one.

func (*State) ChildrenStatus 

func (s *State) ChildrenStatus(ctx context.Context, userID int64, children []core.RatingKey) (Rollup, map[core.RatingKey]core.WatchStatus, error)

ChildrenStatus fetches statuses for children and rolls them up.

func (*State) ClaimSync added in v0.2.57 

func (s *State) ClaimSync(ctx context.Context, holder string, now time.Time, stale time.Duration) (SyncRow, bool, error)

ClaimSync takes the oldest live intent for holder. A row already claimed by a live holder is left alone until stale.

func (*State) Close 

func (s *State) Close(ctx context.Context) error

Close writes progress still held from a database outage (bounded by ctx). It is a no-op on a standalone server.

func (*State) CompactSync added in v0.2.57 

func (s *State) CompactSync(ctx context.Context, now time.Time) error

CompactSync deletes terminal journal rows and idle locks. A live intent, the newest undelivered row for a viewer and title, is kept.

func (*State) Create 

func (s *State) Create(ctx context.Context, userID int64, title string, items []core.RatingKey) (*core.Playlist, error)

Create makes a playlist with the given items in order.

func (*State) DeadSync added in v0.2.57 

func (s *State) DeadSync(ctx context.Context, seq int64, holder string, reason string) error

DeadSync stops delivery of one row without changing local watch state.

func (*State) DeferSync added in v0.2.57 

func (s *State) DeferSync(ctx context.Context, seq int64, holder string, until time.Time, reason string) error

DeferSync puts a claimed row back, later, with an error note.

func (*State) Delete 

func (s *State) Delete(ctx context.Context, userID int64, key core.RatingKey) error

Delete removes a playlist. Deleting another user's playlist is ErrNotFound.

func (*State) DropPendingSync added in v0.2.57 

func (s *State) DropPendingSync(ctx context.Context, userID int64, reason string) error

DropPendingSync cancels every undelivered intent for a viewer.

func (*State) DropPendingSyncTx added in v0.2.57 

func (s *State) DropPendingSyncTx(ctx context.Context, tx *sql.Tx, userID int64, reason string) error

DropPendingSyncTx is DropPendingSync on an open transaction, so a consent change and the cancellation commit together.

func (*State) ExportEntries 

func (s *State) ExportEntries(ctx context.Context, userID int64, ratings bool) ([]ExportEntry, error)

func (*State) Familiar added in v0.3.3 

func (s *State) Familiar(ctx context.Context, userID int64, keys []core.RatingKey) (map[core.RatingKey]int, error)

Familiar reports userID's history with each of keys (movies and shows): core.AffinityViewer when they watched, started or rated it, core.AffinityServer when only someone else on this server did (a PMS library holds what its household watches), absent otherwise. A show counts through any of its episodes. Every probe is an index seek that stops at its first row, so the cost is the number of keys, never the size of anyone's history.

func (*State) FollowedShows added in v0.3.0 

func (s *State) FollowedShows(ctx context.Context, since time.Time) ([]ShowActivity, error)

FollowedShows lists, per viewer, the shows with an episode watched or in progress since the given time, most recently viewed first.

func (*State) Generation added in v0.3.0 

func (s *State) Generation() uint64

Generation changes when watch progress, played state, or a rating is saved.

func (*State) Get 

func (s *State) Get(ctx context.Context, userID int64, key core.RatingKey) (core.WatchStatus, error)

Get returns the status for one key. A never-touched key returns a zero status (with Key set) and no error.

func (*State) GetMany 

func (s *State) GetMany(ctx context.Context, userID int64, keys []core.RatingKey) (map[core.RatingKey]core.WatchStatus, error)

GetMany returns a status for every requested key (zero when untouched).

func (*State) GetPlaylist 

func (s *State) GetPlaylist(ctx context.Context, userID int64, key core.RatingKey) (*core.Playlist, error)

Get returns one playlist owned by the user, or core.ErrNotFound.

func (*State) History 

func (s *State) History(ctx context.Context, userID int64, since time.Time, limit int) ([]core.HistoryEntry, error)

History returns the viewing log, newest first. A zero since means all.

func (*State) ImportTransaction 

func (s *State) ImportTransaction(ctx context.Context, prepare func(*sql.Tx) (ImportBatch, error)) error

ImportTransaction commits a reviewed import with the caller's own bookkeeping and list writes. State owns every watch/history write and emits ordinary state events only after the complete transaction succeeds. Repeated history is inert.

func (*State) InProgress 

func (s *State) InProgress(ctx context.Context, userID int64, limit int) ([]core.WatchStatus, error)

InProgress returns items with a resume point, most recently viewed first.

func (*State) InProgressAny 

func (s *State) InProgressAny(ctx context.Context) ([]core.RatingKey, error)

InProgressAny lists distinct keys in progress for any user.

func (*State) LastSource added in v0.3.0 

func (s *State) LastSource(ctx context.Context, userID int64, key core.RatingKey) (*core.PlaybackSourceHistory, error)

LastSource returns nil for unknown source identity or an untouched item.

func (*State) List 

func (s *State) List(ctx context.Context, userID int64) ([]core.Playlist, error)

List returns the user's playlists (with items), most recently updated first.

func (*State) LockSync added in v0.2.57 

func (s *State) LockSync(ctx context.Context, userID int64, key core.RatingKey, aspect SyncAspect, holder string, now time.Time, ttl time.Duration) (bool, error)

LockSync holds one user/title/aspect so two nodes cannot deliver it at once.

func (*State) Move 

func (s *State) Move(ctx context.Context, userID int64, key core.RatingKey, from, to int) error

Move moves the item at from so it lands at index to (both 0-based, to is the final position in the resulting list).

func (*State) NoteSyncIntents added in v0.2.57 

func (s *State) NoteSyncIntents(ctx context.Context, rows []SyncRow) error

NoteSyncIntents appends explicit intents (an owner-chosen export) in one transaction, superseding older undelivered rows of the same aspect.

func (*State) OutboxStatus 

func (s *State) OutboxStatus() (st store.OutboxStatus, ok bool)

OutboxStatus reports timeline writes held through a database outage; ok is false when progress is written synchronously (standalone).

func (*State) PageUserData added in v0.3.0 

func (s *State) PageUserData(ctx context.Context, userID int64, keys []core.RatingKey) (map[core.RatingKey]core.WatchStatus, error)

PageUserData reads watch progress and the page's favorite, not-interested and watchlist flags in one query. Keys the caller has not authorized must not be passed in.

func (*State) PauseSync added in v0.2.57 

func (s *State) PauseSync(ctx context.Context, seq int64, holder string, until time.Time, reason string) error

PauseSync puts a claimed row back without counting an attempt. Delivery uses it when another node still holds the title lock, so this node can move on to a different row instead of claiming the same one again.

func (*State) PendingSyncCount added in v0.2.57 

func (s *State) PendingSyncCount(ctx context.Context, userID int64) (pending, dead int, err error)

PendingSyncCount is undelivered, not superseded, not dead.

func (*State) Playlists 

func (s *State) Playlists() core.Playlists

Playlists returns s as a core.Playlists.

func (*State) Progress 

func (s *State) Progress(ctx context.Context, userID int64, key core.RatingKey, offset, duration time.Duration, state string) error

Progress records a timeline update. state is Plex's timeline state ("playing", "paused", "stopped", "buffering"). See the package doc for the 5% start and 90% completion rules. A duration of 0 keeps the previously known duration; with no known duration the offset is kept as a resume point and nothing is ever marked watched by progress alone.

On a clustered node a Progress the database cannot take within a short timeout is held and written later (see Options.Availability); it then returns nil.

func (*State) ProgressWithReceipt added in v0.3.0 

func (s *State) ProgressWithReceipt(ctx context.Context, user int64, key core.RatingKey, position, duration time.Duration, status string, receipt func(*sql.Tx) error) error

ProgressWithReceipt commits receipt, watch state and history together. The callback may run again after a row conflict and must use only the supplied transaction. This synchronous path never acknowledges an outbox-only write.

func (*State) Rate 

func (s *State) Rate(ctx context.Context, userID int64, key core.RatingKey, rating float64) error

Rate sets the user's rating (0-10; 0 clears it).

func (*State) RecordLastSource added in v0.3.0 

func (s *State) RecordLastSource(ctx context.Context, userID int64, key core.RatingKey, epoch, sequence int64, source core.PlaybackSourceObservation) error

RecordLastSource stores only source-bound facts. Session epoch precedes the sample order, so old sessions and delayed transactions cannot overwrite newer accepted samples. It describes the last durable sample, which can lag if a later asynchronous write fails. Unknown revision clears prior facts.

func (*State) ReleaseSync added in v0.2.57 

func (s *State) ReleaseSync(ctx context.Context, seq int64, holder string) error

ReleaseSync drops a claim so another worker can take the row.

func (*State) Remove 

func (s *State) Remove(ctx context.Context, userID int64, key core.RatingKey, itemIndex int) error

Remove deletes the item at itemIndex (0-based).

func (*State) RetryDeadSync added in v0.2.57 

func (s *State) RetryDeadSync(ctx context.Context, userID int64, reasons []string) error

RetryDeadSync makes matching dead or deferred rows claimable again. Rejected and unsupported rows stay dead.

func (*State) Scrobble 

func (s *State) Scrobble(ctx context.Context, userID int64, key core.RatingKey) error

Scrobble marks an item watched: view count increments, offset clears.

func (*State) ScrobbleMany 

func (s *State) ScrobbleMany(ctx context.Context, userID int64, keys []core.RatingKey) error

ScrobbleMany marks several items watched in one transaction — the Plex front end expands a show or season scrobble into its episode keys and calls this. Each key gets a history entry and an EventScrobble.

func (*State) SetJournalFault added in v0.2.57 

func (s *State) SetJournalFault(fn func() error)

SetJournalFault installs a test hook that fails the journal write inside the watch transaction. Production leaves it unset.

func (*State) SetJournalGate added in v0.2.57 

func (s *State) SetJournalGate(fn func(ctx context.Context, tx *sql.Tx, userID int64) (generation int64, ok bool, err error))

SetJournalGate installs the per-viewer decision for durable account-sync intents. The generation is the consent fence observed by that decision. The watch transaction still commits when the gate returns false.

func (*State) SetLikes added in v0.3.0 

func (s *State) SetLikes(ctx context.Context, userID int64, key core.RatingKey, likes *bool) error

SetLikes maps a tri-state like onto the shared 0–10 viewer rating atomically.

func (*State) Subscribe 

func (s *State) Subscribe(fn func(Event)) func()

Subscribe registers fn for state events (scrobbles, progress, ratings) so Trakt sync, recommendations, prefetch etc. can react. It returns an unsubscribe function. Handlers run synchronously after the change is committed; a panicking handler is recovered and logged.

func (*State) SyncCurrent added in v0.2.57 

func (s *State) SyncCurrent(ctx context.Context, seq int64, holder string) (bool, error)

SyncCurrent reports whether seq is still the holder's live intent.

func (*State) UnlockSync added in v0.2.57 

func (s *State) UnlockSync(ctx context.Context, userID int64, key core.RatingKey, aspect SyncAspect, holder string) error

UnlockSync releases a lock this holder owns.

func (*State) Unscrobble 

func (s *State) Unscrobble(ctx context.Context, userID int64, key core.RatingKey) error

Unscrobble marks an item unwatched: view count and offset clear. The user's rating and history log are kept (history is a log, not status).

func (*State) UnscrobbleMany 

func (s *State) UnscrobbleMany(ctx context.Context, userID int64, keys []core.RatingKey) error

UnscrobbleMany marks several items unwatched in one transaction (show or season "mark unwatched").

func (*State) WatchSample added in v0.3.3 

func (s *State) WatchSample(ctx context.Context, userID int64, limit int) (sampled, inProgress, watched int, truncated bool, err error)

WatchSample counts, over at most limit of a viewer's watch-state rows, how many are in progress and how many are watched (support diagnostics); truncated reports more rows than limit.

func (*State) WatchedEpisodes added in v0.2.50 

func (s *State) WatchedEpisodes(ctx context.Context, userID int64) ([]core.RatingKey, error)

WatchedEpisodes lists watched episode keys (view_count > 0) for userID.

func (*State) WatchedEpisodesOf added in v0.3.0 

func (s *State) WatchedEpisodesOf(ctx context.Context, userID int64, shows []int64) ([]core.RatingKey, error)

WatchedEpisodesOf lists userID's watched episode keys (view_count > 0) belonging to shows (TMDB ids). Each show's episodes are one key range of the (user_id, rating_key) primary key, so a list page reads only its own shows' history, however long the viewer's history is.

type SyncAspect added in v0.2.57 

type SyncAspect string

SyncAspect is the independent half of a title's account sync: watched state, or a rating. A newer rating must not erase a pending watch, and the reverse. 

const (
	SyncWatch  SyncAspect = "watch"
	SyncRating SyncAspect = "rating"
)

type SyncRow added in v0.2.57 

type SyncRow struct {
	Seq        int64
	UserID     int64
	Key        core.RatingKey
	Aspect     SyncAspect
	Kind       string
	Rating     float64
	At         time.Time
	Bulk       bool
	Dead       bool
	LastErr    string
	Attempts   int
	ConsentGen int64
}

SyncRow is one durable intent. Seq orders intents. Bulk is set when the watch change was one of several keys in the same transaction (a season or show mark).

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.