DeadAir on the server: copies with long silences shortened (#171)

For anyone who turns DeadAir on, a background task makes each episode a
copy with every silence quieter than -38 dB for over 0.35 s shortened to
0.2 s. Nothing is re-encoded: ffmpeg copies the original's frames and
leaves the silent ones out, about 11 s for an hour of MP3 in a release
build. On Brad & Will's episode 359 that is 520 cuts and 3.3 minutes
saved, the copy decodes clean, and it has no isolated spikes in its
silences that the original lacks.

Cuts are made from the middle of each silence, keeping 0.1 s at each
end, so the frames either side of a cut are silent: decoding blends a
frame with the one before it, and an MP3 frame can begin in the bytes of
the frame before, so the first frame after a cut decodes against the
wrong neighbour. Inside a silence nobody hears that; at a word they would.

Positions stay in the original's time and the server converts them both
ways. The iOS and Android apps save positions themselves from their own
player, even while the page is asleep, so converting in the page would
have stored the copy's times as the original's. /media/{id}/deadair
serves the copy at its own address and records that this person is
playing it; a position saved meanwhile is converted from the copy's
clock, as is the length a player measures.

The copy is made by a task of its own, not the worker: a first pass over
a library is hours of ffmpeg, and scans must not wait behind it. It
checks every two minutes, which covers new downloads, files already
downloaded, a switch turned on and a file downloaded again with one rule.
Episodes everyone who would hear the copy has finished get none.
download::remove deletes a copy with its original, and the quota counts
copies.

Settings: the switch is PATCH /api/me {deadair}, a feed kept out is
PATCH /api/feeds/{id} {deadair: false}. GET /api/me gives the switch and
deadair_saved, the seconds saved on episodes heard 75% or more on a copy,
each counted once. Three new tables, so no ALTER.

The image gains Debian's ffmpeg, about 480 MB, in a layer that stays
cached until the base image changes. The player's switch, the feed
setting and the stat come next, after mockups.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/db.rs

@@ -2,7 +2,7 @@
 //! per-feed .ipxd plists, history.dat and qmcache.dat.
 
 use anyhow::{Context, Result};
-use crate::entity::{api_tokens, blocklists, catalogue, enclosures, entries, feeds, hidden, sessions, settings, subscriptions, users};
+use crate::entity::{api_tokens, blocklists, catalogue, deadair_prefs, enclosures, entries, feeds, hidden, sessions, settings, subscriptions, users};
 use sea_orm::sea_query::{Expr, Func};
 use sea_orm::{
     ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set,
@@ -60,6 +60,9 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
         schema.create_table_from_entity(blocklists::Entity),
         schema.create_table_from_entity(hidden::Entity),
         schema.create_table_from_entity(api_tokens::Entity),
+        schema.create_table_from_entity(deadair::Entity),
+        schema.create_table_from_entity(deadair_prefs::Entity),
+        schema.create_table_from_entity(deadair_saved::Entity),
     ] {
         orm.execute(table.if_not_exists()).await.context("creating the schema")?;
     }
@@ -198,6 +201,7 @@ impl From<enclosures::Model> for EncRow {
     fn from(e: enclosures::Model) -> Self {
         EncRow {
             id: e.id,
+            deadair: None,
             feed_id: e.feed_id,
             guid: e.guid,
             url: e.url,
@@ -299,6 +303,13 @@ impl Db {
     /// A SQLite file of its own (two connections to one ":memory:" would be two databases); or,
     /// with IPX_TEST_DATABASE_URL set, a Postgres schema of its own on that database, so the
     /// same tests prove the SQL on both.
+    /// Runs a statement as it is, for a test outside this module setting up what it needs.
+    #[cfg(test)]
+    pub async fn sql(&self, sql: &str) -> Result<()> {
+        self.orm.execute_unprepared(sql).await?;
+        Ok(())
+    }
+
     #[cfg(test)]
     pub async fn memory() -> Result<Self> {
         static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
@@ -1159,6 +1170,10 @@ impl Filter {
 #[derive(Debug, Clone, serde::Serialize)]
 pub struct EncRow {
     pub id: i64,
+    /// Seconds DeadAir's copy of this file saves, when the person asking plays it from one
+    /// (`deadair::for_reader`). None from the database.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub deadair: Option<i64>,
     pub feed_id: String,
     pub guid: String,
     pub url: String,
@@ -2192,6 +2207,188 @@ pub fn step(of: &'static str) -> tracing::Span {
     if tracing::Span::current().is_none() { tracing::Span::none() } else { tracing::info_span!("step", of) }
 }
 
+/// A file on disk with no DeadAir copy made since it was downloaded (issue #171).
+#[derive(Debug, Clone)]
+pub struct DeadAirTodo {
+    pub id: i64,
+    pub feed_id: String,
+    pub guid: String,
+    pub path: String,
+}
+
+/// An enclosure's DeadAir copy: where it is, what it left out, and how much that came to.
+#[derive(Debug, Clone)]
+pub struct Copy {
+    pub enclosure_id: i64,
+    pub path: String,
+    pub cuts: Vec<crate::deadair::Cut>,
+    pub saved_ms: i64,
+}
+
+impl Db {
+    /// DeadAir settings as (person, feed, wanted), everyone's or one person's. An empty feed is
+    /// the switch beside the speed.
+    pub async fn deadair_prefs(&self, user_id: Option<i64>) -> Result<Vec<(i64, String, bool)>> {
+        let mut a = Args::default();
+        let only = match user_id {
+            Some(u) => format!("WHERE user_id = {}", a.p(u)),
+            None => String::new(),
+        };
+        self.rows(&format!("SELECT user_id, feed_id, wanted FROM deadair_prefs {only}"), a.0)
+            .await?
+            .iter()
+            .map(|r| Ok((r.try_get("", "user_id")?, r.try_get("", "feed_id")?, r.try_get("", "wanted")?)))
+            .collect()
+    }
+
+    /// Sets one DeadAir setting, or with None takes it away: for a feed, that is following the
+    /// switch (or the feed's group) again.
+    pub async fn set_deadair(&self, user_id: i64, feed_id: &str, wanted: Option<bool>) -> Result<()> {
+        match wanted {
+            None => {
+                deadair_prefs::Entity::delete_by_id((user_id, feed_id.to_owned())).exec(&self.orm).await?;
+            }
+            Some(w) => {
+                self.exec(
+                    "INSERT INTO deadair_prefs (user_id, feed_id, wanted) VALUES ($1, $2, $3)
+                     ON CONFLICT (user_id, feed_id) DO UPDATE SET wanted = excluded.wanted",
+                    vec![user_id.into(), feed_id.into(), w.into()],
+                )
+                .await?;
+            }
+        }
+        Ok(())
+    }
+
+    /// Files on disk with no copy made since they were downloaded, oldest first.
+    pub async fn deadair_todo(&self) -> Result<Vec<DeadAirTodo>> {
+        self.rows(
+            "SELECT e.id, e.feed_id, e.guid, e.path FROM enclosures e
+               LEFT JOIN deadair d ON d.enclosure_id = e.id
+              WHERE e.path IS NOT NULL AND e.state = 'done'
+                AND (d.enclosure_id IS NULL OR d.made_at < coalesce(e.downloaded_at, 0))
+              ORDER BY e.id",
+            vec![],
+        )
+        .await?
+        .iter()
+        .map(|r| {
+            Ok(DeadAirTodo {
+                id: r.try_get("", "id")?,
+                feed_id: r.try_get("", "feed_id")?,
+                guid: r.try_get("", "guid")?,
+                path: r.try_get("", "path")?,
+            })
+        })
+        .collect()
+    }
+
+    /// Whether any of these people subscribes to the feed.
+    pub async fn subscribed_any(&self, users: &[i64], feed_id: &str) -> Result<bool> {
+        if users.is_empty() {
+            return Ok(false);
+        }
+        let mut a = Args::default();
+        let feed = a.p(feed_id);
+        let ids: Vec<String> = users.iter().map(|&u| a.p(u)).collect();
+        let sql = format!(
+            "SELECT 1 AS one FROM subscriptions WHERE feed_id = {feed} AND user_id IN ({}) LIMIT 1",
+            ids.join(", ")
+        );
+        Ok(!self.rows(&sql, a.0).await?.is_empty())
+    }
+
+    /// Who has an item marked read, which playing it to the end or past 90% does.
+    pub async fn readers(&self, feed_id: &str, guid: &str) -> Result<Vec<i64>> {
+        self.rows(
+            "SELECT user_id FROM entry_state WHERE feed_id = $1 AND guid = $2 AND read",
+            vec![feed_id.into(), guid.into()],
+        )
+        .await?
+        .iter()
+        .map(|r| Ok(r.try_get("", "user_id")?))
+        .collect()
+    }
+
+    /// Records the copy made of an enclosure, or with no path that it has none.
+    pub async fn store_deadair(&self, enclosure_id: i64, path: Option<&str>, cuts: &str, saved_ms: i64) -> Result<()> {
+        self.exec(
+            "INSERT INTO deadair (enclosure_id, path, cuts, saved_ms, made_at) VALUES ($1, $2, $3, $4, $5)
+             ON CONFLICT (enclosure_id) DO UPDATE SET path = excluded.path, cuts = excluded.cuts,
+                 saved_ms = excluded.saved_ms, made_at = excluded.made_at",
+            vec![enclosure_id.into(), path.map(str::to_owned).into(), cuts.into(), saved_ms.into(), now().into()],
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// The copies there are of these enclosures, made since each was last downloaded.
+    pub async fn copies(&self, enclosure_ids: &[i64]) -> Result<Vec<Copy>> {
+        if enclosure_ids.is_empty() {
+            return Ok(vec![]);
+        }
+        let mut a = Args::default();
+        let ids: Vec<String> = enclosure_ids.iter().map(|&i| a.p(i)).collect();
+        let sql = format!(
+            "SELECT d.enclosure_id, d.path, d.cuts, d.saved_ms FROM deadair d
+               JOIN enclosures e ON e.id = d.enclosure_id
+              WHERE d.enclosure_id IN ({}) AND d.path IS NOT NULL AND e.path IS NOT NULL
+                AND d.made_at >= coalesce(e.downloaded_at, 0)",
+            ids.join(", ")
+        );
+        self.rows(&sql, a.0)
+            .await?
+            .iter()
+            .map(|r| {
+                let cuts: String = r.try_get("", "cuts")?;
+                Ok(Copy {
+                    enclosure_id: r.try_get("", "enclosure_id")?,
+                    path: r.try_get("", "path")?,
+                    cuts: serde_json::from_str(&cuts).context("reading DeadAir's cuts")?,
+                    saved_ms: r.try_get("", "saved_ms")?,
+                })
+            })
+            .collect()
+    }
+
+    /// Counts an episode towards the time DeadAir has saved someone, once however often it is
+    /// played.
+    pub async fn count_deadair(&self, user_id: i64, feed_id: &str, guid: &str, saved_ms: i64) -> Result<()> {
+        self.exec(
+            "INSERT INTO deadair_saved (user_id, feed_id, guid, saved_ms, at) VALUES ($1, $2, $3, $4, $5)
+             ON CONFLICT (user_id, feed_id, guid) DO NOTHING",
+            vec![user_id.into(), feed_id.into(), guid.into(), saved_ms.into(), now().into()],
+        )
+        .await?;
+        Ok(())
+    }
+
+    /// An episode's length as someone's player measured it, else as its feed gives it.
+    pub async fn entry_duration(&self, user_id: i64, feed_id: &str, guid: &str) -> Result<Option<i64>> {
+        let rows = self
+            .rows(
+                "SELECT coalesce(s.duration, e.duration) AS duration FROM entries e
+                   LEFT JOIN entry_state s ON s.user_id = $1 AND s.feed_id = e.feed_id AND s.guid = e.guid
+                  WHERE e.feed_id = $2 AND e.guid = $3",
+                vec![user_id.into(), feed_id.into(), guid.into()],
+            )
+            .await?;
+        Ok(rows.first().map(|r| r.try_get::<Option<i64>>("", "duration")).transpose()?.flatten())
+    }
+
+    /// The time DeadAir has saved someone, in ms.
+    pub async fn deadair_total(&self, user_id: i64) -> Result<i64> {
+        let rows = self
+            .rows(
+                "SELECT CAST(coalesce(sum(saved_ms), 0) AS BIGINT) AS total FROM deadair_saved WHERE user_id = $1",
+                vec![user_id.into()],
+            )
+            .await?;
+        // Cast: Postgres sums bigints as NUMERIC, which does not read as an i64.
+        Ok(rows.first().map(|r| r.try_get::<i64>("", "total")).transpose()?.unwrap_or(0))
+    }
+}
+
 /// Unix seconds. Everything time-shaped in the DB is stored this way.
 pub fn now() -> i64 {
     std::time::SystemTime::now()
@@ -2204,6 +2401,51 @@ pub fn now() -> i64 {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn deadair_copies_are_made_once_per_download_and_counted_once_per_episode() {
+        let db = Db::memory().await.unwrap();
+        let u = db.create_user("dee", None, false).await.unwrap();
+        db.set_deadair(u, "", Some(true)).await.unwrap();
+        db.set_deadair(u, "f", Some(false)).await.unwrap();
+        db.set_deadair(u, "f", None).await.unwrap();
+        assert_eq!(db.deadair_prefs(Some(u)).await.unwrap(), vec![(u, String::new(), true)]);
+        for sql in [
+            "INSERT INTO enclosures (id, feed_id, guid, url, path, state, downloaded_at)
+             VALUES (1, 'f', 'g', 'http://x/a.mp3', '/x/a.mp3', 'done', 100)",
+            "INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES (2, 'f', 'h', 'http://x/b.mp3', NULL, 'reaped')",
+            "INSERT INTO entries (feed_id, guid, first_seen, duration) VALUES ('f', 'g', 1, 600)",
+        ] {
+            db.orm.execute_unprepared(sql).await.unwrap();
+        }
+        let todo = db.deadair_todo().await.unwrap();
+        assert_eq!(todo.iter().map(|t| t.id).collect::<Vec<_>>(), [1], "only a file on disk");
+
+        db.store_deadair(1, Some("/x/a.deadair.mp3"), "[[1000,3000]]", 2000).await.unwrap();
+        assert!(db.deadair_todo().await.unwrap().is_empty());
+        let c = db.copies(&[1, 2]).await.unwrap();
+        assert_eq!((c.len(), c[0].cuts.as_slice(), c[0].saved_ms), (1, &[crate::deadair::Cut(1000, 3000)][..], 2000));
+        // Downloaded again since: the copy is stale until made again.
+        db.orm.execute_unprepared(&format!("UPDATE enclosures SET downloaded_at = {} WHERE id = 1", now() + 60)).await.unwrap();
+        assert_eq!(db.deadair_todo().await.unwrap().len(), 1);
+        assert!(db.copies(&[1]).await.unwrap().is_empty());
+
+        assert!(!db.subscribed_any(&[u], "f").await.unwrap());
+        db.subscribe(u, "f").await.unwrap();
+        assert!(db.subscribed_any(&[u], "f").await.unwrap());
+        assert!(db.readers("f", "g").await.unwrap().is_empty());
+        db.set_entry_flag(u, "f", "g", EntryFlag::Read, true).await.unwrap();
+        assert_eq!(db.readers("f", "g").await.unwrap(), [u]);
+
+        assert_eq!(db.entry_duration(u, "f", "g").await.unwrap(), Some(600));
+        db.set_position(u, "f", "g", 10, Some(590)).await.unwrap();
+        assert_eq!(db.entry_duration(u, "f", "g").await.unwrap(), Some(590), "what the player measured");
+
+        assert_eq!(db.deadair_total(u).await.unwrap(), 0);
+        db.count_deadair(u, "f", "g", 2000).await.unwrap();
+        db.count_deadair(u, "f", "g", 2000).await.unwrap();
+        assert_eq!(db.deadair_total(u).await.unwrap(), 2000, "a replay counts once");
+    }
+
     #[tokio::test]
     async fn a_read_does_not_wait_for_a_write_in_progress() {
         use sea_orm::TransactionTrait;