rays/ipx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 20 Wiki Activity

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

Browse Source 
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>
This commit is contained in:
Ray Slakinski
2026-10-16 16:54:50 +00:00
parent 30622fd3e1
commit 70990c70c7
21 changed files with 874 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

285
crates/ipx/src/db.rs
View File

@@ -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();