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

grain.social/grain
Switch repository
Loading…

photography-focused photo sharing platform

grain.social
1 collaborator
0
2

server: indexRecords reads a caller's new records back into the index

Merged chadtmiller.com asked to merge into feature/rust from grain.social/grain:sync/index-records

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

crates/grain-serve/src/lib.rs+138−8
View file
@@ -1,18 +1,95 @@
//! grain's server. For now it answers only atproto's health check; the
//! index, the queries, ingest, login and push join it pull by pull on
//! `feature/rust`, as the plan orders them.
//! grain's server. For now it answers atproto's health check and
//! `social.grain.sync.indexRecords` ([`sync`]); the queries, ingest, login
//! and push join it pull by pull on `feature/rust`, as the plan orders
//! them.
pub mod index;
pub mod sync;
pub mod xrpc;
use axum::http::StatusCode;
use axum::{Json, Router, routing::any, routing::get};
use axum::{Json, Router, routing::any, routing::get, routing::post};
use grain::env::Ctx;
use serde_json::{Value, json};
use std::path::PathBuf;
use std::sync::Arc;
/// What the server is set to do, from its environment.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct Settings {
/// `GRAIN_INDEX`: the index's file. None keeps no index.
pub index: Option<PathBuf>,
/// `GRAIN_PUBLIC_URL`: where the server is reached, which names its
/// DID ([`service_did`]). None has no DID, and takes no service tokens.
pub public_url: Option<String>,
}
impl Settings {
pub fn from_vars(get: impl Fn(&str) -> Option<String>) -> Settings {
let set = |name: &str| get(name).filter(|v| !v.is_empty());
Settings {
index: set("GRAIN_INDEX").map(PathBuf::from),
public_url: set("GRAIN_PUBLIC_URL"),
}
}
/// The server `ctx` and these settings make: its index opened, and
/// its DID named.
pub fn app(&self, ctx: Ctx) -> anyhow::Result<App> {
let did = self.public_url.as_deref().map(service_did).transpose()?;
let index = match &self.index {
Some(path) => Some(Arc::new(index::Index::open(path)?)),
None => None,
};
Ok(App { ctx, index, did })
}
}
/// What the routes share.
pub struct App {
pub ctx: Ctx,
/// The index, if the server keeps one.
pub index: Option<Arc<index::Index>>,
/// The DID service tokens name the server by in `aud`, if it has one:
/// [`service_did`] of `GRAIN_PUBLIC_URL`.
pub did: Option<String>,
}
impl App {
/// A server that keeps no index and has no DID: one that answers its
/// health check.
pub fn bare(ctx: Ctx) -> App {
App {
ctx,
index: None,
did: None,
}
}
}
/// The DID of the server reached at `public_url`, an `https://` or
/// `http://` URL: the `did:web` of its host, as knotless-serve's is of
/// its `--public-url`. `https://grain.social` is `did:web:grain.social`.
pub fn service_did(public_url: &str) -> anyhow::Result<String> {
let url = public_url.trim_end_matches('/');
let host = url
.strip_prefix("https://")
.or_else(|| url.strip_prefix("http://"))
.filter(|h| !h.is_empty() && !h.contains('/'))
.ok_or_else(|| anyhow::anyhow!("{public_url}: a public URL is http(s)://host[:port]"))?;
Ok(format!("did:web:{}", host.replace(':', "%3A")))
}
/// The routes the server answers.
pub fn router() -> Router {
pub fn router(app: Arc<App>) -> Router {
Router::new()
.route("/xrpc/_health", get(health))
.route(
"/xrpc/social.grain.sync.indexRecords",
post(sync::index_records),
)
.route("/xrpc/{*method}", any(not_implemented))
.with_state(app)
}
/// A method the server does not serve, as atproto's XRPC servers refuse it.
@@ -29,12 +106,65 @@ async fn health() -> Json<Value> {
Json(json!({ "version": env!("CARGO_PKG_VERSION") }))
}
/// Serves on `listener` until the process is told to stop.
pub async fn serve(listener: tokio::net::TcpListener) -> anyhow::Result<()> {
axum::serve(listener, router())
/// Serves `app` on `listener` until the process is told to stop.
pub async fn serve(listener: tokio::net::TcpListener, app: Arc<App>) -> anyhow::Result<()> {
axum::serve(listener, router(app))
.with_graceful_shutdown(async {
let _ = tokio::signal::ctrl_c().await;
})
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use grain::env::Env;
#[test]
fn the_did_is_the_public_urls_host() {
let did = |url| service_did(url).unwrap();
assert_eq!(did("https://grain.social"), "did:web:grain.social");
assert_eq!(did("https://grain.social/"), "did:web:grain.social");
assert_eq!(did("http://127.0.0.1:2600"), "did:web:127.0.0.1%3A2600");
for wrong in ["grain.social", "https://", "https://grain.social/xrpc"] {
let why = service_did(wrong).unwrap_err().to_string();
assert!(why.starts_with(wrong), "{why}");
}
}
#[test]
fn settings_are_read_from_the_environment() {
let unset = Settings::from_vars(|_| None);
assert_eq!(unset, Settings::default());
let empty = Settings::from_vars(|_| Some(String::new()));
assert_eq!(empty, Settings::default());
let set = Settings::from_vars(|name| match name {
"GRAIN_INDEX" => Some("/data/index.db".into()),
"GRAIN_PUBLIC_URL" => Some("https://grain.social".into()),
_ => None,
});
assert_eq!(set.index, Some(PathBuf::from("/data/index.db")));
assert_eq!(set.public_url.as_deref(), Some("https://grain.social"));
}
#[test]
fn the_app_opens_its_index_and_names_its_did() {
let ctx = || Ctx::new(Env::from_vars(|_| None));
let dir = tempfile::tempdir().unwrap();
let settings = Settings {
index: Some(dir.path().join("index.db")),
public_url: Some("https://grain.social".into()),
};
let app = settings.app(ctx()).unwrap();
assert!(app.index.is_some());
assert_eq!(app.did.as_deref(), Some("did:web:grain.social"));
let bare = Settings::default().app(ctx()).unwrap();
assert!(bare.index.is_none() && bare.did.is_none());
let wrong = Settings {
public_url: Some("grain.social".into()),
..Settings::default()
};
assert!(wrong.app(ctx()).is_err());
}
}
crates/grain-serve/src/main.rs+9−3
View file
@@ -1,16 +1,22 @@
//! `grain-serve`: listens on `GRAIN_LISTEN`, 127.0.0.1:2600 unless set.
//! `grain-serve`: listens on `GRAIN_LISTEN`, 127.0.0.1:2600 unless set,
//! keeping the index at `GRAIN_INDEX` and taking service tokens for the
//! DID of `GRAIN_PUBLIC_URL` when set ([`grain_serve::Settings`]).
use anyhow::Context;
use grain::env::{Ctx, Env};
use std::sync::Arc;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
.init();
let settings = grain_serve::Settings::from_vars(|name| std::env::var(name).ok());
let app = settings.app(Ctx::new(Env::from_process()))?;
let addr = std::env::var("GRAIN_LISTEN").unwrap_or_else(|_| "127.0.0.1:2600".into());
let listener = tokio::net::TcpListener::bind(&addr)
.await
.with_context(|| format!("listening on {addr}"))?;
tracing::info!(%addr, "serving");
grain_serve::serve(listener).await
tracing::info!(%addr, did = ?app.did, "serving");
grain_serve::serve(listener, Arc::new(app)).await
}
crates/grain-serve/src/sync.rsnew+287−0
View file
@@ -0,0 +1,287 @@
//! Records read back from their PDS into the index, as soon as they are
//! written: `social.grain.sync.indexRecords`, which the apps and the
//! command line call after writing to their own PDS, and [`read_back`],
//! which it calls and the web's own writes will too.
//!
//! What the index takes is only ever what the PDS answers. The caller
//! 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.
use crate::App;
use crate::index::{Index, PutError, record_uri, records};
use crate::xrpc::Failure;
use axum::Json;
use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode, header};
use grain::env::Ctx;
use grain::xrpc::{Xrpc, XrpcError};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::sync::Arc;
/// The method, as a service token for it names it in `lxm`.
pub const LXM: &str = "social.grain.sync.indexRecords";
/// The most URIs one call names, as its lexicon allows.
pub const MAX_URIS: usize = 25;
/// What reading records back did, as `indexRecords` answers it.
#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ReadBack {
/// Read back from the PDS and put in the index.
pub indexed: Vec<String>,
/// Not on the PDS: deleted from the index, if it held them.
pub not_found: Vec<String>,
}
/// Why records were not read back.
#[derive(Debug)]
pub enum ReadBackError {
/// Not a record's AT-URI.
NotARecord(String),
/// A record in another repo than the one being read.
RepoMismatch(String),
/// A record of a collection the index does not keep.
UnknownCollection(String),
/// The PDS failed to answer for the record, other than to say it has
/// no such record.
Pds {
uri: String,
error: anyhow::Error,
},
/// The PDS has the record, and it does not fit its lexicon.
Refused {
uri: String,
why: String,
},
Index(rusqlite::Error),
}
impl std::fmt::Display for ReadBackError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ReadBackError::NotARecord(uri) => write!(f, "{uri} is not a record's AT-URI"),
ReadBackError::RepoMismatch(uri) => write!(f, "{uri} is not in the caller's repo"),
ReadBackError::UnknownCollection(uri) => {
write!(f, "{uri} is of a collection grain does not index")
}
ReadBackError::Pds { uri, error } => write!(f, "reading {uri}: {error:#}"),
ReadBackError::Refused { uri, why } => {
write!(f, "{uri} does not fit its lexicon: {why}")
}
ReadBackError::Index(e) => write!(f, "index: {e}"),
}
}
}
impl std::error::Error for ReadBackError {}
impl From<ReadBackError> for Failure {
fn from(e: ReadBackError) -> Failure {
let message = e.to_string();
let (status, error) = match e {
ReadBackError::NotARecord(_) => (StatusCode::BAD_REQUEST, "InvalidRequest"),
ReadBackError::RepoMismatch(_) => (StatusCode::BAD_REQUEST, "RepoMismatch"),
ReadBackError::UnknownCollection(_) => (StatusCode::BAD_REQUEST, "UnknownCollection"),
ReadBackError::Pds { .. } => (StatusCode::BAD_GATEWAY, "UpstreamFailure"),
ReadBackError::Refused { .. } => (StatusCode::BAD_REQUEST, "InvalidRequest"),
ReadBackError::Index(_) => (StatusCode::INTERNAL_SERVER_ERROR, "InternalServerError"),
};
Failure::new(status, error, message)
}
}
/// Reads each of `uris`, records in `did`'s repo, back from `pds`, that
/// repo's PDS, and puts it in `index` as the PDS has it, or deletes it
/// from `index` if the PDS has no such record. `did` is tracked from then
/// on, so that Jetstream keeps what it writes next.
///
/// A URI outside `did`'s repo, or of a collection the index does not keep,
/// refuses the whole call before anything is read. The records are read
/// at once, and each one read is applied whatever became of the others,
/// so a call that fails part way can be made again as it was. A URI named
/// twice is read once.
pub async fn read_back(
ctx: &Ctx,
index: &Index,
did: &str,
pds: &str,
uris: &[String],
) -> Result<ReadBack, ReadBackError> {
let mut named: Vec<(&str, &str, &str)> = Vec::new();
for uri in uris {
let Some((repo, collection, rkey)) = record_uri(uri) else {
return Err(ReadBackError::NotARecord(uri.clone()));
};
if repo != did {
return Err(ReadBackError::RepoMismatch(uri.clone()));
}
if records::table_of(collection).is_none() {
return Err(ReadBackError::UnknownCollection(uri.clone()));
}
if !named.iter().any(|(u, _, _)| u == uri) {
named.push((uri, collection, rkey));
}
}
let pds = Xrpc::new(ctx.http.clone(), pds);
let read = named
.iter()
.map(|&(_, collection, rkey)| get_record(&pds, did, collection, rkey));
let read = futures::future::join_all(read).await;
index.track(did).map_err(ReadBackError::Index)?;
let mut done = ReadBack::default();
let mut failed = None;
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),
Err(PutError::Refused(why)) => Err(ReadBackError::Refused {
uri: uri.to_string(),
why: why.to_string(),
}),
Err(PutError::Db(e)) => Err(ReadBackError::Index(e)),
},
Ok(None) => match index.delete(uri) {
Ok(_) => Ok(&mut done.not_found),
Err(e) => Err(ReadBackError::Index(e)),
},
Err(error) => Err(ReadBackError::Pds {
uri: uri.to_string(),
error,
}),
};
match applied {
Ok(into) => into.push(uri.to_string()),
Err(e) => failed = failed.or(Some(e)),
}
}
match failed {
Some(e) => Err(e),
None => Ok(done),
}
}
/// The record at `did`/`collection`/`rkey` as its PDS has it, with its
/// CID; none if the PDS says it has no such record.
async fn get_record(
pds: &Xrpc,
did: &str,
collection: &str,
rkey: &str,
) -> anyhow::Result<Option<(String, Value)>> {
#[derive(Deserialize)]
struct Out {
cid: String,
value: Value,
}
let params = [("repo", did), ("collection", collection), ("rkey", rkey)];
let out: Out = match pds.query("com.atproto.repo.getRecord", &params, None).await {
Ok(out) => out,
Err(e) if XrpcError::of(&e).is_some_and(|e| e.error == "RecordNotFound") => {
return Ok(None);
}
Err(e) => return Err(e),
};
Ok(Some((out.cid, out.value)))
}
#[derive(Deserialize)]
struct Input {
uris: Vec<String>,
}
/// `social.grain.sync.indexRecords`: the caller's service token checked,
/// then [`read_back`] from the PDS their DID document names. It answers
/// once every record is indexed, so a read after it sees the write.
pub async fn index_records(
State(app): State<Arc<App>>,
headers: HeaderMap,
body: Bytes,
) -> Result<Json<ReadBack>, Failure> {
let (Some(index), Some(aud)) = (&app.index, &app.did) else {
let why = "this server keeps no index, or has no DID to take service tokens for";
return Err(Failure::new(
StatusCode::NOT_IMPLEMENTED,
"MethodNotImplemented",
why,
));
};
let token = headers
.get(header::AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.ok_or_else(|| {
let why = "a service token is required";
Failure::new(StatusCode::UNAUTHORIZED, "AuthenticationRequired", why)
})?;
let did = grain::auth::verify(&app.ctx, token, aud, LXM)
.await
.map_err(|e| Failure::new(StatusCode::UNAUTHORIZED, "InvalidToken", format!("{e:#}")))?;
let input: Input = serde_json::from_slice(&body)
.map_err(|e| Failure::invalid(format!("the input is not {{uris}}: {e}")))?;
if input.uris.is_empty() || input.uris.len() > MAX_URIS {
let why = format!(
"uris names 1 to {MAX_URIS} records, not {}",
input.uris.len()
);
return Err(Failure::invalid(why));
}
let pds = grain::identity::resolve_pds(&app.ctx, &did)
.await
.map_err(|e| Failure::new(StatusCode::BAD_GATEWAY, "UpstreamFailure", format!("{e:#}")))?;
let done = read_back(&app.ctx, index, &did, &pds, &input.uris).await?;
tracing::info!(%did, indexed = done.indexed.len(), not_found = done.not_found.len(),
"indexRecords");
Ok(Json(done))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn each_failure_answers_as_xrpc_does() {
let uri = || "at://did:plc:a/social.grain.gallery/3l3qo2vuowo2a".to_string();
let cases = [
(ReadBackError::NotARecord(uri()), 400, "InvalidRequest"),
(ReadBackError::RepoMismatch(uri()), 400, "RepoMismatch"),
(
ReadBackError::UnknownCollection(uri()),
400,
"UnknownCollection",
),
(
ReadBackError::Pds {
uri: uri(),
error: anyhow::anyhow!("down"),
},
502,
"UpstreamFailure",
),
(
ReadBackError::Refused {
uri: uri(),
why: "title: required".into(),
},
400,
"InvalidRequest",
),
(
ReadBackError::Index(rusqlite::Error::InvalidQuery),
500,
"InternalServerError",
),
];
for (e, status, error) in cases {
let said = e.to_string();
let failure = Failure::from(e);
assert_eq!(failure.status.as_u16(), status, "{said}");
assert_eq!(failure.error, error);
assert_eq!(failure.message, said);
}
}
}
crates/grain-serve/src/xrpc.rsnew+39−0
View file
@@ -0,0 +1,39 @@
//! XRPC failures as grain-serve answers them: a status and atproto's
//! `{error, message}`.
use axum::Json;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use serde_json::json;
/// A call refused, or one that failed on the way.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Failure {
pub status: StatusCode,
/// The error's name: one the method's lexicon declares, or one of
/// XRPC's own, such as `InvalidRequest`.
pub error: &'static str,
pub message: String,
}
impl Failure {
pub fn new(status: StatusCode, error: &'static str, message: impl Into<String>) -> Failure {
Failure {
status,
error,
message: message.into(),
}
}
/// A request that does not say what it must: 400 `InvalidRequest`.
pub fn invalid(message: impl Into<String>) -> Failure {
Failure::new(StatusCode::BAD_REQUEST, "InvalidRequest", message)
}
}
impl IntoResponse for Failure {
fn into_response(self) -> Response {
let body = json!({ "error": self.error, "message": self.message });
(self.status, Json(body)).into_response()
}
}
crates/grain-serve/tests/common/mod.rs+19−2
View file
@@ -12,11 +12,18 @@ pub struct Served {
http: reqwest::Client,
}
/// grain-serve, serving on a free port until the test's runtime ends.
/// grain-serve with no index and no DID, serving on a free port until the
/// test's runtime ends.
pub async fn served() -> Served {
let ctx = grain::env::Ctx::new(grain::env::Env::from_vars(|_| None));
served_app(grain_serve::App::bare(ctx)).await
}
/// `app`, serving on a free port until the test's runtime ends.
pub async fn served_app(app: grain_serve::App) -> Served {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
tokio::spawn(grain_serve::serve(listener));
tokio::spawn(grain_serve::serve(listener, std::sync::Arc::new(app)));
Served {
base,
http: reqwest::Client::new(),
@@ -36,6 +43,16 @@ impl Served {
self.answer(path, res).await
}
/// A procedure called with `token`, a service token, as its bearer.
pub async fn post_as(&self, path: &str, token: &str, body: &Value) -> (u16, Value) {
let res = self
.http
.post(format!("{}{path}", self.base))
.bearer_auth(token)
.json(body);
self.answer(path, res).await
}
async fn answer(&self, path: &str, req: reqwest::RequestBuilder) -> (u16, Value) {
let res = req.send().await.unwrap();
let status = res.status().as_u16();
crates/grain-serve/tests/index_records.rsnew+349−0
View file
@@ -0,0 +1,349 @@
//! `social.grain.sync.indexRecords` against the reference PDS: an account
//! writes grain's records to its own PDS, asks that PDS for a service token
//! for grain-serve, and names the records, which grain-serve reads back
//! from the PDS into its index.
//!
//! The reference PDS resolves no lexicon over the network: it validates a
//! record only against the schemas it bundles, and keeps one of a `$type`
//! it has none for with `validationStatus: "unknown"` when `validate` is
//! unset, as the apps leave it. It bundles copies of `social.grain.gallery`,
//! `social.grain.photo` and `social.grain.favorite`, which it holds those
//! to; grain's other collections it keeps unchecked. So the tests write as
//! the apps do, and nothing reaches for grain.social.
#[path = "../../grain/tests/common/mod.rs"]
mod net;
mod common;
use grain::auth;
use grain::env::Ctx;
use grain::session::{self, Session};
use grain::xrpc::Xrpc;
use grain_serve::App;
use grain_serve::index::Index;
use grain_serve::index::records::{Follow, Gallery};
use grain_serve::sync::{LXM, ReadBackError, read_back};
use serde_json::{Value, json};
use std::sync::Arc;
const PATH: &str = "/xrpc/social.grain.sync.indexRecords";
/// The DID the server takes service tokens for. Nothing resolves it: a
/// PDS signs a token for any `aud`.
const AUD: &str = "did:web:grain.test";
const NOW: &str = "2026-10-31T12:00:00.000Z";
struct Caller {
ctx: Ctx,
me: Session,
index: Arc<Index>,
served: common::Served,
_dir: tempfile::TempDir,
}
/// A fresh account on the reference PDS, logged in, and grain-serve with
/// an index of its own, taking tokens for [`AUD`].
async fn caller() -> Caller {
let dir = tempfile::tempdir().unwrap();
let ctx = net::ctx(dir.path(), &[]);
let index = Arc::new(Index::memory().unwrap());
let served = common::served_app(App {
ctx: ctx.clone(),
index: Some(index.clone()),
did: Some(AUD.into()),
})
.await;
let account = net::account().await;
let me = session::create(&ctx, &account.did, &account.password)
.await
.unwrap();
Caller {
ctx,
me,
index,
served,
_dir: dir,
}
}
impl Caller {
fn pds(&self) -> Xrpc {
Xrpc::new(self.ctx.http.clone(), &self.me.service)
}
/// Writes `record` to `collection` in the caller's repo, as the apps
/// do: its URI and CID.
async fn write(&self, collection: &str, record: Value) -> (String, String) {
let input = json!({"repo": self.me.did, "collection": collection, "record": record});
self.written("com.atproto.repo.createRecord", input).await
}
/// Writes `record` over the one at `uri`.
async fn rewrite(&self, uri: &str, record: Value) -> (String, String) {
let (collection, rkey) = parts(uri);
let input =
json!({"repo": self.me.did, "collection": collection, "rkey": rkey, "record": record});
self.written("com.atproto.repo.putRecord", input).await
}
async fn written(&self, nsid: &str, input: Value) -> (String, String) {
let out: Value = self
.pds()
.procedure(nsid, Some(&input), Some(&self.me.auth()))
.await
.unwrap();
let field = |name: &str| out[name].as_str().unwrap().to_string();
(field("uri"), field("cid"))
}
async fn delete(&self, uri: &str) {
let (collection, rkey) = parts(uri);
let input = json!({"repo": self.me.did, "collection": collection, "rkey": rkey});
let _: Value = self
.pds()
.procedure(
"com.atproto.repo.deleteRecord",
Some(&input),
Some(&self.me.auth()),
)
.await
.unwrap();
}
/// A token from the caller's PDS for `lxm` on `aud`.
async fn token_for(&self, aud: &str, lxm: &str) -> String {
auth::service_token(&self.ctx, &self.me, aud, lxm)
.await
.unwrap()
}
async fn token(&self) -> String {
self.token_for(AUD, LXM).await
}
/// indexRecords on `uris`, with a token as the apps ask for one.
async fn index_records(&self, uris: &[&str]) -> (u16, Value) {
let token = self.token().await;
self.served
.post_as(PATH, &token, &json!({ "uris": uris }))
.await
}
}
fn parts(uri: &str) -> (&str, &str) {
let mut parts = uri.rsplitn(3, '/');
let rkey = parts.next().unwrap();
(parts.next().unwrap(), rkey)
}
fn gallery(title: &str) -> Value {
json!({"$type": "social.grain.gallery", "title": title, "createdAt": NOW})
}
#[tokio::test]
async fn records_written_are_read_back_into_the_index() {
let c = caller().await;
let (gallery_uri, cid) = c.write("social.grain.gallery", gallery("Fog")).await;
let follow =
json!({"$type": "social.grain.graph.follow", "subject": "did:plc:bob", "createdAt": NOW});
let (follow_uri, _) = c.write("social.grain.graph.follow", follow).await;
assert!(!c.index.tracks(&c.me.did));
let (status, body) = c.index_records(&[&gallery_uri, &follow_uri]).await;
assert_eq!(status, 200, "{body}");
assert_eq!(
body,
json!({"indexed": [gallery_uri, follow_uri], "notFound": []})
);
let held = c.index.get::<Gallery>(&gallery_uri).unwrap().unwrap();
assert_eq!(
(held.cid.as_str(), held.did.as_str()),
(cid.as_str(), c.me.did.as_str())
);
assert_eq!(held.record.title, "Fog");
let follows = c.index.get::<Follow>(&follow_uri).unwrap().unwrap();
assert_eq!(follows.record.subject, "did:plc:bob");
// The caller is tracked, so Jetstream keeps what they write next.
assert!(c.index.tracks(&c.me.did));
// Again, and named twice: the same record by its CID changes nothing.
let (status, again) = c.index_records(&[&gallery_uri, &gallery_uri]).await;
assert_eq!(status, 200, "{again}");
assert_eq!(again, json!({"indexed": [gallery_uri], "notFound": []}));
let still = c.index.get::<Gallery>(&gallery_uri).unwrap().unwrap();
assert_eq!(still, held);
// Edited: the PDS's new version replaces the one held.
let (_, edited) = c.rewrite(&gallery_uri, gallery("Fog, later")).await;
let (status, body) = c.index_records(&[&gallery_uri]).await;
assert_eq!(status, 200, "{body}");
let now = c.index.get::<Gallery>(&gallery_uri).unwrap().unwrap();
assert_eq!(
(now.cid.as_str(), now.record.title.as_str()),
(edited.as_str(), "Fog, later")
);
assert_eq!(now.indexed_at, held.indexed_at);
}
#[tokio::test]
async fn records_gone_from_the_pds_are_deleted() {
let c = caller().await;
let (uri, _) = c.write("social.grain.gallery", gallery("Gone")).await;
let (status, _) = c.index_records(&[&uri]).await;
assert_eq!(status, 200);
assert!(c.index.get::<Gallery>(&uri).unwrap().is_some());
c.delete(&uri).await;
let never = format!("at://{}/social.grain.gallery/3l3qo2vuowo2a", c.me.did);
let (status, body) = c.index_records(&[&uri, &never]).await;
assert_eq!(status, 200, "{body}");
assert_eq!(body, json!({"indexed": [], "notFound": [uri, never]}));
assert!(c.index.get::<Gallery>(&uri).unwrap().is_none());
// Gone is gone however often it is said.
let (status, again) = c.index_records(&[&uri]).await;
assert_eq!(
(status, again),
(200, json!({"indexed": [], "notFound": [uri]}))
);
}
#[tokio::test]
async fn uris_outside_the_callers_repo_or_the_index_are_refused() {
let c = caller().await;
let (mine, _) = c.write("social.grain.gallery", gallery("Mine")).await;
let theirs = "at://did:plc:someoneelse/social.grain.gallery/3l3qo2vuowo2a";
let like = format!("at://{}/app.bsky.feed.like/3l3qo2vuowo2a", c.me.did);
let no_key = format!("at://{}/social.grain.gallery", c.me.did);
for (uris, error) in [
(vec![mine.as_str(), theirs], "RepoMismatch"),
(vec![mine.as_str(), like.as_str()], "UnknownCollection"),
(vec![no_key.as_str()], "InvalidRequest"),
] {
let (status, body) = c.index_records(&uris).await;
assert_eq!(
(status, body["error"].as_str()),
(400, Some(error)),
"{body}"
);
}
// Refused before anything was read, the caller's own record too.
assert!(c.index.get::<Gallery>(&mine).unwrap().is_none());
assert!(!c.index.tracks(&c.me.did));
let token = c.token().await;
let many: Vec<String> = (0..26).map(|_| mine.clone()).collect();
for input in [
json!({"uris": []}),
json!({"uris": many}),
json!({"uri": mine}),
json!("not an object"),
] {
let (status, body) = c.served.post_as(PATH, &token, &input).await;
assert_eq!(
(status, body["error"].as_str()),
(400, Some("InvalidRequest")),
"{input}: {body}"
);
}
}
#[tokio::test]
async fn a_record_that_does_not_fit_its_lexicon_is_not_indexed() {
let c = caller().await;
// A follow of no one: the reference PDS bundles no schema for follows,
// so it keeps one, as a PDS that checks nothing would.
let follow = json!({"$type": "social.grain.graph.follow", "createdAt": NOW});
let (uri, _) = c.write("social.grain.graph.follow", follow).await;
let (status, body) = c.index_records(&[&uri]).await;
assert_eq!(
(status, body["error"].as_str()),
(400, Some("InvalidRequest")),
"{body}"
);
assert!(
body["message"]
.as_str()
.unwrap()
.contains("does not fit its lexicon")
);
assert!(c.index.get::<Follow>(&uri).unwrap().is_none());
}
#[tokio::test]
async fn only_a_service_token_for_this_method_on_this_server_lets_a_call_in() {
let c = caller().await;
let (uri, _) = c.write("social.grain.gallery", gallery("Locked")).await;
let input = json!({ "uris": [uri] });
let (status, body) = c.served.post(PATH, &input).await;
assert_eq!(
(status, body["error"].as_str()),
(401, Some("AuthenticationRequired"))
);
let other_server = c.token_for("did:web:other.test", LXM).await;
let other_method = c.token_for(AUD, "social.grain.sync.other").await;
// Signed by a key that is not the account's.
let stranger = grain::crypto::P256Key::generate().sign_jwt(
&json!({"typ": "JWT"}),
&json!({"iss": c.me.did, "aud": AUD, "lxm": LXM, "exp": grain::keys::now_secs() + 60}),
);
// Good for two seconds.
let exp = grain::keys::now_secs() + 2;
let expired = auth::lasting_token(&c.ctx, &c.me, AUD, LXM, exp)
.await
.unwrap();
while grain::keys::now_secs() <= exp {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
for token in [
other_server,
other_method,
stranger,
expired,
"not.a.token".into(),
] {
let (status, body) = c.served.post_as(PATH, &token, &input).await;
assert_eq!(
(status, body["error"].as_str()),
(401, Some("InvalidToken")),
"{body}"
);
}
assert!(c.index.get::<Gallery>(&uri).unwrap().is_none());
}
#[tokio::test]
async fn a_server_with_no_index_or_did_does_not_serve_it() {
let s = common::served().await;
let (status, body) = s
.post_as(
PATH,
"token",
&json!({"uris": ["at://did:plc:a/social.grain.gallery/3l3qo2vuowo2a"]}),
)
.await;
assert_eq!(
(status, body["error"].as_str()),
(501, Some("MethodNotImplemented"))
);
}
#[tokio::test]
async fn a_pds_that_does_not_answer_fails_the_read_back() {
let c = caller().await;
let (uri, _) = c.write("social.grain.gallery", gallery("Elsewhere")).await;
let nowhere = "http://127.0.0.1:1";
let failed = read_back(
&c.ctx,
&c.index,
&c.me.did,
nowhere,
std::slice::from_ref(&uri),
)
.await;
match failed {
Err(ReadBackError::Pds { uri: failed, .. }) => assert_eq!(failed, uri),
other => panic!("{other:?}"),
}
}
crates/grain-serve/Cargo.toml+1−0
View file
@@ -10,6 +10,7 @@ rust-version.workspace = true
grain.workspace = true
anyhow.workspace = true
axum.workspace = true
futures.workspace = true
rusqlite.workspace = true
serde.workspace = true
serde_json.workspace = true
Cargo.lock+1−0
View file
@@ -598,6 +598,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"axum",
"futures",
"grain",
"reqwest",
"rusqlite",