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