steid
Sign in

@jamesgill /

steid

steid/src/application/push.rs
Rust13.1 KBCode·Blame·History·Raw
1//! Recording what a push did.
2//!
3//! **The event is recorded from the application, not from a `post-receive` hook script
4//! inside the bare repository.** The refs are snapshotted before the receive-pack RPC
5//! and again after it, and every ref whose id moved becomes an event. Two extra
6//! `for-each-ref` per push is the accepted cost.
7//!
8//! The alternative — a hook script in every repository — was considered and rejected
9//! *for now*: the application already knows the actor, whereas a script would have to be
10//! installed on repository creation **and on boot** (so that repositories created by an
11//! older version get it), kept in sync with the binary, and would still need a channel
12//! back into the process to reach the database. It is the right answer if the per-push
13//! cost ever matters; it is two problems where this is none.
14//!
15//! A snapshot diff has one property a hook does not: it reports what *actually* changed
16//! on disk. A hook is told what receive-pack intends, before another hook can reject it.
17
18use std::time::SystemTime;
19
20use crate::domain::{
21 Actor, DomainError, ObjectId, OrgName, PushEvent, PushEventId, RefName, RepoName,
22 repository::{MembershipRepository, OrgRepository, PushEventRepository, RepoRepository},
23};
24
25use super::{
26 access::{Stores, is_org_owner},
27 error::Result,
28 port::{GitQuery, GitQueryError},
29};
30
31/// Every ref and what it pointed at, at one moment.
32///
33/// A plain list rather than a map because it is produced by one port call and consumed
34/// by one comparison, and both are small.
35pub type RefSnapshot = Vec<(RefName, ObjectId)>;
36
37/// One ref that moved between two snapshots.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct RefChange {
40 pub name: RefName,
41 /// The zero id when the ref was created.
42 pub before: ObjectId,
43 /// The zero id when the ref was deleted.
44 pub after: ObjectId,
45}
46
47/// Reads every ref and the object it names.
48///
49/// Fails soft nowhere: a snapshot that could not be taken is a real failure, and the
50/// caller decides whether to give up on recording rather than pretending the repository
51/// had no refs — which would turn a failed read into a flood of spurious "created"
52/// events on the next push.
53pub async fn snapshot_refs(
54 handle: &OrgName,
55 name: &RepoName,
56 git: &impl GitQuery,
57) -> std::result::Result<RefSnapshot, GitQueryError> {
58 git.ref_snapshot(handle, name).await
59}
60
61/// What changed between two snapshots, ordered by ref name.
62///
63/// Pure, so the rule that decides what counts as a change is testable without a
64/// repository. Ordered so a `git push --all` records its events in a stable order
65/// rather than whatever git listed.
66pub fn ref_changes(
67 before: &[(RefName, ObjectId)],
68 after: &[(RefName, ObjectId)],
69) -> Vec<RefChange> {
70 let zero = PushEvent::zero();
71 let find = |snapshot: &[(RefName, ObjectId)], wanted: &RefName| {
72 snapshot
73 .iter()
74 .find(|(name, _)| name == wanted)
75 .map(|(_, id)| id.clone())
76 };
77
78 let mut changes = Vec::new();
79
80 // Moved and deleted refs come from walking `before`; created ones from walking
81 // `after` for names that were not there. Two passes rather than a merged sort,
82 // because a snapshot is tens of entries and clarity is worth more here.
83 for (name, was) in before {
84 let now = find(after, name).unwrap_or_else(|| zero.clone());
85
86 if &now != was {
87 changes.push(RefChange {
88 name: name.clone(),
89 before: was.clone(),
90 after: now,
91 });
92 }
93 }
94
95 for (name, now) in after {
96 if find(before, name).is_none() {
97 changes.push(RefChange {
98 name: name.clone(),
99 before: zero.clone(),
100 after: now.clone(),
101 });
102 }
103 }
104
105 changes.sort_by(|a, b| a.name.as_str().cmp(b.name.as_str()));
106 changes
107}
108
109/// Records one event per ref the push moved.
110///
111/// Takes an `Actor` and re-checks that they own the repository, even though nothing
112/// reaches here without `serve_git` having already said so. Two reasons: the rule lives
113/// in one place regardless of transport, and this runs in a task spawned *after* the
114/// response has begun streaming — far enough from the check that repeating it is
115/// cheaper than reasoning about whether it still holds.
116///
117/// An absent or invisible repository records nothing and is not an error: a repository
118/// can be deleted while a push to it is in flight, and there is nothing useful to do
119/// about that.
120#[allow(clippy::too_many_arguments)]
121pub async fn record_push(
122 actor: &Actor,
123 handle: &OrgName,
124 name: &RepoName,
125 before: &[(RefName, ObjectId)],
126 after: &[(RefName, ObjectId)],
127 now: SystemTime,
128 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
129 events: &impl PushEventRepository,
130) -> Result<Vec<PushEvent>> {
131 let Some(org) = stores.orgs.find_by_name(handle).await? else {
132 return Ok(Vec::new());
133 };
134
135 let Some(repo) = stores.repos.find_by_org_and_name(&org.id, name).await? else {
136 return Ok(Vec::new());
137 };
138
139 if !is_org_owner(&org, actor, stores.memberships).await? {
140 return Err(DomainError::Forbidden.into());
141 }
142
143 let mut recorded = Vec::new();
144
145 for change in ref_changes(before, after) {
146 let event = PushEvent::new(
147 PushEventId::generate(),
148 repo.id.clone(),
149 change.name,
150 change.before,
151 change.after,
152 actor.user_id().cloned(),
153 now,
154 )?;
155
156 events.save(&event).await?;
157 recorded.push(event);
158 }
159
160 Ok(recorded)
161}
162
163#[cfg(test)]
164mod tests {
165 use super::*;
166 use crate::{
167 domain::{
168 Membership, MembershipId, OrgId, Organization, RepoId, Repository, Role, UserId,
169 Visibility,
170 },
171 infrastructure::repository::{
172 InMemoryMembershipRepo, InMemoryOrgRepo, InMemoryPushEventRepo, InMemoryRepoRepo,
173 },
174 };
175
176 fn oid(prefix: &str) -> ObjectId {
177 ObjectId::new(format!("{prefix:0>40}")).expect("valid id")
178 }
179
180 fn snapshot(refs: &[(&str, &str)]) -> RefSnapshot {
181 refs.iter()
182 .map(|(name, id)| (RefName::from_trusted(*name), oid(id)))
183 .collect()
184 }
185
186 // --- what counts as a change ---------------------------------------------------
187
188 #[test]
189 fn nothing_moving_is_no_change() {
190 let refs = snapshot(&[("refs/heads/main", "a"), ("refs/tags/v1", "b")]);
191
192 assert_eq!(ref_changes(&refs, &refs), Vec::new());
193 }
194
195 #[test]
196 fn a_moved_ref_carries_both_ends() {
197 let changes = ref_changes(
198 &snapshot(&[("refs/heads/main", "a")]),
199 &snapshot(&[("refs/heads/main", "b")]),
200 );
201
202 assert_eq!(changes.len(), 1);
203 assert_eq!(changes[0].before, oid("a"));
204 assert_eq!(changes[0].after, oid("b"));
205 }
206
207 #[test]
208 fn a_created_ref_has_the_zero_id_before_it() {
209 let changes = ref_changes(&[], &snapshot(&[("refs/heads/main", "a")]));
210
211 assert_eq!(changes.len(), 1);
212 assert_eq!(changes[0].before, PushEvent::zero());
213 assert_eq!(changes[0].after, oid("a"));
214 }
215
216 #[test]
217 fn a_deleted_ref_has_the_zero_id_after_it() {
218 let changes = ref_changes(&snapshot(&[("refs/heads/gone", "a")]), &[]);
219
220 assert_eq!(changes.len(), 1);
221 assert_eq!(changes[0].before, oid("a"));
222 assert_eq!(changes[0].after, PushEvent::zero());
223 }
224
225 #[test]
226 fn a_push_of_several_refs_is_reported_in_a_stable_order() {
227 // `git push --all` moves everything at once, and a listing that reshuffled per
228 // push would be unreadable.
229 let changes = ref_changes(
230 &snapshot(&[("refs/heads/main", "a"), ("refs/heads/old", "c")]),
231 &snapshot(&[
232 ("refs/heads/main", "b"),
233 ("refs/heads/feat", "d"),
234 ("refs/tags/v1", "e"),
235 ]),
236 );
237
238 assert_eq!(
239 changes
240 .iter()
241 .map(|change| change.name.as_str())
242 .collect::<Vec<_>>(),
243 [
244 "refs/heads/feat",
245 "refs/heads/main",
246 "refs/heads/old",
247 "refs/tags/v1"
248 ]
249 );
250 }
251
252 // --- recording -----------------------------------------------------------------
253
254 struct Fixture {
255 orgs: InMemoryOrgRepo,
256 memberships: InMemoryMembershipRepo,
257 repos: InMemoryRepoRepo,
258 events: InMemoryPushEventRepo,
259 owner: Actor,
260 member: Actor,
261 handle: OrgName,
262 repo_id: RepoId,
263 }
264
265 async fn fixture() -> Fixture {
266 let orgs = InMemoryOrgRepo::new();
267 let memberships = InMemoryMembershipRepo::new();
268 let repos = InMemoryRepoRepo::new();
269
270 let org = Organization::new(OrgId::generate(), "acme", None).expect("valid org");
271 orgs.save(&org).await.expect("save org");
272
273 let owner = UserId::generate();
274 let member = UserId::generate();
275
276 for (user, role) in [(&owner, Role::Owner), (&member, Role::Member)] {
277 memberships
278 .save(&Membership::new(
279 MembershipId::generate(),
280 org.id.clone(),
281 user.clone(),
282 role,
283 ))
284 .await
285 .expect("save membership");
286 }
287
288 let repo = Repository::new(
289 RepoId::generate(),
290 org.id.clone(),
291 "steid",
292 None,
293 Visibility::Public,
294 SystemTime::UNIX_EPOCH,
295 )
296 .expect("valid repo");
297 repos.save(&repo).await.expect("save repo");
298
299 Fixture {
300 orgs,
301 memberships,
302 repos,
303 events: InMemoryPushEventRepo::new(),
304 owner: Actor::User(owner),
305 member: Actor::User(member),
306 handle: org.name,
307 repo_id: repo.id,
308 }
309 }
310
311 impl Fixture {
312 async fn record(
313 &self,
314 actor: &Actor,
315 name: &str,
316 before: &[(&str, &str)],
317 after: &[(&str, &str)],
318 ) -> Result<Vec<PushEvent>> {
319 record_push(
320 actor,
321 &self.handle,
322 &RepoName::new(name).expect("valid name"),
323 &snapshot(before),
324 &snapshot(after),
325 SystemTime::UNIX_EPOCH,
326 &Stores::new(&self.orgs, &self.memberships, &self.repos),
327 &self.events,
328 )
329 .await
330 }
331 }
332
333 #[tokio::test]
334 async fn a_push_records_one_event_per_ref_that_moved() {
335 let f = fixture().await;
336
337 let recorded = f
338 .record(
339 &f.owner,
340 "steid",
341 &[("refs/heads/main", "a")],
342 &[("refs/heads/main", "b"), ("refs/tags/v1", "c")],
343 )
344 .await
345 .expect("should record");
346
347 assert_eq!(recorded.len(), 2);
348
349 let stored = f
350 .events
351 .list_by_repo(&f.repo_id, 10)
352 .await
353 .expect("should list");
354 assert_eq!(stored.len(), 2);
355 assert!(
356 stored
357 .iter()
358 .all(|event| event.pusher == f.owner.user_id().cloned())
359 );
360 }
361
362 #[tokio::test]
363 async fn a_push_that_moved_nothing_records_nothing() {
364 // A `git push` with nothing to send still runs the RPC.
365 let f = fixture().await;
366
367 let recorded = f
368 .record(
369 &f.owner,
370 "steid",
371 &[("refs/heads/main", "a")],
372 &[("refs/heads/main", "a")],
373 )
374 .await
375 .expect("should record");
376
377 assert!(recorded.is_empty());
378 assert!(
379 f.events
380 .list_by_repo(&f.repo_id, 10)
381 .await
382 .expect("list")
383 .is_empty()
384 );
385 }
386
387 #[tokio::test]
388 async fn only_an_owner_can_have_a_push_recorded_for_them() {
389 // Repeated here rather than trusted from `serve_git`, because this runs in a
390 // task spawned after the response has begun streaming.
391 let f = fixture().await;
392
393 for actor in [&f.member, &Actor::Anonymous] {
394 let error = f
395 .record(actor, "steid", &[], &[("refs/heads/main", "a")])
396 .await
397 .expect_err("should refuse");
398
399 assert!(matches!(
400 error,
401 super::super::Error::Domain(DomainError::Forbidden)
402 ));
403 }
404
405 assert!(
406 f.events
407 .list_by_repo(&f.repo_id, 10)
408 .await
409 .expect("list")
410 .is_empty()
411 );
412 }
413
414 #[tokio::test]
415 async fn a_repository_deleted_mid_push_records_nothing_rather_than_failing() {
416 let f = fixture().await;
417
418 let recorded = f
419 .record(&f.owner, "gone", &[], &[("refs/heads/main", "a")])
420 .await
421 .expect("should not fail");
422
423 assert!(recorded.is_empty());
424 }
425}