alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::sync::Arc;
2
3use axum::Json;
4use axum::body::Bytes;
5use axum::extract::State;
6use axum::response::{IntoResponse, Response};
7use http::{HeaderMap, StatusCode};
8use serde::{Deserialize, Serialize};
9
10use knot_events::Reservation;
11use knot_git::{
12 ApplyError, ApplyOutcome, Conflict, Identity, NewCommit, ParsedFile, PatchApplier,
13 PatchParseError, RefUpdate, Repo, StagedChange, Staging, is_format_patch,
14 parse_mailbox_bounded, parse_patch_bounded,
15};
16use knot_index::Resolved;
17use knot_postreceive::{Actor, Ci};
18use knot_runtime::{Clock, HttpTransport};
19use knot_types::{
20 AuthorName, BranchName, Email, Oid, OwnerDid, RefName, RepoDid, RepoName, RepoRkey, UnixSeconds,
21};
22
23use crate::body::{CommitBody, CommitMessage, Patch};
24use crate::error::XrpcError;
25use crate::reads::{open, repo_not_found, warming};
26use crate::{XrpcState, decode, ok_empty, run_blocking};
27
28pub(crate) const MERGE_ROUTE: &str = "/xrpc/sh.tangled.repo.merge";
29pub(crate) const MERGE_CHECK_ROUTE: &str = "/xrpc/sh.tangled.repo.mergeCheck";
30const MERGE_RETRIES: u32 = 3;
31const CONFLICT_MESSAGE: &str = "patch cannot be applied cleanly";
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct Committer {
35 pub name: AuthorName,
36 pub email: Email,
37}
38
39#[derive(Deserialize)]
40#[serde(rename_all = "camelCase")]
41struct MergeInput {
42 did: OwnerDid,
43 name: RepoName,
44 patch: Patch,
45 branch: BranchName,
46 author_name: Option<AuthorName>,
47 author_email: Option<Email>,
48 commit_message: Option<CommitMessage>,
49 commit_body: Option<CommitBody>,
50}
51
52#[derive(Deserialize)]
53struct MergeCheckInput {
54 did: OwnerDid,
55 name: RepoName,
56 patch: Patch,
57 branch: BranchName,
58}
59
60#[derive(Serialize)]
61struct ConflictWire {
62 filename: String,
63 reason: String,
64}
65
66#[derive(Serialize)]
67struct MergeCheckOutput {
68 is_conflicted: bool,
69 #[serde(skip_serializing_if = "Option::is_none")]
70 conflicts: Option<Vec<ConflictWire>>,
71 #[serde(skip_serializing_if = "Option::is_none")]
72 message: Option<String>,
73 #[serde(skip_serializing_if = "Option::is_none")]
74 error: Option<String>,
75}
76
77impl MergeCheckOutput {
78 fn clean() -> Self {
79 Self {
80 is_conflicted: false,
81 conflicts: None,
82 message: None,
83 error: None,
84 }
85 }
86
87 fn conflicted(conflicts: Vec<Conflict>) -> Self {
88 Self {
89 is_conflicted: true,
90 conflicts: Some(
91 conflicts
92 .into_iter()
93 .map(|conflict| ConflictWire {
94 filename: conflict.path,
95 reason: conflict.reason.as_str().to_string(),
96 })
97 .collect(),
98 ),
99 message: Some(CONFLICT_MESSAGE.to_string()),
100 error: None,
101 }
102 }
103
104 fn broken(error: String) -> Self {
105 Self {
106 is_conflicted: true,
107 conflicts: None,
108 message: None,
109 error: Some(error),
110 }
111 }
112}
113
114struct CommitSpec {
115 files: Vec<ParsedFile>,
116 author: Option<MailAuthor>,
117 message: String,
118 change_id: Option<knot_git::CommitChangeId>,
119}
120
121struct MailAuthor {
122 name: AuthorName,
123 email: Email,
124 date: String,
125}
126
127fn parse_specs(
128 patch: &str,
129 message: String,
130 author: Option<MailAuthor>,
131 max_bytes: u64,
132) -> Result<Vec<CommitSpec>, PatchParseError> {
133 match is_format_patch(patch) {
134 true => Ok(parse_mailbox_bounded(patch, max_bytes)?
135 .into_iter()
136 .map(|mail| CommitSpec {
137 message: mail.commit_message(),
138 author: Some(MailAuthor {
139 name: mail.author_name,
140 email: mail.author_email,
141 date: mail.date,
142 }),
143 change_id: mail.change_id,
144 files: mail.files,
145 })
146 .collect()),
147 false => Ok(vec![CommitSpec {
148 files: parse_patch_bounded(patch, max_bytes)?,
149 author,
150 message,
151 change_id: None,
152 }]),
153 }
154}
155
156pub(crate) fn resolve_by_name<H: HttpTransport, C: Clock>(
157 state: &XrpcState<H, C>,
158 owner: &OwnerDid,
159 name: &RepoName,
160) -> Result<RepoDid, XrpcError> {
161 let rkey = RepoRkey::new(name.as_str()).map_err(|_| repo_not_found())?;
162 match state.index.resolve_repo(owner, &rkey) {
163 Resolved::Ready(found) => found.ok_or_else(repo_not_found),
164 Resolved::Warming => Err(warming()),
165 }
166}
167
168fn branch_tip(repo: &Repo, refname: &RefName) -> Result<Oid, XrpcError> {
169 repo.find_ref(refname)?
170 .ok_or_else(|| XrpcError::invalid_request("no such branch to merge into"))
171}
172
173fn mail_time(date: &str, now: UnixSeconds) -> (UnixSeconds, i32) {
174 let trimmed = date.trim();
175 chrono::DateTime::parse_from_rfc2822(trimmed)
176 .or_else(|_| chrono::DateTime::parse_from_rfc3339(trimmed))
177 .map(|parsed| {
178 (
179 UnixSeconds::new(parsed.timestamp()),
180 parsed.offset().local_minus_utc(),
181 )
182 })
183 .unwrap_or((now, 0))
184}
185
186fn spec_identities(
187 spec: &CommitSpec,
188 fallback: &Identity,
189 now: UnixSeconds,
190) -> (Identity, Vec<(String, Vec<u8>)>) {
191 let author = match &spec.author {
192 Some(mail) => {
193 let (time, offset_seconds) = mail_time(&mail.date, now);
194 Identity {
195 name: mail.name.clone(),
196 email: mail.email.clone(),
197 time,
198 offset_seconds,
199 }
200 }
201 None => fallback.clone(),
202 };
203 let extra_headers = spec
204 .change_id
205 .iter()
206 .map(|change_id| {
207 (
208 "change-id".to_string(),
209 change_id.as_str().as_bytes().to_vec(),
210 )
211 })
212 .collect();
213 (author, extra_headers)
214}
215
216enum StageStop {
217 Conflict(Vec<Conflict>),
218 Apply(ApplyError),
219}
220
221fn stage_all(
222 repo: &Repo,
223 tip: Oid,
224 specs: &[CommitSpec],
225) -> Result<Result<Vec<Vec<StagedChange>>, Vec<Conflict>>, ApplyError> {
226 let mut applier = PatchApplier::new(repo, tip);
227 let staged = specs.iter().try_fold(Vec::new(), |mut clean, spec| {
228 match applier.step(&spec.files) {
229 Ok(ApplyOutcome::Clean(staged)) => {
230 clean.push(staged);
231 Ok(clean)
232 }
233 Ok(ApplyOutcome::Conflicted(conflicts)) => Err(StageStop::Conflict(conflicts)),
234 Err(error) => Err(StageStop::Apply(error)),
235 }
236 });
237 match staged {
238 Ok(clean) => Ok(Ok(clean)),
239 Err(StageStop::Conflict(conflicts)) => Ok(Err(conflicts)),
240 Err(StageStop::Apply(error)) => Err(error),
241 }
242}
243
244enum MergeAttempt {
245 Done {
246 old: Oid,
247 new: Oid,
248 reservation: Reservation,
249 },
250 Conflicted(Vec<Conflict>),
251 Raced,
252}
253
254enum Merged {
255 Done {
256 old: Oid,
257 new: Oid,
258 reservation: Reservation,
259 },
260 Conflicted(Vec<Conflict>),
261}
262
263fn attempt_merge(
264 repo: &Repo,
265 refname: &RefName,
266 specs: &[CommitSpec],
267 committer: &Committer,
268 now: UnixSeconds,
269 reserve: &dyn Fn() -> Reservation,
270) -> Result<MergeAttempt, XrpcError> {
271 if specs.iter().any(|spec| spec.message.trim().is_empty()) {
272 return Err(XrpcError::invalid_request("commit message is required"));
273 }
274 let tip = branch_tip(repo, refname)?;
275 let staged = match stage_all(repo, tip, specs).map_err(XrpcError::from)? {
276 Ok(staged) => staged,
277 Err(conflicts) => return Ok(MergeAttempt::Conflicted(conflicts)),
278 };
279 let committer_identity = Identity {
280 name: committer.name.clone(),
281 email: committer.email.clone(),
282 time: now,
283 offset_seconds: 0,
284 };
285 let staging = Staging::new(repo).map_err(XrpcError::from)?;
286 let work = staging.repo();
287 let base_tree = work.find_commit(tip).map_err(XrpcError::from)?.tree;
288 let new_tip = specs.iter().zip(staged).try_fold(
289 (base_tree, tip),
290 |(tree, parent), (spec, staged)| -> Result<(Oid, Oid), XrpcError> {
291 let next_tree = work
292 .write_staged_tree(tree, &staged)
293 .map_err(XrpcError::from)?;
294 let (author, extra_headers) = spec_identities(spec, &committer_identity, now);
295 let commit = work
296 .write_commit(&NewCommit {
297 tree: next_tree,
298 parents: vec![parent],
299 author,
300 committer: committer_identity.clone(),
301 message: spec.message.clone(),
302 extra_headers,
303 })
304 .map_err(XrpcError::from)?;
305 Ok((next_tree, commit))
306 },
307 )?;
308 match repo.find_ref(refname).map_err(XrpcError::from)? {
309 Some(current) if current == tip => {}
310 _ => return Ok(MergeAttempt::Raced),
311 }
312 staging.migrate_into(repo).map_err(XrpcError::from)?;
313 match repo.update_ref_sealed(
314 &RefUpdate::Update {
315 name: refname.clone(),
316 old: tip,
317 new: new_tip.1,
318 },
319 reserve,
320 ) {
321 Ok(reservation) => Ok(MergeAttempt::Done {
322 old: tip,
323 new: new_tip.1,
324 reservation,
325 }),
326 Err(error) => match repo.find_ref(refname) {
327 Ok(Some(current)) if current != tip => Ok(MergeAttempt::Raced),
328 _ => Err(error.into()),
329 },
330 }
331}
332
333fn merge_with_retry(
334 repo: &Repo,
335 refname: &RefName,
336 specs: &[CommitSpec],
337 committer: &Committer,
338 now: UnixSeconds,
339 attempts: u32,
340 reserve: &dyn Fn() -> Reservation,
341) -> Result<Merged, XrpcError> {
342 match attempt_merge(repo, refname, specs, committer, now, reserve)? {
343 MergeAttempt::Done {
344 old,
345 new,
346 reservation,
347 } => Ok(Merged::Done {
348 old,
349 new,
350 reservation,
351 }),
352 MergeAttempt::Conflicted(conflicts) => Ok(Merged::Conflicted(conflicts)),
353 MergeAttempt::Raced if attempts > 1 => {
354 merge_with_retry(repo, refname, specs, committer, now, attempts - 1, reserve)
355 }
356 MergeAttempt::Raced => Err(XrpcError::conflict("branch moved during the merge, retry")),
357 }
358}
359
360fn merge_conflict(conflicts: &[Conflict]) -> XrpcError {
361 let detail = conflicts
362 .first()
363 .map(|conflict| {
364 format!(
365 "{CONFLICT_MESSAGE}: {} {}",
366 conflict.path,
367 conflict.reason.as_str()
368 )
369 })
370 .unwrap_or_else(|| CONFLICT_MESSAGE.to_string());
371 XrpcError::named(
372 StatusCode::CONFLICT,
373 "MergeConflict",
374 format!("Merge failed due to conflicts: {detail}"),
375 )
376}
377
378pub(crate) async fn merge<H: HttpTransport, C: Clock>(
379 State(state): State<Arc<XrpcState<H, C>>>,
380 headers: HeaderMap,
381 method: crate::Method,
382 body: Bytes,
383) -> Result<Response, XrpcError> {
384 let actor = state.authenticate(&headers, &method).await?;
385 let input: MergeInput = decode(&body)?;
386 let repo_did = resolve_by_name(&state, &input.did, &input.name)?;
387 crate::authorize_push(
388 &state,
389 &actor,
390 &repo_did,
391 "only repository owner or a collaborator may merge",
392 )
393 .await?;
394 let refname = input.branch.head_ref();
395 let committer = state.committer.clone();
396 let now = state.now();
397 let layout = state.layout.clone();
398 let max_patch_bytes = state.byte_limits.patch_decompressed.get();
399 let event_repo = repo_did.clone();
400 let event_ref = refname.clone();
401
402 let events = Arc::clone(&state.events);
403 let outcome = run_blocking(move || {
404 let specs = parse_specs(
405 input.patch.as_str(),
406 unified_message(&input),
407 unified_author(&input),
408 max_patch_bytes,
409 )
410 .map_err(|error| XrpcError::invalid_request(error.to_string()))?;
411 let repo = open(&layout, &repo_did)?;
412 let reserve = || events.reserve();
413 merge_with_retry(
414 &repo,
415 &refname,
416 &specs,
417 &committer,
418 now,
419 MERGE_RETRIES,
420 &reserve,
421 )
422 })
423 .await?;
424
425 match outcome {
426 Merged::Conflicted(conflicts) => Err(merge_conflict(&conflicts)),
427 Merged::Done {
428 old,
429 new,
430 reservation,
431 } => {
432 let owner = crate::current_owner(&state, &event_repo);
433 let layout = state.layout.clone();
434 let languages_push_budget = state.budgets.languages_push;
435 let catalog = Arc::clone(&state.catalog);
436 let repo_label = event_repo.as_str().to_string();
437 if let Err(error) = run_blocking(move || -> Result<(), XrpcError> {
438 let repo = open(&layout, &event_repo)?;
439 let update = RefUpdate::Update {
440 name: event_ref,
441 old,
442 new,
443 };
444 let post_actor = Actor {
445 committer: actor,
446 owner,
447 repo: event_repo,
448 };
449 knot_postreceive::post_receive(
450 &repo,
451 &post_actor,
452 vec![(update, reservation)],
453 &Ci::Skip,
454 &knot_types::PushOptions::default(),
455 None,
456 languages_push_budget,
457 &catalog.push,
458 );
459 Ok(())
460 })
461 .await
462 {
463 tracing::warn!(repo = %repo_label, %error, "post-receive after merge failed");
464 }
465 Ok(ok_empty())
466 }
467 }
468}
469
470fn unified_message(input: &MergeInput) -> String {
471 let message = input
472 .commit_message
473 .as_ref()
474 .map(|message| message.as_str().to_string())
475 .unwrap_or_default();
476 match input
477 .commit_body
478 .as_ref()
479 .map(|body| body.as_str())
480 .filter(|body| !body.is_empty())
481 {
482 Some(body) => format!("{message}\n\n{body}"),
483 None => message,
484 }
485}
486
487fn unified_author(input: &MergeInput) -> Option<MailAuthor> {
488 match (input.author_name.as_ref(), input.author_email.as_ref()) {
489 (Some(name), Some(email)) if !name.as_str().is_empty() && !email.as_str().is_empty() => {
490 Some(MailAuthor {
491 name: name.clone(),
492 email: email.clone(),
493 date: String::new(),
494 })
495 }
496 _ => None,
497 }
498}
499
500pub(crate) async fn merge_check<H: HttpTransport, C: Clock>(
501 State(state): State<Arc<XrpcState<H, C>>>,
502 body: Bytes,
503) -> Result<Response, XrpcError> {
504 let input: MergeCheckInput = decode(&body)?;
505 let repo_did = resolve_by_name(&state, &input.did, &input.name)?;
506 let refname = input.branch.head_ref();
507 let layout = state.layout.clone();
508 let max_patch_bytes = state.byte_limits.patch_decompressed.get();
509
510 let output = run_blocking(move || {
511 let specs = match parse_specs(input.patch.as_str(), String::new(), None, max_patch_bytes) {
512 Ok(specs) => specs,
513 Err(error) => return Ok(MergeCheckOutput::broken(error.to_string())),
514 };
515 let repo = open(&layout, &repo_did)?;
516 let tip = branch_tip(&repo, &refname)?;
517 match stage_all(&repo, tip, &specs) {
518 Ok(Ok(_)) => Ok(MergeCheckOutput::clean()),
519 Ok(Err(conflicts)) => Ok(MergeCheckOutput::conflicted(conflicts)),
520 Err(ApplyError::TooLarge) => {
521 Ok(MergeCheckOutput::broken(ApplyError::TooLarge.to_string()))
522 }
523 Err(ApplyError::Git(error)) => Err(error.into()),
524 }
525 })
526 .await?;
527
528 Ok(check_response(output))
529}
530
531fn check_response(output: MergeCheckOutput) -> Response {
532 (StatusCode::OK, Json(output)).into_response()
533}