GKaszewski/movies-diary
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: per-entity federation privacy toggles for reviews and watchlist

Browse Source 
- add federate_reviews + federate_watchlist to UserSettings (default true)
- new UserFederationSettingsQuery port with FederationFlags struct
- remove get_user_federate_goals from LocalApContentQuery
- gate ReviewLogged, ReviewUpdated, WatchlistEntryAdded, on_poster_synced on flags
- goals gating migrated to UserFederationSettingsQuery
- ReviewDeleted and WatchlistEntryRemoved ungated (tombstones always fire)
- sqlite + postgres migrations and adapter impls
- settings API and SPA toggles
This commit is contained in:
Gabriel Kaszewski
2026-06-12 02:26:01 +02:00
parent 33aa5bdab3
commit ca7ca51949
25 changed files with 372 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
crates/adapters/activitypub/src/event_handler.rs
View File

@@ -4,7 +4,7 @@ use domain::ports::EventHandler;
use domain::{
errors::DomainError,
events::DomainEvent,
ports::LocalApContentQuery,
ports::{LocalApContentQuery, UserFederationSettingsQuery},
value_objects::{MovieId, ReviewId, UserId},
};
use std::sync::Arc;
@@ -17,6 +17,7 @@ use crate::urls::{actor_url, goal_url, review_url};
pub struct ActivityPubEventHandler {
ap_service: Arc<ActivityPubService>,
content_query: Arc<dyn LocalApContentQuery>,
federation_settings: Arc<dyn UserFederationSettingsQuery>,
base_url: String,
}
@@ -24,11 +25,13 @@ impl ActivityPubEventHandler {
pub fn new(
ap_service: Arc<ActivityPubService>,
content_query: Arc<dyn LocalApContentQuery>,
federation_settings: Arc<dyn UserFederationSettingsQuery>,
base_url: String,
) -> Self {
Self {
ap_service,
content_query,
federation_settings,
base_url,
}
}
@@ -131,6 +134,12 @@ impl EventHandler for ActivityPubEventHandler {
impl ActivityPubEventHandler {
async fn on_review_logged(&self, user_id: &UserId, review_id: &ReviewId) -> anyhow::Result<()> {
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.reviews {
return Ok(());
}
let review = match self.content_query.get_review_by_id(review_id).await? {
Some(r) => r,
None => return Ok(()),
@@ -184,6 +193,12 @@ impl ActivityPubEventHandler {
user_id: &UserId,
review_id: &ReviewId,
) -> anyhow::Result<()> {
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.reviews {
return Ok(());
}
let review = match self.content_query.get_review_by_id(review_id).await? {
Some(r) => r,
None => return Ok(()),
@@ -250,6 +265,12 @@ impl ActivityPubEventHandler {
external_metadata_id: &Option<String>,
added_at: &chrono::NaiveDateTime,
) -> anyhow::Result<()> {
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.watchlist {
return Ok(());
}
use crate::urls::watchlist_entry_url;
let ap_id = watchlist_entry_url(&self.base_url, user_id.value(), movie_id.value());
let actor = actor_url(&self.base_url, user_id.value());
@@ -316,6 +337,13 @@ impl ActivityPubEventHandler {
for entry in entries {
let review = entry.review();
let user_id = review.user_id();
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.reviews {
continue;
}
let ap_id = review_url(&self.base_url, review.id());
let actor = actor_url(&self.base_url, user_id.value());
@@ -343,12 +371,9 @@ impl ActivityPubEventHandler {
user_id: &UserId,
year: u16,
) -> anyhow::Result<()> {
if !self
.content_query
.get_user_federate_goals(user_id)
.await
.unwrap_or(false)
{
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.goals {
return Ok(());
}
let Some((goal, current)) = self
@@ -384,12 +409,9 @@ impl ActivityPubEventHandler {
target_count: u32,
is_create: bool,
) -> anyhow::Result<()> {
if !self
.content_query
.get_user_federate_goals(user_id)
.await
.unwrap_or(false)
{
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.goals {
return Ok(());
}
let current = self
@@ -418,12 +440,9 @@ impl ActivityPubEventHandler {
}
async fn on_goal_deleted(&self, user_id: &UserId, year: u16) -> anyhow::Result<()> {
if !self
.content_query
.get_user_federate_goals(user_id)
.await
.unwrap_or(false)
{
let flags = self.federation_settings.get_federation_flags(user_id).await
.unwrap_or(domain::ports::FederationFlags { goals: true, reviews: true, watchlist: true });
if !flags.goals {
return Ok(());
}
let ap_id = goal_url(&self.base_url, user_id.value(), year);

3
crates/adapters/activitypub/src/lib.rs
View File

@@ -52,6 +52,7 @@ pub struct ActivityPubDeps {
pub remote_goal_repo: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
pub local_ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
pub user_repo: std::sync::Arc<dyn domain::ports::UserRepository>,
pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
pub base_url: String,
pub allow_registration: bool,
pub event_publisher: std::sync::Arc<dyn domain::ports::EventPublisher>,
@@ -68,6 +69,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
remote_goal_repo,
local_ap_content,
user_repo,
federation_settings,
base_url,
allow_registration,
event_publisher,
@@ -129,6 +131,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
let event_handler = std::sync::Arc::new(ActivityPubEventHandler::new(
std::sync::Arc::clone(&concrete),
local_ap_content,
federation_settings,
base_url,
)) as std::sync::Arc<dyn domain::ports::EventHandler>;

2
crates/adapters/postgres/migrations/0028_federation_privacy.sql Normal file
View File

@@ -0,0 +1,2 @@
ALTER TABLE user_settings ADD COLUMN federate_reviews BOOLEAN NOT NULL DEFAULT TRUE;
ALTER TABLE user_settings ADD COLUMN federate_watchlist BOOLEAN NOT NULL DEFAULT TRUE;

17
crates/adapters/postgres/src/ap_content.rs
View File

@@ -229,23 +229,6 @@ impl LocalApContentQuery for PostgresApContentQuery {
rows.into_iter().map(DiaryRow::into_domain).collect()
}
async fn get_user_federate_goals(&self, user_id: &UserId) -> Result<bool, DomainError> {
let uid = user_id.value().to_string();
let row = sqlx::query("SELECT federate_goals FROM user_settings WHERE user_id = $1")
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
match row {
Some(r) => {
let val: i64 = r.try_get("federate_goals").unwrap_or(0);
Ok(val != 0)
}
None => Ok(false),
}
}
async fn get_goal_with_progress(
&self,
user_id: &UserId,

8
crates/adapters/postgres/src/lib.rs
View File

@@ -93,6 +93,7 @@ pub struct PostgresWireOutput {
pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
pub goal: std::sync::Arc<dyn domain::ports::GoalRepository>,
pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
pub remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
}
@@ -108,6 +109,8 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
.map_err(|e| anyhow::anyhow!("{e}"))
.context("Database migration failed")?;
let user_settings_repo = std::sync::Arc::new(user_settings::PostgresUserSettingsRepository::new(pool.clone()));
Ok(PostgresWireOutput {
pool: pool.clone(),
movie: std::sync::Arc::new(PostgresMovieRepository::new(pool.clone())) as _,
@@ -125,9 +128,8 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
wrapup_repo: std::sync::Arc::new(PostgresWrapUpRepository::new(pool.clone())) as _,
wrapup_stats: std::sync::Arc::new(PostgresWrapUpStatsQuery::new(pool.clone())) as _,
goal: std::sync::Arc::new(goals::PostgresGoalRepository::new(pool.clone())) as _,
user_settings: std::sync::Arc::new(user_settings::PostgresUserSettingsRepository::new(
pool.clone(),
)) as _,
user_settings: std::sync::Arc::clone(&user_settings_repo) as _,
federation_settings: user_settings_repo as _,
remote_goal: std::sync::Arc::new(remote_goals::PostgresRemoteGoalRepository::new(pool))
as _,
})

70
crates/adapters/postgres/src/user_settings.rs
View File

@@ -1,6 +1,9 @@
use async_trait::async_trait;
use domain::{
errors::DomainError, models::UserSettings, ports::UserSettingsRepository, value_objects::UserId,
errors::DomainError,
models::UserSettings,
ports::{FederationFlags, UserFederationSettingsQuery, UserSettingsRepository},
value_objects::UserId,
};
use sqlx::{PgPool, Row};
@@ -23,20 +26,25 @@ impl PostgresUserSettingsRepository {
impl UserSettingsRepository for PostgresUserSettingsRepository {
async fn get(&self, user_id: &UserId) -> Result<UserSettings, DomainError> {
let uid = user_id.value().to_string();
let row =
sqlx::query("SELECT user_id, federate_goals FROM user_settings WHERE user_id = $1")
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
let row = sqlx::query(
"SELECT federate_goals, federate_reviews, federate_watchlist \
FROM user_settings WHERE user_id = $1",
)
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
match row {
Some(r) => {
let federate: i64 = r.try_get("federate_goals").unwrap_or(0);
let goals: bool = r.try_get("federate_goals").unwrap_or(true);
let reviews: bool = r.try_get("federate_reviews").unwrap_or(true);
let watchlist: bool = r.try_get("federate_watchlist").unwrap_or(true);
Ok(UserSettings::from_persistence(
user_id.clone(),
federate != 0,
goals,
reviews,
watchlist,
))
}
None => Ok(UserSettings::new(user_id.clone())),
@@ -45,18 +53,48 @@ impl UserSettingsRepository for PostgresUserSettingsRepository {
async fn save(&self, settings: &UserSettings) -> Result<(), DomainError> {
let uid = settings.user_id().value().to_string();
let federate = if settings.federate_goals() { 1i64 } else { 0 };
sqlx::query(
"INSERT INTO user_settings (user_id, federate_goals) VALUES ($1, $2) \
ON CONFLICT (user_id) DO UPDATE SET federate_goals = $2",
"INSERT INTO user_settings (user_id, federate_goals, federate_reviews, federate_watchlist) \
VALUES ($1, $2, $3, $4) \
ON CONFLICT (user_id) DO UPDATE \
SET federate_goals = $2, federate_reviews = $3, federate_watchlist = $4",
)
.bind(&uid)
.bind(federate)
.bind(settings.federate_goals())
.bind(settings.federate_reviews())
.bind(settings.federate_watchlist())
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
Ok(())
}
}
#[async_trait]
impl UserFederationSettingsQuery for PostgresUserSettingsRepository {
async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError> {
let uid = user_id.value().to_string();
let row = sqlx::query(
"SELECT federate_goals, federate_reviews, federate_watchlist \
FROM user_settings WHERE user_id = $1",
)
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
match row {
Some(r) => {
let goals: bool = r.try_get("federate_goals").unwrap_or(true);
let reviews: bool = r.try_get("federate_reviews").unwrap_or(true);
let watchlist: bool = r.try_get("federate_watchlist").unwrap_or(true);
Ok(FederationFlags { goals, reviews, watchlist })
}
None => Ok(FederationFlags {
goals: true,
reviews: true,
watchlist: true,
}),
}
}
}

2
crates/adapters/sqlite/migrations/0028_federation_privacy.sql Normal file
View File

@@ -0,0 +1,2 @@
ALTER TABLE user_settings ADD COLUMN federate_reviews BOOLEAN NOT NULL DEFAULT TRUE;
ALTER TABLE user_settings ADD COLUMN federate_watchlist BOOLEAN NOT NULL DEFAULT TRUE;

19
crates/adapters/sqlite/src/ap_content.rs
View File

@@ -5,7 +5,7 @@ use domain::{
ports::LocalApContentQuery,
value_objects::{MovieId, ReviewId, UserId},
};
use sqlx::{Row, SqlitePool};
use sqlx::SqlitePool;
use crate::models::{DiaryRow, MovieRow, ReviewRow, WatchlistRow};
@@ -169,23 +169,6 @@ impl LocalApContentQuery for SqliteApContentQuery {
rows.into_iter().map(DiaryRow::into_domain).collect()
}
async fn get_user_federate_goals(&self, user_id: &UserId) -> Result<bool, DomainError> {
let uid = user_id.value().to_string();
let row = sqlx::query("SELECT federate_goals FROM user_settings WHERE user_id = ?")
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
match row {
Some(r) => {
let val: i64 = r.try_get("federate_goals").unwrap_or(0);
Ok(val != 0)
}
None => Ok(false),
}
}
async fn get_goal_with_progress(
&self,
user_id: &UserId,

8
crates/adapters/sqlite/src/lib.rs
View File

@@ -89,6 +89,7 @@ pub struct SqliteWireOutput {
pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
pub goal: std::sync::Arc<dyn domain::ports::GoalRepository>,
pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
pub remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
}
@@ -113,6 +114,8 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
.map_err(|e| anyhow::anyhow!("{e}"))
.context("Database migration failed")?;
let user_settings_repo = std::sync::Arc::new(user_settings::SqliteUserSettingsRepository::new(pool.clone()));
Ok(SqliteWireOutput {
pool: pool.clone(),
movie: std::sync::Arc::new(SqliteMovieRepository::new(pool.clone())) as _,
@@ -128,9 +131,8 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
wrapup_repo: std::sync::Arc::new(SqliteWrapUpRepository::new(pool.clone())) as _,
wrapup_stats: std::sync::Arc::new(SqliteWrapUpStatsQuery::new(pool.clone())) as _,
goal: std::sync::Arc::new(goals::SqliteGoalRepository::new(pool.clone())) as _,
user_settings: std::sync::Arc::new(user_settings::SqliteUserSettingsRepository::new(
pool.clone(),
)) as _,
user_settings: std::sync::Arc::clone(&user_settings_repo) as _,
federation_settings: user_settings_repo as _,
remote_goal: std::sync::Arc::new(remote_goals::SqliteRemoteGoalRepository::new(pool)) as _,
})
}

82
crates/adapters/sqlite/src/user_settings.rs
View File

@@ -1,6 +1,9 @@
use async_trait::async_trait;
use domain::{
errors::DomainError, models::UserSettings, ports::UserSettingsRepository, value_objects::UserId,
errors::DomainError,
models::UserSettings,
ports::{FederationFlags, UserFederationSettingsQuery, UserSettingsRepository},
value_objects::UserId,
};
use sqlx::{Row, SqlitePool};
@@ -23,20 +26,25 @@ impl SqliteUserSettingsRepository {
impl UserSettingsRepository for SqliteUserSettingsRepository {
async fn get(&self, user_id: &UserId) -> Result<UserSettings, DomainError> {
let uid = user_id.value().to_string();
let row =
sqlx::query("SELECT user_id, federate_goals FROM user_settings WHERE user_id = ?")
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
let row = sqlx::query(
"SELECT federate_goals, federate_reviews, federate_watchlist \
FROM user_settings WHERE user_id = ?",
)
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
match row {
Some(r) => {
let federate: i64 = r.try_get("federate_goals").unwrap_or(0);
let goals: i64 = r.try_get("federate_goals").unwrap_or(1);
let reviews: i64 = r.try_get("federate_reviews").unwrap_or(1);
let watchlist: i64 = r.try_get("federate_watchlist").unwrap_or(1);
Ok(UserSettings::from_persistence(
user_id.clone(),
federate != 0,
goals != 0,
reviews != 0,
watchlist != 0,
))
}
None => Ok(UserSettings::new(user_id.clone())),
@@ -45,15 +53,51 @@ impl UserSettingsRepository for SqliteUserSettingsRepository {
async fn save(&self, settings: &UserSettings) -> Result<(), DomainError> {
let uid = settings.user_id().value().to_string();
let federate = if settings.federate_goals() { 1i64 } else { 0 };
sqlx::query("INSERT OR REPLACE INTO user_settings (user_id, federate_goals) VALUES (?, ?)")
.bind(&uid)
.bind(federate)
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
sqlx::query(
"INSERT OR REPLACE INTO user_settings \
(user_id, federate_goals, federate_reviews, federate_watchlist) \
VALUES (?, ?, ?, ?)",
)
.bind(&uid)
.bind(if settings.federate_goals() { 1i64 } else { 0 })
.bind(if settings.federate_reviews() { 1i64 } else { 0 })
.bind(if settings.federate_watchlist() { 1i64 } else { 0 })
.execute(&self.pool)
.await
.map_err(Self::map_err)?;
Ok(())
}
}
#[async_trait]
impl UserFederationSettingsQuery for SqliteUserSettingsRepository {
async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError> {
let uid = user_id.value().to_string();
let row = sqlx::query(
"SELECT federate_goals, federate_reviews, federate_watchlist \
FROM user_settings WHERE user_id = ?",
)
.bind(&uid)
.fetch_optional(&self.pool)
.await
.map_err(Self::map_err)?;
match row {
Some(r) => {
let goals: i64 = r.try_get("federate_goals").unwrap_or(1);
let reviews: i64 = r.try_get("federate_reviews").unwrap_or(1);
let watchlist: i64 = r.try_get("federate_watchlist").unwrap_or(1);
Ok(FederationFlags {
goals: goals != 0,
reviews: reviews != 0,
watchlist: watchlist != 0,
})
}
None => Ok(FederationFlags {
goals: true,
reviews: true,
watchlist: true,
}),
}
}
}