memby/server/internal/library/ingest.go

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
}