From f9bb94a2995ce0a61d16f6037b47056eeef75c9f Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Sun, 7 Jun 2026 17:18:55 -0400 Subject: [PATCH] [codex] Fix autoscan plugin bindings and poll status (#75) * fix(autoscan): bind sources by plugin id * fix(autoscan): skip overlapping source polls --- cmd/silo/main.go | 3 + internal/api/autoscan_wiring.go | 20 +-- internal/api/handlers/autoscan.go | 151 +++++++++++------- internal/api/handlers/autoscan_test.go | 74 ++++++--- internal/autoscan/discovery.go | 48 ++---- internal/autoscan/discovery_test.go | 35 +--- internal/autoscan/errors.go | 4 + internal/autoscan/probe_test.go | 4 +- internal/autoscan/provider.go | 10 +- internal/autoscan/provider_test.go | 25 ++- internal/autoscan/repository.go | 150 ++++++++++++++--- internal/autoscan/service.go | 57 +++---- internal/autoscan/service_test.go | 123 +++++++++----- internal/autoscan/types.go | 19 +-- internal/plugins/service.go | 47 ++++++ internal/plugins/service_scan_source_test.go | 78 +++++++++ ...60607201204_autoscan_plugin_id_binding.sql | 145 +++++++++++++++++ .../20260607204951_autoscan_running_polls.sql | 36 +++++ web/src/api/types.ts | 26 ++- web/src/hooks/queries/useAutoscan.ts | 3 + web/src/lib/autoscanLabels.test.ts | 26 +-- web/src/lib/autoscanLabels.ts | 34 ++-- .../pages/admin/autoscan/ActivityPanel.tsx | 98 ++++++++++-- web/src/pages/admin/autoscan/SourcesPanel.tsx | 28 ++-- 24 files changed, 895 insertions(+), 349 deletions(-) create mode 100644 internal/plugins/service_scan_source_test.go create mode 100644 migrations/sql/20260607201204_autoscan_plugin_id_binding.sql create mode 100644 migrations/sql/20260607204951_autoscan_running_polls.sql diff --git a/cmd/silo/main.go b/cmd/silo/main.go index 93d52829..3f66c88b 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1499,6 +1499,9 @@ func main() { taskMgr.Register(tasks.NewReconcileRequestsTask(requestReconcileSvc, 100)) if deps.FolderRepo != nil && deps.LibraryScanQueue != nil && pluginService != nil && pluginInstallationStore != nil { autoscanRepo := autoscan.NewRepository(deps.DB) + if err := autoscanRepo.MarkInterruptedEvents(appCtx); err != nil { + slog.Warn("autoscan: failed to mark interrupted polls", "err", err) + } autoscanSvc := api.BuildAutoscanService( autoscanRepo, pluginService, diff --git a/internal/api/autoscan_wiring.go b/internal/api/autoscan_wiring.go index e65639d5..ddd10253 100644 --- a/internal/api/autoscan_wiring.go +++ b/internal/api/autoscan_wiring.go @@ -64,8 +64,8 @@ type PluginScanSourceAdapter struct { Svc *plugins.Service } -func (a PluginScanSourceAdapter) ScanSourceClient(ctx context.Context, installationID int, capabilityID string) (autoscan.PollChangesClient, error) { - return a.Svc.ScanSourceClient(ctx, installationID, capabilityID) +func (a PluginScanSourceAdapter) ScanSourceClient(ctx context.Context, pluginID, capabilityID string) (autoscan.PollChangesClient, error) { + return a.Svc.ScanSourceClientByPluginID(ctx, pluginID, capabilityID) } // scanSourceCapabilityType is the plugin capability type autoscan discovery @@ -75,15 +75,6 @@ const scanSourceCapabilityType = "scan_source.v1" // PluginScanSourceLister adapts the plugin installation store to // autoscan.ScanSourceLister: it enumerates every installed scan_source.v1 // capability across ALL installed plugins, regardless of enabled state. -// -// Using List (not ListEnabled) is deliberate for orphan detection: a temporarily -// DISABLED-but-installed plugin must NOT be treated as orphaned. If we dropped -// disabled plugins here, PollOnce would prune their sources and skip them with no -// last_error — they'd vanish silently. Keeping them present means PollOnce still -// attempts them; the plugin client fails to load (not running) and the source -// gets a visible RecordError instead. Only a fully UNINSTALLED plugin (gone from -// List entirely) is a true orphan. The Add-source picker shares this all-installed -// set, which is fine — operators can bind sources against an installed capability. type PluginScanSourceLister struct { Store *plugins.InstallationStore } @@ -104,10 +95,9 @@ func (l PluginScanSourceLister) ListScanSources(ctx context.Context) ([]autoscan continue } out = append(out, autoscan.DiscoveredSource{ - InstallationID: c.InstallationID, - CapabilityID: c.ID, - PluginID: inst.PluginID, - DisplayName: scanSourceDisplayName(inst.PluginID, c), + PluginID: inst.PluginID, + CapabilityID: c.ID, + DisplayName: scanSourceDisplayName(inst.PluginID, c), }) } } diff --git a/internal/api/handlers/autoscan.go b/internal/api/handlers/autoscan.go index d8bb727d..fb91a0cb 100644 --- a/internal/api/handlers/autoscan.go +++ b/internal/api/handlers/autoscan.go @@ -36,6 +36,7 @@ type autoscanStore interface { CountAutoscanScans(ctx context.Context, filter autoscan.ScanListFilter) (int, error) ListEvents(ctx context.Context, filter autoscan.EventListFilter) ([]autoscan.EventWithRuns, error) CountEvents(ctx context.Context, filter autoscan.EventListFilter) (int, error) + ListRunningEvents(ctx context.Context) ([]autoscan.Event, error) GetQueueSummary(ctx context.Context) (autoscan.QueueSummary, error) LatestEventAt(ctx context.Context) (*time.Time, error) } @@ -309,7 +310,7 @@ func (h *AutoscanHandler) HandleDeleteConnection(w http.ResponseWriter, r *http. // credentials: the connection link is by id only. type autoscanSourceResponse struct { ID string `json:"id"` - InstallationID int `json:"installation_id"` + PluginID string `json:"plugin_id"` CapabilityID string `json:"capability_id"` ConnectionID *string `json:"connection_id"` Enabled bool `json:"enabled"` @@ -331,7 +332,7 @@ func sourceResponse(s autoscan.Source) autoscanSourceResponse { config := normalizeSourceConfig(s.SourceConfig) return autoscanSourceResponse{ ID: s.ID, - InstallationID: s.InstallationID, + PluginID: s.PluginID, CapabilityID: s.CapabilityID, ConnectionID: s.ConnectionID, Enabled: s.Enabled, @@ -364,10 +365,9 @@ func (h *AutoscanHandler) HandleListSources(w http.ResponseWriter, r *http.Reque // --- Available scan-source plugins (Add-source picker) --- type autoscanScanSourcePluginResponse struct { - InstallationID int `json:"installation_id"` - CapabilityID string `json:"capability_id"` - PluginID string `json:"plugin_id"` - DisplayName string `json:"display_name"` + PluginID string `json:"plugin_id"` + CapabilityID string `json:"capability_id"` + DisplayName string `json:"display_name"` } // HandleListAvailableScanSources returns every installed scan_source capability @@ -381,10 +381,9 @@ func (h *AutoscanHandler) HandleListAvailableScanSources(w http.ResponseWriter, out := make([]autoscanScanSourcePluginResponse, 0, len(available)) for _, a := range available { out = append(out, autoscanScanSourcePluginResponse{ - InstallationID: a.InstallationID, - CapabilityID: a.CapabilityID, - PluginID: a.PluginID, - DisplayName: a.DisplayName, + PluginID: a.PluginID, + CapabilityID: a.CapabilityID, + DisplayName: a.DisplayName, }) } writeJSON(w, http.StatusOK, struct { @@ -394,12 +393,12 @@ func (h *AutoscanHandler) HandleListAvailableScanSources(w http.ResponseWriter, // --- Create source --- -// autoscanCreateSourceInput is the create payload: it names the installed -// scan_source capability to bind (installation_id + capability_id) plus the -// initial binding/scheduling fields. Unlike the update payload, identity is -// supplied here rather than read from an existing row. +// autoscanCreateSourceInput is the create payload: it names the scan_source +// capability to bind (plugin_id + capability_id) plus the initial +// binding/scheduling fields. Unlike the update payload, identity is supplied +// here rather than read from an existing row. type autoscanCreateSourceInput struct { - InstallationID int `json:"installation_id"` + PluginID string `json:"plugin_id"` CapabilityID string `json:"capability_id"` ConnectionID *string `json:"connection_id"` Enabled bool `json:"enabled"` @@ -420,6 +419,11 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ return } capID := strings.TrimSpace(in.CapabilityID) + pluginID := strings.TrimSpace(in.PluginID) + if pluginID == "" { + writeError(w, http.StatusBadRequest, "bad_request", "plugin_id is required") + return + } if capID == "" { writeError(w, http.StatusBadRequest, "bad_request", "capability_id is required") return @@ -432,7 +436,7 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ writeError(w, http.StatusBadRequest, "bad_request", err.Error()) return } - // Validate the (installation_id, capability_id) is a currently-installed + // Validate the (plugin_id, capability_id) is a currently-installed // scan_source capability — a source may only be created against an installed // plugin capability. available, err := h.svc.ListAvailableScanSources(r.Context()) @@ -440,13 +444,13 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ writeAutoscanError(w, err) return } - if !scanSourceInstalled(available, in.InstallationID, capID) { - writeError(w, http.StatusBadRequest, "bad_request", "installation_id + capability_id is not a currently-installed scan_source capability") + if !scanSourceInstalled(available, pluginID, capID) { + writeError(w, http.StatusBadRequest, "bad_request", "plugin_id + capability_id is not a currently-installed scan_source capability") return } connArg := normalizeConnectionID(in.ConnectionID) created, err := h.repo.CreateSource(r.Context(), autoscan.Source{ - InstallationID: in.InstallationID, + PluginID: pluginID, CapabilityID: capID, ConnectionID: connArg, Enabled: in.Enabled, @@ -462,18 +466,18 @@ func (h *AutoscanHandler) HandleCreateSource(w http.ResponseWriter, r *http.Requ writeJSON(w, http.StatusCreated, sourceResponse(created)) } -// scanSourceInstalled reports whether (installationID, capabilityID) is in the +// scanSourceInstalled reports whether (pluginID, capabilityID) is in the // set of currently-installed scan_source capabilities. -func scanSourceInstalled(available []autoscan.AvailableScanSource, installationID int, capabilityID string) bool { +func scanSourceInstalled(available []autoscan.AvailableScanSource, pluginID, capabilityID string) bool { for _, a := range available { - if a.InstallationID == installationID && a.CapabilityID == capabilityID { + if a.PluginID == pluginID && a.CapabilityID == capabilityID { return true } } return false } -// autoscanSourceInput is the source write payload. The (installation_id, +// autoscanSourceInput is the source write payload. The (plugin_id, // capability_id) identity is read from the existing source row, so only the // schedulable/binding fields are accepted. // @@ -548,7 +552,7 @@ func (h *AutoscanHandler) HandleUpdateSource(w http.ResponseWriter, r *http.Requ // connection_id is a full-state field: nil means unbind, a UUID string // means bind. A whitespace-only string is normalised to nil (unbound). connArg := normalizeConnectionID(in.ConnectionID) - // Update by id; identity (installation_id, capability_id) is immutable and + // Update by id; identity (plugin_id, capability_id) is immutable and // preserved by the repo. A missing source maps to 404. updated, err := h.repo.UpdateSource(r.Context(), autoscan.Source{ ID: id, @@ -581,9 +585,8 @@ func normalizeConnectionID(in *string) *string { } // HandleDeleteSource removes a source row. This is the operator's escape hatch -// for an orphaned source (one whose scan_source plugin was uninstalled): the -// poll loop skips such rows quietly, and this endpoint clears them. An unknown -// id maps to 404 via autoscan.ErrNotFound. +// for an orphaned source (one whose scan_source plugin was uninstalled or no +// longer resolves cleanly). An unknown id maps to 404 via autoscan.ErrNotFound. func (h *AutoscanHandler) HandleDeleteSource(w http.ResponseWriter, r *http.Request) { id := strings.TrimSpace(chi.URLParam(r, "id")) if err := h.repo.DeleteSource(r.Context(), id); err != nil { @@ -710,7 +713,7 @@ type autoscanEventScanRunResponse struct { type autoscanEventResponse struct { ID int64 `json:"id"` SourceID *string `json:"source_id"` - InstallationID int `json:"installation_id"` + PluginID string `json:"plugin_id"` CapabilityID string `json:"capability_id"` StartedAt time.Time `json:"started_at"` CompletedAt time.Time `json:"completed_at"` @@ -753,7 +756,7 @@ func eventResponse(event autoscan.EventWithRuns) autoscanEventResponse { return autoscanEventResponse{ ID: e.ID, SourceID: e.SourceID, - InstallationID: e.InstallationID, + PluginID: e.PluginID, CapabilityID: e.CapabilityID, StartedAt: e.StartedAt, CompletedAt: e.CompletedAt, @@ -794,9 +797,9 @@ func (h *AutoscanHandler) HandleListEvents(w http.ResponseWriter, r *http.Reques filter.Offset = offset } switch filter.Status { - case "", autoscan.EventStatusSuccess, autoscan.EventStatusError, autoscan.EventStatusUnresolved: + case "", autoscan.EventStatusRunning, autoscan.EventStatusSuccess, autoscan.EventStatusError, autoscan.EventStatusUnresolved: default: - writeError(w, http.StatusBadRequest, "bad_request", "status must be success, error, or unresolved") + writeError(w, http.StatusBadRequest, "bad_request", "status must be running, success, error, or unresolved") return } @@ -835,7 +838,7 @@ type autoscanScanResponse struct { CompletedAt *time.Time `json:"completed_at,omitempty"` AutoscanEventID *int64 `json:"autoscan_event_id,omitempty"` SourceID *string `json:"source_id,omitempty"` - InstallationID *int `json:"installation_id,omitempty"` + PluginID string `json:"plugin_id,omitempty"` CapabilityID string `json:"capability_id,omitempty"` EventStatus string `json:"event_status,omitempty"` EventCompletedAt *time.Time `json:"event_completed_at,omitempty"` @@ -862,7 +865,7 @@ func autoscanScanRunResponse(scan autoscan.ScanWithEvent) autoscanScanResponse { CompletedAt: scan.CompletedAt, AutoscanEventID: scan.AutoscanEventID, SourceID: scan.SourceID, - InstallationID: scan.InstallationID, + PluginID: scan.PluginID, CapabilityID: scan.CapabilityID, EventStatus: string(scan.EventStatus), EventCompletedAt: scan.EventCompletedAt, @@ -923,24 +926,35 @@ func (h *AutoscanHandler) HandleListScans(w http.ResponseWriter, r *http.Request // --- Status --- type autoscanStatusSource struct { - ID string `json:"id"` - InstallationID int `json:"installation_id"` - CapabilityID string `json:"capability_id"` - ConnectionID *string `json:"connection_id"` - Enabled bool `json:"enabled"` - Label string `json:"label"` - PathRewrites []autoscan.PathRewrite `json:"path_rewrites"` - LastRunAt *time.Time `json:"last_run_at,omitempty"` - LastError *string `json:"last_error,omitempty"` + ID string `json:"id"` + PluginID string `json:"plugin_id"` + CapabilityID string `json:"capability_id"` + ConnectionID *string `json:"connection_id"` + Enabled bool `json:"enabled"` + Label string `json:"label"` + PathRewrites []autoscan.PathRewrite `json:"path_rewrites"` + LastRunAt *time.Time `json:"last_run_at,omitempty"` + LastError *string `json:"last_error,omitempty"` +} + +type autoscanRunningPollResponse struct { + ID int64 `json:"id"` + SourceID *string `json:"source_id"` + PluginID string `json:"plugin_id"` + CapabilityID string `json:"capability_id"` + StartedAt time.Time `json:"started_at"` + ElapsedMS int64 `json:"elapsed_ms"` + MarkerBefore *string `json:"marker_before,omitempty"` } type autoscanStatusResponse struct { - Enabled bool `json:"enabled"` - Sources []autoscanStatusSource `json:"sources"` - ActiveScans int `json:"active_scans"` - AcceptedScans int `json:"accepted_scans"` - RunningScans int `json:"running_scans"` - LatestEventAt *time.Time `json:"latest_event_at,omitempty"` + Enabled bool `json:"enabled"` + Sources []autoscanStatusSource `json:"sources"` + RunningPolls []autoscanRunningPollResponse `json:"running_polls"` + ActiveScans int `json:"active_scans"` + AcceptedScans int `json:"accepted_scans"` + RunningScans int `json:"running_scans"` + LatestEventAt *time.Time `json:"latest_event_at,omitempty"` } func (h *AutoscanHandler) HandleStatus(w http.ResponseWriter, r *http.Request) { @@ -960,6 +974,11 @@ func (h *AutoscanHandler) HandleStatus(w http.ResponseWriter, r *http.Request) { writeAutoscanError(w, err) return } + runningEvents, err := h.repo.ListRunningEvents(ctx) + if err != nil { + writeAutoscanError(w, err) + return + } latestEventAt, err := h.repo.LatestEventAt(ctx) if err != nil { writeAutoscanError(w, err) @@ -972,20 +991,38 @@ func (h *AutoscanHandler) HandleStatus(w http.ResponseWriter, r *http.Request) { rewrites = []autoscan.PathRewrite{} } trimmed = append(trimmed, autoscanStatusSource{ - ID: src.ID, - InstallationID: src.InstallationID, - CapabilityID: src.CapabilityID, - ConnectionID: src.ConnectionID, - Enabled: src.Enabled, - Label: src.Label, - PathRewrites: rewrites, - LastRunAt: src.LastRunAt, - LastError: src.LastError, + ID: src.ID, + PluginID: src.PluginID, + CapabilityID: src.CapabilityID, + ConnectionID: src.ConnectionID, + Enabled: src.Enabled, + Label: src.Label, + PathRewrites: rewrites, + LastRunAt: src.LastRunAt, + LastError: src.LastError, + }) + } + now := time.Now() + runningPolls := make([]autoscanRunningPollResponse, 0, len(runningEvents)) + for _, event := range runningEvents { + elapsed := now.Sub(event.StartedAt) + if elapsed < 0 { + elapsed = 0 + } + runningPolls = append(runningPolls, autoscanRunningPollResponse{ + ID: event.ID, + SourceID: event.SourceID, + PluginID: event.PluginID, + CapabilityID: event.CapabilityID, + StartedAt: event.StartedAt, + ElapsedMS: elapsed.Milliseconds(), + MarkerBefore: event.MarkerBefore, }) } writeJSON(w, http.StatusOK, autoscanStatusResponse{ Enabled: s.Enabled, Sources: trimmed, + RunningPolls: runningPolls, ActiveScans: queue.Active, AcceptedScans: queue.Accepted, RunningScans: queue.Running, diff --git a/internal/api/handlers/autoscan_test.go b/internal/api/handlers/autoscan_test.go index a3bd0ed6..761aeb19 100644 --- a/internal/api/handlers/autoscan_test.go +++ b/internal/api/handlers/autoscan_test.go @@ -31,6 +31,7 @@ type fakeAutoscanStore struct { countScansFn func(autoscan.ScanListFilter) (int, error) listEventsFn func(autoscan.EventListFilter) ([]autoscan.EventWithRuns, error) countEventsFn func(autoscan.EventListFilter) (int, error) + listRunningFn func() ([]autoscan.Event, error) queueSummaryFn func() (autoscan.QueueSummary, error) latestEventAtFn func() (*time.Time, error) } @@ -140,6 +141,13 @@ func (f *fakeAutoscanStore) CountEvents(_ context.Context, filter autoscan.Event return 0, nil } +func (f *fakeAutoscanStore) ListRunningEvents(context.Context) ([]autoscan.Event, error) { + if f.listRunningFn != nil { + return f.listRunningFn() + } + return nil, nil +} + func (f *fakeAutoscanStore) GetQueueSummary(context.Context) (autoscan.QueueSummary, error) { if f.queueSummaryFn != nil { return f.queueSummaryFn() @@ -273,7 +281,7 @@ func TestAutoscanHandleListSourcesListsOnly(t *testing.T) { store := &fakeAutoscanStore{ listSourcesFn: func() ([]autoscan.Source, error) { listed = true - return []autoscan.Source{{ID: "src-1", InstallationID: 1, CapabilityID: "arr"}}, nil + return []autoscan.Source{{ID: "src-1", PluginID: "silo.autoscan.arr", CapabilityID: "arr"}}, nil }, } h := NewAutoscanHandler(store, &fakeAutoscanTriggerer{}) @@ -300,7 +308,7 @@ func TestAutoscanHandleListSourcesListsOnly(t *testing.T) { func TestAutoscanHandleListAvailableScanSources(t *testing.T) { trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{ - {InstallationID: 1, CapabilityID: "arr", PluginID: "sonarr", DisplayName: "Sonarr"}, + {PluginID: "sonarr", CapabilityID: "arr", DisplayName: "Sonarr"}, }} h := NewAutoscanHandler(&fakeAutoscanStore{}, trig) @@ -331,19 +339,19 @@ func TestAutoscanHandleCreateSourceSucceeds(t *testing.T) { }, } trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{ - {InstallationID: 1, CapabilityID: "arr"}, + {PluginID: "silo.autoscan.arr", CapabilityID: "arr"}, }} h := NewAutoscanHandler(store, trig) req := newAutoscanRequest("POST", "/api/v1/admin/autoscan/sources", - `{"installation_id":1,"capability_id":"arr","connection_id":"conn-1","enabled":true}`, "") + `{"plugin_id":"silo.autoscan.arr","capability_id":"arr","connection_id":"conn-1","enabled":true}`, "") rec := httptest.NewRecorder() h.HandleCreateSource(rec, req) if rec.Code != http.StatusCreated { t.Fatalf("status = %d, want 201; body=%s", rec.Code, rec.Body.String()) } - if got.InstallationID != 1 || got.CapabilityID != "arr" || got.ConnectionID == nil || *got.ConnectionID != "conn-1" { + if got.PluginID != "silo.autoscan.arr" || got.CapabilityID != "arr" || got.ConnectionID == nil || *got.ConnectionID != "conn-1" { t.Fatalf("unexpected created source: %+v", got) } var body autoscanSourceResponse @@ -363,14 +371,14 @@ func TestAutoscanHandleCreateSourceRejectsUnknownCapability(t *testing.T) { return s, nil }, } - // Installed set does NOT include (1, "arr"). + // Installed set does NOT include ("silo.autoscan.arr", "arr"). trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{ - {InstallationID: 2, CapabilityID: "other"}, + {PluginID: "silo.autoscan.other", CapabilityID: "other"}, }} h := NewAutoscanHandler(store, trig) req := newAutoscanRequest("POST", "/api/v1/admin/autoscan/sources", - `{"installation_id":1,"capability_id":"arr","enabled":false}`, "") + `{"plugin_id":"silo.autoscan.arr","capability_id":"arr","enabled":false}`, "") rec := httptest.NewRecorder() h.HandleCreateSource(rec, req) @@ -394,12 +402,12 @@ func TestAutoscanHandleCreateSourceEnableWithoutConnectionSucceeds(t *testing.T) }, } trig := &fakeAutoscanTriggerer{available: []autoscan.AvailableScanSource{ - {InstallationID: 1, CapabilityID: "arr"}, + {PluginID: "silo.autoscan.arr", CapabilityID: "arr"}, }} h := NewAutoscanHandler(store, trig) req := newAutoscanRequest("POST", "/api/v1/admin/autoscan/sources", - `{"installation_id":1,"capability_id":"arr","enabled":true}`, "") + `{"plugin_id":"silo.autoscan.arr","capability_id":"arr","enabled":true}`, "") rec := httptest.NewRecorder() h.HandleCreateSource(rec, req) @@ -615,7 +623,7 @@ func TestAutoscanHandleUpdateSourceEnableWithoutConnectionSucceeds(t *testing.T) var got autoscan.Source store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: nil}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { got = s @@ -641,7 +649,7 @@ func TestAutoscanHandleUpdateSourceEnableWithConnectionSucceeds(t *testing.T) { var got autoscan.Source store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { got = s @@ -669,7 +677,7 @@ func TestAutoscanHandleUpdateSourceBindConnectionSucceeds(t *testing.T) { store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { // Source starts with no connection. - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: nil}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { got = s @@ -697,7 +705,7 @@ func TestAutoscanHandleUpdateSourceUnbindConnectionSucceeds(t *testing.T) { store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { // Source starts with a connection already bound. - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { got = s @@ -727,7 +735,7 @@ func TestAutoscanHandleUpdateSourceUnbindWhileEnabledSucceeds(t *testing.T) { var got autoscan.Source store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { got = s @@ -925,7 +933,7 @@ func TestAutoscanHandleUpdateSourceRoundTripsPathRewrites(t *testing.T) { var got autoscan.Source store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { got = s @@ -990,7 +998,7 @@ func TestAutoscanHandleUpdateSourceRejectsBlankRewrite(t *testing.T) { upserted := false store := &fakeAutoscanStore{ getSourceFn: func(id string) (autoscan.Source, error) { - return autoscan.Source{ID: id, InstallationID: 1, CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil + return autoscan.Source{ID: id, PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: ptr("conn-1")}, nil }, updateSourceFn: func(s autoscan.Source) (autoscan.Source, error) { upserted = true @@ -1020,7 +1028,7 @@ func TestAutoscanHandleListScansSerializesFiltersAndEventContext(t *testing.T) { eventCompleted := completed.Add(-1 * time.Second) eventID := int64(42) sourceID := "src-1" - installationID := 7 + pluginID := "silo.autoscan.cephfs" var gotFilter autoscan.ScanListFilter var gotCountFilter autoscan.ScanListFilter store := &fakeAutoscanStore{ @@ -1040,7 +1048,7 @@ func TestAutoscanHandleListScansSerializesFiltersAndEventContext(t *testing.T) { }, AutoscanEventID: &eventID, SourceID: &sourceID, - InstallationID: &installationID, + PluginID: pluginID, CapabilityID: "cephfs", EventStatus: autoscan.EventStatusSuccess, EventCompletedAt: &eventCompleted, @@ -1084,7 +1092,7 @@ func TestAutoscanHandleListScansSerializesFiltersAndEventContext(t *testing.T) { if scan.AutoscanEventID == nil || *scan.AutoscanEventID != eventID || scan.EventStatus != "success" { t.Fatalf("event context = %+v", scan) } - if scan.SourceID == nil || *scan.SourceID != sourceID || scan.InstallationID == nil || *scan.InstallationID != installationID || scan.CapabilityID != "cephfs" { + if scan.SourceID == nil || *scan.SourceID != sourceID || scan.PluginID != pluginID || scan.CapabilityID != "cephfs" { t.Fatalf("source context = %+v", scan) } } @@ -1128,7 +1136,7 @@ func TestAutoscanHandleListEventsSerializesFiltersAndRuns(t *testing.T) { Event: autoscan.Event{ ID: 42, SourceID: &sourceID, - InstallationID: 7, + PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", StartedAt: started, CompletedAt: completed, @@ -1258,18 +1266,30 @@ func TestAutoscanHandleUpdateSourceNormalizesLabel(t *testing.T) { func TestAutoscanHandleStatusReturnsTrimmedSources(t *testing.T) { latestEventAt := time.Date(2026, 6, 4, 14, 30, 0, 0, time.UTC) + runningStartedAt := time.Now().Add(-90 * time.Second) + sourceID := "src-1" store := &fakeAutoscanStore{ getSettingsFn: func() (autoscan.Settings, error) { return autoscan.Settings{Enabled: true}, nil }, listSourcesFn: func() ([]autoscan.Source, error) { return []autoscan.Source{ - {ID: "src-1", InstallationID: 7, CapabilityID: "scan_source", ConnectionID: ptr("conn-1"), Enabled: true}, + {ID: "src-1", PluginID: "silo.autoscan.cephfs", CapabilityID: "scan_source", ConnectionID: ptr("conn-1"), Enabled: true}, }, nil }, queueSummaryFn: func() (autoscan.QueueSummary, error) { return autoscan.QueueSummary{Active: 3, Accepted: 2, Running: 1}, nil }, + listRunningFn: func() ([]autoscan.Event, error) { + return []autoscan.Event{{ + ID: 44, + SourceID: &sourceID, + PluginID: "silo.autoscan.cephfs", + CapabilityID: "cephfs", + StartedAt: runningStartedAt, + Status: autoscan.EventStatusRunning, + }}, nil + }, latestEventAtFn: func() (*time.Time, error) { return &latestEventAt, nil }, @@ -1296,6 +1316,16 @@ func TestAutoscanHandleStatusReturnsTrimmedSources(t *testing.T) { if body.ActiveScans != 3 || body.AcceptedScans != 2 || body.RunningScans != 1 { t.Errorf("queue summary = %+v", body) } + if len(body.RunningPolls) != 1 { + t.Fatalf("running_polls = %+v", body.RunningPolls) + } + poll := body.RunningPolls[0] + if poll.ID != 44 || poll.SourceID == nil || *poll.SourceID != sourceID || poll.PluginID != "silo.autoscan.cephfs" || poll.CapabilityID != "cephfs" { + t.Fatalf("running poll identity = %+v", poll) + } + if poll.ElapsedMS <= 0 { + t.Fatalf("running poll elapsed_ms = %d, want > 0", poll.ElapsedMS) + } if body.LatestEventAt == nil || !body.LatestEventAt.Equal(latestEventAt) { t.Errorf("latest_event_at = %v, want %v", body.LatestEventAt, latestEventAt) } diff --git a/internal/autoscan/discovery.go b/internal/autoscan/discovery.go index 71bf3e20..4a66d429 100644 --- a/internal/autoscan/discovery.go +++ b/internal/autoscan/discovery.go @@ -9,19 +9,17 @@ import ( // enriched with the metadata the Add-source picker needs (plugin id + a // human-friendly display name). type DiscoveredSource struct { - InstallationID int - CapabilityID string // PluginID is the installation's plugin id (e.g. "sonarr"); empty when the // lister cannot supply it. - PluginID string + PluginID string + CapabilityID string // DisplayName is a human-friendly label for the capability (from the // capability's manifest display_name, falling back to plugin/capability ids). DisplayName string } // ScanSourceLister enumerates every installed scan_source.v1 capability so the -// engine can (a) offer them in the Add-source picker and (b) detect orphaned -// source rows whose plugin has been uninstalled. +// engine can offer them in the Add-source picker. type ScanSourceLister interface { // ListScanSources returns one entry per installed scan_source.v1 capability, // enriched with plugin id + display name. @@ -31,17 +29,9 @@ type ScanSourceLister interface { // AvailableScanSource is one installed scan_source capability an operator can // create a source against (the Add-source picker list). type AvailableScanSource struct { - InstallationID int `json:"installation_id"` - CapabilityID string `json:"capability_id"` - PluginID string `json:"plugin_id"` - DisplayName string `json:"display_name"` -} - -// installedKey identifies an installed scan_source capability for set -// membership tests (orphan detection in PollOnce). -type installedKey struct { - InstallationID int - CapabilityID string + PluginID string `json:"plugin_id"` + CapabilityID string `json:"capability_id"` + DisplayName string `json:"display_name"` } // ListAvailableScanSources enumerates every installed scan_source capability so @@ -59,30 +49,10 @@ func (s *Service) ListAvailableScanSources(ctx context.Context) ([]AvailableScan out := make([]AvailableScanSource, 0, len(discovered)) for _, d := range discovered { out = append(out, AvailableScanSource{ - InstallationID: d.InstallationID, - CapabilityID: d.CapabilityID, - PluginID: d.PluginID, - DisplayName: d.DisplayName, + PluginID: d.PluginID, + CapabilityID: d.CapabilityID, + DisplayName: d.DisplayName, }) } return out, nil } - -// installedScanSources returns the set of currently-installed scan_source -// capabilities so PollOnce can skip orphaned source rows (their plugin is -// gone). A nil set means the set is unavailable (no lister) — callers must treat -// that as "discovery unavailable" and NOT prune. -func (s *Service) installedScanSources(ctx context.Context) (map[installedKey]struct{}, error) { - if s.lister == nil { - return nil, nil - } - discovered, err := s.lister.ListScanSources(ctx) - if err != nil { - return nil, fmt.Errorf("list scan sources: %w", err) - } - present := make(map[installedKey]struct{}, len(discovered)) - for _, d := range discovered { - present[installedKey{InstallationID: d.InstallationID, CapabilityID: d.CapabilityID}] = struct{}{} - } - return present, nil -} diff --git a/internal/autoscan/discovery_test.go b/internal/autoscan/discovery_test.go index 1314732c..630d22aa 100644 --- a/internal/autoscan/discovery_test.go +++ b/internal/autoscan/discovery_test.go @@ -14,8 +14,8 @@ func (f fakeLister) ListScanSources(context.Context) ([]DiscoveredSource, error) func TestListAvailableScanSourcesEnumeratesInstalled(t *testing.T) { lister := fakeLister{sources: []DiscoveredSource{ - {InstallationID: 1, CapabilityID: "arr-a", PluginID: "sonarr", DisplayName: "Sonarr"}, - {InstallationID: 2, CapabilityID: "arr-b", PluginID: "radarr", DisplayName: "Radarr"}, + {PluginID: "sonarr", CapabilityID: "arr-a", DisplayName: "Sonarr"}, + {PluginID: "radarr", CapabilityID: "arr-b", DisplayName: "Radarr"}, }} svc := &Service{lister: lister} @@ -26,7 +26,7 @@ func TestListAvailableScanSourcesEnumeratesInstalled(t *testing.T) { if len(available) != 2 { t.Fatalf("expected 2 available, got %d: %+v", len(available), available) } - if available[0] != (AvailableScanSource{InstallationID: 1, CapabilityID: "arr-a", PluginID: "sonarr", DisplayName: "Sonarr"}) { + if available[0] != (AvailableScanSource{PluginID: "sonarr", CapabilityID: "arr-a", DisplayName: "Sonarr"}) { t.Fatalf("unexpected first available: %+v", available[0]) } } @@ -41,32 +41,3 @@ func TestListAvailableScanSourcesNilListerEmpty(t *testing.T) { t.Fatalf("nil lister must return empty, got %+v", available) } } - -func TestInstalledScanSourcesSetMembership(t *testing.T) { - lister := fakeLister{sources: []DiscoveredSource{ - {InstallationID: 1, CapabilityID: "arr-a"}, - }} - svc := &Service{lister: lister} - - present, err := svc.installedScanSources(context.Background()) - if err != nil { - t.Fatalf("installedScanSources: %v", err) - } - if _, ok := present[installedKey{1, "arr-a"}]; !ok { - t.Fatalf("expected 1/arr-a present: %+v", present) - } - if _, ok := present[installedKey{2, "arr-b"}]; ok { - t.Fatalf("did not expect 2/arr-b present: %+v", present) - } -} - -func TestInstalledScanSourcesNilListerNilSet(t *testing.T) { - svc := &Service{lister: nil} - present, err := svc.installedScanSources(context.Background()) - if err != nil { - t.Fatalf("installedScanSources: %v", err) - } - if present != nil { - t.Fatalf("nil lister must return a nil set, got %+v", present) - } -} diff --git a/internal/autoscan/errors.go b/internal/autoscan/errors.go index 3d720516..4df038a3 100644 --- a/internal/autoscan/errors.go +++ b/internal/autoscan/errors.go @@ -5,3 +5,7 @@ import "errors" // ErrNotFound is returned when an autoscan operation references a connection or // source that does not exist. The admin API surfaces it as a 404. var ErrNotFound = errors.New("autoscan: not found") + +// ErrPollAlreadyRunning is returned when a source already has an active poll +// event. Callers should treat it as a no-op skip, not a source failure. +var ErrPollAlreadyRunning = errors.New("autoscan: source poll already running") diff --git a/internal/autoscan/probe_test.go b/internal/autoscan/probe_test.go index 3410f984..7d21952b 100644 --- a/internal/autoscan/probe_test.go +++ b/internal/autoscan/probe_test.go @@ -69,7 +69,7 @@ func TestTestConnectionNoProbeConfigured(t *testing.T) { func TestSuggestRewritesMatchesArrRootsToSiloFolders(t *testing.T) { store := &fakeStore{ sources: []Source{ - {ID: "src-1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1")}, + {ID: "src-1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1")}, }, } arr := &fakeArrClient{roots: []string{"/data/tv"}} @@ -87,7 +87,7 @@ func TestSuggestRewritesMatchesArrRootsToSiloFolders(t *testing.T) { func TestSuggestRewritesNoConnection(t *testing.T) { store := &fakeStore{ - sources: []Source{{ID: "src-1", InstallationID: 1, CapabilityID: "arr", ConnectionID: nil}}, + sources: []Source{{ID: "src-1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil}}, } svc := &Service{store: store, connres: passthroughConnRes{}, rootFolders: &fakeArrClient{}, folders: fakeFolderLister{}} diff --git a/internal/autoscan/provider.go b/internal/autoscan/provider.go index 7a47a05a..ca8421de 100644 --- a/internal/autoscan/provider.go +++ b/internal/autoscan/provider.go @@ -9,7 +9,7 @@ import ( // ScanSourceProvider yields changed paths for one source. The engine calls // PollChanges; production wraps the plugins.Service scan_source resolver. type ScanSourceProvider interface { - PollChanges(ctx context.Context, installationID int, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) (changes []Change, nextMarker string, err error) + PollChanges(ctx context.Context, pluginID, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) (changes []Change, nextMarker string, err error) } // PollChangesClient is the slice of *pluginhost.ScanSourceClient used here. It @@ -20,10 +20,10 @@ type PollChangesClient interface { PollChanges(ctx context.Context, req *pluginv1.PollChangesRequest) (*pluginv1.PollChangesResponse, error) } -// ScanSourceResolver yields a per-(installation, capability) scan-source client. +// ScanSourceResolver yields a per-(plugin, capability) scan-source client. // Exported for the same cross-package adapter reason as PollChangesClient. type ScanSourceResolver interface { - ScanSourceClient(ctx context.Context, installationID int, capabilityID string) (PollChangesClient, error) + ScanSourceClient(ctx context.Context, pluginID, capabilityID string) (PollChangesClient, error) } type pluginProvider struct{ resolver ScanSourceResolver } @@ -34,8 +34,8 @@ func NewPluginProvider(resolver ScanSourceResolver) ScanSourceProvider { return &pluginProvider{resolver: resolver} } -func (p *pluginProvider) PollChanges(ctx context.Context, installationID int, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) { - client, err := p.resolver.ScanSourceClient(ctx, installationID, capabilityID) +func (p *pluginProvider) PollChanges(ctx context.Context, pluginID, capabilityID, marker string, conn ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) { + client, err := p.resolver.ScanSourceClient(ctx, pluginID, capabilityID) if err != nil { return nil, "", err } diff --git a/internal/autoscan/provider_test.go b/internal/autoscan/provider_test.go index 649e9a7d..87c10755 100644 --- a/internal/autoscan/provider_test.go +++ b/internal/autoscan/provider_test.go @@ -17,10 +17,17 @@ func (c *capturingClient) PollChanges(_ context.Context, req *pluginv1.PollChang return &pluginv1.PollChangesResponse{SourcePaths: []string{"/mnt/media/x"}, NextMarker: "m2"}, nil } -// capturingResolver yields a fixed PollChangesClient. -type capturingResolver struct{ client PollChangesClient } +// capturingResolver yields a fixed PollChangesClient and records the requested +// stable scan-source identity. +type capturingResolver struct { + client PollChangesClient + lastPluginID string + lastCapID string +} -func (r capturingResolver) ScanSourceClient(context.Context, int, string) (PollChangesClient, error) { +func (r *capturingResolver) ScanSourceClient(_ context.Context, pluginID, capabilityID string) (PollChangesClient, error) { + r.lastPluginID = pluginID + r.lastCapID = capabilityID return r.client, nil } @@ -28,11 +35,12 @@ func (r capturingResolver) ScanSourceClient(context.Context, int, string) (PollC // delivered to the plugin on the PollChangesRequest. func TestPluginProviderPopulatesConnection(t *testing.T) { client := &capturingClient{} - prov := NewPluginProvider(capturingResolver{client}) + resolver := &capturingResolver{client: client} + prov := NewPluginProvider(resolver) conn := ResolvedConnection{BaseURL: "https://arr.example", APIKey: "secret-key"} sourceConfig := map[string]string{"exclusions": ".downloads"} - changes, next, err := prov.PollChanges(context.Background(), 1, "cap", "m1", conn, sourceConfig) + changes, next, err := prov.PollChanges(context.Background(), "silo.autoscan.arr", "cap", "m1", conn, sourceConfig) if err != nil { t.Fatalf("PollChanges: %v", err) } @@ -45,6 +53,9 @@ func TestPluginProviderPopulatesConnection(t *testing.T) { if client.last.GetCapabilityId() != "cap" || client.last.GetMarker() != "m1" { t.Fatalf("unexpected request fields: cap=%q marker=%q", client.last.GetCapabilityId(), client.last.GetMarker()) } + if resolver.lastPluginID != "silo.autoscan.arr" || resolver.lastCapID != "cap" { + t.Fatalf("resolver identity = %q/%q", resolver.lastPluginID, resolver.lastCapID) + } rc := client.last.GetConnection() if rc == nil { t.Fatal("expected connection to be populated on the request") @@ -59,9 +70,9 @@ func TestPluginProviderPopulatesConnection(t *testing.T) { func TestPluginProviderPrefersStructuredChanges(t *testing.T) { client := &capturingStructuredClient{} - prov := NewPluginProvider(capturingResolver{client: client}) + prov := NewPluginProvider(&capturingResolver{client: client}) - changes, next, err := prov.PollChanges(context.Background(), 1, "cap", "m1", ResolvedConnection{}, nil) + changes, next, err := prov.PollChanges(context.Background(), "silo.autoscan.arr", "cap", "m1", ResolvedConnection{}, nil) if err != nil { t.Fatalf("PollChanges: %v", err) } diff --git a/internal/autoscan/repository.go b/internal/autoscan/repository.go index ecd4359d..2e042737 100644 --- a/internal/autoscan/repository.go +++ b/internal/autoscan/repository.go @@ -175,14 +175,14 @@ func (r *Repository) GetConnection(ctx context.Context, id string) (Connection, // --- Sources --- -const sourceColumns = `id, installation_id, capability_id, connection_id, enabled, +const sourceColumns = `id, plugin_id, capability_id, connection_id, enabled, poll_interval_seconds, path_rewrites, source_config, label, marker, last_run_at, last_error` func scanSource(row interface{ Scan(...any) error }) (Source, error) { var s Source var pathRewrites []byte var sourceConfig []byte - if err := row.Scan(&s.ID, &s.InstallationID, &s.CapabilityID, &s.ConnectionID, + if err := row.Scan(&s.ID, &s.PluginID, &s.CapabilityID, &s.ConnectionID, &s.Enabled, &s.PollIntervalSeconds, &pathRewrites, &sourceConfig, &s.Label, &s.Marker, &s.LastRunAt, &s.LastError); err != nil { return Source{}, err } @@ -283,11 +283,11 @@ func (r *Repository) CreateSource(ctx context.Context, s Source) (Source, error) } row := r.pool.QueryRow(ctx, ` INSERT INTO autoscan_sources ( - installation_id, capability_id, connection_id, enabled, poll_interval_seconds, path_rewrites, source_config, label + plugin_id, capability_id, connection_id, enabled, poll_interval_seconds, path_rewrites, source_config, label ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING `+sourceColumns, - s.InstallationID, s.CapabilityID, connectionIDArg(s.ConnectionID), s.Enabled, s.PollIntervalSeconds, rewrites, sourceConfig, s.Label) + s.PluginID, s.CapabilityID, connectionIDArg(s.ConnectionID), s.Enabled, s.PollIntervalSeconds, rewrites, sourceConfig, s.Label) out, err := scanSource(row) if err != nil { if connID, ok := connectionFKViolation(err, s.ConnectionID); ok { @@ -299,7 +299,7 @@ func (r *Repository) CreateSource(ctx context.Context, s Source) (Source, error) } // UpdateSource updates a source's binding/scheduling fields by id. Identity -// (installation_id, capability_id) and bookkeeping fields (marker/last_run_at/ +// (plugin_id, capability_id) and bookkeeping fields (marker/last_run_at/ // last_error) are left untouched. An unknown id maps to ErrNotFound; a // non-existent connection trips the FK constraint and also maps to ErrNotFound. func (r *Repository) UpdateSource(ctx context.Context, s Source) (Source, error) { @@ -353,7 +353,7 @@ func connectionFKViolation(err error, connectionID *string) (string, bool) { func (r *Repository) ListSources(ctx context.Context) ([]Source, error) { rows, err := r.pool.Query(ctx, `SELECT `+sourceColumns+` - FROM autoscan_sources ORDER BY installation_id, capability_id`) + FROM autoscan_sources ORDER BY plugin_id, capability_id`) if err != nil { return nil, fmt.Errorf("list autoscan sources: %w", err) } @@ -364,7 +364,7 @@ func (r *Repository) ListSources(ctx context.Context) ([]Source, error) { func (r *Repository) ListEnabledSources(ctx context.Context) ([]Source, error) { rows, err := r.pool.Query(ctx, `SELECT `+sourceColumns+` FROM autoscan_sources WHERE enabled = true - ORDER BY installation_id, capability_id`) + ORDER BY plugin_id, capability_id`) if err != nil { return nil, fmt.Errorf("list enabled autoscan sources: %w", err) } @@ -462,7 +462,7 @@ func (r *Repository) RecordError(ctx context.Context, sourceID, msg string) erro return nil } -const eventColumns = `id, source_id, installation_id, capability_id, started_at, completed_at, +const eventColumns = `id, source_id, plugin_id, capability_id, started_at, completed_at, duration_ms, status, changes_returned, changes_resolved, targets_claimed, scans_created, scans_reused, scans_suppressed, error_message, marker_before, marker_after` @@ -472,7 +472,7 @@ func scanEvent(row interface{ Scan(...any) error }) (Event, error) { if err := row.Scan( &e.ID, &e.SourceID, - &e.InstallationID, + &e.PluginID, &e.CapabilityID, &e.StartedAt, &e.CompletedAt, @@ -499,25 +499,80 @@ func (r *Repository) CreateEvent(ctx context.Context, in EventCreate) (int64, er if started.IsZero() { started = time.Now() } - row := r.pool.QueryRow(ctx, ` + sourceID := strings.TrimSpace(in.SourceID) + if sourceID == "" { + row := r.pool.QueryRow(ctx, ` + INSERT INTO autoscan_events ( + source_id, plugin_id, capability_id, started_at, completed_at, + duration_ms, status, error_message, marker_before + ) + VALUES ($1, $2, $3, $4, $4, 0, $5, $6, $7) + RETURNING id`, + nil, + in.PluginID, + in.CapabilityID, + started, + string(EventStatusRunning), + "", + nullable(in.MarkerBefore), + ) + var id int64 + if err := row.Scan(&id); err != nil { + return 0, fmt.Errorf("create autoscan event: %w", err) + } + return id, nil + } + + tx, err := r.pool.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("begin autoscan event: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(900173, hashtext($1))`, sourceID); err != nil { + return 0, fmt.Errorf("lock autoscan event: %w", err) + } + + var runningID int64 + err = tx.QueryRow(ctx, ` + SELECT id + FROM autoscan_events + WHERE source_id = $1 + AND status = $2 + ORDER BY started_at ASC, id ASC + LIMIT 1`, + sourceID, + string(EventStatusRunning), + ).Scan(&runningID) + if err == nil { + return 0, fmt.Errorf("%w: source %s event %d", ErrPollAlreadyRunning, sourceID, runningID) + } + if !errors.Is(err, pgx.ErrNoRows) { + return 0, fmt.Errorf("check running autoscan event: %w", err) + } + + row := tx.QueryRow(ctx, ` INSERT INTO autoscan_events ( - source_id, installation_id, capability_id, started_at, completed_at, + source_id, plugin_id, capability_id, started_at, completed_at, duration_ms, status, error_message, marker_before ) VALUES ($1, $2, $3, $4, $4, 0, $5, $6, $7) RETURNING id`, - nullable(in.SourceID), - in.InstallationID, + sourceID, + in.PluginID, in.CapabilityID, started, - string(EventStatusError), - "poll started but did not finish", + string(EventStatusRunning), + "", nullable(in.MarkerBefore), ) var id int64 if err := row.Scan(&id); err != nil { return 0, fmt.Errorf("create autoscan event: %w", err) } + if err := tx.Commit(ctx); err != nil { + return 0, fmt.Errorf("commit autoscan event: %w", err) + } return id, nil } @@ -565,6 +620,25 @@ func (r *Repository) FinishEvent(ctx context.Context, in EventFinish) error { return nil } +func (r *Repository) MarkInterruptedEvents(ctx context.Context) error { + msg := "poll started but did not finish" + _, err := r.pool.Exec(ctx, ` + UPDATE autoscan_events + SET completed_at = now(), + duration_ms = GREATEST(0, EXTRACT(EPOCH FROM (now() - started_at)) * 1000)::bigint, + status = $1, + error_message = $2 + WHERE status = $3`, + string(EventStatusError), + msg, + string(EventStatusRunning), + ) + if err != nil { + return fmt.Errorf("mark interrupted autoscan events: %w", err) + } + return nil +} + // clampAutoscanLimit bounds a requested page size to a sane window: the // default page when unset, capped so a single query can never fan out. func clampAutoscanLimit(limit int) int { @@ -758,6 +832,30 @@ func (r *Repository) ListEvents(ctx context.Context, filter EventListFilter) ([] return events, runRows.Err() } +func (r *Repository) ListRunningEvents(ctx context.Context) ([]Event, error) { + rows, err := r.pool.Query(ctx, ` + SELECT `+eventColumns+` + FROM autoscan_events + WHERE status = $1 + ORDER BY started_at ASC, id ASC`, + string(EventStatusRunning), + ) + if err != nil { + return nil, fmt.Errorf("list running autoscan events: %w", err) + } + defer rows.Close() + + events := make([]Event, 0) + for rows.Next() { + event, scanErr := scanEvent(rows) + if scanErr != nil { + return nil, scanErr + } + events = append(events, event) + } + return events, rows.Err() +} + func (r *Repository) ListAutoscanScans(ctx context.Context, filter ScanListFilter) ([]ScanWithEvent, error) { limit := clampAutoscanLimit(filter.Limit) offset := filter.Offset @@ -780,15 +878,15 @@ func (r *Repository) ListAutoscanScans(ctx context.Context, filter ScanListFilte sr.trigger, sr.status, COALESCE(sr.error_message, ''), - sr.requested_at, - sr.started_at, - sr.completed_at, - sr.autoscan_event_id, - e.source_id, - e.installation_id, - COALESCE(e.capability_id, ''), - COALESCE(e.status, ''), - e.completed_at + sr.requested_at, + sr.started_at, + sr.completed_at, + sr.autoscan_event_id, + e.source_id, + COALESCE(e.plugin_id, ''), + COALESCE(e.capability_id, ''), + COALESCE(e.status, ''), + e.completed_at FROM scan_runs sr LEFT JOIN autoscan_events e ON e.id = sr.autoscan_event_id WHERE `+strings.Join(clauses, " AND ")+` @@ -818,7 +916,7 @@ func (r *Repository) ListAutoscanScans(ctx context.Context, filter ScanListFilte &scan.CompletedAt, &scan.AutoscanEventID, &scan.SourceID, - &scan.InstallationID, + &scan.PluginID, &scan.CapabilityID, &eventStatus, &scan.EventCompletedAt, @@ -853,7 +951,7 @@ func (r *Repository) GetQueueSummary(ctx context.Context) (QueueSummary, error) func (r *Repository) LatestEventAt(ctx context.Context) (*time.Time, error) { var latest *time.Time - err := r.pool.QueryRow(ctx, `SELECT max(completed_at) FROM autoscan_events`).Scan(&latest) + err := r.pool.QueryRow(ctx, `SELECT max(completed_at) FROM autoscan_events WHERE status <> $1`, string(EventStatusRunning)).Scan(&latest) if err != nil { return nil, fmt.Errorf("get latest autoscan event time: %w", err) } diff --git a/internal/autoscan/service.go b/internal/autoscan/service.go index 7821adfc..391d4ddd 100644 --- a/internal/autoscan/service.go +++ b/internal/autoscan/service.go @@ -91,8 +91,7 @@ func (s *Service) SetSuggesterDeps(rootFolders RootFolderClient, folders FolderL // from scan_source plugins; connres resolves a source's connection to concrete // credentials; resolver/queue/suppress drive the resolve→suppress→enqueue loop. // lister enumerates installed scan_source capabilities so the Add-source picker -// can offer them and PollOnce can skip orphaned source rows; it may be nil -// (the picker then returns an empty list and orphan-pruning is disabled). +// can offer them; it may be nil (the picker then returns an empty list). func NewService( store Store, provider ScanSourceProvider, @@ -113,24 +112,15 @@ func NewService( } } -// PollOnce runs one autoscan cycle. Per-source failures are logged and skipped; -// only settings/listing errors propagate. The opaque next marker returned by the +// PollOnce runs one autoscan cycle. Per-source failures are logged, recorded on +// the source/event, and the loop continues; only settings/listing errors +// propagate. The opaque next marker returned by the // provider is stored verbatim, but only when the cycle's work is genuinely // consumed: the provider returned no paths, or it returned paths and at least one // resolved+enqueued. When paths come back but NONE resolve to a library folder // (e.g. a freshly-enabled source with unconfigured rewrites) the marker is held // and an error recorded, so those imports aren't skipped forever. func (s *Service) PollOnce(ctx context.Context) error { - // Fetch the set of currently-installed scan_source capabilities so we can skip - // orphaned source rows (their plugin was uninstalled). Listing failures are - // non-fatal: a nil set means the installed set is unavailable, in which case - // we do NOT prune orphans — every enabled source is assumed present. - present, derr := s.installedScanSources(ctx) - if derr != nil { - slog.WarnContext(ctx, "autoscan: list installed scan sources failed", "err", derr) - present = nil - } - settings, err := s.store.GetSettings(ctx) if err != nil { return err @@ -146,16 +136,6 @@ func (s *Service) PollOnce(ctx context.Context) error { now := time.Now() for _, src := range sources { - // Skip orphaned sources: an enabled source whose scan_source plugin has - // been uninstalled/disabled is no longer in the discovered set, so polling - // it would error every cycle. Skip it quietly (no RecordError) so the spam - // stops; the operator can delete it via the source delete endpoint. A nil - // `present` set means discovery is unavailable — don't prune in that case. - if present != nil { - if _, ok := present[installedKey{InstallationID: src.InstallationID, CapabilityID: src.CapabilityID}]; !ok { - continue - } - } // Honor the per-source poll interval as a "poll at most every N seconds" // floor: the global task fires at the default cadence, so a source with a // longer interval is skipped until enough time has elapsed. @@ -170,7 +150,10 @@ func (s *Service) PollOnce(ctx context.Context) error { if src.Marker != nil { marker = *src.Marker } - eventID := s.createEvent(ctx, src, marker, time.Now()) + eventID, started := s.createEvent(ctx, src, marker, time.Now()) + if !started { + continue + } // A connection is OPTIONAL. Server-based providers (Sonarr/Radarr) bind a // connection and the resolved {base_url, api_key} is handed to the plugin. // Other providers (e.g. a filesystem/CephFS watcher) need none and get an @@ -195,7 +178,7 @@ func (s *Service) PollOnce(ctx context.Context) error { } conn = resolved } - changes, next, perr := s.provider.PollChanges(ctx, src.InstallationID, src.CapabilityID, marker, conn, src.SourceConfig) + changes, next, perr := s.provider.PollChanges(ctx, src.PluginID, src.CapabilityID, marker, conn, src.SourceConfig) if perr != nil { slog.WarnContext(ctx, "autoscan: poll changes failed", "source_id", src.ID, "err", perr) if rerr := s.store.RecordError(ctx, src.ID, perr.Error()); rerr != nil { @@ -295,22 +278,26 @@ func (s *Service) PollOnce(ctx context.Context) error { return nil } -func (s *Service) createEvent(ctx context.Context, src Source, marker string, startedAt time.Time) int64 { +func (s *Service) createEvent(ctx context.Context, src Source, marker string, startedAt time.Time) (int64, bool) { if s == nil || s.store == nil { - return 0 + return 0, true } id, err := s.store.CreateEvent(ctx, EventCreate{ - SourceID: src.ID, - InstallationID: src.InstallationID, - CapabilityID: src.CapabilityID, - StartedAt: startedAt, - MarkerBefore: marker, + SourceID: src.ID, + PluginID: src.PluginID, + CapabilityID: src.CapabilityID, + StartedAt: startedAt, + MarkerBefore: marker, }) if err != nil { + if errors.Is(err, ErrPollAlreadyRunning) { + slog.DebugContext(ctx, "autoscan: source poll already running", "source_id", src.ID) + return 0, false + } slog.WarnContext(ctx, "autoscan: create event failed", "source_id", src.ID, "err", err) - return 0 + return 0, true } - return id + return id, true } func (s *Service) finishEvent(ctx context.Context, eventID int64, finish EventFinish) { diff --git a/internal/autoscan/service_test.go b/internal/autoscan/service_test.go index a797cbe6..18343a20 100644 --- a/internal/autoscan/service_test.go +++ b/internal/autoscan/service_test.go @@ -2,6 +2,7 @@ package autoscan import ( "context" + "errors" "path/filepath" "strings" "testing" @@ -12,13 +13,14 @@ import ( ) type fakeStore struct { - settings Settings - sources []Source - connection Connection - advanced map[string]string // source ID -> marker - recorded map[string]string // source ID -> error message - createdEvents []EventCreate - events []EventFinish + settings Settings + sources []Source + connection Connection + advanced map[string]string // source ID -> marker + recorded map[string]string // source ID -> error message + createdEvents []EventCreate + events []EventFinish + createEventErr error } func (f *fakeStore) GetSettings(context.Context) (Settings, error) { return f.settings, nil } @@ -51,6 +53,9 @@ func (f *fakeStore) RecordError(_ context.Context, sourceID, msg string) error { return nil } func (f *fakeStore) CreateEvent(_ context.Context, event EventCreate) (int64, error) { + if f.createEventErr != nil { + return 0, f.createEventErr + } f.createdEvents = append(f.createdEvents, event) return int64(len(f.createdEvents)), nil } @@ -65,14 +70,20 @@ type fakeProvider struct { changes map[string][]Change // key: capabilityID, structured changes nextMarker string err error + errByCap map[string]error lastConfig map[string]string + calls int } -func (f *fakeProvider) PollChanges(_ context.Context, _ int, capabilityID, _ string, _ ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) { +func (f *fakeProvider) PollChanges(_ context.Context, _ string, capabilityID, _ string, _ ResolvedConnection, sourceConfig map[string]string) ([]Change, string, error) { + f.calls++ f.lastConfig = sourceConfig if f.err != nil { return nil, "", f.err } + if err := f.errByCap[capabilityID]; err != nil { + return nil, "", err + } if changes, ok := f.changes[capabilityID]; ok { return changes, f.nextMarker, nil } @@ -87,7 +98,7 @@ func TestPollOncePassesSourceConfigToProvider(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "cephfs", Enabled: true, + ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true, SourceConfig: map[string]string{"exclusions": ".downloads\n.recyclebin"}, }}, } @@ -101,6 +112,38 @@ func TestPollOncePassesSourceConfigToProvider(t *testing.T) { } } +func TestPollOnceSkipsSourceWhenPollAlreadyRunning(t *testing.T) { + store := &fakeStore{ + settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, + createEventErr: ErrPollAlreadyRunning, + sources: []Source{{ + ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true, + }}, + } + prov := &fakeProvider{paths: map[string][]string{"cephfs": {"/mnt/media/Movie/movie.mkv"}}, nextMarker: "m1"} + q := &recordingQueuer{} + svc := newService(store, prov, q, allowSuppressor{}) + + if err := svc.PollOnce(context.Background()); err != nil { + t.Fatalf("PollOnce: %v", err) + } + if prov.calls != 0 { + t.Fatalf("provider calls = %d, want 0", prov.calls) + } + if len(q.enqueued) != 0 { + t.Fatalf("enqueued scans = %+v, want none", q.enqueued) + } + if len(store.events) != 0 { + t.Fatalf("finished events = %+v, want none", store.events) + } + if len(store.recorded) != 0 { + t.Fatalf("recorded source errors = %+v, want none", store.recorded) + } + if len(store.advanced) != 0 { + t.Fatalf("advanced markers = %+v, want none", store.advanced) + } +} + // passthroughConnRes resolves to empty credentials; the engine doesn't inspect // them in tests (the fake provider ignores conn). type passthroughConnRes struct{} @@ -210,7 +253,7 @@ func TestPollOnceEnqueuesDedupedFolders(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, }}, } prov := &fakeProvider{paths: map[string][]string{ @@ -241,7 +284,7 @@ func TestPollOnceRecordsSuccessfulEvent(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, Marker: &marker, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, Marker: &marker, }}, } prov := &fakeProvider{paths: map[string][]string{ @@ -283,7 +326,7 @@ func TestPollOnceRecordsReusedScanCounts(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, }}, } prov := &fakeProvider{paths: map[string][]string{ @@ -318,7 +361,7 @@ func TestPollOnceAppliesSourceRewritesBeforeEnqueue(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, PathRewrites: []PathRewrite{{From: "/data/tv", To: "/mnt/media/tv"}}, }}, } @@ -347,7 +390,7 @@ func TestPollOnceStructuredFileChangeEnqueuesExactFile(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "cephfs", Enabled: true, + ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true, PathRewrites: []PathRewrite{{From: "/ceph/tv", To: "/mnt/media/tv"}}, }}, } @@ -380,7 +423,7 @@ func TestPollOnceStructuredSubtreeChangeEnqueuesExactSubtree(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "cephfs", Enabled: true, + ID: "s1", PluginID: "silo.autoscan.cephfs", CapabilityID: "cephfs", Enabled: true, PathRewrites: []PathRewrite{{From: "/ceph/movies", To: "/mnt/media/movies"}}, }}, } @@ -416,7 +459,7 @@ func TestPollOnceScansDistinctPathsUnderSameFolder(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, }}, } prov := &fakeProvider{paths: map[string][]string{ @@ -446,7 +489,7 @@ func TestPollOnceStoresOpaqueMarkerVerbatim(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, }}, } const opaque = "eyJjdXJzb3IiOiJhYmMxMjMifQ==|2026-06-02T14:10:00Z" @@ -470,7 +513,7 @@ func TestPollOnceHoldsMarkerWhenPathsReturnedButNoneResolve(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, }}, } prov := &fakeProvider{paths: map[string][]string{ @@ -523,7 +566,7 @@ func TestPollOnceAdvancesMarkerWhenResolvedButAllSuppressed(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, // rewrite /data/tv -> /mnt/media/tv so fakeResolver resolves the paths. PathRewrites: []PathRewrite{{From: "/data/tv", To: "/mnt/media/tv"}}, }}, @@ -556,7 +599,7 @@ func TestPollOnceAdvancesMarkerWhenZeroPathsReturned(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, }}, } prov := &fakeProvider{paths: map[string][]string{"arr": {}}, nextMarker: "m1"} @@ -588,7 +631,7 @@ func TestPollOnceDisabledNoop(t *testing.T) { func TestPollOnceProviderErrorKeepsMarker(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, - sources: []Source{{ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}}, + sources: []Source{{ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}}, } prov := &fakeProvider{err: context.DeadlineExceeded} q := &recordingQueuer{} @@ -607,7 +650,7 @@ func TestPollOnceProviderErrorKeepsMarker(t *testing.T) { func TestPollOnceReleasesClaimOnEnqueueFailure(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, - sources: []Source{{ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}}, + sources: []Source{{ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true}}, } prov := &fakeProvider{paths: map[string][]string{"arr": {"/mnt/media/Show/S01/E01.mkv"}}, nextMarker: "m1"} sup := &recordingSuppressor{} @@ -640,7 +683,7 @@ func TestPollOncePollsConnectionlessSource(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: nil, Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: nil, Enabled: true, }}, } prov := &fakeProvider{paths: map[string][]string{"arr": {"/mnt/media/Show/S01/E01.mkv"}}, nextMarker: "m1"} @@ -666,7 +709,7 @@ func TestPollOnceSkipsSourcePolledTooRecently(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, PollIntervalSeconds: &interval, LastRunAt: &recent, }}, } @@ -690,7 +733,7 @@ func TestPollOnceRunsSourcePastItsInterval(t *testing.T) { store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{{ - ID: "s1", InstallationID: 1, CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, + ID: "s1", PluginID: "silo.autoscan.arr", CapabilityID: "arr", ConnectionID: strptr("c1"), Enabled: true, PollIntervalSeconds: &interval, LastRunAt: &old, }}, } @@ -708,25 +751,23 @@ func TestPollOnceRunsSourcePastItsInterval(t *testing.T) { } } -func TestPollOnceSkipsOrphanedSourceQuietly(t *testing.T) { - // Two enabled sources: only "arr-live" is still discovered; "arr-gone"'s - // plugin has been uninstalled. The orphan must be skipped quietly — no poll, - // no RecordError spam — while the live source still polls normally. +func TestPollOnceRecordsScanSourceResolutionFailure(t *testing.T) { + // Two enabled sources: one polls normally, while the other simulates the + // production resolver failing because the plugin cannot be resolved. store := &fakeStore{ settings: Settings{Enabled: true, DefaultPollIntervalSeconds: 600, DebounceSeconds: 60}, sources: []Source{ - {ID: "live", InstallationID: 1, CapabilityID: "arr-live", ConnectionID: strptr("c1"), Enabled: true}, - {ID: "gone", InstallationID: 2, CapabilityID: "arr-gone", ConnectionID: strptr("c1"), Enabled: true}, + {ID: "live", PluginID: "silo.autoscan.live", CapabilityID: "arr-live", ConnectionID: strptr("c1"), Enabled: true}, + {ID: "gone", PluginID: "silo.autoscan.gone", CapabilityID: "arr-gone", ConnectionID: strptr("c1"), Enabled: true}, }, } prov := &fakeProvider{paths: map[string][]string{ "arr-live": {"/mnt/media/Show/S01/E01.mkv"}, - "arr-gone": {"/mnt/media/Other/S01/E01.mkv"}, + }, errByCap: map[string]error{ + "arr-gone": errors.New("scan source plugin \"silo.autoscan.gone\" is not installed"), }, nextMarker: "m1"} q := &recordingQueuer{} - // Lister reports only the live capability; the orphan is absent. - lister := fakeLister{sources: []DiscoveredSource{{InstallationID: 1, CapabilityID: "arr-live"}}} - svc := NewService(store, prov, passthroughConnRes{}, fakeResolver{}, q, allowSuppressor{}, lister) + svc := newService(store, prov, q, allowSuppressor{}) if err := svc.PollOnce(context.Background()); err != nil { t.Fatalf("PollOnce: %v", err) @@ -738,9 +779,15 @@ func TestPollOnceSkipsOrphanedSourceQuietly(t *testing.T) { t.Fatalf("expected live source marker advanced") } if _, ok := store.advanced["gone"]; ok { - t.Fatalf("orphaned source must NOT poll/advance") + t.Fatalf("failed source must NOT advance") } - if _, ok := store.recorded["gone"]; ok { - t.Fatalf("orphaned source must be skipped quietly, but an error was recorded: %q", store.recorded["gone"]) + if got := store.recorded["gone"]; !strings.Contains(got, "not installed") { + t.Fatalf("failed source error = %q", got) + } + if len(store.events) != 2 { + t.Fatalf("expected one event per source, got %+v", store.events) + } + if store.events[1].Status != EventStatusError || !strings.Contains(store.events[1].ErrorMessage, "not installed") { + t.Fatalf("failed source event = %+v", store.events[1]) } } diff --git a/internal/autoscan/types.go b/internal/autoscan/types.go index 64c8d214..7bace525 100644 --- a/internal/autoscan/types.go +++ b/internal/autoscan/types.go @@ -45,11 +45,11 @@ type Change struct { Scope ChangeScope } -// Source ties a scan_source plugin capability instance to a connection plus the +// Source ties a scan_source plugin capability to a connection plus the // host-owned scheduling/bookkeeping state. type Source struct { ID string - InstallationID int + PluginID string CapabilityID string ConnectionID *string // nil until an operator binds a connection Enabled bool @@ -65,6 +65,7 @@ type Source struct { type EventStatus string const ( + EventStatusRunning EventStatus = "running" EventStatusSuccess EventStatus = "success" EventStatusError EventStatus = "error" EventStatusUnresolved EventStatus = "unresolved" @@ -73,7 +74,7 @@ const ( type Event struct { ID int64 SourceID *string - InstallationID int + PluginID string CapabilityID string StartedAt time.Time CompletedAt time.Time @@ -91,11 +92,11 @@ type Event struct { } type EventCreate struct { - SourceID string - InstallationID int - CapabilityID string - StartedAt time.Time - MarkerBefore string + SourceID string + PluginID string + CapabilityID string + StartedAt time.Time + MarkerBefore string } type EventFinish struct { @@ -154,7 +155,7 @@ type ScanWithEvent struct { ScanRunSummary AutoscanEventID *int64 SourceID *string - InstallationID *int + PluginID string CapabilityID string EventStatus EventStatus EventCompletedAt *time.Time diff --git a/internal/plugins/service.go b/internal/plugins/service.go index 76136e30..a7f2f0f0 100644 --- a/internal/plugins/service.go +++ b/internal/plugins/service.go @@ -482,6 +482,53 @@ func (s *Service) ScanSourceClient( return client.ScanSource(capabilityID) } +func (s *Service) ScanSourceClientByPluginID( + ctx context.Context, + pluginID string, + capabilityID string, +) (*pluginhost.ScanSourceClient, error) { + if s == nil || s.installations == nil { + return nil, fmt.Errorf("scan source plugin resolver is not configured") + } + installations, err := s.installations.ListByPluginID(ctx, pluginID) + if err != nil { + return nil, err + } + if len(installations) == 0 { + return nil, fmt.Errorf("scan source plugin %q is not installed", pluginID) + } + if len(installations) > 1 { + return nil, fmt.Errorf("scan source plugin %q is ambiguous across %d installations", pluginID, len(installations)) + } + + var matches []*Installation + for _, installation := range installations { + if installation == nil { + continue + } + capabilities, err := s.installations.ListCapabilities(ctx, installation.ID) + if err != nil { + return nil, err + } + for _, capability := range capabilities { + if capability == nil { + continue + } + if capability.Type == "scan_source.v1" && capability.ID == capabilityID { + matches = append(matches, installation) + break + } + } + } + if len(matches) == 0 { + return nil, fmt.Errorf("scan source capability %q is not installed for plugin %q", capabilityID, pluginID) + } + if !matches[0].Enabled { + return nil, fmt.Errorf("scan source plugin %q is disabled", pluginID) + } + return s.ScanSourceClient(ctx, matches[0].ID, capabilityID) +} + func (s *Service) EventConsumerClient( ctx context.Context, installationID int, diff --git a/internal/plugins/service_scan_source_test.go b/internal/plugins/service_scan_source_test.go new file mode 100644 index 00000000..14fb5f31 --- /dev/null +++ b/internal/plugins/service_scan_source_test.go @@ -0,0 +1,78 @@ +package plugins + +import ( + "context" + "strings" + "testing" + + pluginv1 "github.com/Silo-Server/silo-plugin-sdk/pkg/pluginproto/silo/plugin/v1" + "github.com/Silo-Server/silo-server/internal/pluginhost" +) + +func TestScanSourceClientByPluginIDResolvesCurrentInstallation(t *testing.T) { + manifest := testPluginManifest(t, "silo.autoscan.cephfs", "0.2.2") + manifest.Capabilities = []*pluginv1.CapabilityDescriptor{ + {Type: "scan_source.v1", Id: "cephfs", DisplayName: "CephFS"}, + } + installPath := writeInstalledPluginManifest(t, manifest) + host := &fakeServiceHost{ + clientErr: pluginhost.ErrClientNotFound, + startResult: &fakePluginClient{manifest: manifest}, + } + service := &Service{ + installations: &fakeServiceInstallationStore{ + byID: map[int]*Installation{ + 8: { + ID: 8, + PluginID: "silo.autoscan.cephfs", + Version: "0.2.2", + InstallPath: installPath, + Enabled: true, + }, + }, + byPluginID: map[string][]*Installation{ + "silo.autoscan.cephfs": { + {ID: 8, PluginID: "silo.autoscan.cephfs", Version: "0.2.2", InstallPath: installPath, Enabled: true}, + }, + }, + listCapabilities: []*Capability{{InstallationID: 8, Type: "scan_source.v1", ID: "cephfs"}}, + }, + host: host, + } + + if _, err := service.ScanSourceClientByPluginID(context.Background(), "silo.autoscan.cephfs", "cephfs"); err != nil { + t.Fatalf("ScanSourceClientByPluginID: %v", err) + } + if len(host.started) != 1 || host.started[0].InstallationID != 8 { + t.Fatalf("started installations = %+v, want installation 8", host.started) + } +} + +func TestScanSourceClientByPluginIDRejectsAmbiguousInstallations(t *testing.T) { + manifest := testPluginManifest(t, "silo.autoscan.cephfs", "0.2.2") + manifest.Capabilities = []*pluginv1.CapabilityDescriptor{ + {Type: "scan_source.v1", Id: "cephfs", DisplayName: "CephFS"}, + } + installPath := writeInstalledPluginManifest(t, manifest) + service := &Service{ + installations: &fakeServiceInstallationStore{ + byID: map[int]*Installation{ + 8: {ID: 8, PluginID: "silo.autoscan.cephfs", Version: "0.2.2", InstallPath: installPath, Enabled: true}, + 9: {ID: 9, PluginID: "silo.autoscan.cephfs", Version: "0.2.3", InstallPath: installPath, Enabled: true}, + }, + byPluginID: map[string][]*Installation{ + "silo.autoscan.cephfs": { + {ID: 8, PluginID: "silo.autoscan.cephfs", Version: "0.2.2", InstallPath: installPath, Enabled: true}, + {ID: 9, PluginID: "silo.autoscan.cephfs", Version: "0.2.3", InstallPath: installPath, Enabled: true}, + }, + }, + listCapabilities: []*Capability{{Type: "scan_source.v1", ID: "cephfs"}}, + }, + host: &fakeServiceHost{}, + } + + _, err := service.ScanSourceClientByPluginID(context.Background(), "silo.autoscan.cephfs", "cephfs") + if err == nil || !strings.Contains(err.Error(), "ambiguous") { + t.Fatalf("ScanSourceClientByPluginID error = %v, want ambiguous", err) + } +} diff --git a/migrations/sql/20260607201204_autoscan_plugin_id_binding.sql b/migrations/sql/20260607201204_autoscan_plugin_id_binding.sql new file mode 100644 index 00000000..efbeb7bf --- /dev/null +++ b/migrations/sql/20260607201204_autoscan_plugin_id_binding.sql @@ -0,0 +1,145 @@ +-- +goose Up +-- +goose StatementBegin +ALTER TABLE public.autoscan_sources + ADD COLUMN plugin_id text NOT NULL DEFAULT ''; + +ALTER TABLE public.autoscan_events + ADD COLUMN plugin_id text NOT NULL DEFAULT ''; + +UPDATE public.autoscan_sources s +SET plugin_id = pi.plugin_id +FROM public.plugin_installations pi +WHERE s.installation_id = pi.id + AND s.plugin_id = ''; + +WITH inferred AS ( + SELECT s.id, min(pi.plugin_id) AS plugin_id, count(*) AS matches + FROM public.autoscan_sources s + JOIN public.plugin_capabilities pc + ON pc.capability_type = 'scan_source.v1' + AND pc.capability_id = s.capability_id + JOIN public.plugin_installations pi + ON pi.id = pc.plugin_installation_id + WHERE s.plugin_id = '' + GROUP BY s.id +) +UPDATE public.autoscan_sources s +SET plugin_id = inferred.plugin_id +FROM inferred +WHERE s.id = inferred.id + AND inferred.matches = 1; + +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM public.autoscan_sources + WHERE enabled = true + AND btrim(plugin_id) = '' + ) THEN + RAISE EXCEPTION 'enabled autoscan source could not be migrated to plugin_id'; + END IF; +END $$; + +UPDATE public.autoscan_events e +SET plugin_id = s.plugin_id +FROM public.autoscan_sources s +WHERE e.source_id = s.id + AND e.plugin_id = '' + AND btrim(s.plugin_id) <> ''; + +UPDATE public.autoscan_events e +SET plugin_id = pi.plugin_id +FROM public.plugin_installations pi +WHERE e.installation_id = pi.id + AND e.plugin_id = ''; + +WITH inferred AS ( + SELECT e.id, min(pi.plugin_id) AS plugin_id, count(*) AS matches + FROM public.autoscan_events e + JOIN public.plugin_capabilities pc + ON pc.capability_type = 'scan_source.v1' + AND pc.capability_id = e.capability_id + JOIN public.plugin_installations pi + ON pi.id = pc.plugin_installation_id + WHERE e.plugin_id = '' + GROUP BY e.id +) +UPDATE public.autoscan_events e +SET plugin_id = inferred.plugin_id +FROM inferred +WHERE e.id = inferred.id + AND inferred.matches = 1; + +CREATE INDEX idx_autoscan_sources_plugin_capability + ON public.autoscan_sources (plugin_id, capability_id); + +CREATE INDEX idx_autoscan_events_plugin_capability_completed + ON public.autoscan_events (plugin_id, capability_id, completed_at DESC); + +ALTER TABLE public.autoscan_sources + ALTER COLUMN plugin_id DROP DEFAULT; + +ALTER TABLE public.autoscan_events + ALTER COLUMN plugin_id DROP DEFAULT; + +ALTER TABLE public.autoscan_sources + DROP COLUMN installation_id; + +ALTER TABLE public.autoscan_events + DROP COLUMN installation_id; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE public.autoscan_sources + ADD COLUMN installation_id integer NOT NULL DEFAULT 0; + +ALTER TABLE public.autoscan_events + ADD COLUMN installation_id integer NOT NULL DEFAULT 0; + +WITH resolved AS ( + SELECT s.id, min(pc.plugin_installation_id)::integer AS installation_id, count(*) AS matches + FROM public.autoscan_sources s + JOIN public.plugin_installations pi + ON pi.plugin_id = s.plugin_id + JOIN public.plugin_capabilities pc + ON pc.plugin_installation_id = pi.id + AND pc.capability_type = 'scan_source.v1' + AND pc.capability_id = s.capability_id + WHERE btrim(s.plugin_id) <> '' + GROUP BY s.id +) +UPDATE public.autoscan_sources s +SET installation_id = resolved.installation_id +FROM resolved +WHERE s.id = resolved.id + AND resolved.matches = 1; + +WITH resolved AS ( + SELECT e.id, min(pc.plugin_installation_id)::integer AS installation_id, count(*) AS matches + FROM public.autoscan_events e + JOIN public.plugin_installations pi + ON pi.plugin_id = e.plugin_id + JOIN public.plugin_capabilities pc + ON pc.plugin_installation_id = pi.id + AND pc.capability_type = 'scan_source.v1' + AND pc.capability_id = e.capability_id + WHERE btrim(e.plugin_id) <> '' + GROUP BY e.id +) +UPDATE public.autoscan_events e +SET installation_id = resolved.installation_id +FROM resolved +WHERE e.id = resolved.id + AND resolved.matches = 1; + +DROP INDEX IF EXISTS public.idx_autoscan_events_plugin_capability_completed; +DROP INDEX IF EXISTS public.idx_autoscan_sources_plugin_capability; + +ALTER TABLE public.autoscan_events + DROP COLUMN plugin_id; + +ALTER TABLE public.autoscan_sources + DROP COLUMN plugin_id; +-- +goose StatementEnd diff --git a/migrations/sql/20260607204951_autoscan_running_polls.sql b/migrations/sql/20260607204951_autoscan_running_polls.sql new file mode 100644 index 00000000..4fa8c1b8 --- /dev/null +++ b/migrations/sql/20260607204951_autoscan_running_polls.sql @@ -0,0 +1,36 @@ +-- +goose Up +-- +goose StatementBegin +ALTER TABLE public.autoscan_events + DROP CONSTRAINT IF EXISTS autoscan_events_status_check; + +ALTER TABLE public.autoscan_events + ADD CONSTRAINT autoscan_events_status_check + CHECK (status = ANY (ARRAY[ + 'running'::text, + 'success'::text, + 'error'::text, + 'unresolved'::text + ])); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +UPDATE public.autoscan_events +SET status = 'error', + error_message = CASE + WHEN error_message = '' THEN 'poll started but did not finish' + ELSE error_message + END +WHERE status = 'running'; + +ALTER TABLE public.autoscan_events + DROP CONSTRAINT IF EXISTS autoscan_events_status_check; + +ALTER TABLE public.autoscan_events + ADD CONSTRAINT autoscan_events_status_check + CHECK (status = ANY (ARRAY[ + 'success'::text, + 'error'::text, + 'unresolved'::text + ])); +-- +goose StatementEnd diff --git a/web/src/api/types.ts b/web/src/api/types.ts index f95d8675..61c7cc48 100644 --- a/web/src/api/types.ts +++ b/web/src/api/types.ts @@ -1808,7 +1808,7 @@ export interface AutoscanPathRewrite { export interface AutoscanSource { id: string; - installation_id: number; + plugin_id: string; capability_id: string; connection_id: string | null; enabled: boolean; @@ -1834,9 +1834,8 @@ export interface AutoscanSourcesResponse { } export interface AutoscanAvailableSource { - installation_id: number; - capability_id: string; plugin_id: string; + capability_id: string; display_name: string; } @@ -1845,7 +1844,7 @@ export interface AutoscanAvailableSourcesResponse { } export interface AutoscanSourceCreateInput { - installation_id: number; + plugin_id: string; capability_id: string; connection_id?: string | null; enabled: boolean; @@ -1887,7 +1886,7 @@ export interface AutoscanRewriteSuggestions { export interface AutoscanStatusSource { id: string; - installation_id: number; + plugin_id: string; capability_id: string; connection_id: string | null; enabled: boolean; @@ -1896,16 +1895,27 @@ export interface AutoscanStatusSource { last_error: string | null; } +export interface AutoscanRunningPoll { + id: number; + source_id: string | null; + plugin_id: string; + capability_id: string; + started_at: string; + elapsed_ms: number; + marker_before?: string; +} + export interface AutoscanStatus { enabled: boolean; sources: AutoscanStatusSource[]; + running_polls: AutoscanRunningPoll[]; active_scans: number; accepted_scans: number; running_scans: number; latest_event_at?: string; } -export type AutoscanEventStatus = "success" | "error" | "unresolved"; +export type AutoscanEventStatus = "running" | "success" | "error" | "unresolved"; export interface AutoscanEventScanRun { id: string; @@ -1923,7 +1933,7 @@ export interface AutoscanEventScanRun { export interface AutoscanEvent { id: number; source_id: string | null; - installation_id: number; + plugin_id: string; capability_id: string; started_at: string; completed_at: string; @@ -1961,7 +1971,7 @@ export interface AutoscanScan { completed_at?: string; autoscan_event_id?: number; source_id?: string; - installation_id?: number; + plugin_id?: string; capability_id?: string; event_status?: AutoscanEventStatus; event_completed_at?: string; diff --git a/web/src/hooks/queries/useAutoscan.ts b/web/src/hooks/queries/useAutoscan.ts index b9fdf6be..203d57de 100644 --- a/web/src/hooks/queries/useAutoscan.ts +++ b/web/src/hooks/queries/useAutoscan.ts @@ -312,6 +312,7 @@ export function useAutoscanScans(params?: { // --- Trigger --- export function useTriggerAutoscan() { + const queryClient = useQueryClient(); return useMutation({ mutationFn: () => api<{ status: string }>("/admin/autoscan/trigger", { @@ -319,6 +320,8 @@ export function useTriggerAutoscan() { }), onSuccess: () => { toast.success("Autoscan triggered"); + queryClient.invalidateQueries({ queryKey: adminKeys.autoscanStatus() }); + queryClient.invalidateQueries({ queryKey: ["admin", "autoscan", "events"] }); }, onError: (err) => { toast.error(err instanceof Error ? err.message : "Failed to trigger autoscan"); diff --git a/web/src/lib/autoscanLabels.test.ts b/web/src/lib/autoscanLabels.test.ts index c1caf3f3..95dd7c29 100644 --- a/web/src/lib/autoscanLabels.test.ts +++ b/web/src/lib/autoscanLabels.test.ts @@ -8,38 +8,38 @@ import { } from "./autoscanLabels"; describe("composeSourceLabel", () => { - const base = { capabilityId: "arr", installationId: 4 }; + const base = { capabilityId: "arr", pluginId: "silo.autoscan.arr" }; it("uses the operator label first, demoting connection to detail", () => { expect( composeSourceLabel({ ...base, operatorLabel: "4K Movies", connectionName: "Radarr4k" }), - ).toEqual({ name: "4K Movies", detail: "Radarr4k · plugin #4" }); + ).toEqual({ name: "4K Movies", detail: "Radarr4k · silo.autoscan.arr" }); }); it("uses the connection name when no operator label", () => { expect( composeSourceLabel({ ...base, connectionName: "Radarr4k", displayName: "Arr Watcher" }), - ).toEqual({ name: "Radarr4k", detail: "Arr Watcher · plugin #4" }); + ).toEqual({ name: "Radarr4k", detail: "Arr Watcher · silo.autoscan.arr" }); }); it("uses the manifest display name when no connection", () => { expect( composeSourceLabel({ capabilityId: "cephfs", - installationId: 5, + pluginId: "silo.autoscan.cephfs", displayName: "CephFS Watcher", }), - ).toEqual({ name: "CephFS Watcher", detail: "plugin #5" }); + ).toEqual({ name: "CephFS Watcher", detail: "silo.autoscan.cephfs" }); }); it("falls back to capability id when nothing else is set", () => { - expect(composeSourceLabel(base)).toEqual({ name: "arr", detail: "plugin #4" }); + expect(composeSourceLabel(base)).toEqual({ name: "arr", detail: "silo.autoscan.arr" }); }); it("ignores whitespace-only rungs", () => { expect(composeSourceLabel({ ...base, operatorLabel: " ", connectionName: " " })).toEqual({ name: "arr", - detail: "plugin #4", + detail: "silo.autoscan.arr", }); }); }); @@ -47,7 +47,7 @@ describe("composeSourceLabel", () => { describe("resolveEventSourceName", () => { const source: AutoscanSource = { id: "src-1", - installation_id: 4, + plugin_id: "silo.autoscan.arr", capability_id: "arr", connection_id: "conn-1", enabled: true, @@ -61,13 +61,13 @@ describe("resolveEventSourceName", () => { const lookups: SourceLabelLookups = { sourceByID: new Map([["src-1", source]]), connectionByID: new Map([["conn-1", "Radarr4k"]]), - displayNames: new Map([["4:arr", "Arr Watcher"]]), + displayNames: new Map([["silo.autoscan.arr:arr", "Arr Watcher"]]), }; it("resolves the connection name via the source reference", () => { expect( resolveEventSourceName( - { source_id: "src-1", capability_id: "arr", installation_id: 4 }, + { source_id: "src-1", capability_id: "arr", plugin_id: "silo.autoscan.arr" }, lookups, ), ).toBe("Radarr4k"); @@ -80,7 +80,7 @@ describe("resolveEventSourceName", () => { }; expect( resolveEventSourceName( - { source_id: "src-1", capability_id: "arr", installation_id: 4 }, + { source_id: "src-1", capability_id: "arr", plugin_id: "silo.autoscan.arr" }, withLabel, ), ).toBe("4K Movies"); @@ -89,7 +89,7 @@ describe("resolveEventSourceName", () => { it("falls back to display name when the source was deleted (null source_id)", () => { expect( resolveEventSourceName( - { source_id: null, capability_id: "arr", installation_id: 4 }, + { source_id: null, capability_id: "arr", plugin_id: "silo.autoscan.arr" }, lookups, ), ).toBe("Arr Watcher"); @@ -106,7 +106,7 @@ describe("resolveEventSourceName", () => { }; expect( resolveEventSourceName( - { source_id: "src-1", capability_id: "arr", installation_id: 4 }, + { source_id: "src-1", capability_id: "arr", plugin_id: "silo.autoscan.arr" }, orphaned, ), ).toBe("Arr Watcher"); diff --git a/web/src/lib/autoscanLabels.ts b/web/src/lib/autoscanLabels.ts index a3ffb545..8a485c12 100644 --- a/web/src/lib/autoscanLabels.ts +++ b/web/src/lib/autoscanLabels.ts @@ -5,7 +5,7 @@ export interface SourceLabelParts { connectionName?: string | null; displayName?: string | null; capabilityId: string; - installationId: number; + pluginId: string; } export interface SourceLabel { @@ -22,31 +22,33 @@ export function composeSourceLabel(parts: SourceLabelParts): SourceLabel { const operator = parts.operatorLabel?.trim() ?? ""; const connection = parts.connectionName?.trim() ?? ""; const display = parts.displayName?.trim() ?? ""; - const pluginSuffix = `plugin #${parts.installationId}`; - const pluginIdentity = display || parts.capabilityId; + const pluginIdentity = display || parts.pluginId || parts.capabilityId; + const pluginDetail = parts.pluginId || parts.capabilityId; + const detailWithPlugin = (detail: string) => + detail === pluginDetail ? detail : `${detail} · ${pluginDetail}`; if (operator) { - return { name: operator, detail: `${connection || pluginIdentity} · ${pluginSuffix}` }; + return { name: operator, detail: detailWithPlugin(connection || pluginIdentity) }; } if (connection) { - return { name: connection, detail: `${pluginIdentity} · ${pluginSuffix}` }; + return { name: connection, detail: detailWithPlugin(pluginIdentity) }; } if (display) { - return { name: display, detail: pluginSuffix }; + return { name: display, detail: pluginDetail }; } - return { name: parts.capabilityId, detail: pluginSuffix }; + return { name: parts.capabilityId, detail: pluginDetail }; } -/** Stable key for the (installation, capability) -> manifest display_name map. */ -export function pluginDisplayNameKey(installationId: number, capabilityId: string): string { - return `${installationId}:${capabilityId}`; +/** Stable key for the (plugin, capability) -> manifest display_name map. */ +export function pluginDisplayNameKey(pluginId: string, capabilityId: string): string { + return `${pluginId}:${capabilityId}`; } -/** Build the (installation, capability) -> display_name lookup from the picker list. */ +/** Build the (plugin, capability) -> display_name lookup from the picker list. */ export function buildPluginDisplayNames(available: AutoscanAvailableSource[]): Map { const map = new Map(); for (const a of available) { - map.set(pluginDisplayNameKey(a.installation_id, a.capability_id), a.display_name); + map.set(pluginDisplayNameKey(a.plugin_id, a.capability_id), a.display_name); } return map; } @@ -64,22 +66,22 @@ export interface SourceLabelLookups { * own fallback, e.g. "Autoscan"). */ export function resolveEventSourceName( - ref: { source_id?: string | null; capability_id?: string; installation_id?: number | null }, + ref: { source_id?: string | null; capability_id?: string; plugin_id?: string | null }, lookups: SourceLabelLookups, ): string { - if (!ref.capability_id || ref.installation_id == null) return ""; + if (!ref.capability_id || !ref.plugin_id) return ""; const source = ref.source_id ? lookups.sourceByID.get(ref.source_id) : undefined; const connectionName = source?.connection_id ? lookups.connectionByID.get(source.connection_id) : undefined; const displayName = lookups.displayNames.get( - pluginDisplayNameKey(ref.installation_id, ref.capability_id), + pluginDisplayNameKey(ref.plugin_id, ref.capability_id), ); return composeSourceLabel({ operatorLabel: source?.label, connectionName, displayName, capabilityId: ref.capability_id, - installationId: ref.installation_id, + pluginId: ref.plugin_id, }).name; } diff --git a/web/src/pages/admin/autoscan/ActivityPanel.tsx b/web/src/pages/admin/autoscan/ActivityPanel.tsx index 2aed3428..dc021c2e 100644 --- a/web/src/pages/admin/autoscan/ActivityPanel.tsx +++ b/web/src/pages/admin/autoscan/ActivityPanel.tsx @@ -15,6 +15,7 @@ import type { AutoscanEvent, AutoscanEventScanRun, AutoscanEventStatus, + AutoscanRunningPoll, AutoscanScan, AutoscanScanStatus, Library, @@ -95,6 +96,18 @@ function formatDuration(ms: number): string { return `${minutes}m ${seconds}s`; } +function pollEventDuration(event: AutoscanEvent): string { + if (event.status !== "running") { + return formatDuration(event.duration_ms); + } + const elapsed = Math.max(0, Date.now() - new Date(event.started_at).getTime()); + return formatDuration(elapsed); +} + +function pollEventTimestamp(event: AutoscanEvent): string { + return formatTimestamp(event.status === "running" ? event.started_at : event.completed_at); +} + function eventStatusTone(status: AutoscanEventStatus): { label: string; icon: typeof CheckCircle2; @@ -107,6 +120,12 @@ function eventStatusTone(status: AutoscanEventStatus): { icon: CheckCircle2, className: "border-emerald-500/30 bg-emerald-500/10 text-emerald-500", }; + case "running": + return { + label: "Running", + icon: RefreshCw, + className: "border-primary/30 bg-primary/10 text-primary", + }; case "unresolved": return { label: "Unresolved", @@ -144,9 +163,15 @@ function scanStatusLabel(status: AutoscanScanStatus | ScanRun["status"]) { // arr-plugin sources fan out one-per-connection under a single generic // capability, so resolve every Activity row through the shared label chain: // operator label -> connection name -> manifest display_name -> capability_id. -function pollSourceName(event: AutoscanEvent, lookups: SourceLabelLookups): string { +type PollSourceRef = { + source_id?: string | null; + plugin_id?: string | null; + capability_id?: string; +}; + +function pollSourceName(event: PollSourceRef, lookups: SourceLabelLookups): string { return ( - resolveEventSourceName(event, lookups) || `${event.capability_id} #${event.installation_id}` + resolveEventSourceName(event, lookups) || event.plugin_id || event.capability_id || "Autoscan" ); } @@ -163,7 +188,7 @@ function PollStatusBadge({ status }: { status: AutoscanEventStatus }) { const Icon = tone.icon; return ( - + {tone.label} ); @@ -415,6 +440,55 @@ function AutoscanQueue({ ); } +function RunningPolls({ + polls, + lookups, +}: { + polls: AutoscanRunningPoll[]; + lookups: SourceLabelLookups; +}) { + if (polls.length === 0) return null; + + return ( +
+
+
+
+ + Polling now +
+

+ Scan-source plugin calls currently in progress. +

+
+ + {polls.length} active + +
+ +
+ {polls.map((poll) => ( +
+
+
+
{pollSourceName(poll, lookups)}
+
+ {poll.plugin_id} · {poll.capability_id} +
+
+ +
+
+ Started {formatTimestamp(poll.started_at)} + {formatDuration(poll.elapsed_ms)} elapsed +
+
+ ))} +
+
+ ); +} + function ScanHistoryCard({ scan, librariesByID, @@ -536,8 +610,8 @@ function PollEventCard({ event, lookups }: { event: AutoscanEvent; lookups: Sour
-
{formatTimestamp(event.completed_at)}
-
{formatDuration(event.duration_ms)}
+
{pollEventTimestamp(event)}
+
{pollEventDuration(event)}
{event.error_message ? ( @@ -579,7 +653,7 @@ function PollEventTable({ Counts Scans Duration - Completed + Time } > @@ -608,10 +682,10 @@ function PollEventTable({ - {formatDuration(event.duration_ms)} + {pollEventDuration(event)} - {formatTimestamp(event.completed_at)} + {pollEventTimestamp(event)} ))} @@ -740,10 +814,11 @@ export default function ActivityPanel() { return (
-
+
- + +
+ + All statuses + Running Success Unresolved Error diff --git a/web/src/pages/admin/autoscan/SourcesPanel.tsx b/web/src/pages/admin/autoscan/SourcesPanel.tsx index aa052816..ae8aec25 100644 --- a/web/src/pages/admin/autoscan/SourcesPanel.tsx +++ b/web/src/pages/admin/autoscan/SourcesPanel.tsx @@ -93,10 +93,10 @@ function resolveSourceName( operatorLabel: source.label, connectionName: connectionOptions.find((c) => c.id === (source.connection_id ?? ""))?.name, displayName: pluginDisplayNames.get( - pluginDisplayNameKey(source.installation_id, source.capability_id), + pluginDisplayNameKey(source.plugin_id, source.capability_id), ), capabilityId: source.capability_id, - installationId: source.installation_id, + pluginId: source.plugin_id, }).name; } @@ -152,7 +152,7 @@ const CEPHFS_LEGACY_MOVIE_NESTED_KEY = "movie_nested_paths"; const CEPHFS_LEGACY_TV_NESTED_KEY = "tv_nested_paths"; function isCephFSSource(source: AutoscanSource): boolean { - return source.capability_id === CEPHFS_CAPABILITY_ID; + return source.plugin_id === CEPHFS_PLUGIN_ID || source.capability_id === CEPHFS_CAPABILITY_ID; } function isCephFSPlugin(plugin: { plugin_id: string; capability_id: string } | undefined): boolean { @@ -837,10 +837,10 @@ function SourceRow({ (c) => c.id === (edit.connectionId || source.connection_id || ""), )?.name, displayName: pluginDisplayNames.get( - pluginDisplayNameKey(source.installation_id, source.capability_id), + pluginDisplayNameKey(source.plugin_id, source.capability_id), ), capabilityId: source.capability_id, - installationId: source.installation_id, + pluginId: source.plugin_id, }); const sourceIdentity = (
@@ -1061,7 +1061,7 @@ function SourceRow({ // --------------------------------------------------------------------------- interface AddSourceForm { - /** "installation_id:capability_id" composite key of the chosen plugin. */ + /** "plugin_id:capability_id" composite key of the chosen plugin. */ pluginKey: string; connectionId: string; // "" / "__none__" means no connection intervalStr: string; @@ -1075,8 +1075,8 @@ const BLANK_ADD_SOURCE: AddSourceForm = { sourceConfig: {}, }; -function pluginKey(installationId: number, capabilityId: string): string { - return `${installationId}:${capabilityId}`; +function pluginKey(pluginId: string, capabilityId: string): string { + return `${pluginId}:${capabilityId}`; } function AddSourceDialog({ @@ -1094,7 +1094,7 @@ function AddSourceDialog({ const plugins = available.data ?? []; const selectedPlugin = plugins.find( - (p) => pluginKey(p.installation_id, p.capability_id) === form.pluginKey, + (p) => pluginKey(p.plugin_id, p.capability_id) === form.pluginKey, ); const selectedIsCephFS = isCephFSPlugin(selectedPlugin); @@ -1111,7 +1111,7 @@ function AddSourceDialog({ const pollInterval = raw === "" ? null : Number(raw); createSource.mutate( { - installation_id: selectedPlugin.installation_id, + plugin_id: selectedPlugin.plugin_id, capability_id: selectedPlugin.capability_id, connection_id: connectionId, enabled: false, @@ -1157,9 +1157,7 @@ function AddSourceDialog({