ai_crew_sync/dashboard/
data.rs

1//! Everything the dashboard reads, and nothing about how it looks.
2//!
3//! Split out of the page builder so the queries can be read, changed and
4//! measured without scrolling past HTML — and so the rendering layer receives
5//! named fields instead of seven-element tuples whose meaning lived in the
6//! order of a SELECT.
7
8use sqlx::PgPool;
9use uuid::Uuid;
10
11#[derive(sqlx::FromRow)]
12pub struct Totals {
13    pub agents_online: i64,
14    pub open_tasks: i64,
15    pub claimed_tasks: i64,
16    pub messages_24h: i64,
17}
18
19#[derive(sqlx::FromRow)]
20pub struct AgentRow {
21    pub name: String,
22    pub status: Option<String>,
23    pub repo: Option<String>,
24    pub branch: Option<String>,
25    pub activity: Option<String>,
26    pub updated_at: Option<chrono::DateTime<chrono::Utc>>,
27    pub online: bool,
28}
29
30#[derive(sqlx::FromRow)]
31pub struct TaskRow {
32    pub key: String,
33    pub title: String,
34    pub status: String,
35    pub claimed_by: Option<String>,
36    pub result: Option<String>,
37    pub blocked: bool,
38    pub updated_at: chrono::DateTime<chrono::Utc>,
39}
40
41#[derive(sqlx::FromRow)]
42pub struct MessageRow {
43    pub id: i64,
44    pub channel: String,
45    pub sender: String,
46    pub body: String,
47    pub created_at: chrono::DateTime<chrono::Utc>,
48}
49
50#[derive(sqlx::FromRow)]
51pub struct LockRow {
52    pub name: String,
53    pub holder: String,
54    pub purpose: Option<String>,
55    pub expires_at: chrono::DateTime<chrono::Utc>,
56}
57
58#[derive(sqlx::FromRow)]
59pub struct NoteRow {
60    pub scope: String,
61    pub key: String,
62    pub updated_by: Option<String>,
63    pub updated_at: chrono::DateTime<chrono::Utc>,
64}
65
66/// One refresh of the dashboard.
67pub struct Snapshot {
68    pub team: String,
69    pub totals: Totals,
70    pub agents: Vec<AgentRow>,
71    pub tasks: Vec<TaskRow>,
72    pub messages: Vec<MessageRow>,
73    pub locks: Vec<LockRow>,
74    pub notes: Vec<NoteRow>,
75}
76
77/// Read everything the page shows for `team_id`.
78///
79/// Direct messages are excluded here, at the source, rather than being
80/// filtered later: the rendering layer never receives one, so it cannot leak
81/// one by accident.
82///
83/// The seven queries are independent, so they run concurrently on separate
84/// pool connections rather than one after another — the page refreshes every
85/// 15 seconds, and serialising them made the refresh as slow as their sum.
86/// Each sees its own snapshot, which is correct for a dashboard: the numbers
87/// are already stale by the time they reach a browser, and no derived value
88/// is computed across two of them.
89pub async fn load(pool: &PgPool, team_id: Uuid) -> Result<Snapshot, sqlx::Error> {
90    let (team, totals, agents, tasks, messages, locks, notes) = tokio::try_join!(
91        load_team(pool, team_id),
92        load_totals(pool, team_id),
93        load_agents(pool, team_id),
94        load_tasks(pool, team_id),
95        load_messages(pool, team_id),
96        load_locks(pool, team_id),
97        load_notes(pool, team_id),
98    )?;
99
100    Ok(Snapshot {
101        team,
102        totals,
103        agents,
104        tasks,
105        messages,
106        locks,
107        notes,
108    })
109}
110
111async fn load_team(pool: &PgPool, team_id: Uuid) -> Result<String, sqlx::Error> {
112    sqlx::query_scalar("SELECT slug FROM teams WHERE id = $1")
113        .bind(team_id)
114        .fetch_one(pool)
115        .await
116}
117
118async fn load_totals(pool: &PgPool, team_id: Uuid) -> Result<Totals, sqlx::Error> {
119    // The three stat-tile numbers in one round trip instead of three.
120    sqlx::query_as(
121        r#"
122        SELECT
123            (SELECT count(*) FROM agents a
124              JOIN agent_presence p ON p.agent_id = a.id
125             WHERE a.team_id = $1 AND p.expires_at > now())          AS agents_online,
126            (SELECT count(*) FROM tasks
127             WHERE team_id = $1 AND status = 'open')                 AS open_tasks,
128            (SELECT count(*) FROM tasks
129             WHERE team_id = $1 AND status = 'claimed')              AS claimed_tasks,
130            (SELECT count(*) FROM messages
131             WHERE team_id = $1 AND channel_id IS NOT NULL
132               AND created_at > now() - interval '24 hours')         AS messages_24h
133        "#,
134    )
135    .bind(team_id)
136    .fetch_one(pool)
137    .await
138}
139
140async fn load_agents(pool: &PgPool, team_id: Uuid) -> Result<Vec<AgentRow>, sqlx::Error> {
141    sqlx::query_as(
142        r#"
143        SELECT a.name,
144               p.status,
145               p.repo,
146               p.branch,
147               p.activity,
148               p.updated_at,
149               COALESCE(p.expires_at > now(), false) AS online
150        FROM agents a
151        LEFT JOIN agent_presence p ON p.agent_id = a.id
152        WHERE a.team_id = $1 AND a.disabled_at IS NULL
153        ORDER BY COALESCE(p.expires_at > now(), false) DESC, a.name
154        "#,
155    )
156    .bind(team_id)
157    .fetch_all(pool)
158    .await
159}
160
161/// Claimed and open first — the work in flight is what a human came to see.
162async fn load_tasks(pool: &PgPool, team_id: Uuid) -> Result<Vec<TaskRow>, sqlx::Error> {
163    sqlx::query_as(
164        r#"
165        SELECT t.key,
166               t.title,
167               t.status,
168               cb.name AS claimed_by,
169               t.result,
170               EXISTS (
171                   SELECT 1 FROM task_deps td
172                   JOIN tasks d ON d.id = td.blocked_by_task_id
173                   WHERE td.task_id = t.id AND d.status NOT IN ('done', 'cancelled')
174               ) AS blocked,
175               t.updated_at
176        FROM tasks t
177        LEFT JOIN agents cb ON cb.id = t.claimed_by
178        WHERE t.team_id = $1
179        ORDER BY CASE t.status WHEN 'claimed' THEN 0 WHEN 'open' THEN 1 ELSE 2 END,
180                 t.updated_at DESC
181        LIMIT 30
182        "#,
183    )
184    .bind(team_id)
185    .fetch_all(pool)
186    .await
187}
188
189/// The JOIN on channels is what excludes direct messages: a DM has no
190/// channel_id, so it cannot appear here at all.
191async fn load_messages(pool: &PgPool, team_id: Uuid) -> Result<Vec<MessageRow>, sqlx::Error> {
192    sqlx::query_as(
193        r#"
194        SELECT m.id, ch.name AS channel, s.name AS sender, left(m.body, 240) AS body, m.created_at
195        FROM messages m
196        JOIN channels ch ON ch.id = m.channel_id
197        JOIN agents s    ON s.id = m.sender_agent_id
198        WHERE m.team_id = $1
199        ORDER BY m.id DESC
200        LIMIT 20
201        "#,
202    )
203    .bind(team_id)
204    .fetch_all(pool)
205    .await
206}
207
208async fn load_locks(pool: &PgPool, team_id: Uuid) -> Result<Vec<LockRow>, sqlx::Error> {
209    sqlx::query_as(
210        r#"
211        SELECT l.name, a.name AS holder, l.purpose, l.expires_at
212        FROM locks l
213        JOIN agents a ON a.id = l.holder_agent_id
214        WHERE l.team_id = $1 AND l.expires_at > now()
215        ORDER BY l.name
216        "#,
217    )
218    .bind(team_id)
219    .fetch_all(pool)
220    .await
221}
222
223async fn load_notes(pool: &PgPool, team_id: Uuid) -> Result<Vec<NoteRow>, sqlx::Error> {
224    sqlx::query_as(
225        r#"
226        SELECT n.scope, n.key, a.name AS updated_by, n.updated_at
227        FROM notes n
228        LEFT JOIN agents a ON a.id = n.updated_by
229        WHERE n.team_id = $1
230        ORDER BY n.updated_at DESC
231        LIMIT 15
232        "#,
233    )
234    .bind(team_id)
235    .fetch_all(pool)
236    .await
237}
ai_crew_sync/dashboard/data.rs

ai_crew_sync/dashboard/
data.rs
