~ 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 pushes each new record's notifications

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

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/mod.rs+4−0
View file
@@ -37,6 +37,9 @@ pub struct Index {
/// The accounts in `repo`, kept here too: the stream asks of every
/// Bluesky post on the network whether its author is one.
tracked: Mutex<HashSet<String>>,
/// Where the records live paths newly put are told, for push
/// ([`Index::arrived`]).
arrivals: std::sync::OnceLock<tokio::sync::mpsc::Sender<String>>,
}
/// What putting a record did.
@@ -200,6 +203,7 @@ impl Index {
db: Mutex::new(db),
lexicons: Lexicons::grain(),
tracked: Mutex::new(tracked),
arrivals: std::sync::OnceLock::new(),
})
}
crates/grain-serve/src/index/push.rs+26−0
View file
@@ -2,9 +2,15 @@
//! `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.
//!
//! The paths that put what accounts write as they write it (the stream,
//! `indexRecords` and the web's writes) say which records were new to the
//! index ([`Index::arrived`]), for the sender ([`crate::push::send`]) to
//! push. Backfill and the import do not: what they put is old news.
use super::Index;
use rusqlite::params;
use tokio::sync::mpsc::{Sender, error::TrySendError};
/// A device an account's pushes go to.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -19,6 +25,26 @@ const TOKENS_SQL: &str = "SELECT token, platform FROM push_token WHERE did = ?1
const DROP_SQL: &str = "DELETE FROM push_token WHERE token = ?1";
impl Index {
/// Has the records [`Index::arrived`] names sent to `to`, from then
/// on; whether none was named before.
pub fn push_to(&self, to: Sender<String>) -> bool {
self.arrivals.set(to).is_ok()
}
/// Says that a live path put the record at `uri`, which the index did
/// not hold, for it to be pushed. Where the sender is behind by
/// [`crate::push::send::QUEUED`] records, it is not.
pub fn arrived(&self, uri: &str) {
let Some(to) = self.arrivals.get() else {
return;
};
match to.try_send(uri.to_string()) {
Ok(()) => {}
Err(TrySendError::Full(_)) => tracing::warn!(%uri, "push: behind, not pushed"),
Err(TrySendError::Closed(_)) => tracing::warn!(%uri, "push: stopped, not pushed"),
}
}
/// 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.
crates/grain-serve/src/index/schema.rs+14−1
View file
@@ -87,6 +87,13 @@
//! A migration may drop and remake a record table, which backfill refills;
//! these it only alters in place.
//!
//! # Push
//!
//! `push_record` is each record whose notifications were pushed, and when,
//! so that one the stream delivers again, or deletes and makes again, is
//! not pushed twice ([`crate::push::send`]). A row is kept for a day,
//! past which a record is too old to push anyway.
//!
//! # Indexes
//!
//! Each serves a query grain runs: the feeds (recent, following, actor,
@@ -96,7 +103,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, V5, V6, V7, V8];
pub const MIGRATIONS: &[&str] = &[V1, V2, V3, V4, V5, V6, V7, V8, V9];
const V1: &str = "
CREATE TABLE gallery (
@@ -469,3 +476,9 @@ pub fn unhidden(uri: &str) -> String {
const V8: &str = "
CREATE TABLE carried_cookie_key (key BLOB NOT NULL CHECK (length(key) = 32));
";
/// The records that have been pushed, so that none is pushed twice.
const V9: &str = "
CREATE TABLE push_record (uri TEXT PRIMARY KEY, at TEXT NOT NULL) WITHOUT ROWID;
CREATE INDEX push_record_at ON push_record (at);
";
crates/grain-serve/src/push/mod.rs+3−1
View file
@@ -11,10 +11,12 @@
//! A push goes by the platform's own service: [`apns`] for `ios`, [`fcm`]
//! for `android`, each set up on its own, so a server with only one sends
//! to its devices and leaves the others' tokens be. `web` devices are
//! kept, but nothing is pushed to them.
//! kept, but nothing is pushed to them. [`send`] pushes what each record
//! new to the index notifies.
pub mod apns;
pub mod fcm;
pub mod send;
use crate::App;
use crate::index::Index;
crates/grain-serve/src/push/send.rsnew+258−0
View file
@@ -0,0 +1,258 @@
//! The sender: a task that pushes the notifications each record new to
//! the index gives rise to, off the path that put it.
//!
//! A live path says a record arrived ([`Index::arrived`]); the sender
//! then, in turn:
//!
//! - leaves it be if it says it was made more than [`RECENT`] ago, or
//! says no time it was made: an account's old records put again, as a
//! repo moved to a new host is, are not news;
//! - keeps that it was pushed (`push_record`), and leaves it be if it
//! was already, as one deleted and made again is;
//! - finds whom it notifies, where they want it pushed
//! ([`notification::arising`]), which leaves out an author taken down,
//! blocking or blocked, or muted, and a record a label hides;
//! - makes each a [`Message`]: what the author did, by their profile's
//! name, where a tap goes, and the recipient's unread count as the
//! badge;
//! - pushes it to each of the recipient's devices, each on its own task,
//! trying one again [`Retry::tries`] times in all; a device its service
//! says is gone is forgotten.
use super::{Message, Outcome, Transports};
use crate::index::{Index, record_uri, records};
use crate::read::notification;
use rusqlite::{Connection, OptionalExtension};
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
/// How many records may wait to be pushed; past that, one arriving is not.
pub const QUEUED: usize = 10_000;
/// How long ago a record may say it was made and still be pushed.
pub const RECENT: &str = "-1 hour";
/// How long a pushed record is kept as pushed.
pub const KEPT: &str = "-1 day";
/// The tables of the records that notify anyone ([`notification`]'s
/// arms).
pub const NOTIFYING: [&str; 4] = ["favorite", "follow", "comment", "gallery"];
/// How a push that may go if tried again is tried.
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct Retry {
/// How many times in all.
pub tries: u32,
/// The wait before the second, doubled before each after.
pub wait: Duration,
}
impl Default for Retry {
fn default() -> Retry {
Retry {
tries: 4,
wait: Duration::from_secs(2),
}
}
}
/// What finds the pushes of a record: [`pushes`], but for tests.
type Find = fn(&Connection, &str) -> rusqlite::Result<Vec<(String, Message)>>;
/// Starts pushing what `index` says arrives, by `transports`, on the
/// runtime this is called on.
pub fn start(index: Arc<Index>, transports: Arc<Transports>, retry: Retry) {
start_finding(index, transports, retry, pushes)
}
/// [`start`], finding each record's pushes with `find`. One whose finding
/// panics is logged and left, and the next is pushed as ever.
pub(crate) fn start_finding(
index: Arc<Index>,
transports: Arc<Transports>,
retry: Retry,
find: Find,
) {
let (to, mut from) = mpsc::channel::<String>(QUEUED);
if !index.push_to(to) {
tracing::warn!("push: already started");
return;
}
tokio::spawn(async move {
while let Some(uri) = from.recv().await {
// Off the runtime's threads, as it holds the index's lock; and
// a panic is caught while it does, so the lock is not poisoned.
let found = (index.clone(), uri.clone());
let pushes = tokio::task::spawn_blocking(move || {
let (index, uri) = found;
index.read(|db| {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| find(db, &uri)))
})
});
let pushes = match pushes.await {
Ok(Ok(Ok(pushes))) => pushes,
Ok(Ok(Err(e))) => {
tracing::error!(%uri, error = %e, "push: the index failed");
continue;
}
// Caught, or past catching, as on a lock already poisoned.
Ok(Err(_)) | Err(_) => {
tracing::error!(%uri, "push: finding what to push panicked; not pushed");
continue;
}
};
for (did, message) in pushes {
let tokens = match index.push_tokens(&did) {
Ok(tokens) => tokens,
Err(e) => {
tracing::error!(%did, error = %e, "push: the index failed");
continue;
}
};
for device in tokens {
let (index, transports) = (index.clone(), transports.clone());
let (did, message) = (did.clone(), message.clone());
tokio::spawn(async move {
deliver(&index, &transports, retry, &did, &device, &message).await
});
}
}
}
});
}
/// Pushes `message` to `device`, one of `did`'s, as [`Retry`] says.
async fn deliver(
index: &Index,
transports: &Transports,
retry: Retry,
did: &str,
device: &crate::index::push::PushToken,
message: &Message,
) {
let platform = device.platform.as_str();
let mut wait = retry.wait;
for tried in 1..=retry.tries {
let Some(outcome) = transports.send(platform, &device.token, message).await else {
return;
};
match outcome {
Outcome::Sent => {
tracing::info!(%did, platform, "push: sent");
return;
}
Outcome::Gone => {
if let Err(e) = index.drop_push_token(&device.token) {
tracing::error!(%did, error = %e, "push: the index failed");
}
tracing::info!(%did, platform, "push: the device is gone, and forgotten");
return;
}
Outcome::Failed(why) => {
tracing::warn!(%did, platform, %why, "push: failed");
return;
}
Outcome::Retry(why) if tried == retry.tries => {
tracing::warn!(%did, platform, %why, tried, "push: failed, and not tried again");
}
Outcome::Retry(why) => {
tracing::info!(%did, platform, %why, tried, "push: to try again");
tokio::time::sleep(wait).await;
wait *= 2;
}
}
}
}
/// Whether the record at `uri`, in `table`, says it was made within
/// [`RECENT`]; or later, as a clock set ahead says.
const RECENT_SQL: &str = "SELECT strftime('%Y-%m-%dT%H:%M:%fZ', created_at)
>= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?2) FROM {table} WHERE uri = ?1";
const PUSHED_SQL: &str = "INSERT OR IGNORE INTO push_record VALUES (?1, ?2)";
const FORGET_SQL: &str =
"DELETE FROM push_record WHERE at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?1)";
const NAME_SQL: &str = "SELECT display_name FROM profile WHERE did = ?1 AND display_name <> ''";
const SUBJECT_SQL: &str = "SELECT subject FROM favorite WHERE uri = ?1
UNION ALL SELECT subject FROM comment WHERE uri = ?1";
/// What the record at `uri` pushes, to whom: none if it is not recent,
/// was pushed already, or notifies no one who wants it pushed.
pub fn pushes(db: &Connection, uri: &str) -> rusqlite::Result<Vec<(String, Message)>> {
let Some((author, collection, _)) = record_uri(uri) else {
return Ok(Vec::new());
};
let Some(table) = records::table_of(collection).filter(|t| NOTIFYING.contains(t)) else {
return Ok(Vec::new());
};
let recent: Option<Option<bool>> = db
.query_row(&RECENT_SQL.replace("{table}", table), (uri, RECENT), |r| {
r.get(0)
})
.optional()?;
if recent.flatten() != Some(true) {
return Ok(Vec::new());
}
db.execute(FORGET_SQL, [KEPT])?;
if db.execute(PUSHED_SQL, (uri, grain::keys::now_iso()))? == 0 {
return Ok(Vec::new());
}
let arising = notification::arising(db, uri)?;
if arising.is_empty() {
return Ok(Vec::new());
}
let name: Option<String> = db.query_row(NAME_SQL, [author], |r| r.get(0)).optional()?;
let name = name.as_deref().unwrap_or("Someone");
let subject: Option<String> = db.query_row(SUBJECT_SQL, [uri], |r| r.get(0)).optional()?;
let mut pushes = Vec::new();
for (recipient, reason) in arising {
let Some(mut message) = message(&reason, name) else {
continue;
};
let about = match reason.as_str() {
"follow" => ("did", author),
"gallery-mention" => ("uri", uri),
_ => ("uri", subject.as_deref().unwrap_or(uri)),
};
message.data.insert(about.0.into(), about.1.into());
message.badge = Some(notification::unread_count(db, &recipient)?);
pushes.push((recipient, message));
}
Ok(pushes)
}
/// The notification `reason` names, of something `name` did: its title,
/// body and `type`, without what it is about or the badge.
pub fn message(reason: &str, name: &str) -> Option<Message> {
let (title, body, kind) = match reason {
"gallery-favorite" => ("New favorite", "favorited your gallery", reason),
"story-favorite" => ("New favorite", "favorited your story", reason),
"comment-favorite" => ("New favorite", "favorited your comment", reason),
"follow" => ("New follower", "followed you", reason),
"reply" => ("New reply", "replied to your comment", "comment-reply"),
"gallery-comment" => ("New comment", "commented on your gallery", reason),
"story-comment" => ("New comment", "commented on your story", reason),
"gallery-comment-mention" => ("New mention", "mentioned you in a comment", reason),
"gallery-mention" => ("New mention", "mentioned you in a gallery", reason),
_ => return None,
};
Some(Message {
title: title.into(),
body: format!("{name} {body}"),
data: BTreeMap::from([("type".to_string(), kind.to_string())]),
badge: None,
})
}
#[cfg(test)]
pub fn queries() -> Vec<String> {
let mut all: Vec<String> = NOTIFYING
.iter()
.map(|table| RECENT_SQL.replace("{table}", table))
.collect();
all.extend([PUSHED_SQL, FORGET_SQL, NAME_SQL, SUBJECT_SQL].map(str::to_string));
all
}
crates/grain-serve/src/push/tests.rs+493−0
View file
493 lines changed: not loaded until asked for.
crates/grain-serve/src/read/tests.rs+1−0
View file
@@ -15,6 +15,7 @@ fn the_reads_read_by_index() {
queries.extend(views::queries());
queries.extend(search::queries());
queries.extend(notification::queries());
queries.extend(crate::push::send::queries());
index.read(|db| {
sized(db);
for sql in &queries {
crates/grain-serve/src/jetstream.rs+3−1
View file
@@ -13,6 +13,7 @@
//! stream write and a backfill's read of the same repo leave the later of
//! the two. An account newly tracked, or back after its host said it was
//! gone, is queued for backfill ([`Index::resync`]) to be read whole.
//! A record new to the index is pushed ([`Index::arrived`]).
//!
//! Where it was read to is kept in the index, under `jetstream:<url>`,
//! and a restart resumes there. A cursor
@@ -20,7 +21,7 @@
//! ([`grain::jetstream::PROBE_EVERY`]).
use crate::backfill::SIGNALS;
use crate::index::{Index, PutError, records};
use crate::index::{Applied, Index, PutError, records};
use grain::jetstream::{Event, Follower, Timing, What};
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -140,6 +141,7 @@ pub fn apply(index: &Index, event: Event) -> rusqlite::Result<()> {
None => index.put(&uri, &cid, &record),
};
match put {
Ok(Applied::Created) => index.arrived(&uri),
Ok(_) => {}
Err(PutError::Refused(why)) => {
tracing::info!(%uri, %why, "jetstream: skipped a record");
crates/grain-serve/src/lib.rs+57−0
View file
@@ -60,6 +60,10 @@ pub struct Settings {
/// web carries the cookies browsers held before grain-serve
/// ([`login::carryover`]). None carries none.
pub carry_cookies_until: Option<String>,
/// `GRAIN_APNS_*` and `GRAIN_FCM_*` ([`push::Config::from_vars`]):
/// the services notifications are pushed by. With neither, as tests
/// and dev run, none is pushed.
pub push: push::Config,
}
impl Settings {
@@ -82,6 +86,7 @@ impl Settings {
},
sign_up_pds: set("GRAIN_SIGN_UP_PDS"),
carry_cookies_until: set("GRAIN_CARRY_COOKIES_UNTIL"),
push: push::Config::from_vars(&get),
}
}
@@ -112,6 +117,26 @@ impl Settings {
Ok(true)
}
/// Starts pushing what arrives in `app`'s index by the services these
/// settings name, on the runtime this is called on; whether any is.
pub fn push(&self, app: &App) -> anyhow::Result<bool> {
if self.push == push::Config::default() {
return Ok(false);
}
let Some(index) = &app.index else {
bail!("GRAIN_APNS_KEY_FILE and GRAIN_FCM_KEY_FILE need GRAIN_INDEX");
};
let transports = push::Transports::new(&self.push)?;
tracing::info!(
apns = transports.apns.is_some(),
fcm = transports.fcm.is_some(),
"pushing"
);
let retry = push::send::Retry::default();
push::send::start(index.clone(), Arc::new(transports), retry);
Ok(true)
}
/// The server `ctx` and these settings make: its index opened, and
/// its DID named.
pub fn app(&self, ctx: Ctx) -> anyhow::Result<App> {
@@ -320,6 +345,38 @@ mod tests {
assert_eq!(set.sign_up_pds.as_deref(), Some("https://pds.test"));
}
#[tokio::test]
async fn push_starts_where_a_service_is_named() {
let ctx = || Ctx::new(Env::from_vars(|_| None));
let none = Settings::default();
assert!(!none.push(&none.app(ctx()).unwrap()).unwrap());
let dir = tempfile::tempdir().unwrap();
let named = |index: bool, key_file: &str| Settings {
index: index.then(|| dir.path().join("index.db")),
push: push::Config {
apns: None,
fcm: Some(push::fcm::Config {
key_file: dir.path().join(key_file),
project_id: Some("grain-test".into()),
url: push::fcm::URL.into(),
}),
},
..Settings::default()
};
let unindexed = named(false, "none.json");
let why = unindexed.push(&unindexed.app(ctx()).unwrap()).unwrap_err();
assert!(why.to_string().ends_with("need GRAIN_INDEX"), "{why}");
let unreadable = named(true, "none.json");
assert!(unreadable.push(&unreadable.app(ctx()).unwrap()).is_err());
let account = serde_json::json!({
"client_email": "push@grain-test.iam.gserviceaccount.com",
"private_key": include_str!("push/test-rsa-key.pem"),
});
std::fs::write(dir.path().join("account.json"), account.to_string()).unwrap();
let set = named(true, "account.json");
assert!(set.push(&set.app(ctx()).unwrap()).unwrap());
}
#[test]
fn the_app_opens_its_index_and_names_its_did() {
let ctx = || Ctx::new(Env::from_vars(|_| None));
crates/grain-serve/src/main.rs+3−1
View file
@@ -1,6 +1,7 @@
//! `grain-serve`: listens on `GRAIN_LISTEN`, 127.0.0.1:2600 unless set,
//! keeping the index at `GRAIN_INDEX`, backfilled when `GRAIN_BACKFILL` is
//! set and fed from `GRAIN_JETSTREAM` when that is, and taking service
//! set and fed from `GRAIN_JETSTREAM` when that is, pushing by APNs and
//! FCM when `GRAIN_APNS_*` and `GRAIN_FCM_*` say, and taking service
//! tokens for the DID of `GRAIN_PUBLIC_URL` when set
//! ([`grain_serve::Settings`]).
@@ -18,6 +19,7 @@ async fn main() -> anyhow::Result<()> {
if let Some(backfill) = settings.backfill(&app)? {
tokio::spawn(backfill.run());
}
settings.push(&app)?;
settings.follow(&app)?;
let addr = std::env::var("GRAIN_LISTEN").unwrap_or_else(|_| "127.0.0.1:2600".into());
let listener = tokio::net::TcpListener::bind(&addr)
crates/grain-serve/src/sync.rs+9−3
View file
@@ -7,14 +7,15 @@
//! names URIs, never content: each is read with `com.atproto.repo.getRecord`
//! from the PDS the repo's DID document names, and put by its URI and CID,
//! so Jetstream delivering the same record later changes nothing. A record
//! the PDS no longer has is deleted from the index.
//! the PDS no longer has is deleted from the index. One new to the index
//! is pushed ([`Index::arrived`]).
//!
//! Whoever holds a service token names the PDS asked, by their DID
//! document, so it is asked only as [`grain::guard`] allows: on https, at
//! a public address, following no redirect.
use crate::App;
use crate::index::{Index, PutError, record_uri, records};
use crate::index::{Applied, Index, PutError, record_uri, records};
use crate::xrpc::Failure;
use axum::Json;
use axum::body::Bytes;
@@ -148,7 +149,12 @@ pub async fn read_back(
for ((uri, _, _), read) in named.into_iter().zip(read) {
let applied = match read {
Ok(Some((cid, record))) => match index.put(uri, &cid, &record) {
Ok(_) => Ok(&mut done.indexed),
Ok(applied) => {
if applied == Applied::Created {
index.arrived(uri);
}
Ok(&mut done.indexed)
}
Err(PutError::Refused(why)) => Err(ReadBackError::Refused {
uri: uri.to_string(),
why: why.to_string(),
crates/grain-serve/src/write.rs+5−3
View file
crates/grain-serve/tests/index_records.rs+7−0
View file
crates/grain-serve/tests/write.rs+6−0
View file