~ 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 the devices notifications are pushed to

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:push/register

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

crates/grain-import/src/lib.rs+8−1
View file
@@ -53,6 +53,9 @@
//! 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.
//!
//! Of `_push_tokens`, a device's platform is grain's word for it: hatk's
//! `apns` is `ios` and its `fcm` is `android`; `web`, and any other, stay.
//!
//! `_cursor`'s Jetstream rows (`jetstream:<url>`) are `stream_cursor`'s,
//! so the stream resumes where hatk left it.
//!
@@ -443,7 +446,11 @@ pub const KEPT: &[Kept] = &[
source: "_push_tokens",
from: "_push_tokens",
count: "SELECT count(*) FROM _push_tokens",
select: "SELECT did, token, platform, created_at FROM _push_tokens",
// hatk named a device's platform by the service that delivers to
// it; grain by the platform, as registerPush does.
select: "SELECT did, token,
CASE platform WHEN 'apns' THEN 'ios' WHEN 'fcm' THEN 'android' ELSE platform END,
created_at FROM _push_tokens",
insert: "INSERT INTO push_token (did, token, platform, created_at) VALUES (?1, ?2, ?3, ?4)",
},
Kept {
crates/grain-import/src/tests.rs+6−4
View file
@@ -123,9 +123,10 @@ INSERT INTO _reports VALUES
(1, 'at://did:plc:bob/social.grain.gallery/3l3qo2vuowo2a', 'did:plc:bob', 'spam', 'ads', 'did:plc:alice', 'open', NULL, NULL, '2025-01-06T00:00:00.000Z'),
(4, 'did:plc:carol', 'did:plc:carol', 'rude', NULL, 'did:plc:alice', 'resolved', 'did:plc:admin', '2025-01-08T00:00:00.000Z', '2025-01-07T00:00:00.000Z');
INSERT INTO _push_tokens VALUES
('did:plc:alice', 'apns-token', 'ios', '2025-01-01T00:00:00.000Z'),
('did:plc:alice', 'fcm-token', 'android', '2025-01-02T00:00:00.000Z'),
('did:plc:bob', '{"endpoint":"https://push.example/1"}', 'web', '2025-01-03T00:00:00.000Z');
('did:plc:alice', 'apns-token', 'apns', '2025-01-01T00:00:00.000Z'),
('did:plc:alice', 'fcm-token', 'fcm', '2025-01-02T00:00:00.000Z'),
('did:plc:bob', '{"endpoint":"https://push.example/1"}', 'web', '2025-01-03T00:00:00.000Z'),
('did:plc:carol', 'later-token', 'later', '2025-01-04T00: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'),
@@ -386,6 +387,7 @@ fn every_row_of_what_the_network_cannot_rebuild_is_imported() {
"did:plc:alice|apns-token|ios|2025-01-01T00:00:00.000Z",
"did:plc:alice|fcm-token|android|2025-01-02T00:00:00.000Z",
r#"did:plc:bob|{"endpoint":"https://push.example/1"}|web|2025-01-03T00:00:00.000Z"#,
"did:plc:carol|later-token|later|2025-01-04T00:00:00.000Z",
]
);
assert_eq!(
@@ -464,7 +466,7 @@ fn every_row_of_what_the_network_cannot_rebuild_is_imported() {
("preference", 1, 1),
("notification_seen", 2, 2),
("notification_pref", 4, 4),
("push_token", 3, 3),
("push_token", 4, 4),
("story_view", 2, 2),
("space_invite", 2, 2),
("oauth_key", 1, 1),
crates/grain-serve/src/index/mod.rs+1−0
View file
@@ -17,6 +17,7 @@ pub mod backfill;
#[cfg(test)]
pub mod plan;
pub mod preferences;
pub mod push;
pub mod records;
pub mod schema;
pub mod sessions;
crates/grain-serve/src/index/push.rsnew+125−0
View file
@@ -0,0 +1,125 @@
//! The devices each account's pushes go to (`push_token`), as
//! `registerPush` and `unregisterPush` keep them ([`crate::push`]): a
//! token and the platform whose service delivers to it, a row each of an
//! account's devices. Nothing rebuilds them from the network.
use super::Index;
use rusqlite::params;
/// A device an account's pushes go to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PushToken {
/// What the platform's service names the device by.
pub token: String,
/// `ios`, `android` or `web`.
pub platform: String,
}
const TOKENS_SQL: &str = "SELECT token, platform FROM push_token WHERE did = ?1 ORDER BY token";
const DROP_SQL: &str = "DELETE FROM push_token WHERE token = ?1";
impl Index {
/// Keeps `token`, a device on `platform`, as one of `did`'s, made at
/// `now`. One it keeps already is kept on `platform` from then on, as
/// made when it was first.
pub fn register_push(
&self,
did: &str,
token: &str,
platform: &str,
now: &str,
) -> rusqlite::Result<()> {
let db = self.db.lock().unwrap();
db.execute(
"INSERT INTO push_token (did, token, platform, created_at) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (did, token) DO UPDATE SET platform = excluded.platform",
params![did, token, platform, now],
)?;
Ok(())
}
/// Stops `did`'s pushes going to `token`; whether they went there.
pub fn unregister_push(&self, did: &str, token: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let sql = "DELETE FROM push_token WHERE did = ?1 AND token = ?2";
Ok(db.execute(sql, [did, token])? > 0)
}
/// The devices `did`'s pushes go to.
pub fn push_tokens(&self, did: &str) -> rusqlite::Result<Vec<PushToken>> {
let db = self.db.lock().unwrap();
let mut stmt = db.prepare_cached(TOKENS_SQL)?;
let tokens = stmt.query_map([did], |r| {
Ok(PushToken {
token: r.get(0)?,
platform: r.get(1)?,
})
})?;
tokens.collect()
}
/// Forgets `token` for every account it was kept for, as when its
/// platform's service says no device answers to it any more; for how
/// many.
pub fn drop_push_token(&self, token: &str) -> rusqlite::Result<usize> {
let db = self.db.lock().unwrap();
db.execute(DROP_SQL, [token])
}
}
#[cfg(test)]
mod tests {
use super::super::plan::{scans, sized};
use super::*;
fn token(token: &str, platform: &str) -> PushToken {
PushToken {
token: token.into(),
platform: platform.into(),
}
}
#[test]
fn devices_are_kept_for_their_accounts() {
let index = Index::memory().unwrap();
let (a, b) = ("did:plc:a", "did:plc:b");
assert_eq!(index.push_tokens(a).unwrap(), []);
index.register_push(a, "t1", "ios", "2026-01-01").unwrap();
index
.register_push(a, "t2", "android", "2026-01-02")
.unwrap();
index.register_push(b, "t1", "ios", "2026-01-03").unwrap();
// Registered again, it keeps when it was first.
index.register_push(a, "t1", "web", "2026-01-04").unwrap();
let tokens = index.push_tokens(a).unwrap();
assert_eq!(tokens, [token("t1", "web"), token("t2", "android")]);
let made: String = index.read(|db| {
let sql = "SELECT created_at FROM push_token WHERE did = ?1 AND token = 't1'";
db.query_row(sql, [a], |r| r.get(0)).unwrap()
});
assert_eq!(made, "2026-01-01");
// One account's unregistering leaves another's alone.
assert!(index.unregister_push(a, "t1").unwrap());
assert!(!index.unregister_push(a, "t1").unwrap());
assert_eq!(index.push_tokens(a).unwrap(), [token("t2", "android")]);
assert_eq!(index.push_tokens(b).unwrap(), [token("t1", "ios")]);
// A token no device answers to is dropped for every account.
index.register_push(a, "t1", "ios", "2026-01-05").unwrap();
assert_eq!(index.drop_push_token("t1").unwrap(), 2);
assert_eq!(index.drop_push_token("t1").unwrap(), 0);
assert_eq!(index.push_tokens(b).unwrap(), []);
}
#[test]
fn devices_are_found_by_index() {
let index = Index::memory().unwrap();
index.read(|db| {
sized(db);
for sql in [TOKENS_SQL, DROP_SQL] {
assert_eq!(scans(db, sql), Vec::<String>::new(), "{sql}");
}
});
}
}
crates/grain-serve/src/index/tests.rs+0−1
View file
@@ -463,7 +463,6 @@ fn the_queries_still_to_port_read_by_index() {
"SELECT * FROM report r WHERE r.status = ?1 ORDER BY r.created_at DESC LIMIT 50",
"SELECT * FROM label l WHERE l.src = ?1 AND l.id > ?2 ORDER BY l.id LIMIT 100",
"DELETE FROM story_view WHERE created_at < ?1",
"SELECT did FROM push_token t WHERE t.token = ?1",
// backfill's queue, and the revisions it holds records to
"SELECT did FROM repo r WHERE r.state = 'pending' AND r.retry_after <= ?1
AND r.host_status IS NULL ORDER BY r.retry_after LIMIT ?2",
crates/grain-serve/src/push/mod.rsnew+105−0
View file
@@ -0,0 +1,105 @@
//! Push: the devices an account has its notifications pushed to, which
//! `social.grain.notification.registerPush` and `unregisterPush` keep
//! ([`crate::index::push`]), taken as Bluesky's
//! `app.bsky.notification.registerPush` takes them.
//!
//! The apps call them with a service token for grain-serve's DID, which
//! `serviceDid` names too: a device registered for another service is
//! refused rather than kept for pushes this one would send. `platform` is
//! `ios`, `android` or `web`, and `appId` the app's bundle or package id.
use crate::App;
use crate::index::Index;
use crate::xrpc::Failure;
use axum::Json;
use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use serde::Deserialize;
use serde_json::{Value, json};
use std::sync::Arc;
/// The methods, as a service token for each names it in `lxm`.
pub const REGISTER: &str = "social.grain.notification.registerPush";
pub const UNREGISTER: &str = "social.grain.notification.unregisterPush";
/// The platforms a device registers on.
pub const PLATFORMS: [&str; 3] = ["ios", "android", "web"];
/// The longest token kept, in bytes: an APNs token is 64 hex digits and
/// an FCM one under 200 characters, and a web push subscription, as JSON,
/// some hundreds.
pub const MOST_TOKEN_BYTES: usize = 4096;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Device {
service_did: String,
token: String,
platform: String,
#[allow(dead_code)]
app_id: String,
}
/// The index, the account calling `lxm` and the device it names, or why
/// not.
async fn device<'a>(
app: &'a App,
headers: &HeaderMap,
body: &[u8],
lxm: &str,
) -> Result<(&'a Index, String, Device), Failure> {
let Some(index) = &app.index else {
let why = "this server keeps no index";
return Err(Failure::new(
StatusCode::NOT_IMPLEMENTED,
"MethodNotImplemented",
why,
));
};
let did = crate::viewer::writer(app, headers, lxm).await?;
let device: Device = serde_json::from_slice(body).map_err(|e| {
Failure::invalid(format!(
"the input is not {{serviceDid, token, platform, appId}}: {e}"
))
})?;
if app.did.as_deref() != Some(device.service_did.as_str()) {
let why = format!("serviceDid {} is not this server", device.service_did);
return Err(Failure::invalid(why));
}
if !PLATFORMS.contains(&device.platform.as_str()) {
let why = format!("platform is ios, android or web, not {}", device.platform);
return Err(Failure::invalid(why));
}
if device.token.is_empty() || device.token.len() > MOST_TOKEN_BYTES {
let why = format!("token is 1 to {MOST_TOKEN_BYTES} bytes");
return Err(Failure::invalid(why));
}
Ok((index, did, device))
}
/// `social.grain.notification.registerPush`: the caller's pushes sent to
/// the device from then on.
pub async fn register(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, Failure> {
let (index, did, device) = device(&app, &headers, &body, REGISTER).await?;
let now = grain::keys::now_iso();
index.register_push(&did, &device.token, &device.platform, &now)?;
Ok(Json(json!({})))
}
/// `social.grain.notification.unregisterPush`: the caller's pushes no
/// longer sent to the device, as when it logs out there. One it never
/// registered is no failure.
pub async fn unregister(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, Failure> {
let (index, did, device) = device(&app, &headers, &body, UNREGISTER).await?;
index.unregister_push(&did, &device.token)?;
Ok(Json(json!({})))
}
crates/grain-serve/src/lib.rs+10−4
View file
@@ -1,9 +1,9 @@
//! grain's server. It answers atproto's health check,
//! `social.grain.sync.indexRecords` ([`sync`]), the queries that read
//! the index alone ([`read`]) and what an account has seen of its
//! notifications ([`notification`]), and logs the web in ([`login`]); the
//! web's writes and push join it pull by pull on `feature/rust`, as the
//! plan orders them.
//! the index alone ([`read`]), what an account has seen of its
//! notifications ([`notification`]) and the devices they are pushed to
//! ([`push`]), and the web's writes ([`write`]), and logs the web in
//! ([`login`]).
pub mod backfill;
pub mod cdn;
@@ -12,6 +12,7 @@ pub mod jetstream;
pub mod login;
pub mod notification;
pub mod preferences;
pub mod push;
pub mod read;
pub mod sync;
pub mod viewer;
@@ -198,6 +199,11 @@ pub fn router(app: Arc<App>) -> Router {
&format!("/xrpc/{}", notification::UPDATE_SEEN),
post(notification::update_seen),
)
.route(&format!("/xrpc/{}", push::REGISTER), post(push::register))
.route(
&format!("/xrpc/{}", push::UNREGISTER),
post(push::unregister),
)
.route(
&format!("/xrpc/{}", preferences::GET),
get(preferences::get_preferences),
crates/grain-serve/tests/push.rsnew+147−0
View file
@@ -0,0 +1,147 @@
//! The devices an account's notifications are pushed to, registered as
//! the apps register them: with a service token from the account's own
//! PDS, the reference PDS. Every answer goes through `common`, so
//! `harness/answers.mjs` holds it to its lexicon.
#[path = "../../grain/tests/common/mod.rs"]
mod net;
mod common;
use grain::env::Ctx;
use grain::session::{self, Session};
use grain_serve::App;
use grain_serve::cdn::Cdn;
use grain_serve::index::Index;
use grain_serve::index::push::PushToken;
use serde_json::{Value, json};
use std::sync::Arc;
const AUD: &str = "did:web:grain.test";
const REGISTER: &str = "/xrpc/social.grain.notification.registerPush";
const UNREGISTER: &str = "/xrpc/social.grain.notification.unregisterPush";
struct World {
ctx: Ctx,
me: Session,
index: Arc<Index>,
served: common::Served,
_dir: tempfile::TempDir,
}
async fn world() -> World {
let dir = tempfile::tempdir().unwrap();
let ctx = net::ctx(dir.path(), &[]);
let index = Arc::new(Index::memory().unwrap());
let served = common::served_app(App {
ctx: net::ctx(&dir.path().join("serve"), &[]),
index: Some(index.clone()),
did: Some(AUD.into()),
cdn: Cdn::default(),
bsky_appview: None,
web: None,
})
.await;
let account = net::account().await;
let me = session::create(&ctx, &account.did, &account.password)
.await
.unwrap();
World {
ctx,
me,
index,
served,
_dir: dir,
}
}
/// A device, as the apps name it.
fn device(token: &str, platform: &str) -> Value {
json!({"serviceDid": AUD, "token": token, "platform": platform, "appId": "social.grain"})
}
fn kept(token: &str, platform: &str) -> PushToken {
PushToken {
token: token.into(),
platform: platform.into(),
}
}
impl World {
/// `path`, a procedure, called with a service token for its method.
async fn post_as(&self, path: &str, body: &Value) -> (u16, Value) {
let nsid = &path["/xrpc/".len()..];
let token = grain::auth::service_token(&self.ctx, &self.me, AUD, nsid)
.await
.unwrap();
self.served.post_as(path, &token, body).await
}
fn tokens(&self) -> Vec<PushToken> {
self.index.push_tokens(&self.me.did).unwrap()
}
}
#[tokio::test]
async fn a_device_is_registered_and_unregistered() {
let w = world().await;
let ok = (200, json!({}));
assert_eq!(w.post_as(REGISTER, &device("aa11", "ios")).await, ok);
assert_eq!(w.post_as(REGISTER, &device("fcm-1", "android")).await, ok);
// Again, as the apps register on every launch: kept once.
assert_eq!(w.post_as(REGISTER, &device("aa11", "ios")).await, ok);
assert_eq!(w.tokens(), [kept("aa11", "ios"), kept("fcm-1", "android")]);
assert_eq!(w.post_as(UNREGISTER, &device("aa11", "ios")).await, ok);
assert_eq!(w.tokens(), [kept("fcm-1", "android")]);
// One it never registered, or no longer has, is no failure.
assert_eq!(w.post_as(UNREGISTER, &device("aa11", "ios")).await, ok);
// Another account's device is not the caller's to unregister.
w.index
.register_push("did:plc:other", "web-1", "web", "2026-01-01T00:00:00.000Z")
.unwrap();
assert_eq!(w.post_as(UNREGISTER, &device("web-1", "web")).await, ok);
let other = w.index.push_tokens("did:plc:other").unwrap();
assert_eq!(other, [kept("web-1", "web")]);
}
#[tokio::test]
async fn a_device_is_named_in_full_by_an_account() {
let w = world().await;
for path in [REGISTER, UNREGISTER] {
let (status, body) = w.served.post(path, &device("aa11", "ios")).await;
assert_eq!(
(status, body["error"].as_str()),
(401, Some("AuthenticationRequired")),
"{path}"
);
}
let too_long = "a".repeat(grain_serve::push::MOST_TOKEN_BYTES + 1);
let mut elsewhere = device("aa11", "ios");
elsewhere["serviceDid"] = "did:web:api.bsky.app".into();
for wrong in [
json!({}),
json!({"token": "aa11", "platform": "ios", "appId": "social.grain"}),
elsewhere,
device("aa11", "windows"),
device("", "ios"),
device(&too_long, "ios"),
] {
for path in [REGISTER, UNREGISTER] {
let (status, body) = w.post_as(path, &wrong).await;
assert_eq!(
(status, body["error"].as_str()),
(400, Some("InvalidRequest")),
"{path} {wrong}"
);
}
}
assert_eq!(w.tokens(), []);
// A server with no index keeps no devices.
let bare = common::served().await;
let (status, body) = bare.post(REGISTER, &device("aa11", "ios")).await;
assert_eq!(
(status, body["error"].as_str()),
(501, Some("MethodNotImplemented"))
);
}