Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193mod common;
use axum::body::Body;use axum::http::{Request, StatusCode};use happyview::db::adapt_sql;use http_body_util::BodyExt;use serde_json::{Value, json};use serial_test::serial;use tower::ServiceExt;use uuid::Uuid;
use common::app::TestApp;
async fn json_body(resp: axum::response::Response) -> Value { let body = resp.into_body().collect().await.unwrap().to_bytes(); serde_json::from_slice(&body).unwrap()}
fn admin_get( uri: &str, cookie: (axum::http::HeaderName, axum::http::HeaderValue),) -> Request<Body> { Request::builder() .uri(uri) .header(cookie.0, cookie.1) .body(Body::empty()) .unwrap()}
fn admin_post( uri: &str, cookie: (axum::http::HeaderName, axum::http::HeaderValue), body: &Value,) -> Request<Body> { Request::builder() .method("POST") .uri(uri) .header(cookie.0, cookie.1) .header("content-type", "application/json") .body(Body::from(serde_json::to_vec(body).unwrap())) .unwrap()}
async fn seed_job(app: &TestApp, job_type: &str, status: &str) -> String { seed_job_with_dpop(app, job_type, status, false, None, None).await}
async fn seed_job_with_dpop( app: &TestApp, job_type: &str, status: &str, inherit_auth: bool, api_client_id: Option<&str>, dpop_key_id: Option<&str>,) -> String { let id = Uuid::new_v4().to_string(); let now = happyview::db::now_rfc3339(); let input = serde_json::to_string(&json!({"test": true})).unwrap();
let sql = adapt_sql( "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth, api_client_id, dpop_key_id) VALUES (?, ?, ?, ?, '{}', ?, ?, ?, ?, ?)", app.state.db_backend, ); happyview::db::query(&sql) .bind(&id) .bind(job_type) .bind(status) .bind(&input) .bind(&app.admin_did) .bind(&now) .bind(inherit_auth) .bind(api_client_id) .bind(dpop_key_id) .execute(&app.state.db) .await .expect("seed_job: insert failed");
id}
async fn set_job_status(app: &TestApp, id: &str, status: &str) { let sql = adapt_sql( "UPDATE happyview_jobs SET status = ? WHERE id = ?", app.state.db_backend, ); happyview::db::query(&sql) .bind(status) .bind(id) .execute(&app.state.db) .await .expect("set_job_status failed");}
// ---------------------------------------------------------------------------// List jobs// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn list_jobs_empty() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_get("/admin/jobs", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["jobs"].as_array().unwrap().len(), 0); assert_eq!(body["cursor"], Value::Null);}
#[tokio::test]#[serial]async fn list_jobs_returns_seeded_jobs() { common::require_db!(); let app = TestApp::new().await;
seed_job(&app, "test.export", "pending").await; seed_job(&app, "test.import", "running").await;
let resp = app .router .clone() .oneshot(admin_get("/admin/jobs", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; let jobs = body["jobs"].as_array().unwrap(); assert_eq!(jobs.len(), 2);}
#[tokio::test]#[serial]async fn list_jobs_filters_by_status() { common::require_db!(); let app = TestApp::new().await;
seed_job(&app, "test.export", "pending").await; seed_job(&app, "test.import", "running").await; seed_job(&app, "test.cleanup", "completed").await;
let resp = app .router .clone() .oneshot(admin_get("/admin/jobs?status=running", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; let jobs = body["jobs"].as_array().unwrap(); assert_eq!(jobs.len(), 1); assert_eq!(jobs[0]["job_type"], "test.import"); assert_eq!(jobs[0]["status"], "running");}
// ---------------------------------------------------------------------------// Get job// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn get_job_returns_details() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "pending").await;
let resp = app .router .clone() .oneshot(admin_get(&format!("/admin/jobs/{id}"), app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["id"], id); assert_eq!(body["job_type"], "test.export"); assert_eq!(body["status"], "pending"); assert_eq!(body["input"]["test"], true);}
#[tokio::test]#[serial]async fn get_job_not_found() { common::require_db!(); let app = TestApp::new().await;
let fake_id = Uuid::new_v4(); let resp = app .router .clone() .oneshot(admin_get( &format!("/admin/jobs/{fake_id}"), app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::NOT_FOUND);}
// ---------------------------------------------------------------------------// Cancel job// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn cancel_pending_job_sets_cancelled() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "pending").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["status"], "cancelled");}
#[tokio::test]#[serial]async fn cancel_running_job_sets_cancelling() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "running").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["status"], "cancelling");}
#[tokio::test]#[serial]async fn cancel_paused_job_sets_cancelled() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "paused").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["status"], "cancelled");}
#[tokio::test]#[serial]async fn cancel_completed_job_returns_409() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "completed").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::CONFLICT);}
// ---------------------------------------------------------------------------// Pause job// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn pause_running_job_sets_pausing() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "running").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/pause"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["status"], "pausing");}
#[tokio::test]#[serial]async fn pause_pending_job_returns_409() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "pending").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/pause"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::CONFLICT);}
// ---------------------------------------------------------------------------// Resume job// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn resume_paused_job_sets_pending() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "paused").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/resume"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["status"], "pending");}
#[tokio::test]#[serial]async fn resume_running_job_returns_409() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.export", "running").await;
let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/resume"), app.admin_cookie(), &json!({}), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::CONFLICT);}
// ---------------------------------------------------------------------------// Worker: claim → execute → complete//// This is the only coverage that exercises `claim_next_job`, whose// `UPDATE ... RETURNING` writes `status = 'running'` *before* the returned row// is decoded. A decode failure there leaves the job marked running with nobody// executing it, and every read below goes through raw SQL rather than the admin// API so a failure here points at the worker rather than the read path.// ---------------------------------------------------------------------------
async fn seed_script(app: &TestApp, trigger_id: &str, body: &str) { // created_at/updated_at have no default on Postgres — bind them explicitly. let now = happyview::db::now_rfc3339(); let sql = adapt_sql( "INSERT INTO happyview_scripts (id, body, script_type, created_at, updated_at) VALUES (?, ?, 'lua', ?, ?)", app.state.db_backend, ); happyview::db::query(&sql) .bind(trigger_id) .bind(body) .bind(&now) .bind(&now) .execute(&app.state.db) .await .expect("seed_script: insert failed");}
async fn job_row(app: &TestApp, id: &str) -> (String, Option<String>, Option<String>) { let sql = adapt_sql( "SELECT status, result, error FROM happyview_jobs WHERE id = ?", app.state.db_backend, ); happyview::db::query_as::<(String, Option<String>, Option<String>)>(&sql) .bind(id) .fetch_one(&app.state.db) .await .expect("job_row: select failed")}
/// Poll until the job leaves the pending/running states, or time out.async fn await_terminal_status( app: &TestApp, id: &str,) -> (String, Option<String>, Option<String>) { for _ in 0..100 { let row = job_row(app, id).await; if !matches!(row.0.as_str(), "pending" | "running") { return row; } tokio::time::sleep(std::time::Duration::from_millis(100)).await; } job_row(app, id).await}
#[tokio::test]#[serial]async fn worker_runs_pending_job_to_completion() { common::require_db!(); let app = TestApp::new().await;
seed_script( &app, "job.run:test.worker", "function handle() return { ok = true } end", ) .await; let id = seed_job(&app, "test.worker", "pending").await;
let worker = tokio::spawn(happyview::jobs::worker::run_worker(app.state.clone())); let (status, result, error) = await_terminal_status(&app, &id).await; worker.abort();
assert_eq!(status, "completed", "job error: {error:?}"); let result: Value = serde_json::from_str(&result.expect("no result persisted")).unwrap(); assert_eq!(result["ok"], true);}
// ---------------------------------------------------------------------------// Auth: unauthenticated requests// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn list_jobs_without_auth_returns_401() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot( Request::builder() .uri("/admin/jobs") .body(Body::empty()) .unwrap(), ) .await .unwrap();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);}
// ---------------------------------------------------------------------------// Full lifecycle: pending → running → pausing → paused → pending → cancel// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn full_job_lifecycle() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.lifecycle", "pending").await;
// Simulate worker claiming → running set_job_status(&app, &id, "running").await;
// Pause the running job let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/pause"), app.admin_cookie(), &json!({}), )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(json_body(resp).await["status"], "pausing");
// Simulate worker acknowledging pause set_job_status(&app, &id, "paused").await;
// Resume the paused job let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/resume"), app.admin_cookie(), &json!({}), )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(json_body(resp).await["status"], "pending");
// Cancel the pending job let resp = app .router .clone() .oneshot(admin_post( &format!("/admin/jobs/{id}/cancel"), app.admin_cookie(), &json!({}), )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(json_body(resp).await["status"], "cancelled");
// Verify final state let resp = app .router .clone() .oneshot(admin_get(&format!("/admin/jobs/{id}"), app.admin_cookie())) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let job = json_body(resp).await; assert_eq!(job["status"], "cancelled"); assert!(job["completed_at"].is_string());}
// ---------------------------------------------------------------------------// Job logs// ---------------------------------------------------------------------------
async fn seed_job_log(app: &TestApp, job_id: &str, level: &str, message: &str) { let id = Uuid::new_v4().to_string(); let now = happyview::db::now_rfc3339(); let sql = adapt_sql( "INSERT INTO happyview_job_logs (id, job_id, level, message, created_at) VALUES (?, ?, ?, ?, ?)", app.state.db_backend, ); happyview::db::query(&sql) .bind(&id) .bind(job_id) .bind(level) .bind(message) .bind(&now) .execute(&app.state.db) .await .expect("seed_job_log: insert failed");}
#[tokio::test]#[serial]async fn list_job_logs_empty() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.logs", "running").await;
let resp = app .router .clone() .oneshot(admin_get( &format!("/admin/jobs/{id}/logs"), app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; assert_eq!(body["logs"].as_array().unwrap().len(), 0); assert_eq!(body["cursor"], Value::Null);}
#[tokio::test]#[serial]async fn list_job_logs_returns_seeded_logs() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.logs", "running").await; seed_job_log(&app, &id, "info", "starting migration").await; seed_job_log(&app, &id, "warn", "rate limited").await; seed_job_log(&app, &id, "info", "migration complete").await;
let resp = app .router .clone() .oneshot(admin_get( &format!("/admin/jobs/{id}/logs"), app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; let logs = body["logs"].as_array().unwrap(); assert_eq!(logs.len(), 3); assert_eq!(logs[0]["message"], "starting migration"); assert_eq!(logs[0]["level"], "info"); assert_eq!(logs[1]["message"], "rate limited"); assert_eq!(logs[1]["level"], "warn"); assert_eq!(logs[2]["message"], "migration complete");}
#[tokio::test]#[serial]async fn list_job_logs_pagination() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.logs-page", "running").await; for i in 0..5 { seed_job_log(&app, &id, "info", &format!("log-{i}")).await; }
let resp = app .router .clone() .oneshot(admin_get( &format!("/admin/jobs/{id}/logs?limit=2"), app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; let logs = body["logs"].as_array().unwrap(); assert_eq!(logs.len(), 2); assert_eq!(logs[0]["message"], "log-0"); assert_eq!(logs[1]["message"], "log-1"); assert!(body["cursor"].is_string());
let cursor = body["cursor"].as_str().unwrap(); let resp2 = app .router .clone() .oneshot(admin_get( &format!("/admin/jobs/{id}/logs?limit=2&cursor={cursor}"), app.admin_cookie(), )) .await .unwrap();
assert_eq!(resp2.status(), StatusCode::OK); let body2 = json_body(resp2).await; let logs2 = body2["logs"].as_array().unwrap(); assert_eq!(logs2.len(), 2); assert_eq!(logs2[0]["message"], "log-2"); assert_eq!(logs2[1]["message"], "log-3");}
#[tokio::test]#[serial]async fn list_job_logs_without_auth_returns_401() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.logs-auth", "running").await;
let resp = app .router .clone() .oneshot( Request::builder() .uri(format!("/admin/jobs/{id}/logs")) .body(Body::empty()) .unwrap(), ) .await .unwrap();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);}
// ---------------------------------------------------------------------------// DPoP context persistence// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn create_job_persists_dpop_context() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job_with_dpop( &app, "test.dpop", "pending", true, Some("client_abc"), Some("key_xyz"), ) .await;
let resp = app .router .clone() .oneshot(admin_get(&format!("/admin/jobs/{id}"), app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let job = json_body(resp).await; assert_eq!(job["inherit_auth"], true); assert_eq!(job["api_client_id"], "client_abc"); assert_eq!(job["dpop_key_id"], "key_xyz");}
#[tokio::test]#[serial]async fn create_job_without_dpop_has_null_fields() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job(&app, "test.no-dpop", "pending").await;
let resp = app .router .clone() .oneshot(admin_get(&format!("/admin/jobs/{id}"), app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let job = json_body(resp).await; assert_eq!(job["inherit_auth"], false); assert_eq!(job["api_client_id"], Value::Null); assert_eq!(job["dpop_key_id"], Value::Null);}
#[tokio::test]#[serial]async fn get_job_returns_dpop_fields_in_response() { common::require_db!(); let app = TestApp::new().await;
let id = seed_job_with_dpop( &app, "test.dpop-api", "pending", true, Some("api_client_123"), Some("dpop_key_456"), ) .await;
let resp = app .router .clone() .oneshot(admin_get(&format!("/admin/jobs/{id}"), app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let job = json_body(resp).await; assert_eq!(job["id"], id); assert_eq!(job["inherit_auth"], true); assert_eq!(job["api_client_id"], "api_client_123"); assert_eq!(job["dpop_key_id"], "dpop_key_456");}
#[tokio::test]#[serial]async fn list_jobs_includes_dpop_fields() { common::require_db!(); let app = TestApp::new().await;
seed_job_with_dpop( &app, "test.list-dpop", "pending", true, Some("list_client"), Some("list_key"), ) .await;
let resp = app .router .clone() .oneshot(admin_get("/admin/jobs", app.admin_cookie())) .await .unwrap();
assert_eq!(resp.status(), StatusCode::OK); let body = json_body(resp).await; let jobs = body["jobs"].as_array().unwrap(); assert_eq!(jobs.len(), 1); assert_eq!(jobs[0]["api_client_id"], "list_client"); assert_eq!(jobs[0]["dpop_key_id"], "list_key"); assert_eq!(jobs[0]["inherit_auth"], true);}
// ---------------------------------------------------------------------------// Native job dispatch// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn scripts_cannot_claim_a_reserved_job_trigger() { common::require_db!(); let app = TestApp::new().await;
let resp = app .router .clone() .oneshot(admin_post( "/admin/scripts", app.admin_cookie(), &json!({ "id": "job.run:happyview.delete-collection", "body": "function handle() end", "script_type": "lua" }), )) .await .unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST); let body = json_body(resp).await; assert!( body.to_string().contains("reserved"), "error should explain the reservation: {body}" );}
async fn seed_record(app: &TestApp, uri: &str, collection: &str) { // created_at has no default on Postgres (dropped by the TEXT migration, // see 20260318000000_uuid_to_text.sql), so it must be supplied explicitly // to work on both backends. let sql = adapt_sql( "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, indexed_at, created_at) \ VALUES (?, ?, ?, ?, ?, ?, ?, ?)", app.state.db_backend, ); let now = happyview::db::now_rfc3339(); happyview::db::query(&sql) .bind(uri) .bind("did:plc:seed") .bind(collection) .bind("rkey1") .bind(r#"{"text":"hi"}"#) .bind("bafyseed") .bind(&now) .bind(&now) .execute(&app.state.db) .await .expect("seed_record failed");}
async fn count_in_collection(app: &TestApp, collection: &str) -> i64 { let sql = adapt_sql( "SELECT COUNT(*) FROM happyview_records WHERE collection = ?", app.state.db_backend, ); let (n,): (i64,) = happyview::db::query_as(&sql) .bind(collection) .fetch_one(&app.state.db) .await .expect("count failed"); n}
#[tokio::test]#[serial]async fn native_delete_collection_job_empties_the_collection() { common::require_db!(); let app = TestApp::new().await;
for i in 0..25 { seed_record( &app, &format!("at://did:plc:seed/app.test.post/{i}"), "app.test.post", ) .await; } seed_record(&app, "at://did:plc:seed/app.other.post/1", "app.other.post").await; assert_eq!(count_in_collection(&app, "app.test.post").await, 25);
let id = Uuid::new_v4().to_string(); let now = happyview::db::now_rfc3339(); // inherit_auth is a bool column; on Postgres a literal `0` doesn't // implicitly cast to boolean, so it must be a bound parameter (as // seed_job_with_dpop does above), not embedded as a literal in the SQL. let sql = adapt_sql( "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth) \ VALUES (?, 'happyview.delete-collection', 'pending', ?, '{}', ?, ?, ?)", app.state.db_backend, ); happyview::db::query(&sql) .bind(&id) .bind(r#"{"collection":"app.test.post"}"#) .bind(&app.admin_did) .bind(&now) .bind(false) .execute(&app.state.db) .await .expect("enqueue failed");
let worker = tokio::spawn(happyview::jobs::worker::run_worker(app.state.clone())); let (status, result, error) = await_terminal_status(&app, &id).await; worker.abort();
assert_eq!(status, "completed", "job error: {error:?}"); let result: Value = serde_json::from_str(&result.expect("no result")).unwrap(); assert_eq!(result["deleted"], 25);
assert_eq!(count_in_collection(&app, "app.test.post").await, 0); assert_eq!( count_in_collection(&app, "app.other.post").await, 1, "other collections must be untouched" );}
/// Insert `count` records into `collection` in chunks, staying well under/// SQLite's default bound variable limit (999) per statement.async fn seed_records_bulk(app: &TestApp, collection: &str, count: usize) { const CHUNK: usize = 100; let now = happyview::db::now_rfc3339(); let mut i = 0usize; while i < count { let n = std::cmp::min(CHUNK, count - i); let placeholders = vec!["(?, ?, ?, ?, ?, ?, ?, ?)"; n].join(", "); let sql = adapt_sql( &format!( "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, indexed_at, created_at) \ VALUES {placeholders}" ), app.state.db_backend, ); let mut q = happyview::db::query(&sql); for j in 0..n { let idx = i + j; q = q .bind(format!("at://did:plc:bulk/{collection}/{idx}")) .bind("did:plc:bulk".to_string()) .bind(collection.to_string()) .bind(format!("rkey{idx}")) .bind(r#"{"text":"bulk"}"#.to_string()) .bind(format!("bafybulk{idx}")) .bind(now.clone()) .bind(now.clone()); } q.execute(&app.state.db) .await .expect("seed_records_bulk: insert failed"); i += n; }}
async fn seed_job_row_with_status( app: &TestApp, id: &str, job_type: &str, status: &str, input: &Value,) { let sql = adapt_sql( "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth) \ VALUES (?, ?, ?, ?, '{}', ?, ?, ?)", app.state.db_backend, ); let now = happyview::db::now_rfc3339(); happyview::db::query(&sql) .bind(id) .bind(job_type) .bind(status) .bind(input.to_string()) .bind(&app.admin_did) .bind(&now) .bind(false) .execute(&app.state.db) .await .expect("seed_job_row_with_status failed");}
// Regression guard for B1: the delete loop's only historical exit was// `affected == 0`, which never happens on a collection still receiving// inserts (Jetstream ingest). The fix breaks on a *short* batch instead.// BATCH_SIZE is 5000 (see delete_collection.rs); seeding one row over a full// batch forces two DELETE statements — a full batch (affected == BATCH_SIZE,// loop must continue) then a short one (affected == 1 < BATCH_SIZE, loop must// break) — so an inverted or off-by-one comparison (e.g. `<=`) would leave a// row behind and fail this test's final count assertions.#[tokio::test]#[serial]async fn native_delete_collection_handles_more_than_one_batch() { common::require_db!(); let app = TestApp::new().await;
const SEEDED: usize = 5001; seed_records_bulk(&app, "app.bulk.post", SEEDED).await; assert_eq!( count_in_collection(&app, "app.bulk.post").await, SEEDED as i64 );
let id = Uuid::new_v4().to_string(); let now = happyview::db::now_rfc3339(); let sql = adapt_sql( "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth) \ VALUES (?, 'happyview.delete-collection', 'pending', ?, '{}', ?, ?, ?)", app.state.db_backend, ); happyview::db::query(&sql) .bind(&id) .bind(r#"{"collection":"app.bulk.post"}"#) .bind(&app.admin_did) .bind(&now) .bind(false) .execute(&app.state.db) .await .expect("enqueue failed");
let worker = tokio::spawn(happyview::jobs::worker::run_worker(app.state.clone())); let (status, result, error) = await_terminal_status(&app, &id).await; worker.abort();
assert_eq!(status, "completed", "job error: {error:?}"); let result: Value = serde_json::from_str(&result.expect("no result")).unwrap(); assert_eq!(result["deleted"], SEEDED); assert_eq!(count_in_collection(&app, "app.bulk.post").await, 0);}
// Regression guard for B3: the should_stop branch was untested. Pre-setting// the job row's status to `cancelling` before calling the handler makes// should_stop's very first check (which runs before any batch) fire on// iteration one — deterministic, no need to race the 5000-row batch.#[tokio::test]#[serial]async fn native_delete_collection_stops_on_cancelling_without_failing() { common::require_db!(); let app = TestApp::new().await;
seed_record(&app, "at://did:plc:seed/app.stop.post/1", "app.stop.post").await; seed_record(&app, "at://did:plc:seed/app.stop.post/2", "app.stop.post").await; seed_record(&app, "at://did:plc:seed/app.stop.post/3", "app.stop.post").await;
let id = Uuid::new_v4().to_string(); let input = json!({"collection": "app.stop.post"}); seed_job_row_with_status( &app, &id, "happyview.delete-collection", "cancelling", &input, ) .await;
let job = happyview::jobs::Job { id: id.clone(), job_type: "happyview.delete-collection".into(), status: "cancelling".into(), input, progress: json!({}), result: None, error: None, created_by: app.admin_did.clone(), started_at: None, completed_at: None, created_at: happyview::db::now_rfc3339(), inherit_auth: false, api_client_id: None, dpop_key_id: None, };
let outcome = happyview::jobs::native::delete_collection::run(&app.state, &job).await;
match outcome { happyview::jobs::native::NativeOutcome::Completed(result) => { assert_eq!(result["deleted"], 0); assert_eq!(result["stopped"], "cancelling"); } happyview::jobs::native::NativeOutcome::Failed(e) => { panic!("a deliberate cancel must not be reported as a failure: {e}"); } }
assert_eq!( count_in_collection(&app, "app.stop.post").await, 3, "no records should be deleted once should_stop fires before the first batch" );}
#[tokio::test]#[serial]async fn native_delete_collection_stops_on_pausing_without_failing() { common::require_db!(); let app = TestApp::new().await;
seed_record(&app, "at://did:plc:seed/app.pause.post/1", "app.pause.post").await;
let id = Uuid::new_v4().to_string(); let input = json!({"collection": "app.pause.post"}); seed_job_row_with_status(&app, &id, "happyview.delete-collection", "pausing", &input).await;
let job = happyview::jobs::Job { id: id.clone(), job_type: "happyview.delete-collection".into(), status: "pausing".into(), input, progress: json!({}), result: None, error: None, created_by: app.admin_did.clone(), started_at: None, completed_at: None, created_at: happyview::db::now_rfc3339(), inherit_auth: false, api_client_id: None, dpop_key_id: None, };
let outcome = happyview::jobs::native::delete_collection::run(&app.state, &job).await;
match outcome { happyview::jobs::native::NativeOutcome::Completed(result) => { assert_eq!(result["deleted"], 0); assert_eq!(result["stopped"], "pausing"); } happyview::jobs::native::NativeOutcome::Failed(e) => { panic!("a deliberate pause must not be reported as a failure: {e}"); } }
assert_eq!( count_in_collection(&app, "app.pause.post").await, 1, "no records should be deleted once should_stop fires before the first batch" );}