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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

The web proxy, deletes and the cross-post test land together, checked once

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:batch/two

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+31−1
View file
@@ -135,6 +135,21 @@ pub fn row<T: Record>(r: &rusqlite::Row) -> rusqlite::Result<Row<T>> {
})
}
/// What grain keeps for an account, which goes with its records
/// ([`Index::purge`]): its web session, preferences, how and when it
/// reads its notifications, the devices they are pushed to, whom it
/// muted, the stories it watched, and its invitations to spaces.
pub const KEPT_FOR: &[&str] = &[
"DELETE FROM web_session WHERE did = ?1",
"DELETE FROM preference WHERE did = ?1",
"DELETE FROM notification_seen WHERE did = ?1",
"DELETE FROM notification_pref WHERE did = ?1",
"DELETE FROM push_token WHERE did = ?1",
"DELETE FROM mute WHERE did = ?1",
"DELETE FROM story_view WHERE did = ?1",
"DELETE FROM space_invite WHERE member_did = ?1 OR author_did = ?1",
];
/// The author, collection and key of a record's AT-URI.
pub fn record_uri(uri: &str) -> Option<(&str, &str, &str)> {
let mut parts = uri.strip_prefix("at://")?.split('/');
@@ -408,6 +423,15 @@ impl Index {
Ok(added > 0)
}
/// Stops tracking the account `did`, which deleted what it had here:
/// its row in `repo` goes, handle and all, until it writes again.
pub fn untrack(&self, did: &str) -> rusqlite::Result<bool> {
let db = self.db.lock().unwrap();
let gone = db.execute("DELETE FROM repo WHERE did = ?1", [did])?;
self.tracked.lock().unwrap().remove(did);
Ok(gone > 0)
}
/// Whether the account `did` is tracked.
pub fn tracks(&self, did: &str) -> bool {
self.tracked.lock().unwrap().contains(did)
@@ -449,7 +473,10 @@ impl Index {
Ok(())
}
/// Deletes every record `did` wrote; how many.
/// Deletes every record `did` wrote, and what grain keeps for it
/// ([`KEPT_FOR`]); how many records. What an admin holds of it stays:
/// its takedown, the labels given out and the reports made of it, as
/// do the mutes others made of it, which are theirs.
pub fn purge(&self, did: &str) -> rusqlite::Result<usize> {
let mut db = self.db.lock().unwrap();
let tx = db.transaction()?;
@@ -467,6 +494,9 @@ impl Index {
for (_, table) in records::COLLECTIONS {
purged += tx.execute(&format!("DELETE FROM {table} WHERE did = ?1"), [did])?;
}
for sql in KEPT_FOR {
tx.execute(sql, [did])?;
}
tx.commit()?;
Ok(purged)
}
crates/grain-serve/src/index/tests.rs+72−0
View file
@@ -631,3 +631,75 @@ fn a_repo_backfilled_replaces_what_it_no_longer_has() {
assert_eq!(index.forget_deletions(OLD).unwrap(), 0);
assert_eq!(index.forget_deletions(LATER).unwrap(), 1);
}
#[test]
fn a_purge_takes_what_grain_keeps_for_the_account_and_leaves_what_admins_hold() {
let index = Index::memory().unwrap();
let other = "did:plc:bob";
index.track(DID).unwrap();
index
.put(&uri("social.grain.gallery"), CID, &gallery("Fog"))
.unwrap();
let mine = [
"INSERT INTO web_session VALUES (?1, '{}', 0, 0)",
"INSERT INTO preference VALUES (?1, 'theme', '\"dark\"', '')",
"INSERT INTO notification_seen VALUES (?1, '')",
"INSERT INTO notification_pref VALUES (?1, 'follows', 1, 1, 0)",
"INSERT INTO push_token VALUES (?1, 'token', 'ios', '')",
"INSERT INTO mute VALUES (?1, 'did:plc:carol', '')",
"INSERT INTO story_view VALUES (?1, 'at://did:plc:carol/social.grain.story/s', '')",
"INSERT INTO space_invite VALUES ('ats://space', ?1, 'did:plc:carol', '')",
"INSERT INTO takedown VALUES (?1, '')",
"INSERT INTO report (subject_uri, subject_did, label, reported_by, created_at)
VALUES (?1, ?1, 'spam', 'did:plc:carol', '')",
"INSERT INTO label (src, uri, val, cts) VALUES ('did:web:grain.social', ?1, 'spam', '')",
];
index.read(|db| {
for sql in mine {
db.execute(sql, [DID]).unwrap();
db.execute(sql, [other]).unwrap();
}
// Another's mute of the account is theirs.
db.execute("INSERT INTO mute VALUES (?1, ?2, '')", [other, DID])
.unwrap();
});
assert_eq!(index.purge(DID).unwrap(), 1);
let count = |table: &str, did: &str| -> i64 {
let column = match table {
"space_invite" => "member_did",
"report" => "subject_did",
"label" => "uri",
_ => "did",
};
let sql = format!("SELECT count(*) FROM {table} WHERE {column} = ?1");
index.read(|db| db.query_row(&sql, [did], |r| r.get(0)).unwrap())
};
for table in [
"web_session",
"preference",
"notification_seen",
"notification_pref",
"push_token",
"mute",
"story_view",
"space_invite",
] {
assert_eq!(count(table, DID), 0, "{table}");
assert!(count(table, other) >= 1, "{table} of another");
}
for table in ["takedown", "report", "label"] {
assert_eq!(count(table, DID), 1, "{table} stays");
}
let muted: i64 = index.read(|db| {
db.query_row("SELECT count(*) FROM mute WHERE subject = ?1", [DID], |r| {
r.get(0)
})
.unwrap()
});
assert_eq!(muted, 1);
// Tracked still, as its host may say more of it, until it is untracked.
assert!(index.tracks(DID));
assert!(index.untrack(DID).unwrap());
assert!(!index.tracks(DID) && index.repo(DID).unwrap().is_none());
assert!(!index.untrack(DID).unwrap());
}
crates/grain-serve/src/read/tests.rs+1−0
View file
@@ -17,6 +17,7 @@ fn the_reads_read_by_index() {
queries.extend(sitemap::queries());
queries.extend(notification::queries());
queries.extend(crate::push::send::queries());
queries.push(crate::gallery::going_sql());
index.read(|db| {
sized(db);
for sql in &queries {
crates/grain-serve/src/accounts.rsnew+382−0
View file
@@ -0,0 +1,382 @@
//! Accounts that go, as knotless-serve takes them (its `accounts.rs` and
//! `web/account_data.rs`).
//!
//! **Deleted at its host.** Jetstream says an account was deleted; before
//! anything of it goes, the host its identity names now is asked whether
//! it still has it (`com.atproto.sync.getRepoStatus`): one that moved is
//! deleted from the host it left, and the host it moved to still answers
//! for it, so it stays ([`deletions`]). Only an answer that says so
//! counts as gone: the host has no such repo or says it was deleted, or the
//! identity is a tombstone or nothing at all ([`Asked::Gone`]). Where it
//! cannot be asked, PLC or the host down or answering oddly, it is asked
//! again later, and in the end kept ([`settle`]), since what grain keeps
//! for an account nothing would bring back. A deleted account loses what the
//! index holds from it, its records and what grain keeps for it
//! ([`Index::purge`]); what an admin holds of it stays, as do others'
//! records about it.
//!
//! **Deleting its grain data.** `social.grain.server.deleteAccount`, as
//! knotless's `dev.knotless.deleteAccountData` does. Asked by the web's
//! cookie, from a page of the site, grain-serve deletes the account's
//! records in every `social.grain.*` collection at its PDS itself, with
//! the session it keeps for it, up to `limit` a call, so the page can show
//! how far it has got and a run cut short carries on; once none is left,
//! it logs the browser out of the account, the session revoked at the PDS,
//! and forgets it: [`Index::purge`], and the account no longer tracked.
//! Asked with a service token, as the apps ask once they have deleted the
//! records at their own PDS, it deletes nothing, and forgets the account
//! only once its PDS lists no grain collection. The atproto account, and
//! other apps' records in it, a Bluesky cross-post among them, stay.
use crate::App;
use crate::index::Index;
use crate::login::upstream;
use crate::xrpc::Failure;
use axum::Json;
use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::{IntoResponse, Response};
use grain::env::Ctx;
use grain::xrpc::{Xrpc, XrpcError};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::sync::Arc;
use std::time::Duration;
/// The method, as a service token for it names it in `lxm`.
pub const DELETE_ACCOUNT: &str = "social.grain.server.deleteAccount";
/// What every grain collection's name starts with.
pub const PREFIX: &str = "social.grain.";
/// The records one `listRecords` reads, and one `applyWrites` deletes.
const PAGE: usize = 100;
/// The most records one call deletes unless it says, and the most it may.
const LIMIT: u64 = 400;
const MOST: u64 = 1000;
/// The order collections go in: what hangs off a gallery before the
/// gallery, and the profile last, so that what is left between calls
/// shows as little as it can. The rest go after, by name.
const ORDER: &[&str] = &[
"social.grain.photo.exif",
"social.grain.gallery.item",
"social.grain.photo",
"social.grain.gallery",
"social.grain.story",
"social.grain.favorite",
"social.grain.comment",
"social.grain.graph.follow",
"social.grain.graph.block",
];
const LAST: &str = "social.grain.actor.profile";
/// Takes in that `did` was deleted at its host: its records go, with
/// what grain keeps for it, and it is said to be deleted.
pub fn deleted(index: &Index, did: &str) -> rusqlite::Result<()> {
let purged = index.purge(did)?;
index.set_host_status(did, Some("deleted"))?;
tracing::info!(%did, purged, "an account was deleted");
Ok(())
}
/// What the host `did`'s identity names now says of it.
#[derive(Debug, Clone, PartialEq)]
pub enum Asked {
/// Gone: the host has no such repo (`RepoNotFound`) or says it was
/// deleted (`active: false`, `status: deleted`), or the identity itself
/// is gone (a PLC tombstone, `410`, or nothing there, `404`).
Gone,
/// The host has it: active, as it is where it moved to, or inactive in
/// a way that can be undone (deactivated, suspended, taken down).
Active,
/// Nothing that says either: the identity or the host could not be
/// read, or answered with what says neither. Why, for the log.
Unknown(String),
}
/// Asks after `did`, which Jetstream says was deleted, where its identity
/// says it lives now ([`Asked`]).
pub async fn gone(ctx: &Ctx, did: &str) -> Asked {
let doc = match grain::identity::document(ctx, did).await {
Ok(doc) => doc,
Err(e) if matches!(http_status(&e), Some(404 | 410)) => return Asked::Gone,
Err(e) => return Asked::Unknown(format!("{e:#}")),
};
let pds = match grain::identity::pds(ctx, did, &doc) {
Ok(pds) => pds,
Err(e) => return Asked::Unknown(format!("{e:#}")),
};
let xrpc = Xrpc::new(ctx.guarded.clone(), &pds);
let asked = [("did", did)];
match xrpc
.query::<Value>("com.atproto.sync.getRepoStatus", &asked, None)
.await
{
Ok(status) => match (status["active"].as_bool(), status["status"].as_str()) {
(Some(true), _) => Asked::Active,
(Some(false), Some("deleted")) => Asked::Gone,
// Deactivated, suspended or taken down, which can be undone:
// the account is still there.
(Some(false), why) => {
tracing::info!(%did, status = why, "inactive at its host, not deleted");
Asked::Active
}
(None, _) => Asked::Unknown(format!("{pds} answered {status}")),
},
Err(e) => match XrpcError::of(&e) {
Some(refused) if refused.error == "RepoNotFound" => Asked::Gone,
_ => Asked::Unknown(format!("{e:#}")),
},
}
}
/// The HTTP status a request in `e` was answered with, if it was.
fn http_status(e: &anyhow::Error) -> Option<u16> {
let answered = e.chain().find_map(|c| c.downcast_ref::<reqwest::Error>());
answered
.and_then(reqwest::Error::status)
.map(|s| s.as_u16())
}
/// How long to keep asking after an account no answer has settled.
#[derive(Debug, Clone, Copy)]
pub struct Patience {
/// The wait before asking again, doubled each time.
pub first: Duration,
/// The longest wait.
pub most: Duration,
/// How many times to ask in all.
pub tries: u32,
}
impl Default for Patience {
/// Six asks over about half an hour, the last waits capped at ten
/// minutes: long enough for PLC or a host to come back.
fn default() -> Patience {
Patience {
first: Duration::from_secs(30),
most: Duration::from_secs(600),
tries: 6,
}
}
}
/// What came of an account Jetstream says was deleted.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Settled {
/// Its host has it: it stays.
Kept,
/// Gone: it was forgotten ([`deleted`]).
Purged,
/// No answer said, however often asked: it stays, as nothing it
/// leaves would come back.
GaveUp,
}
/// Asks after `did` until an answer says, waiting longer each time, and
/// forgets it only if it is [`Asked::Gone`].
pub async fn settle(index: &Index, ctx: &Ctx, did: &str, patience: Patience) -> Settled {
let mut wait = patience.first;
for tried in 1..=patience.tries {
match gone(ctx, did).await {
Asked::Active => {
tracing::info!(%did, "an account deleted where it was is active where it is now");
return Settled::Kept;
}
Asked::Gone => {
let _ =
deleted(index, did).inspect_err(|e| tracing::error!(%did, %e, "forgetting it"));
return Settled::Purged;
}
Asked::Unknown(why) if tried < patience.tries => {
tracing::warn!(%did, tried, %why, "asking after an account deleted, again later");
tokio::time::sleep(wait).await;
wait = (wait * 2).min(patience.most);
}
Asked::Unknown(why) => {
tracing::error!(%did, tried, %why, "an account deleted, never settled: kept");
}
}
}
Settled::GaveUp
}
/// Each account Jetstream says was deleted, from `told`, asked after on
/// its own ([`settle`]) so that one waiting holds up none of the rest;
/// done once `told` closes and every one has settled.
pub async fn deletions(
index: Arc<Index>,
ctx: Ctx,
mut told: tokio::sync::mpsc::UnboundedReceiver<String>,
patience: Patience,
) {
let mut asking = tokio::task::JoinSet::new();
while let Some(did) = told.recv().await {
let (index, ctx) = (index.clone(), ctx.clone());
asking.spawn(async move { settle(&index, &ctx, &did, patience).await });
}
asking.join_all().await;
}
#[derive(Deserialize)]
struct Input {
did: String,
limit: Option<u64>,
}
/// What one call did.
#[derive(Serialize, Debug, Default, Clone, Copy, PartialEq)]
pub struct Step {
/// The records it deleted.
pub deleted: u64,
/// Whether none is left: grain has then forgotten the account.
pub done: bool,
}
/// The grain collections `did`'s repo at `pds` holds records in, in the
/// order they go.
async fn collections(pds: &Xrpc, did: &str) -> Result<Vec<String>, Failure> {
#[derive(Deserialize)]
struct Described {
collections: Vec<String>,
}
let asked = [("repo", did)];
let described: Described = pds
.query("com.atproto.repo.describeRepo", &asked, None)
.await
.map_err(|e| upstream(e.context("com.atproto.repo.describeRepo")))?;
let mut ours = described.collections;
ours.retain(|c| c.starts_with(PREFIX));
let at = |c: &str| ORDER.iter().position(|o| *o == c).unwrap_or(ORDER.len());
let rank = |c: &str| (c == LAST, at(c));
ours.sort_by(|a, b| rank(a).cmp(&rank(b)).then(a.cmp(b)));
Ok(ours)
}
/// Whether `did`'s repo still holds a grain record, as its PDS says to
/// anyone.
async fn holds_any(ctx: &Ctx, did: &str) -> Result<bool, Failure> {
let pds = grain::identity::resolve_pds(ctx, did)
.await
.map_err(|e| upstream(e.context("finding the account's PDS")))?;
let pds = Xrpc::new(ctx.guarded.clone(), &pds);
Ok(!collections(&pds, did).await?.is_empty())
}
/// Deletes up to `budget` of `did`'s grain records at its PDS, with
/// its web session, a collection at a time, a page at a time.
async fn delete_some(app: &App, did: &str, budget: u64) -> Result<Step, Failure> {
let session = crate::login::session(app, did).await?;
let pds = Xrpc::new(app.ctx.guarded.clone(), &session.service);
let mut step = Step::default();
for collection in collections(&pds, did).await? {
loop {
if step.deleted >= budget {
return Ok(step);
}
#[derive(Deserialize)]
struct Listed {
records: Vec<Listing>,
}
#[derive(Deserialize)]
struct Listing {
uri: String,
}
let limit = (budget - step.deleted).min(PAGE as u64).to_string();
let asked = [
("repo", did),
("collection", collection.as_str()),
("limit", limit.as_str()),
];
let listed: Listed = pds
.query("com.atproto.repo.listRecords", &asked, None)
.await
.map_err(|e| upstream(e.context("com.atproto.repo.listRecords")))?;
if listed.records.is_empty() {
break;
}
let writes: Vec<Value> = listed
.records
.iter()
.filter_map(|r| r.uri.rsplit('/').next())
.map(|rkey| {
json!({
"$type": "com.atproto.repo.applyWrites#delete",
"collection": collection,
"rkey": rkey,
})
})
.collect();
let body = json!({ "repo": did, "writes": writes });
crate::write::at_pds(app, did, "com.atproto.repo.applyWrites", &body).await?;
step.deleted += writes.len() as u64;
}
}
step.done = true;
Ok(step)
}
/// Forgets `did`, whose grain records are gone: what the index holds of
/// it, and its web session, revoked at its PDS first.
async fn forget(app: &App, did: &str) -> Result<(), Failure> {
let index = app.index.as_deref().expect("deleteAccount holds an index");
crate::login::end_session(app, did).await?;
let purged = index.purge(did)?;
index.untrack(did)?;
tracing::info!(%did, purged, "an account deleted its grain data");
Ok(())
}
/// `social.grain.server.deleteAccount`.
pub async fn delete_account(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Response, Failure> {
if app.index.is_none() {
let why = "this server keeps no index";
let refused = Failure::new(StatusCode::NOT_IMPLEMENTED, "MethodNotImplemented", why);
return Err(refused);
}
let did = crate::viewer::writer(&app, &headers, DELETE_ACCOUNT).await?;
let input: Input = serde_json::from_slice(&body)
.map_err(|e| Failure::invalid(format!("the input is not {{did, limit}}: {e}")))?;
if input.did != did {
let why = format!("{did} asks, but names {}", input.did);
return Err(Failure::new(StatusCode::FORBIDDEN, "Forbidden", why));
}
if headers.contains_key(header::AUTHORIZATION) {
if holds_any(&app.ctx, &did).await? {
let why = format!("{did} still holds grain records: delete them first");
return Err(Failure::new(StatusCode::BAD_REQUEST, "RecordsRemain", why));
}
forget(&app, &did).await?;
let step = Step {
deleted: 0,
done: true,
};
return Ok(Json(step).into_response());
}
let limit = input.limit.unwrap_or(LIMIT);
if !(1..=MOST).contains(&limit) {
return Err(Failure::invalid(format!(
"limit is 1 to {MOST}, not {limit}"
)));
}
let step = delete_some(&app, &did, limit).await?;
if !step.done {
return Ok(Json(step).into_response());
}
// Logged out first, which revokes the session and sets the cookies to
// the next account the browser holds, if any; then forgotten.
let out = crate::login::logout(State(app.clone()), headers).await?;
forget(&app, &did).await?;
let (parts, _) = out.into_parts();
Ok(Response::from_parts(
parts,
axum::body::Body::from(json!(step).to_string()),
))
}
crates/grain-serve/src/gallery.rsnew+120−0
View file
@@ -0,0 +1,120 @@
//! `social.grain.gallery.deleteGallery`: a gallery deleted with what hangs
//! off it, in one go, for the web. The records go from the account's PDS
//! with the session grain-serve keeps for it, as the web's other writes do
//! ([`crate::write`]), and from the index before it answers.
//!
//! What goes is the account's own: the gallery, its items, the photos
//! they place that no other of its galleries does, those photos' EXIF, and
//! its own favorite of the gallery and comments on it. What others wrote
//! about it is theirs, and stays. Children go before the gallery, so a
//! failure part way leaves a gallery with fewer photos, never photos with
//! no gallery.
//!
//! Only the web calls it, by its cookie, from a page of the site: the apps
//! delete at their own PDS and call `social.grain.sync.indexRecords`, and
//! a service token here is refused.
use crate::App;
use crate::xrpc::Failure;
use axum::Json;
use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode, header};
use grain::lexicon;
use rusqlite::params;
use serde::Deserialize;
use serde_json::{Value, json};
use std::sync::Arc;
/// The method, as the web's cookie calls it.
pub const DELETE_GALLERY: &str = "social.grain.gallery.deleteGallery";
#[derive(Deserialize)]
struct Input {
rkey: String,
}
/// The records of `did`'s that go with its gallery `gallery`, children
/// first, the gallery last.
pub const GOING_SQL: &str = "
SELECT 0, e.uri FROM photo_exif e WHERE e.did = ?2 AND e.photo IN (SELECT p FROM photos)
UNION ALL SELECT 1, i.uri FROM gallery_item i WHERE i.gallery = ?1 AND i.did = ?2
UNION ALL SELECT 2, p FROM photos
UNION ALL SELECT 3, f.uri FROM favorite f WHERE f.subject = ?1 AND f.did = ?2
UNION ALL SELECT 4, c.uri FROM comment c WHERE c.subject = ?1 AND c.did = ?2
UNION ALL SELECT 5, g.uri FROM gallery g WHERE g.uri = ?1 AND g.did = ?2
ORDER BY 1, 2";
/// The photos of `did`'s that its gallery `?1` places and no other of its
/// galleries does, for [`GOING_SQL`] to read.
const PHOTOS_SQL: &str = "WITH photos (p) AS (
SELECT DISTINCT i.item FROM gallery_item i JOIN photo ph ON ph.uri = i.item AND ph.did = ?2
WHERE i.gallery = ?1 AND i.did = ?2
AND NOT EXISTS (SELECT 1 FROM gallery_item o
WHERE o.item = i.item AND o.did = ?2 AND o.gallery != ?1))";
/// The whole query: the photos, then what goes.
pub fn going_sql() -> String {
format!("{PHOTOS_SQL} {GOING_SQL}")
}
/// `social.grain.gallery.deleteGallery`.
pub async fn delete_gallery(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<Value>, Failure> {
let Some(index) = app.index.clone() else {
let why = "this server keeps no index";
let refused = Failure::new(StatusCode::NOT_IMPLEMENTED, "MethodNotImplemented", why);
return Err(refused);
};
if headers.contains_key(header::AUTHORIZATION) {
let why = "a service token deletes at its own PDS, then calls \
social.grain.sync.indexRecords";
return Err(Failure::new(StatusCode::FORBIDDEN, "Forbidden", why));
}
let did = crate::viewer::writer(&app, &headers, DELETE_GALLERY).await?;
let input: Input = serde_json::from_slice(&body)
.map_err(|e| Failure::invalid(format!("the input is not {{rkey}}: {e}")))?;
if !lexicon::valid_record_key(&input.rkey) {
return Err(Failure::invalid(format!(
"not a record key: {}",
input.rkey
)));
}
let gallery = format!("at://{did}/social.grain.gallery/{}", input.rkey);
let going: Vec<String> = index.read(|db| {
let mut stmt = db.prepare_cached(&going_sql())?;
let rows = stmt.query_map(params![gallery, did], |r| r.get(1))?;
rows.collect::<rusqlite::Result<_>>()
})?;
if going.last() != Some(&gallery) {
let why = "Gallery not found";
return Err(Failure::new(StatusCode::BAD_REQUEST, "NotFound", why));
}
let writes: Vec<Value> = going
.iter()
.filter_map(|uri| crate::index::record_uri(uri))
.map(|(_, collection, rkey)| {
json!({
"$type": "com.atproto.repo.applyWrites#delete",
"collection": collection,
"rkey": rkey,
})
})
.collect();
let mut missed = Vec::new();
for (chunk, uris) in writes
.chunks(crate::write::MOST_WRITES)
.zip(going.chunks(crate::write::MOST_WRITES))
{
let body = json!({ "repo": did, "writes": chunk });
crate::write::at_pds(&app, &did, "com.atproto.repo.applyWrites", &body).await?;
// What the index fails to drop, Jetstream drops as the PDS says.
missed.extend(uris.iter().filter_map(|uri| index.delete(uri).err()));
}
let records = going.len();
tracing::info!(%did, %gallery, records, unindexed = ?missed, "deleted a gallery");
Ok(Json(json!({})))
}
crates/grain-serve/src/jetstream.rs+79−10
View file
@@ -13,15 +13,20 @@
//! 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`]).
//! A record new to the index is pushed ([`Index::arrived`]). An account
//! its host says was deleted goes, records and all, once the host its
//! identity names now says it has no such account
//! ([`crate::accounts::deletions`]): one that moved stays.
//!
//! Where it was read to is kept in the index, under `jetstream:<url>`,
//! and a restart resumes there. A cursor
//! Jetstream no longer keeps is given up for the live tip
//! ([`grain::jetstream::PROBE_EVERY`]).
use crate::accounts;
use crate::backfill::SIGNALS;
use crate::index::{Applied, Index, PutError, records};
use grain::env::Ctx;
use grain::jetstream::{Event, Follower, Timing, What};
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -42,8 +47,26 @@ pub fn source(url: &str) -> String {
}
/// Follows the Jetstream at `url` into `index` until the task is dropped,
/// saving where it was read to at most every `save`.
pub async fn follow(index: Arc<Index>, url: String, timing: Timing, save: Duration) {
/// saving where it was read to at most every `save`. An account the
/// stream says was deleted is asked after with `asking`
/// ([`crate::accounts::deletions`]), or with none taken at its word.
pub async fn follow(
index: Arc<Index>,
url: String,
timing: Timing,
save: Duration,
asking: Option<Ctx>,
) {
let (to, told) = tokio::sync::mpsc::unbounded_channel();
let asked = asking.map(|ctx| {
tokio::spawn(accounts::deletions(
index.clone(),
ctx,
told,
Default::default(),
))
});
let deletions = asked.is_some().then_some(to);
let collections = records::COLLECTIONS.iter().map(|(nsid, _)| *nsid).collect();
index.feeds.follows(NAME, &url);
let feed = Feed {
@@ -52,10 +75,15 @@ pub async fn follow(index: Arc<Index>, url: String, timing: Timing, save: Durati
seq: None,
saved: None,
save,
deletions,
};
grain::jetstream::follow(url, collections, timing, feed).await
}
/// Where an account the stream says was deleted is sent, for its host to
/// be asked.
pub type Deletions = tokio::sync::mpsc::UnboundedSender<String>;
struct Feed {
index: Arc<Index>,
source: String,
@@ -63,6 +91,7 @@ struct Feed {
seq: Option<i64>,
saved: Option<Instant>,
save: Duration,
deletions: Option<Deletions>,
}
impl Feed {
@@ -97,7 +126,7 @@ impl Follower for Feed {
fn event(&mut self, event: Event) {
self.seq = event.seq.or(self.seq);
self.index.feeds.heard(NAME, event.seq);
logged(apply(&self.index, event));
logged(apply_with(&self.index, event, self.deletions.as_ref()));
if self.saved.is_none_or(|at| at.elapsed() >= self.save) {
logged(self.save());
}
@@ -116,8 +145,19 @@ impl Follower for Feed {
}
}
/// What `event` says, applied to `index`.
/// What `event` says, applied to `index`, an account deleted taken at its
/// word.
pub fn apply(index: &Index, event: Event) -> rusqlite::Result<()> {
apply_with(index, event, None)
}
/// [`apply`], an account deleted sent to `deletions` where there is one,
/// for its host to be asked first.
pub fn apply_with(
index: &Index,
event: Event,
deletions: Option<&Deletions>,
) -> rusqlite::Result<()> {
match event.what {
What::Commit { did, op } => {
if records::table_of(&op.collection).is_none() {
@@ -177,11 +217,12 @@ pub fn apply(index: &Index, event: Event) -> rusqlite::Result<()> {
index.resync(&did)?;
}
}
(false, Some("deleted")) => {
let purged = index.purge(&did)?;
index.set_host_status(&did, Some("deleted"))?;
tracing::info!(%did, purged, "jetstream: an account was deleted");
}
(false, Some("deleted")) => match deletions {
Some(to) => {
let _ = to.send(did);
}
None => accounts::deleted(index, &did)?,
},
// Its host says it is not there for now: its records stay,
// unread, for when it is back.
(false, Some(status @ ("deactivated" | "suspended" | "takendown"))) => {
@@ -198,3 +239,31 @@ pub fn apply(index: &Index, event: Event) -> rusqlite::Result<()> {
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use grain::jetstream::What;
#[test]
fn an_account_deleted_is_asked_after_where_its_host_can_be() {
let index = Index::memory().unwrap();
let did = "did:plc:alice";
index.track(did).unwrap();
let deleted = || Event {
seq: Some(1),
what: What::Account {
did: did.into(),
active: false,
status: Some("deleted".into()),
},
};
let (to, mut told) = tokio::sync::mpsc::unbounded_channel();
apply_with(&index, deleted(), Some(&to)).unwrap();
assert_eq!(told.try_recv().ok().as_deref(), Some(did));
assert_eq!(index.host_status(did).unwrap(), None, "not yet");
// With no host to ask, the event is taken at its word.
apply(&index, deleted()).unwrap();
assert_eq!(index.host_status(did).unwrap().as_deref(), Some("deleted"));
}
}
crates/grain-serve/src/lib.rs+39−7
View file
@@ -4,15 +4,19 @@
//! the index alone ([`read`]), what an account has seen of its
//! notifications ([`notification`]) and the devices they are pushed to
//! ([`push`]), the stories it watched ([`story`]), muting ([`graph`]),
//! moderation ([`moderation`]), permissioned spaces ([`spaces`]), and the
//! web's writes ([`write`]), draws the cards links unfurl into ([`og`]),
//! and logs the web in ([`login`]). Off the path of any request, its
//! classifiers score what arrives for moderators to review ([`classify`]).
//! moderation ([`moderation`]), permissioned spaces ([`spaces`]), the
//! web's writes ([`write`]) and deleting a gallery ([`gallery`]), accounts
//! that go ([`accounts`]), draws the cards links unfurl into ([`og`]),
//! and logs the web in ([`login`]), handing the rest to the web's pages
//! ([`pages`]). Off the path of any request, its classifiers score what
//! arrives for moderators to review ([`classify`]).
pub mod accounts;
pub mod backfill;
pub mod cdn;
pub mod classify;
pub mod did;
pub mod gallery;
pub mod graph;
pub mod index;
pub mod jetstream;
@@ -20,6 +24,7 @@ pub mod login;
pub mod moderation;
pub mod notification;
pub mod og;
pub mod pages;
pub mod preferences;
pub mod push;
pub mod read;
@@ -69,10 +74,11 @@ pub const FLAGS: &[&str] = &[
"--clef-url",
"--check-index",
"--metrics",
"--web",
];
/// What `grain-serve` says of its flags when given one it does not take.
pub const USAGE: &str = "usage: grain-serve [--listen <addr>] [--index <file>] [--public-url <https://host>] [--jetstream <wss://host>] [--backfill <https://directory>[,...]] [--backfill-mirror <https://mirror>] [--backfill-skip <host>[,...]] [--backfill-concurrency <n>] [--cdn <https://imgproxy>] [--bsky-appview <https://appview>] [--sign-up-pds <https://pds>] [--carry-cookies-until <rfc3339 time>] [--apns-key-id <id>] [--apns-team-id <id>] [--apns-topic <bundle id>] [--apns-url <https://apns>] [--fcm-project-id <id>] [--clef-model <model>] [--clef-url <url>] [--check-index <file>] [--metrics <addr>]
pub const USAGE: &str = "usage: grain-serve [--listen <addr>] [--index <file>] [--public-url <https://host>] [--jetstream <wss://host>] [--backfill <https://directory>[,...]] [--backfill-mirror <https://mirror>] [--backfill-skip <host>[,...]] [--backfill-concurrency <n>] [--cdn <https://imgproxy>] [--bsky-appview <https://appview>] [--sign-up-pds <https://pds>] [--carry-cookies-until <rfc3339 time>] [--apns-key-id <id>] [--apns-team-id <id>] [--apns-topic <bundle id>] [--apns-url <https://apns>] [--fcm-project-id <id>] [--clef-model <model>] [--clef-url <url>] [--check-index <file>] [--metrics <addr>] [--web <http://app>]
environment: GRAIN_ADMIN_DIDS, GRAIN_CDN_KEY, GRAIN_CDN_SALT, GRAIN_APNS_KEY_FILE, GRAIN_FCM_KEY_FILE, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_AI_TOKEN, RUST_LOG";
/// What the server is set to do, from its flags ([`FLAGS`]) and, for
@@ -128,6 +134,11 @@ pub struct Settings {
/// `--metrics`: where `/metrics` and `PUT /log` are answered, apart
/// from the rest ([`telemetry::admin`]). None answers them nowhere.
pub metrics: Option<String>,
/// `--web`: the web's app, a URL such as `http://web:3000`, that draws
/// every page and is handed each request the server does not answer
/// itself ([`pages`]). None hands on nothing: what the server does not
/// answer is not found.
pub web: Option<String>,
}
/// `grain-serve`'s settings, from its arguments, the first being the
@@ -181,6 +192,7 @@ impl Settings {
clef: classify::clef::Config::given(given),
check_index: get("--check-index").map(PathBuf::from),
metrics: get("--metrics"),
web: url("--web"),
})
}
@@ -206,7 +218,8 @@ impl Settings {
bail!("--jetstream needs --index");
};
let timing = grain::jetstream::Timing::default();
let follow = jetstream::follow(index.clone(), url.clone(), timing, jetstream::SAVE);
let (index, ctx, url) = (index.clone(), app.ctx.clone(), url.clone());
let follow = jetstream::follow(index, url, timing, jetstream::SAVE, Some(ctx));
tokio::spawn(follow);
Ok(true)
}
@@ -278,6 +291,7 @@ impl Settings {
web,
admins: self.admins.clone(),
cards: og::Cards::default(),
pages: self.web.clone(),
spaces: Default::default(),
})
}
@@ -301,6 +315,9 @@ pub struct App {
pub admins: Vec<String>,
/// The OpenGraph cards drawn lately.
pub cards: og::Cards,
/// The web's app, handed what the server does not answer itself
/// ([`pages`]), if there is one: `--web`.
pub pages: Option<String>,
/// The credentials spaces are read with lately ([`spaces::Spaces`]).
pub spaces: spaces::Spaces,
}
@@ -318,6 +335,7 @@ impl App {
web: None,
admins: Vec::new(),
cards: og::Cards::default(),
pages: None,
spaces: Default::default(),
}
}
@@ -346,8 +364,10 @@ pub fn check_index(path: &std::path::Path) -> anyhow::Result<String> {
}
}
/// The routes the server answers.
/// The routes the server answers, and, with [`App::pages`], the web's app
/// handed everything else ([`pages`]).
pub fn router(app: Arc<App>) -> Router {
let pages = app.pages.as_deref().map(pages::pages).unwrap_or_default();
spaces::routes(moderation::routes(og::routes(read::routes(Router::new()))))
.route("/healthz", get(status::healthz))
.route("/_status", get(status::status))
@@ -400,6 +420,14 @@ pub fn router(app: Arc<App>) -> Router {
"/xrpc/social.grain.repo.uploadBlob",
post(write::upload_blob).layer(DefaultBodyLimit::max(write::MOST_BYTES)),
)
.route(
&format!("/xrpc/{}", gallery::DELETE_GALLERY),
post(gallery::delete_gallery),
)
.route(
&format!("/xrpc/{}", accounts::DELETE_ACCOUNT),
post(accounts::delete_account),
)
.route(login::METADATA, get(login::client_metadata))
.route(login::JWKS, get(login::client_jwks))
.route("/oauth/login", post(login::login))
@@ -408,6 +436,7 @@ pub fn router(app: Arc<App>) -> Router {
.route("/oauth/logout", post(login::logout))
.route("/oauth/session", get(login::whoami))
.route("/xrpc/{*method}", any(not_implemented))
.merge(pages)
.layer(axum::middleware::from_fn_with_state(
app.clone(),
login::carryover::carry,
@@ -499,6 +528,8 @@ mod tests {
"https://pds.test",
"--jetstream",
jetstream::DEFAULT,
"--web",
"http://web:3000/",
]),
|name| match name {
"GRAIN_ADMIN_DIDS" => Some("did:plc:alice, did:web:bob.test".into()),
@@ -510,6 +541,7 @@ mod tests {
)
.unwrap();
assert_eq!(set.listen.as_deref(), Some("0.0.0.0:2600"));
assert_eq!(set.web.as_deref(), Some("http://web:3000"));
assert_eq!(set.admins, ["did:plc:alice", "did:web:bob.test"]);
assert_eq!(
set.bsky_appview.as_deref(),
crates/grain-serve/src/login.rs+17−9
View file
@@ -398,7 +398,7 @@ pub async fn asked(app: &App, pds: &str) -> Result<String, Failure> {
}
/// A PDS or authorization server failing a login: 502, why told.
fn upstream(e: anyhow::Error) -> Failure {
pub(crate) fn upstream(e: anyhow::Error) -> Failure {
Failure::new(StatusCode::BAD_GATEWAY, "UpstreamFailure", format!("{e:#}"))
}
@@ -499,6 +499,21 @@ pub async fn switch(
Ok(res)
}
/// Ends `did`'s web session: revokes it at its authorization server, as
/// far as it lets, and forgets it here, with the space credentials it was
/// reading with.
pub(crate) async fn end_session(app: &App, did: &str) -> Result<(), Failure> {
if let (Some(web), Some(kept)) = (app.web.as_deref(), index(app).web_session(did)?)
&& let Some(o) = &kept.session.oauth
{
// A token the server cannot revoke now expires on its own.
let _ = oauth::revoke(&guarded(app), &web.client, &kept.session, o).await;
}
index(app).end_web_session(did)?;
app.spaces.log_out(did);
Ok(())
}
/// Logs the account the cookie names out: revokes its session at its
/// authorization server, as far as it lets, and forgets it here, which
/// ends every cookie signed for it. The browser goes on as the other
@@ -509,14 +524,7 @@ pub async fn logout(State(app): State<Arc<App>>, headers: HeaderMap) -> Result<R
return Err(not_this_site());
}
if let Some(did) = viewer(&app, &headers) {
if let Some(kept) = index(&app).web_session(&did)?
&& let Some(o) = &kept.session.oauth
{
// A token the server cannot revoke now expires on its own.
let _ = oauth::revoke(&guarded(&app), &web.client, &kept.session, o).await;
}
index(&app).end_web_session(&did)?;
app.spaces.log_out(&did);
end_session(&app, &did).await?;
}
// Read once the account is logged out, so none of its cookies is left.
let mut others = others(&app, &headers, None);
crates/grain-serve/src/main.rs+2−1
View file
@@ -2,7 +2,8 @@
//! given, keeping the index at `--index`, backfilled when `--backfill`
//! names directories and fed from `--jetstream` when that names one,
//! pushing by APNs and FCM when their key files are named, and taking
//! service tokens for the DID of `--public-url` when given
//! service tokens for the DID of `--public-url` when given, and handing
//! the web's pages to the app at `--web` when that names one
//! ([`grain_serve::Settings`], [`grain_serve::USAGE`]).
use anyhow::Context;
crates/grain-serve/src/pages.rsnew+97−0
View file
@@ -0,0 +1,97 @@
//! The web's pages: every request grain-serve does not answer itself,
//! handed to the SvelteKit app in `web/` that draws them, at `--web`
//! ([`crate::Settings::web`]), as knotless-serve hands its pages to its
//! app. The app reads back here for what to show, by `/xrpc`, with the
//! browser's cookie. With no app to hand to, what the server does not
//! answer is not found.
//!
//! A request is handed on whole, but for the headers about one hop of a
//! connection, with the host it was asked for as `x-forwarded-host` and,
//! where nothing in front said, `x-forwarded-proto` as `http`. The app's
//! answer comes back as it gave it, its redirects and cookies too, with
//! `x-content-type-options: nosniff` and `referrer-policy: same-origin`
//! where it set neither. An app that does not answer is a 502.
use axum::Router;
use axum::body::Body;
use axum::extract::Request;
use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
use axum::response::{IntoResponse, Response};
use std::sync::{Arc, LazyLock};
/// The largest request body handed on: the pages take none to speak of.
pub const BODY_LIMIT: usize = 1 << 20;
/// Headers about one hop of a connection, which are not handed on.
const HOP: [header::HeaderName; 6] = [
header::CONNECTION,
header::HOST,
header::TRANSFER_ENCODING,
header::CONTENT_LENGTH,
header::UPGRADE,
header::TE,
];
static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.build()
.expect("a client builds")
});
/// A router that hands every request to the app at `web`.
pub fn pages<S: Clone + Send + Sync + 'static>(web: &str) -> Router<S> {
let web: Arc<str> = web.trim_end_matches('/').into();
Router::new().fallback(move |req: Request| {
let web = web.clone();
async move { page(&web, req).await }
})
}
async fn page(web: &str, req: Request) -> Response {
match handed(web, req).await {
Ok(res) => res,
Err(e) => {
tracing::warn!(error = %format!("{e:#}"), "the pages are not being drawn");
let why = "The pages are not being drawn just now. Try again in a moment.";
(StatusCode::BAD_GATEWAY, why).into_response()
}
}
}
/// `req`, as the app at `web` answers it.
async fn handed(web: &str, req: Request) -> anyhow::Result<Response> {
let (parts, body) = req.into_parts();
let at = parts.uri.path_and_query().map_or("/", |p| p.as_str());
let mut headers = kept(&parts.headers);
if let Some(host) = parts.headers.get(header::HOST) {
headers.insert("x-forwarded-host", host.clone());
}
headers
.entry("x-forwarded-proto")
.or_insert(HeaderValue::from_static("http"));
let body = axum::body::to_bytes(body, BODY_LIMIT).await?;
let asked = CLIENT.request(parts.method, format!("{web}{at}"));
let answer = grain::telemetry::send(asked.headers(headers).body(body), 0).await?;
let mut res = Response::builder().status(answer.status());
let mut headers = kept(answer.headers());
for (name, value) in [
(header::X_CONTENT_TYPE_OPTIONS, "nosniff"),
(header::REFERRER_POLICY, "same-origin"),
] {
headers
.entry(name)
.or_insert(HeaderValue::from_static(value));
}
*res.headers_mut().expect("a builder with a status") = headers;
Ok(res.body(Body::from(answer.bytes().await?))?)
}
/// `headers`, but for those about one hop.
fn kept(headers: &HeaderMap) -> HeaderMap {
let mut out = headers.clone();
for name in HOP {
out.remove(name);
}
out
}
crates/grain-serve/src/write.rs+7−0
View file
crates/grain-serve/tests/web/mod.rs+2−0
View file
crates/grain-serve/tests/cross_post.rsnew+138−0
View file
crates/grain-serve/tests/deletes.rsnew+472−0
View file
472 lines changed: not loaded until asked for.
crates/grain-serve/tests/gone.rsnew+211−0
View file
crates/grain-serve/tests/index_records.rs+1−0
View file
crates/grain-serve/tests/jetstream.rs+11−0
View file
crates/grain-serve/tests/notifications.rs+1−0
View file
crates/grain-serve/tests/pages.rsnew+163−0
View file
crates/grain-serve/tests/push.rs+1−0
View file
crates/grain-serve/tests/read.rs+3−0
View file
lexicons/social/grain/gallery/deleteGallery.json+7−2
View file
lexicons/social/grain/server/deleteAccount.json+53−3
View file
lexicons/README.md+8−0
View file
web/src/routes/settings/account/+page.svelte+17−5
View file
vite.config.ts+4−3
View file