Signals: feeds made from your subscriptions, picked by Jev (#254)

A Signal is a name and keywords. A task beside DeadAir's asks TypeSafe's Jev, on the owner's own
key, about each item of their subscribed feeds from the last 30 days and every new one: one call
an item, one noul question for each of their Signals still to hear about it, 0.5 or more a fit.
Every verdict is kept, fitting or not, in signal_verdicts, so nothing is asked twice, and new
keywords drop a Signal's verdicts to be judged afresh. A key TypeSafe refuses (401/402/403) is
passed over until its owner saves another, so a bad key is not sent the whole queue.

A Signal's items come through entries_where with an EXISTS on its hits, so filters, search, sort
and read state are an ordinary list's, and reading an item in a Signal reads it in its feed.

The page shows Signals under the places, with New signal, and Settings takes the TypeSafe key,
which is never sent back. The browser suite stubs Jev in the fixture server (IPX_TYPESAFE_URL).

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

crates/feed/src/lib.rs

@@ -1140,6 +1140,9 @@ pub fn category_path(name: &str) -> (String, Option<String>) {
     (name.to_string(), None)
 }
 
+/// TypeSafe's System One endpoint, where Jev answers questions about a piece of state.
+pub const JEV_URL: &str = "https://api.typesafe.ai/v1/systemone";
+
 /// A Directory category for a feed that names none of its own, which most blogs do not (#117):
 /// TypeSafe's Jev picks one of Apple's categories or subcategories from the feed's title and
 /// latest items, the subcategory being what a podcast's own category is stored as. The likeliest
@@ -1164,7 +1167,7 @@ pub async fn classify(client: &reqwest::Client, key: &str, feed: &ParsedFeed) ->
         }},
     });
     let answer: serde_json::Value = client
-        .post("https://api.typesafe.ai/v1/systemone")
+        .post(JEV_URL)
         .bearer_auth(key)
         .json(&body)
         .timeout(FEED_TIMEOUT)

crates/ipx/CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Signals: a feed you make from your own subscriptions, with keywords, holding the items TypeSafe's Jev says are about them, from the last 30 days and every new one, judged on your own TypeSafe key (`/api/signals`, and `typesafe_key` on `/api/me`) (#254).
+
 ### Changed
 
 - A feed that sends the same body again, as Patreon and Substack do whatever they are asked, is taken as unchanged instead of read again, and, without a schedule of its own, checked less often the longer it stays the same, up to every four hours, back on schedule when it changes (#242).

crates/ipx/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, deadair_prefs, enclosures, entries, feeds, hidden, sessions, settings, subscriptions, users};
+use crate::entity::{api_tokens, blocklists, catalogue, deadair_prefs, enclosures, entries, feeds, hidden, sessions, settings, signal_verdicts, signals, subscriptions, users};
 use sea_orm::sea_query::{Expr, Func};
 use sea_orm::{
     ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set,
@@ -63,6 +63,8 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
         schema.create_table_from_entity(deadair_plans::Entity),
         schema.create_table_from_entity(deadair_prefs::Entity),
         schema.create_table_from_entity(deadair_saved::Entity),
+        schema.create_table_from_entity(signals::Entity),
+        schema.create_table_from_entity(signal_verdicts::Entity),
     ] {
         orm.execute(table.if_not_exists()).await.context("creating the schema")?;
     }
@@ -100,6 +102,10 @@ async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
             orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
         }
     }
+    if orm.execute_unprepared("SELECT typesafe_key FROM users LIMIT 0").await.is_err() {
+        let sql = "ALTER TABLE users ADD COLUMN typesafe_key TEXT";
+        orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
+    }
     if orm.execute_unprepared("SELECT pause_ms FROM deadair_prefs LIMIT 0").await.is_err() {
         let sql = "ALTER TABLE deadair_prefs ADD COLUMN pause_ms BIGINT";
         orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
@@ -1125,12 +1131,19 @@ fn entries_from(
     filter: Filter,
     search: Option<&str>,
     guid: Option<&str>,
+    signal: Option<i64>,
 ) -> String {
     let user = a.p(user_id);
     let mut scope = match feed_id {
         Some(f) => format!("e.feed_id = {}", a.p(f)),
         None => format!("e.feed_id IN (SELECT feed_id FROM subscriptions WHERE user_id = {user})"),
     };
+    if let Some(id) = signal {
+        scope = format!(
+            "{scope} AND EXISTS (SELECT 1 FROM signal_verdicts v WHERE v.signal_id = {} AND v.feed_id = e.feed_id AND v.guid = e.guid AND v.hit)",
+            a.p(id)
+        );
+    }
     if let Some(g) = guid {
         scope = format!("{scope} AND e.guid = {}", a.p(g));
     }
@@ -1274,14 +1287,37 @@ impl Db {
         limit: i64,
         order: &str,
     ) -> Result<Vec<EntryRow>> {
-        self.entries_where(user_id, feed_id, filter, search, None, offset, limit, order).await
+        self.entries_where(user_id, feed_id, filter, search, None, None, offset, limit, order).await
+    }
+
+    /// A Signal's page of entries, as `entries_in` gives All Subscriptions'. The caller has made
+    /// sure the Signal is this person's.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn signal_entries(
+        &self,
+        user_id: i64,
+        signal: i64,
+        filter: Filter,
+        search: Option<&str>,
+        offset: i64,
+        limit: i64,
+        order: &str,
+    ) -> Result<Vec<EntryRow>> {
+        self.entries_where(user_id, None, filter, search, None, Some(signal), offset, limit, order).await
+    }
+
+    pub async fn signal_count(&self, user_id: i64, signal: i64, filter: Filter, search: Option<&str>) -> Result<i64> {
+        let mut a = Args::default();
+        let sql = format!("SELECT count(*) AS n {}", entries_from(&mut a, user_id, None, filter, search, None, Some(signal)));
+        let r = self.rows(&sql, a.0).await?;
+        Ok(r.first().context("count(*) always returns a row")?.try_get("", "n")?)
     }
 
     /// One entry as `entries_in` gives it: what the player asks for again when the clock it
     /// plays in changes (DeadAir, #171).
     pub async fn entry(&self, user_id: i64, feed_id: &str, guid: &str) -> Result<Option<EntryRow>> {
         let order = order_sql("published", "desc");
-        Ok(self.entries_where(user_id, Some(feed_id), Filter::All, None, Some(guid), 0, 1, &order).await?.pop())
+        Ok(self.entries_where(user_id, Some(feed_id), Filter::All, None, Some(guid), None, 0, 1, &order).await?.pop())
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -1292,12 +1328,13 @@ impl Db {
         filter: Filter,
         search: Option<&str>,
         guid: Option<&str>,
+        signal: Option<i64>,
         offset: i64,
         limit: i64,
         order: &str,
     ) -> Result<Vec<EntryRow>> {
         let mut a = Args::default();
-        let from = entries_from(&mut a, user_id, feed_id, filter, search, guid);
+        let from = entries_from(&mut a, user_id, feed_id, filter, search, guid, signal);
         // The page's keys first, then its rows: the sort then needs only `entries_by_time`, not
         // every item's row, and the show notes are read for the page alone (#188).
         let (limit, offset) = (a.p(limit), a.p(offset));
@@ -1363,7 +1400,7 @@ impl Db {
         search: Option<&str>,
     ) -> Result<i64> {
         let mut a = Args::default();
-        let sql = format!("SELECT count(*) AS n {}", entries_from(&mut a, user_id, feed_id, filter, search, None));
+        let sql = format!("SELECT count(*) AS n {}", entries_from(&mut a, user_id, feed_id, filter, search, None, None));
         let r = self.rows(&sql, a.0).await?;
         Ok(r.first().context("count(*) always returns a row")?.try_get("", "n")?)
     }
@@ -2496,6 +2533,191 @@ impl Db {
 }
 
 /// Unix seconds. Everything time-shaped in the DB is stored this way.
+/// A Signal as its owner's sidebar lists it (#254).
+#[derive(Debug, serde::Serialize)]
+pub struct SignalRow {
+    pub id: i64,
+    pub name: String,
+    pub keywords: Vec<String>,
+    pub unread: i64,
+}
+
+/// An item still to be put to Jev for someone, with each of their Signals that has not been
+/// asked about it.
+#[derive(Debug)]
+pub struct Unjudged {
+    pub user_id: i64,
+    pub key: String,
+    pub feed_id: String,
+    pub guid: String,
+    pub feed_title: Option<String>,
+    pub title: Option<String>,
+    pub description: Option<String>,
+    pub signals: Vec<(i64, Vec<String>)>,
+}
+
+/// How far back a Signal looks when it is made: a month of items, not every back catalogue.
+pub const SIGNAL_DAYS: i64 = 30;
+
+impl Db {
+    pub async fn signals(&self, user_id: i64) -> Result<Vec<SignalRow>> {
+        self.signal_rows(user_id, None).await
+    }
+
+    /// One of this person's Signals: None for anyone else's, as for one that does not exist.
+    pub async fn signal(&self, user_id: i64, id: i64) -> Result<Option<SignalRow>> {
+        Ok(self.signal_rows(user_id, Some(id)).await?.pop())
+    }
+
+    async fn signal_rows(&self, user_id: i64, id: Option<i64>) -> Result<Vec<SignalRow>> {
+        let mut a = Args::default();
+        let user = a.p(user_id);
+        let one = id.map(|i| format!("AND g.id = {}", a.p(i))).unwrap_or_default();
+        let sql = format!(
+            "SELECT g.id, g.name, g.keywords,
+                    (SELECT count(*) FROM signal_verdicts v
+                       JOIN entries e ON e.feed_id = v.feed_id AND e.guid = v.guid
+                       JOIN subscriptions s ON s.user_id = g.user_id AND s.feed_id = v.feed_id
+                       LEFT JOIN entry_state st ON st.user_id = g.user_id AND st.feed_id = v.feed_id AND st.guid = v.guid
+                      WHERE v.signal_id = g.id AND v.hit AND NOT coalesce(st.read, false) AND {}) AS unread
+               FROM signals g WHERE g.user_id = {user} {one} ORDER BY lower(g.name), g.id",
+            not_hidden(&user)
+        );
+        self.rows(&sql, a.0)
+            .await?
+            .iter()
+            .map(|r| {
+                Ok(SignalRow {
+                    id: r.try_get("", "id")?,
+                    name: r.try_get("", "name")?,
+                    keywords: keywords(r.try_get("", "keywords")?).unwrap_or_default(),
+                    unread: r.try_get("", "unread")?,
+                })
+            })
+            .collect()
+    }
+
+    pub async fn add_signal(&self, user_id: i64, name: &str, words: &[String]) -> Result<i64> {
+        let m = signals::ActiveModel {
+            user_id: Set(user_id),
+            name: Set(name.to_owned()),
+            keywords: Set(serde_json::to_string(words)?),
+            created: Set(now()),
+            ..Default::default()
+        }
+        .insert(&self.orm)
+        .await?;
+        Ok(m.id)
+    }
+
+    /// Renames a Signal or gives it new keywords. New keywords put every item to Jev again, since
+    /// what it said was about the old ones. False for a Signal that is not this person's.
+    pub async fn set_signal(&self, user_id: i64, id: i64, name: Option<&str>, words: Option<&[String]>) -> Result<bool> {
+        if self.signal(user_id, id).await?.is_none() {
+            return Ok(false);
+        }
+        let mut m = signals::ActiveModel { id: Set(id), ..Default::default() };
+        if let Some(n) = name {
+            m.name = Set(n.to_owned());
+        }
+        if let Some(w) = words {
+            m.keywords = Set(serde_json::to_string(w)?);
+            signal_verdicts::Entity::delete_many().filter(signal_verdicts::Column::SignalId.eq(id)).exec(&self.orm).await?;
+        }
+        if name.is_some() || words.is_some() {
+            m.update(&self.orm).await?;
+        }
+        Ok(true)
+    }
+
+    pub async fn remove_signal(&self, user_id: i64, id: i64) -> Result<bool> {
+        if self.signal(user_id, id).await?.is_none() {
+            return Ok(false);
+        }
+        // Not left to the cascade: SQLite enforces foreign keys only when asked to.
+        signal_verdicts::Entity::delete_many().filter(signal_verdicts::Column::SignalId.eq(id)).exec(&self.orm).await?;
+        signals::Entity::delete_by_id(id).exec(&self.orm).await?;
+        Ok(true)
+    }
+
+    pub async fn typesafe_key_set(&self, user_id: i64) -> Result<bool> {
+        Ok(users::Entity::find_by_id(user_id).one(&self.orm).await?.is_some_and(|u| u.typesafe_key.is_some()))
+    }
+
+    pub async fn set_typesafe_key(&self, user_id: i64, key: Option<&str>) -> Result<()> {
+        self.update_user(user_id, users::ActiveModel { typesafe_key: Set(key.map(str::to_owned)), ..Default::default() })
+            .await
+    }
+
+    /// Up to `limit` items, newest first, that someone with a key has a Signal not yet asked
+    /// about: from the feeds they subscribe to, the last `SIGNAL_DAYS`, and not hidden from them.
+    /// `skip` are people whose key TypeSafe has refused.
+    pub async fn unjudged(&self, skip: &[i64], limit: i64) -> Result<Vec<Unjudged>> {
+        let mut a = Args::default();
+        let since = a.p(now() - SIGNAL_DAYS * 86_400);
+        let skip = if skip.is_empty() {
+            String::new()
+        } else {
+            format!("AND g.user_id NOT IN ({})", skip.iter().map(|u| a.p(*u)).collect::<Vec<_>>().join(", "))
+        };
+        let limit = a.p(limit);
+        let sql = format!(
+            "SELECT g.id AS signal_id, g.keywords, g.user_id, u.typesafe_key, e.feed_id, e.guid, e.title,
+                    e.description, f.title AS feed_title
+               FROM signals g
+               JOIN users u ON u.id = g.user_id
+               JOIN subscriptions s ON s.user_id = g.user_id
+               JOIN entries e ON e.feed_id = s.feed_id
+               LEFT JOIN feeds f ON f.id = e.feed_id
+              WHERE u.typesafe_key IS NOT NULL {skip}
+                AND coalesce(e.published, e.first_seen) >= {since}
+                AND NOT EXISTS (SELECT 1 FROM signal_verdicts v
+                                 WHERE v.signal_id = g.id AND v.feed_id = e.feed_id AND v.guid = e.guid)
+                AND NOT EXISTS (SELECT 1 FROM hidden h
+                                 WHERE h.user_id = g.user_id AND h.feed_id = e.feed_id AND h.guid = e.guid)
+              ORDER BY coalesce(e.published, e.first_seen) DESC, e.feed_id, e.guid, g.user_id
+              LIMIT {limit}"
+        );
+        let mut out: Vec<Unjudged> = Vec::new();
+        for r in self.rows(&sql, a.0).await? {
+            let (user_id, feed_id, guid): (i64, String, String) =
+                (r.try_get("", "user_id")?, r.try_get("", "feed_id")?, r.try_get("", "guid")?);
+            let signal = (r.try_get("", "signal_id")?, keywords(r.try_get("", "keywords")?).unwrap_or_default());
+            // Rows of one item for one person come together, by the ORDER BY.
+            match out.last_mut() {
+                Some(u) if (u.user_id, &u.feed_id, &u.guid) == (user_id, &feed_id, &guid) => u.signals.push(signal),
+                _ => out.push(Unjudged {
+                    user_id,
+                    key: r.try_get("", "typesafe_key")?,
+                    feed_id,
+                    guid,
+                    feed_title: r.try_get("", "feed_title")?,
+                    title: r.try_get("", "title")?,
+                    description: r.try_get("", "description")?,
+                    signals: vec![signal],
+                }),
+            }
+        }
+        Ok(out)
+    }
+
+    pub async fn record_verdicts(&self, feed_id: &str, guid: &str, verdicts: &[(i64, bool)]) -> Result<()> {
+        for (signal, hit) in verdicts {
+            // Not for a Signal removed while Jev was being asked, which Postgres's foreign key
+            // would refuse and SQLite would keep for nothing.
+            // shortcut: keywords changed mid-call keep a verdict on the old ones for that item.
+            self.exec(
+                "INSERT INTO signal_verdicts (signal_id, feed_id, guid, hit)
+                 SELECT $1, $2, $3, $4 WHERE EXISTS (SELECT 1 FROM signals WHERE id = $1)
+                 ON CONFLICT (signal_id, feed_id, guid) DO UPDATE SET hit = excluded.hit",
+                vec![(*signal).into(), feed_id.into(), guid.into(), (*hit).into()],
+            )
+            .await?;
+        }
+        Ok(())
+    }
+}
+
 pub fn now() -> i64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -3093,6 +3315,59 @@ mod tests {
         assert_eq!(listening().await, ["h", "e"]);
     }
 
+    #[tokio::test]
+    async fn a_signal_shows_what_jev_said_fits_and_only_to_its_owner() {
+        let db = Db::memory().await.unwrap();
+        let t = now();
+        db.exec_for_test(&format!(
+            "INSERT INTO users (id, name, is_admin) VALUES (1,'a',true), (2,'b',false);
+             INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'), (1,'g'), (2,'f');
+             INSERT INTO entries (feed_id, guid, title, published, first_seen) VALUES
+                ('f','x','Monaco qualifying',{t},{t}),
+                ('g','y','Bread',{t},{t}),
+                ('g','old','Silverstone 1998',100,100),
+                ('h','z','Not subscribed',{t},{t});"
+        ))
+        .await
+        .unwrap();
+        let words = |w: &[&str]| w.iter().map(|s| s.to_string()).collect::<Vec<_>>();
+        let f1 = db.add_signal(1, "F1", &words(&["formula 1"])).await.unwrap();
+        let food = db.add_signal(1, "Food", &words(&["baking"])).await.unwrap();
+        db.add_signal(2, "Theirs", &words(&["x"])).await.unwrap();
+
+        assert!(db.unjudged(&[], 50).await.unwrap().is_empty(), "nobody has a key yet");
+        db.set_typesafe_key(1, Some("k")).await.unwrap();
+        let queue = db.unjudged(&[], 50).await.unwrap();
+        let mut seen: Vec<_> = queue.iter().map(|u| (u.guid.as_str(), u.signals.len())).collect();
+        seen.sort();
+        assert_eq!(seen, [("x", 2), ("y", 2)], "each recent subscribed item once, with both Signals; old and unsubscribed left out");
+        assert!(db.unjudged(&[1], 50).await.unwrap().is_empty(), "a refused key is passed over");
+
+        db.record_verdicts("f", "x", &[(f1, true), (food, false)]).await.unwrap();
+        db.record_verdicts("g", "y", &[(f1, false), (food, true)]).await.unwrap();
+        assert!(db.unjudged(&[], 50).await.unwrap().is_empty(), "nothing is asked twice");
+
+        let order = order_sql("published", "desc");
+        let page = db.signal_entries(1, f1, Filter::All, None, 0, 50, &order).await.unwrap();
+        assert_eq!(page.iter().map(|e| e.guid.as_str()).collect::<Vec<_>>(), ["x"]);
+        assert_eq!(db.signal_count(1, food, Filter::All, None).await.unwrap(), 1);
+        assert_eq!(db.signals(1).await.unwrap().iter().map(|s| s.unread).collect::<Vec<_>>(), [1, 1]);
+        db.exec_for_test("INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES (1,'f','x',true,false,0)")
+            .await
+            .unwrap();
+        assert_eq!(db.signal(1, f1).await.unwrap().unwrap().unread, 0, "read in its feed is read in the Signal");
+
+        assert!(db.signal(2, f1).await.unwrap().is_none(), "someone else's Signal is not found");
+        assert!(!db.set_signal(2, f1, Some("mine"), None).await.unwrap());
+        assert!(!db.remove_signal(2, f1).await.unwrap());
+
+        db.set_signal(1, f1, None, Some(&words(&["f1"]))).await.unwrap();
+        assert_eq!(db.signal_count(1, f1, Filter::All, None).await.unwrap(), 0, "new keywords are judged afresh");
+        assert!(db.remove_signal(1, food).await.unwrap());
+        db.record_verdicts("f", "x", &[(food, true)]).await.unwrap();
+        assert_eq!(db.signals(1).await.unwrap().len(), 1, "a verdict for a removed Signal is not kept");
+    }
+
     #[tokio::test]
     async fn block_lists_hide_items_from_whoever_keeps_them() {
         let db = Db::memory().await.unwrap();

crates/ipx/src/entity.rs

@@ -157,6 +157,12 @@ pub mod users {
         pub theme: Option<String>,
         #[sea_orm(column_type = "Text", nullable)]
         pub theme_mode: Option<String>,
+        /// Their own TypeSafe key, which Jev judges their Signals' items on (#254). Never sent
+        /// back to the page.
+        // shortcut: kept as it is, since every call needs it whole; encrypt it with a key of the
+        // server's if the database is ever shared with anything else.
+        #[sea_orm(column_type = "Text", nullable)]
+        pub typesafe_key: Option<String>,
     }
 
     #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
@@ -282,6 +288,59 @@ pub mod hidden {
     owned_by_user!();
 }
 
+/// A feed someone makes from their own subscriptions (#254): the items Jev says fit its
+/// keywords, whichever feed they came from.
+pub mod signals {
+    use sea_orm::entity::prelude::*;
+
+    #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+    #[sea_orm(table_name = "signals")]
+    pub struct Model {
+        #[sea_orm(primary_key)]
+        pub id: i64,
+        pub user_id: i64,
+        #[sea_orm(column_type = "Text")]
+        pub name: String,
+        /// JSON array of strings.
+        #[sea_orm(column_type = "Text")]
+        pub keywords: String,
+        pub created: i64,
+    }
+
+    owned_by_user!();
+}
+
+/// What Jev said of each item a Signal has been asked about, fitting or not: an item with no row
+/// here is still to be asked.
+pub mod signal_verdicts {
+    use sea_orm::entity::prelude::*;
+
+    #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel)]
+    #[sea_orm(table_name = "signal_verdicts")]
+    pub struct Model {
+        #[sea_orm(primary_key, auto_increment = false)]
+        pub signal_id: i64,
+        #[sea_orm(primary_key, auto_increment = false, column_type = "Text")]
+        pub feed_id: String,
+        #[sea_orm(primary_key, auto_increment = false, column_type = "Text")]
+        pub guid: String,
+        pub hit: bool,
+    }
+
+    #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+    pub enum Relation {
+        #[sea_orm(
+            belongs_to = "super::signals::Entity",
+            from = "Column::SignalId",
+            to = "super::signals::Column::Id",
+            on_delete = "Cascade"
+        )]
+        Signal,
+    }
+
+    impl ActiveModelBehavior for ActiveModel {}
+}
+
 pub mod sessions {
     use sea_orm::entity::prelude::*;
 

crates/ipx/src/main.rs

@@ -13,6 +13,7 @@ mod ipc;
 mod logbuf;
 mod metrics;
 mod retention;
+mod signals;
 mod torrent;
 mod web;
 
@@ -159,6 +160,10 @@ pub struct Ctx {
     /// Wakes DeadAir's maker when a show is turned on, so its copies start at once instead of
     /// at the next pass (#171).
     pub deadair: tokio::sync::Notify,
+    /// Wakes the Signals judge when there are new items or a Signal is made or changed (#254).
+    pub signals: tokio::sync::Notify,
+    /// People whose TypeSafe key was refused, passed over until they save another.
+    pub signals_refused: std::sync::Mutex<std::collections::HashSet<i64>>,
     /// When the Directory was last cleared of dead feeds (`clean_directory`), 0 for not since
     /// starting.
     pub directory_cleaned: std::sync::atomic::AtomicI64,
@@ -367,6 +372,8 @@ async fn main() -> Result<()> {
         cfg_write: Default::default(),
         directory_cleaned: Default::default(),
         deadair: Default::default(),
+        signals: Default::default(),
+        signals_refused: Default::default(),
         directory_wake: Default::default(),
         directory_shared: Default::default(),
         directory_resend: Default::default(),
@@ -663,6 +670,7 @@ async fn daemon(
     // Off the worker: the first pass over a library is hours of ffmpeg, and scans and downloads
     // must not wait behind it.
     tokio::spawn(deadair::run(ctx.clone(), rx_stop.clone()));
+    tokio::spawn(signals::run(ctx.clone(), rx_stop.clone()));
     tokio::spawn(directory::run(ctx.clone(), rx_stop.clone()));
     // The library's counts for /metrics. The Docker healthcheck asked for them every 30 s until
     // it was removed: Docker runs it with exec, which failed on Tower whenever root was logged
@@ -1893,6 +1901,7 @@ async fn scan_one(
     }, store).await?;
     if scan.new_entries > 0 {
         ctx.db.rehide(id).await?;   // what is new may hold someone's blocked words
+        ctx.signals.notify_one();
     }
     // A new item takes a place among the newest and the oldest of them leaves the queue; a file a
     // filter lets through again may be outside them.
@@ -2605,6 +2614,8 @@ mod tests {
             retry: Default::default(),
             cfg_write: Default::default(),
             deadair: Default::default(),
+            signals: Default::default(),
+            signals_refused: Default::default(),
             directory_wake: Default::default(),
             directory_shared: Default::default(),
             directory_resend: Default::default(),

crates/ipx/src/signals.rs

@@ -0,0 +1,180 @@
+//! Signals (#254): feeds someone makes from their own subscriptions, holding the items TypeSafe's
+//! Jev says fit the keywords they gave. Jev is asked on their own key, once for each item, with a
+//! question for each of their Signals that has not had it yet, and what it says is kept in
+//! `signal_verdicts`, fitting or not, so nothing is asked twice.
+
+use std::collections::HashSet;
+use std::sync::Arc;
+
+use anyhow::Result;
+use tracing::Instrument;
+
+use crate::Ctx;
+use crate::db::Unjudged;
+use crate::feed;
+
+/// Items put to Jev a pass. A pass ends early once there are none left, so this only says how
+/// long the queue is held before people who got a key later are looked at.
+const BATCH: i64 = 50;
+/// Between passes when nothing wakes it: a scan or a new Signal does, so this is a backstop.
+const IDLE: std::time::Duration = std::time::Duration::from_secs(300);
+/// How sure Jev has to be that an item is about the keywords.
+const THRESHOLD: f64 = 0.5;
+/// Of an item's text, which is a podcast's whole show notes at times.
+const TEXT_CHARS: usize = 2000;
+
+pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
+    loop {
+        let done = tokio::select! {
+            _ = stop.changed() => return,
+            done = pass(&ctx) => done,
+        };
+        match done {
+            Ok(true) => continue,
+            Ok(false) => {}
+            Err(e) => tracing::warn!(error = %format!("{e:#}"), "Signals could not be judged"),
+        }
+        tokio::select! {
+            _ = stop.changed() => return,
+            _ = ctx.signals.notified() => {}
+            _ = tokio::time::sleep(IDLE) => {}
+        }
+    }
+}
+
+/// One batch. True when there may be more waiting.
+async fn pass(ctx: &Ctx) -> Result<bool> {
+    let skip: Vec<i64> = ctx.signals_refused.lock().unwrap().iter().copied().collect();
+    let items = ctx.db.unjudged(&skip, BATCH).await?;
+    if items.is_empty() {
+        return Ok(false);
+    }
+    let span = tracing::info_span!("signals", entries = items.len());
+    async {
+        let mut refused = HashSet::new();
+        for item in &items {
+            if refused.contains(&item.user_id) {
+                continue;
+            }
+            match judge(ctx, item).await {
+                Ok(verdicts) => {
+                    let hits = verdicts.iter().filter(|(_, h)| *h).count();
+                    if hits > 0 {
+                        tracing::info!(ev = "signal", user = item.user_id, feed = %item.feed_id, new = hits, "an item fits a Signal");
+                    }
+                    ctx.db.record_verdicts(&item.feed_id, &item.guid, &verdicts).await?;
+                }
+                // A key TypeSafe will not take is not sent the rest of the queue. It is tried
+                // again once its owner saves one (`web::patch_me`) or the daemon restarts.
+                Err(e) if refuses_key(&e) => {
+                    tracing::warn!(user = item.user_id, error = %format!("{e:#}"), "TypeSafe refused a Signal key");
+                    refused.insert(item.user_id);
+                }
+                // Left unjudged, so it is asked again next pass.
+                Err(e) => tracing::warn!(user = item.user_id, feed = %item.feed_id, error = %format!("{e:#}"), "Jev could not judge an item"),
+            }
+        }
+        let some_refused = !refused.is_empty();
+        ctx.signals_refused.lock().unwrap().extend(refused);
+        // A pass that only failed would otherwise go straight round again on the same items.
+        anyhow::Ok(items.len() as i64 == BATCH || some_refused)
+    }
+    .instrument(span)
+    .await
+}
+
+fn refuses_key(e: &anyhow::Error) -> bool {
+    e.downcast_ref::<reqwest::Error>()
+        .and_then(|e| e.status())
+        .is_some_and(|s| matches!(s.as_u16(), 401 | 402 | 403))
+}
+
+async fn judge(ctx: &Ctx, item: &Unjudged) -> Result<Vec<(i64, bool)>> {
+    let url = std::env::var("IPX_TYPESAFE_URL").unwrap_or_else(|_| feed::JEV_URL.to_owned());
+    let answer: serde_json::Value = ctx
+        .client
+        .post(url)
+        .bearer_auth(&item.key)
+        .json(&request(item))
+        .timeout(feed::FEED_TIMEOUT)
+        .send()
+        .instrument(tracing::info_span!("jev", user = item.user_id, signal_count = item.signals.len()))
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    // A question Jev left unanswered stays unjudged and is asked again.
+    Ok(item.signals.iter().filter_map(|(id, _)| hit(&answer, *id).map(|h| (*id, h))).collect())
+}
+
+/// The item as Jev sees it, and one yes-or-no question for each Signal, keyed by its id.
+fn request(item: &Unjudged) -> serde_json::Value {
+    let text: String = text_of(item.description.as_deref().unwrap_or("")).chars().take(TEXT_CHARS).collect();
+    let questions: serde_json::Map<String, serde_json::Value> =
+        item.signals.iter().map(|(id, words)| (id.to_string(), question(words))).collect();
+    serde_json::json!({
+        "model": "jev-latest",
+        "state": { "feed": item.feed_title, "title": item.title, "text": text },
+        "questions": questions,
+    })
+}
+
+fn question(words: &[String]) -> serde_json::Value {
+    serde_json::json!({
+        "type": "noul",
+        "instructions": format!("Is this item about any of these: {}?", words.join(", ")),
+        "criteria": "Yes if the item's subject is one of them, or plainly part of one. No if one is only mentioned in passing.",
+    })
+}
+
+/// Whether Jev says the item fits Signal `id`; None when the answer does not say.
+fn hit(answer: &serde_json::Value, id: i64) -> Option<bool> {
+    answer["answers"][id.to_string()]["noul"].as_f64().map(|p| p >= THRESHOLD)
+}
+
+/// Show notes are HTML: Jev is given their words.
+fn text_of(html: &str) -> String {
+    let mut out = String::with_capacity(html.len());
+    let mut in_tag = false;
+    for c in html.chars() {
+        match c {
+            '<' => in_tag = true,
+            '>' if in_tag => { in_tag = false; out.push(' '); }
+            _ if !in_tag => out.push(c),
+            _ => {}
+        }
+    }
+    out.split_whitespace().collect::<Vec<_>>().join(" ")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_noul_of_a_half_or_more_is_a_fit() {
+        let answer = serde_json::json!({ "answers": { "3": { "noul": 0.5 }, "4": { "noul": 0.49 } } });
+        assert_eq!(hit(&answer, 3), Some(true));
+        assert_eq!(hit(&answer, 4), Some(false));
+        assert_eq!(hit(&answer, 5), None, "a question not answered is asked again");
+    }
+
+    #[test]
+    fn each_signal_is_its_own_question_on_the_items_words() {
+        let item = Unjudged {
+            user_id: 1,
+            key: "k".into(),
+            feed_id: "f".into(),
+            guid: "g".into(),
+            feed_title: Some("Pit Lane".into()),
+            title: Some("Monaco".into()),
+            description: Some("<p>Qualifying <b>report</b></p>".into()),
+            signals: vec![(3, vec!["formula 1".into(), "f1".into()]), (4, vec!["cooking".into()])],
+        };
+        let r = request(&item);
+        assert_eq!(r["state"]["text"], "Qualifying report");
+        assert_eq!(r["questions"]["3"]["type"], "noul");
+        assert!(r["questions"]["3"]["instructions"].as_str().unwrap().contains("formula 1, f1"));
+        assert!(r["questions"]["4"]["instructions"].as_str().unwrap().contains("cooking"));
+    }
+}

crates/ipx/src/web.rs

@@ -47,6 +47,9 @@ pub fn router(state: WebState) -> Router {
         .route("/api/feeds/{id}", patch(patch_feed).delete(remove_feed))
         .route("/api/feeds/{id}/entries", get(entries))
         .route("/api/entries", get(all_entries))
+        .route("/api/signals", get(list_signals).post(add_signal))
+        .route("/api/signals/{id}", patch(patch_signal).delete(remove_signal))
+        .route("/api/signals/{id}/entries", get(signal_entries))
         .route("/api/entries/{feed_id}/{guid}", get(one_entry))
         .route("/api/entries/{feed_id}/{guid}/flags", post(set_flags))
         .route("/api/entries/{feed_id}/{guid}/position", post(set_position))
@@ -382,6 +385,7 @@ async fn me(
     Json(serde_json::json!({
         "name": user.name, "admin": user.is_admin, "sign_out": sign_out, "blocked": blocked,
         "deadair_saved": saved, "theme": theme, "version": env!("CARGO_PKG_VERSION"),
+        "typesafe_key_set": state.ctx.db.typesafe_key_set(user.id).await.unwrap_or(false),
     }))
 }
 
@@ -389,6 +393,9 @@ async fn me(
 struct MePatch {
     /// Words that hide an item in every feed you read.
     blocked: Option<Vec<String>>,
+    /// The TypeSafe key Signals are judged on: a key saves it, null or empty forgets it.
+    #[serde(default, deserialize_with = "double_option")]
+    typesafe_key: Option<Option<String>>,
 }
 
 /// Saves the block list for every feed. The theme is not the account's: each browser keeps its
@@ -401,9 +408,102 @@ async fn patch_me(
     if let Some(words) = body.blocked {
         state.ctx.db.set_blocklist(user.id, "", &clean_words(words)?).await?;
     }
+    if let Some(key) = body.typesafe_key {
+        let key = key.map(|k| k.trim().to_owned()).filter(|k| !k.is_empty());
+        if key.as_ref().is_some_and(|k| k.len() > 500 || !k.bytes().all(|b| b.is_ascii_graphic())) {
+            return Err(ApiError::bad_request("that does not look like a TypeSafe key"));
+        }
+        state.ctx.db.set_typesafe_key(user.id, key.as_deref()).await?;
+        state.ctx.signals_refused.lock().unwrap().remove(&user.id);
+        state.ctx.signals.notify_one();
+    }
     Ok(StatusCode::NO_CONTENT)
 }
 
+// ---- Signals (#254): each person's own, so each route finds the Signal by its owner ----
+
+#[derive(Deserialize)]
+struct SignalBody {
+    name: Option<String>,
+    keywords: Option<Vec<String>>,
+}
+
+fn signal_name(name: &str) -> Result<String, ApiError> {
+    let name = name.trim();
+    if name.is_empty() || name.chars().count() > 100 {
+        return Err(ApiError::bad_request("a Signal needs a name of up to 100 characters"));
+    }
+    Ok(name.to_owned())
+}
+
+fn signal_words(words: Vec<String>) -> Result<Vec<String>, ApiError> {
+    let words = clean_words(words)?;
+    // Each is in the question Jev is asked about every item.
+    if words.is_empty() || words.len() > 50 {
+        return Err(ApiError::bad_request("a Signal needs between 1 and 50 keywords"));
+    }
+    Ok(words)
+}
+
+async fn list_signals(
+    State(state): State<WebState>,
+    user: crate::db::User,
+) -> Result<Json<Vec<crate::db::SignalRow>>, ApiError> {
+    Ok(Json(state.ctx.db.signals(user.id).await?))
+}
+
+async fn add_signal(
+    State(state): State<WebState>,
+    user: crate::db::User,
+    Json(body): Json<SignalBody>,
+) -> Result<Json<serde_json::Value>, ApiError> {
+    let name = signal_name(body.name.as_deref().unwrap_or(""))?;
+    let words = signal_words(body.keywords.unwrap_or_default())?;
+    let id = state.ctx.db.add_signal(user.id, &name, &words).await?;
+    state.ctx.signals.notify_one();
+    Ok(Json(serde_json::json!({ "id": id })))
+}
+
+async fn patch_signal(
+    State(state): State<WebState>,
+    Path(id): Path<i64>,
+    user: crate::db::User,
+    Json(body): Json<SignalBody>,
+) -> Result<StatusCode, ApiError> {
+    let name = body.name.as_deref().map(signal_name).transpose()?;
+    let words = body.keywords.map(signal_words).transpose()?;
+    if !state.ctx.db.set_signal(user.id, id, name.as_deref(), words.as_deref()).await? {
+        return Err(ApiError::not_found("no such Signal"));
+    }
+    if words.is_some() {
+        state.ctx.signals.notify_one();
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn remove_signal(
+    State(state): State<WebState>,
+    Path(id): Path<i64>,
+    user: crate::db::User,
+) -> Result<StatusCode, ApiError> {
+    if !state.ctx.db.remove_signal(user.id, id).await? {
+        return Err(ApiError::not_found("no such Signal"));
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn signal_entries(
+    State(state): State<WebState>,
+    Path(id): Path<i64>,
+    user: crate::db::User,
+    Query(page): Query<Page>,
+) -> Result<Json<EntryPage>, ApiError> {
+    if state.ctx.db.signal(user.id, id).await?.is_none() {
+        return Err(ApiError::not_found("no such Signal"));
+    }
+    entry_page(&state, user.id, None, Some(id), &page).await
+}
+
 fn theme_ok(theme: &str, mode: &str) -> bool {
     (1..=20).contains(&theme.len())
         && theme.bytes().all(|b| b.is_ascii_lowercase())
@@ -1231,7 +1331,7 @@ async fn entries(
     Query(page): Query<Page>,
 ) -> Result<Json<EntryPage>, ApiError> {
     subscribed(&state, &user, &id).await?;
-    entry_page(&state, user.id, Some(&id), &page).await
+    entry_page(&state, user.id, Some(&id), None, &page).await
 }
 
 /// A feed's items and files are for the people who subscribe to it (#129). Anyone signed in
@@ -1252,24 +1352,26 @@ async fn all_entries(
     user: crate::db::User,
     Query(page): Query<Page>,
 ) -> Result<Json<EntryPage>, ApiError> {
-    entry_page(&state, user.id, None, &page).await
+    entry_page(&state, user.id, None, None, &page).await
 }
 
-/// One feed's page of items, or every subscribed feed's when `feed` is None.
+/// One feed's page of items, or every subscribed feed's when `feed` is None, or a Signal's.
 async fn entry_page(
     state: &WebState,
     user_id: i64,
     feed: Option<&str>,
+    signal: Option<i64>,
     page: &Page,
 ) -> Result<Json<EntryPage>, ApiError> {
     let filter = crate::db::Filter::parse(page.filter.as_deref().unwrap_or("all"));
     let search = page.q.as_deref().map(str::trim).filter(|q| !q.is_empty());
     let db = &state.ctx.db;
     let order = crate::db::order_sql(page.sort.as_deref().unwrap_or("published"), page.dir.as_deref().unwrap_or("desc"));
-    let mut rows = db
+    let limit = page.limit.clamp(1, 200);
-        .entries_in(user_id, feed, filter, search, page.offset, page.limit.clamp(1, 200), &order)
+    let mut rows = match signal {
-        .instrument(step("entries_in"))
+        Some(id) => db.signal_entries(user_id, id, filter, search, page.offset, limit, &order).instrument(step("entries_in")).await?,
-        .await?;
+        None => db.entries_in(user_id, feed, filter, search, page.offset, limit, &order).instrument(step("entries_in")).await?,
+    };
     step("sanitize").in_scope(|| {
         let mut sanitizer = feed_sanitizer();
         for row in &mut rows {
@@ -1279,7 +1381,10 @@ async fn entry_page(
         }
     });
     crate::deadair::for_reader(&state.ctx, user_id, &mut rows).instrument(step("deadair")).await?;
-    let total = db.count_in(user_id, feed, filter, search).instrument(step("count_in")).await?;
+    let total = match signal {
+        Some(id) => db.signal_count(user_id, id, filter, search).instrument(step("count_in")).await?,
+        None => db.count_in(user_id, feed, filter, search).instrument(step("count_in")).await?,
+    };
     Ok(Json(EntryPage { total, entries: rows }))
 }
 

docs/architecture.md

@@ -41,6 +41,7 @@ id, and Caddy's request lines theirs. ipx also serves Prometheus metrics (`[web]
 | `crates/ipx/src/logbuf.rs` | Ring buffer behind the UI's Log view | — |
 | `crates/ipx/src/directory.rs` | Asking the directory service, and sharing this server's public feeds with it | — |
 | `crates/ipx/src/metrics.rs` | `/metrics` for Prometheus | — |
+| `crates/ipx/src/signals.rs` | Signals: asks Jev, on each person's TypeSafe key, which of their items fit their Signals | — |
 | `crates/directory/src/lib.rs` | The directory's HTTP API, its token check, the refresher | — |
 | `crates/directory/src/store.rs` | The directory's database: feeds, snapshots, blocked | — |
 | `crates/directory/src/config.rs` | The directory's config file: its tokens and their access | — |
@@ -96,8 +97,17 @@ subscriptions  user_id, feed_id, keywords, auto_download, allow_explicit,
                max_new_per_check                         PK (user_id, feed_id)
 entry_state    user_id, feed_id, guid, read, flagged, position
                                                          PK (user_id, feed_id, guid)
+signals        id, user_id, name, keywords
+signal_verdicts  signal_id, feed_id, guid, hit           PK (signal_id, feed_id, guid)
 ```
 
+A Signal (#254) is a feed someone makes from their own subscriptions. `signals::run`, a task of its
+own beside DeadAir's, woken by a scan with something new or a Signal made or changed, asks Jev
+about each item of the last 30 days that has no verdict yet for one of its owner's Signals: one
+call an item, a noul question a Signal, on the owner's `users.typesafe_key`. Every answer is kept,
+fitting or not, so nothing is asked twice; new keywords drop the Signal's verdicts. A Signal's
+items are `entries_where` with an `EXISTS` on its hits, so read state is the item's own.
+
 Read state is `entry_state` alone. `entries` had `read`, `flagged` and `position` columns from
 before accounts; two bugs came from queries still reading them, and they were dropped in 0.5.
 

playwright.config.js

@@ -60,6 +60,7 @@ module.exports = defineConfig({
         // production's Tempo and never opens production's database.
         OTEL_EXPORTER_OTLP_ENDPOINT: '',
         IPX_DATABASE_URL: '',
+        IPX_TYPESAFE_URL: 'http://127.0.0.1:8792/jev',
       },
     },
   ],

tests/page-unit.js

@@ -143,3 +143,14 @@ test('a Patreon feed is told by its address', () => {
   assert.equal(run(`isPatreon({ url: 'https://example.com/feed' })`), false);
   assert.equal(run('isPatreon(null)'), false);
 });
+
+test('a Signal is a place of its own, found by its id', () => {
+  run(`S.signals = [{id: 7, name: 'F1', keywords: ['f1'], unread: 2}]`);
+  assert.equal(run(`signalId(':signal:7')`), 7);
+  assert.equal(run(`signalId(':all')`), null);
+  assert.equal(run(`signalId('signal-7')`), null, 'a feed id is never a Signal');
+  assert.equal(run(`signalOf(':signal:7').name`), 'F1');
+  assert.equal(run(`signalOf(':signal:8')`), undefined);
+  assert.match(run(`listUrl(':signal:7')`), /^\/api\/signals\/7\/entries\?/);
+  run(`S.signals = []`);
+});

tests/ui/app.spec.js

@@ -1141,6 +1141,7 @@ test('one action, one icon: the toolbar, the page and every dialog agree', async
     () => page.locator('#content .acts [data-a="settings"]').click(),
     () => page.locator('#content .acts [data-a="dl"]').click(),
     () => page.locator('#content .acts [data-a="rm"]').click(),
+    () => page.locator('#feedlist .place', { hasText: 'New signal' }).click(),
   ];
   for (const open of dialogs) {
     await open();
@@ -2068,3 +2069,36 @@ test('the page comes compressed, and the compressed copies cannot be asked for b
     expect((await page.request.get(p)).status(), p).toBe(404);
   }
 });
+
+test('a Signal holds the items Jev says fit its keywords, read where they are read', async ({ page }) => {
+  // #254. Jev is the fixture server's stand-in; the show is added here and removed at the end.
+  expect((await page.request.patch('/api/me', { data: { typesafe_key: 'test-key' } })).status()).toBe(204);
+  const { id: feed } = await (await page.request.post('/api/feeds', { data: { url: 'http://127.0.0.1:8792/signal.xml' } })).json();
+  await page.reload();
+
+  await page.locator('#feedlist .place', { hasText: 'New signal' }).click();
+  await page.locator('#sgname').fill('F1');
+  await page.locator('#sgkwin').fill('monaco');
+  await page.keyboard.press('Enter');
+  await page.locator('#sgsave').click();
+  const row = page.locator('#feedlist .place', { hasText: 'F1' });
+  await expect(row).toHaveClass(/sel/);
+
+  const signals = async () => (await (await page.request.get('/api/signals')).json());
+  const sg = (await signals()).find(s => s.name === 'F1');
+  const shown = async () => (await (await page.request.get(`/api/signals/${sg.id}/entries`)).json()).entries.map(e => e.title);
+  await expect.poll(shown, { timeout: 15_000 }).toEqual(['Monaco qualifying']);
+
+  await page.reload();
+  await row.click();
+  await page.locator('.ep', { hasText: 'Monaco qualifying' }).click();
+  await expect.poll(async () => (await (await page.request.get(`/api/entries/${encodeURIComponent(feed)}/sig-1`)).json()).read,
+    { timeout: 5_000 }).toBe(true);
+  expect((await signals()).find(s => s.id === sg.id).unread).toBe(0);
+
+  expect((await page.request.get(`/api/signals/${sg.id + 1000}/entries`)).status(), 'no such Signal').toBe(404);
+
+  await page.request.delete(`/api/signals/${sg.id}`);
+  await page.request.delete(`/api/feeds/${encodeURIComponent(feed)}`);
+  await page.request.patch('/api/me', { data: { typesafe_key: null } });
+});

tests/ui/fixtures/serve.js

@@ -7,6 +7,30 @@ const dir = __dirname;
 const port = Number(process.env.FIXTURE_PORT || 8792);
 
 http.createServer((req, res) => {
+  // Jev, for Signals (#254): sure of an item whose title has a question's keyword in it.
+  if (req.method === 'POST' && req.url === '/jev') {
+    let body = '';
+    req.on('data', c => { body += c; });
+    req.on('end', () => {
+      const { state, questions } = JSON.parse(body);
+      const title = (state.title || '').toLowerCase();
+      const answers = {};
+      for (const [id, q] of Object.entries(questions)) {
+        const words = q.instructions.replace(/^.*: /, '').replace(/\?$/, '').split(', ');
+        answers[id] = { noul: words.some(w => title.includes(w.toLowerCase())) ? 0.9 : 0.1 };
+      }
+      res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify({ answers }));
+    });
+    return;
+  }
+  // Dated today, since a Signal looks at the last 30 days only.
+  if (req.url === '/signal.xml') {
+    const now = new Date().toUTCString();
+    const item = (guid, title) => `<item><guid>${guid}</guid><title>${title}</title><pubDate>${now}</pubDate></item>`;
+    res.writeHead(200, { 'content-type': 'application/xml' }).end(`<?xml version="1.0"?><rss version="2.0"><channel>
+      <title>Signal Show</title>${item('sig-1', 'Monaco qualifying')}${item('sig-2', 'Baking bread')}</channel></rss>`);
+    return;
+  }
   const name = decodeURIComponent(req.url.split('?')[0].replace(/^\//, '')) || 'index';
   const file = path.join(dir, path.basename(name));
   fs.readFile(file, (err, body) => {

web/CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Signals, under the places in the sidebar: New signal makes one from a name and keywords, and its page lists the items from your feeds that are about them, read and unread as in their own feeds. Settings takes your TypeSafe key, which Signals need (#254).
 - A list opens at once when you have seen it before or the page fetched it ahead: All Subscriptions and the feeds with something unread are fetched when the page loads, and a feed or filter as the pointer or a finger reaches it, then refreshed behind once shown. A list that has to be waited for shows a spinner instead of nothing (#247). Currently Listening and the Directory's front page are fetched ahead the same way and open at once (#248).
 
 ### Changed

web/build.mjs

@@ -24,7 +24,7 @@ const WEB_VERSION = JSON.parse(fs.readFileSync(path.join(here, '..', 'package.js
 // The files are one script, concatenated in this order, not modules: they share one top-level
 // scope, as the single inline script did, and code that runs at load needs what came before it.
 const PAGES = {
-  'index.html': { script: 'app.js', src: ['util', 'theme', 'feeds', 'feedpage', 'items', 'player', 'dialogs', 'gestures', 'events', 'native'] },
+  'index.html': { script: 'app.js', src: ['util', 'theme', 'feeds', 'feedpage', 'items', 'player', 'dialogs', 'signals', 'gestures', 'events', 'native'] },
   'admin.html': { script: 'admin.js', src: ['util', 'theme', 'admin'] },
   'login.html': { script: 'login.js', src: ['login'] },
 };

web/src/dialogs.ts

@@ -440,6 +440,13 @@ async function prefsModal(){
       <span class="hint">Comma separated words or phrases, in every feed you read. An item with
       one in its title or text is hidden from you and not downloaded for you. Each feed's
       settings can add more.</span></div>
+    <div class="field"><label for="stskey">TypeSafe key</label>
+      <div class="inline"><input type="password" id="stskey" autocomplete="off" spellcheck="false"
+        placeholder="${S.me?.typesafe_key_set?'Saved. Paste a new one to replace it':'Paste your key'}">
+        <button class="btn ico" id="stsforget" title="Forget the key" aria-label="Forget the key"${S.me?.typesafe_key_set?'':' hidden'}>${ICON.trash}</button></div>
+      <span class="hint">Signals are picked by TypeSafe's Jev on your own key, from
+      <a href="https://typesafe.ai" target="_blank" rel="noopener">typesafe.ai</a>. It is kept on this
+      server and never shown again.</span></div>
     <div class="field boxed"><label for="tokname">API tokens</label>
       <div id="tokens"></div>
       <div class="inline"><input type="text" id="tokname" placeholder="What it is for">
@@ -469,6 +476,18 @@ async function prefsModal(){
       $<HTMLInputElement>('#tokval').select();
     }catch(err){ toast(why(err),true); }
   };
+  const saveKey=async(typesafe_key: string|null)=>{
+    try{
+      await api('/api/me',{method:'PATCH',body:JSON.stringify({typesafe_key})});
+      if(S.me) S.me.typesafe_key_set=!!typesafe_key;
+      $<HTMLInputElement>('#stskey').value='';
+      $<HTMLInputElement>('#stskey').placeholder=typesafe_key?'Saved. Paste a new one to replace it':'Paste your key';
+      $<HTMLButtonElement>('#stsforget').hidden=!typesafe_key;
+      toast(typesafe_key?'Saved':'Forgotten');
+    }catch(err){ toast(why(err),true); }
+  };
+  $<HTMLInputElement>('#stskey').onchange=(e: Event)=>{ const k=(e.target as HTMLInputElement).value.trim(); if(k) saveKey(k); };
+  $<HTMLButtonElement>('#stsforget').onclick=()=>saveKey(null);
   $<HTMLInputElement>('#sblock').onchange=async (e: Event)=>{
     const blocked=splitWords((e.target as HTMLInputElement).value);
     try{

web/src/feedpage.ts

@@ -6,10 +6,11 @@ function renderFeed(){
   const v=VIEWS[S.feed ?? ''];
   if(v&&v.url) return renderListed(v);
   const f=v?null:S.feeds.find(x=>x.id===S.feed);
+  const sg=signalOf(S.feed);
   $<HTMLButtonElement>('#tbRemove').disabled=!f;
   syncTools(null);
-  if(!v&&!f){ box.innerHTML='<p class="empty">Add a feed to get started.</p>'; $('#count').textContent=''; return; }
+  if(!v&&!f&&!sg){ box.innerHTML='<p class="empty">Add a feed to get started.</p>'; $('#count').textContent=''; return; }
-  const name=f?(f.title||f.id):v.title;
+  const name=f?(f.title||f.id):sg?sg.name:v.title;
   $<HTMLInputElement>('#epSearch').placeholder=`Search ${name}…`;
   const kids=f?S.feeds.filter(c=>c.group===f.id):[];
   if(kids.length){ box.classList.add('plain'); renderGroup(f!,kids); return; }
@@ -36,7 +37,7 @@ function renderFeed(){
         <button class="btn ico" data-a="settings" title="Settings" aria-label="Settings">${ICON.settings}</button>
         <button class="btn ico danger" data-a="rm" title="Unsubscribe" aria-label="Unsubscribe">${ICON.circleMinus}</button>
       </div>
-    </div>` : `
+    </div>` : sg ? signalHead(sg) : `
     <div class="fhead slim">
       <div class="art">${v.icon}</div>
       <div class="meta">
@@ -69,7 +70,7 @@ function renderFeed(){
   S.sel=null;
   showDetail(null);
 
-  $$('#content .acts .btn').forEach(b=>b.onclick=()=>f?feedAction(b.dataset.a!,f):allAction(b.dataset.a!));
+  $$('#content .acts .btn').forEach(b=>b.onclick=()=>f?feedAction(b.dataset.a!,f):sg?signalAction(b.dataset.a!,sg):allAction(b.dataset.a!));
   $$('#content .tabs button').forEach(b=>b.onpointerenter=b.ontouchstart=()=>{ if(S.feed) prefetch(S.feed,b.dataset.f!); });
   $$('#content .tabs button').forEach(b=>b.onclick=()=>{
     S.filter=b.dataset.f!; S.offset=0;

web/src/feeds.ts

@@ -9,7 +9,8 @@ function patchFeed(row: Feed){
 }
 
 async function loadFeeds(keepSel?: boolean){
-  S.feeds = await api('/api/feeds');
+  // A Signal list that fails leaves the feeds to show.
+  [S.feeds, S.signals] = await Promise.all([api('/api/feeds'), api('/api/signals').catch((): Signal[]=>[])]);
   // Only for a hint in a feed's settings, so once, on the page's first load, not on every reload
   // of the list (#103).
   if(!keepSel) api('/api/settings').then(g=>{globalMax=g.max_new_per_check}).catch(()=>{});
@@ -18,7 +19,7 @@ async function loadFeeds(keepSel?: boolean){
   // All Subscriptions rather than picking one alphabetically. Nothing to land on at all (a
   // brand new account) leaves S.feed alone, so the empty state's own message shows instead.
   if(!keepSel && S.feeds.length){
-    const known = S.feed && (VIEWS[S.feed] || S.feeds.some(f=>f.id===S.feed));
+    const known = S.feed && (VIEWS[S.feed] || signalOf(S.feed) || S.feeds.some(f=>f.id===S.feed));
     selectFeed(known ? S.feed! : ':all');
   }
 }
@@ -102,6 +103,7 @@ function renderFeeds(){
     el.onpointerenter=el.ontouchstart=()=>prefetch(id);
     places.appendChild(el);
   }
+  signalRows(places);
   list.appendChild(places);
   const shown=S.feeds.filter(f=>!q||(f.title||f.id).toLowerCase().includes(q));
   if(!shown.length){ list.insertAdjacentHTML('beforeend','<p class="empty" style="padding:20px 8px">No feeds.</p>'); return done(); }

web/src/items.ts

@@ -12,7 +12,10 @@ const listKey=(feed: string, filter=S.filter)=>[feed,filter,S.sort.col,S.sort.di
 function listUrl(feed: string, filter=S.filter, offset=0){
   const p=new URLSearchParams({offset:String(offset),limit:String(LIMIT),filter,sort:S.sort.col,dir:S.sort.dir});
   if(S.q) p.set('q',S.q);
-  return feed===':all' ? `/api/entries?${p}` : `/api/feeds/${encodeURIComponent(feed)}/entries?${p}`;
+  const sg=signalId(feed);
+  return feed===':all' ? `/api/entries?${p}`
+    : sg!==null ? `/api/signals/${sg}/entries?${p}`
+    : `/api/feeds/${encodeURIComponent(feed)}/entries?${p}`;
 }
 function keepList(key: string, r: ListPage){
   lists.delete(key); lists.set(key,r);
@@ -48,7 +51,8 @@ const prefetchPage=(url: string)=>{ if(!pages.has(url)) fetchPage(url).catch(()=
 /// A feed changed (a scan, a read elsewhere): its kept lists, and every one of All Subscriptions,
 /// are fetched again when next wanted.
 function forgetLists(feed: string){
-  for(const k of [...lists.keys()]) if(k.startsWith(feed+'\n')||k.startsWith(':all\n')) lists.delete(k);
+  // A Signal holds items of any feed.
+  for(const k of [...lists.keys()]) if(k.startsWith(feed+'\n')||k.startsWith(':all\n')||k.startsWith(':signal:')) lists.delete(k);
 }
 /// What is likely to be opened next, fetched two at a time once the page is idle: All Subscriptions
 /// and the feeds with something unread, most first.
@@ -120,10 +124,10 @@ function listLoading(box=$('#eps')){
 }
 function renderEntries(){
   const box=$('#eps'); if(!box) return;
-  const f=S.feeds.find(x=>x.id===S.feed), v=VIEWS[S.feed ?? ''];
+  const f=S.feeds.find(x=>x.id===S.feed), v=VIEWS[S.feed ?? ''], sg=signalOf(S.feed);
-  const unread=f?f.unread:S.feeds.reduce((n,x)=>n+(x.unread||0),0);
+  const unread=f?f.unread:sg?sg.unread:S.feeds.reduce((n,x)=>n+(x.unread||0),0);
   $('#count').textContent=
-    `${f?(f.title||f.id):v?v.title:''}: ${S.total} item${S.total===1?'':'s'}, ${unread} unread`;
+    `${f?(f.title||f.id):sg?sg.name:v?v.title:''}: ${S.total} item${S.total===1?'':'s'}, ${unread} unread`;
   const pane=$('#list'), top=pane?pane.scrollTop:0;
   box.innerHTML='';
   if(!S.entries.length){

web/src/signals.ts

@@ -0,0 +1,90 @@
+/* ---------------- signals ---------------- */
+/// Signals (#254): feeds you make from your own subscriptions, holding the items TypeSafe's Jev
+/// says are about the keywords you gave. A Signal is a place in the sidebar, `:signal:<id>`.
+
+/// The Signal's id in a place's id, or null for anything else.
+function signalId(view: string|null){
+  const m=/^:signal:(\d+)$/.exec(view||'');
+  return m ? Number(m[1]) : null;
+}
+function signalOf(view: string|null){
+  const id=signalId(view);
+  return id===null ? undefined : S.signals.find(s=>s.id===id);
+}
+
+/// Your Signals under the places, and the way to make one.
+function signalRows(places: HTMLElement){
+  for(const sg of S.signals){
+    const id=':signal:'+sg.id, el=document.createElement('div');
+    el.className='place'+(S.feed===id?' sel':'');
+    el.tabIndex=0; el.dataset.id=id;
+    el.innerHTML=`<span class="ico">${ICON.signal}</span><b>${esc(sg.name)}</b>`+
+      `<span class="badge${sg.unread?'':' zero'}" title="${sg.unread} unread">${sg.unread>999?'999+':sg.unread}</span>`;
+    el.onclick=()=>{ selectFeed(id); nav(false); };
+    el.onpointerenter=el.ontouchstart=()=>prefetch(id);
+    places.appendChild(el);
+  }
+  const add=document.createElement('div');
+  add.className='place'; add.tabIndex=0; add.dataset.id=':signal:new';
+  add.innerHTML=`<span class="ico">${ICON.plus}</span><b>New signal</b>`;
+  add.onclick=()=>{ signalModal(); nav(false); };
+  places.appendChild(add);
+}
+
+function signalHead(sg: Signal){
+  const nokey=S.me&&!S.me.typesafe_key_set;
+  return `
+    <div class="fhead slim">
+      <div class="art">${ICON.signal}</div>
+      <div class="meta">
+        <h2>${esc(sg.name)}</h2>
+        <div class="sub">Items from your feeds about ${esc(sg.keywords.join(', '))} · ${sg.unread} unread</div>
+        ${nokey?'<div class="sub" style="color:var(--warn)">Nothing is picked until you add your TypeSafe key in Settings.</div>':''}
+      </div>
+      <div class="acts">
+        <button class="btn ico" data-a="settings" title="Edit this signal" aria-label="Edit this signal">${ICON.settings}</button>
+        <button class="btn ico danger" data-a="rm" title="Delete this signal" aria-label="Delete this signal">${ICON.trash}</button>
+      </div>
+    </div>`;
+}
+
+async function signalAction(a: string, sg: Signal){
+  if(a==='settings') signalModal(sg);
+  if(a==='rm'){
+    if(!confirm(`Delete the signal ${sg.name}? The items stay in their own feeds.`)) return;
+    try{
+      await api(`/api/signals/${sg.id}`,{method:'DELETE'});
+      toast('Deleted'); await loadFeeds(true); selectFeed(':all');
+    }catch(e){ toast(why(e),true); }
+  }
+}
+
+/// Making a Signal, or changing one.
+function signalModal(sg?: Signal){
+  const nokey=S.me&&!S.me.typesafe_key_set;
+  openModal(`<h3>${sg?'Edit signal':'New signal'}</h3>
+    <div class="field"><label for="sgname">Name</label>
+      <input type="text" id="sgname" value="${esc(sg?.name||'')}" placeholder="Formula 1"></div>
+    <div class="field"><label for="sgkwin">About any of</label>
+      <div class="chips" id="sgkw">${chipsHTML(sg?.keywords||[],'sgkwin')}</div>
+      <span class="hint">TypeSafe's Jev reads each item in the feeds you subscribe to, the last 30 days
+      of them and every new one, and puts the ones about any of these here as well as in their own
+      feed. Each item is one call on your TypeSafe key.${sg?' Changing these reads them all again.':''}</span></div>
+    ${nokey?'<p class="hint" style="color:var(--warn)">Add your TypeSafe key in Settings first: nothing is picked without one.</p>':''}
+    <div class="cardacts"><button class="btn ico" onclick="closeModal()" title="Cancel" aria-label="Cancel">${ICON.close}</button>
+      <button class="btn ico primary" id="sgsave" title="${sg?'Save':'Make signal'}" aria-label="${sg?'Save':'Make signal'}">${sg?ICON.check:ICON.plus}</button></div>`);
+  wireChips($('#sgkw'));
+  $<HTMLInputElement>('#sgname').focus();
+  $<HTMLButtonElement>('#sgsave').onclick=async()=>{
+    const name=$<HTMLInputElement>('#sgname').value.trim(), keywords=chipWords($('#sgkw'));
+    if(!name){ $<HTMLInputElement>('#sgname').focus(); return; }
+    if(!keywords.length){ $<HTMLInputElement>('#sgkwin').focus(); return; }
+    const body=JSON.stringify({name,keywords});
+    try{
+      const r=sg ? await api(`/api/signals/${sg.id}`,{method:'PATCH',body})
+        : await api('/api/signals',{method:'POST',body});
+      closeModal(); toast(sg?'Saved':'Made. Items show up here as Jev reads them');
+      await loadFeeds(true); selectFeed(':signal:'+(sg?sg.id:r.id));
+    }catch(e){ toast(why(e),true); }
+  };
+}

web/src/types.ts

@@ -82,6 +82,16 @@ interface Me {
   deadair_saved: number;
   theme: {name: string, mode: string} | null;
   version: string;
+  /// A TypeSafe key is saved for Signals; the key itself never comes back.
+  typesafe_key_set: boolean;
+}
+
+/// A Signal in your sidebar, `/api/signals` (db.rs SignalRow).
+interface Signal {
+  id: number;
+  name: string;
+  keywords: string[];
+  unread: number;
 }
 
 /// A feed as the Directory lists it (web.rs PopularRow): never its address.
@@ -101,6 +111,7 @@ interface DirFeed {
 /// The page's own state (util.ts `S`).
 interface State {
   feeds: Feed[];
+  signals: Signal[];
   /// The feed or place open: a feed's id, or a view such as ':all'.
   feed: string | null;
   entries: Entry[];

web/src/util.ts

@@ -13,6 +13,8 @@ const esc = (s: string|null|undefined) => (s??'').replace(/[&<>"']/g,c=>ESCAPED[
 // the button's own colour. To add one, copy the path from svgs/<style>/<name>.svg at the same tag.
 const fa=(box: string,body: string)=>`<svg class="i" viewBox="${box}" aria-hidden="true">${body}</svg>`;
 const ICON: Record<string,string>={
+// Waves either side of a point: a Signal (#254).
+signal:fa('0 0 512 512','<g fill="none" stroke="currentColor" stroke-width="44" stroke-linecap="round"><path d="M136 136a170 170 0 0 0 0 240"/><path d="M376 136a170 170 0 0 1 0 240"/><path d="M204 196a85 85 0 0 0 0 120"/><path d="M308 196a85 85 0 0 1 0 120"/></g><circle cx="256" cy="256" r="38" fill="currentColor"/>'),
   share:fa('0 0 512 512','<path fill="currentColor" d="M384 192c53 0 96-43 96-96s-43-96-96-96-96 43-96 96c0 5.4 .5 10.8 1.3 16L159.6 184.1c-16.9-15-39.2-24.1-63.6-24.1-53 0-96 43-96 96s43 96 96 96c24.4 0 46.6-9.1 63.6-24.1L289.3 400c-.9 5.2-1.3 10.5-1.3 16 0 53 43 96 96 96s96-43 96-96-43-96-96-96c-24.4 0-46.6 9.1-63.6 24.1L190.7 272c.9-5.2 1.3-10.5 1.3-16s-.5-10.8-1.3-16l129.7-72.1c16.9 15 39.2 24.1 63.6 24.1z"/>'), // solid/share-nodes
   plus:fa('0 0 448 512','<path fill="currentColor" d="M256 64c0-17.7-14.3-32-32-32s-32 14.3-32 32l0 160-160 0c-17.7 0-32 14.3-32 32s14.3 32 32 32l160 0 0 160c0 17.7 14.3 32 32 32s32-14.3 32-32l0-160 160 0c17.7 0 32-14.3 32-32s-14.3-32-32-32l-160 0 0-160z"/>'), // solid/plus
   circleMinus:fa('0 0 512 512','<path fill="currentColor" d="M512 256A256 256 0 1 0 0 256a256 256 0 1 0 512 0zM184 232l144 0c13.3 0 24 10.7 24 24s-10.7 24-24 24l-144 0c-13.3 0-24-10.7-24-24s10.7-24 24-24z"/>'), // solid/circle-minus, for Unsubscribe
@@ -218,7 +220,7 @@ function nav(on: boolean){ $('#sidebar').classList.toggle('open',on); $('#scrim'
 
 /* ---------------- state ---------------- */
 const S: State = {
-  feeds:[],
+  feeds:[], signals:[],
   // Which feed (or place) and which tab were open last time, so a refresh lands back where
   // you were instead of jumping to the first feed alphabetically.
   feed:(()=>{ try{ return localStorage.getItem('ipx.feed'); }catch{ return null; } })(),