headless_lms_models/
agent_changeset_calls.rs

1//! Audit log of agent write calls. A call is also the unit an agent reverts.
2
3use std::collections::HashMap;
4
5use utoipa::ToSchema;
6
7use crate::{
8    agent_changeset_entities::AgentChangesetEntityOperation,
9    agent_changesets::AgentChangesetStatus,
10    library::entity_snapshots::{ContentEntityType, EntityKey, EntityRef, EntitySnapshot},
11    prelude::*,
12};
13
14/// One entity a call touched, as the audit log records it; see [`insert`].
15#[derive(Debug, Clone, PartialEq)]
16pub struct NewEntityChange {
17    pub key: EntityKey,
18    /// See [`EntityRef::label`]; of the entity after the call, or before it for deletes.
19    pub label: String,
20    /// The entity's snapshot JSON just before the call; `None` when it did not exist.
21    pub before_json: Option<String>,
22    /// [`EntitySnapshot::digest`] of the entity just after the call; `None` when the call deleted
23    /// it.
24    pub after_digest: Option<String>,
25}
26
27/// One entity a call touched, as reverting the call needs it.
28#[derive(Debug, Clone, PartialEq)]
29pub struct EntityChange {
30    pub entity_type: ContentEntityType,
31    pub entity_id: Uuid,
32    pub label: String,
33    /// `None` when the entity did not exist before the call.
34    pub before: Option<EntitySnapshot>,
35    /// [`EntitySnapshot::digest`] of the entity just after the call; `None` when the call deleted
36    /// it.
37    pub after_digest: Option<String>,
38}
39
40/// A call as reverting it needs it.
41#[derive(Debug, Clone, PartialEq)]
42pub struct AgentChangesetCall {
43    pub id: Uuid,
44    pub changeset_id: Uuid,
45    pub tool_name: String,
46    pub entity_changes: Vec<EntityChange>,
47}
48
49/// Who made a call: the user and the agent client acting for them.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct CallActor {
52    pub user_id: Uuid,
53    pub oauth_client_id: Uuid,
54    pub oauth_grant_id: Uuid,
55    pub client_name: String,
56}
57
58/// A call to record with [`insert`].
59#[derive(Debug, Clone)]
60pub struct NewAgentChangesetCall<'a> {
61    pub changeset_id: Uuid,
62    pub actor: &'a CallActor,
63    pub tool_name: &'a str,
64    pub ops: &'a serde_json::Value,
65    pub summary: Option<&'a str>,
66    pub entity_changes: &'a [NewEntityChange],
67    pub revert_of_call_id: Option<Uuid>,
68}
69
70/// Records the call and the entities it touched, and returns its id.
71pub async fn insert(conn: &mut PgConnection, call: NewAgentChangesetCall<'_>) -> ModelResult<Uuid> {
72    let id = sqlx::query_scalar!(
73        r#"
74INSERT INTO agent_changeset_calls (
75    changeset_id,
76    oauth_client_id,
77    oauth_grant_id,
78    client_name,
79    tool_name,
80    ops,
81    summary,
82    revert_of_call_id
83  )
84VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
85RETURNING id
86        "#,
87        call.changeset_id,
88        call.actor.oauth_client_id,
89        call.actor.oauth_grant_id,
90        call.actor.client_name,
91        call.tool_name,
92        call.ops,
93        call.summary,
94        call.revert_of_call_id,
95    )
96    .fetch_one(&mut *conn)
97    .await?;
98    let mut entity_types = Vec::with_capacity(call.entity_changes.len());
99    let mut entity_ids = Vec::with_capacity(call.entity_changes.len());
100    let mut labels = Vec::with_capacity(call.entity_changes.len());
101    let mut befores = Vec::with_capacity(call.entity_changes.len());
102    let mut after_digests = Vec::with_capacity(call.entity_changes.len());
103    for change in call.entity_changes {
104        entity_types.push(change.key.entity_type);
105        entity_ids.push(change.key.entity_id);
106        labels.push(change.label.as_str());
107        befores.push(change.before_json.as_deref());
108        after_digests.push(change.after_digest.as_deref());
109    }
110    sqlx::query!(
111        r#"
112INSERT INTO agent_changeset_call_entities (
113    call_id,
114    entity_type,
115    entity_id,
116    label,
117    before_snapshot,
118    after_digest
119  )
120SELECT $1,
121  k.entity_type,
122  k.entity_id,
123  k.label,
124  k.before::jsonb,
125  k.after_digest
126FROM UNNEST(
127    $2::content_entity_type [],
128    $3::uuid [],
129    $4::text [],
130    $5::text [],
131    $6::text []
132  ) AS k(entity_type, entity_id, label, before, after_digest)
133        "#,
134        id,
135        entity_types as Vec<ContentEntityType>,
136        &entity_ids,
137        &labels as &[&str],
138        &befores as &[Option<&str>],
139        &after_digests as &[Option<&str>],
140    )
141    .execute(conn)
142    .await?;
143    Ok(id)
144}
145
146/// The call with the entities it touched.
147pub async fn get_by_id(conn: &mut PgConnection, id: Uuid) -> ModelResult<AgentChangesetCall> {
148    let call = sqlx::query!(
149        r#"
150SELECT id,
151  changeset_id,
152  tool_name
153FROM agent_changeset_calls
154WHERE id = $1
155        "#,
156        id,
157    )
158    .fetch_one(&mut *conn)
159    .await?;
160    let entity_changes = sqlx::query!(
161        r#"
162SELECT entity_type,
163  entity_id,
164  label,
165  before_snapshot::text AS "before_snapshot?",
166  after_digest
167FROM agent_changeset_call_entities
168WHERE call_id = $1
169        "#,
170        id,
171    )
172    .fetch_all(conn)
173    .await?
174    .into_iter()
175    .map(|row| {
176        Ok(EntityChange {
177            entity_type: row.entity_type,
178            entity_id: row.entity_id,
179            label: row.label,
180            before: row
181                .before_snapshot
182                .map(|json| EntitySnapshot::from_json(row.entity_type, &json))
183                .transpose()?,
184            after_digest: row.after_digest,
185        })
186    })
187    .collect::<ModelResult<_>>()?;
188    Ok(AgentChangesetCall {
189        id: call.id,
190        changeset_id: call.changeset_id,
191        tool_name: call.tool_name,
192        entity_changes,
193    })
194}
195
196/// The latest call of the changeset that touched each of its entities, by entity.
197pub async fn get_latest_by_entity(
198    conn: &mut PgConnection,
199    changeset_id: Uuid,
200) -> ModelResult<HashMap<EntityKey, Uuid>> {
201    let rows = sqlx::query!(
202        r#"
203SELECT DISTINCT ON (ce.entity_type, ce.entity_id) ce.entity_type,
204  ce.entity_id,
205  ce.call_id
206FROM agent_changeset_call_entities ce
207  JOIN agent_changeset_calls c ON c.id = ce.call_id
208WHERE c.changeset_id = $1
209ORDER BY ce.entity_type,
210  ce.entity_id,
211  c.created_at DESC,
212  c.id DESC
213        "#,
214        changeset_id,
215    )
216    .fetch_all(conn)
217    .await?;
218    Ok(rows
219        .into_iter()
220        .map(|row| (EntityKey::new(row.entity_type, row.entity_id), row.call_id))
221        .collect())
222}
223
224/// A call as the review page lists it, without its operations and snapshots.
225#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)]
226pub struct AgentChangesetCallSummary {
227    pub id: Uuid,
228    pub created_at: DateTime<Utc>,
229    pub client_name: String,
230    pub tool_name: String,
231    pub summary: Option<String>,
232    pub revert_of_call_id: Option<Uuid>,
233}
234
235/// The changeset's calls, oldest first.
236pub async fn get_summaries_by_changeset_id(
237    conn: &mut PgConnection,
238    changeset_id: Uuid,
239) -> ModelResult<Vec<AgentChangesetCallSummary>> {
240    let res = sqlx::query_as!(
241        AgentChangesetCallSummary,
242        r#"
243SELECT id,
244  created_at,
245  client_name,
246  tool_name,
247  summary,
248  revert_of_call_id
249FROM agent_changeset_calls
250WHERE changeset_id = $1
251ORDER BY created_at,
252  id
253        "#,
254        changeset_id,
255    )
256    .fetch_all(conn)
257    .await?;
258    Ok(res)
259}
260
261/// An entity a call touched, labelled from its snapshot after the call (before it, for deletes).
262#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
263pub struct CallHistoryEntity {
264    #[serde(flatten)]
265    pub entity: EntityRef,
266    pub operation: AgentChangesetEntityOperation,
267}
268
269/// One call in the user's agent history of a course.
270#[derive(Debug, Clone, PartialEq)]
271pub struct CallHistoryEntry {
272    pub id: Uuid,
273    pub created_at: DateTime<Utc>,
274    pub changeset_id: Uuid,
275    pub changeset_status: AgentChangesetStatus,
276    pub client_name: String,
277    pub tool_name: String,
278    pub summary: Option<String>,
279    pub revert_of_call_id: Option<Uuid>,
280    pub entities: Vec<CallHistoryEntity>,
281}
282
283/// The user's most recent calls in a course, newest first; only the calls that touched
284/// `entity_id` when it is given.
285pub async fn get_history(
286    conn: &mut PgConnection,
287    user_id: Uuid,
288    course_id: Uuid,
289    entity_id: Option<Uuid>,
290    limit: i64,
291) -> ModelResult<Vec<CallHistoryEntry>> {
292    let calls = sqlx::query!(
293        r#"
294SELECT c.id,
295  c.created_at,
296  c.changeset_id,
297  cs.status AS "changeset_status: AgentChangesetStatus",
298  cs.expires_at AS changeset_expires_at,
299  c.client_name,
300  c.tool_name,
301  c.summary,
302  c.revert_of_call_id
303FROM agent_changeset_calls c
304  JOIN agent_changesets cs ON cs.id = c.changeset_id
305WHERE cs.user_id = $1
306  AND cs.course_id = $2
307  AND (
308    $3::uuid IS NULL
309    OR EXISTS (
310      SELECT 1
311      FROM agent_changeset_call_entities ce
312      WHERE ce.call_id = c.id
313        AND ce.entity_id = $3
314    )
315  )
316ORDER BY c.created_at DESC,
317  c.id
318LIMIT $4
319        "#,
320        user_id,
321        course_id,
322        entity_id,
323        limit,
324    )
325    .fetch_all(&mut *conn)
326    .await?;
327    let call_ids: Vec<Uuid> = calls.iter().map(|call| call.id).collect();
328    let mut entities_by_call: HashMap<Uuid, Vec<CallHistoryEntity>> = HashMap::new();
329    for row in sqlx::query!(
330        r#"
331SELECT call_id,
332  entity_type,
333  entity_id,
334  label,
335  before_snapshot IS NOT NULL AS "has_before!",
336  after_digest IS NOT NULL AS "has_after!"
337FROM agent_changeset_call_entities
338WHERE call_id = ANY($1)
339ORDER BY entity_type,
340  label,
341  entity_id
342        "#,
343        &call_ids,
344    )
345    .fetch_all(conn)
346    .await?
347    {
348        entities_by_call
349            .entry(row.call_id)
350            .or_default()
351            .push(CallHistoryEntity {
352                entity: EntityRef {
353                    entity_type: row.entity_type,
354                    entity_id: row.entity_id,
355                    label: row.label,
356                },
357                operation: AgentChangesetEntityOperation::between(row.has_before, row.has_after),
358            });
359    }
360    let now = Utc::now();
361    Ok(calls
362        .into_iter()
363        .map(|call| CallHistoryEntry {
364            entities: entities_by_call.remove(&call.id).unwrap_or_default(),
365            id: call.id,
366            created_at: call.created_at,
367            changeset_id: call.changeset_id,
368            changeset_status: call.changeset_status.at(call.changeset_expires_at, now),
369            client_name: call.client_name,
370            tool_name: call.tool_name,
371            summary: call.summary,
372            revert_of_call_id: call.revert_of_call_id,
373        })
374        .collect())
375}
headless_lms_models/agent_changeset_calls.rs

headless_lms_models/
agent_changeset_calls.rs
