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

Signals: an admin chooses whether the server's TypeSafe key pays for everyone (#271)

Browse Source 
With signals_on_server set and TYPESAFE_KEY present, the judge asks Jev
about everyone's Signals on the server's key; otherwise each person turns
Signals on in Settings with a key of their own, which is now checked with
TypeSafe before it is saved. The Signals folder shows only when they are on.

The browser suite's daemon inherited the shell's TYPESAFE_KEY, so fixture
feeds were sent to the real TypeSafe to be categorised; it is blanked now,
and an empty key counts as none.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-10-17 13:18:03 +00:00
parent 33eb1f4db2
commit 9a41641eaa
15 changed files with 242 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/ipx/CHANGELOG.md
View File

@@ -15,6 +15,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- `/metrics` counts Signals: items judged and fitting, calls to Jev by how they ended and the time they took, time since the judge's last pass, and keys refused; a failed call to Jev shows as a failed trace (#270).
- An admin setting, Signals for everyone on the server's TypeSafe key, so the server pays for Jev instead of each person (#271).
### Changed
- A TypeSafe key is checked with TypeSafe before it is saved, and one it refuses is not saved (#271).
## [1.3.0] - 2026-10-16

11
crates/ipx/src/config.rs
View File

@@ -50,6 +50,9 @@ pub struct General {
pub directory_token: String,
/// Send the directory the public feeds people here subscribe to, and how many of them do.
pub directory_share: bool,
/// Signals are judged on the server's TypeSafe key (`TYPESAFE_KEY`), for everyone, rather than
/// on each person's own (#271). Without that key it changes nothing.
pub signals_on_server: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)]
@@ -207,6 +210,7 @@ impl Default for General {
directory_url: String::new(),
directory_token: String::new(),
directory_share: false,
signals_on_server: false,
}
}
}
@@ -331,6 +335,8 @@ pub struct Stored {
pub directory_token: String,
#[serde(default)]
pub directory_share: bool,
#[serde(default)]
pub signals_on_server: bool,
}
fn art_cache_mb() -> u64 {
@@ -338,7 +344,7 @@ fn art_cache_mb() -> u64 {
}
/// `[general]` keys that live in the database once it holds the configuration.
const STORED_KEYS: [&str; 9] = [
const STORED_KEYS: [&str; 10] = [
"schedule",
"max_total_gb",
"max_age_days",
@@ -348,6 +354,7 @@ const STORED_KEYS: [&str; 9] = [
"directory_url",
"directory_token",
"directory_share",
"signals_on_server",
];
impl Stored {
@@ -363,6 +370,7 @@ impl Stored {
directory_url: g.directory_url.clone(),
directory_token: g.directory_token.clone(),
directory_share: g.directory_share,
signals_on_server: g.signals_on_server,
}
}
@@ -377,6 +385,7 @@ impl Stored {
g.directory_url = self.directory_url;
g.directory_token = self.directory_token;
g.directory_share = self.directory_share;
g.signals_on_server = self.signals_on_server;
}
}

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

@@ -2579,7 +2579,8 @@ pub struct Ask {
#[derive(Debug)]
pub struct Unjudged {
pub user_id: i64,
pub key: String,
/// Their own key; None when the server's is used for everyone.
pub key: Option<String>,
pub feed_id: String,
pub guid: String,
pub feed_title: Option<String>,
@@ -2704,8 +2705,9 @@ impl Db {
/// 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>> {
/// `skip` are people whose key TypeSafe has refused. With `everyone`, the server's key is
/// used, and people without a key of their own are asked for too (#271).
pub async fn unjudged(&self, skip: &[i64], limit: i64, everyone: bool) -> Result<Vec<Unjudged>> {
let mut a = Args::default();
let since = a.p(now() - SIGNAL_DAYS * 86_400);
let skip = if skip.is_empty() {
@@ -2714,6 +2716,7 @@ impl Db {
format!("AND g.user_id NOT IN ({})", skip.iter().map(|u| a.p(*u)).collect::<Vec<_>>().join(", "))
};
let limit = a.p(limit);
let keyed = if everyone { "true" } else { "u.typesafe_key IS NOT NULL" };
let sql = format!(
"SELECT g.id AS signal_id, g.name, g.keywords, g.description AS signal_description, g.user_id, u.typesafe_key, e.feed_id, e.guid, e.title,
e.description, f.title AS feed_title
@@ -2722,7 +2725,7 @@ impl Db {
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}
WHERE {keyed} {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)
@@ -3394,19 +3397,23 @@ mod tests {
let food = db.add_signal(1, "Food", &words(&["baking"]), Some("Bread and cakes")).await.unwrap();
db.add_signal(2, "Theirs", &words(&["x"]), None).await.unwrap();
assert!(db.unjudged(&[], 50).await.unwrap().is_empty(), "nobody has a key yet");
assert!(db.unjudged(&[], 50, false).await.unwrap().is_empty(), "nobody has a key yet");
let everyone = db.unjudged(&[], 50, true).await.unwrap();
let mut seen: Vec<_> = everyone.iter().map(|u| (u.user_id, u.guid.as_str(), u.key.is_none())).collect();
seen.sort();
assert_eq!(seen, [(1, "x", true), (1, "y", true), (2, "x", true)], "on the server's key, everyone's Signals, no key of their own needed");
db.set_typesafe_key(1, Some("k")).await.unwrap();
let queue = db.unjudged(&[], 50).await.unwrap();
let queue = db.unjudged(&[], 50, false).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");
let ask = queue[0].signals.iter().find(|a| a.id == food).unwrap();
assert_eq!((ask.name.as_str(), ask.description.as_deref()), ("Food", Some("Bread and cakes")), "Jev is asked about the name and the sentence");
assert!(db.unjudged(&[1], 50).await.unwrap().is_empty(), "a refused key is passed over");
assert!(db.unjudged(&[1], 50, false).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");
assert!(db.unjudged(&[], 50, false).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();

2
crates/ipx/src/main.rs
View File

@@ -1837,7 +1837,7 @@ async fn scan_one(
&& (force || stored.entries == 0)
&& !parsed.entries.is_empty()
&& ctx.cfg().feeds.get(id).is_some_and(|f| f.category.is_none())
&& let Ok(key) = std::env::var("TYPESAFE_KEY")
&& let Some(key) = signals::typesafe_key()
{
match feed::classify(&ctx.client, &key, &parsed).await {
Ok(category) => {

100
crates/ipx/src/signals.rs
View File

@@ -46,10 +46,27 @@ pub async fn run(ctx: Arc<Ctx>, mut stop: tokio::sync::watch::Receiver<bool>) {
}
}
/// The key every Signal is judged on, when an admin has chosen that the server pays for Jev
/// (#271) and the server has a key; None leaves each person to their own.
pub fn server_key(ctx: &Ctx) -> Option<String> {
if !ctx.cfg().general.signals_on_server {
return None;
}
typesafe_key()
}
/// The server's own TypeSafe key, `TYPESAFE_KEY`; set but empty is none, as the browser suite
/// sets it so the test daemon never spends a real one.
pub fn typesafe_key() -> Option<String> {
std::env::var("TYPESAFE_KEY").ok().filter(|k| !k.trim().is_empty())
}
/// 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?;
let server = server_key(ctx);
// People's refused keys are not what is asked with when the server's is.
let skip: Vec<i64> = if server.is_some() { Vec::new() } else { ctx.signals_refused.lock().unwrap().iter().copied().collect() };
let items = ctx.db.unjudged(&skip, BATCH, server.is_some()).await?;
if items.is_empty() {
return Ok(false);
}
@@ -63,7 +80,8 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
if refused.contains(&item.user_id) {
continue;
}
match judge(ctx, item).await {
let Some(key) = server.as_deref().or(item.key.as_deref()) else { continue };
match judge(ctx, item, key).await {
Ok(verdicts) => {
judged += 1;
let hits = verdicts.iter().filter(|(_, h)| *h).count();
@@ -77,6 +95,12 @@ async fn pass(ctx: &Ctx) -> Result<bool> {
}
// 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.
// The server's key refused is everyone's: the queue waits for the next pass.
Err(e) if refuses_key(&e) && server.is_some() => {
tracing::warn!(error = %format!("{e:#}"), "TypeSafe refused the server's key for Signals");
stopped = true;
break;
}
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);
@@ -136,7 +160,7 @@ fn full(items: &[Unjudged]) -> bool {
items.iter().map(|i| i.signals.len()).sum::<usize>() as i64 >= BATCH
}
fn refuses_key(e: &anyhow::Error) -> bool {
pub fn refuses_key(e: &anyhow::Error) -> bool {
matches!(status(e), Some(401 | 402 | 403))
}
@@ -144,8 +168,7 @@ fn malformed(e: &anyhow::Error) -> bool {
matches!(status(e), Some(400 | 422))
}
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());
async fn judge(ctx: &Ctx, item: &Unjudged, key: &str) -> Result<Vec<(i64, bool)>> {
let span = tracing::info_span!(
"jev",
user = item.user_id,
@@ -154,17 +177,7 @@ async fn judge(ctx: &Ctx, item: &Unjudged) -> Result<Vec<(i64, bool)>> {
error = tracing::field::Empty,
"otel.status_code" = tracing::field::Empty,
);
let answer: serde_json::Value = called(span.clone(), async {
let r = jev_post(ctx, url, &item.key).json(&request(item)).send().await?;
let status = r.status().as_u16();
tracing::Span::current().record("http.response.status_code", status);
if !r.status().is_success() {
let body: String = r.text().await.unwrap_or_default().chars().take(500).collect();
return Err(Refused { status, body }.into());
}
anyhow::Ok(r.json().await?)
})
.await?;
let answer = called(span, ask(ctx, key, &request(item))).await?;
// A question Jev left unanswered stays unjudged and is asked again.
Ok(item.signals.iter().filter_map(|a| hit(&answer, a.id).map(|h| (a.id, h))).collect())
}
@@ -217,7 +230,6 @@ fn hit(answer: &serde_json::Value, id: i64) -> Option<bool> {
/// ipx-web's, and sends the likeliest. On the server's key, not the Signal owner's: it is the
/// server's page doing it, once a Signal.
pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], description: Option<&str>, candidates: &[(String, String)]) -> Result<String> {
let url = std::env::var("IPX_TYPESAFE_URL").unwrap_or_else(|_| feed::JEV_URL.to_owned());
let span = tracing::info_span!(
"jev_icon",
candidates = candidates.len(),
@@ -226,26 +238,46 @@ pub async fn pick_icon(ctx: &Ctx, key: &str, name: &str, words: &[String], descr
"otel.status_code" = tracing::field::Empty,
);
called(span, async {
let r = jev_post(ctx, url, key).json(&icon_request(name, words, description, candidates)).send().await?;
let status = r.status().as_u16();
tracing::Span::current().record("http.response.status_code", status);
if !r.status().is_success() {
let body: String = r.text().await.unwrap_or_default().chars().take(500).collect();
return Err(Refused { status, body }.into());
}
let answer: serde_json::Value = r.json().await?;
let answer = ask(ctx, key, &icon_request(name, words, description, candidates)).await?;
chosen(&answer, candidates).ok_or_else(|| anyhow::anyhow!("Jev chose no icon offered: {answer}"))
})
.await
}
/// A request to Jev, carrying this trace's `traceparent` as every call out of ipx does.
fn jev_post(ctx: &Ctx, url: String, key: &str) -> reqwest::RequestBuilder {
let req = ctx.client.post(url).bearer_auth(key).timeout(feed::FEED_TIMEOUT);
match ipx_directory::otel::traceparent() {
Some(tp) => req.header("traceparent", tp),
None => req,
/// Whether TypeSafe takes `key`, by one short question, before it is saved (#271): a key it
/// refuses would sit saved with nothing ever picked out, and nobody told why.
pub async fn check_key(ctx: &Ctx, key: &str) -> Result<()> {
let span = tracing::info_span!(
"jev_key_check",
"http.response.status_code" = tracing::field::Empty,
error = tracing::field::Empty,
"otel.status_code" = tracing::field::Empty,
);
let body = serde_json::json!({
"model": "jev-latest",
"state": "A reader is saving their key.",
"questions": { "ok": { "type": "noul", "instructions": "Is this a key check?",
"criteria": { "true": "It is.", "false": "It is not." } } },
});
called(span, ask(ctx, key, &body)).await.map(|_| ())
}
/// Puts `body` to Jev on `key` and gives its answer, or what TypeSafe refused it with. Carries
/// this trace's `traceparent`, as every call out of ipx does.
async fn ask(ctx: &Ctx, key: &str, body: &serde_json::Value) -> Result<serde_json::Value> {
let url = std::env::var("IPX_TYPESAFE_URL").unwrap_or_else(|_| feed::JEV_URL.to_owned());
let mut req = ctx.client.post(url).bearer_auth(key).json(body).timeout(feed::FEED_TIMEOUT);
if let Some(tp) = ipx_directory::otel::traceparent() {
req = req.header("traceparent", tp);
}
let r = req.send().await?;
let status = r.status().as_u16();
tracing::Span::current().record("http.response.status_code", status);
if !r.status().is_success() {
let body: String = r.text().await.unwrap_or_default().chars().take(500).collect();
return Err(Refused { status, body }.into());
}
Ok(r.json().await?)
}
/// Runs a call to Jev in `span`, counting it in /metrics by how it ended, and marking the span
@@ -389,7 +421,7 @@ mod tests {
#[test]
fn a_batch_is_full_by_its_rows_not_its_items() {
let item = |n: usize| Unjudged {
user_id: 1, key: "k".into(), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None,
user_id: 1, key: Some("k".into()), feed_id: "f".into(), guid: "g".into(), feed_title: None, title: None, description: None,
signals: (0..n as i64).map(|id| Ask { id, name: "s".into(), keywords: vec![], description: None }).collect(),
};
assert!(full(&(0..25).map(|_| item(2)).collect::<Vec<_>>()), "25 items of two Signals each is the 50 asked for");
@@ -400,7 +432,7 @@ mod tests {
fn each_signal_is_its_own_question_on_the_items_words() {
let item = Unjudged {
user_id: 1,
key: "k".into(),
key: Some("k".into()),
feed_id: "f".into(),
guid: "g".into(),
feed_title: Some("Pit Lane".into()),

29
crates/ipx/src/web.rs
View File

@@ -387,6 +387,8 @@ async fn me(
"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),
// Whose key Signals are judged on (#271): "server", for everyone, or "own", each person's.
"signals": if crate::signals::server_key(&state.ctx).is_some() { "server" } else { "own" },
}))
}
@@ -414,6 +416,16 @@ async fn patch_me(
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"));
}
if let Some(k) = &key {
crate::signals::check_key(&state.ctx, k).await.map_err(|e| {
if crate::signals::refuses_key(&e) {
ApiError::bad_request("TypeSafe did not accept that key")
} else {
tracing::warn!(error = %format!("{e:#}"), "a TypeSafe key could not be checked");
ApiError::bad_gateway("TypeSafe could not be asked about that key just now; try again")
}
})?;
}
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();
@@ -517,7 +529,7 @@ async fn pick_signal_icon(
_user: crate::db::User,
Json(body): Json<IconAsk>,
) -> Result<Json<serde_json::Value>, ApiError> {
let Ok(key) = std::env::var("TYPESAFE_KEY") else {
let Some(key) = crate::signals::typesafe_key() else {
return Err(ApiError::not_found("picking an icon is not set up on this server"));
};
let name = signal_name(&body.name)?;
@@ -2319,6 +2331,9 @@ struct Settings {
/// Whether a token is set: the token itself never leaves the server.
directory_token_set: bool,
directory_share: bool,
signals_on_server: bool,
/// Whether the server has a TypeSafe key (`TYPESAFE_KEY`) for `signals_on_server` to use.
typesafe_server_key: bool,
}
async fn get_settings(State(state): State<WebState>) -> Json<Settings> {
@@ -2335,6 +2350,8 @@ async fn get_settings(State(state): State<WebState>) -> Json<Settings> {
directory_url: cfg.general.directory_url.clone(),
directory_token_set: !cfg.general.directory_token.trim().is_empty(),
directory_share: cfg.general.directory_share,
signals_on_server: cfg.general.signals_on_server,
typesafe_server_key: crate::signals::typesafe_key().is_some(),
})
}
@@ -2351,6 +2368,7 @@ struct SettingsPatch {
/// Replaces the token; empty clears it.
directory_token: Option<String>,
directory_share: Option<bool>,
signals_on_server: Option<bool>,
}
async fn patch_settings(
@@ -2403,7 +2421,16 @@ async fn patch_settings(
if let Some(v) = body.directory_share {
cfg.general.directory_share = v;
}
let signals_changed = body.signals_on_server.is_some_and(|v| v != cfg.general.signals_on_server);
if let Some(v) = body.signals_on_server {
cfg.general.signals_on_server = v;
}
state.ctx.store_cfg(cfg).await?;
if signals_changed {
// Whoever's key was refused is tried again on whichever key is now used.
state.ctx.signals_refused.lock().unwrap().clear();
state.ctx.signals.notify_one();
}
Ok(StatusCode::NO_CONTENT)
}