maarcadetweet/crates/appview/tests/api_integration.rs

//! Integration tests for the new read API routes
//! (`/api/timeline/home`, `/api/profile/...`, `/api/search`).
//!
//! These run against a live appview service + DB. Like
//! `appview_integration.rs`, they're fail-open: if the service or DB
//! isn't reachable, the test prints a notice and returns success
//! rather than panicking — so `cargo test --workspace` stays green in
//! environments where the appview hasn't been started.

mod common;

use common::TestAuth;
use serde_json::{json, Value};
use std::time::Duration;

const APPVIEW_URL: &str = "http://127.0.0.1:2584";

async fn client() -> reqwest::Client {
    reqwest::Client::builder()
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap()
}

async fn wait_for_appview_db() -> bool {
    let c = client().await;
    for _ in 0..20 {
        if let Ok(r) = c.get(format!("{APPVIEW_URL}/healthz")).send().await {
            if r.status().is_success() {
                return true;
            }
        }
        tokio::time::sleep(Duration::from_millis(250)).await;
    }
    false
}

async fn try_db_url() -> Option<String> {
    std::env::var("DATABASE_URL_APPVIEW").ok()
}

async fn db_reachable() -> bool {
    let Some(url) = try_db_url().await else {
        return false;
    };
    matches!(
        tokio::time::timeout(Duration::from_secs(2), sqlx::PgPool::connect(&url)).await,
        Ok(Ok(_))
    )
}

/// How this suite authenticates against `/api/timeline/home`, which is
/// no longer public. `None` means the AppView enforces auth but the
/// test process cannot mint a token (no `PDS_JWT_SECRET`), in which
/// case the test skips like it does for a missing service.
async fn auth_or_skip() -> Option<TestAuth> {
    TestAuth::probe(&client().await, APPVIEW_URL).await
}

/// `GET /api/timeline/home` as `did`, with the bearer token attached
/// when the instance requires one. The seeded DIDs are synthetic, so
/// the token is minted from the PDS's own signing secret — see
/// `tests/common/mod.rs`.
async fn get_timeline(
    c: &reqwest::Client,
    auth: &TestAuth,
    did: &str,
    extra: &[(&str, &str)],
) -> reqwest::Response {
    let mut params: Vec<(&str, &str)> = vec![("did", did)];
    params.extend_from_slice(extra);
    auth.apply(
        c.get(format!("{APPVIEW_URL}/api/timeline/home"))
            .query(&params),
        did,
    )
    .send()
    .await
    .unwrap()
}

async fn post_ingest(c: &reqwest::Client, body: Value) -> reqwest::Response {
    c.post(format!("{APPVIEW_URL}/internal/ingest-commit"))
        .json(&body)
        .send()
        .await
        .unwrap()
}

/// Insert a follow row directly via the DB. We bypass the ingest
/// endpoint because (a) 1500 individual HTTP round-trips are
/// prohibitively slow for the cap test, and (b) we don't need the
/// indexer to also re-resolve handles etc. for this test.
async fn insert_follow(
    pool: &sqlx::PgPool,
    follower_did: &str,
    subject_did: &str,
) {
    sqlx::query(
        r#"INSERT INTO follows (follower_did, subject_did, created_at)
           VALUES ($1, $2, now())
           ON CONFLICT (follower_did, subject_did) DO NOTHING"#,
    )
    .bind(follower_did)
    .bind(subject_did)
    .execute(pool)
    .await
    .unwrap();
}

/// Seed N posts for a DID with sequential rkeys and `created_at`
/// timestamps that strictly increase, so the cursor ordering test is
/// deterministic.
async fn seed_posts(c: &reqwest::Client, did: &str, texts: &[&str]) {
    for (i, text) in texts.iter().enumerate() {
        let r = post_ingest(
            c,
            json!({
                "did": did,
                "collection": "app.twi.post",
                "action": "create",
                "rkey": rkey(),
                "cid": "bafyreicid",
                "record": {
                    "text": text,
                    "createdAt": format!("2026-07-01T12:00:{:02}Z", i),
                }
            }),
        )
        .await;
        assert_eq!(r.status().as_u16(), 200);
    }
}

fn did_for_test(name: &str) -> String {
    // Random per-test DID so the tests can run in parallel without
    // colliding on URI primary keys.
    format!("did:plc:test_{}_{}", name, uuid::Uuid::new_v4().simple())
}

fn rkey() -> String {
    uuid::Uuid::new_v4().simple().to_string()
}

#[tokio::test]
async fn timeline_returns_seeded_posts() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let Some(auth) = auth_or_skip().await else { return };
    let did = did_for_test("tl");

    // Seed 3 posts with distinct rkeys.
    for i in 0..3 {
        let r = post_ingest(
            &c,
            json!({
                "did": did,
                "collection": "app.twi.post",
                "action": "create",
                "rkey": rkey(),
                "cid": "bafyreicid",
                "record": {
                    "text": format!("seeded post #{i}"),
                    // Strictly increasing so the ordering assertion
                    // below has something to actually check: the rows
                    // are inserted in this order, so `indexed_at` and
                    // `created_at` agree for *our* posts.
                    "createdAt": format!("2026-07-01T12:00:{:02}Z", i),
                }
            }),
        )
        .await;
        assert_eq!(r.status().as_u16(), 200);
    }

    // Give Jetstream / ingest a beat to settle — `indexed_at` defaults
    // to `now()` on insert, so we want a non-zero chance of seeing all
    // three rows in the first page.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // `limit=100`, not 10: this DID follows nobody, so the endpoint
    // serves the cold-start *global* recent feed. On any database with
    // more than a handful of recent posts (i.e. every developer
    // machine that has run this suite twice) the three rows we just
    // seeded fall outside a 10-row window and the assertions below
    // fail for reasons that have nothing to do with the timeline.
    let resp = get_timeline(&c, &auth, &did, &[("limit", "100")]).await;
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    let posts = body["posts"].as_array().expect("posts is array");
    assert!(posts.len() >= 3, "expected >=3 posts, got {}", posts.len());

    // All three seeded posts must be in the response and all share the
    // same DID.
    let our_uris: Vec<&str> = posts
        .as_slice()
        .iter()
        .filter_map(|p| {
            let uri = p["uri"].as_str()?;
            if uri.starts_with(&format!("at://{did}/")) {
                Some(uri)
            } else {
                None
            }
        })
        .collect();
    assert!(our_uris.len() >= 3, "missing our seeded posts in {posts:?}");

    // Posts come back `indexed_at DESC`, and we can't see `indexed_at`
    // in the response — but for the three rows WE just inserted,
    // insertion order == `indexed_at` order == `created_at` order, so
    // their `created_at` values must be non-increasing.
    //
    // Two fixes over the original assertion:
    //  - the wire field is `created_at`, not `createdAt` (the
    //    `PostRow` wire type in `routes/types.rs` carries no
    //    `rename_all = "camelCase"`), so `p["createdAt"]` was JSON
    //    `null` and `.as_str().unwrap()` panicked on the first row;
    //  - it ran over ALL posts, including other tests' fixtures from
    //    the global cold-start feed, whose `created_at` values have
    //    no relation to their `indexed_at` order. Restricting it to
    //    our own DID is the only version of this claim that holds.
    let mut prev: Option<String> = None;
    for p in posts.iter().filter(|p| p["did"] == json!(did)) {
        let ca = p["created_at"].as_str().unwrap().to_string();
        if let Some(prev) = prev.take() {
            assert!(
                ca <= prev,
                "created_at must be non-increasing: {ca} <= {prev}"
            );
        }
        prev = Some(ca);
    }
}

#[tokio::test]
async fn timeline_paginates_with_cursor() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let Some(auth) = auth_or_skip().await else { return };
    let did = did_for_test("pg");

    // Seed 50 posts.
    for _ in 0..50 {
        let r = post_ingest(
            &c,
            json!({
                "did": did,
                "collection": "app.twi.post",
                "action": "create",
                "rkey": rkey(),
                "cid": "bafyreicid",
                "record": {
                    "text": "page",
                    "createdAt": "2026-07-01T12:00:00Z",
                }
            }),
        )
        .await;
        assert_eq!(r.status().as_u16(), 200);
    }
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Page 1: limit=20.
    let resp = get_timeline(&c, &auth, &did, &[("limit", "20")]).await;
    let body: Value = resp.json().await.unwrap();
    let page1 = body["posts"].as_array().unwrap().clone();
    let cursor1 = body["cursor"].as_str().expect("page1 cursor");
    assert_eq!(page1.len(), 20, "page1 should be exactly 20");

    // Page 2: with cursor.
    let resp = get_timeline(&c, &auth, &did, &[("limit", "20"), ("cursor", cursor1)]).await;
    let body: Value = resp.json().await.unwrap();
    let page2 = body["posts"].as_array().unwrap().clone();
    assert_eq!(page2.len(), 20, "page2 should be exactly 20");

    // Pages must not overlap.
    let p1: std::collections::HashSet<&str> = page1
        .iter()
        .map(|p| p["uri"].as_str().unwrap())
        .collect();
    let p2: std::collections::HashSet<&str> = page2
        .iter()
        .map(|p| p["uri"].as_str().unwrap())
        .collect();
    assert!(p1.is_disjoint(&p2), "page1 and page2 overlap");

    // Page 3: tail — fewer than 20 expected, cursor=null.
    let cursor2 = body["cursor"].as_str().expect("page2 cursor");
    let resp = get_timeline(&c, &auth, &did, &[("limit", "20"), ("cursor", cursor2)]).await;
    let body: Value = resp.json().await.unwrap();
    let page3 = body["posts"].as_array().unwrap().clone();
    assert!(page3.len() <= 20, "page3 should be <= 20");
    // At least one of the three pages should be non-empty.
    assert!(!page1.is_empty() || !page2.is_empty() || !page3.is_empty());
}

#[tokio::test]
async fn profile_returns_posts_for_handle() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let did_a = did_for_test("alice");
    let did_b = did_for_test("bob");
    let handle_a = format!("alice.{}", uuid::Uuid::new_v4().simple());

    // Seed a post for A with handle populated, and a post for B with a
    // different handle. The internal-ingest path doesn't expose a
    // `handle` field, so we update the column directly.
    for did in [&did_a, &did_b] {
        let r = post_ingest(
            &c,
            json!({
                "did": did,
                "collection": "app.twi.post",
                "action": "create",
                "rkey": rkey(),
                "cid": "bafyreicid",
                "record": {
                    "text": "hi",
                    "createdAt": "2026-07-01T12:00:00Z",
                }
            }),
        )
        .await;
        assert_eq!(r.status().as_u16(), 200);
    }

    // Backfill handle for A only.
    let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
    let pool = sqlx::PgPool::connect(&url).await.unwrap();
    sqlx::query("UPDATE posts SET handle = $1 WHERE did = $2")
        .bind(&handle_a)
        .bind(&did_a)
        .execute(&pool)
        .await
        .unwrap();

    // Query by handle (no leading @).
    let resp = c
        .get(format!("{APPVIEW_URL}/api/profile/{handle_a}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["did"], json!(did_a));
    assert_eq!(body["handle"], json!(handle_a));
    let posts = body["posts"].as_array().unwrap();
    assert!(
        posts.iter().any(|p| p["did"] == json!(did_a)),
        "did_a post missing from profile"
    );
    assert!(
        !posts.iter().any(|p| p["did"] == json!(did_b)),
        "did_b post leaked into alice's profile"
    );

    // Same query, with leading @ — must also work.
    let resp = c
        .get(format!("{APPVIEW_URL}/api/profile/@{handle_a}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);

    // An unknown handle is NOT a 404. `resolve_profile` deliberately
    // synthesises an empty profile (empty `did`, zero counts, no
    // posts) so the UI renders an empty profile page instead of an
    // error toast — see the comment on the `let Some(target_did)`
    // else-branch in `routes.rs`. This assertion used to expect 404
    // and contradicted the endpoint it was testing.
    let unknown = format!("nobody_{}", uuid::Uuid::new_v4().simple());
    let resp = c
        .get(format!("{APPVIEW_URL}/api/profile/{unknown}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["did"], json!(""));
    assert_eq!(body["handle"], json!(unknown));
    assert!(body["posts"].as_array().unwrap().is_empty());
    assert_eq!(body["post_count"], json!(0));

    // /api/profile?did=... must work too.
    let resp = c
        .get(format!("{APPVIEW_URL}/api/profile"))
        .query(&[("did", did_a.as_str())])
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["did"], json!(did_a));
}

#[tokio::test]
async fn search_finds_text_match() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let did = did_for_test("srch");

    for text in ["hello world from test", "goodbye cruel world", "x"] {
        let r = post_ingest(
            &c,
            json!({
                "did": did,
                "collection": "app.twi.post",
                "action": "create",
                "rkey": rkey(),
                "cid": "bafyreicid",
                "record": {
                    "text": text,
                    "createdAt": "2026-07-01T12:00:00Z",
                }
            }),
        )
        .await;
        assert_eq!(r.status().as_u16(), 200);
    }
    tokio::time::sleep(Duration::from_millis(50)).await;

    let resp = c
        .get(format!("{APPVIEW_URL}/api/search"))
        .query(&[("q", "hello"), ("limit", "10")])
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["q"], json!("hello"));
    let posts = body["posts"].as_array().unwrap();
    assert!(!posts.is_empty(), "expected at least one match for 'hello'");
    for p in posts {
        let t = p["text"].as_str().unwrap();
        assert!(
            t.to_lowercase().contains("hello"),
            "post in result doesn't contain 'hello': {t}"
        );
    }

    // Empty q is a 400.
    let resp = c
        .get(format!("{APPVIEW_URL}/api/search"))
        .query(&[("q", "")])
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status().as_u16(), 400);
}

/// When alice follows bob and carol but NOT dave, her home timeline
/// must show bob's and carol's posts only — dave's post is invisible
/// to her even though it sits in the global recent feed.
#[tokio::test]
async fn timeline_filters_to_followees() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let Some(auth) = auth_or_skip().await else { return };
    let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
    let pool = sqlx::PgPool::connect(&url).await.unwrap();

    let alice = did_for_test("alice");
    let bob = did_for_test("bob");
    let carol = did_for_test("carol");
    let dave = did_for_test("dave");

    // Alice follows bob + carol (NOT dave).
    insert_follow(&pool, &alice, &bob).await;
    insert_follow(&pool, &alice, &carol).await;

    // Each person posts once.
    seed_posts(&c, &bob, &["bob says hi"]).await;
    seed_posts(&c, &carol, &["carol says hi"]).await;
    seed_posts(&c, &dave, &["dave says hi (alice should NOT see this)"]).await;

    tokio::time::sleep(Duration::from_millis(100)).await;

    let resp = get_timeline(&c, &auth, &alice, &[("limit", "100")]).await;
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    let posts = body["posts"].as_array().expect("posts is array");

    // Collect DIDs of returned posts.
    let returned_dids: std::collections::HashSet<String> = posts
        .iter()
        .map(|p| p["did"].as_str().unwrap().to_string())
        .collect();

    // Bob and carol MUST be present; dave MUST NOT be.
    assert!(
        returned_dids.contains(&bob),
        "bob's post missing from alice's timeline: {posts:?}"
    );
    assert!(
        returned_dids.contains(&carol),
        "carol's post missing from alice's timeline: {posts:?}"
    );
    assert!(
        !returned_dids.contains(&dave),
        "dave's post leaked into alice's timeline: {posts:?}"
    );

    // Stronger: walk every post and assert no `did` matches dave.
    for p in posts {
        let did = p["did"].as_str().unwrap();
        assert_ne!(did, dave, "dave leaked: {p:?}");
    }
}

/// Alice posts without following anyone. The endpoint must still
/// surface her own posts — via the global-recent "cold start"
/// fallback — so a brand-new account with no follows can see what
/// they've posted.
#[tokio::test]
async fn timeline_includes_own_posts() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let Some(auth) = auth_or_skip().await else { return };
    let alice = did_for_test("alone");

    // Alice posts without seeding any follows.
    seed_posts(&c, &alice, &["alice's first post", "alice's second post"]).await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    let resp = get_timeline(&c, &auth, &alice, &[("limit", "100")]).await;
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    let posts = body["posts"].as_array().expect("posts is array");

    // At least alice's two posts must be present. The global fallback
    // will include other recent posts from the DB too — we only
    // assert on alice's visibility here.
    let alice_uris: Vec<&str> = posts
        .iter()
        .filter_map(|p| {
            let uri = p["uri"].as_str()?;
            if uri.starts_with(&format!("at://{alice}/")) {
                Some(uri)
            } else {
                None
            }
        })
        .collect();
    assert!(
        alice_uris.len() >= 2,
        "alice's own posts missing from her own timeline: {posts:?}"
    );
}

/// Alice follows 1500 fake DIDs. The endpoint must NOT blow up — the
/// `target_dids` cap at MAX_FOLLOWED_DIDS=1000 kicks in, the user's
/// own DID is re-inserted, and the SQL `ANY($)` array stays bounded.
#[tokio::test]
async fn timeline_caps_followee_list() {
    if !wait_for_appview_db().await {
        eprintln!("appview not running, skipping");
        return;
    }
    if !db_reachable().await {
        eprintln!("appview DB unreachable, skipping");
        return;
    }
    let c = client().await;
    let Some(auth) = auth_or_skip().await else { return };
    let url = std::env::var("DATABASE_URL_APPVIEW").unwrap();
    let pool = sqlx::PgPool::connect(&url).await.unwrap();

    let alice = did_for_test("poweruser");
    // Seed 1500 follows (well over MAX_FOLLOWED_DIDS=1000).
    for _ in 0..1500 {
        let fake = format!(
            "did:plc:fake_{}_{}",
            uuid::Uuid::new_v4().simple(),
            uuid::Uuid::new_v4().simple()
        );
        insert_follow(&pool, &alice, &fake).await;
    }

    // Alice also posts — to confirm she sees her own DID even though
    // the cap trimmed the lexically-greatest 1000 followees.
    seed_posts(&c, &alice, &["poweruser post"]).await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    let resp = get_timeline(&c, &auth, &alice, &[("limit", "50")]).await;
    assert_eq!(resp.status().as_u16(), 200);
    let body: Value = resp.json().await.unwrap();
    let posts = body["posts"].as_array().expect("posts is array");

    // Alice's own post must be visible — the cap invariant guarantees
    // her own DID is preserved.
    let alice_visible = posts
        .iter()
        .any(|p| p["did"].as_str() == Some(alice.as_str()));
    assert!(
        alice_visible,
        "alice's own post not visible after cap: {posts:?}"
    );

    // The fake followee DIDs have no posts, so nothing else should
    // leak in. We only assert the endpoint didn't error and that
    // alice's own DID is honored.
}