Skip to content
Sign up

flagon-io/g1t

public
0
517 linesCodeBlameRaw
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
10use g1t_contracts::events::SessionAppended;
11use g1t_contracts::repos::Repo;
12use g1t_contracts::time::rfc3339;
13use g1t_contracts::work::*;
14use g1t_contracts::{FailureCode, Outcome, User, new_id};
15use g1t_kit::now_ms;
16use serde::Deserialize;
17use worker::Result;
18use worker::wasm_bindgen::JsValue;
19
20use super::{SESSION_COLUMNS, SessionRow, after_minutes, tidy, unwrap_or_fail};
21use crate::Work;
22use crate::rows::NumberRow;
23
24/// The most entries one look at a session returns.
25const ENTRY_PAGE: u32 = 500;
26/// The most sessions one list returns.
27const LIST_LIMIT: u32 = 100;
28const MAX_TITLE_CHARS: usize = 200;
29const MAX_BRANCH_CHARS: usize = 255;
30const MAX_ID_CHARS: usize = 200;
31const MAX_COMMIT_CHARS: usize = 64;
32
33fn not_found<T>() -> Outcome<T> {
34 Outcome::fail(FailureCode::NotFound, "Session not found.")
35}
36
37fn text_or_null(value: &Option<String>) -> JsValue {
38 value.as_deref().map_or(JsValue::NULL, JsValue::from)
39}
40
41fn 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.
46struct 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)]
55struct 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.
63pub(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.
76pub(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.
85pub(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.
100pub(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.
115pub(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`.
119pub(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.
123pub(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
128impl 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}