| 1 | //! What sessions touched, and the issues and pull requests they worked on. |
| 2 | //! |
| 3 | //! A session is tied to an issue or pull request in as many ways as it |
| 4 | //! was (`LinkVia`): attached when it started or updated, a claim it made, |
| 5 | //! a pull request it opened or a branch or commit of one, a |
| 6 | //! `G1t-Session:` trailer on a commit pushed to one, and `#N` in its |
| 7 | //! prompt, its task or a subagent's. Each way is a row of |
| 8 | //! `session_links`; reading a session gathers its whole tree's |
| 9 | //! (`gather_links`). Files two sessions touched make them overlap |
| 10 | //! (`overlap.detected`, once per pair). |
| 11 | |
| 12 | use std::collections::{HashMap, HashSet}; |
| 13 | |
| 14 | use g1t_contracts::Outcome; |
| 15 | use g1t_contracts::repos::{Commit, LogArgs, RepoPath}; |
| 16 | use g1t_contracts::time::rfc3339; |
| 17 | use g1t_contracts::work::*; |
| 18 | use g1t_kit::now_ms; |
| 19 | use serde::Deserialize; |
| 20 | use worker::Result; |
| 21 | |
| 22 | use super::{SESSION_COLUMNS, SessionRow}; |
| 23 | use crate::Work; |
| 24 | |
| 25 | /// The most `#N` of one text looked up. |
| 26 | const MAX_REFS: usize = 10; |
| 27 | /// The most issues and pull requests a session's links name. |
| 28 | const MAX_LINKED: usize = 40; |
| 29 | /// The most commits of one push read for trailers. |
| 30 | const MAX_TRAILER_COMMITS: u32 = 20; |
| 31 | /// The most subagents a tree shows. |
| 32 | pub(crate) const MAX_TREE: u32 = 200; |
| 33 | |
| 34 | /// Ties session `?1` of repository `?2` to issue or pull request (`?3`) |
| 35 | /// `?4` by way of `?5`, at `?6`; nothing when it is tied that way already. |
| 36 | pub(crate) const INSERT_LINK: &str = "INSERT INTO session_links (session_id, repo_id, kind, number, via, at) |
| 37 | VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT DO NOTHING"; |
| 38 | |
| 39 | /// Records that session `?1` of repository `?2` touched (`?3`) `?4`, at |
| 40 | /// `?5`; returns it only the first time. |
| 41 | pub(crate) const INSERT_TOUCH: &str = "INSERT INTO session_touches (session_id, repo_id, kind, value, at) |
| 42 | VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT DO NOTHING RETURNING kind, value"; |
| 43 | |
| 44 | /// Session `?1`'s subagents, and theirs, in the order they started. |
| 45 | pub(crate) fn tree_sql() -> String { |
| 46 | format!( |
| 47 | "WITH RECURSIVE tree(id) AS ( |
| 48 | SELECT id FROM sessions WHERE parent_id = ?1 |
| 49 | UNION SELECT s.id FROM sessions s JOIN tree ON s.parent_id = tree.id) |
| 50 | SELECT {SESSION_COLUMNS} FROM sessions WHERE id IN (SELECT id FROM tree) |
| 51 | ORDER BY started_at, id LIMIT {MAX_TREE}" |
| 52 | ) |
| 53 | } |
| 54 | |
| 55 | /// Ends session `?1`'s subagents still going, and theirs, at `?2` for |
| 56 | /// `?3`. |
| 57 | pub(crate) fn end_tree_sql() -> String { |
| 58 | format!( |
| 59 | "WITH RECURSIVE tree(id) AS ( |
| 60 | SELECT id FROM sessions WHERE parent_id = ?1 |
| 61 | UNION SELECT s.id FROM sessions s JOIN tree ON s.parent_id = tree.id) |
| 62 | UPDATE sessions SET state = 'ended', ended_at = ?2, end_reason = ?3, last_seen_at = ?2 |
| 63 | WHERE id IN (SELECT id FROM tree) AND state = 'active' |
| 64 | RETURNING {SESSION_COLUMNS}" |
| 65 | ) |
| 66 | } |
| 67 | |
| 68 | /// Session `?1`'s line: its parents up to the top, and its subagents and |
| 69 | /// theirs. Never told of overlapping with each other. |
| 70 | pub(crate) const FAMILY: &str = "WITH RECURSIVE |
| 71 | up(id, parent) AS ( |
| 72 | SELECT id, parent_id FROM sessions WHERE id = ?1 |
| 73 | UNION SELECT s.id, s.parent_id FROM sessions s JOIN up ON s.id = up.parent), |
| 74 | down(id) AS ( |
| 75 | SELECT id FROM sessions WHERE parent_id = ?1 |
| 76 | UNION SELECT s.id FROM sessions s JOIN down ON s.parent_id = down.id) |
| 77 | SELECT id FROM up UNION SELECT id FROM down"; |
| 78 | |
| 79 | /// The files of `?2` (JSON) that active sessions of repository `?1` other |
| 80 | /// than `?3` touched. |
| 81 | pub(crate) const TOUCHED_BY_OTHERS: &str = "SELECT t.session_id AS session_id, t.value AS value |
| 82 | FROM session_touches t JOIN sessions s ON s.id = t.session_id |
| 83 | WHERE t.repo_id = ?1 AND t.kind = 'file' AND t.value IN (SELECT value FROM json_each(?2)) |
| 84 | AND t.session_id <> ?3 AND s.state = 'active'"; |
| 85 | |
| 86 | /// The sessions of repository `?1` that pushed the branch of one of the |
| 87 | /// pull requests `?2` (JSON numbers). |
| 88 | pub(crate) const BRANCH_SESSIONS: &str = "SELECT DISTINCT t.session_id AS id FROM session_touches t |
| 89 | JOIN pulls p ON p.repo_id = t.repo_id AND p.source_branch = t.value AND p.fork_repo_id IS NULL |
| 90 | WHERE t.repo_id = ?1 AND t.kind = 'branch' AND p.number IN (SELECT value FROM json_each(?2)) LIMIT 200"; |
| 91 | |
| 92 | /// Announces the pair `?2`, `?3` of repository `?1` as overlapping on |
| 93 | /// `?4` (JSON); returns it only the first time. |
| 94 | pub(crate) const INSERT_OVERLAP: &str = "INSERT INTO session_overlaps (repo_id, low, high, files, noticed_at) |
| 95 | VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT DO NOTHING RETURNING low"; |
| 96 | |
| 97 | #[derive(Deserialize)] |
| 98 | struct IdRow { |
| 99 | id: String, |
| 100 | } |
| 101 | |
| 102 | #[derive(Deserialize)] |
| 103 | struct NumberOnly { |
| 104 | number: u32, |
| 105 | } |
| 106 | |
| 107 | #[derive(Deserialize)] |
| 108 | pub(crate) struct LinkRow { |
| 109 | session_id: String, |
| 110 | kind: String, |
| 111 | number: u32, |
| 112 | via: String, |
| 113 | } |
| 114 | |
| 115 | #[derive(Deserialize)] |
| 116 | pub(crate) struct TouchRow { |
| 117 | session_id: String, |
| 118 | kind: String, |
| 119 | value: String, |
| 120 | at: String, |
| 121 | } |
| 122 | |
| 123 | #[derive(Deserialize)] |
| 124 | struct PairRow { |
| 125 | session_id: String, |
| 126 | value: String, |
| 127 | } |
| 128 | |
| 129 | #[derive(Deserialize)] |
| 130 | struct SkillRow { |
| 131 | session_id: String, |
| 132 | name: String, |
| 133 | uses: u32, |
| 134 | } |
| 135 | |
| 136 | fn kind_name(kind: LinkKind) -> &'static str { |
| 137 | match kind { |
| 138 | LinkKind::Issue => "issue", |
| 139 | LinkKind::Pull => "pull", |
| 140 | } |
| 141 | } |
| 142 | |
| 143 | impl Work { |
| 144 | /// Ties `session_id` to an issue or pull request one way. Never fails |
| 145 | /// what it follows: a problem is logged. |
| 146 | pub(crate) async fn link(&self, session_id: &str, repo_id: &str, kind: LinkKind, number: u32, via: LinkVia) { |
| 147 | let at = rfc3339(now_ms()); |
| 148 | let linked = async { |
| 149 | self.db |
| 150 | .prepare(INSERT_LINK) |
| 151 | .bind(&[session_id.into(), repo_id.into(), kind_name(kind).into(), number.into(), via.as_str().into(), at.into()])? |
| 152 | .run() |
| 153 | .await |
| 154 | }; |
| 155 | if let Err(error) = linked.await { |
| 156 | worker::console_error!("session {session_id}: not tied to {} #{number}: {error}", kind_name(kind)); |
| 157 | } |
| 158 | } |
| 159 | |
| 160 | /// Ties a session to the issues and pull requests `#N` in `text` names |
| 161 | /// that the repository has. |
| 162 | pub(crate) async fn link_refs(&self, session_id: &str, repo_id: &str, text: &str) -> Result<()> { |
| 163 | for number in refs_in(text).into_iter().take(MAX_REFS) { |
| 164 | if self.issue(repo_id, number).await?.is_some() { |
| 165 | self.link(session_id, repo_id, LinkKind::Issue, number, LinkVia::Ref).await; |
| 166 | } else if self.pull(repo_id, number).await?.is_some() { |
| 167 | self.link(session_id, repo_id, LinkKind::Pull, number, LinkVia::Ref).await; |
| 168 | } |
| 169 | } |
| 170 | Ok(()) |
| 171 | } |
| 172 | |
| 173 | /// Ties a session to the pull requests of `branch`. |
| 174 | pub(crate) async fn link_branch(&self, session_id: &str, repo_id: &str, branch: &str) -> Result<()> { |
| 175 | let pulls = self |
| 176 | .db |
| 177 | .prepare("SELECT number FROM pulls WHERE repo_id = ? AND source_branch = ? AND fork_repo_id IS NULL") |
| 178 | .bind(&[repo_id.into(), branch.into()])? |
| 179 | .all() |
| 180 | .await? |
| 181 | .results::<NumberOnly>()?; |
| 182 | for pull in pulls { |
| 183 | self.link(session_id, repo_id, LinkKind::Pull, pull.number, LinkVia::Branch).await; |
| 184 | } |
| 185 | Ok(()) |
| 186 | } |
| 187 | |
| 188 | /// Ties a session to the pull requests whose head is `commit` (a full |
| 189 | /// or abbreviated hash). |
| 190 | pub(crate) async fn link_commit(&self, session_id: &str, repo_id: &str, commit: &str) -> Result<()> { |
| 191 | let pulls = self |
| 192 | .db |
| 193 | .prepare("SELECT number FROM pulls WHERE repo_id = ? AND head_commit IS NOT NULL AND substr(head_commit, 1, length(?2)) = ?2") |
| 194 | .bind(&[repo_id.into(), commit.into()])? |
| 195 | .all() |
| 196 | .await? |
| 197 | .results::<NumberOnly>()?; |
| 198 | for pull in pulls { |
| 199 | self.link(session_id, repo_id, LinkKind::Pull, pull.number, LinkVia::Commit).await; |
| 200 | } |
| 201 | Ok(()) |
| 202 | } |
| 203 | |
| 204 | /// Records what a session touched, ties it to what that names, and |
| 205 | /// tells it and the other sessions on the same files that they |
| 206 | /// overlap. Returns how many were new. |
| 207 | pub(crate) async fn record_touches(&self, session: &Session, touches: &[NewTouch]) -> Result<u32> { |
| 208 | let touches = tidy_touches(touches); |
| 209 | if touches.is_empty() { |
| 210 | return Ok(0); |
| 211 | } |
| 212 | let at = rfc3339(now_ms()); |
| 213 | let mut statements = Vec::with_capacity(touches.len()); |
| 214 | for touch in &touches { |
| 215 | statements.push(self.db.prepare(INSERT_TOUCH).bind(&[ |
| 216 | session.id.as_str().into(), |
| 217 | session.repo_id.as_str().into(), |
| 218 | touch.kind.as_str().into(), |
| 219 | touch.value.as_str().into(), |
| 220 | at.as_str().into(), |
| 221 | ])?); |
| 222 | } |
| 223 | let mut new: Vec<NewTouch> = Vec::new(); |
| 224 | for result in self.db.batch(statements).await? { |
| 225 | for row in result.results::<serde_json::Value>()? { |
| 226 | if let (Some(kind), Some(value)) = (row["kind"].as_str().and_then(TouchKind::parse), row["value"].as_str()) { |
| 227 | new.push(NewTouch { kind, value: value.to_owned() }); |
| 228 | } |
| 229 | } |
| 230 | } |
| 231 | let mut files = Vec::new(); |
| 232 | for touch in &new { |
| 233 | let number = touch.value.parse::<u32>().ok(); |
| 234 | match (touch.kind, number) { |
| 235 | (TouchKind::Pull, Some(number)) => self.link(&session.id, &session.repo_id, LinkKind::Pull, number, LinkVia::Opened).await, |
| 236 | (TouchKind::Issue, Some(number)) => { |
| 237 | if self.issue(&session.repo_id, number).await?.is_some() { |
| 238 | self.link(&session.id, &session.repo_id, LinkKind::Issue, number, LinkVia::Ref).await; |
| 239 | } |
| 240 | } |
| 241 | (TouchKind::Branch, _) => self.link_branch(&session.id, &session.repo_id, &touch.value).await?, |
| 242 | (TouchKind::Commit, _) => self.link_commit(&session.id, &session.repo_id, &touch.value).await?, |
| 243 | (TouchKind::File, _) => files.push(touch.value.clone()), |
| 244 | _ => {} |
| 245 | } |
| 246 | } |
| 247 | if !files.is_empty() { |
| 248 | self.announce_session_overlaps(session, &files).await?; |
| 249 | } |
| 250 | Ok(new.len() as u32) |
| 251 | } |
| 252 | |
| 253 | /// The ids of a session's line: its parents and its subagents. |
| 254 | pub(crate) async fn family(&self, session_id: &str) -> Result<HashSet<String>> { |
| 255 | Ok(self |
| 256 | .db |
| 257 | .prepare(FAMILY) |
| 258 | .bind(&[session_id.into()])? |
| 259 | .all() |
| 260 | .await? |
| 261 | .results::<IdRow>()? |
| 262 | .into_iter() |
| 263 | .map(|row| row.id) |
| 264 | .collect()) |
| 265 | } |
| 266 | |
| 267 | /// Tells `session` and each other active session that touched one of |
| 268 | /// `files` (outside its own line) that they overlap, once per pair. |
| 269 | async fn announce_session_overlaps(&self, session: &Session, files: &[String]) -> Result<()> { |
| 270 | let touched: Vec<(String, String)> = self |
| 271 | .db |
| 272 | .prepare(TOUCHED_BY_OTHERS) |
| 273 | .bind(&[session.repo_id.as_str().into(), serde_json::to_string(files)?.into(), session.id.as_str().into()])? |
| 274 | .all() |
| 275 | .await? |
| 276 | .results::<PairRow>()? |
| 277 | .into_iter() |
| 278 | .map(|row| (row.session_id, row.value)) |
| 279 | .collect(); |
| 280 | if touched.is_empty() { |
| 281 | return Ok(()); |
| 282 | } |
| 283 | let family = self.family(&session.id).await?; |
| 284 | let at = rfc3339(now_ms()); |
| 285 | for (other_id, files) in session_overlaps(&session.id, &family, &touched) { |
| 286 | let (low, high) = session_pair(&session.id, &other_id); |
| 287 | let new = self |
| 288 | .db |
| 289 | .prepare(INSERT_OVERLAP) |
| 290 | .bind(&[session.repo_id.as_str().into(), low.into(), high.into(), serde_json::to_string(&files)?.into(), at.as_str().into()])? |
| 291 | .first::<serde_json::Value>(None) |
| 292 | .await?; |
| 293 | if new.is_none() { |
| 294 | continue; |
| 295 | } |
| 296 | let Some(other) = self.session_row(&session.repo_id, &other_id).await? else { |
| 297 | continue; |
| 298 | }; |
| 299 | let other = other.at(now_ms()); |
| 300 | let ours = SessionOverlapDetected { |
| 301 | repo_id: session.repo_id.clone(), |
| 302 | number: session.pull.or(session.issue), |
| 303 | session_id: session.id.clone(), |
| 304 | other_session_id: other.id.clone(), |
| 305 | other_actor: other.actor.username.clone(), |
| 306 | files: files.clone(), |
| 307 | issue: session.issue, |
| 308 | other_issue: other.issue, |
| 309 | }; |
| 310 | let theirs = SessionOverlapDetected { |
| 311 | number: other.pull.or(other.issue), |
| 312 | session_id: other.id.clone(), |
| 313 | other_session_id: session.id.clone(), |
| 314 | other_actor: session.actor.username.clone(), |
| 315 | issue: other.issue, |
| 316 | other_issue: session.issue, |
| 317 | ..ours.clone() |
| 318 | }; |
| 319 | self.publish_as("overlap.detected", &session.repo_id, Some(session.actor.id.clone()), ours).await?; |
| 320 | self.publish_as("overlap.detected", &session.repo_id, Some(session.actor.id.clone()), theirs).await?; |
| 321 | } |
| 322 | Ok(()) |
| 323 | } |
| 324 | |
| 325 | /// A session's subagents, and theirs, in the order they started, each |
| 326 | /// with how many it started. |
| 327 | pub(crate) async fn subagents_of(&self, session_id: &str, now: u64) -> Result<Vec<Session>> { |
| 328 | let rows = self.db.prepare(tree_sql()).bind(&[session_id.into()])?.all().await?.results::<SessionRow>()?; |
| 329 | let mut children: Vec<Session> = rows.into_iter().map(|row| row.at(now)).collect(); |
| 330 | count_subagents(&mut children); |
| 331 | Ok(children) |
| 332 | } |
| 333 | |
| 334 | /// How many subagents each of `sessions` started, filled in. |
| 335 | pub(crate) async fn fill_subagents(&self, sessions: &mut [Session]) -> Result<()> { |
| 336 | if sessions.is_empty() { |
| 337 | return Ok(()); |
| 338 | } |
| 339 | let ids: Vec<&str> = sessions.iter().map(|session| session.id.as_str()).collect(); |
| 340 | #[derive(Deserialize)] |
| 341 | struct Count { |
| 342 | parent_id: String, |
| 343 | n: u32, |
| 344 | } |
| 345 | let counts = self |
| 346 | .db |
| 347 | .prepare("SELECT parent_id, COUNT(*) AS n FROM sessions WHERE parent_id IN (SELECT value FROM json_each(?)) GROUP BY parent_id") |
| 348 | .bind(&[serde_json::to_string(&ids)?.into()])? |
| 349 | .all() |
| 350 | .await? |
| 351 | .results::<Count>()?; |
| 352 | let counts: HashMap<String, u32> = counts.into_iter().map(|count| (count.parent_id, count.n)).collect(); |
| 353 | for session in sessions { |
| 354 | session.subagents = counts.get(&session.id).copied().unwrap_or_default(); |
| 355 | } |
| 356 | Ok(()) |
| 357 | } |
| 358 | |
| 359 | /// What a tree of sessions touched, its links, and the skills it used. |
| 360 | pub(crate) async fn tree_facts(&self, repo_id: &str, ids: &[String]) -> Result<(Vec<SessionTouch>, Vec<SessionLink>, Vec<SessionSkill>)> { |
| 361 | let ids_json = serde_json::to_string(ids)?; |
| 362 | let touches = self |
| 363 | .db |
| 364 | .prepare( |
| 365 | "SELECT session_id, kind, value, at FROM session_touches |
| 366 | WHERE session_id IN (SELECT value FROM json_each(?)) ORDER BY at, rowid LIMIT 1000", |
| 367 | ) |
| 368 | .bind(&[ids_json.as_str().into()])? |
| 369 | .all() |
| 370 | .await? |
| 371 | .results::<TouchRow>()?; |
| 372 | // A pull request opened on a branch after the session pushed it. |
| 373 | let branches: HashSet<(String, String)> = touches |
| 374 | .iter() |
| 375 | .filter(|touch| touch.kind == "branch") |
| 376 | .map(|touch| (touch.session_id.clone(), touch.value.clone())) |
| 377 | .collect(); |
| 378 | for (session_id, branch) in branches.iter().take(20) { |
| 379 | self.link_branch(session_id, repo_id, branch).await?; |
| 380 | } |
| 381 | let records: Vec<LinkRecord> = self |
| 382 | .db |
| 383 | .prepare( |
| 384 | "SELECT session_id, kind, number, via FROM session_links |
| 385 | WHERE session_id IN (SELECT value FROM json_each(?)) ORDER BY at", |
| 386 | ) |
| 387 | .bind(&[ids_json.as_str().into()])? |
| 388 | .all() |
| 389 | .await? |
| 390 | .results::<LinkRow>()? |
| 391 | .into_iter() |
| 392 | .filter_map(|row| { |
| 393 | let kind = match row.kind.as_str() { |
| 394 | "issue" => LinkKind::Issue, |
| 395 | "pull" => LinkKind::Pull, |
| 396 | _ => return None, |
| 397 | }; |
| 398 | Some(LinkRecord { session_id: row.session_id, kind, number: row.number, via: LinkVia::parse(&row.via)? }) |
| 399 | }) |
| 400 | .collect(); |
| 401 | let mut known = HashMap::new(); |
| 402 | let mut wanted: Vec<(LinkKind, u32)> = Vec::new(); |
| 403 | for record in &records { |
| 404 | if !wanted.contains(&(record.kind, record.number)) { |
| 405 | wanted.push((record.kind, record.number)); |
| 406 | } |
| 407 | } |
| 408 | for (kind, number) in wanted.into_iter().take(MAX_LINKED) { |
| 409 | let found = match kind { |
| 410 | LinkKind::Issue => self.issue(repo_id, number).await?.map(|issue| (issue.title, state_name(&issue.state))), |
| 411 | LinkKind::Pull => self.pull(repo_id, number).await?.map(|pull| (pull.title, state_name(&pull.status))), |
| 412 | }; |
| 413 | if let Some(found) = found { |
| 414 | known.insert((kind, number), found); |
| 415 | } |
| 416 | } |
| 417 | let links = gather_links(&records, &known); |
| 418 | let skills = self |
| 419 | .db |
| 420 | .prepare( |
| 421 | "SELECT session_id, text AS name, COUNT(*) AS uses FROM session_entries |
| 422 | WHERE session_id IN (SELECT value FROM json_each(?)) AND kind = 'skill' |
| 423 | GROUP BY session_id, text ORDER BY MIN(session_seq)", |
| 424 | ) |
| 425 | .bind(&[ids_json.as_str().into()])? |
| 426 | .all() |
| 427 | .await? |
| 428 | .results::<SkillRow>()? |
| 429 | .into_iter() |
| 430 | .map(|row| SessionSkill { session_id: row.session_id, name: row.name, uses: row.uses }) |
| 431 | .collect(); |
| 432 | let touched = touches |
| 433 | .into_iter() |
| 434 | .filter_map(|row| { |
| 435 | Some(SessionTouch { session_id: row.session_id, kind: TouchKind::parse(&row.kind)?, value: row.value, at: row.at }) |
| 436 | }) |
| 437 | .collect(); |
| 438 | Ok((touched, links, skills)) |
| 439 | } |
| 440 | |
| 441 | /// After a push moved `pulls` on `branch` to `after`: the sessions that |
| 442 | /// pushed that branch, and those its new commits name in |
| 443 | /// `G1t-Session:` trailers, are tied to them. Never fails the push's |
| 444 | /// handling: a problem is logged. |
| 445 | pub(crate) async fn link_push(&self, repo_id: &str, branch: &str, after: &str, commits: Option<u32>, pulls: &[Pull]) { |
| 446 | if pulls.is_empty() { |
| 447 | return; |
| 448 | } |
| 449 | let linked = async { |
| 450 | let pushed_by = self |
| 451 | .db |
| 452 | .prepare("SELECT DISTINCT session_id AS id FROM session_touches WHERE repo_id = ? AND kind = 'branch' AND value = ?") |
| 453 | .bind(&[repo_id.into(), branch.into()])? |
| 454 | .all() |
| 455 | .await? |
| 456 | .results::<IdRow>()?; |
| 457 | for session in &pushed_by { |
| 458 | for pull in pulls { |
| 459 | self.link(&session.id, repo_id, LinkKind::Pull, pull.number, LinkVia::Branch).await; |
| 460 | } |
| 461 | } |
| 462 | // Trailers are read only while someone records sessions here. |
| 463 | let recording = self |
| 464 | .db |
| 465 | .prepare("SELECT id FROM sessions WHERE repo_id = ? AND state = 'active' LIMIT 1") |
| 466 | .bind(&[repo_id.into()])? |
| 467 | .first::<IdRow>(None) |
| 468 | .await?; |
| 469 | if recording.is_none() { |
| 470 | return Ok::<(), worker::Error>(()); |
| 471 | } |
| 472 | let Some(pull) = pulls.first() else { return Ok(()) }; |
| 473 | let path: Option<RepoPath> = |
| 474 | g1t_kit::call(&self.repos, "path_by_id", &g1t_contracts::repos::PathByIdArgs { id: repo_id.to_owned() }).await?; |
| 475 | let Some(path) = path else { return Ok(()) }; |
| 476 | let log = LogArgs { |
| 477 | path, |
| 478 | viewer: self.owner_viewer(pull).await?, |
| 479 | git_ref: Some(after.to_owned()), |
| 480 | limit: commits.unwrap_or(MAX_TRAILER_COMMITS).clamp(1, MAX_TRAILER_COMMITS), |
| 481 | }; |
| 482 | let Outcome::Ok(history) = g1t_kit::call::<_, Outcome<Vec<Commit>>>(&self.repos, "log", &log).await? else { |
| 483 | return Ok(()); |
| 484 | }; |
| 485 | let mut named: Vec<String> = Vec::new(); |
| 486 | for commit in &history { |
| 487 | for id in session_trailers(&commit.message) { |
| 488 | if !named.contains(&id) { |
| 489 | named.push(id); |
| 490 | } |
| 491 | } |
| 492 | } |
| 493 | for id in named { |
| 494 | if self.session_row(repo_id, &id).await?.is_none() { |
| 495 | continue; |
| 496 | } |
| 497 | for pull in pulls { |
| 498 | self.link(&id, repo_id, LinkKind::Pull, pull.number, LinkVia::Trailer).await; |
| 499 | } |
| 500 | } |
| 501 | Ok(()) |
| 502 | }; |
| 503 | if let Err(error) = linked.await { |
| 504 | worker::console_error!("push to {branch}: sessions not tied to its pull requests: {error}"); |
| 505 | } |
| 506 | } |
| 507 | |
| 508 | /// Sessions tied to an issue or pull request (`kind`, `number`) by a |
| 509 | /// link, by id. |
| 510 | pub(crate) async fn linked_sessions(&self, repo_id: &str, kind: LinkKind, number: u32) -> Result<Vec<String>> { |
| 511 | Ok(self |
| 512 | .db |
| 513 | .prepare("SELECT DISTINCT session_id AS id FROM session_links WHERE repo_id = ? AND kind = ? AND number = ? LIMIT 200") |
| 514 | .bind(&[repo_id.into(), kind_name(kind).into(), number.into()])? |
| 515 | .all() |
| 516 | .await? |
| 517 | .results::<IdRow>()? |
| 518 | .into_iter() |
| 519 | .map(|row| row.id) |
| 520 | .collect()) |
| 521 | } |
| 522 | |
| 523 | /// Sessions that pushed one of `pulls`' branches, by id: tied to them |
| 524 | /// even when the pull request was opened after the push. |
| 525 | pub(crate) async fn branch_sessions(&self, repo_id: &str, pulls: &[u32]) -> Result<Vec<String>> { |
| 526 | if pulls.is_empty() { |
| 527 | return Ok(Vec::new()); |
| 528 | } |
| 529 | Ok(self |
| 530 | .db |
| 531 | .prepare(BRANCH_SESSIONS) |
| 532 | .bind(&[repo_id.into(), serde_json::to_string(pulls)?.into()])? |
| 533 | .all() |
| 534 | .await? |
| 535 | .results::<IdRow>()? |
| 536 | .into_iter() |
| 537 | .map(|row| row.id) |
| 538 | .collect()) |
| 539 | } |
| 540 | |
| 541 | /// Sessions by id, as they read now. |
| 542 | pub(crate) async fn sessions_by_id(&self, repo_id: &str, ids: &[String]) -> Result<Vec<Session>> { |
| 543 | if ids.is_empty() { |
| 544 | return Ok(Vec::new()); |
| 545 | } |
| 546 | let now = now_ms(); |
| 547 | Ok(self |
| 548 | .db |
| 549 | .prepare(format!( |
| 550 | "SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id IN (SELECT value FROM json_each(?))" |
| 551 | )) |
| 552 | .bind(&[repo_id.into(), serde_json::to_string(ids)?.into()])? |
| 553 | .all() |
| 554 | .await? |
| 555 | .results::<SessionRow>()? |
| 556 | .into_iter() |
| 557 | .map(|row| row.at(now)) |
| 558 | .collect()) |
| 559 | } |
| 560 | } |
| 561 | |
| 562 | /// An issue's state or a pull request's status, as its JSON name. |
| 563 | fn state_name<T: serde::Serialize>(state: &T) -> String { |
| 564 | serde_json::to_value(state).ok().and_then(|value| value.as_str().map(str::to_owned)).unwrap_or_default() |
| 565 | } |
| 566 | |
| 567 | /// Fills in how many subagents each session of a tree started, from the |
| 568 | /// tree itself. |
| 569 | pub(crate) fn count_subagents(sessions: &mut [Session]) { |
| 570 | let mut counts: HashMap<String, u32> = HashMap::new(); |
| 571 | for session in sessions.iter() { |
| 572 | if let Some(parent) = &session.parent_id { |
| 573 | *counts.entry(parent.clone()).or_default() += 1; |
| 574 | } |
| 575 | } |
| 576 | for session in sessions.iter_mut() { |
| 577 | session.subagents = counts.get(&session.id).copied().unwrap_or_default(); |
| 578 | } |
| 579 | } |