admin/memby
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.2.76 - Icon Packs

Browse Source
This commit is contained in:
ponzischeme89
2026-08-18 14:59:29 +12:00
parent 36d171e51b
commit 8c847c59b8
70 changed files with 5267 additions and 673 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/events.go
+355
View File
@@ -0,0 +1,355 @@
package library
import (
"fmt"
"strings"
)
// Turning a Sonarr or Radarr notification into a piece of work, and all of it pure.
//
// The gateway used to learn that a file had appeared by asking Emby every hour whether
// anything had been saved since the last time it asked. Sonarr and Radarr already know —
// they are the things that put the file there — so this is the translation from what they
// say into the one question the ingest worker answers: which item should be re-read, or
// removed, and how do two deliveries of the same news collapse into one.
//
// Nothing here does I/O, which is what lets every rule below be a table test.
// Ingest actions. A rename is deliberately a refresh like any other: the Emby item id
// survives a file being moved, and so does the credits fingerprint measured against it —
// the only stale thing is the row's payload.
const (
ActionRefresh = "refresh"
ActionRemove = "remove"
)
// Ingest kinds.
const (
KindEpisode = "episode"
KindMovie = "movie"
KindSeries = "series"
)
// Why a request exists. It is carried through to the log and the console, because "this
// episode was re-read because Sonarr upgraded the file" is the sentence an operator needs
// and "an item changed" is not.
const (
ReasonImport = "import"
ReasonUpgrade = "upgrade"
ReasonRename = "rename"
ReasonDelete = "delete"
)
// IngestRequest is one piece of work. It carries what the *arr knew rather than an Emby
// id, because at the moment a webhook arrives Emby has very often not scanned the file in
// yet and there is no id to carry.
type IngestRequest struct {
// Key is the dedupe identity, and it names the *file* rather than the event. Two
// deliveries of one import collapse onto one row; a file deleted and re-imported is a
// different file and therefore its own work. Same reasoning as the alert id in
// radarrImportAlert.
Key string `json:"-"`
Action string `json:"-"`
Kind string `json:"-"`
Reason string `json:"-"`
// Series identity, for an episode or a series-level event.
Series string `json:"series,omitempty"`
SeriesYear int `json:"seriesYear,omitempty"`
Season int `json:"season,omitempty"`
Episode int `json:"episode,omitempty"`
// Film identity.
Title string `json:"title,omitempty"`
Year int `json:"year,omitempty"`
// EmbyItemID is filled in only where the caller already knows it — a delete of
// something the catalogue holds. Empty is the ordinary case.
EmbyItemID string `json:"embyItemId,omitempty"`
}
// IngestResult is a finished piece of ingest work, handed to whoever wants to announce it.
//
// It is what makes "a scan has completed" a thing the gateway can say: a webhook only means
// the *arr has moved a file, and the several minutes between that and Emby having scanned
// it in are exactly the minutes in which a banner saying the title is there would be wrong.
// This is emitted from the other end, once the row is in the catalogue.
//
// It carries both what the *arr said and what Emby turned out to call the thing, because
// the news is about the title and the item id is what can put artwork behind it.
type IngestResult struct {
Source string // sonarr | radarr
Kind string // KindEpisode | KindMovie | KindSeries
Reason string // ReasonImport | ReasonUpgrade | ReasonRename | ReasonDelete
// ItemID and Name are Emby's, filled in from the row that was just written. ItemID is
// empty for a series-wide refresh, which is about a show rather than about one file.
ItemID string
Name string
ImageTag string
// SeriesName is Emby's name for the show an episode belongs to, which is what a banner
// leads with — the episode's own Name is its title.
SeriesName string
Season int
Episode int
Year int
}
// SonarrWebhook is the subset of Sonarr's body this reads. Sonarr sends considerably
// more; anything not named here is ignored on purpose, so a Sonarr upgrade that adds
// fields cannot break the hook.
type SonarrWebhook struct {
EventType string `json:"eventType"`
Series struct {
ID int `json:"id"`
Title string `json:"title"`
Year int `json:"year"`
} `json:"series"`
Episodes []struct {
ID int `json:"id"`
SeasonNumber int `json:"seasonNumber"`
EpisodeNumber int `json:"episodeNumber"`
} `json:"episodes"`
EpisodeFile struct {
ID int `json:"id"`
SeasonNumber int `json:"seasonNumber"`
} `json:"episodeFile"`
// RenamedEpisodeFiles is what On Rename carries: the files that moved, each with the
// id the library already knows them by.
RenamedEpisodeFiles []struct {
ID int `json:"id"`
SeasonNumber int `json:"seasonNumber"`
} `json:"renamedEpisodeFiles"`
IsUpgrade bool `json:"isUpgrade"`
// DeletedFiles marks a series delete that took the media with it. A series removed
// from Sonarr's list while its files stay on disk is not a reason to forget it.
DeletedFiles bool `json:"deletedFiles"`
}
// RadarrWebhook is the same narrow reading of Radarr's body.
type RadarrWebhook struct {
EventType string `json:"eventType"`
IsUpgrade bool `json:"isUpgrade"`
Movie struct {
ID int `json:"id"`
Title string `json:"title"`
Year int `json:"year"`
} `json:"movie"`
MovieFile struct {
ID int `json:"id"`
} `json:"movieFile"`
MovieFileID int `json:"movieFileId"`
DeletedFiles bool `json:"deletedFiles"`
}
// IsTestEvent reports the payload a webhook's Test button sends. It is answered 200 and
// enqueues nothing, which is what makes that button mean "reachable" rather than
// "reachable, and here is a row about a series that does not exist".
func IsTestEvent(eventType string) bool {
return strings.EqualFold(strings.TrimSpace(eventType), "Test")
}
// SonarrRequests turns one Sonarr notification into the work it implies.
//
// A notification can name several episodes — a multi-episode file, or a rename that moved
// a season — so this answers a slice. Each carries its own key, because each is its own
// file and the two may well arrive again separately.
func SonarrRequests(payload SonarrWebhook) []IngestRequest {
event := strings.ToLower(strings.TrimSpace(payload.EventType))
title := strings.TrimSpace(payload.Series.Title)
switch event {
case "download", "episodefileimported":
if title == "" || len(payload.Episodes) == 0 {
return nil
}
reason := ReasonImport
if payload.IsUpgrade {
// The file genuinely changed, so the row must be re-read. That it is not
// *news* is a separate judgement, made by the alert half.
reason = ReasonUpgrade
}
out := make([]IngestRequest, 0, len(payload.Episodes))
for _, episode := range payload.Episodes {
out = append(out, IngestRequest{
Key: sonarrEpisodeKey(payload.EpisodeFile.ID, episode.ID),
Action: ActionRefresh,
Kind: KindEpisode,
Reason: reason,
Series: title,
SeriesYear: payload.Series.Year,
Season: episode.SeasonNumber,
Episode: episode.EpisodeNumber,
})
}
return out
case "rename":
if title == "" {
return nil
}
// A rename names files rather than episodes, and Sonarr does not say which episode
// each file held. The series is the unit of work: one re-read of the show's
// episodes settles every file that moved, and a season rename would otherwise be
// one request per episode for the same answer.
return []IngestRequest{{
Key: fmt.Sprintf("sonarr:series:%d:rename", payload.Series.ID),
Action: ActionRefresh,
Kind: KindSeries,
Reason: ReasonRename,
Series: title,
SeriesYear: payload.Series.Year,
}}
case "episodefiledelete", "episodefiledeleted":
if title == "" {
return nil
}
season, episode := deletedEpisodeNumbers(payload)
if episode <= 0 {
return nil
}
return []IngestRequest{{
Key: fmt.Sprintf("sonarr:episodefile:%d:delete", payload.EpisodeFile.ID),
Action: ActionRemove,
Kind: KindEpisode,
Reason: ReasonDelete,
Series: title,
SeriesYear: payload.Series.Year,
Season: season,
Episode: episode,
}}
case "seriesdelete", "seriesdeleted":
// Only a delete that took the files. A series unfollowed in Sonarr while its
// episodes stay on disk is still in the library and must stay in the catalogue.
if title == "" || !payload.DeletedFiles {
return nil
}
return []IngestRequest{{
Key: fmt.Sprintf("sonarr:series:%d:delete", payload.Series.ID),
Action: ActionRemove,
Kind: KindSeries,
Reason: ReasonDelete,
Series: title,
SeriesYear: payload.Series.Year,
}}
}
return nil
}
// sonarrEpisodeKey prefers the file id, which is the thing that actually changed. Sonarr
// omits it on some versions of the import event, and the episode id is then the only
// stable identity available — coarser, since it does not change when the file is
// replaced, but a repeated delivery still collapses, which is what the key is for.
func sonarrEpisodeKey(fileID, episodeID int) string {
if fileID > 0 {
return fmt.Sprintf("sonarr:episodefile:%d:%d", fileID, episodeID)
}
return fmt.Sprintf("sonarr:episode:%d", episodeID)
}
// deletedEpisodeNumbers reads the position of a deleted file. The episode list is
// preferred because it carries the episode number; the file's own season number stands in
// where the list is absent.
func deletedEpisodeNumbers(payload SonarrWebhook) (int, int) {
for _, episode := range payload.Episodes {
if episode.EpisodeNumber > 0 {
season := episode.SeasonNumber
if season == 0 && payload.EpisodeFile.SeasonNumber > 0 {
season = payload.EpisodeFile.SeasonNumber
}
return season, episode.EpisodeNumber
}
}
return payload.EpisodeFile.SeasonNumber, 0
}
// RadarrRequests turns one Radarr notification into the work it implies.
func RadarrRequests(payload RadarrWebhook) []IngestRequest {
event := strings.ToLower(strings.TrimSpace(payload.EventType))
title := strings.TrimSpace(payload.Movie.Title)
if title == "" || payload.Movie.ID <= 0 {
return nil
}
fileID := payload.MovieFile.ID
if fileID <= 0 {
fileID = payload.MovieFileID
}
switch event {
case "download", "moviefileimported":
reason := ReasonImport
if payload.IsUpgrade {
reason = ReasonUpgrade
}
return []IngestRequest{{
Key: radarrFileKey(payload.Movie.ID, fileID, reason),
Action: ActionRefresh,
Kind: KindMovie,
Reason: reason,
Title: title,
Year: payload.Movie.Year,
}}
case "rename":
return []IngestRequest{{
Key: fmt.Sprintf("radarr:movie:%d:rename", payload.Movie.ID),
Action: ActionRefresh,
Kind: KindMovie,
Reason: ReasonRename,
Title: title,
Year: payload.Movie.Year,
}}
case "moviefiledelete", "moviefiledeleted":
return []IngestRequest{{
Key: radarrFileKey(payload.Movie.ID, fileID, ReasonDelete),
Action: ActionRemove,
Kind: KindMovie,
Reason: ReasonDelete,
Title: title,
Year: payload.Movie.Year,
}}
case "moviedelete", "moviedeleted":
if !payload.DeletedFiles {
return nil
}
return []IngestRequest{{
Key: fmt.Sprintf("radarr:movie:%d:delete", payload.Movie.ID),
Action: ActionRemove,
Kind: KindMovie,
Reason: ReasonDelete,
Title: title,
Year: payload.Movie.Year,
}}
}
return nil
}
func radarrFileKey(movieID, fileID int, reason string) string {
if fileID > 0 {
return fmt.Sprintf("radarr:moviefile:%d:%s", fileID, reason)
}
return fmt.Sprintf("radarr:movie:%d:%s", movieID, reason)
}
// NormalizedTitle strips a title down to the letters and digits it shares with whatever
// the other system calls it, so "Marvel's Daredevil" and "Marvels Daredevil" are one show.
//
// It lives here because both halves of the gateway need it and there must be exactly one
// answer to "which show is this": the schedule row matches Sonarr titles against the Emby
// catalogue with it, and the ingest worker matches the same titles against Emby itself.
func NormalizedTitle(value string) string {
return strings.Map(func(r rune) rune {
if r >= 'A' && r <= 'Z' {
return r + ('a' - 'A')
}
if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
return r
}
return -1
}, value)
}
server/internal/library/events_test.go
+229
View File
@@ -0,0 +1,229 @@
package library
import "testing"
func sonarrDownload(upgrade bool) SonarrWebhook {
var payload SonarrWebhook
payload.EventType = "Download"
payload.IsUpgrade = upgrade
payload.Series.ID = 12
payload.Series.Title = "Blue Bloods"
payload.Series.Year = 2010
payload.EpisodeFile.ID = 8123
payload.Episodes = append(payload.Episodes, struct {
ID int `json:"id"`
SeasonNumber int `json:"seasonNumber"`
EpisodeNumber int `json:"episodeNumber"`
}{ID: 551, SeasonNumber: 6, EpisodeNumber: 7})
return payload
}
func TestSonarrImportBecomesOneEpisodeRefresh(t *testing.T) {
requests := SonarrRequests(sonarrDownload(false))
if len(requests) != 1 {
t.Fatalf("expected one request, got %d", len(requests))
}
request := requests[0]
if request.Action != ActionRefresh || request.Kind != KindEpisode {
t.Fatalf("unexpected shape: %+v", request)
}
if request.Reason != ReasonImport {
t.Fatalf("expected an import, got %q", request.Reason)
}
if request.Series != "Blue Bloods" || request.Season != 6 || request.Episode != 7 {
t.Fatalf("unexpected identity: %+v", request)
}
}
// An upgrade is silent as *news* and is still a reason to re-read the row: the file
// genuinely changed. Conflating those two judgements is how a replaced file would keep a
// catalogue entry describing the copy it replaced.
func TestSonarrUpgradeStillRefreshes(t *testing.T) {
requests := SonarrRequests(sonarrDownload(true))
if len(requests) != 1 || requests[0].Reason != ReasonUpgrade {
t.Fatalf("expected one upgrade refresh, got %+v", requests)
}
if requests[0].Action != ActionRefresh {
t.Fatalf("an upgrade must refresh, got %q", requests[0].Action)
}
}
// The key names the file, so two deliveries of one import are one piece of work. Both
// *arrs re-notify on retry and neither promises exactly-once.
func TestRepeatedDeliveryKeepsOneKey(t *testing.T) {
first := SonarrRequests(sonarrDownload(false))
second := SonarrRequests(sonarrDownload(false))
if first[0].Key != second[0].Key {
t.Fatalf("the same import produced two keys: %q and %q", first[0].Key, second[0].Key)
}
// A different file for the same episode is different work, or a replacement would be
// swallowed by the row its predecessor left behind.
replaced := sonarrDownload(true)
replaced.EpisodeFile.ID = 9001
if SonarrRequests(replaced)[0].Key == first[0].Key {
t.Fatal("a replacement file must not reuse the previous file's key")
}
}
// A multi-episode file names several episodes and each is its own row, because each may
// well be delivered again on its own.
func TestSonarrMultiEpisodeFileProducesOneRequestEach(t *testing.T) {
payload := sonarrDownload(false)
payload.Episodes = append(payload.Episodes, struct {
ID int `json:"id"`
SeasonNumber int `json:"seasonNumber"`
EpisodeNumber int `json:"episodeNumber"`
}{ID: 552, SeasonNumber: 6, EpisodeNumber: 8})
requests := SonarrRequests(payload)
if len(requests) != 2 {
t.Fatalf("expected two requests, got %d", len(requests))
}
if requests[0].Key == requests[1].Key {
t.Fatal("two episodes of one file collapsed onto one key")
}
}
// A rename is series-wide because Sonarr does not say which episode each moved file held,
// and it is a refresh rather than an invalidation: the Emby id survives a move.
func TestSonarrRenameRefreshesTheSeries(t *testing.T) {
var payload SonarrWebhook
payload.EventType = "Rename"
payload.Series.ID = 12
payload.Series.Title = "Blue Bloods"
requests := SonarrRequests(payload)
if len(requests) != 1 {
t.Fatalf("expected one request, got %d", len(requests))
}
if requests[0].Kind != KindSeries || requests[0].Action != ActionRefresh {
t.Fatalf("unexpected shape: %+v", requests[0])
}
if requests[0].Reason != ReasonRename {
t.Fatalf("expected a rename, got %q", requests[0].Reason)
}
}
func TestSonarrEpisodeDeleteRemovesThatEpisode(t *testing.T) {
payload := sonarrDownload(false)
payload.EventType = "EpisodeFileDelete"
requests := SonarrRequests(payload)
if len(requests) != 1 || requests[0].Action != ActionRemove {
t.Fatalf("expected one removal, got %+v", requests)
}
if requests[0].Season != 6 || requests[0].Episode != 7 {
t.Fatalf("unexpected position: %+v", requests[0])
}
}
// A series removed from Sonarr's list while its files stay on disk is still in the
// library. Only a delete that took the media with it removes anything.
func TestSonarrSeriesDeleteOnlyCountsWhenFilesWent(t *testing.T) {
var payload SonarrWebhook
payload.EventType = "SeriesDelete"
payload.Series.ID = 12
payload.Series.Title = "Blue Bloods"
if requests := SonarrRequests(payload); len(requests) != 0 {
t.Fatalf("an unfollowed series must not be removed: %+v", requests)
}
payload.DeletedFiles = true
requests := SonarrRequests(payload)
if len(requests) != 1 || requests[0].Action != ActionRemove || requests[0].Kind != KindSeries {
t.Fatalf("expected a series removal, got %+v", requests)
}
}
func TestSonarrIgnoresEventsThatChangeNothing(t *testing.T) {
for _, event := range []string{"Grab", "Health", "ApplicationUpdate", "", "ManualInteractionRequired"} {
var payload SonarrWebhook
payload.EventType = event
payload.Series.Title = "Blue Bloods"
if requests := SonarrRequests(payload); len(requests) != 0 {
t.Fatalf("%q produced work: %+v", event, requests)
}
}
}
func radarrDownload(upgrade bool) RadarrWebhook {
var payload RadarrWebhook
payload.EventType = "Download"
payload.IsUpgrade = upgrade
payload.Movie.ID = 44
payload.Movie.Title = "Arrival"
payload.Movie.Year = 2016
payload.MovieFile.ID = 441
return payload
}
func TestRadarrImportAndUpgradeBothRefresh(t *testing.T) {
imported := RadarrRequests(radarrDownload(false))
if len(imported) != 1 || imported[0].Reason != ReasonImport || imported[0].Kind != KindMovie {
t.Fatalf("unexpected import: %+v", imported)
}
upgraded := RadarrRequests(radarrDownload(true))
if len(upgraded) != 1 || upgraded[0].Reason != ReasonUpgrade {
t.Fatalf("unexpected upgrade: %+v", upgraded)
}
// The two are separate work: an upgrade of a file already imported must not be
// swallowed by the settled row its import left behind.
if imported[0].Key == upgraded[0].Key {
t.Fatal("an upgrade reused the import's key")
}
}
func TestRadarrDeleteReadsTheTopLevelFileID(t *testing.T) {
var payload RadarrWebhook
payload.EventType = "MovieFileDelete"
payload.Movie.ID = 44
payload.Movie.Title = "Arrival"
payload.MovieFileID = 441
requests := RadarrRequests(payload)
if len(requests) != 1 || requests[0].Action != ActionRemove {
t.Fatalf("expected one removal, got %+v", requests)
}
if requests[0].Key != "radarr:moviefile:441:delete" {
t.Fatalf("unexpected key: %q", requests[0].Key)
}
}
func TestRadarrMovieDeleteOnlyCountsWhenFilesWent(t *testing.T) {
var payload RadarrWebhook
payload.EventType = "MovieDelete"
payload.Movie.ID = 44
payload.Movie.Title = "Arrival"
if requests := RadarrRequests(payload); len(requests) != 0 {
t.Fatalf("an unmonitored film must not be removed: %+v", requests)
}
payload.DeletedFiles = true
if requests := RadarrRequests(payload); len(requests) != 1 {
t.Fatalf("expected a removal, got %+v", requests)
}
}
// The Test button must be answered without recording work about something that does not
// exist, which is what makes it mean "reachable".
func TestTestEventIsRecognisedFromEitherArr(t *testing.T) {
if !IsTestEvent("Test") || !IsTestEvent(" test ") {
t.Fatal("a test event was not recognised")
}
if IsTestEvent("Download") {
t.Fatal("an import was read as a test")
}
var sonarrTest SonarrWebhook
sonarrTest.EventType = "Test"
sonarrTest.Series.Title = "Test Title"
if requests := SonarrRequests(sonarrTest); len(requests) != 0 {
t.Fatalf("the test event produced work: %+v", requests)
}
}
func TestNormalizedTitleIgnoresPunctuationAndCase(t *testing.T) {
if NormalizedTitle("Marvel's Daredevil") != NormalizedTitle("Marvels Daredevil") {
t.Fatal("punctuation changed the answer")
}
if NormalizedTitle("The Pitt") != "thepitt" {
t.Fatalf("unexpected normalisation: %q", NormalizedTitle("The Pitt"))
}
if NormalizedTitle(" ") != "" {
t.Fatal("a blank title must normalise to nothing")
}
}
server/internal/library/ingest.go
+740
View File
@@ -0,0 +1,740 @@
package library
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"net/url"
"strconv"
"strings"
"time"
"github.com/ponzischeme89/memby/server/internal/emby"
"github.com/ponzischeme89/memby/server/internal/store"
)
// The worker that drains what Sonarr and Radarr told us.
//
// The scheduled import asks Emby "what has changed since an hour ago" and pages through
// the answer. This asks Emby "where is this one episode", which is a request whose size
// does not grow with the library, and it asks only because something that actually puts
// files on disk said there was a reason to.
//
// Emby is the *lookup* here and never the discovery mechanism. Nothing in this file
// enumerates a library, and the one thing that still does — Syncer.Schedule — is demoted
// to reconciliation for media the *arrs do not manage.
const (
// defaultSettleDelay is how long after a webhook the first attempt is made. Sonarr
// fires On Import the moment it has moved the file; Emby has not scanned it yet, and
// asking immediately would spend a request to learn that.
defaultSettleDelay = 60 * time.Second
// idlePoll is how often the worker looks for due work. Coarse on purpose: everything
// here is already late by a settle delay, and a tight loop against Postgres on an idle
// NAS is exactly the background cost this replaces.
idlePoll = 20 * time.Second
// claimBatch bounds one pass. A season pack arrives as a dozen notifications at once
// and there is no hurry: draining a few per pass keeps Emby's request rate flat.
claimBatch = 4
// maxAttempts is where a piece of work is given up on. With the backoff below that is
// most of a day, after which the item is the reconciliation sweep's problem — which is
// the honest answer, since something other than timing is wrong by then.
maxAttempts = 7
// jobBudget bounds one piece of work end to end.
jobBudget = 60 * time.Second
)
// IngestStore is the slice of the store this needs. Narrow so the whole worker can be
// exercised against maps in a test, and so it is visible at a glance that the only things
// it writes are catalogue rows and the queue's own state.
type IngestStore interface {
EnqueueIngest(ctx context.Context, job store.IngestJob) (bool, error)
ClaimIngest(ctx context.Context, now time.Time, limit int) ([]store.IngestJob, error)
FinishIngest(ctx context.Context, key, state, outcome, itemID, errorText string, retryAt time.Time) error
UpsertLibraryItems(ctx context.Context, items []store.LibraryItem, syncedAt time.Time) (int64, error)
DeleteLibraryItem(ctx context.Context, itemID string) (int64, error)
SeriesRefs(ctx context.Context) ([]store.SeriesRef, error)
CreditsSeriesEpisodes(ctx context.Context, seriesIDs []string) ([]store.CreditsEpisodeRow, error)
LibraryItemsByName(ctx context.Context, itemType, name string) ([]store.NamedItem, error)
}
// EmbySource is the slice of Emby this needs: two reads and one nudge.
type EmbySource interface {
Items(ctx context.Context, cred emby.Credentials, params url.Values) (*emby.ItemsResult, error)
Episodes(ctx context.Context, cred emby.Credentials, seriesID string, params url.Values) (*emby.ItemsResult, error)
RefreshItem(ctx context.Context, cred emby.Credentials, itemID string) error
}
// Ingester drains the durable queue.
type Ingester struct {
Store IngestStore
Emby EmbySource
Credentials func(ctx context.Context) (emby.Credentials, error)
Log *slog.Logger
// Paused is the server-wide quiet-time gate. The queue is durable precisely so this can
// say no: a webhook that arrives during quiet hours is recorded and read afterwards,
// where the old arrangement answered it 503 and lost the event outright.
Paused func() bool
// Settle is the delay applied when work is enqueued. Held here so the hook and the
// worker cannot disagree about it.
Settle time.Duration
// Announce is told about a finished import, so the news reaches the televisions from
// the moment the title is actually there rather than from the moment the *arr said it
// would be. Installed from main.go, like syncer.SetAfterSync and for the same reason:
// library has no business knowing what an alert is. Nil is ordinary — a gateway with
// nothing to announce to, and every test in this package.
Announce func(ctx context.Context, result IngestResult)
}
func (i *Ingester) log() *slog.Logger {
if i == nil || i.Log == nil {
return slog.Default()
}
return i.Log
}
// SettleDelay is what the hook stamps onto a new row.
func (i *Ingester) SettleDelay() time.Duration {
if i == nil || i.Settle <= 0 {
return defaultSettleDelay
}
return i.Settle
}
// Run is the worker. One goroutine for the whole gateway.
func (i *Ingester) Run(ctx context.Context) {
if i == nil || i.Store == nil || i.Emby == nil || i.Credentials == nil {
return
}
i.log().Info("library ingest worker started", "settle", i.SettleDelay().String())
for {
if ctx.Err() != nil {
return
}
worked := false
if i.Paused == nil || !i.Paused() {
worked = i.drain(ctx)
}
if worked {
continue
}
if !sleep(ctx, idlePoll) {
return
}
}
}
// drain works everything currently due and reports whether it did anything, so a busy
// queue is emptied without waiting a poll interval between rows.
func (i *Ingester) drain(ctx context.Context) bool {
jobs, err := i.Store.ClaimIngest(ctx, time.Now().UTC(), claimBatch)
if err != nil {
if ctx.Err() == nil {
i.log().Warn("could not read the ingest queue", "error", err)
}
return false
}
if len(jobs) == 0 {
return false
}
for _, job := range jobs {
if ctx.Err() != nil {
return false
}
jobCtx, cancel := context.WithTimeout(ctx, jobBudget)
i.work(jobCtx, job)
cancel()
}
return true
}
// work is one row, start to finish. Every exit records an outcome, because the row *is*
// the operator's answer to "why was this item re-read, and did it work".
func (i *Ingester) work(ctx context.Context, job store.IngestJob) {
var request IngestRequest
if err := json.Unmarshal(job.Payload, &request); err != nil {
i.settle(ctx, job, store.IngestFailed, "invalid", "", err)
return
}
request.Key, request.Action = job.Key, job.Action
request.Kind, request.Reason = job.Kind, job.Reason
cred, err := i.Credentials(ctx)
if err != nil {
// Nobody has signed in yet, so there is no way to ask Emby anything. That is a
// deferral rather than a failure: the work is still valid, it simply cannot be
// done until a television signs in.
i.defer_(ctx, job, "no_credentials", err)
return
}
if job.Action == ActionRemove {
i.remove(ctx, job, request)
return
}
i.refresh(ctx, job, request, cred)
}
// refresh is the ordinary path: find the item in Emby and write it into the catalogue.
func (i *Ingester) refresh(
ctx context.Context, job store.IngestJob, request IngestRequest, cred emby.Credentials,
) {
items, itemID, err := i.resolve(ctx, request, cred)
if err != nil {
i.defer_(ctx, job, "lookup_failed", err)
return
}
if len(items) == 0 {
// Emby has not scanned the file in yet, which on a fresh import is the expected
// first answer rather than a fault. One nudge, then wait: the backoff is what turns
// "not yet" into "not ever" without a request per minute in between.
i.nudge(ctx, request, cred)
i.defer_(ctx, job, "not_found", nil)
return
}
written := make([]store.LibraryItem, 0, len(items))
for _, raw := range items {
if item, ok := toLibraryItem(raw); ok {
written = append(written, item)
}
}
if len(written) == 0 {
i.defer_(ctx, job, "not_found", nil)
return
}
// Stamped now, like any other import, so a title written here is never the victim of a
// full pass that happens to be running.
if _, err := i.Store.UpsertLibraryItems(ctx, written, time.Now().UTC()); err != nil {
i.defer_(ctx, job, "write_failed", err)
return
}
i.settle(ctx, job, store.IngestDone, "imported", itemID, nil)
i.log().Info("arr ingest",
"event", "arr_ingest", "key", job.Key, "source", job.Source, "reason", job.Reason,
"kind", job.Kind, "outcome", "imported", "items", len(written),
"item", itemID, "attempts", job.Attempts+1)
// After the row is recorded, never before: the announcement is a claim that the title
// is in the catalogue, and it must not be made by a pass that then failed to record it.
i.announce(ctx, job, request, written, itemID)
}
// announce reports a finished scan, if anybody is listening.
//
// Whether a given import is worth a banner is deliberately not decided here — that is a
// question about what viewers should be told, which belongs with the rest of the alert
// wording. This says what happened; the API package decides what to say about it.
func (i *Ingester) announce(
ctx context.Context, job store.IngestJob, request IngestRequest,
written []store.LibraryItem, itemID string,
) {
if i.Announce == nil {
return
}
result := IngestResult{
Source: job.Source,
Kind: job.Kind,
Reason: job.Reason,
ItemID: itemID,
SeriesName: request.Series,
Season: request.Season,
Episode: request.Episode,
Name: request.Title,
Year: request.Year,
}
// Emby's own record of the item outranks what the *arr called it: they disagree about
// punctuation and about years often enough that the banner and the card underneath it
// would otherwise name the same thing two ways.
if item, found := findWritten(written, itemID); found {
result.Name = item.Name
result.ImageTag = primaryImageTag(item.Payload)
if item.SeriesName != "" {
result.SeriesName = item.SeriesName
}
if item.ProductionYear != nil {
result.Year = *item.ProductionYear
}
}
i.Announce(ctx, result)
}
func findWritten(written []store.LibraryItem, itemID string) (store.LibraryItem, bool) {
if itemID == "" {
return store.LibraryItem{}, false
}
for _, item := range written {
if item.ID == itemID {
return item, true
}
}
return store.LibraryItem{}, false
}
// primaryImageTag digs the poster tag out of the payload that was just stored, so a banner
// can carry artwork without a second lookup. An absent tag is ordinary and costs nothing:
// the alert simply travels without one.
func primaryImageTag(payload json.RawMessage) string {
var parsed struct {
ImageTags map[string]string `json:"ImageTags"`
}
if json.Unmarshal(payload, &parsed) != nil {
return ""
}
return parsed.ImageTags["Primary"]
}
// remove takes a deleted title out of the catalogue.
//
// It resolves against the *local* catalogue rather than against Emby, which is the one
// place in this file that is deliberately the other way round: the thing being removed is
// a row in Memby's copy, and Emby — having had the file deleted underneath it — is the
// least likely place to still be able to name it.
func (i *Ingester) remove(ctx context.Context, job store.IngestJob, request IngestRequest) {
itemID, err := i.localItemID(ctx, request)
if err != nil {
i.defer_(ctx, job, "lookup_failed", err)
return
}
if itemID == "" {
// Nothing to remove. Ordinary rather than a failure: the catalogue may never have
// held it, or a previous delivery of this event already did the work.
i.settle(ctx, job, store.IngestDone, "absent", "", nil)
i.log().Info("arr ingest",
"event", "arr_ingest", "key", job.Key, "source", job.Source, "reason", job.Reason,
"kind", job.Kind, "outcome", "absent")
return
}
removed, err := i.Store.DeleteLibraryItem(ctx, itemID)
if err != nil {
i.defer_(ctx, job, "delete_failed", err)
return
}
i.settle(ctx, job, store.IngestDone, "removed", itemID, nil)
i.log().Info("arr ingest",
"event", "arr_ingest", "key", job.Key, "source", job.Source, "reason", job.Reason,
"kind", job.Kind, "outcome", "removed", "rows", removed, "item", itemID)
}
// resolve turns what the *arr said into Emby items, narrowly.
//
// The second return is the item the work was about, for the log and the console. It is
// empty for a series-wide refresh, which is about a show rather than about one file.
func (i *Ingester) resolve(
ctx context.Context, request IngestRequest, cred emby.Credentials,
) ([]json.RawMessage, string, error) {
switch request.Kind {
case KindMovie:
return i.resolveMovie(ctx, request, cred)
case KindEpisode, KindSeries:
return i.resolveFromSeries(ctx, request, cred)
}
return nil, "", fmt.Errorf("library: unknown ingest kind %q", request.Kind)
}
func (i *Ingester) resolveMovie(
ctx context.Context, request IngestRequest, cred emby.Credentials,
) ([]json.RawMessage, string, error) {
page, err := i.Emby.Items(ctx, cred, itemQuery(url.Values{
"SearchTerm": {request.Title},
"IncludeItemTypes": {"Movie"},
"Recursive": {"true"},
"Limit": {"20"},
}))
if err != nil {
return nil, "", err
}
if page == nil {
return nil, "", nil
}
match, id := pickByTitle(page.Items, request.Title, request.Year)
if match == nil {
return nil, "", nil
}
return []json.RawMessage{match}, id, nil
}
// resolveFromSeries handles both an episode and a whole-series refresh, because they share
// the expensive half: working out which Emby show this is.
func (i *Ingester) resolveFromSeries(
ctx context.Context, request IngestRequest, cred emby.Credentials,
) ([]json.RawMessage, string, error) {
seriesID, seriesPayload, err := i.seriesItem(ctx, request, cred)
if err != nil {
return nil, "", err
}
if seriesID == "" {
return nil, "", nil
}
params := itemQuery(url.Values{})
if request.Kind == KindEpisode && request.Season > 0 {
// One season rather than a show. A long-running series is a thousand records and
// this runs per imported file.
params.Set("Season", strconv.Itoa(request.Season))
}
page, err := i.Emby.Episodes(ctx, cred, seriesID, params)
if err != nil {
return nil, "", err
}
out := make([]json.RawMessage, 0, 8)
if seriesPayload != nil {
// A show Emby has only just created has no row here yet, and its episodes would be
// imported as children of a series the catalogue has never heard of.
out = append(out, seriesPayload)
}
if page == nil {
return out, "", nil
}
if request.Kind == KindSeries {
// A rename moved files; which files is not something Sonarr says, so the show is
// the unit of work and one re-read settles all of them.
return append(out, page.Items...), seriesID, nil
}
for _, raw := range page.Items {
var parsed struct {
ID string `json:"Id"`
IndexNumber *int `json:"IndexNumber"`
ParentIndexNumber *int `json:"ParentIndexNumber"`
}
if json.Unmarshal(raw, &parsed) != nil || parsed.IndexNumber == nil {
continue
}
if *parsed.IndexNumber != request.Episode {
continue
}
if parsed.ParentIndexNumber != nil && *parsed.ParentIndexNumber != request.Season {
continue
}
return append(out, raw), parsed.ID, nil
}
// The series is there and the episode is not: Emby has the show but has not scanned the
// new file. Reporting nothing found keeps that on the deferral path — but the series
// payload is still worth writing if it was new.
if len(out) > 0 {
if _, err := i.Store.UpsertLibraryItems(ctx, seriesItems(out), time.Now().UTC()); err != nil {
i.log().Debug("could not write the series row ahead of its episode", "error", err)
}
}
return nil, "", nil
}
// seriesItem answers which Emby series this is, preferring the catalogue.
//
// The local index is one query the gateway already makes elsewhere and it is right for
// every show that has ever been imported. Emby is asked only when it misses, which is
// exactly the case this feature exists for — a brand-new show whose first episode has just
// landed — and the payload comes back with it so the series row can be written too.
func (i *Ingester) seriesItem(
ctx context.Context, request IngestRequest, cred emby.Credentials,
) (string, json.RawMessage, error) {
if id := i.localSeriesID(ctx, request.Series, request.SeriesYear); id != "" {
return id, nil, nil
}
page, err := i.Emby.Items(ctx, cred, itemQuery(url.Values{
"SearchTerm": {request.Series},
"IncludeItemTypes": {"Series"},
"Recursive": {"true"},
"Limit": {"20"},
}))
if err != nil {
return "", nil, err
}
if page == nil {
return "", nil, nil
}
match, id := pickByTitle(page.Items, request.Series, request.SeriesYear)
return id, match, nil
}
func (i *Ingester) localSeriesID(ctx context.Context, title string, year int) string {
refs, err := i.Store.SeriesRefs(ctx)
if err != nil {
i.log().Debug("series index unavailable for ingest", "error", err)
return ""
}
return matchByTitle(refs, title, year)
}
// localItemID resolves a delete against the catalogue.
func (i *Ingester) localItemID(ctx context.Context, request IngestRequest) (string, error) {
switch request.Kind {
case KindMovie:
named, err := i.Store.LibraryItemsByName(ctx, "Movie", request.Title)
if err != nil {
return "", err
}
return matchNamed(named, request.Title, request.Year), nil
case KindSeries:
return i.localSeriesID(ctx, request.Series, request.SeriesYear), nil
case KindEpisode:
seriesID := i.localSeriesID(ctx, request.Series, request.SeriesYear)
if seriesID == "" {
return "", nil
}
episodes, err := i.Store.CreditsSeriesEpisodes(ctx, []string{seriesID})
if err != nil {
return "", err
}
for _, episode := range episodes {
if episode.Episode == request.Episode && episode.Season == request.Season {
return episode.ItemID, nil
}
}
}
return "", nil
}
// nudge asks Emby to look at the folder the file landed in.
//
// Best-effort and deliberately unreported: it is the same trick the subtitle download uses
// after Bazarr writes a sidecar, and a household whose Emby scans on its own does not need
// it. Refusing to nudge without a parent is the important half — a refresh of nothing is a
// request that cannot help.
func (i *Ingester) nudge(ctx context.Context, request IngestRequest, cred emby.Credentials) {
if request.Kind == KindMovie {
return
}
seriesID := i.localSeriesID(ctx, request.Series, request.SeriesYear)
if seriesID == "" {
return
}
if err := i.Emby.RefreshItem(ctx, cred, seriesID); err != nil {
i.log().Debug("could not ask emby to rescan a series", "series", seriesID, "error", err)
}
}
// defer_ schedules another attempt, or gives up.
func (i *Ingester) defer_(ctx context.Context, job store.IngestJob, outcome string, cause error) {
attempts := job.Attempts + 1
if attempts >= maxAttempts {
i.settle(ctx, job, store.IngestFailed, outcome, "", cause)
i.log().Warn("arr ingest gave up",
"event", "arr_ingest", "key", job.Key, "source", job.Source, "reason", job.Reason,
"kind", job.Kind, "outcome", outcome, "attempts", attempts, "error", errorText(cause))
return
}
retryAt := time.Now().UTC().Add(IngestRetryDelay(attempts))
if err := i.Store.FinishIngest(
ctx, job.Key, store.IngestPending, outcome, "", errorText(cause), retryAt,
); err != nil {
i.log().Warn("could not reschedule ingest work", "key", job.Key, "error", err)
}
i.log().Debug("arr ingest deferred",
"event", "arr_ingest", "key", job.Key, "reason", job.Reason, "outcome", outcome,
"attempts", attempts, "retry_in", IngestRetryDelay(attempts).String(),
"error", errorText(cause))
}
func (i *Ingester) settle(
ctx context.Context, job store.IngestJob, state, outcome, itemID string, cause error,
) {
// Detached from the job's own budget: a row that timed out must still record that it
// did, or the next pass claims it again immediately and the backoff never applies.
writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()
if err := i.Store.FinishIngest(
writeCtx, job.Key, state, outcome, itemID, errorText(cause), time.Now().UTC(),
); err != nil {
i.log().Warn("could not record ingest outcome", "key", job.Key, "error", err)
}
}
// IngestRetryDelay is the backoff, and it is a step function rather than an exponent so
// the schedule can be read off the page: a minute, five, twenty, an hour, then four-hourly
// out to the attempt limit. The early steps are short because the common cause is Emby not
// having scanned yet, which resolves in minutes; the late ones are long because by then the
// cause is something a faster retry cannot fix.
func IngestRetryDelay(attempts int) time.Duration {
switch {
case attempts <= 1:
return time.Minute
case attempts == 2:
return 5 * time.Minute
case attempts == 3:
return 20 * time.Minute
case attempts == 4:
return time.Hour
default:
return 4 * time.Hour
}
}
// itemQuery is the field set every lookup here uses, and it is deliberately the scheduled
// import's own.
//
// Thinning it would leave an event-imported title without People, MediaStreams or
// ProviderIds — so no cast on its page, no ratings lookup and no format badges — until Emby
// next reported it changed, which for a film nobody edits again is never. Syncer.Find makes
// the same promise for the same reason.
func itemQuery(params url.Values) url.Values {
params.Set("Fields", syncFields)
params.Set("ImageTypeLimit", "1")
params.Set("EnableImages", "true")
params.Set("EnableImageTypes", syncImageTypes)
params.Set("EnableTotalRecordCount", "false")
params.Set("EnableUserData", "false")
return params
}
// pickByTitle chooses the item a title and year names.
//
// Year-qualified first and title-only as the fallback, the rule the schedule row's series
// index already applies: an *arr and Emby disagree about a show's year far more often than
// they disagree about its name, but where both know the year it is what separates a remake
// from its original.
func pickByTitle(items []json.RawMessage, title string, year int) (json.RawMessage, string) {
want := NormalizedTitle(title)
if want == "" {
return nil, ""
}
var fallback json.RawMessage
var fallbackID string
for _, raw := range items {
var parsed struct {
ID string `json:"Id"`
Name string `json:"Name"`
ProductionYear *int `json:"ProductionYear"`
}
if json.Unmarshal(raw, &parsed) != nil || parsed.ID == "" {
continue
}
if NormalizedTitle(parsed.Name) != want {
continue
}
if year > 0 && parsed.ProductionYear != nil && *parsed.ProductionYear == year {
return raw, parsed.ID
}
if fallback == nil {
fallback, fallbackID = raw, parsed.ID
}
}
return fallback, fallbackID
}
func matchByTitle(refs []store.SeriesRef, title string, year int) string {
want := NormalizedTitle(title)
if want == "" {
return ""
}
fallback := ""
for _, ref := range refs {
if NormalizedTitle(ref.Name) != want {
continue
}
if year > 0 && ref.Year == year {
return ref.ID
}
if fallback == "" {
fallback = ref.ID
}
}
return fallback
}
func matchNamed(items []store.NamedItem, title string, year int) string {
want := NormalizedTitle(title)
if want == "" {
return ""
}
fallback := ""
for _, item := range items {
if NormalizedTitle(item.Name) != want {
continue
}
if year > 0 && item.Year == year {
return item.ID
}
if fallback == "" {
fallback = item.ID
}
}
return fallback
}
// seriesItems is the series payload on its own, for the case where the episode has not
// appeared yet but the show has.
func seriesItems(payloads []json.RawMessage) []store.LibraryItem {
out := make([]store.LibraryItem, 0, len(payloads))
for _, raw := range payloads {
if item, ok := toLibraryItem(raw); ok && item.Type == "Series" {
out = append(out, item)
}
}
return out
}
func errorText(err error) string {
if err == nil {
return ""
}
return strings.TrimSpace(err.Error())
}
func sleep(ctx context.Context, duration time.Duration) bool {
timer := time.NewTimer(duration)
defer timer.Stop()
select {
case <-ctx.Done():
return false
case <-timer.C:
return true
}
}
// Enqueue records work a webhook implied, and answers how much of it was news.
//
// It is the hook's whole job. Everything expensive happens later, on the worker, which is
// what lets the hook answer Sonarr in a millisecond and — more importantly — what lets it
// answer at all during quiet hours, when the work itself must wait.
func (i *Ingester) Enqueue(
ctx context.Context, source string, requests []IngestRequest,
) (int, error) {
if i == nil || i.Store == nil || len(requests) == 0 {
return 0, nil
}
due := time.Now().UTC().Add(i.SettleDelay())
fresh := 0
var firstErr error
for _, request := range requests {
payload, err := json.Marshal(request)
if err != nil {
if firstErr == nil {
firstErr = err
}
continue
}
inserted, err := i.Store.EnqueueIngest(ctx, store.IngestJob{
Key: request.Key,
Action: request.Action,
Kind: request.Kind,
Reason: request.Reason,
Source: source,
Payload: payload,
DueAt: due,
})
if err != nil {
if firstErr == nil {
firstErr = err
}
continue
}
if inserted {
fresh++
continue
}
// A repeat delivery is ordinary — both *arrs re-notify on retry — so it is DEBUG,
// the same stance the per-keystroke search line takes.
i.log().Debug("arr ingest already queued",
"event", "arr_ingest", "key", request.Key, "source", source, "reason", request.Reason)
}
return fresh, firstErr
}
server/internal/library/ingest_test.go
+507
View File
@@ -0,0 +1,507 @@
package library
import (
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"net/url"
"testing"
"time"
"github.com/ponzischeme89/memby/server/internal/emby"
"github.com/ponzischeme89/memby/server/internal/store"
)
// fakeStore is the queue and the catalogue as maps. Everything the worker does is visible
// in it, which is the point of the store being an interface here.
type fakeStore struct {
jobs map[string]*store.IngestJob
order []string
items map[string]store.LibraryItem
deleted []string
series []store.SeriesRef
episodes []store.CreditsEpisodeRow
named []store.NamedItem
failNext error
}
func newFakeStore() *fakeStore {
return &fakeStore{jobs: map[string]*store.IngestJob{}, items: map[string]store.LibraryItem{}}
}
func (f *fakeStore) EnqueueIngest(_ context.Context, job store.IngestJob) (bool, error) {
existing, found := f.jobs[job.Key]
if found {
// The real table's ON CONFLICT: one row, and a re-delivery never pulls the settle
// delay forward.
if job.DueAt.After(existing.DueAt) {
existing.DueAt = job.DueAt
}
existing.State = store.IngestPending
return false, nil
}
stored := job
stored.State = store.IngestPending
f.jobs[job.Key] = &stored
f.order = append(f.order, job.Key)
return true, nil
}
func (f *fakeStore) ClaimIngest(_ context.Context, now time.Time, limit int) ([]store.IngestJob, error) {
out := []store.IngestJob{}
for _, key := range f.order {
job := f.jobs[key]
if job.State != store.IngestPending || job.DueAt.After(now) {
continue
}
out = append(out, *job)
if len(out) == limit {
break
}
}
return out, nil
}
func (f *fakeStore) FinishIngest(
_ context.Context, key, state, outcome, itemID, errorText string, retryAt time.Time,
) error {
job, found := f.jobs[key]
if !found {
return errors.New("no such job")
}
job.State, job.Outcome, job.ItemID = state, outcome, itemID
job.LastError, job.DueAt = errorText, retryAt
job.Attempts++
return nil
}
func (f *fakeStore) UpsertLibraryItems(
_ context.Context, items []store.LibraryItem, _ time.Time,
) (int64, error) {
if f.failNext != nil {
err := f.failNext
f.failNext = nil
return 0, err
}
for _, item := range items {
f.items[item.ID] = item
}
return int64(len(items)), nil
}
func (f *fakeStore) DeleteLibraryItem(_ context.Context, itemID string) (int64, error) {
f.deleted = append(f.deleted, itemID)
delete(f.items, itemID)
return 1, nil
}
func (f *fakeStore) SeriesRefs(context.Context) ([]store.SeriesRef, error) { return f.series, nil }
func (f *fakeStore) CreditsSeriesEpisodes(
context.Context, []string,
) ([]store.CreditsEpisodeRow, error) {
return f.episodes, nil
}
func (f *fakeStore) LibraryItemsByName(
_ context.Context, _, _ string,
) ([]store.NamedItem, error) {
return f.named, nil
}
// fakeEmby answers the two lookups and counts the nudges.
type fakeEmby struct {
items []json.RawMessage
episodes []json.RawMessage
refreshed []string
itemQueries []url.Values
seasons []string
err error
}
func (f *fakeEmby) Items(
_ context.Context, _ emby.Credentials, params url.Values,
) (*emby.ItemsResult, error) {
f.itemQueries = append(f.itemQueries, params)
if f.err != nil {
return nil, f.err
}
return &emby.ItemsResult{Items: f.items}, nil
}
func (f *fakeEmby) Episodes(
_ context.Context, _ emby.Credentials, _ string, params url.Values,
) (*emby.ItemsResult, error) {
f.seasons = append(f.seasons, params.Get("Season"))
if f.err != nil {
return nil, f.err
}
return &emby.ItemsResult{Items: f.episodes}, nil
}
func (f *fakeEmby) RefreshItem(_ context.Context, _ emby.Credentials, itemID string) error {
f.refreshed = append(f.refreshed, itemID)
return nil
}
func testIngester(st *fakeStore, source *fakeEmby) *Ingester {
return &Ingester{
Store: st,
Emby: source,
Credentials: func(context.Context) (emby.Credentials, error) {
return emby.Credentials{UserID: "u", Token: "t"}, nil
},
Log: slog.New(slog.NewTextHandler(io.Discard, nil)),
Settle: time.Minute,
}
}
func episodePayload(id string, season, episode int) json.RawMessage {
raw, _ := json.Marshal(map[string]any{
"Id": id, "Name": "The Job", "Type": "Episode", "SeriesId": "emby-series",
"ParentIndexNumber": season, "IndexNumber": episode,
})
return raw
}
func enqueueOne(t *testing.T, ingester *Ingester, request IngestRequest) {
t.Helper()
if _, err := ingester.Enqueue(context.Background(), "sonarr", []IngestRequest{request}); err != nil {
t.Fatalf("enqueue: %v", err)
}
}
func episodeRequest() IngestRequest {
return IngestRequest{
Key: "sonarr:episodefile:8123:551", Action: ActionRefresh, Kind: KindEpisode,
Reason: ReasonImport, Series: "Blue Bloods", SeriesYear: 2010, Season: 6, Episode: 7,
}
}
// The ordinary path: the series is already in the catalogue, so Emby is asked for one
// season and the episode is written.
func TestImportWritesTheEpisodeFromOneSeasonLookup(t *testing.T) {
st := newFakeStore()
st.series = []store.SeriesRef{{ID: "emby-series", Name: "Blue Bloods", Year: 2010}}
source := &fakeEmby{episodes: []json.RawMessage{
episodePayload("emby-ep-6", 6, 6),
episodePayload("emby-ep-7", 6, 7),
}}
ingester := testIngester(st, source)
enqueueOne(t, ingester, episodeRequest())
ingester.work(context.Background(), *st.jobs["sonarr:episodefile:8123:551"])
if _, written := st.items["emby-ep-7"]; !written {
t.Fatalf("the episode was not written: %v", st.items)
}
if _, extra := st.items["emby-ep-6"]; extra {
t.Fatal("an episode nobody asked about was written")
}
if len(source.seasons) != 1 || source.seasons[0] != "6" {
t.Fatalf("expected one season-scoped lookup, got %v", source.seasons)
}
// A series already in the catalogue costs no search at all.
if len(source.itemQueries) != 0 {
t.Fatalf("the catalogue was not used for the series: %v", source.itemQueries)
}
job := st.jobs["sonarr:episodefile:8123:551"]
if job.State != store.IngestDone || job.Outcome != "imported" {
t.Fatalf("unexpected outcome: %+v", job)
}
}
// The field set must be the scheduled import's own, or an event-imported title arrives
// without the cast, streams and provider ids everything downstream reads.
func TestLookupsAskForTheFullSyncFields(t *testing.T) {
st := newFakeStore()
source := &fakeEmby{}
ingester := testIngester(st, source)
enqueueOne(t, ingester, IngestRequest{
Key: "radarr:moviefile:441:import", Action: ActionRefresh, Kind: KindMovie,
Reason: ReasonImport, Title: "Arrival", Year: 2016,
})
ingester.work(context.Background(), *st.jobs["radarr:moviefile:441:import"])
if len(source.itemQueries) != 1 {
t.Fatalf("expected one lookup, got %d", len(source.itemQueries))
}
query := source.itemQueries[0]
if query.Get("Fields") != syncFields {
t.Fatalf("a thinner field set was requested: %q", query.Get("Fields"))
}
if query.Get("EnableUserData") != "false" {
t.Fatal("the shared catalogue must never carry one viewer's user data")
}
}
// Emby not having scanned the file yet is the expected first answer, not a fault: the row
// waits, one nudge is sent, and the backoff widens.
func TestNotFoundDefersWithABackoffRatherThanFailing(t *testing.T) {
st := newFakeStore()
st.series = []store.SeriesRef{{ID: "emby-series", Name: "Blue Bloods", Year: 2010}}
source := &fakeEmby{}
ingester := testIngester(st, source)
enqueueOne(t, ingester, episodeRequest())
key := "sonarr:episodefile:8123:551"
before := time.Now().UTC()
ingester.work(context.Background(), *st.jobs[key])
job := st.jobs[key]
if job.State != store.IngestPending || job.Outcome != "not_found" {
t.Fatalf("expected a deferral, got %+v", job)
}
if !job.DueAt.After(before) {
t.Fatal("the next attempt was not scheduled into the future")
}
if len(source.refreshed) != 1 || source.refreshed[0] != "emby-series" {
t.Fatalf("expected one rescan nudge at the series, got %v", source.refreshed)
}
if len(st.items) != 0 {
t.Fatal("nothing should have been written")
}
}
// Attempts are given up on eventually, because past the last step of the backoff the cause
// is not timing and a row retrying for ever is one nobody looks at.
func TestRepeatedFailureIsEventuallyGivenUpOn(t *testing.T) {
st := newFakeStore()
source := &fakeEmby{err: errors.New("emby is not answering")}
ingester := testIngester(st, source)
enqueueOne(t, ingester, episodeRequest())
key := "sonarr:episodefile:8123:551"
for attempt := 0; attempt < maxAttempts; attempt++ {
ingester.work(context.Background(), *st.jobs[key])
}
if st.jobs[key].State != store.IngestFailed {
t.Fatalf("expected the row to be given up on, got %+v", st.jobs[key])
}
if st.jobs[key].LastError == "" {
t.Fatal("a failed row must record why")
}
}
func TestRetryDelayWidensAndSettles(t *testing.T) {
previous := time.Duration(0)
for attempt := 1; attempt <= 6; attempt++ {
delay := IngestRetryDelay(attempt)
if delay < previous {
t.Fatalf("the backoff narrowed at attempt %d: %s after %s", attempt, delay, previous)
}
previous = delay
}
if IngestRetryDelay(1) != time.Minute {
t.Fatalf("the first retry should be quick, got %s", IngestRetryDelay(1))
}
}
// A delete resolves against the catalogue, not against Emby: the file is gone, and Emby is
// the least likely thing to still be able to name it.
func TestDeleteRemovesTheEpisodeFromTheCatalogue(t *testing.T) {
st := newFakeStore()
st.series = []store.SeriesRef{{ID: "emby-series", Name: "Blue Bloods", Year: 2010}}
st.episodes = []store.CreditsEpisodeRow{
{ItemID: "emby-ep-6", SeriesID: "emby-series", Season: 6, Episode: 6},
{ItemID: "emby-ep-7", SeriesID: "emby-series", Season: 6, Episode: 7},
}
source := &fakeEmby{}
ingester := testIngester(st, source)
request := episodeRequest()
request.Action, request.Reason, request.Key = ActionRemove, ReasonDelete, "sonarr:episodefile:8123:delete"
enqueueOne(t, ingester, request)
ingester.work(context.Background(), *st.jobs[request.Key])
if len(st.deleted) != 1 || st.deleted[0] != "emby-ep-7" {
t.Fatalf("unexpected deletions: %v", st.deleted)
}
if len(source.itemQueries) != 0 || len(source.seasons) != 0 {
t.Fatal("a delete must not need to ask Emby anything")
}
}
// A delete of something the catalogue never held is settled rather than retried: there is
// nothing to remove and no later attempt could change that.
func TestDeleteOfSomethingAbsentSettlesQuietly(t *testing.T) {
st := newFakeStore()
ingester := testIngester(st, &fakeEmby{})
request := IngestRequest{
Key: "radarr:moviefile:9:delete", Action: ActionRemove, Kind: KindMovie,
Reason: ReasonDelete, Title: "Never Imported", Year: 1999,
}
enqueueOne(t, ingester, request)
ingester.work(context.Background(), *st.jobs[request.Key])
job := st.jobs[request.Key]
if job.State != store.IngestDone || job.Outcome != "absent" {
t.Fatalf("expected a quiet settle, got %+v", job)
}
if len(st.deleted) != 0 {
t.Fatalf("something was deleted: %v", st.deleted)
}
}
// A brand-new show is the case the local index cannot answer, and it is exactly the case
// this feature exists for. Emby is asked, and the series row is written beside its episode
// so the episode is not a child of a show the catalogue has never heard of.
func TestANewSeriesIsResolvedThroughEmbyAndWrittenToo(t *testing.T) {
st := newFakeStore()
seriesRaw, _ := json.Marshal(map[string]any{
"Id": "emby-series", "Name": "Blue Bloods", "Type": "Series", "ProductionYear": 2010,
})
source := &fakeEmby{
items: []json.RawMessage{seriesRaw},
episodes: []json.RawMessage{episodePayload("emby-ep-7", 6, 7)},
}
ingester := testIngester(st, source)
enqueueOne(t, ingester, episodeRequest())
ingester.work(context.Background(), *st.jobs["sonarr:episodefile:8123:551"])
if _, written := st.items["emby-series"]; !written {
t.Fatalf("the new series row was not written: %v", st.items)
}
if _, written := st.items["emby-ep-7"]; !written {
t.Fatal("the episode was not written")
}
}
// The year separates a remake from its original where both systems know it, and the title
// alone is the fallback because they disagree about years more often than about names.
func TestMovieMatchingPrefersTheYearAndFallsBackToTheTitle(t *testing.T) {
original, _ := json.Marshal(map[string]any{
"Id": "old", "Name": "The Thing", "Type": "Movie", "ProductionYear": 1982,
})
remake, _ := json.Marshal(map[string]any{
"Id": "new", "Name": "The Thing", "Type": "Movie", "ProductionYear": 2011,
})
items := []json.RawMessage{original, remake}
if _, id := pickByTitle(items, "The Thing", 2011); id != "new" {
t.Fatalf("the year did not decide: %q", id)
}
if _, id := pickByTitle(items, "The Thing", 0); id != "old" {
t.Fatalf("expected the first title match as the fallback, got %q", id)
}
if _, id := pickByTitle(items, "Something Else", 0); id != "" {
t.Fatalf("an unrelated title matched: %q", id)
}
}
// A repeated webhook is one row, and it never pulls the settle delay forward — the whole
// point of the delay is that the file has finished being written.
func TestRepeatedEnqueueIsOneRowAndKeepsTheSettleDelay(t *testing.T) {
st := newFakeStore()
ingester := testIngester(st, &fakeEmby{})
request := episodeRequest()
fresh, err := ingester.Enqueue(context.Background(), "sonarr", []IngestRequest{request})
if err != nil || fresh != 1 {
t.Fatalf("first delivery: fresh=%d err=%v", fresh, err)
}
first := st.jobs[request.Key].DueAt
fresh, err = ingester.Enqueue(context.Background(), "sonarr", []IngestRequest{request})
if err != nil || fresh != 0 {
t.Fatalf("a repeat was treated as news: fresh=%d err=%v", fresh, err)
}
if len(st.jobs) != 1 {
t.Fatalf("a repeat produced %d rows", len(st.jobs))
}
if st.jobs[request.Key].DueAt.Before(first) {
t.Fatal("a repeat pulled the settle delay forward")
}
}
// Quiet time stands the worker down without touching the queue, which is the arrangement
// that lets the hook accept an event at any hour.
func TestQuietTimeStopsTheWorkerAndNotTheQueue(t *testing.T) {
st := newFakeStore()
st.series = []store.SeriesRef{{ID: "emby-series", Name: "Blue Bloods", Year: 2010}}
source := &fakeEmby{episodes: []json.RawMessage{episodePayload("emby-ep-7", 6, 7)}}
ingester := testIngester(st, source)
ingester.Paused = func() bool { return true }
enqueueOne(t, ingester, episodeRequest())
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { ingester.Run(ctx); close(done) }()
time.Sleep(50 * time.Millisecond)
cancel()
<-done
if len(st.items) != 0 {
t.Fatal("work was done during quiet time")
}
if st.jobs["sonarr:episodefile:8123:551"].State != store.IngestPending {
t.Fatal("the queued work was lost rather than deferred")
}
}
// A finished scan is the moment there is something truthful to announce, which is why the
// hook fires from here rather than from the webhook. It must carry Emby's own names: the
// *arr and Emby disagree about punctuation often enough that a banner built from the
// webhook would name the same thing differently from the card underneath it.
func TestAFinishedImportIsAnnouncedWithEmbysOwnNames(t *testing.T) {
st := newFakeStore()
st.series = []store.SeriesRef{{ID: "emby-series", Name: "Blue Bloods", Year: 2010}}
source := &fakeEmby{episodes: []json.RawMessage{episodeWithArtwork("emby-ep-7", 6, 7)}}
ingester := testIngester(st, source)
var announced []IngestResult
ingester.Announce = func(_ context.Context, result IngestResult) {
announced = append(announced, result)
}
enqueueOne(t, ingester, episodeRequest())
ingester.work(context.Background(), *st.jobs["sonarr:episodefile:8123:551"])
if len(announced) != 1 {
t.Fatalf("expected one announcement, got %d", len(announced))
}
result := announced[0]
if result.ItemID != "emby-ep-7" || result.Name != "The Job" {
t.Errorf("announced %q/%q, want Emby's id and episode title", result.ItemID, result.Name)
}
if result.SeriesName != "Blue Bloods" {
t.Errorf("series = %q, want Emby's series name", result.SeriesName)
}
if result.Season != 6 || result.Episode != 7 {
t.Errorf("position = S%02dE%02d, want S06E07", result.Season, result.Episode)
}
if result.ImageTag != "poster-tag" {
t.Errorf("image tag = %q, want the poster from the stored payload", result.ImageTag)
}
if result.Reason != ReasonImport || result.Kind != KindEpisode {
t.Errorf("result = %+v, want the import reason and kind carried through", result)
}
}
// Nothing is announced for work that did not land. The banner claims the title is in the
// catalogue, so a lookup that found nothing must stay silent and simply be retried.
func TestNothingIsAnnouncedWhenEmbyHasNotScannedYet(t *testing.T) {
st := newFakeStore()
st.series = []store.SeriesRef{{ID: "emby-series", Name: "Blue Bloods", Year: 2010}}
ingester := testIngester(st, &fakeEmby{})
announcements := 0
ingester.Announce = func(context.Context, IngestResult) { announcements++ }
enqueueOne(t, ingester, episodeRequest())
ingester.work(context.Background(), *st.jobs["sonarr:episodefile:8123:551"])
if announcements != 0 {
t.Fatalf("announced %d times for an episode Emby has not scanned", announcements)
}
}
func episodeWithArtwork(id string, season, episode int) json.RawMessage {
raw, _ := json.Marshal(map[string]any{
"Id": id, "Name": "The Job", "Type": "Episode", "SeriesId": "emby-series",
"SeriesName": "Blue Bloods",
"ParentIndexNumber": season, "IndexNumber": episode,
"ImageTags": map[string]string{"Primary": "poster-tag"},
})
return raw
}
server/internal/library/syncer.go
+61 -23
View File
@@ -250,6 +250,14 @@ func (s *Syncer) run(
return result, nil
}
// EmbyCredentials is how the ingest worker borrows the same account the scheduled import
// uses. One rule for "who does the gateway talk to Emby as" rather than two, so a
// household with a service account configured never has an event-driven read appear in
// somebody's Emby history as their television.
func (s *Syncer) EmbyCredentials(ctx context.Context) (emby.Credentials, error) {
return s.credentials(ctx)
}
// credentials prefers the configured service account and otherwise borrows the most
// recent TV session.
func (s *Syncer) credentials(ctx context.Context) (emby.Credentials, error) {
@@ -333,44 +341,74 @@ func (s *Syncer) Find(ctx context.Context, term string, limit int) ([]json.RawMe
return found, nil
}
// disabledSyncPoll is how often a switched-off schedule wakes to ask whether it still is.
// A setting an operator has just changed must not need a restart, which is the same reason
// the Emby reachability probe keeps ticking slowly while it is off.
const disabledSyncPoll = 5 * time.Minute
// Schedule runs an incremental import on an interval until ctx is cancelled.
//
// New episodes tend to land through the day and films weekly; an hourly incremental pass
// covers both without ever asking Emby for the whole catalogue again.
func (s *Syncer) Schedule(ctx context.Context, interval time.Duration, paused ...func() bool) {
if interval <= 0 {
// It is reconciliation now rather than discovery. Where the *arr webhooks are configured, a
// file is in the catalogue within a minute of Sonarr or Radarr putting it there and this
// pass exists for what they do not manage — media dropped in by hand, a title edited in
// Emby, a webhook that never arrived because the gateway was down. Where they are not, it
// is still the only thing that notices anything, which is why the interval is the
// operator's rather than a constant.
//
// interval is a function rather than a value because it is read every cycle: an operator
// who has just lengthened the sweep must see that take effect without restarting the
// container. Zero means switched off, and this keeps waking to ask.
func (s *Syncer) Schedule(
ctx context.Context, interval func() time.Duration, paused ...func() bool,
) {
if interval == nil {
s.log.Info("library auto-sync disabled")
return
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
s.log.Info("library auto-sync scheduled", "interval", interval.String())
s.log.Info("library auto-sync scheduled", "interval", durationLabel(interval()))
for {
wait := interval()
disabled := wait <= 0
if disabled {
wait = disabledSyncPoll
}
timer := time.NewTimer(wait)
select {
case <-ctx.Done():
timer.Stop()
return
case <-ticker.C:
if len(paused) > 0 && paused[0] != nil && paused[0]() {
s.log.Debug("skipping scheduled sync; quiet time is active")
case <-timer.C:
}
timer.Stop()
if disabled {
continue
}
if len(paused) > 0 && paused[0] != nil && paused[0]() {
s.log.Debug("skipping scheduled sync; quiet time is active")
continue
}
if s.Running() {
s.log.Info("skipping scheduled sync; one is already running")
continue
}
if _, err := s.Sync(ctx, "incremental", "schedule"); err != nil {
if errors.Is(err, ErrNoCredentials) {
// Nobody has signed in yet. Not worth an error-level log every hour.
s.log.Info("skipping scheduled sync; no credentials yet")
continue
}
if s.Running() {
s.log.Info("skipping scheduled sync; one is already running")
continue
}
if _, err := s.Sync(ctx, "incremental", "schedule"); err != nil {
if errors.Is(err, ErrNoCredentials) {
// Nobody has signed in yet. Not worth an error-level log every hour.
s.log.Info("skipping scheduled sync; no credentials yet")
continue
}
s.log.Error("scheduled sync failed", "error", err)
}
s.log.Error("scheduled sync failed", "error", err)
}
}
}
func durationLabel(value time.Duration) string {
if value <= 0 {
return "off"
}
return value.String()
}
// syncItem mirrors the Emby fields promoted to columns.
type syncItem struct {
ID string `json:"Id"`