maarcadetweet/crates/appview/tests/notifications_integration.rs

//! 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<sqlx::PgPool> {
    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<String> = 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<String>| {
        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::<Value>().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<i64> {
        p["notifications"]
            .as_array()
            .unwrap()
            .iter()
            .map(|n| n["id"].as_i64().unwrap())
            .collect()
    };
    let mut all: Vec<i64> = ids(&p1);
    all.extend(ids(&p2));
    all.extend(ids(&p3));
    let unique: std::collections::HashSet<i64> = 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<String> = 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<String> = 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<String> = Vec::new();
    let mut cursor: Option<String> = 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/<rkey>` 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
}