feat(scrobbling): give a listen a queue it can leave the server by

Cargo.toml

@@ -123,3 +123,7 @@ path = "tests/uploads.rs"
 [[test]]
 name = "canvas"
 path = "tests/canvas.rs"
+
+[[test]]
+name = "scrobbling"
+path = "tests/scrobbling.rs"

docs/rfcs/RFC-010-external-scrobbling.md

@@ -1,9 +1,14 @@
 # RFC-010 — Le scrobbling externe
 
 - **Statut** : Proposed
-- **Implémentée par** : rien encore. Quand du code existera, c'est cette ligne
-  qui nommera les PR, et le champ *Statut* ci-dessus ne basculera pas — il ne
-  bascule jamais dans ce projet.
+- **Implémentée par** :
+  [#191](https://github.com/InstaZDLL/waveflow-server/pull/191), la moitié
+  durable — les deux tables, la mise en file dans la transaction qui écrit
+  `play_event`, les cinq verdicts, et la tâche de drainage. **Aucun adaptateur,
+  donc aucun appel sortant** : le serveur ne parle toujours à personne, et
+  `Cargo.toml` ne porte toujours pas de client HTTP. ListenBrainz, les routes
+  natives et la CLI viennent ensuite, par la décision 11. Le champ *Statut*
+  ci-dessus ne bascule pas — il ne bascule jamais dans ce projet.
 - **Date** : 2026-09-13
 - **Révisée** : 2026-09-13, après revue externe. Les décisions 2 à 6 ont changé,
   et chacune dit ce que la version antérieure affirmait de faux plutôt que de
@@ -339,5 +344,19 @@ version laissait pendantes sont tranchées ci-dessus, et ce qui reste appartient
 à l'implémentation : le nom exact des routes, le seuil au-delà duquel une file
 qui n'avance pas devient `degraded`, et la forme précise du JSON.
 
-C'est la ligne *Implémentée par* de l'en-tête qu'il faudra lire ensuite. Elle
-dit encore « rien encore ».
+**Une question que cette RFC n'avait pas posée** est apparue en relisant le code
+de [#191](https://github.com/InstaZDLL/waveflow-server/pull/191) : la file ne se
+purge jamais. Une ligne terminale — `sent`, `rejected`, `abandoned`, `cancelled`,
+`discarded` — reste indéfiniment, donc la table grandit d'une ligne par écoute et
+par destination, sans fin, chez un auditeur actif. Ce n'est pas une décision
+d'architecture et cela n'a pas retenu #191, parce que la rétention ne demande
+aucune migration : un réglage de déploiement et une tâche de purge, sur le patron
+de [RFC-007](RFC-007-library-event-stream.md) — une fenêtre, un plancher, et la
+septième tâche de fond qui existe déjà.
+
+Une seule contrainte en sort, et elle vient d'ici : **`uncertain` ne se purge
+jamais.** Cette ligne-là attend la décision d'une personne, et la lui retirer
+au bout de trente jours, c'est décider à sa place — exactement ce que la
+décision 13 refuse.
+
+C'est la ligne *Implémentée par* de l'en-tête qu'il faut lire pour le reste.

migrations-v2/20260913000000_scrobble_outbox.sql

@@ -0,0 +1,148 @@
+-- Where an account's authorisation to submit a listen lives, and what is
+-- waiting to be submitted under it. RFC-010 carries the reasoning; what follows
+-- is the shape.
+
+-- One generation of one account's authorisation at one destination.
+--
+-- The row *is* the generation, which is why relinking inserts a new one rather
+-- than updating this. RFC-010 decision 4: twenty listens are waiting, the
+-- account unlinks Last.fm and links a different one — the account and the
+-- destination are the same pair, the authorisation is not, and a queue keyed on
+-- that pair would send the twenty to the second profile. `scrobble_outbox`
+-- references this id, so the old rows can only ever name the authorisation they
+-- were queued under.
+--
+-- The secret is sealed by `SecretBox` under the instance key, exactly as the
+-- dedicated Subsonic password is: `data/waveflow.db` and `data/instance.key`
+-- are one backup or neither is.
+--
+-- No destination URL here. Decision 10: a base URL is the operator's setting,
+-- never an account's — a member picks among the destinations the server knows
+-- and does not get to describe one.
+CREATE TABLE scrobble_link (
+    id TEXT PRIMARY KEY NOT NULL,
+    user_id TEXT NOT NULL REFERENCES account(id) ON DELETE CASCADE,
+    provider TEXT NOT NULL CHECK (provider IN ('listenbrainz', 'maloja', 'lastfm')),
+    -- `unlinked` is terminal and keeps the generation readable; `broken` is what
+    -- an adapter's `AuthBroken` verdict writes, and it still holds its secret so
+    -- the account is told which link stopped working rather than finding it
+    -- gone.
+    status TEXT NOT NULL CHECK (status IN ('active', 'broken', 'unlinked')),
+    credential_nonce BLOB NOT NULL CHECK (length(credential_nonce) = 12),
+    credential_ciphertext BLOB NOT NULL,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL,
+    last_success_at INTEGER,
+    -- A normalised cause, never the destination's own words. Decision 12: what
+    -- the API shows is a state, not an echo.
+    last_failure TEXT
+) STRICT;
+
+-- At most one live generation per account and destination, held by the schema
+-- rather than by whoever remembers to look first. Past generations stay
+-- readable because `unlinked` is excluded.
+CREATE UNIQUE INDEX scrobble_link_live_idx
+    ON scrobble_link(user_id, provider) WHERE status <> 'unlinked';
+
+-- One listen, frozen as it was heard, waiting for one destination.
+--
+-- The envelope is a copy and that is the point. RFC-010 decision 2: since #186 a
+-- member corrects titles and artists, and a correction rewrites
+-- `track_participant` — reading the track at drain time would submit what it has
+-- become rather than what was played. A track deleted between the listen and the
+-- send no longer empties the row either.
+CREATE TABLE scrobble_outbox (
+    -- The rowid stays the internal identity: `retry_of` points at it, the drain
+    -- orders by it, and the jitter that keeps two rows from returning together
+    -- is derived from it. None of that wants a UUID.
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    -- What the outside is allowed to name. Public ids are UUIDs here, and the
+    -- two gestures decision 13 gives a person — discard this entry, retry that
+    -- one — are the only things that ever name a row from outside.
+    --
+    -- A sequential integer would work and would also say how many listens this
+    -- whole server has ever queued, to anyone holding one of their own. Added
+    -- now rather than when the routes arrive: while this migration is unmerged
+    -- it is one column, and afterwards it is a second migration and a backfill.
+    public_id TEXT NOT NULL UNIQUE,
+    link_id TEXT NOT NULL REFERENCES scrobble_link(id) ON DELETE CASCADE,
+    -- Provenance, and nullable on purpose: `ON DELETE SET NULL` is what lets the
+    -- envelope outlive the track it came from. A cascade here would undo the
+    -- whole reason the envelope is a copy.
+    play_event_id INTEGER REFERENCES play_event(id) ON DELETE SET NULL,
+    -- Set only by the one deliberate duplicate this design accepts: decision 13,
+    -- where a person retries an `uncertain` entry knowing the destination may
+    -- already hold it. Nothing else ever writes it, so `retry_of IS NOT NULL`
+    -- *means* a person asked for this row.
+    retry_of INTEGER REFERENCES scrobble_outbox(id) ON DELETE SET NULL,
+    played_at INTEGER NOT NULL,
+    title TEXT NOT NULL,
+    artists_json TEXT NOT NULL CHECK (json_valid(artists_json)),
+    album TEXT,
+    album_artist TEXT,
+    duration_ms INTEGER,
+    musicbrainz_recording_id TEXT,
+    -- `sent`, `rejected`, `abandoned`, `cancelled`, `uncertain` and `discarded`
+    -- are all terminal. They are six words rather than one because each names a
+    -- different thing that happened, and the counters decision 12 describes
+    -- cannot be computed from a single `done`.
+    --
+    -- `sending` is the one transient state, and it is what makes a submission
+    -- at-most-once rather than merely usually-once. A row is moved into it by
+    -- an `UPDATE … WHERE state='pending'` before anything is emitted, so two
+    -- passes cannot both take it — and a row found in it long after any
+    -- deadline could have expired belongs to a process that died mid-flight,
+    -- which is `uncertain` by decision 5: nobody knows whether the destination
+    -- recorded it. Left as `pending`, that row would simply be sent again.
+    state TEXT NOT NULL CHECK (
+        state IN (
+            'pending', 'sending', 'sent', 'uncertain', 'discarded', 'rejected',
+            'abandoned', 'cancelled'
+        )
+    ),
+    attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    next_attempt_at INTEGER NOT NULL,
+    last_failure TEXT,
+    created_at INTEGER NOT NULL,
+    updated_at INTEGER NOT NULL
+) STRICT;
+
+-- Exactly one queueing per listen and per authorisation — decision 5, held by
+-- the schema and not only by the code.
+--
+-- `WHERE retry_of IS NULL` is what leaves room for the single exception:
+-- a retry a person asked for names the same listen on purpose, and refusing it
+-- here would make decision 13 unimplementable. Rows whose `play_event_id` has
+-- been nulled by a deleted track are all distinct to SQLite, which is correct:
+-- there is no longer a listen anything could queue a second time.
+CREATE UNIQUE INDEX scrobble_outbox_once_idx
+    ON scrobble_outbox(play_event_id, link_id) WHERE retry_of IS NULL;
+
+-- One deliberate duplicate per ambiguous entry, and not two.
+--
+-- Decision 13 lets a person retry a listen whose fate is unknown, accepting
+-- that the destination may already hold it. That acceptance is given once. The
+-- original stays `uncertain` afterwards — it must, because erasing it would
+-- falsify the only trace explaining why a duplicate exists — so nothing in the
+-- row itself says it has already been answered, and a second call would queue a
+-- second copy. Here rather than only in the service, for the reason decision 5
+-- gives about the index above: the one path in this design that can manufacture
+-- duplicates on demand should be shut by the schema.
+CREATE UNIQUE INDEX scrobble_outbox_one_retry_idx
+    ON scrobble_outbox(retry_of) WHERE retry_of IS NOT NULL;
+
+-- What the drain asks for: the due rows, in the order they became due.
+CREATE INDEX scrobble_outbox_due_idx ON scrobble_outbox(state, next_attempt_at);
+
+-- What the counters ask for: one link's queue, by state.
+CREATE INDEX scrobble_outbox_link_idx ON scrobble_outbox(link_id, state);
+
+-- What the foreign key asks for when a listen is deleted.
+--
+-- `play_event_id` is `ON DELETE SET NULL`, so removing a `play_event` row makes
+-- SQLite look for the outbox rows pointing at it. The only other index leading
+-- with this column is `scrobble_outbox_once_idx`, which is partial and so cannot
+-- answer for a retry row — leaving a scan. Deleting tracks is not a rare event
+-- here: an ordinary rescan that finds files gone cascades track -> play_event
+-- -> this lookup.
+CREATE INDEX scrobble_outbox_play_event_idx ON scrobble_outbox(play_event_id);

src/config.rs

@@ -71,6 +71,57 @@ pub struct CanvasLimits {
     pub library_quota_bytes: i64,
 }
 
+/// How the server drains what it owes a third party. RFC-010.
+///
+/// Apart from the credentials, which belong to an account and are posed through
+/// the API — decision 9. What is left here is what belongs to the deployment:
+/// how often the queue moves, how many times a failure is worth repeating, and
+/// how long a queue may sit still before the operator should be told it has
+/// stopped.
+#[derive(Debug, Clone, Copy)]
+pub struct ScrobbleLimits {
+    /// How often the drain walks the queue.
+    pub drain_interval: Duration,
+    /// How long one submission may take before the drain stops waiting for it.
+    ///
+    /// A deployment setting rather than a constant because decision 9 says so,
+    /// and it is load-bearing: the drain is a background task, so a destination
+    /// that accepts a connection and then never answers would hold the queue
+    /// still for the life of the process — the silent failure `degraded` exists
+    /// to surface, arriving by the one route that would also stop `degraded`
+    /// from ever being computed.
+    pub request_timeout: Duration,
+    /// How many times one listen may be submitted before it is abandoned and
+    /// counted. Bounded because a queue that never empties is a fault and not a
+    /// state — decision 6.
+    pub max_attempts: u32,
+    /// How long the oldest waiting listen may be waiting before the link is
+    /// reported `degraded`. A valid token with a queue that has not moved for
+    /// hours is the silent failure a durable queue exists to make visible —
+    /// decision 12.
+    pub stale_after: Duration,
+    /// How many listens one drain pass takes. It bounds the pass, never the
+    /// request: decision 11 sends one listen per request whatever this says.
+    pub batch: usize,
+}
+
+/// How many times a listen is offered before the queue gives up on it.
+///
+/// Thirty, because the waits grow to an hour and stop there: six doublings from
+/// a minute, then an hour apiece. Thirty submissions therefore span a little
+/// over **a day** of a destination being down, which is the span worth
+/// surviving — a nightly maintenance window, a regional outage, a certificate
+/// nobody renewed until morning.
+///
+/// It was eight, with a comment claiming the same day; eight delivers two hours
+/// and three minutes. The comment was the honest statement of what this is for,
+/// so the number moved to meet it rather than the sentence being trimmed to fit.
+/// Giving up early buys nothing here: abandoning a listen is a permanent hole in
+/// someone's history, and unlike a retry after an ambiguous answer it risks no
+/// duplicate at all. `the_default_attempt_cap_carries_a_listen_across_a_day_of_outage`
+/// keeps this paragraph and the arithmetic from drifting apart again.
+pub const DEFAULT_SCROBBLE_MAX_ATTEMPTS: u32 = 30;
+
 #[derive(Clone)]
 pub struct Config {
     pub bind_addr: SocketAddr,
@@ -140,6 +191,16 @@ pub struct Config {
     /// there is — sharing the flag made this whole feature inert exactly where
     /// it is most wanted.
     pub canvas: CanvasLimits,
+    /// How the server drains what it owes a third party.
+    ///
+    /// `WAVEFLOW_SCROBBLE_DRAIN_INTERVAL_SECS`,
+    /// `WAVEFLOW_SCROBBLE_REQUEST_TIMEOUT_SECS`,
+    /// `WAVEFLOW_SCROBBLE_MAX_ATTEMPTS`, `WAVEFLOW_SCROBBLE_STALE_AFTER_SECS`,
+    /// `WAVEFLOW_SCROBBLE_BATCH`.
+    ///
+    /// None of these matter until an account links a destination: a server that
+    /// has only been upgraded makes no outbound request at all.
+    pub scrobbling: ScrobbleLimits,
     pub allowed_origins: Vec<axum::http::HeaderValue>,
     /// How the catalogue decides which row a scanned file belongs to.
     ///
@@ -180,6 +241,7 @@ impl std::fmt::Debug for Config {
             .field("library_event_retention", &self.library_event_retention)
             .field("canvas_dir", &self.canvas_dir)
             .field("canvas", &self.canvas)
+            .field("scrobbling", &self.scrobbling)
             .field("allowed_origins", &self.allowed_origins)
             .field("pid", &self.pid)
             .finish()
@@ -262,6 +324,32 @@ impl Config {
             )?,
         };
         validate_canvas(&canvas)?;
+        let scrobbling = ScrobbleLimits {
+            // A minute is far below what any destination considers a listening
+            // history's resolution, and far above what a queue of a few rows
+            // costs to walk.
+            drain_interval: Duration::from_secs(parse_positive_env(
+                "WAVEFLOW_SCROBBLE_DRAIN_INTERVAL_SECS",
+                60u64,
+            )?),
+            max_attempts: parse_positive_env(
+                "WAVEFLOW_SCROBBLE_MAX_ATTEMPTS",
+                DEFAULT_SCROBBLE_MAX_ATTEMPTS,
+            )?,
+            stale_after: Duration::from_secs(parse_positive_env(
+                "WAVEFLOW_SCROBBLE_STALE_AFTER_SECS",
+                3_600u64,
+            )?),
+            batch: parse_positive_env("WAVEFLOW_SCROBBLE_BATCH", 50usize)?,
+            // Thirty seconds is far past any of the three destinations'
+            // ordinary latency and far short of a drain that has stopped. It
+            // bounds one submission, never the pass: the pass is bounded by
+            // `batch`.
+            request_timeout: Duration::from_secs(parse_positive_env(
+                "WAVEFLOW_SCROBBLE_REQUEST_TIMEOUT_SECS",
+                30u64,
+            )?),
+        };
         // Both refuse zero and negatives at startup rather than falling back:
         // every fallback for a bound is wrong, and the operator is turned away
         // where they can see why. No ceiling — an enormous value means "purge
@@ -342,6 +430,7 @@ impl Config {
             library_event_retention,
             canvas_dir,
             canvas,
+            scrobbling,
             allowed_origins,
             pid,
         })
@@ -398,6 +487,20 @@ impl Config {
                 max_duration_secs: 15,
                 library_quota_bytes: 1024 * 1024,
             },
+            // Same reasoning again: small enough that a test can exhaust the
+            // attempts without waiting out eight doublings, shaped like
+            // production rather than unlimited. The interval is never reached —
+            // `spawn_scrobble_drain` is started by `main`, and tests run the
+            // pass themselves.
+            scrobbling: ScrobbleLimits {
+                drain_interval: Duration::from_secs(60),
+                // Long enough that no test double ever meets it by accident;
+                // the one test that means to meet it shortens this first.
+                request_timeout: Duration::from_secs(30),
+                max_attempts: 3,
+                stale_after: Duration::from_secs(3600),
+                batch: 8,
+            },
             allowed_origins: Vec::new(),
             // The real defaults, so the whole test suite exercises the specs
             // production runs under rather than a simplified stand-in.

src/main.rs

@@ -38,6 +38,10 @@ async fn serve(config: Config, state: waveflow_server::AppState) -> anyhow::Resu
     state.services.spawn_canvas_sweeper();
     state.services.spawn_artwork_sweeper();
     state.services.spawn_library_event_purge();
+    // Nothing to drain until an account links a destination, and no adapter is
+    // registered yet — a server that was merely upgraded makes no outbound
+    // request. RFC-010.
+    state.services.spawn_scrobble_drain();
     state.db.spawn_authorization_pruning();
     let router = waveflow_server::app(&config, state);
     let listener = tokio::net::TcpListener::bind(bind_addr)

src/services/mod.rs

@@ -976,6 +976,14 @@ pub struct DomainServices {
     /// no business happening while the process-wide gate is held — the same
     /// reason `upload_locks` exists.
     canvas_locks: Arc<dashmap::DashMap<String, Arc<tokio::sync::Mutex<()>>>>,
+    scrobbling: crate::config::ScrobbleLimits,
+    /// The destinations this process knows how to reach.
+    ///
+    /// Filled after construction rather than at it, which is what lets a test
+    /// drive the whole queue against a double and lets a server run with no
+    /// outbound adapter at all — the ordinary case, since RFC-010 decision 4
+    /// forbids shipping any provider credential in an AGPL binary.
+    scrobble_targets: scrobbling::ScrobbleTargets,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -1024,13 +1032,22 @@ mod library_events;
 mod playback;
 mod playlists;
 mod scan;
+mod scrobbling;
 mod search;
 mod shares;
 mod songs;
 mod sync;
 mod track_metadata;
 mod uploads;
 
+/// The scrobbling vocabulary, which is the one thing in this module tree that
+/// has to be spoken outside it: an adapter implements [`ScrobbleTarget`]
+/// without being part of the domain, and the API reads the link states.
+pub use scrobbling::{
+    ScrobbleDrain, ScrobbleEnvelope, ScrobbleLinkState, ScrobbleProvider, ScrobbleTarget,
+    ScrobbleVerdict,
+};
+
 impl DomainServices {
     pub fn new(
         db: Database,
@@ -1056,6 +1073,8 @@ impl DomainServices {
             ffprobe_path: config.ffprobe_path.clone(),
             ffmpeg_path: config.ffmpeg_path.clone(),
             canvas_locks: Arc::new(dashmap::DashMap::new()),
+            scrobbling: config.scrobbling,
+            scrobble_targets: Arc::new(dashmap::DashMap::new()),
         }
     }