diff --git a/crates/appview/src/indexer.rs b/crates/appview/src/indexer.rs index 69263be..d5b75a0 100644 --- a/crates/appview/src/indexer.rs +++ b/crates/appview/src/indexer.rs @@ -348,6 +348,31 @@ pub async fn upsert_post(db: &PgPool, row: &mut PostRow) -> Result<()> { .bind(&row.avatar_cid) .execute(db) .await?; + + // Reply notification. A post with a `parent_uri` is a reply, so the + // parent's author gets a "someone replied to you" row. + // + // `subject_uri` is the REPLY's own URI, not the parent's: the + // notification list hydrates `subject_uri`'s text, and what the + // recipient wants to read is what the replier wrote — they already + // know the content of their own post. It also makes the row a + // direct link target for "open this reply in the thread view". + // + // The dedupe index keys on the reply URI, so re-indexing (or an + // edit that re-runs the upsert) can't produce a second row. + if let Some(parent_uri) = row.parent_uri.as_deref() { + if let Some(recipient) = post_author_did(db, parent_uri).await? { + record_notification( + db, + &recipient, + &row.did, + NOTIF_REPLY, + Some(&row.uri), + row.created_at, + ) + .await?; + } + } Ok(()) } @@ -414,6 +439,23 @@ pub async fn upsert_like( .bind(&post_uri) .execute(&mut *tx) .await?; + + // Notify the post's author. Runs inside the same transaction as + // the counter bump so a crash can't leave a like counted but + // un-notified (or vice versa). `post_author_did` returns None + // for a post we haven't indexed — then there's nobody to + // notify and we quietly skip. + if let Some(recipient) = post_author_did(&mut *tx, &post_uri).await? { + record_notification( + &mut *tx, + &recipient, + did, + NOTIF_LIKE, + Some(&post_uri), + created_at, + ) + .await?; + } } tx.commit().await?; Ok(()) @@ -489,6 +531,20 @@ pub async fn upsert_repost( .bind(&post_uri) .execute(&mut *tx) .await?; + + // Same transactional notification write as the like path — see + // `upsert_like` for the rationale. + if let Some(recipient) = post_author_did(&mut *tx, &post_uri).await? { + record_notification( + &mut *tx, + &recipient, + did, + NOTIF_REPOST, + Some(&post_uri), + created_at, + ) + .await?; + } } tx.commit().await?; Ok(()) @@ -517,6 +573,124 @@ pub async fn delete_repost(db: &PgPool, did: &str, rkey: &str) -> Result<()> { Ok(()) } +// -- notifications --------------------------------------------------------- + +/// The four notification kinds the AppView produces. Kept as `&str` +/// constants rather than an enum because the value is a plain `TEXT` +/// column guarded by a CHECK constraint (migration 0008) and every +/// call site is a literal — an enum would only add a `to_string()`. +pub const NOTIF_LIKE: &str = "like"; +pub const NOTIF_REPOST: &str = "repost"; +pub const NOTIF_FOLLOW: &str = "follow"; +pub const NOTIF_REPLY: &str = "reply"; + +/// Should an interaction by `author_did` aimed at `recipient_did` +/// produce a notification row? +/// +/// Pure so it can be unit-tested without a database. Two rules: +/// +/// 1. **No self-interactions.** Liking your own post, reposting +/// yourself, self-following or replying to yourself must stay +/// silent — the user already knows they did it, and a "you liked +/// your own post" row is pure noise. +/// 2. **No empty DIDs.** An empty recipient means we failed to resolve +/// the target (e.g. a like against a post that isn't in our index), +/// and an empty author means the event was malformed. Either way +/// the row would be unattributable in the UI. +/// +/// The same two rules are ALSO enforced in SQL inside +/// [`record_notification`], so a caller that forgets this guard still +/// can't write a bad row — this function exists to skip the round trip +/// in the common self-interaction case and to make the rule testable. +pub fn should_notify(recipient_did: &str, author_did: &str) -> bool { + !recipient_did.is_empty() && !author_did.is_empty() && recipient_did != author_did +} + +/// Insert one notification row, if and only if it is warranted. +/// +/// Returns `Ok(true)` when a row was actually written, `Ok(false)` when +/// the write was skipped — either because the interaction failed +/// [`should_notify`], because the recipient isn't a user this AppView +/// serves, or because the same notification already exists. +/// +/// **"Local user"**: the AppView deliberately has no `users` table — +/// the PDS owns account state. The closest thing we have is "a DID this +/// AppView knows about", i.e. one with a row in the `profiles` cache +/// (written by the PDS profile push / the Jetstream `app.bsky.actor.profile` +/// arm) or at least one indexed post. That is exactly the set of users +/// a client could ever poll notifications for, so restricting writes to +/// it keeps us from materialising a notification row for every like on +/// the entire public firehose while never dropping one a local user +/// would actually see. +/// +/// **Idempotency**: the `ON CONFLICT` target is the expression index +/// `notifications_dedupe_idx` from migration 0008, keyed on +/// `(recipient_did, author_did, kind, COALESCE(subject_uri, ''))`. The +/// `COALESCE` is load-bearing: a plain unique index would let two +/// `follow` rows (whose `subject_uri` is NULL) coexist, because SQL +/// NULLs never collide. Re-indexing the same Jetstream event — on a +/// reconnect replay, or via the PDS `/internal/ingest-commit` push that +/// races the firehose — is therefore a no-op. +/// +/// The executor is generic so this can run either on the pool (the +/// reply path) or inside the caller's transaction (the like / repost +/// paths, where the notification must commit atomically with the +/// counter bump). +pub async fn record_notification<'e, E>( + exec: E, + recipient_did: &str, + author_did: &str, + kind: &str, + subject_uri: Option<&str>, + created_at: chrono::DateTime, +) -> Result +where + E: sqlx::PgExecutor<'e>, +{ + if !should_notify(recipient_did, author_did) { + return Ok(false); + } + // One statement, so there's no TOCTOU window between "is the + // recipient local?" and "insert". Every parameter is explicitly + // cast because `INSERT ... SELECT $1, $2, ...` gives Postgres no + // column context to infer the placeholder types from. + let res = sqlx::query( + r#"INSERT INTO notifications + (recipient_did, author_did, kind, subject_uri, created_at) + SELECT $1::text, $2::text, $3::text, $4::text, $5::timestamptz + WHERE $1::text <> $2::text + AND $1::text <> '' + AND $2::text <> '' + AND (EXISTS (SELECT 1 FROM profiles WHERE did = $1::text) + OR EXISTS (SELECT 1 FROM posts WHERE did = $1::text)) + ON CONFLICT (recipient_did, author_did, kind, COALESCE(subject_uri, '')) + DO NOTHING"#, + ) + .bind(recipient_did) + .bind(author_did) + .bind(kind) + .bind(subject_uri) + .bind(created_at) + .execute(exec) + .await?; + Ok(res.rows_affected() > 0) +} + +/// Look up the author DID of an indexed post. `None` when the post +/// isn't in our index — which is the normal case for a like/reply +/// aimed at a post hosted somewhere we don't follow. The caller then +/// simply skips the notification rather than guessing a recipient. +async fn post_author_did<'e, E>(exec: E, uri: &str) -> Result> +where + E: sqlx::PgExecutor<'e>, +{ + let did: Option = sqlx::query_scalar("SELECT did FROM posts WHERE uri = $1") + .bind(uri) + .fetch_optional(exec) + .await?; + Ok(did) +} + // -- follows --------------------------------------------------------------- pub async fn upsert_follow( @@ -540,6 +714,22 @@ pub async fn upsert_follow( .bind(created_at) .execute(db) .await?; + + // Notify the followed user. `subject_uri` is NULL — a follow isn't + // about a post — which is exactly the case the dedupe index's + // `COALESCE(subject_uri, '')` exists for. Not wrapped in a + // transaction with the follow upsert: the follow row is the source + // of truth and a missed notification is recoverable noise, whereas + // taking a transaction here would serialise every follow write. + record_notification( + db, + subject_did, + follower_did, + NOTIF_FOLLOW, + None, + created_at, + ) + .await?; Ok(()) } @@ -1181,6 +1371,373 @@ mod tests { } } +#[cfg(test)] +mod notification_tests { + use super::*; + use serde_json::json; + + fn did(tag: &str) -> String { + format!("did:plc:notif_{}_{}", tag, uuid::Uuid::new_v4().simple()) + } + + /// Seed one post so `did` counts as a user this AppView knows + /// about (see [`record_notification`]'s "local user" note) and so + /// there's something to like / reply to. + /// + /// The `handle` is deliberately non-empty. `handle_sync`'s tests + /// assert exact counts over a *global* scan of empty-handle rows, + /// so a fixture that left the column blank would silently break + /// them whenever both suites share a database. + async fn seed_post(db: &PgPool, author: &str, rkey: &str) -> String { + let uri = format!("at://{author}/app.twi.post/{rkey}"); + sqlx::query( + r#"INSERT INTO posts + (uri, did, handle, rkey, collection, text, cid, + parent_uri, root_uri, langs, created_at) + VALUES ($1,$2,'notif-fixture.test','x','app.twi.post','seed','bafy', + NULL,NULL,NULL, now()) + ON CONFLICT (uri) DO NOTHING"#, + ) + .bind(&uri) + .bind(author) + .execute(db) + .await + .unwrap(); + uri + } + + async fn count_notifications(db: &PgPool, recipient: &str, kind: &str) -> i64 { + sqlx::query_scalar( + "SELECT COUNT(*)::BIGINT FROM notifications \ + WHERE recipient_did = $1 AND kind = $2", + ) + .bind(recipient) + .bind(kind) + .fetch_one(db) + .await + .unwrap() + } + + async fn cleanup(db: &PgPool, dids: &[&str]) { + for d in dids { + let _ = sqlx::query( + "DELETE FROM notifications WHERE recipient_did = $1 OR author_did = $1", + ) + .bind(d) + .execute(db) + .await; + let _ = sqlx::query("DELETE FROM likes WHERE did = $1") + .bind(d) + .execute(db) + .await; + let _ = sqlx::query("DELETE FROM reposts WHERE did = $1") + .bind(d) + .execute(db) + .await; + let _ = sqlx::query("DELETE FROM follows WHERE follower_did = $1 OR subject_did = $1") + .bind(d) + .execute(db) + .await; + let _ = sqlx::query("DELETE FROM posts WHERE did = $1") + .bind(d) + .execute(db) + .await; + } + } + + #[test] + fn should_notify_rejects_self_and_empty() { + assert!(should_notify("did:plc:a", "did:plc:b")); + // Self-interaction: liking / replying to / following yourself. + assert!(!should_notify("did:plc:a", "did:plc:a")); + // Unresolvable ends of the edge. + assert!(!should_notify("", "did:plc:b")); + assert!(!should_notify("did:plc:a", "")); + assert!(!should_notify("", "")); + // Case matters — DIDs are compared verbatim, never folded. + assert!(should_notify("did:plc:A", "did:plc:a")); + } + + /// A like by someone else must produce exactly one notification, + /// and re-indexing the same event (Jetstream replay racing the PDS + /// push) must not produce a second. + #[tokio::test] + async fn like_notifies_author_once() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let author = did("author"); + let liker = did("liker"); + // The liker also needs to exist for nothing in particular — + // only the *recipient* is checked — but seeding both keeps the + // fixture symmetric with reality. + let post_uri = seed_post(&db, &author, "p1").await; + seed_post(&db, &liker, "p1").await; + + let record = json!({ + "subject": { "uri": post_uri, "cid": "bafysubject" }, + "createdAt": "2026-07-01T12:00:00Z" + }); + upsert_like(&db, &liker, "lrk1", Some("bafylike"), Some(&record)) + .await + .unwrap(); + assert_eq!(count_notifications(&db, &author, NOTIF_LIKE).await, 1); + + // Replay the exact same event. + upsert_like(&db, &liker, "lrk1", Some("bafylike"), Some(&record)) + .await + .unwrap(); + assert_eq!( + count_notifications(&db, &author, NOTIF_LIKE).await, + 1, + "replayed like must not duplicate the notification" + ); + + // Unlike + re-like under a NEW rkey. The like row is recreated + // but the notification tuple is unchanged, so the dedupe index + // suppresses it — see migration 0008's "Idempotency" note. + delete_like(&db, &liker, "lrk1").await.unwrap(); + upsert_like(&db, &liker, "lrk2", Some("bafylike2"), Some(&record)) + .await + .unwrap(); + assert_eq!( + count_notifications(&db, &author, NOTIF_LIKE).await, + 1, + "toggling a like must not be a notification-spam vector" + ); + + // The stored row must point at the liked post and name the + // liker as the author. + let row: (String, Option) = sqlx::query_as( + "SELECT author_did, subject_uri FROM notifications \ + WHERE recipient_did = $1 AND kind = $2", + ) + .bind(&author) + .bind(NOTIF_LIKE) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(row.0, liker); + assert_eq!(row.1.as_deref(), Some(post_uri.as_str())); + // Unread by default — that's what /api/notifications/count sees. + let unread: i64 = sqlx::query_scalar( + "SELECT COUNT(*)::BIGINT FROM notifications \ + WHERE recipient_did = $1 AND read_at IS NULL", + ) + .bind(&author) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(unread, 1); + + cleanup(&db, &[&author, &liker]).await; + } + + /// Liking / reposting your own post is silent. + #[tokio::test] + async fn self_interaction_writes_no_notification() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let author = did("selfie"); + let post_uri = seed_post(&db, &author, "p1").await; + let record = json!({ + "subject": { "uri": post_uri, "cid": "bafysubject" }, + "createdAt": "2026-07-01T12:00:00Z" + }); + + upsert_like(&db, &author, "lrk1", None, Some(&record)) + .await + .unwrap(); + upsert_repost(&db, &author, "rrk1", None, Some(&record)) + .await + .unwrap(); + // Self-follow is legal in the protocol; it must stay silent too. + upsert_follow(&db, &author, &author, None).await.unwrap(); + + let total: i64 = sqlx::query_scalar( + "SELECT COUNT(*)::BIGINT FROM notifications WHERE recipient_did = $1", + ) + .bind(&author) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(total, 0, "self-interactions must not notify"); + + // The like/repost themselves still landed — the notification + // suppression must not swallow the interaction. + let likes: i64 = sqlx::query_scalar("SELECT COUNT(*)::BIGINT FROM likes WHERE did = $1") + .bind(&author) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(likes, 1); + + cleanup(&db, &[&author]).await; + } + + /// A repost notifies, and a reply notifies the *parent's* author + /// with the reply's own URI as the subject. + #[tokio::test] + async fn repost_and_reply_notify() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let author = did("parent"); + let other = did("replier"); + let parent_uri = seed_post(&db, &author, "p1").await; + seed_post(&db, &other, "p1").await; + + let record = json!({ + "subject": { "uri": parent_uri, "cid": "bafysubject" }, + "createdAt": "2026-07-01T12:00:00Z" + }); + upsert_repost(&db, &other, "rrk1", None, Some(&record)) + .await + .unwrap(); + assert_eq!(count_notifications(&db, &author, NOTIF_REPOST).await, 1); + + // Now a reply from `other` to `author`'s post. + let reply_record = json!({ + "text": "nice one", + "createdAt": "2026-07-01T12:05:00Z", + "reply": { + "parent": { "uri": parent_uri, "cid": "bafyparent" }, + "root": { "uri": parent_uri, "cid": "bafyparent" } + } + }); + let mut row = PostRow::from_record( + &other, + "replykey", + "app.twi.post", + "bafyreply", + &reply_record, + None, + ); + let reply_uri = row.uri.clone(); + upsert_post(&db, &mut row).await.unwrap(); + + let subject: Option = sqlx::query_scalar( + "SELECT subject_uri FROM notifications \ + WHERE recipient_did = $1 AND kind = $2", + ) + .bind(&author) + .bind(NOTIF_REPLY) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!( + subject.as_deref(), + Some(reply_uri.as_str()), + "a reply notification's subject is the REPLY, so the list can \ + preview what was written" + ); + + // Re-indexing the reply must not duplicate. + upsert_post(&db, &mut row).await.unwrap(); + assert_eq!(count_notifications(&db, &author, NOTIF_REPLY).await, 1); + + cleanup(&db, &[&author, &other]).await; + } + + /// A follow notifies the followed user with a NULL subject_uri — + /// the case the dedupe index's `COALESCE(subject_uri, '')` exists + /// for, since plain SQL NULLs never collide. + #[tokio::test] + async fn follow_notifies_subject_and_dedupes_on_null_subject() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let subject = did("followee"); + let follower = did("follower"); + seed_post(&db, &subject, "p1").await; + + let record = json!({ "subject": subject, "createdAt": "2026-01-01T00:00:00Z" }); + upsert_follow(&db, &follower, &subject, Some(&record)) + .await + .unwrap(); + upsert_follow(&db, &follower, &subject, Some(&record)) + .await + .unwrap(); + assert_eq!( + count_notifications(&db, &subject, NOTIF_FOLLOW).await, + 1, + "two NULL-subject follow rows must collide, not coexist" + ); + + let subject_uri: Option = sqlx::query_scalar( + "SELECT subject_uri FROM notifications \ + WHERE recipient_did = $1 AND kind = $2", + ) + .bind(&subject) + .bind(NOTIF_FOLLOW) + .fetch_one(&db) + .await + .unwrap(); + assert!(subject_uri.is_none(), "a follow is not about a post"); + + cleanup(&db, &[&subject, &follower]).await; + } + + /// A recipient the AppView has never seen (no profile row, no + /// posts) gets nothing — this is what keeps us from materialising + /// a row for every like on the public firehose. + #[tokio::test] + async fn unknown_recipient_is_skipped() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let stranger = did("stranger"); + let author = did("author"); + + let wrote = record_notification( + &db, + &stranger, + &author, + NOTIF_FOLLOW, + None, + chrono::Utc::now(), + ) + .await + .unwrap(); + assert!(!wrote, "unknown recipient must not get a notification row"); + + // Give the recipient a post — now they're a user we serve. + seed_post(&db, &stranger, "p1").await; + let wrote = record_notification( + &db, + &stranger, + &author, + NOTIF_FOLLOW, + None, + chrono::Utc::now(), + ) + .await + .unwrap(); + assert!(wrote, "known recipient must get the row"); + + // And the second call is a no-op thanks to ON CONFLICT. + let wrote = record_notification( + &db, + &stranger, + &author, + NOTIF_FOLLOW, + None, + chrono::Utc::now(), + ) + .await + .unwrap(); + assert!(!wrote, "duplicate must report false, not error"); + + cleanup(&db, &[&stranger, &author]).await; + } +} + /// Backfill the `posts.handle` column for every row belonging to /// `did`. Used by the Jetstream `identity` handler when Jetstream /// tells us a DID's handle has changed — every existing post row diff --git a/crates/appview/src/routes.rs b/crates/appview/src/routes.rs index 8995f0a..f8da41a 100644 --- a/crates/appview/src/routes.rs +++ b/crates/appview/src/routes.rs @@ -29,7 +29,11 @@ use crate::state::AppState; pub mod cursor; pub mod types; -use types::{PostRow, PostRowWithIndexed, ProfileResponse, SearchResponse, TimelineResponse}; +use types::{ + ActorListResponse, ActorProfile, NotificationCountResponse, NotificationItem, + NotificationSeenResponse, NotificationsResponse, PostRow, PostRowWithIndexed, + ProfileResponse, SearchResponse, ThreadFullResponse, TimelineResponse, +}; pub fn router(state: AppState) -> Router { // CORS: the Tauri webview's origin is the Vite dev server @@ -54,6 +58,19 @@ pub fn router(state: AppState) -> Router { .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/` + // 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/notifications", get(notifications)) + .route("/api/notifications/count", get(notifications_count)) + .route("/api/notifications/seen", post(notifications_seen)) + .route("/api/followers", get(followers)) + .route("/api/following", get(following)) .route("/healthz", get(healthz)) .route("/internal/ingest-commit", post(crate::ingest::ingest_commit)) .layer(cors) @@ -105,10 +122,7 @@ async fn timeline_home( if q.did.is_empty() { return Err(bad_request("did is required")); } - let limit = q - .limit - .unwrap_or(DEFAULT_LIMIT) - .clamp(1, MAX_LIMIT); + 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: @@ -170,30 +184,9 @@ async fn timeline_home( target_dids.sort(); } - let decode = match q.cursor.as_deref().map(cursor::decode) { - Some(Ok(c)) => Some(c), - Some(Err(e)) => return Err(bad_request(&e)), - None => None, - }; - // Convert the cursor's microsecond timestamp to a DateTime so - // sqlx binds it as `timestamptz` rather than `bigint` (which would - // fail the `(indexed_at, uri) < ($1, $2)` row comparison). - // - // If the timestamp is out of chrono::Utc's representable range - // (e.g. i64::MAX from a malicious cursor) we reject with 400 instead - // of silently falling back to page 1, which would lose the user's - // pagination state. - let cursor_ts: Option> = match decode.as_ref() { - Some(c) => Some( - chrono::Utc - .timestamp_micros(c.ts) - .single() - .ok_or_else(|| bad_request("invalid cursor timestamp"))?, - ), - None => None, - }; - let cursor_uri: Option = - decode.as_ref().map(|c| c.uri.clone()); + let keyset = decode_cursor(q.cursor.as_deref())?; + let cursor_ts: Option> = keyset.as_ref().map(|(ts, _)| *ts); + let cursor_uri: Option = keyset.map(|(_, uri)| uri); // Fetch limit+1 to know if there's a next page without a second // round trip. let fetch = limit + 1; @@ -444,9 +437,9 @@ async fn resolve_profile( .map(|p| p.handle.clone()) }) .unwrap_or_else(|| { - // Last-resort synthetic handle. The schema note says - // "@" is acceptable; we keep it short and safe. - short_did_for_display(&target_did) + // Last-resort synthetic handle. Bare, without the `@`: + // every consumer of this field renders `@{handle}` itself. + short_did_bare(&target_did) }); decorate_handles(&mut posts); @@ -588,39 +581,154 @@ async fn post_by_uri( return Err(bad_request("uri is required")); } let uri = percent_decode(&uri); - let post: Option = 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 uri = $1 - LIMIT 1"#, - ) - .bind(&uri) - .fetch_optional(&state.db) - .await - .map_err(db_err)?; - // Pull the two refs off the row before we move it into the response - // — `post` is consumed by the `Some(p)` arm but we still need - // `parent_uri` / `root_uri` for the hydration lookups. - let (parent_uri, root_uri, like_count, repost_count) = match post.as_ref() { - Some(p) => ( - p.parent_uri.clone(), - p.root_uri.clone(), - Some(p.like_count), - Some(p.repost_count), - ), - None => (None, None, None, None), + // 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, + /// Ancestor chain, **root first / immediate parent last**. + parents: Vec, + root: Option, + replies: Vec, +} + +/// 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)> { + 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), }; - let parent: Option = match parent_uri.as_deref() { - Some(u) => fetch_one_post(&state, u).await?, - 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 = + std::collections::HashSet::new(); + visited.insert(uri.to_string()); + let mut chain: Vec = 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 = match root_uri.as_deref() { Some(u) if Some(u) != parent_uri.as_deref() => { - fetch_one_post(&state, u).await? + 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 @@ -629,41 +737,88 @@ async fn post_by_uri( None => None, }; - // Viewer-scoped engagement state. We only run these queries when - // (a) the post was found (otherwise `None` so the UI can ignore - // viewer state on a missing post) and (b) the caller actually - // passed a viewer_did. - let (viewer_liked, viewer_reposted) = if post.is_some() { - if let Some(viewer) = q.viewer_did.as_deref() { - 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)?; - (Some(liked), Some(reposted)) - } else { - (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 = 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 { - (None, None) + Vec::new() }; - // We hand `parent` / `root` to the response and `post` last so the - // borrow on `parent_uri` / `root_uri` is already released. - Ok(Json(ThreadResponse { + Ok(ThreadContext { post, - thread: ThreadView { parent, root }, + 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/`. + #[serde(default)] + uri: Option, + #[serde(default)] + viewer_did: Option, +} + +/// `GET /api/thread?uri=…` +async fn thread_query( + State(state): State, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + let uri = q.uri.clone().unwrap_or_default(); + thread_inner(&state, &uri, q.viewer_did.as_deref()).await +} + +/// `GET /api/thread/` — path form, mirroring `/api/post/*uri`. +async fn thread_path( + State(state): State, + Path(uri): Path, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + thread_inner(&state, &uri, q.viewer_did.as_deref()).await +} + +async fn thread_inner( + state: &AppState, + uri: &str, + viewer_did: Option<&str>, +) -> Result, (StatusCode, Json)> { + 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, @@ -671,18 +826,54 @@ async fn post_by_uri( })) } +/// 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, Option), (StatusCode, Json)> { + 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, (StatusCode, Json)> { - 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 + sqlx::query_as::<_, PostRow>(&format!( + r#"SELECT {POST_COLUMNS} FROM posts WHERE uri = $1 - LIMIT 1"#, - ) + LIMIT 1"# + )) .bind(uri) .fetch_optional(&state.db) .await @@ -783,6 +974,423 @@ async fn search( })) } +// -- 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, + #[serde(default)] + cursor: Option, +} + +/// `GET /api/notifications?did=…&limit=…&cursor=…` +/// +/// Same pagination contract as `/api/timeline/home`: newest first, +/// keyset-ordered on `(indexed_at, )`, `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. +async fn notifications( + State(state): State, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + 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())?; + // 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 = match keyset.as_ref() { + Some((_, raw)) => Some( + raw.parse::() + .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 = + 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. +async fn notifications_count( + State(state): State, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + if q.did.is_empty() { + return Err(bad_request("did is required")); + } + 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>, +} + +/// `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). +async fn notifications_seen( + State(state): State, + Json(req): Json, +) -> Result, (StatusCode, Json)> { + if req.did.is_empty() { + return Err(bad_request("did is required")); + } + 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, + #[serde(default)] + cursor: Option, +} + +/// `GET /api/followers?did=…&limit=&cursor=` +async fn followers( + State(state): State, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + actor_list(&state, FollowDirection::Followers, &q).await +} + +/// `GET /api/following?did=…&limit=&cursor=` +async fn following( + State(state): State, + Query(q): Query, +) -> Result, (StatusCode, Json)> { + actor_list(&state, FollowDirection::Following, &q).await +} + +/// Shared body of `/api/followers` and `/api/following`. +/// +/// The cursor is the `follows` edge's `(indexed_at, )` +/// 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, (StatusCode, Json)> { + 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, Option, DateTime), + >(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 = 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) -> impl IntoResponse { @@ -804,16 +1412,61 @@ async fn healthz(State(state): State) -> impl IntoResponse { fn decorate_handles(posts: &mut [PostRow]) { for p in posts.iter_mut() { if p.handle.is_empty() { - p.handle = short_did_for_display(&p.did); + p.handle = short_did_bare(&p.did); } } } -fn short_did_for_display(did: &str) -> String { - // Match the spec: "@{first-12-chars-of-did}…". Use char-based slicing - // so we never panic on a UTF-8 boundary (e.g. `did:web:münchen.de`). +/// 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}…") + 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 { + 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` (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, String)>, (StatusCode, Json)> { + 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 '\'` @@ -857,6 +1510,7 @@ fn bad_request(msg: &str) -> (StatusCode, Json) { #[cfg(test)] mod tests { use super::*; + use base64::Engine; #[test] fn escape_like_handles_wildcards() { @@ -868,10 +1522,8 @@ mod tests { #[test] fn short_did_format_matches_spec() { - let s = short_did_for_display("did:plc:abcdefghijklmnop"); - assert_eq!(s, "@did:plc:abcd…"); - let s = short_did_for_display("short"); - assert_eq!(s, "@short…"); + assert_eq!(short_did_bare("did:plc:abcdefghijklmnop"), "did:plc:abcd…"); + assert_eq!(short_did_bare("short"), "short…"); } #[test] @@ -891,6 +1543,130 @@ mod tests { 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::().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::().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![ @@ -930,7 +1706,10 @@ mod tests { }, ]; decorate_handles(&mut rows); - assert!(rows[0].handle.starts_with('@')); + // 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"); } diff --git a/crates/appview/src/routes/types.rs b/crates/appview/src/routes/types.rs index ce9f62e..f755ac6 100644 --- a/crates/appview/src/routes/types.rs +++ b/crates/appview/src/routes/types.rs @@ -199,6 +199,165 @@ pub struct SearchResponse { pub q: String, } +// -- thread ----------------------------------------------------------------- + +/// `GET /api/thread` response. +/// +/// This is the *full* thread shape: the ancestor chain above the post +/// and the direct replies below it. `/api/post/{uri}` keeps its own, +/// narrower `{ post, thread: { parent, root } }` shape for backwards +/// compatibility — both are built from the same +/// `routes::load_thread_context` call, so they can never disagree +/// about what the parent or root is. +/// +/// `post` is `None` when the URI isn't in our index. In that case +/// `parents` / `replies` are empty and the counters are omitted, so a +/// client can render "post not found" from a single field check. +#[derive(Debug, Serialize)] +pub struct ThreadFullResponse { + pub post: Option, + /// Ancestor chain ordered **root first, immediate parent last**. + /// Empty for a top-level post, and truncated (from the top) when + /// the chain is longer than the walk limit or when an ancestor + /// isn't in our index — the client should treat a `parents[0]` + /// whose `parent_uri` is non-null as "chain continues above, + /// not loaded". + pub parents: Vec, + /// The thread root as named by the post's own `root_uri`. May be + /// the same row as `parents[0]`, and is `None` for a top-level + /// post (which is its own root). + pub root: Option, + /// Direct replies to `post`, oldest first. Only direct children — + /// the client re-requests `/api/thread` for a nested branch. + pub replies: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub like_count: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub repost_count: Option, + /// Same semantics as on `/api/post/{uri}`: `None` means "no + /// viewer_did was supplied, state unknown" — not "not liked". + #[serde(skip_serializing_if = "Option::is_none")] + pub viewer_liked: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub viewer_reposted: Option, +} + +// -- notifications ---------------------------------------------------------- + +/// One hydrated row of `GET /api/notifications`. +/// +/// The DB row only stores DIDs and a subject URI; the read query joins +/// the `profiles` cache (falling back to the newest non-empty +/// `posts.handle` for authors we've seen post but never had a profile +/// record for) and the `posts` table so the client can render a full +/// notification line without any follow-up fetch. +/// +/// `read_at` is serialised even when `None` — unlike the optional +/// fields around it — because "unread" is the state the client's badge +/// keys off, and a *missing* key would be indistinguishable from a +/// field the client forgot to read. +#[derive(Debug, Clone, Serialize)] +pub struct NotificationItem { + /// Row id. Also the tiebreaker inside the opaque cursor. + pub id: i64, + /// `"like"` | `"repost"` | `"follow"` | `"reply"`. + pub kind: String, + pub author_did: String, + /// Best known handle for the author, WITHOUT a leading `@` (the UI + /// renders `@{handle}`). Falls back to a truncated DID, and is + /// never null. + pub author_handle: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub author_display_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub author_avatar_cid: Option, + /// The post this notification is about. `null` for `"follow"`. For + /// `"like"`/`"repost"` it's the recipient's own post; for + /// `"reply"` it's the reply itself. + pub subject_uri: Option, + /// Text of `subject_uri`'s post. Omitted when the subject is not + /// (or no longer) in our index, and always for `"follow"`. + #[serde(skip_serializing_if = "Option::is_none")] + pub subject_text: Option, + /// The interaction's own `createdAt` from the AT record. + pub created_at: DateTime, + /// When the AppView indexed it. This is the list's sort key, and + /// the value a client should echo back as `seenAt`. + pub indexed_at: DateTime, + /// `null` while unread. + pub read_at: Option>, +} + +impl<'r> FromRow<'r, sqlx::postgres::PgRow> for NotificationItem { + fn from_row(row: &'r sqlx::postgres::PgRow) -> sqlx::Result { + Ok(NotificationItem { + id: row.try_get("id")?, + kind: row.try_get("kind")?, + author_did: row.try_get("author_did")?, + author_handle: row.try_get("author_handle")?, + author_display_name: row.try_get("author_display_name")?, + author_avatar_cid: row.try_get("author_avatar_cid")?, + subject_uri: row.try_get("subject_uri")?, + subject_text: row.try_get("subject_text")?, + created_at: row.try_get("created_at")?, + indexed_at: row.try_get("indexed_at")?, + read_at: row.try_get("read_at")?, + }) + } +} + +/// `GET /api/notifications` response. `cursor` is `None` at the end of +/// the list — same contract as [`TimelineResponse`]. +#[derive(Debug, Serialize)] +pub struct NotificationsResponse { + pub notifications: Vec, + pub cursor: Option, +} + +/// `GET /api/notifications/count` response. +#[derive(Debug, Serialize)] +pub struct NotificationCountResponse { + /// Number of rows with `read_at IS NULL` for this DID. + pub count: i64, +} + +/// `POST /api/notifications/seen` response. +#[derive(Debug, Serialize)] +pub struct NotificationSeenResponse { + pub ok: bool, + /// How many previously-unread rows this call flipped to read. + /// Zero is a normal, successful outcome (nothing was unread). + pub updated: i64, +} + +// -- actor lists (followers / following) ------------------------------------ + +/// A minimal profile card, as returned by `GET /api/followers` and +/// `GET /api/following`. +/// +/// Deliberately *not* [`ProfileResponse`]: those endpoints return a +/// list, and shipping each entry's posts + counts would turn one page +/// of 30 followers into 30 post queries. The client renders a row with +/// avatar + name + handle and navigates to `/api/profile` on click. +#[derive(Debug, Clone, Serialize)] +pub struct ActorProfile { + pub did: String, + /// Without a leading `@`; falls back to a truncated DID and is + /// never null. + pub handle: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub display_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub avatar_cid: Option, +} + +/// `GET /api/followers` / `GET /api/following` response. +#[derive(Debug, Serialize)] +pub struct ActorListResponse { + pub profiles: Vec, + pub cursor: Option, +} + // -- Langs newtype ---------------------------------------------------------- /// A list of language tags. Always serialises as `Vec`, never diff --git a/crates/appview/tests/api_integration.rs b/crates/appview/tests/api_integration.rs index 33641c0..99bb9d5 100644 --- a/crates/appview/tests/api_integration.rs +++ b/crates/appview/tests/api_integration.rs @@ -134,7 +134,11 @@ async fn timeline_returns_seeded_posts() { "cid": "bafyreicid", "record": { "text": format!("seeded post #{i}"), - "createdAt": "2026-07-01T12:00:00Z", + // Strictly increasing so the ordering assertion + // below has something to actually check: the rows + // are inserted in this order, so `indexed_at` and + // `created_at` agree for *our* posts. + "createdAt": format!("2026-07-01T12:00:{:02}Z", i), } }), ) @@ -147,9 +151,15 @@ async fn timeline_returns_seeded_posts() { // three rows in the first page. tokio::time::sleep(Duration::from_millis(50)).await; + // `limit=100`, not 10: this DID follows nobody, so the endpoint + // serves the cold-start *global* recent feed. On any database with + // more than a handful of recent posts (i.e. every developer + // machine that has run this suite twice) the three rows we just + // seeded fall outside a 10-row window and the assertions below + // fail for reasons that have nothing to do with the timeline. let resp = c .get(format!("{APPVIEW_URL}/api/timeline/home")) - .query(&[("did", did.as_str()), ("limit", "10")]) + .query(&[("did", did.as_str()), ("limit", "100")]) .send() .await .unwrap(); @@ -174,15 +184,28 @@ async fn timeline_returns_seeded_posts() { .collect(); assert!(our_uris.len() >= 3, "missing our seeded posts in {posts:?}"); - // Posts must be sorted with `indexed_at DESC`. We can't see - // indexed_at directly in the response, but the URI order in - // `app.twi.post/` is rkey-random here, so we only assert - // `created_at` is non-increasing. + // Posts come back `indexed_at DESC`, and we can't see `indexed_at` + // in the response — but for the three rows WE just inserted, + // insertion order == `indexed_at` order == `created_at` order, so + // their `created_at` values must be non-increasing. + // + // Two fixes over the original assertion: + // - the wire field is `created_at`, not `createdAt` (the + // `PostRow` wire type in `routes/types.rs` carries no + // `rename_all = "camelCase"`), so `p["createdAt"]` was JSON + // `null` and `.as_str().unwrap()` panicked on the first row; + // - it ran over ALL posts, including other tests' fixtures from + // the global cold-start feed, whose `created_at` values have + // no relation to their `indexed_at` order. Restricting it to + // our own DID is the only version of this claim that holds. let mut prev: Option = None; - for p in posts { - let ca = p["createdAt"].as_str().unwrap().to_string(); - if let Some(p) = prev.take() { - assert!(ca <= p, "createdAt must be non-increasing: {ca} <= {p}"); + for p in posts.iter().filter(|p| p["did"] == json!(did)) { + let ca = p["created_at"].as_str().unwrap().to_string(); + if let Some(prev) = prev.take() { + assert!( + ca <= prev, + "created_at must be non-increasing: {ca} <= {prev}" + ); } prev = Some(ca); } @@ -354,16 +377,24 @@ async fn profile_returns_posts_for_handle() { .unwrap(); assert_eq!(resp.status().as_u16(), 200); - // 404 for an unknown handle. + // An unknown handle is NOT a 404. `resolve_profile` deliberately + // synthesises an empty profile (empty `did`, zero counts, no + // posts) so the UI renders an empty profile page instead of an + // error toast — see the comment on the `let Some(target_did)` + // else-branch in `routes.rs`. This assertion used to expect 404 + // and contradicted the endpoint it was testing. + let unknown = format!("nobody_{}", uuid::Uuid::new_v4().simple()); let resp = c - .get(format!( - "{APPVIEW_URL}/api/profile/nobody_{}", - uuid::Uuid::new_v4().simple() - )) + .get(format!("{APPVIEW_URL}/api/profile/{unknown}")) .send() .await .unwrap(); - assert_eq!(resp.status().as_u16(), 404); + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["did"], json!("")); + assert_eq!(body["handle"], json!(unknown)); + assert!(body["posts"].as_array().unwrap().is_empty()); + assert_eq!(body["post_count"], json!(0)); // /api/profile?did=... must work too. let resp = c diff --git a/crates/appview/tests/notifications_integration.rs b/crates/appview/tests/notifications_integration.rs new file mode 100644 index 0000000..666ecbb --- /dev/null +++ b/crates/appview/tests/notifications_integration.rs @@ -0,0 +1,757 @@ +//! Integration tests for the Phase-5c read API additions: +//! `/api/notifications` (+ `/count`, `/seen`), `/api/followers`, +//! `/api/following` and `/api/thread`. +//! +//! Same contract as `api_integration.rs`: these run against a live +//! appview service + DB and are **fail-open**. If the service or the +//! database isn't reachable the test prints a notice and returns +//! success, so `cargo test --workspace` stays green on a machine where +//! `docker compose up` hasn't been run. +//! +//! Notification rows are written by the *indexer*, not by any HTTP +//! endpoint, so every test here seeds through `/internal/ingest-commit` +//! (the same path the PDS uses) and then reads back through the public +//! API. That's deliberate: it's the only way to catch a mismatch +//! between what the write path stores and what the read path joins. + +use serde_json::{json, Value}; +use std::time::Duration; + +/// Base URL of the appview under test. Overridable so the suite can be +/// pointed at a throwaway instance on a scratch database instead of +/// whatever the developer happens to have running on the default port. +fn appview_url() -> String { + std::env::var("APPVIEW_TEST_URL") + .unwrap_or_else(|_| "http://127.0.0.1:2584".to_string()) +} + +async fn client() -> reqwest::Client { + reqwest::Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .unwrap() +} + +async fn wait_for_appview_db() -> bool { + let base = appview_url(); + let c = client().await; + for _ in 0..20 { + if let Ok(r) = c.get(format!("{base}/healthz")).send().await { + if r.status().is_success() { + return true; + } + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + false +} + +async fn db_pool() -> Option { + let url = std::env::var("DATABASE_URL_APPVIEW").ok()?; + match tokio::time::timeout(Duration::from_secs(2), sqlx::PgPool::connect(&url)).await { + Ok(Ok(pool)) => Some(pool), + _ => None, + } +} + +/// Guard used at the top of every test. Returns `None` (→ skip) unless +/// both the HTTP service and the database are up. +async fn ready() -> Option<(reqwest::Client, sqlx::PgPool)> { + if !wait_for_appview_db().await { + eprintln!("appview not running, skipping"); + return None; + } + let Some(pool) = db_pool().await else { + eprintln!("appview DB unreachable, skipping"); + return None; + }; + Some((client().await, pool)) +} + +async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response { + let base = appview_url(); + c.post(format!("{base}/internal/ingest-commit")) + .json(&body) + .send() + .await + .unwrap() +} + +fn did_for_test(name: &str) -> String { + format!("did:plc:ntf_{}_{}", name, uuid::Uuid::new_v4().simple()) +} + +fn rkey() -> String { + uuid::Uuid::new_v4().simple().to_string() +} + +/// Create one post through the ingest path and return its URI. +async fn seed_post(c: &reqwest::Client, did: &str, text: &str) -> String { + let rk = rkey(); + let r = post_ingest( + c, + json!({ + "did": did, + "handle": "ntf-fixture.test", + "collection": "app.twi.post", + "action": "create", + "rkey": rk, + "cid": "bafyreicid", + "record": { "text": text, "createdAt": "2026-07-01T12:00:00Z" } + }), + ) + .await; + assert_eq!(r.status().as_u16(), 200); + format!("at://{did}/app.twi.post/{rk}") +} + +/// Create a reply to `parent_uri` and return the reply's URI. +async fn seed_reply( + c: &reqwest::Client, + did: &str, + parent_uri: &str, + root_uri: &str, + text: &str, +) -> String { + let rk = rkey(); + let r = post_ingest( + c, + json!({ + "did": did, + "handle": "ntf-fixture.test", + "collection": "app.twi.post", + "action": "create", + "rkey": rk, + "cid": "bafyreicid", + "record": { + "text": text, + "createdAt": "2026-07-01T12:05:00Z", + "reply": { + "parent": { "uri": parent_uri, "cid": "bafyparent" }, + "root": { "uri": root_uri, "cid": "bafyroot" } + } + } + }), + ) + .await; + assert_eq!(r.status().as_u16(), 200); + format!("at://{did}/app.twi.post/{rk}") +} + +async fn seed_like(c: &reqwest::Client, did: &str, post_uri: &str) { + let r = post_ingest( + c, + json!({ + "did": did, + "collection": "app.bsky.feed.like", + "action": "create", + "rkey": rkey(), + "cid": "bafylike", + "record": { + "subject": { "uri": post_uri, "cid": "bafyreicid" }, + "createdAt": "2026-07-01T12:01:00Z" + } + }), + ) + .await; + assert_eq!(r.status().as_u16(), 200); +} + +async fn seed_follow(c: &reqwest::Client, follower: &str, subject: &str) { + let r = post_ingest( + c, + json!({ + "did": follower, + "collection": "app.bsky.graph.follow", + "action": "create", + "rkey": rkey(), + "subject_did": subject, + "record": { "subject": subject, "createdAt": "2026-01-01T00:00:00Z" } + }), + ) + .await; + assert_eq!(r.status().as_u16(), 200); +} + +// -- notifications ---------------------------------------------------------- + +/// A like, a repost-free reply and a follow from three different +/// people must show up as three hydrated notification rows, and the +/// unread count must agree with the list. +#[tokio::test] +async fn notifications_list_count_and_seen() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let alice = did_for_test("alice"); + let bob = did_for_test("bob"); + let carol = did_for_test("carol"); + + // Alice posts; bob likes it and carol replies; bob also follows her. + let post_uri = seed_post(&c, &alice, "alice's original").await; + seed_post(&c, &bob, "bob exists").await; + seed_post(&c, &carol, "carol exists").await; + seed_like(&c, &bob, &post_uri).await; + let reply_uri = seed_reply(&c, &carol, &post_uri, &post_uri, "carol's reply").await; + seed_follow(&c, &bob, &alice).await; + + let resp = c + .get(format!("{base}/api/notifications")) + .query(&[("did", alice.as_str()), ("limit", "50")]) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await.unwrap(); + let items = body["notifications"].as_array().expect("notifications array"); + assert_eq!(items.len(), 3, "expected like + reply + follow, got {items:?}"); + + // Newest first: `indexed_at` must be non-increasing down the list. + let mut prev: Option = None; + for n in items { + let at = n["indexed_at"].as_str().unwrap().to_string(); + if let Some(p) = prev.take() { + assert!(at <= p, "indexed_at must be non-increasing: {at} <= {p}"); + } + prev = Some(at); + } + + let by_kind = |k: &str| -> Value { + items + .iter() + .find(|n| n["kind"] == json!(k)) + .unwrap_or_else(|| panic!("missing {k} notification in {items:?}")) + .clone() + }; + + // The like points at alice's own post and previews its text. + let like = by_kind("like"); + assert_eq!(like["author_did"], json!(bob)); + assert_eq!(like["subject_uri"], json!(post_uri)); + assert_eq!(like["subject_text"], json!("alice's original")); + assert!(like["read_at"].is_null(), "new notifications start unread"); + + // The reply points at the REPLY (not the parent), so the preview + // shows what carol wrote. + let reply = by_kind("reply"); + assert_eq!(reply["author_did"], json!(carol)); + assert_eq!(reply["subject_uri"], json!(reply_uri)); + assert_eq!(reply["subject_text"], json!("carol's reply")); + + // A follow has no subject at all. + let follow = by_kind("follow"); + assert_eq!(follow["author_did"], json!(bob)); + assert!(follow["subject_uri"].is_null()); + + // `author_handle` is never empty and never carries a leading '@' + // (the UI renders `@{handle}` itself). + for n in items { + let h = n["author_handle"].as_str().expect("author_handle is a string"); + assert!(!h.is_empty(), "author_handle must never be empty: {n:?}"); + assert!(!h.starts_with('@'), "author_handle must not carry a sigil: {h}"); + } + + // The unread count agrees with the list. + let resp = c + .get(format!("{base}/api/notifications/count")) + .query(&[("did", alice.as_str())]) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["count"], json!(3)); + + // Mark everything seen. + let resp = c + .post(format!("{base}/api/notifications/seen")) + .json(&json!({ "did": alice })) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["ok"], json!(true)); + assert_eq!(body["updated"], json!(3)); + + // Idempotent: a second call updates nothing and still succeeds. + let resp = c + .post(format!("{base}/api/notifications/seen")) + .json(&json!({ "did": alice })) + .send() + .await + .unwrap(); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["updated"], json!(0)); + + // Count is now zero and the rows carry a read_at. + let resp = c + .get(format!("{base}/api/notifications/count")) + .query(&[("did", alice.as_str())]) + .send() + .await + .unwrap(); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["count"], json!(0)); + + let resp = c + .get(format!("{base}/api/notifications")) + .query(&[("did", alice.as_str())]) + .send() + .await + .unwrap(); + let body: Value = resp.json().await.unwrap(); + for n in body["notifications"].as_array().unwrap() { + assert!(!n["read_at"].is_null(), "row should be read now: {n:?}"); + } + + // `did` is mandatory on all three. + for path in [ + "/api/notifications", + "/api/notifications/count", + ] { + let resp = c + .get(format!("{base}{path}")) + .query(&[("did", "")]) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 400, "{path} must reject an empty did"); + } +} + +/// Self-interactions produce nothing: alice liking and replying to her +/// own post leaves her notification list empty. +#[tokio::test] +async fn notifications_skip_self_interactions() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let alice = did_for_test("solo"); + let post_uri = seed_post(&c, &alice, "talking to myself").await; + seed_like(&c, &alice, &post_uri).await; + seed_reply(&c, &alice, &post_uri, &post_uri, "and replying too").await; + seed_follow(&c, &alice, &alice).await; + + let resp = c + .get(format!("{base}/api/notifications")) + .query(&[("did", alice.as_str())]) + .send() + .await + .unwrap(); + let body: Value = resp.json().await.unwrap(); + assert_eq!( + body["notifications"].as_array().unwrap().len(), + 0, + "self-interactions must not notify: {body:?}" + ); +} + +/// Cursor pagination over the notification list: pages must be +/// disjoint, exactly `limit` long while more remain, and the cursor +/// must go null at the end. +#[tokio::test] +async fn notifications_paginate_with_cursor() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let alice = did_for_test("popular"); + let post_uri = seed_post(&c, &alice, "the post everyone likes").await; + + // 12 distinct likers → 12 notifications. (Distinct DIDs matter: + // the dedupe index is per (recipient, author, kind, subject).) + for i in 0..12 { + let liker = did_for_test(&format!("fan{i}")); + seed_like(&c, &liker, &post_uri).await; + } + + let page = |cursor: Option| { + let c = c.clone(); + let alice = alice.clone(); + let base = base.clone(); + async move { + let mut req = c + .get(format!("{base}/api/notifications")) + .query(&[("did", alice.as_str()), ("limit", "5")]); + if let Some(cur) = cursor { + req = req.query(&[("cursor", cur.as_str())]); + } + let resp = req.send().await.unwrap(); + assert_eq!(resp.status().as_u16(), 200); + resp.json::().await.unwrap() + } + }; + + let p1 = page(None).await; + assert_eq!(p1["notifications"].as_array().unwrap().len(), 5); + let c1 = p1["cursor"].as_str().expect("page1 cursor").to_string(); + + let p2 = page(Some(c1)).await; + assert_eq!(p2["notifications"].as_array().unwrap().len(), 5); + let c2 = p2["cursor"].as_str().expect("page2 cursor").to_string(); + + let p3 = page(Some(c2)).await; + assert_eq!(p3["notifications"].as_array().unwrap().len(), 2); + assert!( + p3["cursor"].is_null(), + "cursor must be null on the last page: {p3:?}" + ); + + // No id may appear on two pages. + let ids = |p: &Value| -> Vec { + p["notifications"] + .as_array() + .unwrap() + .iter() + .map(|n| n["id"].as_i64().unwrap()) + .collect() + }; + let mut all: Vec = ids(&p1); + all.extend(ids(&p2)); + all.extend(ids(&p3)); + let unique: std::collections::HashSet = all.iter().copied().collect(); + assert_eq!(unique.len(), all.len(), "pages overlap: {all:?}"); + assert_eq!(all.len(), 12); + + // A mangled cursor is a 400, not a silent restart at page 1. + let resp = c + .get(format!("{base}/api/notifications")) + .query(&[("did", alice.as_str()), ("cursor", "!!!garbage!!!")]) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 400); +} + +/// `seenAt` is a watermark: only rows indexed at or before it flip to +/// read. Accepted in both camelCase and snake_case. +#[tokio::test] +async fn notifications_seen_respects_watermark() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let alice = did_for_test("watermark"); + let post_uri = seed_post(&c, &alice, "watermark subject").await; + + let first = did_for_test("early"); + seed_like(&c, &first, &post_uri).await; + + // Read back the first notification's indexed_at — that's the + // watermark a client would echo after rendering page 1. + let body: Value = c + .get(format!("{base}/api/notifications")) + .query(&[("did", alice.as_str())]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let watermark = body["notifications"][0]["indexed_at"] + .as_str() + .unwrap() + .to_string(); + + // A second interaction lands *after* the watermark. + tokio::time::sleep(Duration::from_millis(20)).await; + let second = did_for_test("late"); + seed_like(&c, &second, &post_uri).await; + + let resp = c + .post(format!("{base}/api/notifications/seen")) + .json(&json!({ "did": alice, "seenAt": watermark })) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await.unwrap(); + assert_eq!( + body["updated"], json!(1), + "only the row at/before the watermark may flip to read" + ); + + // The later one is still unread. + let body: Value = c + .get(format!("{base}/api/notifications/count")) + .query(&[("did", alice.as_str())]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(body["count"], json!(1)); + + // snake_case spelling must work identically. + let resp = c + .post(format!("{base}/api/notifications/seen")) + .json(&json!({ "did": alice, "seen_at": null })) + .send() + .await + .unwrap(); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body["updated"], json!(1)); +} + +// -- follower / following lists --------------------------------------------- + +#[tokio::test] +async fn followers_and_following_lists() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let hub = did_for_test("hub"); + seed_post(&c, &hub, "hub post").await; + + // Three people follow the hub; the hub follows one of them back. + let mut fans = Vec::new(); + for i in 0..3 { + let fan = did_for_test(&format!("fan{i}")); + seed_post(&c, &fan, "fan post").await; + seed_follow(&c, &fan, &hub).await; + fans.push(fan); + } + seed_follow(&c, &hub, &fans[0]).await; + + // Followers. + let body: Value = c + .get(format!("{base}/api/followers")) + .query(&[("did", hub.as_str()), ("limit", "50")]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let got: std::collections::HashSet = body["profiles"] + .as_array() + .expect("profiles array") + .iter() + .map(|p| p["did"].as_str().unwrap().to_string()) + .collect(); + for fan in &fans { + assert!(got.contains(fan), "follower {fan} missing from {body:?}"); + } + assert!( + !got.contains(&hub), + "the hub must not appear in its own follower list" + ); + + // Following — exactly one edge. + let body: Value = c + .get(format!("{base}/api/following")) + .query(&[("did", hub.as_str()), ("limit", "50")]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + let following: Vec = body["profiles"] + .as_array() + .unwrap() + .iter() + .map(|p| p["did"].as_str().unwrap().to_string()) + .collect(); + assert_eq!(following, vec![fans[0].clone()]); + + // Never an empty or '@'-prefixed handle. + for p in body["profiles"].as_array().unwrap() { + let h = p["handle"].as_str().expect("handle is a string"); + assert!(!h.is_empty()); + assert!(!h.starts_with('@')); + } + + // Pagination: limit=1 must page through all three followers + // without repeats. + let mut seen: Vec = Vec::new(); + let mut cursor: Option = None; + for _ in 0..5 { + let mut req = c + .get(format!("{base}/api/followers")) + .query(&[("did", hub.as_str()), ("limit", "1")]); + if let Some(cur) = cursor.as_deref() { + req = req.query(&[("cursor", cur)]); + } + let body: Value = req.send().await.unwrap().json().await.unwrap(); + for p in body["profiles"].as_array().unwrap() { + seen.push(p["did"].as_str().unwrap().to_string()); + } + match body["cursor"].as_str() { + Some(c) => cursor = Some(c.to_string()), + None => break, + } + } + let unique: std::collections::HashSet<&String> = seen.iter().collect(); + assert_eq!(unique.len(), seen.len(), "paged followers repeat: {seen:?}"); + assert_eq!(seen.len(), 3, "paging lost a follower: {seen:?}"); + + // `did` is mandatory. + for path in ["/api/followers", "/api/following"] { + let resp = c + .get(format!("{base}{path}")) + .query(&[("did", "")]) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 400); + } +} + +// -- thread ----------------------------------------------------------------- + +/// `/api/thread` returns the ancestor chain above a post and its +/// direct replies, in both the query-param and the path spelling. +#[tokio::test] +async fn thread_returns_parents_and_replies() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let a = did_for_test("root"); + let b = did_for_test("mid"); + let d = did_for_test("leaf"); + + let root = seed_post(&c, &a, "root post").await; + let mid = seed_reply(&c, &b, &root, &root, "middle reply").await; + let leaf = seed_reply(&c, &d, &mid, &root, "leaf reply").await; + + for url in [ + format!("{base}/api/thread?uri={}", urlencoding(&mid)), + format!("{base}/api/thread/{mid}"), + ] { + let resp = c.get(&url).send().await.unwrap(); + assert_eq!(resp.status().as_u16(), 200, "GET {url}"); + let body: Value = resp.json().await.unwrap(); + + assert_eq!(body["post"]["uri"], json!(mid), "GET {url}"); + // One ancestor, and it's the root. + let parents = body["parents"].as_array().unwrap(); + assert_eq!(parents.len(), 1, "GET {url}: {parents:?}"); + assert_eq!(parents[0]["uri"], json!(root)); + assert_eq!(body["root"]["uri"], json!(root)); + // One direct reply: the leaf. + let replies = body["replies"].as_array().unwrap(); + assert_eq!(replies.len(), 1, "GET {url}: {replies:?}"); + assert_eq!(replies[0]["uri"], json!(leaf)); + assert_eq!(body["like_count"], json!(0)); + } + + // The root's thread has no parents and one reply (the middle). + let body: Value = c + .get(format!("{base}/api/thread/{root}")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(body["parents"].as_array().unwrap().len(), 0); + assert!(body["root"].is_null(), "a top-level post has no root ref"); + let replies = body["replies"].as_array().unwrap(); + assert_eq!(replies.len(), 1); + assert_eq!(replies[0]["uri"], json!(mid)); + + // An unknown URI is a 200 with a null post, not a 404 — the UI + // renders "not in index" from one field check. + let body: Value = c + .get(format!( + "{base}/api/thread/at://did:plc:nobody/app.twi.post/{}", + rkey() + )) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(body["post"].is_null()); + assert_eq!(body["parents"].as_array().unwrap().len(), 0); + assert_eq!(body["replies"].as_array().unwrap().len(), 0); + + // A missing `uri` is a 400. + let resp = c + .get(format!("{base}/api/thread")) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 400); +} + +/// `/api/post/{uri}` must keep its historical shape after the thread +/// refactor — the Tauri client reads `thread.parent` / `thread.root` +/// and has no `parents` / `replies` fields. +#[tokio::test] +async fn post_by_uri_stays_backwards_compatible() { + let base = appview_url(); + let Some((c, _pool)) = ready().await else { + return; + }; + let a = did_for_test("compat_a"); + let b = did_for_test("compat_b"); + let root = seed_post(&c, &a, "compat root").await; + let mid = seed_reply(&c, &b, &root, &root, "compat reply").await; + seed_like(&c, &a, &mid).await; + + let body: Value = c + .get(format!("{base}/api/post/{mid}")) + .query(&[("viewer_did", a.as_str())]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + + assert_eq!(body["post"]["uri"], json!(mid)); + // The legacy nested shape: immediate parent + root, both hydrated. + assert_eq!(body["thread"]["parent"]["uri"], json!(root)); + assert_eq!(body["thread"]["root"]["uri"], json!(root)); + assert_eq!(body["like_count"], json!(1)); + assert_eq!(body["repost_count"], json!(0)); + assert_eq!(body["viewer_liked"], json!(true)); + assert_eq!(body["viewer_reposted"], json!(false)); + // The endpoint must NOT have grown the thread route's fields. + assert!(body.get("replies").is_none(), "unexpected `replies`: {body:?}"); + assert!(body.get("parents").is_none(), "unexpected `parents`: {body:?}"); + + // And the two endpoints must agree about the parent / root. + let thread: Value = c + .get(format!("{base}/api/thread/{mid}")) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!( + thread["parents"].as_array().unwrap().last().unwrap()["uri"], + body["thread"]["parent"]["uri"], + "/api/thread and /api/post disagree about the parent" + ); + assert_eq!(thread["root"]["uri"], body["thread"]["root"]["uri"]); +} + +/// Minimal percent-encoder for the `?uri=` form. Only the characters +/// an `at://did:plc:…/app.twi.post/` URI can contain that a query +/// string would otherwise eat. +fn urlencoding(s: &str) -> String { + let mut out = String::with_capacity(s.len() * 2); + for ch in s.chars() { + match ch { + 'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' => out.push(ch), + other => { + let mut buf = [0u8; 4]; + for b in other.encode_utf8(&mut buf).as_bytes() { + out.push_str(&format!("%{b:02X}")); + } + } + } + } + out +} diff --git a/migrations/appview/0008_notifications.sql b/migrations/appview/0008_notifications.sql new file mode 100644 index 0000000..c2092fc --- /dev/null +++ b/migrations/appview/0008_notifications.sql @@ -0,0 +1,113 @@ +-- AppView database schema 0008: notifications + follow-list pagination. +-- +-- Why +-- +-- Phase 5 shipped the read API (timeline / profile / search / post) but +-- the client had no way to learn that *someone else* interacted with +-- the user: a like, a repost, a follow or a reply never produced a +-- durable record. The Tauri client's tray/notification path (Phase 7) +-- therefore had nothing to poll. This migration adds the table the +-- Jetstream indexer writes into and the read API serves from. +-- +-- Table shape +-- id BIGSERIAL — monotonic tiebreaker for the keyset +-- cursor. The API's opaque cursor is +-- `(indexed_at, id)`, mirroring the +-- `(indexed_at, uri)` pair used by /api/timeline/home, +-- so the same `routes::cursor` encoder is reused. +-- recipient_did the user who should SEE the notification (the post +-- author for like/repost/reply, the followed user for +-- follow). +-- author_did the user who CAUSED it (the liker / reposter / +-- follower / replier). +-- kind 'like' | 'repost' | 'follow' | 'reply'. Enforced by +-- a CHECK rather than a Postgres ENUM so adding a +-- variant later is an ALTER ... DROP/ADD CONSTRAINT +-- instead of a type migration that locks every +-- dependent object. +-- subject_uri the post the notification is *about*. NULL for +-- 'follow' (there is no post). For 'like'/'repost' +-- it's the liked/reposted post (the recipient's own +-- post); for 'reply' it's the REPLY itself, because +-- the interesting text to show in the notification +-- list is what the replier wrote, not what the +-- recipient already knows they posted. +-- created_at the interaction's own `createdAt` from the AT record. +-- indexed_at when WE saw it. This is what the list is ordered by, +-- for the same reason the timeline orders by +-- `posts.indexed_at`: a client-supplied `created_at` +-- can be arbitrarily far in the past or future and +-- would break keyset pagination. +-- read_at NULL = unread. Set in bulk by +-- `POST /api/notifications/seen`. +-- +-- Deliberately NO `CHECK (recipient_did <> author_did)` +-- ------------------------------------------------------ +-- Self-interactions must not produce notifications, and the indexer +-- enforces that in two places (a pure `should_notify` guard in Rust +-- plus a `WHERE $1 <> $2` in the INSERT ... SELECT). A CHECK would +-- turn a future slip into a constraint violation that aborts the +-- surrounding like/repost transaction — i.e. it would lose the *like* +-- because of a notification bug. Filtering is strictly better than +-- failing here. +-- +-- Idempotency +-- ----------- +-- `notifications_dedupe_idx` is the unique constraint the indexer's +-- `ON CONFLICT ... DO NOTHING` infers. `subject_uri` is nullable and +-- NULLs never collide in a plain unique index, so the index is on +-- `COALESCE(subject_uri, '')` — that makes the two follow rows +-- (subject_uri IS NULL) for the same (recipient, author) pair collide +-- as intended. +-- +-- Consequence worth knowing: unlike-then-relike (or unfollow-then- +-- refollow) does NOT produce a second notification, because the tuple +-- is identical. That is the desired behaviour — it makes notification +-- spam via toggling impossible — but it does mean a notification is +-- "once per (recipient, author, kind, subject)" for all time. + +CREATE TABLE notifications ( + id BIGSERIAL PRIMARY KEY, + recipient_did TEXT NOT NULL, + author_did TEXT NOT NULL, + kind TEXT NOT NULL + CHECK (kind IN ('like', 'repost', 'follow', 'reply')), + subject_uri TEXT, + created_at TIMESTAMPTZ NOT NULL, + indexed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + read_at TIMESTAMPTZ +); + +-- Primary read path: `WHERE recipient_did = $1 ORDER BY indexed_at DESC, +-- id DESC`. The trailing `id DESC` makes the index cover the keyset +-- predicate `(indexed_at, id) < ($2, $3)` end-to-end, so a page fetch +-- never sorts. +CREATE INDEX notifications_recipient_indexed_at_idx + ON notifications (recipient_did, indexed_at DESC, id DESC); + +-- Dedupe / ON CONFLICT target. See the "Idempotency" note above. +CREATE UNIQUE INDEX notifications_dedupe_idx + ON notifications (recipient_did, author_did, kind, COALESCE(subject_uri, '')); + +-- `GET /api/notifications/count` is a hot poll from the client's tray +-- badge, so the unread slice gets its own partial index. It stays tiny +-- because rows leave it as soon as they're marked seen. +CREATE INDEX notifications_unread_idx + ON notifications (recipient_did) + WHERE read_at IS NULL; + +-- ===================================================== +-- follows: pagination indexes for the follower/following lists +-- ===================================================== +-- +-- `GET /api/followers` and `GET /api/following` page with the same +-- keyset scheme as the timeline: `(indexed_at, )`. +-- The pre-existing indexes cover only the equality half (the PK covers +-- `follower_did`, `follows_subject_idx` covers `subject_did`), which +-- leaves Postgres sorting the whole follower set on every page. These +-- two make both directions index-ordered. +CREATE INDEX IF NOT EXISTS follows_subject_indexed_at_idx + ON follows (subject_did, indexed_at DESC, follower_did DESC); + +CREATE INDEX IF NOT EXISTS follows_follower_indexed_at_idx + ON follows (follower_did, indexed_at DESC, subject_did DESC);