015f300941651035945b5cdd98aeada97e4337aa / frogue / src / activitypub / forge / inbound.rs
Raw
//! ForgeFed inbound activity handlers for repository inboxes.
//!
//! Handles Offer (ticket creation), Resolve (ticket closure), Edit (ticket updates),
//! Create (comments), and Add (patch sets) activities from remote actors.
use actix_web::HttpResponse;
use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter};
use tracing::{debug, error, info, instrument, warn};
use crate::activitypub::types::{
AddActivity, CreateActivity, CreateObject, EditActivity, ObjectRef, OfferActivity, OfferObject,
ResolveActivity,
};
use crate::entities::prelude::{ApRemoteActors, Tickets};
use crate::entities::{ap_remote_actors, tickets as tickets_entity};
/// Handle an incoming Offer activity (ticket creation).
///
/// Creates a local ticket from the embedded Ticket object and sends Accept.
/// Per decision 7, Offer with inline Patch is rejected.
#[instrument(skip(db, http_client, offer))]
pub async fn handle_offer(
db: &DatabaseConnection,
http_client: &reqwest_middleware::ClientWithMiddleware,
public_url: &str,
repo_actor: &crate::entities::ap_actors::Model,
verified_actor_url: &str,
offer: &OfferActivity,
) -> HttpResponse {
let repo_id = match repo_actor.repo_id {
Some(id) => id,
None => {
debug!("offer handler called with non-repo actor");
return HttpResponse::Accepted().finish();
}
};
// Resolve the remote actor (check cache first, then fetch if needed)
let remote_actor = match ApRemoteActors::find()
.filter(ap_remote_actors::Column::ActorUrl.eq(verified_actor_url))
.one(db)
.await
{
Ok(Some(actor)) => actor,
Ok(None) => {
// Not in cache, try to fetch
match crate::activitypub::remote::resolve_actor(db, http_client, verified_actor_url)
.await
{
Ok(actor) => actor,
Err(e) => {
error!(error = %e, actor_url = %verified_actor_url, "failed to resolve remote actor");
return HttpResponse::Accepted().finish();
}
}
}
Err(e) => {
error!(error = %e, "failed to look up remote actor in cache");
return HttpResponse::InternalServerError().finish();
}
};
// Check if this is a duplicate offer
let existing = Tickets::find()
.filter(tickets_entity::Column::RemoteOfferId.eq(&offer.id))
.one(db)
.await;
match existing {
Ok(Some(_)) => {
debug!(offer_id = %offer.id, "duplicate offer, ignoring");
return HttpResponse::Accepted().finish();
}
Ok(None) => {}
Err(e) => {
error!(error = %e, "failed to check for duplicate offer");
return HttpResponse::InternalServerError().finish();
}
}
// Extract the Ticket object
let ticket_embed = match &offer.object {
OfferObject::Url(url) => {
// TODO: fetch via signed_get - for now, reject
warn!(url = %url, "URL-only offers not yet supported");
return HttpResponse::Accepted().finish();
}
OfferObject::Embedded(embed) => embed,
};
// Check if this is a Patch (MR) - reject per decision 7
if ticket_embed.kind == "Patch" {
info!("rejecting offer with inline patch (decision 7)");
// Get repo actor URL for the Reject activity
let urls = crate::activitypub::ApUrls::new(public_url);
let repo = match crate::entities::prelude::Repositories::find_by_id(repo_id)
.one(db)
.await
{
Ok(Some(r)) => r,
Ok(None) => {
error!("repository not found");
return HttpResponse::InternalServerError().finish();
}
Err(e) => {
error!(error = %e, "failed to look up repository");
return HttpResponse::InternalServerError().finish();
}
};
let owner = match crate::entities::prelude::Users::find_by_id(repo.owner_id)
.one(db)
.await
{
Ok(Some(u)) => u,
Ok(None) => {
error!("repository owner not found");
return HttpResponse::InternalServerError().finish();
}
Err(e) => {
error!(error = %e, "failed to look up repository owner");
return HttpResponse::InternalServerError().finish();
}
};
let repo_actor_url = urls.repo_actor(&owner.name, &repo.slug);
// Send Reject activity
if let Err(e) = crate::activitypub::forge::activity::send_reject(
db,
repo_actor.id,
&repo_actor_url,
&offer.id,
&remote_actor.inbox_url,
)
.await
{
error!(error = %e, "failed to send Reject for offer");
}
return HttpResponse::Accepted().finish();
}
// Extract ticket fields
let summary = ticket_embed
.summary
.clone()
.unwrap_or_else(|| "Untitled".to_string());
let description = ticket_embed.content.clone();
let offer_id = offer.id.clone();
// Allocate ticket number
let next_number = match crate::tickets::store::take_ticket_number(db, repo_id).await {
Ok(n) => n,
Err(e) => {
error!(error = %e, "failed to allocate ticket number");
return HttpResponse::InternalServerError().finish();
}
};
// Create the ticket
match crate::tickets::store::create_ticket_from_remote(
db,
repo_id,
remote_actor.id,
next_number,
&summary,
description.as_deref(),
crate::tickets::State::Open,
&offer_id,
)
.await
{
Ok(ticket) => {
info!(
ticket_id = ticket.id,
number = ticket.number,
remote_actor = %verified_actor_url,
"created ticket from remote offer"
);
// Look up repository and owner to construct ticket URL
let repo = match crate::entities::prelude::Repositories::find_by_id(repo_id)
.one(db)
.await
{
Ok(Some(r)) => r,
Ok(None) => {
error!("repository not found");
return HttpResponse::InternalServerError().finish();
}
Err(e) => {
error!(error = %e, "failed to look up repository");
return HttpResponse::InternalServerError().finish();
}
};
let owner = match crate::entities::prelude::Users::find_by_id(repo.owner_id)
.one(db)
.await
{
Ok(Some(u)) => u,
Ok(None) => {
error!("repository owner not found");
return HttpResponse::InternalServerError().finish();
}
Err(e) => {
error!(error = %e, "failed to look up repository owner");
return HttpResponse::InternalServerError().finish();
}
};
// Construct ticket URL using the ApUrls helper
let urls = crate::activitypub::ApUrls::new(public_url);
let ticket_url = urls.repo_ticket(&owner.name, &repo.slug, ticket.number);
let repo_actor_url = urls.repo_actor(&owner.name, &repo.slug);
// Send Accept activity to the offerer
if let Err(e) = crate::activitypub::forge::activity::announce_accept(
db,
repo_actor.id,
&repo_actor_url,
&offer.id,
&ticket_url,
Some(&remote_actor.inbox_url),
)
.await
{
error!(error = %e, "failed to send Accept for offer");
}
HttpResponse::Accepted().finish()
}
Err(e) => {
error!(error = %e, "failed to create ticket from offer");
HttpResponse::InternalServerError().finish()
}
}
}
/// Handle an incoming Resolve activity (ticket closure).
///
/// Closes the referenced ticket if the remote actor is authorized.
#[instrument(skip(db, http_client, resolve))]
pub async fn handle_resolve(
db: &DatabaseConnection,
http_client: &reqwest_middleware::ClientWithMiddleware,
public_url: &str,
repo_actor: &crate::entities::ap_actors::Model,
verified_actor_url: &str,
resolve: &ResolveActivity,
) -> HttpResponse {
let repo_id = match repo_actor.repo_id {
Some(id) => id,
None => {
debug!("resolve handler called with non-repo actor");
return HttpResponse::Accepted().finish();
}
};
// Resolve the remote actor
let remote_actor = match ApRemoteActors::find()
.filter(ap_remote_actors::Column::ActorUrl.eq(verified_actor_url))
.one(db)
.await
{
Ok(Some(actor)) => actor,
Ok(None) => {
// Not in cache, try to fetch
match crate::activitypub::remote::resolve_actor(db, http_client, verified_actor_url)
.await
{
Ok(actor) => actor,
Err(e) => {
error!(error = %e, actor_url = %verified_actor_url, "failed to resolve remote actor");
return HttpResponse::Accepted().finish();
}
}
}
Err(e) => {
error!(error = %e, "failed to look up remote actor in cache");
return HttpResponse::InternalServerError().finish();
}
};
// Extract ticket URL from the object
let ticket_url = resolve.object.id();
// Parse ticket URL to extract number
// Expected format: {base}/~{owner}/{repo}/tickets/{number}
let ticket_number = match parse_ticket_url(ticket_url, public_url, repo_id) {
Some(n) => n,
None => {
debug!(url = %ticket_url, "could not parse ticket URL");
return HttpResponse::Accepted().finish();
}
};
// Find the ticket
let ticket = match Tickets::find()
.filter(tickets_entity::Column::RepositoryId.eq(repo_id))
.filter(tickets_entity::Column::Number.eq(ticket_number))
.one(db)
.await
{
Ok(Some(t)) => t,
Ok(None) => {
debug!(number = ticket_number, "ticket not found");
return HttpResponse::Accepted().finish();
}
Err(e) => {
error!(error = %e, "failed to look up ticket");
return HttpResponse::InternalServerError().finish();
}
};
// Authorize: only the remote creator can resolve
if ticket.remote_actor_id != Some(remote_actor.id) {
warn!(
ticket_id = ticket.id,
remote_actor_id = remote_actor.id,
creator_id = ?ticket.remote_actor_id,
"unauthorized resolve attempt"
);
return HttpResponse::Accepted().finish();
}
// Update ticket state to Closed
match crate::tickets::store::update_ticket(
db,
ticket.id,
&ticket.summary,
ticket.description.as_deref(),
crate::tickets::State::Closed,
Some(resolve.id.clone()),
)
.await
{
Ok(_) => {
info!(
ticket_id = ticket.id,
number = ticket.number,
"ticket closed via remote resolve"
);
// TODO: send Accept activity
HttpResponse::Accepted().finish()
}
Err(e) => {
error!(error = %e, "failed to close ticket");
HttpResponse::InternalServerError().finish()
}
}
}
/// Handle an incoming Edit activity (ticket update).
///
/// Updates the referenced ticket if the remote actor is authorized.
#[instrument(skip(db, http_client, edit))]
pub async fn handle_edit(
db: &DatabaseConnection,
http_client: &reqwest_middleware::ClientWithMiddleware,
public_url: &str,
repo_actor: &crate::entities::ap_actors::Model,
verified_actor_url: &str,
edit: &EditActivity,
) -> HttpResponse {
let repo_id = match repo_actor.repo_id {
Some(id) => id,
None => {
debug!("edit handler called with non-repo actor");
return HttpResponse::Accepted().finish();
}
};
// Resolve the remote actor
let remote_actor = match ApRemoteActors::find()
.filter(ap_remote_actors::Column::ActorUrl.eq(verified_actor_url))
.one(db)
.await
{
Ok(Some(actor)) => actor,
Ok(None) => {
// Not in cache, try to fetch
match crate::activitypub::remote::resolve_actor(db, http_client, verified_actor_url)
.await
{
Ok(actor) => actor,
Err(e) => {
error!(error = %e, actor_url = %verified_actor_url, "failed to resolve remote actor");
return HttpResponse::Accepted().finish();
}
}
}
Err(e) => {
error!(error = %e, "failed to look up remote actor in cache");
return HttpResponse::InternalServerError().finish();
}
};
// Extract ticket URL from the object
let ticket_url = edit.object.id();
// Parse ticket URL to extract number
let ticket_number = match parse_ticket_url(ticket_url, public_url, repo_id) {
Some(n) => n,
None => {
debug!(url = %ticket_url, "could not parse ticket URL");
return HttpResponse::Accepted().finish();
}
};
// Find the ticket
let ticket = match Tickets::find()
.filter(tickets_entity::Column::RepositoryId.eq(repo_id))
.filter(tickets_entity::Column::Number.eq(ticket_number))
.one(db)
.await
{
Ok(Some(t)) => t,
Ok(None) => {
debug!(number = ticket_number, "ticket not found");
return HttpResponse::Accepted().finish();
}
Err(e) => {
error!(error = %e, "failed to look up ticket");
return HttpResponse::InternalServerError().finish();
}
};
// Authorize: only the remote creator can edit
if ticket.remote_actor_id != Some(remote_actor.id) {
warn!(
ticket_id = ticket.id,
remote_actor_id = remote_actor.id,
creator_id = ?ticket.remote_actor_id,
"unauthorized edit attempt"
);
return HttpResponse::Accepted().finish();
}
// Extract updated fields from the embedded object
let (new_summary, new_description) = match &edit.object {
ObjectRef::Embedded(embedded) => {
let summary = embedded
.summary
.clone()
.unwrap_or_else(|| ticket.summary.clone());
let description = embedded
.content
.clone()
.or_else(|| ticket.description.clone());
(summary, description)
}
ObjectRef::Url(_) => {
// URL-only edit, keep current values
(ticket.summary.clone(), ticket.description.clone())
}
};
// Update the ticket
match crate::tickets::store::update_ticket(
db,
ticket.id,
&new_summary,
new_description.as_deref(),
ticket.state,
ticket.resolve_activity_url.clone(),
)
.await
{
Ok(_) => {
info!(
ticket_id = ticket.id,
number = ticket.number,
"ticket updated via remote edit"
);
// TODO: send Accept activity
HttpResponse::Accepted().finish()
}
Err(e) => {
error!(error = %e, "failed to update ticket");
HttpResponse::InternalServerError().finish()
}
}
}
/// Handle an incoming Create activity for comments.
///
/// Creates a comment on a ticket if the inReplyTo references a local ticket.
#[instrument(skip(db, http_client, create))]
pub async fn handle_create(
db: &DatabaseConnection,
http_client: &reqwest_middleware::ClientWithMiddleware,
public_url: &str,
repo_actor: &crate::entities::ap_actors::Model,
verified_actor_url: &str,
create: &CreateActivity,
) -> HttpResponse {
let repo_id = match repo_actor.repo_id {
Some(id) => id,
None => {
// User actors handle Create differently (for notes)
debug!("create handler called with non-repo actor, delegating");
return HttpResponse::Accepted().finish();
}
};
// Resolve the remote actor
let remote_actor = match ApRemoteActors::find()
.filter(ap_remote_actors::Column::ActorUrl.eq(verified_actor_url))
.one(db)
.await
{
Ok(Some(actor)) => actor,
Ok(None) => {
// Not in cache, try to fetch
match crate::activitypub::remote::resolve_actor(db, http_client, verified_actor_url)
.await
{
Ok(actor) => actor,
Err(e) => {
error!(error = %e, actor_url = %verified_actor_url, "failed to resolve remote actor");
return HttpResponse::Accepted().finish();
}
}
}
Err(e) => {
error!(error = %e, "failed to look up remote actor in cache");
return HttpResponse::InternalServerError().finish();
}
};
// Extract the Note object
let note = match &create.object {
CreateObject::Embedded(embed) => embed,
CreateObject::Url(url) => {
debug!(url = %url, "URL-only creates not yet supported");
return HttpResponse::Accepted().finish();
}
};
// Check if it's a Note
if note.kind != "Note" {
debug!(kind = %note.kind, "create for non-Note object");
return HttpResponse::Accepted().finish();
}
// Check if it's in reply to a ticket
let in_reply_to = match ¬e.in_reply_to {
Some(url) => url,
None => {
debug!("note without inReplyTo, ignoring");
return HttpResponse::Accepted().finish();
}
};
// Parse ticket URL to extract number
let ticket_number = match parse_ticket_url(in_reply_to, public_url, repo_id) {
Some(n) => n,
None => {
debug!(url = %in_reply_to, "could not parse ticket URL");
return HttpResponse::Accepted().finish();
}
};
// Find the ticket
let ticket = match Tickets::find()
.filter(tickets_entity::Column::RepositoryId.eq(repo_id))
.filter(tickets_entity::Column::Number.eq(ticket_number))
.one(db)
.await
{
Ok(Some(t)) => t,
Ok(None) => {
debug!(number = ticket_number, "ticket not found");
return HttpResponse::Accepted().finish();
}
Err(e) => {
error!(error = %e, "failed to look up ticket");
return HttpResponse::InternalServerError().finish();
}
};
// Extract comment content
let content = note.content.clone().unwrap_or_default();
let object_id = note.id.clone();
// Determine content format from source if available
let content_format = note
.source
.as_ref()
.map(|s| s.media_type.as_str())
.unwrap_or("text/markdown");
// Create the comment
match crate::comments::store::create_on_ticket_from_remote(
db,
remote_actor.id,
ticket.id,
&content,
content_format,
&object_id,
)
.await
{
Ok(comment) => {
info!(
comment_id = comment.id,
ticket_id = ticket.id,
"created comment from remote create"
);
// TODO: send Accept activity and relay to followers
HttpResponse::Accepted().finish()
}
Err(e) => {
error!(error = %e, "failed to create comment");
HttpResponse::InternalServerError().finish()
}
}
}
/// Handle an incoming Add activity (patch set update).
///
/// Per decision 7, inline patches are rejected. Origin-only patches are also rejected.
#[instrument(skip(db, add))]
pub async fn handle_add(
db: &DatabaseConnection,
public_url: &str,
repo_actor: &crate::entities::ap_actors::Model,
verified_actor_url: &str,
add: &AddActivity,
) -> HttpResponse {
let _repo_id = match repo_actor.repo_id {
Some(id) => id,
None => {
debug!("add handler called with non-repo actor");
return HttpResponse::Accepted().finish();
}
};
// Resolve the remote actor
let _remote_actor = match ApRemoteActors::find()
.filter(ap_remote_actors::Column::ActorUrl.eq(verified_actor_url))
.one(db)
.await
{
Ok(Some(actor)) => actor,
Ok(None) => {
warn!(actor_url = %verified_actor_url, "add from unknown remote actor");
return HttpResponse::Accepted().finish();
}
Err(e) => {
error!(error = %e, "failed to look up remote actor");
return HttpResponse::InternalServerError().finish();
}
};
// Extract target MR URL
let target_mr_url = match &add.target {
Some(target) => target.id(),
None => {
debug!("add activity without target, ignoring");
return HttpResponse::Accepted().finish();
}
};
// TODO: Parse MR URL and look up the patch proposal
// For now, reject all Add activities per decision 7
info!(
target_mr_url = %target_mr_url,
remote_actor = %verified_actor_url,
"rejecting add activity (decision 7: inline patches not supported yet)"
);
// TODO: send Reject activity with explanation
// Reject{object: add.id, summary: "inline patch merge requests are not supported yet"}
HttpResponse::Accepted().finish()
}
/// Parse a ticket URL to extract the ticket number.
///
/// Expected format: {base}/~{owner}/{repo}/tickets/{number}
fn parse_ticket_url(url: &str, public_url: &str, _repo_id: i64) -> Option<i64> {
let base = public_url.trim_end_matches('/');
if !url.starts_with(base) {
return None;
}
let suffix = &url[base.len()..];
// Expected: /~{owner}/{repo}/tickets/{number}
// Split gives: ["~alice", "testrepo", "tickets", "1"]
let parts: Vec<&str> = suffix.trim_start_matches('/').split('/').collect();
if parts.len() < 4 || !parts[0].starts_with('~') || parts[2] != "tickets" {
return None;
}
parts[3].parse().ok()
}