Files
maarcadetweet/crates/appview/src/routes.rs
T
tomdeboneandEifelduds 92f2357637 perf(appview): Profil, Cold-Start-Feed und Follow-Timeline entlasten
Gemessen gegen die Dev-Instanz (3,3 Mio. Posts):

* GET /api/profile/<handle>  9,5 s → 0,04 s
* Cold-Start-Timeline        7,4 s → 0,006 s
* Timeline mit 2300 Follows   28 s → 0,02 s

Drei unabhängige Ursachen, alle drei ein Seq-Scan über die posts-Tabelle:

1. resolve_profile sucht die DID über profiles.LOWER(handle) und, als
   Fallback, über posts.handle. Für beides gab es keinen Index. Auf
   profiles hatte Migration 0007 genau diesen Index entfernt, mit der
   Begründung, jeder Aufrufer leite ohnehin zuerst eine DID ab — das
   stimmt nicht mehr, seit resolve_profile den profiles-Cache zuerst
   befragt.
2. Der Cold-Start-Feed filtert `collection IN (…)` und sortiert nach
   indexed_at. Der vorhandene (collection, indexed_at, uri)-Index taugt
   dafür nicht: mit zwei führenden Werten liefert er keine
   indexed_at-Ordnung mehr. Ein partieller Index über genau das
   Prädikat schiebt den Filter in die Definition und lässt
   (indexed_at DESC, uri DESC) als Sortierschlüssel übrig.
3. Genau dieser neue Index wurde dann zur Falle für den Graph-Zweig:
   der Planer sah einen Index, der schon in indexed_at-Ordnung liefert,
   und nahm an, er treffe früh genug auf n passende Zeilen — bei dünn
   besetzten Followees hieß "früh" 2,87 Mio. verworfene Zeilen. Je nach
   Anzahl bisheriger Ausführungen des Prepared Statements kippte er
   zwischen diesem und dem guten Plan, was intermittierend aussah.

Der Graph-Zweig formuliert die Absicht jetzt aus: pro Followee die
neuesten Posts über ein LATERAL, dann mergen. Damit ist der globale
Scan kein wählbarer Plan mehr, und jede Iteration ist ein begrenzter
Range-Scan auf posts_did_indexed_at_uri_idx. Korrekt ist das, weil die
globalen Top-N immer eine Teilmenge der Vereinigung der Top-N je
Followee sind — deshalb wird pro Followee limit+1 geholt.

Co-Authored-By: Eifelduds <noreply@eifelduds.de>
Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX
2026-09-09 23:02:47 +02:00

1862 lines
69 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! 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 ----------------------------------------------------------------
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(),
}))
}
// -- 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");
}
}