silo-server/internal/notifications/availability_detector.go

package notifications

import (
	"context"
	"log/slog"
	"time"
)

const availabilityDetectTimeout = 2 * time.Minute

// AvailabilityDetector turns completed ingest runs into episode_availability
// facts and release events. It runs after matching/reconcile is complete so
// a release is tied to an actual resolved episode, and it never blocks or
// fails the ingest itself.
type AvailabilityDetector struct {
	releases *ReleaseRepository
	settings *Settings
	logger   *slog.Logger
	// nudge wakes the fanout worker after new release events land; may be nil.
	nudge func()
}

// NewAvailabilityDetector creates an AvailabilityDetector.
func NewAvailabilityDetector(releases *ReleaseRepository, settings *Settings) *AvailabilityDetector {
	return &AvailabilityDetector{
		releases: releases,
		settings: settings,
		logger:   slog.Default().With("component", "notifications.availability"),
	}
}

// SetFanoutNudge wires the fanout worker wake signal.
func (d *AvailabilityDetector) SetFanoutNudge(nudge func()) {
	if d != nil {
		d.nudge = nudge
	}
}

// AvailabilityKinds selects which content kinds an ingest scope covers.
// Each kind keeps its own seed marker and silent-seeding semantics.
type AvailabilityKinds struct {
	Episodes   bool
	Movies     bool
	Audiobooks bool
	Ebooks     bool
}

// Any reports whether at least one kind is selected.
func (k AvailabilityKinds) Any() bool {
	return k.Episodes || k.Movies || k.Audiobooks || k.Ebooks
}

// availabilityKindOps abstracts the per-kind recording calls so episode and
// movie passes share one detection flow; seed state is kind-keyed in the
// repository itself.
type availabilityKindOps struct {
	kind             string
	recordForLibrary func(ctx context.Context, libraryID int, emitEvents bool) (int, int, error)
	recordForPaths   func(ctx context.Context, libraryID int, scopePaths []string, emitEvents bool) (int, int, error)
}

// HandleIngestCompleted records newly available content for a completed
// ingest scope. fullLibrary distinguishes whole-library scans (set-based
// detection, and the scan that seeds a new library) from subtree/file scans
// (path-bounded detection).
//
// Seeding semantics: a library without a seed marker records availability
// silently — "newly available" means newly released to this server, not newly
// seen by the notifications feature. The marker is written when a full scan
// completes successfully, so the next scan onward emits release events.
func (d *AvailabilityDetector) HandleIngestCompleted(ctx context.Context, libraryID int, fullLibrary bool, scopePaths []string, kinds AvailabilityKinds) {
	if d == nil || d.releases == nil {
		return
	}
	// The scan context is done once the scan finishes; detection runs on its
	// own deadline so cancellation of the parent does not drop availability.
	detectCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), availabilityDetectTimeout)
	defer cancel()

	if kinds.Episodes {
		d.runKind(detectCtx, libraryID, fullLibrary, scopePaths, availabilityKindOps{
			kind:             EventKindEpisode,
			recordForLibrary: d.releases.RecordAvailabilityForLibrary,
			recordForPaths:   d.releases.RecordAvailabilityForPaths,
		})
	}
	for _, k := range flatItemKinds {
		if k.Selected(kinds) {
			d.runKind(detectCtx, libraryID, fullLibrary, scopePaths, d.flatKindOps(k))
		}
	}
}

// flatKindOps binds one flat item kind's registry entry to the shared
// detection flow.
func (d *AvailabilityDetector) flatKindOps(k flatItemKind) availabilityKindOps {
	return availabilityKindOps{
		kind: k.Kind,
		recordForLibrary: func(ctx context.Context, libraryID int, emitEvents bool) (int, int, error) {
			return d.releases.RecordItemAvailabilityForLibrary(ctx, k, libraryID, emitEvents)
		},
		recordForPaths: func(ctx context.Context, libraryID int, scopePaths []string, emitEvents bool) (int, int, error) {
			return d.releases.RecordItemAvailabilityForPaths(ctx, k, libraryID, scopePaths, emitEvents)
		},
	}
}

func (d *AvailabilityDetector) runKind(ctx context.Context, libraryID int, fullLibrary bool, scopePaths []string, ops availabilityKindOps) {
	seeded, err := d.releases.IsContentSeeded(ctx, libraryID, ops.kind)
	if err != nil {
		d.logger.WarnContext(ctx, "seed state lookup failed",
			"library_id", libraryID, "kind", ops.kind, "error", err)
		return
	}
	emitEvents := seeded && d.settings.ReleaseEventsEnabled(ctx)

	var inserted, events int
	if fullLibrary {
		inserted, events, err = ops.recordForLibrary(ctx, libraryID, emitEvents)
	} else if seeded {
		inserted, events, err = ops.recordForPaths(ctx, libraryID, scopePaths, emitEvents)
	} else {
		// Subtree/file ingest on an unseeded library: record silently but do
		// not seed-mark — only a successful full scan proves the back catalog
		// has been captured.
		inserted, events, err = ops.recordForPaths(ctx, libraryID, scopePaths, false)
	}
	if err != nil {
		d.logger.WarnContext(ctx, "availability detection failed",
			"library_id", libraryID, "kind", ops.kind, "error", err)
		return
	}

	if fullLibrary && !seeded {
		if err := d.releases.MarkContentSeeded(ctx, libraryID, ops.kind); err != nil {
			d.logger.WarnContext(ctx, "seed marker write failed",
				"library_id", libraryID, "kind", ops.kind, "error", err)
		} else {
			d.logger.InfoContext(ctx, "library availability seeded",
				"library_id", libraryID, "kind", ops.kind, "availability_rows", inserted)
		}
	}

	if inserted > 0 || events > 0 {
		d.logger.InfoContext(ctx, "availability recorded",
			"library_id", libraryID,
			"kind", ops.kind,
			"full_library", fullLibrary,
			"availability_rows", inserted,
			"release_events", events,
		)
	}
	if events > 0 && d.nudge != nil {
		d.nudge()
	}
}