OXYGENHUB
◆ v0.1.0 spectra / crates / radar / src / lib.rs
21.9 KiB · Rust · last changed Spectra stands alone: its own identities, signatures and CBOR; projects of any kind of work (#26, #27) 1 day ago source blame history raw
1
//! Radar (spec §17): the human's pages, rendered with askama, refreshed
2
//! from the event stream, working without JavaScript. Linked by `spectra`
3
//! and, later, mounted by SpectraHub.
4
//!
5
//! The owner signs in with a one-time link their CLI signed (`spectra
6
//! open`); everything they do here is recorded with that signed request as
7
//! its proof (spec §8.3). Every unsafe request carries the session's CSRF
8
//! token and a same-origin `Origin`.
9
10
use askama::Template;
11
use axum::body::Bytes;
12
use axum::extract::connect_info::ConnectInfo;
13
use axum::extract::{Form, Path, Query, State};
14
use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
15
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
16
use axum::response::{Html, IntoResponse, Redirect, Response};
17
use axum::routing::{get, post};
18
use axum::Router;
19
use futures::StreamExt;
20
use serde::Deserialize;
21
use spectra_core::conversation::{self as conv, Author, Post, Signer};
22
use spectra_core::event::{Auth, Method};
23
use spectra_core::ids::{format_time, now_ms};
24
use spectra_core::principals::{Principal, Principals};
25
use spectra_mcp::http::{AppState, PeerInfo};
26
use spectra_mcp::session::{self, Session};
27
use std::time::Duration;
28
29
const MARK: &str = include_str!("../assets/spectra.svg");
30
const STYLE: &str = include_str!("../assets/style.css");
31
const SCRIPT: &str = include_str!("../assets/radar.js");
32
const COOKIE: &str = "spectra_session";
33
34
#[derive(Clone)]
35
pub struct RadarState {
36
    pub app: AppState,
37
}
38
39
impl RadarState {
40
    fn sessions(&self) -> &session::Sessions {
41
        &self.app.sessions
42
    }
43
}
44
45
/// Radar's routes.
46
pub fn router(app: AppState) -> Router {
47
    let state = RadarState { app };
48
    Router::new()
49
        .route("/", get(live))
50
        .route("/channels", get(channels))
51
        .route("/c/{name}", get(channel).post(post_channel))
52
        .route("/c/{name}/t/{id}", get(thread).post(post_thread))
53
        .route("/log", get(log))
54
        .route("/log/verify", post(verify))
55
        .route("/radar/link", post(link))
56
        .route("/signin", get(signin))
57
        .route("/signout", post(signout))
58
        .route("/radar/events", get(events))
59
        .route("/radar/style.css", get(|| async { asset("text/css; charset=utf-8", STYLE) }))
60
        .route("/radar/radar.js", get(|| async { asset("text/javascript; charset=utf-8", SCRIPT) }))
61
        .route("/radar/spectra.svg", get(|| async { asset("image/svg+xml", MARK) }))
62
        .layer(axum::middleware::map_response(security_headers))
63
        .with_state(state)
64
}
65
66
fn asset(kind: &'static str, body: &'static str) -> Response {
67
    ([(header::CONTENT_TYPE, kind), (header::CACHE_CONTROL, "max-age=300")], body).into_response()
68
}
69
70
/// CSP and friends on every response (spec §21.2).
71
async fn security_headers(mut r: Response) -> Response {
72
    let h = r.headers_mut();
73
    h.insert("content-security-policy", HeaderValue::from_static("default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; frame-ancestors 'none'; form-action 'self'; base-uri 'none'"));
74
    h.insert("x-content-type-options", HeaderValue::from_static("nosniff"));
75
    // `same-origin`: a form post still carries our Origin (no-referrer makes it `null`).
76
    h.insert("referrer-policy", HeaderValue::from_static("same-origin"));
77
    h.insert("x-frame-options", HeaderValue::from_static("DENY"));
78
    r
79
}
80
81
/// What every page's frame shows.
82
pub struct Shell {
83
    pub page: &'static str,
84
    pub owner: String,
85
    pub head: u64,
86
    pub version: &'static str,
87
    pub mark: &'static str,
88
}
89
90
/// The signed-in owner, or the page that says how to sign in.
91
struct Owner {
92
    session: Session,
93
    principal: Principal,
94
    shell: Shell,
95
}
96
97
impl Owner {
98
    fn author(&self) -> Author<'static> {
99
        Author {
100
            principal: self.principal.clone(),
101
            signer: Signer::Request(Auth {
102
                key: Some(self.session.key),
103
                method: Method::Session,
104
                assurance: 2,
105
                proof: Some(self.session.proof.clone()),
106
            }),
107
        }
108
    }
109
}
110
111
fn cookie(headers: &HeaderMap) -> Option<String> {
112
    let c = headers.get(header::COOKIE)?.to_str().ok()?;
113
    c.split(';').map(str::trim).find_map(|kv| kv.strip_prefix(&format!("{COOKIE}=")).map(String::from))
114
}
115
116
async fn owner(st: &RadarState, headers: &HeaderMap, page: &'static str) -> Result<Owner, Response> {
117
    let now = now_ms();
118
    let session = cookie(headers)
119
        .and_then(|id| st.sessions().get(&id, now))
120
        .ok_or_else(|| signin_page("Radar shows Spectra to its owner only.", StatusCode::UNAUTHORIZED))?;
121
    let d = &st.app.daemon;
122
    let principal = d
123
        .principals
124
        .read()
125
        .await
126
        .principal(&key_id(session.key), now)
127
        .map_err(|why| signin_page(&format!("This session no longer belongs to the owner: {why}."), StatusCode::UNAUTHORIZED))?;
128
    if !matches!(principal, Principal::Owner { .. }) {
129
        return Err(signin_page("Radar shows Spectra to its owner only.", StatusCode::FORBIDDEN));
130
    }
131
    let shell = Shell { page, owner: principal.display(), head: d.ledger.head().await.0, version: env!("CARGO_PKG_VERSION"), mark: MARK };
132
    Ok(Owner { session, principal, shell })
133
}
134
135
fn key_id(k: [u8; 32]) -> spectra_core::identity::KeyId {
136
    spectra_core::identity::KeyId(k)
137
}
138
139
#[derive(Template)]
140
#[template(path = "signin.html")]
141
struct SigninPage<'a> {
142
    mark: &'static str,
143
    message: &'a str,
144
}
145
146
fn signin_page(message: &str, status: StatusCode) -> Response {
147
    (status, render(&SigninPage { mark: MARK, message })).into_response()
148
}
149
150
fn render<T: Template>(t: &T) -> Response {
151
    match t.render() {
152
        Ok(html) => Html(html).into_response(),
153
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("could not render the page: {e}")).into_response(),
154
    }
155
}
156
157
/// Unsafe requests: the session's CSRF token and our own origin.
158
/// The refusal, if the form may not go through.
159
fn check_form(st: &RadarState, o: &Owner, headers: &HeaderMap, csrf: &str) -> Option<Response> {
160
    if !spectra_mcp::http::origin_ok(headers, &st.app.origins) {
161
        return Some((StatusCode::FORBIDDEN, "this origin may not change Spectra").into_response());
162
    }
163
    if csrf != o.session.csrf {
164
        return Some((StatusCode::FORBIDDEN, "the form is stale: reload the page").into_response());
165
    }
166
    None
167
}
168
169
/// Markdown, sanitised: what a message's body becomes on a page.
170
fn markdown(body: &str) -> String {
171
    let mut html = String::new();
172
    pulldown_cmark::html::push_html(
173
        &mut html,
174
        pulldown_cmark::Parser::new_ext(body, pulldown_cmark::Options::ENABLE_STRIKETHROUGH | pulldown_cmark::Options::ENABLE_TABLES),
175
    );
176
    ammonia::clean(&html)
177
}
178
179
// ---- Live ----
180
181
struct AgentCard {
182
    name: String,
183
    owner: String,
184
    kind: String,
185
    project: String,
186
    presence: String,
187
    status: Option<String>,
188
    last_tool: Option<String>,
189
    last_at: String,
190
}
191
192
struct EventRow {
193
    seq: i64,
194
    at: String,
195
    kind: String,
196
    actor: String,
197
    subject: Option<String>,
198
}
199
200
#[derive(Template)]
201
#[template(path = "live.html")]
202
struct LivePage {
203
    shell: Shell,
204
    agents: Vec<AgentCard>,
205
    unread: i64,
206
    events: Vec<EventRow>,
207
}
208
209
async fn live(State(st): State<RadarState>, headers: HeaderMap) -> Response {
210
    let o = match owner(&st, &headers, "live").await {
211
        Ok(o) => o,
212
        Err(r) => return r,
213
    };
214
    let d = &st.app.daemon;
215
    let pool = d.ledger.pool();
216
    let rows: Vec<(String, String, String, String, Option<String>)> =
217
        sqlx::query_as("SELECT a.name, a.owner_name, a.kind, a.project, p.status_text FROM agents a LEFT JOIN presence p ON p.agent = a.name WHERE a.disabled_ms IS NULL ORDER BY a.name")
218
            .fetch_all(pool)
219
            .await
220
            .unwrap_or_default();
221
    let mut agents = Vec::new();
222
    {
223
        let now = now_ms();
224
        for (name, owner, kind, project, status) in rows {
225
            let last: Option<(String, i64)> =
226
                sqlx::query_as("SELECT tool, at_ms FROM tool_calls WHERE agent = ? ORDER BY seq DESC LIMIT 1")
227
                    .bind(&name)
228
                    .fetch_optional(pool)
229
                    .await
230
                    .ok()
231
                    .flatten();
232
            let presence = d.presence.lock().await.state(&name, now).to_string();
233
            agents.push(AgentCard {
234
                name,
235
                owner,
236
                kind,
237
                project,
238
                presence,
239
                status,
240
                last_tool: last.as_ref().map(|(t, _)| t.clone()),
241
                last_at: last.map(|(_, at)| format_time(at as u64)).unwrap_or_default(),
242
            });
243
        }
244
    }
245
    let unread = conv::unread(&d.ledger, "human").await.unwrap_or(0);
246
    let events = recent(&st, None, None, 50)
247
        .await
248
        .into_iter()
249
        .map(|e| EventRow { seq: e.seq, at: e.at, kind: e.kind, actor: e.actor, subject: e.subject })
250
        .collect();
251
    render(&LivePage { shell: o.shell, agents, unread, events })
252
}
253
254
// ---- Channels ----
255
256
struct ChannelRow {
257
    name: String,
258
    display: String,
259
    kind: String,
260
    purpose: String,
261
    count: i64,
262
    last: String,
263
}
264
265
/// A channel with its message count and last message time.
266
type ChannelCount = (String, String, String, Option<String>, i64, Option<i64>);
267
268
#[derive(Template)]
269
#[template(path = "channels.html")]
270
struct ChannelsPage {
271
    shell: Shell,
272
    channels: Vec<ChannelRow>,
273
}
274
275
async fn channels(State(st): State<RadarState>, headers: HeaderMap) -> Response {
276
    let o = match owner(&st, &headers, "channels").await {
277
        Ok(o) => o,
278
        Err(r) => return r,
279
    };
280
    let rows: Vec<ChannelCount> = sqlx::query_as(
281
        "SELECT c.id, c.name, c.kind, c.purpose, (SELECT COUNT(*) FROM messages m WHERE m.channel = c.id), (SELECT MAX(at_ms) FROM messages m WHERE m.channel = c.id) \
282
         FROM channels c ORDER BY c.kind = 'direct', c.kind = 'system', c.name",
283
    )
284
    .fetch_all(st.app.daemon.ledger.pool())
285
    .await
286
    .unwrap_or_default();
287
    let channels = rows
288
        .into_iter()
289
        .map(|(_, name, kind, purpose, count, last)| ChannelRow {
290
            display: match name.strip_prefix("dm:") {
291
                Some(p) => format!("@{p}"),
292
                None => format!("#{name}"),
293
            },
294
            name,
295
            kind,
296
            purpose: purpose.unwrap_or_default(),
297
            count,
298
            last: last.map(|t| format_time(t as u64)).unwrap_or_default(),
299
        })
300
        .collect();
301
    render(&ChannelsPage { shell: o.shell, channels })
302
}
303
304
// ---- A channel and a thread ----
305
306
struct MessageRow {
307
    id: String,
308
    author: String,
309
    at: String,
310
    kind: String,
311
    replies: i64,
312
    redacted: bool,
313
    body_html: String,
314
}
315
316
#[derive(Template)]
317
#[template(path = "channel.html")]
318
struct ChannelPage {
319
    shell: Shell,
320
    name: String,
321
    display: String,
322
    purpose: Option<String>,
323
    thread: Option<String>,
324
    messages: Vec<MessageRow>,
325
    can_post: bool,
326
    action: String,
327
    csrf: String,
328
}
329
330
async fn show(st: &RadarState, o: Owner, name: &str, thread: Option<&str>) -> Response {
331
    let d = &st.app.daemon;
332
    let (ch, msgs) = match conv::read(&d.ledger, &o.principal, if thread.is_some() { None } else { Some(name) }, thread, None, 200).await {
333
        Ok(r) => r,
334
        Err(e) => return (StatusCode::NOT_FOUND, format!("{e:#}")).into_response(),
335
    };
336
    let mut messages = Vec::new();
337
    for m in msgs {
338
        messages.push(MessageRow {
339
            author: spectra_core::conversation::display_name(&d.ledger, &m.author).await,
340
            at: format_time(m.at_ms as u64),
341
            redacted: m.redacted.is_some(),
342
            body_html: if m.redacted.is_some() { ammonia::clean_text(&m.body) } else { markdown(&m.body) },
343
            id: m.id,
344
            kind: m.kind,
345
            replies: m.replies,
346
        });
347
    }
348
    let can_post = conv::can_post(&o.principal, &ch).is_ok();
349
    let action = match thread {
350
        Some(t) => format!("/c/{}/t/{t}", ch.name),
351
        None => format!("/c/{}", ch.name),
352
    };
353
    let csrf = o.session.csrf.clone();
354
    render(&ChannelPage {
355
        shell: o.shell,
356
        name: ch.name.clone(),
357
        display: ch.display(),
358
        purpose: ch.purpose.clone(),
359
        thread: thread.map(String::from),
360
        messages,
361
        can_post,
362
        action,
363
        csrf,
364
    })
365
}
366
367
async fn channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>) -> Response {
368
    match owner(&st, &headers, "channels").await {
369
        Ok(o) => show(&st, o, &name, None).await,
370
        Err(r) => r,
371
    }
372
}
373
374
async fn thread(State(st): State<RadarState>, headers: HeaderMap, Path((name, id)): Path<(String, String)>) -> Response {
375
    match owner(&st, &headers, "channels").await {
376
        Ok(o) => show(&st, o, &name, Some(&id)).await,
377
        Err(r) => r,
378
    }
379
}
380
381
#[derive(Deserialize)]
382
struct PostForm {
383
    _csrf: String,
384
    body: String,
385
    reply_to: Option<String>,
386
}
387
388
async fn post_to(st: &RadarState, headers: &HeaderMap, name: &str, form: PostForm, back: String) -> Response {
389
    let o = match owner(st, headers, "channels").await {
390
        Ok(o) => o,
391
        Err(r) => return r,
392
    };
393
    if let Some(r) = check_form(st, &o, headers, &form._csrf) {
394
        return r;
395
    }
396
    let post = Post {
397
        channel: if form.reply_to.is_some() { None } else { Some(name.to_string()) },
398
        body: form.body,
399
        reply_to: form.reply_to,
400
        ..Default::default()
401
    };
402
    match conv::post(&st.app.daemon.ledger, &o.author(), post, now_ms()).await {
403
        Ok(_) => Redirect::to(&back).into_response(),
404
        Err(e) => (StatusCode::BAD_REQUEST, format!("{e:#}")).into_response(),
405
    }
406
}
407
408
async fn post_channel(State(st): State<RadarState>, headers: HeaderMap, Path(name): Path<String>, Form(form): Form<PostForm>) -> Response {
409
    let back = format!("/c/{name}");
410
    post_to(&st, &headers, &name, form, back).await
411
}
412
413
async fn post_thread(
414
    State(st): State<RadarState>,
415
    headers: HeaderMap,
416
    Path((name, id)): Path<(String, String)>,
417
    Form(mut form): Form<PostForm>,
418
) -> Response {
419
    form.reply_to = Some(id.clone());
420
    let back = format!("/c/{name}/t/{id}");
421
    post_to(&st, &headers, &name, form, back).await
422
}
423
424
// ---- The log ----
425
426
struct LogRow {
427
    seq: i64,
428
    id: String,
429
    at: String,
430
    kind: String,
431
    actor: String,
432
    subject: Option<String>,
433
    method: String,
434
    assurance: i64,
435
    hash: String,
436
    prev: String,
437
    key: String,
438
    proof: String,
439
    payload: String,
440
}
441
442
async fn recent(st: &RadarState, kind: Option<&str>, before: Option<i64>, limit: i64) -> Vec<LogRow> {
443
    #[allow(clippy::type_complexity)]
444
    let rows: Vec<(
445
        i64,
446
        String,
447
        i64,
448
        String,
449
        String,
450
        Option<String>,
451
        i64,
452
        i64,
453
        Vec<u8>,
454
        Vec<u8>,
455
        Option<Vec<u8>>,
456
        Option<Vec<u8>>,
457
        Option<Vec<u8>>,
458
    )> = sqlx::query_as(
459
        "SELECT seq, id, at_ms, kind, actor, subject, auth_method, auth_assurance, hash, prev, auth_key, auth_proof, payload FROM events \
460
         WHERE (? IS NULL OR kind = ?) AND (? IS NULL OR seq < ?) ORDER BY seq DESC LIMIT ?",
461
    )
462
    .bind(kind)
463
    .bind(kind)
464
    .bind(before)
465
    .bind(before)
466
    .bind(limit)
467
    .fetch_all(st.app.daemon.ledger.pool())
468
    .await
469
    .unwrap_or_default();
470
    let mut out = Vec::new();
471
    for (seq, id, at, kind, actor, subject, method, assurance, hash, prev, key, proof, payload) in rows {
472
        let payload = payload
473
            .and_then(|p| serde_json::from_slice::<serde_json::Value>(&p).ok())
474
            .map(|v| serde_json::to_string_pretty(&v).unwrap_or_default())
475
            .unwrap_or_else(|| "(pruned to its stub)".into());
476
        out.push(LogRow {
477
            seq,
478
            id,
479
            at: format_time(at as u64),
480
            kind,
481
            actor: spectra_core::conversation::display_name(&st.app.daemon.ledger, &actor).await,
482
            subject,
483
            method: Method::from_u8(method as u8).map(|m| m.name()).unwrap_or("?").to_string(),
484
            assurance,
485
            hash: hex::encode(hash),
486
            prev: hex::encode(prev),
487
            key: key.map(hex::encode).unwrap_or_else(|| "none (Spectra itself; covered by the next checkpoint)".into()),
488
            proof: proof.map(|p| format!("{} bytes", p.len())).unwrap_or_else(|| "none".into()),
489
            payload,
490
        });
491
    }
492
    out
493
}
494
495
#[derive(Template)]
496
#[template(path = "log.html")]
497
struct LogPage {
498
    shell: Shell,
499
    head_hash: String,
500
    verified: String,
501
    csrf: String,
502
    kinds: Vec<String>,
503
    kind: String,
504
    events: Vec<LogRow>,
505
    older: Option<i64>,
506
}
507
508
#[derive(Deserialize)]
509
struct LogQuery {
510
    kind: Option<String>,
511
    before: Option<i64>,
512
    around: Option<i64>,
513
}
514
515
async fn log(State(st): State<RadarState>, headers: HeaderMap, Query(q): Query<LogQuery>) -> Response {
516
    let o = match owner(&st, &headers, "log").await {
517
        Ok(o) => o,
518
        Err(r) => return r,
519
    };
520
    let d = &st.app.daemon;
521
    let (_, head) = d.ledger.head().await;
522
    let last: Option<(Vec<u8>, i64)> =
523
        sqlx::query_as("SELECT payload, at_ms FROM events WHERE kind = 'ledger.verified' ORDER BY seq DESC LIMIT 1")
524
            .fetch_optional(d.ledger.pool())
525
            .await
526
            .ok()
527
            .flatten();
528
    let verified = match last {
529
        Some((p, at)) => {
530
            let v: serde_json::Value = serde_json::from_slice(&p).unwrap_or_default();
531
            let n = v["findings"].as_u64().unwrap_or(0);
532
            format!(
533
                "{} at {} over {} events",
534
                if n == 0 { "clean".to_string() } else { format!("{n} finding(s)") },
535
                format_time(at as u64),
536
                v["events"].as_u64().unwrap_or(0)
537
            )
538
        }
539
        None => "not yet".into(),
540
    };
541
    let kinds: Vec<String> =
542
        sqlx::query_scalar("SELECT kind FROM event_counts ORDER BY n DESC LIMIT 12").fetch_all(d.ledger.pool()).await.unwrap_or_default();
543
    let kind = q.kind.unwrap_or_default();
544
    let before = q.before.or(q.around.map(|a| a + 25));
545
    let events = recent(&st, (!kind.is_empty()).then_some(kind.as_str()), before, 100).await;
546
    let older = (events.len() == 100).then(|| events.last().map(|e| e.seq)).flatten();
547
    let csrf = o.session.csrf.clone();
548
    render(&LogPage { shell: o.shell, head_hash: hex::encode(head), verified, csrf, kinds, kind, events, older })
549
}
550
551
#[derive(Deserialize)]
552
struct CsrfForm {
553
    _csrf: String,
554
}
555
556
/// Verify the ledger now and record the result (`ledger.verified`).
557
async fn verify(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response {
558
    let o = match owner(&st, &headers, "log").await {
559
        Ok(o) => o,
560
        Err(r) => return r,
561
    };
562
    if let Some(r) = check_form(&st, &o, &headers, &f._csrf) {
563
        return r;
564
    }
565
    let ledger = &st.app.daemon.ledger;
566
    match ledger.verify(&mut Principals::default()).await {
567
        Ok(v) => {
568
            let e = spectra_core::event::NewEvent::system(
569
                "ledger.verified",
570
                serde_json::json!({ "events": v.events, "checkpoints": v.checkpoints, "findings": v.findings.len(), "head": hex::encode(v.head), "first": v.findings.first().map(|f| format!("#{}: {}", f.seq, f.detail)) }),
571
            );
572
            let _ = ledger.append_one(e).await;
573
            Redirect::to("/log").into_response()
574
        }
575
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:#}")).into_response(),
576
    }
577
}
578
579
// ---- Signing in ----
580
581
/// `POST /radar/link`, signed by the owner's CLI: a one-time sign-in link.
582
async fn link(State(st): State<RadarState>, ConnectInfo(peer): ConnectInfo<PeerInfo>, headers: HeaderMap, body: Bytes) -> Response {
583
    let caller = match spectra_mcp::http::authenticate(&st.app, peer, &headers, "POST /radar/link", &body).await {
584
        Ok(c) => c,
585
        Err(r) => return r,
586
    };
587
    if !matches!(caller.principal, Principal::Owner { .. }) {
588
        return (StatusCode::FORBIDDEN, axum::Json(serde_json::json!({ "title": "only the owner signs in to Radar" }))).into_response();
589
    }
590
    let (Some(key), Some(proof)) = (caller.auth.key, caller.auth.proof) else {
591
        return StatusCode::UNAUTHORIZED.into_response();
592
    };
593
    let code = st.sessions().link(key, proof, now_ms());
594
    let base = st.app.origins.first().cloned().unwrap_or_default();
595
    axum::Json(serde_json::json!({ "url": format!("{base}/signin?code={code}"), "valid_ms": session::LINK_MS })).into_response()
596
}
597
598
#[derive(Deserialize)]
599
struct SigninQuery {
600
    code: Option<String>,
601
}
602
603
async fn signin(State(st): State<RadarState>, Query(q): Query<SigninQuery>) -> Response {
604
    let Some(id) = q.code.and_then(|c| st.sessions().open(&c, now_ms())) else {
605
        return signin_page("That sign-in link has been used or has expired.", StatusCode::UNAUTHORIZED);
606
    };
607
    let cookie = format!("{COOKIE}={id}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}", session::SESSION_MS / 1000);
608
    let mut r = Redirect::to("/").into_response();
609
    r.headers_mut().insert(header::SET_COOKIE, HeaderValue::from_str(&cookie).unwrap());
610
    r
611
}
612
613
async fn signout(State(st): State<RadarState>, headers: HeaderMap, Form(f): Form<CsrfForm>) -> Response {
614
    if let Ok(o) = owner(&st, &headers, "live").await {
615
        if check_form(&st, &o, &headers, &f._csrf).is_none() {
616
            if let Some(id) = cookie(&headers) {
617
                st.sessions().close(&id);
618
            }
619
        }
620
    }
621
    signin_page("Signed out.", StatusCode::OK)
622
}
623
624
// ---- The event stream ----
625
626
/// `GET /radar/events`: one SSE message per appended event, carrying its
627
/// sequence number, kind, actor and subject, never its data (spec §17.3).
628
async fn events(State(st): State<RadarState>, headers: HeaderMap) -> Response {
629
    if let Err(r) = owner(&st, &headers, "live").await {
630
        return r;
631
    }
632
    let rx = st.app.daemon.ledger.subscribe();
633
    let stream = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(|e| async move {
634
        let e = e.ok()?;
635
        let data = serde_json::json!({ "seq": e.seq, "kind": e.kind, "actor": e.actor, "subject": e.subject });
636
        Some(Ok::<_, std::convert::Infallible>(SseEvent::default().data(data.to_string())))
637
    });
638
    Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))).into_response()
639
}