From 1d5402517b6a2e2c21e885e0b1965be54236b0bc Mon Sep 17 00:00:00 2001 From: crypt0rr <57799908+crypt0rr@users.noreply.github.com> Date: Fri, 14 Aug 2026 07:18:08 +0200 Subject: [PATCH 1/2] Harden recovery and notification reliability --- .dockerignore | 3 + .gitignore | 3 + Makefile | 3 +- README.md | 53 +- internal/jobs/jobs.go | 83 ++- internal/jobs/jobs_test.go | 54 ++ internal/jobs/providers.go | 57 ++- internal/notify/notify.go | 55 ++ internal/notify/notify_test.go | 42 ++ internal/store/auth.go | 47 ++ internal/store/calendar.go | 10 +- .../store/migrations/025_recovery_leases.sql | 76 +++ internal/store/notification_history_test.go | 271 ++++++++++ internal/store/notifications.go | 473 ++++++++++++++++-- internal/store/operations.go | 126 ++++- internal/store/release_helpers.go | 4 +- internal/store/schema.go | 3 +- internal/store/store.go | 80 +-- internal/store/store_test.go | 5 +- internal/web/admin.go | 18 + internal/web/auth.go | 21 +- internal/web/core.go | 32 +- internal/web/settings.go | 11 +- internal/web/static/app.js | 5 +- internal/web/templates/admin.html | 9 +- internal/web/templates/partials.html | 13 +- internal/web/web.go | 40 +- scripts/backup.sh | 5 + scripts/restore-smoke.sh | 55 +- 29 files changed, 1481 insertions(+), 176 deletions(-) create mode 100644 internal/store/migrations/025_recovery_leases.sql diff --git a/.dockerignore b/.dockerignore index a1d14d9..869574f 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,3 +4,6 @@ artist-tracker artist-trackarr coverage.out +*.tgz +*.sha256 +.artist-trackarr-last-* diff --git a/.gitignore b/.gitignore index cc5a2c7..a4d303c 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,6 @@ /artist-tracker /artist-trackarr /coverage.out +*.tgz +*.sha256 +.artist-trackarr-last-* diff --git a/Makefile b/Makefile index fbab70c..00c453f 100644 --- a/Makefile +++ b/Makefile @@ -9,7 +9,8 @@ build: docker build -t artist-trackarr:local . run: - docker compose up --build + $(MAKE) build + ARTIST_TRACKARR_IMAGE=artist-trackarr:local docker compose up fmt-check: @test -z "$$(gofmt -l internal cmd)" diff --git a/README.md b/README.md index b2e8aea..7b73af2 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,7 @@ releases without creating releases or notifications. Use the moon/sun button in the header to switch between light and dark mode; your choice is remembered in the browser. The running application version and project repository are available in the -footer. The current release is `v0.41.0`; release images display the injected +footer. The current release is `v0.42.0`; release images display the injected semantic version while local builds identify themselves as `dev`. Operational timestamps are stored in UTC and rendered in the configured system timezone; existing databases are @@ -57,7 +57,7 @@ error when a data lookup fails, while the detailed cause remains in structured logs. Static assets use immutable, version-stamped URLs and continue to serve their unversioned paths for compatibility. -The v0.41.0 persistence-hardening release routes production SQLite writes +The v0.42.0 reliability-guardrails release routes production SQLite writes through a bounded busy/locked retry path, rejects malformed operational timestamps instead of silently showing zero values, and preflights database paths and MusicBrainz contact input before startup. The v0.40.0 operations @@ -212,7 +212,7 @@ GitHub Actions builds and publishes the Docker image to - `latest` and `main` follow the current `main` branch. - `sha-` identifies an exact source revision. -- Pushing a tag such as `v0.41.0` publishes `0.41.0`, `0.41`, and `latest`. +- Pushing a tag such as `v0.42.0` publishes `0.42.0`, `0.42`, and `latest`. Release images receive their version through the Docker build's `APP_VERSION` argument. Tag builds inject the semantic tag (without the leading `v`), while @@ -222,7 +222,7 @@ not confused with a release. Pin a deployment to a release by setting the Compose image before starting: ```console -ARTIST_TRACKARR_IMAGE=ghcr.io/crypt0rr/artist-trackarr:0.41.0 docker compose up -d +ARTIST_TRACKARR_IMAGE=ghcr.io/crypt0rr/artist-trackarr:0.42.0 docker compose up -d ``` ## Configuration @@ -348,10 +348,22 @@ filters. Genres come from MusicBrainz tags and are normalized locally. ## Notification destinations -Users can add guided Email, Discord, Telegram, ntfy, Gotify, and generic -webhook destinations. Any service supported by Shoutrrr can be added with its -raw service URL. Credentials are encrypted in SQLite and redacted from the UI -and logs. Use the **Send test** action after adding a destination. +Users can add Discord, Telegram, ntfy, and generic HTTP(S) webhook +destinations. Advanced Shoutrrr URLs are limited to those same audited +transports; SMTP, Gotify, and unknown schemes are rejected because the +application cannot apply its connection-time SSRF policy to them. Existing +legacy destinations remain visible as **Unsupported**, are never contacted, +and must be replaced explicitly. Credentials are encrypted in SQLite and +redacted from the UI and logs. Use the **Send test** action after adding a +destination. + +Delivery is at-least-once. A process crash after an external provider accepts a +message can result in a duplicate, but durable claims and recovery avoid +silently losing queued work. Paused or unsupported destinations receive a +blocked queue row instead of disappearing from an event; an administrator or +owner can retry after replacing/recovering the destination. A newly added +destination receives future events only and is not backfilled with historical +notifications. Users can choose whether albums, EPs, singles, announcements, and release-day reminders should be delivered. Followed artists show their last and next @@ -397,6 +409,11 @@ persistent data directory and is accompanied by a restrictive-permission ./scripts/backup.sh artist-trackarr-backup.tgz ``` +Successful backups write a non-sensitive timestamp marker into the persistent +volume so administrator diagnostics can show an approximate backup age. The +marker is archived with the next backup and is not a substitute for an +off-host backup inventory. + Restore into an empty Compose volume while the app is stopped, keep the original `APP_ENCRYPTION_KEY` available, and run the temporary restore rehearsal before replacing production data. The key is required to decrypt @@ -404,11 +421,16 @@ existing notification destinations. Embedded migrations run automatically during upgrades; the rehearsal must pass SQLite foreign-key checks and `/readyz` before the restored instance is considered usable. -The rehearsal verifies the checksum sidecar when present, fingerprints the -durable database state, and compares that fingerprint after a clean restart. A -mismatch fails the rehearsal rather than declaring the restore usable. Legacy -archives without a sidecar are accepted with a warning; new backups should -always retain both files. +The rehearsal requires an immutable image digest (`@sha256:`), verifies the +checksum sidecar, runs SQLite `integrity_check` and `foreign_key_check`, +validates that encrypted destinations can be opened with the original key, +fingerprints the durable logical database state, and compares that fingerprint +after a clean restart. A mismatch fails the rehearsal rather than declaring the +restore usable. Legacy archives without a sidecar or mutable images are +accepted only when explicitly opted in with `RESTORE_ALLOW_LEGACY_ARCHIVE=true` +or `RESTORE_ALLOW_MUTABLE_IMAGE=true`; new backups should always use the +immutable path. Backup archives and encryption keys are confidential operator +artifacts. The rehearsal uses an isolated Docker volume, starts the selected image, stops it with the configured grace period, and starts it again to verify that @@ -416,10 +438,13 @@ the restored data remains usable: ```console APP_ENCRYPTION_KEY="$APP_ENCRYPTION_KEY" \ - ARTIST_TRACKARR_IMAGE=ghcr.io/crypt0rr/artist-trackarr:0.41.0 \ + ARTIST_TRACKARR_IMAGE=ghcr.io/crypt0rr/artist-trackarr@sha256: \ ./scripts/restore-smoke.sh artist-trackarr-backup.tgz ``` +The rehearsal records a non-sensitive restore result marker in its temporary +volume; that volume is removed when the rehearsal exits. + ## Development The test suite runs in the pinned Go toolchain from the build image: diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index 6d2100e..9555fbd 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -54,6 +54,7 @@ type Runner struct { musicBrainzFailureStreak int running atomic.Bool lastActivity atomic.Int64 + workerID string } // RunnerStatus is a process-local scheduler snapshot for the admin assurance @@ -153,6 +154,11 @@ func New(s *store.Store, provider catalog.CatalogProvider, normalizer catalog.Re cipher: cipher, interval: interval, spotifyInterval: 24 * time.Hour, logger: logger, metrics: metrics.New(), } + if token, err := security.Token(12); err == nil { + runner.workerID = "runner-" + token + } else { + runner.workerID = fmt.Sprintf("runner-%d", time.Now().UnixNano()) + } runner.initLifecycle() for _, option := range options { option(runner) @@ -340,6 +346,16 @@ func (r *Runner) Run(ctx context.Context) { r.tasks.Wait() close(r.done) }() + if recovered, err := r.store.RecoverExpiredWork(ctx, time.Now().UTC()); err != nil { + r.logger.Warn("durable work recovery failed", "error", err) + } else if recovered > 0 { + r.logger.Info("durable work recovered", "rows", recovered) + } + if reconciled, err := r.store.ReconcileStaleDeliveryAttempts(ctx, time.Now().UTC(), 10*time.Minute); err != nil { + r.logger.Warn("stale delivery attempt reconciliation failed", "error", err) + } else if reconciled > 0 { + r.logger.Info("stale delivery attempts reconciled", "attempts", reconciled) + } if ctx.Err() == nil { r.launchSync(ctx) r.launchReleaseDayQueue(ctx) @@ -545,6 +561,16 @@ func (r *Runner) runSyncCadence(ctx context.Context) { func (r *Runner) runMaintenance(ctx context.Context) { r.metrics.RecordMaintenance() + if recovered, err := r.store.RecoverExpiredWork(ctx, time.Now().UTC()); err != nil { + r.logger.Warn("durable work recovery failed", "error", err) + } else if recovered > 0 { + r.logger.Info("durable work recovered", "rows", recovered) + } + if reconciled, err := r.store.ReconcileStaleDeliveryAttempts(ctx, time.Now().UTC(), 10*time.Minute); err != nil { + r.logger.Warn("stale delivery attempt reconciliation failed", "error", err) + } else if reconciled > 0 { + r.logger.Info("stale delivery attempts reconciled", "attempts", reconciled) + } if err := r.store.PruneApplicationLogs(ctx, time.Now().UTC().Add(-7*24*time.Hour)); err != nil { r.logger.Debug("application log pruning failed", "error", err) } @@ -702,7 +728,7 @@ func (r *Runner) refreshListenBrainz(ctx context.Context, now time.Time) (int, e } func (r *Runner) processManualSyncRequests(ctx context.Context, now time.Time) int { - requests, err := r.store.ClaimManualSyncRequests(ctx, 3) + requests, err := r.store.ClaimManualSyncRequestsWithLease(ctx, 3, r.workerID, 5*time.Minute) if err != nil { r.logger.Warn("manual synchronization queue failed", "error", err) return 0 @@ -722,7 +748,7 @@ func (r *Runner) processManualSyncRequests(ctx context.Context, now time.Time) i _, syncErr = r.syncArtists(ctx, now) } } - if err := r.store.CompleteManualSyncRequest(ctx, req.ID, syncErr); err != nil { + if err := r.store.CompleteManualSyncRequestOwned(ctx, req.ID, r.workerID, syncErr); err != nil { r.logger.Warn("manual synchronization completion failed", "request_id", req.ID, "error", err) } } @@ -953,6 +979,33 @@ func (r *Runner) syncOne(ctx context.Context, artist store.Artist, now time.Time return outcome, nil } if len(strategy.batches) == 0 { + // Empty provider catalogs are successful health checks but are not + // actionable release observations. Once every fallback provider has + // answered successfully, advance the normal artist cadence instead of + // treating the empty response as a failure. + // A successful empty catalog is safe to treat as a normal cadence only + // when every provider that was attempted completed cleanly. If a + // fallback provider failed or is cooling down, keep the artist on a + // bounded retry cadence instead of letting an empty response mask the + // outage. + if len(strategy.providerErrors) == 0 && (strategy.spotifyHealthy || strategy.itunesHealthy) { + if err := r.store.MarkArtistChecked(ctx, artist.ID, now, r.interval); err != nil { + return outcome, err + } + if spotifyWasDue && strategy.spotifyAttempted { + // Empty Spotify results are a healthy request but not an + // actionable catalog. They must not enter adaptive backoff; + // retry on the bounded failure cadence instead. + retryAt := now.Add(providerFailureRetryDelay(strategy.spotifyRateLimit, r.interval)) + if strategy.spotifyRateLimit != nil { + retryAt = now.Add(syncRetryDelay(strategy.spotifyRateLimit, r.spotifyInterval)) + } + if err := r.store.ScheduleSpotifyCheck(ctx, artist.ID, retryAt); err != nil { + return outcome, err + } + } + return outcome, nil + } retryAt := now.Add(providerFailureRetryDelay(strategy.spotifyRateLimit, r.interval)) if strategy.spotifyRateLimit != nil { r.logger.Debug("Spotify check retry scheduled", "artist_id", artist.ID, @@ -979,7 +1032,7 @@ func (r *Runner) syncOne(ctx context.Context, artist store.Artist, now time.Time } return outcome, nil } - if spotifyWasDue { + if spotifyWasDue && strategy.spotifySucceeded { if strategy.spotifyRateLimit != nil { r.logger.Debug("Spotify check retry scheduled", "artist_id", artist.ID, "retry_after", syncRetryDelay(strategy.spotifyRateLimit, r.spotifyInterval).String(), @@ -1002,6 +1055,18 @@ func (r *Runner) syncOne(ctx context.Context, artist store.Artist, now time.Time return outcome, err } outcome.SpotifyBackoff = outcome.SpotifyUnchanged && !upcoming + } else if spotifyWasDue && strategy.spotifyAttempted && !strategy.spotifySucceeded { + // A failed or empty Spotify response must not be counted as an + // unchanged catalog merely because a fallback provider produced data. + // Keep Spotify on the bounded retry cadence and leave adaptive streaks + // untouched. + retryAt := now.Add(providerFailureRetryDelay(strategy.spotifyRateLimit, r.interval)) + if strategy.spotifyRateLimit != nil { + retryAt = now.Add(syncRetryDelay(strategy.spotifyRateLimit, r.spotifyInterval)) + } + if err := r.store.ScheduleSpotifyCheck(ctx, artist.ID, retryAt); err != nil { + return outcome, err + } } return outcome, nil } @@ -1072,7 +1137,7 @@ func providerFailureRetryDelay(rateLimit *catalog.SpotifyRateLimitError, interva func (r *Runner) deliver(ctx context.Context, now time.Time) (deliveryStats, error) { var summary deliveryStats - deliveries, err := r.store.DueDeliveries(ctx, now, 25) + deliveries, err := r.store.ClaimDueDeliveries(ctx, now, 25, r.workerID, 5*time.Minute) if err != nil { return summary, err } @@ -1080,7 +1145,7 @@ func (r *Runner) deliver(ctx context.Context, now time.Time) (deliveryStats, err if digestLimit < 0 { digestLimit = 0 } - digestDeliveries, err := r.store.DueDigestDeliveries(ctx, now, digestLimit) + digestDeliveries, err := r.store.ClaimDueDigestDeliveries(ctx, now, digestLimit, r.workerID, 5*time.Minute) if err != nil { return summary, err } @@ -1167,7 +1232,7 @@ func (r *Runner) deliverDigestOne(ctx context.Context, now time.Time, delivery s } } if err == nil { - if markErr := r.store.MarkDigestDeliverySent(ctx, delivery.ID, now); markErr != nil { + if markErr := r.store.MarkDigestDeliverySentOwned(ctx, delivery.ID, delivery.ClaimOwner, now); markErr != nil { if attemptID > 0 { _ = r.store.FinishDeliveryAttempt(ctx, attemptID, delivery.Destination.ID, false, markErr.Error(), nil, time.Now().UTC()) } @@ -1186,7 +1251,7 @@ func (r *Runner) deliverDigestOne(ctx context.Context, now time.Time, delivery s redactedError := notify.RedactError(err) r.logger.Warn("release digest delivery attempt failed", "digest_delivery_id", delivery.ID, "destination_id", delivery.Destination.ID, "error", redactedError) - if markErr := r.store.MarkDigestDeliveryFailed(ctx, delivery.ID, delivery.Attempts+1, redactedError, now); markErr != nil { + if markErr := r.store.MarkDigestDeliveryFailedOwned(ctx, delivery.ID, delivery.Attempts+1, redactedError, delivery.ClaimOwner, now); markErr != nil { result.err = markErr } if attemptID > 0 { @@ -1222,7 +1287,7 @@ func (r *Runner) deliverOne(ctx context.Context, now time.Time, delivery store.D } } if err == nil { - if markErr := r.store.MarkDeliverySent(ctx, delivery.ID, now); markErr != nil { + if markErr := r.store.MarkDeliverySentOwned(ctx, delivery.ID, delivery.ClaimOwner, now); markErr != nil { if attemptID > 0 { _ = r.store.FinishDeliveryAttempt(ctx, attemptID, delivery.Destination.ID, false, markErr.Error(), nil, time.Now().UTC()) } @@ -1241,7 +1306,7 @@ func (r *Runner) deliverOne(ctx context.Context, now time.Time, delivery store.D redactedError := notify.RedactError(err) r.logger.Warn("notification attempt failed", "delivery_id", delivery.ID, "destination_id", delivery.Destination.ID, "error", redactedError) - if markErr := r.store.MarkDeliveryFailed(ctx, delivery.ID, delivery.Attempts+1, redactedError, now); markErr != nil { + if markErr := r.store.MarkDeliveryFailedOwned(ctx, delivery.ID, delivery.Attempts+1, redactedError, delivery.ClaimOwner, now); markErr != nil { result.err = markErr } if attemptID > 0 { diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index 1b1a043..c045348 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -29,6 +29,17 @@ type resolutionCatalog struct { releaseCalls atomic.Int32 } +type musicBrainzCreditCatalog struct { + resolutionCatalog + creditErr error + creditCalls atomic.Int32 +} + +func (f *musicBrainzCreditCatalog) ArtistReleaseCredits(context.Context, string, []store.Release) ([]store.Release, error) { + f.creditCalls.Add(1) + return nil, f.creditErr +} + type perArtistCatalog struct { releases map[string][]store.Release errors map[string]error @@ -1181,6 +1192,49 @@ func TestPersistedSpotifyCooldownSkipsCallsAfterRestart(t *testing.T) { } } +func TestMusicBrainzCreditFailurePersistsProviderCooldown(t *testing.T) { + ctx := context.Background() + database := resolutionTestStore(t) + userID, err := database.CreateUser(ctx, "credit-cooldown@example.com", "unused", "member", "UTC", "credit-cooldown") + if err != nil { + t.Fatal(err) + } + artist, err := database.UpsertArtist(ctx, store.Artist{MBID: "credit-cooldown-artist", Name: "Credit Cooldown Artist"}) + if err != nil { + t.Fatal(err) + } + if _, err := database.Follow(ctx, userID, artist.ID); err != nil { + t.Fatal(err) + } + provider := &musicBrainzCreditCatalog{ + resolutionCatalog: resolutionCatalog{releases: []store.Release{{ + MBID: "credit-cooldown-release", Title: "Release", PrimaryType: "Album", + FirstReleaseDate: "2026-08-01", DatePrecision: 3, + }}}, + creditErr: errors.New("MusicBrainz credit endpoint unavailable"), + } + runner := testRunner(database, provider) + if err := runner.SyncArtistNow(ctx, artist); err != nil { + t.Fatal(err) + } + if provider.releaseCalls.Load() != 1 || provider.creditCalls.Load() != 1 { + t.Fatalf("MusicBrainz calls release=%d credits=%d", provider.releaseCalls.Load(), provider.creditCalls.Load()) + } + health, err := database.ProviderHealthByName(ctx, "musicbrainz") + if err != nil { + t.Fatal(err) + } + if health.NextCheckAt == nil || health.LastError == "" { + t.Fatalf("credit failure did not persist provider cooldown: %#v", health) + } + if err := runner.SyncArtistNow(ctx, artist); err != nil { + t.Fatal(err) + } + if provider.releaseCalls.Load() != 1 || provider.creditCalls.Load() != 1 { + t.Fatalf("persisted MusicBrainz cooldown was ignored: release=%d credits=%d", provider.releaseCalls.Load(), provider.creditCalls.Load()) + } +} + func TestTotalProviderFailureSchedulesBoundedRetry(t *testing.T) { ctx := context.Background() database := resolutionTestStore(t) diff --git a/internal/jobs/providers.go b/internal/jobs/providers.go index 463371a..5e83bfb 100644 --- a/internal/jobs/providers.go +++ b/internal/jobs/providers.go @@ -22,7 +22,12 @@ type providerObservation struct { nextCheckAt *time.Time attempted bool + // healthy means the provider request completed successfully. succeeded + // means it returned an actionable release batch; an empty catalog is + // healthy but intentionally hands control to the next fallback provider. + healthy bool succeeded bool + empty bool suppressed bool deferred bool cooldown time.Time @@ -38,7 +43,10 @@ type providerStrategyResult struct { providerErrors []error spotifySucceeded bool + spotifyHealthy bool + spotifyAttempted bool itunesSucceeded bool + itunesHealthy bool spotifySuppressed bool spotifyDeferred bool spotifyCooldown time.Time @@ -62,6 +70,8 @@ func (r *Runner) observeReleaseProviders(ctx context.Context, artist store.Artis } r.recordProviderStatus(ctx, artist.ID, spotify, now) result.spotifySucceeded = spotify.succeeded + result.spotifyHealthy = spotify.healthy + result.spotifyAttempted = spotify.attempted result.spotifySuppressed = spotify.suppressed result.spotifyDeferred = spotify.deferred result.spotifyCooldown = spotify.cooldown @@ -88,6 +98,7 @@ func (r *Runner) observeReleaseProviders(ctx context.Context, artist store.Artis } r.recordProviderStatus(ctx, artist.ID, itunes, now) result.itunesSucceeded = itunes.succeeded + result.itunesHealthy = itunes.healthy result.itunesRateLimit = itunes.itunesRateLimit if itunes.succeeded { result.batches = append(result.batches, store.ReleaseBatch{ @@ -164,10 +175,16 @@ func (r *Runner) observeSpotify(ctx context.Context, artist store.Artist, now ti return observation, changedErr } r.clearSpotifyProviderCooldown() - observation.succeeded = true + observation.healthy = true + observation.succeeded = len(releases) > 0 + observation.empty = len(releases) == 0 observation.releases = releases observation.spotifyChanged = changed - observation.spotifyUnchanged = !changed + // An empty response is a healthy request, but not an unchanged + // catalog. Keeping it out of the unchanged signal prevents callers + // from treating a fallback-triggering empty result as adaptive + // Spotify success. + observation.spotifyUnchanged = len(releases) > 0 && !changed observation.status = "healthy" observation.nextCheckAt = timePtr(now.Add(r.spotifyInterval)) _ = r.store.UpsertProviderHealth(ctx, "spotify", true, nil, false, false, "") @@ -237,7 +254,9 @@ func (r *Runner) observeITunes(ctx context.Context, artist store.Artist, now tim releases = append(releases, credits...) } } - observation.succeeded = true + observation.healthy = true + observation.succeeded = len(releases) > 0 + observation.empty = len(releases) == 0 observation.releases = releases observation.status = "healthy" observation.nextCheckAt = timePtr(now.Add(r.interval)) @@ -289,17 +308,33 @@ func (r *Runner) observeMusicBrainz(ctx context.Context, artist store.Artist, no if creditProvider, ok := r.catalog.(catalog.ReleaseCreditProvider); ok { credits, creditErr := creditProvider.ArtistReleaseCredits(ctx, artist.MBID, releases) if creditErr != nil { - r.logger.Debug("MusicBrainz credit enrichment failed", "artist_id", artist.ID, "error", creditErr) + // Credit discovery is a second scheduled MusicBrainz call. Keep + // its outage state in the same persisted provider cooldown as the + // release lookup so a restart cannot immediately repeat a burst. + retryAt := now.Add(r.musicBrainzFailureDelay()) + r.setMusicBrainzCooldown(retryAt) + observation.cooldown = retryAt + observation.nextCheckAt = &retryAt + observation.status = "degraded" + observation.lastError = sanitizedProviderError(creditErr) + observation.err = creditErr + _ = r.store.UpsertProviderHealth(ctx, "musicbrainz", false, &retryAt, false, false, observation.lastError) + r.logger.Warn("MusicBrainz credit observation failed", "artist_id", artist.ID, + "retry_after", retryAt.Sub(now).String(), "error", observation.lastError) } else { releases = append(releases, credits...) } } - observation.succeeded = true + observation.healthy = true + observation.succeeded = len(releases) > 0 + observation.empty = len(releases) == 0 observation.releases = releases - observation.status = "healthy" - observation.nextCheckAt = timePtr(now.Add(r.interval)) - r.clearMusicBrainzCooldown() - _ = r.store.UpsertProviderHealth(ctx, "musicbrainz", true, nil, false, false, "") + if observation.status == "" { + observation.status = "healthy" + observation.nextCheckAt = timePtr(now.Add(r.interval)) + r.clearMusicBrainzCooldown() + _ = r.store.UpsertProviderHealth(ctx, "musicbrainz", true, nil, false, false, "") + } return observation, nil } retryAt := now.Add(r.musicBrainzFailureDelay()) @@ -326,14 +361,14 @@ func (r *Runner) recordProviderStatus(ctx context.Context, artistID int64, obser if observation.attempted { attempt = timePtr(now) } - if observation.succeeded { + if observation.healthy { success = timePtr(now) } if observation.err != nil { failure = timePtr(now) } releaseCount := -1 - if observation.succeeded { + if observation.healthy { releaseCount = len(observation.releases) } if err := r.store.RecordArtistProviderStatus(ctx, store.ArtistProviderStatus{ diff --git a/internal/notify/notify.go b/internal/notify/notify.go index b9e94a1..0d0cee8 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -22,6 +22,55 @@ type NotificationSender interface { Send(context.Context, string, string, string) error } +// ErrUnsupportedTransport is returned before Shoutrrr is allowed to create a +// sender for transports whose networking cannot be brought under the +// application's resolver, redirect, and timeout policy. Keeping this check +// at both validation and send time also protects destinations created by an +// older release. +var ErrUnsupportedTransport = errors.New("notification transport is not supported") + +// ValidateTransportPolicy admits only transports for which the application +// can enforce connection-time target validation. Discord and Telegram use +// fixed upstream endpoints; ntfy and generic HTTP(S) use the scoped client +// below. Gotify and SMTP are intentionally rejected until transport-owned +// adapters are available. +func ValidateTransportPolicy(serviceURL string) error { + parsed, err := url.Parse(strings.TrimSpace(serviceURL)) + if err != nil || parsed == nil { + return ErrUnsupportedTransport + } + scheme := strings.ToLower(strings.TrimSpace(parsed.Scheme)) + switch scheme { + case "discord", "telegram", "ntfy": + return nil + case "generic+http", "generic+https": + if parsed.Host == "" { + return ErrUnsupportedTransport + } + return nil + default: + return ErrUnsupportedTransport + } +} + +// CanonicalTransportService gives persistence a stable service label even +// when a member used the advanced raw-URL field. Generic HTTP(S) URLs are +// stored as "generic" rather than the UI's "advanced" sentinel. +func CanonicalTransportService(serviceURL string) string { + parsed, err := url.Parse(strings.TrimSpace(serviceURL)) + if err != nil || parsed == nil || strings.TrimSpace(parsed.Scheme) == "" { + return "unknown" + } + switch strings.ToLower(parsed.Scheme) { + case "generic+http", "generic+https": + return "generic" + case "discord", "telegram", "ntfy": + return strings.ToLower(parsed.Scheme) + default: + return strings.ToLower(parsed.Scheme) + } +} + type ShoutrrrSender struct { // AllowPrivateTargets is an explicit opt-in for self-hosted notification // services on a trusted LAN. It is disabled by default to keep a member @@ -176,6 +225,9 @@ func (s ShoutrrrSender) Validate(serviceURL string) error { if strings.TrimSpace(serviceURL) == "" { return errors.New("notification URL is required") } + if err := ValidateTransportPolicy(serviceURL); err != nil { + return err + } if err := validateOutboundTarget(context.Background(), serviceURL, s.AllowPrivateTargets, false); err != nil { return err } @@ -197,6 +249,9 @@ func (s ShoutrrrSender) send(ctx context.Context, serviceURL, title, body string if err := ctx.Err(); err != nil { return err } + if err := ValidateTransportPolicy(serviceURL); err != nil { + return err + } sendCtx, cancel := context.WithTimeout(ctx, s.sendTimeout()) defer cancel() if err := validateOutboundTargetWithLookup(sendCtx, serviceURL, s.AllowPrivateTargets, true, s.lookupIP); err != nil { diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go index 6f1dba1..7413fd3 100644 --- a/internal/notify/notify_test.go +++ b/internal/notify/notify_test.go @@ -120,6 +120,48 @@ func TestBuildURLRejectsIncompleteInput(t *testing.T) { } } +func TestValidateTransportPolicyMatrix(t *testing.T) { + for _, serviceURL := range []string{ + "discord://token@123456", + "telegram://token@telegram?chats=-100123", + "ntfy://ntfy.sh/releases", + "generic+http://hooks.example/releases", + "generic+https://hooks.example/releases", + } { + if err := ValidateTransportPolicy(serviceURL); err != nil { + t.Errorf("supported transport %q rejected: %v", serviceURL, err) + } + } + for _, serviceURL := range []string{ + "gotify://push.example/token", + "smtp://mail.example:587/", + "matrix://room.example/room", + "http://hooks.example/releases", + "generic+ftp://hooks.example/releases", + } { + if err := ValidateTransportPolicy(serviceURL); !errors.Is(err, ErrUnsupportedTransport) { + t.Errorf("unsupported transport %q error=%v, want ErrUnsupportedTransport", serviceURL, err) + } + } +} + +func TestCanonicalTransportService(t *testing.T) { + tests := map[string]string{ + "generic+http://hooks.example/releases": "generic", + "generic+https://hooks.example/releases": "generic", + "DISCORD://token@123456": "discord", + "telegram://token@telegram?chats=-100": "telegram", + "ntfy://ntfy.sh/releases": "ntfy", + "smtp://mail.example:587/": "smtp", + "not a URL": "unknown", + } + for serviceURL, want := range tests { + if got := CanonicalTransportService(serviceURL); got != want { + t.Errorf("CanonicalTransportService(%q)=%q, want %q", serviceURL, got, want) + } + } +} + func TestRedactErrorRemovesDestinationCredentials(t *testing.T) { message := RedactError(errors.New(`send failed to ntfy://user:pass@example.test/topic?token=abc: password=secret`)) if strings.Contains(message, "example.test") || strings.Contains(message, "pass@") || strings.Contains(message, "?token=abc") || strings.Contains(message, "=secret") { diff --git a/internal/store/auth.go b/internal/store/auth.go index b8ea363..44747a3 100644 --- a/internal/store/auth.go +++ b/internal/store/auth.go @@ -141,6 +141,53 @@ func (s *Store) CreateUser(ctx context.Context, email, hash, role, timezone, use return id, nil } +// CreateInitialAdmin atomically establishes the first account. The setup +// handler may be reached concurrently by two browser sessions; checking the +// count on a read connection before inserting is not sufficient to protect +// the one-time setup invariant. +func (s *Store) CreateInitialAdmin(ctx context.Context, email, hash, timezone, username string) (int64, error) { + email = strings.ToLower(strings.TrimSpace(email)) + if email == "" || !strings.Contains(email, "@") { + return 0, errors.New("a valid email address is required") + } + if _, err := time.LoadLocation(timezone); err != nil { + return 0, errors.New("invalid IANA timezone") + } + tx, err := s.beginWriteTx(ctx) + if err != nil { + return 0, err + } + defer func() { _ = tx.Rollback() }() + var count int + if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM users`).Scan(&count); err != nil { + return 0, err + } + if count != 0 { + return 0, ErrSetupCompleted + } + username = strings.TrimSpace(username) + if username == "" { + username = derivedUsername(email, 1, nil) + } + username, err = validateUsername(username) + if err != nil { + return 0, err + } + result, err := tx.ExecContext(ctx, `INSERT INTO users(email,username,password_hash,role,timezone,created_at) + VALUES(?,?,?,?,?,?)`, email, username, hash, "admin", timezone, nowText()) + if err != nil { + return 0, err + } + id, err := result.LastInsertId() + if err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, err + } + return id, nil +} + func usernameTakenTx(ctx context.Context, tx *sql.Tx, username string, exceptID int64) (bool, error) { query := `SELECT EXISTS(SELECT 1 FROM users WHERE username COLLATE NOCASE=?` args := []any{username} diff --git a/internal/store/calendar.go b/internal/store/calendar.go index a524cdb..1151f1d 100644 --- a/internal/store/calendar.go +++ b/internal/store/calendar.go @@ -172,11 +172,11 @@ func (s *Store) QueueDueReleaseDigests(ctx context.Context, now time.Time) (int, _ = tx.Rollback() return queued, err } - deliveryResult, err := tx.ExecContext(ctx, `INSERT INTO release_digest_deliveries + deliveryResult, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO release_digest_deliveries (run_id,destination_id,status,next_attempt_at) - SELECT ?,d.id,'pending',? FROM destinations d + SELECT ?,d.id,`+destinationQueueStatus("d")+`,? FROM destinations d LEFT JOIN destination_health dh ON dh.destination_id=d.id - WHERE d.user_id=? AND d.enabled=1 AND `+destinationAdmissionPredicate, runID, timeText(now), user.ID) + WHERE d.user_id=? AND d.enabled=1`, runID, timeText(now), user.ID) if err != nil { _ = tx.Rollback() return queued, err @@ -187,7 +187,9 @@ func (s *Store) QueueDueReleaseDigests(ctx context.Context, now time.Time) (int, return queued, err } if deliveryCount == 0 { - if _, err := tx.ExecContext(ctx, `UPDATE release_digest_runs SET status='sent' WHERE id=?`, runID); err != nil { + // No admitted destination is not a successful send. Keep the run + // pending so it remains visible for an explicit replay/recovery action. + if _, err := tx.ExecContext(ctx, `UPDATE release_digest_runs SET status='pending' WHERE id=?`, runID); err != nil { _ = tx.Rollback() return queued, err } diff --git a/internal/store/migrations/025_recovery_leases.sql b/internal/store/migrations/025_recovery_leases.sql new file mode 100644 index 0000000..d9fd37d --- /dev/null +++ b/internal/store/migrations/025_recovery_leases.sql @@ -0,0 +1,76 @@ +-- Recovery leases and explicit blocked delivery state. The queue remains +-- at-least-once: a worker claims a row for a short period, and an expired +-- claim becomes runnable again during the next maintenance pass. + +ALTER TABLE destinations ADD COLUMN transport_status TEXT NOT NULL DEFAULT 'supported'; +ALTER TABLE destinations ADD COLUMN transport_message TEXT NOT NULL DEFAULT ''; +UPDATE destinations + SET transport_status='unsupported', + transport_message='This destination uses a transport that is no longer supported; replace it.' + WHERE lower(service) NOT IN ('ntfy','discord','telegram','generic'); + +ALTER TABLE manual_sync_requests ADD COLUMN lease_owner TEXT; +ALTER TABLE manual_sync_requests ADD COLUMN lease_expires_at TEXT; +ALTER TABLE manual_sync_requests ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0; + +ALTER TABLE delivery_attempts ADD COLUMN abandoned_at TEXT; + +-- SQLite cannot alter a CHECK constraint in place. Rebuild the two queue +-- tables once so blocked rows can be retained and replayed after recovery. +CREATE TABLE deliveries_recovery_new ( + id INTEGER PRIMARY KEY, + event_id INTEGER NOT NULL REFERENCES notification_events(id) ON DELETE CASCADE, + destination_id INTEGER NOT NULL REFERENCES destinations(id) ON DELETE CASCADE, + status TEXT NOT NULL CHECK(status IN ('pending','sent','failed','blocked')), + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TEXT NOT NULL, + last_error TEXT NOT NULL DEFAULT '', + sent_at TEXT, + claim_owner TEXT, + claim_expires_at TEXT, + UNIQUE(event_id, destination_id) +); +INSERT INTO deliveries_recovery_new + (id,event_id,destination_id,status,attempts,next_attempt_at,last_error,sent_at) +SELECT id,event_id,destination_id,status,attempts,next_attempt_at,last_error,sent_at + FROM deliveries; +DROP TABLE deliveries; +ALTER TABLE deliveries_recovery_new RENAME TO deliveries; +CREATE INDEX deliveries_due ON deliveries(status,next_attempt_at); +CREATE INDEX deliveries_status_due_destination + ON deliveries(status,next_attempt_at,destination_id); + +CREATE TABLE release_digest_deliveries_recovery_new ( + id INTEGER PRIMARY KEY, + run_id INTEGER NOT NULL REFERENCES release_digest_runs(id) ON DELETE CASCADE, + destination_id INTEGER NOT NULL REFERENCES destinations(id) ON DELETE CASCADE, + status TEXT NOT NULL CHECK(status IN ('pending','sent','failed','blocked')), + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TEXT NOT NULL, + last_error TEXT NOT NULL DEFAULT '', + sent_at TEXT, + claim_owner TEXT, + claim_expires_at TEXT, + UNIQUE(run_id, destination_id) +); +INSERT INTO release_digest_deliveries_recovery_new + (id,run_id,destination_id,status,attempts,next_attempt_at,last_error,sent_at) +SELECT id,run_id,destination_id,status,attempts,next_attempt_at,last_error,sent_at + FROM release_digest_deliveries; +DROP TABLE release_digest_deliveries; +ALTER TABLE release_digest_deliveries_recovery_new RENAME TO release_digest_deliveries; +CREATE INDEX release_digest_deliveries_due + ON release_digest_deliveries(status,next_attempt_at); +CREATE INDEX release_digest_deliveries_status_due_destination + ON release_digest_deliveries(status,next_attempt_at,destination_id); + +CREATE INDEX IF NOT EXISTS destinations_transport_status + ON destinations(transport_status,enabled); +CREATE INDEX IF NOT EXISTS manual_sync_leases + ON manual_sync_requests(status,lease_expires_at); +CREATE INDEX IF NOT EXISTS deliveries_claim_expiry + ON deliveries(claim_expires_at,status); +CREATE INDEX IF NOT EXISTS release_digest_deliveries_claim_expiry + ON release_digest_deliveries(claim_expires_at,status); +CREATE INDEX IF NOT EXISTS delivery_attempts_started + ON delivery_attempts(status,started_at); diff --git a/internal/store/notification_history_test.go b/internal/store/notification_history_test.go index dcee1e6..6925f95 100644 --- a/internal/store/notification_history_test.go +++ b/internal/store/notification_history_test.go @@ -161,6 +161,20 @@ func TestDestinationHealthAndDeliveryAttemptAudit(t *testing.T) { if err != nil || health[destination.ID].Status != "paused" { t.Fatalf("destination was not paused after five failures: health=%#v err=%v", health, err) } + // A send that was already in flight when the fifth failure paused the + // circuit must not silently reopen it. Recovery is an explicit operator + // action through RetryFailedDeliveries. + inFlight, err := s.StartDeliveryAttempt(ctx, 99, 0, destination, 1, now.Add(10*time.Second)) + if err != nil { + t.Fatal(err) + } + if err := s.FinishDeliveryAttempt(ctx, inFlight, destination.ID, true, "", nil, now.Add(11*time.Second)); err != nil { + t.Fatal(err) + } + health, err = s.DestinationHealthByUser(ctx, userID) + if err != nil || health[destination.ID].Status != "paused" { + t.Fatalf("in-flight success reopened paused destination: health=%#v err=%v", health, err) + } admin, err := s.AdminDestinationHealth(ctx) if err != nil || len(admin) != 1 || admin[0].UserEmail != "health@example.com" || admin[0].Status != "paused" { t.Fatalf("admin health=%#v err=%v", admin, err) @@ -225,3 +239,260 @@ func TestDestinationHealthAndDeliveryAttemptAudit(t *testing.T) { t.Fatalf("cross-user retry error=%v", err) } } + +func TestUnsupportedDestinationRemainsVisibleAndQueuesBlockedWork(t *testing.T) { + ctx := context.Background() + s := testStore(t) + userID, err := s.CreateUser(ctx, "unsupported-destination@example.com", "hash", "member", "UTC", "unsupported") + if err != nil { + t.Fatal(err) + } + artist, err := s.UpsertArtist(ctx, Artist{MBID: "unsupported-destination-artist", Name: "Unsupported Artist"}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Follow(ctx, userID, artist.ID); err != nil { + t.Fatal(err) + } + if err := s.AddDestination(ctx, userID, "Legacy push", "gotify", []byte("encrypted")); err != nil { + t.Fatal(err) + } + destinations, err := s.Destinations(ctx, userID) + if err != nil || len(destinations) != 1 { + t.Fatalf("destinations=%#v err=%v", destinations, err) + } + destination := destinations[0] + if destination.TransportStatus != "unsupported" || destination.TransportMessage == "" { + t.Fatalf("legacy destination status=%q message=%q", destination.TransportStatus, destination.TransportMessage) + } + + now := time.Date(2026, time.August, 13, 12, 0, 0, 0, time.UTC) + if err := s.ApplyReleaseSync(ctx, artist, []Release{{ + MBID: "unsupported-destination-release", Title: "Blocked Album", PrimaryType: "Album", + FirstReleaseDate: "2026-09-01", DatePrecision: 3, + }}, now); err != nil { + t.Fatal(err) + } + var status string + if err := s.DB.QueryRowContext(ctx, `SELECT d.status FROM deliveries d JOIN destinations dst ON dst.id=d.destination_id WHERE dst.id=?`, destination.ID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "blocked" { + t.Fatalf("unsupported destination delivery status=%q, want blocked", status) + } + if due, err := s.DueDeliveries(ctx, now.Add(24*time.Hour), 10); err != nil || len(due) != 0 { + t.Fatalf("blocked destination returned due deliveries=%#v err=%v", due, err) + } + if count, err := s.RetryFailedDeliveries(ctx, userID, destination.ID, now); err != nil || count != 1 { + t.Fatalf("blocked destination retry count=%d err=%v", count, err) + } + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM deliveries WHERE destination_id=?`, destination.ID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "blocked" { + t.Fatalf("unsupported retry status=%q, want blocked", status) + } + // Replacing the destination's transport is an explicit operator action; + // replay then admits the retained event without creating a duplicate. + if _, err := s.DB.ExecContext(ctx, `UPDATE destinations SET service='generic',transport_status='supported',transport_message='' WHERE id=?`, destination.ID); err != nil { + t.Fatal(err) + } + if count, err := s.RetryFailedDeliveries(ctx, userID, destination.ID, now); err != nil || count != 1 { + t.Fatalf("recovered destination retry count=%d err=%v", count, err) + } + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM deliveries WHERE destination_id=?`, destination.ID).Scan(&status); err != nil { + t.Fatal(err) + } + if status != "pending" { + t.Fatalf("recovered destination status=%q, want pending", status) + } +} + +func TestPausingDestinationBlocksQueuedNormalAndDigestWork(t *testing.T) { + ctx := context.Background() + s := testStore(t) + userID, err := s.CreateUser(ctx, "paused-queue@example.com", "hash", "member", "UTC", "paused-queue") + if err != nil { + t.Fatal(err) + } + if err := s.AddDestination(ctx, userID, "Primary", "ntfy", []byte("encrypted")); err != nil { + t.Fatal(err) + } + destinations, err := s.Destinations(ctx, userID) + if err != nil || len(destinations) != 1 { + t.Fatalf("destinations=%#v err=%v", destinations, err) + } + destination := destinations[0] + artist, err := s.UpsertArtist(ctx, Artist{MBID: "paused-queue-artist", Name: "Paused Queue Artist"}) + if err != nil { + t.Fatal(err) + } + now := time.Date(2026, time.August, 13, 10, 0, 0, 0, time.UTC) + release, err := s.DB.ExecContext(ctx, `INSERT INTO release_groups + (mbid,artist_id,title,primary_type,secondary_types,first_release_date,date_precision,musicbrainz_url,first_observed_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,?)`, "paused-queue-release", artist.ID, "Paused Album", "Album", "[]", "2026-08-13", 3, "https://musicbrainz.org/release-group/paused-queue", nowText(), nowText()) + if err != nil { + t.Fatal(err) + } + releaseID, err := release.LastInsertId() + if err != nil { + t.Fatal(err) + } + event, err := s.DB.ExecContext(ctx, `INSERT INTO notification_events(user_id,release_group_id,event_type,title,body,created_at) VALUES(?,?,?,?,?,?)`, + userID, releaseID, "announcement", "Paused Album", "body", nowText()) + if err != nil { + t.Fatal(err) + } + eventID, err := event.LastInsertId() + if err != nil { + t.Fatal(err) + } + if _, err := s.DB.ExecContext(ctx, `INSERT INTO deliveries(event_id,destination_id,status,attempts,next_attempt_at,last_error) VALUES(?,?,?,?,?,?)`, + eventID, destination.ID, "pending", 0, timeText(now), ""); err != nil { + t.Fatal(err) + } + digest, err := s.DB.ExecContext(ctx, `INSERT INTO release_digest_runs(user_id,frequency,period_start,title,body,release_count,status,created_at) VALUES(?,?,?,?,?,?,?,?)`, + userID, "daily", "2026-08-13", "Digest", "Body", 1, "pending", nowText()) + if err != nil { + t.Fatal(err) + } + digestID, err := digest.LastInsertId() + if err != nil { + t.Fatal(err) + } + if _, err := s.DB.ExecContext(ctx, `INSERT INTO release_digest_deliveries(run_id,destination_id,status,next_attempt_at) VALUES(?,?,?,?)`, + digestID, destination.ID, "pending", timeText(now)); err != nil { + t.Fatal(err) + } + + for attemptNumber := 1; attemptNumber <= 5; attemptNumber++ { + attempt, err := s.StartDeliveryAttempt(ctx, int64(500+attemptNumber), 0, destination, attemptNumber, now.Add(time.Duration(attemptNumber)*time.Second)) + if err != nil { + t.Fatal(err) + } + if err := s.FinishDeliveryAttempt(ctx, attempt, destination.ID, false, "delivery failed", &now, now.Add(time.Duration(attemptNumber)*time.Second)); err != nil { + t.Fatal(err) + } + } + var normalStatus, digestStatus string + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM deliveries WHERE event_id=?`, eventID).Scan(&normalStatus); err != nil { + t.Fatal(err) + } + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM release_digest_deliveries WHERE run_id=?`, digestID).Scan(&digestStatus); err != nil { + t.Fatal(err) + } + if normalStatus != "blocked" || digestStatus != "blocked" { + t.Fatalf("queued work statuses normal=%q digest=%q, want blocked", normalStatus, digestStatus) + } + if count, err := s.RetryFailedDeliveries(ctx, userID, destination.ID, now); err != nil || count != 2 { + t.Fatalf("retry count=%d err=%v, want both blocked rows", count, err) + } + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM deliveries WHERE event_id=?`, eventID).Scan(&normalStatus); err != nil { + t.Fatal(err) + } + if err := s.DB.QueryRowContext(ctx, `SELECT status FROM release_digest_deliveries WHERE run_id=?`, digestID).Scan(&digestStatus); err != nil { + t.Fatal(err) + } + if normalStatus != "pending" || digestStatus != "pending" { + t.Fatalf("replayed work statuses normal=%q digest=%q, want pending", normalStatus, digestStatus) + } +} + +func TestDurableDeliveryClaimsRecoverExpiredWork(t *testing.T) { + ctx := context.Background() + s := testStore(t) + userID, err := s.CreateInitialAdmin(ctx, "claims@example.com", "hash", "UTC", "claims-admin") + if err != nil { + t.Fatal(err) + } + if _, err := s.CreateInitialAdmin(ctx, "second@example.com", "hash", "UTC", "second-admin"); !errors.Is(err, ErrSetupCompleted) { + t.Fatalf("second initial admin error=%v, want ErrSetupCompleted", err) + } + artist, err := s.UpsertArtist(ctx, Artist{MBID: "claims-artist", Name: "Claims Artist"}) + if err != nil { + t.Fatal(err) + } + if _, err := s.Follow(ctx, userID, artist.ID); err != nil { + t.Fatal(err) + } + if err := s.AddDestination(ctx, userID, "Primary", "ntfy", []byte("encrypted")); err != nil { + t.Fatal(err) + } + destinations, err := s.Destinations(ctx, userID) + if err != nil || len(destinations) != 1 { + t.Fatalf("destinations=%#v err=%v", destinations, err) + } + destination := destinations[0] + now := time.Date(2026, time.August, 13, 12, 0, 0, 0, time.UTC) + release, err := s.DB.ExecContext(ctx, `INSERT INTO release_groups + (mbid,artist_id,title,primary_type,secondary_types,first_release_date,date_precision,musicbrainz_url,first_observed_at,updated_at) + VALUES(?,?,?,?,?,?,?,?,?,?)`, "claims-release", artist.ID, "Claims Album", "Album", "[]", "2026-08-13", 3, "https://musicbrainz.org/release-group/claims", nowText(), nowText()) + if err != nil { + t.Fatal(err) + } + releaseID, err := release.LastInsertId() + if err != nil { + t.Fatal(err) + } + event, err := s.DB.ExecContext(ctx, `INSERT INTO notification_events(user_id,release_group_id,event_type,title,body,created_at) VALUES(?,?,?,?,?,?)`, + userID, releaseID, "announcement", "Claims Album", "body", nowText()) + if err != nil { + t.Fatal(err) + } + eventID, err := event.LastInsertId() + if err != nil { + t.Fatal(err) + } + if _, err := s.DB.ExecContext(ctx, `INSERT INTO deliveries(event_id,destination_id,status,attempts,next_attempt_at,last_error) VALUES(?,?,?,?,?,?)`, + eventID, destination.ID, "pending", 0, timeText(now.Add(-time.Minute)), ""); err != nil { + t.Fatal(err) + } + digest, err := s.DB.ExecContext(ctx, `INSERT INTO release_digest_runs(user_id,frequency,period_start,title,body,release_count,status,created_at) VALUES(?,?,?,?,?,?,?,?)`, + userID, "daily", "2026-08-13", "Digest", "Body", 1, "pending", nowText()) + if err != nil { + t.Fatal(err) + } + digestID, err := digest.LastInsertId() + if err != nil { + t.Fatal(err) + } + if _, err := s.DB.ExecContext(ctx, `INSERT INTO release_digest_deliveries(run_id,destination_id,status,next_attempt_at) VALUES(?,?,?,?)`, + digestID, destination.ID, "pending", timeText(now.Add(-time.Minute))); err != nil { + t.Fatal(err) + } + + normal, err := s.ClaimDueDeliveries(ctx, now, 10, "worker-one", time.Minute) + if err != nil || len(normal) != 1 || normal[0].ClaimOwner != "worker-one" { + t.Fatalf("claimed normal deliveries=%#v err=%v", normal, err) + } + if again, err := s.ClaimDueDeliveries(ctx, now, 10, "worker-two", time.Minute); err != nil || len(again) != 0 { + t.Fatalf("second worker claimed normal delivery=%#v err=%v", again, err) + } + digests, err := s.ClaimDueDigestDeliveries(ctx, now, 10, "worker-one", time.Minute) + if err != nil || len(digests) != 1 || digests[0].ClaimOwner != "worker-one" { + t.Fatalf("claimed digest deliveries=%#v err=%v", digests, err) + } + if _, err := s.DB.ExecContext(ctx, `UPDATE deliveries SET claim_expires_at=? WHERE event_id=?`, timeText(now.Add(-time.Second)), eventID); err != nil { + t.Fatal(err) + } + if _, err := s.DB.ExecContext(ctx, `UPDATE release_digest_deliveries SET claim_expires_at=? WHERE run_id=?`, timeText(now.Add(-time.Second)), digestID); err != nil { + t.Fatal(err) + } + recovered, err := s.RecoverExpiredWork(ctx, now) + if err != nil || recovered != 2 { + t.Fatalf("recovered=%d err=%v, want normal and digest claims", recovered, err) + } + var owner sql.NullString + if err := s.DB.QueryRowContext(ctx, `SELECT claim_owner FROM deliveries WHERE event_id=?`, eventID).Scan(&owner); err != nil { + t.Fatal(err) + } + if owner.Valid { + t.Fatalf("normal delivery claim owner=%q after recovery", owner.String) + } + if err := s.DB.QueryRowContext(ctx, `SELECT claim_owner FROM release_digest_deliveries WHERE run_id=?`, digestID).Scan(&owner); err != nil { + t.Fatal(err) + } + if owner.Valid { + t.Fatalf("digest delivery claim owner=%q after recovery", owner.String) + } +} diff --git a/internal/store/notifications.go b/internal/store/notifications.go index cea7c90..6f88945 100644 --- a/internal/store/notifications.go +++ b/internal/store/notifications.go @@ -14,9 +14,32 @@ import ( // destinationAdmissionPredicate is shared by event/digest fan-out and both // due-delivery readers. A destination's historical failures are informational; -// only the explicit persisted paused state blocks new work. +// only the explicit persisted paused state blocks new work. Transport policy +// is applied alongside it so legacy Gotify/SMTP/unknown advanced destinations +// remain visible but cannot initiate network activity. const destinationAdmissionPredicate = `COALESCE(dh.status,'healthy')<>'paused'` +func supportedDestinationServicePredicate(alias string) string { + return `COALESCE(` + alias + `.transport_status,'supported')='supported' AND LOWER(` + alias + `.service) IN ('discord','telegram','ntfy','generic')` +} + +func destinationTransportStatus(service string) (string, string) { + switch strings.ToLower(strings.TrimSpace(service)) { + case "ntfy", "discord", "telegram", "generic": + return "supported", "" + default: + return "unsupported", "This destination uses a transport that is no longer supported; replace it." + } +} + +// destinationQueueStatus is used at event/digest creation time. A paused or +// legacy-unsupported destination still receives a durable blocked row so the +// event is never silently lost; due readers only admit supported/presently +// healthy destinations. +func destinationQueueStatus(alias string) string { + return `CASE WHEN COALESCE(` + alias + `.transport_status,'supported')<>'supported' OR LOWER(` + alias + `.service) NOT IN ('discord','telegram','ntfy','generic') OR COALESCE(dh.status,'healthy')='paused' THEN 'blocked' ELSE 'pending' END` +} + var ( deliveryURLPattern = regexp.MustCompile(`(?i)\b[a-z][a-z0-9+.-]*://[^\s"'<>]+`) deliveryCredentialPattern = regexp.MustCompile(`(?i)(password|passwd|token|secret|api[_-]?key|key)=([^&\s]+)`) @@ -33,8 +56,9 @@ func (s *Store) AddDestination(ctx context.Context, userID int64, name, service if err != nil { return err } - _, err = s.execWriteContext(ctx, `INSERT INTO destinations(user_id,name,service,encrypted_url,created_at) - VALUES(?,?,?,?,?)`, userID, name, service, encrypted, nowText()) + status, message := destinationTransportStatus(service) + _, err = s.execWriteContext(ctx, `INSERT INTO destinations(user_id,name,service,encrypted_url,created_at,transport_status,transport_message) + VALUES(?,?,?,?,?,?,?)`, userID, name, service, encrypted, nowText(), status, message) return err } @@ -86,7 +110,7 @@ func (s *Store) RenameDestination(ctx context.Context, userID, destinationID int return nil } func (s *Store) Destinations(ctx context.Context, userID int64) ([]Destination, error) { - rows, err := s.readerDB().QueryContext(ctx, `SELECT id,user_id,name,service,encrypted_url,enabled + rows, err := s.readerDB().QueryContext(ctx, `SELECT id,user_id,name,service,encrypted_url,enabled,COALESCE(transport_status,'supported'),COALESCE(transport_message,'') FROM destinations WHERE user_id=? ORDER BY name`, userID) if err != nil { return nil, err @@ -95,7 +119,7 @@ func (s *Store) Destinations(ctx context.Context, userID int64) ([]Destination, var result []Destination for rows.Next() { var d Destination - if err := rows.Scan(&d.ID, &d.UserID, &d.Name, &d.Service, &d.EncryptedURL, &d.Enabled); err != nil { + if err := rows.Scan(&d.ID, &d.UserID, &d.Name, &d.Service, &d.EncryptedURL, &d.Enabled, &d.TransportStatus, &d.TransportMessage); err != nil { return nil, err } result = append(result, d) @@ -108,10 +132,12 @@ func (s *Store) Destinations(ctx context.Context, userID int64) ([]Destination, // healthy until their first delivery attempt. func (s *Store) DestinationHealthByUser(ctx context.Context, userID int64) (map[int64]DestinationHealth, error) { rows, err := s.readerDB().QueryContext(ctx, `SELECT d.id, - COALESCE(h.status,'healthy'), + CASE WHEN `+supportedDestinationServicePredicate("d")+` THEN COALESCE(h.status,'healthy') ELSE 'unsupported' END, COALESCE(h.consecutive_failures,0), (SELECT COUNT(*) FROM deliveries WHERE destination_id=d.id AND status='pending')+ (SELECT COUNT(*) FROM release_digest_deliveries WHERE destination_id=d.id AND status='pending'), + (SELECT COUNT(*) FROM deliveries WHERE destination_id=d.id AND status='blocked')+ + (SELECT COUNT(*) FROM release_digest_deliveries WHERE destination_id=d.id AND status='blocked'), (SELECT COUNT(*) FROM deliveries WHERE destination_id=d.id AND status='failed')+ (SELECT COUNT(*) FROM release_digest_deliveries WHERE destination_id=d.id AND status='failed'), h.last_success_at,h.last_failure_at,h.next_retry_at,COALESCE(h.last_error,''), @@ -128,7 +154,7 @@ func (s *Store) DestinationHealthByUser(ctx context.Context, userID int64) (map[ var status string var lastSuccess, lastFailure, nextRetry, lastError, updated sql.NullString if err := rows.Scan(&health.DestinationID, &status, &health.ConsecutiveFailures, - &health.PendingCount, &health.FailedCount, &lastSuccess, &lastFailure, &nextRetry, &lastError, &updated); err != nil { + &health.PendingCount, &health.BlockedCount, &health.FailedCount, &lastSuccess, &lastFailure, &nextRetry, &lastError, &updated); err != nil { return nil, err } health.Status = status @@ -155,16 +181,18 @@ func (s *Store) DestinationHealthByUser(ctx context.Context, userID int64) (map[ // dashboard. It never includes encrypted destination URLs or message bodies. func (s *Store) AdminDestinationHealth(ctx context.Context) ([]AdminDestinationHealth, error) { rows, err := s.readerDB().QueryContext(ctx, `SELECT d.id,u.email,d.name,d.service, - COALESCE(h.status,'healthy'),COALESCE(h.consecutive_failures,0), + CASE WHEN `+supportedDestinationServicePredicate("d")+` THEN COALESCE(h.status,'healthy') ELSE 'unsupported' END,COALESCE(h.consecutive_failures,0), (SELECT COUNT(*) FROM deliveries WHERE destination_id=d.id AND status='pending')+ (SELECT COUNT(*) FROM release_digest_deliveries WHERE destination_id=d.id AND status='pending'), + (SELECT COUNT(*) FROM deliveries WHERE destination_id=d.id AND status='blocked')+ + (SELECT COUNT(*) FROM release_digest_deliveries WHERE destination_id=d.id AND status='blocked'), (SELECT COUNT(*) FROM deliveries WHERE destination_id=d.id AND status='failed')+ (SELECT COUNT(*) FROM release_digest_deliveries WHERE destination_id=d.id AND status='failed'), h.last_success_at,h.last_failure_at,h.next_retry_at,COALESCE(h.last_error,''), COALESCE(h.updated_at,d.created_at) FROM destinations d JOIN users u ON u.id=d.user_id LEFT JOIN destination_health h ON h.destination_id=d.id - ORDER BY CASE COALESCE(h.status,'healthy') WHEN 'paused' THEN 0 WHEN 'degraded' THEN 1 ELSE 2 END, + ORDER BY CASE WHEN `+supportedDestinationServicePredicate("d")+` THEN CASE COALESCE(h.status,'healthy') WHEN 'paused' THEN 1 WHEN 'degraded' THEN 2 ELSE 3 END ELSE 0 END, d.name,u.email`) if err != nil { return nil, err @@ -176,7 +204,7 @@ func (s *Store) AdminDestinationHealth(ctx context.Context) ([]AdminDestinationH var lastSuccess, lastFailure, nextRetry, lastError, updated sql.NullString if err := rows.Scan(&health.DestinationID, &health.UserEmail, &health.DestinationName, &health.Service, &health.Status, &health.ConsecutiveFailures, &health.PendingCount, - &health.FailedCount, &lastSuccess, &lastFailure, &nextRetry, &lastError, &updated); err != nil { + &health.BlockedCount, &health.FailedCount, &lastSuccess, &lastFailure, &nextRetry, &lastError, &updated); err != nil { return nil, err } var parseErr error @@ -246,7 +274,13 @@ func (s *Store) FinishDeliveryAttempt(ctx context.Context, attemptID, destinatio if success { if _, err := tx.ExecContext(ctx, `INSERT INTO destination_health(destination_id,status,consecutive_failures,last_success_at,next_retry_at,last_error,updated_at) VALUES(?,'healthy',0,?,NULL,'',?) - ON CONFLICT(destination_id) DO UPDATE SET status='healthy',consecutive_failures=0,last_success_at=excluded.last_success_at,next_retry_at=NULL,last_error='',updated_at=excluded.updated_at`, + ON CONFLICT(destination_id) DO UPDATE SET + status=CASE WHEN destination_health.status='paused' THEN 'paused' ELSE 'healthy' END, + consecutive_failures=CASE WHEN destination_health.status='paused' THEN destination_health.consecutive_failures ELSE 0 END, + last_success_at=excluded.last_success_at, + next_retry_at=CASE WHEN destination_health.status='paused' THEN destination_health.next_retry_at ELSE NULL END, + last_error=CASE WHEN destination_health.status='paused' THEN destination_health.last_error ELSE '' END, + updated_at=excluded.updated_at`, destinationID, timeText(finished), timeText(finished)); err != nil { return err } @@ -260,6 +294,32 @@ func (s *Store) FinishDeliveryAttempt(ctx context.Context, attemptID, destinatio destinationID, timeText(finished), nullableTime(nextRetry), message, timeText(finished)); err != nil { return err } + // Once the circuit pauses a destination, work that was already queued + // must remain durable but cannot stay in the runnable queue. Convert all + // currently pending rows to blocked atomically with the health update so + // a maintenance tick cannot repeatedly claim work that is guaranteed to + // fail. RetryFailedDeliveries explicitly moves these rows back to pending + // after an operator recovers the destination. + var healthStatus string + if err := tx.QueryRowContext(ctx, `SELECT status FROM destination_health WHERE destination_id=?`, destinationID).Scan(&healthStatus); err != nil { + return err + } + if healthStatus == "paused" { + for _, query := range []string{ + `UPDATE deliveries + SET status='blocked',claim_owner=NULL,claim_expires_at=NULL, + last_error=CASE WHEN last_error='' THEN 'destination paused after repeated failures' ELSE last_error END + WHERE destination_id=? AND status='pending'`, + `UPDATE release_digest_deliveries + SET status='blocked',claim_owner=NULL,claim_expires_at=NULL, + last_error=CASE WHEN last_error='' THEN 'destination paused after repeated failures' ELSE last_error END + WHERE destination_id=? AND status='pending'`, + } { + if _, err := tx.ExecContext(ctx, query, destinationID); err != nil { + return err + } + } + } } return tx.Commit() } @@ -282,8 +342,10 @@ func (s *Store) RetryFailedDeliveries(ctx context.Context, userID, destinationID } count := 0 for _, query := range []string{ - `UPDATE deliveries SET status='pending',attempts=0,next_attempt_at=?,last_error='' WHERE destination_id=? AND status='failed'`, - `UPDATE release_digest_deliveries SET status='pending',attempts=0,next_attempt_at=?,last_error='' WHERE destination_id=? AND status='failed'`, + `UPDATE deliveries SET status=CASE WHEN COALESCE((SELECT transport_status FROM destinations WHERE id=deliveries.destination_id),'supported')='supported' AND LOWER(COALESCE((SELECT service FROM destinations WHERE id=deliveries.destination_id),'')) IN ('discord','telegram','ntfy','generic') THEN 'pending' ELSE 'blocked' END, + attempts=0,next_attempt_at=?,last_error='',claim_owner=NULL,claim_expires_at=NULL WHERE destination_id=? AND status IN ('failed','blocked')`, + `UPDATE release_digest_deliveries SET status=CASE WHEN COALESCE((SELECT transport_status FROM destinations WHERE id=release_digest_deliveries.destination_id),'supported')='supported' AND LOWER(COALESCE((SELECT service FROM destinations WHERE id=release_digest_deliveries.destination_id),'')) IN ('discord','telegram','ntfy','generic') THEN 'pending' ELSE 'blocked' END, + attempts=0,next_attempt_at=?,last_error='',claim_owner=NULL,claim_expires_at=NULL WHERE destination_id=? AND status IN ('failed','blocked')`, } { result, updateErr := tx.ExecContext(ctx, query, timeText(now), destinationID) if updateErr != nil { @@ -307,9 +369,9 @@ func (s *Store) RetryFailedDeliveries(ctx context.Context, userID, destinationID } func (s *Store) Destination(ctx context.Context, userID, id int64) (Destination, error) { var d Destination - err := s.readerDB().QueryRowContext(ctx, `SELECT id,user_id,name,service,encrypted_url,enabled + err := s.readerDB().QueryRowContext(ctx, `SELECT id,user_id,name,service,encrypted_url,enabled,COALESCE(transport_status,'supported'),COALESCE(transport_message,'') FROM destinations WHERE user_id=? AND id=?`, userID, id).Scan( - &d.ID, &d.UserID, &d.Name, &d.Service, &d.EncryptedURL, &d.Enabled) + &d.ID, &d.UserID, &d.Name, &d.Service, &d.EncryptedURL, &d.Enabled, &d.TransportStatus, &d.TransportMessage) return d, err } func (s *Store) DeleteDestination(ctx context.Context, userID, id int64) error { @@ -318,14 +380,15 @@ func (s *Store) DeleteDestination(ctx context.Context, userID, id int64) error { } func (s *Store) DueDeliveries(ctx context.Context, now time.Time, limit int) ([]Delivery, error) { rows, err := s.readerDB().QueryContext(ctx, `SELECT d.id,d.event_id,d.attempts,d.next_attempt_at, - dst.id,dst.user_id,dst.name,dst.service,dst.encrypted_url,dst.enabled, + dst.id,dst.user_id,dst.name,dst.service,dst.encrypted_url,dst.enabled,COALESCE(dst.transport_status,'supported'),COALESCE(dst.transport_message,''), e.title,e.body,e.event_type,rg.title FROM deliveries d JOIN destinations dst ON dst.id=d.destination_id LEFT JOIN destination_health dh ON dh.destination_id=dst.id JOIN notification_events e ON e.id=d.event_id JOIN release_groups rg ON rg.id=e.release_group_id WHERE d.status='pending' AND d.next_attempt_at<=? AND dst.enabled=1 - AND `+destinationAdmissionPredicate+` ORDER BY d.next_attempt_at LIMIT ?`, - timeText(now), limit) + AND (d.claim_expires_at IS NULL OR d.claim_expires_at<=?) + AND `+supportedDestinationServicePredicate("dst")+` AND `+destinationAdmissionPredicate+` ORDER BY d.next_attempt_at LIMIT ?`, + timeText(now), timeText(now), limit) if err != nil { return nil, err } @@ -336,7 +399,8 @@ func (s *Store) DueDeliveries(ctx context.Context, now time.Time, limit int) ([] var next string if err := rows.Scan(&d.ID, &d.EventID, &d.Attempts, &next, &d.Destination.ID, &d.Destination.UserID, &d.Destination.Name, &d.Destination.Service, - &d.Destination.EncryptedURL, &d.Destination.Enabled, &d.Title, &d.Body, &d.EventType, &d.ReleaseTitle); err != nil { + &d.Destination.EncryptedURL, &d.Destination.Enabled, &d.Destination.TransportStatus, &d.Destination.TransportMessage, + &d.Title, &d.Body, &d.EventType, &d.ReleaseTitle); err != nil { return nil, err } d.NextAttempt, err = parseStoredTime(next, "delivery next_attempt_at") @@ -355,15 +419,16 @@ func (s *Store) DueDigestDeliveries(ctx context.Context, now time.Time, limit in return nil, nil } rows, err := s.readerDB().QueryContext(ctx, `SELECT dd.id,dd.run_id,dd.attempts,dd.next_attempt_at, - dst.id,dst.user_id,dst.name,dst.service,dst.encrypted_url,dst.enabled, + dst.id,dst.user_id,dst.name,dst.service,dst.encrypted_url,dst.enabled,COALESCE(dst.transport_status,'supported'),COALESCE(dst.transport_message,''), r.title,r.body FROM release_digest_deliveries dd JOIN release_digest_runs r ON r.id=dd.run_id JOIN destinations dst ON dst.id=dd.destination_id LEFT JOIN destination_health dh ON dh.destination_id=dst.id WHERE dd.status='pending' AND dd.next_attempt_at<=? AND dst.enabled=1 - AND `+destinationAdmissionPredicate+` - ORDER BY dd.next_attempt_at,dd.id LIMIT ?`, timeText(now), limit) + AND (dd.claim_expires_at IS NULL OR dd.claim_expires_at<=?) + AND `+supportedDestinationServicePredicate("dst")+` AND `+destinationAdmissionPredicate+` + ORDER BY dd.next_attempt_at,dd.id LIMIT ?`, timeText(now), timeText(now), limit) if err != nil { return nil, err } @@ -374,7 +439,8 @@ func (s *Store) DueDigestDeliveries(ctx context.Context, now time.Time, limit in var next string if err := rows.Scan(&d.ID, &d.RunID, &d.Attempts, &next, &d.Destination.ID, &d.Destination.UserID, &d.Destination.Name, &d.Destination.Service, - &d.Destination.EncryptedURL, &d.Destination.Enabled, &d.Title, &d.Body); err != nil { + &d.Destination.EncryptedURL, &d.Destination.Enabled, &d.Destination.TransportStatus, &d.Destination.TransportMessage, + &d.Title, &d.Body); err != nil { return nil, err } d.NextAttempt, err = parseStoredTime(next, "digest delivery next_attempt_at") @@ -385,12 +451,241 @@ func (s *Store) DueDigestDeliveries(ctx context.Context, now time.Time, limit in } return result, rows.Err() } + +// ClaimDueDeliveries atomically leases runnable normal deliveries for one +// runner instance. A second runner can see the row only after the lease +// expires, which gives us at-least-once processing without duplicate claims. +func (s *Store) ClaimDueDeliveries(ctx context.Context, now time.Time, limit int, owner string, lease time.Duration) ([]Delivery, error) { + if limit < 1 { + return nil, nil + } + owner = strings.TrimSpace(owner) + if owner == "" { + owner = "legacy-worker" + } + if lease <= 0 { + lease = 2 * time.Minute + } + expires := now.Add(lease) + tx, err := s.beginWriteTx(ctx) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + rows, err := tx.QueryContext(ctx, `SELECT d.id FROM deliveries d + JOIN destinations dst ON dst.id=d.destination_id + LEFT JOIN destination_health dh ON dh.destination_id=dst.id + WHERE d.status='pending' AND d.next_attempt_at<=? AND dst.enabled=1 + AND (d.claim_expires_at IS NULL OR d.claim_expires_at<=?) + AND `+supportedDestinationServicePredicate("dst")+` AND `+destinationAdmissionPredicate+` + ORDER BY d.next_attempt_at,d.id LIMIT ?`, timeText(now), timeText(now), limit) + if err != nil { + return nil, err + } + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + _ = rows.Close() + return nil, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + for _, id := range ids { + if _, err := tx.ExecContext(ctx, `UPDATE deliveries SET claim_owner=?,claim_expires_at=? WHERE id=? AND status='pending' AND (claim_expires_at IS NULL OR claim_expires_at<=?)`, owner, timeText(expires), id, timeText(now)); err != nil { + return nil, err + } + } + if len(ids) == 0 { + if err := tx.Commit(); err != nil { + return nil, err + } + return nil, nil + } + placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",") + args := make([]any, 0, len(ids)+1) + for _, id := range ids { + args = append(args, id) + } + args = append(args, owner) + rows, err = tx.QueryContext(ctx, `SELECT d.id,d.event_id,d.attempts,d.next_attempt_at, + dst.id,dst.user_id,dst.name,dst.service,dst.encrypted_url,dst.enabled,COALESCE(dst.transport_status,'supported'),COALESCE(dst.transport_message,''), + e.title,e.body,e.event_type,rg.title + FROM deliveries d JOIN destinations dst ON dst.id=d.destination_id + JOIN notification_events e ON e.id=d.event_id JOIN release_groups rg ON rg.id=e.release_group_id + WHERE d.id IN (`+placeholders+`) AND d.claim_owner=? ORDER BY d.id`, args...) + if err != nil { + return nil, err + } + var result []Delivery + for rows.Next() { + var d Delivery + var next string + if err := rows.Scan(&d.ID, &d.EventID, &d.Attempts, &next, + &d.Destination.ID, &d.Destination.UserID, &d.Destination.Name, &d.Destination.Service, + &d.Destination.EncryptedURL, &d.Destination.Enabled, &d.Destination.TransportStatus, &d.Destination.TransportMessage, + &d.Title, &d.Body, &d.EventType, &d.ReleaseTitle); err != nil { + _ = rows.Close() + return nil, err + } + d.NextAttempt, err = parseStoredTime(next, "delivery next_attempt_at") + if err != nil { + _ = rows.Close() + return nil, err + } + d.ClaimOwner = owner + result = append(result, d) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + if err := tx.Commit(); err != nil { + return nil, err + } + return result, nil +} + +// ClaimDueDigestDeliveries is the digest counterpart of ClaimDueDeliveries. +func (s *Store) ClaimDueDigestDeliveries(ctx context.Context, now time.Time, limit int, owner string, lease time.Duration) ([]DigestDelivery, error) { + if limit < 1 { + return nil, nil + } + owner = strings.TrimSpace(owner) + if owner == "" { + owner = "legacy-worker" + } + if lease <= 0 { + lease = 2 * time.Minute + } + expires := now.Add(lease) + tx, err := s.beginWriteTx(ctx) + if err != nil { + return nil, err + } + defer func() { _ = tx.Rollback() }() + rows, err := tx.QueryContext(ctx, `SELECT dd.id FROM release_digest_deliveries dd + JOIN destinations dst ON dst.id=dd.destination_id + LEFT JOIN destination_health dh ON dh.destination_id=dst.id + WHERE dd.status='pending' AND dd.next_attempt_at<=? AND dst.enabled=1 + AND (dd.claim_expires_at IS NULL OR dd.claim_expires_at<=?) + AND `+supportedDestinationServicePredicate("dst")+` AND `+destinationAdmissionPredicate+` + ORDER BY dd.next_attempt_at,dd.id LIMIT ?`, timeText(now), timeText(now), limit) + if err != nil { + return nil, err + } + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + _ = rows.Close() + return nil, err + } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + for _, id := range ids { + if _, err := tx.ExecContext(ctx, `UPDATE release_digest_deliveries SET claim_owner=?,claim_expires_at=? WHERE id=? AND status='pending' AND (claim_expires_at IS NULL OR claim_expires_at<=?)`, owner, timeText(expires), id, timeText(now)); err != nil { + return nil, err + } + } + if len(ids) == 0 { + if err := tx.Commit(); err != nil { + return nil, err + } + return nil, nil + } + placeholders := strings.TrimRight(strings.Repeat("?,", len(ids)), ",") + args := make([]any, 0, len(ids)+1) + for _, id := range ids { + args = append(args, id) + } + args = append(args, owner) + rows, err = tx.QueryContext(ctx, `SELECT dd.id,dd.run_id,dd.attempts,dd.next_attempt_at, + dst.id,dst.user_id,dst.name,dst.service,dst.encrypted_url,dst.enabled,COALESCE(dst.transport_status,'supported'),COALESCE(dst.transport_message,''), + r.title,r.body + FROM release_digest_deliveries dd JOIN release_digest_runs r ON r.id=dd.run_id + JOIN destinations dst ON dst.id=dd.destination_id + WHERE dd.id IN (`+placeholders+`) AND dd.claim_owner=? ORDER BY dd.id`, args...) + if err != nil { + return nil, err + } + var result []DigestDelivery + for rows.Next() { + var d DigestDelivery + var next string + if err := rows.Scan(&d.ID, &d.RunID, &d.Attempts, &next, + &d.Destination.ID, &d.Destination.UserID, &d.Destination.Name, &d.Destination.Service, + &d.Destination.EncryptedURL, &d.Destination.Enabled, &d.Destination.TransportStatus, &d.Destination.TransportMessage, + &d.Title, &d.Body); err != nil { + _ = rows.Close() + return nil, err + } + d.NextAttempt, err = parseStoredTime(next, "digest delivery next_attempt_at") + if err != nil { + _ = rows.Close() + return nil, err + } + d.ClaimOwner = owner + result = append(result, d) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + if err := tx.Commit(); err != nil { + return nil, err + } + return result, nil +} func (s *Store) MarkDeliverySent(ctx context.Context, id int64, now time.Time) error { - _, err := s.execWriteContext(ctx, `UPDATE deliveries SET status='sent',attempts=attempts+1,sent_at=?,last_error='' WHERE id=?`, - timeText(now), id) + err := s.MarkDeliverySentOwned(ctx, id, "", now) + if errors.Is(err, sql.ErrNoRows) { + return nil + } return err } + +func (s *Store) MarkDeliverySentOwned(ctx context.Context, id int64, owner string, now time.Time) error { + query := `UPDATE deliveries SET status='sent',attempts=attempts+1,sent_at=?,last_error='',claim_owner=NULL,claim_expires_at=NULL WHERE id=?` + args := []any{timeText(now), id} + if strings.TrimSpace(owner) != "" { + query += ` AND claim_owner=?` + args = append(args, owner) + } + result, err := s.execWriteContext(ctx, query, args...) + if err != nil { + return err + } + changed, err := result.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + return nil + } + return nil +} func (s *Store) MarkDeliveryFailed(ctx context.Context, id int64, attempts int, message string, now time.Time) error { + err := s.MarkDeliveryFailedOwned(ctx, id, attempts, message, "", now) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + return err +} + +func (s *Store) MarkDeliveryFailedOwned(ctx context.Context, id int64, attempts int, message, owner string, now time.Time) error { message = safeDeliveryError(message) status := "pending" if attempts >= 5 { @@ -400,32 +695,77 @@ func (s *Store) MarkDeliveryFailed(ctx context.Context, id int64, attempts int, if len(message) > 500 { message = message[:500] } - _, err := s.execWriteContext(ctx, `UPDATE deliveries SET status=?,attempts=?,next_attempt_at=?,last_error=? WHERE id=?`, - status, attempts, timeText(now.Add(delay)), message, id) - return err + query := `UPDATE deliveries SET status=?,attempts=?,next_attempt_at=?,last_error=?,claim_owner=NULL,claim_expires_at=NULL WHERE id=?` + args := []any{status, attempts, timeText(now.Add(delay)), message, id} + if strings.TrimSpace(owner) != "" { + query += ` AND claim_owner=?` + args = append(args, owner) + } + result, err := s.execWriteContext(ctx, query, args...) + if err != nil { + return err + } + changed, err := result.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + return nil + } + return nil } func (s *Store) MarkDigestDeliverySent(ctx context.Context, id int64, now time.Time) error { + err := s.MarkDigestDeliverySentOwned(ctx, id, "", now) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + return err +} + +func (s *Store) MarkDigestDeliverySentOwned(ctx context.Context, id int64, owner string, now time.Time) error { tx, err := s.beginWriteTx(ctx) if err != nil { return err } defer func() { _ = tx.Rollback() }() - if _, err := tx.ExecContext(ctx, `UPDATE release_digest_deliveries - SET status='sent',attempts=attempts+1,sent_at=?,last_error='' WHERE id=?`, timeText(now), id); err != nil { + query := `UPDATE release_digest_deliveries + SET status='sent',attempts=attempts+1,sent_at=?,last_error='',claim_owner=NULL,claim_expires_at=NULL WHERE id=?` + args := []any{timeText(now), id} + if strings.TrimSpace(owner) != "" { + query += ` AND claim_owner=?` + args = append(args, owner) + } + result, err := tx.ExecContext(ctx, query, args...) + if err != nil { + return err + } + changed, err := result.RowsAffected() + if err != nil { return err } + if changed == 0 { + return nil + } if _, err := tx.ExecContext(ctx, `UPDATE release_digest_runs SET status=CASE WHEN EXISTS ( SELECT 1 FROM release_digest_deliveries WHERE run_id=release_digest_runs.id AND status='failed' ) THEN 'failed' ELSE 'sent' END WHERE id=(SELECT run_id FROM release_digest_deliveries WHERE id=?) - AND NOT EXISTS (SELECT 1 FROM release_digest_deliveries WHERE run_id=release_digest_runs.id AND status='pending')`, id); err != nil { + AND NOT EXISTS (SELECT 1 FROM release_digest_deliveries WHERE run_id=release_digest_runs.id AND status IN ('pending','blocked'))`, id); err != nil { return err } return tx.Commit() } func (s *Store) MarkDigestDeliveryFailed(ctx context.Context, id int64, attempts int, message string, now time.Time) error { + err := s.MarkDigestDeliveryFailedOwned(ctx, id, attempts, message, "", now) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + return err +} + +func (s *Store) MarkDigestDeliveryFailedOwned(ctx context.Context, id int64, attempts int, message, owner string, now time.Time) error { message = safeDeliveryError(message) status := "pending" if attempts >= 5 { @@ -440,20 +780,83 @@ func (s *Store) MarkDigestDeliveryFailed(ctx context.Context, id int64, attempts return err } defer func() { _ = tx.Rollback() }() - if _, err := tx.ExecContext(ctx, `UPDATE release_digest_deliveries - SET status=?,attempts=?,next_attempt_at=?,last_error=? WHERE id=?`, - status, attempts, timeText(now.Add(delay)), message, id); err != nil { + query := `UPDATE release_digest_deliveries + SET status=?,attempts=?,next_attempt_at=?,last_error=?,claim_owner=NULL,claim_expires_at=NULL WHERE id=?` + args := []any{status, attempts, timeText(now.Add(delay)), message, id} + if strings.TrimSpace(owner) != "" { + query += ` AND claim_owner=?` + args = append(args, owner) + } + result, err := tx.ExecContext(ctx, query, args...) + if err != nil { + return err + } + changed, err := result.RowsAffected() + if err != nil { return err } + if changed == 0 { + return nil + } if status == "failed" { if _, err := tx.ExecContext(ctx, `UPDATE release_digest_runs SET status='failed' WHERE id=(SELECT run_id FROM release_digest_deliveries WHERE id=?) - AND NOT EXISTS (SELECT 1 FROM release_digest_deliveries WHERE run_id=release_digest_runs.id AND status='pending')`, id); err != nil { + AND NOT EXISTS (SELECT 1 FROM release_digest_deliveries WHERE run_id=release_digest_runs.id AND status IN ('pending','blocked'))`, id); err != nil { return err } } return tx.Commit() } + +// RecoverExpiredWork makes abandoned claims runnable again. It is safe to +// call at startup and during maintenance: only expired ownership metadata is +// cleared, and sent/failed rows are never resurrected. +func (s *Store) RecoverExpiredWork(ctx context.Context, now time.Time) (int, error) { + tx, err := s.beginWriteTx(ctx) + if err != nil { + return 0, err + } + defer func() { _ = tx.Rollback() }() + recovered := 0 + statements := []string{ + `UPDATE manual_sync_requests SET status='queued',started_at=NULL,lease_owner=NULL,lease_expires_at=NULL,last_error='worker lease expired' WHERE status='running' AND lease_expires_at IS NOT NULL AND lease_expires_at<=?`, + `UPDATE deliveries SET claim_owner=NULL,claim_expires_at=NULL WHERE status='pending' AND claim_expires_at IS NOT NULL AND claim_expires_at<=?`, + `UPDATE release_digest_deliveries SET claim_owner=NULL,claim_expires_at=NULL WHERE status='pending' AND claim_expires_at IS NOT NULL AND claim_expires_at<=?`, + } + for _, statement := range statements { + result, err := tx.ExecContext(ctx, statement, timeText(now)) + if err != nil { + return 0, err + } + changed, err := result.RowsAffected() + if err != nil { + return 0, err + } + recovered += int(changed) + } + if err := tx.Commit(); err != nil { + return 0, err + } + return recovered, nil +} + +// ReconcileStaleDeliveryAttempts turns attempts that have no completion +// record after a bounded window into failed audit entries. The queue row is +// left pending and will be retried by its normal lease/attempt policy. +func (s *Store) ReconcileStaleDeliveryAttempts(ctx context.Context, now time.Time, staleAfter time.Duration) (int, error) { + if staleAfter <= 0 { + staleAfter = 10 * time.Minute + } + cutoff := now.Add(-staleAfter) + result, err := s.execWriteContext(ctx, `UPDATE delivery_attempts + SET status='failed',finished_at=?,abandoned_at=?,last_error='worker attempt expired' + WHERE status='started' AND started_at<=?`, timeText(now), timeText(now), timeText(cutoff)) + if err != nil { + return 0, err + } + changed, err := result.RowsAffected() + return int(changed), err +} func (s *Store) DeliveryHistory(ctx context.Context, userID int64, limit int) ([]DeliveryHistory, error) { rows, err := s.readerDB().QueryContext(ctx, `SELECT title,event_type,destination,status,attempts,last_error,created_at,sent_at FROM ( diff --git a/internal/store/operations.go b/internal/store/operations.go index b856bd9..213195b 100644 --- a/internal/store/operations.go +++ b/internal/store/operations.go @@ -6,6 +6,8 @@ import ( "encoding/json" "errors" "fmt" + "os" + "path/filepath" "strings" "time" @@ -34,6 +36,38 @@ func (s *Store) InsertApplicationLog(ctx context.Context, entry logging.Entry) e _, err = s.execWriteContext(ctx, `INSERT INTO application_logs(created_at,level,message,attributes_json) VALUES(?,?,?,?)`, entry.Time.UTC().Format(time.RFC3339Nano), level, entry.Message, string(attrs)) return err } + +const ( + backupMarkerFile = ".artist-trackarr-last-backup" + restoreMarkerFile = ".artist-trackarr-last-restore" +) + +// operationalMarker reads a deliberately small, operator-written marker. The +// marker is not a source of truth for application data; it only lets the +// diagnostics page report whether an external backup or restore rehearsal has +// recently completed. Missing markers are normal for fresh installations. +func (s *Store) operationalMarker(name string) (*time.Time, string) { + if strings.TrimSpace(s.dataDir) == "" { + return nil, "" + } + contents, err := os.ReadFile(filepath.Join(s.dataDir, name)) + if err != nil { + return nil, "" + } + lines := strings.Split(strings.TrimSpace(string(contents)), "\n") + if len(lines) == 0 || strings.TrimSpace(lines[0]) == "" { + return nil, "" + } + parsed, err := parseTime(strings.TrimSpace(lines[0])) + if err != nil { + return nil, "" + } + status := "" + if len(lines) > 1 { + status = strings.TrimSpace(lines[1]) + } + return &parsed, status +} func (s *Store) ApplicationLogs(ctx context.Context, limit int) ([]logging.Entry, error) { if limit <= 0 { limit = 200 @@ -125,15 +159,33 @@ func (s *Store) CreateManualSyncRequest(ctx context.Context, userID int64, scope return ManualSyncRequest{ID: id, RequestedBy: userID, Scope: scope, ArtistID: artistID, Status: "queued", CreatedAt: created}, nil } func (s *Store) ClaimManualSyncRequests(ctx context.Context, limit int) ([]ManualSyncRequest, error) { + return s.ClaimManualSyncRequestsWithLease(ctx, limit, "legacy-worker", 5*time.Minute) +} + +// ClaimManualSyncRequestsWithLease atomically claims queued work and recovers +// running rows whose lease expired. The owner token prevents two runner +// instances from completing the same durable request concurrently. +func (s *Store) ClaimManualSyncRequestsWithLease(ctx context.Context, limit int, owner string, lease time.Duration) ([]ManualSyncRequest, error) { if limit < 1 { limit = 1 } + owner = strings.TrimSpace(owner) + if owner == "" { + owner = "legacy-worker" + } + if lease <= 0 { + lease = 5 * time.Minute + } tx, err := s.beginWriteTx(ctx) if err != nil { return nil, err } defer func() { _ = tx.Rollback() }() - rows, err := tx.QueryContext(ctx, `SELECT id,requested_by,scope,artist_id,created_at FROM manual_sync_requests WHERE status='queued' ORDER BY id LIMIT ?`, limit) + now := time.Now().UTC() + expires := now.Add(lease) + rows, err := tx.QueryContext(ctx, `SELECT id,requested_by,scope,artist_id,created_at FROM manual_sync_requests + WHERE status='queued' OR (status='running' AND lease_expires_at IS NOT NULL AND lease_expires_at<=?) + ORDER BY id LIMIT ?`, timeText(now), limit) if err != nil { return nil, err } @@ -163,25 +215,32 @@ func (s *Store) ClaimManualSyncRequests(ctx context.Context, limit int) ([]Manua if err := rows.Err(); err != nil { return nil, err } - now := nowText() + nowTextValue := timeText(now) for _, id := range ids { - if _, err := tx.ExecContext(ctx, `UPDATE manual_sync_requests SET status='running',started_at=? WHERE id=?`, now, id); err != nil { + if _, err := tx.ExecContext(ctx, `UPDATE manual_sync_requests SET status='running',started_at=?,lease_owner=?,lease_expires_at=?,attempt_count=attempt_count+1 WHERE id=? AND (status='queued' OR (status='running' AND lease_expires_at IS NOT NULL AND lease_expires_at<=?))`, nowTextValue, owner, timeText(expires), id, nowTextValue); err != nil { return nil, err } } if err := tx.Commit(); err != nil { return nil, err } - started, err := parseStoredTime(now, "manual sync started_at") + started, err := parseStoredTime(nowTextValue, "manual sync started_at") if err != nil { return nil, err } for i := range out { out[i].StartedAt = &started + out[i].LeaseOwner = owner + expiresCopy := expires + out[i].LeaseExpiresAt = &expiresCopy } return out, nil } func (s *Store) CompleteManualSyncRequest(ctx context.Context, id int64, syncErr error) error { + return s.CompleteManualSyncRequestOwned(ctx, id, "", syncErr) +} + +func (s *Store) CompleteManualSyncRequestOwned(ctx context.Context, id int64, owner string, syncErr error) error { status, msg := "completed", "" if syncErr != nil { status = "failed" @@ -190,14 +249,30 @@ func (s *Store) CompleteManualSyncRequest(ctx context.Context, id int64, syncErr if len(msg) > 500 { msg = msg[:500] } - _, err := s.execWriteContext(ctx, `UPDATE manual_sync_requests SET status=?,finished_at=?,last_error=? WHERE id=?`, status, nowText(), msg, id) - return err + query := `UPDATE manual_sync_requests SET status=?,finished_at=?,last_error=?,lease_owner=NULL,lease_expires_at=NULL WHERE id=?` + args := []any{status, nowText(), msg, id} + if strings.TrimSpace(owner) != "" { + query += ` AND lease_owner=?` + args = append(args, owner) + } + result, err := s.execWriteContext(ctx, query, args...) + if err != nil { + return err + } + changed, err := result.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + return sql.ErrNoRows + } + return nil } func (s *Store) ManualSyncRequests(ctx context.Context, limit int) ([]ManualSyncRequest, error) { if limit < 1 { limit = 20 } - rows, err := s.readerDB().QueryContext(ctx, `SELECT id,requested_by,scope,artist_id,status,created_at,started_at,finished_at,last_error FROM manual_sync_requests ORDER BY id DESC LIMIT ?`, limit) + rows, err := s.readerDB().QueryContext(ctx, `SELECT id,requested_by,scope,artist_id,status,created_at,started_at,finished_at,last_error,lease_owner,lease_expires_at,attempt_count FROM manual_sync_requests ORDER BY id DESC LIMIT ?`, limit) if err != nil { return nil, err } @@ -207,10 +282,11 @@ func (s *Store) ManualSyncRequests(ctx context.Context, limit int) ([]ManualSync var r ManualSyncRequest var aid sql.NullInt64 var c string - var st, ft sql.NullString - if err := rows.Scan(&r.ID, &r.RequestedBy, &r.Scope, &aid, &r.Status, &c, &st, &ft, &r.LastError); err != nil { + var st, ft, leaseExpires, leaseOwner sql.NullString + if err := rows.Scan(&r.ID, &r.RequestedBy, &r.Scope, &aid, &r.Status, &c, &st, &ft, &r.LastError, &leaseOwner, &leaseExpires, &r.AttemptCount); err != nil { return nil, err } + r.LeaseOwner = leaseOwner.String created, parseErr := parseStoredTime(c, "manual sync created_at") if parseErr != nil { return nil, parseErr @@ -226,6 +302,9 @@ func (s *Store) ManualSyncRequests(ctx context.Context, limit int) ([]ManualSync if r.FinishedAt, parseErr = parseStoredNullableTime(ft, "manual sync finished_at"); parseErr != nil { return nil, parseErr } + if r.LeaseExpiresAt, parseErr = parseStoredNullableTime(leaseExpires, "manual sync lease_expires_at"); parseErr != nil { + return nil, parseErr + } out = append(out, r) } return out, rows.Err() @@ -330,6 +409,35 @@ func (s *Store) Diagnostics(ctx context.Context) (DiagnosticsSnapshot, error) { return DiagnosticsSnapshot{}, err } snapshot.DatabaseHealthy = true + var oldest sql.NullString + if err := s.readerDB().QueryRowContext(ctx, `SELECT MIN(value) FROM ( + SELECT next_attempt_at AS value FROM deliveries WHERE status IN ('pending','blocked') + UNION ALL SELECT next_attempt_at FROM release_digest_deliveries WHERE status IN ('pending','blocked'))`).Scan(&oldest); err != nil { + return DiagnosticsSnapshot{}, err + } + if snapshot.OldestQueueAt, err = parseStoredNullableTime(oldest, "oldest queued delivery"); err != nil { + return DiagnosticsSnapshot{}, err + } + if err := s.readerDB().QueryRowContext(ctx, `SELECT + (SELECT COUNT(*) FROM manual_sync_requests WHERE status='running' AND lease_expires_at IS NOT NULL AND lease_expires_at<=?)+ + (SELECT COUNT(*) FROM deliveries WHERE status='pending' AND claim_expires_at IS NOT NULL AND claim_expires_at<=?)+ + (SELECT COUNT(*) FROM release_digest_deliveries WHERE status='pending' AND claim_expires_at IS NOT NULL AND claim_expires_at<=?), + (SELECT COUNT(*) FROM destination_health WHERE status='paused'), + (SELECT COUNT(*) FROM provider_health WHERE last_failure_at IS NOT NULL AND (last_success_at IS NULL OR last_failure_at>last_success_at)), + (SELECT COUNT(*) FROM release_digest_deliveries WHERE status IN ('pending','blocked'))`, timeText(snapshot.CheckedAt), timeText(snapshot.CheckedAt), timeText(snapshot.CheckedAt)). + Scan(&snapshot.StaleClaims, &snapshot.PausedDestinations, &snapshot.ProviderFailures, &snapshot.DigestBacklog); err != nil { + return DiagnosticsSnapshot{}, err + } + var pageCount, pageSize int64 + if err := s.readerDB().QueryRowContext(ctx, `PRAGMA page_count`).Scan(&pageCount); err != nil { + return DiagnosticsSnapshot{}, err + } + if err := s.readerDB().QueryRowContext(ctx, `PRAGMA page_size`).Scan(&pageSize); err != nil { + return DiagnosticsSnapshot{}, err + } + snapshot.DatabaseBytes = pageCount * pageSize + snapshot.LastBackupAt, _ = s.operationalMarker(backupMarkerFile) + snapshot.LastRestoreAt, snapshot.LastRestoreResult = s.operationalMarker(restoreMarkerFile) health, err := s.ProviderHealth(ctx) if err != nil { return DiagnosticsSnapshot{}, err diff --git a/internal/store/release_helpers.go b/internal/store/release_helpers.go index d8b03d1..dd9f60c 100644 --- a/internal/store/release_helpers.go +++ b/internal/store/release_helpers.go @@ -750,9 +750,9 @@ func insertNotificationEventTxMode(ctx context.Context, tx *sql.Tx, userID, rele // repeated release-day queue run should never turn an already queued event // into a duplicate-delivery error (or duplicate rows). _, err = tx.ExecContext(ctx, `INSERT OR IGNORE INTO deliveries(event_id,destination_id,status,next_attempt_at) - SELECT ?,d.id,'pending',? FROM destinations d + SELECT ?,d.id,`+destinationQueueStatus("d")+`,? FROM destinations d LEFT JOIN destination_health dh ON dh.destination_id=d.id - WHERE d.user_id=? AND d.enabled=1 AND `+destinationAdmissionPredicate, eventID, timeText(now), userID) + WHERE d.user_id=? AND d.enabled=1`, eventID, timeText(now), userID) return err } diff --git a/internal/store/schema.go b/internal/store/schema.go index f4f4760..73c1129 100644 --- a/internal/store/schema.go +++ b/internal/store/schema.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io/fs" + "path/filepath" "sort" "strconv" "strings" @@ -39,7 +40,7 @@ func Open(path string) (*Store, error) { _ = db.Close() return nil, err } - s := &Store{DB: db} + s := &Store{DB: db, dataDir: filepath.Dir(path)} if err := s.migrate(context.Background()); err != nil { _ = db.Close() return nil, err diff --git a/internal/store/store.go b/internal/store/store.go index 8ad5ea1..034296f 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -21,8 +21,13 @@ var migrations embed.FS // is committing, which keeps dashboard requests from queueing behind provider // synchronization work. type Store struct { - DB *sql.DB - Reader *sql.DB + DB *sql.DB + Reader *sql.DB + // dataDir is set for stores opened through Open and is intentionally empty + // for lightweight test fixtures that wrap an existing *sql.DB. Operational + // marker files in this directory contain only non-sensitive backup/recovery + // timestamps and status labels. + dataDir string readerMu sync.RWMutex healthMu sync.RWMutex pollInterval time.Duration @@ -236,6 +241,7 @@ var ( ErrCannotDeleteSelf = errors.New("you cannot delete your own account") ErrLastAdmin = errors.New("the last administrator cannot be deleted") ErrManualSyncQueueFull = errors.New("manual synchronization queue is full; try again later") + ErrSetupCompleted = errors.New("setup has already completed") ErrInvalidUsername = errors.New("username must be 3-32 characters using letters, numbers, dots, underscores, or hyphens") ErrUsernameTaken = errors.New("that username is already in use") ErrInvalidNotificationHoldAction = errors.New("invalid notification hold action") @@ -389,6 +395,7 @@ type DigestDelivery struct { Body string Attempts int NextAttempt time.Time + ClaimOwner string } // NotificationHold is an owner-scoped notification that was kept out of the @@ -544,17 +551,26 @@ type AssuranceSummary struct { // support view. It intentionally excludes provider error text, credentials, // notification bodies, and destination URLs. type DiagnosticsSnapshot struct { - CheckedAt time.Time - DatabaseHealthy bool - SchemaVersion int - FollowedArtists int - Releases int - QueuedSyncs int - RunningSyncs int - PendingDeliveries int - FailedDeliveries int - RecentLogEntries int - Providers []DiagnosticsProvider + CheckedAt time.Time + DatabaseHealthy bool + SchemaVersion int + FollowedArtists int + Releases int + QueuedSyncs int + RunningSyncs int + PendingDeliveries int + FailedDeliveries int + RecentLogEntries int + OldestQueueAt *time.Time + StaleClaims int + PausedDestinations int + ProviderFailures int + DigestBacklog int + DatabaseBytes int64 + LastBackupAt *time.Time + LastRestoreAt *time.Time + LastRestoreResult string + Providers []DiagnosticsProvider } // DiagnosticsProvider is the redacted provider projection used by support @@ -580,12 +596,14 @@ type SpotifyPollingState struct { } type Destination struct { - ID int64 - UserID int64 - Name string - Service string - EncryptedURL []byte - Enabled bool + ID int64 + UserID int64 + Name string + Service string + EncryptedURL []byte + Enabled bool + TransportStatus string + TransportMessage string } // DestinationHealth is durable, owner-visible delivery state. It is kept @@ -596,6 +614,7 @@ type DestinationHealth struct { Status string ConsecutiveFailures int PendingCount int + BlockedCount int FailedCount int LastSuccessAt *time.Time LastFailureAt *time.Time @@ -612,6 +631,7 @@ type AdminDestinationHealth struct { Status string ConsecutiveFailures int PendingCount int + BlockedCount int FailedCount int LastSuccessAt *time.Time LastFailureAt *time.Time @@ -630,6 +650,7 @@ type Delivery struct { NextAttempt time.Time EventType string ReleaseTitle string + ClaimOwner string } type DeliveryHistory struct { @@ -660,15 +681,18 @@ type AdminDeliveryHistory struct { } type ManualSyncRequest struct { - ID int64 - RequestedBy int64 - Scope string - ArtistID *int64 - Status string - CreatedAt time.Time - StartedAt *time.Time - FinishedAt *time.Time - LastError string + ID int64 + RequestedBy int64 + Scope string + ArtistID *int64 + Status string + CreatedAt time.Time + StartedAt *time.Time + FinishedAt *time.Time + LastError string + LeaseOwner string + LeaseExpiresAt *time.Time + AttemptCount int } type ProviderHealth struct { diff --git a/internal/store/store_test.go b/internal/store/store_test.go index b94cc6b..62bd900 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -222,7 +222,10 @@ func TestITunesMigrationPreservesExistingProviderData(t *testing.T) { if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=24`).Scan(&migrationsApplied); err != nil || migrationsApplied != 1 { t.Fatalf("assurance indexes migration marker=%d err=%v", migrationsApplied, err) } - for _, indexName := range []string{"idx_provider_observations_release_observed", "idx_follows_artist_user", "idx_import_rows_job_id", "release_credits_release_artist", "release_credits_artist_release", "destinations_user_enabled", "deliveries_status_due_destination", "release_digest_deliveries_status_due_destination"} { + if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=25`).Scan(&migrationsApplied); err != nil || migrationsApplied != 1 { + t.Fatalf("recovery leases migration marker=%d err=%v", migrationsApplied, err) + } + for _, indexName := range []string{"idx_provider_observations_release_observed", "idx_follows_artist_user", "idx_import_rows_job_id", "release_credits_release_artist", "release_credits_artist_release", "destinations_user_enabled", "deliveries_status_due_destination", "release_digest_deliveries_status_due_destination", "destinations_transport_status", "manual_sync_leases", "deliveries_claim_expiry", "release_digest_deliveries_claim_expiry", "delivery_attempts_started"} { var found string if err := db.QueryRow(`SELECT name FROM sqlite_master WHERE type='index' AND name=?`, indexName).Scan(&found); err != nil { t.Fatalf("migration index %q missing: %v", indexName, err) diff --git a/internal/web/admin.go b/internal/web/admin.go index be5d291..36838d4 100644 --- a/internal/web/admin.go +++ b/internal/web/admin.go @@ -156,7 +156,25 @@ func diagnosticReport(snapshot store.DiagnosticsSnapshot, runner jobs.RunnerStat fmt.Fprintf(&report, "Queued syncs: %d\n", snapshot.QueuedSyncs) fmt.Fprintf(&report, "Running syncs: %d\n", snapshot.RunningSyncs) fmt.Fprintf(&report, "Pending deliveries: %d\n", snapshot.PendingDeliveries) + fmt.Fprintf(&report, "Digest backlog: %d\n", snapshot.DigestBacklog) fmt.Fprintf(&report, "Failed deliveries: %d\n", snapshot.FailedDeliveries) + fmt.Fprintf(&report, "Stale work claims: %d\n", snapshot.StaleClaims) + fmt.Fprintf(&report, "Paused destinations: %d\n", snapshot.PausedDestinations) + fmt.Fprintf(&report, "Provider failures: %d\n", snapshot.ProviderFailures) + fmt.Fprintf(&report, "Database size: %d bytes\n", snapshot.DatabaseBytes) + if snapshot.OldestQueueAt != nil { + fmt.Fprintf(&report, "Oldest queued delivery: %s\n", snapshot.OldestQueueAt.Format(time.RFC3339)) + } + if snapshot.LastBackupAt != nil { + fmt.Fprintf(&report, "Last backup: %s\n", snapshot.LastBackupAt.Format(time.RFC3339)) + } else { + fmt.Fprintln(&report, "Last backup: not recorded") + } + if snapshot.LastRestoreAt != nil { + fmt.Fprintf(&report, "Last restore rehearsal: %s (%s)\n", snapshot.LastRestoreAt.Format(time.RFC3339), snapshot.LastRestoreResult) + } else { + fmt.Fprintln(&report, "Last restore rehearsal: not recorded") + } fmt.Fprintf(&report, "Application events (24h): %d\n", snapshot.RecentLogEntries) fmt.Fprintf(&report, "Scheduler: %s\n", diagnosticHealthLabel(runner.Running)) if runner.LastActivityAt != nil { diff --git a/internal/web/auth.go b/internal/web/auth.go index dae5768..110ae0d 100644 --- a/internal/web/auth.go +++ b/internal/web/auth.go @@ -4,6 +4,7 @@ import ( "crypto/subtle" "database/sql" "errors" + "fmt" "net" "net/http" "strings" @@ -35,16 +36,6 @@ func (a *App) setup(w http.ResponseWriter, r *http.Request) { return } defer release() - count, err := a.store.UserCount(r.Context()) - if err != nil { - a.logger.Error("setup user count failed", "page", "Create administrator", "path", r.URL.Path, "error", err) - http.Error(w, "could not load setup", http.StatusInternalServerError) - return - } - if count > 0 { - http.Error(w, "setup has already completed", http.StatusConflict) - return - } if a.cfg.SetupToken == "" || subtle.ConstantTimeCompare([]byte(r.FormValue("setup_token")), []byte(a.cfg.SetupToken)) != 1 { d := a.data(r, "Create administrator") d.Error = "The setup token is incorrect." @@ -57,10 +48,14 @@ func (a *App) setup(w http.ResponseWriter, r *http.Request) { if _, supplied := r.Form["username"]; supplied && username == "" { err = store.ErrInvalidUsername } else { - _, err = a.store.CreateUser(r.Context(), r.FormValue("email"), hash, "admin", r.FormValue("timezone"), username) + _, err = a.store.CreateInitialAdmin(r.Context(), r.FormValue("email"), hash, r.FormValue("timezone"), username) } } if err != nil { + if errors.Is(err, store.ErrSetupCompleted) { + http.Error(w, "setup has already completed", http.StatusConflict) + return + } d := a.data(r, "Create administrator") d.Error = err.Error() a.render(w, "setup", d, http.StatusBadRequest) @@ -190,7 +185,9 @@ func (a *App) acceptReset(w http.ResponseWriter, r *http.Request) { } if err != nil { if !errors.Is(err, sql.ErrNoRows) { - a.logger.Warn("password reset failed", "page", "Reset password", "path", r.URL.Path, "error", err) + digest := security.Digest(chi.URLParam(r, "token")) + a.logger.Warn("password reset failed", "page", "Reset password", "route", "/reset/{token}", + "token_fingerprint", fmt.Sprintf("%x", digest[:6]), "error", err) } d := a.data(r, "Reset password") d.Error, d.Token, d.TokenKind = "Reset link is invalid, expired, or already used. Please request a new link.", chi.URLParam(r, "token"), "reset" diff --git a/internal/web/core.go b/internal/web/core.go index 065020a..17572a0 100644 --- a/internal/web/core.go +++ b/internal/web/core.go @@ -221,7 +221,7 @@ func New(cfg config.Config, s *store.Store, mb catalog.CatalogProvider, spotify switch strings.ToLower(status) { case "healthy": return "sent" - case "paused": + case "paused", "unsupported": return "failed" default: return "ambiguous" @@ -235,6 +235,8 @@ func New(cfg config.Config, s *store.Store, mb catalog.CatalogProvider, spotify return "Paused" case "degraded": return "Degraded" + case "unsupported": + return "Unsupported" default: return "Unknown" } @@ -464,14 +466,16 @@ func New(cfg config.Config, s *store.Store, mb catalog.CatalogProvider, spotify cfg: cfg, store: s, mb: mb, spotify: spotify, sender: sender, itunes: itunesProvider, cipher: cipher, artwork: art, jobs: runner, logger: logger, templates: tmpl, - setupLimiter: newFixedWindowLimiter(10, 15*time.Minute), - loginLimiter: newFixedWindowLimiter(20, 5*time.Minute), - tokenLimiter: newFixedWindowLimiter(20, 5*time.Minute), - discoveryLimiter: newFixedWindowLimiter(30, 5*time.Minute), - importLimiter: newFixedWindowLimiter(5, time.Hour), - importSlots: make(chan struct{}, maxConcurrentImports), - providerLimiter: newFixedWindowLimiter(30, 10*time.Minute), - loginSlots: make(chan struct{}, 8), + setupLimiter: newFixedWindowLimiter(10, 15*time.Minute), + loginLimiter: newFixedWindowLimiter(20, 5*time.Minute), + tokenLimiter: newFixedWindowLimiter(20, 5*time.Minute), + discoveryLimiter: newFixedWindowLimiter(30, 5*time.Minute), + importLimiter: newFixedWindowLimiter(5, time.Hour), + importSlots: make(chan struct{}, maxConcurrentImports), + providerLimiter: newFixedWindowLimiter(30, 10*time.Minute), + destinationTestLimiter: newFixedWindowLimiter(5, 15*time.Minute), + destinationRetryLimiter: newFixedWindowLimiter(10, 15*time.Minute), + loginSlots: make(chan struct{}, 8), }, nil } @@ -892,5 +896,15 @@ func (a *App) ready(w http.ResponseWriter, r *http.Request) { http.Error(w, "not ready", http.StatusServiceUnavailable) return } + w.Header().Set("X-ArtistTrackarr-Database", "healthy") + runnerState := "unknown" + if a.jobs != nil { + if a.jobs.Status().Running { + runnerState = "running" + } else { + runnerState = "stopped" + } + } + w.Header().Set("X-ArtistTrackarr-Runner", runnerState) w.WriteHeader(http.StatusNoContent) } diff --git a/internal/web/settings.go b/internal/web/settings.go index 34d9fa4..aad20cc 100644 --- a/internal/web/settings.go +++ b/internal/web/settings.go @@ -43,12 +43,13 @@ func (a *App) addDestination(w http.ResponseWriter, r *http.Request) { if err == nil { err = a.sender.Validate(serviceURL) } + storedService := notify.CanonicalTransportService(serviceURL) var encrypted []byte if err == nil { encrypted, err = a.cipher.Encrypt(serviceURL) } if err == nil { - err = a.store.AddDestination(r.Context(), session.User.ID, r.FormValue("name"), input.Service, encrypted) + err = a.store.AddDestination(r.Context(), session.User.ID, r.FormValue("name"), storedService, encrypted) } if err != nil { d := a.data(r, "Settings") @@ -69,6 +70,10 @@ func (a *App) testDestination(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } + if a.destinationTestLimiter != nil && !a.destinationTestLimiter.Allow(fmt.Sprintf("%d|%d", session.User.ID, id)) { + rateLimited(w, 900, "destination tests are temporarily rate limited; try again later") + return + } destination, err := a.store.Destination(r.Context(), session.User.ID, id) if errors.Is(err, sql.ErrNoRows) { http.NotFound(w, r) @@ -95,6 +100,10 @@ func (a *App) retryDestination(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) return } + if a.destinationRetryLimiter != nil && !a.destinationRetryLimiter.Allow(fmt.Sprintf("%d|%d", session.User.ID, id)) { + rateLimited(w, 900, "destination retries are temporarily rate limited; try again later") + return + } count, err := a.store.RetryFailedDeliveries(r.Context(), session.User.ID, id, time.Now().UTC()) if err != nil { if errors.Is(err, sql.ErrNoRows) { diff --git a/internal/web/static/app.js b/internal/web/static/app.js index 3a8907d..4d68741 100644 --- a/internal/web/static/app.js +++ b/internal/web/static/app.js @@ -83,17 +83,16 @@ if (form) { const visible = { ntfy: ["host", "username", "password", "topic"], - email: ["host", "port", "username", "password", "from", "to"], discord: ["token", "target"], telegram: ["token", "target"], - gotify: ["host", "token"], generic: ["target"], advanced: ["raw_url"] }; const refresh = () => { const service = form.querySelector("[data-service]").value; + const fields = visible[service] || []; form.querySelectorAll("[data-field]").forEach((field) => { - field.hidden = !visible[service].includes(field.dataset.field); + field.hidden = !fields.includes(field.dataset.field); }); }; form.querySelector("[data-service]").addEventListener("change", refresh); diff --git a/internal/web/templates/admin.html b/internal/web/templates/admin.html index 1f54e00..669a6f7 100644 --- a/internal/web/templates/admin.html +++ b/internal/web/templates/admin.html @@ -26,7 +26,14 @@
{{.Diagnostics.QueuedSyncs}}queued syncs
{{.Diagnostics.RunningSyncs}}running syncs
{{.Diagnostics.PendingDeliveries}}pending deliveries
+
{{.Diagnostics.DigestBacklog}}digest backlog
{{.Diagnostics.FailedDeliveries}}failed deliveries
+
{{.Diagnostics.StaleClaims}}stale work claims
+
{{.Diagnostics.PausedDestinations}}paused destinations
+
{{.Diagnostics.ProviderFailures}}provider failures
+
{{.Diagnostics.DatabaseBytes}}database bytes
+
{{if .Diagnostics.LastBackupAt}}{{formatProviderTime .Diagnostics.LastBackupAt}}{{else}}—{{end}}last backup
+
{{if .Diagnostics.LastRestoreAt}}{{.Diagnostics.LastRestoreResult}}{{else}}—{{end}}restore rehearsal
{{.Diagnostics.RecentLogEntries}}application events (24h)
{{.Diagnostics.FollowedArtists}}followed artists
{{.Diagnostics.Releases}}known releases
@@ -63,7 +70,7 @@

Delivery assurance

Destination health

{{len .AdminDestinationHealth}} destination{{if ne (len .AdminDestinationHealth) 1}}s{{end}}
{{if .AdminDestinationHealth}}
- {{range .AdminDestinationHealth}}{{end}} + {{range .AdminDestinationHealth}}{{end}}
AccountDestinationHealthQueueLast errorUpdated
{{.UserEmail}}{{.DestinationName}}{{.Service}}{{destinationHealthLabel .Status}}{{if .ConsecutiveFailures}}{{.ConsecutiveFailures}} consecutive failures{{end}}{{.PendingCount}} pending{{.FailedCount}} failed{{if .LastError}}{{.LastError}}{{else}}—{{end}}{{if .UpdatedAt}}{{formatTime .UpdatedAt}}{{end}}
{{.UserEmail}}{{.DestinationName}}{{.Service}}{{destinationHealthLabel .Status}}{{if .ConsecutiveFailures}}{{.ConsecutiveFailures}} consecutive failures{{end}}{{.PendingCount}} pending{{.BlockedCount}} blocked{{.FailedCount}} failed{{if eq .Status "unsupported"}}Transport is unsupported; replace the destination.{{else if .LastError}}{{.LastError}}{{else}}—{{end}}{{if .UpdatedAt}}{{formatTime .UpdatedAt}}{{end}}
{{else}}
No notification destinations configured.
{{end}}
diff --git a/internal/web/templates/partials.html b/internal/web/templates/partials.html index 93ea7ac..46bc960 100644 --- a/internal/web/templates/partials.html +++ b/internal/web/templates/partials.html @@ -53,8 +53,8 @@

Delivery

Your destinations

Credentials are encrypted at rest. Send a test after adding a destination.

{{len .Destinations}} configured
{{if .Destinations}}
{{range .Destinations}}{{$health := index $.DestinationHealth .ID}}
{{.Name}}{{.Service}} · credentials hidden - {{destinationHealthLabel $health.Status}}{{if $health.PendingCount}}{{$health.PendingCount}} pending{{end}}{{if $health.FailedCount}}{{$health.FailedCount}} failed{{end}}{{if $health.ConsecutiveFailures}}{{$health.ConsecutiveFailures}} consecutive failure{{if ne $health.ConsecutiveFailures 1}}s{{end}}{{end}} - {{if $health.LastError}}{{$health.LastError}}{{end}} + {{destinationHealthLabel $health.Status}}{{if $health.PendingCount}}{{$health.PendingCount}} pending{{end}}{{if $health.BlockedCount}}{{$health.BlockedCount}} blocked{{end}}{{if $health.FailedCount}}{{$health.FailedCount}} failed{{end}}{{if $health.ConsecutiveFailures}}{{$health.ConsecutiveFailures}} consecutive failure{{if ne $health.ConsecutiveFailures 1}}s{{end}}{{end}} + {{if eq $health.Status "unsupported"}}This transport is no longer supported. Replace it with an HTTP(S)-based destination.{{else if $health.LastError}}{{$health.LastError}}{{end}} {{if $health.NextRetryAt}}Retry after {{formatProviderTime $health.NextRetryAt}}{{end}}
Rename
@@ -73,10 +73,10 @@ {{template "csrf" .}}
@@ -85,9 +85,8 @@
-
-
+ Advanced URLs must use Discord, Telegram, ntfy, or generic HTTP(S). Gotify and SMTP are not accepted until safe transport adapters are available.
diff --git a/internal/web/web.go b/internal/web/web.go index 5b180db..317d798 100644 --- a/internal/web/web.go +++ b/internal/web/web.go @@ -24,25 +24,27 @@ const ( ) type App struct { - cfg config.Config - store *store.Store - mb catalog.CatalogProvider - spotify catalog.SpotifyProvider - itunes catalog.ITunesProvider - sender notify.NotificationSender - cipher *security.Cipher - artwork artwork.Provider - jobs *jobs.Runner - logger *slog.Logger - templates *template.Template - setupLimiter *fixedWindowLimiter - loginLimiter *fixedWindowLimiter - tokenLimiter *fixedWindowLimiter - discoveryLimiter *fixedWindowLimiter - importLimiter *fixedWindowLimiter - importSlots chan struct{} - providerLimiter *fixedWindowLimiter - loginSlots chan struct{} + cfg config.Config + store *store.Store + mb catalog.CatalogProvider + spotify catalog.SpotifyProvider + itunes catalog.ITunesProvider + sender notify.NotificationSender + cipher *security.Cipher + artwork artwork.Provider + jobs *jobs.Runner + logger *slog.Logger + templates *template.Template + setupLimiter *fixedWindowLimiter + loginLimiter *fixedWindowLimiter + tokenLimiter *fixedWindowLimiter + discoveryLimiter *fixedWindowLimiter + importLimiter *fixedWindowLimiter + importSlots chan struct{} + providerLimiter *fixedWindowLimiter + destinationTestLimiter *fixedWindowLimiter + destinationRetryLimiter *fixedWindowLimiter + loginSlots chan struct{} } // UserView is the deliberately narrow projection exposed to templates. The diff --git a/scripts/backup.sh b/scripts/backup.sh index 130dc94..23312b0 100755 --- a/scripts/backup.sh +++ b/scripts/backup.sh @@ -69,4 +69,9 @@ docker run --rm --volumes-from "$container_id" -v "$output_dir:/backup" "$HELPER sha256sum "$(basename -- "$output_file")" > "$(basename -- "$checksum_file")" ) chmod 600 "$output_file" "$checksum_file" +# Write the marker only after both the archive and checksum are complete. It is +# deliberately not part of the just-created archive; the next backup will +# include it, while the live application immediately sees the latest result. +docker run --rm --volumes-from "$container_id" "$HELPER_IMAGE" sh -ec \ + 'printf "%s\\n" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" > /data/.artist-trackarr-last-backup && chmod 600 /data/.artist-trackarr-last-backup' echo "backup: wrote $output_file and $checksum_file from volume $mount_name" diff --git a/scripts/restore-smoke.sh b/scripts/restore-smoke.sh index b595b4a..24922dc 100755 --- a/scripts/restore-smoke.sh +++ b/scripts/restore-smoke.sh @@ -18,7 +18,21 @@ if [ -z "${APP_ENCRYPTION_KEY:-}" ] || [ "${#APP_ENCRYPTION_KEY}" -lt 32 ]; then exit 1 fi -image=${ARTIST_TRACKARR_IMAGE:-ghcr.io/crypt0rr/artist-trackarr:latest} +image=${ARTIST_TRACKARR_IMAGE:-} +if [ -z "$image" ]; then + echo "restore: set ARTIST_TRACKARR_IMAGE to an immutable image digest (for example ghcr.io/crypt0rr/artist-trackarr@sha256:...)" >&2 + exit 1 +fi +case "$image" in + *@sha256:*) ;; + *) + if [ "${RESTORE_ALLOW_MUTABLE_IMAGE:-false}" != "true" ]; then + echo "restore: image must use an immutable @sha256 digest (set RESTORE_ALLOW_MUTABLE_IMAGE=true only for local rehearsal)" >&2 + exit 1 + fi + echo "restore: warning: mutable image explicitly allowed" >&2 + ;; +esac port=${RESTORE_SMOKE_PORT:-18080} volume="artist-trackarr-restore-$$" container="artist-trackarr-restore-$$" @@ -38,18 +52,39 @@ if [ -s "$checksum_path" ]; then sha256sum -c "$(basename -- "$checksum_path")" ) else - echo "restore: checksum sidecar is absent; accepting a legacy archive" >&2 + if [ "${RESTORE_ALLOW_LEGACY_ARCHIVE:-false}" != "true" ]; then + echo "restore: checksum sidecar is required (set RESTORE_ALLOW_LEGACY_ARCHIVE=true only for a legacy archive)" >&2 + exit 1 + fi + echo "restore: warning: checksum sidecar is absent; accepting an explicitly allowed legacy archive" >&2 fi docker run --rm -v "$volume:/data" -v "$archive_path:/backup/restore.tgz:ro" "$HELPER_IMAGE" \ - sh -ec 'tar xzf /backup/restore.tgz -C /data; test -s /data/artist-tracker.db; apk add --no-cache sqlite >/dev/null; test -z "$(sqlite3 /data/artist-tracker.db "PRAGMA foreign_key_check;")"' + sh -ec 'tar xzf /backup/restore.tgz -C /data; test -s /data/artist-tracker.db; apk add --no-cache sqlite >/dev/null; test "$(sqlite3 /data/artist-tracker.db "PRAGMA integrity_check;")" = ok; test -z "$(sqlite3 /data/artist-tracker.db "PRAGMA foreign_key_check;")"' state_fingerprint() { - docker run --rm -v "$volume:/data" "$HELPER_IMAGE" sh -ec ' - apk add --no-cache sqlite >/dev/null - sqlite3 -noheader -separator "|" /data/artist-tracker.db \ - "SELECT (SELECT count(*) FROM schema_migrations),(SELECT count(*) FROM users),(SELECT count(*) FROM artists),(SELECT count(*) FROM follows),(SELECT count(*) FROM release_groups),(SELECT count(*) FROM destinations);" \ - | sha256sum | cut -d" " -f1 - ' + docker run --rm -v "$volume:/data" "$HELPER_IMAGE" sh -ec " +set -eu +apk add --no-cache sqlite >/dev/null +fingerprint_file=\$(mktemp) +sqlite3 -noheader -separator '|' /data/artist-tracker.db >\"\$fingerprint_file\" <<'SQL' +-- Fingerprint durable logical state, not runner-owned timestamps, claims, +-- observations, sessions, or queue status that may legitimately change +-- while the restored process starts and performs maintenance. +SELECT 'schema',version FROM schema_migrations ORDER BY version; +SELECT 'users',id,email,username,password_hash,role,timezone,reminder_time FROM users ORDER BY id; +SELECT 'artists',id,mbid,name,sort_name,artist_type,country,disambiguation,spotify_id,spotify_url,spotify_image_url FROM artists ORDER BY id; +SELECT 'follows',user_id,artist_id FROM follows ORDER BY user_id,artist_id; +SELECT 'releases',id,mbid,artist_id,title,primary_type,secondary_types,first_release_date,date_precision,musicbrainz_url,spotify_url,itunes_url,spotify_id,itunes_id,source,artist_credit_role,itunes_artwork_url FROM release_groups ORDER BY id; +SELECT 'destinations',id,user_id,name,service,hex(encrypted_url),enabled,transport_status,transport_message FROM destinations ORDER BY id; +SELECT 'events',id,user_id,release_group_id,event_type,title,body FROM notification_events ORDER BY id; +SELECT 'prefs',user_id,albums,eps,singles,announcements,release_day,release_digest_enabled,release_digest_frequency,hold_conflicting_notifications FROM notification_preferences ORDER BY user_id; +SELECT 'genres',artist_id,genre,source FROM artist_genres ORDER BY artist_id,genre,source; +SELECT 'credits',release_group_id,artist_id,provider,provider_id,role,track_title,credit_name,provider_url,confidence FROM release_credits ORDER BY release_group_id,artist_id,provider,provider_id,role,track_title; +SQL +sha256sum \"\$fingerprint_file\" > \"\$fingerprint_file.sha256\" +fingerprint_hash=\$(cut -d' ' -f1 \"\$fingerprint_file.sha256\") +printf '%s\\n' "\$fingerprint_hash" +" } docker run -d --name "$container" --network host -v "$volume:/data" \ @@ -95,4 +130,6 @@ if [ "$before_fingerprint" != "$after_fingerprint" ]; then echo "restore: database state changed across restart" >&2 exit 1 fi +docker run --rm -v "$volume:/data" "$HELPER_IMAGE" sh -ec \ + 'printf "%s\\n%s\\n" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "ok" > /data/.artist-trackarr-last-restore && chmod 600 /data/.artist-trackarr-last-restore' echo "restore: readiness, foreign-key check, key-preserving startup, and restart persistence passed (state $after_fingerprint)" From c7487cd53fd6849bea22847c7f60e452448b38d7 Mon Sep 17 00:00:00 2001 From: crypt0rr <57799908+crypt0rr@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:01:54 +0200 Subject: [PATCH 2/2] Pin patched Go toolchain for CI --- .github/workflows/docker.yml | 5 +++-- Dockerfile | 2 +- Makefile | 4 ++-- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index 0f96514..18fac99 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -61,8 +61,9 @@ jobs: with: # Keep CI on the patched toolchain used by the Docker build. The # go.mod language version remains 1.25 for source compatibility; - # reading it verbatim would install the vulnerable 1.25.0 release. - go-version: '1.26.x' + # pin the patch version so a newly published standard-library + # vulnerability cannot silently select an older toolchain. + go-version: '1.26.6' cache: true - name: Run race detector diff --git a/Dockerfile b/Dockerfile index 06371d7..c7211eb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,5 @@ # syntax=docker/dockerfile:1.26@sha256:ecfaec9ed6d810b56388c508f4121597bfbba70d41a6dfeee4d8cad5f295fc32 -FROM golang:1.26-alpine@sha256:0178a641fbb4858c5f1b48e34bdaabe0350a330a1b1149aabd498d0699ff5fb2 AS dependencies +FROM golang:1.26.6-alpine@sha256:af8d6740070b8906d12eae1c3e3ea0957fb63f492051ea05e354c38ef9fe88df AS dependencies WORKDIR /src RUN apk add --no-cache ca-certificates tzdata COPY go.mod go.sum* ./ diff --git a/Makefile b/Makefile index 00c453f..8e121ef 100644 --- a/Makefile +++ b/Makefile @@ -16,11 +16,11 @@ fmt-check: @test -z "$$(gofmt -l internal cmd)" tooling-check: - @marker=$$(printf '@%s' latest); if rg -n "$$marker" .github Dockerfile scripts; then \ + @marker=$$(printf '@%s' latest); if git grep -n -F "$$marker" -- .github Dockerfile scripts; then \ echo "unpinned tool reference found" >&2; \ exit 1; \ fi - @if rg -n --pcre2 '\b(?:golang|alpine):[^@[:space:]]+(?:[[:space:]]|$$)' Dockerfile scripts; then \ + @if git grep -n -E '(^|[^[:alnum:]_-])(golang|alpine):[^@[:space:]]+([[:space:]]|$$)' -- Dockerfile scripts; then \ echo "unpinned container helper reference found" >&2; \ exit 1; \ fi