| 1 | //! Sessions: one run of someone's own harness on a repository. |
| 2 | //! |
| 3 | //! Starting one is idempotent on the harness's own session id, so a hook |
| 4 | //! that fires twice, or a harness that reconnects, gets the same session |
| 5 | //! back. Every update, and every batch of entries, is a heartbeat: it keeps |
| 6 | //! the session active, brings back one that timed out, and renews the |
| 7 | //! claims made from it. Only the person whose token started a session |
| 8 | //! changes it. |
| 9 | |
| 10 | use g1t_contracts::events::SessionAppended; |
| 11 | use g1t_contracts::repos::Repo; |
| 12 | use g1t_contracts::time::rfc3339; |
| 13 | use g1t_contracts::work::*; |
| 14 | use g1t_contracts::{FailureCode, Outcome, User, new_id}; |
| 15 | use g1t_kit::now_ms; |
| 16 | use serde::Deserialize; |
| 17 | use worker::Result; |
| 18 | use worker::wasm_bindgen::JsValue; |
| 19 | |
| 20 | use super::{SESSION_COLUMNS, SessionRow, after_minutes, tidy, unwrap_or_fail}; |
| 21 | use crate::Work; |
| 22 | use crate::rows::NumberRow; |
| 23 | |
| 24 | /// The most entries one look at a session returns. |
| 25 | const ENTRY_PAGE: u32 = 500; |
| 26 | /// The most sessions one list returns. |
| 27 | const LIST_LIMIT: u32 = 100; |
| 28 | const MAX_TITLE_CHARS: usize = 200; |
| 29 | const MAX_BRANCH_CHARS: usize = 255; |
| 30 | const MAX_ID_CHARS: usize = 200; |
| 31 | const MAX_COMMIT_CHARS: usize = 64; |
| 32 | |
| 33 | fn not_found<T>() -> Outcome<T> { |
| 34 | Outcome::fail(FailureCode::NotFound, "Session not found.") |
| 35 | } |
| 36 | |
| 37 | fn text_or_null(value: &Option<String>) -> JsValue { |
| 38 | value.as_deref().map_or(JsValue::NULL, JsValue::from) |
| 39 | } |
| 40 | |
| 41 | fn number_or_null(value: Option<u32>) -> JsValue { |
| 42 | value.map_or(JsValue::NULL, JsValue::from) |
| 43 | } |
| 44 | |
| 45 | /// What a session's heartbeat may change, tidied. |
| 46 | struct Fields { |
| 47 | issue: Option<u32>, |
| 48 | pull: Option<u32>, |
| 49 | branch: Option<String>, |
| 50 | head_commit: Option<String>, |
| 51 | title: Option<String>, |
| 52 | } |
| 53 | |
| 54 | #[derive(Deserialize)] |
| 55 | struct RenewRow { |
| 56 | id: String, |
| 57 | ttl_minutes: u32, |
| 58 | expires_at: String, |
| 59 | } |
| 60 | |
| 61 | /// Starts a session; nothing when one with the same harness session id |
| 62 | /// was started already. |
| 63 | pub(crate) fn insert_session_sql() -> String { |
| 64 | format!( |
| 65 | "INSERT INTO sessions |
| 66 | (id, repo_id, issue, pull, branch, actor_id, actor_name, harness, harness_session_id, |
| 67 | title, state, head_commit, started_at, last_seen_at) |
| 68 | VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, 'active', ?11, ?12, ?13) |
| 69 | ON CONFLICT (repo_id, actor_id, harness, harness_session_id) |
| 70 | WHERE harness_session_id IS NOT NULL DO NOTHING |
| 71 | RETURNING {SESSION_COLUMNS}" |
| 72 | ) |
| 73 | } |
| 74 | |
| 75 | /// A person's session by its harness's id for it. |
| 76 | pub(crate) fn harness_session_sql() -> String { |
| 77 | format!( |
| 78 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 79 | WHERE repo_id = ?1 AND actor_id = ?2 AND harness = ?3 AND harness_session_id = ?4" |
| 80 | ) |
| 81 | } |
| 82 | |
| 83 | /// A heartbeat: seen now (`?6`), active again, with whichever fields are |
| 84 | /// given. |
| 85 | pub(crate) fn resume_sql() -> String { |
| 86 | format!( |
| 87 | "UPDATE sessions |
| 88 | SET issue = COALESCE(?1, issue), pull = COALESCE(?2, pull), branch = COALESCE(?3, branch), |
| 89 | head_commit = COALESCE(?4, head_commit), title = COALESCE(?5, title), |
| 90 | last_seen_at = ?6, state = 'active', ended_at = NULL, end_reason = NULL |
| 91 | WHERE id = ?7 |
| 92 | RETURNING {SESSION_COLUMNS}" |
| 93 | ) |
| 94 | } |
| 95 | |
| 96 | /// Records one entry in session `?7`, and in pull request `?1` when it |
| 97 | /// is on one. Each takes the next number of the session, and of its pull |
| 98 | /// request, itself, so two batches at once cannot collide; an entry whose |
| 99 | /// key (`?8`) the session has recorded already is skipped. |
| 100 | pub(crate) const INSERT_ENTRY: &str = "INSERT INTO session_entries |
| 101 | (pull_id, seq, kind, text, tool, \"commit\", at, session_id, session_seq, key) |
| 102 | SELECT ?1, |
| 103 | CASE WHEN ?1 IS NULL |
| 104 | THEN (SELECT COALESCE(MAX(session_seq), 0) + 1 FROM session_entries WHERE session_id = ?7) |
| 105 | ELSE (SELECT COALESCE(MAX(seq), 0) + 1 FROM session_entries WHERE pull_id = ?1) END, |
| 106 | ?2, ?3, ?4, ?5, ?6, ?7, |
| 107 | (SELECT COALESCE(MAX(session_seq), 0) + 1 FROM session_entries WHERE session_id = ?7), |
| 108 | ?8 |
| 109 | WHERE true |
| 110 | ON CONFLICT (session_id, key) WHERE session_id IS NOT NULL AND key IS NOT NULL DO NOTHING |
| 111 | RETURNING session_seq AS n"; |
| 112 | |
| 113 | /// Entries of session `?3` recorded before it named its pull request |
| 114 | /// `?1` join it, numbered after its last (`?2`), in their order. |
| 115 | pub(crate) const MOVE_ENTRIES: &str = "UPDATE session_entries SET pull_id = ?1, seq = ?2 + session_seq |
| 116 | WHERE session_id = ?3 AND pull_id IS NULL"; |
| 117 | |
| 118 | /// Renews active claim `?2` to `?1`. |
| 119 | pub(crate) const RENEW_CLAIM: &str = "UPDATE claims SET expires_at = ?1 WHERE id = ?2 AND state = 'active'"; |
| 120 | |
| 121 | /// When a claim of `minutes` renewed at `now` lapses, if that is later |
| 122 | /// than it does already (`expires_at`): a heartbeat never shortens one. |
| 123 | pub(crate) fn renewal(now: u64, minutes: u32, expires_at: &str) -> Option<String> { |
| 124 | let renewed = after_minutes(now, minutes); |
| 125 | (renewed.as_str() > expires_at).then_some(renewed) |
| 126 | } |
| 127 | |
| 128 | impl Work { |
| 129 | /// The issue and pull request a session names, checked: each is one of |
| 130 | /// the repository's, of the right kind. A pull request for an issue |
| 131 | /// names that issue when no issue was given. |
| 132 | async fn session_subjects(&self, repo: &Repo, issue: Option<u32>, pull: Option<u32>) -> Result<Outcome<(Option<u32>, Option<u32>)>> { |
| 133 | let mut issue = issue.filter(|number| *number > 0); |
| 134 | let pull = pull.filter(|number| *number > 0); |
| 135 | if let Some(number) = issue |
| 136 | && self.issue(&repo.id, number).await?.is_none() |
| 137 | { |
| 138 | return Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is not an issue in this repository."))); |
| 139 | } |
| 140 | if let Some(number) = pull { |
| 141 | match self.pull(&repo.id, number).await? { |
| 142 | Some(found) => issue = issue.or(found.issue), |
| 143 | None => { |
| 144 | return Ok(Outcome::fail(FailureCode::Invalid, format!("#{number} is not a pull request in this repository."))); |
| 145 | } |
| 146 | } |
| 147 | } |
| 148 | Ok(Outcome::Ok((issue, pull))) |
| 149 | } |
| 150 | |
| 151 | pub(crate) async fn start_session(&self, a: StartSessionArgs) -> Result<Outcome<Session>> { |
| 152 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 153 | let (issue, pull) = unwrap_or_fail!(self.session_subjects(&repo, a.issue, a.pull).await?); |
| 154 | let fields = Fields { |
| 155 | issue, |
| 156 | pull, |
| 157 | branch: tidy(a.branch.as_deref(), MAX_BRANCH_CHARS), |
| 158 | head_commit: tidy(a.head_commit.as_deref(), MAX_COMMIT_CHARS), |
| 159 | title: tidy(a.title.as_deref(), MAX_TITLE_CHARS), |
| 160 | }; |
| 161 | let harness_session_id = tidy(a.harness_session_id.as_deref(), MAX_ID_CHARS); |
| 162 | let now = now_ms(); |
| 163 | let at = rfc3339(now); |
| 164 | // The same harness session again: the one already started. |
| 165 | if let Some(harness_id) = &harness_session_id |
| 166 | && let Some(row) = self.harness_session(&repo.id, &a.actor, a.harness, harness_id).await? |
| 167 | { |
| 168 | return self.resume(&repo, row, &a.actor, fields).await; |
| 169 | } |
| 170 | let inserted = self |
| 171 | .db |
| 172 | .prepare(insert_session_sql()) |
| 173 | .bind(&[ |
| 174 | new_id("ses", now).into(), |
| 175 | repo.id.as_str().into(), |
| 176 | number_or_null(fields.issue), |
| 177 | number_or_null(fields.pull), |
| 178 | text_or_null(&fields.branch), |
| 179 | a.actor.id.as_str().into(), |
| 180 | a.actor.username.as_str().into(), |
| 181 | a.harness.as_str().into(), |
| 182 | text_or_null(&harness_session_id), |
| 183 | text_or_null(&fields.title), |
| 184 | text_or_null(&fields.head_commit), |
| 185 | at.as_str().into(), |
| 186 | at.as_str().into(), |
| 187 | ])? |
| 188 | .first::<SessionRow>(None) |
| 189 | .await?; |
| 190 | let Some(row) = inserted else { |
| 191 | // Started at the same moment by another call: that one. |
| 192 | let Some(harness_id) = &harness_session_id else { |
| 193 | return Ok(not_found()); |
| 194 | }; |
| 195 | return match self.harness_session(&repo.id, &a.actor, a.harness, harness_id).await? { |
| 196 | Some(row) => Ok(Outcome::Ok(row.at(now))), |
| 197 | None => Ok(not_found()), |
| 198 | }; |
| 199 | }; |
| 200 | let session = row.at(now); |
| 201 | self.publish("session.started", &repo.id, &a.actor, SessionEvent::of(&session)).await?; |
| 202 | Ok(Outcome::Ok(session)) |
| 203 | } |
| 204 | |
| 205 | async fn harness_session(&self, repo_id: &str, actor: &User, harness: Harness, harness_id: &str) -> Result<Option<SessionRow>> { |
| 206 | self.db |
| 207 | .prepare(harness_session_sql()) |
| 208 | .bind(&[repo_id.into(), actor.id.as_str().into(), harness.as_str().into(), harness_id.into()])? |
| 209 | .first::<SessionRow>(None) |
| 210 | .await |
| 211 | } |
| 212 | |
| 213 | /// A heartbeat, with whatever `fields` change: the session is seen now, |
| 214 | /// brought back if it had ended (announced as started again), its |
| 215 | /// claims renewed, and entries recorded before it named its pull |
| 216 | /// request moved onto it. |
| 217 | async fn resume(&self, repo: &Repo, row: SessionRow, actor: &User, fields: Fields) -> Result<Outcome<Session>> { |
| 218 | let now = now_ms(); |
| 219 | let was_ended = row.state == "ended"; |
| 220 | let had_pull = row.pull; |
| 221 | let id = row.id.clone(); |
| 222 | let at = rfc3339(now); |
| 223 | let updated = self |
| 224 | .db |
| 225 | .prepare(resume_sql()) |
| 226 | .bind(&[ |
| 227 | number_or_null(fields.issue), |
| 228 | number_or_null(fields.pull), |
| 229 | text_or_null(&fields.branch), |
| 230 | text_or_null(&fields.head_commit), |
| 231 | text_or_null(&fields.title), |
| 232 | at.as_str().into(), |
| 233 | id.as_str().into(), |
| 234 | ])? |
| 235 | .first::<SessionRow>(None) |
| 236 | .await?; |
| 237 | let Some(updated) = updated else { |
| 238 | return Ok(not_found()); |
| 239 | }; |
| 240 | let session = updated.at(now); |
| 241 | self.renew_claims(&id, now).await?; |
| 242 | if let Some(number) = session.pull |
| 243 | && had_pull != Some(number) |
| 244 | { |
| 245 | self.move_entries_onto_pull(&repo.id, &id, number).await?; |
| 246 | } |
| 247 | if was_ended { |
| 248 | self.publish("session.started", &repo.id, actor, SessionEvent::of(&session)).await?; |
| 249 | } |
| 250 | Ok(Outcome::Ok(session)) |
| 251 | } |
| 252 | |
| 253 | /// Renews the active claims made from a session by their own length. |
| 254 | async fn renew_claims(&self, session_id: &str, now: u64) -> Result<()> { |
| 255 | let claims = self |
| 256 | .db |
| 257 | .prepare("SELECT id, ttl_minutes, expires_at FROM claims WHERE session_id = ? AND state = 'active'") |
| 258 | .bind(&[session_id.into()])? |
| 259 | .all() |
| 260 | .await? |
| 261 | .results::<RenewRow>()?; |
| 262 | let mut statements = Vec::new(); |
| 263 | for claim in claims { |
| 264 | if let Some(renewed) = renewal(now, claim.ttl_minutes, &claim.expires_at) { |
| 265 | statements.push(self.db.prepare(RENEW_CLAIM).bind(&[renewed.into(), claim.id.into()])?); |
| 266 | } |
| 267 | } |
| 268 | if !statements.is_empty() { |
| 269 | self.db.batch(statements).await?; |
| 270 | } |
| 271 | Ok(()) |
| 272 | } |
| 273 | |
| 274 | /// Entries a session recorded before it named its pull request join |
| 275 | /// that pull request's session, after what it has, in their order. |
| 276 | async fn move_entries_onto_pull(&self, repo_id: &str, session_id: &str, number: u32) -> Result<()> { |
| 277 | let Some(pull) = self.pull(repo_id, number).await? else { |
| 278 | return Ok(()); |
| 279 | }; |
| 280 | let base = self |
| 281 | .db |
| 282 | .prepare("SELECT COALESCE(MAX(seq), 0) AS n FROM session_entries WHERE pull_id = ?") |
| 283 | .bind(&[pull.id.as_str().into()])? |
| 284 | .first::<NumberRow>(None) |
| 285 | .await? |
| 286 | .map_or(0, |row| row.n); |
| 287 | let moved = self |
| 288 | .db |
| 289 | .prepare(MOVE_ENTRIES) |
| 290 | .bind(&[pull.id.as_str().into(), base.into(), session_id.into()])? |
| 291 | .run() |
| 292 | .await; |
| 293 | // An entry recorded on the pull request at the same moment took a |
| 294 | // number these needed: they stay the session's alone. |
| 295 | if let Err(error) = moved { |
| 296 | worker::console_error!("session {session_id}: entries not moved onto #{number}: {error}"); |
| 297 | } |
| 298 | Ok(()) |
| 299 | } |
| 300 | |
| 301 | /// The caller's own session, as stored. |
| 302 | async fn own_session(&self, repo: &Repo, actor: &User, id: &str) -> Result<Outcome<SessionRow>> { |
| 303 | Ok(match self.session_row(&repo.id, id.trim()).await? { |
| 304 | Some(row) if row.actor_id == actor.id => Outcome::Ok(row), |
| 305 | _ => not_found(), |
| 306 | }) |
| 307 | } |
| 308 | |
| 309 | pub(crate) async fn update_session(&self, a: UpdateSessionArgs) -> Result<Outcome<Session>> { |
| 310 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 311 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 312 | if row.state == "ended" && row.end_reason.as_deref() != Some("timed_out") { |
| 313 | return Ok(Outcome::fail(FailureCode::Conflict, "This session has ended. Start a new one.")); |
| 314 | } |
| 315 | let (issue, pull) = unwrap_or_fail!(self.session_subjects(&repo, a.issue, a.pull).await?); |
| 316 | let fields = Fields { |
| 317 | issue, |
| 318 | pull, |
| 319 | branch: tidy(a.branch.as_deref(), MAX_BRANCH_CHARS), |
| 320 | head_commit: tidy(a.head_commit.as_deref(), MAX_COMMIT_CHARS), |
| 321 | title: tidy(a.title.as_deref(), MAX_TITLE_CHARS), |
| 322 | }; |
| 323 | self.resume(&repo, row, &a.actor, fields).await |
| 324 | } |
| 325 | |
| 326 | pub(crate) async fn end_session(&self, a: EndSessionArgs) -> Result<Outcome<Session>> { |
| 327 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 328 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 329 | let now = now_ms(); |
| 330 | if row.state == "ended" { |
| 331 | return Ok(Outcome::Ok(row.at(now))); |
| 332 | } |
| 333 | let reason = tidy(a.reason.as_deref(), MAX_NOTE_CHARS).unwrap_or_else(|| "ended".to_owned()); |
| 334 | let at = rfc3339(now); |
| 335 | let ended = self |
| 336 | .db |
| 337 | .prepare(format!( |
| 338 | "UPDATE sessions SET state = 'ended', ended_at = ?, end_reason = ?, last_seen_at = ? |
| 339 | WHERE id = ? AND state = 'active' |
| 340 | RETURNING {SESSION_COLUMNS}" |
| 341 | )) |
| 342 | .bind(&[at.as_str().into(), reason.into(), at.as_str().into(), row.id.as_str().into()])? |
| 343 | .first::<SessionRow>(None) |
| 344 | .await?; |
| 345 | let Some(ended) = ended else { |
| 346 | // Ended by another call meanwhile. |
| 347 | return Ok(match self.session_row(&repo.id, &row.id).await? { |
| 348 | Some(row) => Outcome::Ok(row.at(now)), |
| 349 | None => not_found(), |
| 350 | }); |
| 351 | }; |
| 352 | let session = ended.at(now); |
| 353 | self.publish("session.ended", &repo.id, &a.actor, SessionEvent::of(&session)).await?; |
| 354 | Ok(Outcome::Ok(session)) |
| 355 | } |
| 356 | |
| 357 | pub(crate) async fn get_work_session(&self, a: WorkSessionArgs) -> Result<Outcome<SessionDetail>> { |
| 358 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 359 | let Some(row) = self.session_row(&repo.id, a.id.trim()).await? else { |
| 360 | return Ok(not_found()); |
| 361 | }; |
| 362 | let entries = self |
| 363 | .db |
| 364 | .prepare( |
| 365 | "SELECT session_seq AS seq, kind, text, tool, \"commit\", at FROM session_entries |
| 366 | WHERE session_id = ? AND session_seq > ? ORDER BY session_seq LIMIT ?", |
| 367 | ) |
| 368 | .bind(&[row.id.as_str().into(), a.after_seq.into(), ENTRY_PAGE.into()])? |
| 369 | .all() |
| 370 | .await? |
| 371 | .results::<crate::rows::SessionRow>()? |
| 372 | .into_iter() |
| 373 | .map(SessionEntry::from) |
| 374 | .collect(); |
| 375 | Ok(Outcome::Ok(SessionDetail { session: row.at(now_ms()), entries })) |
| 376 | } |
| 377 | |
| 378 | pub(crate) async fn list_work_sessions(&self, a: ListWorkSessionsArgs) -> Result<Outcome<Vec<Session>>> { |
| 379 | let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?); |
| 380 | let now = now_ms(); |
| 381 | let idle_since = rfc3339(now.saturating_sub(SESSION_IDLE_MS)); |
| 382 | // Every statement names each bind it is given. |
| 383 | let state = match a.state { |
| 384 | None => "AND ?2 IS NOT NULL", |
| 385 | Some(SessionState::Active) => "AND state = 'active' AND last_seen_at > ?2", |
| 386 | Some(SessionState::Idle) => "AND state = 'active' AND last_seen_at <= ?2", |
| 387 | Some(SessionState::Ended) => "AND state = 'ended' AND ?2 IS NOT NULL", |
| 388 | }; |
| 389 | let rows = self |
| 390 | .db |
| 391 | .prepare(format!( |
| 392 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 393 | WHERE repo_id = ?1 {state} |
| 394 | AND (?3 IS NULL OR issue = ?3) AND (?4 IS NULL OR pull = ?4) |
| 395 | ORDER BY last_seen_at DESC LIMIT ?5" |
| 396 | )) |
| 397 | .bind(&[ |
| 398 | repo.id.as_str().into(), |
| 399 | idle_since.as_str().into(), |
| 400 | number_or_null(a.issue), |
| 401 | number_or_null(a.pull), |
| 402 | LIST_LIMIT.into(), |
| 403 | ])? |
| 404 | .all() |
| 405 | .await? |
| 406 | .results::<SessionRow>()?; |
| 407 | Ok(Outcome::Ok(rows.into_iter().map(|row| row.at(now)).collect())) |
| 408 | } |
| 409 | |
| 410 | /// Notes `text` (from g1t) in the live harness sessions on `pull` or |
| 411 | /// its issue, so an agent reading its session hears of it. Never fails |
| 412 | /// what it follows: a problem is logged. |
| 413 | pub(crate) async fn note_sessions(&self, pull: &Pull, text: &str) { |
| 414 | let noted = async { |
| 415 | let sessions: Vec<SessionRow> = self |
| 416 | .db |
| 417 | .prepare(format!( |
| 418 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 419 | WHERE repo_id = ?1 AND state = 'active' AND (pull = ?2 OR (?3 IS NOT NULL AND issue = ?3))" |
| 420 | )) |
| 421 | .bind(&[pull.repo_id.as_str().into(), pull.number.into(), number_or_null(pull.issue)])? |
| 422 | .all() |
| 423 | .await? |
| 424 | .results()?; |
| 425 | let at = rfc3339(now_ms()); |
| 426 | let mut statements = Vec::new(); |
| 427 | for session in sessions { |
| 428 | // Kept by the session alone: the pull request's own session |
| 429 | // has the note already. |
| 430 | statements.push(self.db.prepare(INSERT_ENTRY).bind(&[ |
| 431 | JsValue::NULL, |
| 432 | "note".into(), |
| 433 | text.into(), |
| 434 | JsValue::NULL, |
| 435 | text_or_null(&pull.head_commit), |
| 436 | at.as_str().into(), |
| 437 | session.id.as_str().into(), |
| 438 | JsValue::NULL, |
| 439 | ])?); |
| 440 | } |
| 441 | if !statements.is_empty() { |
| 442 | self.db.batch(statements).await?; |
| 443 | } |
| 444 | Ok::<(), worker::Error>(()) |
| 445 | }; |
| 446 | if let Err(error) = noted.await { |
| 447 | worker::console_error!("#{}: sessions were not told: {error}", pull.number); |
| 448 | } |
| 449 | } |
| 450 | |
| 451 | pub(crate) async fn record_session_entries(&self, a: RecordEntriesArgs) -> Result<Outcome<Recorded>> { |
| 452 | if a.entries.len() > crate::MAX_ENTRY_BATCH { |
| 453 | return Ok(Outcome::fail( |
| 454 | FailureCode::Invalid, |
| 455 | format!("Send at most {} entries at a time.", crate::MAX_ENTRY_BATCH), |
| 456 | )); |
| 457 | } |
| 458 | let repo = unwrap_or_fail!(self.participating_repo(&a.repo, &a.actor).await?); |
| 459 | let row = unwrap_or_fail!(self.own_session(&repo, &a.actor, &a.id).await?); |
| 460 | if row.state == "ended" && row.end_reason.as_deref() != Some("timed_out") { |
| 461 | return Ok(Outcome::fail(FailureCode::Conflict, "This session has ended. Start a new one.")); |
| 462 | } |
| 463 | // Recording is a heartbeat too. |
| 464 | let fields = Fields { issue: None, pull: None, branch: None, head_commit: None, title: None }; |
| 465 | let session = unwrap_or_fail!(self.resume(&repo, row, &a.actor, fields).await?); |
| 466 | if a.entries.is_empty() { |
| 467 | return Ok(Outcome::Ok(Recorded::default())); |
| 468 | } |
| 469 | let pull = match session.pull { |
| 470 | Some(number) => self.pull(&repo.id, number).await?, |
| 471 | None => None, |
| 472 | }; |
| 473 | let pull_id: JsValue = pull.as_ref().map_or(JsValue::NULL, |pull| pull.id.as_str().into()); |
| 474 | let at = rfc3339(now_ms()); |
| 475 | let total = a.entries.len() as u32; |
| 476 | let mut statements = Vec::with_capacity(a.entries.len()); |
| 477 | for keyed in a.entries { |
| 478 | let entry = keyed.entry; |
| 479 | let kind = serde_json::to_value(entry.kind)?; |
| 480 | let text: String = entry.text.chars().take(crate::MAX_ENTRY_CHARS).collect(); |
| 481 | let key = tidy(keyed.key.as_deref(), MAX_KEY_CHARS); |
| 482 | // Each takes the next number of the session, and of its pull |
| 483 | // request when it has one, itself, so two batches at once |
| 484 | // cannot collide; a key already recorded is skipped. |
| 485 | statements.push( |
| 486 | self.db |
| 487 | .prepare(INSERT_ENTRY) |
| 488 | .bind(&[ |
| 489 | pull_id.clone(), |
| 490 | kind.as_str().unwrap_or("note").into(), |
| 491 | text.into(), |
| 492 | text_or_null(&entry.tool), |
| 493 | text_or_null(&entry.commit.or_else(|| session.head_commit.clone())), |
| 494 | at.as_str().into(), |
| 495 | session.id.as_str().into(), |
| 496 | text_or_null(&key), |
| 497 | ])?, |
| 498 | ); |
| 499 | } |
| 500 | let mut recorded = 0; |
| 501 | for result in self.db.batch(statements).await? { |
| 502 | recorded += result.results::<NumberRow>()?.len() as u32; |
| 503 | } |
| 504 | if let Some(pull) = pull |
| 505 | && recorded > 0 |
| 506 | { |
| 507 | self.publish( |
| 508 | "session.appended", |
| 509 | &repo.id, |
| 510 | &a.actor, |
| 511 | SessionAppended { pull_id: pull.id, repo_id: repo.id.clone(), number: pull.number, count: recorded }, |
| 512 | ) |
| 513 | .await?; |
| 514 | } |
| 515 | Ok(Outcome::Ok(Recorded { recorded, duplicates: total - recorded })) |
| 516 | } |
| 517 | } |