tomdeboneandClaude Opus 5
c4ca218d97
Bisher erfuhr ein Nutzer nie, dass jemand anderes mit ihm interagiert hat:
Like, Repost, Follow und Reply hinterließen keine Spur, an der der Client
hätte pollen können. Der Tray-/Notification-Pfad im Desktop-Client (Phase 7)
hing damit in der Luft.
Migration 0008:
* notifications(recipient, author, kind, subject_uri, created_at,
indexed_at, read_at) mit Keyset-Index (recipient, indexed_at DESC, id DESC)
und Partial-Index auf ungelesene Zeilen für den Badge-Poll.
* Dedupe-Unique-Index über COALESCE(subject_uri, '') — plain NULLs
kollidieren nicht, sonst gäbe es pro Follow beliebig viele Zeilen.
Folge: Unlike-Relike erzeugt keine zweite Notification, Toggle-Spam ist
damit ausgeschlossen.
* Bewusst kein CHECK (recipient <> author): ein Ausrutscher dort würde die
umgebende Like-Transaktion abbrechen, also das Like wegen eines
Notification-Bugs verlieren. Gefiltert wird in Rust und im INSERT.
Indexer: record_notification() hängt an upsert_like/-repost (in derselben
Transaktion wie die Counter) sowie upsert_follow/-post. Selbst-Interaktionen
sind still. Empfänger muss uns bekannt sein (profiles- oder posts-Zeile),
sonst würden wir für den gesamten öffentlichen Firehose Zeilen anlegen —
als ein INSERT ... SELECT ... WHERE EXISTS, also ohne TOCTOU-Fenster.
Reply-Notifications tragen die URI der *Antwort* als subject_uri, weil die
Liste den Text zeigt, den der Empfänger noch nicht kennt.
Endpoints: GET /api/notifications, /api/notifications/count,
POST /api/notifications/seen (seenAt als Wasserzeichen),
GET /api/followers, /api/following, GET /api/thread (beide Schreibweisen).
Cursor-Codec, Limit-Clamping und Fehlerform sind die der bestehenden
Endpoints.
/api/post/*uri bleibt wire-kompatibel und teilt sich jetzt
load_thread_context() mit /api/thread — mit max_parents = 1, weil es nur
den direkten Parent serialisiert; die volle Ahnenkette wären bis zu 20
sequenzielle Queries für Zeilen, die danach verworfen werden.
Nebenbei ein Darstellungsfehler: der synthetische Platzhalter-Handle für
Actors ohne bekannten Handle trug ein führendes '@', während jeder Consumer
selbst '@{handle}' rendert — im Feed kam '@@did:plc:abcd…' heraus. Der
Platzhalter ist jetzt durchgängig sigil-frei.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX
2093 lines
72 KiB
Rust
2093 lines
72 KiB
Rust
//! DB upsert functions for the AppView indexer.
|
|
//!
|
|
//! These are pure: given a [`PgPool`] and a parsed Jetstream (or PDS-pushed)
|
|
//! `CommitOp`, each function performs one idempotent write against the
|
|
//! `posts` / `likes` / `reposts` / `follows` tables.
|
|
//!
|
|
//! Every write is idempotent: UPSERT for create/update and DELETE for
|
|
//! removals, so re-applying the same event multiple times (e.g. on a Jetstream
|
|
//! reconnect that replays recent messages) is safe.
|
|
|
|
use anyhow::{anyhow, Result};
|
|
use at_firehose::{CommitOp, JetstreamEvent};
|
|
use serde_json::Value;
|
|
use sqlx::{
|
|
encode::IsNull,
|
|
error::BoxDynError,
|
|
PgPool, Postgres, Type,
|
|
};
|
|
#[cfg(test)]
|
|
use std::time::Duration;
|
|
#[cfg(test)]
|
|
use tokio::time::timeout;
|
|
|
|
/// Collections whose `create`/`delete` events we persist to the `posts` table.
|
|
#[allow(dead_code)]
|
|
pub const POST_COLLECTIONS: &[&str] = &["app.twi.post", "app.bsky.feed.post"];
|
|
|
|
/// Build a `at://did/collection/rkey` URI from the event's `did` + the
|
|
/// commit op's `rkey` (falling back to the trailing component of `path`).
|
|
#[allow(dead_code)]
|
|
pub fn build_uri(did: &str, collection: &str, op: &CommitOp) -> Option<String> {
|
|
let rkey = op
|
|
.rkey
|
|
.clone()
|
|
.or_else(|| {
|
|
op.path
|
|
.as_deref()
|
|
.and_then(|p| p.rsplit('/').next().map(str::to_string))
|
|
})?;
|
|
Some(format!("at://{did}/{collection}/{rkey}"))
|
|
}
|
|
|
|
/// Best-effort extraction of the collection NSID from a Jetstream commit
|
|
/// payload. Jetstream single-op events have it at `commit.collection`;
|
|
/// firehose-style batched events have it per-op under `collection`.
|
|
pub fn extract_collection(commit: &Value) -> Option<String> {
|
|
if let Some(c) = commit.get("collection").and_then(|v| v.as_str()) {
|
|
return Some(c.to_string());
|
|
}
|
|
if let Some(ops) = commit.get("ops").and_then(|v| v.as_array()) {
|
|
if let Some(c) = ops
|
|
.first()
|
|
.and_then(|o| o.get("collection"))
|
|
.and_then(|v| v.as_str())
|
|
{
|
|
return Some(c.to_string());
|
|
}
|
|
}
|
|
None
|
|
}
|
|
|
|
/// Convert the wire-format `commit` value into a flat list of
|
|
/// [`CommitOp`]s. Accepts both the Jetstream single-op shape (where the
|
|
/// commit object itself carries `operation`/`rkey`/`record`/...) and the
|
|
/// firehose batched shape (where `commit.ops` is an array of ops).
|
|
///
|
|
/// The wire field name is `operation`, which we map to `CommitOp::action`
|
|
/// because that's the name the at-firehose crate settled on.
|
|
pub fn commit_op_from_jetstream_value(commit: &Value) -> Vec<CommitOp> {
|
|
if let Some(ops) = commit.get("ops").and_then(|v| v.as_array()) {
|
|
return ops.iter().filter_map(parse_single_op).collect();
|
|
}
|
|
parse_single_op(commit).into_iter().collect()
|
|
}
|
|
|
|
fn parse_single_op(op: &Value) -> Option<CommitOp> {
|
|
let action = op
|
|
.get("operation")
|
|
.or_else(|| op.get("action"))
|
|
.and_then(|v| v.as_str())?
|
|
.to_string();
|
|
let rkey = op.get("rkey").and_then(|v| v.as_str()).map(String::from);
|
|
let path = op.get("path").and_then(|v| v.as_str()).map(String::from);
|
|
let cid = op.get("cid").and_then(|v| v.as_str()).map(String::from);
|
|
let record = op.get("record").cloned();
|
|
Some(CommitOp {
|
|
action,
|
|
rkey,
|
|
path,
|
|
cid,
|
|
record,
|
|
})
|
|
}
|
|
|
|
/// Parse an ISO-8601 timestamp string from a record's `createdAt`. We treat
|
|
/// any parse failure as "now" rather than dropping the record — the event
|
|
/// itself is still useful; we just lose the client's timestamp.
|
|
pub fn parse_created_at(s: Option<&str>) -> chrono::DateTime<chrono::Utc> {
|
|
match s.and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) {
|
|
Some(dt) => dt.with_timezone(&chrono::Utc),
|
|
None => chrono::Utc::now(),
|
|
}
|
|
}
|
|
|
|
// -- cursor ----------------------------------------------------------------
|
|
|
|
const CURSOR_ROW_ID: i32 = 1;
|
|
|
|
/// Read the persisted jetstream cursor (microseconds since epoch). Returns 0
|
|
/// if the row is missing for any reason (e.g. fresh DB before the migration
|
|
/// inserted it).
|
|
#[allow(dead_code)]
|
|
pub async fn cursor_get(db: &PgPool) -> Result<i64> {
|
|
let row: Option<(i64,)> = sqlx::query_as(
|
|
"SELECT cursor FROM jetstream_cursor WHERE id = $1",
|
|
)
|
|
.bind(CURSOR_ROW_ID)
|
|
.fetch_optional(db)
|
|
.await?;
|
|
Ok(row.map(|(c,)| c).unwrap_or(0))
|
|
}
|
|
|
|
/// Advance the persisted cursor, but never to a lower value.
|
|
pub async fn cursor_advance(db: &PgPool, new_value: i64) -> Result<()> {
|
|
sqlx::query(
|
|
r#"UPDATE jetstream_cursor
|
|
SET cursor = GREATEST(cursor, $1), updated_at = now()
|
|
WHERE id = $2"#,
|
|
)
|
|
.bind(new_value)
|
|
.bind(CURSOR_ROW_ID)
|
|
.execute(db)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// -- posts -----------------------------------------------------------------
|
|
|
|
/// A nullable JSONB column wrapper. We store `embed` as the full
|
|
/// AT-Protocol embed object verbatim — the alternative (normalising into
|
|
/// a separate `embeds` table) would cost an extra round trip per post
|
|
/// and require schema migrations for every new embed variant.
|
|
///
|
|
/// `Option<Value>` already implements `Encode<Json>` for `serde_json::Value`,
|
|
/// so we only need the Decode/Type pair to handle SQL NULL → `None`.
|
|
#[derive(Debug, Clone, Default)]
|
|
pub struct EmbedColumn(pub Option<Value>);
|
|
|
|
impl From<Option<Value>> for EmbedColumn {
|
|
fn from(v: Option<Value>) -> Self {
|
|
EmbedColumn(v)
|
|
}
|
|
}
|
|
|
|
impl From<Value> for EmbedColumn {
|
|
fn from(v: Value) -> Self {
|
|
EmbedColumn(Some(v))
|
|
}
|
|
}
|
|
|
|
impl<'r> sqlx::Decode<'r, Postgres> for EmbedColumn {
|
|
fn decode(
|
|
value: <Postgres as sqlx::Database>::ValueRef<'r>,
|
|
) -> Result<Self, sqlx::error::BoxDynError> {
|
|
let v: Option<Value> = <Option<Value> as sqlx::Decode<Postgres>>::decode(value)?;
|
|
Ok(EmbedColumn(v))
|
|
}
|
|
}
|
|
|
|
impl<'q> sqlx::Encode<'q, Postgres> for EmbedColumn {
|
|
fn encode_by_ref(
|
|
&self,
|
|
buf: &mut <Postgres as sqlx::Database>::ArgumentBuffer<'q>,
|
|
) -> Result<IsNull, BoxDynError> {
|
|
match &self.0 {
|
|
Some(v) => <&Value as sqlx::Encode<Postgres>>::encode_by_ref(&v, buf),
|
|
None => Ok(IsNull::Yes),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Type<Postgres> for EmbedColumn {
|
|
fn type_info() -> <Postgres as sqlx::Database>::TypeInfo {
|
|
<Value as Type<Postgres>>::type_info()
|
|
}
|
|
fn compatible(ty: &<Postgres as sqlx::Database>::TypeInfo) -> bool {
|
|
<Value as Type<Postgres>>::compatible(ty)
|
|
}
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
pub struct PostRow {
|
|
pub uri: String,
|
|
pub did: String,
|
|
pub handle: String,
|
|
pub rkey: String,
|
|
pub collection: String,
|
|
pub text: String,
|
|
pub cid: String,
|
|
pub parent_uri: Option<String>,
|
|
pub root_uri: Option<String>,
|
|
pub embed: Option<Value>,
|
|
pub langs: Option<Vec<String>>,
|
|
pub created_at: chrono::DateTime<chrono::Utc>,
|
|
/// Resolved author avatar CID from the `profiles` cache. Populated
|
|
/// at `upsert_post` time so the PostCard can render an avatar
|
|
/// without a per-row PDS round trip. NULL for users whose profile
|
|
/// hasn't been pushed yet.
|
|
pub avatar_cid: Option<String>,
|
|
}
|
|
|
|
impl PostRow {
|
|
/// Extract a PostRow from the Jetstream / PDS event.
|
|
/// `handle` is not present in Jetstream events and resolving it would
|
|
/// require a PLC directory lookup; we store an empty placeholder and
|
|
/// expect a future handle-sync job to backfill it.
|
|
///
|
|
/// `embed` is captured verbatim from the record — the UI renders it
|
|
/// by sniffing for `$type` (`app.bsky.embed.images` / `.external` /
|
|
/// `.record`). Keeping it as raw JSON means we don't have to mirror
|
|
/// every embed variant in Rust.
|
|
///
|
|
/// `pds_handle` is the optional handle forwarded by the PDS through
|
|
/// the `/internal/ingest-commit` payload. Local-PDS users have
|
|
/// `did:key:` DIDs that no PLC directory can resolve, so the PDS is
|
|
/// the only authoritative source for their handle. Pass `Some(handle)`
|
|
/// when you have it; pass `None` (e.g. Jetstream path) and the
|
|
/// `upsert_post` COALESCE guard ensures the empty value won't
|
|
/// clobber a backfilled handle from `firehose::handle_identity`.
|
|
pub fn from_record(
|
|
did: &str,
|
|
rkey: &str,
|
|
collection: &str,
|
|
cid: &str,
|
|
record: &Value,
|
|
pds_handle: Option<&str>,
|
|
) -> Self {
|
|
let text = record
|
|
.get("text")
|
|
.and_then(|v| v.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let created_at = parse_created_at(
|
|
record.get("createdAt").and_then(|v| v.as_str()),
|
|
);
|
|
let reply = record.get("reply");
|
|
let parent_uri = reply
|
|
.and_then(|r| r.get("parent"))
|
|
.and_then(|p| p.get("uri"))
|
|
.and_then(|u| u.as_str())
|
|
.map(str::to_string);
|
|
let root_uri = reply
|
|
.and_then(|r| r.get("root"))
|
|
.and_then(|p| p.get("uri"))
|
|
.and_then(|u| u.as_str())
|
|
.map(str::to_string);
|
|
let embed = record.get("embed").cloned().filter(|v| !v.is_null());
|
|
let langs = record
|
|
.get("langs")
|
|
.and_then(|v| v.as_array())
|
|
.map(|arr| {
|
|
arr.iter()
|
|
.filter_map(|v| v.as_str().map(str::to_string))
|
|
.collect::<Vec<_>>()
|
|
});
|
|
let uri = format!("at://{did}/{collection}/{rkey}");
|
|
let handle = pds_handle
|
|
.map(|h| h.trim().to_string())
|
|
.filter(|h| !h.is_empty())
|
|
.unwrap_or_default();
|
|
Self {
|
|
uri,
|
|
did: did.to_string(),
|
|
handle,
|
|
rkey: rkey.to_string(),
|
|
collection: collection.to_string(),
|
|
text,
|
|
cid: cid.to_string(),
|
|
parent_uri,
|
|
root_uri,
|
|
embed,
|
|
langs,
|
|
created_at,
|
|
// Avatar CID is populated later by the upsert path via a
|
|
// `SELECT avatar_cid FROM profiles WHERE did = $1` lookup,
|
|
// so a freshly indexed post starts at None. (The lookup
|
|
// happens in `upsert_post_with_avatar` below.)
|
|
avatar_cid: None,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Insert or update a post row keyed by URI. Idempotent.
|
|
///
|
|
/// `row.avatar_cid` is filled in-place with the current profile-avatar
|
|
/// CID for the row's author (from the `profiles` cache, NULL if the
|
|
/// profile hasn't been pushed yet). The ON CONFLICT clause uses
|
|
/// `COALESCE(EXCLUDED, posts)` so a backfill on a re-indexed post
|
|
/// won't overwrite an avatar we already had.
|
|
///
|
|
/// IMPORTANT: `indexed_at` is NOT touched on conflict. We deliberately
|
|
/// preserve the original insert time so the `(indexed_at, uri)` keyset
|
|
/// pagination order is stable across Jetstream replays / PDS re-syncs.
|
|
pub async fn upsert_post(db: &PgPool, row: &mut PostRow) -> Result<()> {
|
|
if row.avatar_cid.is_none() {
|
|
// Look up the latest avatar CID for this author from the
|
|
// profiles cache (populated by the PDS push path on profile
|
|
// updates). NULL if the profile hasn't been ingested yet —
|
|
// the post will display as the initial-letter avatar until
|
|
// the user uploads one.
|
|
row.avatar_cid = sqlx::query_scalar::<_, Option<String>>(
|
|
"SELECT avatar_cid FROM profiles WHERE did = $1",
|
|
)
|
|
.bind(&row.did)
|
|
.fetch_optional(db)
|
|
.await?
|
|
.flatten();
|
|
}
|
|
sqlx::query(
|
|
r#"INSERT INTO posts
|
|
(uri, did, handle, rkey, collection, text, cid,
|
|
parent_uri, root_uri, embed, langs, created_at,
|
|
avatar_cid)
|
|
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
|
|
ON CONFLICT (uri) DO UPDATE SET
|
|
text = EXCLUDED.text,
|
|
cid = EXCLUDED.cid,
|
|
handle = COALESCE(NULLIF(EXCLUDED.handle,''), posts.handle),
|
|
parent_uri = EXCLUDED.parent_uri,
|
|
root_uri = EXCLUDED.root_uri,
|
|
embed = EXCLUDED.embed,
|
|
langs = EXCLUDED.langs,
|
|
created_at = EXCLUDED.created_at,
|
|
avatar_cid = COALESCE(EXCLUDED.avatar_cid, posts.avatar_cid)"#,
|
|
)
|
|
.bind(&row.uri)
|
|
.bind(&row.did)
|
|
.bind(&row.handle)
|
|
.bind(&row.rkey)
|
|
.bind(&row.collection)
|
|
.bind(&row.text)
|
|
.bind(&row.cid)
|
|
.bind(&row.parent_uri)
|
|
.bind(&row.root_uri)
|
|
.bind(EmbedColumn(row.embed.clone()))
|
|
.bind(&row.langs)
|
|
.bind(row.created_at)
|
|
.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(())
|
|
}
|
|
|
|
/// Delete a post by URI. Idempotent (returns Ok even if the row is gone).
|
|
pub async fn delete_post(db: &PgPool, uri: &str) -> Result<()> {
|
|
sqlx::query("DELETE FROM posts WHERE uri = $1")
|
|
.bind(uri)
|
|
.execute(db)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// -- likes -----------------------------------------------------------------
|
|
|
|
/// Insert or ignore a like row keyed by URI. The partial unique index
|
|
/// on `(did, post_uri)` lets us reject double-likes at the DB level
|
|
/// instead of relying on caller discipline. We also maintain the
|
|
/// denormalized `posts.like_count` so the post endpoint doesn't have
|
|
/// to `COUNT(*)` over the entire `likes` table on every read.
|
|
pub async fn upsert_like(
|
|
db: &PgPool,
|
|
did: &str,
|
|
rkey: &str,
|
|
cid: Option<&str>,
|
|
record: Option<&Value>,
|
|
) -> Result<()> {
|
|
let uri = format!("at://{did}/app.bsky.feed.like/{rkey}");
|
|
let _cid = cid.unwrap_or("");
|
|
let post_uri = record
|
|
.and_then(|r| r.get("subject"))
|
|
.and_then(|s| s.get("uri"))
|
|
.and_then(|u| u.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let post_cid = record
|
|
.and_then(|r| r.get("subject"))
|
|
.and_then(|s| s.get("cid"))
|
|
.and_then(|u| u.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let created_at = parse_created_at(
|
|
record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()),
|
|
);
|
|
let mut tx = db.begin().await?;
|
|
let inserted = sqlx::query(
|
|
r#"INSERT INTO likes (uri, did, post_uri, post_cid, created_at)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (did, post_uri) DO NOTHING
|
|
RETURNING uri"#,
|
|
)
|
|
.bind(&uri)
|
|
.bind(did)
|
|
.bind(&post_uri)
|
|
.bind(&post_cid)
|
|
.bind(created_at)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
if inserted.is_some() && !post_uri.is_empty() {
|
|
// Increment the denormalized counter so the post endpoint can
|
|
// serve like_count from `posts` without scanning `likes`.
|
|
sqlx::query(
|
|
"UPDATE posts SET like_count = like_count + 1 WHERE uri = $1",
|
|
)
|
|
.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(())
|
|
}
|
|
|
|
pub async fn delete_like(db: &PgPool, did: &str, rkey: &str) -> Result<()> {
|
|
let uri = format!("at://{did}/app.bsky.feed.like/{rkey}");
|
|
let mut tx = db.begin().await?;
|
|
let row: Option<(String,)> = sqlx::query_as(
|
|
"DELETE FROM likes WHERE uri = $1 RETURNING post_uri",
|
|
)
|
|
.bind(&uri)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
if let Some((post_uri,)) = row {
|
|
if !post_uri.is_empty() {
|
|
sqlx::query(
|
|
"UPDATE posts SET like_count = GREATEST(like_count - 1, 0) WHERE uri = $1",
|
|
)
|
|
.bind(&post_uri)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
}
|
|
tx.commit().await?;
|
|
Ok(())
|
|
}
|
|
|
|
// -- reposts ---------------------------------------------------------------
|
|
|
|
pub async fn upsert_repost(
|
|
db: &PgPool,
|
|
did: &str,
|
|
rkey: &str,
|
|
cid: Option<&str>,
|
|
record: Option<&Value>,
|
|
) -> Result<()> {
|
|
let uri = format!("at://{did}/app.bsky.feed.repost/{rkey}");
|
|
let _cid = cid.unwrap_or("");
|
|
let post_uri = record
|
|
.and_then(|r| r.get("subject"))
|
|
.and_then(|s| s.get("uri"))
|
|
.and_then(|u| u.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let post_cid = record
|
|
.and_then(|r| r.get("subject"))
|
|
.and_then(|s| s.get("cid"))
|
|
.and_then(|u| u.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let created_at = parse_created_at(
|
|
record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()),
|
|
);
|
|
let mut tx = db.begin().await?;
|
|
let inserted = sqlx::query(
|
|
r#"INSERT INTO reposts (uri, did, post_uri, post_cid, created_at)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (did, post_uri) DO NOTHING
|
|
RETURNING uri"#,
|
|
)
|
|
.bind(&uri)
|
|
.bind(did)
|
|
.bind(&post_uri)
|
|
.bind(&post_cid)
|
|
.bind(created_at)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
if inserted.is_some() && !post_uri.is_empty() {
|
|
sqlx::query(
|
|
"UPDATE posts SET repost_count = repost_count + 1 WHERE uri = $1",
|
|
)
|
|
.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(())
|
|
}
|
|
|
|
pub async fn delete_repost(db: &PgPool, did: &str, rkey: &str) -> Result<()> {
|
|
let uri = format!("at://{did}/app.bsky.feed.repost/{rkey}");
|
|
let mut tx = db.begin().await?;
|
|
let row: Option<(String,)> = sqlx::query_as(
|
|
"DELETE FROM reposts WHERE uri = $1 RETURNING post_uri",
|
|
)
|
|
.bind(&uri)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
if let Some((post_uri,)) = row {
|
|
if !post_uri.is_empty() {
|
|
sqlx::query(
|
|
"UPDATE posts SET repost_count = GREATEST(repost_count - 1, 0) WHERE uri = $1",
|
|
)
|
|
.bind(&post_uri)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
}
|
|
tx.commit().await?;
|
|
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<chrono::Utc>,
|
|
) -> Result<bool>
|
|
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<Option<String>>
|
|
where
|
|
E: sqlx::PgExecutor<'e>,
|
|
{
|
|
let did: Option<String> = sqlx::query_scalar("SELECT did FROM posts WHERE uri = $1")
|
|
.bind(uri)
|
|
.fetch_optional(exec)
|
|
.await?;
|
|
Ok(did)
|
|
}
|
|
|
|
// -- follows ---------------------------------------------------------------
|
|
|
|
pub async fn upsert_follow(
|
|
db: &PgPool,
|
|
follower_did: &str,
|
|
subject_did: &str,
|
|
record: Option<&Value>,
|
|
) -> Result<()> {
|
|
let created_at = parse_created_at(
|
|
record.and_then(|r| r.get("createdAt")).and_then(|v| v.as_str()),
|
|
);
|
|
sqlx::query(
|
|
r#"INSERT INTO follows (follower_did, subject_did, created_at)
|
|
VALUES ($1, $2, $3)
|
|
ON CONFLICT (follower_did, subject_did) DO UPDATE SET
|
|
created_at = EXCLUDED.created_at,
|
|
indexed_at = now()"#,
|
|
)
|
|
.bind(follower_did)
|
|
.bind(subject_did)
|
|
.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(())
|
|
}
|
|
|
|
pub async fn delete_follow(
|
|
db: &PgPool,
|
|
follower_did: &str,
|
|
subject_did: &str,
|
|
) -> Result<()> {
|
|
sqlx::query(
|
|
"DELETE FROM follows WHERE follower_did = $1 AND subject_did = $2",
|
|
)
|
|
.bind(follower_did)
|
|
.bind(subject_did)
|
|
.execute(db)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Extract the subject DID from a follow record (`{ "subject": "did:..."}`).
|
|
pub fn follow_subject_did(record: Option<&Value>) -> Option<String> {
|
|
record?
|
|
.get("subject")?
|
|
.as_str()
|
|
.map(|s| s.to_string())
|
|
}
|
|
|
|
// -- test harness ----------------------------------------------------------
|
|
|
|
#[cfg(test)]
|
|
async fn try_test_db() -> Option<PgPool> {
|
|
let url = std::env::var("DATABASE_URL_APPVIEW").ok()?;
|
|
match timeout(Duration::from_secs(2), sqlx::PgPool::connect(&url)).await {
|
|
Ok(Ok(pool)) => match sqlx::migrate!("../../migrations/appview")
|
|
.run(&pool)
|
|
.await
|
|
{
|
|
Ok(()) => Some(pool),
|
|
Err(_) => None,
|
|
},
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
/// Route a `JetstreamEvent::kind == "commit"` event to the right upsert /
|
|
/// delete function. Returns `Ok(false)` if the event was recognized but
|
|
/// not actionable (e.g. unsupported collection), `Ok(true)` if it was
|
|
/// applied, `Err(_)` if DB write failed.
|
|
pub async fn apply_commit(
|
|
db: &PgPool,
|
|
ev: &JetstreamEvent,
|
|
) -> Result<bool> {
|
|
let commit = match &ev.commit {
|
|
Some(c) => c,
|
|
None => return Ok(false),
|
|
};
|
|
let collection = match extract_collection(commit) {
|
|
Some(c) => c,
|
|
None => return Ok(false),
|
|
};
|
|
let ops = commit_op_from_jetstream_value(commit);
|
|
if ops.is_empty() {
|
|
return Ok(false);
|
|
}
|
|
let mut applied = false;
|
|
for op in &ops {
|
|
match collection.as_str() {
|
|
"app.twi.post" | "app.bsky.feed.post" => {
|
|
if op.action == "create" {
|
|
let rkey = op
|
|
.rkey
|
|
.clone()
|
|
.or_else(|| {
|
|
op.path
|
|
.as_deref()
|
|
.and_then(|p| p.rsplit('/').next().map(str::to_string))
|
|
})
|
|
.ok_or_else(|| {
|
|
anyhow!("create op missing rkey for post")
|
|
})?;
|
|
let cid = op.cid.clone().unwrap_or_default();
|
|
let record = op.record.clone().unwrap_or(Value::Null);
|
|
// Jetstream `commit` events don't carry the
|
|
// handle — leave it empty so the upsert
|
|
// COALESCE guard preserves the row's existing
|
|
// (or backfilled-from-identity) handle.
|
|
let mut row = PostRow::from_record(
|
|
&ev.did,
|
|
&rkey,
|
|
&collection,
|
|
&cid,
|
|
&record,
|
|
None,
|
|
);
|
|
upsert_post(db, &mut row).await?;
|
|
applied = true;
|
|
} else if op.action == "delete" {
|
|
let rkey = op
|
|
.rkey
|
|
.clone()
|
|
.or_else(|| {
|
|
op.path
|
|
.as_deref()
|
|
.and_then(|p| p.rsplit('/').next().map(str::to_string))
|
|
})
|
|
.ok_or_else(|| {
|
|
anyhow!("delete op missing rkey for post")
|
|
})?;
|
|
let uri = format!(
|
|
"at://{}/{}/{}",
|
|
ev.did, collection, rkey
|
|
);
|
|
delete_post(db, &uri).await?;
|
|
applied = true;
|
|
}
|
|
}
|
|
"app.bsky.feed.like" => {
|
|
let rkey = op
|
|
.rkey
|
|
.clone()
|
|
.or_else(|| {
|
|
op.path
|
|
.as_deref()
|
|
.and_then(|p| p.rsplit('/').next().map(str::to_string))
|
|
});
|
|
let rkey = match rkey {
|
|
Some(r) => r,
|
|
None => {
|
|
tracing::warn!("like op missing rkey; skipping");
|
|
continue;
|
|
}
|
|
};
|
|
if op.action == "create" {
|
|
upsert_like(
|
|
db,
|
|
&ev.did,
|
|
&rkey,
|
|
op.cid.as_deref(),
|
|
op.record.as_ref(),
|
|
)
|
|
.await?;
|
|
} else if op.action == "delete" {
|
|
delete_like(db, &ev.did, &rkey).await?;
|
|
}
|
|
applied = true;
|
|
}
|
|
"app.bsky.feed.repost" => {
|
|
let rkey = op
|
|
.rkey
|
|
.clone()
|
|
.or_else(|| {
|
|
op.path
|
|
.as_deref()
|
|
.and_then(|p| p.rsplit('/').next().map(str::to_string))
|
|
});
|
|
let rkey = match rkey {
|
|
Some(r) => r,
|
|
None => {
|
|
tracing::warn!("repost op missing rkey; skipping");
|
|
continue;
|
|
}
|
|
};
|
|
if op.action == "create" {
|
|
upsert_repost(
|
|
db,
|
|
&ev.did,
|
|
&rkey,
|
|
op.cid.as_deref(),
|
|
op.record.as_ref(),
|
|
)
|
|
.await?;
|
|
} else if op.action == "delete" {
|
|
delete_repost(db, &ev.did, &rkey).await?;
|
|
}
|
|
applied = true;
|
|
}
|
|
"app.bsky.graph.follow" => {
|
|
let subject_did = match op.action.as_str() {
|
|
"create" => match follow_subject_did(op.record.as_ref()) {
|
|
Some(s) => s,
|
|
None => {
|
|
tracing::warn!(
|
|
"follow create missing subject; skipping"
|
|
);
|
|
continue;
|
|
}
|
|
},
|
|
"delete" => {
|
|
// Jetstream delete on follows carries no record
|
|
// value, so we can't know which subject was
|
|
// unfollowed. The PDS-driven internal ingest path
|
|
// handles this — it knows the subject from its
|
|
// own snapshot.
|
|
tracing::warn!(
|
|
"follow delete via Jetstream lacks subject; \
|
|
route through /internal/ingest-commit instead"
|
|
);
|
|
continue;
|
|
}
|
|
_ => continue,
|
|
};
|
|
if op.action == "create" {
|
|
upsert_follow(
|
|
db,
|
|
&ev.did,
|
|
&subject_did,
|
|
op.record.as_ref(),
|
|
)
|
|
.await?;
|
|
} else if op.action == "delete" {
|
|
delete_follow(db, &ev.did, &subject_did).await?;
|
|
}
|
|
applied = true;
|
|
}
|
|
"app.bsky.actor.profile" => {
|
|
// Jetstream carries profile records as plain
|
|
// commit ops (no separate collection). We treat
|
|
// any rkey — usually `self`, but spec allows
|
|
// rkey-rotation — as the user's authoritative
|
|
// profile and upsert into the `profiles` cache.
|
|
//
|
|
// The Jetstream `commit` envelope doesn't carry
|
|
// the handle; we look it up from the `posts`
|
|
// table (backfilled there by the `identity`
|
|
// event stream). Empty is fine — the next
|
|
// handle_sync pass will populate it.
|
|
if op.action == "create" {
|
|
let record = match op.record.clone() {
|
|
Some(r) if !r.is_null() => r,
|
|
_ => continue,
|
|
};
|
|
let handle: String = sqlx::query_scalar(
|
|
"SELECT handle FROM posts \
|
|
WHERE did = $1 AND handle <> '' \
|
|
ORDER BY indexed_at DESC LIMIT 1",
|
|
)
|
|
.bind(&ev.did)
|
|
.fetch_optional(db)
|
|
.await?
|
|
.unwrap_or_default();
|
|
upsert_profile(db, &ev.did, &handle, &record).await?;
|
|
applied = true;
|
|
} else if op.action == "delete" {
|
|
sqlx::query("DELETE FROM profiles WHERE did = $1")
|
|
.bind(&ev.did)
|
|
.execute(db)
|
|
.await?;
|
|
applied = true;
|
|
}
|
|
}
|
|
_ => {
|
|
// Unrecognised collection — ignore (may happen when Jetstream
|
|
// sends something we didn't subscribe to).
|
|
}
|
|
}
|
|
}
|
|
Ok(applied)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use serde_json::json;
|
|
|
|
#[test]
|
|
fn commit_op_from_jetstream_value_single_create() {
|
|
let commit = json!({
|
|
"operation": "create",
|
|
"collection": "app.bsky.feed.post",
|
|
"rkey": "3k2abc",
|
|
"cid": "bafyreicid",
|
|
"record": {"text": "hi", "createdAt": "2026-07-01T12:00:00Z"}
|
|
});
|
|
let ops = commit_op_from_jetstream_value(&commit);
|
|
assert_eq!(ops.len(), 1);
|
|
assert_eq!(ops[0].action, "create");
|
|
assert_eq!(ops[0].rkey.as_deref(), Some("3k2abc"));
|
|
assert_eq!(ops[0].cid.as_deref(), Some("bafyreicid"));
|
|
assert_eq!(
|
|
ops[0].record.as_ref().unwrap()["text"].as_str(),
|
|
Some("hi")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn commit_op_from_jetstream_value_batched_ops() {
|
|
let commit = json!({
|
|
"ops": [
|
|
{"action": "create", "collection": "app.bsky.feed.like", "rkey": "1",
|
|
"cid": "c1", "record": {"subject": {"uri": "at://x/y/z", "cid": "cz"},
|
|
"createdAt": "2026-07-01T12:00:00Z"}},
|
|
{"action": "delete", "collection": "app.bsky.feed.like", "rkey": "0"}
|
|
]
|
|
});
|
|
let ops = commit_op_from_jetstream_value(&commit);
|
|
assert_eq!(ops.len(), 2);
|
|
assert_eq!(ops[0].action, "create");
|
|
assert_eq!(ops[0].rkey.as_deref(), Some("1"));
|
|
assert_eq!(ops[1].action, "delete");
|
|
assert_eq!(ops[1].rkey.as_deref(), Some("0"));
|
|
}
|
|
|
|
#[test]
|
|
fn commit_op_from_jetstream_value_missing_action_returns_empty() {
|
|
let commit = json!({ "collection": "app.bsky.feed.post", "rkey": "x" });
|
|
let ops = commit_op_from_jetstream_value(&commit);
|
|
assert!(ops.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn extract_collection_top_level_and_batched() {
|
|
let top = json!({"operation":"create","collection":"app.twi.post","rkey":"r"});
|
|
assert_eq!(extract_collection(&top).as_deref(), Some("app.twi.post"));
|
|
|
|
let batched = json!({"ops": [{"collection":"app.bsky.feed.like"}]});
|
|
assert_eq!(
|
|
extract_collection(&batched).as_deref(),
|
|
Some("app.bsky.feed.like")
|
|
);
|
|
|
|
let none = json!({});
|
|
assert!(extract_collection(&none).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn build_uri_uses_rkey_or_path() {
|
|
let op_rkey = CommitOp {
|
|
action: "create".into(),
|
|
rkey: Some("rk".into()),
|
|
path: None,
|
|
cid: None,
|
|
record: None,
|
|
};
|
|
assert_eq!(
|
|
build_uri("did:plc:a", "app.twi.post", &op_rkey).as_deref(),
|
|
Some("at://did:plc:a/app.twi.post/rk")
|
|
);
|
|
|
|
let op_path = CommitOp {
|
|
action: "delete".into(),
|
|
rkey: None,
|
|
path: Some("app.twi.post/last".into()),
|
|
cid: None,
|
|
record: None,
|
|
};
|
|
assert_eq!(
|
|
build_uri("did:plc:a", "app.twi.post", &op_path).as_deref(),
|
|
Some("at://did:plc:a/app.twi.post/last")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn follow_subject_did_extracts() {
|
|
let rec = json!({ "subject": "did:plc:b", "createdAt": "2026-01-01T00:00:00Z" });
|
|
assert_eq!(
|
|
follow_subject_did(Some(&rec)).as_deref(),
|
|
Some("did:plc:b")
|
|
);
|
|
assert!(follow_subject_did(None).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn from_record_captures_embed() {
|
|
let rec = json!({
|
|
"text": "look at this",
|
|
"createdAt": "2026-07-01T12:00:00Z",
|
|
"embed": {
|
|
"$type": "app.bsky.embed.images",
|
|
"images": [
|
|
{"alt": "a cat", "image": {"$type": "blob", "ref": {"$link": "bafy"}}}
|
|
]
|
|
}
|
|
});
|
|
let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
|
|
let embed = row.embed.expect("embed must be captured");
|
|
assert_eq!(embed["$type"], "app.bsky.embed.images");
|
|
assert_eq!(embed["images"][0]["alt"], "a cat");
|
|
}
|
|
|
|
#[test]
|
|
fn from_record_captures_external_embed() {
|
|
let rec = json!({
|
|
"text": "see link",
|
|
"createdAt": "2026-07-01T12:00:00Z",
|
|
"embed": {
|
|
"$type": "app.bsky.embed.external",
|
|
"external": {
|
|
"uri": "https://example.com",
|
|
"title": "Example",
|
|
"description": "An example"
|
|
}
|
|
}
|
|
});
|
|
let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
|
|
let embed = row.embed.expect("embed must be captured");
|
|
assert_eq!(embed["$type"], "app.bsky.embed.external");
|
|
assert_eq!(embed["external"]["uri"], "https://example.com");
|
|
}
|
|
|
|
#[test]
|
|
fn from_record_omits_embed_when_missing() {
|
|
let rec = json!({
|
|
"text": "no embed here",
|
|
"createdAt": "2026-07-01T12:00:00Z"
|
|
});
|
|
let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
|
|
assert!(row.embed.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn from_record_captures_reply_uris() {
|
|
let rec = json!({
|
|
"text": "a reply",
|
|
"createdAt": "2026-07-01T12:00:00Z",
|
|
"reply": {
|
|
"parent": {"uri": "at://did:plc:b/app.twi.post/p", "cid": "cp"},
|
|
"root": {"uri": "at://did:plc:b/app.twi.post/r", "cid": "cr"}
|
|
}
|
|
});
|
|
let row = PostRow::from_record("did:plc:abc", "rkey", "app.bsky.feed.post", "cid", &rec, None);
|
|
assert_eq!(row.parent_uri.as_deref(), Some("at://did:plc:b/app.twi.post/p"));
|
|
assert_eq!(row.root_uri.as_deref(), Some("at://did:plc:b/app.twi.post/r"));
|
|
}
|
|
|
|
fn fake_event(collection: &str, action: &str, rkey: &str) -> JetstreamEvent {
|
|
let commit = if action == "create" {
|
|
json!({
|
|
"operation": action,
|
|
"collection": collection,
|
|
"rkey": rkey,
|
|
"cid": "bafyreicid",
|
|
"record": {
|
|
"text": "hi there",
|
|
"createdAt": "2026-07-01T12:00:00Z"
|
|
}
|
|
})
|
|
} else {
|
|
json!({
|
|
"operation": action,
|
|
"collection": collection,
|
|
"rkey": rkey
|
|
})
|
|
};
|
|
JetstreamEvent {
|
|
did: "did:plc:test".into(),
|
|
time_us: 1_700_000_000_000_000,
|
|
kind: "commit".into(),
|
|
commit: Some(commit),
|
|
identity: None,
|
|
account: None,
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn upsert_and_delete_post_round_trip() {
|
|
let Some(db) = try_test_db().await else {
|
|
eprintln!("appview DB unavailable; skipping");
|
|
return;
|
|
};
|
|
// Clean any previous row in this test's slot.
|
|
let _ = sqlx::query("DELETE FROM posts WHERE did = 'did:plc:test'")
|
|
.execute(&db)
|
|
.await
|
|
.unwrap();
|
|
|
|
let ev = fake_event("app.twi.post", "create", "abc123");
|
|
let applied = apply_commit(&db, &ev).await.unwrap();
|
|
assert!(applied);
|
|
let (uri, text): (String, String) = sqlx::query_as(
|
|
"SELECT uri, text FROM posts WHERE uri = $1",
|
|
)
|
|
.bind("at://did:plc:test/app.twi.post/abc123")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(uri, "at://did:plc:test/app.twi.post/abc123");
|
|
assert_eq!(text, "hi there");
|
|
|
|
// Verify that the (currently null) embed column is readable —
|
|
// we want to fail loud if the column is missing from the
|
|
// schema rather than silently falling back to text-only posts.
|
|
let embed_json: Option<serde_json::Value> = sqlx::query_scalar(
|
|
"SELECT embed FROM posts WHERE uri = $1",
|
|
)
|
|
.bind("at://did:plc:test/app.twi.post/abc123")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
embed_json.is_none(),
|
|
"embed should be null for plain-text posts, got: {embed_json:?}"
|
|
);
|
|
|
|
// Apply again — must be idempotent (no error, still one row).
|
|
let _ = apply_commit(&db, &ev).await.unwrap();
|
|
let count: i64 = sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM posts WHERE uri = $1",
|
|
)
|
|
.bind("at://did:plc:test/app.twi.post/abc123")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(count, 1);
|
|
|
|
// Delete it.
|
|
let del = fake_event("app.twi.post", "delete", "abc123");
|
|
apply_commit(&db, &del).await.unwrap();
|
|
let remaining: i64 = sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM posts WHERE uri = $1",
|
|
)
|
|
.bind("at://did:plc:test/app.twi.post/abc123")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(remaining, 0);
|
|
}
|
|
|
|
/// Seed a post with an `app.bsky.embed.images` embed and confirm
|
|
/// the JSONB column round-trips the embed object back as JSON.
|
|
#[tokio::test]
|
|
async fn upsert_post_stores_embed_jsonb() {
|
|
let Some(db) = try_test_db().await else {
|
|
eprintln!("appview DB unavailable; skipping");
|
|
return;
|
|
};
|
|
// Clean any previous row in this test's slot.
|
|
let _ = sqlx::query("DELETE FROM posts WHERE did = 'did:plc:embed'")
|
|
.execute(&db)
|
|
.await
|
|
.unwrap();
|
|
|
|
let record = json!({
|
|
"text": "with an image",
|
|
"createdAt": "2026-07-01T12:00:00Z",
|
|
"embed": {
|
|
"$type": "app.bsky.embed.images",
|
|
"images": [
|
|
{"alt": "first alt", "image": {"$type": "blob", "ref": {"$link": "bafy1"}}},
|
|
{"alt": "second alt", "image": {"$type": "blob", "ref": {"$link": "bafy2"}}}
|
|
]
|
|
}
|
|
});
|
|
let mut row = PostRow::from_record(
|
|
"did:plc:embed",
|
|
"embedkey",
|
|
"app.twi.post",
|
|
"cid-embed",
|
|
&record,
|
|
None,
|
|
);
|
|
upsert_post(&db, &mut row).await.unwrap();
|
|
|
|
let embed: serde_json::Value = sqlx::query_scalar(
|
|
"SELECT embed FROM posts WHERE uri = $1",
|
|
)
|
|
.bind("at://did:plc:embed/app.twi.post/embedkey")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(embed["$type"], "app.bsky.embed.images");
|
|
let imgs = embed["images"].as_array().expect("images array");
|
|
assert_eq!(imgs.len(), 2);
|
|
assert_eq!(imgs[0]["alt"], "first alt");
|
|
assert_eq!(imgs[1]["alt"], "second alt");
|
|
|
|
// Cleanup.
|
|
let _ = sqlx::query("DELETE FROM posts WHERE did = 'did:plc:embed'")
|
|
.execute(&db)
|
|
.await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn upsert_and_delete_follow_round_trip() {
|
|
let Some(db) = try_test_db().await else {
|
|
eprintln!("appview DB unavailable; skipping");
|
|
return;
|
|
};
|
|
// Wipe test data so the test is order-independent.
|
|
let _ = sqlx::query(
|
|
"DELETE FROM follows WHERE follower_did = $1 AND subject_did = $2",
|
|
)
|
|
.bind("did:plc:test")
|
|
.bind("did:plc:b")
|
|
.execute(&db)
|
|
.await
|
|
.unwrap();
|
|
|
|
let ev = JetstreamEvent {
|
|
did: "did:plc:test".into(),
|
|
time_us: 1_700_000_000_000_000,
|
|
kind: "commit".into(),
|
|
commit: Some(json!({
|
|
"operation": "create",
|
|
"collection": "app.bsky.graph.follow",
|
|
"rkey": "frk1",
|
|
"cid": "bafyfollow",
|
|
"record": {
|
|
"subject": "did:plc:b",
|
|
"createdAt": "2026-01-01T00:00:00Z"
|
|
}
|
|
})),
|
|
identity: None,
|
|
account: None,
|
|
};
|
|
apply_commit(&db, &ev).await.unwrap();
|
|
let (count,): (i64,) = sqlx::query_as(
|
|
"SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2",
|
|
)
|
|
.bind("did:plc:test")
|
|
.bind("did:plc:b")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(count, 1);
|
|
|
|
// Idempotent re-apply.
|
|
apply_commit(&db, &ev).await.unwrap();
|
|
let (count,): (i64,) = sqlx::query_as(
|
|
"SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2",
|
|
)
|
|
.bind("did:plc:test")
|
|
.bind("did:plc:b")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(count, 1);
|
|
|
|
// Delete via the internal API (not via Jetstream — Jetstream
|
|
// delete on follows doesn't carry the subject).
|
|
delete_follow(&db, "did:plc:test", "did:plc:b").await.unwrap();
|
|
let (count,): (i64,) = sqlx::query_as(
|
|
"SELECT COUNT(*) FROM follows WHERE follower_did = $1 AND subject_did = $2",
|
|
)
|
|
.bind("did:plc:test")
|
|
.bind("did:plc:b")
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(count, 0);
|
|
}
|
|
}
|
|
|
|
#[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<String>) = 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<String> = 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<String> = 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
|
|
/// needs the new value.
|
|
///
|
|
/// **Race-safety**: this only writes if the row's current handle is
|
|
/// empty OR doesn't match, so concurrent PDS pushes (which carry
|
|
/// the same handle) and concurrent identity replays don't fight.
|
|
/// The `WHERE handle IS DISTINCT FROM $1` makes the update a
|
|
/// no-op when the value is already correct, which Postgres treats
|
|
/// cheaply.
|
|
///
|
|
/// Returns the number of rows updated.
|
|
pub async fn backfill_handle(
|
|
db: &PgPool,
|
|
did: &str,
|
|
new_handle: &str,
|
|
) -> Result<u64> {
|
|
let res = sqlx::query(
|
|
"UPDATE posts SET handle = $1 WHERE did = $2 AND handle IS DISTINCT FROM $1",
|
|
)
|
|
.bind(new_handle)
|
|
.bind(did)
|
|
.execute(db)
|
|
.await?;
|
|
Ok(res.rows_affected())
|
|
}
|
|
|
|
/// Upsert a `profiles` row for `did`. The caller has just ingested
|
|
/// the profile record body (decoded CBOR), and the denormalised
|
|
/// counts are computed here (a single `SELECT COUNT(*)` over each
|
|
/// side-table — cheap with the existing PK indexes on `posts.did` and
|
|
/// `follows.{follower,subject}_did`).
|
|
pub async fn upsert_profile(
|
|
db: &PgPool,
|
|
did: &str,
|
|
handle: &str,
|
|
record: &Value,
|
|
) -> Result<()> {
|
|
let display_name = record
|
|
.get("displayName")
|
|
.and_then(|v| v.as_str())
|
|
.map(str::to_string);
|
|
let description = record
|
|
.get("description")
|
|
.and_then(|v| v.as_str())
|
|
.map(str::to_string);
|
|
let avatar_cid = blob_link_of(record, "avatar");
|
|
let banner_cid = blob_link_of(record, "banner");
|
|
|
|
// Denormalised counts. Cheap with the existing PKs.
|
|
let post_count: i64 = sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM posts WHERE did = $1 \
|
|
AND collection IN ('app.twi.post','app.bsky.feed.post')",
|
|
)
|
|
.bind(did)
|
|
.fetch_one(db)
|
|
.await?;
|
|
let follower_count: i64 = sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM follows WHERE subject_did = $1",
|
|
)
|
|
.bind(did)
|
|
.fetch_one(db)
|
|
.await?;
|
|
let following_count: i64 = sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM follows WHERE follower_did = $1",
|
|
)
|
|
.bind(did)
|
|
.fetch_one(db)
|
|
.await?;
|
|
|
|
sqlx::query(
|
|
r#"INSERT INTO profiles
|
|
(did, handle, display_name, description,
|
|
avatar_cid, banner_cid,
|
|
post_count, follower_count, following_count, indexed_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, now())
|
|
ON CONFLICT (did) DO UPDATE SET
|
|
handle = EXCLUDED.handle,
|
|
display_name = EXCLUDED.display_name,
|
|
description = EXCLUDED.description,
|
|
avatar_cid = EXCLUDED.avatar_cid,
|
|
banner_cid = EXCLUDED.banner_cid,
|
|
post_count = EXCLUDED.post_count,
|
|
follower_count= EXCLUDED.follower_count,
|
|
following_count= EXCLUDED.following_count,
|
|
indexed_at = now()"#,
|
|
)
|
|
.bind(did)
|
|
.bind(handle)
|
|
.bind(display_name)
|
|
.bind(description)
|
|
.bind(avatar_cid)
|
|
.bind(banner_cid)
|
|
.bind(post_count)
|
|
.bind(follower_count)
|
|
.bind(following_count)
|
|
.execute(db)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Pull a blob-ref `$link` out of a profile record's field.
|
|
/// Accepts both the modern shape
|
|
/// (`{ $type: "blob", ref: { $link: "..." } }`)
|
|
/// and the legacy shape (`{ $link: "..." }`) for robustness.
|
|
fn blob_link_of(record: &Value, field: &str) -> Option<String> {
|
|
let v = record.get(field)?;
|
|
// Try `ref.$link` first, then flat `$link`.
|
|
if let Some(link) = v.get("ref").and_then(|r| r.get("$link")).and_then(|s| s.as_str()) {
|
|
return Some(link.to_string());
|
|
}
|
|
v.get("$link").and_then(|s| s.as_str()).map(str::to_string)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod profile_tests {
|
|
use super::*;
|
|
use serde_json::json;
|
|
|
|
/// Open the appview DB used by integration tests, running
|
|
/// migrations first. Returns `None` when no DB is reachable so
|
|
/// the test can `eprintln!` and bail (no panic).
|
|
async fn try_test_db() -> Option<PgPool> {
|
|
let url = std::env::var("DATABASE_URL_APPVIEW").ok()?;
|
|
match timeout(
|
|
Duration::from_secs(2),
|
|
sqlx::PgPool::connect(&url),
|
|
)
|
|
.await
|
|
{
|
|
Ok(Ok(pool)) => match sqlx::migrate!("../../migrations/appview").run(&pool).await {
|
|
Ok(()) => Some(pool),
|
|
Err(_) => None,
|
|
},
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn blob_link_of_modern_shape() {
|
|
let rec = json!({
|
|
"avatar": {
|
|
"$type": "blob",
|
|
"ref": { "$link": "bafyavatar" },
|
|
"mimeType": "image/png",
|
|
"size": 1234
|
|
}
|
|
});
|
|
assert_eq!(blob_link_of(&rec, "avatar").as_deref(), Some("bafyavatar"));
|
|
}
|
|
|
|
#[test]
|
|
fn blob_link_of_legacy_flat_link() {
|
|
let rec = json!({ "banner": { "$link": "bafybanner" } });
|
|
assert_eq!(blob_link_of(&rec, "banner").as_deref(), Some("bafybanner"));
|
|
}
|
|
|
|
#[test]
|
|
fn blob_link_of_missing_field() {
|
|
let rec = json!({ "displayName": "x" });
|
|
assert_eq!(blob_link_of(&rec, "avatar"), None);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn upsert_profile_round_trip() {
|
|
let Some(db) = try_test_db().await else {
|
|
eprintln!("appview DB unavailable; skipping");
|
|
return;
|
|
};
|
|
// Use a unique DID per test run so we don't collide with the
|
|
// migration backfill (which seeded a row for every distinct
|
|
// DID in `posts`).
|
|
let did = format!(
|
|
"did:plc:profile_test_{}",
|
|
uuid::Uuid::new_v4().simple()
|
|
);
|
|
let handle = format!("user.{}.test", uuid::Uuid::new_v4().simple());
|
|
|
|
// Seed a couple of posts so post_count is non-zero.
|
|
for rkey in &["p1", "p2"] {
|
|
sqlx::query(
|
|
r#"INSERT INTO posts
|
|
(uri, did, handle, rkey, collection, text, cid,
|
|
parent_uri, root_uri, langs, created_at)
|
|
VALUES ($1,$2,$3,$4,'app.twi.post','seed','bafy',NULL,NULL,NULL, now())
|
|
ON CONFLICT (uri) DO NOTHING"#,
|
|
)
|
|
.bind(format!("at://{did}/app.twi.post/{rkey}"))
|
|
.bind(&did)
|
|
.bind(&handle)
|
|
.bind(rkey)
|
|
.execute(&db)
|
|
.await
|
|
.unwrap();
|
|
}
|
|
|
|
let record = json!({
|
|
"displayName": "Alice",
|
|
"description": "tester",
|
|
"avatar": { "$type": "blob", "ref": { "$link": "bafyavatar" } },
|
|
"banner": { "$type": "blob", "ref": { "$link": "bafybanner" } }
|
|
});
|
|
upsert_profile(&db, &did, &handle, &record).await.unwrap();
|
|
|
|
let row: (
|
|
String, // handle
|
|
Option<String>, // display_name
|
|
Option<String>, // description
|
|
Option<String>, // avatar_cid
|
|
Option<String>, // banner_cid
|
|
i64, // post_count
|
|
i64, // follower_count
|
|
i64, // following_count
|
|
) = sqlx::query_as(
|
|
"SELECT handle, display_name, description, avatar_cid, banner_cid, \
|
|
post_count, follower_count, following_count \
|
|
FROM profiles WHERE did = $1",
|
|
)
|
|
.bind(&did)
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(row.0, handle);
|
|
assert_eq!(row.1.as_deref(), Some("Alice"));
|
|
assert_eq!(row.2.as_deref(), Some("tester"));
|
|
assert_eq!(row.3.as_deref(), Some("bafyavatar"));
|
|
assert_eq!(row.4.as_deref(), Some("bafybanner"));
|
|
assert_eq!(row.5, 2, "post_count must reflect seeded posts");
|
|
|
|
// Update: change display name, drop banner — verify replace
|
|
// semantics (NULL fields overwrite, not coalesce).
|
|
let record2 = json!({ "displayName": "Alice 2" });
|
|
upsert_profile(&db, &did, &handle, &record2).await.unwrap();
|
|
let name: Option<String> = sqlx::query_scalar(
|
|
"SELECT display_name FROM profiles WHERE did = $1",
|
|
)
|
|
.bind(&did)
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(name.as_deref(), Some("Alice 2"));
|
|
|
|
// Cleanup.
|
|
let _ = sqlx::query("DELETE FROM profiles WHERE did = $1")
|
|
.bind(&did)
|
|
.execute(&db)
|
|
.await;
|
|
let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
|
|
.bind(&did)
|
|
.execute(&db)
|
|
.await;
|
|
}
|
|
|
|
/// `apply_commit` must dispatch an `app.bsky.actor.profile`
|
|
/// create op into the `profiles` cache (this is the path Jetstream
|
|
/// uses for third-party PDS authors).
|
|
#[tokio::test]
|
|
async fn apply_commit_indexes_profile_create() {
|
|
let Some(db) = try_test_db().await else {
|
|
eprintln!("appview DB unavailable; skipping");
|
|
return;
|
|
};
|
|
let did = format!(
|
|
"did:plc:profile_commit_{}",
|
|
uuid::Uuid::new_v4().simple()
|
|
);
|
|
let _ = sqlx::query("DELETE FROM profiles WHERE did = $1")
|
|
.bind(&did)
|
|
.execute(&db)
|
|
.await;
|
|
|
|
// Seed a post so the indexer can find a known handle.
|
|
sqlx::query(
|
|
r#"INSERT INTO posts
|
|
(uri, did, handle, rkey, collection, text, cid,
|
|
parent_uri, root_uri, langs, created_at)
|
|
VALUES ($1,$2,$3,'seed','app.twi.post','hi','bafy',NULL,NULL,NULL, now())
|
|
ON CONFLICT (uri) DO NOTHING"#,
|
|
)
|
|
.bind(format!("at://{did}/app.twi.post/seed"))
|
|
.bind(&did)
|
|
.bind("alice.test")
|
|
.execute(&db)
|
|
.await
|
|
.unwrap();
|
|
|
|
let commit = json!({
|
|
"operation": "create",
|
|
"collection": "app.bsky.actor.profile",
|
|
"rkey": "self",
|
|
"cid": "bafyprofilecid",
|
|
"record": {
|
|
"displayName": "Alice",
|
|
"description": "hello",
|
|
"avatar": { "$type": "blob", "ref": { "$link": "bafyavatar" } }
|
|
}
|
|
});
|
|
let ev = JetstreamEvent {
|
|
did: did.clone(),
|
|
time_us: 1_700_000_000_000_000,
|
|
kind: "commit".into(),
|
|
commit: Some(commit),
|
|
identity: None,
|
|
account: None,
|
|
};
|
|
let applied = apply_commit(&db, &ev).await.unwrap();
|
|
assert!(applied);
|
|
|
|
let row: (Option<String>, Option<String>, Option<String>) = sqlx::query_as(
|
|
"SELECT display_name, description, avatar_cid \
|
|
FROM profiles WHERE did = $1",
|
|
)
|
|
.bind(&did)
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(row.0.as_deref(), Some("Alice"));
|
|
assert_eq!(row.1.as_deref(), Some("hello"));
|
|
assert_eq!(row.2.as_deref(), Some("bafyavatar"));
|
|
|
|
// Delete op should wipe the row.
|
|
let del = json!({
|
|
"operation": "delete",
|
|
"collection": "app.bsky.actor.profile",
|
|
"rkey": "self"
|
|
});
|
|
let ev_del = JetstreamEvent {
|
|
did: did.clone(),
|
|
time_us: 1_700_000_001_000_000,
|
|
kind: "commit".into(),
|
|
commit: Some(del),
|
|
identity: None,
|
|
account: None,
|
|
};
|
|
apply_commit(&db, &ev_del).await.unwrap();
|
|
let remaining: i64 = sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM profiles WHERE did = $1",
|
|
)
|
|
.bind(&did)
|
|
.fetch_one(&db)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(remaining, 0, "delete op must remove profile row");
|
|
|
|
// Cleanup.
|
|
let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
|
|
.bind(&did)
|
|
.execute(&db)
|
|
.await;
|
|
}
|
|
}
|