maarcadetweet/crates/appview/tests/handle_sync_integration.rs

//! Integration tests for `HandleSyncWorker::run_once()`.
//!
//! These exercise the worker's SQL against a live appview DB. The
//! resolver is substituted for a stub so the tests do not depend on
//! `plc.directory` being reachable (and so we can deterministically
//! prove the "don't overwrite" race protection works).
//!
//! Like the sibling `api_integration.rs`, every test is fail-open: if
//! `DATABASE_URL_APPVIEW` is unset or the DB isn't reachable, the test
//! prints a notice and returns. This keeps `cargo test --workspace`
//! green in environments without the appview stack running.

use anyhow::Result;
use async_trait::async_trait;
use at_identity::DidHandleResolver;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::timeout;
use uuid::Uuid;

use appview::handle_sync::{HandleSyncWorker, SyncReport};

/// In-process test double for the PLC client. We never want these
/// tests to talk to the real PLC.
#[derive(Default)]
struct StubResolver {
    /// DID → resolved handle (or `None` for an unresolvable DID).
    /// `Some("")` is treated as "no result" by the worker.
    mapping: Mutex<HashMap<String, Option<String>>>,
    /// How many times each DID was queried — used by the limit test.
    queries: Mutex<Vec<String>>,
}

impl StubResolver {
    fn new(map: HashMap<String, Option<String>>) -> Self {
        Self {
            mapping: Mutex::new(map),
            queries: Mutex::new(Vec::new()),
        }
    }
    fn into_arc(self) -> Arc<StubResolver> {
        Arc::new(self)
    }
    fn query_count(&self, did: &str) -> usize {
        self.queries
            .lock()
            .unwrap()
            .iter()
            .filter(|d| d.as_str() == did)
            .count()
    }
}

#[async_trait]
impl DidHandleResolver for StubResolver {
    async fn resolve_handle(&self, did: &str) -> Result<Option<String>> {
        self.queries.lock().unwrap().push(did.to_string());
        // Snapshot the mapping out so the worker sees a consistent view
        // even if another writer fiddles mid-call.
        let m = self.mapping.lock().unwrap();
        // None → unknown; Some("") → unknown; Some("h") → resolved.
        match m.get(did) {
            Some(Some(h)) if !h.is_empty() => Ok(Some(h.clone())),
            _ => Ok(None),
        }
    }
}

/// Build a worker whose PDS, PLC and web resolvers are all the same
/// stub. The integration tests in this file don't care which method
/// the DID uses — the stub answers for any prefix.
fn worker_with(db: sqlx::PgPool, stub: Arc<StubResolver>) -> HandleSyncWorker {
    let r: Arc<dyn DidHandleResolver> = stub;
    HandleSyncWorker {
        db,
        pds_resolver: Arc::clone(&r),
        plc_resolver: Arc::clone(&r),
        web_resolver: Arc::clone(&r),
        interval_secs: 999,
    }
}

async fn try_test_db() -> Option<sqlx::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,
    }
}

fn unique_did(prefix: &str) -> String {
    format!("did:plc:hsync_{}_{}", prefix, Uuid::new_v4().simple())
}

async fn seed_post(
    db: &sqlx::PgPool,
    did: &str,
    rkey: &str,
    handle: &str,
    text: &str,
) -> Result<()> {
    let uri = format!("at://{did}/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,$3,$4,'app.twi.post',$5,'bafy',NULL,NULL,NULL, now())
           ON CONFLICT (uri) DO NOTHING"#,
    )
    .bind(&uri)
    .bind(did)
    .bind(handle)
    .bind(rkey)
    .bind(text)
    .execute(db)
    .await?;
    Ok(())
}

async fn fetch_handle(
    db: &sqlx::PgPool,
    did: &str,
) -> Result<Option<String>> {
    let row: Option<(String,)> = sqlx::query_as(
        "SELECT handle FROM posts WHERE did = $1 \
         ORDER BY indexed_at DESC LIMIT 1",
    )
    .bind(did)
    .fetch_optional(db)
    .await?;
    Ok(row.and_then(|(s,)| if s.is_empty() { None } else { Some(s) }))
}

async fn count_empty_handle_for(db: &sqlx::PgPool, did: &str) -> Result<i64> {
    let (n,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM posts WHERE did = $1 AND handle = ''",
    )
    .bind(did)
    .fetch_one(db)
    .await?;
    Ok(n)
}

/// Seed two posts for one DID with empty handles, point the stub
/// resolver at a known handle, and assert the worker fills both rows.
#[tokio::test]
async fn sync_resolves_known_did() {
    let Some(db) = try_test_db().await else {
        eprintln!("appview DB unavailable; skipping");
        return;
    };
    let did = unique_did("known");
    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await
        .unwrap();

    let expected = format!("known.{}", Uuid::new_v4().simple());
    let stub = StubResolver::new(HashMap::from([(
        did.clone(),
        Some(expected.clone()),
    )]))
    .into_arc();

    // Two posts → two rows must be updated.
    seed_post(&db, &did, "rka", "", "first").await.unwrap();
    seed_post(&db, &did, "rkb", "", "second").await.unwrap();
    assert_eq!(count_empty_handle_for(&db, &did).await.unwrap(), 2);

    // Drive the resolve half with our own DID. `run_once()` scans
    // globally, ordered by DID and capped at BATCH_SIZE, so on a
    // database that a live indexer keeps topping up, a freshly seeded
    // DID isn't guaranteed to make the batch — the assertions below
    // would then be measuring someone else's rows.
    let worker = worker_with(db.clone(), stub.clone());
    let report: SyncReport = worker.resolve_batch(vec![did.clone()]).await.unwrap();
    assert_eq!(report.resolved, 2, "{report:?}");
    assert_eq!(report.failed, 0);
    assert_eq!(report.skipped, 0);

    // No empty-handle rows remain for this DID and the handle matches.
    assert_eq!(count_empty_handle_for(&db, &did).await.unwrap(), 0);
    let got = fetch_handle(&db, &did).await.unwrap();
    assert_eq!(got.as_deref(), Some(expected.as_str()));

    // Resolver was consulted exactly once for this DID.
    assert_eq!(stub.query_count(&did), 1);

    // Cleanup.
    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await;
}

/// A DID whose posts already carry a handle must NOT be re-queried
/// or overwritten — the worker's `SELECT … WHERE handle = ''` filters
/// it out entirely.
#[tokio::test]
async fn sync_skips_already_resolved() {
    let Some(db) = try_test_db().await else {
        eprintln!("appview DB unavailable; skipping");
        return;
    };
    let did = unique_did("already");
    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await
        .unwrap();

    let pre = "preset.handle".to_string();
    seed_post(&db, &did, "rkA", &pre, "alpha").await.unwrap();
    seed_post(&db, &did, "rkB", &pre, "beta").await.unwrap();

    // The stub would overwrite with a different handle if asked.
    let stub = StubResolver::new(HashMap::from([(
        did.clone(),
        Some("wrong.handle".into()),
    )]))
    .into_arc();

    let worker = worker_with(db.clone(), stub.clone());
    // This test is about the SELECT: a DID whose rows already carry a
    // handle must never reach a resolver in the first place. So assert
    // on `select_candidates()` rather than forcing the DID through
    // `resolve_batch` — that would consult the resolver by definition
    // and defeat the `query_count == 0` check below.
    let candidates = worker.select_candidates().await.unwrap();
    assert!(
        !candidates.contains(&did),
        "a DID that already has a handle must not be selected"
    );

    // Both rows must still carry the pre-existing handle.
    let (cnt,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM posts WHERE did = $1 AND handle = $2",
    )
    .bind(&did)
    .bind(&pre)
    .fetch_one(&db)
    .await
    .unwrap();
    assert_eq!(cnt, 2);

    // Resolver was NOT consulted for this DID.
    assert_eq!(stub.query_count(&did), 0);

    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await;
}

/// Seed more than `BATCH_SIZE` distinct empty-handle DIDs and verify
/// only the first batch is processed this pass. The leftover DIDs
/// remain empty (will be picked up next pass).
#[tokio::test]
async fn sync_respects_limit() {
    use appview::handle_sync::BATCH_SIZE;

    let Some(db) = try_test_db().await else {
        eprintln!("appview DB unavailable; skipping");
        return;
    };

    let prefix = unique_did("limit");
    // Seed BATCH_SIZE + 5 distinct DIDs, each with one empty-handle post.
    let total = BATCH_SIZE as usize + 5;
    let mut all_dids = Vec::with_capacity(total);
    for i in 0..total {
        let did = format!("{prefix}_{i}");
        seed_post(&db, &did, "rk", "", "x").await.unwrap();
        all_dids.push(did);
    }
    let mut mapping = HashMap::new();
    for did in &all_dids {
        mapping.insert(
            did.clone(),
            Some(format!("resolved.{}", &did[did.len() - 6..])),
        );
    }
    let stub = StubResolver::new(mapping).into_arc();

    let worker = worker_with(db.clone(), stub.clone());
    // The cap lives in the SELECT, so assert it there; the report of a
    // full `run_once()` depends on what else is pending database-wide.
    let candidates = worker.select_candidates().await.unwrap();
    assert!(
        candidates.len() as i64 <= BATCH_SIZE,
        "select must never exceed BATCH_SIZE, got {}",
        candidates.len()
    );
    let batch: Vec<String> = all_dids.iter().take(BATCH_SIZE as usize).cloned().collect();
    let report = worker.resolve_batch(batch).await.unwrap();
    assert_eq!(
        report.resolved as i64,
        BATCH_SIZE,
        "expected exactly BATCH_SIZE rows resolved, got {report:?}"
    );
    assert_eq!(report.failed, 0);
    assert_eq!(report.skipped, 0);

    // Exactly 5 empty-handle posts remain (the capped overflow).
    let remaining: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM posts WHERE did LIKE $1 AND handle = ''",
    )
    .bind(format!("{prefix}_%"))
    .fetch_one(&db)
    .await
    .unwrap();
    assert_eq!(remaining, 5, "expected 5 unresolved rows left");

    // The stub resolver was consulted for exactly BATCH_SIZE DIDs.
    // (Note: the worker can't know which 5 were left out — the query
    // count is process-wide; we count the total below.)
    let total_qs = {
        let guard = stub.queries.lock().unwrap();
        guard.len()
    };
    assert_eq!(
        total_qs as i64,
        BATCH_SIZE,
        "resolver must be called at most BATCH_SIZE times, got {total_qs}"
    );

    // Cleanup so repeated runs stay hygienic.
    let _ = sqlx::query("DELETE FROM posts WHERE did LIKE $1")
        .bind(format!("{prefix}_%"))
        .execute(&db)
        .await;
}

/// Unresolvable DIDs (stub returns `Ok(None)`) count as `skipped`,
/// not `failed`, so a temporary PLC outage doesn't poison
/// observability dashboards.
#[tokio::test]
async fn sync_skips_unresolvable_dids() {
    let Some(db) = try_test_db().await else {
        eprintln!("appview DB unavailable; skipping");
        return;
    };
    let did = unique_did("unres");
    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await
        .unwrap();
    seed_post(&db, &did, "rk", "", "").await.unwrap();

    // DID deliberately absent from the stub's mapping → Ok(None).
    let stub = StubResolver::new(HashMap::new()).into_arc();

    let worker = worker_with(db.clone(), stub.clone());
    let report = worker.resolve_batch(vec![did.clone()]).await.unwrap();
    assert_eq!(report.resolved, 0);
    assert_eq!(report.failed, 0);
    assert_eq!(report.skipped, 1, "{report:?}");
    assert_eq!(
        count_empty_handle_for(&db, &did).await.unwrap(),
        1,
        "post must remain empty until resolver succeeds"
    );

    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await;
}

/// End-to-end test for the `did:web:` dispatch path: a `did:web:`
/// DID with an empty post handle must be routed to the **web**
/// resolver (not the PLC one) and the post handle must be updated
/// from the web resolver's answer.
#[tokio::test]
async fn sync_resolves_did_web_via_web_resolver() {
    let Some(db) = try_test_db().await else {
        eprintln!("appview DB unavailable; skipping");
        return;
    };
    let did = format!("did:web:example.com:user:{}", Uuid::new_v4().simple());
    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await
        .unwrap();
    seed_post(&db, &did, "rk", "", "web post").await.unwrap();

    // Two stubs that disagree. The dispatcher MUST pick the web one
    // for a did:web DID — choosing the PLC one would write the wrong
    // handle.
    let expected = format!("web-handle.{}", Uuid::new_v4().simple());
    let plc = StubResolver::new(HashMap::from([(
        did.clone(),
        Some("WRONG-PLC-HANDLE".into()),
    )]));
    let web = StubResolver::new(HashMap::from([(
        did.clone(),
        Some(expected.clone()),
    )]));
    let plc_arc: Arc<dyn DidHandleResolver> = plc.into_arc();
    let web_arc: Arc<dyn DidHandleResolver> = web.into_arc();

    // The local PDS is consulted before the method dispatch and does
    // not host a foreign `did:web:` — wiring one of the other stubs in
    // here would make it claim a DID it doesn't have, and the test
    // would assert against the PDS-first rule instead of the dispatch.
    let pds_arc: Arc<dyn DidHandleResolver> =
        StubResolver::new(HashMap::new()).into_arc();

    let worker = HandleSyncWorker {
        db: db.clone(),
        pds_resolver: pds_arc,
        plc_resolver: plc_arc,
        web_resolver: web_arc,
        interval_secs: 999,
    };
    let report = worker.resolve_batch(vec![did.clone()]).await.unwrap();
    assert_eq!(
        report.resolved, 1,
        "did:web must resolve through the web resolver, got {report:?}"
    );
    assert_eq!(report.failed, 0);
    let h = fetch_handle(&db, &did).await.unwrap();
    assert_eq!(h.as_deref(), Some(expected.as_str()));

    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await;
}

/// `did:plc:` DIDs must still flow through the PLC resolver — the
/// web resolver must NOT be consulted (which would otherwise issue
/// a `https://plc.directory/.../did.json` request and fail).
#[tokio::test]
async fn sync_resolves_did_plc_via_plc_resolver() {
    let Some(db) = try_test_db().await else {
        eprintln!("appview DB unavailable; skipping");
        return;
    };
    let did = unique_did("plcpath");
    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await
        .unwrap();
    seed_post(&db, &did, "rk", "", "plc post").await.unwrap();

    let expected = format!("plc-handle.{}", Uuid::new_v4().simple());
    let plc = StubResolver::new(HashMap::from([(
        did.clone(),
        Some(expected.clone()),
    )]));
    // The web stub deliberately holds the wrong handle. If dispatch
    // wrongly routed a did:plc DID to the web resolver, the row would
    // end up with "WRONG-WEB-HANDLE".
    let web = StubResolver::new(HashMap::from([(
        did.clone(),
        Some("WRONG-WEB-HANDLE".into()),
    )]));
    let plc_arc: Arc<dyn DidHandleResolver> = plc.into_arc();
    let web_arc: Arc<dyn DidHandleResolver> = web.into_arc();

    // Same reasoning as the did:web test: the PDS doesn't host this
    // DID, so the method dispatch is what's under test.
    let pds_arc: Arc<dyn DidHandleResolver> =
        StubResolver::new(HashMap::new()).into_arc();

    let worker = HandleSyncWorker {
        db: db.clone(),
        pds_resolver: pds_arc,
        plc_resolver: plc_arc,
        web_resolver: web_arc,
        interval_secs: 999,
    };
    let report = worker.resolve_batch(vec![did.clone()]).await.unwrap();
    assert_eq!(
        report.resolved, 1,
        "did:plc must resolve through the PLC resolver, got {report:?}"
    );
    let h = fetch_handle(&db, &did).await.unwrap();
    assert_eq!(h.as_deref(), Some(expected.as_str()));

    let _ = sqlx::query("DELETE FROM posts WHERE did = $1")
        .bind(&did)
        .execute(&db)
        .await;
}