feat: transaction for follow

6 files changed, 47 insertions, 9 deletions

diff --git a/contrib/bruno/environments/sellershut.bru b/contrib/bruno/environments/sellershut.bru
index 82d0e7c..bb9a8b7 100644
--- a/contrib/bruno/environments/sellershut.bru
+++ b/contrib/bruno/environments/sellershut.bru
@@ -1,3 +1,5 @@
 vars {
-  HUT_HOSTNAME: http://localhost:2210
+  HUT_DOMAIN: localhost
+  HUT_PORT: 2210
+  HUT_HOSTNAME: http://{{HUT_DOMAIN}}:{{HUT_PORT}}
 }
diff --git a/contrib/bruno/users/webfinger.bru b/contrib/bruno/users/webfinger.bru
index ea238f9..a6d15bf 100644
--- a/contrib/bruno/users/webfinger.bru
+++ b/contrib/bruno/users/webfinger.bru
@@ -5,11 +5,11 @@ meta {
 }
 
 get {
-  url: {{HUT_HOSTNAME}}/.well-known/webfinger?resource=acct:sellershut@localhost
+  url: {{HUT_HOSTNAME}}/.well-known/webfinger?resource=acct:sellershut@{{HUT_DOMAIN}}
   body: none
   auth: inherit
 }
 
 params:query {
-  resource: acct:sellershut@localhost
+  resource: acct:sellershut@{{HUT_DOMAIN}}
 }
diff --git a/crates/sellershut/migrations/20250719223814_activity.sql b/crates/sellershut/migrations/20250719223814_activity.sql
new file mode 100644
index 0000000..107a21b
--- /dev/null
+++ b/crates/sellershut/migrations/20250719223814_activity.sql
@@ -0,0 +1,12 @@
+create table activity (
+    ap_id text references account(ap_id),
+    activity jsonb not null,
+    activity_id text generated always as (
+        activity->>'id'
+    ) stored,
+    constraint unique_activity_id unique (activity_id),
+    created_at timestamptz not null default now()
+);
+
+create index idx_activity_owner_id on activity(ap_id);
+create index idx_activity_created on activity(created_at);
diff --git a/crates/sellershut/src/entity/user.rs b/crates/sellershut/src/entity/user.rs
index b420682..2d09acc 100644
--- a/crates/sellershut/src/entity/user.rs
+++ b/crates/sellershut/src/entity/user.rs
@@ -14,8 +14,8 @@ use activitypub_federation::{
 };
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
-use sqlx::types::time::OffsetDateTime;
 use stack_up::{Environment, Services};
+use time::OffsetDateTime;
 use tracing::trace;
 use url::Url;
 use uuid::Uuid;
@@ -180,6 +180,7 @@ impl User {
         A: Activity + Serialize + std::fmt::Debug + Send + Sync,
         <A as Activity>::Error: From<anyhow::Error> + From<serde_json::Error>,
     {
+        trace!("sending accept activity");
         let activity = WithContext::new_default(activity);
         // Send through queue in some cases and bypass it in others to test both code paths
         if use_queue {
diff --git a/crates/sellershut/src/main.rs b/crates/sellershut/src/main.rs
index f0540bf..b51c9f0 100644
--- a/crates/sellershut/src/main.rs
+++ b/crates/sellershut/src/main.rs
@@ -10,7 +10,7 @@ use clap::Parser;
 use stack_up::{Configuration, Services, tracing::Tracing};
 
 use crate::{error::AppError, state::AppState};
-use tracing::{error, info};
+use tracing::{error, info, trace};
 
 /// sellershut
 #[derive(Parser, Debug)]
@@ -45,6 +45,11 @@ async fn main() -> Result<(), AppError> {
         .inspect_err(|e| error!("database: {e}"))?
         .build();
 
+    trace!("running migrations");
+    sqlx::migrate!("./migrations")
+        .run(&services.postgres)
+        .await?;
+
     let state = AppState::create(services, &config).await?;
 
     let addr = SocketAddr::from((Ipv6Addr::UNSPECIFIED, config.application.port));
diff --git a/crates/sellershut/src/server/activities/follow.rs b/crates/sellershut/src/server/activities/follow.rs
index e39b81e..48d9caa 100644
--- a/crates/sellershut/src/server/activities/follow.rs
+++ b/crates/sellershut/src/server/activities/follow.rs
@@ -6,6 +6,7 @@ use activitypub_federation::{
 };
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
+use tracing::trace;
 use url::Url;
 use uuid::Uuid;
 
@@ -71,21 +72,38 @@ impl Activity for Follow {
     async fn receive(self, data: &Data<Self::DataType>) -> Result<(), Self::Error> {
         let id = Uuid::now_v7();
 
-        sqlx::query!("insert into following (id, follower, followee) values ($1, $2, $3) on conflict (follower, followee) do nothing"
-            ,id,
+        let mut transaction = data.services.postgres.begin().await?;
+
+        trace!("adding a follower");
+        sqlx::query!(
+            "insert into following (id, follower, followee) values ($1, $2, $3)",
+            id,
             self.actor.inner().as_str(),
             self.object.inner().as_str(),
-            ).execute(&data.services.postgres).await?;
+        )
+        .execute(&mut *transaction)
+        .await?;
 
         let follower = self.actor.dereference(data).await?;
         let id = generate_object_id(data.domain(), data.environment)?;
 
         let local_user = self.object.dereference(data).await?;
-        let accept = Accept::new(self.object.clone(), self, id.clone());
+        let accept = Accept::new(self.object.clone(), self.clone(), id.clone());
 
         local_user
             .send(accept, vec![follower.shared_inbox_or_inbox()], false, data)
             .await?;
+
+        sqlx::query!(
+            "insert into activity (ap_id, activity) values ($1, $2)",
+            self.actor.inner().as_str(),
+            sqlx::types::Json(&self) as _
+        )
+        .execute(&mut *transaction)
+        .await?;
+
+        transaction.commit().await?;
+
         Ok(())
     }
 }