feat: goals — "watch N movies in YEAR" with progress bar
Domain: Goal entity, UserSettings (federation toggle), RemoteGoalEntry.
Ports: GoalRepository, UserSettingsRepository, RemoteGoalRepository.
Adapters: sqlite + postgres repos, migrations, AP content query extensions.
Application: CRUD use cases (create/update/delete/get/list), settings use cases.
API: 7 endpoints (/goals CRUD, /users/{id}/goals, /settings) with utoipa docs.
Federation: GoalObject (Note + goal discriminator), outbound broadcast with
per-user toggle, inbound GoalObjectHandler in CompositeObjectHandler.
SPA: API client + hooks, GoalCard (shadcn Card+Progress+DropdownMenu),
GoalSheet (Drawer), profile integration (editable own, read-only others),
federation toggle in settings (Switch).
Classic HTML: glassmorphic goal card on profile, Frutiger Aero styling.
Progress computed from existing reviews — backwards compatible.
This commit is contained in:
@@ -5,11 +5,15 @@ use chrono::{DateTime, Utc};
|
||||
use k_ap::{ApContentReader, ApObjectHandler};
|
||||
use url::Url;
|
||||
|
||||
use crate::{review_handler::ReviewObjectHandler, watchlist_handler::WatchlistObjectHandler};
|
||||
use crate::{
|
||||
goal_handler::GoalObjectHandler, review_handler::ReviewObjectHandler,
|
||||
watchlist_handler::WatchlistObjectHandler,
|
||||
};
|
||||
|
||||
pub struct CompositeObjectHandler {
|
||||
pub review: Arc<ReviewObjectHandler>,
|
||||
pub watchlist: Arc<WatchlistObjectHandler>,
|
||||
pub goal: Arc<GoalObjectHandler>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -40,8 +44,11 @@ impl ApObjectHandler for CompositeObjectHandler {
|
||||
) -> anyhow::Result<()> {
|
||||
let is_watchlist = object.get("watchlistEntry").and_then(|v| v.as_bool()) == Some(true)
|
||||
|| (object.get("movieTitle").is_some() && object.get("rating").is_none());
|
||||
let is_goal = object.get("goal").and_then(|v| v.as_bool()) == Some(true);
|
||||
if object.get("rating").is_some() {
|
||||
self.review.on_create(ap_id, actor_url, object).await
|
||||
} else if is_goal {
|
||||
self.goal.on_create(ap_id, actor_url, object).await
|
||||
} else if is_watchlist {
|
||||
self.watchlist.on_create(ap_id, actor_url, object).await
|
||||
} else {
|
||||
@@ -56,8 +63,11 @@ impl ApObjectHandler for CompositeObjectHandler {
|
||||
actor_url: &Url,
|
||||
object: serde_json::Value,
|
||||
) -> anyhow::Result<()> {
|
||||
let is_goal = object.get("goal").and_then(|v| v.as_bool()) == Some(true);
|
||||
if object.get("rating").is_some() {
|
||||
self.review.on_update(ap_id, actor_url, object).await
|
||||
} else if is_goal {
|
||||
self.goal.on_update(ap_id, actor_url, object).await
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
@@ -66,12 +76,14 @@ impl ApObjectHandler for CompositeObjectHandler {
|
||||
async fn on_delete(&self, ap_id: &Url, actor_url: &Url) -> anyhow::Result<()> {
|
||||
self.review.on_delete(ap_id, actor_url).await?;
|
||||
self.watchlist.on_delete(ap_id, actor_url).await?;
|
||||
self.goal.on_delete(ap_id, actor_url).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_actor_removed(&self, actor_url: &Url) -> anyhow::Result<()> {
|
||||
self.review.on_actor_removed(actor_url).await?;
|
||||
self.watchlist.on_actor_removed(actor_url).await?;
|
||||
self.goal.on_actor_removed(actor_url).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -10,8 +10,8 @@ use std::sync::Arc;
|
||||
|
||||
use k_ap::{ActivityPubService, ApVisibility};
|
||||
|
||||
use crate::objects::review_to_ap_object;
|
||||
use crate::urls::{actor_url, review_url};
|
||||
use crate::objects::{goal_to_ap_object, review_to_ap_object};
|
||||
use crate::urls::{actor_url, goal_url, review_url};
|
||||
|
||||
pub struct ActivityPubEventHandler {
|
||||
ap_service: Arc<ActivityPubService>,
|
||||
@@ -101,6 +101,28 @@ impl EventHandler for ActivityPubEventHandler {
|
||||
.on_poster_synced(movie_id)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(e.to_string())),
|
||||
DomainEvent::GoalCreated {
|
||||
user_id,
|
||||
year,
|
||||
target_count,
|
||||
..
|
||||
} => self
|
||||
.broadcast_goal(user_id, *year, *target_count, true)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(e.to_string())),
|
||||
DomainEvent::GoalUpdated {
|
||||
user_id,
|
||||
year,
|
||||
target_count,
|
||||
..
|
||||
} => self
|
||||
.broadcast_goal(user_id, *year, *target_count, false)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(e.to_string())),
|
||||
DomainEvent::GoalDeleted { user_id, year, .. } => self
|
||||
.on_goal_deleted(user_id, *year)
|
||||
.await
|
||||
.map_err(|e| DomainError::InfrastructureError(e.to_string())),
|
||||
_ => Ok(()),
|
||||
}
|
||||
}
|
||||
@@ -311,4 +333,60 @@ impl ActivityPubEventHandler {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn broadcast_goal(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
year: u16,
|
||||
target_count: u32,
|
||||
is_create: bool,
|
||||
) -> anyhow::Result<()> {
|
||||
if !self
|
||||
.content_query
|
||||
.get_user_federate_goals(user_id)
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
let current = self
|
||||
.content_query
|
||||
.get_goal_with_progress(user_id, year)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.map(|(_, c)| c)
|
||||
.unwrap_or(0);
|
||||
|
||||
let ap_id = goal_url(&self.base_url, user_id.value(), year);
|
||||
let actor = actor_url(&self.base_url, user_id.value());
|
||||
let obj = goal_to_ap_object(ap_id, actor, year, target_count, current, &self.base_url);
|
||||
let json = serde_json::to_value(obj)?;
|
||||
if is_create {
|
||||
self.ap_service
|
||||
.broadcast_create_note(user_id.value(), json, ApVisibility::Public, vec![])
|
||||
.await?;
|
||||
} else {
|
||||
self.ap_service
|
||||
.broadcast_update_note(user_id.value(), json, ApVisibility::Public, vec![])
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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)
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
let ap_id = goal_url(&self.base_url, user_id.value(), year);
|
||||
self.ap_service
|
||||
.broadcast_delete_to_followers(user_id.value(), ap_id)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
89
crates/adapters/activitypub/src/goal_handler.rs
Normal file
89
crates/adapters/activitypub/src/goal_handler.rs
Normal file
@@ -0,0 +1,89 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use domain::{models::RemoteGoalEntry, ports::RemoteGoalRepository};
|
||||
use k_ap::ApObjectHandler;
|
||||
use url::Url;
|
||||
|
||||
use crate::objects::GoalObject;
|
||||
|
||||
pub struct GoalObjectHandler {
|
||||
pub remote_goal_repo: Arc<dyn RemoteGoalRepository>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ApObjectHandler for GoalObjectHandler {
|
||||
async fn on_create(
|
||||
&self,
|
||||
ap_id: &Url,
|
||||
actor_url: &Url,
|
||||
object: serde_json::Value,
|
||||
) -> anyhow::Result<()> {
|
||||
let obj: GoalObject = match serde_json::from_value(object) {
|
||||
Ok(o) => o,
|
||||
Err(e) => {
|
||||
tracing::warn!(ap_id = %ap_id, "ignoring malformed goal Create: {}", e);
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
let entry = RemoteGoalEntry {
|
||||
ap_id: ap_id.as_str().to_string(),
|
||||
actor_url: actor_url.as_str().to_string(),
|
||||
year: obj.goal_year,
|
||||
target_count: obj.goal_target,
|
||||
current_count: obj.goal_current,
|
||||
received_at: chrono::Utc::now(),
|
||||
};
|
||||
self.remote_goal_repo.save(entry).await?;
|
||||
tracing::info!(ap_id = %ap_id, year = obj.goal_year, "saved remote goal");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_update(
|
||||
&self,
|
||||
ap_id: &Url,
|
||||
_actor_url: &Url,
|
||||
object: serde_json::Value,
|
||||
) -> anyhow::Result<()> {
|
||||
let obj: GoalObject = match serde_json::from_value(object) {
|
||||
Ok(o) => o,
|
||||
Err(e) => {
|
||||
tracing::warn!(ap_id = %ap_id, "ignoring malformed goal Update: {}", e);
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
self.remote_goal_repo
|
||||
.update_by_ap_id(ap_id.as_str(), obj.goal_target, obj.goal_current)
|
||||
.await?;
|
||||
tracing::info!(ap_id = %ap_id, "updated remote goal progress");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_delete(&self, ap_id: &Url, actor_url: &Url) -> anyhow::Result<()> {
|
||||
self.remote_goal_repo
|
||||
.remove_by_ap_id(ap_id.as_str(), actor_url.as_str())
|
||||
.await?;
|
||||
tracing::info!(ap_id = %ap_id, "removed remote goal");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_actor_removed(&self, _actor_url: &Url) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn on_like(&self, _: &Url, _: &Url) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
async fn on_announce_received(&self, _: &Url, _: &Url) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
async fn on_announce_of_remote(&self, _: &Url, _: &Url) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
async fn on_unlike(&self, _: &Url, _: &Url) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
async fn on_mention(&self, _: &Url, _: uuid::Uuid, _: &Url) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
pub mod composite_handler;
|
||||
pub mod event_handler;
|
||||
pub mod federation_event_bridge;
|
||||
pub mod goal_handler;
|
||||
pub mod objects;
|
||||
pub mod port;
|
||||
pub mod remote_review_repository;
|
||||
@@ -48,6 +49,7 @@ pub struct ActivityPubDeps {
|
||||
pub blocklist_repo: std::sync::Arc<dyn BlocklistRepository>,
|
||||
pub review_store: std::sync::Arc<dyn RemoteReviewRepository>,
|
||||
pub remote_watchlist_repo: std::sync::Arc<dyn domain::ports::RemoteWatchlistRepository>,
|
||||
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 base_url: String,
|
||||
@@ -63,6 +65,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
|
||||
blocklist_repo,
|
||||
review_store,
|
||||
remote_watchlist_repo,
|
||||
remote_goal_repo,
|
||||
local_ap_content,
|
||||
user_repo,
|
||||
base_url,
|
||||
@@ -79,9 +82,11 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
|
||||
content_query: std::sync::Arc::clone(&local_ap_content),
|
||||
base_url: base_url.clone(),
|
||||
});
|
||||
let goal_handler = std::sync::Arc::new(goal_handler::GoalObjectHandler { remote_goal_repo });
|
||||
let composite = std::sync::Arc::new(composite_handler::CompositeObjectHandler {
|
||||
review: review_handler,
|
||||
watchlist: watchlist_handler,
|
||||
goal: goal_handler,
|
||||
});
|
||||
|
||||
let federation_debug = std::env::var("FEDERATION_DEBUG")
|
||||
|
||||
@@ -191,6 +191,64 @@ pub fn watchlist_to_ap_object(input: WatchlistApInput) -> WatchlistObject {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Goal object ──────────────────────────────────────────────────────────────
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct GoalObject {
|
||||
#[serde(rename = "type")]
|
||||
pub(crate) kind: NoteType,
|
||||
pub(crate) id: Url,
|
||||
pub(crate) attributed_to: Url,
|
||||
pub(crate) content: String,
|
||||
pub(crate) published: chrono::DateTime<chrono::Utc>,
|
||||
pub(crate) goal_year: u16,
|
||||
pub(crate) goal_target: u32,
|
||||
pub(crate) goal_current: u32,
|
||||
#[serde(default)]
|
||||
pub(crate) goal: bool,
|
||||
#[serde(default)]
|
||||
pub(crate) tag: Vec<ApHashtag>,
|
||||
#[serde(skip_serializing_if = "Vec::is_empty", default)]
|
||||
pub(crate) to: Vec<String>,
|
||||
#[serde(skip_serializing_if = "Vec::is_empty", default)]
|
||||
pub(crate) cc: Vec<String>,
|
||||
}
|
||||
|
||||
pub fn goal_to_ap_object(
|
||||
ap_id: Url,
|
||||
actor_url: Url,
|
||||
year: u16,
|
||||
target: u32,
|
||||
current: u32,
|
||||
base_url: &str,
|
||||
) -> GoalObject {
|
||||
let content = format!(
|
||||
"🎯 Goal: Watch {} movies in {} ({}/{})",
|
||||
target, year, current, target
|
||||
);
|
||||
let tag = vec![ApHashtag {
|
||||
kind: "Hashtag".to_string(),
|
||||
href: Url::parse(&format!("{}/tags/moviesdiary", base_url)).expect("valid base_url"),
|
||||
name: "#MoviesDiary".to_string(),
|
||||
}];
|
||||
|
||||
GoalObject {
|
||||
kind: NoteType::default(),
|
||||
id: ap_id,
|
||||
attributed_to: actor_url.clone(),
|
||||
content,
|
||||
published: chrono::Utc::now(),
|
||||
goal_year: year,
|
||||
goal_target: target,
|
||||
goal_current: current,
|
||||
goal: true,
|
||||
tag,
|
||||
to: vec![AS_PUBLIC.to_string()],
|
||||
cc: vec![format!("{}/followers", actor_url)],
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "tests/objects.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -13,6 +13,11 @@ pub fn review_url(base_url: &str, review_id: &ReviewId) -> Url {
|
||||
.expect("base_url is always a valid URL prefix")
|
||||
}
|
||||
|
||||
pub fn goal_url(base_url: &str, user_id: uuid::Uuid, year: u16) -> Url {
|
||||
Url::parse(&format!("{}/users/{}/goals/{}", base_url, user_id, year))
|
||||
.expect("base_url is always a valid URL prefix")
|
||||
}
|
||||
|
||||
/// Builds the canonical watchlist entry URL: `{base_url}/users/{user_id}/watchlist/{movie_id}`
|
||||
pub fn watchlist_entry_url(base_url: &str, user_id: uuid::Uuid, movie_id: uuid::Uuid) -> Url {
|
||||
Url::parse(&format!(
|
||||
|
||||
@@ -2,7 +2,9 @@ use chrono::NaiveDateTime;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
events::DomainEvent,
|
||||
value_objects::{ExternalMetadataId, MovieId, PosterPath, Rating, ReviewId, UserId, WrapUpId},
|
||||
value_objects::{
|
||||
ExternalMetadataId, GoalId, MovieId, PosterPath, Rating, ReviewId, UserId, WrapUpId,
|
||||
},
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
@@ -90,6 +92,23 @@ pub enum EventPayload {
|
||||
PosterSynced {
|
||||
movie_id: String,
|
||||
},
|
||||
GoalCreated {
|
||||
goal_id: String,
|
||||
user_id: String,
|
||||
year: u16,
|
||||
target_count: u32,
|
||||
},
|
||||
GoalUpdated {
|
||||
goal_id: String,
|
||||
user_id: String,
|
||||
year: u16,
|
||||
target_count: u32,
|
||||
},
|
||||
GoalDeleted {
|
||||
goal_id: String,
|
||||
user_id: String,
|
||||
year: u16,
|
||||
},
|
||||
}
|
||||
|
||||
impl EventPayload {
|
||||
@@ -113,6 +132,9 @@ impl EventPayload {
|
||||
EventPayload::WrapUpCompleted { .. } => "WrapUpCompleted",
|
||||
EventPayload::SearchReindexRequested => "SearchReindexRequested",
|
||||
EventPayload::PosterSynced { .. } => "PosterSynced",
|
||||
EventPayload::GoalCreated { .. } => "GoalCreated",
|
||||
EventPayload::GoalUpdated { .. } => "GoalUpdated",
|
||||
EventPayload::GoalDeleted { .. } => "GoalDeleted",
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -258,6 +280,37 @@ impl From<&DomainEvent> for EventPayload {
|
||||
DomainEvent::PosterSynced { movie_id } => EventPayload::PosterSynced {
|
||||
movie_id: movie_id.value().to_string(),
|
||||
},
|
||||
DomainEvent::GoalCreated {
|
||||
goal_id,
|
||||
user_id,
|
||||
year,
|
||||
target_count,
|
||||
} => EventPayload::GoalCreated {
|
||||
goal_id: goal_id.value().to_string(),
|
||||
user_id: user_id.value().to_string(),
|
||||
year: *year,
|
||||
target_count: *target_count,
|
||||
},
|
||||
DomainEvent::GoalUpdated {
|
||||
goal_id,
|
||||
user_id,
|
||||
year,
|
||||
target_count,
|
||||
} => EventPayload::GoalUpdated {
|
||||
goal_id: goal_id.value().to_string(),
|
||||
user_id: user_id.value().to_string(),
|
||||
year: *year,
|
||||
target_count: *target_count,
|
||||
},
|
||||
DomainEvent::GoalDeleted {
|
||||
goal_id,
|
||||
user_id,
|
||||
year,
|
||||
} => EventPayload::GoalDeleted {
|
||||
goal_id: goal_id.value().to_string(),
|
||||
user_id: user_id.value().to_string(),
|
||||
year: *year,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -412,6 +465,37 @@ impl TryFrom<EventPayload> for DomainEvent {
|
||||
EventPayload::PosterSynced { movie_id } => Ok(DomainEvent::PosterSynced {
|
||||
movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?),
|
||||
}),
|
||||
EventPayload::GoalCreated {
|
||||
goal_id,
|
||||
user_id,
|
||||
year,
|
||||
target_count,
|
||||
} => Ok(DomainEvent::GoalCreated {
|
||||
goal_id: GoalId::from_uuid(parse_uuid(&goal_id, "goal_id")?),
|
||||
user_id: UserId::from_uuid(parse_uuid(&user_id, "user_id")?),
|
||||
year,
|
||||
target_count,
|
||||
}),
|
||||
EventPayload::GoalUpdated {
|
||||
goal_id,
|
||||
user_id,
|
||||
year,
|
||||
target_count,
|
||||
} => Ok(DomainEvent::GoalUpdated {
|
||||
goal_id: GoalId::from_uuid(parse_uuid(&goal_id, "goal_id")?),
|
||||
user_id: UserId::from_uuid(parse_uuid(&user_id, "user_id")?),
|
||||
year,
|
||||
target_count,
|
||||
}),
|
||||
EventPayload::GoalDeleted {
|
||||
goal_id,
|
||||
user_id,
|
||||
year,
|
||||
} => Ok(DomainEvent::GoalDeleted {
|
||||
goal_id: GoalId::from_uuid(parse_uuid(&goal_id, "goal_id")?),
|
||||
user_id: UserId::from_uuid(parse_uuid(&user_id, "user_id")?),
|
||||
year,
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,6 +20,9 @@ pub fn event_to_subject(prefix: &str, event: &DomainEvent) -> String {
|
||||
DomainEvent::WrapUpCompleted { .. } => "wrapup.completed",
|
||||
DomainEvent::SearchReindexRequested => "search.reindex.requested",
|
||||
DomainEvent::PosterSynced { .. } => "poster.synced",
|
||||
DomainEvent::GoalCreated { .. } => "goal.created",
|
||||
DomainEvent::GoalUpdated { .. } => "goal.updated",
|
||||
DomainEvent::GoalDeleted { .. } => "goal.deleted",
|
||||
};
|
||||
format!("{prefix}.{suffix}")
|
||||
}
|
||||
|
||||
25
crates/adapters/postgres/migrations/0025_goals.sql
Normal file
25
crates/adapters/postgres/migrations/0025_goals.sql
Normal file
@@ -0,0 +1,25 @@
|
||||
CREATE TABLE IF NOT EXISTS goals (
|
||||
id TEXT PRIMARY KEY NOT NULL,
|
||||
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
year BIGINT NOT NULL,
|
||||
target_count BIGINT NOT NULL,
|
||||
goal_type TEXT NOT NULL DEFAULT 'movies',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
UNIQUE(user_id, year)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_goals_user ON goals(user_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS user_settings (
|
||||
user_id TEXT PRIMARY KEY NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
federate_goals BIGINT NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS remote_goals (
|
||||
ap_id TEXT PRIMARY KEY NOT NULL,
|
||||
actor_url TEXT NOT NULL,
|
||||
year BIGINT NOT NULL,
|
||||
target_count BIGINT NOT NULL,
|
||||
current_count BIGINT NOT NULL DEFAULT 0,
|
||||
received_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_remote_goals_actor ON remote_goals(actor_url);
|
||||
@@ -228,4 +228,48 @@ 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,
|
||||
year: u16,
|
||||
) -> Result<Option<(domain::models::Goal, u32)>, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
let y = year as i64;
|
||||
|
||||
let row = sqlx::query(
|
||||
"SELECT id, user_id, year, target_count, goal_type, \
|
||||
to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS created_at \
|
||||
FROM goals WHERE user_id = $1 AND year = $2",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(y)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
let Some(r) = row else { return Ok(None) };
|
||||
|
||||
let goal = crate::goals::row_to_goal(&r)?;
|
||||
let count = crate::goals::count_reviews_in_year(&self.pool, user_id, year).await?;
|
||||
|
||||
Ok(Some((goal, count)))
|
||||
}
|
||||
}
|
||||
|
||||
195
crates/adapters/postgres/src/goals.rs
Normal file
195
crates/adapters/postgres/src/goals.rs
Normal file
@@ -0,0 +1,195 @@
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::{Goal, GoalType},
|
||||
ports::GoalRepository,
|
||||
value_objects::{GoalId, UserId},
|
||||
};
|
||||
use sqlx::{PgPool, Row};
|
||||
|
||||
use crate::models::{datetime_to_str, parse_datetime, parse_uuid};
|
||||
|
||||
pub struct PostgresGoalRepository {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresGoalRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
fn map_err(e: sqlx::Error) -> DomainError {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GoalRepository for PostgresGoalRepository {
|
||||
async fn save(&self, goal: &Goal) -> Result<(), DomainError> {
|
||||
let id = goal.id().value().to_string();
|
||||
let user_id = goal.user_id().value().to_string();
|
||||
let year = goal.year() as i64;
|
||||
let target = goal.target_count() as i64;
|
||||
let goal_type = goal.goal_type().as_str();
|
||||
let created_at = datetime_to_str(goal.created_at());
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO goals (id, user_id, year, target_count, goal_type, created_at) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6::timestamptz)",
|
||||
)
|
||||
.bind(&id)
|
||||
.bind(&user_id)
|
||||
.bind(year)
|
||||
.bind(target)
|
||||
.bind(goal_type)
|
||||
.bind(&created_at)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn update(&self, goal: &Goal) -> Result<(), DomainError> {
|
||||
let id = goal.id().value().to_string();
|
||||
let target = goal.target_count() as i64;
|
||||
|
||||
let result = sqlx::query("UPDATE goals SET target_count = $1 WHERE id = $2")
|
||||
.bind(target)
|
||||
.bind(&id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(DomainError::NotFound("Goal not found".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete(&self, id: &GoalId, user_id: &UserId) -> Result<(), DomainError> {
|
||||
let gid = id.value().to_string();
|
||||
let uid = user_id.value().to_string();
|
||||
|
||||
let result = sqlx::query("DELETE FROM goals WHERE id = $1 AND user_id = $2")
|
||||
.bind(&gid)
|
||||
.bind(&uid)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(DomainError::NotFound("Goal not found".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn find_by_user_and_year(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
year: u16,
|
||||
) -> Result<Option<Goal>, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
let y = year as i64;
|
||||
|
||||
let row = sqlx::query(
|
||||
"SELECT id, user_id, year, target_count, goal_type, \
|
||||
to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS created_at \
|
||||
FROM goals WHERE user_id = $1 AND year = $2",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(y)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
row.map(|r| row_to_goal(&r)).transpose()
|
||||
}
|
||||
|
||||
async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<Goal>, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
|
||||
let rows = sqlx::query(
|
||||
"SELECT id, user_id, year, target_count, goal_type, \
|
||||
to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS created_at \
|
||||
FROM goals WHERE user_id = $1 ORDER BY year DESC",
|
||||
)
|
||||
.bind(&uid)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
rows.iter().map(row_to_goal).collect()
|
||||
}
|
||||
|
||||
async fn count_reviews_in_year(&self, user_id: &UserId, year: u16) -> Result<u32, DomainError> {
|
||||
count_reviews_in_year(&self.pool, user_id, year).await
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn count_reviews_in_year(
|
||||
pool: &PgPool,
|
||||
user_id: &UserId,
|
||||
year: u16,
|
||||
) -> Result<u32, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
let start = format!("{year}-01-01 00:00:00");
|
||||
let end = format!("{}-01-01 00:00:00", year + 1);
|
||||
|
||||
let count: i64 = sqlx::query(
|
||||
"SELECT COUNT(*) FROM reviews \
|
||||
WHERE user_id = $1 \
|
||||
AND watched_at >= $2::timestamptz \
|
||||
AND watched_at < $3::timestamptz \
|
||||
AND remote_actor_url IS NULL",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(&start)
|
||||
.bind(&end)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
})?
|
||||
.try_get(0)
|
||||
.unwrap_or(0);
|
||||
|
||||
Ok(count as u32)
|
||||
}
|
||||
|
||||
pub(crate) fn row_to_goal(r: &sqlx::postgres::PgRow) -> Result<Goal, DomainError> {
|
||||
let id_str: String = r
|
||||
.try_get("id")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read goal id: {e}")))?;
|
||||
let user_id_str: String = r
|
||||
.try_get("user_id")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read user_id: {e}")))?;
|
||||
let year: i64 = r
|
||||
.try_get("year")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read year: {e}")))?;
|
||||
let target: i64 = r.try_get("target_count").map_err(|e| {
|
||||
DomainError::InfrastructureError(format!("Failed to read target_count: {e}"))
|
||||
})?;
|
||||
let goal_type_str: String = r
|
||||
.try_get("goal_type")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read goal_type: {e}")))?;
|
||||
let created_at_str: String = r
|
||||
.try_get("created_at")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read created_at: {e}")))?;
|
||||
|
||||
let id = GoalId::from_uuid(parse_uuid(&id_str)?);
|
||||
let user_id = UserId::from_uuid(parse_uuid(&user_id_str)?);
|
||||
let goal_type: GoalType = goal_type_str.parse()?;
|
||||
let created_at = parse_datetime(&created_at_str)?;
|
||||
|
||||
Ok(Goal::from_persistence(
|
||||
id,
|
||||
user_id,
|
||||
year as u16,
|
||||
target as u32,
|
||||
goal_type,
|
||||
created_at,
|
||||
))
|
||||
}
|
||||
@@ -13,6 +13,7 @@ use domain::{
|
||||
use sqlx::PgPool;
|
||||
|
||||
mod ap_content;
|
||||
mod goals;
|
||||
mod image_ref;
|
||||
mod import_profile;
|
||||
mod import_session;
|
||||
@@ -20,6 +21,8 @@ mod models;
|
||||
mod persons;
|
||||
mod profile;
|
||||
mod profile_fields;
|
||||
mod remote_goals;
|
||||
mod user_settings;
|
||||
mod users;
|
||||
mod watch_event;
|
||||
mod watchlist;
|
||||
@@ -1008,6 +1011,9 @@ pub struct PostgresWireOutput {
|
||||
pub ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
|
||||
pub wrapup_repo: std::sync::Arc<dyn domain::ports::WrapUpRepository>,
|
||||
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 remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
|
||||
}
|
||||
|
||||
pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
|
||||
@@ -1038,6 +1044,12 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
|
||||
watchlist: std::sync::Arc::new(PostgresWatchlistRepository::new(pool.clone())) as _,
|
||||
ap_content: std::sync::Arc::new(PostgresApContentQuery::new(pool.clone())) as _,
|
||||
wrapup_repo: std::sync::Arc::new(PostgresWrapUpRepository::new(pool.clone())) as _,
|
||||
wrapup_stats: std::sync::Arc::new(PostgresWrapUpStatsQuery::new(pool)) 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 _,
|
||||
remote_goal: std::sync::Arc::new(remote_goals::PostgresRemoteGoalRepository::new(pool))
|
||||
as _,
|
||||
})
|
||||
}
|
||||
|
||||
109
crates/adapters/postgres/src/remote_goals.rs
Normal file
109
crates/adapters/postgres/src/remote_goals.rs
Normal file
@@ -0,0 +1,109 @@
|
||||
use async_trait::async_trait;
|
||||
use chrono::TimeZone;
|
||||
use domain::{errors::DomainError, models::RemoteGoalEntry, ports::RemoteGoalRepository};
|
||||
use sqlx::{PgPool, Row};
|
||||
|
||||
pub struct PostgresRemoteGoalRepository {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresRemoteGoalRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
fn map_err(e: sqlx::Error) -> DomainError {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl RemoteGoalRepository for PostgresRemoteGoalRepository {
|
||||
async fn save(&self, entry: RemoteGoalEntry) -> Result<(), DomainError> {
|
||||
let received = entry.received_at.format("%Y-%m-%d %H:%M:%S").to_string();
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO remote_goals \
|
||||
(ap_id, actor_url, year, target_count, current_count, received_at) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6::timestamptz) \
|
||||
ON CONFLICT (ap_id) DO UPDATE SET \
|
||||
target_count = $4, current_count = $5",
|
||||
)
|
||||
.bind(&entry.ap_id)
|
||||
.bind(&entry.actor_url)
|
||||
.bind(entry.year as i64)
|
||||
.bind(entry.target_count as i64)
|
||||
.bind(entry.current_count as i64)
|
||||
.bind(&received)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn update_by_ap_id(
|
||||
&self,
|
||||
ap_id: &str,
|
||||
target: u32,
|
||||
current: u32,
|
||||
) -> Result<(), DomainError> {
|
||||
sqlx::query(
|
||||
"UPDATE remote_goals SET target_count = $1, current_count = $2 WHERE ap_id = $3",
|
||||
)
|
||||
.bind(target as i64)
|
||||
.bind(current as i64)
|
||||
.bind(ap_id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn remove_by_ap_id(&self, ap_id: &str, actor_url: &str) -> Result<(), DomainError> {
|
||||
sqlx::query("DELETE FROM remote_goals WHERE ap_id = $1 AND actor_url = $2")
|
||||
.bind(ap_id)
|
||||
.bind(actor_url)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_by_actor_url(&self, actor_url: &str) -> Result<Vec<RemoteGoalEntry>, DomainError> {
|
||||
let rows = sqlx::query(
|
||||
"SELECT ap_id, actor_url, year, target_count, current_count, \
|
||||
to_char(received_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS received_at \
|
||||
FROM remote_goals WHERE actor_url = $1 ORDER BY year DESC",
|
||||
)
|
||||
.bind(actor_url)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
rows.iter()
|
||||
.map(|r| {
|
||||
let year: i64 = r.try_get("year").unwrap_or(0);
|
||||
let target: i64 = r.try_get("target_count").unwrap_or(0);
|
||||
let current: i64 = r.try_get("current_count").unwrap_or(0);
|
||||
let received_str: String = r.try_get("received_at").unwrap_or_default();
|
||||
let received_at =
|
||||
chrono::NaiveDateTime::parse_from_str(&received_str, "%Y-%m-%d %H:%M:%S")
|
||||
.map(|ndt| chrono::Utc.from_utc_datetime(&ndt))
|
||||
.unwrap_or_else(|_| chrono::Utc::now());
|
||||
|
||||
Ok(RemoteGoalEntry {
|
||||
ap_id: r.try_get("ap_id").unwrap_or_default(),
|
||||
actor_url: r.try_get("actor_url").unwrap_or_default(),
|
||||
year: year as u16,
|
||||
target_count: target as u32,
|
||||
current_count: current as u32,
|
||||
received_at,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
62
crates/adapters/postgres/src/user_settings.rs
Normal file
62
crates/adapters/postgres/src/user_settings.rs
Normal file
@@ -0,0 +1,62 @@
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError, models::UserSettings, ports::UserSettingsRepository, value_objects::UserId,
|
||||
};
|
||||
use sqlx::{PgPool, Row};
|
||||
|
||||
pub struct PostgresUserSettingsRepository {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
impl PostgresUserSettingsRepository {
|
||||
pub fn new(pool: PgPool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
fn map_err(e: sqlx::Error) -> DomainError {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
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)?;
|
||||
|
||||
match row {
|
||||
Some(r) => {
|
||||
let federate: i64 = r.try_get("federate_goals").unwrap_or(0);
|
||||
Ok(UserSettings::from_persistence(
|
||||
user_id.clone(),
|
||||
federate != 0,
|
||||
))
|
||||
}
|
||||
None => Ok(UserSettings::new(user_id.clone())),
|
||||
}
|
||||
}
|
||||
|
||||
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",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(federate)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
25
crates/adapters/sqlite/migrations/0025_goals.sql
Normal file
25
crates/adapters/sqlite/migrations/0025_goals.sql
Normal file
@@ -0,0 +1,25 @@
|
||||
CREATE TABLE IF NOT EXISTS goals (
|
||||
id TEXT PRIMARY KEY NOT NULL,
|
||||
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
year INTEGER NOT NULL,
|
||||
target_count INTEGER NOT NULL,
|
||||
goal_type TEXT NOT NULL DEFAULT 'movies',
|
||||
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%d %H:%M:%S', 'now')),
|
||||
UNIQUE(user_id, year)
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_goals_user ON goals(user_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS user_settings (
|
||||
user_id TEXT PRIMARY KEY NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
federate_goals INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS remote_goals (
|
||||
ap_id TEXT PRIMARY KEY NOT NULL,
|
||||
actor_url TEXT NOT NULL,
|
||||
year INTEGER NOT NULL,
|
||||
target_count INTEGER NOT NULL,
|
||||
current_count INTEGER NOT NULL DEFAULT 0,
|
||||
received_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%d %H:%M:%S', 'now'))
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_remote_goals_actor ON remote_goals(actor_url);
|
||||
@@ -1,11 +1,11 @@
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::{DiaryEntry, Movie, Review, WatchlistWithMovie},
|
||||
models::{DiaryEntry, Goal, Movie, Review, WatchlistWithMovie},
|
||||
ports::LocalApContentQuery,
|
||||
value_objects::{MovieId, ReviewId, UserId},
|
||||
};
|
||||
use sqlx::SqlitePool;
|
||||
use sqlx::{Row, SqlitePool};
|
||||
|
||||
use crate::models::{DiaryRow, MovieRow, ReviewRow, WatchlistRow};
|
||||
|
||||
@@ -168,4 +168,47 @@ 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,
|
||||
year: u16,
|
||||
) -> Result<Option<(Goal, u32)>, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
let y = year as i64;
|
||||
|
||||
let row = sqlx::query(
|
||||
"SELECT id, user_id, year, target_count, goal_type, created_at \
|
||||
FROM goals WHERE user_id = ? AND year = ?",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(y)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
let Some(r) = row else { return Ok(None) };
|
||||
|
||||
let goal = crate::goals::row_to_goal(&r)?;
|
||||
let count = crate::goals::count_reviews_in_year(&self.pool, user_id, year).await?;
|
||||
|
||||
Ok(Some((goal, count)))
|
||||
}
|
||||
}
|
||||
|
||||
196
crates/adapters/sqlite/src/goals.rs
Normal file
196
crates/adapters/sqlite/src/goals.rs
Normal file
@@ -0,0 +1,196 @@
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::{Goal, GoalType},
|
||||
ports::GoalRepository,
|
||||
value_objects::{GoalId, UserId},
|
||||
};
|
||||
use sqlx::{Row, SqlitePool};
|
||||
|
||||
pub struct SqliteGoalRepository {
|
||||
pool: SqlitePool,
|
||||
}
|
||||
|
||||
impl SqliteGoalRepository {
|
||||
pub fn new(pool: SqlitePool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
fn map_err(e: sqlx::Error) -> DomainError {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl GoalRepository for SqliteGoalRepository {
|
||||
async fn save(&self, goal: &Goal) -> Result<(), DomainError> {
|
||||
let id = goal.id().value().to_string();
|
||||
let user_id = goal.user_id().value().to_string();
|
||||
let year = goal.year() as i64;
|
||||
let target = goal.target_count() as i64;
|
||||
let goal_type = goal.goal_type().as_str();
|
||||
let created_at = goal.created_at().format("%Y-%m-%d %H:%M:%S").to_string();
|
||||
|
||||
sqlx::query(
|
||||
"INSERT INTO goals (id, user_id, year, target_count, goal_type, created_at) \
|
||||
VALUES (?, ?, ?, ?, ?, ?)",
|
||||
)
|
||||
.bind(&id)
|
||||
.bind(&user_id)
|
||||
.bind(year)
|
||||
.bind(target)
|
||||
.bind(goal_type)
|
||||
.bind(&created_at)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn update(&self, goal: &Goal) -> Result<(), DomainError> {
|
||||
let id = goal.id().value().to_string();
|
||||
let target = goal.target_count() as i64;
|
||||
|
||||
let result = sqlx::query("UPDATE goals SET target_count = ? WHERE id = ?")
|
||||
.bind(target)
|
||||
.bind(&id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(DomainError::NotFound("Goal not found".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete(&self, id: &GoalId, user_id: &UserId) -> Result<(), DomainError> {
|
||||
let gid = id.value().to_string();
|
||||
let uid = user_id.value().to_string();
|
||||
|
||||
let result = sqlx::query("DELETE FROM goals WHERE id = ? AND user_id = ?")
|
||||
.bind(&gid)
|
||||
.bind(&uid)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(DomainError::NotFound("Goal not found".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn find_by_user_and_year(
|
||||
&self,
|
||||
user_id: &UserId,
|
||||
year: u16,
|
||||
) -> Result<Option<Goal>, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
let y = year as i64;
|
||||
|
||||
let row = sqlx::query(
|
||||
"SELECT id, user_id, year, target_count, goal_type, created_at \
|
||||
FROM goals WHERE user_id = ? AND year = ?",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(y)
|
||||
.fetch_optional(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
row.map(|r| row_to_goal(&r)).transpose()
|
||||
}
|
||||
|
||||
async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<Goal>, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
|
||||
let rows = sqlx::query(
|
||||
"SELECT id, user_id, year, target_count, goal_type, created_at \
|
||||
FROM goals WHERE user_id = ? ORDER BY year DESC",
|
||||
)
|
||||
.bind(&uid)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
rows.iter().map(row_to_goal).collect()
|
||||
}
|
||||
|
||||
async fn count_reviews_in_year(&self, user_id: &UserId, year: u16) -> Result<u32, DomainError> {
|
||||
count_reviews_in_year(&self.pool, user_id, year).await
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn count_reviews_in_year(
|
||||
pool: &SqlitePool,
|
||||
user_id: &UserId,
|
||||
year: u16,
|
||||
) -> Result<u32, DomainError> {
|
||||
let uid = user_id.value().to_string();
|
||||
let start = format!("{year}-01-01 00:00:00");
|
||||
let end = format!("{}-01-01 00:00:00", year + 1);
|
||||
|
||||
let count: i64 = sqlx::query(
|
||||
"SELECT COUNT(*) FROM reviews \
|
||||
WHERE user_id = ? AND watched_at >= ? AND watched_at < ? \
|
||||
AND remote_actor_url IS NULL",
|
||||
)
|
||||
.bind(&uid)
|
||||
.bind(&start)
|
||||
.bind(&end)
|
||||
.fetch_one(pool)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
})?
|
||||
.try_get(0)
|
||||
.unwrap_or(0);
|
||||
|
||||
Ok(count as u32)
|
||||
}
|
||||
|
||||
pub(crate) fn row_to_goal(r: &sqlx::sqlite::SqliteRow) -> Result<Goal, DomainError> {
|
||||
let id_str: String = r
|
||||
.try_get("id")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read goal id: {e}")))?;
|
||||
let user_id_str: String = r
|
||||
.try_get("user_id")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read user_id: {e}")))?;
|
||||
let year: i64 = r
|
||||
.try_get("year")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read year: {e}")))?;
|
||||
let target: i64 = r.try_get("target_count").map_err(|e| {
|
||||
DomainError::InfrastructureError(format!("Failed to read target_count: {e}"))
|
||||
})?;
|
||||
let goal_type_str: String = r
|
||||
.try_get("goal_type")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read goal_type: {e}")))?;
|
||||
let created_at_str: String = r
|
||||
.try_get("created_at")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Failed to read created_at: {e}")))?;
|
||||
|
||||
let id = GoalId::from_uuid(
|
||||
uuid::Uuid::parse_str(&id_str)
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Invalid goal UUID: {e}")))?,
|
||||
);
|
||||
let user_id = UserId::from_uuid(
|
||||
uuid::Uuid::parse_str(&user_id_str)
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Invalid user UUID: {e}")))?,
|
||||
);
|
||||
let goal_type: GoalType = goal_type_str.parse()?;
|
||||
let created_at = chrono::NaiveDateTime::parse_from_str(&created_at_str, "%Y-%m-%d %H:%M:%S")
|
||||
.map_err(|e| DomainError::InfrastructureError(format!("Invalid datetime: {e}")))?;
|
||||
|
||||
Ok(Goal::from_persistence(
|
||||
id,
|
||||
user_id,
|
||||
year as u16,
|
||||
target as u32,
|
||||
goal_type,
|
||||
created_at,
|
||||
))
|
||||
}
|
||||
@@ -13,6 +13,7 @@ use domain::{
|
||||
use sqlx::SqlitePool;
|
||||
|
||||
mod ap_content;
|
||||
mod goals;
|
||||
mod image_ref;
|
||||
mod import_profile;
|
||||
mod import_session;
|
||||
@@ -21,6 +22,8 @@ mod models;
|
||||
mod persons;
|
||||
mod profile;
|
||||
mod profile_fields;
|
||||
mod remote_goals;
|
||||
mod user_settings;
|
||||
mod users;
|
||||
mod watch_event;
|
||||
mod watchlist;
|
||||
@@ -978,6 +981,9 @@ pub struct SqliteWireOutput {
|
||||
pub ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
|
||||
pub wrapup_repo: std::sync::Arc<dyn domain::ports::WrapUpRepository>,
|
||||
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 remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
|
||||
}
|
||||
|
||||
pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
|
||||
@@ -1015,7 +1021,12 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
|
||||
watchlist: std::sync::Arc::new(SqliteWatchlistRepository::new(pool.clone())) as _,
|
||||
ap_content: std::sync::Arc::new(SqliteApContentQuery::new(pool.clone())) as _,
|
||||
wrapup_repo: std::sync::Arc::new(SqliteWrapUpRepository::new(pool.clone())) as _,
|
||||
wrapup_stats: std::sync::Arc::new(SqliteWrapUpStatsQuery::new(pool)) 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 _,
|
||||
remote_goal: std::sync::Arc::new(remote_goals::SqliteRemoteGoalRepository::new(pool)) as _,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
104
crates/adapters/sqlite/src/remote_goals.rs
Normal file
104
crates/adapters/sqlite/src/remote_goals.rs
Normal file
@@ -0,0 +1,104 @@
|
||||
use async_trait::async_trait;
|
||||
use chrono::TimeZone;
|
||||
use domain::{errors::DomainError, models::RemoteGoalEntry, ports::RemoteGoalRepository};
|
||||
use sqlx::{Row, SqlitePool};
|
||||
|
||||
pub struct SqliteRemoteGoalRepository {
|
||||
pool: SqlitePool,
|
||||
}
|
||||
|
||||
impl SqliteRemoteGoalRepository {
|
||||
pub fn new(pool: SqlitePool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
fn map_err(e: sqlx::Error) -> DomainError {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl RemoteGoalRepository for SqliteRemoteGoalRepository {
|
||||
async fn save(&self, entry: RemoteGoalEntry) -> Result<(), DomainError> {
|
||||
let received = entry.received_at.format("%Y-%m-%d %H:%M:%S").to_string();
|
||||
|
||||
sqlx::query(
|
||||
"INSERT OR REPLACE INTO remote_goals \
|
||||
(ap_id, actor_url, year, target_count, current_count, received_at) \
|
||||
VALUES (?, ?, ?, ?, ?, ?)",
|
||||
)
|
||||
.bind(&entry.ap_id)
|
||||
.bind(&entry.actor_url)
|
||||
.bind(entry.year as i64)
|
||||
.bind(entry.target_count as i64)
|
||||
.bind(entry.current_count as i64)
|
||||
.bind(&received)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn update_by_ap_id(
|
||||
&self,
|
||||
ap_id: &str,
|
||||
target: u32,
|
||||
current: u32,
|
||||
) -> Result<(), DomainError> {
|
||||
sqlx::query("UPDATE remote_goals SET target_count = ?, current_count = ? WHERE ap_id = ?")
|
||||
.bind(target as i64)
|
||||
.bind(current as i64)
|
||||
.bind(ap_id)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn remove_by_ap_id(&self, ap_id: &str, actor_url: &str) -> Result<(), DomainError> {
|
||||
sqlx::query("DELETE FROM remote_goals WHERE ap_id = ? AND actor_url = ?")
|
||||
.bind(ap_id)
|
||||
.bind(actor_url)
|
||||
.execute(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get_by_actor_url(&self, actor_url: &str) -> Result<Vec<RemoteGoalEntry>, DomainError> {
|
||||
let rows = sqlx::query(
|
||||
"SELECT ap_id, actor_url, year, target_count, current_count, received_at \
|
||||
FROM remote_goals WHERE actor_url = ? ORDER BY year DESC",
|
||||
)
|
||||
.bind(actor_url)
|
||||
.fetch_all(&self.pool)
|
||||
.await
|
||||
.map_err(Self::map_err)?;
|
||||
|
||||
rows.iter()
|
||||
.map(|r| {
|
||||
let year: i64 = r.try_get("year").unwrap_or(0);
|
||||
let target: i64 = r.try_get("target_count").unwrap_or(0);
|
||||
let current: i64 = r.try_get("current_count").unwrap_or(0);
|
||||
let received_str: String = r.try_get("received_at").unwrap_or_default();
|
||||
let received_at =
|
||||
chrono::NaiveDateTime::parse_from_str(&received_str, "%Y-%m-%d %H:%M:%S")
|
||||
.map(|ndt| chrono::Utc.from_utc_datetime(&ndt))
|
||||
.unwrap_or_else(|_| chrono::Utc::now());
|
||||
|
||||
Ok(RemoteGoalEntry {
|
||||
ap_id: r.try_get("ap_id").unwrap_or_default(),
|
||||
actor_url: r.try_get("actor_url").unwrap_or_default(),
|
||||
year: year as u16,
|
||||
target_count: target as u32,
|
||||
current_count: current as u32,
|
||||
received_at,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
59
crates/adapters/sqlite/src/user_settings.rs
Normal file
59
crates/adapters/sqlite/src/user_settings.rs
Normal file
@@ -0,0 +1,59 @@
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError, models::UserSettings, ports::UserSettingsRepository, value_objects::UserId,
|
||||
};
|
||||
use sqlx::{Row, SqlitePool};
|
||||
|
||||
pub struct SqliteUserSettingsRepository {
|
||||
pool: SqlitePool,
|
||||
}
|
||||
|
||||
impl SqliteUserSettingsRepository {
|
||||
pub fn new(pool: SqlitePool) -> Self {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
fn map_err(e: sqlx::Error) -> DomainError {
|
||||
tracing::error!("Database error: {:?}", e);
|
||||
DomainError::InfrastructureError("Database operation failed".into())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
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)?;
|
||||
|
||||
match row {
|
||||
Some(r) => {
|
||||
let federate: i64 = r.try_get("federate_goals").unwrap_or(0);
|
||||
Ok(UserSettings::from_persistence(
|
||||
user_id.clone(),
|
||||
federate != 0,
|
||||
))
|
||||
}
|
||||
None => Ok(UserSettings::new(user_id.clone())),
|
||||
}
|
||||
}
|
||||
|
||||
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)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -209,6 +209,15 @@ pub struct ProfileTemplate<'a> {
|
||||
pub pending_followers: Vec<RemoteActorData>,
|
||||
pub sort_by: String,
|
||||
pub search: String,
|
||||
pub goals: Vec<GoalViewData>,
|
||||
}
|
||||
|
||||
pub struct GoalViewData {
|
||||
pub year: u16,
|
||||
pub target_count: u32,
|
||||
pub current_count: u32,
|
||||
pub percentage: f64,
|
||||
pub is_complete: bool,
|
||||
}
|
||||
|
||||
impl<'a> ProfileTemplate<'a> {
|
||||
|
||||
@@ -24,6 +24,25 @@
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{% if !goals.is_empty() %}
|
||||
<div class="goals-section">
|
||||
{% for g in goals %}
|
||||
<div class="goal-card">
|
||||
<div class="goal-header">
|
||||
<span class="goal-label">{{ g.year }} Goal</span>
|
||||
<span class="goal-count">{{ g.current_count }} / {{ g.target_count }} movies</span>
|
||||
</div>
|
||||
<div class="progress-track">
|
||||
<div class="progress-fill" style="width: {{ g.percentage }}%"></div>
|
||||
</div>
|
||||
{% if g.is_complete %}
|
||||
<span class="goal-complete">✦ Goal reached!</span>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endfor %}
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
{% if is_own_profile %}
|
||||
<section class="follow-section">
|
||||
<h3>Follow remote user</h3>
|
||||
|
||||
Reference in New Issue
Block a user