diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 8f4fb2d0..08b7564a 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -22,8 +22,6 @@ jobs: with: components: rustfmt, clippy - # Caches ~/.cargo registry/git and target/, keyed on toolchain + Cargo.lock; - # prunes workspace-crate artifacts so the cache holds only compiled deps. - name: Cache cargo uses: Swatinem/rust-cache@v2 @@ -33,7 +31,6 @@ jobs: - name: clippy run: cargo clippy --workspace --all-targets -- -D warnings - # --workspace covers tui (it is outside default-members). - name: test run: cargo test --workspace diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cef57497..48b0ea2b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,8 +22,6 @@ jobs: with: components: rustfmt, clippy - # Caches ~/.cargo registry/git and target/, keyed on toolchain + Cargo.lock; - # prunes workspace-crate artifacts so the cache holds only compiled deps. - name: Cache cargo uses: Swatinem/rust-cache@v2 @@ -33,7 +31,6 @@ jobs: - name: clippy run: cargo clippy --workspace --all-targets -- -D warnings - # --workspace covers tui (it is outside default-members). - name: test run: cargo test --workspace diff --git a/Dockerfile b/Dockerfile index 8388fe57..f3370e67 100644 --- a/Dockerfile +++ b/Dockerfile @@ -16,7 +16,6 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ RUN cargo install cargo-chef --locked --version 0.1.78 WORKDIR /build -# ----- planner: dependency recipe (changes only when manifests/lockfile change) ----- FROM chef AS planner COPY . . RUN cargo chef prepare --recipe-path recipe.json @@ -27,14 +26,10 @@ FROM chef AS builder # Features: federation (ActivityPub), nats (EVENT_BUS_BACKEND=nats). # Build without federation: --build-arg FEATURES= # To add NATS support: --build-arg FEATURES=federation,nats -# Declared before cook so cook and build compile identical feature sets. ARG FEATURES=federation -# .cargo carries the registry config needed to fetch private crates COPY .cargo ./.cargo COPY --from=planner /build/recipe.json recipe.json -# target/ is a normal layer (no cache mount) so the cooked deps live in the image -# layer cache; only the registry/git download caches are mounted. RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \ --mount=type=cache,target=/usr/local/cargo/git,sharing=locked \ cargo chef cook --release -p server -p worker --no-default-features --features "${FEATURES}" --recipe-path recipe.json diff --git a/crates/adapters/activitypub/src/deletion_notifier.rs b/crates/adapters/activitypub/src/deletion_notifier.rs index dcf368f9..2254c94f 100644 --- a/crates/adapters/activitypub/src/deletion_notifier.rs +++ b/crates/adapters/activitypub/src/deletion_notifier.rs @@ -10,11 +10,6 @@ use k_ap::ActivityPubService; use crate::urls::actor_url; -/// Broadcasts the actor `Delete` to the user's accepted followers. -/// -/// k-ap only enqueues the per-inbox deliveries; they are signed later from the -/// tombstoned user row and keypair, which the purge keeps until the tombstone -/// cleanup (see `AccountPurger`). pub struct ApDeletionNotifier { service: Arc, instance: InstanceIdentity, diff --git a/crates/adapters/activitypub/src/event_handler.rs b/crates/adapters/activitypub/src/event_handler.rs index c08af08a..c2dda4d5 100644 --- a/crates/adapters/activitypub/src/event_handler.rs +++ b/crates/adapters/activitypub/src/event_handler.rs @@ -63,9 +63,6 @@ impl EventHandler for ActivityPubEventHandler { } async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { - // A deleted account must not speak again: its queued events would - // otherwise sign and send from the tombstone after the Delete. Only - // the Delete itself (sent by the deletion notifier) goes out. if let Some(author) = self.outbound_author(event) && self.is_tombstoned(&author).await { @@ -173,8 +170,6 @@ impl EventHandler for ActivityPubEventHandler { } impl ActivityPubEventHandler { - /// The local account an outbound activity would be sent as. A queued - /// delivery counts unless it is the account's `Delete`. fn outbound_author(&self, event: &DomainEvent) -> Option { match event { DomainEvent::ReviewLogged { user_id, .. } diff --git a/crates/adapters/activitypub/src/remote_review_repository.rs b/crates/adapters/activitypub/src/remote_review_repository.rs index db63262c..88df1a06 100644 --- a/crates/adapters/activitypub/src/remote_review_repository.rs +++ b/crates/adapters/activitypub/src/remote_review_repository.rs @@ -14,7 +14,6 @@ pub struct RemoteReviewUpdate<'a> { pub spoilers: bool, } -/// A remote review to persist, with the events that ride the same write. pub struct NewRemoteReview<'a> { pub review: &'a Review, pub ap_id: &'a str, diff --git a/crates/adapters/activitypub/src/review_handler.rs b/crates/adapters/activitypub/src/review_handler.rs index f1e019be..b9a42f8f 100644 --- a/crates/adapters/activitypub/src/review_handler.rs +++ b/crates/adapters/activitypub/src/review_handler.rs @@ -159,7 +159,6 @@ impl ApObjectHandler for ReviewObjectHandler { spoilers: obj.contains_spoilers, }); - // The enrichment request rides the review write (one transaction). let events: Vec = obj .external_metadata_id .as_ref() diff --git a/crates/adapters/activitypub/src/social_adapter.rs b/crates/adapters/activitypub/src/social_adapter.rs index e0a54ca2..e90dff9e 100644 --- a/crates/adapters/activitypub/src/social_adapter.rs +++ b/crates/adapters/activitypub/src/social_adapter.rs @@ -39,7 +39,6 @@ impl CompositeSocialAdapter { } } - /// Best-effort publish for writes k_ap owns (no shared transaction). async fn announce(&self, events: &[DomainEvent]) { for event in events { if let Err(e) = self.event_publisher.publish(event).await { @@ -84,8 +83,6 @@ impl SocialCommand for CompositeSocialAdapter { SocialIdentity::Remote { actor_url } => actor_url.clone(), }, }; - // Not atomic: k_ap owns the transaction, so the events are published - // best-effort after its write succeeds. self.ap_service .follow(follower.value(), &handle) .await @@ -102,7 +99,6 @@ impl SocialCommand for CompositeSocialAdapter { ) -> Result<(), DomainError> { match target { SocialIdentity::Local(_) => self.local.unfollow(follower, target, events).await, - // Not atomic: k_ap owns the transaction. SocialIdentity::Remote { .. } => { self.ap_service .unfollow(follower.value(), &self.instance.actor_url_of(target)) @@ -122,7 +118,6 @@ impl SocialCommand for CompositeSocialAdapter { ) -> Result<(), DomainError> { match requester { SocialIdentity::Local(_) => self.local.accept_follow(owner, requester, events).await, - // Not atomic: k_ap owns the transaction. SocialIdentity::Remote { .. } => { self.ap_service .accept_follower(owner.value(), &self.instance.actor_url_of(requester)) @@ -142,7 +137,6 @@ impl SocialCommand for CompositeSocialAdapter { ) -> Result<(), DomainError> { match requester { SocialIdentity::Local(_) => self.local.reject_follow(owner, requester, events).await, - // Not atomic: k_ap owns the transaction. SocialIdentity::Remote { .. } => { self.ap_service .reject_follower(owner.value(), &self.instance.actor_url_of(requester)) @@ -162,7 +156,6 @@ impl SocialCommand for CompositeSocialAdapter { ) -> Result<(), DomainError> { match follower { SocialIdentity::Local(_) => self.local.remove_follower(owner, follower, events).await, - // Not atomic: k_ap owns the transaction. SocialIdentity::Remote { .. } => { self.ap_service .remove_follower(owner.value(), &self.instance.actor_url_of(follower)) @@ -174,7 +167,6 @@ impl SocialCommand for CompositeSocialAdapter { } } - /// Not atomic: k_ap owns the transaction. async fn block( &self, blocker: &UserId, @@ -190,7 +182,6 @@ impl SocialCommand for CompositeSocialAdapter { Ok(()) } - /// Not atomic: k_ap owns the transaction. async fn unblock( &self, blocker: &UserId, diff --git a/crates/adapters/activitypub/src/tombstone_gate.rs b/crates/adapters/activitypub/src/tombstone_gate.rs index ae6f00f5..100471bb 100644 --- a/crates/adapters/activitypub/src/tombstone_gate.rs +++ b/crates/adapters/activitypub/src/tombstone_gate.rs @@ -8,9 +8,6 @@ use axum::{ }; use domain::{ports::UserRepository, value_objects::UserId}; -/// Answers `410 Gone` for the per-user ActivityPub endpoints of a deleted -/// account. k-ap resolves tombstones so it can sign the account's `Delete` -/// deliveries; this keeps that from making the actor look alive. pub async fn tombstone_gate( State(users): State>, request: Request, @@ -27,9 +24,6 @@ pub async fn tombstone_gate( next.run(request).await } -/// The user id in any `/users/{id}/...` path (outbox, inbox, collections, -/// content objects such as `/users/{id}/reviews/{rid}`). The bare actor path -/// has its own tombstone response. pub fn gated_user(path: &str) -> Option { let rest = path.strip_prefix("/users/")?; let (id, endpoint) = rest.split_once('/')?; diff --git a/crates/adapters/activitypub/src/user_adapter.rs b/crates/adapters/activitypub/src/user_adapter.rs index ff5ec6f1..4a5e298a 100644 --- a/crates/adapters/activitypub/src/user_adapter.rs +++ b/crates/adapters/activitypub/src/user_adapter.rs @@ -64,10 +64,6 @@ impl DomainUserRepoAdapter { impl ApUserRepository for DomainUserRepoAdapter { async fn find_by_id(&self, id: uuid::Uuid) -> anyhow::Result> { let user_id = UserId::from_uuid(id); - // A deleted account still resolves here, so k-ap can sign the queued - // `Delete` deliveries until the tombstone cleanup. Username lookups - // below stay live-only, and the AP endpoints answer 410 for it - // (`tombstone_gate`, `actor_document_response`). let user = match self.repo.find_by_id(&user_id).await? { Some(u) => u, None => match self.repo.find_tombstoned_by_id(&user_id).await? { diff --git a/crates/adapters/activitypub/tests/activity_ids.rs b/crates/adapters/activitypub/tests/activity_ids.rs index d160376f..303096ac 100644 --- a/crates/adapters/activitypub/tests/activity_ids.rs +++ b/crates/adapters/activitypub/tests/activity_ids.rs @@ -1,11 +1,3 @@ -//! Outbound activity ids across a redelivered event, through the real k-ap -//! `ActivityPubService` (delivery captured via its `EventPublisher` seam). -//! -//! k-ap 0.5.0 derives Create ids from the object id (UUIDv5, -//! `service/broadcast.rs:285-288`) but mints Update/Delete/Move/actor-update ids -//! randomly (`broadcast.rs:81,129,157,210,321,343,377` -> `url_scheme.rs:53`, -//! `Uuid::new_v4`) with no override. - use std::sync::Arc; use std::time::Duration; @@ -189,8 +181,6 @@ async fn activitypub_is_idempotent_when_run_twice() { assert_eq!(first, second, "a redelivered Create keeps its activity id"); } -/// Characterises the k-ap 0.5.0 limitation recorded in ADR-0015. If k-ap starts -/// minting deterministic Delete ids this fails: upgrade the ADR and flip it. #[tokio::test] async fn redelivered_delete_gets_a_fresh_activity_id_known_k_ap_limitation() { let mut h = harness().await; @@ -207,8 +197,6 @@ async fn redelivered_delete_gets_a_fresh_activity_id_known_k_ap_limitation() { assert_ne!(first, second); } -/// Queued events of a deleted account must not sign and send from its -/// tombstone; the account's own Delete still goes out. #[tokio::test] async fn a_deleted_account_sends_nothing_but_its_delete() { let mut h = harness().await; @@ -240,7 +228,6 @@ async fn a_deleted_account_sends_nothing_but_its_delete() { signing_actor_id: h.user.value(), }; - // The review repository panics if the handler gets past the guard. for event in [&logged, &added, &queued] { h.handler.handle(event).await.unwrap(); } diff --git a/crates/adapters/http-axum/Cargo.toml b/crates/adapters/http-axum/Cargo.toml index 8a2158da..952f7c6e 100644 --- a/crates/adapters/http-axum/Cargo.toml +++ b/crates/adapters/http-axum/Cargo.toml @@ -7,7 +7,6 @@ license = "MIT" [features] default = ["federation"] -# Enables federation-gated handlers; forwards to application/federation federation = ["application/federation"] [dependencies] diff --git a/crates/adapters/http-axum/src/api/mod.rs b/crates/adapters/http-axum/src/api/mod.rs index 5ab3d146..ef9f44ac 100644 --- a/crates/adapters/http-axum/src/api/mod.rs +++ b/crates/adapters/http-axum/src/api/mod.rs @@ -16,9 +16,6 @@ use axum::{Router, routing}; use crate::state::AppState; -/// JSON API routes (mounted under `/api/v1`), minus the webhook ingestion -/// endpoints ([`webhook_router`], separately rate limited) and the -/// federation-gated admin endpoints (`federation::api_routes`). pub fn router() -> Router { let base = Router::new() .route("/diary", routing::get(diary::get_diary)) @@ -150,7 +147,6 @@ pub fn router() -> Router { base.merge(social_routes()) } -/// Media-server webhook ingestion endpoints (mounted under `/api/v1`). pub fn webhook_router() -> Router { Router::new() .route( diff --git a/crates/adapters/http-axum/src/errors.rs b/crates/adapters/http-axum/src/errors.rs index d9b6b44e..7f7e6728 100644 --- a/crates/adapters/http-axum/src/errors.rs +++ b/crates/adapters/http-axum/src/errors.rs @@ -14,7 +14,6 @@ pub const CONFLICT: &str = "CONFLICT"; pub const RATE_LIMITED: &str = "RATE_LIMITED"; pub const INTERNAL: &str = "INTERNAL"; -/// The only constructor of JSON error bodies in this crate. pub fn refuse(status: StatusCode, code: &'static str, message: impl Into) -> Response { ( status, diff --git a/crates/adapters/http-axum/src/extractors/json.rs b/crates/adapters/http-axum/src/extractors/json.rs index 27c8d994..1eb12f66 100644 --- a/crates/adapters/http-axum/src/extractors/json.rs +++ b/crates/adapters/http-axum/src/extractors/json.rs @@ -1,5 +1,3 @@ -//! JSON API extractors whose rejections are refused in the one error shape. - use axum::{ extract::{ FromRequest, FromRequestParts, Multipart, Path, Query, Request, @@ -13,10 +11,6 @@ use serde::de::DeserializeOwned; use crate::errors::{INTERNAL, INVALID_INPUT, refuse}; -/// Malformed request (bad JSON, query or path), refused as `INVALID_INPUT`. -/// Status follows axum: 400 syntax, 415 content type, 422 data errors, 413 size. -/// A rejection with a 5xx status is a server-side fault (e.g. a route that -/// declares no path parameter), refused as `INTERNAL`. pub struct MalformedRequest { status: StatusCode, message: String, @@ -69,7 +63,6 @@ impl From for MalformedRequest { } } -/// `axum::extract::Multipart` with refusals in the one error shape. pub struct Upload(pub Multipart); impl FromRequest for Upload @@ -83,7 +76,6 @@ where } } -/// `axum::Json` with refusals in the one error shape. pub struct Body(pub T); impl FromRequest for Body @@ -99,7 +91,6 @@ where } } -/// `axum::extract::Query` with refusals in the one error shape. pub struct Params(pub T); impl FromRequestParts for Params @@ -115,7 +106,6 @@ where } } -/// `axum::extract::Path` with refusals in the one error shape. pub struct PathId(pub T); impl FromRequestParts for PathId diff --git a/crates/adapters/http-axum/src/extractors/mod.rs b/crates/adapters/http-axum/src/extractors/mod.rs index 38d4ec45..7000f100 100644 --- a/crates/adapters/http-axum/src/extractors/mod.rs +++ b/crates/adapters/http-axum/src/extractors/mod.rs @@ -146,7 +146,6 @@ where ) .await .map_err(ApiError)? - // `None` only if the account vanished between `ensure_active` and here. .ok_or_else(|| ApiError(DomainError::NotFound("user not found".into())))?; if is_admin { Ok(AdminApiUser(user_id)) @@ -175,7 +174,6 @@ where ) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response())? - // `None` only if the account vanished between `ensure_active` and here. .ok_or_else(|| StatusCode::UNAUTHORIZED.into_response())?; if is_admin { Ok(AdminUser(user_id)) diff --git a/crates/adapters/http-axum/src/federation/mod.rs b/crates/adapters/http-axum/src/federation/mod.rs index a182b0a2..3f205b2f 100644 --- a/crates/adapters/http-axum/src/federation/mod.rs +++ b/crates/adapters/http-axum/src/federation/mod.rs @@ -1,7 +1,3 @@ -//! Federation-only HTTP surface. This is the one module that decides, per -//! build, whether federation routes exist: both builds expose the same -//! functions, so `router.rs` and the api/html modules stay free of `cfg`. - #[cfg(feature = "federation")] pub(crate) mod blocklist_api; #[cfg(feature = "federation")] @@ -18,7 +14,6 @@ use axum::{Router, routing}; use crate::state::AppState; -/// Classic-UI routes that only exist with federation (AP collections, admin pages). #[cfg(feature = "federation")] pub fn html_routes() -> Router { Router::new() @@ -53,7 +48,6 @@ pub fn html_routes() -> Router { ) } -/// JSON API routes that only exist with federation (blocklist administration). #[cfg(feature = "federation")] pub fn api_routes() -> Router { Router::new() @@ -80,8 +74,6 @@ pub fn api_routes() -> Router { ) } -/// Without federation the follower/following collection URLs fall back to the -/// classic list pages. #[cfg(not(feature = "federation"))] pub fn html_routes() -> Router { Router::new() @@ -108,8 +100,6 @@ pub fn api_routes() -> Router { Router::new() } -/// ActivityPub content negotiation for `GET /users/{id}`: federation builds -/// answer AP clients with the actor document, others always render HTML. #[cfg(not(feature = "federation"))] pub async fn actor_document_response( _state: &AppState, diff --git a/crates/adapters/http-axum/src/federation/negotiation.rs b/crates/adapters/http-axum/src/federation/negotiation.rs index 831b9775..64cb5ecc 100644 --- a/crates/adapters/http-axum/src/federation/negotiation.rs +++ b/crates/adapters/http-axum/src/federation/negotiation.rs @@ -6,8 +6,6 @@ use uuid::Uuid; use crate::state::AppState; -/// Answers ActivityPub clients (`Accept: application/activity+json` or -/// `application/ld+json`) with the actor document; `None` means "render HTML". pub async fn actor_document_response( state: &AppState, headers: &HeaderMap, diff --git a/crates/adapters/http-axum/src/html/csrf.rs b/crates/adapters/http-axum/src/html/csrf.rs index aa607835..52fd9ba9 100644 --- a/crates/adapters/http-axum/src/html/csrf.rs +++ b/crates/adapters/http-axum/src/html/csrf.rs @@ -19,7 +19,6 @@ pub fn extract_from_cookie(headers: &axum::http::HeaderMap) -> Option { }) } -/// Router state for [`csrf_middleware`]: whether cookies carry the `Secure` attribute. #[derive(Clone, Copy)] pub struct SecureCookies(pub bool); diff --git a/crates/adapters/http-axum/src/html/mod.rs b/crates/adapters/http-axum/src/html/mod.rs index 52e40020..5691afde 100644 --- a/crates/adapters/http-axum/src/html/mod.rs +++ b/crates/adapters/http-axum/src/html/mod.rs @@ -23,8 +23,6 @@ use axum::{Router, routing}; use crate::state::AppState; -/// Classic server-rendered UI routes, minus the rate-limited auth forms -/// ([`auth_router`]) and the federation-gated pages (`federation::html_routes`). pub fn router() -> Router { let base = Router::new() .route("/", routing::get(diary::get_activity_feed_html)) @@ -127,7 +125,6 @@ pub fn router() -> Router { base.merge(social_routes()) } -/// Login, logout and registration forms; the caller adds the auth rate limiter. pub fn auth_router() -> Router { Router::new() .route( diff --git a/crates/adapters/http-axum/src/html/users.rs b/crates/adapters/http-axum/src/html/users.rs index fce35519..50785572 100644 --- a/crates/adapters/http-axum/src/html/users.rs +++ b/crates/adapters/http-axum/src/html/users.rs @@ -233,11 +233,8 @@ pub async fn post_delete_user( let target_id = UserId::from_uuid(target); let own = actor == target_id; match delete_account(&state, actor, target_id, &form.password).await { - // An admin removing their own account ends their own session too. Ok(()) if own => super::auth::get_logout(State(state)).await.into_response(), Ok(()) => axum::response::Redirect::to("/users").into_response(), - // Same as other forbidden classic actions: a plain 403, not a redirect - // to a page that would not show the error. Err(e @ DomainError::Forbidden(_)) => crate::errors::domain_error_response(e), Err(e) => axum::response::Redirect::to(&format!( "/users/{target}?error={}", diff --git a/crates/adapters/http-axum/src/router.rs b/crates/adapters/http-axum/src/router.rs index 7a9a02f5..57448687 100644 --- a/crates/adapters/http-axum/src/router.rs +++ b/crates/adapters/http-axum/src/router.rs @@ -48,7 +48,6 @@ async fn health_handler() -> axum::Json { axum::Json(serde_json::json!({ "status": "ok" })) } -/// JSON API rate-limit refusals use the one error shape. fn slow_down(reason: RejectionReason) -> axum::response::Response { match reason { RejectionReason::QuotaExceeded { .. } => refuse( @@ -64,7 +63,6 @@ fn slow_down(reason: RejectionReason) -> axum::response::Response { } } -/// Every `/api/v1` limiter is built here so its refusals use the one error shape. fn api_governor(quota: Quota) -> axum_governor::GovernorConfig { GovernorConfigBuilder::default() .with_extractor(PeerIp::default()) diff --git a/crates/adapters/http-axum/src/support.rs b/crates/adapters/http-axum/src/support.rs index 75420146..1461ab17 100644 --- a/crates/adapters/http-axum/src/support.rs +++ b/crates/adapters/http-axum/src/support.rs @@ -1,5 +1,3 @@ -//! Request/response helpers shared by the JSON API and the classic HTML UI. - use crate::state::AppState; use domain::models::import::DomainField; diff --git a/crates/adapters/http-axum/src/test_support.rs b/crates/adapters/http-axum/src/test_support.rs index d92cac11..ff1300b8 100644 --- a/crates/adapters/http-axum/src/test_support.rs +++ b/crates/adapters/http-axum/src/test_support.rs @@ -1,5 +1,3 @@ -//! Fixtures shared by the in-crate mapper tests. - use domain::value_objects::{SocialActor, SocialIdentity, UserId}; pub(crate) fn local_actor(uid: &UserId) -> SocialActor { diff --git a/crates/adapters/image-converter/src/handler.rs b/crates/adapters/image-converter/src/handler.rs index 046562cb..17d1c3ae 100644 --- a/crates/adapters/image-converter/src/handler.rs +++ b/crates/adapters/image-converter/src/handler.rs @@ -45,9 +45,6 @@ impl EventHandler for ImageConversionHandler { return Ok(()); } - // `get` falls back to `{key}.webp` / `{key}.avif`, which would re-encode - // an already converted image on redelivery. Probe the exact key first: - // once the original is gone, an earlier delivery finished the conversion. if !self.storage.exists(&key).await? { tracing::warn!(%key, "original image already gone, treating conversion as done"); return Ok(()); diff --git a/crates/adapters/image-converter/tests/conversion_handler_test.rs b/crates/adapters/image-converter/tests/conversion_handler_test.rs index c23e48c6..bbd0cc12 100644 --- a/crates/adapters/image-converter/tests/conversion_handler_test.rs +++ b/crates/adapters/image-converter/tests/conversion_handler_test.rs @@ -181,7 +181,6 @@ async fn image_conversion_is_idempotent_when_run_twice() { handler.handle(&event).await.unwrap(); let converted = storage.get(&new_key).await.unwrap(); - // Redelivery after the original was swapped and deleted. handler.handle(&event).await.unwrap(); assert_eq!(image_ref.swaps().len(), 1, "one swap"); diff --git a/crates/adapters/metadata/src/omdb.rs b/crates/adapters/metadata/src/omdb.rs index 46ccc98f..39c64767 100644 --- a/crates/adapters/metadata/src/omdb.rs +++ b/crates/adapters/metadata/src/omdb.rs @@ -16,7 +16,6 @@ pub(crate) struct OmdbProvider { impl OmdbProvider { pub(crate) fn new(api_key: String) -> Self { - // reqwest is built with rustls-no-provider; install ring so this works without bootstrap. let _ = rustls::crypto::ring::default_provider().install_default(); Self { client: reqwest::Client::new(), diff --git a/crates/adapters/metadata/src/tmdb.rs b/crates/adapters/metadata/src/tmdb.rs index 655f966d..855d5af4 100644 --- a/crates/adapters/metadata/src/tmdb.rs +++ b/crates/adapters/metadata/src/tmdb.rs @@ -15,7 +15,6 @@ pub(crate) struct TmdbProvider { impl TmdbProvider { pub(crate) fn new(api_key: String) -> Self { - // reqwest is built with rustls-no-provider; install ring so this works without bootstrap. let _ = rustls::crypto::ring::default_provider().install_default(); Self { client: reqwest::Client::new(), diff --git a/crates/adapters/metadata/tests/client_tls_omdb.rs b/crates/adapters/metadata/tests/client_tls_omdb.rs index a2594f52..d2a9ec54 100644 --- a/crates/adapters/metadata/tests/client_tls_omdb.rs +++ b/crates/adapters/metadata/tests/client_tls_omdb.rs @@ -1,5 +1,3 @@ -//! Fresh test binary, no bootstrap: constructing the client must not panic -//! for lack of a rustls CryptoProvider. #[test] fn omdb_client_builds_without_bootstrap() { let _ = metadata::MetadataClientImpl::new_omdb("key".into()); diff --git a/crates/adapters/metadata/tests/client_tls_tmdb.rs b/crates/adapters/metadata/tests/client_tls_tmdb.rs index 9a63e670..f05c8190 100644 --- a/crates/adapters/metadata/tests/client_tls_tmdb.rs +++ b/crates/adapters/metadata/tests/client_tls_tmdb.rs @@ -1,5 +1,3 @@ -//! Fresh test binary, no bootstrap: constructing the client must not panic -//! for lack of a rustls CryptoProvider. #[test] fn tmdb_client_builds_without_bootstrap() { let _ = metadata::MetadataClientImpl::new_tmdb("key".into()); diff --git a/crates/adapters/nats/src/config.rs b/crates/adapters/nats/src/config.rs index e4f31404..e63a7023 100644 --- a/crates/adapters/nats/src/config.rs +++ b/crates/adapters/nats/src/config.rs @@ -11,7 +11,6 @@ pub struct NatsConfig { pub subject_prefix: String, pub stream_name: String, pub consumer_name: String, - /// JetStream redelivery bound, same as the DB queue's attempt limit. pub max_deliver: i64, } diff --git a/crates/adapters/nats/src/consumer.rs b/crates/adapters/nats/src/consumer.rs index 261e6a06..5259f2a1 100644 --- a/crates/adapters/nats/src/consumer.rs +++ b/crates/adapters/nats/src/consumer.rs @@ -25,11 +25,8 @@ struct NatsJetStreamAckHandle { max_deliver: i64, } -/// How to nak a failed delivery: the same backoff schedule as the SQLite -/// queue, keyed by JetStream's delivery count (1 on first delivery). struct NakPlan { delay: std::time::Duration, - /// This was the last delivery `max_deliver` allows. exhausted: bool, } @@ -82,9 +79,6 @@ impl AckHandle for NatsJetStreamAckHandle { .map_err(|e| DomainError::InfrastructureError(e.to_string())) } - /// JetStream counts every redelivery toward `max_deliver`, deferrals - /// included. The delay is at least the remaining claim lease, so the - /// deliveries allowed by `max_deliver` still span more than one lease. async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError> { self.message .ack_with(AckKind::Nak(Some(delay))) @@ -159,20 +153,10 @@ fn consume_from_rx(rx: EnvelopeRx) -> BoxStream<'static, Result Option { (existing.duplicate_window != DEDUP_WINDOW).then(|| StreamConfig { duplicate_window: DEDUP_WINDOW, @@ -180,10 +164,6 @@ fn stream_config_update(existing: &StreamConfig) -> Option { }) } -/// `Some(updated)` when an existing durable consumer disagrees on `max_deliver` -/// or `ack_wait`. The update is a clone of the existing config with only those -/// two fields changed, so operator-set fields survive and immutable ones -/// (durable name, filter subject, deliver policy) cannot mismatch. fn consumer_config_update( existing: &jetstream::consumer::Config, desired_max_deliver: i64, diff --git a/crates/adapters/nats/src/lib.rs b/crates/adapters/nats/src/lib.rs index ca2125fc..29ba505e 100644 --- a/crates/adapters/nats/src/lib.rs +++ b/crates/adapters/nats/src/lib.rs @@ -12,8 +12,6 @@ use std::sync::Arc; use domain::ports::{EventConsumer, EventRelaySink}; -/// The worker's NATS side: where handlers read events from, and where the -/// outbox relay writes them to. pub async fn create_bus( cfg: NatsConfig, ) -> anyhow::Result<(Arc, Arc)> { @@ -29,7 +27,6 @@ pub async fn create_bus( Arc::new(NatsRelaySink::new_core(client, cfg.subject_prefix)), ) } - // The consumer creates the stream the relay publishes into. NatsMode::JetStream => ( Arc::new(NatsJetStreamConsumer::create(&cfg, client.clone()).await?), Arc::new(NatsRelaySink::new_jetstream(client, cfg.subject_prefix)), diff --git a/crates/adapters/nats/src/relay.rs b/crates/adapters/nats/src/relay.rs index 52454ffe..5cf26c18 100644 --- a/crates/adapters/nats/src/relay.rs +++ b/crates/adapters/nats/src/relay.rs @@ -9,13 +9,9 @@ use event_payload::EventPayload; use crate::{payload::NatsEventMessage, subject::event_to_subject}; -/// What goes on the wire for one outbox row. Built purely from the row, so -/// relaying it again (after a crash between publish and mark-done) yields the -/// same `Nats-Msg-Id` and JetStream drops the duplicate. pub(crate) struct RelayMessage { pub subject: String, pub payload: Vec, - /// `None` for rows queued before identities existed: those get no dedup. pub msg_id: Option, } @@ -42,8 +38,6 @@ enum Transport { JetStream(jetstream::Context), } -/// Publishes outbox rows to NATS. JetStream mode waits for the broker ack and -/// sets `Nats-Msg-Id`; core mode is fire-and-forget with no broker-side dedup. pub struct NatsRelaySink { transport: Transport, subject_prefix: String, diff --git a/crates/adapters/poster-fetcher/src/lib.rs b/crates/adapters/poster-fetcher/src/lib.rs index 022a4b3f..ff8f0d30 100644 --- a/crates/adapters/poster-fetcher/src/lib.rs +++ b/crates/adapters/poster-fetcher/src/lib.rs @@ -16,7 +16,6 @@ pub struct ReqwestPosterFetcher { impl ReqwestPosterFetcher { pub fn new(config: PosterFetcherConfig) -> anyhow::Result { - // reqwest is built with rustls-no-provider; install ring so this works without bootstrap. let _ = rustls::crypto::ring::default_provider().install_default(); let client = reqwest::Client::builder() .timeout(Duration::from_secs(config.timeout_seconds)) diff --git a/crates/adapters/poster-fetcher/tests/client_tls.rs b/crates/adapters/poster-fetcher/tests/client_tls.rs index 74013f71..4db6a8b0 100644 --- a/crates/adapters/poster-fetcher/tests/client_tls.rs +++ b/crates/adapters/poster-fetcher/tests/client_tls.rs @@ -1,5 +1,3 @@ -//! Fresh test binary, no bootstrap: constructing the client must not panic -//! for lack of a rustls CryptoProvider. use poster_fetcher::{PosterFetcherConfig, ReqwestPosterFetcher}; #[test] diff --git a/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs b/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs index 5a82873f..6dffaba5 100644 --- a/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs +++ b/crates/adapters/sqlite-federation/tests/federation/remote_review_test.rs @@ -1,5 +1,3 @@ -//! A remote review and the enrichment request it triggers commit together. - use activitypub::{NewRemoteReview, RemoteReviewRepository}; use domain::{ events::DomainEvent, diff --git a/crates/adapters/sqlite/src/event_queue/mod.rs b/crates/adapters/sqlite/src/event_queue/mod.rs index 5253f1af..2a1cb46a 100644 --- a/crates/adapters/sqlite/src/event_queue/mod.rs +++ b/crates/adapters/sqlite/src/event_queue/mod.rs @@ -16,9 +16,6 @@ use tokio::sync::{Mutex, mpsc}; use crate::shared::enqueue_events; pub(crate) use payload::DbEventPayload; -/// How long a claim holds a row. A worker that dies between claim and -/// ack/nack strands its rows in `processing`; once the lease lapses the claim -/// query takes them again. Must exceed the slowest handler. pub(crate) const CLAIM_LEASE_SECONDS: i64 = 600; pub struct DbEventQueueConfig { @@ -140,9 +137,6 @@ struct QueueRow { claimed_at: Option, } -/// Claims a batch in one atomic statement. A read-then-write transaction can -/// fail with SQLITE_BUSY_SNAPSHOT (517) when another connection commits in -/// between, which `busy_timeout` cannot retry. async fn claim_batch( pool: &SqlitePool, config: &DbEventQueueConfig, @@ -168,13 +162,10 @@ async fn claim_batch( .await .map_err(|e| DomainError::InfrastructureError(format!("claim batch: {e}")))?; - // RETURNING order is unspecified; deliver oldest first. rows.sort_by(|a, b| (&a.next_attempt_at, a.id).cmp(&(&b.next_attempt_at, b.id))); Ok(rows) } -/// A row that cannot be decoded can never succeed, and has no ack handle to -/// settle it with; park it so it is neither retried forever nor left claimed. async fn dead_letter_undecodable(pool: &SqlitePool, row_id: i64, error: &DomainError) { tracing::error!( row_id, @@ -223,14 +214,10 @@ struct DbAckHandle { row_id: i64, attempts: i32, max_attempts: i32, - /// The claim this handle belongs to. Settling is conditional on it, so a - /// worker whose lease lapsed and was reclaimed cannot flip the new claim's row. claimed_at: Option, } impl DbAckHandle { - /// Runs a settle statement whose last two binds are the row id and claim - /// stamp. Zero rows changed means the claim is no longer ours. async fn settle<'q>( &self, what: &str, diff --git a/crates/adapters/sqlite/src/event_queue/tests.rs b/crates/adapters/sqlite/src/event_queue/tests.rs index 94f86cd9..24a6c731 100644 --- a/crates/adapters/sqlite/src/event_queue/tests.rs +++ b/crates/adapters/sqlite/src/event_queue/tests.rs @@ -1,7 +1,3 @@ -//! Inline (not under `tests/`) because `claim_batch` is private and the race -//! only shows on its claim path; two pools on a file-backed WAL database are -//! needed, as in production. - use std::str::FromStr; use sqlx::SqlitePool; @@ -73,7 +69,6 @@ async fn claiming_while_another_connection_writes_does_not_fail() { .await .expect("claim must not fail while another connection writes"); if batch.is_empty() { - // an empty UPDATE still takes the write lock; don't starve the writer tokio::task::yield_now().await; } claimed += batch.len(); @@ -315,8 +310,6 @@ mod stale_handle { use super::{claim_batch, config}; use crate::event_queue::{CLAIM_LEASE_SECONDS, SqliteEventQueue}; - /// Claims a row, lets its lease lapse, and has another worker reclaim it. - /// Returns the first (now stale) worker's envelope. async fn reclaimed() -> (sqlx::SqlitePool, EventEnvelope) { let pool = sqlx::sqlite::SqlitePoolOptions::new() .max_connections(1) @@ -340,8 +333,6 @@ mod stale_handle { .await .unwrap(); assert_eq!(claim_batch(&pool, &config(10)).await.unwrap().len(), 1); - // Real reclaims happen a lease later, so their timestamp differs from - // the stale handle's; the claim clock has only second resolution. let second_claim = (chrono::Utc::now() - chrono::Duration::seconds(5)) .format("%Y-%m-%dT%H:%M:%SZ") .to_string(); diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index 46e00d49..b33e09ee 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -20,8 +20,6 @@ pub use repositories::{ create_federated_profile_query, create_image_ref, create_person_adapter, create_search_adapter, }; -/// Row helpers shared with the `sqlite-federation` crate (which cannot live -/// here because it depends on `activitypub`/`k-ap`). pub mod persistence { pub use crate::shared::{datetime_to_str, enqueue_events}; } @@ -79,7 +77,6 @@ pub async fn wire(database_url: &str) -> anyhow::Result { wire_pool(pool).await } -/// Migrates an already-connected pool and builds every repository on it. pub async fn wire_pool(pool: SqlitePool) -> anyhow::Result { use anyhow::Context; diff --git a/crates/adapters/sqlite/src/migrations.rs b/crates/adapters/sqlite/src/migrations.rs index afcae43b..37d141e8 100644 --- a/crates/adapters/sqlite/src/migrations.rs +++ b/crates/adapters/sqlite/src/migrations.rs @@ -3,10 +3,6 @@ use std::time::Duration; use domain::errors::DomainError; use sqlx::SqlitePool; -/// Server and worker both migrate at startup, so two processes can race on the -/// same database. The loser's migration fails (e.g. duplicate column) because -/// the winner already applied it; re-running re-reads `_sqlx_migrations` and -/// skips what is done. Only an error that persists across all retries is returned. const RETRY_BACKOFFS: [Duration; 5] = [ Duration::from_millis(100), Duration::from_millis(250), diff --git a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs index eea71c14..694a6657 100644 --- a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs +++ b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs @@ -9,10 +9,6 @@ use crate::shared::map_sqlx_error; const TIMESTAMP_FORMAT: &str = "%Y-%m-%dT%H:%M:%SZ"; -/// Every table the purge empties for the account, with the column that holds -/// the user id. `reviews` is further restricted to the user's own (non-remote) -/// rows. A guard test compares this list with the schema, so a new -/// user-owned table cannot be forgotten. pub const PURGED_USER_TABLES: &[(&str, &str)] = &[ ("reviews", "user_id"), ("watchlist_entries", "user_id"), @@ -30,13 +26,8 @@ pub const PURGED_USER_TABLES: &[(&str, &str)] = &[ ("blocked_actors", "local_user_id"), ]; -/// Tables that reference a user but are kept until the tombstone cleanup: -/// the keypair is needed to sign queued federation deliveries. pub const RETAINED_USER_TABLES: &[&str] = &["ap_local_actors"]; -/// Columns that identify the user by actor or content URL rather than by id. -/// The purge matches rows on them; the guard test asserts each still exists so -/// a rename cannot silently stop the purge from clearing them. pub const URL_KEYED_USER_COLUMNS: &[(&str, &str)] = &[ ("ap_announces", "object_url"), ("ap_announces", "actor_url"), @@ -44,9 +35,6 @@ pub const URL_KEYED_USER_COLUMNS: &[(&str, &str)] = &[ ("ap_following", "remote_actor_url"), ]; -/// Columns holding serialized events and activities that carry the user's id -/// (the outbox log keeps finished rows, including authored activity JSON) and -/// the global wrap-up reports, whose `most_active_user` names a member. pub const PAYLOAD_KEYED_USER_COLUMNS: &[(&str, &str)] = &[ ("event_queue", "payload"), ("wrap_up_records", "report_json"), @@ -95,8 +83,6 @@ impl AccountPurger for SqliteAccountPurger { let base = self.instance.base_url().to_string(); let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; - // Strip first: it claims the purge. Only a tombstoned, not yet purged - // account matches, so a live account (or a re-run) changes nothing. let claimed = sqlx::query( "UPDATE users SET email = 'deleted-' || id || '@deleted.invalid', password_hash = '!purged', display_name = NULL, bio = NULL, @@ -113,7 +99,6 @@ impl AccountPurger for SqliteAccountPurger { return Ok(()); } - // Announces of the user's content, matched before the reviews go. sqlx::query( "DELETE FROM ap_announces WHERE object_url IN ( @@ -130,11 +115,6 @@ impl AccountPurger for SqliteAccountPurger { .await .map_err(map_sqlx_error)?; - // The outbox log: finished events and activities authored as, or - // mentioning, this user. Unfinished rows are never touched (they - // belong to other users' work too and finish or dead-letter on their - // own); the tombstone cleanup removes their history later. The - // `UserDeleted` row that triggered this run stays until then as well. sqlx::query( "DELETE FROM event_queue WHERE instr(payload, ?1) > 0 @@ -146,8 +126,6 @@ impl AccountPurger for SqliteAccountPurger { .await .map_err(map_sqlx_error)?; - // Global wrap-up reports name their most active member; redact it - // (the type allows none) and keep the rest of the report. sqlx::query( "UPDATE wrap_up_records SET report_json = json_set(report_json, '$.most_active_user', NULL) @@ -159,7 +137,6 @@ impl AccountPurger for SqliteAccountPurger { .await .map_err(map_sqlx_error)?; - // Other accounts' follows of this one. for table in ["ap_followers", "ap_following"] { sqlx::query(&format!("DELETE FROM {table} WHERE remote_actor_url = ?")) .bind(&actor_url) @@ -201,9 +178,6 @@ impl AccountPurger for SqliteAccountPurger { ) -> Result { let cutoff = cutoff.format(TIMESTAMP_FORMAT).to_string(); let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; - // A tombstone whose id is still named by an unfinished event waits for - // the next run: that event may yet be handled, and its history row - // must be erased with the rest. let expiring: Vec = sqlx::query_scalar( "SELECT id FROM users u WHERE purged_at IS NOT NULL AND purged_at < ? @@ -217,15 +191,11 @@ impl AccountPurger for SqliteAccountPurger { .await .map_err(map_sqlx_error)?; for id in &expiring { - // Includes the `UserDeleted` row and the `Delete` deliveries. sqlx::query("DELETE FROM event_queue WHERE instr(payload, ?) > 0") .bind(id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; - // A Follow arriving on the shared inbox after the purge can leave - // a late row for the tombstone; delete it explicitly rather than - // rely on the foreign-key cascade of the final `users` delete. for table in ["ap_followers", "ap_following"] { sqlx::query(&format!("DELETE FROM {table} WHERE local_user_id = ?")) .bind(id) diff --git a/crates/adapters/sqlite/src/repositories/diary/query.rs b/crates/adapters/sqlite/src/repositories/diary/query.rs index 2d53de26..997fe139 100644 --- a/crates/adapters/sqlite/src/repositories/diary/query.rs +++ b/crates/adapters/sqlite/src/repositories/diary/query.rs @@ -131,8 +131,6 @@ impl DiaryQuery for SqliteDiaryRepository { FeedSortBy::RatingAsc => "r.rating ASC, r.watched_at ASC", }; - // Tombstoned local authors vanish from the feed; remote and orphan - // rows have no `users` match, so `deleted_at` is NULL for them. where_parts.push("u.deleted_at IS NULL".to_string()); let where_clause = where_parts.join(" AND "); diff --git a/crates/adapters/sqlite/src/repositories/image_ref/mod.rs b/crates/adapters/sqlite/src/repositories/image_ref/mod.rs index 5601b667..0a1397c5 100644 --- a/crates/adapters/sqlite/src/repositories/image_ref/mod.rs +++ b/crates/adapters/sqlite/src/repositories/image_ref/mod.rs @@ -26,8 +26,6 @@ pub fn create_image_ref(pool: SqlitePool) -> (Arc, Arc Result<(), DomainError> { let mut tx = self .pool @@ -54,8 +52,6 @@ impl ImageRefCommand for SqliteImageRefAdapter { #[async_trait] impl ImageRefQuery for SqliteImageRefAdapter { - // Tombstoned users are intentionally visible here until the purge removes - // the row, so their avatar is not garbage-collected early. async fn list_keys(&self) -> Result, DomainError> { let rows: Vec<(String,)> = sqlx::query_as( "SELECT avatar_path FROM users WHERE avatar_path IS NOT NULL diff --git a/crates/adapters/sqlite/src/repositories/movie_dedup/mod.rs b/crates/adapters/sqlite/src/repositories/movie_dedup/mod.rs index f380ef56..6e1d0697 100644 --- a/crates/adapters/sqlite/src/repositories/movie_dedup/mod.rs +++ b/crates/adapters/sqlite/src/repositories/movie_dedup/mod.rs @@ -37,8 +37,6 @@ impl MovieDeduplicator for SqliteMovieDeduplicator { .await .map_err(crate::shared::map_sqlx_error)?; - // The old row still holds the external id (it is deleted last); release it - // so the canonical upsert does not violate UNIQUE(external_metadata_id). sqlx::query("UPDATE movies SET external_metadata_id = NULL WHERE id = ?") .bind(&old) .execute(&mut *tx) diff --git a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs index 756b46b9..6e410fc5 100644 --- a/crates/adapters/sqlite/src/repositories/processed_events/mod.rs +++ b/crates/adapters/sqlite/src/repositories/processed_events/mod.rs @@ -6,7 +6,6 @@ use domain::{ }; use sqlx::SqlitePool; -/// Matches the column default in migration 0101, so string comparison orders rows by time. const TIMESTAMP_FORMAT: &str = "%Y-%m-%dT%H:%M:%SZ"; const CLAIM_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%.3fZ"; @@ -23,8 +22,6 @@ impl SqliteProcessedEventLedger { ) } - /// A shorter lease than production's, so lease takeover can be tested - /// without waiting half an hour. pub fn with_lease(pool: SqlitePool, lease: std::time::Duration) -> Self { Self { pool, @@ -37,7 +34,6 @@ fn stamp(at: chrono::NaiveDateTime) -> String { at.format(TIMESTAMP_FORMAT).to_string() } -/// Millisecond precision, so two claims taken in one second have distinct tokens. fn claim_stamp(at: chrono::NaiveDateTime) -> String { at.format(CLAIM_FORMAT).to_string() } @@ -51,13 +47,9 @@ fn parse_claim_stamp(value: &str) -> Option { #[async_trait] impl ProcessedEventLedger for SqliteProcessedEventLedger { async fn begin(&self, id: &EventId, handler: &str) -> Result { - // Retried once: the row can vanish (released) between the claim attempt - // and the state lookup. for _ in 0..2 { let now = chrono::Utc::now().naive_utc(); let started = claim_stamp(now); - // One statement, so two workers cannot both win: insert if absent, - // or take over an in_progress claim past its lease. let claimed = sqlx::query( "INSERT INTO processed_events (event_id, handler, status, started_at, processed_at) VALUES (?, ?, 'in_progress', ?, ?) @@ -214,7 +206,6 @@ mod tests { token(ledger.begin(&id, "b").await.unwrap()); token(ledger.begin(&EventId::new(), "a").await.unwrap()); - // The blocked delivery is told to come back when the lease runs out. let lease = std::time::Duration::from_secs(HANDLER_LEASE_SECONDS as u64); assert!(retry_after <= lease && retry_after > lease - std::time::Duration::from_secs(5)); } @@ -260,7 +251,6 @@ mod tests { insert(&pool, &id, "in_progress", "2020-01-01T00:00:00Z").await; token(ledger.begin(&id, "h").await.unwrap()); - // The takeover refreshed the lease, so a third delivery is blocked. blocked(ledger.begin(&id, "h").await.unwrap()); let processed_at: String = sqlx::query_scalar("SELECT processed_at FROM processed_events") .fetch_one(&pool) @@ -295,7 +285,6 @@ mod tests { ledger.release(&id, "a", &lapsed).await.unwrap(); ledger.complete(&id, "a", &lapsed).await.unwrap(); - // Still the new owner's live claim: neither call touched it. blocked(ledger.begin(&id, "a").await.unwrap()); ledger.complete(&id, "a", &owner).await.unwrap(); assert_eq!(ledger.begin(&id, "a").await.unwrap(), Begin::AlreadyDone); diff --git a/crates/adapters/sqlite/src/repositories/users/mod.rs b/crates/adapters/sqlite/src/repositories/users/mod.rs index ef91e249..776e291e 100644 --- a/crates/adapters/sqlite/src/repositories/users/mod.rs +++ b/crates/adapters/sqlite/src/repositories/users/mod.rs @@ -169,9 +169,6 @@ impl UserRepository for SqliteUserRepository { .begin() .await .map_err(crate::shared::map_sqlx_error)?; - // The last-admin guard lives in this single statement, so two admins - // deleting each other concurrently cannot both pass it: the write lock - // serialises them and the loser sees the winner's tombstone. let stamped = sqlx::query( "UPDATE users SET deleted_at = ?1 WHERE id = ?2 AND deleted_at IS NULL AND (role != 'admin' OR EXISTS ( @@ -283,8 +280,6 @@ impl UserRepository for SqliteUserRepository { } impl SqliteUserRepository { - /// Uniqueness check over every row, tombstones included: a deleted - /// account keeps its email and username until the purge removes the row. async fn row_exists(&self, column: &str, value: &str) -> Result { sqlx::query_scalar::<_, i64>(&format!( "SELECT EXISTS(SELECT 1 FROM users WHERE {column} = ?)" diff --git a/crates/adapters/sqlite/src/repositories/watchlist/mod.rs b/crates/adapters/sqlite/src/repositories/watchlist/mod.rs index 8d747023..78c96abf 100644 --- a/crates/adapters/sqlite/src/repositories/watchlist/mod.rs +++ b/crates/adapters/sqlite/src/repositories/watchlist/mod.rs @@ -52,7 +52,6 @@ impl WatchlistRepository for SqliteWatchlistRepository { .map_err(crate::shared::map_sqlx_error)? .rows_affected() > 0; - // A repeat add changed nothing, so it announces nothing. if inserted { enqueue_events(&mut tx, events).await?; } diff --git a/crates/adapters/sqlite/src/shared.rs b/crates/adapters/sqlite/src/shared.rs index 035e240f..eb79feb9 100644 --- a/crates/adapters/sqlite/src/shared.rs +++ b/crates/adapters/sqlite/src/shared.rs @@ -10,9 +10,6 @@ use domain::{ }, }; -/// THE outbox write path: inserts one `event_queue` row per event, each with a -/// fresh `EventId`, inside the caller's transaction so a change and the events -/// announcing it commit or roll back together. pub async fn enqueue_events( tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, events: &[domain::events::DomainEvent], diff --git a/crates/adapters/sqlite/tests/account_purge_test.rs b/crates/adapters/sqlite/tests/account_purge_test.rs index 91bc76ba..ec2a5e29 100644 --- a/crates/adapters/sqlite/tests/account_purge_test.rs +++ b/crates/adapters/sqlite/tests/account_purge_test.rs @@ -1,6 +1,3 @@ -//! The purge removes every row a deleted account owns, strips the user row to -//! a data-free tombstone, and the cleanup later drops the tombstone itself. - use std::collections::BTreeSet; use std::str::FromStr; @@ -69,8 +66,6 @@ struct Seeded { movie_id: String, } -/// Alice owns a row in every table; bob has a little of everything, including -/// a follow of alice and an announce of his own review, which must survive. async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { let alice = test_user("alice"); let bob = test_user("bob"); @@ -205,7 +200,6 @@ async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { follower_actor_url: REMOTE_ACTOR.into(), status: FollowStatus::Accepted, }, - // bob follows alice, and alice follows bob FollowWrite::AddFollow { follower_id: b.value(), target_actor_url: alice_actor.clone(), @@ -221,7 +215,6 @@ async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { ) .await .unwrap(); - // A remote review attributed to the same account id is not the user's. exec( pool, "INSERT INTO reviews (id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url) @@ -234,7 +227,6 @@ async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { ], ) .await; - // Announces of alice's review and goal, of bob's review, and one by alice. let alice_review: String = sqlx::query_scalar("SELECT id FROM reviews WHERE user_id = ? AND remote_actor_url IS NULL") .bind(&a_str) @@ -352,7 +344,6 @@ async fn purge_removes_every_user_owned_row() { purger(&pool).purge(&s.user).await.unwrap(); for (table, column) in PURGED_USER_TABLES { - // Reviews: only the remote one, asserted below, is left. let left = if *table == "reviews" { 1 } else { 0 }; assert_eq!( count_for(&pool, table, column, &s.user).await, @@ -406,14 +397,12 @@ async fn purge_removes_every_user_owned_row() { "finished rows mentioning the user are gone; pending/processing rows (incl. other users' events about her) and bob's own rows stay" ); - // The other account is untouched. for (table, column) in PURGED_USER_TABLES { assert!( count_for(&pool, table, column, &s.other).await > 0, "{table} of the other user must survive" ); } - // The keypair stays for queued deliveries. for table in RETAINED_USER_TABLES { assert_eq!(count_for(&pool, table, "user_id", &s.user).await, 1); } @@ -458,8 +447,6 @@ async fn the_purge_table_lists_cover_every_user_referencing_column_in_the_schema "{t} has a foreign key to users but is not in the purge lists" ); } - // Columns the purge matches by URL or payload must still exist, so a - // rename fails here instead of silently leaving rows behind. for (table, column) in URL_KEYED_USER_COLUMNS .iter() .chain(PAYLOAD_KEYED_USER_COLUMNS) @@ -474,9 +461,6 @@ async fn the_purge_table_lists_cover_every_user_referencing_column_in_the_schema } } -/// `ap_activities` (k-ap's inbound dedup) holds only remote activity ids and a -/// timestamp: nothing in it is attributable to a local user. If it ever grows a -/// column, the purge must start clearing it. #[tokio::test] async fn ap_activities_holds_nothing_attributable_to_a_local_user() { let pool = migrated_pool().await; @@ -555,7 +539,6 @@ async fn purge_strips_the_user_row_but_keeps_the_username_and_state() { .await .unwrap() ); - // Images stay referenced until the objects are gone. let mut keys = p.owned_image_keys(&s.user).await.unwrap(); keys.sort(); assert_eq!(keys, vec!["avatars/a.webp", "banners/a.webp"]); @@ -571,8 +554,6 @@ async fn purging_twice_is_a_no_op() { parts.user.mark_deleted(&s.user, &[]).await.unwrap(); let p = purger(&pool); p.purge(&s.user).await.unwrap(); - // New data cannot appear for a purged account, but a re-run must not - // touch anything else or fail. let before: i64 = count(&pool, "SELECT COUNT(*) FROM reviews").await; let stamp: String = sqlx::query_scalar("SELECT purged_at FROM users WHERE id = ?") .bind(s.user.value().to_string()) @@ -589,7 +570,6 @@ async fn purging_twice_is_a_no_op() { .await .unwrap(); assert_eq!(stamp, again); - // And an id that never existed is fine too. p.purge(&UserId::generate()).await.unwrap(); } @@ -666,7 +646,6 @@ async fn tombstone_cleanup_removes_rows_older_than_retention_and_keeps_younger_o ) .await; - // Younger than the cutoff: kept. let removed = p .purge_expired_tombstones(now - chrono::Duration::days(9)) .await @@ -674,9 +653,6 @@ async fn tombstone_cleanup_removes_rows_older_than_retention_and_keeps_younger_o assert_eq!(removed, 0); assert_eq!(count_for(&pool, "users", "id", &s.user).await, 1); - // Unfinished events still name the user (the pending Delete, bob's - // pending FollowRequested, a processing row): the tombstone waits and no - // unfinished row is ever deleted. let alice = s.user.value().to_string(); let unfinished = || async { sqlx::query_scalar::<_, i64>( @@ -696,7 +672,6 @@ async fn tombstone_cleanup_removes_rows_older_than_retention_and_keeps_younger_o assert_eq!(count_for(&pool, "users", "id", &s.user).await, 1); assert_eq!(unfinished().await, before); - // A Follow that reached the shared inbox after the purge left a row. exec( &pool, "INSERT INTO ap_followers (local_user_id, remote_actor_url, status, created_at) @@ -704,7 +679,6 @@ async fn tombstone_cleanup_removes_rows_older_than_retention_and_keeps_younger_o &[&alice], ) .await; - // Once they finish, the cleanup takes the tombstone and the history. exec(&pool, "UPDATE event_queue SET status = 'done'", &[]).await; let removed = p .purge_expired_tombstones(now - chrono::Duration::days(7)) diff --git a/crates/adapters/sqlite/tests/cascade_test.rs b/crates/adapters/sqlite/tests/cascade_test.rs index 54eac773..66500f0d 100644 --- a/crates/adapters/sqlite/tests/cascade_test.rs +++ b/crates/adapters/sqlite/tests/cascade_test.rs @@ -1,7 +1,3 @@ -//! Real-database counterpart of the `InMemoryStore` cascade tests in the domain -//! crate. Both iterate the same table lists (`domain::testing::*_CASCADE_TABLES`), -//! so the fake and the schema cannot drift apart unnoticed. - use std::collections::BTreeSet; use std::str::FromStr; @@ -28,8 +24,6 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; const REMOTE_ACTOR: &str = "https://remote.example/users/ann"; -/// Private in-memory database with plain `SqliteConnectOptions`; relies on -/// sqlx's default `foreign_keys = ON`, which production also depends on. async fn migrated_pool() -> SqlitePool { let opts = SqliteConnectOptions::from_str("sqlite::memory:").unwrap(); let pool = SqlitePoolOptions::new() @@ -202,7 +196,6 @@ async fn seed_user_with_dependents( ) .await .unwrap(); - // No repository in this crate owns these two tables (federation crate). sqlx::query("INSERT INTO blocked_actors (local_user_id, remote_actor_url, blocked_at) VALUES (?, ?, 'now')") .bind(user_id.value().to_string()) .bind(REMOTE_ACTOR) diff --git a/crates/adapters/sqlite/tests/repositories/movie_dedup_test.rs b/crates/adapters/sqlite/tests/repositories/movie_dedup_test.rs index f243ae69..15cc38e4 100644 --- a/crates/adapters/sqlite/tests/repositories/movie_dedup_test.rs +++ b/crates/adapters/sqlite/tests/repositories/movie_dedup_test.rs @@ -46,8 +46,6 @@ async fn insert_review(pool: &SqlitePool, movie: &MovieId) { .unwrap(); } -/// Mirrors what `application::movies::merge_duplicates` builds when the -/// canonical row does not exist. fn canonical_from(old: &Movie, canonical_id: MovieId) -> Movie { Movie::from_persistence( canonical_id, diff --git a/crates/adapters/sqlite/tests/repositories/outbox_support.rs b/crates/adapters/sqlite/tests/repositories/outbox_support.rs index 60b4ff3d..7c3416be 100644 --- a/crates/adapters/sqlite/tests/repositories/outbox_support.rs +++ b/crates/adapters/sqlite/tests/repositories/outbox_support.rs @@ -1,5 +1,3 @@ -//! Shared helpers for the transactional-outbox repository tests. - use std::str::FromStr; use sqlx::{ @@ -25,8 +23,6 @@ pub async fn count(pool: &SqlitePool, table: &str) -> i64 { .unwrap() } -/// Makes every later `event_queue` insert abort, so a write whose own row -/// already landed inside the transaction must be rolled back with it. pub async fn fail_event_inserts(pool: &SqlitePool) { sqlx::query( "CREATE TRIGGER fail_event_insert BEFORE INSERT ON event_queue \ diff --git a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs index 14f46307..c69e9144 100644 --- a/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs +++ b/crates/adapters/sqlite/tests/repositories/outbox_writes_test.rs @@ -1,5 +1,3 @@ -//! Goal, watchlist and profile writes commit with their events or not at all. - use domain::{ events::DomainEvent, models::{ProfileField, UserProfile, WatchlistEntry}, @@ -796,8 +794,6 @@ async fn a_failing_follow_write_rolls_back_the_rows_already_applied() { use domain::ports::{FollowCommand, FollowWrite}; let (pool, user) = pool_with_user().await; let mut writes = follow_writes(user.id()); - // An unknown local user violates the ap_following foreign key after the - // follower row was already written. writes.push(FollowWrite::AddFollow { follower_id: uuid::Uuid::new_v4(), target_actor_url: "https://md.example/users/other".into(), diff --git a/crates/adapters/sqlite/tests/repositories/user_repository_test.rs b/crates/adapters/sqlite/tests/repositories/user_repository_test.rs index 9798efc9..7897f16b 100644 --- a/crates/adapters/sqlite/tests/repositories/user_repository_test.rs +++ b/crates/adapters/sqlite/tests/repositories/user_repository_test.rs @@ -120,7 +120,6 @@ async fn marking_a_user_deleted_hides_them_from_every_lookup() { repo.save(&admin).await.unwrap(); let other = test_user("other"); repo.save(&other).await.unwrap(); - // A second admin, so the first is not the last one standing. sqlx::query("UPDATE users SET role = 'admin' WHERE id = ?") .bind(other.id().value().to_string()) .execute(&pool) @@ -145,7 +144,6 @@ async fn marking_a_user_deleted_hides_them_from_every_lookup() { assert!(emails.contains(&other.email().value())); assert!(!emails.contains(&admin.email().value())); assert_eq!(repo.count_active_admins().await.unwrap(), 1); - // The purge can still read the tombstone; a live user is not one. assert!( repo.find_tombstoned_by_id(admin.id()) .await diff --git a/crates/adapters/tmdb-enrichment/src/client.rs b/crates/adapters/tmdb-enrichment/src/client.rs index 4ae720e6..11f0568a 100644 --- a/crates/adapters/tmdb-enrichment/src/client.rs +++ b/crates/adapters/tmdb-enrichment/src/client.rs @@ -12,7 +12,6 @@ impl TmdbEnrichmentClient { .tmdb_api_key .clone() .ok_or_else(|| DomainError::InfrastructureError("TMDB_API_KEY is not set".into()))?; - // reqwest is built with rustls-no-provider; install ring so this works without bootstrap. let _ = rustls::crypto::ring::default_provider().install_default(); Ok(Self { api_key, diff --git a/crates/adapters/tmdb-enrichment/tests/client_tls.rs b/crates/adapters/tmdb-enrichment/tests/client_tls.rs index 711abe61..3f554897 100644 --- a/crates/adapters/tmdb-enrichment/tests/client_tls.rs +++ b/crates/adapters/tmdb-enrichment/tests/client_tls.rs @@ -1,5 +1,3 @@ -//! Fresh test binary, no bootstrap: constructing the client must not panic -//! for lack of a rustls CryptoProvider. use tmdb_enrichment::TmdbEnrichmentClient; #[test] diff --git a/crates/api-types/src/common.rs b/crates/api-types/src/common.rs index e009b46a..0025fa13 100644 --- a/crates/api-types/src/common.rs +++ b/crates/api-types/src/common.rs @@ -22,11 +22,8 @@ pub type WatchlistResponse = PaginatedResponse; pub type PaginatedPersonHits = PaginatedResponse; -/// The one body every JSON API refusal carries. #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)] pub struct ErrorResponse { - /// Human-readable message. pub error: String, - /// Stable machine-readable code, e.g. `NOT_FOUND`. pub code: String, } diff --git a/crates/api-types/src/mappers/mod.rs b/crates/api-types/src/mappers/mod.rs index 20b918b2..3221636e 100644 --- a/crates/api-types/src/mappers/mod.rs +++ b/crates/api-types/src/mappers/mod.rs @@ -1,4 +1,3 @@ -//! Pure domain -> DTO mappers (depend only on `domain` and this crate). mod diary; mod goals; mod movies; diff --git a/crates/api-types/src/users.rs b/crates/api-types/src/users.rs index 4ccef828..b1266671 100644 --- a/crates/api-types/src/users.rs +++ b/crates/api-types/src/users.rs @@ -126,7 +126,6 @@ pub struct UserProfileResponse { pub goals: Option>, #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub is_federated: bool, - /// The instance's service account: never deletable, so UIs hide the action. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub is_service_account: bool, #[serde(skip_serializing_if = "Option::is_none")] @@ -156,7 +155,6 @@ pub struct UpdateProfileFieldsRequest { pub fields: Vec, } -/// Proof of intent for account deletion: the acting user's own password. #[derive(Debug, Clone, Deserialize, utoipa::ToSchema)] pub struct DeleteAccountRequest { pub password: String, diff --git a/crates/application/src/diary/delete_review.rs b/crates/application/src/diary/delete_review.rs index dcc878c2..e5573b46 100644 --- a/crates/application/src/diary/delete_review.rs +++ b/crates/application/src/diary/delete_review.rs @@ -20,7 +20,6 @@ pub async fn execute(deps: &DeleteReviewDeps, cmd: DeleteReviewCommand) -> Resul } let movie_id = review.movie_id().clone(); - // Read before the write: the orphan event carries the poster to clean up. let poster_path = deps .diary .get_review_history(&movie_id) @@ -29,7 +28,6 @@ pub async fn execute(deps: &DeleteReviewDeps, cmd: DeleteReviewCommand) -> Resul .poster_path() .cloned(); - // One write: the review, plus its movie when this was the last review. deps.review .delete_review_and_orphaned_movie( &review_id, diff --git a/crates/application/src/diary/review_logger.rs b/crates/application/src/diary/review_logger.rs index 7d95fd3c..7e5ed39e 100644 --- a/crates/application/src/diary/review_logger.rs +++ b/crates/application/src/diary/review_logger.rs @@ -76,7 +76,6 @@ impl ReviewLogger for DefaultReviewLogger { watched_at: *review.watched_at(), }; - // Events ride the review save (the decisive write) so they commit with it. let mut events = Vec::with_capacity(2); if let Some(ext_id) = movie.external_metadata_id() { events.push(DomainEvent::MovieEnrichmentRequested { @@ -87,7 +86,6 @@ impl ReviewLogger for DefaultReviewLogger { events.push(review_event); self.review_repo.save_review(&review, &events).await?; - // The removal event is recorded only if an entry was actually removed. self.watchlist_repo .remove_if_present( review.user_id(), diff --git a/crates/application/src/goals/create.rs b/crates/application/src/goals/create.rs index 7c5c0701..05a7815e 100644 --- a/crates/application/src/goals/create.rs +++ b/crates/application/src/goals/create.rs @@ -29,7 +29,6 @@ pub async fn execute( cmd.target_count, GoalType::Movies, )?; - // Reads that can fail happen before the write, so an error never follows a commit. let current_count = deps.stats.count_reviews_in_year(&user_id, cmd.year).await?; let event = DomainEvent::GoalCreated { goal_id: g.id().clone(), diff --git a/crates/application/src/goals/update.rs b/crates/application/src/goals/update.rs index 24151a85..52c2c5a9 100644 --- a/crates/application/src/goals/update.rs +++ b/crates/application/src/goals/update.rs @@ -17,7 +17,6 @@ pub async fn execute( .ok_or_else(|| DomainError::NotFound(format!("Goal for year {}", cmd.year)))?; g.update_target(cmd.target_count)?; - // Reads that can fail happen before the write, so an error never follows a commit. let current_count = deps.stats.count_reviews_in_year(&user_id, cmd.year).await?; let event = DomainEvent::GoalUpdated { goal_id: g.id().clone(), diff --git a/crates/application/src/import/execute.rs b/crates/application/src/import/execute.rs index 808e3276..f29bb51c 100644 --- a/crates/application/src/import/execute.rs +++ b/crates/application/src/import/execute.rs @@ -88,7 +88,6 @@ pub async fn execute( }) } -/// Row-level validation failures keep their bare message in the summary. fn import_error_message(e: DomainError) -> String { match e { DomainError::ValidationError(msg) => msg, diff --git a/crates/application/src/jobs/processed_events_cleanup.rs b/crates/application/src/jobs/processed_events_cleanup.rs index 7e83302d..e85e0563 100644 --- a/crates/application/src/jobs/processed_events_cleanup.rs +++ b/crates/application/src/jobs/processed_events_cleanup.rs @@ -7,8 +7,6 @@ use domain::{ ports::{PeriodicJob, ProcessedEventLedger}, }; -/// Ledger rows older than this can no longer matter: the queue gives up on an -/// event long before, so it cannot be redelivered. pub const PROCESSED_EVENTS_RETENTION_DAYS: i64 = 30; pub struct ProcessedEventsCleanupJob { diff --git a/crates/application/src/jobs/tombstone_cleanup.rs b/crates/application/src/jobs/tombstone_cleanup.rs index 76952d4e..d8816136 100644 --- a/crates/application/src/jobs/tombstone_cleanup.rs +++ b/crates/application/src/jobs/tombstone_cleanup.rs @@ -7,9 +7,6 @@ use domain::{ ports::{AccountPurger, PeriodicJob}, }; -/// How long a purged account's stripped user row and signing keypair stay. -/// Federation deliveries of the account's `Delete` are signed from them, and -/// their retries are long over by then. pub const TOMBSTONE_RETENTION_DAYS: i64 = 7; pub struct TombstoneCleanupJob { diff --git a/crates/application/src/movies/enrich_movie.rs b/crates/application/src/movies/enrich_movie.rs index c0510e20..caa08302 100644 --- a/crates/application/src/movies/enrich_movie.rs +++ b/crates/application/src/movies/enrich_movie.rs @@ -14,8 +14,6 @@ pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result< } let Some(movie) = deps.movie_query.get_movie_by_id(&cmd.movie_id).await? else { - // The movie row is written with the event, so a missing row means it was - // deleted. Do not store the profile: it is the "enrichment done" marker. tracing::warn!(movie_id = %cmd.movie_id.value(), "enrich_movie: movie not found, skipping"); return Ok(()); }; @@ -37,8 +35,6 @@ pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result< .await?; } - // Written last: a stored fresh profile is what makes a redelivered - // enrichment skip, so it must only exist once every other effect landed. deps.movie_profile.upsert(&cmd.profile).await?; tracing::info!( diff --git a/crates/application/src/movies/poster_sync_handler.rs b/crates/application/src/movies/poster_sync_handler.rs index 1f9e4620..e63323d5 100644 --- a/crates/application/src/movies/poster_sync_handler.rs +++ b/crates/application/src/movies/poster_sync_handler.rs @@ -110,10 +110,6 @@ impl EventHandler for PosterSyncHandler { movie_id, external_metadata_id, } => { - // The poster is stored under the movie id (the converter later - // appends `.webp`/`.avif`) and the movie row is updated - // atomically with `PosterSynced`; exactly that path on the movie - // means an earlier delivery completed. let id_str = movie_id.value().to_string(); let canonical = [ id_str.clone(), diff --git a/crates/application/src/social/execute.rs b/crates/application/src/social/execute.rs index a773da24..0b428bf5 100644 --- a/crates/application/src/social/execute.rs +++ b/crates/application/src/social/execute.rs @@ -2,8 +2,6 @@ use domain::{errors::DomainError, events::DomainEvent, value_objects::UserId}; use super::{commands::SocialCmd, deps::SocialCommandDeps}; -/// The command's event rides the write: the social port records it with the -/// change (atomically where the adapter owns the transaction). pub async fn execute_command(deps: &SocialCommandDeps, cmd: SocialCmd) -> Result<(), DomainError> { let command = &deps.social_command; match cmd { diff --git a/crates/application/src/users/ensure_active.rs b/crates/application/src/users/ensure_active.rs index 6a448247..03411f6c 100644 --- a/crates/application/src/users/ensure_active.rs +++ b/crates/application/src/users/ensure_active.rs @@ -2,8 +2,6 @@ use domain::{errors::DomainError, value_objects::UserId}; use crate::users::deps::EnsureActiveDeps; -/// A token outlives its account; this is the check that stops a deleted -/// (or vanished) user from acting on one. pub async fn execute(deps: &EnsureActiveDeps, user: &UserId) -> Result<(), DomainError> { match deps.user.find_by_id(user).await? { Some(_) => Ok(()), diff --git a/crates/application/src/users/purge_account.rs b/crates/application/src/users/purge_account.rs index 3a4d6550..b42e2a7b 100644 --- a/crates/application/src/users/purge_account.rs +++ b/crates/application/src/users/purge_account.rs @@ -9,21 +9,6 @@ use domain::{ }, }; -/// Purges a deleted account once its `UserDeleted` event arrives. -/// -/// Order matters. Federation is told first (deliveries are only enqueued, and -/// signed later from the tombstone the purge leaves behind). A notifier -/// failure never blocks the purge: unreachable servers must not keep a -/// person's data. Then the database rows go in one transaction, and the -/// image objects after the commit. Image references are dropped only once the -/// objects are gone, so a crash in between is recovered by the redelivery. -/// -/// Idempotent: every step is a no-op on a re-run, and the notification is -/// skipped once the account is purged. -/// -/// A purge that fails after the notification was queued is redelivered and -/// queues the `Delete` again. That is harmless: a repeated `Delete` of an -/// already deleted actor is idempotent on remote servers. pub struct AccountPurgeHandler { user: Arc, notifier: Arc, @@ -57,7 +42,6 @@ impl EventHandler for AccountPurgeHandler { let DomainEvent::UserDeleted { user_id } = event else { return Ok(()); }; - // Gone already (removed by the tombstone cleanup): nothing left to do. if self.user.find_tombstoned_by_id(user_id).await?.is_none() { return Ok(()); } diff --git a/crates/application/src/users/request_account_deletion.rs b/crates/application/src/users/request_account_deletion.rs index c018b83b..f91e26f2 100644 --- a/crates/application/src/users/request_account_deletion.rs +++ b/crates/application/src/users/request_account_deletion.rs @@ -2,10 +2,6 @@ use domain::{errors::DomainError, models::UserRole, value_objects::UserId}; use crate::users::deps::RequestAccountDeletionDeps; -/// Tombstones `target`. A user may delete themselves; an admin may delete -/// anyone. Either way the *actor's* password is the proof, and the last -/// active admin can never be removed. `UserDeleted` rides the repository -/// write, so the tombstone and its event commit together. pub async fn execute( deps: &RequestAccountDeletionDeps, actor: UserId, diff --git a/crates/application/src/worker.rs b/crates/application/src/worker.rs index 9288c2e8..6279ab7d 100644 --- a/crates/application/src/worker.rs +++ b/crates/application/src/worker.rs @@ -19,8 +19,6 @@ pub struct WorkerService { } impl WorkerService { - /// Fails if two handlers share a name: the name keys the processed-event - /// ledger, so a collision would make one handler's success skip the other. pub fn new( consumer: Arc, handlers: Vec>, @@ -89,8 +87,6 @@ impl WorkerService { } } -/// Margin added to a claim's remaining lease before a blocked delivery is -/// retried, so the retry lands after the lease has lapsed rather than on it. const DEFER_MARGIN: Duration = Duration::from_secs(5); async fn dispatch( @@ -100,7 +96,6 @@ async fn dispatch( ) { let id = envelope.id().copied(); let mut any_transient = false; - // Longest wait among handlers blocked behind another delivery's live claim. let mut blocked_for: Option = None; for handler in handlers.iter() { let name = handler.name(); @@ -118,7 +113,6 @@ async fn dispatch( continue; } Err(e) => { - // Running anyway could repeat a side effect; retry the delivery instead. tracing::warn!(handler = name, "ledger claim failed, nacking delivery: {e}"); any_transient = true; continue; @@ -131,15 +125,10 @@ async fn dispatch( if let (Some(id), Some(token)) = (&id, &claim) && let Err(e) = ledger.complete(id, name, token).await { - // The work is done; failing the delivery would only repeat it. - // The claim lapses after its lease and the handler may run once more. tracing::warn!(handler = name, "completing ledger claim failed: {e}"); } } Err(e) => { - // Permanent failures are released too, not recorded: if a sibling - // handler's transient failure redelivers this event, a handler that - // failed permanently runs (and fails) again. if let (Some(id), Some(token)) = (&id, &claim) && let Err(le) = ledger.release(id, name, token).await { @@ -160,10 +149,6 @@ async fn dispatch( } } } - // A blocked handler cannot be retried before its claim is done or lapsed, - // so wait that long without spending a retry attempt; any transient failure - // in the same delivery is retried then too. A plain nack would burn the - // retry budget (and dead-letter) long before a crashed claim's lease ends. if let Some(wait) = blocked_for { if let Err(e) = envelope.defer(wait + DEFER_MARGIN).await { tracing::error!("defer failed: {e}"); diff --git a/crates/application/src/wrapup/handle_requested.rs b/crates/application/src/wrapup/handle_requested.rs index 0332f5a8..6709fff9 100644 --- a/crates/application/src/wrapup/handle_requested.rs +++ b/crates/application/src/wrapup/handle_requested.rs @@ -17,9 +17,6 @@ pub async fn execute( tracing::debug!("wrapup {} already ready, skipping", wrapup_id.value()); return Ok(()); } - // A record still `Generating` is an earlier delivery that died mid-way. The - // ledger guarantees no concurrent run, so resume it: generation is a pure - // read of stats and `set_complete` is the single atomic completion write. deps.wrapup_repo .update_status(&wrapup_id, &WrapUpStatus::Generating, None) diff --git a/crates/application/tests/auth/register_test.rs b/crates/application/tests/auth/register_test.rs index cb93db58..cda0b887 100644 --- a/crates/application/tests/auth/register_test.rs +++ b/crates/application/tests/auth/register_test.rs @@ -90,7 +90,6 @@ async fn registration_refuses_a_retired_username() { .unwrap() .unwrap(); users.mark_deleted(old.id(), &[]).await.unwrap(); - // The purge removes the row; only the retired name keeps the username taken. users.delete_user(old.id()); let err = register::execute(&deps, register_command("new@example.com", "Ghost")) diff --git a/crates/application/tests/diary/edit_review_test.rs b/crates/application/tests/diary/edit_review_test.rs index 102bb43b..328ddb69 100644 --- a/crates/application/tests/diary/edit_review_test.rs +++ b/crates/application/tests/diary/edit_review_test.rs @@ -78,7 +78,6 @@ async fn edit_other_users_review_returns_forbidden() { assert!(matches!(err, domain::errors::DomainError::Forbidden(_))); } -/// Loads from the store but refuses every write. pub(crate) struct WriteRefusingReviews(pub(crate) Arc); #[async_trait::async_trait] diff --git a/crates/application/tests/handlers/idempotency_test.rs b/crates/application/tests/handlers/idempotency_test.rs index 0028e324..c77aac7d 100644 --- a/crates/application/tests/handlers/idempotency_test.rs +++ b/crates/application/tests/handlers/idempotency_test.rs @@ -1,6 +1,3 @@ -//! Every event handler may run twice for the same event (lease takeover after -//! a crash mid-handler). Running it twice must leave the world as if it ran once. - use std::collections::HashMap; use std::sync::{ Arc, Mutex, @@ -36,7 +33,6 @@ use application::person::event_handler::PersonEnrichmentHandler; // ── fakes ─────────────────────────────────────────────────────────────────── -/// Search index keyed like a real one: indexing the same id twice is one doc. #[derive(Default)] struct RecordingSearch { docs: Mutex>, @@ -79,7 +75,6 @@ impl SearchCommand for RecordingSearch { } } -/// Fails the first `index` call, succeeds afterwards. struct FlakySearch { inner: Arc, failed_once: AtomicUsize, @@ -103,7 +98,6 @@ fn person(tmdb: u64, name: &str) -> Person { Person::basic(PersonId::from_external(&ext), ext, name.into(), None, None) } -/// One person whose `enriched_at` is set by `update_enrichment`, like the real store. struct FakePersons { person: Mutex, enrich_writes: AtomicUsize, @@ -239,7 +233,6 @@ impl domain::ports::ImageFetcher for NoImages { } } -/// Returns the key it was given, like the real adapter. struct KeyStorage; #[async_trait] @@ -371,8 +364,6 @@ async fn a_movie_enrichment_that_failed_midway_is_redone_on_redelivery() { }; assert!(handler.handle(&event).await.is_err()); - // The profile row is the "enrichment done" marker: it must not exist yet, - // or the redelivery below would skip and the search index would stay empty. assert!(store.get_by_movie_id(movie.id()).await.unwrap().is_none()); handler.handle(&event).await.unwrap(); @@ -517,13 +508,11 @@ async fn poster_sync_skips_converted_poster_keys_but_not_lookalike_paths() { external_metadata_id: ext, }; - // The converter's output is the canonical synced key: skipped. movie.update_poster(PosterPath::new(format!("{id}.webp")).unwrap()); store.upsert_movie(&movie, &[]).await.unwrap(); handler.handle(&event).await.unwrap(); assert!(publisher.published().is_empty()); - // A foreign path that merely contains the id is not a synced poster. movie.update_poster(PosterPath::new(format!("https://cdn.example/{id}/p.jpg")).unwrap()); store.upsert_movie(&movie, &[]).await.unwrap(); handler.handle(&event).await.unwrap(); diff --git a/crates/application/tests/support/commands.rs b/crates/application/tests/support/commands.rs index 9cf0d28b..4a70acc2 100644 --- a/crates/application/tests/support/commands.rs +++ b/crates/application/tests/support/commands.rs @@ -1,7 +1,3 @@ -//! Command builders for tests. Each builder returns a command with sensible -//! defaults; tests override individual fields with struct-update syntax: -//! `LogReviewCommand { rating: 6, ..log_review_command(user, movie) }`. - use chrono::NaiveDate; use uuid::Uuid; @@ -66,7 +62,6 @@ pub fn log_review_command_manual(user_id: Uuid, title: &str, year: u16) -> LogRe } } -/// Edits only the rating; `review_id` and `requesting_user_id` are random. pub fn edit_review_command(rating: u8) -> EditReviewCommand { EditReviewCommand { review_id: Uuid::new_v4(), @@ -115,7 +110,6 @@ pub fn update_goal_command(year: u16, target_count: u32) -> UpdateGoalCommand { } } -/// Confirms indices `[0]`. pub fn execute_import_command(user_id: Uuid, session_id: Uuid) -> ExecuteImportCommand { ExecuteImportCommand { user_id, @@ -132,7 +126,6 @@ pub fn create_import_session_command(user_id: Uuid, csv: &[u8]) -> CreateImportS } } -/// No field mappings. pub fn apply_import_mapping_command(user_id: Uuid, session_id: Uuid) -> ApplyImportMappingCommand { ApplyImportMappingCommand { user_id, @@ -153,7 +146,6 @@ pub fn apply_import_profile_command( } } -/// One confirmation, rating as given, no comment. pub fn confirm_watch_events_command( user_id: Uuid, watch_event_id: Uuid, @@ -176,7 +168,6 @@ pub fn dismiss_watch_events_command( DismissWatchEventsCommand { user_id, event_ids } } -/// Jellyfin provider, no label. pub fn generate_webhook_token_command(user_id: Uuid) -> GenerateWebhookTokenCommand { GenerateWebhookTokenCommand { user_id, @@ -185,7 +176,6 @@ pub fn generate_webhook_token_command(user_id: Uuid) -> GenerateWebhookTokenComm } } -/// Every field unset; override what the test cares about. pub fn update_profile_command(user_id: Uuid) -> UpdateProfileCommand { UpdateProfileCommand { user_id, @@ -199,7 +189,6 @@ pub fn update_profile_command(user_id: Uuid) -> UpdateProfileCommand { } } -/// All federation toggles on. pub fn update_user_settings_command(user_id: Uuid) -> UpdateUserSettingsCommand { UpdateUserSettingsCommand { user_id, @@ -216,7 +205,6 @@ pub fn add_to_watchlist_command(user_id: Uuid, movie_id: Uuid) -> AddToWatchlist } } -/// Window is 1 Jan of `year` to 1 Jan of `year + 1`. pub fn request_wrap_up_command(user_id: Option, year: i32) -> RequestWrapUpCommand { RequestWrapUpCommand { user_id, diff --git a/crates/application/tests/support/mod.rs b/crates/application/tests/support/mod.rs index fd6d9819..8c1464e8 100644 --- a/crates/application/tests/support/mod.rs +++ b/crates/application/tests/support/mod.rs @@ -1,6 +1,3 @@ -//! Shared fakes and builders for the application integration tests. -//! Included via `mod support;` from each `tests/.rs`; not every -//! binary uses every helper. #![allow(dead_code, unused_imports)] mod commands; diff --git a/crates/application/tests/users/purge_account_test.rs b/crates/application/tests/users/purge_account_test.rs index e29f6b6c..24652256 100644 --- a/crates/application/tests/users/purge_account_test.rs +++ b/crates/application/tests/users/purge_account_test.rs @@ -139,7 +139,6 @@ fn entries(log: &Log) -> Vec { #[test] fn the_handler_name_is_stable() { - // The name keys the processed-event ledger; renaming re-runs events. let f = tokio::runtime::Runtime::new() .unwrap() .block_on(fixture(false, true)); @@ -255,9 +254,6 @@ async fn a_redelivered_event_is_skipped_by_the_ledger() { assert_eq!(purges, 1); } -/// The rows were committed but the worker died before deleting the image -/// objects: the redelivery finds the account purged, still sees the image -/// references, deletes the objects and clears the references. #[tokio::test] async fn a_rerun_after_a_crash_following_the_commit_deletes_the_images() { let f = fixture(false, true).await; diff --git a/crates/application/tests/users/request_account_deletion_test.rs b/crates/application/tests/users/request_account_deletion_test.rs index 94eb8b0c..f0c5e5e9 100644 --- a/crates/application/tests/users/request_account_deletion_test.rs +++ b/crates/application/tests/users/request_account_deletion_test.rs @@ -21,8 +21,6 @@ use domain::models::{User, UserSummary}; use domain::ports::UserRepository as UserRepositoryPort; use domain::value_objects::Username; -/// Delegates to a store but reports a stale admin count, as a concurrent -/// deletion would leave the use case's pre-check. struct StaleAdminCount(Arc); #[async_trait] diff --git a/crates/application/tests/users/update_profile_test.rs b/crates/application/tests/users/update_profile_test.rs index 2f22e6df..cb92f595 100644 --- a/crates/application/tests/users/update_profile_test.rs +++ b/crates/application/tests/users/update_profile_test.rs @@ -273,7 +273,6 @@ async fn changing_also_known_as_records_the_move_with_the_update() { assert!(matches!(recorded[1], DomainEvent::UserAccountMoved { .. })); } -/// Loads from the store but refuses the profile write. struct ProfileWriteRefusingUsers(Arc); #[async_trait::async_trait] diff --git a/crates/application/tests/worker/worker_test.rs b/crates/application/tests/worker/worker_test.rs index 34a24e64..399cbb91 100644 --- a/crates/application/tests/worker/worker_test.rs +++ b/crates/application/tests/worker/worker_test.rs @@ -377,7 +377,6 @@ impl EventHandler for CountingHandler { } } -/// Delivers the same event (same id) `times` times, recording ack/nack per delivery. struct RedeliveringConsumer { id: Option, event: DomainEvent, @@ -416,8 +415,6 @@ impl EventConsumer for RedeliveringConsumer { Box::new(CountNack(Arc::clone(&nacks), Arc::clone(&defers))), ); async move { - // Let the previous delivery's handlers finish first: a real - // broker redelivers only after the nack. tokio::time::sleep(std::time::Duration::from_millis(50)).await; Ok(envelope) } @@ -498,7 +495,6 @@ async fn a_handler_already_running_elsewhere_is_not_run_concurrently() { let (handler, calls) = CountingHandler::new("slow", 0); let store = InMemoryStore::new(); let id = domain::events::EventId::new(); - // Another delivery holds a live claim on this (event, handler). assert!(matches!( store.begin(&id, "slow").await.unwrap(), Begin::Run(_) @@ -560,7 +556,6 @@ async fn a_delivery_blocked_by_a_live_claim_is_deferred_without_consuming_an_att assert_eq!(*nacks.lock().unwrap(), 0, "a nack would consume an attempt"); let defers = defers.lock().unwrap(); assert_eq!(defers.len(), 1); - // Retried only once the claim's lease has run out (plus a margin). let lease = std::time::Duration::from_secs(HANDLER_LEASE_SECONDS as u64); assert!( defers[0] > lease - std::time::Duration::from_secs(5), @@ -588,7 +583,6 @@ async fn a_lapsed_owner_cannot_release_a_claim_taken_over_by_another() { let Begin::Run(lapsed) = store.begin(&id, "h").await.unwrap() else { panic!("stale claim should be taken over"); }; - // Simulate: the original owner's token is from before the takeover. let original = domain::ports::ClaimToken::new("2020-01-01T00:00:00.000Z".into()); assert_ne!(original, lapsed); @@ -652,7 +646,6 @@ async fn a_failed_handler_releases_its_claim_so_redelivery_retries() { .run(tokio::sync::watch::channel(false).1) .await; - // Delivery 1 fails and releases; 2 succeeds; 3 is skipped as done. assert_eq!(*calls.lock().unwrap(), 2); assert_eq!(*nacks.lock().unwrap(), 1); } diff --git a/crates/application/tests/wrapup/handle_requested_test.rs b/crates/application/tests/wrapup/handle_requested_test.rs index ed69f321..0ca398ce 100644 --- a/crates/application/tests/wrapup/handle_requested_test.rs +++ b/crates/application/tests/wrapup/handle_requested_test.rs @@ -82,7 +82,6 @@ async fn generates_wrapup_and_marks_complete() { assert_eq!(final_rec.status, WrapUpStatus::Ready); assert!(final_rec.report.is_some()); - // The completion event is recorded together with the completed row. let recorded = repo.recorded_events(); assert_eq!(recorded.len(), 1); assert!(matches!( @@ -93,8 +92,6 @@ async fn generates_wrapup_and_marks_complete() { #[tokio::test] async fn a_wrapup_left_generating_is_completed_on_redelivery() { - // A crash after `Generating` was written (or a failed `set_complete`) leaves - // the record stuck; the redelivered event must finish it, not skip it. let repo = InMemoryWrapUpRepository::new(); let wrapup_id = WrapUpId::generate(); diff --git a/crates/bootstrap/src/database.rs b/crates/bootstrap/src/database.rs index eb795d6f..fcfab37f 100644 --- a/crates/bootstrap/src/database.rs +++ b/crates/bootstrap/src/database.rs @@ -58,8 +58,6 @@ pub async fn build_database_adapters( assemble(w, instance) } -/// Like [`build_database_adapters`], on a pool the caller already connected -/// (migrations still run). Lets tests drive the real wiring on an in-memory pool. pub async fn build_database_adapters_on( pool: sqlx::SqlitePool, instance: &domain::value_objects::InstanceIdentity, diff --git a/crates/bootstrap/src/event_bus.rs b/crates/bootstrap/src/event_bus.rs index ba7ded83..da337d30 100644 --- a/crates/bootstrap/src/event_bus.rs +++ b/crates/bootstrap/src/event_bus.rs @@ -27,8 +27,6 @@ impl EventBusBackend { } } -/// The server's publisher is always the outbox writer: it never talks to the -/// bus, whatever `EVENT_BUS_BACKEND` says. The worker owns delivery. pub async fn create_publisher( db_pool: &sqlx::SqlitePool, config: &config::AppConfig, @@ -37,18 +35,12 @@ pub async fn create_publisher( sqlite::SqliteEventQueue::create_publisher(db_pool.clone(), &config.events).await } -/// What the worker gets from the bus: a publisher for its own events, the -/// consumer its handlers read, and, when the bus is not the outbox, the relay -/// that moves outbox rows onto it. struct Wiring { publisher: Arc, consumer: Arc, relay: Option, } -/// The worker always writes to the outbox. With an external bus -/// (`bus = Some((consumer, sink))`) handlers read from the bus and the relay -/// drains the outbox into it; without one, handlers drain the outbox directly. fn compose( (publisher, outbox): (Arc, Arc), bus: Option<(Arc, Arc)>, diff --git a/crates/bootstrap/src/federation.rs b/crates/bootstrap/src/federation.rs index 64aba88d..801bb45d 100644 --- a/crates/bootstrap/src/federation.rs +++ b/crates/bootstrap/src/federation.rs @@ -1,8 +1,3 @@ -//! Social and federation wiring for the server and worker. -//! -//! Federation off still wires real local social behaviour (`LocalSocialService`): -//! the follow graph and block queries must never silently report empty. - use std::sync::Arc; use domain::ports::{ @@ -123,9 +118,6 @@ pub async fn wire_server_social( }) } -/// The worker's federation event handlers (the ActivityPub delivery handler and -/// the follow-accepted outbox backfill) and the notifier that tells followers -/// an account was deleted. #[cfg(feature = "federation")] pub async fn wire_worker_handlers( db: &DatabaseOutput, diff --git a/crates/bootstrap/src/follow_backfill_handler.rs b/crates/bootstrap/src/follow_backfill_handler.rs index 96f91c2f..bd3db1b1 100644 --- a/crates/bootstrap/src/follow_backfill_handler.rs +++ b/crates/bootstrap/src/follow_backfill_handler.rs @@ -62,11 +62,6 @@ mod tests { use super::*; - /// The handler has no state of its own: idempotency lives behind the port - /// (remote imports are keyed upserts; backfill Creates use the deterministic - /// k-ap id `activities/create/{uuidv5(object id)}`, see activitypub - /// `tests/activity_ids.rs`). What this proves is that a redelivery replays - /// exactly the same port calls, with no accumulated or derived state. #[derive(Default)] struct CallLog(Mutex>); diff --git a/crates/bootstrap/src/lib.rs b/crates/bootstrap/src/lib.rs index 812adf63..c81e81c0 100644 --- a/crates/bootstrap/src/lib.rs +++ b/crates/bootstrap/src/lib.rs @@ -1,9 +1,3 @@ -//! The composition root: the only crate that names concrete adapters. -//! -//! `server` and `worker` are thin binaries over [`build_server`] and -//! [`build_worker`]; everything between configuration and a running service is -//! assembled here (ADR-0005, ADR-0013). - mod build; mod database; mod event_bus; diff --git a/crates/bootstrap/src/outbox_relay.rs b/crates/bootstrap/src/outbox_relay.rs index 5d3f99cd..9c087e4a 100644 --- a/crates/bootstrap/src/outbox_relay.rs +++ b/crates/bootstrap/src/outbox_relay.rs @@ -1,22 +1,3 @@ -//! Moves outbox rows onto the message bus. The worker runs it when the bus is -//! not the outbox itself (`EVENT_BUS_BACKEND=nats`). -//! -//! Rows are claimed by the SQLite queue's own atomic claim, so a row is -//! relayed by one claimant at a time, and failures take the queue's backoff -//! and dead-letter path. A row is marked done only after the sink reports the -//! bus accepted it. -//! -//! A sink (transport) failure is the bus's fault, not the row's: the row is -//! deferred without spending an attempt, and the relay pauses with a growing -//! backoff instead of churning through the rest of the outbox. Attempts, and -//! so dead-lettering, are only ever spent by consumers on handler failures. -//! -//! Delivery is at-least-once. A crash between publish and mark-done leaves the -//! row `processing`; when its claim lease lapses it is claimed and relayed -//! again. The bus drops that repeat while it is inside the broker's dedup -//! window (same `Nats-Msg-Id`); beyond the window, or in core mode, the -//! consumer ledger makes the handlers skip it. - use std::{sync::Arc, time::Duration}; use domain::{ @@ -41,16 +22,11 @@ impl OutboxRelay { } } - /// Overrides the pause/defer schedule, keyed by consecutive sink failures. pub fn with_backoff(mut self, backoff: fn(u32) -> Duration) -> Self { self.backoff = backoff; self } - /// Runs until `shutdown` fires or the source ends. A shutdown never - /// interrupts a row mid-relay: it is acked or nacked before this returns. - /// Rows the claim already took but the relay had not started stay - /// `processing` and come back when their lease lapses. pub async fn run(self, mut shutdown: watch::Receiver) { let mut rows = self.source.consume(); let mut sink_failures: u32 = 0; @@ -80,7 +56,6 @@ impl OutboxRelay { } } } - // The queue already dead-lettered it and logged the row id. Some(Err(e)) => tracing::error!("outbox relay: skipped undecodable row: {e}"), None => { tracing::error!("outbox relay: source closed"); @@ -92,7 +67,6 @@ impl OutboxRelay { } } -/// Returns how long the relay should pause when the sink failed. async fn relay_envelope( envelope: EventEnvelope, sink: &dyn EventRelaySink, diff --git a/crates/bootstrap/src/server.rs b/crates/bootstrap/src/server.rs index 6cb49955..604ef088 100644 --- a/crates/bootstrap/src/server.rs +++ b/crates/bootstrap/src/server.rs @@ -19,7 +19,6 @@ pub struct ServerApp { pub addr: SocketAddr, } -/// Wires the whole HTTP application from configuration. pub async fn build_server(config: &config::AppConfig) -> anyhow::Result { database::warn_if_database_backend_set(); let instance = InstanceIdentity::new(config.server.base_url.clone()); @@ -28,7 +27,6 @@ pub async fn build_server(config: &config::AppConfig) -> anyhow::Result anyhow::Result, } impl WorkerApp { - /// Names of the registered event handlers, in registration order. pub fn handler_names(&self) -> Vec<&'static str> { self.service.handler_names() } - /// Runs the handlers and, when there is one, the outbox relay, until - /// `shutdown` fires. Both stop together and finish their in-flight work. - /// - /// If the relay stops or panics on its own, the worker stops too and - /// returns an error so the process exits non-zero and the supervisor - /// restarts it: a worker that keeps handling events but no longer - /// delivers the outbox would fail silently. pub async fn run(self, mut shutdown: watch::Receiver) -> anyhow::Result<()> { let (stop_tx, stop_rx) = watch::channel(false); let mut relay = self.relay.map(|r| tokio::spawn(r.run(stop_rx.clone()))); @@ -74,8 +65,6 @@ impl WorkerApp { } } -/// Wires the background worker: event handlers, periodic jobs (already -/// spawned) and the consumer. Run it with `service.run(shutdown)`. pub async fn build_worker(config: &config::AppConfig) -> anyhow::Result { database::warn_if_database_backend_set(); let instance = InstanceIdentity::new(config.server.base_url.clone()); @@ -83,8 +72,6 @@ pub async fn build_worker(config: &config::AppConfig) -> anyhow::Result Router { support::server().await.0 } -/// An app with one registered user; returns its pool, id and bearer token. async fn test_app_with_user() -> (Router, SqlitePool, Uuid, String) { let (app, pool) = support::server().await; let session = support::register(&app, "insights", "insights@example.com").await; diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 175f672c..c74ea805 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/crates/bootstrap/tests/bootstrap_test.rs @@ -1,5 +1,3 @@ -//! The composition root assembles on a fresh database, with every adapter real. - mod support; #[tokio::test] @@ -21,9 +19,6 @@ async fn worker_builds_on_a_fresh_database() { .expect("worker builds"); } -/// Without the federation feature, following another local user still works -/// (the request lands as pending for the target): -/// the local social ports are wired in place of the ActivityPub ones. #[cfg(not(feature = "federation"))] #[tokio::test] async fn federation_off_build_wires_local_social() { @@ -68,8 +63,6 @@ async fn federation_off_build_wires_local_social() { ); } -/// The server never talks to the bus: whatever `EVENT_BUS_BACKEND` says, a -/// change and its event land in the outbox table. No NATS server is reachable here. #[tokio::test] async fn server_publishes_to_the_outbox_even_when_bus_is_nats() { let pool = support::memory_pool().await; @@ -108,8 +101,6 @@ async fn server_publishes_to_the_outbox_even_when_bus_is_nats() { assert_eq!(queued, 1); } -/// Handler names key the processed-event ledger, so each must be unique. The -/// check runs over the real wiring, in both federation builds. #[tokio::test] async fn handler_names_are_unique() { let pool = support::memory_pool().await; @@ -126,8 +117,6 @@ async fn handler_names_are_unique() { ); let mut sorted = names.clone(); sorted.sort_unstable(); - // Names key the ledger: a rename re-runs events inside the retention - // window, so changing this list must be a deliberate edit. let mut expected = vec![ "account_purge", "image_cleanup", @@ -143,7 +132,6 @@ async fn handler_names_are_unique() { expected.extend(["activitypub", "follow_backfill"]); } expected.sort_unstable(); - // Image conversion ("image_conversion") is disabled in the test config. assert_eq!(sorted, expected); } @@ -157,8 +145,6 @@ async fn queued(pool: &sqlx::SqlitePool, event_type: &str) -> i64 { .unwrap() } -/// A local follow writes both follow rows and its event in one commit, in -/// both federation builds. #[tokio::test] async fn a_local_follow_lands_with_its_event_in_the_outbox() { let (router, pool) = support::server().await; @@ -189,8 +175,6 @@ async fn a_local_follow_lands_with_its_event_in_the_outbox() { assert_eq!(queued(&pool, "FollowRequested").await, 1); } -/// The k_ap path owns its transaction, so the adapter publishes the event -/// itself once the k_ap write has succeeded. #[cfg(feature = "federation")] #[tokio::test] async fn a_remote_block_publishes_its_event_after_the_k_ap_write() { @@ -217,9 +201,6 @@ async fn a_remote_block_publishes_its_event_after_the_k_ap_write() { assert_eq!(queued(&pool, "ActorBlocked").await, 1); } -/// A token outlives the account that owns it, so every authenticated route -/// must re-check the account: a deleted user gets exactly what an anonymous -/// caller gets. #[tokio::test] async fn a_deleted_users_token_is_rejected() { let (router, pool) = support::server().await; @@ -255,9 +236,6 @@ async fn a_deleted_users_token_is_rejected() { assert_eq!(headers.get("location").unwrap(), "/login"); } -/// A deleted account stays resolvable inside k-ap (to sign its `Delete`), but -/// every ActivityPub document for it answers 410, while the classic profile -/// page stays 404. #[cfg(feature = "federation")] #[tokio::test] async fn actor_document_for_a_tombstone_is_gone() { @@ -301,8 +279,6 @@ async fn actor_document_for_a_tombstone_is_gone() { assert_eq!(status, 404, "the classic profile page stays 404"); } -/// Remote servers posting to a deleted account's inbox are told it is gone, -/// before any signature check. #[cfg(feature = "federation")] #[tokio::test] async fn posting_to_a_tombstones_inbox_is_gone() { diff --git a/crates/bootstrap/tests/config_propagation_test.rs b/crates/bootstrap/tests/config_propagation_test.rs index f7a4af4f..3f7f21ac 100644 --- a/crates/bootstrap/tests/config_propagation_test.rs +++ b/crates/bootstrap/tests/config_propagation_test.rs @@ -1,5 +1,3 @@ -//! Configuration values reach the running app through the composition root. - mod support; use axum::{ diff --git a/crates/bootstrap/tests/http_handlers/api_handlers_test.rs b/crates/bootstrap/tests/http_handlers/api_handlers_test.rs index 700725e2..62bf6074 100644 --- a/crates/bootstrap/tests/http_handlers/api_handlers_test.rs +++ b/crates/bootstrap/tests/http_handlers/api_handlers_test.rs @@ -395,8 +395,6 @@ async fn following_collection_returns_500_on_federation_error() { assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR); } -// --- One refusal shape --- - async fn refusal_body(resp: axum::response::Response) -> serde_json::Value { let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await diff --git a/crates/bootstrap/tests/http_handlers/extractors_test.rs b/crates/bootstrap/tests/http_handlers/extractors_test.rs index 15d428a2..cdd4696b 100644 --- a/crates/bootstrap/tests/http_handlers/extractors_test.rs +++ b/crates/bootstrap/tests/http_handlers/extractors_test.rs @@ -246,7 +246,6 @@ async fn admin_api_user_rejects_an_unknown_user() { ) .await .unwrap(); - // An account the store does not hold is inactive, so it is unauthenticated. assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); } @@ -338,7 +337,6 @@ async fn admin_user_rejects_an_unknown_user() { ) .await .unwrap(); - // An account the store does not hold is inactive: back to the login page. assert_eq!(resp.status(), StatusCode::SEE_OTHER); assert_eq!(resp.headers().get("location").unwrap(), "/login"); } diff --git a/crates/bootstrap/tests/http_handlers/support.rs b/crates/bootstrap/tests/http_handlers/support.rs index e53faeb1..8beb3235 100644 --- a/crates/bootstrap/tests/http_handlers/support.rs +++ b/crates/bootstrap/tests/http_handlers/support.rs @@ -1,5 +1,3 @@ -//! `AppState` fixtures: `Panic*` stubs everywhere, real prebuilt `deps` built through -//! `bootstrap::build_deps`, with per-port overrides injected BEFORE the deps are built. #![allow(dead_code)] use application::config::AppSettings; @@ -25,8 +23,6 @@ use domain::{ use http_axum::context::AppContext; use std::sync::Arc; -// --- Local stubs --- - struct PanicReviewLogger; #[async_trait::async_trait] @@ -168,9 +164,6 @@ pub fn test_app_state_with_search_port( build_test_state_from(repos, Arc::new(PanicAuthService)) } -/// Test state whose repositories are adjusted by `adjust` BEFORE `build_deps` -/// runs, for tests that need more than one stub swapped in (e.g. a follow graph -/// plus the live user that authenticated routes now look up). pub fn test_app_state_with_repos( adjust: impl FnOnce(&mut Repositories), auth_service: Arc, diff --git a/crates/bootstrap/tests/ledger_recovery_test.rs b/crates/bootstrap/tests/ledger_recovery_test.rs index 72e02ac8..d84d0eed 100644 --- a/crates/bootstrap/tests/ledger_recovery_test.rs +++ b/crates/bootstrap/tests/ledger_recovery_test.rs @@ -1,5 +1,3 @@ -//! Crash recovery through the real worker dispatch and the real sqlite ledger. - mod support; use std::sync::atomic::{AtomicU32, Ordering}; @@ -56,9 +54,6 @@ impl EventConsumer for Once { } } -/// A claim abandoned mid-handler blocks redeliveries (deferred, not failed) -/// until its lease lapses; then the next delivery takes it over and runs the -/// handler once, and later deliveries skip it. #[tokio::test] async fn an_abandoned_claim_defers_deliveries_until_the_lease_lapses_then_runs_once() { let pool = support::memory_pool().await; @@ -68,7 +63,6 @@ async fn an_abandoned_claim_defers_deliveries_until_the_lease_lapses_then_runs_o Duration::from_secs(1), )); let id = EventId::new(); - // The crashed worker: claimed, never completed or released. assert!(matches!( ledger.begin(&id, "crashy").await.unwrap(), Begin::Run(_) diff --git a/crates/bootstrap/tests/outbox_relay_test.rs b/crates/bootstrap/tests/outbox_relay_test.rs index 4c453261..c1d9d5cd 100644 --- a/crates/bootstrap/tests/outbox_relay_test.rs +++ b/crates/bootstrap/tests/outbox_relay_test.rs @@ -1,6 +1,3 @@ -//! The relay claims outbox rows with the SQLite queue's own atomic claim and -//! settles each row from the sink's verdict, with fakes at the sink boundary. - mod support; use std::sync::{Arc, Mutex}; @@ -20,9 +17,7 @@ use uuid::Uuid; #[derive(Default)] struct FakeSink { fail: bool, - /// Fail this many calls, then succeed (a bus outage that ends). fail_first: std::sync::atomic::AtomicUsize, - /// When set, `relay` signals `started` then takes this long. slow: Option, started: Notify, seen: Mutex>>, @@ -133,8 +128,6 @@ async fn relay_marks_rows_done_only_after_publish_succeeds() { #[tokio::test] async fn a_failing_bus_never_spends_the_rows_attempts_and_the_row_relays_on_recovery() { - // Budget of 2 attempts, 3 consecutive bus failures: were failures counted - // as attempts the row would be dead-lettered before the bus recovers. let (pool, publisher, queue) = outbox_with_attempts(2).await; publisher.publish(&event()).await.unwrap(); let sink = Arc::new(FakeSink { diff --git a/crates/bootstrap/tests/route_smoke_test.rs b/crates/bootstrap/tests/route_smoke_test.rs index e2c89c49..e99ba9d8 100644 --- a/crates/bootstrap/tests/route_smoke_test.rs +++ b/crates/bootstrap/tests/route_smoke_test.rs @@ -1,6 +1,3 @@ -//! Characterization of the classic HTML routes, feeds and the OpenAPI surface, -//! recorded before the HTTP layer moves. Statuses are what the app returns today. - mod support; use axum::{ @@ -13,7 +10,6 @@ use support::{get, json_request, send}; const EMAIL: &str = "owner@example.com"; const USERNAME: &str = "owner"; -/// Public classic pages. `{id}`, `{username}`, `{movie}` are replaced by seeded values. const CLASSIC_VISITOR_PAGES: &[(&str, u16)] = &[ ("/", 200), ("/users", 200), @@ -36,7 +32,6 @@ const CLASSIC_VISITOR_PAGES: &[(&str, u16)] = &[ ("/health", 200), ]; -/// Pages that need the signed-in owner. const CLASSIC_OWNER_PAGES: &[(&str, u16)] = &[ ("/reviews/new", 200), ("/diary/export", 200), @@ -64,12 +59,10 @@ const FEDERATION_VISITOR_PAGES: &[(&str, u16)] = &[ const FEDERATION_OWNER_PAGES: &[(&str, u16)] = &[ ("/users/{id}/followers", 303), ("/users/{id}/following", 303), - // The registered owner is not an admin. ("/admin/blocked-domains", 403), ("/social/blocked", 200), ]; -/// Without federation the collection URLs permanently redirect to the HTML lists. #[cfg(not(feature = "federation"))] const FEDERATION_OFF_PAGES: &[(&str, u16)] = &[ ("/users/{id}/followers", 308), @@ -80,11 +73,9 @@ struct Seed { router: Router, user_id: String, movie_id: String, - /// Cookie header for the signed-in owner (csrf + token). owner_cookie: String, } -/// One user (registered through the classic form), one movie with a spoiler review, one goal. async fn seed() -> Seed { let (router, pool) = support::server().await; let session = support::register(&router, USERNAME, EMAIL).await; @@ -149,8 +140,6 @@ async fn seed() -> Seed { } } -/// Inserts a READY 2025 wrap-up record (user scope, or global when `user_id` is None), -/// the state the worker leaves behind after generating a report. async fn seed_ready_wrapup(pool: &sqlx::SqlitePool, user_id: Option<&str>) { use chrono::NaiveDate; use domain::models::wrapup::{DateRange, WrapUpScope}; @@ -210,7 +199,6 @@ async fn every_classic_page_renders_for_a_visitor() { #[cfg(not(feature = "federation"))] check_table(&seed, FEDERATION_OFF_PAGES, None).await; - // A dropped wrapup route would also 404, so require the template itself. for template in ["/wrapups/{id}/2025", "/wrapups/global/2025"] { let (status, _, body) = send(&seed.router, get(&expand(&seed, template), None)).await; assert_eq!(status, 200, "{template}"); @@ -220,7 +208,6 @@ async fn every_classic_page_renders_for_a_visitor() { ); } - // The owner embeds this widget in an iframe on their personal website. let (status, _, body) = send( &seed.router, get(&expand(&seed, "/users/{id}?embed=true"), None), @@ -245,7 +232,6 @@ async fn every_classic_page_renders_for_the_signed_in_owner() { #[tokio::test] async fn feeds_and_openapi_are_served() { let seed = seed().await; - // `/docs` (what users type) redirects to the Swagger UI at `/docs/`. for (template, expected) in [ ("/feed.rss", 200), ("/users/{id}/feed.rss", 200), diff --git a/crates/bootstrap/tests/support/mod.rs b/crates/bootstrap/tests/support/mod.rs index 9429d37e..089a009a 100644 --- a/crates/bootstrap/tests/support/mod.rs +++ b/crates/bootstrap/tests/support/mod.rs @@ -1,5 +1,3 @@ -//! Shared helpers for the bootstrap integration tests: a production app built -//! through `bootstrap` on an in-memory database, driven over HTTP. #![allow(dead_code)] use axum::{ @@ -29,7 +27,6 @@ pub fn test_config() -> config::AppConfig { nats: Default::default(), storage: Default::default(), image_conversion: Default::default(), - // Required to build; never called (seeding logs reviews with manual titles). metadata: config::MetadataConfig { tmdb_api_key: Some("test-key".into()), ..Default::default() @@ -94,7 +91,6 @@ pub fn json_request( b.body(Body::from(body.to_string())).unwrap() } -/// A single-connection in-memory pool (every connection would otherwise get its own database). pub async fn memory_pool() -> sqlx::SqlitePool { sqlx::sqlite::SqlitePoolOptions::new() .max_connections(1) @@ -103,7 +99,6 @@ pub async fn memory_pool() -> sqlx::SqlitePool { .unwrap() } -/// The production server router on a fresh in-memory database, plus its pool. pub async fn server() -> (Router, sqlx::SqlitePool) { let pool = memory_pool().await; let router = bootstrap::build_server_on(&test_config(), pool.clone()) @@ -114,14 +109,10 @@ pub async fn server() -> (Router, sqlx::SqlitePool) { pub struct Session { pub user_id: String, - /// Bearer token for the JSON API. pub token: String, - /// Cookie header for the classic UI (csrf + session). pub cookie: String, } -/// Registers through the classic form (GET for the csrf cookie, then POST with -/// the token), then logs in over the JSON API for a bearer token. pub async fn register(router: &Router, username: &str, email: &str) -> Session { let (status, headers, _) = send(router, get("/register", None)).await; assert_eq!(status, 200, "GET /register"); diff --git a/crates/config/src/lenient.rs b/crates/config/src/lenient.rs index 74377255..4401ef8e 100644 --- a/crates/config/src/lenient.rs +++ b/crates/config/src/lenient.rs @@ -1,6 +1,3 @@ -//! Deserializers for values that arrive as strings from the environment but -//! may be native numbers/bools in a TOML file. - use std::fmt; use std::marker::PhantomData; use std::str::FromStr; @@ -68,12 +65,10 @@ impl Visitor<'_> for Flag { } } -/// `true` only for `true` or `1` (anything else is false, like the old readers). pub fn true_or_one<'de, D: Deserializer<'de>>(d: D) -> Result { d.deserialize_any(Flag { accepts_one: true }) } -/// `true` only for `true`. pub fn true_only<'de, D: Deserializer<'de>>(d: D) -> Result { d.deserialize_any(Flag { accepts_one: false }) } diff --git a/crates/config/src/loader.rs b/crates/config/src/loader.rs index 5d5e6b83..a06a4d4d 100644 --- a/crates/config/src/loader.rs +++ b/crates/config/src/loader.rs @@ -10,8 +10,6 @@ use crate::errors::ConfigError; const DEFAULT_CONFIG_FILE: &str = "config.toml"; const CONFIG_FILE_VARIABLE: &str = "CONFIG_FILE"; -/// Every environment variable the application has always honoured, mapped to -/// its `section.field` key. No prefix: the names are the public contract. const LEGACY_ENV: &[(&str, &str)] = &[ ("ALLOW_REGISTRATION", "server.allow_registration"), ("BASE_URL", "server.base_url"), @@ -58,9 +56,6 @@ const LEGACY_ENV: &[(&str, &str)] = &[ ("WRAPUP_LOGO_PATH", "wrapup.logo_path"), ]; -/// Reads the legacy variables as plain strings; typing is left to the config -/// structs so that e.g. `JWT_SECRET=12345` stays a string. Empty or -/// whitespace-only values are skipped as unset. struct LegacyEnv; impl Provider for LegacyEnv { @@ -77,8 +72,6 @@ impl Provider for LegacyEnv { let value = raw .into_string() .map_err(|_| figment::Error::from(format!("{variable} is not valid unicode")))?; - // An empty value (`PORT=` in a .env template) means "not set": the - // default applies, or the optional field stays `None`. if value.trim().is_empty() { continue; } diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index 25218ff7..ac6f31a5 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -148,22 +148,14 @@ pub enum DomainEvent { pub trait AckHandle: Send + Sync { async fn ack(&self) -> Result<(), DomainError>; async fn nack(&self) -> Result<(), DomainError>; - /// Redeliver after `delay` without counting a failed attempt: for a - /// delivery that was not tried (it is blocked behind a live claim), so it - /// must not burn the retry budget that `nack` spends. A broker may still - /// count the redelivery itself: JetStream does, toward `max_deliver`. async fn defer(&self, delay: std::time::Duration) -> Result<(), DomainError>; } -/// Wait before retrying an event whose `attempt`-th delivery just failed: -/// 5s doubling per attempt, capped at 5 minutes. Every backend (SQLite queue, -/// JetStream nak, outbox relay) uses this one schedule. pub fn retry_backoff(attempt: u32) -> std::time::Duration { let secs = 5u64 * (1u64 << attempt.min(6)); std::time::Duration::from_secs(secs.min(300)) } -/// Stable identity of one queued event; the key for de-duplicating deliveries. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub struct EventId(uuid::Uuid); @@ -194,7 +186,6 @@ pub struct EventEnvelope { } impl EventEnvelope { - /// `id` is `None` only for events queued before identities existed. pub fn new(id: Option, event: DomainEvent, ack: Box) -> Self { Self { event, id, ack } } diff --git a/crates/domain/src/ports/account_purge.rs b/crates/domain/src/ports/account_purge.rs index ec991468..e1c8f563 100644 --- a/crates/domain/src/ports/account_purge.rs +++ b/crates/domain/src/ports/account_purge.rs @@ -2,41 +2,21 @@ use async_trait::async_trait; use crate::{errors::DomainError, value_objects::UserId}; -/// Tells the fediverse that a local account is gone (an ActivityPub `Delete` -/// to followers). Federation-off builds wire a no-op. -/// -/// Delivery is asynchronous: the implementation may only enqueue deliveries. -/// Those are signed later with the actor's keypair and the tombstoned user row, -/// which is why the purge keeps both until the tombstone cleanup. #[async_trait] pub trait FederationDeletionNotifier: Send + Sync { async fn notify_deleted(&self, user: &UserId) -> Result<(), DomainError>; } -/// Removes everything a deleted account owns, in one transaction, and strips -/// the user row down to a data-free tombstone. -/// -/// What stays: the username (retired for good), role, `deleted_at`, the -/// `purged_at` stamp and the signing keypair, so queued federation deliveries -/// can still be signed. `purge_expired_tombstones` removes those later. #[async_trait] pub trait AccountPurger: Send + Sync { - /// Whether the account's data has already been purged. async fn is_purged(&self, user: &UserId) -> Result; - /// Object-storage keys the account owns (avatar, banner). Still readable - /// after `purge`, until `forget_images`. async fn owned_image_keys(&self, user: &UserId) -> Result, DomainError>; - /// One transaction; every delete is "if present", so a re-run is a no-op. - /// Frees the email. Does nothing for an account that is gone or purged. async fn purge(&self, user: &UserId) -> Result<(), DomainError>; - /// Drops the image references once their objects are deleted. async fn forget_images(&self, user: &UserId) -> Result<(), DomainError>; - /// Deletes tombstones (user row and keypair) purged before `cutoff`. - /// Returns how many were removed. async fn purge_expired_tombstones( &self, cutoff: chrono::NaiveDateTime, diff --git a/crates/domain/src/ports/auth.rs b/crates/domain/src/ports/auth.rs index e96f0152..352d1198 100644 --- a/crates/domain/src/ports/auth.rs +++ b/crates/domain/src/ports/auth.rs @@ -14,8 +14,6 @@ pub trait AuthService: Send + Sync { #[async_trait] pub trait UserRepository: Send + Sync { - // The finders and `list_with_stats` serve live users only: tombstoned - // (deleted) accounts are invisible to them. async fn find_by_email(&self, email: &Email) -> Result, DomainError>; async fn find_by_username(&self, username: &Username) -> Result, DomainError>; async fn save(&self, user: &User) -> Result<(), DomainError>; @@ -27,17 +25,12 @@ pub trait UserRepository: Send + Sync { profile: &crate::models::UserProfile, events: &[crate::events::DomainEvent], ) -> Result<(), DomainError>; - /// Tombstones the account in one transaction: stamps `deleted_at`, - /// retires the username, drops the user's refresh sessions and records - /// `events`. Afterwards every reader below behaves as if the user is gone. async fn mark_deleted( &self, user_id: &UserId, events: &[crate::events::DomainEvent], ) -> Result<(), DomainError>; - /// The tombstoned user, for the purge. Live users are not returned. async fn find_tombstoned_by_id(&self, id: &UserId) -> Result, DomainError>; - /// Whether the name belonged to a deleted account (case-insensitive). async fn is_username_retired(&self, username: &Username) -> Result; async fn count_active_admins(&self) -> Result; } diff --git a/crates/domain/src/ports/diary.rs b/crates/domain/src/ports/diary.rs index 1417b72f..15c2af5a 100644 --- a/crates/domain/src/ports/diary.rs +++ b/crates/domain/src/ports/diary.rs @@ -51,9 +51,6 @@ pub trait ReviewRepository: Send + Sync { review: &Review, events: &[DomainEvent], ) -> Result<(), DomainError>; - /// Deletes the review and, in the same transaction, its movie when no - /// review of any kind references it any more. `review_events` are always - /// recorded; `orphan_events` only when the movie row was deleted. async fn delete_review_and_orphaned_movie( &self, review_id: &ReviewId, diff --git a/crates/domain/src/ports/events.rs b/crates/domain/src/ports/events.rs index c3b18e81..904abc69 100644 --- a/crates/domain/src/ports/events.rs +++ b/crates/domain/src/ports/events.rs @@ -10,8 +10,6 @@ pub trait EventPublisher: Send + Sync { async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError>; } -/// Delivers one outbox row to the message bus. `Ok` means the bus accepted it -/// (for a brokered bus: acknowledged), so the row may be marked done. #[async_trait] pub trait EventRelaySink: Send + Sync { async fn relay(&self, id: Option<&EventId>, event: &DomainEvent) -> Result<(), DomainError>; @@ -23,8 +21,6 @@ pub trait EventConsumer: Send + Sync { #[async_trait] pub trait EventHandler: Send + Sync { - /// Stable, unique snake_case identity. It keys the processed-event ledger, - /// so renaming a handler makes it re-run events still inside the retention window. fn name(&self) -> &'static str; async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError>; } diff --git a/crates/domain/src/ports/federation.rs b/crates/domain/src/ports/federation.rs index b81d9f42..064af5f7 100644 --- a/crates/domain/src/ports/federation.rs +++ b/crates/domain/src/ports/federation.rs @@ -8,8 +8,6 @@ use crate::{ #[async_trait] pub trait ApDocumentPort: Send + Sync { - /// Whether the actor belonged to a deleted account: its documents are - /// answered `410 Gone` instead of being served. async fn actor_is_gone(&self, _user_id: Uuid) -> Result { Ok(false) } diff --git a/crates/domain/src/ports/follow.rs b/crates/domain/src/ports/follow.rs index 197b5dd9..1af12c17 100644 --- a/crates/domain/src/ports/follow.rs +++ b/crates/domain/src/ports/follow.rs @@ -6,8 +6,6 @@ use crate::{ value_objects::{FollowRelation, FollowStatus, SocialActor}, }; -/// One row-level change to the local follow tables. A local follow operation -/// is a batch of these applied in a single transaction with its events. #[derive(Clone, Debug, PartialEq)] pub enum FollowWrite { AddFollow { @@ -42,8 +40,6 @@ pub enum FollowWrite { #[async_trait] pub trait FollowCommand: Send + Sync { - /// Applies every write and records `events` in one transaction: all rows - /// and events land, or none do. async fn apply( &self, writes: &[FollowWrite], diff --git a/crates/domain/src/ports/images.rs b/crates/domain/src/ports/images.rs index be06f48a..4e0e5ee8 100644 --- a/crates/domain/src/ports/images.rs +++ b/crates/domain/src/ports/images.rs @@ -11,7 +11,6 @@ pub trait ObjectStorage: Send + Sync { key: &str, ) -> Result>, DomainError>; async fn delete(&self, key: &str) -> Result<(), DomainError>; - /// Whether exactly `key` exists. Unlike `get`, no extension fallback. async fn exists(&self, key: &str) -> Result; } diff --git a/crates/domain/src/ports/noop.rs b/crates/domain/src/ports/noop.rs index 5a1e13fb..9bde4778 100644 --- a/crates/domain/src/ports/noop.rs +++ b/crates/domain/src/ports/noop.rs @@ -36,7 +36,6 @@ impl super::RemoteWatchlistRepository for NoopRemoteWatchlistRepository { // ── NoopFederationDeletionNotifier ─────────────────────────────────────────── -/// Federation compiled out: nobody remote to tell. pub struct NoopFederationDeletionNotifier; #[async_trait] diff --git a/crates/domain/src/ports/processed_events.rs b/crates/domain/src/ports/processed_events.rs index ea542000..352eae40 100644 --- a/crates/domain/src/ports/processed_events.rs +++ b/crates/domain/src/ports/processed_events.rs @@ -2,14 +2,8 @@ use async_trait::async_trait; use crate::{errors::DomainError, events::EventId}; -/// How long an `in_progress` claim blocks other deliveries of the same -/// (event, handler). Past it the claim is presumed dead (crashed worker) and -/// the next `begin` takes it over. Must exceed any handler's runtime. pub const HANDLER_LEASE_SECONDS: i64 = 1800; -/// Proof of ownership of one claim, returned by `begin`. `complete` and -/// `release` act only while the claim still carries this token, so an owner -/// whose lease lapsed cannot disturb whoever took the claim over. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClaimToken(String); @@ -25,31 +19,15 @@ impl ClaimToken { #[derive(Debug, Clone, PartialEq, Eq)] pub enum Begin { - /// The caller now holds the claim and must run the handler, then - /// `complete` or `release` with the token. Run(ClaimToken), - /// The handler already finished this event; skip it. AlreadyDone, - /// Another delivery holds a live claim; do not run. Retry the delivery - /// after `retry_after`, when the claim is done or its lease has lapsed. InProgressElsewhere { retry_after: std::time::Duration }, } -/// Per-(event, handler) claim ledger that stops redelivered events from -/// re-running handlers. -/// -/// Guarantee: two deliveries never run the same handler for the same event -/// concurrently (`begin` is atomic across workers). After a crash mid-handler -/// the claim lapses after `HANDLER_LEASE_SECONDS` and the handler runs again, -/// so delivery is at-least-once and handlers must be idempotent. #[async_trait] pub trait ProcessedEventLedger: Send + Sync { - /// Atomically claims (event, handler): inserts a claim if there is none, or - /// takes over an `in_progress` claim older than the lease; otherwise - /// reports the existing state. async fn begin(&self, id: &EventId, handler: &str) -> Result; - /// Marks the claim done, if `token` still owns it: the handler will not run for this event again. async fn complete( &self, id: &EventId, @@ -57,7 +35,6 @@ pub trait ProcessedEventLedger: Send + Sync { token: &ClaimToken, ) -> Result<(), DomainError>; - /// Drops an `in_progress` claim, if `token` still owns it, so a retry can run immediately. async fn release( &self, id: &EventId, @@ -65,7 +42,5 @@ pub trait ProcessedEventLedger: Send + Sync { token: &ClaimToken, ) -> Result<(), DomainError>; - /// Removes rows (done, or abandoned `in_progress`) first recorded before - /// `cutoff`. Returns the number removed. async fn delete_older_than(&self, cutoff: chrono::NaiveDateTime) -> Result; } diff --git a/crates/domain/src/ports/rss.rs b/crates/domain/src/ports/rss.rs index 1e4a37f0..999163c8 100644 --- a/crates/domain/src/ports/rss.rs +++ b/crates/domain/src/ports/rss.rs @@ -1,13 +1,9 @@ use crate::errors::DomainError; use crate::models::DiaryEntry; -/// Channel-level metadata for a rendered feed. Paths are site-relative -/// (e.g. `/users/{id}`); the renderer resolves them against its base URL. pub struct FeedChannel<'a> { pub title: &'a str, - /// Human-readable page the channel represents. pub link_path: &'a str, - /// Where the feed itself is served, advertised as `atom:link rel="self"`. pub self_path: &'a str, } diff --git a/crates/domain/src/ports/watchlist.rs b/crates/domain/src/ports/watchlist.rs index 7fe6ace9..cfb53272 100644 --- a/crates/domain/src/ports/watchlist.rs +++ b/crates/domain/src/ports/watchlist.rs @@ -21,7 +21,6 @@ pub trait WatchlistRepository: Send + Sync { events: &[DomainEvent], ) -> Result<(), DomainError>; - /// `events` are recorded only when an entry was actually removed. async fn remove_if_present( &self, user_id: &UserId, diff --git a/crates/domain/src/testing/factories.rs b/crates/domain/src/testing/factories.rs index 4ad9921b..81c2a3cc 100644 --- a/crates/domain/src/testing/factories.rs +++ b/crates/domain/src/testing/factories.rs @@ -11,7 +11,6 @@ use crate::value_objects::{ const DEFAULT_YEAR: u16 = 2000; -/// Fixed watch timestamp used by review factories (2024-03-15 20:00:00). pub fn test_watched_at() -> NaiveDateTime { NaiveDate::from_ymd_opt(2024, 3, 15) .unwrap() diff --git a/crates/domain/src/testing/fakes/refusing.rs b/crates/domain/src/testing/fakes/refusing.rs index fedac6d7..3bf5a25a 100644 --- a/crates/domain/src/testing/fakes/refusing.rs +++ b/crates/domain/src/testing/fakes/refusing.rs @@ -1,5 +1,3 @@ -//! Fakes whose every call fails with an infrastructure error, for failure-path tests. - use async_trait::async_trait; use chrono::NaiveDateTime; diff --git a/crates/domain/src/testing/fakes/store.rs b/crates/domain/src/testing/fakes/store.rs index 91d5efae..1516bbed 100644 --- a/crates/domain/src/testing/fakes/store.rs +++ b/crates/domain/src/testing/fakes/store.rs @@ -9,7 +9,6 @@ use crate::models::{ }; use crate::value_objects::SocialIdentity; -/// One `processed_events` row. pub(super) struct LedgerRow { pub(super) done: bool, pub(super) started_at: chrono::NaiveDateTime, @@ -22,9 +21,6 @@ pub(super) enum FollowState { Accepted, } -/// Tables whose rows the schema removes when a user row is deleted -/// (`ON DELETE CASCADE` on a column referencing `users(id)`), and which -/// `InMemoryStore` models. The sqlite adapter's cascade test asserts the same list. pub const USER_CASCADE_TABLES: &[&str] = &[ "watchlist_entries", "goals", @@ -37,11 +33,8 @@ pub const USER_CASCADE_TABLES: &[&str] = &[ "blocked_actors", ]; -/// Tables that cascade from a user but whose aggregate this store does not hold -/// (they belong to the federation crate). pub const USER_CASCADE_TABLES_OUTSIDE_STORE: &[&str] = &["ap_local_actors"]; -/// Tables whose rows the schema removes when a movie row is deleted. pub const MOVIE_CASCADE_TABLES: &[&str] = &[ "watchlist_entries", "movie_profiles", @@ -51,17 +44,12 @@ pub const MOVIE_CASCADE_TABLES: &[&str] = &[ "movie_crew", ]; -/// One in-memory implementation of every persistence port, mirroring the -/// relational constraints of the sqlite schema (cascades, uniqueness, FK -/// restrictions). Share one `Arc` across all port slots. pub struct InMemoryStore { pub(super) movies: RwLock>, pub(super) movie_profiles: RwLock>, pub(super) reviews: RwLock>, pub(super) users: RwLock>, - /// Ids of `users` rows carrying a `deleted_at` tombstone. pub(super) tombstoned: RwLock>, - /// Lower-cased `retired_usernames` (the column is NOCASE). pub(super) retired_usernames: RwLock>, pub(super) profile_fields: RwLock>>, pub(super) user_settings: RwLock>, @@ -111,8 +99,6 @@ impl InMemoryStore { store } - /// Events recorded atomically with a change (the outbox rows the sqlite - /// adapter would have written in the same transaction), in write order. pub fn recorded_events(&self) -> Vec { self.recorded_events.read().unwrap().clone() } @@ -156,7 +142,6 @@ impl InMemoryStore { self.processed_events.read().unwrap().len() } - /// Inserts a ledger row with explicit timestamps (retention and lease tests). pub fn seed_processed_event( &self, id: &crate::events::EventId, @@ -178,13 +163,10 @@ impl InMemoryStore { self.refresh_sessions.read().unwrap().len() } - /// Inserts a user bypassing the uniqueness checks of `UserRepository::save`. pub fn seed_user(&self, user: User) { self.users.write().unwrap().insert(user.id().value(), user); } - /// Row count of the sqlite table of the same name, for the tables this - /// store models. Panics on a table the store does not hold. pub fn rows_in(&self, table: &str) -> usize { match table { "watchlist_entries" => self.watchlist_count(), diff --git a/crates/domain/src/testing/fakes/store_goal.rs b/crates/domain/src/testing/fakes/store_goal.rs index be5ec066..581c3678 100644 --- a/crates/domain/src/testing/fakes/store_goal.rs +++ b/crates/domain/src/testing/fakes/store_goal.rs @@ -12,7 +12,6 @@ use super::InMemoryStore; #[async_trait] impl GoalCommand for InMemoryStore { - /// `goals` is UNIQUE(user_id, year). async fn save(&self, goal: &Goal, events: &[DomainEvent]) -> Result<(), DomainError> { let mut goals = self.goals.write().unwrap(); let duplicate = goals.values().any(|g| { diff --git a/crates/domain/src/testing/fakes/store_movie.rs b/crates/domain/src/testing/fakes/store_movie.rs index 139cdae0..e5afba50 100644 --- a/crates/domain/src/testing/fakes/store_movie.rs +++ b/crates/domain/src/testing/fakes/store_movie.rs @@ -15,9 +15,6 @@ use super::InMemoryStore; #[async_trait] impl MovieCommand for InMemoryStore { - /// `movies.external_metadata_id` is UNIQUE: another movie row holding the - /// same external id is a constraint violation, which the sqlite adapter - /// surfaces as an infrastructure error. async fn upsert_movie(&self, movie: &Movie, events: &[DomainEvent]) -> Result<(), DomainError> { let mut movies = self.movies.write().unwrap(); if let Some(external) = movie.external_metadata_id() { @@ -37,9 +34,6 @@ impl MovieCommand for InMemoryStore { Ok(()) } - /// Schema: `reviews.movie_id REFERENCES movies(id)` has no cascade, so a - /// movie with reviews cannot be deleted. `watchlist_entries` and the - /// `movie_*` profile tables cascade; `watch_events.movie_id` is set to NULL. async fn delete_movie( &self, movie_id: &MovieId, @@ -67,7 +61,6 @@ impl MovieCommand for InMemoryStore { } impl InMemoryStore { - /// Removes the movie row and applies the schema's cascades. pub(super) fn remove_movie_with_cascades(&self, id: uuid::Uuid) { self.movies.write().unwrap().remove(&id); self.watchlist @@ -155,7 +148,6 @@ impl MovieQuery for InMemoryStore { page: &PageParams, filter: &MovieFilter, ) -> Result, DomainError> { - // Mirrors the sqlite adapter: filter, ORDER BY title ASC, then page. let movies = self.movies.read().unwrap(); let profiles = self.movie_profiles.read().unwrap(); let search = filter.search.as_deref().map(str::to_lowercase); diff --git a/crates/domain/src/testing/fakes/store_processed_events.rs b/crates/domain/src/testing/fakes/store_processed_events.rs index f26c71d7..ee706a7d 100644 --- a/crates/domain/src/testing/fakes/store_processed_events.rs +++ b/crates/domain/src/testing/fakes/store_processed_events.rs @@ -10,7 +10,6 @@ use crate::{ use super::InMemoryStore; use super::store::LedgerRow; -/// Claim stamps are millisecond-precision UTC, as in the sqlite adapter. const STAMP: &str = "%Y-%m-%dT%H:%M:%S%.3fZ"; fn now_millis() -> NaiveDateTime { @@ -23,9 +22,6 @@ fn token(at: NaiveDateTime) -> ClaimToken { ClaimToken::new(at.format(STAMP).to_string()) } -/// Mirrors the sqlite adapter: `begin` inserts, or takes over only a stale -/// `in_progress` claim; `(event_id, handler)` is the key; `complete` and -/// `release` act only for the token that owns the claim. #[async_trait] impl ProcessedEventLedger for InMemoryStore { async fn begin(&self, id: &EventId, handler: &str) -> Result { diff --git a/crates/domain/src/testing/fakes/store_session.rs b/crates/domain/src/testing/fakes/store_session.rs index 514e56af..4627fb2e 100644 --- a/crates/domain/src/testing/fakes/store_session.rs +++ b/crates/domain/src/testing/fakes/store_session.rs @@ -10,7 +10,6 @@ use super::InMemoryStore; #[async_trait] impl RefreshSessionRepository for InMemoryStore { - /// `refresh_sessions.token` is UNIQUE. async fn create(&self, session: &RefreshSession) -> Result<(), DomainError> { let mut sessions = self.refresh_sessions.write().unwrap(); if sessions.iter().any(|s| s.token == session.token) { diff --git a/crates/domain/src/testing/fakes/store_user.rs b/crates/domain/src/testing/fakes/store_user.rs index 4e59b0c4..12fb02c4 100644 --- a/crates/domain/src/testing/fakes/store_user.rs +++ b/crates/domain/src/testing/fakes/store_user.rs @@ -34,8 +34,6 @@ impl UserRepository for InMemoryStore { .cloned()) } - /// `users.email` and `users.username` are UNIQUE; the sqlite adapter checks - /// both up front and reports a validation error. async fn save(&self, user: &User) -> Result<(), DomainError> { let mut users = self.users.write().unwrap(); if users @@ -129,8 +127,6 @@ impl UserRepository for InMemoryStore { .count() as u64) } - /// Ordered by email ascending like the real query. Diary stats are stubbed - /// (review count 0, no last activity); only the identity fields are real. async fn list_with_stats(&self) -> Result, DomainError> { let store = self.users.read().unwrap(); let gone = self.tombstoned.read().unwrap(); @@ -229,10 +225,6 @@ impl UserFederationSettingsQuery for InMemoryStore { } impl InMemoryStore { - /// Deletes a user the way `DELETE FROM users` does under the production - /// schema: rows in `USER_CASCADE_TABLES` go with it. Reviews, refresh - /// sessions, import sessions and import profiles carry a plain `user_id` - /// column with no foreign key, so they stay. pub fn delete_user(&self, user_id: &UserId) { let id = user_id.value(); self.users.write().unwrap().remove(&id); diff --git a/crates/domain/src/testing/fakes/store_watchlist.rs b/crates/domain/src/testing/fakes/store_watchlist.rs index ab523cad..ef85b55b 100644 --- a/crates/domain/src/testing/fakes/store_watchlist.rs +++ b/crates/domain/src/testing/fakes/store_watchlist.rs @@ -16,8 +16,6 @@ use super::InMemoryStore; #[async_trait] impl WatchlistRepository for InMemoryStore { - /// `INSERT OR IGNORE` on UNIQUE(user_id, movie_id): adding twice is a no-op - /// and, like the adapter, records events only for the row it inserted. async fn add(&self, entry: &WatchlistEntry, events: &[DomainEvent]) -> Result<(), DomainError> { let key = (entry.user_id.value(), entry.movie_id.value()); let inserted = { diff --git a/crates/domain/src/testing/fakes/wrapup.rs b/crates/domain/src/testing/fakes/wrapup.rs index e771cde2..2da7e916 100644 --- a/crates/domain/src/testing/fakes/wrapup.rs +++ b/crates/domain/src/testing/fakes/wrapup.rs @@ -81,8 +81,6 @@ impl InMemoryWrapUpRepository { }) } - /// Events recorded atomically with a change (the outbox rows the sqlite - /// adapter writes in the same transaction). pub fn recorded_events(&self) -> Vec { self.recorded.lock().unwrap().clone() } @@ -268,9 +266,6 @@ impl WrapUpRepository for PanicWrapUpRepository { // ── RefusingWrapUpWrites ─────────────────────────────────────────────────── -/// Delegates to an in-memory repository but refuses the two writes that carry -/// outbox events (`create`, `set_complete`), leaving `update_status` and reads -/// working so a refusal can be observed mid-flow. pub struct RefusingWrapUpWrites { pub inner: Arc, } diff --git a/crates/domain/src/value_objects/ids.rs b/crates/domain/src/value_objects/ids.rs index 293b67c5..b4908b66 100644 --- a/crates/domain/src/value_objects/ids.rs +++ b/crates/domain/src/value_objects/ids.rs @@ -33,8 +33,6 @@ impl MovieId { uuid_id!(ReviewId); uuid_id!(UserId); -/// Id of the instance's service actor (seeded by migration 0022). It signs -/// fetches for the whole instance and is never a deletable account. pub const SERVICE_ACTOR_UUID: Uuid = Uuid::from_bytes([0, 0, 0, 0, 0, 0, 0x40, 0, 0x80, 0, 0, 0, 0, 0, 0, 0]); diff --git a/crates/domain/tests/testing/in_memory_store_test.rs b/crates/domain/tests/testing/in_memory_store_test.rs index 3e8ed07f..dab8d1d5 100644 --- a/crates/domain/tests/testing/in_memory_store_test.rs +++ b/crates/domain/tests/testing/in_memory_store_test.rs @@ -61,8 +61,6 @@ fn movie_profile(movie_id: &MovieId) -> MovieProfile { } } -/// Saves a user and one row in every table the schema cascades from a user, -/// plus rows in the tables that hang off a user without a foreign key. async fn seed_user_with_dependents(store: &Arc) -> (UserId, Movie) { let user = test_user("alice"); let user_id = user.id().clone(); diff --git a/crates/event-payload/src/lib.rs b/crates/event-payload/src/lib.rs index e63059d0..7d6dd3af 100644 --- a/crates/event-payload/src/lib.rs +++ b/crates/event-payload/src/lib.rs @@ -154,8 +154,6 @@ pub enum EventPayload { }, } -/// Wire form of an event: the tagged payload plus its optional identity. -/// Messages without `event_id` (queued before identities existed) still decode. #[derive(Debug, Serialize, Deserialize, PartialEq)] pub struct EventMessage { #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/crates/event-payload/tests/event_payload_test.rs b/crates/event-payload/tests/event_payload_test.rs index 5252bedc..34ee6356 100644 --- a/crates/event-payload/tests/event_payload_test.rs +++ b/crates/event-payload/tests/event_payload_test.rs @@ -132,9 +132,6 @@ fn a_message_round_trips_its_event_id_and_decodes_without_one() { assert_eq!(serde_json::from_str::(&json).unwrap(), unit); } -/// One sample per `DomainEvent` variant. `assert_covers_every_variant` is an -/// exhaustive match, so adding a variant breaks the build until it is added -/// here too. fn every_event() -> Vec { use domain::{ models::{ExternalPersonId, PersonId}, diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 5bea41a6..acca556c 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -9,7 +9,6 @@ license = "MIT" default = ["federation"] federation = ["bootstrap/federation"] nats = ["bootstrap/nats"] -# deprecated aliases kept so existing deploy commands keep working sqlite = [] sqlite-federation = ["federation"] diff --git a/crates/tui/src/client.rs b/crates/tui/src/client.rs index 1364bf14..e74aeba9 100644 --- a/crates/tui/src/client.rs +++ b/crates/tui/src/client.rs @@ -46,7 +46,6 @@ pub struct ApiClient { impl ApiClient { pub fn new(url: &str) -> Self { - // reqwest is built without a bundled TLS provider; install ring first. let _ = rustls::crypto::ring::default_provider().install_default(); Self { base_url: std::sync::RwLock::new(url.to_string()), diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 685b713b..822554bf 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -7,7 +7,6 @@ edition.workspace = true default = ["federation"] federation = ["bootstrap/federation"] nats = ["bootstrap/nats"] -# deprecated aliases kept so existing deploy commands keep working sqlite = [] sqlite-federation = ["federation"]