defmodule GitGud.PullRequests do
@moduledoc """
Pull requests + mergeability + merging.
Merge strategy currently implemented: a non-fast-forward 3-way merge
using `git merge-tree --write-tree` to produce the resulting tree
followed by `git commit-tree` + `git update-ref` for an atomic CAS on
the target ref. Squash/rebase strategies live in a later phase.
"""
import Ecto.Query, warn: false
alias GitGud.Accounts.User
alias GitGud.BranchProtections
alias GitGud.Federation.Actor, as: FedActor
alias GitGud.Git
alias GitGud.Labels
alias GitGud.PullRequests.PullRequest
alias GitGud.PullRequests.PrComment
alias GitGud.PullRequests.PrReview
alias GitGud.Repo
alias GitGud.Repositories
alias GitGud.Repositories.Numbering
alias GitGud.Repositories.Repository
@target_type "pull_request"
# ── queries ──────────────────────────────────────────────────────────
def list_pull_requests(%Repository{id: rid}, opts \\ []) do
state = Keyword.get(opts, :state, "open")
label_ids = Keyword.get(opts, :label_ids, [])
search = Keyword.get(opts, :search)
limit = Keyword.get(opts, :limit, 50)
offset = Keyword.get(opts, :offset, 0)
base =
PullRequest
|> where([p], p.repository_id == ^rid)
|> filter_state(state)
|> filter_labels(label_ids)
|> filter_search(search)
|> order_by([p], desc: p.inserted_at)
|> limit(^limit)
|> offset(^offset)
|> preload([:author])
prs = Repo.all(base)
labels_by_id = Labels.labels_for_many(@target_type, Enum.map(prs, & &1.id))
Enum.map(prs, &Map.put(&1, :labels, Map.get(labels_by_id, &1.id, [])))
end
defp filter_state(q, "all"), do: q
defp filter_state(q, s) when s in ["open", "closed", "merged"],
do: where(q, [p], p.state == ^s)
defp filter_state(q, _), do: q
defp filter_labels(q, []), do: q
defp filter_labels(q, label_ids) do
from(p in q,
join: lab in "labelings",
on:
lab.target_type == ^@target_type and lab.target_id == p.id and lab.label_id in ^label_ids
)
|> distinct(true)
end
defp filter_search(q, nil), do: q
defp filter_search(q, ""), do: q
defp filter_search(q, term) when is_binary(term) do
websearch = String.trim(term)
from p in q,
where: fragment("? @@ websearch_to_tsquery('english', ?)", p.search_vector, ^websearch)
end
def get_pull_request!(%Repository{id: rid}, number) do
comments_q = from c in PrComment, order_by: [asc: c.inserted_at, asc: c.id]
reviews_q = from r in PrReview, order_by: [asc: r.inserted_at, asc: r.id]
PullRequest
|> where([p], p.repository_id == ^rid and p.number == ^number)
|> preload([
:author,
:merged_by,
comments: ^{comments_q, [:author, :source_actor]},
reviews: ^{reviews_q, [:reviewer]}
])
|> Repo.one!()
|> attach_labels()
end
defp attach_labels(%PullRequest{id: id} = pr),
do: Map.put(pr, :labels, Labels.labels_for({@target_type, id}))
# ── lifecycle ────────────────────────────────────────────────────────
@doc """
Open a new PR. Resolves source/target to SHAs, computes the merge base,
and snapshots both head and base into the PR row.
"""
def create_pull_request(%Repository{} = repo, %User{id: uid} = author, attrs) do
source_ref = attrs["source_ref"] || attrs[:source_ref]
target_ref = attrs["target_ref"] || attrs[:target_ref]
source_repo_id = attrs["source_repository_id"] || attrs[:source_repository_id]
{source_repo, cross_repo?} = resolve_source_repo(repo, source_repo_id)
result =
with {:ok, head_sha} <- Repositories.resolve(source_repo, source_ref),
{:ok, target_sha} <- Repositories.resolve(repo, target_ref),
# For cross-repo PRs we need to fetch the source's commit into
# the target's object database before merge_base / merge_tree
# can see both SHAs from one repo's perspective.
:ok <- maybe_fetch_cross_repo(cross_repo?, repo, source_repo, source_ref),
{:ok, base_sha} <- Git.merge_base(repo.disk_path, head_sha, target_sha) do
Repo.transaction(fn ->
next = Numbering.next_for(repo.id)
changeset =
%PullRequest{
repository_id: repo.id,
source_repository_id: source_repo.id,
author_id: uid,
number: next,
state: "open",
head_sha: head_sha,
base_sha: base_sha,
source_ref: source_ref,
target_ref: target_ref
}
|> PullRequest.create_changeset(attrs)
case Repo.insert(changeset) do
{:ok, pr} ->
# Pin the cross-repo head into a stable ref on the target so
# later operations (merge, refresh) don't depend on the
# fork's path being reachable at that moment.
if cross_repo? do
pin_pull_head(repo, source_repo, source_ref, pr.number)
end
attach_labels(pr)
{:error, cs} ->
Repo.rollback(cs)
end
end)
else
err -> {:error, err}
end
with {:ok, pr} <- result do
_ = GitGud.Federation.Publishing.publish_pr_opened(repo, pr, author)
{:ok, pr}
end
end
@doc """
Open a PR coming from a remote ActivityPub actor via an `Offer` activity.
Unlike `create_pull_request/3`, the source is *not* a local fork — we
fetch the source branch directly from a remote git URL into the target
repo under a stable `refs/pull/<N>/head`. The PR row has no local
`author_id` or `source_repository_id`; instead `source_actor_id`
points to the remote actor and `activity_url` pins the Offer for
idempotency.
`params` must include: `title, body, source_ref, target_ref, head_hex,
source_clone_url, activity_url`.
"""
def create_federated_pull_request(
%Repository{} = target_repo,
%FedActor{} = source_actor,
params
) do
with :ok <- validate_federated_params(params),
{:ok, _} <-
fetch_federated_head(
target_repo,
params.source_clone_url,
params.source_ref,
params.head_hex
),
{:ok, head_sha} <- Git.from_hex(params.head_hex),
{:ok, target_sha} <- Repositories.resolve(target_repo, params.target_ref),
{:ok, base_sha} <- Git.merge_base(target_repo.disk_path, head_sha, target_sha) do
Repo.transaction(fn ->
next = Numbering.next_for(target_repo.id)
attrs = %{
"title" => params.title,
"body" => params.body,
"source_ref" => params.source_ref,
"target_ref" => params.target_ref
}
changeset =
%PullRequest{
repository_id: target_repo.id,
source_repository_id: nil,
source_actor_id: source_actor.id,
activity_url: params.activity_url,
source_clone_url: params.source_clone_url,
author_id: nil,
number: next,
state: "open",
head_sha: head_sha,
base_sha: base_sha
}
|> PullRequest.create_changeset(attrs)
case Repo.insert(changeset) do
{:ok, pr} ->
:ok =
Git.update_ref(
target_repo.disk_path,
"refs/pull/#{pr.number}/head",
head_sha,
nil
)
attach_labels(pr)
{:error, cs} ->
Repo.rollback(cs)
end
end)
end
end
defp validate_federated_params(%{
title: t,
source_ref: s,
target_ref: tgt,
head_hex: h,
source_clone_url: u,
activity_url: a
})
when is_binary(t) and is_binary(s) and is_binary(tgt) and is_binary(h) and
is_binary(u) and is_binary(a) and byte_size(h) == 40 do
:ok
end
defp validate_federated_params(_), do: {:error, :invalid_federated_params}
defp fetch_federated_head(target_repo, clone_url, source_ref, head_hex) do
args = [
"-C",
target_repo.disk_path,
"fetch",
"--no-tags",
"--depth=50",
clone_url,
"+refs/heads/" <> source_ref <> ":refs/pull-tmp/" <> source_ref
]
case System.cmd("git", args, stderr_to_stdout: true) do
{_, 0} ->
# The Offer claims `head_hex` as the head; verify it's actually
# reachable now. Mismatch = mistrust the Offer.
case System.cmd(
"git",
["-C", target_repo.disk_path, "cat-file", "-e", head_hex],
stderr_to_stdout: true
) do
{_, 0} -> {:ok, head_hex}
_ -> {:error, :head_not_in_fetch}
end
{out, _} ->
{:error, {:remote_fetch_failed, out}}
end
end
@doc """
Re-snapshot a federated PR's `head_sha` after the remote source pushed
more commits. Re-fetches the source branch from `source_clone_url`,
re-pins `refs/pull/<N>/head`, and recomputes `base_sha`.
"""
def refresh_federated_head(%PullRequest{} = pr, head_hex) when is_binary(head_hex) do
repo = Repositories.get_repository!(pr.repository_id)
with {:ok, _} <-
fetch_federated_head(repo, pr.source_clone_url, pr.source_ref, head_hex),
{:ok, head_sha} <- Git.from_hex(head_hex),
{:ok, target_sha} <- Repositories.resolve(repo, pr.target_ref),
{:ok, base_sha} <- Git.merge_base(repo.disk_path, head_sha, target_sha) do
:ok =
Git.update_ref(
repo.disk_path,
"refs/pull/#{pr.number}/head",
head_sha,
pr.head_sha
)
pr
|> Ecto.Changeset.change(head_sha: head_sha, base_sha: base_sha, mergeable: nil)
|> Repo.update()
end
end
@doc """
Insert a comment authored by a remote AP actor (no local `User`).
`attrs` must include `:body` and `:activity_url` (the Create(Note)'s
`id`). Idempotent via the unique partial index on `activity_url`.
"""
def add_federated_comment(%PullRequest{} = pr, %FedActor{} = source_actor, attrs)
when is_map(attrs) do
%PrComment{
pull_request_id: pr.id,
author_id: nil,
source_actor_id: source_actor.id
}
|> PrComment.federated_changeset(%{
body: Map.get(attrs, :body) || Map.get(attrs, "body"),
activity_url: Map.get(attrs, :activity_url) || Map.get(attrs, "activity_url"),
source_actor_id: source_actor.id,
pull_request_id: pr.id
})
|> Repo.insert(
on_conflict: :nothing,
conflict_target: {:unsafe_fragment, "(activity_url) WHERE activity_url IS NOT NULL"}
)
end
defp resolve_source_repo(target_repo, nil), do: {target_repo, false}
defp resolve_source_repo(target_repo, id) when is_integer(id) or is_binary(id) do
src = Repositories.get_repository!(id)
{src, src.id != target_repo.id}
end
defp maybe_fetch_cross_repo(false, _, _, _), do: :ok
defp maybe_fetch_cross_repo(true, target_repo, source_repo, source_ref) do
args = [
"-C",
target_repo.disk_path,
"fetch",
"--no-tags",
source_repo.disk_path,
"+refs/heads/" <> source_ref <> ":refs/pull-tmp/" <> source_ref
]
case System.cmd("git", args, stderr_to_stdout: true) do
{_, 0} -> :ok
{out, _} -> {:error, {:cross_repo_fetch, out}}
end
end
@doc """
Called when a fork has been pushed. Updates `refs/pull/<N>/head` on
the target repo for every open cross-repo PR that has `source_repo` +
`branch` as its source, and re-snapshots `head_sha` on the PR row.
"""
def refresh_cross_repo_heads(%Repository{} = source_repo, branch, new_sha) do
prs =
Repo.all(
from p in PullRequest,
where:
p.source_repository_id == ^source_repo.id and
p.source_ref == ^branch and
p.state == "open"
)
for pr <- prs do
target_repo = Repositories.get_repository!(pr.repository_id)
_ =
System.cmd(
"git",
[
"-C",
target_repo.disk_path,
"fetch",
"--no-tags",
source_repo.disk_path,
"+refs/heads/" <> branch <> ":refs/pull/#{pr.number}/head"
],
stderr_to_stdout: true
)
pr
|> Ecto.Changeset.change(head_sha: new_sha)
|> Repo.update!()
end
:ok
end
defp pin_pull_head(target_repo, source_repo, source_ref, pr_number) do
# We've already brought the commit object in via the temp ref above.
# Pin it under a stable `refs/pull/<N>/head` and clean up the temp.
{:ok, sha} = Repositories.resolve(source_repo, source_ref)
:ok = Git.update_ref(target_repo.disk_path, "refs/pull/#{pr_number}/head", sha, nil)
_ =
System.cmd(
"git",
["-C", target_repo.disk_path, "update-ref", "-d", "refs/pull-tmp/" <> source_ref],
stderr_to_stdout: true
)
:ok
end
def update_pull_request(%PullRequest{} = pr, attrs) do
pr
|> PullRequest.update_changeset(attrs)
|> Repo.update()
end
def set_state(%PullRequest{} = pr, state) when state in ["open", "closed"] do
pr
|> PullRequest.state_changeset(state)
|> Repo.update()
end
# ── mergeability ─────────────────────────────────────────────────────
@doc """
Run the merge probe and cache the result. Cheap to call repeatedly —
only re-runs git when head/base SHAs have changed since the last
probe. Returns `{:ok, true|false}` or `{:error, reason}`.
"""
def check_mergeability(%PullRequest{} = pr) do
repo = Repositories.get_repository!(pr.repository_id)
case Git.merge_tree(repo.disk_path, pr.base_sha, pr.head_sha, target_head_sha!(repo, pr)) do
{:ok, _tree} -> apply_mergeability(pr, true)
{:error, :conflict} -> apply_mergeability(pr, false)
{:error, reason} -> {:error, reason}
end
end
defp target_head_sha!(repo, pr) do
{:ok, sha} = Repositories.resolve(repo, pr.target_ref)
sha
end
defp apply_mergeability(pr, result) do
case pr |> PullRequest.mergeability_changeset(result) |> Repo.update() do
{:ok, updated} -> {:ok, updated.mergeable}
err -> err
end
end
# ── merging ──────────────────────────────────────────────────────────
@doc """
Perform a 3-way merge of `pr.head_sha` into the target ref. Atomic via
`git update-ref <ref> <new> <expected_old>`: if someone else moved the
ref between probe and merge, we surface a `:concurrent_update` error
and leave the PR untouched.
"""
def merge!(%PullRequest{} = pr, %User{} = actor, message \\ nil) do
repo = Repositories.get_repository!(pr.repository_id)
path = repo.disk_path
with {:ok, current_target_sha} <- Repositories.resolve(repo, pr.target_ref),
:ok <-
BranchProtections.check_ref_update(
repo,
pr.target_ref,
current_target_sha,
pr.head_sha,
pull_request_id: pr.id
),
{:ok, base_sha} <- Git.merge_base(path, pr.head_sha, current_target_sha),
{:ok, tree_sha} <-
Git.merge_tree(path, base_sha, current_target_sha, pr.head_sha),
{:ok, commit_sha} <-
Git.commit_tree(
path,
tree_sha,
[current_target_sha, pr.head_sha],
message || default_merge_message(pr),
author: actor_identity(actor)
),
:ok <-
Git.update_ref(
path,
"refs/heads/" <> pr.target_ref,
commit_sha,
current_target_sha
) do
# Reflect the new state in DB.
with {:ok, updated} <-
pr |> PullRequest.merged_changeset(commit_sha, actor) |> Repo.update() do
# Keep the refs cache in sync without waiting for post-receive
# to land.
:ok =
Repositories.apply_ref_update(
repo,
"refs/heads/" <> pr.target_ref,
current_target_sha,
commit_sha
)
_ = GitGud.Federation.Publishing.publish_pr_merged(repo, updated)
{:ok, updated}
end
else
{:error, :conflict} -> {:error, :conflict}
err -> err
end
end
defp default_merge_message(pr), do: "Merge pull request ##{pr.number}: #{pr.title}"
defp actor_identity(%User{email: email}) do
name = email |> String.split("@") |> hd()
{name, email, DateTime.utc_now(:second)}
end
# ── comments + reviews ───────────────────────────────────────────────
def add_comment(%PullRequest{} = pr, %User{id: uid} = author, attrs) do
case %PrComment{pull_request_id: pr.id, author_id: uid}
|> PrComment.changeset(attrs)
|> Repo.insert() do
{:ok, comment} ->
repo = Repositories.get_repository!(pr.repository_id)
_ = GitGud.Federation.Publishing.publish_pr_comment(repo, pr, comment, author)
{:ok, comment}
err ->
err
end
end
def add_review(%PullRequest{} = pr, %User{id: uid} = reviewer, attrs) do
case %PrReview{pull_request_id: pr.id, reviewer_id: uid}
|> PrReview.changeset(attrs)
|> Repo.insert() do
{:ok, review} ->
# Reviewing a federated PR? Fire an outbound Like / Dislike to
# the source actor so the originating instance sees the verdict.
if pr.source_actor_id && review.state in ["approved", "changes_requested"] do
_ =
GitGud.Federation.Publishing.publish_pr_review(
pr,
reviewer,
review
)
end
{:ok, review}
err ->
err
end
end
@doc """
Insert a PR review that arrived via federation. `attrs` must include
`:state`, `:activity_url`; `:body` optional. Idempotent on
`activity_url` via the unique partial index.
"""
def add_federated_review(%PullRequest{} = pr, %FedActor{} = source_actor, attrs) do
%PrReview{
pull_request_id: pr.id,
source_actor_id: source_actor.id,
reviewer_id: nil
}
|> PrReview.federated_changeset(%{
state: Map.get(attrs, :state) || Map.get(attrs, "state"),
body: Map.get(attrs, :body) || Map.get(attrs, "body"),
activity_url: Map.get(attrs, :activity_url) || Map.get(attrs, "activity_url"),
source_actor_id: source_actor.id,
pull_request_id: pr.id
})
|> Repo.insert(
on_conflict: :nothing,
conflict_target: {:unsafe_fragment, "(activity_url) WHERE activity_url IS NOT NULL"}
)
end
# ── labels ───────────────────────────────────────────────────────────
def attach_label(%PullRequest{id: id}, label), do: Labels.attach(label, {@target_type, id})
def detach_label(%PullRequest{id: id}, label), do: Labels.detach(label, {@target_type, id})
def target_type, do: @target_type
end
19.2 KiB · text
5af55d9