~ 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 answers profiles, galleries, comments, follows and stories

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:read/views

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+15−10
View file
@@ -15,6 +15,8 @@
pub mod backfill;
pub mod import;
#[cfg(test)]
pub mod plan;
pub mod records;
pub mod schema;
@@ -104,6 +106,18 @@ pub struct Label {
pub exp: Option<String>,
}
/// The record a result row holds, one that names the columns of `T`'s
/// table, envelope and all.
pub fn row<T: Record>(r: &rusqlite::Row) -> rusqlite::Result<Row<T>> {
Ok(Row {
uri: r.get("uri")?,
cid: r.get("cid")?,
did: r.get("did")?,
indexed_at: r.get("indexed_at")?,
record: T::from_row(r)?,
})
}
/// 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('/');
@@ -314,16 +328,7 @@ impl Index {
pub fn get<T: Record>(&self, uri: &str) -> rusqlite::Result<Option<Row<T>>> {
let db = self.db.lock().unwrap();
let sql = format!("SELECT * FROM {} WHERE uri = ?1", T::TABLE);
db.query_row(&sql, [uri], |r| {
Ok(Row {
uri: r.get("uri")?,
cid: r.get("cid")?,
did: r.get("did")?,
indexed_at: r.get("indexed_at")?,
record: T::from_row(r)?,
})
})
.optional()
db.query_row(&sql, [uri], row).optional()
}
/// The self-labels the record at `uri` carries.
crates/grain-serve/src/index/plan.rsnew+98−0
View file
@@ -0,0 +1,98 @@
//! How SQLite plans a query on an index the size of prod's: what the
//! plan checks of the feeds and the reads hold queries to.
use rusqlite::Connection;
use std::collections::{HashMap, HashSet};
/// Each table said to hold a million rows, and each index's first column
/// to pick out a hundred, so SQLite plans as it would on prod's.
pub fn sized(db: &Connection) {
db.execute_batch("ANALYZE; DELETE FROM sqlite_stat1;")
.unwrap();
let names = |sql: &str, arg: &str| -> Vec<(String, bool)> {
let mut stmt = db.prepare(sql).unwrap();
let rows = stmt
.query_map([arg], |r| Ok((r.get(0)?, r.get(1)?)))
.unwrap();
rows.collect::<rusqlite::Result<_>>().unwrap()
};
let tables = names(
"SELECT name, 0 FROM sqlite_master WHERE type = 'table' AND name NOT LIKE ?1",
"sqlite_%",
);
for (table, _) in tables {
db.execute(
"INSERT INTO sqlite_stat1 VALUES (?1, NULL, '1000000')",
[&table],
)
.unwrap();
for (index, unique) in names("SELECT name, \"unique\" FROM pragma_index_list(?1)", &table) {
let columns: u32 = db
.query_row(
"SELECT count(*) FROM pragma_index_info(?1)",
[&index],
|r| r.get(0),
)
.unwrap();
let mut stat = "1000000".to_string();
for i in 0..columns {
let picked = if unique && i + 1 == columns {
1
} else {
(100 / 10u64.pow(i)).max(2)
};
stat += &format!(" {picked}");
}
db.execute(
"INSERT INTO sqlite_stat1 VALUES (?1, ?2, ?3)",
[&table, &index, &stat],
)
.unwrap();
}
}
db.execute_batch("ANALYZE sqlite_schema;").unwrap();
}
/// The tables `sql`'s plan reads whole, rather than by an index: what no
/// query of a table that grows with the network may do.
pub fn scans(db: &Connection, sql: &str) -> Vec<String> {
let words: Vec<&str> = sql
.split(|c: char| c.is_whitespace() || "(),".contains(c))
.filter(|w| !w.is_empty())
.collect();
let mut aliases = HashMap::new();
for (i, w) in words.iter().enumerate() {
if (w.eq_ignore_ascii_case("FROM") || w.eq_ignore_ascii_case("JOIN")) && i + 2 < words.len()
{
aliases.insert(words[i + 2], words[i + 1]);
aliases.insert(words[i + 1], words[i + 1]);
}
}
let tables: HashSet<String> = {
let mut stmt = db
.prepare("SELECT name FROM sqlite_master WHERE type = 'table'")
.unwrap();
stmt.query_map([], |r| r.get(0))
.unwrap()
.map(Result::unwrap)
.collect()
};
let mut stmt = db.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
// Unbound, as its parameters do not change its plan.
let mut rows = stmt.raw_query();
let mut details: Vec<String> = Vec::new();
while let Some(row) = rows.next().unwrap() {
details.push(row.get(3).unwrap());
}
details
.into_iter()
.filter(|d| {
let Some(rest) = d.strip_prefix("SCAN ") else {
return false;
};
let name = rest.split_whitespace().next().unwrap_or_default();
let table = aliases.get(name).copied().unwrap_or(name);
tables.contains(table) && !rest.contains(" USING ")
})
.collect()
}
crates/grain-serve/src/index/schema.rs+8−1
View file
@@ -60,7 +60,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];
pub const MIGRATIONS: &[&str] = &[V1, V2, V3];
const V1: &str = "
CREATE TABLE gallery (
@@ -241,6 +241,13 @@ CREATE TABLE record_rev (
CREATE INDEX record_rev_gone ON record_rev (rev) WHERE gone = 1;
";
/// For the lists of accounts and favorites, paged by when each was made.
const V3: &str = "
CREATE INDEX follow_subject_created ON follow (subject, created_at);
CREATE INDEX follow_did_created ON follow (did, created_at);
CREATE INDEX favorite_subject_created ON favorite (subject, created_at);
";
/// 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+1−94
View file
@@ -1,7 +1,7 @@
use super::plan::{scans, sized};
use super::records::*;
use super::*;
use serde_json::json;
use std::collections::HashMap;
const DID: &str = "did:plc:alice";
const CID: &str = "bafyreia4w4bbvc5lqzkvexqlbwbagz5yfjfzzrqhfwqkhkk2xgekh4c6ai";
@@ -447,99 +447,6 @@ fn migrations() {
assert!(failed.starts_with("opening "), "{failed}");
}
/// Each table said to hold a million rows, and each index's first column
/// to pick out a hundred, so SQLite plans as it would on prod's.
fn sized(db: &Connection) {
db.execute_batch("ANALYZE; DELETE FROM sqlite_stat1;")
.unwrap();
let names = |sql: &str, arg: &str| -> Vec<(String, bool)> {
let mut stmt = db.prepare(sql).unwrap();
let rows = stmt
.query_map([arg], |r| Ok((r.get(0)?, r.get(1)?)))
.unwrap();
rows.collect::<rusqlite::Result<_>>().unwrap()
};
let tables = names(
"SELECT name, 0 FROM sqlite_master WHERE type = 'table' AND name NOT LIKE ?1",
"sqlite_%",
);
for (table, _) in tables {
db.execute(
"INSERT INTO sqlite_stat1 VALUES (?1, NULL, '1000000')",
[&table],
)
.unwrap();
for (index, unique) in names("SELECT name, \"unique\" FROM pragma_index_list(?1)", &table) {
let columns: u32 = db
.query_row(
"SELECT count(*) FROM pragma_index_info(?1)",
[&index],
|r| r.get(0),
)
.unwrap();
let mut stat = "1000000".to_string();
for i in 0..columns {
let picked = if unique && i + 1 == columns {
1
} else {
(100 / 10u64.pow(i)).max(2)
};
stat += &format!(" {picked}");
}
db.execute(
"INSERT INTO sqlite_stat1 VALUES (?1, ?2, ?3)",
[&table, &index, &stat],
)
.unwrap();
}
}
db.execute_batch("ANALYZE sqlite_schema;").unwrap();
}
/// The tables `sql`'s plan reads whole, rather than by an index: what no
/// query of a table that grows with the network may do.
fn scans(db: &Connection, sql: &str) -> Vec<String> {
let words: Vec<&str> = sql
.split(|c: char| c.is_whitespace() || "(),".contains(c))
.filter(|w| !w.is_empty())
.collect();
let mut aliases = HashMap::new();
for (i, w) in words.iter().enumerate() {
if (w.eq_ignore_ascii_case("FROM") || w.eq_ignore_ascii_case("JOIN")) && i + 2 < words.len()
{
aliases.insert(words[i + 2], words[i + 1]);
aliases.insert(words[i + 1], words[i + 1]);
}
}
let tables: HashSet<String> = {
let mut stmt = db
.prepare("SELECT name FROM sqlite_master WHERE type = 'table'")
.unwrap();
stmt.query_map([], |r| r.get(0))
.unwrap()
.map(Result::unwrap)
.collect()
};
let mut stmt = db.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
// Unbound, as its parameters do not change its plan.
let mut rows = stmt.raw_query();
let mut details: Vec<String> = Vec::new();
while let Some(row) = rows.next().unwrap() {
details.push(row.get(3).unwrap());
}
details
.into_iter()
.filter(|d| {
let Some(rest) = d.strip_prefix("SCAN ") else {
return false;
};
let name = rest.split_whitespace().next().unwrap_or_default();
let table = aliases.get(name).copied().unwrap_or(name);
tables.contains(table) && !rest.contains(" USING ")
})
.collect()
}
#[test]
fn the_feeds_read_by_index() {
let index = Index::memory().unwrap();
crates/grain-serve/src/read/actor.rsnew+181−0
View file
@@ -0,0 +1,181 @@
//! `social.grain.actor.*`: an account's profile, and the galleries it
//! favorited.
use super::moderation::{shown, unhidden};
use super::views::put;
use super::{Params, Reader, page};
use crate::cdn::AVATAR;
use crate::index::records::Json;
use crate::xrpc::Failure;
use axum::http::StatusCode;
use rusqlite::params;
use serde_json::{Map, Value, json};
/// Why an account's profile is not shown, as Bluesky's appview says it:
/// taken down, here or by its host, deactivated, or deleted.
fn withheld(r: &Reader, did: &str) -> Result<(), Failure> {
let (taken, host): (bool, Option<String>) = r.db.query_row(
"SELECT EXISTS (SELECT 1 FROM takedown WHERE did = ?1),
(SELECT host_status FROM repo WHERE did = ?1)",
[did],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
let refused = |error, why: &str| Err(Failure::new(StatusCode::BAD_REQUEST, error, why));
match host.as_deref() {
_ if taken => refused("AccountTakedown", "Account has been suspended"),
Some("deactivated") => refused("AccountDeactivated", "Account is deactivated"),
Some("deleted") => refused("InvalidRequest", "Profile not found"),
Some(_) => refused("AccountTakedown", "Account has been suspended"),
None => Ok(()),
}
}
/// An account's galleries with photos in them, whether its feed shows
/// them or not; its followers; and whom it follows: accounts, however
/// many records say so.
const COUNTS: [&str; 3] = [
"SELECT count(*) FROM gallery g WHERE g.did = ?1
AND EXISTS (SELECT 1 FROM gallery_item i WHERE i.gallery = g.uri)",
"SELECT count(DISTINCT did) FROM follow WHERE subject = ?1",
"SELECT count(DISTINCT subject) FROM follow WHERE did = ?1",
];
/// Whom an account takes messages from on Germ.
const GERM: &str =
"SELECT message_me FROM germ_declaration d WHERE d.did = ?1 ORDER BY d.uri LIMIT 1";
/// The URI of the one record of `table` by `did` naming `subject`.
const RELATION: &[(&str, &str)] = &[
(
"following",
"SELECT uri FROM follow WHERE did = ?1 AND subject = ?2 ORDER BY uri LIMIT 1",
),
(
"followedBy",
"SELECT uri FROM follow WHERE did = ?2 AND subject = ?1 ORDER BY uri LIMIT 1",
),
(
"blocking",
"SELECT uri FROM block WHERE did = ?1 AND subject = ?2 ORDER BY uri LIMIT 1",
),
(
"blockedBy",
"SELECT uri FROM block WHERE did = ?2 AND subject = ?1 ORDER BY uri LIMIT 1",
),
(
"muted",
"SELECT subject FROM mute WHERE did = ?1 AND subject = ?2",
),
];
/// `social.grain.actor.getProfile`: the account, its counts, whom it
/// takes messages from on Germ, and, for a viewer other than itself, how
/// the two stand. A follow either way is not said where either blocks
/// the other.
pub fn get_profile(r: &Reader, p: &Params) -> Result<Value, Failure> {
let did = r.actor(p)?;
withheld(r, &did)?;
let actor = r
.actors(&[did.as_str()])?
.remove(&did)
.expect("an actor for each DID asked");
let count = |sql: &str| r.db.query_row(sql, [&did], |row| row.get::<_, i64>(0));
let [galleries, followers, follows] = COUNTS;
let (galleries, followers, follows) = (count(galleries)?, count(followers)?, count(follows)?);
let germ: Vec<(Option<Json>,)> = r.all(GERM, [&did])?;
let mut view = json!({
"did": did,
"handle": actor.handle,
"galleryCount": galleries,
"followersCount": followers,
"followsCount": follows,
});
put(
&mut view,
"messageMe",
germ.into_iter().next().and_then(|(m,)| m).map(|m| m.0),
);
let mut labelled = vec![did.as_str()];
if let Some(profile) = &actor.profile {
let record = &profile.record;
view["cid"] = profile.cid.as_str().into();
put(&mut view, "displayName", record.display_name.clone());
put(&mut view, "description", record.description.clone());
let avatar = record
.avatar
.as_ref()
.and_then(|a| r.cdn.url(&did, &a.0, AVATAR));
put(&mut view, "avatar", avatar);
put(&mut view, "createdAt", record.created_at.clone());
labelled.push(&profile.uri);
}
let labels: Vec<Value> = r.labels(&labelled)?.into_values().flatten().collect();
put(&mut view, "labels", (!labels.is_empty()).then_some(labels));
// The viewer, unless they are the account itself.
let other = r.viewer.filter(|v| *v != did);
let mut stand = Map::new();
for (field, sql) in RELATION {
let found: Vec<(String,)> = r.all(sql, params![other, did])?;
if let Some((uri,)) = found.into_iter().next() {
stand.insert(field.to_string(), uri.into());
}
}
if stand.contains_key("blocking") || stand.contains_key("blockedBy") {
stand.remove("following");
stand.remove("followedBy");
}
for flag in ["blockedBy", "muted"] {
if stand.contains_key(flag) {
stand.insert(flag.into(), true.into());
}
}
put(&mut view, "viewer", (!stand.is_empty()).then_some(stand));
Ok(view)
}
/// The galleries the viewer favorited, the latest favorite first: one
/// row a gallery however many times it was favorited, of authors shown,
/// with photos in it, and not labelled to hide.
pub fn favorites_sql() -> String {
format!(
"SELECT f.subject, max(f.created_at) AS at FROM favorite f JOIN gallery g ON g.uri = f.subject
WHERE f.did = ?1 AND {} AND {}
AND EXISTS (SELECT 1 FROM gallery_item i WHERE i.gallery = g.uri)
GROUP BY f.subject HAVING (at, f.subject) < (?2, ?3)
ORDER BY at DESC, f.subject DESC LIMIT ?4",
shown("g.did"),
unhidden("g.uri"),
)
}
/// `social.grain.actor.getFavorites`: the galleries an account favorited,
/// which only the account itself may ask for.
pub fn get_favorites(r: &Reader, p: &Params) -> Result<Value, Failure> {
let did = r.actor(p)?;
if r.signed_in()? != did {
return Err(Failure::invalid(
"an account's favorites are its own to list",
));
}
let limit = p.limit(30, 100)?;
let (at, after) = p.after(false)?;
let rows: Vec<(String, String)> = r.all(
&favorites_sql(),
params![did, at, after, (limit + 1) as i64],
)?;
let (rows, cursor) = page(rows, limit, |(uri, at)| (at.as_str(), uri.as_str()));
let uris: Vec<&str> = rows.iter().map(|(uri, _)| uri.as_str()).collect();
let mut answer = json!({ "items": r.galleries_at(&uris)? });
put(&mut answer, "cursor", cursor);
Ok(answer)
}
/// The SQL these queries run, for the plan check.
#[cfg(test)]
pub fn queries() -> Vec<String> {
let mut all: Vec<String> = RELATION.iter().map(|(_, sql)| sql.to_string()).collect();
all.extend(COUNTS.iter().map(|s| s.to_string()));
all.push(GERM.into());
all.push(favorites_sql());
all
}
crates/grain-serve/src/read/comment.rsnew+113−0
View file
@@ -0,0 +1,113 @@
//! `social.grain.comment.getThread`: the comments on a gallery or story.
use super::moderation::{not_orphaned, shown, unblocked};
use super::views::put;
use super::{Params, Reader, list, page};
use crate::index::records::Comment;
use crate::xrpc::Failure;
use rusqlite::params;
use serde_json::{Value, json};
use std::collections::HashMap;
/// A page of the comments on `?1`, the oldest first, and how many they
/// are: of accounts shown, neither blocking nor blocked by the viewer
/// `?2`, and not answering a comment since deleted. One whose author the
/// viewer muted is kept, and said to be muted, for the client to fold.
pub fn thread_sql() -> (String, String) {
let filter = format!(
"c.subject = ?1 AND {} AND {} AND {}",
shown("c.did"),
not_orphaned("c"),
unblocked("c.did", "?2")
);
let rows = format!(
"SELECT * FROM comment c WHERE {filter} AND (c.created_at, c.uri) > (?3, ?4)
ORDER BY c.created_at, c.uri LIMIT ?5"
);
(
rows,
format!("SELECT count(*) FROM comment c WHERE {filter}"),
)
}
/// `social.grain.comment.getThread`.
pub fn get_thread(r: &Reader, p: &Params) -> Result<Value, Failure> {
let subject = r.record_uri(p, "subject")?;
let limit = p.limit(20, 100)?;
let (at, after) = p.after(true)?;
let (rows_sql, count_sql) = thread_sql();
let rows = r.rows::<Comment>(
&rows_sql,
params![subject, r.viewer, at, after, (limit + 1) as i64],
)?;
let total: i64 =
r.db.query_row(&count_sql, params![subject, r.viewer], |row| row.get(0))?;
let (rows, cursor) = page(rows, limit, |c| {
(c.record.created_at.as_str(), c.uri.as_str())
});
let mut dids: Vec<&str> = rows.iter().map(|c| c.did.as_str()).collect();
dids.sort_unstable();
dids.dedup();
let authors = r.profiles(&dids)?;
let muted: Vec<(String,)> = r.all(
"SELECT subject FROM mute WHERE did = ?1 AND subject IN (SELECT value FROM json_each(?2))",
params![r.viewer, list(&dids)],
)?;
let focus: Vec<&str> = rows
.iter()
.filter_map(|c| c.record.focus.as_deref())
.collect();
let photos = r.photos(&focus)?;
let uris: Vec<&str> = rows.iter().map(|c| c.uri.as_str()).collect();
let favorites = r.viewer_favorites(&uris)?;
let fav_counts: HashMap<String, i64> = r
.all::<(String, i64)>(
"SELECT subject, count(DISTINCT did) FROM favorite
WHERE subject IN (SELECT value FROM json_each(?1)) GROUP BY subject",
[list(&uris)],
)?
.into_iter()
.collect();
let comments: Vec<Value> = rows
.iter()
.map(|c| {
let record = &c.record;
let mut view = json!({
"uri": c.uri,
"cid": c.cid,
"author": authors[&c.did],
"text": record.text,
"createdAt": record.created_at,
"favCount": fav_counts.get(&c.uri).copied().unwrap_or(0),
});
put(
&mut view,
"facets",
record.facets.as_ref().map(|f| f.0.clone()),
);
put(&mut view, "replyTo", record.reply_to.clone());
let photo = record.focus.as_ref().and_then(|f| photos.get(f));
put(&mut view, "focus", photo.map(|photo| r.photo_view(photo)));
put(
&mut view,
"viewer",
favorites.get(&c.uri).map(|fav| json!({"fav": fav})),
);
put(
&mut view,
"muted",
muted.iter().any(|(m,)| *m == c.did).then_some(true),
);
view
})
.collect();
let mut answer = json!({"comments": comments, "totalCount": total});
put(&mut answer, "cursor", cursor);
Ok(answer)
}
#[cfg(test)]
pub fn queries() -> Vec<String> {
let (rows, count) = thread_sql();
vec![rows, count]
}
crates/grain-serve/src/read/gallery.rsnew+78−0
View file
@@ -0,0 +1,78 @@
//! `social.grain.gallery.*`: one gallery, and who favorited it.
use super::moderation::{shown, unblocked};
use super::views::put;
use super::{Params, Reader, graph, page};
use crate::index::records::Gallery;
use crate::xrpc::Failure;
use axum::http::StatusCode;
use rusqlite::params;
use serde_json::{Value, json};
/// The gallery `?1`, of an account shown.
pub fn gallery_sql() -> String {
format!(
"SELECT * FROM gallery g WHERE g.uri = ?1 AND {}",
shown("g.did")
)
}
/// `social.grain.gallery.getGallery`: the gallery, as its URI names it by
/// DID or handle, labels and all; none where its author is not shown, as
/// a link to it must not show what no feed does.
pub fn get_gallery(r: &Reader, p: &Params) -> Result<Value, Failure> {
let uri = r.record_uri(p, "gallery")?;
let rows = r.rows::<Gallery>(&gallery_sql(), [&uri])?;
if rows.is_empty() {
return Err(Failure::new(
StatusCode::BAD_REQUEST,
"NotFound",
"Gallery not found",
));
}
let gallery = r.galleries(rows)?.remove(0);
Ok(json!({ "gallery": gallery }))
}
/// A page of who favorited `?1`, the latest first, and how many they are:
/// of accounts shown, neither blocking nor blocked by the viewer `?2`.
pub fn favorites_sql() -> (String, String) {
let filter = format!(
"f.subject = ?1 AND {} AND {}",
shown("f.did"),
unblocked("f.did", "?2")
);
let rows = format!(
"SELECT f.did, f.created_at, f.uri FROM favorite f WHERE {filter}
AND (f.created_at, f.uri) < (?3, ?4) ORDER BY f.created_at DESC, f.uri DESC LIMIT ?5"
);
(
rows,
format!("SELECT count(DISTINCT f.did) FROM favorite f WHERE {filter}"),
)
}
/// `social.grain.gallery.getFavorites`: who favorited the gallery.
pub fn get_favorites(r: &Reader, p: &Params) -> Result<Value, Failure> {
let uri = r.record_uri(p, "gallery")?;
let limit = p.limit(50, 100)?;
let (at, after) = p.after(false)?;
let (rows_sql, count_sql) = favorites_sql();
let rows: Vec<(String, String, String)> = r.all(
&rows_sql,
params![uri, r.viewer, at, after, (limit + 1) as i64],
)?;
let total: i64 =
r.db.query_row(&count_sql, params![uri, r.viewer], |row| row.get(0))?;
let (rows, cursor) = page(rows, limit, |(_, at, uri)| (at.as_str(), uri.as_str()));
let dids: Vec<&str> = rows.iter().map(|(did, _, _)| did.as_str()).collect();
let mut answer = json!({"totalCount": total, "items": graph::accounts(r, &dids)?});
put(&mut answer, "cursor", cursor);
Ok(answer)
}
#[cfg(test)]
pub fn queries() -> Vec<String> {
let (rows, count) = favorites_sql();
vec![gallery_sql(), rows, count]
}
crates/grain-serve/src/read/graph.rsnew+158−0
View file
@@ -0,0 +1,158 @@
//! `social.grain.graph.*`: who follows an account, whom it follows, and
//! which of its followers the viewer follows too.
//!
//! Lists of accounts leave out those not shown and those that block, or
//! are blocked by, the viewer; not those the viewer muted. Each page holds
//! an account once, however many records it holds for the one follow.
use super::moderation::{shown, unblocked};
use super::views::put;
use super::{Params, Reader, list, page};
use crate::cdn::AVATAR;
use crate::xrpc::Failure;
use rusqlite::params;
use serde_json::{Value, json};
use std::collections::HashMap;
/// Each of `dids`, in order and once, as a list's item: who it is, and,
/// where there is a viewer, whether they follow it.
pub fn accounts(r: &Reader, dids: &[&str]) -> rusqlite::Result<Vec<Value>> {
let mut once: Vec<&str> = Vec::new();
for did in dids {
if !once.contains(did) {
once.push(did);
}
}
let actors = r.actors(&once)?;
let follows: HashMap<String, String> = r
.all::<(String, String)>(
"SELECT subject, uri FROM follow WHERE did = ?1 AND subject IN (SELECT value FROM json_each(?2))
ORDER BY uri DESC",
params![r.viewer, list(&once)],
)?
.into_iter()
.collect();
Ok(once
.iter()
.map(|did| {
let actor = &actors[*did];
let mut item = json!({"did": did, "handle": actor.handle});
if let Some(p) = &actor.profile {
put(&mut item, "displayName", p.record.display_name.clone());
put(&mut item, "description", p.record.description.clone());
let avatar = p
.record
.avatar
.as_ref()
.and_then(|a| r.cdn.url(did, &a.0, AVATAR));
put(&mut item, "avatar", avatar);
}
if r.viewer.is_some() {
let mut viewer = json!({});
put(&mut viewer, "following", follows.get(*did).cloned());
item["viewer"] = viewer;
}
item
})
.collect())
}
/// Which way a list of follows runs: to an account, or from it.
#[derive(Clone, Copy)]
pub enum Way {
Followers,
Following,
}
impl Way {
/// The column naming the account listed, and the one naming the
/// account the list is of.
fn columns(self) -> (&'static str, &'static str) {
match self {
Way::Followers => ("did", "subject"),
Way::Following => ("subject", "did"),
}
}
/// A page of the accounts listed, the latest follow first, and how
/// many they are in all.
pub fn sql(self) -> (String, String) {
let (listed, of) = self.columns();
let filter = format!(
"f.{of} = ?1 AND {} AND {}",
shown(&format!("f.{listed}")),
unblocked(&format!("f.{listed}"), "?2")
);
let rows = format!(
"SELECT f.{listed}, f.created_at, f.uri FROM follow f WHERE {filter}
AND (f.created_at, f.uri) < (?3, ?4) ORDER BY f.created_at DESC, f.uri DESC LIMIT ?5"
);
let count = format!("SELECT count(DISTINCT f.{listed}) FROM follow f WHERE {filter}");
(rows, count)
}
}
fn listing(r: &Reader, p: &Params, way: Way) -> Result<Value, Failure> {
let did = r.actor(p)?;
let limit = p.limit(50, 100)?;
let (at, after) = p.after(false)?;
let (rows_sql, count_sql) = way.sql();
let rows: Vec<(String, String, String)> = r.all(
&rows_sql,
params![did, r.viewer, at, after, (limit + 1) as i64],
)?;
let total: i64 =
r.db.query_row(&count_sql, params![did, r.viewer], |row| row.get(0))?;
let (rows, cursor) = page(rows, limit, |(_, at, uri)| (at.as_str(), uri.as_str()));
let dids: Vec<&str> = rows.iter().map(|(did, _, _)| did.as_str()).collect();
let mut answer = json!({"totalCount": total, "items": accounts(r, &dids)?});
put(&mut answer, "cursor", cursor);
Ok(answer)
}
/// `social.grain.graph.getFollowers`.
pub fn get_followers(r: &Reader, p: &Params) -> Result<Value, Failure> {
listing(r, p, Way::Followers)
}
/// `social.grain.graph.getFollowing`.
pub fn get_following(r: &Reader, p: &Params) -> Result<Value, Failure> {
listing(r, p, Way::Following)
}
/// The followers of `?1` the viewer `?2` follows, the latest follow
/// first, one row an account.
pub fn known_sql() -> String {
format!(
"SELECT f.did FROM follow f JOIN follow v ON v.did = ?2 AND v.subject = f.did
WHERE f.subject = ?1 AND {} AND {}
GROUP BY f.did ORDER BY max(f.created_at) DESC, f.did LIMIT ?3",
shown("f.did"),
unblocked("f.did", "?2")
)
}
/// `social.grain.graph.getKnownFollowers`: of an account's followers,
/// those the viewer follows, which only an account may ask.
pub fn get_known_followers(r: &Reader, p: &Params) -> Result<Value, Failure> {
let did = r.actor(p)?;
let viewer = r.signed_in()?;
let limit = p.limit(50, 50)?;
if did == viewer {
return Ok(json!({ "items": [] }));
}
let known: Vec<(String,)> = r.all(&known_sql(), params![did, viewer, limit as i64])?;
let dids: Vec<&str> = known.iter().map(|(d,)| d.as_str()).collect();
let mut items = accounts(r, &dids)?;
for item in items.iter_mut().filter_map(Value::as_object_mut) {
item.remove("viewer");
}
Ok(json!({ "items": items }))
}
#[cfg(test)]
pub fn queries() -> Vec<String> {
let (a, b) = Way::Followers.sql();
let (c, d) = Way::Following.sql();
vec![a, b, c, d, known_sql()]
}
crates/grain-serve/src/read/mod.rsnew+292−0
View file
@@ -0,0 +1,292 @@
//! grain's queries: the XRPC methods that answer from the index alone,
//! each one a function of a [`Reader`] and the call's [`Params`], ported
//! from hatk's `server/xrpc` handler by handler.
//!
//! A call is answered as Bluesky's appview answers one: its parameters
//! held to its lexicon (a `limit` out of bounds is refused, not clamped),
//! pages turned by a cursor naming the last row's sort key and URI rather
//! than an offset, and a caller named by a service token ([`viewer`])
//! whose follows, mutes, blocks and favorites fill each view's `viewer`.
//! What a read leaves out is written once, in [`moderation`]; how a
//! record reads as a view, once, in [`views`].
//!
//! [`viewer`]: crate::viewer
pub mod actor;
pub mod comment;
pub mod gallery;
pub mod graph;
pub mod moderation;
pub mod place;
pub mod story;
pub mod views;
use crate::App;
use crate::cdn::Cdn;
use crate::index::records::Record;
use crate::index::{Row, row};
use crate::xrpc::Failure;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, Uri};
use axum::{Json, Router, routing::get};
use grain::lexicon;
use rusqlite::Connection;
use rusqlite::types::FromSql;
use serde_json::Value;
use std::sync::Arc;
/// What a query reads with: the index, where blobs are drawn, and who is
/// asking.
pub struct Reader<'a> {
pub db: &'a Connection,
pub cdn: &'a Cdn,
/// The caller, if they said who they are.
pub viewer: Option<&'a str>,
}
/// A query: its answer, from the index, for its parameters.
pub type Method = fn(&Reader, &Params) -> Result<Value, Failure>;
/// Each query served, by its NSID.
pub const METHODS: &[(&str, Method)] = &[
("social.grain.actor.getProfile", actor::get_profile),
("social.grain.actor.getFavorites", actor::get_favorites),
("social.grain.gallery.getGallery", gallery::get_gallery),
("social.grain.gallery.getFavorites", gallery::get_favorites),
("social.grain.comment.getThread", comment::get_thread),
("social.grain.graph.getFollowers", graph::get_followers),
("social.grain.graph.getFollowing", graph::get_following),
(
"social.grain.graph.getKnownFollowers",
graph::get_known_followers,
),
("social.grain.story.getStories", story::get_stories),
("social.grain.story.getStory", story::get_story),
("social.grain.story.getAuthors", story::get_authors),
];
/// `router` with each of [`METHODS`] at `/xrpc/<nsid>`.
pub fn routes(router: Router<Arc<App>>) -> Router<Arc<App>> {
METHODS.iter().fold(router, |router, &(nsid, method)| {
let path = format!("/xrpc/{nsid}");
router.route(
&path,
get(
move |State(app): State<Arc<App>>, headers: HeaderMap, uri: Uri| {
answer(app, headers, uri, nsid, method)
},
),
)
})
}
/// `method`'s answer to a call of `nsid`: who is asking found first, then
/// the index read as no write runs.
async fn answer(
app: Arc<App>,
headers: HeaderMap,
uri: Uri,
nsid: &'static str,
method: Method,
) -> 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 params = Params(
Query::try_from_uri(&uri)
.map(|Query(q)| q)
.unwrap_or_default(),
);
let viewer = crate::viewer::viewer(&app, &headers, nsid).await?;
index.read(|db| {
let reader = Reader {
db,
cdn: &app.cdn,
viewer: viewer.as_deref(),
};
method(&reader, &params).map(Json)
})
}
/// A query's parameters, as its URL names them.
#[derive(Debug, Default, Clone)]
pub struct Params(pub Vec<(String, String)>);
/// The cursor a page starts from when it is the first: past every key a
/// row has, whichever way the rows run.
const PAST_THE_END: &str = "\u{10FFFF}";
impl Params {
pub fn get(&self, name: &str) -> Option<&str> {
self.0
.iter()
.find(|(k, _)| k == name)
.map(|(_, v)| v.as_str())
}
/// The parameter `name`, which the call must give.
pub fn required(&self, name: &str) -> Result<&str, Failure> {
let why = || Failure::invalid(format!("Params must have the property \"{name}\""));
self.get(name).ok_or_else(why)
}
/// The `at-uri` parameter `name`, which the call must give.
pub fn at_uri(&self, name: &str) -> Result<&str, Failure> {
let uri = self.required(name)?;
match lexicon::valid_at_uri(uri) {
true => Ok(uri),
false => Err(Failure::invalid(format!("{name} must be a valid at-uri"))),
}
}
/// `limit`: `default` where the call gives none, and otherwise from 1
/// to `max`, as the method's lexicon bounds it.
pub fn limit(&self, default: usize, max: usize) -> Result<usize, Failure> {
let Some(given) = self.get("limit") else {
return Ok(default);
};
match given.parse::<i64>() {
Err(_) => Err(Failure::invalid("limit must be an integer")),
Ok(n) if n < 1 => Err(Failure::invalid("limit can not be less than 1")),
Ok(n) if n as usize > max => Err(Failure::invalid(format!(
"limit can not be greater than {max}"
))),
Ok(n) => Ok(n as usize),
}
}
/// Where a page of rows ordered by a sort key and then their URI picks
/// up: the key and URI of the last row the cursor names, or, with no
/// cursor, a key past every row's, from whichever end `ascending`
/// starts.
pub fn after(&self, ascending: bool) -> Result<(String, String), Failure> {
let Some(cursor) = self.get("cursor") else {
let end = if ascending { "" } else { PAST_THE_END };
return Ok((end.to_string(), end.to_string()));
};
let (sort, uri) = cursor
.split_once("::")
.ok_or_else(|| Failure::invalid("malformed cursor"))?;
Ok((sort.to_string(), uri.to_string()))
}
}
/// The page of `rows`, fetched one past `limit`, and the cursor to the
/// next if there is one: the last row's sort key and URI, as `key` has
/// them.
pub fn page<T>(
mut rows: Vec<T>,
limit: usize,
key: impl Fn(&T) -> (&str, &str),
) -> (Vec<T>, Option<String>) {
if rows.len() <= limit {
return (rows, None);
}
rows.truncate(limit);
let cursor = rows.last().map(|last| {
let (sort, uri) = key(last);
format!("{sort}::{uri}")
});
(rows, cursor)
}
/// `items` as the JSON array a `json_each` parameter reads.
pub fn list<S: AsRef<str>>(items: &[S]) -> String {
Value::from(items.iter().map(|s| s.as_ref()).collect::<Vec<_>>()).to_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)
})
}
/// The DID `actor`, a DID or a handle, names, if the index knows the
/// handle: handles are kept in lower case, and one may be written with
/// an `@` before it.
pub fn resolve(&self, actor: &str) -> rusqlite::Result<Option<String>> {
if actor.starts_with("did:") {
return Ok(Some(actor.to_string()));
}
let handle = actor.trim_start_matches('@').to_lowercase();
let sql = "SELECT did FROM repo WHERE handle = ?1";
let dids: Vec<(String,)> = self.all(sql, [handle])?;
Ok(dids.into_iter().next().map(|(did,)| did))
}
/// The `actor` parameter, as the DID it names; refused where it names
/// no account the index knows, as Bluesky refuses one.
pub fn actor(&self, params: &Params) -> Result<String, Failure> {
let actor = params.required("actor")?;
if !(lexicon::valid_did(actor) || lexicon::valid_handle(actor.trim_start_matches('@'))) {
return Err(Failure::invalid("actor must be a valid did or a handle"));
}
self.resolve(actor)?
.ok_or_else(|| Failure::invalid("Profile not found"))
}
/// The `at-uri` parameter `name`, its authority made a DID where it is
/// a handle the index knows, as records are kept by their DID's URI.
pub fn record_uri(&self, params: &Params, name: &str) -> Result<String, Failure> {
let uri = params.at_uri(name)?;
let rest = &uri["at://".len()..];
let (authority, path) = rest.split_once('/').unwrap_or((rest, ""));
Ok(match self.resolve(authority)? {
Some(did) if !path.is_empty() => format!("at://{did}/{path}"),
_ => uri.to_string(),
})
}
/// The records of `T` `sql` selects, whole.
pub fn rows<T: Record>(
&self,
sql: &str,
params: impl rusqlite::Params,
) -> rusqlite::Result<Vec<Row<T>>> {
let mut stmt = self.db.prepare_cached(sql)?;
let rows = stmt.query_map(params, row::<T>)?;
rows.collect()
}
/// The rows `sql` selects, each as a tuple of its columns.
pub fn all<T: Columns>(
&self,
sql: &str,
params: impl rusqlite::Params,
) -> rusqlite::Result<Vec<T>> {
let mut stmt = self.db.prepare_cached(sql)?;
let rows = stmt.query_map(params, T::from_row)?;
rows.collect()
}
}
/// A result row as a tuple of its columns, in order.
pub trait Columns: Sized {
fn from_row(row: &rusqlite::Row) -> rusqlite::Result<Self>;
}
macro_rules! columns {
($($t:ident $i:tt),*) => {
impl<$($t: FromSql),*> Columns for ($($t,)*) {
fn from_row(row: &rusqlite::Row) -> rusqlite::Result<Self> {
Ok(($(row.get($i)?,)*))
}
}
};
}
columns!(A 0);
columns!(A 0, B 1);
columns!(A 0, B 1, C 2);
columns!(A 0, B 1, C 2, D 3, E 4);
#[cfg(test)]
mod tests;
crates/grain-serve/src/read/moderation.rsnew+53−0
View file
@@ -0,0 +1,53 @@
//! What reads leave out, written once: every query a read runs builds its
//! filters from these, as hatk's built theirs from `server/filters` and
//! `server/labels/_hidden.ts`.
//!
//! - An account an admin took down, or whose host says it is gone, is
//! shown to no one ([`shown`]).
//! - A record carrying a label that hides ([`HIDDEN_LABELS`]) is left out
//! of lists, from grain's labeler ([`unhidden`]) or its author
//! ([`unhidden_by_author`]), though one asked for by its URI still
//! answers, with its labels, for the client to warn of.
//! - Accounts that block each other see nothing of one another
//! ([`unblocked`]), and one the viewer muted is left out of feeds and
//! lists of content ([`unmuted`]), but not of lists of who follows or
//! favorited whom: a mute hides what someone says, not that they are
//! there.
//!
//! The viewer is a parameter, bound as NULL for no one: `did = NULL` holds
//! for no row, so the same SQL leaves out nothing for a caller who did not
//! say who they are.
pub use crate::index::schema::{HIDDEN_LABELS, shown, unhidden};
/// SQL that holds where neither the account `did` nor `viewer` blocks
/// the other.
pub fn unblocked(did: &str, viewer: &str) -> String {
format!(
"NOT EXISTS (SELECT 1 FROM block WHERE did = {viewer} AND subject = {did})
AND NOT EXISTS (SELECT 1 FROM block WHERE did = {did} AND subject = {viewer})"
)
}
/// SQL that holds where `viewer` has not muted the account `did`.
pub fn unmuted(did: &str, viewer: &str) -> String {
format!("NOT EXISTS (SELECT 1 FROM mute WHERE did = {viewer} AND subject = {did})")
}
/// SQL that holds where the record `uri`'s author gave it no self-label
/// that hides.
pub fn unhidden_by_author(uri: &str) -> String {
format!(
"NOT EXISTS (SELECT 1 FROM self_label WHERE uri = {uri} AND val IN ({}))",
hidden_list()
)
}
/// SQL that holds where the comment `c` answers no comment since deleted.
pub fn not_orphaned(c: &str) -> String {
format!("({c}.reply_to IS NULL OR EXISTS (SELECT 1 FROM comment p WHERE p.uri = {c}.reply_to))")
}
fn hidden_list() -> String {
HIDDEN_LABELS.map(|v| format!("'{v}'")).join(",")
}
crates/grain-serve/src/read/place.rsnew+395−0
View file
crates/grain-serve/src/read/story.rsnew+102−0
View file
crates/grain-serve/src/read/tests.rsnew+80−0
View file
crates/grain-serve/src/read/views.rsnew+509−0
View file
509 lines changed: not loaded until asked for.
crates/grain-serve/src/cdn.rsnew+161−0
View file
crates/grain-serve/src/lib.rs+30−6
View file
crates/grain-serve/src/viewer.rsnew+31−0
View file
crates/grain-serve/src/xrpc.rs+25−0
View file
crates/grain-serve/tests/common/mod.rs+10−0
View file
crates/grain-serve/tests/index_records.rs+1−0
View file
crates/grain-serve/tests/read.rsnew+920−0
View file
920 lines changed: not loaded until asked for.
crates/grain-serve/Cargo.toml+1−0
View file
lexicons/social/grain/actor/defs.json+2−2
View file
lexicons/social/grain/actor/getProfile.json+12−3
View file
lexicons/social/grain/gallery/getFavorites.json+0−1
View file
lexicons/social/grain/gallery/getGallery.json+7−1
View file
lexicons/social/grain/graph/getFollowers.json+0−1
View file
lexicons/social/grain/graph/getFollowing.json+0−1
View file
lexicons/social/grain/graph/getKnownFollowers.json+2−3
View file
Cargo.lock+1−0
View file
hatk.generated.ts+7−7
View file