defmodule GitGud.Federation.Workers.ProcessInbox do
@moduledoc """
Dispatches an accepted inbound activity by type.
Supported types (more to come):
* `Follow` → record follow, fire back `Accept`
* `Undo` of `Follow` → drop the follow
* `Accept` → mark a Follow we sent as accepted
* `Like`, `Create`, … → recorded as an Activity row; no further
action yet (UI surfacing comes later)
Unknown / unsupported types are stored as activities and acked as
`noop` on the delivery so the audit trail is complete.
"""
use Oban.Worker, queue: :default, max_attempts: 5
import Ecto.Query, warn: false
alias GitGud.Federation
alias GitGud.Federation.Activity
alias GitGud.Federation.Actor
alias GitGud.Federation.Follow
alias GitGud.Federation.InboxDelivery
alias GitGud.Federation.RemoteActors
alias GitGud.Federation.Workers.DeliverOutbound
alias GitGud.Repo
def new_job(delivery_id) do
__MODULE__.new(%{"delivery_id" => delivery_id})
end
@impl Oban.Worker
def perform(%Oban.Job{args: %{"delivery_id" => id}}) do
case Repo.get(InboxDelivery, id) do
nil ->
:ok
%InboxDelivery{status: "accepted"} = delivery ->
_ = process(delivery)
:ok
_other ->
# Not in an actionable state — already marked dropped/rejected.
:ok
end
end
defp process(%InboxDelivery{raw_body: body} = delivery) do
case Jason.decode(body) do
{:ok, %{"type" => type} = activity_doc} ->
target_actor = Repo.get!(Actor, delivery.target_actor_id)
with {:ok, source_actor} <- RemoteActors.fetch(delivery.source_actor_url),
:ok <- gate_by_policy(type, target_actor, source_actor, delivery),
{:ok, _} <- record_activity(source_actor, activity_doc) do
dispatch(type, activity_doc, source_actor, target_actor)
mark_processed(delivery)
else
{:gated, status, reason} ->
mark_gated(delivery, status, reason)
_ ->
mark_processed(delivery)
end
_ ->
mark_processed(delivery)
end
end
# Repo-targeted Create activities (issues, PRs, comments) are subject
# to the repo's `interaction_policy`. Follows / Likes / Undo etc. are
# not gated here — Follow is consent-only, others either don't have
# write effects or are gated elsewhere.
defp gate_by_policy("Create", target_actor, source_actor, _delivery) do
case repo_for_actor(target_actor) do
%GitGud.Repositories.Repository{interaction_policy: "public"} ->
:ok
%GitGud.Repositories.Repository{interaction_policy: "followers"} = repo ->
if follower?(source_actor, target_actor) do
:ok
else
{:gated, "rejected",
"policy_followers: source is not an accepted follower of #{repo.name}"}
end
%GitGud.Repositories.Repository{interaction_policy: "approved"} ->
if follower?(source_actor, target_actor) do
:ok
else
{:gated, "quarantined", "policy_approved"}
end
_ ->
# Not a repo target — Person or Group inboxes don't have a
# write-side policy yet.
:ok
end
end
defp gate_by_policy(_type, _target, _source, _delivery), do: :ok
defp repo_for_actor(%Actor{local_repository_id: rid}) when not is_nil(rid),
do: Repo.get(GitGud.Repositories.Repository, rid)
defp repo_for_actor(_), do: nil
defp follower?(%Actor{id: src_id}, %Actor{id: tgt_id}) do
Repo.exists?(
from f in Follow,
where:
f.follower_actor_id == ^src_id and
f.target_actor_id == ^tgt_id and
not is_nil(f.accepted_at)
)
end
defp mark_gated(delivery, status, reason) do
delivery
|> Ecto.Changeset.change(
status: status,
status_reason: reason,
processed_at: DateTime.utc_now(:second)
)
|> Repo.update!()
end
defp dispatch("Follow", doc, source_actor, target_actor) do
case build_follow(source_actor, target_actor, doc) do
{:ok, follow} ->
maybe_send_accept(follow, doc, target_actor)
_ ->
:ok
end
end
defp dispatch("Undo", %{"object" => %{"type" => "Follow"} = follow_obj}, source_actor, _target) do
target_url = Map.get(follow_obj, "object")
case Federation.get_actor_by_url(target_url) do
%Actor{} = target ->
Repo.delete_all(
from f in Follow,
where: f.follower_actor_id == ^source_actor.id and f.target_actor_id == ^target.id
)
:ok
_ ->
:ok
end
end
defp dispatch("Accept", %{"object" => follow_id}, _source, target_actor)
when is_binary(follow_id) do
# `follow_id` is the activity URL of the Follow we sent. We stored
# that as fed_activities.activity_url; mark the corresponding row
# accepted.
case Repo.get_by(Activity, activity_url: follow_id) do
%Activity{id: aid} ->
Repo.update_all(
from(f in Follow,
where: f.follow_activity_id == ^aid and f.follower_actor_id == ^target_actor.id
),
set: [accepted_at: DateTime.utc_now(:second)]
)
:ok
_ ->
:ok
end
end
defp dispatch("Offer", %{"object" => %{} = object} = doc, source_actor, target_actor) do
handle_offer(object, doc, source_actor, target_actor)
end
defp dispatch("Update", %{"object" => %{} = object}, source_actor, _target_actor) do
handle_update(object, source_actor)
end
defp dispatch("Reject", %{"object" => object}, source_actor, _target_actor) do
handle_reject(object, source_actor)
end
defp dispatch("Create", %{"object" => %{} = object} = doc, source_actor, target_actor) do
handle_create(object, doc, source_actor, target_actor)
end
defp dispatch("Like", doc, source_actor, target_actor) do
handle_review(:approved, doc, source_actor, target_actor)
end
defp dispatch("Dislike", doc, source_actor, target_actor) do
handle_review(:changes_requested, doc, source_actor, target_actor)
end
defp dispatch(_type, _doc, _source, _target), do: :ok
# ── Offer (federated PR) ─────────────────────────────────────────────
# ForgeFed represents a PR as a `Ticket` with `isMerge: true` (or
# equivalent: branch + head fields present). The Offer wraps the
# Ticket and identifies the target repo via `target` or via the
# delivery's target_actor (which is what we already resolved).
defp handle_offer(object, doc, source_actor, target_actor) do
cond do
not ticket_with_merge?(object) ->
:ok
true ->
case repo_for_actor(target_actor) do
%GitGud.Repositories.Repository{} = repo ->
create_federated_pr_from_offer(repo, source_actor, object, doc)
_ ->
:ok
end
end
end
defp ticket_with_merge?(%{"type" => type} = obj) do
type_str = if is_list(type), do: Enum.join(type, ","), else: to_string(type)
String.contains?(type_str, "Ticket") and
(Map.get(obj, "isMerge") == true or
(Map.has_key?(obj, "sourceBranch") and Map.has_key?(obj, "targetBranch")))
end
defp ticket_with_merge?(_), do: false
defp create_federated_pr_from_offer(repo, source_actor, object, doc) do
activity_url = Map.get(doc, "id")
with :ok <- ensure_not_duplicate(activity_url),
{:ok, clone_url} <- resolve_clone_url(object, source_actor),
{:ok, head_hex} <- extract_head_hex(object),
{:ok, source_ref} <- fetch_required(object, "sourceBranch"),
{:ok, target_ref} <- fetch_required(object, "targetBranch") do
params = %{
title: Map.get(object, "name") || Map.get(object, "summary") || "Pull request",
body: Map.get(object, "content") || "",
source_ref: source_ref,
target_ref: target_ref,
head_hex: head_hex,
source_clone_url: clone_url,
activity_url: activity_url
}
_ = GitGud.PullRequests.create_federated_pull_request(repo, source_actor, params)
:ok
else
_ -> :ok
end
end
defp ensure_not_duplicate(nil), do: :ok
defp ensure_not_duplicate(url) do
import Ecto.Query, warn: false
if Repo.exists?(from p in GitGud.PullRequests.PullRequest, where: p.activity_url == ^url) do
{:error, :duplicate}
else
:ok
end
end
# Prefer the explicit `cloneUri` on the Ticket's `source`; fall back
# to the source actor's `endpoints.cloneUri` (Forgejo/Gitea convention);
# last resort derive from the source actor's URL by appending `.git`.
defp resolve_clone_url(object, source_actor) do
case get_in(object, ["source", "cloneUri"]) || Map.get(object, "cloneUri") do
url when is_binary(url) and url != "" ->
{:ok, url}
_ ->
case actor_clone_uri(source_actor) do
url when is_binary(url) and url != "" -> {:ok, url}
_ -> {:error, :no_clone_url}
end
end
end
defp actor_clone_uri(%Actor{remote_doc: %{} = doc, actor_url: url}) do
get_in(doc, ["endpoints", "cloneUri"]) || (url && url <> ".git")
end
defp actor_clone_uri(%Actor{actor_url: url}) when is_binary(url), do: url <> ".git"
defp actor_clone_uri(_), do: nil
defp extract_head_hex(object) do
case Map.get(object, "head") || get_in(object, ["headCommit", "hash"]) do
hex when is_binary(hex) and byte_size(hex) == 40 -> {:ok, hex}
_ -> {:error, :no_head}
end
end
defp fetch_required(obj, key) do
case Map.get(obj, key) do
v when is_binary(v) and v != "" -> {:ok, v}
_ -> {:error, {:missing, key}}
end
end
# ── Update (federated PR head moved) ─────────────────────────────────
# An Update wrapping a Ticket is a "remote pushed more commits to the
# source branch" signal. Re-fetch the source branch and re-snapshot
# head_sha + base_sha on the local PR row. We identify the PR by the
# Ticket's `id` matching the PR's stored `activity_url` *or* by the
# source actor + source clone URL + branch (some implementations
# re-use the Ticket id and don't echo the original Offer URL).
defp handle_update(object, source_actor) do
cond do
ticket_with_merge?(object) ->
with {:ok, pr} <- find_federated_pr(object, source_actor) do
refresh_federated_pr(pr, object)
end
true ->
:ok
end
end
defp refresh_federated_pr(pr, object) do
import Ecto.Query, warn: false
head_hex = Map.get(object, "head") || get_in(object, ["headCommit", "hash"])
if is_binary(head_hex) and byte_size(head_hex) == 40 do
GitGud.PullRequests.refresh_federated_head(pr, head_hex)
else
:ok
end
end
# ── Reject (federated PR closed by remote) ───────────────────────────
defp handle_reject(object, source_actor) do
case find_federated_pr(object, source_actor) do
{:ok, pr} ->
_ = GitGud.PullRequests.set_state(pr, "closed")
:ok
_ ->
:ok
end
end
defp find_federated_pr(object, source_actor) when is_map(object) do
import Ecto.Query, warn: false
ticket_id = Map.get(object, "id")
by_ticket =
if is_binary(ticket_id) do
Repo.one(
from p in GitGud.PullRequests.PullRequest,
where: p.activity_url == ^ticket_id and p.source_actor_id == ^source_actor.id,
limit: 1
)
end
cond do
by_ticket ->
{:ok, by_ticket}
true ->
# Fallback: source actor + source branch (best we can do when
# implementations don't echo the original activity URL).
case Map.get(object, "sourceBranch") do
ref when is_binary(ref) ->
Repo.one(
from p in GitGud.PullRequests.PullRequest,
where: p.source_actor_id == ^source_actor.id and p.source_ref == ^ref,
order_by: [desc: p.inserted_at],
limit: 1
)
|> case do
nil -> {:error, :no_match}
pr -> {:ok, pr}
end
_ ->
{:error, :no_match}
end
end
end
defp find_federated_pr(object, source_actor) when is_binary(object) do
# Reject can carry the Ticket id as a bare string instead of an
# embedded object.
find_federated_pr(%{"id" => object}, source_actor)
end
defp find_federated_pr(_, _), do: {:error, :no_match}
# ── Create (Note inReplyTo a federated PR) ───────────────────────────
defp handle_create(%{"type" => type} = object, doc, source_actor, target_actor) do
if note_type?(type) do
with {:ok, parent_url} <- fetch_required(object, "inReplyTo"),
{:ok, pr} <- find_pr_by_object_url(parent_url, target_actor) do
create_federated_pr_comment(pr, source_actor, object, doc)
else
_ -> :ok
end
else
:ok
end
end
defp note_type?(type) when is_binary(type), do: String.contains?(type, "Note")
defp note_type?(list) when is_list(list), do: Enum.any?(list, ¬e_type?/1)
defp note_type?(_), do: false
defp find_pr_by_object_url(url, target_actor) when is_binary(url) do
# Two URL shapes can reach us:
#
# 1. **Local PR Ticket URL** — `<repo_actor>/pull_requests/<N>`.
# The URL we hand out on outbound activities; remote peers use
# it when replying to one of our PRs.
#
# 2. **Remote-minted Ticket URL** — stored as `activity_url` on a
# federated PR row when we accepted the inbound `Offer`. Remote
# peers send Like / Dislike / Note with this URL as `object` /
# `inReplyTo` to reference the same PR they proposed.
case find_pr_local_ticket(url, target_actor) do
{:ok, pr} ->
{:ok, pr}
_ ->
case Repo.get_by(GitGud.PullRequests.PullRequest, activity_url: url) do
nil -> {:error, :no_pr}
pr -> {:ok, pr}
end
end
end
defp find_pr_by_object_url(_, _), do: {:error, :no_url}
defp find_pr_local_ticket(url, target_actor) do
repo_url = target_actor.actor_url
prefix = repo_url <> "/pull_requests/"
case String.starts_with?(url, prefix) and
Integer.parse(String.replace_prefix(url, prefix, "")) do
{n, ""} when is_integer(n) ->
case Repo.get_by(GitGud.PullRequests.PullRequest,
repository_id: target_actor.local_repository_id,
number: n
) do
nil -> :error
pr -> {:ok, pr}
end
_ ->
:error
end
end
defp create_federated_pr_comment(pr, source_actor, object, _doc) do
activity_url = Map.get(object, "id")
body = Map.get(object, "content") || ""
if Repo.exists?(
Ecto.Query.from(c in GitGud.PullRequests.PrComment,
where: c.activity_url == ^activity_url
)
) do
:ok
else
_ =
GitGud.PullRequests.add_federated_comment(pr, source_actor, %{
body: body,
activity_url: activity_url
})
:ok
end
end
# ── Like / Dislike → PR review ──────────────────────────────────────
# `Like` becomes an `approved` review; `Dislike` becomes
# `changes_requested`. The activity's `object` is the PR's Ticket
# URL; we find the matching local PR and insert a federated review
# row with `source_actor_id` set. Idempotent on the activity URL.
defp handle_review(state, %{"object" => object} = doc, source_actor, target_actor) do
object_url =
cond do
is_binary(object) -> object
is_map(object) -> Map.get(object, "id")
true -> nil
end
with {:ok, pr} <- find_pr_by_object_url(object_url, target_actor),
{:ok, _review} <-
GitGud.PullRequests.add_federated_review(pr, source_actor, %{
state: Atom.to_string(state),
body: Map.get(doc, "content"),
activity_url: Map.get(doc, "id")
}) do
:ok
else
_ -> :ok
end
end
defp handle_review(_state, _doc, _source, _target), do: :ok
# ── helpers ──────────────────────────────────────────────────────────
defp record_activity(source_actor, %{"id" => url, "type" => type} = doc) do
%Activity{}
|> Activity.changeset(%{
activity_url: url,
type: type,
actor_id: source_actor.id,
raw: doc,
is_local: false,
published_at: parse_published(Map.get(doc, "published"))
})
|> Repo.insert(on_conflict: :nothing, conflict_target: [:activity_url])
end
defp record_activity(_source_actor, _doc), do: {:error, :no_id_or_type}
defp parse_published(nil), do: nil
defp parse_published(s) when is_binary(s) do
case DateTime.from_iso8601(s) do
{:ok, dt, _} -> DateTime.truncate(dt, :second)
_ -> nil
end
end
defp build_follow(source_actor, target_actor, doc) do
activity_url = Map.get(doc, "id")
activity_row =
activity_url && Repo.get_by(Activity, activity_url: activity_url)
%Follow{}
|> Follow.changeset(%{
follower_actor_id: source_actor.id,
target_actor_id: target_actor.id,
follow_activity_id: activity_row && activity_row.id
})
|> Repo.insert(
on_conflict: :nothing,
conflict_target: [:follower_actor_id, :target_actor_id],
returning: true
)
end
defp maybe_send_accept(follow, follow_doc, target_actor) do
accept_url =
target_actor.actor_url <>
"/activities/accept/" <> Integer.to_string(follow.id) <> "-" <> rand()
accept_doc = %{
"@context" => "https://www.w3.org/ns/activitystreams",
"id" => accept_url,
"type" => "Accept",
"actor" => target_actor.actor_url,
"object" => follow_doc
}
{:ok, activity} =
%Activity{}
|> Activity.changeset(%{
activity_url: accept_url,
type: "Accept",
actor_id: target_actor.id,
target_actor_id: follow.follower_actor_id,
raw: accept_doc,
is_local: true,
published_at: DateTime.utc_now(:second)
})
|> Repo.insert()
# Auto-accept the follow on our side too so UI sees them as a follower.
follow |> Follow.accept() |> Repo.update!()
# Queue the outbound delivery.
{:ok, _job} =
DeliverOutbound.new_job(activity.id, source_inbox_for(follow)) |> Oban.insert()
:ok
end
defp source_inbox_for(%Follow{follower_actor_id: aid}) do
actor = Repo.get!(Actor, aid)
actor.inbox_url
end
defp rand, do: :crypto.strong_rand_bytes(8) |> Base.url_encode64(padding: false)
defp mark_processed(%InboxDelivery{} = delivery) do
delivery
|> Ecto.Changeset.change(processed_at: DateTime.utc_now(:second))
|> Repo.update!()
end
end
18.8 KiB · text
5af55d9