dogfood
Jump to…
snowcorepublic2 bookmarks
snowattribute changes to their author, and index SSH pushesowzkxxuxzulu1mo
mainroot / crates / df-worker / src / index_push.rsBlameRaw
1//! The `IndexPush` job (spec §4).
2//!
3//! Turns pushed Git objects into the change index. Everything here must be
4//! **idempotent** — running it twice over the same repository produces the same
5//! rows — because `dogfood-admin reindex` runs it against live repositories
6//! every time indexing logic changes.
7
8use std::collections::{HashMap, HashSet};
9
10use anyhow::{Context, Result};
11use df_db::ids::new_id;
12use df_index::indexer::{self, ChangeState, IndexedCommit, StackEdge};
13use df_store::{DiffOpts, RepoId, RepoStore, RevId};
14use sqlx::PgPool;
15use uuid::Uuid;
16
17/// How far back to walk from each bookmark.
18///
19/// Bounded so indexing a repository with a very long history cannot run
20/// unboundedly. Older commits were indexed when they were pushed.
21const WALK_LIMIT: usize = 1000;
22
23pub struct Outcome {
24 pub changes_seen: usize,
25 pub revisions_added: usize,
26}
27
28/// Index everything reachable from a repository's bookmarks.
29pub async fn run(
30 db: &PgPool,
31 store: &dyn RepoStore,
32 repo_id: Uuid,
33 pushed_by: Option<Uuid>,
34) -> Result<Outcome> {
35 let sid = RepoId(repo_id);
36
37 let (default_bookmark,): (String,) =
38 sqlx::query_as("SELECT default_bookmark FROM repos WHERE id = $1")
39 .bind(repo_id)
40 .fetch_one(db)
41 .await
42 .context("loading repository")?;
43
44 // ── 1. walk commits reachable from every bookmark ────────────────────────
45 let bookmarks = store.bookmarks(sid).await.context("listing bookmarks")?;
46
47 let mut commits: HashMap<String, IndexedCommit> = HashMap::new();
48 for b in &bookmarks {
49 let revs = match store.log(sid, &b.target, WALK_LIMIT).await {
50 Ok(r) => r,
51 Err(e) => {
52 tracing::warn!(bookmark = %b.name, "walking bookmark failed: {e}");
53 continue;
54 }
55 };
56 for r in revs {
57 commits.entry(r.rev.as_str().to_string()).or_insert_with(|| {
58 IndexedCommit {
59 rev: r.rev.as_str().to_string(),
60 change_id: r.change_id.as_deref().and_then(df_index::ChangeId::parse),
61 parents: r.parents.iter().map(|p| p.as_str().to_string()).collect(),
62 author_name: r.author.name.clone(),
63 author_email: r.author.email.clone(),
64 authored_at: r.author.when,
65 message: r.message.clone(),
66 conflicted: r.conflicted,
67 conflict_sides: r.conflict_sides.clone(),
68 conflict_bases: r.conflict_bases.clone(),
69 }
70 });
71 }
72 }
73
74 // Persist the bookmarks themselves.
75 sync_bookmarks(db, repo_id, &bookmarks).await?;
76
77 // ── 2. which changes are already on the default bookmark? ────────────────
78 // A change reachable from the target has landed.
79 let mut merged: HashSet<String> = HashSet::new();
80 if let Some(target) = bookmarks.iter().find(|b| b.name == default_bookmark) {
81 if let Ok(revs) = store.log(sid, &target.target, WALK_LIMIT).await {
82 for r in revs {
83 if let Some(c) = r.change_id {
84 merged.insert(c);
85 }
86 }
87 }
88 }
89
90 // ── 3. upsert changes and revisions ──────────────────────────────────────
91 let commit_list: Vec<IndexedCommit> = commits.into_values().collect();
92
93 let mut change_rows: HashMap<String, Uuid> = HashMap::new();
94 let mut revisions_added = 0usize;
95 // One lookup per distinct author email rather than per commit: a thousand
96 // -commit walk is usually a handful of people.
97 let mut author_cache: HashMap<String, Option<Uuid>> = HashMap::new();
98
99 for c in &commit_list {
100 // jj commits carry their change id. Plain-git commits do not, and get a
101 // patch-derived synthetic identity instead so `git push` users are not
102 // second-class (spec §4).
103 let (change_id, synthetic) = match c.change_id.as_ref() {
104 Some(id) => (id.as_str().to_string(), false),
105 None => match synthetic_identity(store, sid, c).await {
106 Some(id) => (id, true),
107 None => {
108 tracing::warn!(rev = %c.rev, "could not derive a synthetic identity; skipping");
109 continue;
110 }
111 },
112 };
113
114 // The root change (all-z) is jj's synthetic root, not user work.
115 if change_id.bytes().all(|b| b == b'z') {
116 continue;
117 }
118
119 let author_user_id = resolve_author(db, &mut author_cache, &c.author_email).await;
120
121 let change_uuid = upsert_change(
122 db,
123 repo_id,
124 &change_id,
125 c,
126 &default_bookmark,
127 merged.contains(&change_id),
128 synthetic,
129 author_user_id,
130 )
131 .await?;
132
133 change_rows.insert(change_id.clone(), change_uuid);
134
135 // Step 6: rebase comment anchors onto the new revision. Done here,
136 // while the previous head is still known, rather than in a second pass
137 // that would have to reconstruct it (spec §4).
138 match insert_revision(db, change_uuid, c, pushed_by).await? {
139 Inserted::Existing => {}
140 Inserted::New { id, previous_head } => {
141 revisions_added += 1;
142
143 if let Some(prev) = previous_head {
144 match crate::anchors::rebase(
145 db, store, sid, change_uuid, &prev, &c.rev, id,
146 )
147 .await
148 {
149 Ok(summary) if summary.touched() > 0 => {
150 emit_event(
151 db,
152 repo_id,
153 // Genuinely nobody's action: the indexer
154 // re-anchored these, not a person.
155 None,
156 "comments.rebased",
157 change_uuid,
158 serde_json::json!({
159 "moved": summary.moved,
160 "outdated": summary.outdated,
161 "orphaned": summary.orphaned,
162 }),
163 )
164 .await;
165 }
166 Ok(_) => {}
167 // A failure here must not lose the revision that was
168 // just indexed. The comments keep their old anchors,
169 // which are stale but not wrong about anything.
170 Err(e) => tracing::error!(
171 change = %change_uuid,
172 "rebasing comment anchors failed: {e}"
173 ),
174 }
175 }
176
177 // A push is the *pusher's* action, so it is attributed to them
178 // rather than to the commit's author — those differ whenever
179 // somebody lands work written by someone else. The author is
180 // the fallback only when the transport did not tell us who
181 // pushed (an admin reindex, for one).
182 emit_event(
183 db,
184 repo_id,
185 pushed_by.or(author_user_id),
186 "change.pushed",
187 change_uuid,
188 serde_json::json!({ "rev": c.rev }),
189 )
190 .await;
191
192 if c.conflicted {
193 emit_event(
194 db,
195 repo_id,
196 pushed_by.or(author_user_id),
197 "change.conflicted",
198 change_uuid,
199 serde_json::json!({ "rev": c.rev }),
200 )
201 .await;
202 }
203 }
204 }
205 }
206
207 // ── 4. stack edges ───────────────────────────────────────────────────────
208 let edges = indexer::stack_edges(&commit_list, &merged);
209 sync_stack_edges(db, repo_id, &edges, &change_rows).await?;
210
211 // ── 5. notify open pages ─────────────────────────────────────────────────
212 if let Err(e) = sqlx::query("SELECT pg_notify('dogfood_repo', $1)")
213 .bind(repo_id.to_string())
214 .execute(db)
215 .await
216 {
217 tracing::warn!("NOTIFY failed: {e}");
218 }
219
220 Ok(Outcome {
221 changes_seen: change_rows.len(),
222 revisions_added,
223 })
224}
225
226/// Derive a patch-based identity for a commit with no jj change id.
227///
228/// Returns `None` when the diff cannot be read, in which case the commit is
229/// skipped rather than given an identity we cannot reproduce on a later run —
230/// an unstable identity is worse than no row, because it would split one change
231/// into a new one on every reindex.
232async fn synthetic_identity(
233 store: &dyn RepoStore,
234 repo: RepoId,
235 c: &IndexedCommit,
236) -> Option<String> {
237 // Bounded: a synthetic identity is not worth reading a pathological diff for.
238 let opts = DiffOpts {
239 context_lines: 0,
240 max_files: 1_000,
241 max_lines: 50_000,
242 };
243
244 let diff = match store
245 .diff_from_parent(repo, &RevId::from_stored(c.rev.clone()), opts)
246 .await
247 {
248 Ok(d) => d,
249 Err(e) => {
250 tracing::warn!(rev = %c.rev, "diffing for synthetic identity failed: {e}");
251 return None;
252 }
253 };
254
255 // A truncated diff would hash differently depending on where the limit fell,
256 // so refuse rather than mint an unstable id.
257 if diff.truncated {
258 tracing::warn!(rev = %c.rev, "diff was truncated; not deriving a synthetic identity");
259 return None;
260 }
261
262 let patch = crate::patch_id::canonical_patch(&diff);
263 Some(
264 df_index::synthetic_change_id(&df_index::PatchIdentity {
265 diff: &patch,
266 author_email: &c.author_email,
267 author_date: c.authored_at.timestamp(),
268 })
269 .as_str()
270 .to_owned(),
271 )
272}
273
274/// Resolve the Dogfood account that wrote a commit, by its author email.
275///
276/// Email is the only link a pushed commit carries back to an account — the
277/// commit knows nothing about Dogfood — so this is the same rule every forge
278/// uses. `users.email` is `citext`, so the comparison is case-insensitive in
279/// the database rather than here.
280///
281/// `None` is an ordinary outcome, not a failure: commits pushed by somebody
282/// with no account, or written under an email the account has not recorded,
283/// keep the name the commit gave them and simply do not link anywhere. Never
284/// falls back to the *pusher* — attributing Alice's commit to Bob because Bob
285/// pushed it would be worse than not linking at all.
286async fn resolve_author(
287 db: &PgPool,
288 cache: &mut HashMap<String, Option<Uuid>>,
289 author_email: &str,
290) -> Option<Uuid> {
291 let email = author_email.trim();
292 if email.is_empty() {
293 return None;
294 }
295
296 if let Some(hit) = cache.get(email) {
297 return *hit;
298 }
299
300 let found: Option<(Uuid,)> = sqlx::query_as("SELECT id FROM users WHERE email = $1")
301 .bind(email)
302 .fetch_optional(db)
303 .await
304 .unwrap_or_else(|e| {
305 // A lookup failure must not fail the whole index job; the change is
306 // still worth recording, just unattributed.
307 tracing::warn!("resolving a commit author failed: {e}");
308 None
309 });
310
311 let id = found.map(|(id,)| id);
312 cache.insert(email.to_string(), id);
313 id
314}
315
316/// Insert or update the `changes` row, returning its id.
317async fn upsert_change(
318 db: &PgPool,
319 repo_id: Uuid,
320 change_id: &str,
321 c: &IndexedCommit,
322 default_bookmark: &str,
323 on_target: bool,
324 synthetic: bool,
325 author_user_id: Option<Uuid>,
326) -> Result<Uuid> {
327 let existing: Option<(Uuid, ChangeStateSql)> = sqlx::query_as(
328 "SELECT id, state FROM changes WHERE repo_id = $1 AND change_id = $2",
329 )
330 .bind(repo_id)
331 .bind(change_id)
332 .fetch_optional(db)
333 .await?;
334
335 let title = c.summary();
336 let description = c.message.split_once('\n').map(|(_, r)| r.trim()).unwrap_or("");
337
338 if let Some((id, state)) = existing {
339 // Indexing must never overwrite an author's explicit draft/abandoned
340 // choice (spec §4, and the decided draft behaviour).
341 let next = indexer::next_state(state.into(), on_target);
342
343 sqlx::query(
344 // `COALESCE` on the author so a reindex *fills in* an attribution
345 // that could not be resolved before — the user has since recorded
346 // the email — without ever overwriting one that is already set,
347 // which would undo the web UI's explicit author on a created change.
348 "UPDATE changes
349 SET title = $2, description = $3, conflicted = $4,
350 state = $5::change_state,
351 author_user_id = COALESCE(author_user_id, $6),
352 merged_at = CASE WHEN $5 = 'merged' AND merged_at IS NULL
353 THEN now() ELSE merged_at END,
354 updated_at = now()
355 WHERE id = $1",
356 )
357 .bind(id)
358 .bind(title)
359 .bind(description)
360 .bind(c.conflicted)
361 .bind(state_str(next))
362 .bind(author_user_id)
363 .execute(db)
364 .await?;
365
366 return Ok(id);
367 }
368
369 // New change. Allocate a per-repo display number under a row lock so two
370 // concurrent index jobs cannot mint the same number.
371 let mut tx = db.begin().await?;
372
373 let (number,): (i64,) = sqlx::query_as(
374 "UPDATE repo_counters SET next_change = next_change + 1
375 WHERE repo_id = $1
376 RETURNING next_change - 1",
377 )
378 .bind(repo_id)
379 .fetch_one(&mut *tx)
380 .await
381 .context("allocating a change number (is repo_counters seeded?)")?;
382
383 let id = new_id();
384 let state = if on_target { "merged" } else { "open" };
385
386 // ON CONFLICT makes a concurrent insert of the same change harmless, which
387 // is what keeps the job idempotent.
388 let inserted: Option<(Uuid,)> = sqlx::query_as(
389 "INSERT INTO changes (id, repo_id, change_id, number, title, description,
390 state, conflicted, target_bookmark, synthetic,
391 author_user_id, merged_at)
392 VALUES ($1, $2, $3, $4, $5, $6, $7::change_state, $8, $9, $10, $11,
393 CASE WHEN $7 = 'merged' THEN now() ELSE NULL END)
394 ON CONFLICT (repo_id, change_id) DO NOTHING
395 RETURNING id",
396 )
397 .bind(id)
398 .bind(repo_id)
399 .bind(change_id)
400 .bind(number)
401 .bind(title)
402 .bind(description)
403 .bind(state)
404 .bind(c.conflicted)
405 .bind(default_bookmark)
406 .bind(synthetic)
407 .bind(author_user_id)
408 .fetch_optional(&mut *tx)
409 .await?;
410
411 tx.commit().await?;
412
413 match inserted {
414 Some((id,)) => {
415 // Timeline event for a newly seen change.
416 let _ = sqlx::query(
417 "INSERT INTO events (id, repo_id, actor_id, kind, subject_type, subject_id, payload)
418 VALUES ($1, $2, $3, 'change.opened', 'change', $4, $5)",
419 )
420 .bind(new_id())
421 .bind(repo_id)
422 .bind(author_user_id)
423 .bind(id)
424 .bind(serde_json::json!({ "change_id": change_id }))
425 .execute(db)
426 .await;
427 Ok(id)
428 }
429 None => {
430 // Lost the race; read back the winner.
431 let (id,): (Uuid,) =
432 sqlx::query_as("SELECT id FROM changes WHERE repo_id = $1 AND change_id = $2")
433 .bind(repo_id)
434 .bind(change_id)
435 .fetch_one(db)
436 .await?;
437 Ok(id)
438 }
439 }
440}
441
442/// What appending a revision did.
443enum Inserted {
444 /// Already present. The indexer is re-runnable, so this is the normal case
445 /// on a reindex and must have no side effects at all.
446 Existing,
447 New {
448 /// Row id of the revision just written.
449 id: Uuid,
450 /// The revision that was the change's head before this one, when there
451 /// was one. This is what comment anchors are rebased *from*.
452 previous_head: Option<String>,
453 },
454}
455
456/// Append a revision.
457async fn insert_revision(
458 db: &PgPool,
459 change_uuid: Uuid,
460 c: &IndexedCommit,
461 pushed_by: Option<Uuid>,
462) -> Result<Inserted> {
463 // Read before inserting: after the insert the new revision *is* the head.
464 let previous_head: Option<String> = sqlx::query_scalar(
465 "SELECT rev FROM revisions WHERE change_id_fk = $1 ORDER BY seq DESC LIMIT 1",
466 )
467 .bind(change_uuid)
468 .fetch_optional(db)
469 .await?;
470
471 let conflict_data = c.conflicted.then(|| {
472 serde_json::json!({
473 "sides": c.conflict_sides,
474 "bases": c.conflict_bases,
475 })
476 });
477
478 // seq is per-change and 1-based. Computing it from the existing max keeps
479 // re-runs stable: an already-present revision hits the conflict clause and
480 // does not consume a number.
481 let inserted: Option<(Uuid,)> = sqlx::query_as(
482 "INSERT INTO revisions (id, change_id_fk, rev, seq, parents, author_name,
483 author_email, authored_at, message, conflicted,
484 conflict_data, pushed_by)
485 SELECT $1, $2, $3,
486 COALESCE((SELECT MAX(seq) FROM revisions WHERE change_id_fk = $2), 0) + 1,
487 $4, $5, $6, $7, $8, $9, $10, $11
488 ON CONFLICT (change_id_fk, rev) DO NOTHING
489 RETURNING id",
490 )
491 .bind(new_id())
492 .bind(change_uuid)
493 .bind(&c.rev)
494 .bind(&c.parents)
495 .bind(&c.author_name)
496 .bind(&c.author_email)
497 .bind(c.authored_at)
498 .bind(&c.message)
499 .bind(c.conflicted)
500 .bind(conflict_data)
501 .bind(pushed_by)
502 .fetch_optional(db)
503 .await?;
504
505 let Some((rev_uuid,)) = inserted else {
506 return Ok(Inserted::Existing);
507 };
508
509 // The head is the most recently pushed revision of the change.
510 sqlx::query("UPDATE changes SET head_revision_id = $2, updated_at = now() WHERE id = $1")
511 .bind(change_uuid)
512 .bind(rev_uuid)
513 .execute(db)
514 .await?;
515
516 Ok(Inserted::New { id: rev_uuid, previous_head })
517}
518
519/// Write a timeline event (spec §4 step 8).
520///
521/// Best-effort and unauthored: these events describe what a *push* did, and the
522/// pusher is recorded on the revision itself. Attributing them to a user here
523/// would claim more than the indexer knows — a push can carry somebody else's
524/// commits.
525async fn emit_event(
526 db: &PgPool,
527 repo_id: Uuid,
528 actor: Option<Uuid>,
529 kind: &str,
530 change_uuid: Uuid,
531 payload: serde_json::Value,
532) {
533 if let Err(e) = sqlx::query(
534 "INSERT INTO events (id, repo_id, actor_id, kind, subject_type, subject_id, payload)
535 VALUES ($1, $2, $3, $4, 'change', $5, $6)",
536 )
537 .bind(new_id())
538 .bind(repo_id)
539 .bind(actor)
540 .bind(kind)
541 .bind(change_uuid)
542 .bind(payload)
543 .execute(db)
544 .await
545 {
546 tracing::error!(%kind, "writing a timeline event failed: {e}");
547 }
548}
549
550async fn sync_bookmarks(db: &PgPool, repo_id: Uuid, marks: &[df_store::Bookmark]) -> Result<()> {
551 for b in marks {
552 // `protected` is set for the default bookmark and otherwise left alone:
553 // it is an operator's choice, and re-asserting it on every push would
554 // undo an unprotect. Marking the default keeps the settings page in
555 // step with the hook, which protects it whether or not the flag is set.
556 sqlx::query(
557 "INSERT INTO bookmarks (repo_id, name, target, protected, updated_at)
558 SELECT $1, $2, $3, (r.default_bookmark = $2), now()
559 FROM repos r WHERE r.id = $1
560 ON CONFLICT (repo_id, name)
561 DO UPDATE SET target = EXCLUDED.target,
562 protected = bookmarks.protected OR EXCLUDED.protected,
563 updated_at = now()",
564 )
565 .bind(repo_id)
566 .bind(&b.name)
567 .bind(b.target.as_str())
568 .execute(db)
569 .await?;
570 }
571
572 // Drop bookmarks that no longer exist, so a deleted branch disappears.
573 let names: Vec<String> = marks.iter().map(|b| b.name.clone()).collect();
574 sqlx::query("DELETE FROM bookmarks WHERE repo_id = $1 AND NOT (name = ANY($2))")
575 .bind(repo_id)
576 .bind(&names)
577 .execute(db)
578 .await?;
579
580 Ok(())
581}
582
583/// Replace this repository's stack edges with the freshly computed set.
584///
585/// Deleting and reinserting is what makes the job idempotent: a rebase that
586/// removes an edge must actually remove it, which an insert-only pass cannot do.
587async fn sync_stack_edges(
588 db: &PgPool,
589 repo_id: Uuid,
590 edges: &HashSet<StackEdge>,
591 change_rows: &HashMap<String, Uuid>,
592) -> Result<()> {
593 let mut tx = db.begin().await?;
594
595 sqlx::query("DELETE FROM change_edges WHERE repo_id = $1")
596 .bind(repo_id)
597 .execute(&mut *tx)
598 .await?;
599
600 for e in edges {
601 let (Some(parent), Some(child)) = (change_rows.get(&e.parent), change_rows.get(&e.child))
602 else {
603 continue;
604 };
605 sqlx::query(
606 "INSERT INTO change_edges (repo_id, parent_change, child_change)
607 VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
608 )
609 .bind(repo_id)
610 .bind(parent)
611 .bind(child)
612 .execute(&mut *tx)
613 .await?;
614 }
615
616 tx.commit().await?;
617 Ok(())
618}
619
620// ─── enum bridging ───────────────────────────────────────────────────────────
621
622#[derive(sqlx::Type, Clone, Copy)]
623#[sqlx(type_name = "change_state", rename_all = "lowercase")]
624enum ChangeStateSql {
625 Draft,
626 Open,
627 Merged,
628 Abandoned,
629}
630
631impl From<ChangeStateSql> for ChangeState {
632 fn from(s: ChangeStateSql) -> Self {
633 match s {
634 ChangeStateSql::Draft => ChangeState::Draft,
635 ChangeStateSql::Open => ChangeState::Open,
636 ChangeStateSql::Merged => ChangeState::Merged,
637 ChangeStateSql::Abandoned => ChangeState::Abandoned,
638 }
639 }
640}
641
642fn state_str(s: ChangeState) -> &'static str {
643 match s {
644 ChangeState::Draft => "draft",
645 ChangeState::Open => "open",
646 ChangeState::Merged => "merged",
647 ChangeState::Abandoned => "abandoned",
648 }
649}

649 lines · Rust