sessions

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: 48 Imported by: 0

Documentation

Overview

Package sessions owns protocol-independent playback sessions.

Index

Constants

View Source 
const (
	VersionRead = iota
	VersionOffer
	VersionFreeze
	VersionEnd
	VersionTouch
)
View Source 
const MaxPlayQueues = 2000
View Source 
const VersionPreplayTTL = 90 * time.Second

Variables

This section is empty.

Functions

func BindEvents 

func BindEvents(events frontend.Subscriber, protocol frontend.Protocol) frontend.Subscriber

BindEvents applies the originating surface filter to user-data events.

func BindSurface 

func BindSurface(b frontend.Sessions, protocol frontend.Protocol) frontend.Sessions

BindSurface prevents an adapter from submitting another surface's identity.

func Decide 

func Decide(playback core.Playback, transcoder core.Transcoder, r DecisionRequest) (core.Decision, error)

Decide preserves the playback remux override for on-demand sources and stops the source when a required converter is unavailable.

func EncodeQueue 

func EncodeQueue(q *PlayQueue) (string, error)

func Estimate 

func Estimate(ctx context.Context, playback Estimator, viewer *core.Viewer, clientID string, title *core.Title, caps func() core.ClientCaps) (core.MediaInfo, *core.PlaySession)

Estimate returns an existing source's media information or a virtual estimate. It never prepares or resolves a release.

func HLSDecision 

func HLSDecision(dec core.Decision, allowDirect bool, reconsider func() core.Decision, forcedStream func(core.Decision) core.Decision) core.Decision

HLSDecision applies the v39 start-time decision rules. A client may reject direct play after its initial decision; in that case reconsider is called before selecting the segmented output. forcedStream, when supplied, applies adapter output constraints only when a still-direct decision is forced to HLS.

func NewLANLazyStreams 

func NewLANLazyStreams(db *sql.DB, broker *Broker, clientIP func(*http.Request) (netip.Addr, bool), viewer func(context.Context, *http.Request) (*core.Viewer, error), ttl time.Duration) (frontend.LANLazyStreams, error)

NewLANLazyStreams derives authority from host-owned peer and viewer checks. Capabilities are bound to that viewer, canonical peer device and item.

func NewLazyStreams 

func NewLazyStreams(db *sql.DB, b *Broker, protocol frontend.Protocol, grants LazyGrants, ttl time.Duration, remote func(*http.Request) bool) (frontend.LazyStreams, error)

NewLazyStreams creates a surface-bound durable, reusable capability issuer. Capabilities expire after at most one hour of inactivity and are revalidated on every request.

func NewSourceOffers 

func NewSourceOffers(db *sql.DB, b *Broker, protocol frontend.Protocol) (frontend.SourceOffers, error)

func NewVersionHandles 

func NewVersionHandles(db *sql.DB, b *Broker, protocol frontend.Protocol, key []byte) (frontend.VersionHandles, error)

NewVersionHandles creates a surface's version identifiers. An id is a keyed hash of (surface, title, release) under key (a server secret), so it is the same for every viewer and device and reveals nothing about the release. Ids are not credentials: every lookup rechecks viewer and title policy and the release must be one the viewer is offered.

func NotifyTranscodeActivity 

func NotifyTranscodeActivity(transcoder core.Transcoder, id, state string)

NotifyTranscodeActivity forwards timeline state to an active output. It is deliberately a no-op for transcoders and outputs without this capability.

func PlexWireID 

func PlexWireID(key core.RatingKey, slot string) int64

PlexWireID is the positive int32-range id Plex clients already parse. Offer rows use it directly and are not stream bindings.

func PruneVersionIDs 

func PruneVersionIDs(ctx context.Context, db *sql.DB, now time.Time) (int64, error)

PruneVersionIDs deletes ids nobody listed or resolved for versionIDRetention, in batches.

func QueueItemIndex 

func QueueItemIndex(items []QueueItem, id int64) int

func RecordProgress 

func RecordProgress(ctx context.Context, writer ProgressWriter, viewer int64, key core.RatingKey, position, duration time.Duration, state string) error

RecordProgress preserves the existing treatment of extras, which do not enter the viewer's library watch history.

func Recover 

func Recover[R any](ctx context.Context, table *BindingTable[R], playback any, id, inputURL string, at time.Duration) error

Recover maps an encoder identifier to the source that owns it. Playback verifies the input generation before it attempts failover.

func ShuffleQueue 

func ShuffleQueue(p *PlayQueue)

func TranscodeFor 

func TranscodeFor[R any](table *BindingTable[R], transcoder core.Transcoder, playID string) (core.HLSSession, bool)

TranscodeFor finds the running encoder fed by a playback source.

func UnshuffleQueue 

func UnshuffleQueue(p *PlayQueue)

func VariantHash 

func VariantHash(key core.RatingKey, releaseID string) int64

func WithPlexAdmission 

func WithPlexAdmission(ctx context.Context, admit func(context.Context) (PlexAdmission, error)) context.Context

WithPlexAdmission attaches the prepared source for one Plex Plan call.

func WithPlexServe 

func WithPlexServe(ctx context.Context, remote bool, format string) context.Context

WithPlexServe carries the request's remote bit and subtitle format into ResolvePlexStream and OpenPlexSidecar without changing their signatures.

Types

type Binding 

type Binding[R any] struct {
	ID          string // client "session" param
	StreamToken string // independent random capability; never a client-chosen ID
	PlayID      string // core.PlaySession.ID
	PlaybackID  string // explicit X-Plex-Session-IDentifier owning this encoder
	UserID      int64
	ClientID    string
	Key         core.RatingKey
	Requested   R // original Decision/start inputs, before policy refinement
	Decision    core.Decision
	Info        core.MediaInfo
	Subtitle    int // selected Subtitle's position in Info, -1 = none
	Caps        core.ClientCaps
	Opts        core.DecideOptions
	Offset      time.Duration
	Started     bool // StartHLS ran
	UpdatedAt   time.Time
	// A failed Decision for this binding (FailKey, FailErr at FailedAt):
	// the start request that follows it answers with the same error at
	// once instead of resolving again. One-shot and short-lived.
	FailKey  core.RatingKey
	FailErr  error
	FailedAt time.Time
}

type BindingTable 

type BindingTable[R any] struct {
	// contains filtered or unexported fields
}

func NewBindingTable 

func NewBindingTable[R any]() *BindingTable[R]

func (*BindingTable[R]) ByPlayID 

func (t *BindingTable[R]) ByPlayID(PlayID string) []string

ByPlayID returns the transcode session ids fed by a playback session.

func (*BindingTable[R]) Claim 

func (t *BindingTable[R]) Claim(ID string, UserID int64, ClientID string, now time.Time) bool

Claim reserves a client-chosen ID before any playback/transcoder work. A different viewer or device cannot replace, reuse, stop, or probe that binding.

func (*BindingTable[R]) Expire 

func (t *BindingTable[R]) Expire(now time.Time, ttl time.Duration) []string

func (*BindingTable[R]) Fail 

func (t *BindingTable[R]) Fail(ID string, UserID int64, ClientID string, Key core.RatingKey, err error, now time.Time)

Fail records a failed Decision on the binding the Decision claimed. A binding owned by another viewer or device is left alone.

func (*BindingTable[R]) ForClient 

func (t *BindingTable[R]) ForClient(UserID int64, ClientID string, Key core.RatingKey) []string

ForClient lists the transcode sessions of a client for a title.

func (*BindingTable[R]) Get 

func (t *BindingTable[R]) Get(ID string) (*Binding[R], bool)

func (*BindingTable[R]) IDs 

func (t *BindingTable[R]) IDs() []string

IDs lists every binding this table still holds.

func (*BindingTable[R]) Put 

func (t *BindingTable[R]) Put(e *Binding[R])

func (*BindingTable[R]) Remove 

func (t *BindingTable[R]) Remove(ID string)

func (*BindingTable[R]) TakeFailure 

func (t *BindingTable[R]) TakeFailure(ID string, UserID int64, Key core.RatingKey, now time.Time) (error, bool)

TakeFailure returns and clears a recent failed Decision for this viewer's binding and title.

func (*BindingTable[R]) Touch 

func (t *BindingTable[R]) Touch(ID string, now time.Time)

type Broker 

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

Broker coordinates playback admission for every protocol, including Plex.

func NewBroker 

func NewBroker(d BrokerDeps) (*Broker, error)

func UnderlyingBroker 

func UnderlyingBroker(s frontend.Sessions) *Broker

UnderlyingBroker unwraps a surface binding to the broker that stores Plex activities. A non-broker Sessions value has none.

func (*Broker) BindPlexDelivery 

func (b *Broker) BindPlexDelivery(handler http.Handler) error

BindPlexDelivery preserves the original authenticated Plex handler at owner.

func (*Broker) BindPlexStream 

func (b *Broker) BindPlexStream(_ context.Context, access frontend.SessionAccess, binding PlexStreamBinding) (int64, error)

BindPlexStream assigns the visible id. WireID on the binding must be zero. Library metadata calls this before a viewer is known, so it does not authorize. Resolve and Open do.

func (*Broker) Capabilities 

func (b *Broker) Capabilities(ctx context.Context, a frontend.SessionAccess) (frontend.SessionCapabilities, error)

func (*Broker) Close 

func (b *Broker) Close(ctx context.Context)

Close stops local broker sessions and prevents new plans or opens.

func (*Broker) Command 

func (b *Broker) Command(ctx context.Context, actor *core.Viewer, id string, cmd frontend.Command) error

Command targets one authenticated viewer's session. An active, non-delegated owner may target any viewer's session.

func (*Broker) ConfirmPlexAdmission 

func (b *Broker) ConfirmPlexAdmission(ctx context.Context, v *core.Viewer, remote bool) (*core.Viewer, error)

ConfirmPlexAdmission is the fenced policy check a Plex plan uses. planPlexAdmission already calls confirm with the request caps. This is the same check when the caller has only the remote bit.

func (*Broker) CurrentSubtitleSession 

func (b *Broker) CurrentSubtitleSession(ctx context.Context, a frontend.SessionAccess, id string) (PlayingSubtitleSession, error)

CurrentSubtitleSession revalidates viewer/session policy and gets the current selected options. Call it from a worker, never while holding a database fence.

func (*Broker) DeletePlexBinding 

func (b *Broker) DeletePlexBinding(ctx context.Context, id string) error

func (*Broker) DetailPlan 

func (b *Broker) DetailPlan(ctx context.Context, req frontend.PlanRequest) (frontend.PlayPlan, error)

DetailPlan is for a single explicitly requested playable item, never a listing. The flight is scoped to the authenticated request and bounded. An informational result cannot be opened as a session by the caller.

func (*Broker) EnsurePlexActivity 

func (b *Broker) EnsurePlexActivity(id string, key int)

func (*Broker) Events 

func (b *Broker) Events(ctx context.Context, scope frontend.EventScope) (<-chan frontend.Event, error)

func (*Broker) ExpirePlexActivities 

func (b *Broker) ExpirePlexActivities(now time.Time, ttl time.Duration) []PlexActivity

func (*Broker) ForwardPlexGrantMedia 

func (b *Broker) ForwardPlexGrantMedia(w http.ResponseWriter, r *http.Request, id string) bool

func (*Broker) ForwardPlexMedia 

func (b *Broker) ForwardPlexMedia(w http.ResponseWriter, r *http.Request, id string) bool

func (*Broker) HasAuthoredASSSubtitle 

func (b *Broker) HasAuthoredASSSubtitle(ctx context.Context, a frontend.SessionAccess, id string, index int) bool

HasAuthoredASSSubtitle reports whether the subtitle stream at index has a complete, non-truncated authored ASS/SSA script available.

func (*Broker) HasBinding 

func (b *Broker) HasBinding(id string) bool

HasBinding reports whether this broker owns an encoder ID.

func (*Broker) HasPlexActivity 

func (b *Broker) HasPlexActivity(id string) bool

func (*Broker) ListPlexActivities 

func (b *Broker) ListPlexActivities() []PlexActivity

func (*Broker) LoadPlexBinding 

func (b *Broker) LoadPlexBinding(ctx context.Context, id string) ([]byte, error)

func (*Broker) Markers 

func (b *Broker) Markers(ctx context.Context, access frontend.SessionAccess, id string) (frontend.SessionMarkers, error)

Markers never borrows title-only or duration-only marker caches. It requires a complete provider result for the session's exact current source revision.

func (*Broker) MatchDetailSession 

func (b *Broker) MatchDetailSession(ctx context.Context, scope frontend.VersionScope, handle, sessionID string) error

MatchDetailSession authorizes an existing session only when it is pinned to the offered source and the authenticated viewer/device/surface still match.

func (*Broker) NotePlexEncoder 

func (b *Broker) NotePlexEncoder(id string, format frontend.StreamFormat, decision core.Decision, offset time.Duration)

NotePlexEncoder records that the Plex adapter started ffmpeg under the client transcode id, which is also the broker session id.

func (*Broker) Open 

func (b *Broker) Open(ctx context.Context, req frontend.OpenRequest) (frontend.Session, error)

func (*Broker) OpenPlexSidecar 

func (b *Broker) OpenPlexSidecar(ctx context.Context, access frontend.SessionAccess, receipt auth.Acceptance, wireID int64) (io.ReadCloser, core.StreamInfo, error)

OpenPlexSidecar returns an external sidecar only after the acceptance, fenced policy, stream binding, and session agree. Failure returns no reader.

func (*Broker) Plan 

func (b *Broker) Plan(ctx context.Context, req frontend.PlanRequest) (frontend.PlayPlan, error)

func (*Broker) PlexHLSDistributed 

func (b *Broker) PlexHLSDistributed(id string) bool

PlexHLSDistributed reports allocation before protocol code reads any binding or source snapshot. Local sessions retain their original publication path.

func (*Broker) PlexHLSSession 

func (b *Broker) PlexHLSSession(id string) (core.HLSSession, bool)

PlexHLSSession reports ownership independently of current availability. A nil session with true means retained remote responsibility, never local fallback.

func (*Broker) PlexSessionForPlayback 

func (b *Broker) PlexSessionForPlayback(ctx context.Context, a frontend.SessionAccess, item frontend.RatingKey, playbackID, transcodeID string) (string, error)

PlexSessionForPlayback resolves the broker session for a Plex timeline. A client transcode id wins. Otherwise the playback header selects the newest match. An empty header matches only when exactly one session exists.

func (*Broker) PlexSessionForSource 

func (b *Broker) PlexSessionForSource(ctx context.Context, a frontend.SessionAccess, item frontend.RatingKey, playID string) (string, error)

PlexSessionForSource is the broker session that owns a prepared source for this viewer and item. The newest match wins when more than one remains.

func (*Broker) PlexSessionSource 

func (b *Broker) PlexSessionSource(id string) (string, bool)

PlexSessionSource is the playback source owned by a published Plex session.

func (*Broker) Prune 

func (b *Broker) Prune(ctx context.Context) (int64, error)

Prune deletes delivery tokens older than the newest deliveryTokenKeep, except tokens a delivery authority row still names, and reports the rows removed. Ordering never depends on deleted rows: ids come from an AUTOINCREMENT key (a sequence on PostgreSQL) that never reuses a value, and the newest row is always kept.

func (*Broker) RecordAuthorizedSubtitleIntent 

func (b *Broker) RecordAuthorizedSubtitleIntent(ctx context.Context, viewer int64, key core.RatingKey, manual bool) error

RecordAuthorizedSubtitleIntent is for an adapter that has already authorized the viewer/title and validated the selector. This is deliberately not an HTTP endpoint. Missing DB keeps legacy fixtures working; automation fails closed without its store. Reset-default passes manual=false; explicit off passes true.

func (*Broker) Recover 

func (b *Broker) Recover(ctx context.Context, id, url string, at time.Duration) error

func (*Broker) RemovePlexActivity 

func (b *Broker) RemovePlexActivity(id string) (PlexActivity, bool)

func (*Broker) Report 

func (b *Broker) Report(ctx context.Context, a frontend.SessionAccess, id string, t frontend.Timeline) error

func (*Broker) ReportItemProgress 

func (b *Broker) ReportItemProgress(ctx context.Context, a frontend.SessionAccess, item core.RatingKey, t frontend.Timeline) error

ReportItemProgress records a Jellyfin or Emby timeline that names an item instead of a broker session.

func (*Broker) ReportProgress 

func (b *Broker) ReportProgress(ctx context.Context, a frontend.SessionAccess, item core.RatingKey, t frontend.Timeline) error

ReportProgress records a Plex timeline that has no playback source. It does not prepare or stop one.

func (*Broker) ReportSample 

func (b *Broker) ReportSample(ctx context.Context, a frontend.SessionAccess, id string, sample frontend.ProgressSample) (frontend.ReportOutcome, error)

ReportSample uses the same policy and ownership gate as ordinary reporting, but requires a synchronous receipt writer instead of the availability outbox.

func (*Broker) ResolveDetailSource 

func (b *Broker) ResolveDetailSource(ctx context.Context, scope frontend.VersionScope, handle string) (frontend.DetailSourceTarget, error)

func (*Broker) ResolvePlexStream 

func (b *Broker) ResolvePlexStream(ctx context.Context, access frontend.SessionAccess, wireID int64) (PlexStreamBinding, error)

ResolvePlexStream returns a bound stream. An unknown id is ErrNotFound. A viewer who may not play is ErrUnauthorized, not a miss.

func (*Broker) SavePlexBinding 

func (b *Broker) SavePlexBinding(ctx context.Context, id string, raw []byte) error

func (*Broker) ServeASSSubtitle 

func (b *Broker) ServeASSSubtitle(ctx context.Context, a frontend.SessionAccess, id string, index int, syncOffsetMS int, requireAuthored bool) (frontend.SubtitleASSArtifact, error)

ServeASSSubtitle extracts or converts the subtitle stream at index as ASS/SSA. When requireAuthored is true, any missing or truncated authored ASS track fails closed with frontend.ErrSubtitleAuthoringUnavailable instead of falling back to simplified WebVTT conversion.

func (*Broker) ServeMedia 

func (b *Broker) ServeMedia(w http.ResponseWriter, r *http.Request)

ServeMedia validates a session-scoped capability on every request.

func (*Broker) ServeMediaAccess 

func (b *Broker) ServeMediaAccess(w http.ResponseWriter, r *http.Request, a frontend.SessionAccess)

ServeMediaAccess additionally binds the grant to the authenticated device.

func (*Broker) ServeMediaFor 

func (b *Broker) ServeMediaFor(protocol frontend.Protocol, w http.ResponseWriter, r *http.Request)

ServeMediaFor binds a mounted handler to its protocol surface.

func (*Broker) ServeSessionMedia 

func (b *Broker) ServeSessionMedia(w http.ResponseWriter, r *http.Request, a frontend.SessionAccess, sessionID, resource string)

ServeSessionMedia serves one session resource for a device that already proved it owns the session. The grant stays on the internal path; callers must not copy it into a client URL.

func (*Broker) Sessions 

func (b *Broker) Sessions(ctx context.Context, a frontend.SessionAccess) ([]frontend.Session, error)

func (*Broker) StartPlexHLS 

func (b *Broker) StartPlexHLS(ctx context.Context, viewer *core.User, playID, publicID string, q core.HLSRequest) (core.HLSSession, error)

StartPlexHLS is the shared universal encoder seam after the Plex adapter has performed its existing acceptance, policy, selection and grant checks. Errors never authorize caller fallback or a second encoder.

func (*Broker) StartStream 

func (b *Broker) StartStream(ctx context.Context, a frontend.SessionAccess, id string, req frontend.StreamRequest) (string, error)

func (*Broker) Stop 

func (b *Broker) Stop(ctx context.Context, a frontend.SessionAccess, id string) error

func (*Broker) StopPlayback 

func (b *Broker) StopPlayback(ctx context.Context, playID string) bool

StopPlayback stops every local broker session fed by a source ID.

func (*Broker) StopPlexHLS 

func (b *Broker) StopPlexHLS(id string) bool

StopPlexHLS handles local and remote encoder retirement through the broker. Remote ownership includes uncertain start/stop phases; true reports ownership, not proof that an asynchronous remote retirement has completed.

func (*Broker) StopProtocol 

func (b *Broker) StopProtocol(ctx context.Context, protocol frontend.Protocol) error

StopProtocol releases every local session for one mounted surface.

func (*Broker) SubtitleFontAttachment 

func (b *Broker) SubtitleFontAttachment(ctx context.Context, a frontend.SessionAccess, id string, index int, fontIndex int) (frontend.SubtitleFontAttachment, error)

SubtitleFontAttachment returns a single binary Matroska font attachment by its attachment index for an embedded ASS/SSA subtitle stream.

func (*Broker) SubtitleFontBundle 

func (b *Broker) SubtitleFontBundle(ctx context.Context, a frontend.SessionAccess, id string, index int) ([]frontend.SubtitleFontBundleItem, error)

SubtitleFontBundle returns the base64-encoded Matroska font attachment bundle for an embedded ASS/SSA subtitle stream.

func (*Broker) SubtitleURL 

func (b *Broker) SubtitleURL(ctx context.Context, a frontend.SessionAccess, id string, index int, format string) (string, error)

SubtitleURL authorizes a subtitle by the session's probed stream index. The returned URL contains only the scoped stream capability.

func (*Broker) TranscodeFor 

func (b *Broker) TranscodeFor(_ context.Context, id string) (core.HLSSession, bool)

func (*Broker) TrickplayOffer 

func (b *Broker) TrickplayOffer(ctx context.Context, a frontend.SessionAccess, item core.RatingKey) (frontend.TrickplayOffer, bool)

TrickplayOffer revalidates the viewer, item, version, and source generation before advertising tiles. A false offer leaves the Jellyfin map empty.

func (*Broker) TrickplaySheet 

func (b *Broker) TrickplaySheet(ctx context.Context, a frontend.SessionAccess, item core.RatingKey, index int) ([]byte, error)

TrickplaySheet composes one JPEG grid from the borrowed provider's frames.

func (*Broker) UpdateCapabilities 

func (b *Broker) UpdateCapabilities(ctx context.Context, a frontend.SessionAccess, c frontend.SessionCapabilities) error

UpdateCapabilities stores declarations for the authenticated device only. URL metadata is never dereferenced by the broker.

func (*Broker) UpsertPlexActivity 

func (b *Broker) UpsertPlexActivity(id string, fn func(*PlexActivity)) PlexActivity

func (*Broker) WarmDetail 

func (b *Broker) WarmDetail(ctx context.Context, req frontend.PlanRequest) bool

WarmDetail implements frontend.DetailWarmer. A device that has not yet declared its playback caps on this surface is not warmed, and neither is one whose network locality changed since: remote and LAN plays select differently, so a warm pick could be lower than Play's.

type BrokerDeps 

type BrokerDeps struct {
	// OnActualPlaying runs in a bounded worker after authenticated playing telemetry.
	OnActualPlaying func(context.Context, *Broker, PlayingSubtitleSession)
	// DistributedHLS is opt-in; unsupported plans stay local before reservation.
	DistributedHLS DistributedHLS
	Coordination   *Coordination
	// MaxPendingStarts bounds active or waiting neutral stream starts.
	// Zero selects the default; the broker rejects values above 4096.
	MaxPendingStarts int
	DB               *sql.DB
	Cluster          clusterbus.Node
	Live             LiveBackend
	Variants         func(context.Context, *core.Viewer, *core.Title, core.ClientCaps) ([]core.ReleaseVariant, error)
	SelectTracks     func(context.Context, *core.Viewer, *core.Title, core.MediaInfo, core.DecideOptions, bool, bool) (core.DecideOptions, error)
	// SelectScopedTracks receives the authenticated device as well. If present,
	// it replaces SelectTracks and must preserve the unscoped behavior.
	SelectScopedTracks func(context.Context, *core.Viewer, core.Device, *core.Title, core.MediaInfo, core.DecideOptions, bool, bool) (core.DecideOptions, error)
	Lookup             func(context.Context, *core.Viewer, core.Device, core.RatingKey) (*core.Title, error)
	// MediaLookup checks current title policy using local facts only; it must
	// never resolve a release or fetch remote metadata on a segment request.
	// If nil, Lookup must satisfy the same requirement.
	MediaLookup   func(context.Context, *core.Viewer, core.Device, core.RatingKey) (*core.Title, error)
	Playback      core.Playback
	Transcoder    core.Transcoder
	Progress      ProgressWriter
	SourceHistory SourceHistoryWriter
	ActiveViewer  func(context.Context, int64) (*core.Viewer, error)
	CanPlay       func(*core.Viewer, bool, time.Time) bool
	// FreshPlay re-reads fenced membership and safety policy. A returned viewer
	// replaces the cached one for the rest of the call. ErrPolicyUnconfirmed
	// keeps the cached CanPlay decision for an unrestricted viewer.
	FreshPlay func(context.Context, *core.Viewer, bool, time.Time) (*core.Viewer, error)
	// ReceiptCurrent rechecks the acceptance Plex stored for this request.
	ReceiptCurrent     func(context.Context, auth.Acceptance) error
	BitrateLimit       func(*core.Viewer) int
	RemoteBitrateLimit func() int
	OnTimeline         func(*core.Viewer, core.RatingKey, time.Duration, time.Duration, string)
	Events             *EventBus
	OnEventError       func(error)
	// OnProfileDiagnostic is an optional fixture hook. Values are closed enums
	// only; the callback receives no media, identity or client-supplied data.
	OnProfileDiagnostic func(ProfileDiagnostic)
	// SourceMarkers returns a complete marker set for the exact source revision.
	SourceMarkers    func(context.Context, *core.Viewer, core.RatingKey, string, core.MediaInfo) ([]frontend.Marker, bool, error)
	OpenSubtitle     func(string, string) (io.ReadCloser, string, error)
	OpenSubtitleFrom func(string, string, string) (io.ReadCloser, string, error)
	ExtractSubtitle  func(context.Context, string, int, string, io.Writer) error
	// Trickplay is borrowed from the runtime owner (internal/trickplay/owner.go).
	// It is the same provider Plex uses. The broker releases its own leases;
	// only the runtime owner may Close the provider.
	Trickplay core.TrickplayProvider
	Now       func() time.Time
}

BrokerDeps are the explicit dependencies shared by all protocol surfaces.

type Coordination 

type Coordination struct {
	Owners *PlaybackOwners
	Gates  *Lifecycle
}

Coordination holds the v39 playback owner history and lifecycle gates shared by every protocol adapter in one Ether process.

func NewCoordination 

func NewCoordination() *Coordination

type DecisionArms 

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

DecisionArms owns per-player decision authorization and durable publication. The host cancels ctx and joins wg before clearing its durable rows on shutdown.

func NewDecisionArms 

func NewDecisionArms(ctx context.Context, db *sql.DB, now func() time.Time, node string, log *slog.Logger, wg *sync.WaitGroup) *DecisionArms

func (*DecisionArms) Arm 

func (s *DecisionArms) Arm(pr DecisionOwner, key core.RatingKey)

The public player session owns its decision state; the encoder's session= value is an independent delivery handle and cannot authorize another player.

func (*DecisionArms) Armed 

func (s *DecisionArms) Armed(pr DecisionOwner, key core.RatingKey) bool

func (*DecisionArms) ClearOwn 

func (s *DecisionArms) ClearOwn()

func (*DecisionArms) Disarm 

func (s *DecisionArms) Disarm(pr DecisionOwner, key core.RatingKey)

Stopping one item clears its start/bandwidth authorization while retaining the session's history of having made a decision.

func (*DecisionArms) Expire 

func (s *DecisionArms) Expire(now time.Time)

func (*DecisionArms) Refresh 

func (s *DecisionArms) Refresh(now time.Time)

The session janitor refreshes live arms and removes expired durable entries. Timeline and read handlers update only memory; SQLite writes stay here.

type DecisionOwner 

type DecisionOwner struct {
	ID                  string
	UserID              int64
	ClientID, SessionID string
}

DecisionOwner identifies the authenticated public player, independently of its encoder delivery handle.

type DecisionRequest 

type DecisionRequest struct {
	Session              *core.PlaySession
	Caps                 core.ClientCaps
	Options              core.DecideOptions
	PreferRemux          bool
	Protocol             string
	RemoteMaxBitrateKbps int
	// PreserveSourceOnFailure lets a broker avoid stopping a reused source
	// when an attempted decision cannot be served without a transcoder.
	PreserveSourceOnFailure bool
}

DecisionRequest contains the source and client inputs for a playback decision. PreferRemux is the adapter's signal that source changes need refreshed container headers; the source and requested direct-stream mode still determine whether the override applies.

type DistributedHLS 

type DistributedHLS interface {
	Supports(core.HLSRequest) bool
	Allocated(string) bool
	StartDistributedHLS(context.Context, int64, string, string, core.HLSRequest) (core.HLSSession, error)
	Session(string) (core.HLSSession, bool)
	Stop(string)
}

DistributedHLS preserves broker authentication and stable public URLs. The implementation must retain uncertain starts and fence all replacements.

type DownloadItem 

type DownloadItem struct {
	ID      int64
	QueueID int64
	Key     core.RatingKey
	Title   string // display Title, for the served filename

	Caps core.ClientCaps // as requested at add time (Download set at Run)
	User *core.User

	Status   string
	Error    string
	Progress float64 // 0..1
	Started  bool
	Cancel   context.CancelFunc
	// Run numbers each start of the Item, so a restarted job gets its own
	// playback session instead of sharing (and later stopping) the
	// previous Run's.
	Run int64

	Path       string
	Size       int64
	ReadyAt    time.Time
	VideoCodec string
	AudioCodec string
	Duration   time.Duration
	Bandwidth  int // requested maxVideoBitrate, kbps
	Burn       bool
}

type DownloadJob 

type DownloadJob struct {
	Playback core.Playback
	Remuxer  core.DownloadRemuxer
}

DownloadJob owns one offline source lease, including bounded quality retries and retirement on every exit. Queue IDs, status rendering and track intent stay with the caller. A run must be unique within that caller's queue store.

func (DownloadJob) Run 

func (j DownloadJob) Run(ctx context.Context, q DownloadRequest) (string, int64, error)

type DownloadPreparationError 

type DownloadPreparationError struct{ Err error }

func (*DownloadPreparationError) Error 

func (e *DownloadPreparationError) Error() string

func (*DownloadPreparationError) Unwrap 

func (e *DownloadPreparationError) Unwrap() error

type DownloadQueue 

type DownloadQueue struct {
	ID       int64
	UserID   int64
	ClientID string
	Items    []*DownloadItem
}

func (*DownloadQueue) Item 

func (q *DownloadQueue) Item(ID int64) *DownloadItem

type DownloadQueuePersistence 

type DownloadQueuePersistence struct {
	Queue     *DownloadQueueStore
	DB        *sql.DB
	Artifacts core.DownloadRemuxer
	Log       *slog.Logger
	Expiry    func() time.Duration
	StorageID func() string
	Users     func(context.Context) ([]core.User, error)
}

DownloadQueuePersistence owns coalesced snapshots and retained-file recovery. Callbacks supply current host configuration without exposing an adapter.

func (DownloadQueuePersistence) Load 

func (p DownloadQueuePersistence) Load(ctx context.Context)

func (DownloadQueuePersistence) Retain 

func (p DownloadQueuePersistence) Retain(path string, ready time.Time)

func (DownloadQueuePersistence) Run 

func (p DownloadQueuePersistence) Run(ctx context.Context)

func (DownloadQueuePersistence) Save 

func (p DownloadQueuePersistence) Save() error

type DownloadQueueStore 

type DownloadQueueStore struct {
	Mu        sync.Mutex
	Queues    map[int64]*DownloadQueue
	ByClient  map[string]int64 // "UserID\x00clientID" -> queue ID
	NextQueue int64
	NextItem  int64
	NextRun   int64
	Wake      chan struct{}
}

func NewDownloadQueueStore 

func NewDownloadQueueStore() *DownloadQueueStore

func (*DownloadQueueStore) Add 

func (st *DownloadQueueStore) Add(q *DownloadQueue, u *core.User, t core.Title, caps core.ClientCaps, burn bool) *DownloadItem

func (*DownloadQueueStore) CancelItem 

func (st *DownloadQueueStore) CancelItem(it *DownloadItem) string

func (*DownloadQueueStore) Complete 

func (st *DownloadQueueStore) Complete(ctx context.Context, it *DownloadItem, path string, size int64, now time.Time) bool

func (*DownloadQueueStore) Expire 

func (st *DownloadQueueStore) Expire(now time.Time, expiry time.Duration) []string

Expire clears retained artifacts under the queue lock; callers release files after unlocking, keeping I/O outside the queue's state transition.

func (*DownloadQueueStore) Fail 

func (st *DownloadQueueStore) Fail(ctx context.Context, it *DownloadItem, message string)

func (*DownloadQueueStore) Prepared 

func (st *DownloadQueueStore) Prepared(ctx context.Context, it *DownloadItem, info core.MediaInfo) bool

func (*DownloadQueueStore) Progress 

func (st *DownloadQueueStore) Progress(ctx context.Context, it *DownloadItem, fraction float64)

func (*DownloadQueueStore) QueueByID 

func (st *DownloadQueueStore) QueueByID(ID, UserID int64, ClientID string) *DownloadQueue

QueueByID returns the queue when it belongs to this User and client.

func (*DownloadQueueStore) QueueFor 

func (st *DownloadQueueStore) QueueFor(UserID int64, ClientID string, create bool) *DownloadQueue

func (*DownloadQueueStore) Remove 

func (st *DownloadQueueStore) Remove(q *DownloadQueue, ids []int64) []*DownloadItem

func (*DownloadQueueStore) Restart 

func (st *DownloadQueueStore) Restart(q *DownloadQueue, ids []int64, release func(string))

func (*DownloadQueueStore) StartReady 

func (st *DownloadQueueStore) StartReady(ctx context.Context, perUser int, allowed func(*core.User) bool) []DownloadStart

StartReady atomically reserves each eligible run before any worker starts.

func (*DownloadQueueStore) StopForRestart 

func (st *DownloadQueueStore) StopForRestart()

type DownloadRequest 

type DownloadRequest struct {
	Viewer    *core.User
	Item      core.RatingKey
	Caps      core.ClientCaps
	Run       int64
	NoRelease func(error) bool
	Prepared  func(*core.PlaySession) (core.DecideOptions, error)
	Progress  func(float64)
}

type DownloadStart 

type DownloadStart struct {
	Item    *DownloadItem
	Context context.Context
	Run     int64
}

type Estimator 

type Estimator interface {
	Estimate(context.Context, *core.Title, core.ClientCaps) core.MediaInfo
	SessionFor(int64, string, core.RatingKey) (*core.PlaySession, bool)
}

Estimator is the non-resolving playback surface for detail pages.

type EventBus 

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

EventBus distributes viewer-scoped events locally and across cluster nodes. ActiveViewer must return the current active viewer, not a cached credential.

func NewEventBus 

func NewEventBus(bus clusterbus.Bus, activeViewer func(context.Context, int64) (*core.Viewer, error)) *EventBus

NewEventBus subscribes to cluster events when bus is non-nil. Call Close when the application shuts down.

func (*EventBus) Close 

func (b *EventBus) Close()

func (*EventBus) Publish 

func (b *EventBus) Publish(ctx context.Context, event frontend.Event) error

Publish delivers locally and sends an origin-tagged event to other nodes.

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe(ctx context.Context, scope frontend.EventScope) (<-chan frontend.Event, error)

Subscribe ends on cancellation, revoked access, bus shutdown or a slow subscriber overflowing its buffer. A closed channel tells clients to resync.

type LazyGrants 

type LazyGrants interface {
	GrantReference(context.Context, string, string) (string, *core.Viewer, error)
	ValidateGrantReference(context.Context, string, string, int64) error
}

LazyGrants authenticates an issuing surface credential and rechecks the non-secret grant reference without extending its lifetime.

type Lifecycle 

type Lifecycle = playbackLifecycle

Lifecycle owns the v39 owner and encoder gates.

func NewLifecycle 

func NewLifecycle() *Lifecycle

type LiveBackend 

type LiveBackend interface {
	Allowed(context.Context, *core.Viewer, string) error
	Tune(context.Context, *core.Viewer, string, core.ClientCaps, string) (string, error)
	Serve(http.ResponseWriter, *http.Request, *core.Viewer, string, string, string)
	Stop(context.Context, *core.Viewer, string) error
}

LiveBackend adapts the existing channel playout service. Serve receives a grant-validated request and a path relative to the scoped live stream.

type LiveMediaBackend 

type LiveMediaBackend interface {
	LiveBackend
	TuneMedia(context.Context, *core.Viewer, string, core.ClientCaps, string) (string, *core.MediaInfo, error)
}

LiveMediaBackend optionally returns the live session's probed or output MediaInfo.

type PlayQueue 

type PlayQueue struct {
	ID               int64
	UserID           int64
	ClientID         string
	Items            []QueueItem
	Original         []QueueItem // order before shuffle
	Selected         int         // index into items
	Shuffled         bool
	SourceURI        string
	PlaylistID       core.RatingKey
	PlaylistTitle    string
	Version          int
	LastAdded        int64
	UpdatedAt        time.Time
	SuppressItem     int64
	SuppressRows     bool // mutation response only
	SuppressMetadata bool
	ForceOffsetZero  bool
	LiveSession      string // live queues resolve through LiveTV, never the catalogue

	Rev      int64     // clustered: stored row revision
	LoadedAt time.Time // clustered: when the cached copy was read or written
}

func DecodeQueue 

func DecodeQueue(id, userID, rev, updated int64, data string) (*PlayQueue, error)

func (*PlayQueue) ReanchorAdded 

func (q *PlayQueue) ReanchorAdded()

writePlayQueue renders a queue window around the selected item.

func (*PlayQueue) Snapshot 

func (q *PlayQueue) Snapshot() PlayQueue

type PlayQueuePersistence 

type PlayQueuePersistence struct {
	DB           *sql.DB
	Counter      func(context.Context, string, int) (int64, error)
	Invalidate   func(int64)
	Warn         func(string, error)
	ExpireResult func(error)
}

PlayQueuePersistence connects shared queue transactions to node invalidation and outage reporting. Configure it before serving requests.

type PlayQueues 

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

func NewPlayQueues 

func NewPlayQueues() *PlayQueues

func (*PlayQueues) AddItems 

func (s *PlayQueues) AddItems(q *PlayQueue, keys []core.RatingKey, pre []QueueItem, next bool)

AddItems appends items for keys (or inserts them after the selected item). pre holds IDs preallocated on a clustered node; otherwise the caller holds s.mu (it runs inside With) and IDs are numbered here.

func (*PlayQueues) Configure 

func (s *PlayQueues) Configure(now func() time.Time, persistence *PlayQueuePersistence)

func (*PlayQueues) CreateWith 

func (s *PlayQueues) CreateWith(userID int64, clientID string, keys []core.RatingKey, now time.Time, fn func(q *PlayQueue)) (PlayQueue, error)

CreateWith creates a queue of keys for the user's client, applies fn to it and returns a snapshot.

func (*PlayQueues) DropCache 

func (s *PlayQueues) DropCache()

DropCache forgets every cached queue (bus resync).

func (*PlayQueues) Expire 

func (s *PlayQueues) Expire(now time.Time, ttl time.Duration)

func (*PlayQueues) FollowingPlayable 

func (s *PlayQueues) FollowingPlayable(userID int64, key core.RatingKey, playable func(core.RatingKey) bool) core.RatingKey

FollowingPlayable returns the first playable item after the given extra in the viewer's cached queues. It does not perform catalogue or database reads.

func (*PlayQueues) Forget 

func (s *PlayQueues) Forget(id int64)

Forget drops a cached queue another node changed.

func (*PlayQueues) Get 

func (s *PlayQueues) Get(id, userID int64) (PlayQueue, bool)

func (*PlayQueues) Len 

func (s *PlayQueues) Len() int

func (*PlayQueues) Persistent 

func (s *PlayQueues) Persistent() bool

func (*PlayQueues) Preallocate 

func (s *PlayQueues) Preallocate(keys []core.RatingKey) ([]QueueItem, error)

Preallocate reserves item IDs for keys before an edit on a clustered node (nil on a standalone server, which numbers items inside the edit).

func (*PlayQueues) Reshuffle 

func (s *PlayQueues) Reshuffle(q *PlayQueue, shuffled bool)

Controls preserve the playing identity while every other slot receives a newly allocated identity. This also applies when the requested state repeats.

func (*PlayQueues) SelectItem 

func (s *PlayQueues) SelectItem(userID, itemID int64)

SelectItem marks the item a client reports playing (timeline).

func (*PlayQueues) Take 

func (s *PlayQueues) Take(id, userID int64, clientID string, now time.Time) (PlayQueue, bool)

func (*PlayQueues) TimelineQueue 

func (s *PlayQueues) TimelineQueue(user int64, client string, key core.RatingKey, raw string, present bool) core.RatingKey

TimelineQueue resolves an explicit queue item before watch persistence. An omitted item selector follows the owning client's first matching queue row. Cluster fallback is capped at the same 50 recent queues as selectShared.

func (*PlayQueues) With 

func (s *PlayQueues) With(id, userID int64, now time.Time, fn func(q *PlayQueue)) (PlayQueue, bool)

With runs fn on the queue and returns a snapshot (fn nil: read only).

type PlaybackOwners 

type PlaybackOwners = playbackOwners

PlaybackOwners retains the v39 playback handoff and supersession rules.

func NewPlaybackOwners 

func NewPlaybackOwners() *PlaybackOwners

type PlayingSubtitleSession 

type PlayingSubtitleSession struct {
	Access                       frontend.SessionAccess
	ID, SourceID, SourceRevision string
	Item                         core.RatingKey
	Options                      core.DecideOptions
	ManualSubtitle               bool
}

PlayingSubtitleSession identifies an authenticated playing session. It carries no media capability, provider secret or mutable source handle.

type PlexActivity 

type PlexActivity struct {
	ID                                                     string
	DecisionSeen                                           bool
	PlayingSeen                                            bool
	SessionKey                                             int
	User                                                   core.User
	ClientID, Product, Version, Platform, PlatformVersion  string
	Device, DeviceName, Model, Vendor, SessionID, RemoteIP string
	ProfileName                                            string
	Remote                                                 bool
	Key                                                    core.RatingKey
	State                                                  string
	Offset, Duration                                       time.Duration
	PlayQueueItemID                                        int64
	TranscodeID                                            string
	StartedAt, UpdatedAt                                   time.Time
}

PlexActivity is one Plex client's playback as the session list shows it. The broker is the only store. Cluster rows for these activities use protocol plex.

type PlexAdmission 

type PlexAdmission struct {
	Source               *core.PlaySession
	Media                core.MediaInfo
	Caps                 core.ClientCaps
	Options              core.DecideOptions
	PreferRemux          bool
	Protocol             string
	RemoteMaxBitrateKbps int
}

PlexAdmission is the source and the version and track selections for one Plex plan. Plan makes the direct, remux, or transcode decision from them. Open publishes that decision. The adapter renders it and does not decide again.

type PlexPartSource 

type PlexPartSource struct {
	Session               *core.PlaySession
	Matches               bool
	PreviousID            string
	AnonymousContinuation bool
}

type PlexSources 

type PlexSources struct {
	Playback core.Playback
	Log      *slog.Logger
}

PlexSources is the trusted source-lifetime boundary used after the adapter's authentication, title policy and issued-part checks. It does not publish a second broker session for a HEAD, download or anonymous continuation.

func (PlexSources) Part 

func (s PlexSources) Part(viewer int64, client string, key core.RatingKey, release string) PlexPartSource

Part retains an existing source where possible, including the unique anonymous part continuation. A variant mismatch records the previous source so admission can recognize a fallback to that same live session.

func (PlexSources) Prepare 

func (s PlexSources) Prepare(ctx context.Context, viewer *core.User, key core.RatingKey, caps core.ClientCaps, release string, wait time.Duration) (*core.PlaySession, error)

Prepare bounds acquisition while leaving shared source flights and normal admission/supersession with Playback. A successful wait is followed by normal admission; an error never falls through to a second acquisition attempt.

func (PlexSources) ServePart 

func (s PlexSources) ServePart(w http.ResponseWriter, r *http.Request, sourceID string)

ServePart preserves the Plex range grammar and request cancellation. The admitted source ID is retained; delivery must not open or supersede a session.

type PlexStreamAPI 

type PlexStreamAPI interface {
	BindPlexStream(context.Context, frontend.SessionAccess, PlexStreamBinding) (int64, error)
	ResolvePlexStream(context.Context, frontend.SessionAccess, int64) (PlexStreamBinding, error)
	OpenPlexSidecar(context.Context, frontend.SessionAccess, auth.Acceptance, int64) (io.ReadCloser, core.StreamInfo, error)
}

PlexStreamAPI is the visible-id and sidecar surface Plex calls.

func BindPlexStreams 

func BindPlexStreams(b *Broker) PlexStreamAPI

BindPlexStreams is BindSurface for the Plex stream methods. The returned value rejects any access whose protocol is not plex.

type PlexStreamBinding 

type PlexStreamBinding struct {
	WireID    int64
	SessionID string
	Key       core.RatingKey
	Index     int
	Info      core.StreamInfo
}

PlexStreamBinding is one Plex-visible stream id. WireID is zero on bind; BindPlexStream assigns it. SessionID is empty for library metadata and is the broker session id after Open.

type PlexStreamTable 

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

PlexStreamTable is the visible-id map. Plex offer ids are not stored here.

func NewPlexStreamTable 

func NewPlexStreamTable() *PlexStreamTable

NewPlexStreamTable is the visible-id map used by the broker and by Plex fixtures that have no broker mounted.

func (*PlexStreamTable) Bind 

func (t *PlexStreamTable) Bind(binding PlexStreamBinding) (int64, error)

Bind assigns or reuses the wire id for this slot. It does not authorize.

func (*PlexStreamTable) Lookup 

func (t *PlexStreamTable) Lookup(id int64) (PlexStreamBinding, bool)

Lookup reports the binding. False means the id was never bound.

type ProfileDiagnostic 

type ProfileDiagnostic string

ProfileDiagnostic is a redacted reason emitted only to an explicitly installed diagnostic hook. It never includes client profile values. 

const (
	ProfileRequiredConditionUnknown ProfileDiagnostic = "required_condition_unknown"
	ProfileDirectRejected           ProfileDiagnostic = "direct_rejected"
	ProfileDeliveryRejected         ProfileDiagnostic = "delivery_rejected"
	ProfileFallbackTranscode        ProfileDiagnostic = "fallback_transcode"
	ProfileFallbackRemux            ProfileDiagnostic = "fallback_remux"
	ProfileNoDelivery               ProfileDiagnostic = "no_delivery"
)

type ProgressWriter 

type ProgressWriter interface {
	Progress(context.Context, int64, core.RatingKey, time.Duration, time.Duration, string) error
}

ProgressWriter persists the timeline after the caller has checked ownership.

type QueueItem 

type QueueItem struct {
	ID    int64
	Key   core.RatingKey
	Added bool // explicit Up Next addition, distinct from original queue members
}

type Selections 

type Selections[T any] struct {
	// contains filtered or unexported fields
}

Selections stores per-viewer track choices. The adapter owns the value shape and translates wire stream identifiers before updating a choice.

func NewSelections 

func NewSelections[T any]() *Selections[T]

func (*Selections[T]) Get 

func (s *Selections[T]) Get(viewer, item int64) T

func (*Selections[T]) Update 

func (s *Selections[T]) Update(viewer, item int64, fn func(*T))

type SourceHistoryWriter 

type SourceHistoryWriter interface {
	BeginSourceSession(context.Context) (int64, error)
	RecordLastSource(context.Context, int64, core.RatingKey, int64, int64, core.PlaybackSourceObservation) error
}

SourceHistoryWriter stores facts from accepted samples, ordered first by session epoch and then by the sample number assigned before persistence. Implementations must honor cancellation, including during transaction commit. The production state writer uses context-aware SQLite/PostgreSQL operations. Go cannot forcibly stop an implementation or driver that ignores context.

type StartGuard 

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

StartGuard holds bounded per-session watches and per-title fallbacks.

func (*StartGuard) BaseOnly 

func (g *StartGuard) BaseOnly(k StartHintKey, now time.Time) bool

BaseOnly reports a remembered, unexpired hint for k.

func (*StartGuard) Expire 

func (g *StartGuard) Expire(now time.Time)

func (*StartGuard) Forget 

func (g *StartGuard) Forget(tid string)

func (*StartGuard) Observe 

func (g *StartGuard) Observe(tid, state string, offset, ahead time.Duration, now time.Time) bool

Observe records one timeline for transcode session tid. It returns true exactly once, when the session's start has stalled. ahead is the media the client has requested past its reported position.

func (*StartGuard) Remember 

func (g *StartGuard) Remember(k StartHintKey, now time.Time)

type StartHintKey 

type StartHintKey struct {
	UserID   int64
	ClientID string
	Key      core.RatingKey
}

StartHintKey names one remembered fallback. A title Key is the Dolby Vision HDR10-base hint for that title; a zero Key is client-wide: this viewer's client stalled on surround audio copied into segmented output.

type StartObservation 

type StartObservation interface {
	core.PlaybackPosition
	Decision() core.Decision
	Info() core.MediaInfo
}

StartObservation is the evidence used by stalled-start detection.

func ObserveStart 

func ObserveStart(ctx context.Context, source core.HLSSession, id string) (StartObservation, bool)

ObserveStart accepts local evidence directly; remote evidence must be fresh and belong to the requested execution. A failed read never uses cached data.

type SubtitleIntent 

type SubtitleIntent struct {
	Revision int64
	Manual   bool
}

SubtitleIntent is shared across a viewer's devices, preserving Plex's existing per-title manual selection scope. Device-specific choices may conservatively suppress automation on another device; they never grant extra download quota.

type SubtitleIntents 

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

SubtitleIntents fences manual changes against publication. The broker and adapters must authorize the title and validate selectors BEFORE recording an intent. The fence does not replace that authorization or store track choices.

func NewSubtitleIntents 

func NewSubtitleIntents(db *sql.DB) *SubtitleIntents

func (*SubtitleIntents) Publish 

func (s *SubtitleIntents) Publish(ctx context.Context, viewer int64, key core.RatingKey, expected SubtitleIntent, fn func() error) error

Publish runs a short in-memory source attachment while holding the database row lock. No provider/network call or other database query belongs in fn. Manual writers wait until publication finishes; older snapshots fail closed.

func (*SubtitleIntents) Record 

func (s *SubtitleIntents) Record(ctx context.Context, viewer int64, key core.RatingKey, manual bool) error

func (*SubtitleIntents) Snapshot 

func (s *SubtitleIntents) Snapshot(ctx context.Context, viewer int64, key core.RatingKey) (SubtitleIntent, error)

type VariantIDs 

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

func NewVariantIDs 

func NewVariantIDs(db *sql.DB, now func() time.Time) *VariantIDs

func (*VariantIDs) Flush 

func (r *VariantIDs) Flush(ctx context.Context) error

flush writes queued touches. A row another node's prune removed while this node still serves it is restored, so the table never loses an id a registry hands out. Failed writes are queued again.

func (*VariantIDs) IDFor 

func (r *VariantIDs) IDFor(key core.RatingKey, releaseID string) int64

idFor returns the stable Media/Part id of a non-Auto variant.

func (*VariantIDs) Lookup 

func (r *VariantIDs) Lookup(id int64) (VariantRef, bool)

func (*VariantIDs) Prune 

func (r *VariantIDs) Prune(ctx context.Context, liveIDs []int64, liveRefs []VariantRef) (int64, error)

prune deletes ids unseen for variantRetention, except those in live (ids still held by a play identity: a live or admitting session, a pinned version list), which are touched first. It then drops pruned ids from memory.

func (*VariantIDs) Start 

func (r *VariantIDs) Start(ctx context.Context, wg *sync.WaitGroup)

start runs the touch flusher until ctx ends, flushing once more on the way out. A registry without a database has nothing to flush.

type VariantRef 

type VariantRef struct {
	Key       core.RatingKey
	ReleaseID string
}

type VersionFlowKey 

type VersionFlowKey struct {
	UserID   int64
	ClientID string
	Key      core.RatingKey
}

VersionFlowKey scopes issued versions to one viewer, client and title.

type VersionFlows 

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

func NewVersionFlows 

func NewVersionFlows(ctx context.Context, db *sql.DB) *VersionFlows

NewVersionFlows binds the worker to the host lifetime. The host must run Run and wait for it during shutdown before closing its database.

func (*VersionFlows) Call 

func (v *VersionFlows) Call(ctx context.Context, k VersionFlowKey, vs []core.ReleaseVariant, mode int, now time.Time, ttl time.Duration) (VersionSnapshot, error)

call never writes SQL on the request goroutine. A publication barrier is essential: returning a pre-play response before its snapshot is committed allows the next request on another node to race publication. Concurrent operations are batched in one bounded transaction by the worker.

func (*VersionFlows) Run 

func (v *VersionFlows) Run()

type VersionSnapshot 

type VersionSnapshot struct {
	Variants []core.ReleaseVariant
	Frozen   bool
	Until    time.Time
}

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.