memby/server/internal/store/library_ingest.go

package store

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
)

// The durable side of event-driven ingest.
//
// One table, five queries, and the only interesting one is the insert: it is written
// ON CONFLICT on a key derived from the file, which is the whole of what makes repeated
// webhook delivery safe. Sonarr and Radarr both re-notify on retry and neither guarantees
// exactly-once, so "the same news twice" has to be an ordinary event rather than a
// duplicate row and a duplicate Emby lookup.

// Ingest states.
const (
	IngestPending = "pending"
	IngestDone    = "done"
	IngestFailed  = "failed"
)

// IngestRetention is how long settled rows are kept. Long enough that an operator asking
// "did the webhook fire when that episode landed last week" gets an answer, short enough
// that a household importing all day does not accumulate a table nobody reads.
const IngestRetention = 14 * 24 * time.Hour

// IngestJob is one row of work.
type IngestJob struct {
	Key       string          `json:"key"`
	Action    string          `json:"action"`
	Kind      string          `json:"kind"`
	Reason    string          `json:"reason"`
	Source    string          `json:"source"`
	Payload   json.RawMessage `json:"payload"`
	State     string          `json:"state"`
	Outcome   string          `json:"outcome"`
	ItemID    string          `json:"itemId"`
	Attempts  int             `json:"attempts"`
	LastError string          `json:"lastError"`
	DueAt     time.Time       `json:"dueAt"`
	CreatedAt time.Time       `json:"createdAt"`
	UpdatedAt time.Time       `json:"updatedAt"`
}

// IngestCounts is what the console reads beside the list.
type IngestCounts struct {
	Pending int `json:"pending"`
	Done    int `json:"done"`
	Failed  int `json:"failed"`
}

// EnqueueIngest records a piece of work, or refreshes one already waiting.
//
// The second return reports whether this delivery was news. A repeat is not an error and
// not a second row — it moves the existing row's due time no earlier and is logged at
// DEBUG, because a Sonarr that retried is an ordinary occurrence and not something an
// operator needs told about.
//
// A key that has already been *settled* is deliberately re-opened: the same file can
// legitimately be imported, deleted and imported again, and a row left at 'done' would
// swallow the second import for ever.
func (s *Store) EnqueueIngest(ctx context.Context, job IngestJob) (bool, error) {
	if job.Key == "" || job.Action == "" {
		return false, fmt.Errorf("store: ingest job needs a key and an action")
	}
	if len(job.Payload) == 0 {
		job.Payload = json.RawMessage(`{}`)
	}
	if job.DueAt.IsZero() {
		job.DueAt = time.Now().UTC()
	}
	var inserted bool
	err := s.pool.QueryRow(ctx, `
		INSERT INTO library_ingest_queue
			(key, action, kind, reason, source, payload, state, due_at, created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, 'pending', $7, now(), now())
		ON CONFLICT (key) DO UPDATE SET
			action     = EXCLUDED.action,
			kind       = EXCLUDED.kind,
			reason     = EXCLUDED.reason,
			source     = EXCLUDED.source,
			payload    = EXCLUDED.payload,
			state      = 'pending',
			outcome    = '',
			last_error = '',
			-- A re-delivery must never pull the settle delay forward: the point of it is
			-- that the file has finished being written, and an eager retry would ask Emby
			-- about a file it has not scanned yet.
			due_at     = GREATEST(library_ingest_queue.due_at, EXCLUDED.due_at),
			-- Attempts reset only when the row had settled. A retry storm against a row
			-- still being worked must not reset its backoff.
			attempts   = CASE WHEN library_ingest_queue.state = 'pending'
			                  THEN library_ingest_queue.attempts ELSE 0 END,
			updated_at = now()
		RETURNING (xmax = 0)`,
		job.Key, job.Action, job.Kind, job.Reason, job.Source, job.Payload, job.DueAt.UTC(),
	).Scan(&inserted)
	if err != nil {
		return false, fmt.Errorf("store: enqueue ingest: %w", err)
	}
	return inserted, nil
}

// ClaimIngest takes the work that is due, oldest first.
//
// It marks nothing: the worker is single and in-process, so a claim flag would be state to
// get wrong (a row left claimed by a container that was killed) in exchange for protecting
// against a second worker that does not exist. FinishIngest is what moves a row on.
func (s *Store) ClaimIngest(ctx context.Context, now time.Time, limit int) ([]IngestJob, error) {
	if limit <= 0 {
		limit = 10
	}
	rows, err := s.pool.Query(ctx, `
		SELECT key, action, kind, reason, source, payload, state, outcome, item_id,
		       attempts, last_error, due_at, created_at, updated_at
		FROM library_ingest_queue
		WHERE state = 'pending' AND due_at <= $1
		ORDER BY due_at
		LIMIT $2`, now.UTC(), limit)
	if err != nil {
		return nil, fmt.Errorf("store: claim ingest: %w", err)
	}
	defer rows.Close()
	return scanIngestJobs(rows)
}

// FinishIngest settles a row, or schedules the next attempt.
//
// state is 'done', 'failed' or 'pending' — the last being a deferral, which is the
// ordinary answer for a file Emby has not scanned in yet.
func (s *Store) FinishIngest(
	ctx context.Context, key, state, outcome, itemID, errorText string, retryAt time.Time,
) error {
	due := retryAt
	if due.IsZero() {
		due = time.Now().UTC()
	}
	_, err := s.pool.Exec(ctx, `
		UPDATE library_ingest_queue
		SET state = $2, outcome = $3, item_id = $4, last_error = $5,
		    attempts = attempts + 1, due_at = $6, updated_at = now()
		WHERE key = $1`, key, state, outcome, itemID, errorText, due.UTC())
	if err != nil {
		return fmt.Errorf("store: finish ingest: %w", err)
	}
	return nil
}

// RecentIngests is the console's read: newest activity first, whatever its state.
func (s *Store) RecentIngests(ctx context.Context, limit int) ([]IngestJob, error) {
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	rows, err := s.pool.Query(ctx, `
		SELECT key, action, kind, reason, source, payload, state, outcome, item_id,
		       attempts, last_error, due_at, created_at, updated_at
		FROM library_ingest_queue
		ORDER BY updated_at DESC
		LIMIT $1`, limit)
	if err != nil {
		return nil, fmt.Errorf("store: recent ingests: %w", err)
	}
	defer rows.Close()
	return scanIngestJobs(rows)
}

// IngestStateCounts is the summary above that list.
func (s *Store) IngestStateCounts(ctx context.Context) (IngestCounts, error) {
	var counts IngestCounts
	err := s.pool.QueryRow(ctx, `
		SELECT
			COUNT(*) FILTER (WHERE state = 'pending'),
			COUNT(*) FILTER (WHERE state = 'done'),
			COUNT(*) FILTER (WHERE state = 'failed')
		FROM library_ingest_queue`).Scan(&counts.Pending, &counts.Done, &counts.Failed)
	if err != nil {
		return IngestCounts{}, fmt.Errorf("store: ingest counts: %w", err)
	}
	return counts, nil
}

// PruneIngests removes settled rows past their retention. Pending work is never pruned:
// a row still waiting is work nobody has done, however old it is.
func (s *Store) PruneIngests(ctx context.Context, retention time.Duration) (int64, error) {
	if retention <= 0 {
		return 0, nil
	}
	tag, err := s.pool.Exec(ctx, `
		DELETE FROM library_ingest_queue
		WHERE state <> 'pending' AND updated_at < $1`, time.Now().UTC().Add(-retention))
	if err != nil {
		return 0, fmt.Errorf("store: prune ingests: %w", err)
	}
	return tag.RowsAffected(), nil
}

func scanIngestJobs(rows pgx.Rows) ([]IngestJob, error) {
	out := []IngestJob{}
	for rows.Next() {
		var job IngestJob
		if err := rows.Scan(
			&job.Key, &job.Action, &job.Kind, &job.Reason, &job.Source, &job.Payload,
			&job.State, &job.Outcome, &job.ItemID, &job.Attempts, &job.LastError,
			&job.DueAt, &job.CreatedAt, &job.UpdatedAt,
		); err != nil {
			return nil, err
		}
		out = append(out, job)
	}
	return out, rows.Err()
}