sessions

package
v0.3.8 Latest Latest
Warning

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

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

Documentation

Overview

Package sessions owns protocol-independent playback sessions.

Index

Constants

View Source 
const (
	VersionRead       = frontend.VersionRead
	VersionOffer      = frontend.VersionOffer
	VersionFreeze     = frontend.VersionFreeze
	VersionEnd        = frontend.VersionEnd
	VersionTouch      = frontend.VersionTouch
	VersionPreplayTTL = frontend.VersionPreplayTTL
)
View Source 
const MaxPlayQueues = frontend.MaxPlayQueues

Variables

View Source 
var Estimate = frontend.Estimate
View Source 
var HLSDecision = frontend.HLSDecision

HLSDecision lives in internal/frontend (host_sessions_experience.go).

View Source 
var NotifyTranscodeActivity = frontend.NotifyTranscodeActivity

NotifyTranscodeActivity lives in internal/frontend (host_sessions_experience.go).

View Source 
var VariantHash = frontend.VariantHash

VariantHash lives in internal/frontend (host_sessions_experience.go).

View Source 
var WireID = frontend.WireID

WireID lives in internal/frontend (host_sessions_experience.go); this alias keeps the broker's stream table on the same ids.

View Source 
var WithPlanAdmission = frontend.WithPlanAdmission
View Source 
var WithStreamServe = frontend.WithStreamServe

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 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 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)

Types

type Binding 

type Binding[R any] = frontend.Binding[R]

Binding and BindingTable live in internal/frontend (host_sessions_experience.go); generic types cannot be constructed through the factory interface.

type BindingTable 

type BindingTable[R any] = frontend.BindingTable[R]

func NewBindingTable 

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

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) ActivityStore 

func (b *Broker) ActivityStore() frontend.PlaybackActivityStore

func (*Broker) BindPlexDelivery 

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

BindPlexDelivery preserves the original authenticated Plex handler at owner.

func (*Broker) BindStream 

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

BindStream 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) CancelTrickplay 

func (b *Broker) CancelTrickplay(id string)

CancelTrickplay cancels idle preview generation for id when supported.

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) Cluster 

func (b *Broker) Cluster() frontend.ClusterNode

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) ConfirmAdmission 

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

ConfirmAdmission 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) DB 

func (b *Broker) DB() *sql.DB

func (*Broker) DeleteBinding 

func (b *Broker) DeleteBinding(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) Downloads 

func (b *Broker) Downloads() frontend.DownloadRemuxer

func (*Broker) EnsureActivity 

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

func (*Broker) Events 

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

func (*Broker) ExpireActivities 

func (b *Broker) ExpireActivities(now time.Time, ttl time.Duration) []PlaybackActivity

func (*Broker) ForwardEncodedMedia 

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

func (*Broker) ForwardGrantMedia 

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

func (*Broker) HasActivity 

func (b *Broker) HasActivity(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) HasBroker 

func (b *Broker) HasBroker() bool

func (*Broker) LatestDecision 

func (b *Broker) LatestDecision(key core.RatingKey) (core.DecisionTrace, bool)

func (*Broker) LifecycleGates 

func (b *Broker) LifecycleGates() frontend.Lifecycle

func (*Broker) ListActivities 

func (b *Broker) ListActivities() []PlaybackActivity

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) NewDecisionArms 

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

func (*Broker) NewDownloadJob 

func (b *Broker) NewDownloadJob(playback frontend.Playback, remuxer frontend.DownloadRemuxer) frontend.DownloadJob

func (*Broker) NewDownloadQueuePersistence 

func (b *Broker) NewDownloadQueuePersistence(queue frontend.DownloadQueueStore, db *sql.DB, artifacts frontend.DownloadRemuxer, log *slog.Logger, expiry func() time.Duration, storageID func() string, users func(context.Context) ([]core.User, error)) frontend.DownloadQueuePersistence

func (*Broker) NewDownloadQueueStore 

func (b *Broker) NewDownloadQueueStore() frontend.DownloadQueueStore

func (*Broker) NewPlayQueues 

func (b *Broker) NewPlayQueues() frontend.PlayQueues

func (*Broker) NewStartGuard 

func (b *Broker) NewStartGuard() frontend.StartGuard

func (*Broker) NewVariantIDs 

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

func (*Broker) NewVersionFlows 

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

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) ObserveStart 

func (b *Broker) ObserveStart(ctx context.Context, source frontend.HLSSession, id string) (frontend.StartObservation, bool)

func (*Broker) Open 

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

func (*Broker) OpenSidecar 

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

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

func (*Broker) PartSource 

func (b *Broker) PartSource(viewer int64, client string, key core.RatingKey, release string) frontend.PartSource

func (*Broker) Plan 

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

func (*Broker) Playback 

func (b *Broker) Playback() frontend.Playback

func (*Broker) PlaybackOwners 

func (b *Broker) PlaybackOwners() frontend.PlaybackOwners

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) 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) PrepareSource 

func (b *Broker) PrepareSource(ctx context.Context, viewer *core.User, key core.RatingKey, caps core.ClientCaps, release string, wait time.Duration) (*core.PlaySession, error)

func (*Broker) Prune 

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

Prune deletes delivery tokens older than the newest deliveryTokenKeep, expired transient session handles/connections/flows, and stale device capability rows, reporting the rows removed.

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) RemoveActivity 

func (b *Broker) RemoveActivity(id string) (PlaybackActivity, 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) ResolveStream 

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

ResolveStream 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) SegmentedEncoder 

func (b *Broker) SegmentedEncoder() frontend.SegmentedEncoder

func (*Broker) SegmentedHLSSession 

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

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

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) ServePartSource 

func (b *Broker) ServePartSource(w http.ResponseWriter, r *http.Request, sourceID string)

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) SessionHeartbeat 

func (b *Broker) SessionHeartbeat() time.Duration

func (*Broker) Sessions 

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

func (*Broker) SetDownloads 

func (b *Broker) SetDownloads(remuxer frontend.DownloadRemuxer)

func (*Broker) StartSegmentedHLS 

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

StartSegmentedHLS 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) StopProtocol 

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

StopProtocol releases every local session for one mounted surface.

func (*Broker) StopSegmentedHLS 

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

StopSegmentedHLS 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) 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) Transcoder 

func (b *Broker) Transcoder() frontend.Transcoder

func (*Broker) TrickplayAvailable 

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

TrickplayAvailable reports whether the borrowed provider has previews for id.

func (*Broker) TrickplayBIF 

func (b *Broker) TrickplayBIF(id string) ([]byte, bool)

TrickplayBIF returns the sparse BIF snapshot for id from the borrowed provider.

func (*Broker) TrickplayFrame 

func (b *Broker) TrickplayFrame(id string, at time.Duration) ([]byte, bool)

TrickplayFrame returns a single generated JPEG frame for id at offset at.

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) UpsertActivity 

func (b *Broker) UpsertActivity(id string, fn func(*PlaybackActivity)) PlaybackActivity

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
	SessionHeartbeat time.Duration
	Logger           *slog.Logger
	Downloads        core.DownloadRemuxer
	DecisionTraces   core.DecisionTraceSource
	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

func (*Coordination) LifecycleGates 

func (c *Coordination) LifecycleGates() frontend.Lifecycle

LifecycleGates returns the lifecycle gates as the frontend host contract.

func (*Coordination) PlaybackOwners 

func (c *Coordination) PlaybackOwners() frontend.PlaybackOwners

PlaybackOwners returns the owner history as the frontend host contract.

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 = frontend.DecisionOwner

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 = frontend.DownloadItem

Download value types live in internal/frontend (host_sessions_experience.go).

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 = frontend.DownloadPreparationError

type DownloadQueue 

type DownloadQueue = frontend.DownloadQueue

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) AllItems 

func (s *DownloadQueueStore) AllItems() []*DownloadItem

AllItems snapshots every queued item under the store lock.

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) Lock 

func (s *DownloadQueueStore) Lock()

Lock locks the store mutex (frontend.DownloadQueueStore contract).

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()

func (*DownloadQueueStore) Unlock 

func (s *DownloadQueueStore) Unlock()

Unlock unlocks the store mutex.

func (*DownloadQueueStore) WakeDownloads 

func (s *DownloadQueueStore) WakeDownloads()

WakeDownloads nudges the persistence writer without blocking.

type DownloadRequest 

type DownloadRequest = frontend.DownloadRequest

DownloadRequest and DownloadPreparationError live in internal/frontend (host_sessions_experience.go).

type DownloadStart 

type DownloadStart = frontend.DownloadStart

type Estimator 

type Estimator = frontend.Estimator

Estimator and Estimate live in internal/frontend (host_sessions_experience.go).

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 LocalRuntime 

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

LocalRuntime implements frontend.Sessions and frontend.SessionRuntime for standalone servers, spike fixtures, and unit tests.

func NewLocalRuntime 

func NewLocalRuntime(d LocalRuntimeDeps) *LocalRuntime

NewLocalRuntime constructs a SessionRuntime from local dependencies.

func (*LocalRuntime) ActivityStore 

func (r *LocalRuntime) ActivityStore() frontend.PlaybackActivityStore

func (*LocalRuntime) BindStream 

func (r *LocalRuntime) BindStream(ctx context.Context, access frontend.SessionAccess, binding frontend.StreamBinding) (int64, error)

func (*LocalRuntime) CancelTrickplay 

func (r *LocalRuntime) CancelTrickplay(id string)

func (*LocalRuntime) Capabilities 

func (r *LocalRuntime) Capabilities(ctx context.Context, a frontend.SessionAccess) (frontend.SessionCapabilities, error)

func (*LocalRuntime) Cluster 

func (r *LocalRuntime) Cluster() frontend.ClusterNode

func (*LocalRuntime) Command 

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

func (*LocalRuntime) ConfirmAdmission 

func (r *LocalRuntime) ConfirmAdmission(ctx context.Context, viewer *core.Viewer, remote bool) (*core.Viewer, error)

func (*LocalRuntime) DB 

func (r *LocalRuntime) DB() *sql.DB

func (*LocalRuntime) DecodeQueue 

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

func (*LocalRuntime) Deps 

func (r *LocalRuntime) Deps() LocalRuntimeDeps

func (*LocalRuntime) Downloads 

func (r *LocalRuntime) Downloads() frontend.DownloadRemuxer

func (*LocalRuntime) EncodeQueue 

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

func (*LocalRuntime) Events 

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

func (*LocalRuntime) FillDefaults 

func (r *LocalRuntime) FillDefaults(d LocalRuntimeDeps)

func (*LocalRuntime) HasBroker 

func (r *LocalRuntime) HasBroker() bool

func (*LocalRuntime) LatestDecision 

func (r *LocalRuntime) LatestDecision(key core.RatingKey) (core.DecisionTrace, bool)

func (*LocalRuntime) LifecycleGates 

func (r *LocalRuntime) LifecycleGates() frontend.Lifecycle

func (*LocalRuntime) Markers 

func (r *LocalRuntime) Markers(ctx context.Context, a frontend.SessionAccess, id string) (frontend.SessionMarkers, error)

func (*LocalRuntime) NewDecisionArms 

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

func (*LocalRuntime) NewDownloadJob 

func (r *LocalRuntime) NewDownloadJob(playback frontend.Playback, remuxer frontend.DownloadRemuxer) frontend.DownloadJob

func (*LocalRuntime) NewDownloadQueuePersistence 

func (r *LocalRuntime) NewDownloadQueuePersistence(queue frontend.DownloadQueueStore, db *sql.DB, artifacts frontend.DownloadRemuxer, log *slog.Logger, expiry func() time.Duration, storageID func() string, users func(context.Context) ([]core.User, error)) frontend.DownloadQueuePersistence

func (*LocalRuntime) NewDownloadQueueStore 

func (r *LocalRuntime) NewDownloadQueueStore() frontend.DownloadQueueStore

func (*LocalRuntime) NewPlayQueues 

func (r *LocalRuntime) NewPlayQueues() frontend.PlayQueues

func (*LocalRuntime) NewStartGuard 

func (r *LocalRuntime) NewStartGuard() frontend.StartGuard

func (*LocalRuntime) NewVariantIDs 

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

func (*LocalRuntime) NewVersionFlows 

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

func (*LocalRuntime) NotePlexEncoder 

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

func (*LocalRuntime) ObserveStart 

func (r *LocalRuntime) ObserveStart(ctx context.Context, source frontend.HLSSession, id string) (frontend.StartObservation, bool)

func (*LocalRuntime) Open 

func (r *LocalRuntime) Open(ctx context.Context, req frontend.OpenRequest) (frontend.Session, error)

func (*LocalRuntime) OpenSidecar 

func (r *LocalRuntime) OpenSidecar(ctx context.Context, access frontend.SessionAccess, receipt auth.Acceptance, wireID int64) (io.ReadCloser, core.StreamInfo, error)

func (*LocalRuntime) PartSource 

func (r *LocalRuntime) PartSource(viewer int64, client string, key core.RatingKey, release string) frontend.PartSource

func (*LocalRuntime) Plan 

func (r *LocalRuntime) Plan(ctx context.Context, req frontend.PlanRequest) (frontend.PlayPlan, error)

func (*LocalRuntime) Playback 

func (r *LocalRuntime) Playback() frontend.Playback

func (*LocalRuntime) PlaybackOwners 

func (r *LocalRuntime) PlaybackOwners() frontend.PlaybackOwners

func (*LocalRuntime) PlexSessionForPlayback 

func (r *LocalRuntime) PlexSessionForPlayback(ctx context.Context, a frontend.SessionAccess, item core.RatingKey, playbackID, transcodeID string) (string, error)

func (*LocalRuntime) PlexSessionForSource 

func (r *LocalRuntime) PlexSessionForSource(ctx context.Context, a frontend.SessionAccess, item core.RatingKey, playID string) (string, error)

func (*LocalRuntime) PlexSessionSource 

func (r *LocalRuntime) PlexSessionSource(id string) (string, bool)

func (*LocalRuntime) PrepareSource 

func (r *LocalRuntime) PrepareSource(ctx context.Context, viewer *core.User, key core.RatingKey, caps core.ClientCaps, release string, wait time.Duration) (*core.PlaySession, error)

func (*LocalRuntime) RecordAuthorizedSubtitleIntent 

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

func (*LocalRuntime) Recover 

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

func (*LocalRuntime) Report 

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

func (*LocalRuntime) ReportProgress 

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

func (*LocalRuntime) ResolveStream 

func (r *LocalRuntime) ResolveStream(ctx context.Context, access frontend.SessionAccess, wireID int64) (frontend.StreamBinding, error)

func (*LocalRuntime) SegmentedEncoder 

func (r *LocalRuntime) SegmentedEncoder() frontend.SegmentedEncoder

func (*LocalRuntime) ServeMedia 

func (r *LocalRuntime) ServeMedia(w http.ResponseWriter, req *http.Request)

func (*LocalRuntime) ServePartSource 

func (r *LocalRuntime) ServePartSource(w http.ResponseWriter, req *http.Request, sourceID string)

func (*LocalRuntime) SessionHeartbeat 

func (r *LocalRuntime) SessionHeartbeat() time.Duration

func (*LocalRuntime) Sessions 

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

func (*LocalRuntime) SetDownloads 

func (r *LocalRuntime) SetDownloads(remuxer frontend.DownloadRemuxer)

func (*LocalRuntime) StartStream 

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

func (*LocalRuntime) Stop 

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

func (*LocalRuntime) StopPlayback 

func (r *LocalRuntime) StopPlayback(ctx context.Context, playID string) bool

func (*LocalRuntime) StopProtocol 

func (r *LocalRuntime) StopProtocol(ctx context.Context, protocol frontend.Protocol) error

func (*LocalRuntime) SubtitleURL 

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

func (*LocalRuntime) TranscodeFor 

func (r *LocalRuntime) TranscodeFor(ctx context.Context, id string) (core.HLSSession, bool)

func (*LocalRuntime) Transcoder 

func (r *LocalRuntime) Transcoder() frontend.Transcoder

func (*LocalRuntime) TrickplayAvailable 

func (r *LocalRuntime) TrickplayAvailable(id string) bool

func (*LocalRuntime) TrickplayBIF 

func (r *LocalRuntime) TrickplayBIF(id string) ([]byte, bool)

func (*LocalRuntime) TrickplayFrame 

func (r *LocalRuntime) TrickplayFrame(id string, at time.Duration) ([]byte, bool)

func (*LocalRuntime) UpdateCapabilities 

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

type LocalRuntimeDeps 

type LocalRuntimeDeps struct {
	Sessions         frontend.Sessions
	Encoder          frontend.SegmentedEncoder
	Logger           *slog.Logger
	Playback         core.Playback
	PlaybackFunc     func() core.Playback
	Transcoder       core.Transcoder
	TranscoderFunc   func() core.Transcoder
	Downloads        core.DownloadRemuxer
	Trickplay        core.TrickplayProvider
	DecisionTraces   core.DecisionTraceSource
	DB               *sql.DB
	Cluster          frontend.ClusterNode
	SessionHeartbeat time.Duration
	Coordination     *Coordination
}

LocalRuntimeDeps configures a standalone SessionRuntime when no full session broker is mounted.

type PartSource 

type PartSource = frontend.PartSource

type PartSources 

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

PartSources 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 (PartSources) Part 

func (s PartSources) Part(viewer int64, client string, key core.RatingKey, release string) PartSource

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 (PartSources) Prepare 

func (s PartSources) 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 (PartSources) ServePart 

func (s PartSources) 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 PlanAdmission 

type PlanAdmission = frontend.PlanAdmission

PlanAdmission and its context helpers live in internal/frontend (host_sessions_experience.go).

type PlayQueue 

type PlayQueue = frontend.PlayQueue

func DecodeQueue 

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

type PlayQueuePersistence 

type PlayQueuePersistence = frontend.PlayQueuePersistence

Play-queue value types live in internal/frontend (host_sessions_experience.go); aliases keep this package on them.

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 PlaybackActivity 

type PlaybackActivity = frontend.PlaybackActivity

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 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 = frontend.QueueItem

type Selections 

type Selections[T any] = frontend.Selections[T]

Selections lives in internal/frontend (host_sessions_experience.go); generic types cannot be constructed through the factory interface.

func NewSelections 

func NewSelections[T any]() *Selections[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 = frontend.StartHintKey

type StartObservation 

type StartObservation = frontend.StartObservation

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 StreamBinding 

type StreamBinding = frontend.StreamBinding

StreamBinding and the PlexServe context helpers live in internal/frontend (host_sessions_experience.go).

type StreamTable 

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

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

func NewStreamTable 

func NewStreamTable() *StreamTable

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

func (*StreamTable) Bind 

func (t *StreamTable) Bind(binding StreamBinding) (int64, error)

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

func (*StreamTable) Lookup 

func (t *StreamTable) Lookup(id int64) (StreamBinding, bool)

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

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 = frontend.VariantRef

type VersionFlowKey 

type VersionFlowKey = frontend.VersionFlowKey

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 = frontend.VersionSnapshot

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.