GKaszewski/thoughts
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor database error handling across repositories to use IntoDbResult for improved error management
Some checks failed
lint / lint (push) Has been cancelled
Details
test / unit (push) Has been cancelled
Details
test / integration (push) Has been cancelled
Details
lint / lint (pull_request) Failing after 9m30s
Details
test / unit (pull_request) Successful in 16m10s
Details
test / integration (pull_request) Failing after 16m44s
Details

Browse Source 
- Updated PgNotificationRepository to utilize IntoDbResult for error handling in various methods.
- Refactored PgRemoteActorRepository to replace manual error mapping with IntoDbResult.
- Modified PgRemoteActorConnectionRepository to implement IntoDbResult for error handling.
- Adjusted PgTagRepository to use IntoDbResult for consistent error management.
- Introduced test_helpers module for seeding users and thoughts in tests.
- Enhanced PgThoughtRepository to leverage IntoDbResult for error handling.
- Updated PgTopFriendRepository to utilize IntoDbResult for error management.
- Refactored PgUserRepository to implement IntoDbResult for error handling.
- Added constants for pagination defaults in requests.
- Introduced MAX_TOP_FRIENDS constant for top friends validation.
- Refactored JWT expiration time to use a constant.
- Improved rate limiter configuration with constants for better readability.
- Added utility methods for FollowState and Visibility enums for string conversions.
- Introduced maximum length constants for Username, Email, and Content value objects.
- Cleaned up test modules by removing redundant code and utilizing a shared testing state.
This commit is contained in:
Gabriel Kaszewski
2026-05-15 12:31:25 +02:00
parent a040a38036
commit 314dad5451
40 changed files with 456 additions and 690 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
crates/adapters/activitypub-base/src/actors.rs
View File

@@ -91,6 +91,29 @@ pub struct Person {
attachment: Vec<ProfileFieldObject>,
}
struct ActorUrls {
ap_id: Url,
inbox_url: Url,
shared_inbox_url: Option<Url>,
outbox_url: Url,
followers_url: Url,
following_url: Url,
}
impl ActorUrls {
fn build(base_url: &str, user_id: uuid::Uuid) -> Self {
let ap_id = crate::urls::actor_url(base_url, user_id);
Self {
inbox_url: Url::parse(&format!("{}/inbox", &ap_id)).expect("valid url"),
shared_inbox_url: Url::parse(&format!("{}/inbox", base_url)).ok(),
outbox_url: Url::parse(&format!("{}/outbox", &ap_id)).expect("valid url"),
followers_url: Url::parse(&format!("{}/followers", &ap_id)).expect("valid url"),
following_url: Url::parse(&format!("{}/following", &ap_id)).expect("valid url"),
ap_id,
}
}
}
pub async fn get_local_actor(
user_id: uuid::Uuid,
data: &Data<FederationData>,
@@ -117,12 +140,14 @@ pub async fn get_local_actor(
}
};
let ap_id = crate::urls::actor_url(&data.base_url, user_id);
let inbox_url = Url::parse(&format!("{}/inbox", &ap_id)).expect("valid inbox url");
let shared_inbox_url = Url::parse(&format!("{}/inbox", data.base_url)).ok();
let outbox_url = Url::parse(&format!("{}/outbox", &ap_id)).expect("valid outbox url");
let followers_url = Url::parse(&format!("{}/followers", &ap_id)).expect("valid followers url");
let following_url = Url::parse(&format!("{}/following", &ap_id)).expect("valid following url");
let ActorUrls {
ap_id,
inbox_url,
shared_inbox_url,
outbox_url,
followers_url,
following_url,
} = ActorUrls::build(&data.base_url, user_id);
Ok(DbActor {
user_id,
@@ -182,12 +207,14 @@ impl Object for DbActor {
None => return Ok(None),
};
let ap_id = crate::urls::actor_url(&data.base_url, user_id);
let inbox_url = Url::parse(&format!("{}/inbox", &ap_id)).expect("valid url");
let shared_inbox_url = Url::parse(&format!("{}/inbox", data.base_url)).ok();
let outbox_url = Url::parse(&format!("{}/outbox", &ap_id)).expect("valid url");
let followers_url = Url::parse(&format!("{}/followers", &ap_id)).expect("valid url");
let following_url = Url::parse(&format!("{}/following", &ap_id)).expect("valid url");
let ActorUrls {
ap_id,
inbox_url,
shared_inbox_url,
outbox_url,
followers_url,
following_url,
} = ActorUrls::build(&data.base_url, user_id);
Ok(Some(DbActor {
user_id,

183
crates/adapters/activitypub-base/src/followers_handler.rs
View File

@@ -5,70 +5,95 @@ use serde_json::json;
use crate::data::FederationData;
use crate::error::Error;
const PAGE_SIZE: usize = 20;
use crate::urls::AP_PAGE_SIZE;
#[derive(Deserialize)]
pub struct PageQuery {
page: Option<u32>,
}
async fn collection_handler(
user_id_str: &str,
query: PageQuery,
data: Data<FederationData>,
collection_type: &str,
) -> Result<FederationJson<serde_json::Value>, Error> {
let user_id = uuid::Uuid::parse_str(user_id_str)
.map_err(|_| Error::bad_request(anyhow::anyhow!("invalid user id")))?;
data.user_repo
.find_by_id(user_id)
.await
.map_err(Error::from)?
.ok_or_else(|| Error::not_found(anyhow::anyhow!("user not found")))?;
let collection_id = format!(
"{}/users/{}/{}",
data.base_url, user_id_str, collection_type
);
let total = match collection_type {
"followers" => data.federation_repo.count_followers(user_id).await,
_ => data.federation_repo.count_following(user_id).await,
}
.map_err(Error::from)?;
if let Some(page) = query.page {
let page = page.max(1);
let offset = (page.saturating_sub(1) as usize) * AP_PAGE_SIZE;
let items: Vec<String> = match collection_type {
"followers" => data
.federation_repo
.get_followers_page(user_id, offset as u32, AP_PAGE_SIZE)
.await
.map_err(Error::from)?
.into_iter()
.map(|f| f.actor.url)
.collect(),
_ => data
.federation_repo
.get_following_page(user_id, offset as u32, AP_PAGE_SIZE)
.await
.map_err(Error::from)?
.into_iter()
.map(|a| a.url)
.collect(),
};
let has_next = offset + items.len() < total;
let mut obj = json!({
"@context": crate::urls::AP_CONTEXT,
"type": "OrderedCollectionPage",
"id": format!("{}?page={}", collection_id, page),
"partOf": collection_id,
"totalItems": total,
"orderedItems": items,
});
if has_next {
obj["next"] = json!(format!("{}?page={}", collection_id, page + 1));
}
Ok(FederationJson(obj))
} else {
Ok(FederationJson(json!({
"@context": crate::urls::AP_CONTEXT,
"type": "OrderedCollection",
"id": collection_id,
"totalItems": total,
"first": format!("{}?page=1", collection_id),
})))
}
}
pub async fn followers_handler(
Path(user_id_str): Path<String>,
Query(query): Query<PageQuery>,
data: Data<FederationData>,
) -> Result<FederationJson<serde_json::Value>, Error> {
let user_id = uuid::Uuid::parse_str(&user_id_str)
.map_err(|_| Error::bad_request(anyhow::anyhow!("invalid user id")))?;
data.user_repo
.find_by_id(user_id)
.await
.map_err(Error::from)?
.ok_or_else(|| Error::not_found(anyhow::anyhow!("user not found")))?;
let collection_id = format!("{}/users/{}/followers", data.base_url, user_id_str);
let total = data
.federation_repo
.count_followers(user_id)
.await
.map_err(Error::from)?;
if let Some(page) = query.page {
let page = page.max(1);
let offset = (page.saturating_sub(1) as usize) * PAGE_SIZE;
let followers = data
.federation_repo
.get_followers_page(user_id, offset as u32, PAGE_SIZE)
.await
.map_err(Error::from)?;
let has_next = offset + followers.len() < total;
let items: Vec<String> = followers.into_iter().map(|f| f.actor.url).collect();
let mut obj = json!({
"@context": "https://www.w3.org/ns/activitystreams",
"type": "OrderedCollectionPage",
"id": format!("{}?page={}", collection_id, page),
"partOf": collection_id,
"totalItems": total,
"orderedItems": items,
});
if has_next {
obj["next"] = json!(format!("{}?page={}", collection_id, page + 1));
}
Ok(FederationJson(obj))
} else {
Ok(FederationJson(json!({
"@context": "https://www.w3.org/ns/activitystreams",
"type": "OrderedCollection",
"id": collection_id,
"totalItems": total,
"first": format!("{}?page=1", collection_id),
})))
}
collection_handler(&user_id_str, query, data, "followers").await
}
pub async fn following_handler(
@@ -76,55 +101,5 @@ pub async fn following_handler(
Query(query): Query<PageQuery>,
data: Data<FederationData>,
) -> Result<FederationJson<serde_json::Value>, Error> {
let user_id = uuid::Uuid::parse_str(&user_id_str)
.map_err(|_| Error::bad_request(anyhow::anyhow!("invalid user id")))?;
data.user_repo
.find_by_id(user_id)
.await
.map_err(Error::from)?
.ok_or_else(|| Error::not_found(anyhow::anyhow!("user not found")))?;
let collection_id = format!("{}/users/{}/following", data.base_url, user_id_str);
let total = data
.federation_repo
.count_following(user_id)
.await
.map_err(Error::from)?;
if let Some(page) = query.page {
let page = page.max(1);
let offset = (page.saturating_sub(1) as usize) * PAGE_SIZE;
let following = data
.federation_repo
.get_following_page(user_id, offset as u32, PAGE_SIZE)
.await
.map_err(Error::from)?;
let has_next = offset + following.len() < total;
let items: Vec<String> = following.into_iter().map(|a| a.url).collect();
let mut obj = json!({
"@context": "https://www.w3.org/ns/activitystreams",
"type": "OrderedCollectionPage",
"id": format!("{}?page={}", collection_id, page),
"partOf": collection_id,
"totalItems": total,
"orderedItems": items,
});
if has_next {
obj["next"] = json!(format!("{}?page={}", collection_id, page + 1));
}
Ok(FederationJson(obj))
} else {
Ok(FederationJson(json!({
"@context": "https://www.w3.org/ns/activitystreams",
"type": "OrderedCollection",
"id": collection_id,
"totalItems": total,
"first": format!("{}?page=1", collection_id),
})))
}
collection_handler(&user_id_str, query, data, "following").await
}

4
crates/adapters/activitypub-base/src/nodeinfo.rs
View File

@@ -5,6 +5,8 @@ use serde::Serialize;
use crate::data::FederationData;
use crate::error::Error;
const NODEINFO_2_0_REL: &str = "http://nodeinfo.diaspora.software/ns/schema/2.0";
#[derive(Serialize)]
pub struct NodeInfoWellKnown {
pub links: Vec<NodeInfoLink>,
@@ -50,7 +52,7 @@ pub async fn nodeinfo_well_known_handler(
let href = format!("{}/nodeinfo/2.0", data.base_url);
Ok(Json(NodeInfoWellKnown {
links: vec![NodeInfoLink {
rel: "http://nodeinfo.diaspora.software/ns/schema/2.0".to_string(),
rel: NODEINFO_2_0_REL.to_string(),
href,
}],
}))

12
crates/adapters/activitypub-base/src/outbox.rs
View File

@@ -9,9 +9,7 @@ use activitypub_federation::{
protocol::context::WithContext,
};
use crate::{activities::CreateActivity, data::FederationData, error::Error};
const PAGE_SIZE: usize = 20;
use crate::{activities::CreateActivity, data::FederationData, error::Error, urls::AP_PAGE_SIZE};
#[derive(Deserialize)]
pub struct OutboxQuery {
@@ -66,7 +64,7 @@ pub async fn outbox_handler(
let items = data
.object_handler
.get_local_objects_page(uuid, before, PAGE_SIZE)
.get_local_objects_page(uuid, before, AP_PAGE_SIZE)
.await
.map_err(|e| Error::from(anyhow::anyhow!("{}", e)))?;
@@ -74,7 +72,7 @@ pub async fn outbox_handler(
.parse()
.expect("valid url");
let has_more = items.len() == PAGE_SIZE;
let has_more = items.len() == AP_PAGE_SIZE;
let oldest_ts = items.last().map(|(_, _, ts)| *ts);
let followers_url = format!("{}/followers", actor_url);
@@ -110,7 +108,7 @@ pub async fn outbox_handler(
};
Ok(axum::Json(OrderedCollectionPage {
context: "https://www.w3.org/ns/activitystreams".to_string(),
context: crate::urls::AP_CONTEXT.to_string(),
kind: "OrderedCollectionPage".to_string(),
id: page_id,
part_of: outbox_url,
@@ -127,7 +125,7 @@ pub async fn outbox_handler(
.len() as u64;
Ok(axum::Json(OrderedCollection {
context: "https://www.w3.org/ns/activitystreams".to_string(),
context: crate::urls::AP_CONTEXT.to_string(),
kind: "OrderedCollection".to_string(),
id: outbox_url.clone(),
total_items: total,

31
crates/adapters/activitypub-base/src/service.rs
View File

@@ -27,6 +27,11 @@ use crate::{
webfinger::webfinger_handler,
};
const DELIVERY_MAX_ATTEMPTS: u32 = 3;
const DELIVERY_INITIAL_DELAY_SECS: u64 = 1;
const HTTP_FETCH_TIMEOUT_SECS: u64 = 30;
const BATCH_FETCH_SLEEP_MS: u64 = 100;
fn content_to_html(text: &str) -> String {
let escaped = text
.replace('&', "&amp;")
@@ -139,11 +144,11 @@ pub(crate) async fn send_with_retry(
) -> Vec<anyhow::Error> {
let mut failures = vec![];
for send in sends {
let mut delay = std::time::Duration::from_secs(1);
for attempt in 1..=3u32 {
let mut delay = std::time::Duration::from_secs(DELIVERY_INITIAL_DELAY_SECS);
for attempt in 1..=DELIVERY_MAX_ATTEMPTS {
match send.clone().sign_and_send(data).await {
Ok(()) => break,
Err(e) if attempt < 3 => {
Err(e) if attempt < DELIVERY_MAX_ATTEMPTS => {
tracing::warn!(attempt, error = %e, "delivery failed, retrying");
tokio::time::sleep(delay).await;
delay *= 2;
@@ -1206,7 +1211,7 @@ impl ActivityPubService {
pub async fn backfill_outbox(&self, outbox_url: &str, actor_url: &str) -> anyhow::Result<()> {
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(30))
.timeout(std::time::Duration::from_secs(HTTP_FETCH_TIMEOUT_SECS))
.build()?;
let data = self.federation_config.to_request_data();
let actor = url::Url::parse(actor_url)?;
@@ -1384,7 +1389,7 @@ impl ActivityPubService {
failure_count += 1;
}
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
tokio::time::sleep(std::time::Duration::from_millis(BATCH_FETCH_SLEEP_MS)).await;
}
tracing::info!(
@@ -1705,16 +1710,16 @@ impl domain::ports::FederationActionPort for ActivityPubService {
.map_err(|e| domain::errors::DomainError::ExternalService(e.to_string()))?;
let obj = if let Some(p) = page {
let p = p.max(1);
let offset = (p.saturating_sub(1) as usize) * 20;
let offset = (p.saturating_sub(1) as usize) * crate::urls::AP_PAGE_SIZE;
let followers = data
.federation_repo
.get_followers_page(uuid, offset as u32, 20)
.get_followers_page(uuid, offset as u32, crate::urls::AP_PAGE_SIZE)
.await
.map_err(|e| domain::errors::DomainError::ExternalService(e.to_string()))?;
let has_next = offset + followers.len() < total;
let items: Vec<String> = followers.into_iter().map(|f| f.actor.url).collect();
let mut obj = serde_json::json!({
"@context": "https://www.w3.org/ns/activitystreams",
"@context": crate::urls::AP_CONTEXT,
"type": "OrderedCollectionPage",
"id": format!("{}?page={}", collection_id, p),
"partOf": collection_id,
@@ -1727,7 +1732,7 @@ impl domain::ports::FederationActionPort for ActivityPubService {
obj
} else {
serde_json::json!({
"@context": "https://www.w3.org/ns/activitystreams",
"@context": crate::urls::AP_CONTEXT,
"type": "OrderedCollection",
"id": collection_id,
"totalItems": total,
@@ -1753,16 +1758,16 @@ impl domain::ports::FederationActionPort for ActivityPubService {
.map_err(|e| domain::errors::DomainError::ExternalService(e.to_string()))?;
let obj = if let Some(p) = page {
let p = p.max(1);
let offset = (p.saturating_sub(1) as usize) * 20;
let offset = (p.saturating_sub(1) as usize) * crate::urls::AP_PAGE_SIZE;
let following = data
.federation_repo
.get_following_page(uuid, offset as u32, 20)
.get_following_page(uuid, offset as u32, crate::urls::AP_PAGE_SIZE)
.await
.map_err(|e| domain::errors::DomainError::ExternalService(e.to_string()))?;
let has_next = offset + following.len() < total;
let items: Vec<String> = following.into_iter().map(|a| a.url).collect();
let mut obj = serde_json::json!({
"@context": "https://www.w3.org/ns/activitystreams",
"@context": crate::urls::AP_CONTEXT,
"type": "OrderedCollectionPage",
"id": format!("{}?page={}", collection_id, p),
"partOf": collection_id,
@@ -1775,7 +1780,7 @@ impl domain::ports::FederationActionPort for ActivityPubService {
obj
} else {
serde_json::json!({
"@context": "https://www.w3.org/ns/activitystreams",
"@context": crate::urls::AP_CONTEXT,
"type": "OrderedCollection",
"id": collection_id,
"totalItems": total,

2
crates/adapters/activitypub-base/src/urls.rs
View File

@@ -3,6 +3,8 @@ use url::Url;
use crate::error::Error;
pub const AS_PUBLIC: &str = "https://www.w3.org/ns/activitystreams#Public";
pub const AP_CONTEXT: &str = "https://www.w3.org/ns/activitystreams";
pub const AP_PAGE_SIZE: usize = 20;
pub fn extract_user_id_from_url(url: &Url) -> Option<uuid::Uuid> {
let path = url.path();

11
crates/adapters/activitypub/src/handler.rs
View File

@@ -1,5 +1,8 @@
use anyhow::{anyhow, Result};
use async_trait::async_trait;
const USERS_PATH_PREFIX: &str = "/users/";
const THOUGHTS_PATH_PREFIX: &str = "/thoughts/";
use chrono::{DateTime, Utc};
use std::sync::Arc;
use url::Url;
@@ -172,7 +175,7 @@ impl ApObjectHandler for ThoughtsObjectHandler {
}
let user_uuid = href_url
.path()
.strip_prefix("/users/")
.strip_prefix(USERS_PATH_PREFIX)
.and_then(|s| s.split('/').next())
.and_then(|s| uuid::Uuid::parse_str(s).ok());
if let Some(uuid) = user_uuid {
@@ -217,7 +220,7 @@ impl ApObjectHandler for ThoughtsObjectHandler {
async fn on_like(&self, object_url: &Url, actor_url: &Url) -> Result<()> {
let thought_uuid = object_url
.path()
.strip_prefix("/thoughts/")
.strip_prefix(THOUGHTS_PATH_PREFIX)
.and_then(|s| s.split('/').next())
.and_then(|s| uuid::Uuid::parse_str(s).ok());
@@ -280,7 +283,7 @@ impl ApObjectHandler for ThoughtsObjectHandler {
let thought_uuid = thought_ap_id
.path()
.strip_prefix("/thoughts/")
.strip_prefix(THOUGHTS_PATH_PREFIX)
.and_then(|s| s.split('/').next())
.and_then(|s| uuid::Uuid::parse_str(s).ok());
@@ -305,7 +308,7 @@ impl ApObjectHandler for ThoughtsObjectHandler {
async fn on_announce_received(&self, object_url: &Url, actor_url: &Url) -> Result<()> {
let thought_uuid = object_url
.path()
.strip_prefix("/thoughts/")
.strip_prefix(THOUGHTS_PATH_PREFIX)
.and_then(|s| s.split('/').next())
.and_then(|s| uuid::Uuid::parse_str(s).ok());

1
crates/adapters/postgres-search/Cargo.toml
View File

@@ -5,6 +5,7 @@ edition = "2021"
[dependencies]
domain = { workspace = true }
postgres = { workspace = true }
sqlx = { workspace = true }
uuid = { workspace = true }
chrono = { workspace = true }

58
crates/adapters/postgres-search/src/lib.rs
View File

@@ -1,26 +1,17 @@
use async_trait::async_trait;
use chrono::{DateTime, Utc};
fn visibility_from_str(s: &str) -> domain::models::thought::Visibility {
use domain::models::thought::Visibility;
match s {
"followers" => Visibility::Followers,
"unlisted" => Visibility::Unlisted,
"direct" => Visibility::Direct,
_ => Visibility::Public,
}
}
use domain::{
errors::DomainError,
models::{
feed::{FeedEntry, PageParams, Paginated},
thought::Thought,
thought::{Thought, Visibility},
user::User,
},
ports::SearchPort,
value_objects::{Content, Email, PasswordHash, ThoughtId, UserId, Username},
};
use postgres::user::{UserRow, USER_SELECT};
use sqlx::PgPool;
pub struct PgSearchRepository {
@@ -88,7 +79,7 @@ fn row_to_entry(r: FeedRow) -> FeedEntry {
in_reply_to_id: r.in_reply_to_id.map(ThoughtId::from_uuid),
in_reply_to_url: r.in_reply_to_url,
ap_id: r.t_ap_id,
visibility: visibility_from_str(&r.visibility),
visibility: Visibility::from_db_str(&r.visibility),
content_warning: r.content_warning,
sensitive: r.sensitive,
local: r.t_local,
@@ -122,49 +113,6 @@ fn row_to_entry(r: FeedRow) -> FeedEntry {
}
}
#[derive(sqlx::FromRow)]
struct UserRow {
id: uuid::Uuid,
username: String,
email: String,
password_hash: String,
display_name: Option<String>,
bio: Option<String>,
avatar_url: Option<String>,
header_url: Option<String>,
custom_css: Option<String>,
local: bool,
ap_id: Option<String>,
inbox_url: Option<String>,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
}
impl From<UserRow> for User {
fn from(r: UserRow) -> Self {
User {
id: UserId::from_uuid(r.id),
username: Username::from_trusted(r.username),
email: Email::from_trusted(r.email),
password_hash: PasswordHash(r.password_hash),
display_name: r.display_name,
bio: r.bio,
avatar_url: r.avatar_url,
header_url: r.header_url,
custom_css: r.custom_css,
local: r.local,
ap_id: r.ap_id,
inbox_url: r.inbox_url,
created_at: r.created_at,
updated_at: r.updated_at,
}
}
}
const USER_SELECT: &str =
"SELECT id,username,email,password_hash,display_name,bio,avatar_url,header_url,\
custom_css,local,ap_id,inbox_url,created_at,updated_at FROM users";
#[async_trait]
impl SearchPort for PgSearchRepository {
async fn search_thoughts(

30
crates/adapters/postgres/src/activitypub.rs
View File

@@ -1,4 +1,8 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
const MAX_REMOTE_CONTENT_CHARS: usize = 500;
const THOUGHTS_PATH_PREFIX: &str = "/thoughts/";
use chrono::{DateTime, Utc};
use sqlx::PgPool;
use url::Url;
@@ -47,7 +51,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(user_id.as_uuid())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|rows| {
rows.into_iter()
.map(|r| OutboxEntry {
@@ -113,7 +117,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.fetch_all(&self.pool)
.await
}
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(rows
.into_iter()
@@ -145,7 +149,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(actor_ap_url.as_str())
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(UserId::from_uuid))
}
@@ -179,7 +183,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(actor_ap_url.as_str())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
// Re-fetch to get whichever id won the race
self.find_remote_actor_id(actor_ap_url)
.await?
@@ -205,7 +209,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(user_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -220,13 +224,13 @@ impl ActivityPubRepository for PgActivityPubRepository {
visibility: &str,
in_reply_to: Option<&Url>,
) -> Result<(), DomainError> {
let capped: String = content.chars().take(500).collect();
let capped: String = content.chars().take(MAX_REMOTE_CONTENT_CHARS).collect();
let (in_reply_to_id, in_reply_to_url) = match in_reply_to {
Some(url) => {
// If the parent is a local thought, extract its UUID for in_reply_to_id.
let local_uuid = url
.path()
.strip_prefix("/thoughts/")
.strip_prefix(THOUGHTS_PATH_PREFIX)
.and_then(|s| s.split('/').next())
.and_then(|s| uuid::Uuid::parse_str(s).ok());
(local_uuid, Some(url.as_str().to_string()))
@@ -249,12 +253,12 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(&in_reply_to_url)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
async fn apply_note_update(&self, ap_id: &Url, new_content: &str) -> Result<(), DomainError> {
let capped: String = new_content.chars().take(500).collect();
let capped: String = new_content.chars().take(MAX_REMOTE_CONTENT_CHARS).collect();
sqlx::query(
"UPDATE thoughts SET content=$2,updated_at=NOW() WHERE ap_id=$1 AND local=false",
)
@@ -262,7 +266,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(&capped)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -271,7 +275,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(ap_id.as_str())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -282,7 +286,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
.bind(actor_ap_url.as_str())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -290,7 +294,7 @@ impl ActivityPubRepository for PgActivityPubRepository {
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM thoughts WHERE local=true")
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(n as u64)
}
}

9
crates/adapters/postgres/src/api_key.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use domain::{
@@ -30,7 +31,7 @@ impl ApiKeyRepository for PgApiKeyRepository {
.bind(k.created_at)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -49,7 +50,7 @@ impl ApiKeyRepository for PgApiKeyRepository {
.bind(hash)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| {
o.map(|r| ApiKey {
id: ApiKeyId::from_uuid(r.id),
@@ -72,7 +73,7 @@ impl ApiKeyRepository for PgApiKeyRepository {
}
sqlx::query_as::<_, Row>("SELECT id,user_id,key_hash,name,created_at FROM api_keys WHERE user_id=$1 ORDER BY created_at DESC")
.bind(user_id.as_uuid()).fetch_all(&self.pool).await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|rows| rows.into_iter().map(|r| ApiKey { id: ApiKeyId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), key_hash: r.key_hash, name: r.name, created_at: r.created_at }).collect())
}
@@ -82,7 +83,7 @@ impl ApiKeyRepository for PgApiKeyRepository {
.bind(user_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
}

24
crates/adapters/postgres/src/block.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use domain::{
errors::DomainError, models::social::Block, ports::BlockRepository, value_objects::UserId,
@@ -24,7 +25,7 @@ impl BlockRepository for PgBlockRepository {
.bind(b.created_at)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -34,7 +35,7 @@ impl BlockRepository for PgBlockRepository {
.bind(blocked_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -45,7 +46,7 @@ impl BlockRepository for PgBlockRepository {
.bind(blocked_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(count > 0)
}
}
@@ -53,22 +54,9 @@ impl BlockRepository for PgBlockRepository {
#[cfg(test)]
mod tests {
use super::*;
use crate::user::PgUserRepository;
use crate::test_helpers::seed_user;
use chrono::Utc;
use domain::ports::UserRepository;
use domain::{models::user::User, value_objects::*};
async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
let repo = PgUserRepository::new(pool.clone());
let u = User::new_local(
UserId::new(),
Username::new(username).unwrap(),
Email::new(email).unwrap(),
PasswordHash("h".into()),
);
repo.save(&u).await.unwrap();
u
}
use domain::value_objects::*;
#[sqlx::test(migrations = "./migrations")]
async fn block_exists(pool: sqlx::PgPool) {

47
crates/adapters/postgres/src/boost.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use domain::{
@@ -24,7 +25,7 @@ impl BoostRepository for PgBoostRepository {
"INSERT INTO boosts(id,user_id,thought_id,ap_id,created_at) VALUES($1,$2,$3,$4,$5) ON CONFLICT(user_id,thought_id) DO NOTHING"
)
.bind(b.id.as_uuid()).bind(b.user_id.as_uuid()).bind(b.thought_id.as_uuid()).bind(&b.ap_id).bind(b.created_at)
.execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
.execute(&self.pool).await.into_domain().map(|_| ())
}
async fn delete(&self, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
@@ -33,7 +34,7 @@ impl BoostRepository for PgBoostRepository {
.bind(thought_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
if r.rows_affected() == 0 {
return Err(DomainError::NotFound);
}
@@ -56,7 +57,7 @@ impl BoostRepository for PgBoostRepository {
sqlx::query_as::<_, Row>("SELECT id,user_id,thought_id,ap_id,created_at FROM boosts WHERE user_id=$1 AND thought_id=$2")
.bind(user_id.as_uuid()).bind(thought_id.as_uuid())
.fetch_optional(&self.pool).await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(|r| Boost { id: BoostId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), thought_id: ThoughtId::from_uuid(r.thought_id), ap_id: r.ap_id, created_at: r.created_at }))
}
@@ -65,50 +66,20 @@ impl BoostRepository for PgBoostRepository {
.bind(thought_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{thought::PgThoughtRepository, user::PgUserRepository};
use crate::test_helpers::seed_user_and_thought;
use chrono::Utc;
use domain::ports::{ThoughtRepository, UserRepository};
use domain::{
models::{
thought::{Thought, Visibility},
user::User,
},
value_objects::*,
};
async fn seed(pool: &sqlx::PgPool) -> (User, Thought) {
let urepo = PgUserRepository::new(pool.clone());
let trepo = PgThoughtRepository::new(pool.clone());
let u = User::new_local(
UserId::new(),
Username::new("alice").unwrap(),
Email::new("alice@ex.com").unwrap(),
PasswordHash("h".into()),
);
urepo.save(&u).await.unwrap();
let t = Thought::new_local(
ThoughtId::new(),
u.id.clone(),
Content::new_local("hi").unwrap(),
None,
Visibility::Public,
None,
false,
);
trepo.save(&t).await.unwrap();
(u, t)
}
use domain::value_objects::*;
#[sqlx::test(migrations = "./migrations")]
async fn boost_and_count(pool: sqlx::PgPool) {
let (user, thought) = seed(&pool).await;
let (user, thought) = seed_user_and_thought(&pool).await;
let repo = PgBoostRepository::new(pool);
let boost = Boost {
id: BoostId::new(),
@@ -123,7 +94,7 @@ mod tests {
#[sqlx::test(migrations = "./migrations")]
async fn unboost(pool: sqlx::PgPool) {
let (user, thought) = seed(&pool).await;
let (user, thought) = seed_user_and_thought(&pool).await;
let repo = PgBoostRepository::new(pool);
let boost = Boost {
id: BoostId::new(),

11
crates/adapters/postgres/src/db_error.rs Normal file
View File

@@ -0,0 +1,11 @@
use domain::errors::DomainError;
pub(crate) trait IntoDbResult<T> {
fn into_domain(self) -> Result<T, DomainError>;
}
impl<T> IntoDbResult<T> for Result<T, sqlx::Error> {
fn into_domain(self) -> Result<T, DomainError> {
self.map_err(|e| DomainError::Internal(e.to_string()))
}
}

35
crates/adapters/postgres/src/feed.rs
View File

@@ -1,21 +1,12 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
fn visibility_from_str(s: &str) -> domain::models::thought::Visibility {
use domain::models::thought::Visibility;
match s {
"followers" => Visibility::Followers,
"unlisted" => Visibility::Unlisted,
"direct" => Visibility::Direct,
_ => Visibility::Public,
}
}
use domain::{
errors::DomainError,
models::{
feed::{FeedEntry, PageParams, Paginated},
thought::Thought,
thought::{Thought, Visibility},
user::User,
},
ports::FeedRepository,
@@ -127,7 +118,7 @@ fn row_to_entry(r: FeedRow) -> FeedEntry {
in_reply_to_id: r.in_reply_to_id.map(ThoughtId::from_uuid),
in_reply_to_url: r.in_reply_to_url,
ap_id: r.t_ap_id,
visibility: visibility_from_str(&r.visibility),
visibility: Visibility::from_db_str(&r.visibility),
content_warning: r.content_warning,
sensitive: r.sensitive,
local: r.t_local,
@@ -180,7 +171,7 @@ impl FeedRepository for PgFeedRepository {
.bind(&ids)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let sel = feed_select(viewer);
let sql = format!("{sel} WHERE (t.user_id=ANY($1){}) AND t.visibility != 'direct' ORDER BY t.created_at DESC LIMIT $2 OFFSET $3", fed_clause);
@@ -190,7 +181,7 @@ impl FeedRepository for PgFeedRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(row_to_entry).collect(),
@@ -211,7 +202,7 @@ impl FeedRepository for PgFeedRepository {
)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let sel = feed_select(viewer);
let sql = format!("{sel} WHERE t.local=true AND t.visibility='public' ORDER BY t.created_at DESC LIMIT $1 OFFSET $2");
@@ -220,7 +211,7 @@ impl FeedRepository for PgFeedRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(row_to_entry).collect(),
@@ -243,7 +234,7 @@ impl FeedRepository for PgFeedRepository {
.bind(query)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let sel = feed_select(viewer);
let sql = format!("{sel} WHERE t.content % $1 AND t.visibility='public' ORDER BY similarity(t.content, $1) DESC LIMIT $2 OFFSET $3");
@@ -253,7 +244,7 @@ impl FeedRepository for PgFeedRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(row_to_entry).collect(),
@@ -279,7 +270,7 @@ impl FeedRepository for PgFeedRepository {
.bind(tag_name)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let sel = feed_select(viewer);
let sql = format!(
@@ -295,7 +286,7 @@ impl FeedRepository for PgFeedRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(row_to_entry).collect(),
@@ -324,7 +315,7 @@ impl FeedRepository for PgFeedRepository {
.bind(viewer_uuid)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let sel = feed_select(viewer);
let sql = format!("{sel} WHERE t.user_id = $1 AND ($4::uuid = $1 OR (t.visibility != 'direct' AND (t.visibility IN ('public', 'unlisted') OR (t.visibility = 'followers' AND EXISTS(SELECT 1 FROM follows WHERE follower_id = $4 AND following_id = $1 AND state = 'accepted'))))) ORDER BY t.created_at DESC LIMIT $2 OFFSET $3");
@@ -335,7 +326,7 @@ impl FeedRepository for PgFeedRepository {
.bind(viewer_uuid)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(row_to_entry).collect(),

60
crates/adapters/postgres/src/follow.rs
View File

@@ -1,24 +1,7 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
fn follow_state_from_str(s: &str) -> domain::models::social::FollowState {
use domain::models::social::FollowState;
match s {
"pending" => FollowState::Pending,
"rejected" => FollowState::Rejected,
_ => FollowState::Accepted,
}
}
fn follow_state_as_str(state: &domain::models::social::FollowState) -> &'static str {
use domain::models::social::FollowState;
match state {
FollowState::Pending => "pending",
FollowState::Accepted => "accepted",
FollowState::Rejected => "rejected",
}
}
use domain::{
errors::DomainError,
models::{
@@ -50,12 +33,12 @@ impl FollowRepository for PgFollowRepository {
)
.bind(f.follower_id.as_uuid())
.bind(f.following_id.as_uuid())
.bind(follow_state_as_str(&f.state))
.bind(f.state.as_str())
.bind(&f.ap_id)
.bind(f.created_at)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -65,7 +48,7 @@ impl FollowRepository for PgFollowRepository {
.bind(following_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
if r.rows_affected() == 0 {
return Err(DomainError::NotFound);
}
@@ -92,11 +75,11 @@ impl FollowRepository for PgFollowRepository {
.bind(following_id.as_uuid())
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(|r| Follow {
follower_id: UserId::from_uuid(r.follower_id),
following_id: UserId::from_uuid(r.following_id),
state: follow_state_from_str(&r.state),
state: FollowState::from_db_str(&r.state),
ap_id: r.ap_id,
created_at: r.created_at,
}))
@@ -111,10 +94,10 @@ impl FollowRepository for PgFollowRepository {
sqlx::query("UPDATE follows SET state=$3 WHERE follower_id=$1 AND following_id=$2")
.bind(follower_id.as_uuid())
.bind(following_id.as_uuid())
.bind(follow_state_as_str(state))
.bind(state.as_str())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -129,7 +112,7 @@ impl FollowRepository for PgFollowRepository {
.bind(user_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let rows = sqlx::query_as::<_, crate::user::UserRow>(
"SELECT u.id,u.username,u.email,u.password_hash,u.display_name,u.bio,u.avatar_url,u.header_url,u.custom_css,u.local,u.ap_id,u.inbox_url,u.created_at,u.updated_at
@@ -142,7 +125,7 @@ impl FollowRepository for PgFollowRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(User::from).collect(),
@@ -163,7 +146,7 @@ impl FollowRepository for PgFollowRepository {
.bind(user_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let rows = sqlx::query_as::<_, crate::user::UserRow>(
"SELECT u.id,u.username,u.email,u.password_hash,u.display_name,u.bio,u.avatar_url,u.header_url,u.custom_css,u.local,u.ap_id,u.inbox_url,u.created_at,u.updated_at
@@ -176,7 +159,7 @@ impl FollowRepository for PgFollowRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(User::from).collect(),
@@ -196,7 +179,7 @@ impl FollowRepository for PgFollowRepository {
.bind(user_id.as_uuid())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(ids.into_iter().map(UserId::from_uuid).collect())
}
}
@@ -204,22 +187,9 @@ impl FollowRepository for PgFollowRepository {
#[cfg(test)]
mod tests {
use super::*;
use crate::user::PgUserRepository;
use crate::test_helpers::seed_user;
use chrono::Utc;
use domain::ports::UserRepository;
use domain::{models::user::User, value_objects::*};
async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
let repo = PgUserRepository::new(pool.clone());
let u = User::new_local(
UserId::new(),
Username::new(username).unwrap(),
Email::new(email).unwrap(),
PasswordHash("h".into()),
);
repo.save(&u).await.unwrap();
u
}
use domain::value_objects::*;
#[sqlx::test(migrations = "./migrations")]
async fn save_and_find_follow(pool: sqlx::PgPool) {

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

@@ -2,6 +2,7 @@ pub mod activitypub;
pub mod api_key;
pub mod block;
pub mod boost;
mod db_error;
pub mod feed;
pub mod follow;
pub mod like;
@@ -9,6 +10,8 @@ pub mod notification;
pub mod remote_actor;
pub mod remote_actor_connections;
pub mod tag;
#[cfg(test)]
pub(crate) mod test_helpers;
pub mod thought;
pub mod top_friend;
pub mod user;

47
crates/adapters/postgres/src/like.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use domain::{
@@ -24,7 +25,7 @@ impl LikeRepository for PgLikeRepository {
"INSERT INTO likes(id,user_id,thought_id,ap_id,created_at) VALUES($1,$2,$3,$4,$5) ON CONFLICT(user_id,thought_id) DO NOTHING"
)
.bind(l.id.as_uuid()).bind(l.user_id.as_uuid()).bind(l.thought_id.as_uuid()).bind(&l.ap_id).bind(l.created_at)
.execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
.execute(&self.pool).await.into_domain().map(|_| ())
}
async fn delete(&self, user_id: &UserId, thought_id: &ThoughtId) -> Result<(), DomainError> {
@@ -33,7 +34,7 @@ impl LikeRepository for PgLikeRepository {
.bind(thought_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
if r.rows_affected() == 0 {
return Err(DomainError::NotFound);
}
@@ -56,7 +57,7 @@ impl LikeRepository for PgLikeRepository {
sqlx::query_as::<_, Row>("SELECT id,user_id,thought_id,ap_id,created_at FROM likes WHERE user_id=$1 AND thought_id=$2")
.bind(user_id.as_uuid()).bind(thought_id.as_uuid())
.fetch_optional(&self.pool).await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(|r| Like { id: LikeId::from_uuid(r.id), user_id: UserId::from_uuid(r.user_id), thought_id: ThoughtId::from_uuid(r.thought_id), ap_id: r.ap_id, created_at: r.created_at }))
}
@@ -65,50 +66,20 @@ impl LikeRepository for PgLikeRepository {
.bind(thought_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{thought::PgThoughtRepository, user::PgUserRepository};
use crate::test_helpers::seed_user_and_thought;
use chrono::Utc;
use domain::ports::{ThoughtRepository, UserRepository};
use domain::{
models::{
thought::{Thought, Visibility},
user::User,
},
value_objects::*,
};
async fn seed(pool: &sqlx::PgPool) -> (User, Thought) {
let urepo = PgUserRepository::new(pool.clone());
let trepo = PgThoughtRepository::new(pool.clone());
let u = User::new_local(
UserId::new(),
Username::new("alice").unwrap(),
Email::new("alice@ex.com").unwrap(),
PasswordHash("h".into()),
);
urepo.save(&u).await.unwrap();
let t = Thought::new_local(
ThoughtId::new(),
u.id.clone(),
Content::new_local("hi").unwrap(),
None,
Visibility::Public,
None,
false,
);
trepo.save(&t).await.unwrap();
(u, t)
}
use domain::value_objects::*;
#[sqlx::test(migrations = "./migrations")]
async fn like_and_count(pool: sqlx::PgPool) {
let (user, thought) = seed(&pool).await;
let (user, thought) = seed_user_and_thought(&pool).await;
let repo = PgLikeRepository::new(pool);
let like = Like {
id: LikeId::new(),
@@ -123,7 +94,7 @@ mod tests {
#[sqlx::test(migrations = "./migrations")]
async fn unlike(pool: sqlx::PgPool) {
let (user, thought) = seed(&pool).await;
let (user, thought) = seed_user_and_thought(&pool).await;
let repo = PgLikeRepository::new(pool);
let like = Like {
id: LikeId::new(),

13
crates/adapters/postgres/src/notification.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
@@ -53,7 +54,7 @@ impl NotificationRepository for PgNotificationRepository {
.bind(n.from_user_id.as_ref().map(|u| u.as_uuid()))
.bind(n.thought_id.as_ref().map(|t| t.as_uuid()))
.bind(n.read).bind(n.created_at)
.execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
.execute(&self.pool).await.into_domain().map(|_| ())
}
async fn list_for_user(
@@ -65,7 +66,7 @@ impl NotificationRepository for PgNotificationRepository {
.bind(user_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
#[derive(sqlx::FromRow)]
struct Row {
id: uuid::Uuid,
@@ -79,7 +80,7 @@ impl NotificationRepository for PgNotificationRepository {
let rows = sqlx::query_as::<_, Row>(
"SELECT id,user_id,type,from_user_id,thought_id,read,created_at FROM notifications WHERE user_id=$1 ORDER BY created_at DESC LIMIT $2 OFFSET $3"
).bind(user_id.as_uuid()).bind(page.limit()).bind(page.offset())
.fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
.fetch_all(&self.pool).await.into_domain()?;
let items = rows
.into_iter()
.map(|r| Notification {
@@ -107,7 +108,7 @@ impl NotificationRepository for PgNotificationRepository {
.bind(user_id.as_uuid())
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(count as u64)
}
@@ -117,7 +118,7 @@ impl NotificationRepository for PgNotificationRepository {
.bind(user_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -126,7 +127,7 @@ impl NotificationRepository for PgNotificationRepository {
.bind(user_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
}

5
crates/adapters/postgres/src/remote_actor.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use domain::{
@@ -26,7 +27,7 @@ impl RemoteActorRepository for PgRemoteActorRepository {
)
.bind(&a.url).bind(&a.handle).bind(&a.display_name).bind(&a.inbox_url)
.bind(&a.shared_inbox_url).bind(&a.public_key).bind(&a.avatar_url).bind(a.last_fetched_at)
.execute(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string())).map(|_| ())
.execute(&self.pool).await.into_domain().map(|_| ())
}
async fn find_by_url(&self, url: &str) -> Result<Option<RemoteActor>, DomainError> {
@@ -44,7 +45,7 @@ impl RemoteActorRepository for PgRemoteActorRepository {
sqlx::query_as::<_, Row>(
"SELECT url,handle,display_name,inbox_url,shared_inbox_url,public_key,avatar_url,last_fetched_at FROM remote_actors WHERE url=$1"
).bind(url).fetch_optional(&self.pool).await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(|r| RemoteActor { url: r.url, handle: r.handle, display_name: r.display_name, inbox_url: r.inbox_url, shared_inbox_url: r.shared_inbox_url, public_key: r.public_key, avatar_url: r.avatar_url, last_fetched_at: r.last_fetched_at, bio: None, banner_url: None, also_known_as: None, outbox_url: None, followers_url: None, following_url: None, attachment: vec![] }))
}
}

7
crates/adapters/postgres/src/remote_actor_connections.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use domain::{
errors::DomainError, models::actor_connection_summary::ActorConnectionSummary,
@@ -46,7 +47,7 @@ impl RemoteActorConnectionRepository for PgRemoteActorConnectionRepository {
.bind(&actor.avatar_url)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
}
Ok(())
}
@@ -75,7 +76,7 @@ impl RemoteActorConnectionRepository for PgRemoteActorConnectionRepository {
.bind(page as i32)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(rows
.into_iter()
@@ -103,7 +104,7 @@ impl RemoteActorConnectionRepository for PgRemoteActorConnectionRepository {
.bind(page as i32)
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(row.and_then(|(ts,)| ts))
}

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

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use domain::{
errors::DomainError,
@@ -28,7 +29,7 @@ impl TagRepository for PgTagRepository {
.bind(&name)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
#[derive(sqlx::FromRow)]
struct Row {
id: i32,
@@ -38,7 +39,7 @@ impl TagRepository for PgTagRepository {
.bind(&name)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Tag {
id: row.id,
name: row.name,
@@ -57,7 +58,7 @@ impl TagRepository for PgTagRepository {
.bind(tag_id)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -66,7 +67,7 @@ impl TagRepository for PgTagRepository {
.bind(thought_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -79,7 +80,7 @@ impl TagRepository for PgTagRepository {
sqlx::query_as::<_, Row>(
"SELECT t.id,t.name FROM tags t JOIN thought_tags tt ON tt.tag_id=t.id WHERE tt.thought_id=$1"
).bind(thought_id.as_uuid()).fetch_all(&self.pool).await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|rows| rows.into_iter().map(|r| Tag { id: r.id, name: r.name }).collect())
}
@@ -94,14 +95,14 @@ impl TagRepository for PgTagRepository {
.bind(tag_name)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let rows = sqlx::query_as::<_, crate::thought::ThoughtRow>(
"SELECT th.id,th.user_id,th.content,th.in_reply_to_id,th.in_reply_to_url,th.ap_id,th.visibility,th.content_warning,th.sensitive,th.local,th.created_at,th.updated_at
FROM thoughts th JOIN thought_tags tt ON tt.thought_id=th.id JOIN tags t ON t.id=tt.tag_id
WHERE t.name=$1 ORDER BY th.created_at DESC LIMIT $2 OFFSET $3"
).bind(tag_name).bind(page.limit()).bind(page.offset())
.fetch_all(&self.pool).await.map_err(|e| DomainError::Internal(e.to_string()))?;
.fetch_all(&self.pool).await.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(Thought::from).collect(),
@@ -123,7 +124,7 @@ impl TagRepository for PgTagRepository {
.bind(limit as i64)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
}
}

37
crates/adapters/postgres/src/test_helpers.rs Normal file
View File

@@ -0,0 +1,37 @@
use crate::{thought::PgThoughtRepository, user::PgUserRepository};
use domain::{
models::{
thought::{Thought, Visibility},
user::User,
},
ports::{ThoughtRepository, UserRepository},
value_objects::{Content, Email, PasswordHash, ThoughtId, UserId, Username},
};
pub async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
let repo = PgUserRepository::new(pool.clone());
let u = User::new_local(
UserId::new(),
Username::new(username).unwrap(),
Email::new(email).unwrap(),
PasswordHash("h".into()),
);
repo.save(&u).await.unwrap();
u
}
pub async fn seed_user_and_thought(pool: &sqlx::PgPool) -> (User, Thought) {
let user = seed_user(pool, "alice", "alice@ex.com").await;
let trepo = PgThoughtRepository::new(pool.clone());
let t = Thought::new_local(
ThoughtId::new(),
user.id.clone(),
Content::new_local("hi").unwrap(),
None,
Visibility::Public,
None,
false,
);
trepo.save(&t).await.unwrap();
(user, t)
}

61
crates/adapters/postgres/src/thought.rs
View File

@@ -1,31 +1,12 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
fn visibility_from_str(s: &str) -> domain::models::thought::Visibility {
use domain::models::thought::Visibility;
match s {
"followers" => Visibility::Followers,
"unlisted" => Visibility::Unlisted,
"direct" => Visibility::Direct,
_ => Visibility::Public,
}
}
fn visibility_as_str(v: &domain::models::thought::Visibility) -> &'static str {
use domain::models::thought::Visibility;
match v {
Visibility::Public => "public",
Visibility::Followers => "followers",
Visibility::Unlisted => "unlisted",
Visibility::Direct => "direct",
}
}
use domain::{
errors::DomainError,
models::{
feed::{PageParams, Paginated},
thought::Thought,
thought::{Thought, Visibility},
},
ports::ThoughtRepository,
value_objects::{Content, ThoughtId, UserId},
@@ -66,7 +47,7 @@ impl From<ThoughtRow> for Thought {
in_reply_to_id: r.in_reply_to_id.map(ThoughtId::from_uuid),
in_reply_to_url: r.in_reply_to_url,
ap_id: r.ap_id,
visibility: visibility_from_str(&r.visibility),
visibility: Visibility::from_db_str(&r.visibility),
content_warning: r.content_warning,
sensitive: r.sensitive,
local: r.local,
@@ -93,14 +74,14 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(t.in_reply_to_id.as_ref().map(|x| x.as_uuid()))
.bind(&t.in_reply_to_url)
.bind(&t.ap_id)
.bind(visibility_as_str(&t.visibility))
.bind(t.visibility.as_str())
.bind(&t.content_warning)
.bind(t.sensitive)
.bind(t.local)
.bind(t.created_at)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -109,7 +90,7 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(id.as_uuid())
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(Thought::from))
}
@@ -119,7 +100,7 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(user_id.as_uuid())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
if r.rows_affected() == 0 {
return Err(DomainError::NotFound);
}
@@ -132,7 +113,7 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(content.as_str())
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -153,7 +134,7 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(id.as_uuid())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|rows| rows.into_iter().map(Thought::from).collect())
}
@@ -167,7 +148,7 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(uid)
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
let rows = sqlx::query_as::<_, ThoughtRow>(&format!(
"{THOUGHT_SELECT} WHERE user_id=$1 ORDER BY created_at DESC LIMIT $2 OFFSET $3"
@@ -177,7 +158,7 @@ impl ThoughtRepository for PgThoughtRepository {
.bind(page.offset())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(Paginated {
items: rows.into_iter().map(Thought::from).collect(),
@@ -191,28 +172,12 @@ impl ThoughtRepository for PgThoughtRepository {
#[cfg(test)]
mod tests {
use super::*;
use crate::user::PgUserRepository;
use domain::ports::UserRepository;
use crate::test_helpers::seed_user;
use domain::{
models::{
thought::{Thought, Visibility},
user::User,
},
models::thought::{Thought, Visibility},
value_objects::*,
};
async fn seed_user(pool: &sqlx::PgPool, username: &str, email: &str) -> User {
let repo = PgUserRepository::new(pool.clone());
let u = User::new_local(
UserId::new(),
Username::new(username).unwrap(),
Email::new(email).unwrap(),
PasswordHash("h".into()),
);
repo.save(&u).await.unwrap();
u
}
#[sqlx::test(migrations = "./migrations")]
async fn save_and_find_thought(pool: sqlx::PgPool) {
let user = seed_user(&pool, "alice", "alice@ex.com").await;

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

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use domain::{
errors::DomainError,
@@ -23,16 +24,12 @@ impl TopFriendRepository for PgTopFriendRepository {
user_id: &UserId,
friends: Vec<(UserId, i16)>,
) -> Result<(), DomainError> {
let mut tx = self
.pool
.begin()
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
let mut tx = self.pool.begin().await.into_domain()?;
sqlx::query("DELETE FROM top_friends WHERE user_id=$1")
.bind(user_id.as_uuid())
.execute(&mut *tx)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
for (friend_id, pos) in friends {
sqlx::query("INSERT INTO top_friends(user_id,friend_id,position) VALUES($1,$2,$3)")
.bind(user_id.as_uuid())
@@ -40,11 +37,9 @@ impl TopFriendRepository for PgTopFriendRepository {
.bind(pos)
.execute(&mut *tx)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
}
tx.commit()
.await
.map_err(|e| DomainError::Internal(e.to_string()))
tx.commit().await.into_domain()
}
async fn list_for_user(&self, user_id: &UserId) -> Result<Vec<(TopFriend, User)>, DomainError> {
@@ -79,7 +74,7 @@ impl TopFriendRepository for PgTopFriendRepository {
.bind(user_id.as_uuid())
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(rows
.into_iter()

19
crates/adapters/postgres/src/user.rs
View File

@@ -1,3 +1,4 @@
use crate::db_error::IntoDbResult;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use domain::{
@@ -18,7 +19,7 @@ impl PgUserRepository {
}
#[derive(sqlx::FromRow)]
pub(crate) struct UserRow {
pub struct UserRow {
pub id: uuid::Uuid,
pub username: String,
pub email: String,
@@ -56,7 +57,7 @@ impl From<UserRow> for User {
}
}
const USER_SELECT: &str = "SELECT id,username,email,password_hash,display_name,bio,avatar_url,header_url,custom_css,local,ap_id,inbox_url,created_at,updated_at FROM users";
pub const USER_SELECT: &str = "SELECT id,username,email,password_hash,display_name,bio,avatar_url,header_url,custom_css,local,ap_id,inbox_url,created_at,updated_at FROM users";
#[async_trait]
impl UserRepository for PgUserRepository {
@@ -65,7 +66,7 @@ impl UserRepository for PgUserRepository {
.bind(id.as_uuid())
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(User::from))
}
@@ -74,7 +75,7 @@ impl UserRepository for PgUserRepository {
.bind(username.as_str())
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(User::from))
}
@@ -83,7 +84,7 @@ impl UserRepository for PgUserRepository {
.bind(email.as_str())
.fetch_optional(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|o| o.map(User::from))
}
@@ -115,7 +116,7 @@ impl UserRepository for PgUserRepository {
.bind(user.updated_at)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -139,7 +140,7 @@ impl UserRepository for PgUserRepository {
.bind(custom_css)
.execute(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
.map(|_| ())
}
@@ -170,7 +171,7 @@ impl UserRepository for PgUserRepository {
)
.fetch_all(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))?;
.into_domain()?;
Ok(rows
.into_iter()
@@ -191,7 +192,7 @@ impl UserRepository for PgUserRepository {
sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users WHERE local = true")
.fetch_one(&self.pool)
.await
.map_err(|e| DomainError::Internal(e.to_string()))
.into_domain()
}
}