tomdeboneandClaude Opus 5
124a90dc07
Gegenstück zum subscribeRepos-Endpoint: WebSocket-Consumer mit persistiertem seq-Cursor, Reconnect-Backoff und Behandlung von #info/OutdatedCursor. Eigene Cursor-Tabelle statt einer Zeile in jetstream_cursor: dort steht ein time_us in der Größenordnung 1.7e15, die seq ist ein kleiner Zähler ab 1. Geteilt hätte GREATEST den PDS-Cursor sofort in eine Zukunft geschoben, die die PDS nie erreicht. Kein neuer Indexer-Pfad — jede Op wird in die Single-Op-Form übersetzt, die apply_commit schon vom Jetstream kennt. Push und Firehose liefern denselben Commit doppelt; das ist unkritisch, weil die Schreibpfade Upserts sind und der Dedupe-Index der Notifications den Rest abfängt. Mit einem Test festgehalten statt vorausgesetzt. Der CAR-Reader ist neu (es gab nur einen Writer, und der liegt in einem Binary-Crate ohne lib-Target). Der CBOR-Reader arbeitet mit explizitem Offset, weil ein Frame zwei hintereinander geschriebene Werte sind, und akzeptiert CID-Links in beiden Schreibweisen — die Blöcke tragen Strings. /healthz meldet beide Ströme getrennt; sie fallen unabhängig voneinander aus. Verifiziert mit totem Push-Ziel: der Post kam trotzdem an. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX
1876 lines
70 KiB
Rust
1876 lines
70 KiB
Rust
//! AppView HTTP routes.
|
||
//!
|
||
//! Three groups:
|
||
//! - `root` + `healthz`: public liveness / info probes.
|
||
//! - `timeline_*` / `profile_*` / `search`: read API the Tauri client
|
||
//! calls to render the UI. Reads from `posts` (and the helper
|
||
//! `follows` / `likes` tables) — never writes.
|
||
//! - `ingest_commit`: the internal-only writer used by the PDS, owned
|
||
//! in `crate::ingest`.
|
||
//!
|
||
//! ## Public vs. private
|
||
//!
|
||
//! Most of what the AppView serves is public by construction: in AT
|
||
//! Proto a post, a profile, a follow edge and a like are records in a
|
||
//! public repo, replicated over the firehose. Those endpoints
|
||
//! (`/api/profile*`, `/api/search`, `/api/post/*`, `/api/thread*`,
|
||
//! `/api/followers`, `/api/following`) need no credential.
|
||
//!
|
||
//! Two things are *not* public, and they are the reason this service
|
||
//! has authentication at all:
|
||
//!
|
||
//! - `/api/notifications*` — who interacted with you, and the read
|
||
//! state of that list. Nothing in the protocol makes it readable by
|
||
//! anyone but the recipient.
|
||
//! - `/api/timeline/home` — the timeline is assembled from the
|
||
//! viewer's follow graph, so serving it to an arbitrary `did`
|
||
//! parameter answers "what does this person's feed look like" for
|
||
//! any DID a caller cares to type.
|
||
//!
|
||
//! Both require a valid PDS-issued access token whose `sub` equals the
|
||
//! `did` in the request ([`crate::auth`]). The check is two steps on
|
||
//! purpose: the extractor proves *who* the caller is, and
|
||
//! `ensure_matches` proves they are asking about themselves.
|
||
//!
|
||
//! The cursor format used by `timeline_home` is opaque: it's a
|
||
//! `base64url(micros):uri` pair, which is what [`cursor::encode`] and
|
||
//! [`cursor::decode`] produce/consume.
|
||
|
||
use axum::{
|
||
extract::{Path, Query, State},
|
||
http::StatusCode,
|
||
response::IntoResponse,
|
||
routing::{get, post},
|
||
Json, Router,
|
||
};
|
||
use axum::http::{header, HeaderValue, Method};
|
||
use chrono::{DateTime, TimeZone, Utc};
|
||
use serde::{Deserialize, Serialize};
|
||
use serde_json::{json, Value};
|
||
use tower_http::cors::{AllowOrigin, Any, CorsLayer};
|
||
use tracing::warn;
|
||
|
||
use crate::auth::AuthedDid;
|
||
use crate::state::AppState;
|
||
|
||
pub mod cursor;
|
||
pub mod types;
|
||
|
||
use types::{
|
||
ActorListResponse, ActorProfile, NotificationCountResponse, NotificationItem,
|
||
NotificationSeenResponse, NotificationsResponse, PostRow, PostRowWithIndexed,
|
||
ProfileResponse, SearchResponse, ThreadFullResponse, TimelineResponse,
|
||
};
|
||
|
||
/// Build the CORS layer for the browser-facing routes.
|
||
///
|
||
/// The Tauri webview is a cross-origin caller: in dev its origin is the
|
||
/// Vite server (`http://127.0.0.1:1430`), in a packaged build it is a
|
||
/// platform-specific scheme — `tauri://localhost` on macOS/iOS,
|
||
/// `http://tauri.localhost` on Windows. Either way the browser drops
|
||
/// the response unless we send `Access-Control-Allow-Origin`.
|
||
///
|
||
/// `APPVIEW_CORS_ORIGINS` is a comma-separated allowlist, e.g.
|
||
/// `tauri://localhost,http://127.0.0.1:1430`. When it is unset we keep
|
||
/// the historic wildcard so no existing deployment breaks on upgrade —
|
||
/// [`crate::auth::log_startup_posture`] warns about that at startup.
|
||
///
|
||
/// `Authorization` has to be in `allow_headers`: it is not a
|
||
/// CORS-safelisted header, so without it the browser's preflight fails
|
||
/// and the authenticated endpoints become unreachable from the webview
|
||
/// — with an error that looks nothing like an auth problem.
|
||
///
|
||
/// `allow_credentials` stays off. We authenticate with a bearer token
|
||
/// the client attaches deliberately, never with an ambient cookie, so
|
||
/// there is nothing for a hostile page to replay — and turning it on
|
||
/// would additionally make the wildcard origin illegal.
|
||
fn cors_layer(cfg: &at_shared::config::AppConfig) -> CorsLayer {
|
||
let base = CorsLayer::new()
|
||
.allow_methods([Method::GET, Method::POST, Method::OPTIONS])
|
||
.allow_headers([
|
||
header::AUTHORIZATION,
|
||
header::CONTENT_TYPE,
|
||
header::ACCEPT,
|
||
]);
|
||
|
||
if cfg.appview_cors_origins.is_empty() {
|
||
return base.allow_origin(Any);
|
||
}
|
||
|
||
// Anything unparseable as a header value is dropped with a warning
|
||
// rather than panicking the process — a stray quote in the env var
|
||
// must not take the service down. If *every* entry is bad we fall
|
||
// back to the wildcard and say so, because an empty allowlist would
|
||
// silently break every browser client.
|
||
let mut origins: Vec<HeaderValue> = Vec::new();
|
||
for raw in &cfg.appview_cors_origins {
|
||
match HeaderValue::from_str(raw) {
|
||
Ok(v) => origins.push(v),
|
||
Err(_) => warn!(origin = %raw, "APPVIEW_CORS_ORIGINS: ignoring unparseable origin"),
|
||
}
|
||
}
|
||
if origins.is_empty() {
|
||
warn!("APPVIEW_CORS_ORIGINS contained no usable origin; falling back to allow-any");
|
||
return base.allow_origin(Any);
|
||
}
|
||
base.allow_origin(AllowOrigin::list(origins))
|
||
}
|
||
|
||
pub fn router(state: AppState) -> Router {
|
||
let cors = cors_layer(&state.cfg);
|
||
|
||
// Browser-facing surface. Everything here answers a `fetch()` from
|
||
// the desktop client, so it carries the CORS layer.
|
||
let api = Router::new()
|
||
.route("/", get(root))
|
||
// --- private: token required, `sub` must equal `did` ---
|
||
.route("/api/timeline/home", get(timeline_home))
|
||
.route("/api/notifications", get(notifications))
|
||
.route("/api/notifications/count", get(notifications_count))
|
||
.route("/api/notifications/seen", post(notifications_seen))
|
||
// --- public: AT Proto public records ---
|
||
.route("/api/profile", get(profile_query))
|
||
.route("/api/profile/:handle", get(profile_path))
|
||
.route("/api/search", get(search))
|
||
.route("/api/post/*uri", get(post_by_uri))
|
||
// Two spellings of the same handler. `/api/thread/<uri>`
|
||
// mirrors the existing `/api/post/*uri` (the wildcard capture
|
||
// keeps the `did:plc:…` colons verbatim), while
|
||
// `/api/thread?uri=…` is the form that survives an
|
||
// over-eager URL normaliser in a proxy. They share one
|
||
// implementation, so they can't drift.
|
||
.route("/api/thread", get(thread_query))
|
||
.route("/api/thread/*uri", get(thread_path))
|
||
.route("/api/followers", get(followers))
|
||
.route("/api/following", get(following))
|
||
.route("/healthz", get(healthz))
|
||
.layer(cors);
|
||
|
||
// Server-to-server surface. `/internal/ingest-commit` is called by
|
||
// the PDS with a shared secret, never by a browser, so it stays
|
||
// outside the CORS layer: handing it an
|
||
// `Access-Control-Allow-Origin` header would only ever help a web
|
||
// page try to write to the index.
|
||
let internal = Router::new()
|
||
.route("/internal/ingest-commit", post(crate::ingest::ingest_commit));
|
||
|
||
api.merge(internal).with_state(state)
|
||
}
|
||
|
||
async fn root() -> Json<Value> {
|
||
Json(json!({
|
||
"name": "maarcadetweet-appview",
|
||
"version": env!("CARGO_PKG_VERSION"),
|
||
}))
|
||
}
|
||
|
||
// -- timeline ---------------------------------------------------------------
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct TimelineQuery {
|
||
/// The DID of the user whose timeline we're building. Required —
|
||
/// without it we have no way to do (future) follow-graph filtering
|
||
/// and no place to anchor the pagination.
|
||
did: String,
|
||
/// Page size. Defaults to 30, hard-capped at 100.
|
||
#[serde(default)]
|
||
limit: Option<i64>,
|
||
/// Opaque cursor returned by a previous call.
|
||
#[serde(default)]
|
||
cursor: Option<String>,
|
||
}
|
||
|
||
const DEFAULT_LIMIT: i64 = 30;
|
||
const MAX_LIMIT: i64 = 100;
|
||
/// Hard cap on the number of DIDs we'll filter a timeline query by.
|
||
///
|
||
/// Prevents an unbounded `did = ANY($1::text[])` array from a power
|
||
/// user with thousands of follows — SQL injection isn't the worry
|
||
/// (the bind is parameterised) but a 50k-element array still has to be
|
||
/// shipped across the wire and parsed by Postgres on every page
|
||
/// request. 1000 covers essentially every realistic follow graph; if
|
||
/// a user exceeds it we cap deterministically (sorted by DID) so the
|
||
/// result set is stable across requests, and we always keep the
|
||
/// requesting user's own DID in the set so their own posts still
|
||
/// surface.
|
||
const MAX_FOLLOWED_DIDS: usize = 1000;
|
||
|
||
/// `GET /api/timeline/home?did=…`
|
||
///
|
||
/// **Authenticated.** The timeline is derived from the viewer's follow
|
||
/// graph, so `did` must be the caller's own DID — otherwise this
|
||
/// endpoint would answer "what does this account's feed look like" for
|
||
/// any DID at all.
|
||
async fn timeline_home(
|
||
State(state): State<AppState>,
|
||
auth: AuthedDid,
|
||
Query(q): Query<TimelineQuery>,
|
||
) -> Result<Json<TimelineResponse>, (StatusCode, Json<Value>)> {
|
||
if q.did.is_empty() {
|
||
return Err(bad_request("did is required"));
|
||
}
|
||
auth.ensure_matches(&q.did)?;
|
||
let limit = clamp_limit(q.limit);
|
||
|
||
// Look up the set of DIDs this user follows, then build the
|
||
// `target_dids` set we'll filter `posts` by:
|
||
//
|
||
// 1. Start with the followees from the `follows` table.
|
||
// 2. Add the user's own DID so they always see their own posts.
|
||
// 3. Deduplicate.
|
||
// 4. Cap at MAX_FOLLOWED_DIDS. If we'd exceed the cap, sort by
|
||
// DID (stable across requests), trim to the cap, then if the
|
||
// requesting user's DID was trimmed out and there's still
|
||
// room in the cap, put them back in. This guarantees the
|
||
// user's own posts are visible regardless of follow graph
|
||
// size.
|
||
//
|
||
// If the followee set is empty *and* adding the user's own DID
|
||
// leaves us with only that one entry, we treat it as the cold
|
||
// start case and fall back to the global recent feed — new users
|
||
// see the world before they have a graph.
|
||
let followed_dids: Vec<String> = sqlx::query_scalar(
|
||
"SELECT subject_did FROM follows WHERE follower_did = $1",
|
||
)
|
||
.bind(&q.did)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
|
||
// Build the deduped target set. `followed_dids` may contain the
|
||
// user's own DID (self-follow is rare but legal in the protocol),
|
||
// so we dedup with full sort+dedup rather than relying on a
|
||
// presence-check.
|
||
let has_real_follows = followed_dids.iter().any(|d| d != &q.did);
|
||
let mut target_dids: Vec<String> = if has_real_follows {
|
||
let mut v = followed_dids;
|
||
v.push(q.did.clone());
|
||
v.sort();
|
||
v.dedup();
|
||
v
|
||
} else {
|
||
// Cold start: no real follow graph yet. Fall through to the
|
||
// global-recent branch below.
|
||
vec![]
|
||
};
|
||
|
||
// Cap deterministically. After capping, the user's own DID must
|
||
// still be in the set — that's a hard invariant for "user sees
|
||
// their own posts".
|
||
if target_dids.len() > MAX_FOLLOWED_DIDS {
|
||
// Drop own, sort, trim to leave room for own, re-add own.
|
||
// Sorting makes the truncation deterministic (we drop the
|
||
// lex-greatest (N − MAX_FOLLOWED_DIDS) followees, not a random
|
||
// subset).
|
||
let own = q.did.clone();
|
||
target_dids.retain(|d| d != &own);
|
||
target_dids.sort();
|
||
// Reserve one slot for `own` so the final length is exactly
|
||
// MAX_FOLLOWED_DIDS.
|
||
target_dids.truncate(MAX_FOLLOWED_DIDS - 1);
|
||
target_dids.push(own);
|
||
target_dids.sort();
|
||
}
|
||
|
||
let keyset = decode_cursor(q.cursor.as_deref())?;
|
||
let cursor_ts: Option<DateTime<Utc>> = keyset.as_ref().map(|(ts, _)| *ts);
|
||
let cursor_uri: Option<String> = keyset.map(|(_, uri)| uri);
|
||
// Fetch limit+1 to know if there's a next page without a second
|
||
// round trip.
|
||
let fetch = limit + 1;
|
||
|
||
// We select `indexed_at` alongside the post row so the cursor
|
||
// builder can use it directly without a second round trip. The
|
||
// helper inner function collapses the four (followed? + cursor?)
|
||
// branches into one query_as per shape.
|
||
// Cold-start branch: user has no real follow graph (only self, or
|
||
// nothing). Show the global recent feed.
|
||
let mut rows: Vec<PostRowWithIndexed> = if target_dids.is_empty() {
|
||
match cursor_ts {
|
||
Some(ts) => sqlx::query_as::<_, PostRowWithIndexed>(
|
||
r#"SELECT uri, did, handle, rkey, collection, text, cid,
|
||
parent_uri, root_uri, embed, langs, created_at,
|
||
indexed_at
|
||
FROM posts
|
||
WHERE collection IN ('app.twi.post','app.bsky.feed.post')
|
||
AND (indexed_at, uri) < ($1, $2)
|
||
ORDER BY indexed_at DESC, uri DESC
|
||
LIMIT $3"#,
|
||
)
|
||
.bind(ts)
|
||
.bind(cursor_uri.as_deref().unwrap())
|
||
.bind(fetch)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?,
|
||
None => sqlx::query_as::<_, PostRowWithIndexed>(
|
||
r#"SELECT uri, did, handle, rkey, collection, text, cid,
|
||
parent_uri, root_uri, embed, langs, created_at,
|
||
indexed_at
|
||
FROM posts
|
||
WHERE collection IN ('app.twi.post','app.bsky.feed.post')
|
||
ORDER BY indexed_at DESC, uri DESC
|
||
LIMIT $1"#,
|
||
)
|
||
.bind(fetch)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?,
|
||
}
|
||
} else {
|
||
// Graph-aware branch: the followee set plus the requesting
|
||
// user's own DID. `target_dids` has been deduped and capped at
|
||
// MAX_FOLLOWED_DIDS, and the user's own DID is guaranteed to
|
||
// be in the set.
|
||
//
|
||
// ## Why this is a LATERAL and not `did = ANY($2)`
|
||
//
|
||
// The straightforward `WHERE did = ANY($2) ORDER BY indexed_at
|
||
// DESC LIMIT n` is a plan-stability trap once
|
||
// `posts_feed_indexed_at_uri_idx` exists (migration 0009, added
|
||
// for the cold-start feed). The planner sees an index that
|
||
// already yields rows in `indexed_at DESC` order and assumes it
|
||
// will hit `n` matching rows early — so it walks the global
|
||
// feed and filters. When the followees are sparse (a fresh
|
||
// account following accounts that haven't posted), "early"
|
||
// means millions of rows: measured on the dev instance, 2.87 M
|
||
// rows discarded and 28 s per request, while the same query
|
||
// took 62 ms with the per-DID index. It also flipped between
|
||
// the two plans depending on how often the prepared statement
|
||
// had run, so it looked intermittent.
|
||
//
|
||
// Expressing the intent — "for each followee, their newest
|
||
// posts, merged" — takes that plan off the table: `unnest` is a
|
||
// relation the planner can size, and each iteration is a bounded
|
||
// range scan on `posts_did_indexed_at_uri_idx`. Fetching `$1`
|
||
// per followee is what makes it correct: the global top-N is
|
||
// always a subset of the union of the per-followee top-Ns.
|
||
match cursor_ts {
|
||
Some(ts) => sqlx::query_as::<_, PostRowWithIndexed>(
|
||
r#"SELECT t.uri, t.did, t.handle, t.rkey, t.collection,
|
||
t.text, t.cid, t.parent_uri, t.root_uri,
|
||
t.embed, t.langs, t.created_at, t.indexed_at
|
||
FROM unnest($2::text[]) AS f(did)
|
||
CROSS JOIN LATERAL (
|
||
SELECT p.uri, p.did, p.handle, p.rkey, p.collection,
|
||
p.text, p.cid, p.parent_uri, p.root_uri,
|
||
p.embed, p.langs, p.created_at, p.indexed_at
|
||
FROM posts p
|
||
WHERE p.did = f.did
|
||
AND p.collection IN ('app.twi.post','app.bsky.feed.post')
|
||
AND (p.indexed_at, p.uri) < ($3, $4)
|
||
ORDER BY p.indexed_at DESC, p.uri DESC
|
||
LIMIT $1
|
||
) t
|
||
ORDER BY t.indexed_at DESC, t.uri DESC
|
||
LIMIT $1"#,
|
||
)
|
||
.bind(fetch)
|
||
.bind(&target_dids)
|
||
.bind(ts)
|
||
.bind(cursor_uri.as_deref().unwrap())
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?,
|
||
None => sqlx::query_as::<_, PostRowWithIndexed>(
|
||
r#"SELECT t.uri, t.did, t.handle, t.rkey, t.collection,
|
||
t.text, t.cid, t.parent_uri, t.root_uri,
|
||
t.embed, t.langs, t.created_at, t.indexed_at
|
||
FROM unnest($2::text[]) AS f(did)
|
||
CROSS JOIN LATERAL (
|
||
SELECT p.uri, p.did, p.handle, p.rkey, p.collection,
|
||
p.text, p.cid, p.parent_uri, p.root_uri,
|
||
p.embed, p.langs, p.created_at, p.indexed_at
|
||
FROM posts p
|
||
WHERE p.did = f.did
|
||
AND p.collection IN ('app.twi.post','app.bsky.feed.post')
|
||
ORDER BY p.indexed_at DESC, p.uri DESC
|
||
LIMIT $1
|
||
) t
|
||
ORDER BY t.indexed_at DESC, t.uri DESC
|
||
LIMIT $1"#,
|
||
)
|
||
.bind(fetch)
|
||
.bind(&target_dids)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?,
|
||
}
|
||
};
|
||
|
||
// We fetched limit+1 to peek for a next page. If we got more than
|
||
// `limit`, drop the extra and remember its `indexed_at` to encode
|
||
// into the next cursor.
|
||
let next_ts_uri: Option<(DateTime<Utc>, String)> = if rows.len() as i64 > limit {
|
||
rows.truncate(limit as usize);
|
||
rows.last()
|
||
.map(|p| (p.indexed_at, p.uri.clone()))
|
||
} else {
|
||
None
|
||
};
|
||
|
||
// Strip the `indexed_at` companion column from the response.
|
||
let mut posts: Vec<PostRow> = rows.into_iter().map(Into::into).collect();
|
||
|
||
// Derive a `display_handle` for posts that have an empty `handle`
|
||
// column (Jetstream-only path). We mutate a clone with a populated
|
||
// `handle` so the client doesn't have to guess.
|
||
decorate_handles(&mut posts);
|
||
|
||
let next_cursor = next_ts_uri
|
||
.map(|(ts, uri)| cursor::encode(ts, &uri));
|
||
|
||
Ok(Json(TimelineResponse {
|
||
posts,
|
||
cursor: next_cursor,
|
||
}))
|
||
}
|
||
|
||
// -- profile ----------------------------------------------------------------
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct ProfileQuery {
|
||
#[serde(default)]
|
||
did: Option<String>,
|
||
#[serde(default)]
|
||
handle: Option<String>,
|
||
}
|
||
|
||
async fn profile_query(
|
||
State(state): State<AppState>,
|
||
Query(q): Query<ProfileQuery>,
|
||
) -> Result<Json<ProfileResponse>, (StatusCode, Json<Value>)> {
|
||
let did = q.did.clone().or(q.handle.clone());
|
||
let handle = q.handle.clone();
|
||
resolve_profile(&state, did.as_deref(), handle.as_deref()).await
|
||
}
|
||
|
||
async fn profile_path(
|
||
State(state): State<AppState>,
|
||
Path(handle): Path<String>,
|
||
) -> Result<Json<ProfileResponse>, (StatusCode, Json<Value>)> {
|
||
resolve_profile(&state, None, Some(&handle)).await
|
||
}
|
||
|
||
async fn resolve_profile(
|
||
state: &AppState,
|
||
did: Option<&str>,
|
||
handle: Option<&str>,
|
||
) -> Result<Json<ProfileResponse>, (StatusCode, Json<Value>)> {
|
||
// Reject empty params up front — otherwise the empty-string DID/handle
|
||
// produces a valid-looking 200 with zero posts.
|
||
if let Some(d) = did {
|
||
if d.is_empty() {
|
||
return Err(bad_request("did is required"));
|
||
}
|
||
}
|
||
if let Some(h) = handle {
|
||
if h.is_empty() {
|
||
return Err(bad_request("handle is required"));
|
||
}
|
||
}
|
||
|
||
// Strip a leading '@' on the handle — the UI passes `@alice` style.
|
||
let handle_clean = handle.map(|h| h.trim_start_matches('@').to_string());
|
||
|
||
// Try by DID first if we have it; fall back to handle lookup.
|
||
let target_did: Option<String> = if let Some(d) = did {
|
||
Some(d.to_string())
|
||
} else if let Some(ref h) = handle_clean {
|
||
// Order by `indexed_at DESC` so we get the most recent DID for
|
||
// this handle (a single user can re-use a handle if account
|
||
// history allows, but the latest is the active one).
|
||
//
|
||
// Prefer the `profiles` cache over `posts` — a user can have
|
||
// a profile row (set via PDS push before posting) but no posts
|
||
// yet, and we want the profile page to render with the right
|
||
// DID rather than synthesise an empty one.
|
||
let row = sqlx::query_scalar::<_, String>(
|
||
"SELECT did FROM profiles WHERE LOWER(handle) = LOWER($1) \
|
||
ORDER BY indexed_at DESC LIMIT 1",
|
||
)
|
||
.bind(h)
|
||
.fetch_optional(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
if row.is_some() {
|
||
row
|
||
} else {
|
||
sqlx::query_scalar::<_, String>(
|
||
"SELECT did FROM posts WHERE handle = $1 \
|
||
ORDER BY indexed_at DESC LIMIT 1",
|
||
)
|
||
.bind(h)
|
||
.fetch_optional(&state.db)
|
||
.await
|
||
.map_err(db_err)?
|
||
}
|
||
} else {
|
||
None
|
||
};
|
||
|
||
let Some(target_did) = target_did else {
|
||
// No posts indexed for this handle yet — typical for
|
||
// local-PDS users whose posts haven't been ingested into
|
||
// the Jetstream yet (and the AppView indexes only ingested
|
||
// posts, never queries upstream PDSs for handle→DID
|
||
// resolution). Instead of bubbling a 404 to the UI which
|
||
// then shows an error toast instead of an empty profile,
|
||
// synthesise a profile row with the requested handle and
|
||
// zero counts. The `did` is left empty; the UI's
|
||
// `displayHandle()` falls back to the handle and the
|
||
// "copy did" button just copies an empty string.
|
||
let display = handle_clean.unwrap_or_default();
|
||
return Ok(Json(ProfileResponse {
|
||
did: String::new(),
|
||
handle: display,
|
||
posts: Vec::new(),
|
||
followers: 0,
|
||
following: 0,
|
||
display_name: None,
|
||
description: None,
|
||
avatar_cid: None,
|
||
banner_cid: None,
|
||
post_count: 0,
|
||
}));
|
||
};
|
||
|
||
// Fetch the user's most recent posts (newest first). We return up to
|
||
// 50 — enough for a profile view, and the client can paginate with
|
||
// /api/timeline/home if it needs more.
|
||
let mut posts: Vec<PostRow> = sqlx::query_as::<_, PostRow>(
|
||
r#"SELECT uri, did, handle, rkey, collection, text, cid,
|
||
parent_uri, root_uri, embed, langs, created_at,
|
||
like_count, repost_count
|
||
FROM posts
|
||
WHERE did = $1
|
||
AND collection IN ('app.twi.post','app.bsky.feed.post')
|
||
ORDER BY indexed_at DESC, uri DESC
|
||
LIMIT 50"#,
|
||
)
|
||
.bind(&target_did)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
|
||
// Pick the best available handle: explicit query handle, then the
|
||
// first non-empty handle from the posts we just pulled.
|
||
let display_handle = handle_clean
|
||
.clone()
|
||
.or_else(|| {
|
||
posts
|
||
.iter()
|
||
.find(|p| !p.handle.is_empty())
|
||
.map(|p| p.handle.clone())
|
||
})
|
||
.unwrap_or_else(|| {
|
||
// Last-resort synthetic handle. Bare, without the `@`:
|
||
// every consumer of this field renders `@{handle}` itself.
|
||
short_did_bare(&target_did)
|
||
});
|
||
|
||
decorate_handles(&mut posts);
|
||
|
||
let followers: i64 = sqlx::query_scalar(
|
||
"SELECT COUNT(*)::BIGINT FROM follows WHERE subject_did = $1",
|
||
)
|
||
.bind(&target_did)
|
||
.fetch_one(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
|
||
let following: i64 = sqlx::query_scalar(
|
||
"SELECT COUNT(*)::BIGINT FROM follows WHERE follower_did = $1",
|
||
)
|
||
.bind(&target_did)
|
||
.fetch_one(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
|
||
// Look up the denormalised profile metadata for this DID. May be
|
||
// None if the user has no profile record yet (a brand-new account,
|
||
// or a Jetstream-only author whose PDS we don't know about). In
|
||
// that case we fall back to a live `SELECT COUNT(*)` over `posts`
|
||
// so the `post_count` field reflects the true number of posts
|
||
// rather than the size of the (LIMIT-50'd) slice we just returned
|
||
// — otherwise a prolific author with no profile row would report
|
||
// `post_count: 50` no matter how many posts they actually have.
|
||
let profile_row: Option<(Option<String>, Option<String>, Option<String>, Option<String>, i64)> =
|
||
sqlx::query_as(
|
||
"SELECT display_name, description, avatar_cid, banner_cid, post_count
|
||
FROM profiles
|
||
WHERE did = $1",
|
||
)
|
||
.bind(&target_did)
|
||
.fetch_optional(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
let (display_name, description, avatar_cid, banner_cid, post_count) = match profile_row {
|
||
Some(row) => row,
|
||
None => {
|
||
let real_count: i64 = sqlx::query_scalar(
|
||
"SELECT COUNT(*)::BIGINT FROM posts \
|
||
WHERE did = $1 \
|
||
AND collection IN ('app.twi.post','app.bsky.feed.post')",
|
||
)
|
||
.bind(&target_did)
|
||
.fetch_one(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
(None, None, None, None, real_count)
|
||
}
|
||
};
|
||
|
||
Ok(Json(ProfileResponse {
|
||
did: target_did,
|
||
handle: display_handle,
|
||
posts,
|
||
followers,
|
||
following,
|
||
display_name,
|
||
description,
|
||
avatar_cid,
|
||
banner_cid,
|
||
post_count,
|
||
}))
|
||
}
|
||
|
||
// -- post by uri (thread hydration) ----------------------------------------
|
||
|
||
#[derive(Debug, Serialize)]
|
||
struct ThreadResponse {
|
||
post: Option<PostRow>,
|
||
thread: ThreadView,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
like_count: Option<i64>,
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
repost_count: Option<i64>,
|
||
/// `true` when the requesting viewer (`viewer_did` query param)
|
||
/// has a like row for this post. `None` if the viewer was not
|
||
/// specified — the client should treat `None` as "unknown, hide
|
||
/// the liked state" rather than "not liked". (Phase 5b review
|
||
/// fix H4 — without this, the UI shows a generic count but
|
||
/// can't tell whether the user has already liked the post.)
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
viewer_liked: Option<bool>,
|
||
/// Same as `viewer_liked` but for reposts.
|
||
#[serde(skip_serializing_if = "Option::is_none")]
|
||
viewer_reposted: Option<bool>,
|
||
}
|
||
|
||
#[derive(Debug, Serialize)]
|
||
struct ThreadView {
|
||
parent: Option<PostRow>,
|
||
root: Option<PostRow>,
|
||
}
|
||
|
||
/// Optional query parameters for `/api/post/{uri}`. The Tauri client
|
||
/// passes its current session DID so the response can include
|
||
/// `viewer_liked` / `viewer_reposted` booleans.
|
||
#[derive(Debug, Default, Deserialize)]
|
||
struct PostQuery {
|
||
#[serde(default)]
|
||
viewer_did: Option<String>,
|
||
}
|
||
|
||
/// `GET /api/post/{uri}` — single-post lookup with parent + root
|
||
/// hydrated in one round trip.
|
||
///
|
||
/// Used by the UI when the user clicks a "show thread" link on a reply:
|
||
/// rather than three sequential fetches (`/api/post/{uri}` → fetch parent
|
||
/// → fetch root), the server returns the post plus its `parent_uri` /
|
||
/// `root_uri` rows in one go. Missing parents / roots come back as
|
||
/// `null` so the client can render "unknown / not in index" without
|
||
/// retrying.
|
||
///
|
||
/// When the post is found, the response also includes `like_count`
|
||
/// and `repost_count` from the `likes` / `reposts` tables so the UI
|
||
/// can render engagement numbers next to the action buttons. (We
|
||
/// skip the count queries entirely when the post is missing so
|
||
/// the "not in index" path stays cheap.)
|
||
///
|
||
/// When the caller supplies `viewer_did`, the response also includes
|
||
/// `viewer_liked` / `viewer_reposted` so the client can highlight the
|
||
/// like/repost button when the viewer has already engaged. Lookups
|
||
/// run as a single `EXISTS` query each, hitting the
|
||
/// `likes_did_post_uri_idx` / `reposts_did_post_uri_idx` unique
|
||
/// indexes, so the cost is O(1) per viewer.
|
||
///
|
||
/// The path parameter is captured by `axum::extract::Path` as the raw
|
||
/// remainder after `/api/post/`, so the colons in a `did:plc:…` URI are
|
||
/// preserved verbatim — we never need URL decoding.
|
||
async fn post_by_uri(
|
||
State(state): State<AppState>,
|
||
Path(uri): Path<String>,
|
||
Query(q): Query<PostQuery>,
|
||
) -> Result<Json<ThreadResponse>, (StatusCode, Json<Value>)> {
|
||
if uri.is_empty() {
|
||
return Err(bad_request("uri is required"));
|
||
}
|
||
let uri = percent_decode(&uri);
|
||
|
||
// Shared with `/api/thread` — see `load_thread_context`. Two
|
||
// things are deliberately capped here, because this endpoint's
|
||
// wire shape predates the thread view and its callers read
|
||
// neither field:
|
||
// * no reply fetch — `replies` has never been part of the shape;
|
||
// * `max_parents = 1` — only `thread.parent` (the immediate
|
||
// parent) is serialised, so walking the full ancestor chain
|
||
// would spend up to MAX_THREAD_PARENTS sequential round trips
|
||
// on rows that are then dropped. One step is exactly what this
|
||
// endpoint did before the two handlers were merged.
|
||
let ctx = load_thread_context(&state, &uri, 1, false).await?;
|
||
|
||
let (like_count, repost_count) = match ctx.post.as_ref() {
|
||
Some(p) => (Some(p.like_count), Some(p.repost_count)),
|
||
None => (None, None),
|
||
};
|
||
// The legacy shape carries only the *immediate* parent, which is
|
||
// the last entry of the root-first ancestor chain.
|
||
let parent = ctx.parents.last().cloned();
|
||
let (viewer_liked, viewer_reposted) =
|
||
viewer_state(&state, ctx.post.as_ref(), &uri, q.viewer_did.as_deref()).await?;
|
||
|
||
Ok(Json(ThreadResponse {
|
||
post: ctx.post,
|
||
thread: ThreadView {
|
||
parent,
|
||
root: ctx.root,
|
||
},
|
||
like_count,
|
||
repost_count,
|
||
viewer_liked,
|
||
viewer_reposted,
|
||
}))
|
||
}
|
||
|
||
// -- thread -----------------------------------------------------------------
|
||
|
||
/// Upper bound for `load_thread_context`'s `max_parents` argument —
|
||
/// how far up the reply chain `/api/thread` walks. (`/api/post` passes
|
||
/// 1, since it only serialises the immediate parent.)
|
||
///
|
||
/// Each step is one indexed `SELECT ... WHERE uri = $1`, so the walk is
|
||
/// O(depth) round trips. 20 covers any conversation a human will read
|
||
/// in one screen and bounds the worst case for a maliciously deep
|
||
/// (or accidentally self-referential) chain. The walk also stops early
|
||
/// on the first ancestor that isn't in our index.
|
||
const MAX_THREAD_PARENTS: usize = 20;
|
||
|
||
/// Hard cap on direct replies returned by `/api/thread`. Matches
|
||
/// [`MAX_LIMIT`] in spirit: one screenful plus headroom, and the
|
||
/// client re-requests the thread rooted at a reply to go deeper.
|
||
const MAX_THREAD_REPLIES: i64 = 100;
|
||
|
||
/// The `SELECT` list every post lookup shares.
|
||
///
|
||
/// Kept as one constant so the four call sites (single post, ancestor
|
||
/// walk, root, replies) can't drift into returning different column
|
||
/// sets for the same `PostRow` wire type. `avatar_cid` is deliberately
|
||
/// absent: it was never in the `/api/post` response and `PostRow`'s
|
||
/// `FromRow` treats a missing column as `None`, so adding it here
|
||
/// would silently change an existing endpoint's payload.
|
||
const POST_COLUMNS: &str = r#"uri, did, handle, rkey, collection, text, cid,
|
||
parent_uri, root_uri, embed, langs, created_at,
|
||
like_count, repost_count"#;
|
||
|
||
/// Everything the two thread-shaped endpoints need, resolved once.
|
||
///
|
||
/// `/api/post/{uri}` projects this down to its historical
|
||
/// `{ post, thread: { parent, root } }` shape; `/api/thread` returns
|
||
/// the whole thing. Having one loader means the two endpoints can
|
||
/// never disagree about which row is the parent or the root.
|
||
struct ThreadContext {
|
||
post: Option<PostRow>,
|
||
/// Ancestor chain, **root first / immediate parent last**.
|
||
parents: Vec<PostRow>,
|
||
root: Option<PostRow>,
|
||
replies: Vec<PostRow>,
|
||
}
|
||
|
||
/// Resolve a post plus its thread context.
|
||
///
|
||
/// The ancestor walk follows `parent_uri` upwards one row at a time,
|
||
/// stopping after `max_parents` steps, at the first ancestor missing
|
||
/// from our index, or on a URI we've already visited. That last guard
|
||
/// matters: `parent_uri` comes from a user-authored record, so a
|
||
/// record claiming to be its own parent (or a two-post cycle) is
|
||
/// something a hostile PDS can produce at will, and without the
|
||
/// visited-set this loop would never terminate.
|
||
///
|
||
/// `root` reproduces the pre-existing `/api/post` semantics exactly:
|
||
/// it is the row named by the post's own `root_uri`, it collapses to
|
||
/// the parent row when `root_uri == parent_uri` (including when that
|
||
/// parent isn't indexed and is therefore `None`), and it is `None` for
|
||
/// a top-level post. We reuse a row already pulled by the ancestor
|
||
/// walk when the root is in it, so the common case costs no extra
|
||
/// query.
|
||
///
|
||
/// `with_replies` is opt-in so `/api/post` doesn't pay for a query
|
||
/// whose result it never serialises.
|
||
async fn load_thread_context(
|
||
state: &AppState,
|
||
uri: &str,
|
||
max_parents: usize,
|
||
with_replies: bool,
|
||
) -> Result<ThreadContext, (StatusCode, Json<Value>)> {
|
||
let post = fetch_one_post(state, uri).await?;
|
||
|
||
let (parent_uri, root_uri) = match post.as_ref() {
|
||
Some(p) => (p.parent_uri.clone(), p.root_uri.clone()),
|
||
None => (None, None),
|
||
};
|
||
|
||
// Walk upwards. `chain` is built child-first and reversed at the
|
||
// end, so the caller gets a root-first list it can render top-down.
|
||
let mut visited: std::collections::HashSet<String> =
|
||
std::collections::HashSet::new();
|
||
visited.insert(uri.to_string());
|
||
let mut chain: Vec<PostRow> = Vec::new();
|
||
let mut next = parent_uri.clone();
|
||
while let Some(u) = next {
|
||
if chain.len() >= max_parents {
|
||
break;
|
||
}
|
||
if !visited.insert(u.clone()) {
|
||
// Cycle (or a post that names itself as its parent).
|
||
break;
|
||
}
|
||
match fetch_one_post(state, &u).await? {
|
||
Some(p) => {
|
||
next = p.parent_uri.clone();
|
||
chain.push(p);
|
||
}
|
||
// Ancestor not indexed — the chain is truncated here.
|
||
None => break,
|
||
}
|
||
}
|
||
// `chain[0]` is the immediate parent while the list is child-first.
|
||
let parent = chain.first().cloned();
|
||
chain.reverse();
|
||
let parents = chain;
|
||
|
||
let root: Option<PostRow> = match root_uri.as_deref() {
|
||
Some(u) if Some(u) != parent_uri.as_deref() => {
|
||
match parents.iter().find(|p| p.uri == u) {
|
||
Some(p) => Some(p.clone()),
|
||
None => fetch_one_post(state, u).await?,
|
||
}
|
||
}
|
||
// Self-thread (single-post thread): `root` == `parent`. Avoid the
|
||
// duplicate fetch — surface the parent row as the root too so the
|
||
// UI can render the chain without an extra round trip.
|
||
Some(_) => parent.clone(),
|
||
None => None,
|
||
};
|
||
|
||
// Direct replies only. Ordered oldest-first because a conversation
|
||
// reads top-down, which is the opposite of the timeline's
|
||
// newest-first ordering. `uri` breaks ties so the order is total
|
||
// and stable across requests.
|
||
let replies: Vec<PostRow> = if with_replies && post.is_some() {
|
||
sqlx::query_as::<_, PostRow>(&format!(
|
||
r#"SELECT {POST_COLUMNS}
|
||
FROM posts
|
||
WHERE parent_uri = $1
|
||
AND collection IN ('app.twi.post','app.bsky.feed.post')
|
||
ORDER BY created_at ASC, uri ASC
|
||
LIMIT $2"#
|
||
))
|
||
.bind(uri)
|
||
.bind(MAX_THREAD_REPLIES)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?
|
||
} else {
|
||
Vec::new()
|
||
};
|
||
|
||
Ok(ThreadContext {
|
||
post,
|
||
parents,
|
||
root,
|
||
replies,
|
||
})
|
||
}
|
||
|
||
/// Query parameters for both `/api/thread` spellings.
|
||
#[derive(Debug, Default, Deserialize)]
|
||
struct ThreadQuery {
|
||
/// Only read by the `/api/thread?uri=…` form; ignored (and
|
||
/// unnecessary) on `/api/thread/<uri>`.
|
||
#[serde(default)]
|
||
uri: Option<String>,
|
||
#[serde(default)]
|
||
viewer_did: Option<String>,
|
||
}
|
||
|
||
/// `GET /api/thread?uri=…`
|
||
async fn thread_query(
|
||
State(state): State<AppState>,
|
||
Query(q): Query<ThreadQuery>,
|
||
) -> Result<Json<ThreadFullResponse>, (StatusCode, Json<Value>)> {
|
||
let uri = q.uri.clone().unwrap_or_default();
|
||
thread_inner(&state, &uri, q.viewer_did.as_deref()).await
|
||
}
|
||
|
||
/// `GET /api/thread/<uri>` — path form, mirroring `/api/post/*uri`.
|
||
async fn thread_path(
|
||
State(state): State<AppState>,
|
||
Path(uri): Path<String>,
|
||
Query(q): Query<ThreadQuery>,
|
||
) -> Result<Json<ThreadFullResponse>, (StatusCode, Json<Value>)> {
|
||
thread_inner(&state, &uri, q.viewer_did.as_deref()).await
|
||
}
|
||
|
||
async fn thread_inner(
|
||
state: &AppState,
|
||
uri: &str,
|
||
viewer_did: Option<&str>,
|
||
) -> Result<Json<ThreadFullResponse>, (StatusCode, Json<Value>)> {
|
||
if uri.is_empty() {
|
||
return Err(bad_request("uri is required"));
|
||
}
|
||
let uri = percent_decode(uri);
|
||
let ctx = load_thread_context(state, &uri, MAX_THREAD_PARENTS, true).await?;
|
||
|
||
let (like_count, repost_count) = match ctx.post.as_ref() {
|
||
Some(p) => (Some(p.like_count), Some(p.repost_count)),
|
||
None => (None, None),
|
||
};
|
||
let (viewer_liked, viewer_reposted) =
|
||
viewer_state(state, ctx.post.as_ref(), &uri, viewer_did).await?;
|
||
|
||
Ok(Json(ThreadFullResponse {
|
||
post: ctx.post,
|
||
parents: ctx.parents,
|
||
root: ctx.root,
|
||
replies: ctx.replies,
|
||
like_count,
|
||
repost_count,
|
||
viewer_liked,
|
||
viewer_reposted,
|
||
}))
|
||
}
|
||
|
||
/// Viewer-scoped engagement state for one post.
|
||
///
|
||
/// Returns `(None, None)` when the post is missing (nothing to have an
|
||
/// opinion about) or when the caller passed no `viewer_did` — the
|
||
/// client must read `None` as "unknown", never as "not liked".
|
||
/// Both lookups are `EXISTS` against the `likes_did_post_uri_idx` /
|
||
/// `reposts_did_post_uri_idx` unique indexes, so this is O(1).
|
||
async fn viewer_state(
|
||
state: &AppState,
|
||
post: Option<&PostRow>,
|
||
uri: &str,
|
||
viewer_did: Option<&str>,
|
||
) -> Result<(Option<bool>, Option<bool>), (StatusCode, Json<Value>)> {
|
||
if post.is_none() {
|
||
return Ok((None, None));
|
||
}
|
||
let Some(viewer) = viewer_did.filter(|v| !v.is_empty()) else {
|
||
return Ok((None, None));
|
||
};
|
||
let liked: bool = sqlx::query_scalar(
|
||
"SELECT EXISTS(SELECT 1 FROM likes WHERE did = $1 AND post_uri = $2)",
|
||
)
|
||
.bind(viewer)
|
||
.bind(uri)
|
||
.fetch_one(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
let reposted: bool = sqlx::query_scalar(
|
||
"SELECT EXISTS(SELECT 1 FROM reposts WHERE did = $1 AND post_uri = $2)",
|
||
)
|
||
.bind(viewer)
|
||
.bind(uri)
|
||
.fetch_one(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
Ok((Some(liked), Some(reposted)))
|
||
}
|
||
|
||
async fn fetch_one_post(
|
||
state: &AppState,
|
||
uri: &str,
|
||
) -> Result<Option<PostRow>, (StatusCode, Json<Value>)> {
|
||
sqlx::query_as::<_, PostRow>(&format!(
|
||
r#"SELECT {POST_COLUMNS}
|
||
FROM posts
|
||
WHERE uri = $1
|
||
LIMIT 1"#
|
||
))
|
||
.bind(uri)
|
||
.fetch_optional(&state.db)
|
||
.await
|
||
.map_err(db_err)
|
||
}
|
||
|
||
/// Minimal percent-decode for the path capture. axum's `Path<String>`
|
||
/// already decodes percent-escapes for us, so this is a no-op for
|
||
/// the happy path. It only fires if the URI itself contains a stray
|
||
/// `%XX` sequence the caller wants kept verbatim (e.g. a literal `%`
|
||
/// in a path component, which we don't have here).
|
||
#[allow(dead_code)]
|
||
fn percent_decode(s: &str) -> String {
|
||
let bytes = s.as_bytes();
|
||
let mut out = String::with_capacity(s.len());
|
||
let mut i = 0;
|
||
while i < bytes.len() {
|
||
if bytes[i] == b'%' && i + 2 < bytes.len() {
|
||
if let (Some(h), Some(l)) = (
|
||
hex_digit(bytes[i + 1]),
|
||
hex_digit(bytes[i + 2]),
|
||
) {
|
||
out.push((h * 16 + l) as char);
|
||
i += 3;
|
||
continue;
|
||
}
|
||
}
|
||
// Push ASCII bytes verbatim; for any multi-byte UTF-8 sequence
|
||
// we push the lead byte as a char (which is valid because the
|
||
// resulting char's code point is `< 128` only for ASCII). For
|
||
// non-ASCII bytes the call sites never reach this branch
|
||
// because `Path<String>` already decoded the URI for us.
|
||
out.push(bytes[i] as char);
|
||
i += 1;
|
||
}
|
||
out
|
||
}
|
||
|
||
#[allow(dead_code)]
|
||
fn hex_digit(b: u8) -> Option<u8> {
|
||
match b {
|
||
b'0'..=b'9' => Some(b - b'0'),
|
||
b'a'..=b'f' => Some(b - b'a' + 10),
|
||
b'A'..=b'F' => Some(b - b'A' + 10),
|
||
_ => None,
|
||
}
|
||
}
|
||
|
||
// -- search -----------------------------------------------------------------
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct SearchQuery {
|
||
q: String,
|
||
#[serde(default)]
|
||
limit: Option<i64>,
|
||
}
|
||
|
||
async fn search(
|
||
State(state): State<AppState>,
|
||
Query(q): Query<SearchQuery>,
|
||
) -> Result<Json<SearchResponse>, (StatusCode, Json<Value>)> {
|
||
let needle = q.q.trim();
|
||
if needle.is_empty() {
|
||
return Err(bad_request("q is required"));
|
||
}
|
||
// Cap query length to avoid pathological ILIKE patterns on huge input.
|
||
if needle.len() > 100 {
|
||
return Err(bad_request("q must be at most 100 characters"));
|
||
}
|
||
let limit = q.limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT);
|
||
|
||
// ILIKE search. We escape LIKE wildcards in the user input so a
|
||
// search for "10%" doesn't suddenly act as a glob.
|
||
let escaped = escape_like(needle);
|
||
let pattern = format!("%{escaped}%");
|
||
|
||
let mut posts: Vec<PostRow> = sqlx::query_as::<_, PostRow>(
|
||
r#"SELECT uri, did, handle, rkey, collection, text, cid,
|
||
parent_uri, root_uri, embed, langs, created_at,
|
||
like_count, repost_count
|
||
FROM posts
|
||
WHERE text ILIKE $1 ESCAPE '\'
|
||
AND collection IN ('app.twi.post','app.bsky.feed.post')
|
||
ORDER BY indexed_at DESC
|
||
LIMIT $2"#,
|
||
)
|
||
.bind(&pattern)
|
||
.bind(limit)
|
||
.fetch_all(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
|
||
decorate_handles(&mut posts);
|
||
|
||
Ok(Json(SearchResponse {
|
||
posts,
|
||
q: needle.to_string(),
|
||
}))
|
||
}
|
||
|
||
// -- notifications ----------------------------------------------------------
|
||
|
||
/// The hydration `SELECT` behind `GET /api/notifications`.
|
||
///
|
||
/// Three joins, all optional:
|
||
///
|
||
/// - `profiles pr` is the primary source for the author's handle /
|
||
/// display name / avatar.
|
||
/// - `ap` is a lateral fallback to the newest non-empty `posts.handle`
|
||
/// for that author. It exists because a DID can be known to us
|
||
/// purely through the Jetstream `identity` backfill (which writes
|
||
/// `posts.handle`) without ever having produced an
|
||
/// `app.bsky.actor.profile` record, and a notification with a blank
|
||
/// author is useless.
|
||
/// - `posts sp` resolves `subject_uri` to the post's text. NULL for
|
||
/// `kind = 'follow'` (no subject) and for a subject that has since
|
||
/// been deleted — the row still renders, just without a preview.
|
||
///
|
||
/// `{cursor}` is substituted with either the empty string or the
|
||
/// keyset predicate. It is NOT user input — see `notifications` for
|
||
/// why that's the only thing interpolated.
|
||
const NOTIFICATIONS_SQL: &str = r#"
|
||
SELECT n.id,
|
||
n.kind,
|
||
n.author_did,
|
||
n.subject_uri,
|
||
n.created_at,
|
||
n.indexed_at,
|
||
n.read_at,
|
||
COALESCE(NULLIF(pr.handle, ''), ap.handle, '') AS author_handle,
|
||
pr.display_name AS author_display_name,
|
||
COALESCE(pr.avatar_cid, ap.avatar_cid) AS author_avatar_cid,
|
||
sp.text AS subject_text
|
||
FROM notifications n
|
||
LEFT JOIN profiles pr ON pr.did = n.author_did
|
||
LEFT JOIN posts sp ON sp.uri = n.subject_uri
|
||
LEFT JOIN LATERAL (
|
||
SELECT p.handle, p.avatar_cid
|
||
FROM posts p
|
||
WHERE p.did = n.author_did AND p.handle <> ''
|
||
ORDER BY p.indexed_at DESC
|
||
LIMIT 1
|
||
) ap ON TRUE
|
||
WHERE n.recipient_did = $1
|
||
{cursor}
|
||
ORDER BY n.indexed_at DESC, n.id DESC
|
||
LIMIT $2
|
||
"#;
|
||
|
||
/// The keyset half of [`NOTIFICATIONS_SQL`]. `$3`/`$4` are the cursor's
|
||
/// `(indexed_at, id)` pair; `$2` stays the LIMIT in both variants so
|
||
/// the two bind orders differ only by the two extra binds at the end.
|
||
const NOTIFICATIONS_CURSOR_PREDICATE: &str = "AND (n.indexed_at, n.id) < ($3, $4)";
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct NotificationsQuery {
|
||
/// DID whose notifications to list. This is the *recipient*.
|
||
did: String,
|
||
#[serde(default)]
|
||
limit: Option<i64>,
|
||
#[serde(default)]
|
||
cursor: Option<String>,
|
||
}
|
||
|
||
/// `GET /api/notifications?did=…&limit=…&cursor=…`
|
||
///
|
||
/// Same pagination contract as `/api/timeline/home`: newest first,
|
||
/// keyset-ordered on `(indexed_at, <tiebreak>)`, `limit + 1` fetched to
|
||
/// detect a next page, opaque base64 cursor, `cursor: null` at the end
|
||
/// of the list. The tiebreak here is the row's `id` rather than a URI
|
||
/// (a notification has no URI of its own), which the shared
|
||
/// [`cursor`] codec carries in its string slot.
|
||
///
|
||
/// **Authenticated**: `did` is the recipient, so the caller has to be
|
||
/// that recipient. This is the endpoint that made authentication
|
||
/// necessary in the first place.
|
||
async fn notifications(
|
||
State(state): State<AppState>,
|
||
auth: AuthedDid,
|
||
Query(q): Query<NotificationsQuery>,
|
||
) -> Result<Json<NotificationsResponse>, (StatusCode, Json<Value>)> {
|
||
if q.did.is_empty() {
|
||
return Err(bad_request("did is required"));
|
||
}
|
||
auth.ensure_matches(&q.did)?;
|
||
let limit = clamp_limit(q.limit);
|
||
let keyset = decode_cursor(q.cursor.as_deref())?;
|
||
// The cursor's string slot holds the row id. A client that hands
|
||
// back a mangled cursor gets a 400 rather than silently restarting
|
||
// at page 1 and re-showing notifications it already scrolled past.
|
||
let cursor_id: Option<i64> = match keyset.as_ref() {
|
||
Some((_, raw)) => Some(
|
||
raw.parse::<i64>()
|
||
.map_err(|_| bad_request("invalid cursor id"))?,
|
||
),
|
||
None => None,
|
||
};
|
||
let fetch = limit + 1;
|
||
|
||
// The only interpolated fragment is one of two compile-time
|
||
// constants; every value is bound. (`sqlx` can't parameterise the
|
||
// presence of a WHERE clause, and building two full copies of a
|
||
// 25-line query would be the thing that actually drifts.)
|
||
let sql = NOTIFICATIONS_SQL.replace(
|
||
"{cursor}",
|
||
if cursor_id.is_some() {
|
||
NOTIFICATIONS_CURSOR_PREDICATE
|
||
} else {
|
||
""
|
||
},
|
||
);
|
||
let mut query = sqlx::query_as::<_, NotificationItem>(&sql)
|
||
.bind(&q.did)
|
||
.bind(fetch);
|
||
if let (Some((ts, _)), Some(id)) = (keyset.as_ref(), cursor_id) {
|
||
query = query.bind(*ts).bind(id);
|
||
}
|
||
let mut rows: Vec<NotificationItem> =
|
||
query.fetch_all(&state.db).await.map_err(db_err)?;
|
||
|
||
let next = if rows.len() as i64 > limit {
|
||
rows.truncate(limit as usize);
|
||
rows.last().map(|n| (n.indexed_at, n.id))
|
||
} else {
|
||
None
|
||
};
|
||
|
||
// Last-resort author label. The SQL fallbacks already tried the
|
||
// profiles cache and the author's own posts; a still-empty handle
|
||
// means we've genuinely never seen a handle for this DID.
|
||
for n in rows.iter_mut() {
|
||
if n.author_handle.is_empty() {
|
||
n.author_handle = short_did_bare(&n.author_did);
|
||
}
|
||
}
|
||
|
||
Ok(Json(NotificationsResponse {
|
||
notifications: rows,
|
||
cursor: next.map(|(ts, id)| cursor::encode(ts, &id.to_string())),
|
||
}))
|
||
}
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct NotificationCountQuery {
|
||
did: String,
|
||
}
|
||
|
||
/// `GET /api/notifications/count?did=…` → unread count.
|
||
///
|
||
/// Backed by the `notifications_unread_idx` partial index, so the cost
|
||
/// scales with the number of *unread* rows, not the user's lifetime
|
||
/// notification history. That matters because the client polls this
|
||
/// for its tray badge.
|
||
///
|
||
/// **Authenticated**, same rule as the list itself: an unread count is
|
||
/// still information about someone else's inbox.
|
||
async fn notifications_count(
|
||
State(state): State<AppState>,
|
||
auth: AuthedDid,
|
||
Query(q): Query<NotificationCountQuery>,
|
||
) -> Result<Json<NotificationCountResponse>, (StatusCode, Json<Value>)> {
|
||
if q.did.is_empty() {
|
||
return Err(bad_request("did is required"));
|
||
}
|
||
auth.ensure_matches(&q.did)?;
|
||
let count: i64 = sqlx::query_scalar(
|
||
"SELECT COUNT(*)::BIGINT FROM notifications \
|
||
WHERE recipient_did = $1 AND read_at IS NULL",
|
||
)
|
||
.bind(&q.did)
|
||
.fetch_one(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
Ok(Json(NotificationCountResponse { count }))
|
||
}
|
||
|
||
/// Body of `POST /api/notifications/seen`.
|
||
///
|
||
/// `seenAt` is accepted in both spellings: the Tauri client is
|
||
/// TypeScript and sends camelCase, while a curl-driven test or the
|
||
/// Rust IPC layer naturally writes snake_case. Aliasing costs nothing
|
||
/// and removes a whole class of "why is nothing marked read" bug.
|
||
#[derive(Debug, Deserialize)]
|
||
struct NotificationsSeenReq {
|
||
did: String,
|
||
/// Watermark: only rows the AppView indexed at or before this
|
||
/// instant are marked read. Omit to mark *everything* currently
|
||
/// unread as read.
|
||
///
|
||
/// A client that renders a page and then posts the newest visible
|
||
/// row's `indexed_at` will never accidentally mark a notification
|
||
/// that arrived mid-scroll as already seen.
|
||
#[serde(default, alias = "seenAt")]
|
||
seen_at: Option<DateTime<Utc>>,
|
||
}
|
||
|
||
/// `POST /api/notifications/seen` — bulk-mark notifications as read.
|
||
///
|
||
/// Idempotent: `read_at IS NULL` in the predicate means a second call
|
||
/// updates nothing and reports `updated: 0`. `read_at` is set to
|
||
/// `now()` (when we recorded the ack), not to `seenAt` (which is a
|
||
/// client-supplied watermark and could be arbitrarily far in the past).
|
||
///
|
||
/// **Authenticated**, and the only *write* among the private
|
||
/// endpoints: without the check anyone could clear another user's
|
||
/// unread badge. `AuthedDid` runs before `Json` because the body
|
||
/// extractor consumes the request — axum requires body extractors last.
|
||
async fn notifications_seen(
|
||
State(state): State<AppState>,
|
||
auth: AuthedDid,
|
||
Json(req): Json<NotificationsSeenReq>,
|
||
) -> Result<Json<NotificationSeenResponse>, (StatusCode, Json<Value>)> {
|
||
if req.did.is_empty() {
|
||
return Err(bad_request("did is required"));
|
||
}
|
||
auth.ensure_matches(&req.did)?;
|
||
let res = sqlx::query(
|
||
r#"UPDATE notifications
|
||
SET read_at = now()
|
||
WHERE recipient_did = $1
|
||
AND read_at IS NULL
|
||
AND ($2::timestamptz IS NULL OR indexed_at <= $2::timestamptz)"#,
|
||
)
|
||
.bind(&req.did)
|
||
.bind(req.seen_at)
|
||
.execute(&state.db)
|
||
.await
|
||
.map_err(db_err)?;
|
||
Ok(Json(NotificationSeenResponse {
|
||
ok: true,
|
||
updated: res.rows_affected() as i64,
|
||
}))
|
||
}
|
||
|
||
// -- follower / following lists ---------------------------------------------
|
||
|
||
/// Which side of the `follows` edge the caller wants listed.
|
||
///
|
||
/// The two queries are structurally identical and differ only in which
|
||
/// column is the anchor (bound as `$1`) and which is projected as the
|
||
/// listed actor. Column names can't be bound as parameters, so each
|
||
/// direction owns its own `&'static str` rather than being assembled
|
||
/// from fragments — four fixed strings, zero interpolation of anything
|
||
/// that came from a request.
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||
enum FollowDirection {
|
||
/// Actors who follow the anchor DID.
|
||
Followers,
|
||
/// Actors the anchor DID follows.
|
||
Following,
|
||
}
|
||
|
||
impl FollowDirection {
|
||
fn sql(self, with_cursor: bool) -> &'static str {
|
||
match (self, with_cursor) {
|
||
(FollowDirection::Followers, false) => FOLLOWERS_SQL,
|
||
(FollowDirection::Followers, true) => FOLLOWERS_SQL_CURSOR,
|
||
(FollowDirection::Following, false) => FOLLOWING_SQL,
|
||
(FollowDirection::Following, true) => FOLLOWING_SQL_CURSOR,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Followers of `$1`, newest edge first. The lateral `ap` join is the
|
||
/// same profiles-cache-then-posts fallback the notification list uses.
|
||
const FOLLOWERS_SQL: &str = r#"
|
||
SELECT f.follower_did AS did,
|
||
COALESCE(NULLIF(pr.handle, ''), ap.handle, '') AS handle,
|
||
pr.display_name,
|
||
COALESCE(pr.avatar_cid, ap.avatar_cid) AS avatar_cid,
|
||
f.indexed_at
|
||
FROM follows f
|
||
LEFT JOIN profiles pr ON pr.did = f.follower_did
|
||
LEFT JOIN LATERAL (
|
||
SELECT p.handle, p.avatar_cid
|
||
FROM posts p
|
||
WHERE p.did = f.follower_did AND p.handle <> ''
|
||
ORDER BY p.indexed_at DESC
|
||
LIMIT 1
|
||
) ap ON TRUE
|
||
WHERE f.subject_did = $1
|
||
ORDER BY f.indexed_at DESC, f.follower_did DESC
|
||
LIMIT $2
|
||
"#;
|
||
|
||
const FOLLOWERS_SQL_CURSOR: &str = r#"
|
||
SELECT f.follower_did AS did,
|
||
COALESCE(NULLIF(pr.handle, ''), ap.handle, '') AS handle,
|
||
pr.display_name,
|
||
COALESCE(pr.avatar_cid, ap.avatar_cid) AS avatar_cid,
|
||
f.indexed_at
|
||
FROM follows f
|
||
LEFT JOIN profiles pr ON pr.did = f.follower_did
|
||
LEFT JOIN LATERAL (
|
||
SELECT p.handle, p.avatar_cid
|
||
FROM posts p
|
||
WHERE p.did = f.follower_did AND p.handle <> ''
|
||
ORDER BY p.indexed_at DESC
|
||
LIMIT 1
|
||
) ap ON TRUE
|
||
WHERE f.subject_did = $1
|
||
AND (f.indexed_at, f.follower_did) < ($3, $4)
|
||
ORDER BY f.indexed_at DESC, f.follower_did DESC
|
||
LIMIT $2
|
||
"#;
|
||
|
||
const FOLLOWING_SQL: &str = r#"
|
||
SELECT f.subject_did AS did,
|
||
COALESCE(NULLIF(pr.handle, ''), ap.handle, '') AS handle,
|
||
pr.display_name,
|
||
COALESCE(pr.avatar_cid, ap.avatar_cid) AS avatar_cid,
|
||
f.indexed_at
|
||
FROM follows f
|
||
LEFT JOIN profiles pr ON pr.did = f.subject_did
|
||
LEFT JOIN LATERAL (
|
||
SELECT p.handle, p.avatar_cid
|
||
FROM posts p
|
||
WHERE p.did = f.subject_did AND p.handle <> ''
|
||
ORDER BY p.indexed_at DESC
|
||
LIMIT 1
|
||
) ap ON TRUE
|
||
WHERE f.follower_did = $1
|
||
ORDER BY f.indexed_at DESC, f.subject_did DESC
|
||
LIMIT $2
|
||
"#;
|
||
|
||
const FOLLOWING_SQL_CURSOR: &str = r#"
|
||
SELECT f.subject_did AS did,
|
||
COALESCE(NULLIF(pr.handle, ''), ap.handle, '') AS handle,
|
||
pr.display_name,
|
||
COALESCE(pr.avatar_cid, ap.avatar_cid) AS avatar_cid,
|
||
f.indexed_at
|
||
FROM follows f
|
||
LEFT JOIN profiles pr ON pr.did = f.subject_did
|
||
LEFT JOIN LATERAL (
|
||
SELECT p.handle, p.avatar_cid
|
||
FROM posts p
|
||
WHERE p.did = f.subject_did AND p.handle <> ''
|
||
ORDER BY p.indexed_at DESC
|
||
LIMIT 1
|
||
) ap ON TRUE
|
||
WHERE f.follower_did = $1
|
||
AND (f.indexed_at, f.subject_did) < ($3, $4)
|
||
ORDER BY f.indexed_at DESC, f.subject_did DESC
|
||
LIMIT $2
|
||
"#;
|
||
|
||
#[derive(Debug, Deserialize)]
|
||
struct ActorListQuery {
|
||
did: String,
|
||
#[serde(default)]
|
||
limit: Option<i64>,
|
||
#[serde(default)]
|
||
cursor: Option<String>,
|
||
}
|
||
|
||
/// `GET /api/followers?did=…&limit=&cursor=`
|
||
async fn followers(
|
||
State(state): State<AppState>,
|
||
Query(q): Query<ActorListQuery>,
|
||
) -> Result<Json<ActorListResponse>, (StatusCode, Json<Value>)> {
|
||
actor_list(&state, FollowDirection::Followers, &q).await
|
||
}
|
||
|
||
/// `GET /api/following?did=…&limit=&cursor=`
|
||
async fn following(
|
||
State(state): State<AppState>,
|
||
Query(q): Query<ActorListQuery>,
|
||
) -> Result<Json<ActorListResponse>, (StatusCode, Json<Value>)> {
|
||
actor_list(&state, FollowDirection::Following, &q).await
|
||
}
|
||
|
||
/// Shared body of `/api/followers` and `/api/following`.
|
||
///
|
||
/// The cursor is the `follows` edge's `(indexed_at, <listed did>)`
|
||
/// pair, encoded by the same [`cursor`] codec the timeline uses — an
|
||
/// edge, unlike a post, has no natural single-column key, and
|
||
/// `indexed_at` alone is not unique when a batch of follows lands in
|
||
/// the same transaction.
|
||
async fn actor_list(
|
||
state: &AppState,
|
||
direction: FollowDirection,
|
||
q: &ActorListQuery,
|
||
) -> Result<Json<ActorListResponse>, (StatusCode, Json<Value>)> {
|
||
if q.did.is_empty() {
|
||
return Err(bad_request("did is required"));
|
||
}
|
||
let limit = clamp_limit(q.limit);
|
||
let keyset = decode_cursor(q.cursor.as_deref())?;
|
||
let fetch = limit + 1;
|
||
|
||
let mut query = sqlx::query_as::<
|
||
_,
|
||
(String, String, Option<String>, Option<String>, DateTime<Utc>),
|
||
>(direction.sql(keyset.is_some()))
|
||
.bind(&q.did)
|
||
.bind(fetch);
|
||
if let Some((ts, did)) = keyset.as_ref() {
|
||
query = query.bind(*ts).bind(did.clone());
|
||
}
|
||
let mut rows = query.fetch_all(&state.db).await.map_err(db_err)?;
|
||
|
||
let next = if rows.len() as i64 > limit {
|
||
rows.truncate(limit as usize);
|
||
rows.last().map(|(did, _, _, _, ts)| (*ts, did.clone()))
|
||
} else {
|
||
None
|
||
};
|
||
|
||
let profiles: Vec<ActorProfile> = rows
|
||
.into_iter()
|
||
.map(|(did, handle, display_name, avatar_cid, _)| {
|
||
// Same last-resort label as the notification list: never
|
||
// hand the UI an empty `handle`, and never prefix it with
|
||
// '@' (the UI adds that itself).
|
||
let handle = if handle.is_empty() {
|
||
short_did_bare(&did)
|
||
} else {
|
||
handle
|
||
};
|
||
ActorProfile {
|
||
did,
|
||
handle,
|
||
display_name,
|
||
avatar_cid,
|
||
}
|
||
})
|
||
.collect();
|
||
|
||
Ok(Json(ActorListResponse {
|
||
profiles,
|
||
cursor: next.map(|(ts, did)| cursor::encode(ts, &did)),
|
||
}))
|
||
}
|
||
|
||
// -- healthz ----------------------------------------------------------------
|
||
|
||
/// Liveness probe.
|
||
///
|
||
/// Both ingest streams report separately. `jetstream_connected` is the
|
||
/// public network's view; `pds_firehose_connected` is the local PDS's
|
||
/// guaranteed delivery path for our own users' records. A deployment
|
||
/// can be perfectly healthy for reads with the first one down, but a
|
||
/// `pds_firehose_enabled: true, pds_firehose_connected: false` pair
|
||
/// means local commits are riding on the best-effort push alone — which
|
||
/// is exactly the state an operator wants to see in a probe rather than
|
||
/// discover from a missing post.
|
||
async fn healthz(State(state): State<AppState>) -> impl IntoResponse {
|
||
let stats = &state.stats;
|
||
Json(json!({
|
||
"ok": true,
|
||
"lag_ms": stats.lag_ms(),
|
||
"events_processed": stats.events_processed(),
|
||
"jetstream_connected": stats.jetstream_connected(),
|
||
"pds_firehose_enabled": state.cfg.pds_firehose_enabled,
|
||
"pds_firehose_connected": stats.pds_connected(),
|
||
"pds_firehose_frames": stats.pds_frames_processed(),
|
||
"pds_firehose_seq": stats.pds_last_seq(),
|
||
}))
|
||
}
|
||
|
||
// -- helpers ----------------------------------------------------------------
|
||
|
||
/// Fill in a synthetic `handle` for posts that have an empty one
|
||
/// (Jetstream-indexed posts without a handle backfill). Operates in
|
||
/// place. We deliberately do not mutate the database row — this is a
|
||
/// display-time concern only.
|
||
fn decorate_handles(posts: &mut [PostRow]) {
|
||
for p in posts.iter_mut() {
|
||
if p.handle.is_empty() {
|
||
p.handle = short_did_bare(&p.did);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Synthetic stand-in for a handle we don't know yet:
|
||
/// `{first-12-chars-of-did}…`.
|
||
///
|
||
/// Deliberately *without* a leading `@`. Every consumer renders the
|
||
/// sigil itself — `PostCard.svelte` and `ProfileView.svelte` both
|
||
/// interpolate `@{handle}` — so a prefixed fallback came out as
|
||
/// `@@did:plc:abcd…` in the post feed.
|
||
///
|
||
/// Uses char-based slicing so we never panic on a UTF-8 boundary
|
||
/// (e.g. `did:web:münchen.de`).
|
||
fn short_did_bare(did: &str) -> String {
|
||
let snip: String = did.chars().take(12).collect();
|
||
format!("{snip}…")
|
||
}
|
||
|
||
/// Clamp a caller-supplied `limit` into the range every list endpoint
|
||
/// shares: [`DEFAULT_LIMIT`] when absent, hard-capped at [`MAX_LIMIT`],
|
||
/// and never below 1 (a `limit=0` would otherwise return an empty page
|
||
/// forever and a negative one would make `limit + 1` underflow the
|
||
/// "is there a next page" peek).
|
||
fn clamp_limit(limit: Option<i64>) -> i64 {
|
||
limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
|
||
}
|
||
|
||
/// Decode an optional opaque cursor into the `(indexed_at, tiebreak)`
|
||
/// pair every keyset predicate in this module binds.
|
||
///
|
||
/// Two failure modes, both 400 rather than a silent reset to page 1
|
||
/// (which would make a client re-render rows the user already scrolled
|
||
/// past, or skip rows entirely):
|
||
///
|
||
/// - the cursor isn't a well-formed `base64url(micros:tiebreak)` pair;
|
||
/// - its microsecond timestamp is outside `chrono::Utc`'s
|
||
/// representable range — e.g. `i64::MAX` from a hand-crafted cursor.
|
||
///
|
||
/// The timestamp is returned as a `DateTime<Utc>` (not raw micros) so
|
||
/// sqlx binds it as `timestamptz`; binding a `bigint` would make the
|
||
/// `(indexed_at, …) < ($1, $2)` row comparison fail at the type level.
|
||
fn decode_cursor(
|
||
raw: Option<&str>,
|
||
) -> Result<Option<(DateTime<Utc>, String)>, (StatusCode, Json<Value>)> {
|
||
let Some(raw) = raw else {
|
||
return Ok(None);
|
||
};
|
||
let state = cursor::decode(raw).map_err(|e| bad_request(&e))?;
|
||
let ts = Utc
|
||
.timestamp_micros(state.ts)
|
||
.single()
|
||
.ok_or_else(|| bad_request("invalid cursor timestamp"))?;
|
||
Ok(Some((ts, state.uri)))
|
||
}
|
||
|
||
/// Escape `%`, `_`, and `\` for use inside a `LIKE ... ESCAPE '\'`
|
||
/// pattern.
|
||
fn escape_like(s: &str) -> String {
|
||
let mut out = String::with_capacity(s.len());
|
||
for c in s.chars() {
|
||
match c {
|
||
'\\' | '%' | '_' => {
|
||
out.push('\\');
|
||
out.push(c);
|
||
}
|
||
other => out.push(other),
|
||
}
|
||
}
|
||
out
|
||
}
|
||
|
||
fn db_err(e: impl std::fmt::Display) -> (StatusCode, Json<Value>) {
|
||
(
|
||
StatusCode::INTERNAL_SERVER_ERROR,
|
||
Json(json!({
|
||
"error": "InternalServerError",
|
||
"message": e.to_string(),
|
||
})),
|
||
)
|
||
}
|
||
|
||
fn bad_request(msg: &str) -> (StatusCode, Json<Value>) {
|
||
(
|
||
StatusCode::BAD_REQUEST,
|
||
Json(json!({
|
||
"error": "InvalidRequest",
|
||
"message": msg,
|
||
})),
|
||
)
|
||
}
|
||
|
||
// -- tests ------------------------------------------------------------------
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use base64::Engine;
|
||
|
||
#[test]
|
||
fn escape_like_handles_wildcards() {
|
||
assert_eq!(escape_like("hello"), "hello");
|
||
assert_eq!(escape_like("100%"), "100\\%");
|
||
assert_eq!(escape_like("a_b"), "a\\_b");
|
||
assert_eq!(escape_like("back\\slash"), "back\\\\slash");
|
||
}
|
||
|
||
#[test]
|
||
fn short_did_format_matches_spec() {
|
||
assert_eq!(short_did_bare("did:plc:abcdefghijklmnop"), "did:plc:abcd…");
|
||
assert_eq!(short_did_bare("short"), "short…");
|
||
}
|
||
|
||
#[test]
|
||
fn percent_decode_handles_common_escapes() {
|
||
assert_eq!(
|
||
percent_decode("at%3A%2F%2Fdid%3Aplc%3Aabc"),
|
||
"at://did:plc:abc"
|
||
);
|
||
// No escapes → identity.
|
||
assert_eq!(percent_decode("at://did:plc:abc"), "at://did:plc:abc");
|
||
// Truncated `%` at the end → kept verbatim (don't crash).
|
||
assert_eq!(percent_decode("abc%"), "abc%");
|
||
// Non-hex after `%` → kept verbatim.
|
||
assert_eq!(percent_decode("abc%zz"), "abc%zz");
|
||
// Mixed case hex digits.
|
||
assert_eq!(percent_decode("a%2Bb"), "a+b");
|
||
assert_eq!(percent_decode("a%2bb"), "a+b");
|
||
}
|
||
|
||
#[test]
|
||
fn short_did_bare_has_no_sigil() {
|
||
// The notification / follower lists hand this to a UI that
|
||
// renders `@{handle}` itself, so a leading '@' here would
|
||
// double up.
|
||
let s = short_did_bare("did:plc:abcdefghijklmnop");
|
||
assert_eq!(s, "did:plc:abcd…");
|
||
assert!(!s.starts_with('@'));
|
||
// Multi-byte DIDs must not panic on a char boundary.
|
||
assert_eq!(short_did_bare("did:web:münchen.de"), "did:web:münc…");
|
||
}
|
||
|
||
#[test]
|
||
fn clamp_limit_applies_default_and_bounds() {
|
||
assert_eq!(clamp_limit(None), DEFAULT_LIMIT);
|
||
assert_eq!(clamp_limit(Some(10)), 10);
|
||
// Above the cap → capped.
|
||
assert_eq!(clamp_limit(Some(MAX_LIMIT + 1)), MAX_LIMIT);
|
||
assert_eq!(clamp_limit(Some(i64::MAX)), MAX_LIMIT);
|
||
// Zero / negative would make `limit + 1` a useless (or
|
||
// underflowing) next-page peek.
|
||
assert_eq!(clamp_limit(Some(0)), 1);
|
||
assert_eq!(clamp_limit(Some(-5)), 1);
|
||
assert_eq!(clamp_limit(Some(i64::MIN)), 1);
|
||
}
|
||
|
||
#[test]
|
||
fn decode_cursor_round_trips_and_rejects_garbage() {
|
||
// Absent cursor → no keyset predicate.
|
||
assert!(decode_cursor(None).unwrap().is_none());
|
||
|
||
let ts = Utc
|
||
.timestamp_micros(1_700_000_000_123_456)
|
||
.single()
|
||
.expect("valid ts");
|
||
let encoded = cursor::encode(ts, "at://did:plc:abc/app.twi.post/3k2");
|
||
let (got_ts, got_tiebreak) = decode_cursor(Some(&encoded)).unwrap().unwrap();
|
||
assert_eq!(got_ts, ts);
|
||
assert_eq!(got_tiebreak, "at://did:plc:abc/app.twi.post/3k2");
|
||
|
||
// Garbage → 400, not a silent reset to page 1.
|
||
let (status, _) = decode_cursor(Some("!!!not-base64!!!")).unwrap_err();
|
||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||
}
|
||
|
||
#[test]
|
||
fn decode_cursor_rejects_out_of_range_timestamp() {
|
||
// A hand-crafted cursor whose micros overflow chrono's range
|
||
// must 400 rather than silently paginate from the top.
|
||
let raw = base64::engine::general_purpose::URL_SAFE_NO_PAD
|
||
.encode(format!("{}:at://x/y/z", i64::MAX).as_bytes());
|
||
let (status, body) = decode_cursor(Some(&raw)).unwrap_err();
|
||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||
assert_eq!(body.0["error"], "InvalidRequest");
|
||
}
|
||
|
||
/// The notification cursor reuses the shared codec but stores a
|
||
/// numeric row id in the string slot, so the id must survive the
|
||
/// round trip and parse back to an `i64`.
|
||
#[test]
|
||
fn notification_cursor_carries_row_id() {
|
||
let ts = Utc
|
||
.timestamp_micros(1_700_000_000_000_007)
|
||
.single()
|
||
.expect("valid ts");
|
||
let encoded = cursor::encode(ts, &4242i64.to_string());
|
||
let (got_ts, raw_id) = decode_cursor(Some(&encoded)).unwrap().unwrap();
|
||
assert_eq!(got_ts, ts);
|
||
assert_eq!(raw_id.parse::<i64>().unwrap(), 4242);
|
||
|
||
// A non-numeric tiebreak is what the route turns into
|
||
// "invalid cursor id" — assert the parse actually fails so the
|
||
// route's guard isn't dead code.
|
||
let bogus = cursor::encode(ts, "at://not/a/number");
|
||
let (_, raw) = decode_cursor(Some(&bogus)).unwrap().unwrap();
|
||
assert!(raw.parse::<i64>().is_err());
|
||
}
|
||
|
||
/// `FollowDirection` must map to four distinct queries: the
|
||
/// followers/following split times the with/without-cursor split.
|
||
/// A copy-paste slip here silently returns the wrong side of the
|
||
/// follow graph, which no type check would catch.
|
||
#[test]
|
||
fn follow_direction_picks_distinct_queries() {
|
||
let f = FollowDirection::Followers;
|
||
let g = FollowDirection::Following;
|
||
assert_ne!(f.sql(false), g.sql(false));
|
||
assert_ne!(f.sql(false), f.sql(true));
|
||
assert_ne!(g.sql(false), g.sql(true));
|
||
|
||
// Followers anchor on `subject_did` and list `follower_did`.
|
||
assert!(f.sql(false).contains("WHERE f.subject_did = $1"));
|
||
assert!(f.sql(false).contains("f.follower_did AS did"));
|
||
// Following is the mirror image.
|
||
assert!(g.sql(false).contains("WHERE f.follower_did = $1"));
|
||
assert!(g.sql(false).contains("f.subject_did AS did"));
|
||
|
||
// Only the cursor variants carry the keyset predicate, and
|
||
// each on its own listed column.
|
||
assert!(!f.sql(false).contains("indexed_at, f."));
|
||
assert!(f.sql(true).contains("(f.indexed_at, f.follower_did) < ($3, $4)"));
|
||
assert!(g.sql(true).contains("(f.indexed_at, f.subject_did) < ($3, $4)"));
|
||
}
|
||
|
||
/// The one place this module interpolates into SQL is the
|
||
/// notification list's `{cursor}` placeholder. Assert that both
|
||
/// substitutions produce the query we expect and that the
|
||
/// placeholder never survives into a statement.
|
||
#[test]
|
||
fn notifications_sql_cursor_substitution() {
|
||
let without = NOTIFICATIONS_SQL.replace("{cursor}", "");
|
||
let with = NOTIFICATIONS_SQL.replace("{cursor}", NOTIFICATIONS_CURSOR_PREDICATE);
|
||
assert!(!without.contains("{cursor}"));
|
||
assert!(!with.contains("{cursor}"));
|
||
assert!(!without.contains("n.indexed_at, n.id) <"));
|
||
assert!(with.contains("AND (n.indexed_at, n.id) < ($3, $4)"));
|
||
// Both variants keep `$1` = recipient and `$2` = LIMIT so the
|
||
// bind order in `notifications` is valid for either.
|
||
assert!(without.contains("WHERE n.recipient_did = $1"));
|
||
assert!(with.contains("WHERE n.recipient_did = $1"));
|
||
assert!(without.contains("LIMIT $2"));
|
||
assert!(with.contains("LIMIT $2"));
|
||
}
|
||
|
||
#[test]
|
||
fn decorate_handles_fills_empty_only() {
|
||
let mut rows = vec![
|
||
PostRow {
|
||
uri: "at://x/app.twi.post/1".into(),
|
||
did: "did:plc:abcdefghij".into(),
|
||
handle: String::new(),
|
||
rkey: "1".into(),
|
||
collection: "app.twi.post".into(),
|
||
text: "hi".into(),
|
||
cid: "c".into(),
|
||
parent_uri: None,
|
||
root_uri: None,
|
||
embed: None,
|
||
langs: crate::routes::types::Langs(vec![]),
|
||
created_at: Utc::now(),
|
||
like_count: 0,
|
||
repost_count: 0,
|
||
avatar_cid: None,
|
||
},
|
||
PostRow {
|
||
uri: "at://x/app.twi.post/2".into(),
|
||
did: "did:plc:abc".into(),
|
||
handle: "alice".into(),
|
||
rkey: "2".into(),
|
||
collection: "app.twi.post".into(),
|
||
text: "hi".into(),
|
||
cid: "c".into(),
|
||
parent_uri: None,
|
||
root_uri: None,
|
||
embed: None,
|
||
langs: crate::routes::types::Langs(vec![]),
|
||
created_at: Utc::now(),
|
||
like_count: 0,
|
||
repost_count: 0,
|
||
avatar_cid: None,
|
||
},
|
||
];
|
||
decorate_handles(&mut rows);
|
||
// Bare, no sigil: the UI renders `@{handle}` itself, so a
|
||
// prefixed placeholder showed up as `@@did:plc:abcd…`.
|
||
assert!(!rows[0].handle.starts_with('@'));
|
||
assert!(rows[0].handle.starts_with("did:"));
|
||
assert!(rows[0].handle.ends_with('…'));
|
||
assert_eq!(rows[1].handle, "alice");
|
||
}
|
||
}
|