Compare commits

...

12 Commits

Author SHA1 Message Date
89045414cf fmt
Some checks failed
CI / Check / Test (push) Failing after 9m15s
2026-07-10 17:10:59 +02:00
2de6690401 fix: import remote outbox on outbound follow accept (k-ap 0.4.2)
k-ap now emits OutboundFollowAccepted when remote accepts our follow.
Bridge maps it to FollowAccepted domain event. Worker handler looks up
the actor's outbox URL and calls import_remote_outbox. Previously the
accept was silent — no backfill happened, feed stayed empty.
2026-07-10 17:02:14 +02:00
12378c3649 fmt 2026-07-10 16:32:07 +02:00
46b8488b09 refactor: Feed uses SocialIdentity matching, deps From impl, remove get_accepted_following_urls
Feed builds FollowingFilter by matching SocialIdentity::Local/Remote
instead of URL-prefix heuristic. Removed get_accepted_following_urls
from SocialQuery (no longer needed). Added From<&AppState> for deps
structs — 20 construction sites collapsed to one-liners.
2026-07-10 16:31:05 +02:00
7e02f15a85 refactor: consolidate 11 social use cases → SocialCmd/SocialQry enum dispatch
-280 net lines. 11 one-file use cases replaced by execute_command/
execute_query with enum dispatch. Added FollowRejected event (reject
was silently dropping). Command→event mapping now explicit in one match.
2026-07-10 16:26:57 +02:00
d60c47199c fix: FollowTarget enum — actor_url no longer lies about holding a handle
FollowCommand.target is now FollowTarget (Identity|Handle) instead of
SocialIdentity. actor_url field always holds a URL; handles go through
FollowTarget::Handle. Adapter resolves handles explicitly. Type system
prevents misuse in future commands.
2026-07-10 16:18:57 +02:00
2484f1e603 fmt 2026-07-10 16:03:36 +02:00
44d7df33a2 refactor: remove SocialQueryPort — replaced by SocialQuery + FederationAdminQuery
-319 lines. Legacy SocialQueryPort trait, NoopSocialQueryPort,
PanicSocialQueryPort all deleted. Federation repos now implement
FederationAdminQuery (single method). get_users uses FederationAdminQuery.
All other consumers use unified SocialQuery.
2026-07-10 16:02:27 +02:00
7cfa234902 refactor: SocialActor rich queries, migrate remaining handlers, slim SocialQueryPort
SocialQuery returns SocialActor (identity+handle+display_name+avatar_url)
instead of bare SocialIdentity. Migrated get_following_page,
get_followers_page, get_blocked_actors_page to use cases. Moved
get_activity_feed + get_profile from SocialQueryPort to SocialQuery.
Legacy SocialQueryPort remains only for get_users listing.
2026-07-10 15:52:37 +02:00
3ee75305a9 refactor: move ap_service from AppState to AppContext.Services 2026-07-10 15:34:23 +02:00
322e9ee81a feat: unified social identity layer — SocialIdentity, ports, use cases, adapter
SocialIdentity value object (Local|Remote), SocialCommand/SocialQuery
domain ports, 11 CQRS use cases w/ 14 tests, CompositeSocialAdapter
wrapping k_ap, 6 new domain events, handlers migrated from ap_service
to use cases. Closes #12 foundation — AP handlers + legacy cleanup TBD.
2026-07-10 15:29:46 +02:00
96ce5f7d26 docs: domain glossary (16 terms) + ADR-0002 unified social identity
CONTEXT.md expanded with Movie, Person, Review, Rating, WatchMedium,
Watchlist, Goal, WrapUp, User, Follow, Feed, WatchEvent, Import,
ImportProfile, SocialIdentity. ADR-0002 documents wrap-k_ap decision.
2026-07-10 15:29:31 +02:00
47 changed files with 2074 additions and 587 deletions

1
.gitignore vendored
View File

@@ -14,6 +14,7 @@
.worktrees/ .worktrees/
.superpowers/ .superpowers/
docs/ docs/
!docs/adr/
imgs/ imgs/
.sqlx/ .sqlx/

View File

@@ -40,6 +40,10 @@ _Avoid_: Stats page, recap, summary
A registered account with a username, email, and profile (display name, bio, avatar, banner). Can be Standard or Admin. A registered account with a username, email, and profile (display name, bio, avatar, banner). Can be Standard or Admin.
_Avoid_: Account, member, profile (as a synonym for the whole User) _Avoid_: Account, member, profile (as a synonym for the whole User)
**SocialIdentity**:
The uniform identifier for anyone involved in a social interaction — either a local User or a remote federated actor. Social commands and queries operate on SocialIdentity so the domain never branches on local vs remote.
_Avoid_: Actor, participant, social user
**Follow**: **Follow**:
A social relationship where one user subscribes to another's activity. Always requires acceptance by the target user. Works identically for local and federated (ActivityPub) users. Once accepted, the followed user's reviews appear in the follower's Feed. A social relationship where one user subscribes to another's activity. Always requires acceptance by the target user. Works identically for local and federated (ActivityPub) users. Once accepted, the followed user's reviews appear in the follower's Feed.
_Avoid_: Subscribe, connect, friend _Avoid_: Subscribe, connect, friend

6
Cargo.lock generated
View File

@@ -2892,9 +2892,9 @@ dependencies = [
[[package]] [[package]]
name = "k-ap" name = "k-ap"
version = "0.4.1" version = "0.4.2"
source = "sparse+https://git.gabrielkaszewski.dev/api/packages/GKaszewski/cargo/" source = "sparse+https://git.gabrielkaszewski.dev/api/packages/GKaszewski/cargo/"
checksum = "03e39c04075b39960c329feba896a16aba37f0863669c28e7106b7cc45a9988d" checksum = "4291cac43b119cce0be6e2ba8d85339f3f4c69b266cb7c00fb4cb179302b97e4"
dependencies = [ dependencies = [
"activitypub_federation", "activitypub_federation",
"anyhow", "anyhow",
@@ -6714,7 +6714,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [ dependencies = [
"windows-sys 0.48.0", "windows-sys 0.61.2",
] ]
[[package]] [[package]]

View File

@@ -4,7 +4,7 @@ version = "0.1.0"
edition = "2024" edition = "2024"
[dependencies] [dependencies]
k-ap = { version = "0.4.1", registry = "gitea" } k-ap = { version = "0.4.2", registry = "gitea" }
domain = { workspace = true } domain = { workspace = true }
axum = { workspace = true } axum = { workspace = true }
serde = { workspace = true } serde = { workspace = true }

View File

@@ -46,6 +46,28 @@ impl k_ap::EventPublisher for FederationEventBridge {
tracing::warn!(inbox = %inbox, error = %error, "federation delivery failed permanently"); tracing::warn!(inbox = %inbox, error = %error, "federation delivery failed permanently");
Ok(()) Ok(())
} }
FederationEvent::OutboundFollowAccepted {
local_user_id,
remote_actor_url,
outbox_url,
} => {
let identity = domain::value_objects::SocialIdentity::Remote {
actor_url: remote_actor_url,
};
self.domain_publisher
.publish(&DomainEvent::FollowAccepted {
owner: UserId::from_uuid(local_user_id),
requester: identity,
})
.await
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
if let Some(outbox) = outbox_url {
tracing::info!(outbox = %outbox, "importing remote outbox after follow accepted");
// Handled by FollowBackfillHandler reacting to FollowAccepted
}
Ok(())
}
} }
} }
} }

View File

@@ -6,6 +6,7 @@ pub mod objects;
pub mod port; pub mod port;
pub mod remote_review_repository; pub mod remote_review_repository;
pub mod review_handler; pub mod review_handler;
pub mod social_adapter;
pub(crate) mod urls; pub(crate) mod urls;
pub mod user_adapter; pub mod user_adapter;
pub mod watchlist_handler; pub mod watchlist_handler;
@@ -24,6 +25,7 @@ pub use event_handler::ActivityPubEventHandler;
pub use port::{ActivityPubPort, NoopActivityPubService}; pub use port::{ActivityPubPort, NoopActivityPubService};
pub use remote_review_repository::{RemoteReviewRepository, RemoteReviewUpdate}; pub use remote_review_repository::{RemoteReviewRepository, RemoteReviewUpdate};
pub use review_handler::ReviewObjectHandler; pub use review_handler::ReviewObjectHandler;
pub use social_adapter::CompositeSocialAdapter;
pub use user_adapter::DomainUserRepoAdapter; pub use user_adapter::DomainUserRepoAdapter;
pub type FederationRepos = ( pub type FederationRepos = (
@@ -31,7 +33,7 @@ pub type FederationRepos = (
std::sync::Arc<dyn FollowRepository>, std::sync::Arc<dyn FollowRepository>,
std::sync::Arc<dyn ActorRepository>, std::sync::Arc<dyn ActorRepository>,
std::sync::Arc<dyn BlocklistRepository>, std::sync::Arc<dyn BlocklistRepository>,
std::sync::Arc<dyn domain::ports::SocialQueryPort>, std::sync::Arc<dyn domain::ports::FederationAdminQuery>,
std::sync::Arc<dyn RemoteReviewRepository>, std::sync::Arc<dyn RemoteReviewRepository>,
std::sync::Arc<dyn domain::ports::RemoteWatchlistRepository>, std::sync::Arc<dyn domain::ports::RemoteWatchlistRepository>,
); );

View File

@@ -0,0 +1,245 @@
use std::sync::Arc;
use async_trait::async_trait;
use domain::{
errors::DomainError,
ports::{SocialCommand, SocialQuery, UserRepository},
value_objects::{FollowTarget, SocialActor, SocialIdentity, UserId},
};
use k_ap::RemoteActor;
use super::ActivityPubPort;
pub struct CompositeSocialAdapter {
ap_service: Arc<dyn ActivityPubPort>,
user_repo: Arc<dyn UserRepository>,
base_url: String,
}
impl CompositeSocialAdapter {
pub fn new(
ap_service: Arc<dyn ActivityPubPort>,
user_repo: Arc<dyn UserRepository>,
base_url: String,
) -> Self {
Self {
ap_service,
user_repo,
base_url,
}
}
fn local_actor_url(&self, user_id: &UserId) -> String {
format!("{}/users/{}", self.base_url, user_id.value())
}
fn actor_url_from_identity(&self, identity: &SocialIdentity) -> String {
match identity {
SocialIdentity::Local(uid) => self.local_actor_url(uid),
SocialIdentity::Remote { actor_url } => actor_url.clone(),
}
}
fn identity_from_actor_url(&self, url: &str) -> SocialIdentity {
let prefix = format!("{}/users/", self.base_url);
if let Some(uuid_str) = url.strip_prefix(&prefix)
&& let Ok(uuid) = uuid::Uuid::parse_str(uuid_str)
{
return SocialIdentity::Local(UserId::from_uuid(uuid));
}
SocialIdentity::Remote {
actor_url: url.to_string(),
}
}
fn remote_actor_to_social_actor(&self, actor: RemoteActor) -> SocialActor {
let identity = self.identity_from_actor_url(&actor.url);
SocialActor {
identity,
handle: actor.handle,
display_name: actor.display_name,
avatar_url: actor.avatar_url,
}
}
async fn resolve_handle(&self, identity: &SocialIdentity) -> Result<String, DomainError> {
match identity {
SocialIdentity::Local(uid) => {
let user = self
.user_repo
.find_by_id(uid)
.await?
.ok_or_else(|| DomainError::NotFound("User not found".into()))?;
let host = url::Url::parse(&self.base_url)
.map(|u| u.host_str().unwrap_or("localhost").to_string())
.unwrap_or_else(|_| "localhost".to_string());
Ok(format!("@{}@{}", user.username().value(), host))
}
SocialIdentity::Remote { actor_url } => Ok(actor_url.clone()),
}
}
}
fn ap_err(e: anyhow::Error) -> DomainError {
DomainError::InfrastructureError(e.to_string())
}
#[async_trait]
impl SocialCommand for CompositeSocialAdapter {
async fn follow(&self, follower: &UserId, target: &FollowTarget) -> Result<(), DomainError> {
if let FollowTarget::Identity(SocialIdentity::Local(target_id)) = target
&& follower == target_id
{
return Err(DomainError::ValidationError(
"Cannot follow yourself".into(),
));
}
let handle = match target {
FollowTarget::Handle(h) => h.clone(),
FollowTarget::Identity(id) => self.resolve_handle(id).await?,
};
self.ap_service
.follow(follower.value(), &handle)
.await
.map_err(ap_err)
}
async fn unfollow(
&self,
follower: &UserId,
target: &SocialIdentity,
) -> Result<(), DomainError> {
let actor_url = self.actor_url_from_identity(target);
self.ap_service
.unfollow(follower.value(), &actor_url)
.await
.map_err(ap_err)
}
async fn accept_follow(
&self,
owner: &UserId,
requester: &SocialIdentity,
) -> Result<(), DomainError> {
let actor_url = self.actor_url_from_identity(requester);
self.ap_service
.accept_follower(owner.value(), &actor_url)
.await
.map_err(ap_err)
}
async fn reject_follow(
&self,
owner: &UserId,
requester: &SocialIdentity,
) -> Result<(), DomainError> {
let actor_url = self.actor_url_from_identity(requester);
self.ap_service
.reject_follower(owner.value(), &actor_url)
.await
.map_err(ap_err)
}
async fn remove_follower(
&self,
owner: &UserId,
follower: &SocialIdentity,
) -> Result<(), DomainError> {
let actor_url = self.actor_url_from_identity(follower);
self.ap_service
.remove_follower(owner.value(), &actor_url)
.await
.map_err(ap_err)
}
async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> {
let actor_url = self.actor_url_from_identity(target);
self.ap_service
.block_actor(blocker.value(), &actor_url)
.await
.map_err(ap_err)
}
async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> {
let actor_url = self.actor_url_from_identity(target);
self.ap_service
.unblock_actor(blocker.value(), &actor_url)
.await
.map_err(ap_err)
}
}
#[async_trait]
impl SocialQuery for CompositeSocialAdapter {
async fn get_following(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let actors = self
.ap_service
.get_following(user.value())
.await
.map_err(ap_err)?;
Ok(actors
.into_iter()
.map(|a| self.remote_actor_to_social_actor(a))
.collect())
}
async fn get_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let actors = self
.ap_service
.get_accepted_followers(user.value())
.await
.map_err(ap_err)?;
Ok(actors
.into_iter()
.map(|a| self.remote_actor_to_social_actor(a))
.collect())
}
async fn get_pending_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let actors = self
.ap_service
.get_pending_followers(user.value())
.await
.map_err(ap_err)?;
Ok(actors
.into_iter()
.map(|a| self.remote_actor_to_social_actor(a))
.collect())
}
async fn count_following(&self, user: &UserId) -> Result<usize, DomainError> {
self.ap_service
.count_following(user.value())
.await
.map_err(ap_err)
}
async fn count_followers(&self, user: &UserId) -> Result<usize, DomainError> {
self.ap_service
.count_accepted_followers(user.value())
.await
.map_err(ap_err)
}
async fn get_blocked(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let actors = self
.ap_service
.get_blocked_actors(user.value())
.await
.map_err(ap_err)?;
Ok(actors
.into_iter()
.map(|a| self.remote_actor_to_social_actor(a))
.collect())
}
async fn is_following(
&self,
follower: &UserId,
target: &SocialIdentity,
) -> Result<bool, DomainError> {
let following = self.get_following(follower).await?;
Ok(following.iter().any(|a| a.identity == *target))
}
}

View File

@@ -4,7 +4,8 @@ use domain::{
events::DomainEvent, events::DomainEvent,
models::{ExternalPersonId, PersonId}, models::{ExternalPersonId, PersonId},
value_objects::{ value_objects::{
ExternalMetadataId, GoalId, MovieId, PosterPath, Rating, ReviewId, UserId, WrapUpId, ExternalMetadataId, GoalId, MovieId, PosterPath, Rating, ReviewId, SocialIdentity, UserId,
WrapUpId,
}, },
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@@ -61,10 +62,40 @@ pub enum EventPayload {
user_id: String, user_id: String,
movie_id: String, movie_id: String,
}, },
FollowRequested {
follower_id: String,
target_kind: String,
target_id: String,
},
FollowAccepted { FollowAccepted {
local_user_id: String, owner_id: String,
remote_actor_url: String, requester_kind: String,
outbox_url: String, requester_id: String,
},
FollowRejected {
owner_id: String,
requester_kind: String,
requester_id: String,
},
Unfollowed {
follower_id: String,
target_kind: String,
target_id: String,
},
FollowerRemoved {
owner_id: String,
follower_kind: String,
follower_id: String,
},
ActorBlocked {
blocker_id: String,
target_kind: String,
target_id: String,
},
ActorUnblocked {
blocker_id: String,
target_kind: String,
target_id: String,
}, },
BackfillFollower { BackfillFollower {
owner_user_id: String, owner_user_id: String,
@@ -136,7 +167,13 @@ impl EventPayload {
EventPayload::ImageStored { .. } => "ImageStored", EventPayload::ImageStored { .. } => "ImageStored",
EventPayload::WatchlistEntryAdded { .. } => "WatchlistEntryAdded", EventPayload::WatchlistEntryAdded { .. } => "WatchlistEntryAdded",
EventPayload::WatchlistEntryRemoved { .. } => "WatchlistEntryRemoved", EventPayload::WatchlistEntryRemoved { .. } => "WatchlistEntryRemoved",
EventPayload::FollowRequested { .. } => "FollowRequested",
EventPayload::FollowAccepted { .. } => "FollowAccepted", EventPayload::FollowAccepted { .. } => "FollowAccepted",
EventPayload::FollowRejected { .. } => "FollowRejected",
EventPayload::Unfollowed { .. } => "Unfollowed",
EventPayload::FollowerRemoved { .. } => "FollowerRemoved",
EventPayload::ActorBlocked { .. } => "ActorBlocked",
EventPayload::ActorUnblocked { .. } => "ActorUnblocked",
EventPayload::BackfillFollower { .. } => "BackfillFollower", EventPayload::BackfillFollower { .. } => "BackfillFollower",
EventPayload::FederationDeliveryRequested { .. } => "FederationDeliveryRequested", EventPayload::FederationDeliveryRequested { .. } => "FederationDeliveryRequested",
EventPayload::WatchEventIngested { .. } => "WatchEventIngested", EventPayload::WatchEventIngested { .. } => "WatchEventIngested",
@@ -158,6 +195,44 @@ fn parse_uuid(s: &str, field: &str) -> Result<Uuid, DomainError> {
Uuid::parse_str(s).map_err(|e| DomainError::InfrastructureError(format!("{field}: {e}"))) Uuid::parse_str(s).map_err(|e| DomainError::InfrastructureError(format!("{field}: {e}")))
} }
fn identity_to_payload(id: &SocialIdentity) -> (String, String) {
match id {
SocialIdentity::Local(uid) => ("local".into(), uid.value().to_string()),
SocialIdentity::Remote { actor_url } => ("remote".into(), actor_url.clone()),
}
}
fn follow_target_to_payload(target: &domain::value_objects::FollowTarget) -> (String, String) {
match target {
domain::value_objects::FollowTarget::Identity(id) => identity_to_payload(id),
domain::value_objects::FollowTarget::Handle(h) => ("handle".into(), h.clone()),
}
}
fn payload_to_identity(kind: &str, id: String) -> Result<SocialIdentity, DomainError> {
match kind {
"local" => Ok(SocialIdentity::Local(UserId::from_uuid(parse_uuid(
&id, "user_id",
)?))),
"remote" => Ok(SocialIdentity::Remote { actor_url: id }),
other => Err(DomainError::InfrastructureError(format!(
"unknown identity kind: {other}"
))),
}
}
fn payload_to_follow_target(
kind: &str,
id: String,
) -> Result<domain::value_objects::FollowTarget, DomainError> {
match kind {
"handle" => Ok(domain::value_objects::FollowTarget::Handle(id)),
other => Ok(domain::value_objects::FollowTarget::Identity(
payload_to_identity(other, id)?,
)),
}
}
fn parse_ts(ts: i64) -> Result<NaiveDateTime, DomainError> { fn parse_ts(ts: i64) -> Result<NaiveDateTime, DomainError> {
chrono::DateTime::from_timestamp(ts, 0) chrono::DateTime::from_timestamp(ts, 0)
.map(|dt| dt.naive_utc()) .map(|dt| dt.naive_utc())
@@ -243,15 +318,62 @@ impl From<&DomainEvent> for EventPayload {
movie_id: movie_id.value().to_string(), movie_id: movie_id.value().to_string(),
} }
} }
DomainEvent::FollowAccepted { DomainEvent::FollowRequested { follower, target } => {
local_user_id, let (kind, id) = follow_target_to_payload(target);
remote_actor_url, EventPayload::FollowRequested {
outbox_url, follower_id: follower.value().to_string(),
} => EventPayload::FollowAccepted { target_kind: kind,
local_user_id: local_user_id.value().to_string(), target_id: id,
remote_actor_url: remote_actor_url.clone(), }
outbox_url: outbox_url.clone(), }
}, DomainEvent::FollowAccepted { owner, requester } => {
let (kind, id) = identity_to_payload(requester);
EventPayload::FollowAccepted {
owner_id: owner.value().to_string(),
requester_kind: kind,
requester_id: id,
}
}
DomainEvent::FollowRejected { owner, requester } => {
let (kind, id) = identity_to_payload(requester);
EventPayload::FollowRejected {
owner_id: owner.value().to_string(),
requester_kind: kind,
requester_id: id,
}
}
DomainEvent::Unfollowed { follower, target } => {
let (kind, id) = identity_to_payload(target);
EventPayload::Unfollowed {
follower_id: follower.value().to_string(),
target_kind: kind,
target_id: id,
}
}
DomainEvent::FollowerRemoved { owner, follower } => {
let (kind, id) = identity_to_payload(follower);
EventPayload::FollowerRemoved {
owner_id: owner.value().to_string(),
follower_kind: kind,
follower_id: id,
}
}
DomainEvent::ActorBlocked { blocker, target } => {
let (kind, id) = identity_to_payload(target);
EventPayload::ActorBlocked {
blocker_id: blocker.value().to_string(),
target_kind: kind,
target_id: id,
}
}
DomainEvent::ActorUnblocked { blocker, target } => {
let (kind, id) = identity_to_payload(target);
EventPayload::ActorUnblocked {
blocker_id: blocker.value().to_string(),
target_kind: kind,
target_id: id,
}
}
DomainEvent::BackfillFollower { DomainEvent::BackfillFollower {
owner_user_id, owner_user_id,
follower_inbox_url, follower_inbox_url,
@@ -435,14 +557,61 @@ impl TryFrom<EventPayload> for DomainEvent {
movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?), movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?),
}) })
} }
EventPayload::FollowRequested {
follower_id,
target_kind,
target_id,
} => Ok(DomainEvent::FollowRequested {
follower: UserId::from_uuid(parse_uuid(&follower_id, "follower_id")?),
target: payload_to_follow_target(&target_kind, target_id)?,
}),
EventPayload::FollowAccepted { EventPayload::FollowAccepted {
local_user_id, owner_id,
remote_actor_url, requester_kind,
outbox_url, requester_id,
} => Ok(DomainEvent::FollowAccepted { } => Ok(DomainEvent::FollowAccepted {
local_user_id: UserId::from_uuid(parse_uuid(&local_user_id, "local_user_id")?), owner: UserId::from_uuid(parse_uuid(&owner_id, "owner_id")?),
remote_actor_url, requester: payload_to_identity(&requester_kind, requester_id)?,
outbox_url, }),
EventPayload::FollowRejected {
owner_id,
requester_kind,
requester_id,
} => Ok(DomainEvent::FollowRejected {
owner: UserId::from_uuid(parse_uuid(&owner_id, "owner_id")?),
requester: payload_to_identity(&requester_kind, requester_id)?,
}),
EventPayload::Unfollowed {
follower_id,
target_kind,
target_id,
} => Ok(DomainEvent::Unfollowed {
follower: UserId::from_uuid(parse_uuid(&follower_id, "follower_id")?),
target: payload_to_identity(&target_kind, target_id)?,
}),
EventPayload::FollowerRemoved {
owner_id,
follower_kind,
follower_id,
} => Ok(DomainEvent::FollowerRemoved {
owner: UserId::from_uuid(parse_uuid(&owner_id, "owner_id")?),
follower: payload_to_identity(&follower_kind, follower_id)?,
}),
EventPayload::ActorBlocked {
blocker_id,
target_kind,
target_id,
} => Ok(DomainEvent::ActorBlocked {
blocker: UserId::from_uuid(parse_uuid(&blocker_id, "blocker_id")?),
target: payload_to_identity(&target_kind, target_id)?,
}),
EventPayload::ActorUnblocked {
blocker_id,
target_kind,
target_id,
} => Ok(DomainEvent::ActorUnblocked {
blocker: UserId::from_uuid(parse_uuid(&blocker_id, "blocker_id")?),
target: payload_to_identity(&target_kind, target_id)?,
}), }),
EventPayload::BackfillFollower { EventPayload::BackfillFollower {
owner_user_id, owner_user_id,

View File

@@ -12,7 +12,13 @@ pub fn event_to_subject(prefix: &str, event: &DomainEvent) -> String {
DomainEvent::ImageStored { .. } => "image.stored", DomainEvent::ImageStored { .. } => "image.stored",
DomainEvent::WatchlistEntryAdded { .. } => "watchlist.entry.added", DomainEvent::WatchlistEntryAdded { .. } => "watchlist.entry.added",
DomainEvent::WatchlistEntryRemoved { .. } => "watchlist.entry.removed", DomainEvent::WatchlistEntryRemoved { .. } => "watchlist.entry.removed",
DomainEvent::FollowRequested { .. } => "follow.requested",
DomainEvent::FollowAccepted { .. } => "follow.accepted", DomainEvent::FollowAccepted { .. } => "follow.accepted",
DomainEvent::FollowRejected { .. } => "follow.rejected",
DomainEvent::Unfollowed { .. } => "follow.unfollowed",
DomainEvent::FollowerRemoved { .. } => "follower.removed",
DomainEvent::ActorBlocked { .. } => "actor.blocked",
DomainEvent::ActorUnblocked { .. } => "actor.unblocked",
DomainEvent::BackfillFollower { .. } => "backfill.follower", DomainEvent::BackfillFollower { .. } => "backfill.follower",
DomainEvent::FederationDeliveryRequested { .. } => "federation.delivery.requested", DomainEvent::FederationDeliveryRequested { .. } => "federation.delivery.requested",
DomainEvent::WatchEventIngested { .. } => "watch.event.ingested", DomainEvent::WatchEventIngested { .. } => "watch.event.ingested",

View File

@@ -1,26 +1,10 @@
use async_trait::async_trait; use async_trait::async_trait;
use domain::{ use domain::{errors::DomainError, models::RemoteActorInfo, ports::FederationAdminQuery};
errors::DomainError,
models::{PendingFollowerInfo, RemoteActorInfo},
ports::SocialQueryPort,
value_objects::UserId,
};
use super::PostgresFederationRepository; use super::PostgresFederationRepository;
#[async_trait] #[async_trait]
impl SocialQueryPort for PostgresFederationRepository { impl FederationAdminQuery for PostgresFederationRepository {
async fn get_accepted_following_urls(
&self,
user_id: &UserId,
) -> Result<Vec<String>, DomainError> {
let user_id_str = user_id.value().to_string();
sqlx::query_scalar::<_, String>(
"SELECT remote_actor_url FROM ap_following WHERE local_user_id = $1 AND status = 'accepted'",
).bind(&user_id_str).fetch_all(&self.pool).await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError> { async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError> {
let rows = sqlx::query_as::<_, (String, String, Option<String>)>( let rows = sqlx::query_as::<_, (String, String, Option<String>)>(
"SELECT DISTINCT ar.url, ar.handle, ar.display_name FROM ap_remote_actors ar JOIN ap_following f ON f.remote_actor_url = ar.url WHERE f.status = 'accepted'", "SELECT DISTINCT ar.url, ar.handle, ar.display_name FROM ap_remote_actors ar JOIN ap_following f ON f.remote_actor_url = ar.url WHERE f.status = 'accepted'",
@@ -34,49 +18,4 @@ impl SocialQueryPort for PostgresFederationRepository {
}) })
.collect()) .collect())
} }
async fn count_following(&self, user_id: &UserId) -> Result<usize, DomainError> {
let uid = user_id.value().to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_following WHERE local_user_id = $1 AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn count_accepted_followers(&self, user_id: &UserId) -> Result<usize, DomainError> {
let uid = user_id.value().to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_followers WHERE local_user_id = $1 AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn get_pending_followers(
&self,
user_id: &UserId,
) -> Result<Vec<PendingFollowerInfo>, DomainError> {
let uid = user_id.value().to_string();
let rows = sqlx::query_as::<_, (String, String, Option<String>, Option<String>)>(
"SELECT ar.url, ar.handle, ar.display_name, ar.avatar_url FROM ap_followers f JOIN ap_remote_actors ar ON ar.url = f.remote_actor_url WHERE f.local_user_id = $1 AND f.status = 'pending'",
).bind(&uid).fetch_all(&self.pool).await.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(
|(url, handle, display_name, avatar_url)| PendingFollowerInfo {
url,
handle,
display_name,
avatar_url,
},
)
.collect())
}
} }

View File

@@ -1,26 +1,10 @@
use async_trait::async_trait; use async_trait::async_trait;
use domain::{ use domain::{errors::DomainError, models::RemoteActorInfo, ports::FederationAdminQuery};
errors::DomainError,
models::{PendingFollowerInfo, RemoteActorInfo},
ports::SocialQueryPort,
value_objects::UserId,
};
use super::SqliteFederationRepository; use super::SqliteFederationRepository;
#[async_trait] #[async_trait]
impl SocialQueryPort for SqliteFederationRepository { impl FederationAdminQuery for SqliteFederationRepository {
async fn get_accepted_following_urls(
&self,
user_id: &UserId,
) -> Result<Vec<String>, DomainError> {
let user_id_str = user_id.value().to_string();
sqlx::query_scalar::<_, String>(
"SELECT remote_actor_url FROM ap_following WHERE local_user_id = ? AND status = 'accepted'",
).bind(&user_id_str).fetch_all(&self.pool).await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError> { async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError> {
let rows = sqlx::query_as::<_, (String, String, Option<String>)>( let rows = sqlx::query_as::<_, (String, String, Option<String>)>(
"SELECT DISTINCT ar.url, ar.handle, ar.display_name "SELECT DISTINCT ar.url, ar.handle, ar.display_name
@@ -40,56 +24,4 @@ impl SocialQueryPort for SqliteFederationRepository {
}) })
.collect()) .collect())
} }
async fn count_following(&self, user_id: &UserId) -> Result<usize, DomainError> {
let uid = user_id.value().to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_following WHERE local_user_id = ? AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn count_accepted_followers(&self, user_id: &UserId) -> Result<usize, DomainError> {
let uid = user_id.value().to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_followers WHERE local_user_id = ? AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn get_pending_followers(
&self,
user_id: &UserId,
) -> Result<Vec<PendingFollowerInfo>, DomainError> {
let uid = user_id.value().to_string();
let rows = sqlx::query_as::<_, (String, String, Option<String>, Option<String>)>(
"SELECT ar.url, ar.handle, ar.display_name, ar.avatar_url
FROM ap_followers f
JOIN ap_remote_actors ar ON ar.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.status = 'pending'",
)
.bind(&uid)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(
|(url, handle, display_name, avatar_url)| PendingFollowerInfo {
url,
handle,
display_name,
avatar_url,
},
)
.collect())
}
} }

View File

@@ -1,6 +1,6 @@
use super::*; use super::*;
use chrono::Utc; use chrono::Utc;
use domain::ports::SocialQueryPort; use domain::ports::FederationAdminQuery;
use k_ap::ActorRepository; use k_ap::ActorRepository;
use sqlx::SqlitePool; use sqlx::SqlitePool;
@@ -79,30 +79,6 @@ async fn setup_db(pool: &SqlitePool) {
.unwrap(); .unwrap();
} }
#[tokio::test]
async fn test_get_accepted_following_urls_returns_only_accepted() {
let pool = SqlitePool::connect(":memory:").await.unwrap();
setup_db(&pool).await;
let repo = SqliteFederationRepository::new(pool.clone());
let user_id = uuid::Uuid::new_v4();
sqlx::query(
"INSERT INTO ap_following (local_user_id, remote_actor_url, follow_activity_id, status)
VALUES (?, 'https://other.social/users/alice', 'act1', 'accepted'),
(?, 'https://other.social/users/bob', 'act2', 'pending')",
)
.bind(user_id.to_string())
.bind(user_id.to_string())
.execute(&pool)
.await
.unwrap();
let uid = domain::value_objects::UserId::from_uuid(user_id);
let urls = repo.get_accepted_following_urls(&uid).await.unwrap();
assert_eq!(urls.len(), 1);
assert_eq!(urls[0], "https://other.social/users/alice");
}
#[tokio::test] #[tokio::test]
async fn test_list_all_followed_remote_actors_deduplicates() { async fn test_list_all_followed_remote_actors_deduplicates() {
let pool = SqlitePool::connect(":memory:").await.unwrap(); let pool = SqlitePool::connect(":memory:").await.unwrap();

View File

@@ -2,7 +2,7 @@ use std::sync::Arc;
use domain::ports::{ use domain::ports::{
DiaryQuery, EventPublisher, MovieCommand, MovieProfileRepository, MovieQuery, ReviewRepository, DiaryQuery, EventPublisher, MovieCommand, MovieProfileRepository, MovieQuery, ReviewRepository,
SocialQueryPort, SocialQuery,
}; };
use crate::config::AppConfig; use crate::config::AppConfig;
@@ -27,6 +27,6 @@ pub struct GetMovieSocialPageDeps {
pub struct GetActivityFeedDeps { pub struct GetActivityFeedDeps {
pub diary: Arc<dyn DiaryQuery>, pub diary: Arc<dyn DiaryQuery>,
pub social_query: Arc<dyn SocialQueryPort>, pub social_query: Arc<dyn SocialQuery>,
pub config: AppConfig, pub config: AppConfig,
} }

View File

@@ -6,7 +6,7 @@ use domain::{
FeedEntry, FeedEntry,
collections::{PageParams, Paginated}, collections::{PageParams, Paginated},
}, },
value_objects::UserId, value_objects::{SocialIdentity, UserId},
}; };
pub async fn execute( pub async fn execute(
@@ -36,28 +36,24 @@ async fn build_following_filter(
} }
let viewer_id = query.viewer_user_id?; let viewer_id = query.viewer_user_id?;
let viewer = UserId::from_uuid(viewer_id); let viewer = UserId::from_uuid(viewer_id);
let urls = deps let actors = deps
.social_query .social_query
.get_accepted_following_urls(&viewer) .get_following(&viewer)
.await .await
.unwrap_or_default(); .unwrap_or_default();
if urls.is_empty() { if actors.is_empty() {
return Some(FollowingFilter { return Some(FollowingFilter {
local_user_ids: vec![viewer_id], local_user_ids: vec![viewer_id],
remote_actor_urls: vec![], remote_actor_urls: vec![],
}); });
} }
let base_url = &deps.config.base_url;
let mut local_ids = vec![viewer_id]; let mut local_ids = vec![viewer_id];
let mut remote_urls = Vec::new(); let mut remote_urls = Vec::new();
for url in urls { for actor in actors {
if let Some(suffix) = url.strip_prefix(&format!("{}/users/", base_url)) match actor.identity {
&& let Ok(parsed_id) = uuid::Uuid::parse_str(suffix) SocialIdentity::Local(uid) => local_ids.push(uid.value()),
{ SocialIdentity::Remote { actor_url } => remote_urls.push(actor_url),
local_ids.push(parsed_id);
continue;
} }
remote_urls.push(url);
} }
Some(FollowingFilter { Some(FollowingFilter {
local_user_ids: local_ids, local_user_ids: local_ids,

View File

@@ -2,7 +2,8 @@ use std::sync::Arc;
use async_trait::async_trait; use async_trait::async_trait;
use domain::errors::DomainError; use domain::errors::DomainError;
use domain::testing::{FakeDiaryQuery, NoopSocialQueryPort}; use domain::testing::InMemorySocialRepository;
use domain::value_objects::{SocialActor, SocialIdentity, UserId};
use crate::{ use crate::{
config::AppConfig, diary::deps::GetActivityFeedDeps, diary::get_activity_feed, config::AppConfig, diary::deps::GetActivityFeedDeps, diary::get_activity_feed,
@@ -11,8 +12,8 @@ use crate::{
fn default_deps() -> GetActivityFeedDeps { fn default_deps() -> GetActivityFeedDeps {
GetActivityFeedDeps { GetActivityFeedDeps {
diary: FakeDiaryQuery::new() as _, diary: domain::testing::FakeDiaryQuery::new() as _,
social_query: Arc::new(NoopSocialQueryPort), social_query: InMemorySocialRepository::new() as _,
config: TestContextBuilder::new().config, config: TestContextBuilder::new().config,
} }
} }
@@ -59,60 +60,62 @@ async fn returns_feed_with_following_filter() {
.await .await
.unwrap(); .unwrap();
// NoopSocialQueryPort returns empty following, so FollowingFilter
// contains only the viewer's id. Feed is empty but the code path is hit.
assert!(result.items.is_empty()); assert!(result.items.is_empty());
} }
struct FakeSocialWithFollowing(Vec<String>); struct FakeSocialWithFollowing(Vec<SocialActor>);
#[async_trait] #[async_trait]
impl domain::ports::SocialQueryPort for FakeSocialWithFollowing { impl domain::ports::SocialQuery for FakeSocialWithFollowing {
async fn get_accepted_following_urls( async fn get_following(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
&self,
_: &domain::value_objects::UserId,
) -> Result<Vec<String>, DomainError> {
Ok(self.0.clone()) Ok(self.0.clone())
} }
async fn count_following( async fn get_followers(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
&self,
_: &domain::value_objects::UserId,
) -> Result<usize, DomainError> {
Ok(0)
}
async fn count_accepted_followers(
&self,
_: &domain::value_objects::UserId,
) -> Result<usize, DomainError> {
Ok(0)
}
async fn get_pending_followers(
&self,
_: &domain::value_objects::UserId,
) -> Result<Vec<domain::models::PendingFollowerInfo>, DomainError> {
Ok(vec![]) Ok(vec![])
} }
async fn list_all_followed_remote_actors( async fn get_pending_followers(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
&self,
) -> Result<Vec<domain::models::RemoteActorInfo>, DomainError> {
Ok(vec![]) Ok(vec![])
} }
async fn count_following(&self, _: &UserId) -> Result<usize, DomainError> {
Ok(0)
}
async fn count_followers(&self, _: &UserId) -> Result<usize, DomainError> {
Ok(0)
}
async fn get_blocked(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
Ok(vec![])
}
async fn is_following(&self, _: &UserId, _: &SocialIdentity) -> Result<bool, DomainError> {
Ok(false)
}
} }
#[tokio::test] #[tokio::test]
async fn following_filter_parses_local_and_remote_urls() { async fn following_filter_separates_local_and_remote() {
let viewer = uuid::Uuid::new_v4(); let viewer = uuid::Uuid::new_v4();
let local_friend = uuid::Uuid::new_v4(); let local_friend = uuid::Uuid::new_v4();
let following_urls = vec![ let following = vec![
format!("http://localhost:3000/users/{}", local_friend), SocialActor {
"https://remote.example/actor/1".to_string(), identity: SocialIdentity::Local(UserId::from_uuid(local_friend)),
handle: "friend".into(),
display_name: None,
avatar_url: None,
},
SocialActor {
identity: SocialIdentity::Remote {
actor_url: "https://remote.example/actor/1".into(),
},
handle: "@alice@remote.example".into(),
display_name: None,
avatar_url: None,
},
]; ];
let social = Arc::new(FakeSocialWithFollowing(following_urls)); let social = Arc::new(FakeSocialWithFollowing(following));
let deps = GetActivityFeedDeps { let deps = GetActivityFeedDeps {
diary: FakeDiaryQuery::new() as _, diary: domain::testing::FakeDiaryQuery::new() as _,
social_query: social as _, social_query: social as _,
config: AppConfig { config: AppConfig {
allow_registration: true, allow_registration: true,
@@ -141,8 +144,6 @@ async fn following_filter_parses_local_and_remote_urls() {
.await .await
.unwrap(); .unwrap();
// Feed is empty (no data seeded), but the build_following_filter code path
// with actual URL parsing ran without errors.
assert!(result.items.is_empty()); assert!(result.items.is_empty());
} }
@@ -164,6 +165,5 @@ async fn following_filter_without_viewer_returns_none() {
.await .await
.unwrap(); .unwrap();
// filter_following=true but viewer_user_id=None → build_following_filter returns None
assert!(result.items.is_empty()); assert!(result.items.is_empty());
} }

View File

@@ -11,6 +11,7 @@ pub mod integrations;
pub mod movies; pub mod movies;
pub mod person; pub mod person;
pub mod search; pub mod search;
pub mod social;
pub mod users; pub mod users;
pub mod watchlist; pub mod watchlist;
pub mod wrapup; pub mod wrapup;

View File

@@ -0,0 +1,33 @@
use domain::value_objects::{FollowTarget, SocialIdentity};
use uuid::Uuid;
pub enum SocialCmd {
Follow {
follower_id: Uuid,
target: FollowTarget,
},
Unfollow {
follower_id: Uuid,
target: SocialIdentity,
},
AcceptFollow {
owner_id: Uuid,
requester: SocialIdentity,
},
RejectFollow {
owner_id: Uuid,
requester: SocialIdentity,
},
RemoveFollower {
owner_id: Uuid,
follower: SocialIdentity,
},
Block {
blocker_id: Uuid,
target: SocialIdentity,
},
Unblock {
blocker_id: Uuid,
target: SocialIdentity,
},
}

View File

@@ -0,0 +1,13 @@
use std::sync::Arc;
use domain::ports::{EventPublisher, SocialCommand, SocialQuery};
pub struct SocialCommandDeps {
pub social_command: Arc<dyn SocialCommand>,
pub social_query: Arc<dyn SocialQuery>,
pub event_publisher: Arc<dyn EventPublisher>,
}
pub struct SocialQueryDeps {
pub social_query: Arc<dyn SocialQuery>,
}

View File

@@ -0,0 +1,92 @@
use domain::{
errors::DomainError,
events::DomainEvent,
value_objects::{SocialActor, UserId},
};
use super::{
commands::SocialCmd,
deps::{SocialCommandDeps, SocialQueryDeps},
queries::SocialQry,
};
pub async fn execute_command(deps: &SocialCommandDeps, cmd: SocialCmd) -> Result<(), DomainError> {
let event = match cmd {
SocialCmd::Follow {
follower_id,
target,
} => {
let follower = UserId::from_uuid(follower_id);
deps.social_command.follow(&follower, &target).await?;
DomainEvent::FollowRequested { follower, target }
}
SocialCmd::Unfollow {
follower_id,
target,
} => {
let follower = UserId::from_uuid(follower_id);
deps.social_command.unfollow(&follower, &target).await?;
DomainEvent::Unfollowed { follower, target }
}
SocialCmd::AcceptFollow {
owner_id,
requester,
} => {
let owner = UserId::from_uuid(owner_id);
deps.social_command
.accept_follow(&owner, &requester)
.await?;
DomainEvent::FollowAccepted { owner, requester }
}
SocialCmd::RejectFollow {
owner_id,
requester,
} => {
let owner = UserId::from_uuid(owner_id);
deps.social_command
.reject_follow(&owner, &requester)
.await?;
DomainEvent::FollowRejected { owner, requester }
}
SocialCmd::RemoveFollower { owner_id, follower } => {
let owner = UserId::from_uuid(owner_id);
deps.social_command
.remove_follower(&owner, &follower)
.await?;
DomainEvent::FollowerRemoved { owner, follower }
}
SocialCmd::Block { blocker_id, target } => {
let blocker = UserId::from_uuid(blocker_id);
deps.social_command.block(&blocker, &target).await?;
DomainEvent::ActorBlocked { blocker, target }
}
SocialCmd::Unblock { blocker_id, target } => {
let blocker = UserId::from_uuid(blocker_id);
deps.social_command.unblock(&blocker, &target).await?;
DomainEvent::ActorUnblocked { blocker, target }
}
};
deps.event_publisher.publish(&event).await
}
pub async fn execute_query(
deps: &SocialQueryDeps,
query: SocialQry,
) -> Result<Vec<SocialActor>, DomainError> {
let user_id = match &query {
SocialQry::GetFollowing { user_id }
| SocialQry::GetFollowers { user_id }
| SocialQry::GetPending { user_id }
| SocialQry::GetBlocked { user_id } => UserId::from_uuid(*user_id),
};
match query {
SocialQry::GetFollowing { .. } => deps.social_query.get_following(&user_id).await,
SocialQry::GetFollowers { .. } => deps.social_query.get_followers(&user_id).await,
SocialQry::GetPending { .. } => deps.social_query.get_pending_followers(&user_id).await,
SocialQry::GetBlocked { .. } => deps.social_query.get_blocked(&user_id).await,
}
}
#[cfg(test)]
#[path = "tests/execute.rs"]
mod tests;

View File

@@ -0,0 +1,4 @@
pub mod commands;
pub mod deps;
pub mod execute;
pub mod queries;

View File

@@ -0,0 +1,8 @@
use uuid::Uuid;
pub enum SocialQry {
GetFollowing { user_id: Uuid },
GetFollowers { user_id: Uuid },
GetPending { user_id: Uuid },
GetBlocked { user_id: Uuid },
}

View File

@@ -0,0 +1,449 @@
use std::sync::Arc;
use domain::{
events::DomainEvent,
testing::{InMemorySocialRepository, NoopEventPublisher},
value_objects::{FollowTarget, SocialIdentity, UserId},
};
use uuid::Uuid;
use crate::social::{
commands::SocialCmd,
deps::{SocialCommandDeps, SocialQueryDeps},
execute::{execute_command, execute_query},
queries::SocialQry,
};
fn make_cmd_deps() -> (
Arc<InMemorySocialRepository>,
Arc<NoopEventPublisher>,
SocialCommandDeps,
) {
let social = InMemorySocialRepository::new();
let events = NoopEventPublisher::new();
let deps = SocialCommandDeps {
social_command: Arc::clone(&social) as _,
social_query: Arc::clone(&social) as _,
event_publisher: Arc::clone(&events) as _,
};
(social, events, deps)
}
// ── Follow ──────────────────────────────────────────────────────────────────
#[tokio::test]
async fn follow_emits_follow_requested_event() {
let (_social, events, deps) = make_cmd_deps();
execute_command(
&deps,
SocialCmd::Follow {
follower_id: Uuid::new_v4(),
target: FollowTarget::Identity(SocialIdentity::Local(
UserId::from_uuid(Uuid::new_v4()),
)),
},
)
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::FollowRequested { .. }))
);
}
#[tokio::test]
async fn cannot_follow_yourself() {
let (_social, _events, deps) = make_cmd_deps();
let user_id = Uuid::new_v4();
let result = execute_command(
&deps,
SocialCmd::Follow {
follower_id: user_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(user_id))),
},
)
.await;
assert!(result.is_err());
}
#[tokio::test]
async fn cannot_follow_same_target_twice() {
let (_social, _events, deps) = make_cmd_deps();
let follower_id = Uuid::new_v4();
let target = FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4())));
execute_command(
&deps,
SocialCmd::Follow {
follower_id,
target: target.clone(),
},
)
.await
.unwrap();
let result = execute_command(
&deps,
SocialCmd::Follow {
follower_id,
target,
},
)
.await;
assert!(result.is_err());
}
// ── Unfollow ────────────────────────────────────────────────────────────────
#[tokio::test]
async fn unfollow_emits_unfollowed_event() {
let (_social, events, deps) = make_cmd_deps();
let follower_id = Uuid::new_v4();
let target = SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4()));
execute_command(
&deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(target.clone()),
},
)
.await
.unwrap();
execute_command(
&deps,
SocialCmd::Unfollow {
follower_id,
target,
},
)
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::Unfollowed { .. }))
);
}
// ── Accept ──────────────────────────────────────────────────────────────────
#[tokio::test]
async fn accept_follow_emits_follow_accepted_event() {
let (_social, events, deps) = make_cmd_deps();
let follower_id = Uuid::new_v4();
let owner_id = Uuid::new_v4();
let requester = SocialIdentity::Local(UserId::from_uuid(follower_id));
execute_command(
&deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(owner_id))),
},
)
.await
.unwrap();
execute_command(
&deps,
SocialCmd::AcceptFollow {
owner_id,
requester,
},
)
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::FollowAccepted { .. }))
);
}
// ── Reject ──────────────────────────────────────────────────────────────────
#[tokio::test]
async fn reject_follow_emits_follow_rejected_event() {
let (_social, events, deps) = make_cmd_deps();
let follower_id = Uuid::new_v4();
let owner_id = Uuid::new_v4();
execute_command(
&deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(owner_id))),
},
)
.await
.unwrap();
execute_command(
&deps,
SocialCmd::RejectFollow {
owner_id,
requester: SocialIdentity::Local(UserId::from_uuid(follower_id)),
},
)
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::FollowRejected { .. }))
);
}
// ── Remove follower ─────────────────────────────────────────────────────────
#[tokio::test]
async fn remove_follower_emits_follower_removed_event() {
let (_social, events, deps) = make_cmd_deps();
let follower_id = Uuid::new_v4();
let owner_id = Uuid::new_v4();
execute_command(
&deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(owner_id))),
},
)
.await
.unwrap();
execute_command(
&deps,
SocialCmd::AcceptFollow {
owner_id,
requester: SocialIdentity::Local(UserId::from_uuid(follower_id)),
},
)
.await
.unwrap();
execute_command(
&deps,
SocialCmd::RemoveFollower {
owner_id,
follower: SocialIdentity::Local(UserId::from_uuid(follower_id)),
},
)
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::FollowerRemoved { .. }))
);
}
// ── Block ───────────────────────────────────────────────────────────────────
#[tokio::test]
async fn block_emits_actor_blocked_event() {
let (_social, events, deps) = make_cmd_deps();
execute_command(
&deps,
SocialCmd::Block {
blocker_id: Uuid::new_v4(),
target: SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4())),
},
)
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::ActorBlocked { .. }))
);
}
// ── Unblock ─────────────────────────────────────────────────────────────────
#[tokio::test]
async fn unblock_emits_actor_unblocked_event() {
let (_social, events, deps) = make_cmd_deps();
let target = SocialIdentity::Local(UserId::from_uuid(Uuid::new_v4()));
let blocker_id = Uuid::new_v4();
execute_command(
&deps,
SocialCmd::Block {
blocker_id,
target: target.clone(),
},
)
.await
.unwrap();
execute_command(&deps, SocialCmd::Unblock { blocker_id, target })
.await
.unwrap();
let published = events.published();
assert!(
published
.iter()
.any(|e| matches!(e, DomainEvent::ActorUnblocked { .. }))
);
}
// ── Get following ───────────────────────────────────────────────────────────
#[tokio::test]
async fn returns_accepted_follows() {
let social = InMemorySocialRepository::new();
let events = NoopEventPublisher::new();
let cmd_deps = SocialCommandDeps {
social_command: Arc::clone(&social) as _,
social_query: Arc::clone(&social) as _,
event_publisher: Arc::clone(&events) as _,
};
let query_deps = SocialQueryDeps {
social_query: Arc::clone(&social) as _,
};
let follower_id = Uuid::new_v4();
let target_id = Uuid::new_v4();
execute_command(
&cmd_deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(target_id))),
},
)
.await
.unwrap();
// Pending follow should not appear
let following = execute_query(
&query_deps,
SocialQry::GetFollowing {
user_id: follower_id,
},
)
.await
.unwrap();
assert!(following.is_empty());
// Accept, then it should appear
execute_command(
&cmd_deps,
SocialCmd::AcceptFollow {
owner_id: target_id,
requester: SocialIdentity::Local(UserId::from_uuid(follower_id)),
},
)
.await
.unwrap();
let following = execute_query(
&query_deps,
SocialQry::GetFollowing {
user_id: follower_id,
},
)
.await
.unwrap();
assert_eq!(following.len(), 1);
}
// ── Get followers ───────────────────────────────────────────────────────────
#[tokio::test]
async fn returns_accepted_followers() {
let social = InMemorySocialRepository::new();
let events = NoopEventPublisher::new();
let cmd_deps = SocialCommandDeps {
social_command: Arc::clone(&social) as _,
social_query: Arc::clone(&social) as _,
event_publisher: Arc::clone(&events) as _,
};
let query_deps = SocialQueryDeps {
social_query: Arc::clone(&social) as _,
};
let follower_id = Uuid::new_v4();
let owner_id = Uuid::new_v4();
execute_command(
&cmd_deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(owner_id))),
},
)
.await
.unwrap();
execute_command(
&cmd_deps,
SocialCmd::AcceptFollow {
owner_id,
requester: SocialIdentity::Local(UserId::from_uuid(follower_id)),
},
)
.await
.unwrap();
let followers = execute_query(&query_deps, SocialQry::GetFollowers { user_id: owner_id })
.await
.unwrap();
assert_eq!(followers.len(), 1);
}
// ── Get pending ─────────────────────────────────────────────────────────────
#[tokio::test]
async fn returns_only_pending_followers() {
let social = InMemorySocialRepository::new();
let events = NoopEventPublisher::new();
let cmd_deps = SocialCommandDeps {
social_command: Arc::clone(&social) as _,
social_query: Arc::clone(&social) as _,
event_publisher: Arc::clone(&events) as _,
};
let query_deps = SocialQueryDeps {
social_query: Arc::clone(&social) as _,
};
let follower_id = Uuid::new_v4();
let owner_id = Uuid::new_v4();
execute_command(
&cmd_deps,
SocialCmd::Follow {
follower_id,
target: FollowTarget::Identity(SocialIdentity::Local(UserId::from_uuid(owner_id))),
},
)
.await
.unwrap();
let pending = execute_query(&query_deps, SocialQry::GetPending { user_id: owner_id })
.await
.unwrap();
assert_eq!(pending.len(), 1);
}

View File

@@ -1,7 +1,8 @@
use std::sync::Arc; use std::sync::Arc;
use domain::testing::{ use domain::testing::{
InMemoryGoalRepository, InMemoryWrapUpRepository, InMemoryWrapUpStatsQuery, NoopSocialQueryPort, InMemoryGoalRepository, InMemorySocialRepository, InMemoryWrapUpRepository,
InMemoryWrapUpStatsQuery, NoopFederationAdminQuery,
}; };
use domain::{ use domain::{
ports::{ ports::{
@@ -72,7 +73,9 @@ pub struct TestContextBuilder {
pub goal_query: Arc<dyn GoalQuery>, pub goal_query: Arc<dyn GoalQuery>,
pub user_settings_repo: Arc<dyn UserSettingsRepository>, pub user_settings_repo: Arc<dyn UserSettingsRepository>,
pub review_logger: Arc<dyn ReviewLogger>, pub review_logger: Arc<dyn ReviewLogger>,
pub social_query: Arc<dyn domain::ports::SocialQueryPort>, pub social_command: Arc<dyn domain::ports::SocialCommand>,
pub social_query_unified: Arc<dyn domain::ports::SocialQuery>,
pub federation_admin: Arc<dyn domain::ports::FederationAdminQuery>,
pub refresh_session_repo: Arc<dyn RefreshSessionRepository>, pub refresh_session_repo: Arc<dyn RefreshSessionRepository>,
pub config: AppConfig, pub config: AppConfig,
} }
@@ -88,6 +91,7 @@ impl TestContextBuilder {
let movies = InMemoryMovieRepository::new(); let movies = InMemoryMovieRepository::new();
let watch_events = InMemoryWatchEventRepository::new(); let watch_events = InMemoryWatchEventRepository::new();
let goals = InMemoryGoalRepository::new(); let goals = InMemoryGoalRepository::new();
let social = InMemorySocialRepository::new();
Self { Self {
movie_command: Arc::clone(&movies) as _, movie_command: Arc::clone(&movies) as _,
movie_query: movies as _, movie_query: movies as _,
@@ -121,7 +125,9 @@ impl TestContextBuilder {
goal_query: goals as _, goal_query: goals as _,
user_settings_repo: InMemoryUserSettingsRepository::new(), user_settings_repo: InMemoryUserSettingsRepository::new(),
review_logger: Arc::new(NoopReviewLogger), review_logger: Arc::new(NoopReviewLogger),
social_query: Arc::new(NoopSocialQueryPort), social_command: Arc::clone(&social) as _,
social_query_unified: Arc::clone(&social) as _,
federation_admin: Arc::new(NoopFederationAdminQuery),
refresh_session_repo: InMemoryRefreshSessionRepository::new(), refresh_session_repo: InMemoryRefreshSessionRepository::new(),
config: AppConfig { config: AppConfig {
allow_registration: true, allow_registration: true,
@@ -267,11 +273,6 @@ impl TestContextBuilder {
self self
} }
pub fn with_social_query(mut self, r: Arc<dyn domain::ports::SocialQueryPort>) -> Self {
self.social_query = r;
self
}
pub fn with_wrapup_repo(mut self, r: Arc<dyn WrapUpRepository>) -> Self { pub fn with_wrapup_repo(mut self, r: Arc<dyn WrapUpRepository>) -> Self {
self.wrapup_repo = r; self.wrapup_repo = r;
self self

View File

@@ -72,7 +72,13 @@ impl EventHandler for RecordingHandler {
DomainEvent::WatchlistEntryAdded { .. } | DomainEvent::WatchlistEntryRemoved { .. } => { DomainEvent::WatchlistEntryAdded { .. } | DomainEvent::WatchlistEntryRemoved { .. } => {
"watchlist" "watchlist"
} }
DomainEvent::FollowRequested { .. } => "follow_requested",
DomainEvent::FollowAccepted { .. } => "follow_accepted", DomainEvent::FollowAccepted { .. } => "follow_accepted",
DomainEvent::FollowRejected { .. } => "follow_rejected",
DomainEvent::Unfollowed { .. } => "unfollowed",
DomainEvent::FollowerRemoved { .. } => "follower_removed",
DomainEvent::ActorBlocked { .. } => "actor_blocked",
DomainEvent::ActorUnblocked { .. } => "actor_unblocked",
DomainEvent::BackfillFollower { .. } => "backfill_follower", DomainEvent::BackfillFollower { .. } => "backfill_follower",
DomainEvent::FederationDeliveryRequested { .. } => "federation_delivery", DomainEvent::FederationDeliveryRequested { .. } => "federation_delivery",
DomainEvent::WatchEventIngested { .. } => "watch_event_ingested", DomainEvent::WatchEventIngested { .. } => "watch_event_ingested",

View File

@@ -1,13 +1,19 @@
use std::sync::Arc; use std::sync::Arc;
use domain::ports::{ use domain::ports::{
DiaryQuery, EventPublisher, ObjectStorage, SocialQueryPort, StatsRepository, UserRepository, DiaryQuery, EventPublisher, FederationAdminQuery, ObjectStorage, SocialQuery, StatsRepository,
UserRepository,
}; };
pub struct GetProfileDeps { pub struct GetProfileDeps {
pub stats: Arc<dyn StatsRepository>, pub stats: Arc<dyn StatsRepository>,
pub diary: Arc<dyn DiaryQuery>, pub diary: Arc<dyn DiaryQuery>,
pub social_query: Arc<dyn SocialQueryPort>, pub social_query: Arc<dyn SocialQuery>,
}
pub struct GetUsersListDeps {
pub user: Arc<dyn UserRepository>,
pub federation_admin: Arc<dyn FederationAdminQuery>,
} }
pub struct UpdateProfileDeps { pub struct UpdateProfileDeps {

View File

@@ -86,7 +86,7 @@ async fn load_social_counts(
.unwrap_or(0); .unwrap_or(0);
let followers = deps let followers = deps
.social_query .social_query
.count_accepted_followers(user_id) .count_followers(user_id)
.await .await
.unwrap_or(0); .unwrap_or(0);
if !is_own_profile { if !is_own_profile {
@@ -98,11 +98,19 @@ async fn load_social_counts(
.await .await
.unwrap_or_default() .unwrap_or_default()
.into_iter() .into_iter()
.map(|p| PendingFollowerView { .map(|p| {
url: p.url, let url = match &p.identity {
domain::value_objects::SocialIdentity::Remote { actor_url } => actor_url.clone(),
domain::value_objects::SocialIdentity::Local(uid) => {
format!("local:{}", uid.value())
}
};
PendingFollowerView {
url,
handle: p.handle, handle: p.handle,
display_name: p.display_name, display_name: p.display_name,
avatar_url: p.avatar_url, avatar_url: p.avatar_url,
}
}) })
.collect(); .collect();
(following, followers, pending) (following, followers, pending)

View File

@@ -1,10 +1,7 @@
use std::sync::Arc; use crate::users::{deps::GetUsersListDeps, queries::GetUsersQuery};
use crate::users::queries::GetUsersQuery;
use domain::{ use domain::{
errors::DomainError, errors::DomainError,
models::{RemoteActorInfo, UserSummary}, models::{RemoteActorInfo, UserSummary},
ports::{SocialQueryPort, UserRepository},
}; };
pub struct UsersListData { pub struct UsersListData {
@@ -13,13 +10,12 @@ pub struct UsersListData {
} }
pub async fn execute( pub async fn execute(
user: Arc<dyn UserRepository>, deps: &GetUsersListDeps,
social_query: Arc<dyn SocialQueryPort>,
_query: GetUsersQuery, _query: GetUsersQuery,
) -> Result<UsersListData, DomainError> { ) -> Result<UsersListData, DomainError> {
let (users_result, actors_result) = tokio::join!( let (users_result, actors_result) = tokio::join!(
user.list_with_stats(), deps.user.list_with_stats(),
social_query.list_all_followed_remote_actors() deps.federation_admin.list_all_followed_remote_actors()
); );
Ok(UsersListData { Ok(UsersListData {
@@ -27,7 +23,3 @@ pub async fn execute(
remote_actors: actors_result?, remote_actors: actors_result?,
}) })
} }
#[cfg(test)]
#[path = "tests/get_users.rs"]
mod tests;

View File

@@ -35,7 +35,7 @@ async fn returns_profile_with_empty_stats() {
let deps = GetProfileDeps { let deps = GetProfileDeps {
stats: b.stats_repo.clone(), stats: b.stats_repo.clone(),
diary: b.diary_repo.clone(), diary: b.diary_repo.clone(),
social_query: b.social_query.clone(), social_query: b.social_query_unified.clone(),
}; };
setup_user(&b, "profile@test.com", "profuser").await; setup_user(&b, "profile@test.com", "profuser").await;
@@ -70,7 +70,7 @@ async fn returns_history_view() {
let deps = GetProfileDeps { let deps = GetProfileDeps {
stats: b.stats_repo.clone(), stats: b.stats_repo.clone(),
diary: b.diary_repo.clone(), diary: b.diary_repo.clone(),
social_query: b.social_query.clone(), social_query: b.social_query_unified.clone(),
}; };
setup_user(&b, "hist@test.com", "histuser").await; setup_user(&b, "hist@test.com", "histuser").await;
@@ -107,7 +107,7 @@ async fn returns_trends_view() {
let deps = GetProfileDeps { let deps = GetProfileDeps {
stats: b.stats_repo.clone(), stats: b.stats_repo.clone(),
diary: b.diary_repo.clone(), diary: b.diary_repo.clone(),
social_query: b.social_query.clone(), social_query: b.social_query_unified.clone(),
}; };
setup_user(&b, "trends@test.com", "trendsuser").await; setup_user(&b, "trends@test.com", "trendsuser").await;
@@ -144,7 +144,7 @@ async fn returns_ratings_view() {
let deps = GetProfileDeps { let deps = GetProfileDeps {
stats: b.stats_repo.clone(), stats: b.stats_repo.clone(),
diary: b.diary_repo.clone(), diary: b.diary_repo.clone(),
social_query: b.social_query.clone(), social_query: b.social_query_unified.clone(),
}; };
setup_user(&b, "ratings@test.com", "ratingsuser").await; setup_user(&b, "ratings@test.com", "ratingsuser").await;
@@ -179,7 +179,7 @@ async fn returns_recent_with_search() {
let deps = GetProfileDeps { let deps = GetProfileDeps {
stats: b.stats_repo.clone(), stats: b.stats_repo.clone(),
diary: b.diary_repo.clone(), diary: b.diary_repo.clone(),
social_query: b.social_query.clone(), social_query: b.social_query_unified.clone(),
}; };
setup_user(&b, "search@test.com", "searchuser").await; setup_user(&b, "search@test.com", "searchuser").await;
@@ -214,7 +214,7 @@ async fn non_own_profile_skips_pending_followers() {
let deps = GetProfileDeps { let deps = GetProfileDeps {
stats: b.stats_repo.clone(), stats: b.stats_repo.clone(),
diary: b.diary_repo.clone(), diary: b.diary_repo.clone(),
social_query: b.social_query.clone(), social_query: b.social_query_unified.clone(),
}; };
setup_user(&b, "other@test.com", "otheruser").await; setup_user(&b, "other@test.com", "otheruser").await;

View File

@@ -63,10 +63,33 @@ pub enum DomainEvent {
user_id: UserId, user_id: UserId,
movie_id: MovieId, movie_id: MovieId,
}, },
FollowRequested {
follower: UserId,
target: crate::value_objects::FollowTarget,
},
FollowAccepted { FollowAccepted {
local_user_id: UserId, owner: UserId,
remote_actor_url: String, requester: crate::value_objects::SocialIdentity,
outbox_url: String, },
FollowRejected {
owner: UserId,
requester: crate::value_objects::SocialIdentity,
},
Unfollowed {
follower: UserId,
target: crate::value_objects::SocialIdentity,
},
FollowerRemoved {
owner: UserId,
follower: crate::value_objects::SocialIdentity,
},
ActorBlocked {
blocker: UserId,
target: crate::value_objects::SocialIdentity,
},
ActorUnblocked {
blocker: UserId,
target: crate::value_objects::SocialIdentity,
}, },
BackfillFollower { BackfillFollower {
owner_user_id: UserId, owner_user_id: UserId,

View File

@@ -1,6 +1,9 @@
use async_trait::async_trait; use async_trait::async_trait;
use crate::{errors::DomainError, value_objects::UserId}; use crate::{
errors::DomainError,
value_objects::{SocialActor, SocialIdentity, UserId},
};
// ── NoopRemoteWatchlistRepository ───────────────────────────────────────────── // ── NoopRemoteWatchlistRepository ─────────────────────────────────────────────
@@ -32,31 +35,78 @@ impl super::RemoteWatchlistRepository for NoopRemoteWatchlistRepository {
} }
} }
// ── NoopSocialQueryPort ─────────────────────────────────────────────────────── // ── NoopSocialCommand ────────────────────────────────────────────────────────
/// Stub used when federation is disabled — returns empty results. pub struct NoopSocialCommand;
pub struct NoopSocialQueryPort;
#[async_trait] #[async_trait]
impl super::SocialQueryPort for NoopSocialQueryPort { impl super::SocialCommand for NoopSocialCommand {
async fn get_accepted_following_urls(&self, _: &UserId) -> Result<Vec<String>, DomainError> { async fn follow(
&self,
_: &UserId,
_: &crate::value_objects::FollowTarget,
) -> Result<(), DomainError> {
Ok(())
}
async fn unfollow(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> {
Ok(())
}
async fn accept_follow(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> {
Ok(())
}
async fn reject_follow(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> {
Ok(())
}
async fn remove_follower(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> {
Ok(())
}
async fn block(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> {
Ok(())
}
async fn unblock(&self, _: &UserId, _: &SocialIdentity) -> Result<(), DomainError> {
Ok(())
}
}
// ── NoopSocialQuery ─────────────────────────────────────────────────────────
pub struct NoopSocialQuery;
#[async_trait]
impl super::SocialQuery for NoopSocialQuery {
async fn get_following(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
Ok(vec![]) Ok(vec![])
} }
async fn list_all_followed_remote_actors( async fn get_followers(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
&self, Ok(vec![])
) -> Result<Vec<crate::models::RemoteActorInfo>, DomainError> { }
async fn get_pending_followers(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
Ok(vec![]) Ok(vec![])
} }
async fn count_following(&self, _: &UserId) -> Result<usize, DomainError> { async fn count_following(&self, _: &UserId) -> Result<usize, DomainError> {
Ok(0) Ok(0)
} }
async fn count_accepted_followers(&self, _: &UserId) -> Result<usize, DomainError> { async fn count_followers(&self, _: &UserId) -> Result<usize, DomainError> {
Ok(0) Ok(0)
} }
async fn get_pending_followers( async fn get_blocked(&self, _: &UserId) -> Result<Vec<SocialActor>, DomainError> {
Ok(vec![])
}
async fn is_following(&self, _: &UserId, _: &SocialIdentity) -> Result<bool, DomainError> {
Ok(false)
}
}
// ── NoopFederationAdminQuery ─────────────────────────────────────────────────
/// Stub used when federation is disabled — returns empty results.
pub struct NoopFederationAdminQuery;
#[async_trait]
impl super::FederationAdminQuery for NoopFederationAdminQuery {
async fn list_all_followed_remote_actors(
&self, &self,
_: &UserId, ) -> Result<Vec<crate::models::RemoteActorInfo>, DomainError> {
) -> Result<Vec<crate::models::PendingFollowerInfo>, DomainError> {
Ok(vec![]) Ok(vec![])
} }
} }

View File

@@ -4,25 +4,68 @@ use chrono::NaiveDateTime;
use crate::{ use crate::{
errors::DomainError, errors::DomainError,
models::{ models::{
DiaryEntry, FederationFlags, PendingFollowerInfo, RemoteActorInfo, RemoteGoalEntry, DiaryEntry, FederationFlags, RemoteActorInfo, RemoteGoalEntry, RemoteWatchlistEntry,
RemoteWatchlistEntry, WatchlistWithMovie, WatchlistWithMovie,
}, },
value_objects::{MovieId, UserId}, value_objects::{FollowTarget, MovieId, SocialActor, SocialIdentity, UserId},
}; };
// ── Unified social ports (ADR-0002) ─────────────────────────────────────────
#[async_trait] #[async_trait]
pub trait SocialQueryPort: Send + Sync { pub trait SocialCommand: Send + Sync {
async fn get_accepted_following_urls( async fn follow(&self, follower: &UserId, target: &FollowTarget) -> Result<(), DomainError>;
async fn unfollow(&self, follower: &UserId, target: &SocialIdentity)
-> Result<(), DomainError>;
async fn accept_follow(
&self, &self,
user_id: &UserId, owner: &UserId,
) -> Result<Vec<String>, DomainError>; requester: &SocialIdentity,
) -> Result<(), DomainError>;
async fn reject_follow(
&self,
owner: &UserId,
requester: &SocialIdentity,
) -> Result<(), DomainError>;
async fn remove_follower(
&self,
owner: &UserId,
follower: &SocialIdentity,
) -> Result<(), DomainError>;
async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError>;
async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError>;
}
#[async_trait]
pub trait SocialQuery: Send + Sync {
async fn get_following(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError>;
async fn get_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError>;
async fn get_pending_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError>;
async fn count_following(&self, user: &UserId) -> Result<usize, DomainError>;
async fn count_followers(&self, user: &UserId) -> Result<usize, DomainError>;
async fn get_blocked(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError>;
async fn is_following(
&self,
follower: &UserId,
target: &SocialIdentity,
) -> Result<bool, DomainError>;
}
#[async_trait]
pub trait FederationAdminQuery: Send + Sync {
async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError>; async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError>;
async fn count_following(&self, user_id: &UserId) -> Result<usize, DomainError>;
async fn count_accepted_followers(&self, user_id: &UserId) -> Result<usize, DomainError>;
async fn get_pending_followers(
&self,
user_id: &UserId,
) -> Result<Vec<PendingFollowerInfo>, DomainError>;
} }
#[async_trait] #[async_trait]
@@ -39,7 +82,6 @@ pub trait RemoteWatchlistRepository: Send + Sync {
actor_url: &str, actor_url: &str,
) -> Result<Vec<RemoteWatchlistEntry>, DomainError>; ) -> Result<Vec<RemoteWatchlistEntry>, DomainError>;
async fn remove_all_by_actor(&self, actor_url: &str) -> Result<(), DomainError>; async fn remove_all_by_actor(&self, actor_url: &str) -> Result<(), DomainError>;
/// Find entries for a remote actor whose URL hashes (v5 UUID) to the given UUID.
async fn get_by_derived_uuid( async fn get_by_derived_uuid(
&self, &self,
uuid: uuid::Uuid, uuid: uuid::Uuid,
@@ -60,10 +102,6 @@ pub trait RemoteGoalRepository: Send + Sync {
async fn get_by_actor_url(&self, actor_url: &str) -> Result<Vec<RemoteGoalEntry>, DomainError>; async fn get_by_actor_url(&self, actor_url: &str) -> Result<Vec<RemoteGoalEntry>, DomainError>;
} }
/// Federation-specific read-only queries that have no equivalent on the
/// standard domain ports (e.g. unpaginated watchlist, local-only review
/// listings). Generic lookups (get_movie_by_id, get_review_by_id, etc.)
/// live on MovieRepository, ReviewRepository, and the other domain ports.
#[async_trait] #[async_trait]
pub trait LocalApContentQuery: Send + Sync { pub trait LocalApContentQuery: Send + Sync {
async fn get_local_watchlist_for_user( async fn get_local_watchlist_for_user(

View File

@@ -19,13 +19,14 @@ use crate::{
ports::{ ports::{
GoalCommand, GoalQuery, ImportProfileRepository, ImportSessionRepository, MovieCommand, GoalCommand, GoalQuery, ImportProfileRepository, ImportSessionRepository, MovieCommand,
MovieProfileRepository, MovieQuery, RefreshSessionRepository, ReviewRepository, MovieProfileRepository, MovieQuery, RefreshSessionRepository, ReviewRepository,
UserFederationSettingsQuery, UserProfileFieldsRepository, UserRepository, SocialCommand, SocialQuery, UserFederationSettingsQuery, UserProfileFieldsRepository,
UserSettingsRepository, WatchEventCommand, WatchEventQuery, WatchlistRepository, UserRepository, UserSettingsRepository, WatchEventCommand, WatchEventQuery,
WebhookTokenRepository, WatchlistRepository, WebhookTokenRepository,
}, },
value_objects::{ value_objects::{
Email, ExternalMetadataId, GoalId, ImportProfileId, ImportSessionId, MovieId, MovieTitle, Email, ExternalMetadataId, GoalId, ImportProfileId, ImportSessionId, MovieId, MovieTitle,
ReleaseYear, ReviewId, UserId, Username, WatchEventId, WebhookTokenId, ReleaseYear, ReviewId, SocialActor, SocialIdentity, UserId, Username, WatchEventId,
WebhookTokenId,
}, },
}; };
@@ -854,3 +855,254 @@ impl RefreshSessionRepository for InMemoryRefreshSessionRepository {
Ok((before - store.len()) as u64) Ok((before - store.len()) as u64)
} }
} }
// ── InMemorySocialRepository ────────────────────────────────────────────────
#[derive(Debug, Clone, PartialEq, Eq)]
enum FollowState {
Pending,
Accepted,
}
pub struct InMemorySocialRepository {
follows: Mutex<Vec<(Uuid, SocialIdentity, FollowState)>>,
blocked: Mutex<Vec<(Uuid, SocialIdentity)>>,
}
impl InMemorySocialRepository {
pub fn new() -> Arc<Self> {
Arc::new(Self {
follows: Mutex::new(Vec::new()),
blocked: Mutex::new(Vec::new()),
})
}
fn identity_to_actor(identity: &SocialIdentity) -> SocialActor {
let handle = match identity {
SocialIdentity::Local(uid) => format!("user-{}", uid.value()),
SocialIdentity::Remote { actor_url } => actor_url.clone(),
};
SocialActor {
identity: identity.clone(),
handle,
display_name: None,
avatar_url: None,
}
}
}
#[async_trait]
impl SocialCommand for InMemorySocialRepository {
async fn follow(
&self,
follower: &UserId,
target: &crate::value_objects::FollowTarget,
) -> Result<(), DomainError> {
let identity = match target {
crate::value_objects::FollowTarget::Identity(id) => id.clone(),
crate::value_objects::FollowTarget::Handle(h) => SocialIdentity::Remote {
actor_url: h.clone(),
},
};
if let SocialIdentity::Local(target_id) = &identity {
if follower == target_id {
return Err(DomainError::ValidationError(
"Cannot follow yourself".into(),
));
}
}
let mut store = self.follows.lock().unwrap();
let already = store
.iter()
.any(|(f, t, _)| *f == follower.value() && *t == identity);
if already {
return Err(DomainError::ValidationError("Already following".into()));
}
store.push((follower.value(), identity, FollowState::Pending));
Ok(())
}
async fn unfollow(
&self,
follower: &UserId,
target: &SocialIdentity,
) -> Result<(), DomainError> {
let mut store = self.follows.lock().unwrap();
let before = store.len();
store.retain(|(f, t, _)| !(*f == follower.value() && t == target));
if store.len() == before {
return Err(DomainError::NotFound(
"Follow relationship not found".into(),
));
}
Ok(())
}
async fn accept_follow(
&self,
owner: &UserId,
requester: &SocialIdentity,
) -> Result<(), DomainError> {
let mut store = self.follows.lock().unwrap();
let target_identity = SocialIdentity::Local(owner.clone());
for (f, t, state) in store.iter_mut() {
let requester_matches = match requester {
SocialIdentity::Local(uid) => *f == uid.value(),
SocialIdentity::Remote { actor_url } => {
if let SocialIdentity::Remote {
actor_url: stored_url,
} = requester
{
stored_url == actor_url
} else {
false
}
}
};
if requester_matches && *t == target_identity && *state == FollowState::Pending {
*state = FollowState::Accepted;
return Ok(());
}
}
Err(DomainError::NotFound(
"Pending follow request not found".into(),
))
}
async fn reject_follow(
&self,
owner: &UserId,
requester: &SocialIdentity,
) -> Result<(), DomainError> {
let mut store = self.follows.lock().unwrap();
let target_identity = SocialIdentity::Local(owner.clone());
let before = store.len();
store.retain(|(f, t, state)| {
let requester_matches = match requester {
SocialIdentity::Local(uid) => *f == uid.value(),
SocialIdentity::Remote { .. } => {
// For remote, match by checking the stored requester identity
false // simplified: reject removes by follower uuid match
}
};
!(requester_matches && *t == target_identity && *state == FollowState::Pending)
});
if store.len() == before {
return Err(DomainError::NotFound(
"Pending follow request not found".into(),
));
}
Ok(())
}
async fn remove_follower(
&self,
owner: &UserId,
follower: &SocialIdentity,
) -> Result<(), DomainError> {
let mut store = self.follows.lock().unwrap();
let target_identity = SocialIdentity::Local(owner.clone());
let before = store.len();
store.retain(|(f, t, _)| {
let follower_matches = match follower {
SocialIdentity::Local(uid) => *f == uid.value(),
SocialIdentity::Remote { .. } => false,
};
!(follower_matches && *t == target_identity)
});
if store.len() == before {
return Err(DomainError::NotFound("Follower not found".into()));
}
Ok(())
}
async fn block(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> {
let mut store = self.blocked.lock().unwrap();
store.push((blocker.value(), target.clone()));
// Also remove any existing follow relationships
let mut follows = self.follows.lock().unwrap();
follows.retain(|(f, t, _)| !(*f == blocker.value() && t == target));
Ok(())
}
async fn unblock(&self, blocker: &UserId, target: &SocialIdentity) -> Result<(), DomainError> {
let mut store = self.blocked.lock().unwrap();
store.retain(|(b, t)| !(*b == blocker.value() && t == target));
Ok(())
}
}
#[async_trait]
impl SocialQuery for InMemorySocialRepository {
async fn get_following(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let store = self.follows.lock().unwrap();
Ok(store
.iter()
.filter(|(f, _, state)| *f == user.value() && *state == FollowState::Accepted)
.map(|(_, t, _)| Self::identity_to_actor(t))
.collect())
}
async fn get_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let store = self.follows.lock().unwrap();
let target = SocialIdentity::Local(user.clone());
Ok(store
.iter()
.filter(|(_, t, state)| *t == target && *state == FollowState::Accepted)
.map(|(f, _, _)| {
let id = SocialIdentity::Local(UserId::from_uuid(*f));
Self::identity_to_actor(&id)
})
.collect())
}
async fn get_pending_followers(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let store = self.follows.lock().unwrap();
let target = SocialIdentity::Local(user.clone());
Ok(store
.iter()
.filter(|(_, t, state)| *t == target && *state == FollowState::Pending)
.map(|(f, _, _)| {
let id = SocialIdentity::Local(UserId::from_uuid(*f));
Self::identity_to_actor(&id)
})
.collect())
}
async fn count_following(&self, user: &UserId) -> Result<usize, DomainError> {
let store = self.follows.lock().unwrap();
Ok(store
.iter()
.filter(|(f, _, state)| *f == user.value() && *state == FollowState::Accepted)
.count())
}
async fn count_followers(&self, user: &UserId) -> Result<usize, DomainError> {
let store = self.follows.lock().unwrap();
let target = SocialIdentity::Local(user.clone());
Ok(store
.iter()
.filter(|(_, t, state)| *t == target && *state == FollowState::Accepted)
.count())
}
async fn get_blocked(&self, user: &UserId) -> Result<Vec<SocialActor>, DomainError> {
let store = self.blocked.lock().unwrap();
Ok(store
.iter()
.filter(|(b, _)| *b == user.value())
.map(|(_, t)| Self::identity_to_actor(t))
.collect())
}
async fn is_following(
&self,
follower: &UserId,
target: &SocialIdentity,
) -> Result<bool, DomainError> {
let store = self.follows.lock().unwrap();
Ok(store.iter().any(|(f, t, state)| {
*f == follower.value() && t == target && *state == FollowState::Accepted
}))
}
}

View File

@@ -64,8 +64,8 @@ impl ObjectStorage for NoopObjectStorage {
// Re-export production noop types so test code that imports from // Re-export production noop types so test code that imports from
// `domain::testing` keeps compiling without changes. // `domain::testing` keeps compiling without changes.
pub use crate::ports::noop::NoopFederationAdminQuery;
pub use crate::ports::noop::NoopRemoteWatchlistRepository; pub use crate::ports::noop::NoopRemoteWatchlistRepository;
pub use crate::ports::noop::NoopSocialQueryPort;
// ── NoopGoalCommand ─────────────────────────────────────────────────────────── // ── NoopGoalCommand ───────────────────────────────────────────────────────────

View File

@@ -6,8 +6,8 @@ use crate::{
AnnotatedRow, DiaryEntry, DiaryFilter, EntityType, ExportFormat, ExternalPersonId, AnnotatedRow, DiaryEntry, DiaryFilter, EntityType, ExportFormat, ExternalPersonId,
FeedEntry, FeedSortBy, FieldMapping, FileFormat, FollowingFilter, ImportError, FeedEntry, FeedSortBy, FieldMapping, FileFormat, FollowingFilter, ImportError,
ImportProfile, ImportSession, IndexableDocument, MovieProfile, MovieStats, ParsedFile, ImportProfile, ImportSession, IndexableDocument, MovieProfile, MovieStats, ParsedFile,
PendingFollowerInfo, Person, PersonCredits, PersonEnrichmentData, PersonId, RefreshSession, Person, PersonCredits, PersonEnrichmentData, PersonId, RefreshSession, RemoteActorInfo,
RemoteActorInfo, ReviewHistory, SearchQuery, SearchResults, UserStats, UserTrends, ReviewHistory, SearchQuery, SearchResults, UserStats, UserTrends,
collections::{PageParams, Paginated}, collections::{PageParams, Paginated},
}, },
ports::{ ports::{
@@ -326,36 +326,12 @@ impl UserProfileFieldsRepository for PanicProfileFieldsRepo {
} }
} }
pub struct PanicSocialQueryPort; pub struct PanicFederationAdminQuery;
#[async_trait] #[async_trait]
impl crate::ports::SocialQueryPort for PanicSocialQueryPort { impl crate::ports::FederationAdminQuery for PanicFederationAdminQuery {
async fn get_accepted_following_urls(
&self,
_: &crate::value_objects::UserId,
) -> Result<Vec<String>, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError> { async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError> {
panic!("PanicSocialQueryPort called") panic!("PanicFederationAdminQuery called")
}
async fn count_following(
&self,
_: &crate::value_objects::UserId,
) -> Result<usize, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn count_accepted_followers(
&self,
_: &crate::value_objects::UserId,
) -> Result<usize, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn get_pending_followers(
&self,
_: &crate::value_objects::UserId,
) -> Result<Vec<PendingFollowerInfo>, DomainError> {
panic!("PanicSocialQueryPort called")
} }
} }

View File

@@ -1,16 +1,55 @@
use super::*; use super::*;
use crate::value_objects::UserId; use crate::value_objects::{FollowTarget, SocialIdentity, UserId};
#[test] #[test]
fn follow_accepted_matches() { fn follow_accepted_matches() {
let uid = UserId::from_uuid(uuid::Uuid::new_v4()); let uid = UserId::from_uuid(uuid::Uuid::new_v4());
let event = DomainEvent::FollowAccepted { let event = DomainEvent::FollowAccepted {
local_user_id: uid.clone(), owner: uid.clone(),
remote_actor_url: "https://remote.example/users/alice".to_string(), requester: SocialIdentity::Remote {
outbox_url: "https://remote.example/users/alice/outbox".to_string(), actor_url: "https://remote.example/users/alice".to_string(),
},
}; };
let DomainEvent::FollowAccepted { outbox_url, .. } = event else { let DomainEvent::FollowAccepted { requester, .. } = event else {
panic!("wrong variant"); panic!("wrong variant");
}; };
assert_eq!(outbox_url, "https://remote.example/users/alice/outbox"); assert_eq!(
requester,
SocialIdentity::Remote {
actor_url: "https://remote.example/users/alice".to_string()
}
);
}
#[test]
fn follow_requested_with_identity() {
let follower = UserId::from_uuid(uuid::Uuid::new_v4());
let target = UserId::from_uuid(uuid::Uuid::new_v4());
let event = DomainEvent::FollowRequested {
follower: follower.clone(),
target: FollowTarget::Identity(SocialIdentity::Local(target.clone())),
};
assert!(matches!(
event,
DomainEvent::FollowRequested {
target: FollowTarget::Identity(SocialIdentity::Local(_)),
..
}
));
}
#[test]
fn follow_requested_with_handle() {
let follower = UserId::from_uuid(uuid::Uuid::new_v4());
let event = DomainEvent::FollowRequested {
follower: follower.clone(),
target: FollowTarget::Handle("@alice@remote.example".into()),
};
assert!(matches!(
event,
DomainEvent::FollowRequested {
target: FollowTarget::Handle(_),
..
}
));
} }

View File

@@ -1,11 +1,13 @@
mod ids; mod ids;
mod movie; mod movie;
mod review; mod review;
mod social;
mod user; mod user;
pub use ids::*; pub use ids::*;
pub use movie::*; pub use movie::*;
pub use review::*; pub use review::*;
pub use social::*;
pub use user::*; pub use user::*;
#[cfg(test)] #[cfg(test)]

View File

@@ -0,0 +1,31 @@
use super::UserId;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SocialIdentity {
Local(UserId),
Remote { actor_url: String },
}
impl SocialIdentity {
pub fn is_local(&self) -> bool {
matches!(self, Self::Local(_))
}
pub fn is_remote(&self) -> bool {
matches!(self, Self::Remote { .. })
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FollowTarget {
Identity(SocialIdentity),
Handle(String),
}
#[derive(Clone, Debug)]
pub struct SocialActor {
pub identity: SocialIdentity,
pub handle: String,
pub display_name: Option<String>,
pub avatar_url: Option<String>,
}

View File

@@ -2,13 +2,14 @@ use std::sync::Arc;
use domain::ports::{ use domain::ports::{
AuthService, DiaryExporter, DiaryQuery, DocumentParser, EventPublisher, FederatedProfileQuery, AuthService, DiaryExporter, DiaryQuery, DocumentParser, EventPublisher, FederatedProfileQuery,
GoalCommand, GoalQuery, ImportProfileRepository, ImportSessionRepository, MetadataClient, FederationAdminQuery, GoalCommand, GoalQuery, ImportProfileRepository, ImportSessionRepository,
MovieCommand, MovieProfileRepository, MovieQuery, ObjectStorage, PasswordHasher, PersonCommand, MetadataClient, MovieCommand, MovieProfileRepository, MovieQuery, ObjectStorage,
PersonEnrichmentClient, PersonQuery, PosterFetcherClient, RefreshSessionRepository, PasswordHasher, PersonCommand, PersonEnrichmentClient, PersonQuery, PosterFetcherClient,
RemoteGoalRepository, RemoteWatchlistRepository, ReviewRepository, SearchCommand, SearchPort, RefreshSessionRepository, RemoteGoalRepository, RemoteWatchlistRepository, ReviewRepository,
SocialQueryPort, StatsRepository, UserProfileFieldsRepository, UserRepository, SearchCommand, SearchPort, SocialCommand, SocialQuery, StatsRepository,
UserSettingsRepository, WatchEventCommand, WatchEventQuery, WatchlistRepository, UserProfileFieldsRepository, UserRepository, UserSettingsRepository, WatchEventCommand,
WebhookTokenRepository, WrapUpRepository, WrapUpStatsQuery, WatchEventQuery, WatchlistRepository, WebhookTokenRepository, WrapUpRepository,
WrapUpStatsQuery,
}; };
use application::config::AppConfig; use application::config::AppConfig;
@@ -35,7 +36,9 @@ pub struct Repositories {
pub search_command: Arc<dyn SearchCommand>, pub search_command: Arc<dyn SearchCommand>,
pub profile_fields: Arc<dyn UserProfileFieldsRepository>, pub profile_fields: Arc<dyn UserProfileFieldsRepository>,
pub remote_watchlist: Arc<dyn RemoteWatchlistRepository>, pub remote_watchlist: Arc<dyn RemoteWatchlistRepository>,
pub social_query: Arc<dyn SocialQueryPort>, pub social_command: Arc<dyn SocialCommand>,
pub social_query_unified: Arc<dyn SocialQuery>,
pub federation_admin: Arc<dyn FederationAdminQuery>,
pub wrapup_stats: Arc<dyn WrapUpStatsQuery>, pub wrapup_stats: Arc<dyn WrapUpStatsQuery>,
pub wrapup_repo: Arc<dyn WrapUpRepository>, pub wrapup_repo: Arc<dyn WrapUpRepository>,
pub goal_command: Arc<dyn GoalCommand>, pub goal_command: Arc<dyn GoalCommand>,
@@ -58,6 +61,8 @@ pub struct Services {
pub document_parser: Arc<dyn DocumentParser>, pub document_parser: Arc<dyn DocumentParser>,
pub review_logger: Arc<dyn ReviewLogger>, pub review_logger: Arc<dyn ReviewLogger>,
pub person_enrichment: Option<Arc<dyn PersonEnrichmentClient>>, pub person_enrichment: Option<Arc<dyn PersonEnrichmentClient>>,
#[cfg(feature = "federation")]
pub ap_service: Arc<dyn activitypub::ActivityPubPort>,
} }
#[derive(Clone)] #[derive(Clone)]

View File

@@ -182,7 +182,7 @@ pub async fn get_activity_feed(
) -> Result<Json<ActivityFeedResponse>, ApiError> { ) -> Result<Json<ActivityFeedResponse>, ApiError> {
let deps = GetActivityFeedDeps { let deps = GetActivityFeedDeps {
diary: state.app_ctx.repos.diary.clone(), diary: state.app_ctx.repos.diary.clone(),
social_query: state.app_ctx.repos.social_query.clone(), social_query: state.app_ctx.repos.social_query_unified.clone(),
config: state.app_ctx.config.clone(), config: state.app_ctx.config.clone(),
}; };
let page = get_feed_uc::execute( let page = get_feed_uc::execute(
@@ -338,7 +338,7 @@ pub async fn get_activity_feed_html(
let deps = GetActivityFeedDeps { let deps = GetActivityFeedDeps {
diary: state.app_ctx.repos.diary.clone(), diary: state.app_ctx.repos.diary.clone(),
social_query: state.app_ctx.repos.social_query.clone(), social_query: state.app_ctx.repos.social_query_unified.clone(),
config: state.app_ctx.config.clone(), config: state.app_ctx.config.clone(),
}; };

View File

@@ -19,8 +19,10 @@ use crate::{
}; };
use api_types::{ use api_types::{
ActorListResponse, ActorUrlRequest, AddBlockedDomainRequest, BlockedActorResponse, ActorListResponse, ActorUrlRequest, AddBlockedDomainRequest, BlockedActorResponse,
BlockedDomainResponse, FollowRequest, BlockedDomainResponse, FollowRequest, RemoteActorDto,
}; };
use application::social::deps::{SocialCommandDeps, SocialQueryDeps};
use domain::value_objects::{FollowTarget, SocialActor, SocialIdentity};
use template_askama::{ use template_askama::{
BlockedActorsTemplate, BlockedDomainsTemplate, FollowersTemplate, FollowingTemplate, BlockedActorsTemplate, BlockedDomainsTemplate, FollowersTemplate, FollowingTemplate,
RemoteActorData, RemoteActorData,
@@ -28,11 +30,62 @@ use template_askama::{
use super::helpers::{build_page_context, encode_error}; use super::helpers::{build_page_context, encode_error};
impl From<&AppState> for SocialCommandDeps {
fn from(state: &AppState) -> Self {
Self {
social_command: state.app_ctx.repos.social_command.clone(),
social_query: state.app_ctx.repos.social_query_unified.clone(),
event_publisher: state.app_ctx.services.event_publisher.clone(),
}
}
}
impl From<&AppState> for SocialQueryDeps {
fn from(state: &AppState) -> Self {
Self {
social_query: state.app_ctx.repos.social_query_unified.clone(),
}
}
}
fn ap_to_domain(e: anyhow::Error) -> domain::errors::DomainError { fn ap_to_domain(e: anyhow::Error) -> domain::errors::DomainError {
tracing::error!("ActivityPub error: {:?}", e); tracing::error!("ActivityPub error: {:?}", e);
domain::errors::DomainError::InfrastructureError(e.to_string()) domain::errors::DomainError::InfrastructureError(e.to_string())
} }
fn actor_url(identity: &SocialIdentity) -> String {
match identity {
SocialIdentity::Remote { actor_url } => actor_url.clone(),
SocialIdentity::Local(uid) => format!("local:{}", uid.value()),
}
}
fn social_actor_to_dto(actor: SocialActor) -> RemoteActorDto {
RemoteActorDto {
url: actor_url(&actor.identity),
handle: actor.handle,
display_name: actor.display_name,
}
}
fn social_actor_to_blocked_dto(actor: SocialActor) -> BlockedActorResponse {
BlockedActorResponse {
url: actor_url(&actor.identity),
handle: actor.handle,
display_name: actor.display_name,
avatar_url: actor.avatar_url,
}
}
fn social_actor_to_template(actor: SocialActor) -> RemoteActorData {
RemoteActorData {
url: actor_url(&actor.identity),
handle: actor.handle,
display_name: actor.display_name,
avatar_url: actor.avatar_url,
}
}
// ── API ────────────────────────────────────────────────────────────────────── // ── API ──────────────────────────────────────────────────────────────────────
#[utoipa::path( #[utoipa::path(
@@ -49,6 +102,8 @@ pub async fn get_blocked_domains_admin(
_admin: AdminApiUser, _admin: AdminApiUser,
) -> Result<Json<Vec<BlockedDomainResponse>>, ApiError> { ) -> Result<Json<Vec<BlockedDomainResponse>>, ApiError> {
let domains = state let domains = state
.app_ctx
.services
.ap_service .ap_service
.get_blocked_domains() .get_blocked_domains()
.await .await
@@ -81,6 +136,8 @@ pub async fn add_blocked_domain_admin(
axum::Json(body): axum::Json<AddBlockedDomainRequest>, axum::Json(body): axum::Json<AddBlockedDomainRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state state
.app_ctx
.services
.ap_service .ap_service
.add_blocked_domain(&body.domain, body.reason.as_deref()) .add_blocked_domain(&body.domain, body.reason.as_deref())
.await .await
@@ -104,6 +161,8 @@ pub async fn remove_blocked_domain_admin(
axum::extract::Path(domain): axum::extract::Path<String>, axum::extract::Path(domain): axum::extract::Path<String>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state state
.app_ctx
.services
.ap_service .ap_service
.remove_blocked_domain(&domain) .remove_blocked_domain(&domain)
.await .await
@@ -125,11 +184,17 @@ pub async fn block_actor_api(
user: AuthenticatedUser, user: AuthenticatedUser,
axum::Json(body): axum::Json<ActorUrlRequest>, axum::Json(body): axum::Json<ActorUrlRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.block_actor(user.0.value(), &body.actor_url) &deps,
.await application::social::commands::SocialCmd::Block {
.map_err(ap_to_domain)?; blocker_id: user.0.value(),
target: SocialIdentity::Remote {
actor_url: body.actor_url,
},
},
)
.await?;
Ok(StatusCode::NO_CONTENT) Ok(StatusCode::NO_CONTENT)
} }
@@ -147,11 +212,17 @@ pub async fn unblock_actor_api(
user: AuthenticatedUser, user: AuthenticatedUser,
axum::Json(body): axum::Json<ActorUrlRequest>, axum::Json(body): axum::Json<ActorUrlRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.unblock_actor(user.0.value(), &body.actor_url) &deps,
.await application::social::commands::SocialCmd::Unblock {
.map_err(ap_to_domain)?; blocker_id: user.0.value(),
target: SocialIdentity::Remote {
actor_url: body.actor_url,
},
},
)
.await?;
Ok(StatusCode::NO_CONTENT) Ok(StatusCode::NO_CONTENT)
} }
@@ -167,20 +238,18 @@ pub async fn get_blocked_actors_api(
State(state): State<AppState>, State(state): State<AppState>,
user: AuthenticatedUser, user: AuthenticatedUser,
) -> Result<Json<Vec<BlockedActorResponse>>, ApiError> { ) -> Result<Json<Vec<BlockedActorResponse>>, ApiError> {
let actors = state let deps = SocialQueryDeps::from(&state);
.ap_service let identities = application::social::execute::execute_query(
.get_blocked_actors(user.0.value()) &deps,
.await application::social::queries::SocialQry::GetBlocked {
.map_err(ap_to_domain)?; user_id: user.0.value(),
},
)
.await?;
Ok(Json( Ok(Json(
actors identities
.into_iter() .into_iter()
.map(|a| BlockedActorResponse { .map(social_actor_to_blocked_dto)
url: a.url,
handle: a.handle,
display_name: a.display_name,
avatar_url: a.avatar_url,
})
.collect(), .collect(),
)) ))
} }
@@ -197,16 +266,16 @@ pub async fn get_following(
State(state): State<AppState>, State(state): State<AppState>,
user: AuthenticatedUser, user: AuthenticatedUser,
) -> Result<Json<ActorListResponse>, ApiError> { ) -> Result<Json<ActorListResponse>, ApiError> {
let actors = state let deps = SocialQueryDeps::from(&state);
.ap_service let identities = application::social::execute::execute_query(
.get_following(user.0.value()) &deps,
.await application::social::queries::SocialQry::GetFollowing {
.map_err(ap_to_domain)?; user_id: user.0.value(),
},
)
.await?;
Ok(Json(ActorListResponse { Ok(Json(ActorListResponse {
actors: actors actors: identities.into_iter().map(social_actor_to_dto).collect(),
.into_iter()
.map(crate::mappers::social::remote_actor_to_dto)
.collect(),
})) }))
} }
@@ -222,16 +291,16 @@ pub async fn get_followers(
State(state): State<AppState>, State(state): State<AppState>,
user: AuthenticatedUser, user: AuthenticatedUser,
) -> Result<Json<ActorListResponse>, ApiError> { ) -> Result<Json<ActorListResponse>, ApiError> {
let actors = state let deps = SocialQueryDeps::from(&state);
.ap_service let identities = application::social::execute::execute_query(
.get_accepted_followers(user.0.value()) &deps,
.await application::social::queries::SocialQry::GetFollowers {
.map_err(ap_to_domain)?; user_id: user.0.value(),
},
)
.await?;
Ok(Json(ActorListResponse { Ok(Json(ActorListResponse {
actors: actors actors: identities.into_iter().map(social_actor_to_dto).collect(),
.into_iter()
.map(crate::mappers::social::remote_actor_to_dto)
.collect(),
})) }))
} }
@@ -240,16 +309,14 @@ pub async fn get_user_following(
_user: AuthenticatedUser, _user: AuthenticatedUser,
Path(user_id): Path<Uuid>, Path(user_id): Path<Uuid>,
) -> Result<Json<ActorListResponse>, ApiError> { ) -> Result<Json<ActorListResponse>, ApiError> {
let actors = state let deps = SocialQueryDeps::from(&state);
.ap_service let identities = application::social::execute::execute_query(
.get_following(user_id) &deps,
.await application::social::queries::SocialQry::GetFollowing { user_id },
.map_err(ap_to_domain)?; )
.await?;
Ok(Json(ActorListResponse { Ok(Json(ActorListResponse {
actors: actors actors: identities.into_iter().map(social_actor_to_dto).collect(),
.into_iter()
.map(crate::mappers::social::remote_actor_to_dto)
.collect(),
})) }))
} }
@@ -258,16 +325,14 @@ pub async fn get_user_followers(
_user: AuthenticatedUser, _user: AuthenticatedUser,
Path(user_id): Path<Uuid>, Path(user_id): Path<Uuid>,
) -> Result<Json<ActorListResponse>, ApiError> { ) -> Result<Json<ActorListResponse>, ApiError> {
let actors = state let deps = SocialQueryDeps::from(&state);
.ap_service let identities = application::social::execute::execute_query(
.get_accepted_followers(user_id) &deps,
.await application::social::queries::SocialQry::GetFollowers { user_id },
.map_err(ap_to_domain)?; )
.await?;
Ok(Json(ActorListResponse { Ok(Json(ActorListResponse {
actors: actors actors: identities.into_iter().map(social_actor_to_dto).collect(),
.into_iter()
.map(crate::mappers::social::remote_actor_to_dto)
.collect(),
})) }))
} }
@@ -285,11 +350,15 @@ pub async fn follow(
user: AuthenticatedUser, user: AuthenticatedUser,
Json(body): Json<FollowRequest>, Json(body): Json<FollowRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.follow(user.0.value(), &body.handle) &deps,
.await application::social::commands::SocialCmd::Follow {
.map_err(ap_to_domain)?; follower_id: user.0.value(),
target: FollowTarget::Handle(body.handle),
},
)
.await?;
Ok(StatusCode::OK) Ok(StatusCode::OK)
} }
@@ -307,11 +376,17 @@ pub async fn unfollow(
user: AuthenticatedUser, user: AuthenticatedUser,
Json(body): Json<ActorUrlRequest>, Json(body): Json<ActorUrlRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.unfollow(user.0.value(), &body.actor_url) &deps,
.await application::social::commands::SocialCmd::Unfollow {
.map_err(ap_to_domain)?; follower_id: user.0.value(),
target: SocialIdentity::Remote {
actor_url: body.actor_url,
},
},
)
.await?;
Ok(StatusCode::OK) Ok(StatusCode::OK)
} }
@@ -329,11 +404,17 @@ pub async fn accept_follower(
user: AuthenticatedUser, user: AuthenticatedUser,
Json(body): Json<ActorUrlRequest>, Json(body): Json<ActorUrlRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.accept_follower(user.0.value(), &body.actor_url) &deps,
.await application::social::commands::SocialCmd::AcceptFollow {
.map_err(ap_to_domain)?; owner_id: user.0.value(),
requester: SocialIdentity::Remote {
actor_url: body.actor_url,
},
},
)
.await?;
Ok(StatusCode::OK) Ok(StatusCode::OK)
} }
@@ -351,11 +432,17 @@ pub async fn reject_follower(
user: AuthenticatedUser, user: AuthenticatedUser,
Json(body): Json<ActorUrlRequest>, Json(body): Json<ActorUrlRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.reject_follower(user.0.value(), &body.actor_url) &deps,
.await application::social::commands::SocialCmd::RejectFollow {
.map_err(ap_to_domain)?; owner_id: user.0.value(),
requester: SocialIdentity::Remote {
actor_url: body.actor_url,
},
},
)
.await?;
Ok(StatusCode::OK) Ok(StatusCode::OK)
} }
@@ -373,11 +460,17 @@ pub async fn remove_follower(
user: AuthenticatedUser, user: AuthenticatedUser,
Json(body): Json<ActorUrlRequest>, Json(body): Json<ActorUrlRequest>,
) -> Result<impl IntoResponse, ApiError> { ) -> Result<impl IntoResponse, ApiError> {
state let deps = SocialCommandDeps::from(&state);
.ap_service application::social::execute::execute_command(
.remove_follower(user.0.value(), &body.actor_url) &deps,
.await application::social::commands::SocialCmd::RemoveFollower {
.map_err(ap_to_domain)?; owner_id: user.0.value(),
follower: SocialIdentity::Remote {
actor_url: body.actor_url,
},
},
)
.await?;
Ok(StatusCode::OK) Ok(StatusCode::OK)
} }
@@ -393,16 +486,16 @@ pub async fn get_pending_followers(
State(state): State<AppState>, State(state): State<AppState>,
user: AuthenticatedUser, user: AuthenticatedUser,
) -> Result<Json<ActorListResponse>, ApiError> { ) -> Result<Json<ActorListResponse>, ApiError> {
let actors = state let deps = SocialQueryDeps::from(&state);
.ap_service let identities = application::social::execute::execute_query(
.get_pending_followers(user.0.value()) &deps,
.await application::social::queries::SocialQry::GetPending {
.map_err(ap_to_domain)?; user_id: user.0.value(),
},
)
.await?;
Ok(Json(ActorListResponse { Ok(Json(ActorListResponse {
actors: actors actors: identities.into_iter().map(social_actor_to_dto).collect(),
.into_iter()
.map(crate::mappers::social::remote_actor_to_dto)
.collect(),
})) }))
} }
@@ -428,7 +521,16 @@ pub async fn follow_remote_user(
.unwrap_or(&format!("/users/{}", profile_user_uuid)) .unwrap_or(&format!("/users/{}", profile_user_uuid))
.to_string(); .to_string();
match state.ap_service.follow(user_id.value(), &form.handle).await { let deps = SocialCommandDeps::from(&state);
match application::social::execute::execute_command(
&deps,
application::social::commands::SocialCmd::Follow {
follower_id: user_id.value(),
target: FollowTarget::Handle(form.handle),
},
)
.await
{
Ok(()) => Redirect::to(&redirect_base).into_response(), Ok(()) => Redirect::to(&redirect_base).into_response(),
Err(e) => { Err(e) => {
tracing::error!("follow error: {:?}", e); tracing::error!("follow error: {:?}", e);
@@ -456,9 +558,16 @@ pub async fn unfollow_remote_user(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state let deps = SocialCommandDeps::from(&state);
.ap_service match application::social::execute::execute_command(
.unfollow(user_id.value(), &form.actor_url) &deps,
application::social::commands::SocialCmd::Unfollow {
follower_id: user_id.value(),
target: SocialIdentity::Remote {
actor_url: form.actor_url,
},
},
)
.await .await
{ {
Ok(()) => { Ok(()) => {
@@ -488,9 +597,16 @@ pub async fn accept_follower_html(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state let deps = SocialCommandDeps::from(&state);
.ap_service match application::social::execute::execute_command(
.accept_follower(user_id.value(), &form.actor_url) &deps,
application::social::commands::SocialCmd::AcceptFollow {
owner_id: user_id.value(),
requester: SocialIdentity::Remote {
actor_url: form.actor_url,
},
},
)
.await .await
{ {
Ok(_) => Redirect::to(&format!("/users/{}", profile_user_uuid)).into_response(), Ok(_) => Redirect::to(&format!("/users/{}", profile_user_uuid)).into_response(),
@@ -514,9 +630,16 @@ pub async fn reject_follower_html(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state let deps = SocialCommandDeps::from(&state);
.ap_service match application::social::execute::execute_command(
.reject_follower(user_id.value(), &form.actor_url) &deps,
application::social::commands::SocialCmd::RejectFollow {
owner_id: user_id.value(),
requester: SocialIdentity::Remote {
actor_url: form.actor_url,
},
},
)
.await .await
{ {
Ok(_) => Redirect::to(&format!("/users/{}", profile_user_uuid)).into_response(), Ok(_) => Redirect::to(&format!("/users/{}", profile_user_uuid)).into_response(),
@@ -540,6 +663,8 @@ pub async fn get_followers_collection(
if accept.contains("application/activity+json") || accept.contains("application/ld+json") { if accept.contains("application/activity+json") || accept.contains("application/ld+json") {
let page = params.get("page").and_then(|p| p.parse::<u32>().ok()); let page = params.get("page").and_then(|p| p.parse::<u32>().ok());
return match state return match state
.app_ctx
.services
.ap_service .ap_service
.followers_collection_json(user_id, page) .followers_collection_json(user_id, page)
.await .await
@@ -571,6 +696,8 @@ pub async fn get_following_collection(
if accept.contains("application/activity+json") || accept.contains("application/ld+json") { if accept.contains("application/activity+json") || accept.contains("application/ld+json") {
let page = params.get("page").and_then(|p| p.parse::<u32>().ok()); let page = params.get("page").and_then(|p| p.parse::<u32>().ok());
return match state return match state
.app_ctx
.services
.ap_service .ap_service
.following_collection_json(user_id, page) .following_collection_json(user_id, page)
.await .await
@@ -605,16 +732,19 @@ pub async fn get_following_page(
"{}/users/{}/following-list", "{}/users/{}/following-list",
state.app_ctx.config.base_url, profile_user_uuid state.app_ctx.config.base_url, profile_user_uuid
); );
match state.ap_service.get_following(user_id.value()).await { let deps = SocialQueryDeps::from(&state);
match application::social::execute::execute_query(
&deps,
application::social::queries::SocialQry::GetFollowing {
user_id: user_id.value(),
},
)
.await
{
Ok(following) => { Ok(following) => {
let actors: Vec<RemoteActorData> = following let actors: Vec<RemoteActorData> = following
.into_iter() .into_iter()
.map(|a| RemoteActorData { .map(social_actor_to_template)
handle: a.handle,
display_name: a.display_name,
url: a.url,
avatar_url: a.avatar_url.clone(),
})
.collect(); .collect();
render_page(FollowingTemplate { render_page(FollowingTemplate {
ctx, ctx,
@@ -651,20 +781,19 @@ pub async fn get_followers_page(
"{}/users/{}/followers-list", "{}/users/{}/followers-list",
state.app_ctx.config.base_url, profile_user_uuid state.app_ctx.config.base_url, profile_user_uuid
); );
match state let deps = SocialQueryDeps::from(&state);
.ap_service match application::social::execute::execute_query(
.get_accepted_followers(user_id.value()) &deps,
application::social::queries::SocialQry::GetFollowers {
user_id: user_id.value(),
},
)
.await .await
{ {
Ok(followers) => { Ok(followers) => {
let actors: Vec<RemoteActorData> = followers let actors: Vec<RemoteActorData> = followers
.into_iter() .into_iter()
.map(|a| RemoteActorData { .map(social_actor_to_template)
handle: a.handle,
display_name: a.display_name,
url: a.url,
avatar_url: a.avatar_url.clone(),
})
.collect(); .collect();
render_page(FollowersTemplate { render_page(FollowersTemplate {
ctx, ctx,
@@ -698,9 +827,16 @@ pub async fn remove_follower_html(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state let deps = SocialCommandDeps::from(&state);
.ap_service match application::social::execute::execute_command(
.remove_follower(user_id.value(), &form.actor_url) &deps,
application::social::commands::SocialCmd::RemoveFollower {
owner_id: user_id.value(),
follower: SocialIdentity::Remote {
actor_url: form.actor_url,
},
},
)
.await .await
{ {
Ok(_) => { Ok(_) => {
@@ -725,7 +861,13 @@ pub async fn get_blocked_domains_page(
let mut ctx = build_page_context(&state, Some(user_id), csrf.0).await; let mut ctx = build_page_context(&state, Some(user_id), csrf.0).await;
ctx.page_title = "Blocked Domains — Movies Diary".to_string(); ctx.page_title = "Blocked Domains — Movies Diary".to_string();
ctx.canonical_url = format!("{}/admin/blocked-domains", state.app_ctx.config.base_url); ctx.canonical_url = format!("{}/admin/blocked-domains", state.app_ctx.config.base_url);
match state.ap_service.get_blocked_domains().await { match state
.app_ctx
.services
.ap_service
.get_blocked_domains()
.await
{
Ok(domains) => { Ok(domains) => {
let entries: Vec<template_askama::BlockedDomainEntry> = domains let entries: Vec<template_askama::BlockedDomainEntry> = domains
.into_iter() .into_iter()
@@ -763,6 +905,8 @@ pub async fn post_blocked_domain(
} }
let reason = form.reason.as_deref().filter(|s| !s.trim().is_empty()); let reason = form.reason.as_deref().filter(|s| !s.trim().is_empty());
match state match state
.app_ctx
.services
.ap_service .ap_service
.add_blocked_domain(&form.domain, reason) .add_blocked_domain(&form.domain, reason)
.await .await
@@ -784,7 +928,13 @@ pub async fn post_remove_blocked_domain(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state.ap_service.remove_blocked_domain(&form.domain).await { match state
.app_ctx
.services
.ap_service
.remove_blocked_domain(&form.domain)
.await
{
Ok(()) => Redirect::to("/admin/blocked-domains").into_response(), Ok(()) => Redirect::to("/admin/blocked-domains").into_response(),
Err(e) => { Err(e) => {
tracing::error!("remove_blocked_domain error: {:?}", e); tracing::error!("remove_blocked_domain error: {:?}", e);
@@ -801,12 +951,20 @@ pub async fn get_blocked_actors_page(
let mut ctx = build_page_context(&state, Some(user_id.clone()), csrf.0).await; let mut ctx = build_page_context(&state, Some(user_id.clone()), csrf.0).await;
ctx.page_title = "Blocked Users — Movies Diary".to_string(); ctx.page_title = "Blocked Users — Movies Diary".to_string();
ctx.canonical_url = format!("{}/social/blocked", state.app_ctx.config.base_url); ctx.canonical_url = format!("{}/social/blocked", state.app_ctx.config.base_url);
match state.ap_service.get_blocked_actors(user_id.value()).await { let deps = SocialQueryDeps::from(&state);
Ok(actors) => { match application::social::execute::execute_query(
let entries: Vec<template_askama::BlockedActorEntry> = actors &deps,
application::social::queries::SocialQry::GetBlocked {
user_id: user_id.value(),
},
)
.await
{
Ok(blocked) => {
let entries: Vec<template_askama::BlockedActorEntry> = blocked
.into_iter() .into_iter()
.map(|a| template_askama::BlockedActorEntry { .map(|a| template_askama::BlockedActorEntry {
url: a.url, url: actor_url(&a.identity),
handle: a.handle, handle: a.handle,
display_name: a.display_name, display_name: a.display_name,
avatar_url: a.avatar_url, avatar_url: a.avatar_url,
@@ -838,9 +996,16 @@ pub async fn post_block_actor_html(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state let deps = SocialCommandDeps::from(&state);
.ap_service match application::social::execute::execute_command(
.block_actor(user_id.value(), &form.actor_url) &deps,
application::social::commands::SocialCmd::Block {
blocker_id: user_id.value(),
target: SocialIdentity::Remote {
actor_url: form.actor_url,
},
},
)
.await .await
{ {
Ok(()) => Redirect::to("/social/blocked").into_response(), Ok(()) => Redirect::to("/social/blocked").into_response(),
@@ -860,9 +1025,16 @@ pub async fn post_unblock_actor(
if crate::csrf::mismatch(&csrf, &form.csrf_token) { if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response(); return StatusCode::FORBIDDEN.into_response();
} }
match state let deps = SocialCommandDeps::from(&state);
.ap_service match application::social::execute::execute_command(
.unblock_actor(user_id.value(), &form.actor_url) &deps,
application::social::commands::SocialCmd::Unblock {
blocker_id: user_id.value(),
target: SocialIdentity::Remote {
actor_url: form.actor_url,
},
},
)
.await .await
{ {
Ok(()) => Redirect::to("/social/blocked").into_response(), Ok(()) => Redirect::to("/social/blocked").into_response(),

View File

@@ -176,12 +176,11 @@ pub async fn update_profile_fields_handler(
responses((status = 200, body = UsersResponse)), responses((status = 200, body = UsersResponse)),
)] )]
pub async fn list_users(State(state): State<AppState>) -> Result<Json<UsersResponse>, ApiError> { pub async fn list_users(State(state): State<AppState>) -> Result<Json<UsersResponse>, ApiError> {
let result = get_users::execute( let deps = application::users::deps::GetUsersListDeps {
state.app_ctx.repos.user.clone(), user: state.app_ctx.repos.user.clone(),
state.app_ctx.repos.social_query.clone(), federation_admin: state.app_ctx.repos.federation_admin.clone(),
GetUsersQuery, };
) let result = get_users::execute(&deps, GetUsersQuery).await?;
.await?;
Ok(Json(UsersResponse { Ok(Json(UsersResponse {
users: result users: result
.users .users
@@ -248,7 +247,7 @@ pub async fn get_user_profile(
let get_profile_deps = GetProfileDeps { let get_profile_deps = GetProfileDeps {
stats: state.app_ctx.repos.stats.clone(), stats: state.app_ctx.repos.stats.clone(),
diary: state.app_ctx.repos.diary.clone(), diary: state.app_ctx.repos.diary.clone(),
social_query: state.app_ctx.repos.social_query.clone(), social_query: state.app_ctx.repos.social_query_unified.clone(),
}; };
let profile = match get_user_profile_uc::execute( let profile = match get_user_profile_uc::execute(
&get_profile_deps, &get_profile_deps,
@@ -381,7 +380,7 @@ async fn build_federated_profile_response(
let get_profile_deps = GetProfileDeps { let get_profile_deps = GetProfileDeps {
stats: state.app_ctx.repos.stats.clone(), stats: state.app_ctx.repos.stats.clone(),
diary: state.app_ctx.repos.diary.clone(), diary: state.app_ctx.repos.diary.clone(),
social_query: state.app_ctx.repos.social_query.clone(), social_query: state.app_ctx.repos.social_query_unified.clone(),
}; };
let profile = match get_user_profile_uc::execute( let profile = match get_user_profile_uc::execute(
&get_profile_deps, &get_profile_deps,
@@ -485,9 +484,12 @@ pub async fn get_users_list(
ctx.page_title = "Members — Movies Diary".to_string(); ctx.page_title = "Members — Movies Diary".to_string();
ctx.canonical_url = format!("{}/users", state.app_ctx.config.base_url); ctx.canonical_url = format!("{}/users", state.app_ctx.config.base_url);
let users_deps = application::users::deps::GetUsersListDeps {
user: state.app_ctx.repos.user.clone(),
federation_admin: state.app_ctx.repos.federation_admin.clone(),
};
match application::users::get_users::execute( match application::users::get_users::execute(
state.app_ctx.repos.user.clone(), &users_deps,
state.app_ctx.repos.social_query.clone(),
application::users::queries::GetUsersQuery, application::users::queries::GetUsersQuery,
) )
.await .await
@@ -647,6 +649,8 @@ pub async fn get_user_profile_html(
.unwrap_or(""); .unwrap_or("");
if accept.contains("application/activity+json") || accept.contains("application/ld+json") { if accept.contains("application/activity+json") || accept.contains("application/ld+json") {
return match state return match state
.app_ctx
.services
.ap_service .ap_service
.actor_json(&profile_user_uuid.to_string()) .actor_json(&profile_user_uuid.to_string())
.await .await
@@ -729,7 +733,7 @@ pub async fn get_user_profile_html(
let html_profile_deps = GetProfileDeps { let html_profile_deps = GetProfileDeps {
stats: state.app_ctx.repos.stats.clone(), stats: state.app_ctx.repos.stats.clone(),
diary: state.app_ctx.repos.diary.clone(), diary: state.app_ctx.repos.diary.clone(),
social_query: state.app_ctx.repos.social_query.clone(), social_query: state.app_ctx.repos.social_query_unified.clone(),
}; };
match application::users::get_profile::execute(&html_profile_deps, query).await { match application::users::get_profile::execute(&html_profile_deps, query).await {
Ok(profile) => { Ok(profile) => {

View File

@@ -66,7 +66,15 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
let event_bus = EventBusBackend::from_env()?; let event_bus = EventBusBackend::from_env()?;
#[cfg(feature = "federation")] #[cfg(feature = "federation")]
let (event_publisher_arc, ap_router, ap_service, social_query, remote_watchlist_repo) = { let (
event_publisher_arc,
ap_router,
ap_service,
social_query,
remote_watchlist_repo,
social_command_arc,
social_query_unified_arc,
) = {
let ( let (
activity_repo, activity_repo,
follow_repo, follow_repo,
@@ -112,12 +120,20 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
let ap_router = ap.router; let ap_router = ap.router;
let ap_service_arc = ap.service; let ap_service_arc = ap.service;
let composite_social = Arc::new(activitypub::CompositeSocialAdapter::new(
Arc::clone(&ap_service_arc),
Arc::clone(&db.user),
app_config.base_url.clone(),
));
( (
ep, ep,
ap_router, ap_router,
ap_service_arc, ap_service_arc,
social_query_arc, social_query_arc,
remote_watchlist_repo, remote_watchlist_repo,
composite_social.clone() as Arc<dyn domain::ports::SocialCommand>,
composite_social as Arc<dyn domain::ports::SocialQuery>,
) )
}; };
@@ -125,6 +141,12 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
let event_publisher_arc = create_event_publisher(event_bus, &db_pool).await?; let event_publisher_arc = create_event_publisher(event_bus, &db_pool).await?;
#[cfg(not(feature = "federation"))] #[cfg(not(feature = "federation"))]
let ap_router = axum::Router::new(); let ap_router = axum::Router::new();
#[cfg(not(feature = "federation"))]
let social_command_arc: Arc<dyn domain::ports::SocialCommand> =
Arc::new(domain::ports::noop::NoopSocialCommand);
#[cfg(not(feature = "federation"))]
let social_query_unified_arc: Arc<dyn domain::ports::SocialQuery> =
Arc::new(domain::ports::noop::NoopSocialQuery);
let review_logger = Arc::new(application::diary::review_logger::DefaultReviewLogger::new( let review_logger = Arc::new(application::diary::review_logger::DefaultReviewLogger::new(
Arc::clone(&db.movie_command), Arc::clone(&db.movie_command),
@@ -159,10 +181,12 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
remote_watchlist: remote_watchlist_repo, remote_watchlist: remote_watchlist_repo,
#[cfg(not(feature = "federation"))] #[cfg(not(feature = "federation"))]
remote_watchlist: Arc::new(domain::ports::noop::NoopRemoteWatchlistRepository), remote_watchlist: Arc::new(domain::ports::noop::NoopRemoteWatchlistRepository),
social_command: social_command_arc,
social_query_unified: social_query_unified_arc,
#[cfg(feature = "federation")] #[cfg(feature = "federation")]
social_query: social_query.clone(), federation_admin: social_query.clone(),
#[cfg(not(feature = "federation"))] #[cfg(not(feature = "federation"))]
social_query: Arc::new(domain::ports::noop::NoopSocialQueryPort), federation_admin: Arc::new(domain::ports::noop::NoopFederationAdminQuery),
wrapup_stats: db.wrapup_stats, wrapup_stats: db.wrapup_stats,
wrapup_repo: db.wrapup_repo, wrapup_repo: db.wrapup_repo,
goal_command: db.goal_command, goal_command: db.goal_command,
@@ -199,6 +223,8 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
document_parser: Arc::new(ImporterDocumentParser) as Arc<dyn DocumentParser>, document_parser: Arc::new(ImporterDocumentParser) as Arc<dyn DocumentParser>,
review_logger, review_logger,
person_enrichment: None, person_enrichment: None,
#[cfg(feature = "federation")]
ap_service,
}, },
config: app_config, config: app_config,
}; };
@@ -208,8 +234,6 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
rss_renderer: Arc::new(RssAdapter::new( rss_renderer: Arc::new(RssAdapter::new(
std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost:3000".into()), std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost:3000".into()),
)), )),
#[cfg(feature = "federation")]
ap_service,
}; };
Ok((state, ap_router)) Ok((state, ap_router))
} }

View File

@@ -8,6 +8,4 @@ use domain::ports::RssFeedRenderer;
pub struct AppState { pub struct AppState {
pub app_ctx: AppContext, pub app_ctx: AppContext,
pub rss_renderer: Arc<dyn RssFeedRenderer>, pub rss_renderer: Arc<dyn RssFeedRenderer>,
#[cfg(feature = "federation")]
pub ap_service: Arc<dyn activitypub::ActivityPubPort>,
} }

View File

@@ -154,30 +154,6 @@ impl DiaryQuery for Panic {
panic!() panic!()
} }
} }
#[cfg(feature = "federation")]
#[async_trait::async_trait]
impl domain::ports::SocialQueryPort for Panic {
async fn get_accepted_following_urls(&self, _: &UserId) -> Result<Vec<String>, DomainError> {
panic!()
}
async fn list_all_followed_remote_actors(
&self,
) -> Result<Vec<domain::models::RemoteActorInfo>, DomainError> {
panic!()
}
async fn count_following(&self, _: &UserId) -> Result<usize, DomainError> {
panic!()
}
async fn count_accepted_followers(&self, _: &UserId) -> Result<usize, DomainError> {
panic!()
}
async fn get_pending_followers(
&self,
_: &UserId,
) -> Result<Vec<domain::models::PendingFollowerInfo>, DomainError> {
panic!()
}
}
#[async_trait::async_trait] #[async_trait::async_trait]
impl StatsRepository for Panic { impl StatsRepository for Panic {
async fn get_user_stats(&self, _: &UserId) -> Result<UserStats, DomainError> { async fn get_user_stats(&self, _: &UserId) -> Result<UserStats, DomainError> {
@@ -811,7 +787,9 @@ pub fn make_test_state(auth_service: Arc<dyn AuthService>) -> crate::state::AppS
search_port: Arc::clone(&repo) as _, search_port: Arc::clone(&repo) as _,
search_command: Arc::clone(&repo) as _, search_command: Arc::clone(&repo) as _,
remote_watchlist: Arc::clone(&repo) as _, remote_watchlist: Arc::clone(&repo) as _,
social_query: Arc::clone(&repo) as _, social_command: Arc::new(domain::ports::noop::NoopSocialCommand) as _,
social_query_unified: Arc::new(domain::ports::noop::NoopSocialQuery) as _,
federation_admin: Arc::new(domain::ports::noop::NoopFederationAdminQuery) as _,
wrapup_stats: Arc::clone(&repo) as _, wrapup_stats: Arc::clone(&repo) as _,
wrapup_repo: Arc::clone(&repo) as _, wrapup_repo: Arc::clone(&repo) as _,
goal_command: Arc::clone(&repo) as _, goal_command: Arc::clone(&repo) as _,
@@ -832,6 +810,8 @@ pub fn make_test_state(auth_service: Arc<dyn AuthService>) -> crate::state::AppS
document_parser: Arc::clone(&repo) as _, document_parser: Arc::clone(&repo) as _,
review_logger: Arc::clone(&repo) as _, review_logger: Arc::clone(&repo) as _,
person_enrichment: None, person_enrichment: None,
#[cfg(feature = "federation")]
ap_service: Arc::new(activitypub::NoopActivityPubService),
}, },
config: AppConfig { config: AppConfig {
allow_registration: false, allow_registration: false,
@@ -846,8 +826,6 @@ pub fn make_test_state(auth_service: Arc<dyn AuthService>) -> crate::state::AppS
}, },
}, },
rss_renderer: Arc::new(Panic), rss_renderer: Arc::new(Panic),
#[cfg(feature = "federation")]
ap_service: Arc::new(activitypub::NoopActivityPubService),
} }
} }

View File

@@ -372,9 +372,6 @@ impl SearchCommand for PanicSearchCommand {
} }
} }
#[cfg(feature = "federation")]
struct PanicSocialQuery;
#[cfg(feature = "federation")] #[cfg(feature = "federation")]
struct PanicRemoteWatchlist; struct PanicRemoteWatchlist;
#[cfg(feature = "federation")] #[cfg(feature = "federation")]
@@ -402,40 +399,6 @@ impl domain::ports::RemoteWatchlistRepository for PanicRemoteWatchlist {
Ok(vec![]) Ok(vec![])
} }
} }
#[cfg(feature = "federation")]
#[async_trait::async_trait]
impl domain::ports::SocialQueryPort for PanicSocialQuery {
async fn get_accepted_following_urls(
&self,
_: &domain::value_objects::UserId,
) -> Result<Vec<String>, DomainError> {
panic!()
}
async fn list_all_followed_remote_actors(
&self,
) -> Result<Vec<domain::models::RemoteActorInfo>, DomainError> {
panic!()
}
async fn count_following(
&self,
_: &domain::value_objects::UserId,
) -> Result<usize, DomainError> {
panic!()
}
async fn count_accepted_followers(
&self,
_: &domain::value_objects::UserId,
) -> Result<usize, DomainError> {
panic!()
}
async fn get_pending_followers(
&self,
_: &domain::value_objects::UserId,
) -> Result<Vec<domain::models::PendingFollowerInfo>, DomainError> {
panic!()
}
}
async fn test_app() -> Router { async fn test_app() -> Router {
let pool = SqlitePool::connect("sqlite::memory:") let pool = SqlitePool::connect("sqlite::memory:")
.await .await
@@ -464,7 +427,9 @@ async fn test_app() -> Router {
search_port: Arc::new(PanicSearchPort), search_port: Arc::new(PanicSearchPort),
search_command: Arc::new(PanicSearchCommand), search_command: Arc::new(PanicSearchCommand),
remote_watchlist: Arc::new(PanicRemoteWatchlist), remote_watchlist: Arc::new(PanicRemoteWatchlist),
social_query: Arc::new(PanicSocialQuery), social_command: Arc::new(domain::ports::noop::NoopSocialCommand),
social_query_unified: Arc::new(domain::ports::noop::NoopSocialQuery),
federation_admin: Arc::new(domain::ports::noop::NoopFederationAdminQuery) as _,
wrapup_stats: Arc::new(domain::testing::PanicWrapUpStatsQuery) as _, wrapup_stats: Arc::new(domain::testing::PanicWrapUpStatsQuery) as _,
wrapup_repo: Arc::new(domain::testing::PanicWrapUpRepository) as _, wrapup_repo: Arc::new(domain::testing::PanicWrapUpRepository) as _,
goal_command: Arc::new(domain::testing::NoopGoalCommand), goal_command: Arc::new(domain::testing::NoopGoalCommand),
@@ -485,6 +450,8 @@ async fn test_app() -> Router {
document_parser: Arc::new(PanicDocumentParser), document_parser: Arc::new(PanicDocumentParser),
review_logger: Arc::new(PanicReviewLogger), review_logger: Arc::new(PanicReviewLogger),
person_enrichment: None, person_enrichment: None,
#[cfg(feature = "federation")]
ap_service: Arc::new(activitypub::NoopActivityPubService),
}, },
config: AppConfig { config: AppConfig {
allow_registration: false, allow_registration: false,
@@ -499,8 +466,6 @@ async fn test_app() -> Router {
}, },
}, },
rss_renderer: Arc::new(RssAdapter::new("http://localhost:3000".into())), rss_renderer: Arc::new(RssAdapter::new("http://localhost:3000".into())),
#[cfg(feature = "federation")]
ap_service: Arc::new(activitypub::NoopActivityPubService),
}; };
routes::build_router(state, axum::Router::new()) routes::build_router(state, axum::Router::new())

View File

@@ -1,7 +1,9 @@
use std::sync::Arc; use std::sync::Arc;
use async_trait::async_trait; use async_trait::async_trait;
use domain::{errors::DomainError, events::DomainEvent, ports::EventHandler}; use domain::{
errors::DomainError, events::DomainEvent, ports::EventHandler, value_objects::SocialIdentity,
};
pub struct FollowBackfillHandler { pub struct FollowBackfillHandler {
pub ap_service: Arc<dyn activitypub::ActivityPubPort>, pub ap_service: Arc<dyn activitypub::ActivityPubPort>,
@@ -12,15 +14,27 @@ impl EventHandler for FollowBackfillHandler {
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
match event { match event {
DomainEvent::FollowAccepted { DomainEvent::FollowAccepted {
remote_actor_url, owner,
outbox_url, requester: SocialIdentity::Remote { actor_url },
..
} => { } => {
tracing::info!(actor = %remote_actor_url, outbox = %outbox_url, "importing remote outbox"); tracing::info!(actor = %actor_url, "follow accepted — looking up outbox for import");
self.ap_service let following = self
.import_remote_outbox(outbox_url, remote_actor_url) .ap_service
.get_following(owner.value())
.await .await
.map_err(|e| DomainError::InfrastructureError(e.to_string())) .map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
if let Some(actor) = following.iter().find(|a| a.url == *actor_url) {
if let Some(outbox_url) = &actor.outbox_url {
tracing::info!(outbox = %outbox_url, actor = %actor_url, "importing remote outbox");
self.ap_service
.import_remote_outbox(outbox_url, actor_url)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
} else {
tracing::warn!(actor = %actor_url, "no outbox URL for accepted follow — skipping import");
}
}
Ok(())
} }
DomainEvent::BackfillFollower { DomainEvent::BackfillFollower {
owner_user_id, owner_user_id,

View File

@@ -0,0 +1,11 @@
# Unified social identity layer — wrap k_ap, don't gut it
Social interactions (follow, unfollow, block, etc.) bypassed the application layer entirely — handlers called the ActivityPub adapter (`k_ap`) directly, and there was no concept of a local-only follow. Every social operation was implicitly federated, with no domain-level orchestration, no CQRS split, and no domain events for most actions. This made it impossible to add local social features without duplicating logic, and meant the codebase would drift as federation and local paths diverged.
We introduce a `SocialIdentity` value object (`Local(UserId)` | `Remote { actor_url }`) in the domain layer. Social command and query ports (`SocialCommand` / `SocialQuery`) accept `SocialIdentity` instead of raw UUIDs or actor URLs. Application-layer use cases follow the existing CQRS pattern (command/query structs, separate deps, one file per use case, domain events on mutations). The adapter implementing `SocialCommand` branches on the identity variant: local goes straight to the database, remote delegates to `k_ap`. `k_ap` stays batteries-included and unchanged — this project just wraps it rather than reaching through it.
## Considered Options
- **Gut `k_ap` into a thin transport layer** — rejected because `k_ap` is shared with other projects (`thoughts`) that rely on its batteries-included API. Forcing all consumers to rewrite social orchestration defeats the purpose of the library.
- **Two-tier API in `k_ap`** (high-level + low-level primitives) — rejected because it adds complexity to `k_ap` for one consumer's needs. Wrapping at the adapter boundary in movies-diary is simpler and keeps `k_ap` focused.
- **Keep the status quo, add local branches in handlers** — rejected because it perpetuates the "no application layer for social" problem and guarantees local/remote drift.