//! Activity emission and delivery for ForgeFed objects.
use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set};
use tracing::instrument;
use uuid::Uuid;
use crate::activitypub::types::{Accept, FORGEFED_CONTEXT};
use crate::entities::prelude::ApRemoteActors;
use crate::entities::{ap_activities, ap_deliveries, ap_followers};
/// Publish an activity to the outbox and deliver it to all followers.
///
/// This creates a single activity row and fans out deliveries to all followers,
/// deduplicating by shared inbox URL to avoid duplicate deliveries.
///
/// # Arguments
/// * `db` - Database connection
/// * `actor_id` - The actor (user or repository) publishing the activity
/// * `activity_json` - The serialized activity JSON
/// * `object_url` - The URL of the object being acted upon
///
/// # Returns
/// The ID of the created activity
#[instrument(skip(db, activity_json), fields(actor_id = %actor_id))]
pub async fn publish_activity(
db: &DatabaseConnection,
actor_id: i64,
activity_json: &str,
object_url: &str,
) -> Result<i64, sea_orm::DbErr> {
let activity_uuid = Uuid::new_v4();
// Insert the activity into the outbox
let activity = ap_activities::ActiveModel {
uuid: Set(activity_uuid),
actor_id: Set(actor_id),
activity_json: Set(activity_json.to_string()),
object_url: Set(object_url.to_string()),
..Default::default()
};
let activity = activity.insert(db).await?;
// Get all followers for this actor
let followers = ap_followers::Entity::find()
.filter(ap_followers::Column::LocalActorId.eq(actor_id))
.all(db)
.await?;
// Collect unique inbox URLs, preferring shared inbox when available
let mut inbox_urls = std::collections::HashSet::new();
for follower in followers {
let remote_actor = ApRemoteActors::find_by_id(follower.remote_actor_id)
.one(db)
.await?;
if let Some(remote_actor) = remote_actor {
// Use shared inbox if available, otherwise use individual inbox
let inbox_url = remote_actor
.shared_inbox_url
.unwrap_or(remote_actor.inbox_url);
inbox_urls.insert(inbox_url);
}
}
// Create a delivery row for each unique inbox
for inbox_url in inbox_urls {
let delivery = ap_deliveries::ActiveModel {
activity_id: Set(activity.id),
signer_actor_id: Set(actor_id),
inbox_url: Set(inbox_url),
status: Set("pending".to_string()),
attempts: Set(0),
next_attempt_at: Set(chrono::Utc::now().into()),
last_error: Set(None),
..Default::default()
};
delivery.insert(db).await?;
}
Ok(activity.id)
}
/// Publish an activity signed by one actor but delivered to another actor's followers.
///
/// This is used for comments: the comment is signed by the user actor but delivered
/// to the repository's followers (since the user may have zero followers while the
/// repo has many).
///
/// # Arguments
/// * `db` - Database connection
/// * `signer_actor_id` - The actor who signs the activity (user actor for comments)
/// * `follower_source_actor_id` - The actor whose followers receive the delivery (repo actor)
/// * `activity_json` - The serialized activity JSON
/// * `object_url` - The URL of the object being acted upon
///
/// # Returns
/// The ID of the created activity
#[instrument(skip(db, activity_json), fields(signer = %signer_actor_id, followers_of = %follower_source_actor_id))]
pub async fn publish_activity_with_separate_signer(
db: &DatabaseConnection,
signer_actor_id: i64,
follower_source_actor_id: i64,
activity_json: &str,
object_url: &str,
) -> Result<i64, sea_orm::DbErr> {
let activity_uuid = Uuid::new_v4();
// Insert the activity into the outbox (attributed to the signer)
let activity = ap_activities::ActiveModel {
uuid: Set(activity_uuid),
actor_id: Set(signer_actor_id),
activity_json: Set(activity_json.to_string()),
object_url: Set(object_url.to_string()),
..Default::default()
};
let activity = activity.insert(db).await?;
// Get all followers for the follower source actor (e.g., repo followers)
let followers = ap_followers::Entity::find()
.filter(ap_followers::Column::LocalActorId.eq(follower_source_actor_id))
.all(db)
.await?;
// Collect unique inbox URLs, preferring shared inbox when available
let mut inbox_urls = std::collections::HashSet::new();
for follower in followers {
let remote_actor = ApRemoteActors::find_by_id(follower.remote_actor_id)
.one(db)
.await?;
if let Some(remote_actor) = remote_actor {
// Use shared inbox if available, otherwise use individual inbox
let inbox_url = remote_actor
.shared_inbox_url
.unwrap_or(remote_actor.inbox_url);
inbox_urls.insert(inbox_url);
}
}
// Create a delivery row for each unique inbox (signed by the user actor)
for inbox_url in inbox_urls {
let delivery = ap_deliveries::ActiveModel {
activity_id: Set(activity.id),
signer_actor_id: Set(signer_actor_id),
inbox_url: Set(inbox_url),
status: Set("pending".to_string()),
attempts: Set(0),
next_attempt_at: Set(chrono::Utc::now().into()),
last_error: Set(None),
..Default::default()
};
delivery.insert(db).await?;
}
Ok(activity.id)
}
/// Announce acceptance of an Offer (for both local and remote Offers).
///
/// This builds an Accept activity, records it in the repository's outbox,
/// and fans it out to all followers. If offerer_inbox is provided (remote Offer),
/// also enqueues delivery to the offerer.
///
/// # Arguments
/// * `db` - Database connection
/// * `repo_actor_id` - The repository actor's ID
/// * `repo_actor_url` - The repository actor's URL
/// * `offer_id` - The ID of the Offer being accepted
/// * `result_url` - The URL of the created ticket/patch
/// * `offerer_inbox` - Optional inbox URL for the remote offerer
#[instrument(skip(db), fields(repo_actor_id = %repo_actor_id, offer_id = %offer_id))]
pub async fn announce_accept(
db: &DatabaseConnection,
repo_actor_id: i64,
repo_actor_url: &str,
offer_id: &str,
result_url: &str,
offerer_inbox: Option<&str>,
) -> Result<i64, String> {
let activity_id_url = format!("{}/activities/{}", repo_actor_url, Uuid::new_v4());
let accept = Accept {
context: FORGEFED_CONTEXT.to_vec(),
id: activity_id_url.clone(),
kind: "Accept",
actor: repo_actor_url.to_string(),
object: serde_json::Value::String(offer_id.to_string()),
result: Some(result_url.to_string()),
to: None,
cc: None,
published: Some(chrono::Utc::now().to_rfc3339()),
};
let activity_json =
serde_json::to_string(&accept).map_err(|e| format!("Failed to serialize Accept: {}", e))?;
// Publish to repository outbox and fan out to followers
let activity_db_id = publish_activity(db, repo_actor_id, &activity_json, result_url)
.await
.map_err(|e| format!("Failed to publish Accept: {}", e))?;
// If this is a remote Offer, also deliver to the offerer
if let Some(inbox_url) = offerer_inbox {
let delivery = ap_deliveries::ActiveModel {
activity_id: Set(activity_db_id),
signer_actor_id: Set(repo_actor_id),
inbox_url: Set(inbox_url.to_string()),
status: Set("pending".to_string()),
attempts: Set(0),
next_attempt_at: Set(chrono::Utc::now().into()),
last_error: Set(None),
..Default::default()
};
delivery
.insert(db)
.await
.map_err(|e| format!("Failed to enqueue delivery to offerer: {}", e))?;
}
Ok(activity_db_id)
}
/// Send a Reject activity for an Offer.
///
/// Used to reject unsupported offers (e.g., inline patches).
#[instrument(skip(db), fields(repo_actor_id = %repo_actor_id, offer_id = %offer_id))]
pub async fn send_reject(
db: &DatabaseConnection,
repo_actor_id: i64,
repo_actor_url: &str,
offer_id: &str,
offerer_inbox: &str,
) -> Result<i64, String> {
let activity_id_url = format!("{}/activities/{}", repo_actor_url, Uuid::new_v4());
let reject = serde_json::json!({
"@context": FORGEFED_CONTEXT,
"id": activity_id_url,
"type": "Reject",
"actor": repo_actor_url,
"object": offer_id,
"published": chrono::Utc::now().to_rfc3339(),
});
let activity_json =
serde_json::to_string(&reject).map_err(|e| format!("Failed to serialize Reject: {}", e))?;
// Publish to repository outbox
let activity_db_id = publish_activity(db, repo_actor_id, &activity_json, offer_id)
.await
.map_err(|e| format!("Failed to publish Reject: {}", e))?;
// Deliver to the offerer
let delivery = ap_deliveries::ActiveModel {
activity_id: Set(activity_db_id),
signer_actor_id: Set(repo_actor_id),
inbox_url: Set(offerer_inbox.to_string()),
status: Set("pending".to_string()),
attempts: Set(0),
next_attempt_at: Set(chrono::Utc::now().into()),
last_error: Set(None),
..Default::default()
};
delivery
.insert(db)
.await
.map_err(|e| format!("Failed to enqueue delivery to offerer: {}", e))?;
Ok(activity_db_id)
}