Documentation
¶
Overview ¶
Package debrid streams torrent releases through owner-configured debrid services (Real-Debrid, TorBox, AllDebrid, Premiumize, Debrid-Link, Offcloud), and Stremio addons' own HTTP stream links (direct.go). A torrent the service already holds ("cached") is added to the owner's account and its main video file is read over HTTPS range requests; nothing is downloaded in full and Ether never seeds.
The rest of Ether treats a release payload as opaque bytes (normally an NZB). A debrid release's payload is a small descriptor (see Payload), and the Indexer, Engine and health-check wrappers in this package dispatch on it, delegating everything else to the Usenet implementations.
Index ¶
- Constants
- Variables
- func IsPayload(b []byte) bool
- func NewStreamingTransport() *http.Transport
- type Account
- type AcquireRequest
- type Acquired
- type DirectLink
- type Engine
- func (e *Engine) Check(ctx context.Context, payload []byte) (bool, error)
- func (e *Engine) CheckRelease(ctx context.Context, releaseID string, payload []byte) (bool, error)
- func (e *Engine) Close() error
- func (e *Engine) DurablePrefetch(ctx context.Context, id string, payload []byte, off, n int64) error
- func (e *Engine) Inner() core.Engine
- func (e *Engine) Open(ctx context.Context, releaseID string, payload []byte) (core.MediaSource, error)
- func (e *Engine) Prefetch(ctx context.Context, releaseID string, payload []byte, offset, length int64) error
- func (e *Engine) RefreshLeases()
- func (e *Engine) Stats() EngineStats
- type EngineOptions
- type EngineStats
- type File
- type Handle
- type Indexer
- func (x *Indexer) FetchNZB(ctx context.Context, r core.Release) ([]byte, error)
- func (x *Indexer) Inner() core.Indexer
- func (x *Indexer) MissingSources() error
- func (x *Indexer) Name() string
- func (x *Indexer) PrepareSources(ctx context.Context, releases []core.Release) []core.Release
- func (x *Indexer) Search(ctx context.Context, req core.SearchRequest) ([]core.Release, error)
- func (x *Indexer) SearchRescue(ctx context.Context, req core.SearchRequest) ([]core.Release, error)
- func (x *Indexer) SearchWith(ctx context.Context, req core.SearchRequest, arrived func([]core.Release)) ([]core.Release, error)
- func (x *Indexer) SetUsenetReady(fn func() bool)
- type Latency
- type Options
- type Payload
- type Provider
- type ProviderOptions
- type ProviderStats
- type Service
- func (s *Service) Acquire(ctx context.Context, rel core.Release) ([]byte, error)
- func (s *Service) ApplyHints(ctx context.Context, rels []core.Release)
- func (s *Service) CheckCached(ctx context.Context, hashes []string) (map[string]bool, bool)
- func (s *Service) Config() config.Debrid
- func (s *Service) DirectPayload(ctx context.Context, rel core.Release) ([]byte, error)
- func (s *Service) Enabled() bool
- func (s *Service) KnownCached(hash string) bool
- func (s *Service) Latencies() map[string]Latency
- func (s *Service) RecordStream(providerID string, connect, ttfb time.Duration, bytes int64, ...)
- func (s *Service) Refresh(ctx context.Context, p Payload, force bool) (Payload, error)
- func (s *Service) SetCluster(node clusterbus.Node)
- func (s *Service) SetConcurrencyShare(share func(configured int) int)
- func (s *Service) Stats() []ProviderStats
- func (s *Service) Status() Status
- func (s *Service) Test(ctx context.Context, cfg config.DebridProvider) error
- func (s *Service) UpdateConfig(cfg config.Debrid)
- type Status
- type TorrentResolver
Constants ¶
const KindHTTP = "http"
KindHTTP is the Payload.Kind of a direct HTTP stream.
const TopicBackoff = "debrid:backoff"
TopicBackoff carries provider pauses between nodes.
Variables ¶
var ( // ErrAuth is a rejected API key or an account that cannot use the API // (expired premium). A configuration problem: never a release failure. // It is core.ErrDebridAuth so a playback error can name the cause. ErrAuth = core.ErrDebridAuth // ErrRateLimited is HTTP 429 or the provider's equivalent. ErrRateLimited = errors.New("debrid: rate limited") // ErrGone is a torrent, file or link the provider no longer has; the // payload's stable identifiers can usually re-add it. ErrGone = errors.New("debrid: torrent or link no longer available") // ErrBlocked is a file the provider holds but refuses to serve, such // as Real-Debrid's infringing_file answer to a cached torrent. Re-adding // the torrent cannot help, so on this provider it counts as not cached; // another provider may still serve it. ErrBlocked = fmt.Errorf("debrid: the provider blocks this file: %w", core.ErrNotCached) // ErrAddRefused is a provider refusing to add a torrent at all, as // Real-Debrid does with infringing_file both for filtered names and, // after a burst of adds, for any torrent for a while. The release is // skipped on that provider for now and repeated refusals pause adds. ErrAddRefused = fmt.Errorf("debrid: the provider refused to add the torrent: %w", core.ErrNotCached) // ErrNoProvider means no debrid provider is enabled. ErrNoProvider = errors.New("debrid: no debrid provider is enabled") // ErrUnsupported is an operation a provider does not offer. ErrUnsupported = errors.New("debrid: not supported by this provider") )
Provider errors. Callers classify with errors.Is.
Functions ¶
func NewStreamingTransport ¶
NewStreamingTransport is an HTTP transport tuned for media streaming: keep-alive, HTTP/2 when offered, many idle connections per host and short connect/TLS timeouts.
Types ¶
type AcquireRequest ¶
type AcquireRequest struct {
InfoHash string
Magnet string
// TorrentID reuses a torrent already on the account ("" adds anew).
TorrentID string
// Choose picks the wanted file; providers that select files (Real-Debrid)
// call it before selection.
Choose func([]File) (File, error)
CachedOnly bool
RemoveUncached bool
// MayRemove, when set, is asked before a torrent this call added is
// removed again (RemoveUncached); false keeps it on the account because
// a cluster node is streaming from it.
MayRemove func(ctx context.Context, torrentID string) bool
}
AcquireRequest asks a provider to make a torrent available.
type Acquired ¶
type Acquired struct {
TorrentID string
File File
// Added reports whether this call added the torrent to the account.
Added bool
}
Acquired is a torrent made available by a provider.
type DirectLink ¶
DirectLink is a time-limited HTTPS URL for a file.
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine implements core.Engine and core.HealthChecker: debrid payloads are served from HTTPS range requests; anything else goes to Inner.
func NewEngine ¶
func NewEngine(svc *Service, opts EngineOptions) *Engine
NewEngine wraps inner with debrid streaming.
func (*Engine) Check ¶
Check implements core.HealthChecker. For a debrid payload it confirms a direct link is obtainable and that a range request answers 206 with the expected size, caching the head block it read; others go to the inner checker.
func (*Engine) CheckRelease ¶
CheckRelease is Check with the release ID, passed through to the inner checker for Usenet payloads so its reads are cached under that ID. Implements core.ReleaseHealthChecker.
func (*Engine) Close ¶
Close cancels in-flight fetches and drops this node's streaming leases. The inner engine is closed by its owner.
func (*Engine) DurablePrefetch ¶ added in v0.2.4
func (e *Engine) DurablePrefetch(ctx context.Context, id string, payload []byte, off, n int64) error
DurablePrefetch also persists memory hits and verifies disk presence. Ordinary Prefetch is intentionally best effort; a managed shelf must not call a memory buffer or a failed cache write a completed preload.
func (*Engine) Open ¶
func (e *Engine) Open(ctx context.Context, releaseID string, payload []byte) (core.MediaSource, error)
Open implements core.Engine. With a fresh link already known (the usual case after a detail-page prewarm) it makes no provider API call and no request: the first read is served from the cache or one range GET.
func (*Engine) Prefetch ¶
func (e *Engine) Prefetch(ctx context.Context, releaseID string, payload []byte, offset, length int64) error
Prefetch implements core.Engine. For a debrid payload it pulls [offset, offset+length) into the segment cache; from offset 0 it also pulls the file's tail, where MKV Cues and MP4 moov usually live, so the container index is warm too.
func (*Engine) RefreshLeases ¶
func (e *Engine) RefreshLeases()
RefreshLeases renews this node's streaming leases now (clustered only), e.g. as soon as the database is reachable again after an outage longer than the lease lifetime, so no node removes a torrent this node still streams from.
type EngineOptions ¶
type EngineOptions struct {
// Inner serves every non-debrid payload (the Usenet stream engine).
Inner core.Engine
// Health checks non-debrid payloads; nil when Inner does not implement
// core.HealthChecker.
Health core.HealthChecker
// Cache stores fetched blocks on disk (nil keeps them in memory only).
Cache core.SegmentCache
// HTTPClient and HedgeClient read media; they must use separate
// connection pools so a hedged request opens a fresh connection.
HTTPClient *http.Client
HedgeClient *http.Client
Logger *slog.Logger
Now func() time.Time
HedgeDelay time.Duration // 0 = 400ms; negative disables hedging
BlockSize int64 // tests; 0 = 512 KiB
}
EngineOptions configure an Engine.
type EngineStats ¶
type EngineStats struct {
OpenFiles int `json:"openFiles"`
BlockBytes int64 `json:"blockBytes"`
MemoryHits int64 `json:"memoryHits"`
DiskHits int64 `json:"diskHits"`
BlocksFetched int64 `json:"blocksFetched"`
BytesFetched int64 `json:"bytesFetched"`
Hedges int64 `json:"hedges"`
HedgeWins int64 `json:"hedgeWins"`
LinkRefreshes int64 `json:"linkRefreshes"`
Errors int64 `json:"errors"`
}
EngineStats is the engine part of the debrid admin status.
type File ¶
type File struct {
ID string // provider file id ("" when the provider has none)
Path string
Size int64
// Link is the provider-side link for the file (Real-Debrid hoster
// link, AllDebrid file link, Premiumize, Debrid-Link and Offcloud
// direct link).
Link string
// contains filtered or unexported fields
}
File is one file of a torrent as a provider lists it.
func ChooseFile ¶
func ChooseFile(files []File, target *core.ReleaseTarget) (File, error)
ChooseFile picks the main video file of a torrent: for a target episode the largest file whose name matches it, else the largest video.
type Indexer ¶
type Indexer struct {
// contains filtered or unexported fields
}
Indexer wraps the aggregate indexer: torrent results are kept only while a debrid provider is enabled (annotated with cache availability where a provider offers a bulk check), and FetchNZB returns a debrid payload for torrents while Usenet releases go to the inner indexer unchanged.
A Stremio addon's own HTTP stream links (core.ProtocolHTTP) need no debrid account. Without one, a torrent the addon also linked plays through that link instead; with cached-only on, links the addon labels only as uncached are left out (the label is a hint, never proof: the probe in FetchNZB decides). Usenet results are left out while no Usenet provider is enabled.
func NewIndexer ¶
NewIndexer wraps inner.
func (*Indexer) FetchNZB ¶
FetchNZB implements core.Indexer. For a torrent it returns the debrid payload of the release's main video file (or the wanted episode of a season pack); for anything else, the inner indexer's NZB.
func (*Indexer) MissingSources ¶ added in v0.2.20
MissingSources reports core.ErrNoSources when no indexer is enabled, and core.ErrNoTorrentSources when debrid is the only way to play (no Usenet provider) but no enabled indexer returns torrents.
func (*Indexer) PrepareSources ¶ added in v0.2.4
PrepareSources resolves a bounded, already ranked shortlist's torrent descriptors together, then checks cache availability in bulk. It never adds a torrent to an account or starts an uncached download. Unknown and failed lookups remain candidates for the normal validation path.
func (*Indexer) SearchRescue ¶ added in v0.2.4
SearchRescue implements second-pass rescue search through the inner indexer, annotating discovered torrent releases with debrid cache status.
func (*Indexer) SearchWith ¶ added in v0.2.9
func (x *Indexer) SearchWith(ctx context.Context, req core.SearchRequest, arrived func([]core.Release)) ([]core.Release, error)
SearchWith exposes arrivals from both sources while cache discovery continues. The resolver targets and validates speculative torrents before selecting one.
func (*Indexer) SetUsenetReady ¶ added in v0.2.20
SetUsenetReady wires the "is a Usenet provider enabled" check used by MissingSources. Call before the indexer is used.
type Latency ¶
type Latency struct {
ResolveMs float64 `json:"resolveMs"`
LinkMs float64 `json:"linkMs"`
ConnectMs float64 `json:"connectMs"`
TTFBMs float64 `json:"ttfbMs"`
ThroughputBps float64 `json:"throughputBps"`
Samples int64 `json:"samples"`
}
Latency holds exponentially weighted averages of a provider's timings: API resolution (add/select/poll), link generation, CDN connect (TCP+TLS) and time to first byte of a range request, plus read throughput.
type Options ¶
type Options struct {
DB *sql.DB // nil keeps state in memory
Logger *slog.Logger
Now func() time.Time
// Provider options (tests point them at httptest servers).
ProviderOptions ProviderOptions
// NewProvider overrides provider construction (tests).
NewProvider func(config.DebridProvider, ProviderOptions) (Provider, error)
PayloadTTL time.Duration // stored payload reuse, 0 = 7 days
CachedTTL time.Duration // known-cached memory, 0 = 12h
UncachedTTL time.Duration // known-uncached memory, 0 = 30m
// DirectGuard guards direct HTTP stream requests (tests map fake public
// hosts onto local servers); nil uses a zero safefetch.Guard.
DirectGuard *safefetch.Guard
}
Options configure a Service.
type Payload ¶
type Payload struct {
Key string `json:"key,omitempty"` // payload cache key (release + target)
ReleaseID string `json:"releaseId,omitempty"` // core.Release.ID (segment cache key)
Provider string `json:"provider"` // config.DebridProvider.ID
Kind string `json:"kind"` // config.DebridKinds
InfoHash string `json:"infoHash"` // lower-case hex v1 info hash
Magnet string `json:"magnet,omitempty"` // magnet URI used to (re)add the torrent
TorrentID string `json:"torrentId,omitempty"` // provider torrent / magnet id
FileID string `json:"fileId,omitempty"` // provider file id
Path string `json:"path"` // file path inside the torrent
Size int64 `json:"size"` // file size in bytes
Link string `json:"link,omitempty"` // provider-side file link (RD/AllDebrid), unrestricted on demand
URL string `json:"url,omitempty"` // last direct download URL
Expires time.Time `json:"expires,omitempty"` // when URL should be regenerated
Target *core.ReleaseTarget `json:"target,omitempty"`
Release string `json:"release,omitempty"` // release name, for logs
// Direct HTTP streams (Kind KindHTTP, direct.go): Link is the addon's
// stream link, Headers the request headers it requires and Origin the
// owner-configured addon origin. Link and Headers can carry the
// addon's credentials.
Headers map[string]string `json:"headers,omitempty"`
Origin string `json:"origin,omitempty"`
}
Payload identifies one file of a torrent held by a debrid provider. It is persisted by playback for resume, so it stores stable identifiers (info hash, magnet, provider torrent and file ids, path and size) from which a fresh link can be generated after the last one expires. It never holds an API key.
func DecodePayload ¶
DecodePayload parses an encoded Payload.
type Provider ¶
type Provider interface {
Kind() string
// CheckCached reports which hashes are cached; ErrUnsupported when the
// service has no bulk availability check.
CheckCached(ctx context.Context, hashes []string) (map[string]bool, error)
// Find returns the id of a torrent with this hash already on the
// account ("" when none), so re-adds are avoided; ErrUnsupported when
// the service keeps no torrent list.
Find(ctx context.Context, infoHash string) (string, error)
Acquire(ctx context.Context, req AcquireRequest) (*Acquired, error)
// Link returns a direct URL for a file; ErrGone when the torrent or
// file must be re-acquired first.
Link(ctx context.Context, h Handle) (DirectLink, error)
Remove(ctx context.Context, torrentID string) error
Account(ctx context.Context) (Account, error)
}
Provider is one debrid service. Implementations must never wait for an actual torrent download: Acquire returns core.ErrNotCached for a torrent the service cannot serve instantly (removing a torrent it just added when req.CachedOnly and req.RemoveUncached are set).
func NewProvider ¶
func NewProvider(cfg config.DebridProvider, opts ProviderOptions) (Provider, error)
NewProvider builds the provider for a config entry.
type ProviderOptions ¶
type ProviderOptions struct {
HTTPClient *http.Client
BaseURL string
PollInterval time.Duration // 0 = 500ms
ReadyPolls int // 0 = 6: how long a cached torrent may take to report ready
Now func() time.Time
}
ProviderOptions are shared by provider constructors (tests point BaseURL at an httptest server and shorten the polls).
type ProviderStats ¶
type ProviderStats struct {
ID string `json:"id"`
Name string `json:"name"`
Kind string `json:"kind"`
Enabled bool `json:"enabled"`
Priority int `json:"priority"`
State string `json:"state"` // ok, backoff, auth_failed, disabled
LastError string `json:"lastError,omitempty"`
BackoffUntil *time.Time `json:"backoffUntil,omitempty"`
Requests int64 `json:"requests"`
Failures int64 `json:"failures"`
Cached int64 `json:"cached"`
NotCached int64 `json:"notCached"`
Blocked int64 `json:"blocked"` // cached files the provider refused to serve
LinksRefreshed int64 `json:"linksRefreshed"`
Reused int64 `json:"reused"`
// Slots is how many API calls this node makes to the provider at once:
// AccountSlots on a standalone server, this node's share of it in a
// cluster.
Slots int `json:"slots"`
AccountSlots int `json:"accountSlots"`
LastSuccess *time.Time `json:"lastSuccess,omitempty"`
Latency Latency `json:"latency"`
}
ProviderStats are one provider's counters and latency EWMAs.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service holds the configured debrid providers and applies priority, backoff, bounded concurrency and caching to every provider call.
func NewService ¶
NewService builds a Service from the debrid config section.
func (*Service) Acquire ¶
Acquire makes a torrent release available on the first provider (by priority) that has it cached, and returns its payload. A payload stored for the same release within PayloadTTL is reused without API calls. The error wraps core.ErrNotCached when every provider answered "not cached", errNoVideo (core.ErrReleaseRejected) when the torrent has no playable file, and is a plain (transient) error otherwise.
func (*Service) ApplyHints ¶ added in v0.2.17
ApplyHints records the cached markers sources put on torrent releases (core.Release.DebridCached and DebridUncached) for providers that have no answer of their own about the hash. A provider the source reported as holding the torrent is tried ahead of unknown ones; with CachedOnly, one it reported as not holding it is skipped without adding the torrent. Hints live in memory only and never replace a provider's answer.
func (*Service) CheckCached ¶
CheckCached asks providers with bulk availability checks about hashes, remembering the answers. It returns hash -> cached on some provider, and definitive=true when every enabled provider could answer (so a false entry means no provider has the torrent).
func (*Service) DirectPayload ¶ added in v0.2.44
DirectPayload probes an HTTP stream release and returns its payload: one ranged request learns the file's exact size and final location and rejects what is not the release (an error page, a "downloading" clip, a server without range requests or a size). Every answer the addon gives is transient (core.ErrNotCached or core.ErrSourceUnavailable): an addon that is still fetching a file, or behind a misbehaving proxy, may serve it properly later, so none marks the release bad. Only a release whose own link is malformed is rejected (core.ErrReleaseRejected).
func (*Service) KnownCached ¶
KnownCached reports, without any API call, whether an enabled provider recently reported the hash as cached.
func (*Service) Latencies ¶
Latencies returns each enabled provider's latency EWMAs by provider id, for source routing.
func (*Service) RecordStream ¶
func (s *Service) RecordStream(providerID string, connect, ttfb time.Duration, bytes int64, elapsed time.Duration)
RecordStream feeds CDN timings observed by the engine into the provider's EWMAs (connect and ttfb may be zero when unknown).
func (*Service) Refresh ¶
Refresh returns p with a usable direct link. Unless force is set, a fresher link stored by another process path is reused without an API call. The same provider regenerates the link (re-adding the torrent by magnet when it was removed); otherwise another provider holding the same file takes over. Errors wrap core.ErrSourceUnavailable for transient or no-longer-cached outcomes (failover without marking the release bad) and core.ErrArticleMissing when the file itself is gone from the torrent.
func (*Service) SetCluster ¶
func (s *Service) SetCluster(node clusterbus.Node)
SetCluster connects the cluster bus (nil disconnects). A standalone server never calls it.
func (*Service) SetConcurrencyShare ¶
SetConcurrencyShare divides each provider's account-wide API concurrency among cluster nodes: share receives the account limit and returns this node's part, clamped to 1..limit. Nil restores the account limit. Calls in flight finish; the new limit applies to new calls.
func (*Service) Stats ¶
func (s *Service) Stats() []ProviderStats
Stats reports every configured provider.
func (*Service) Status ¶
Status is the admin status document (engine stats are filled in by the caller when an Engine is wired).
func (*Service) Test ¶
Test checks one provider config for the admin "test" button: the key works and the account can use the API.
func (*Service) UpdateConfig ¶
UpdateConfig hot-applies an edited debrid section. Providers whose identity and key are unchanged keep their breaker state and counters.
type Status ¶
type Status struct {
Enabled bool `json:"enabled"`
CachedOnly bool `json:"cachedOnly"`
Providers []ProviderStats `json:"providers"`
Engine *EngineStats `json:"engine,omitempty"`
}
Status is the admin "debrid" status.