~ knotless Search/
↑↓ move↵ open⇧↵ all results

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

grain-serve keeps when an account last saw its notifications

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:notifications/state

Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.

crates/grain-serve/src/index/import/tests.rs+46−5
View file
@@ -125,7 +125,11 @@ INSERT INTO _push_tokens VALUES
('did:plc:bob', '{"endpoint":"https://push.example/1"}', 'web', '2025-01-03T00:00:00.000Z');
INSERT INTO _preferences VALUES
('did:plc:alice', 'lastSeenNotifications', '"2025-01-01T00:00:00.000Z"', '2025-01-01T00:00:00.000Z'),
('did:plc:alice', 'contentLabels', '{"nudity":"warn"}', '2025-01-02 00:00:00');
('did:plc:alice', 'contentLabels', '{"nudity":"warn"}', '2025-01-02 00:00:00'),
('did:plc:alice', 'notificationPrefs', '{"favorites":{"push":true,"inApp":false,"from":"all"},"follows":{"push":false,"inApp":true,"from":"follows"},"comments":{"push":true,"inApp":true,"from":"all"}}', '2025-01-03T00:00:00.000Z'),
('did:plc:carol', 'lastSeenNotifications', '"2025-01-04T05:30:00+05:30"', '2025-01-04T00:00:00.000Z'),
('did:plc:carol', 'notificationPrefs', '{"mentions":{"inApp":false},"favorites":"off","later":{"push":false}}', '2025-01-05T00:00:00.000Z'),
('did:plc:dave', 'notificationPrefs', 'not json', '2025-01-06T00:00:00.000Z');
INSERT INTO _mutes VALUES
('did:plc:alice', 'did:plc:bob', '2025-01-01T00:00:00.000Z'),
('did:plc:carol', 'did:plc:bob', '2025-01-02T00:00:00.000Z');
@@ -362,9 +366,27 @@ fn every_row_of_what_the_network_cannot_rebuild_is_imported() {
);
assert_eq!(
rows("SELECT did, key, value, updated_at FROM preference ORDER BY key"),
[r#"did:plc:alice|contentLabels|{"nudity":"warn"}|2025-01-02 00:00:00"#]
);
// Notifications' preferences, tables of their own: a seen time in
// UTC, and a row a kind, as hatk read each.
assert_eq!(
rows("SELECT did, seen_at FROM notification_seen ORDER BY did"),
[
"did:plc:alice|2025-01-01T00:00:00.000Z",
"did:plc:carol|2025-01-04T00:00:00.000Z",
]
);
assert_eq!(
rows(
"SELECT did, kind, push, in_app, from_follows FROM notification_pref ORDER BY did, kind"
),
[
r#"did:plc:alice|contentLabels|{"nudity":"warn"}|2025-01-02 00:00:00"#,
r#"did:plc:alice|lastSeenNotifications|"2025-01-01T00:00:00.000Z"|2025-01-01T00:00:00.000Z"#,
"did:plc:alice|comments|1|1|0",
"did:plc:alice|favorites|1|0|0",
"did:plc:alice|follows|0|1|1",
// Left out, a kind is on.
"did:plc:carol|mentions|1|0|0",
]
);
assert_eq!(
@@ -415,7 +437,9 @@ fn every_row_of_what_the_network_cannot_rebuild_is_imported() {
("label", 4, 4),
("report", 2, 2),
("mute", 2, 2),
("preference", 2, 2),
("preference", 1, 1),
("notification_seen", 2, 2),
("notification_pref", 4, 4),
("push_token", 3, 3),
("story_view", 2, 2),
("space_invite", 2, 2),
@@ -633,7 +657,7 @@ fn a_row_the_index_does_not_take_fails_the_import() {
let summary = import(&old, &into, false).unwrap();
assert!(!summary.written);
let preference = kept(&summary, "preference");
assert_eq!((preference.source, preference.imported), (3, 2));
assert_eq!((preference.source, preference.imported), (2, 1));
assert_eq!(summary.failed.len(), 1, "{:?}", summary.failed);
assert!(summary.failed[0].starts_with("preference Text(\"did:plc:erin\")"));
let said = summary.to_string();
@@ -656,6 +680,23 @@ fn a_row_the_index_does_not_take_fails_the_import() {
assert_eq!(main(&args, |_| None), 1);
}
#[test]
fn a_seen_time_that_is_no_time_fails_the_import() {
let dir = tempfile::tempdir().unwrap();
let odd = "INSERT INTO _preferences VALUES
('did:plc:erin', 'lastSeenNotifications', '\"yesterday\"', '2025-01-01T00:00:00.000Z');";
let old = hatk(dir.path(), &format!("{KEPT_ROWS}{odd}"));
let summary = import(&old, &dir.path().join("index.db"), false).unwrap();
assert!(!summary.written);
let seen = kept(&summary, "notification_seen");
assert_eq!((seen.source, seen.imported), (3, 2));
assert!(
summary.failed[0].starts_with("notification_seen Text(\"did:plc:erin\")"),
"{:?}",
summary.failed
);
}
#[test]
fn hatk_with_its_repos_alone_imports() {
let dir = tempfile::tempdir().unwrap();
crates/grain-serve/src/index/import.rs+41−3
View file
@@ -39,6 +39,15 @@
//! `pending` if not. A status hatk never wrote is `pending`, and named in
//! the summary and in `last_error`.
//!
//! Of `_preferences`, the two keys notifications kept are tables of their
//! own: `lastSeenNotifications` is `notification_seen`, its time in UTC
//! as [`grain::keys::iso`] writes it, and `notificationPrefs` is a row of
//! `notification_pref` a kind of notification it set, as hatk read it: a
//! kind is pushed, or shown in the app, unless it says `false`, and from
//! anyone unless it says `from: "follows"`. A kind it kept as anything
//! but an object, or a value that is not JSON, said nothing hatk read, and
//! is not counted. The rest of `_preferences` is `preference`, as it was.
//!
//! `_cursor`'s Jetstream rows (`jetstream:<url>`) are `stream_cursor`'s,
//! so the stream resumes where hatk left it.
//!
@@ -325,11 +334,40 @@ pub const KEPT: &[Kept] = &[
Kept {
table: "preference",
source: "_preferences",
from: "_preferences",
count: "SELECT count(*) FROM _preferences",
select: "SELECT did, key, value, updated_at FROM _preferences",
from: "_preferences, other keys",
count: "SELECT count(*) FROM _preferences
WHERE key NOT IN ('lastSeenNotifications', 'notificationPrefs')",
select: "SELECT did, key, value, updated_at FROM _preferences
WHERE key NOT IN ('lastSeenNotifications', 'notificationPrefs')",
insert: "INSERT INTO preference (did, key, value, updated_at) VALUES (?1, ?2, ?3, ?4)",
},
Kept {
table: "notification_seen",
source: "_preferences",
from: "_preferences, lastSeenNotifications",
count: "SELECT count(*) FROM _preferences WHERE key = 'lastSeenNotifications'",
// A time that is not one is null, which the index does not take.
select: "SELECT did, CASE WHEN json_valid(value)
THEN strftime('%Y-%m-%dT%H:%M:%fZ', value ->> '$') END
FROM _preferences WHERE key = 'lastSeenNotifications'",
insert: "INSERT INTO notification_seen (did, seen_at) VALUES (?1, ?2)",
},
Kept {
table: "notification_pref",
source: "_preferences",
from: "_preferences, notificationPrefs, a row a kind",
count: "SELECT count(*) FROM _preferences p, json_each(CASE WHEN json_valid(p.value) THEN p.value ELSE '{}' END) k
WHERE p.key = 'notificationPrefs' AND k.type = 'object'
AND k.key IN ('favorites', 'follows', 'comments', 'mentions')",
select: "SELECT p.did, k.key, coalesce(json_type(k.value, '$.push') <> 'false', 1),
coalesce(json_type(k.value, '$.inApp') <> 'false', 1),
coalesce(k.value ->> '$.from' = 'follows', 0)
FROM _preferences p, json_each(CASE WHEN json_valid(p.value) THEN p.value ELSE '{}' END) k
WHERE p.key = 'notificationPrefs' AND k.type = 'object'
AND k.key IN ('favorites', 'follows', 'comments', 'mentions')",
insert: "INSERT INTO notification_pref (did, kind, push, in_app, from_follows)
VALUES (?1, ?2, ?3, ?4, ?5)",
},
Kept {
table: "push_token",
source: "_push_tokens",
crates/grain-serve/src/index/mod.rs+22−0
View file
@@ -431,6 +431,28 @@ impl Index {
.exists([did])
}
/// Marks `did`'s notifications seen as of `seen_at`, an RFC 3339
/// time, kept in UTC to the millisecond as SQLite's `strftime` writes
/// it, so that it compares as text with the times records are kept
/// at. A later call replaces it, earlier or not, as Bluesky's
/// `updateSeen` does. Whether it was kept: not where it is a time
/// UTC cannot write in four digits of year, such as the first moments
/// of year 0 east of Greenwich.
pub fn set_seen(&self, did: &str, seen_at: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let sql = "SELECT strftime('%Y-%m-%dT%H:%M:%fZ', ?1)";
let utc: Option<String> = db.query_row(sql, [seen_at], |r| r.get(0))?;
let Some(utc) = utc.filter(|t| lexicon::valid_datetime(t)) else {
return Ok(false);
};
db.execute(
"INSERT INTO notification_seen VALUES (?1, ?2)
ON CONFLICT (did) DO UPDATE SET seen_at = excluded.seen_at",
[did, &utc],
)?;
Ok(true)
}
/// The last sequence number read from the stream `source`.
pub fn cursor(&self, source: &str) -> rusqlite::Result<Option<i64>> {
let db = self.db.lock().unwrap();
crates/grain-serve/src/index/schema.rs+33−2
View file
@@ -57,7 +57,13 @@
//! has them; one on an account names it by its bare DID, not
//! `at://<did>`, as Bluesky's do;
//! - `report`, `mute`, `preference`, `push_token`, `story_view`,
//! `space_invite`, `space_support`.
//! `space_invite`, `space_support`;
//! - `notification_seen`, when each account last saw its notifications
//! (`updateSeen`'s `seenAt`, in UTC as [`grain::keys::iso`] writes it),
//! and `notification_pref`, how each kind of notification reaches it
//! (`favorites`, `follows`, `comments`, `mentions`): pushed or not,
//! shown in the app or not, and from anyone or only from accounts it
//! follows. A kind with no row reaches it every way, from anyone.
//!
//! A migration may drop and remake a record table, which backfill refills;
//! these it only alters in place.
@@ -71,7 +77,7 @@
/// Each step of the schema, run once, in order, by `PRAGMA user_version`.
/// A step is only ever appended: one that has run is never changed.
pub const MIGRATIONS: &[&str] = &[V1, V2, V3, V4];
pub const MIGRATIONS: &[&str] = &[V1, V2, V3, V4, V5];
const V1: &str = "
CREATE TABLE gallery (
@@ -331,6 +337,31 @@ INSERT INTO gallery_search (rowid, handle, title, description, place)
FROM gallery g JOIN search_doc d ON d.uri = g.uri LEFT JOIN repo r ON r.did = g.did;
";
/// Notifications' own state, apart from `preference`, and what an index
/// imported before it kept there moved over: hatk's `lastSeenNotifications`
/// and `notificationPrefs` preferences. A kind of notification hatk kept
/// as anything but an object said nothing, and a seen time that is no
/// time is dropped.
const V5: &str = "
CREATE TABLE notification_seen (did TEXT PRIMARY KEY, seen_at TEXT NOT NULL) WITHOUT ROWID;
CREATE TABLE notification_pref (
did TEXT NOT NULL, kind TEXT NOT NULL, push INTEGER NOT NULL, in_app INTEGER NOT NULL,
from_follows INTEGER NOT NULL, PRIMARY KEY (did, kind)) WITHOUT ROWID;
INSERT INTO notification_seen (did, seen_at)
SELECT did, strftime('%Y-%m-%dT%H:%M:%fZ', value ->> '$') FROM preference
WHERE key = 'lastSeenNotifications' AND json_valid(value)
AND strftime('%Y-%m-%dT%H:%M:%fZ', value ->> '$') IS NOT NULL;
INSERT INTO notification_pref (did, kind, push, in_app, from_follows)
SELECT p.did, k.key, coalesce(json_type(k.value, '$.push') <> 'false', 1),
coalesce(json_type(k.value, '$.inApp') <> 'false', 1),
coalesce(k.value ->> '$.from' = 'follows', 0)
FROM preference p, json_each(CASE WHEN json_valid(p.value) THEN p.value ELSE '{}' END) k
WHERE p.key = 'notificationPrefs' AND k.type = 'object'
AND k.key IN ('favorites', 'follows', 'comments', 'mentions');
DELETE FROM preference WHERE key IN ('lastSeenNotifications', 'notificationPrefs');
";
/// The label values that leave what carries them out of feeds: the
/// imperative ones, and those whose default is to hide.
pub const HIDDEN_LABELS: [&str; 9] = [
crates/grain-serve/src/index/tests.rs+37−0
View file
@@ -447,6 +447,43 @@ fn migrations() {
assert!(failed.starts_with("opening "), "{failed}");
}
#[test]
fn notifications_preferences_move_to_their_own_tables() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("index.db");
let before = &schema::MIGRATIONS[..4];
drop(Index::with(Connection::open(&path).unwrap(), before).unwrap());
let db = Connection::open(&path).unwrap();
db.execute_batch(
r#"INSERT INTO preference VALUES
('did:plc:a', 'lastSeenNotifications', '"2026-10-31T12:00:00Z"', 'x'),
('did:plc:b', 'lastSeenNotifications', '"never"', 'x'),
('did:plc:c', 'lastSeenNotifications', 'not json', 'x'),
('did:plc:a', 'notificationPrefs', '{"follows":{"push":false,"inApp":true,"from":"follows"},"comments":[]}', 'x'),
('did:plc:b', 'notificationPrefs', 'not json', 'x'),
('did:plc:a', 'pinnedFeeds', '[]', 'x');"#,
)
.unwrap();
drop(db);
let index = Index::open(&path).unwrap();
let rows = |sql: &str| -> Vec<String> {
index.read(|db| {
let mut stmt = db.prepare(sql).unwrap();
let rows = stmt.query_map([], |r| r.get(0)).unwrap();
rows.collect::<rusqlite::Result<_>>().unwrap()
})
};
assert_eq!(
rows("SELECT did || ' ' || seen_at FROM notification_seen"),
["did:plc:a 2026-10-31T12:00:00.000Z"]
);
assert_eq!(
rows("SELECT concat_ws(' ', did, kind, push, in_app, from_follows) FROM notification_pref"),
["did:plc:a follows 0 1 1"]
);
assert_eq!(rows("SELECT key FROM preference"), ["pinnedFeeds"]);
}
#[test]
fn the_queries_still_to_port_read_by_index() {
let index = Index::memory().unwrap();
crates/grain-serve/src/read/mod.rs+1−4
View file
@@ -234,10 +234,7 @@ pub fn list<S: AsRef<str>>(items: &[S]) -> String {
impl Reader<'_> {
/// The caller, for a method only an account may call.
pub fn signed_in(&self) -> Result<&str, Failure> {
self.viewer.ok_or_else(|| {
let why = "a service token is required";
Failure::new(StatusCode::UNAUTHORIZED, "AuthenticationRequired", why)
})
self.viewer.ok_or_else(Failure::authentication_required)
}
/// The DID `actor`, a DID or a handle, names, if the index knows the
crates/grain-serve/src/lib.rs+8−2
View file
@@ -1,12 +1,14 @@
//! grain's server. It answers atproto's health check,
//! `social.grain.sync.indexRecords` ([`sync`]) and the queries that read
//! the index alone ([`read`]); the feeds, login, writes and push join it
//! `social.grain.sync.indexRecords` ([`sync`]), the queries that read
//! the index alone ([`read`]) and what an account has seen of its
//! notifications ([`notification`]); the feeds, login, writes and push join it
//! pull by pull on `feature/rust`, as the plan orders them.
pub mod backfill;
pub mod cdn;
pub mod index;
pub mod jetstream;
pub mod notification;
pub mod read;
pub mod sync;
pub mod viewer;
@@ -164,6 +166,10 @@ pub fn router(app: Arc<App>) -> Router {
"/xrpc/social.grain.sync.indexRecords",
post(sync::index_records),
)
.route(
&format!("/xrpc/{}", notification::UPDATE_SEEN),
post(notification::update_seen),
)
.route("/xrpc/{*method}", any(not_implemented))
.with_state(app)
}
crates/grain-serve/src/notification.rsnew+56−0
View file
@@ -0,0 +1,56 @@
//! What an account has seen of its notifications:
//! `social.grain.notification.updateSeen`, as Bluesky's
//! `app.bsky.notification.updateSeen` takes it. The time is kept in
//! `notification_seen` ([`Index::set_seen`]), which the index cannot
//! rebuild from the network.
//!
//! [`Index::set_seen`]: crate::index::Index::set_seen
use crate::App;
use crate::xrpc::Failure;
use axum::Json;
use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use grain::lexicon;
use serde::Deserialize;
use serde_json::{Value, json};
use std::sync::Arc;
/// The method, as a service token for it names it in `lxm`.
pub const UPDATE_SEEN: &str = "social.grain.notification.updateSeen";
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Seen {
seen_at: String,
}
/// `social.grain.notification.updateSeen`: the caller's notifications
/// marked seen as of `seenAt`, which only an account may do.
pub async fn update_seen(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, Failure> {
let Some(index) = &app.index else {
let why = "this server keeps no index";
return Err(Failure::new(
StatusCode::NOT_IMPLEMENTED,
"MethodNotImplemented",
why,
));
};
let viewer = crate::viewer::viewer(&app, &headers, UPDATE_SEEN).await?;
let did = viewer.ok_or_else(Failure::authentication_required)?;
let input: Seen = serde_json::from_slice(&body)
.map_err(|e| Failure::invalid(format!("the input is not {{seenAt}}: {e}")))?;
let invalid = || Failure::invalid("seenAt must be a valid datetime");
if !lexicon::valid_datetime(&input.seen_at) {
return Err(invalid());
}
match index.set_seen(&did, &input.seen_at)? {
true => Ok(Json(json!({}))),
false => Err(invalid()),
}
}
crates/grain-serve/src/sync.rs+1−4
View file
@@ -224,10 +224,7 @@ pub async fn index_records(
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.ok_or_else(|| {
let why = "a service token is required";
Failure::new(StatusCode::UNAUTHORIZED, "AuthenticationRequired", why)
})?;
.ok_or_else(Failure::authentication_required)?;
let did = grain::auth::verify(&app.ctx, token, aud, LXM)
.await
.map_err(|e| Failure::new(StatusCode::UNAUTHORIZED, "InvalidToken", format!("{e:#}")))?;
crates/grain-serve/src/xrpc.rs+7−0
View file
@@ -25,6 +25,13 @@ impl Failure {
}
}
/// A method only an account may call, called by no one: 401
/// `AuthenticationRequired`.
pub fn authentication_required() -> Failure {
let why = "a service token is required";
Failure::new(StatusCode::UNAUTHORIZED, "AuthenticationRequired", why)
}
/// A request that does not say what it must: 400 `InvalidRequest`.
pub fn invalid(message: impl Into<String>) -> Failure {
Failure::new(StatusCode::BAD_REQUEST, "InvalidRequest", message)
crates/grain-serve/tests/notifications.rsnew+132−0
View file