lean_ctx/core/dashboard_pro/
bridge.rs

1// SPDX-License-Identifier: Apache-2.0
2//! One cockpit analysis in two engine phases around the cockpit's own
3//! computation: `prepare` checks the request and collects its inputs,
4//! `finish` checks the cockpit's answer and returns the body, a file, or the
5//! executed plans.
6
7use chrono::{DateTime, Utc};
8use serde_json::{Value, json};
9
10use super::collect::{self, RouteSource};
11use super::effects::{self, Ctx, Notifier, Outcome};
12use super::request::{self, BridgeRequest, Op};
13use super::wire::{self, GuardEvent};
14use super::{ProError, ProResult, ProjectScope, watch};
15
16/// One analysis frame: requests and answers are at most 1 MiB.
17pub(crate) const MAX_FRAME_BYTES: usize = 1 << 20;
18
19pub struct Env<'a> {
20    pub scope: &'a ProjectScope,
21    pub routes: &'a dyn RouteSource,
22    pub notifier: &'a dyn Notifier,
23    pub now: DateTime<Utc>,
24}
25
26/// An HTTP answer for the cockpit.
27#[derive(Debug, Clone, PartialEq)]
28pub struct Reply {
29    pub status: u16,
30    pub body: Value,
31}
32
33impl Reply {
34    fn ok(body: Value) -> Self {
35        Self { status: 200, body }
36    }
37
38    fn error(error: &ProError) -> Self {
39        Self {
40            status: error.status,
41            body: super::error_body(error),
42        }
43    }
44}
45
46/// `POST /api/dashboard-analysis/inputs` `{op, method, query, body}` →
47/// `{request}` for the cockpit's computation. `alerts.evaluate` answers
48/// `{skip: true}` when there is nothing to evaluate, else also
49/// `guard_cursor`, which the cockpit hands back to `apply`.
50pub fn handle_inputs(env: &Env<'_>, raw: &str) -> Reply {
51    let request = match request::parse(raw).and_then(|r| scoped(env, r)) {
52        Ok(request) => request,
53        Err(error) => return Reply::error(&error),
54    };
55    let (guard_new, guard_cursor) = if request.op == Op::AlertsEvaluate {
56        match watch::pending(env.scope) {
57            Ok(Some((events, cursor))) => (Some(events), Some(cursor)),
58            Ok(None) => return Reply::ok(json!({ "skip": true })),
59            Err(error) => return Reply::error(&error),
60        }
61    } else {
62        (None, None)
63    };
64    match prepare(env, &request, guard_new) {
65        Ok(prepared) => {
66            let mut body = json!({ "request": prepared });
67            if let Some(cursor) = guard_cursor {
68                body["guard_cursor"] = json!(cursor);
69            }
70            Reply::ok(body)
71        }
72        Err(error) => Reply::error(&error),
73    }
74}
75
76/// `POST /api/dashboard-analysis/apply` `{op, method, query, body, response,
77/// guard_cursor?}` → `{status, body | file | applied}`; a refusal keeps its
78/// status and answers `{error}`.
79pub fn handle_apply(env: &Env<'_>, raw: &str) -> Reply {
80    if raw.len() > MAX_FRAME_BYTES + request::MAX_REQUEST_BYTES {
81        return Reply::error(&ProError::new(413, "Request too large"));
82    }
83    let Ok(Value::Object(mut object)) = serde_json::from_str::<Value>(raw) else {
84        return Reply::error(&ProError::bad("Invalid JSON request"));
85    };
86    let response = object.remove("response");
87    let guard_cursor = object.remove("guard_cursor");
88    let request = match request::from_object(object).and_then(|r| scoped(env, r)) {
89        Ok(request) => request,
90        Err(error) => return Reply::error(&error),
91    };
92    let Some(Ok(response)) = response.map(serde_json::from_value::<wire::Response>) else {
93        return Reply::error(&ProError::bad(
94            "The analysis answer is missing or unreadable",
95        ));
96    };
97    let cursor = match (request.op, guard_cursor.as_ref().map(Value::as_u64)) {
98        (Op::AlertsEvaluate, Some(Some(cursor))) => Some(cursor),
99        (Op::AlertsEvaluate, _) => {
100            return Reply::error(&ProError::bad("alerts.evaluate needs its guard_cursor"));
101        }
102        (_, Some(_)) => return Reply::error(&ProError::bad("Unknown request field")),
103        (_, None) => None,
104    };
105    let mut body = match finish(env, &request, response) {
106        Ok(body) => body,
107        Err(error) => return Reply::error(&error),
108    };
109    // The cursor moves only once the evaluated events were answered.
110    if let Some(cursor) = cursor
111        && let Err(error) = watch::advance(env.scope, cursor)
112    {
113        return Reply::error(&error);
114    }
115    body["status"] = json!(200);
116    Reply::ok(body)
117}
118
119/// The device scope reads every project's stores; it serves only ops that
120/// read device-wide stores, whichever route built it.
121fn scoped(env: &Env<'_>, request: BridgeRequest) -> ProResult<BridgeRequest> {
122    if env.scope.is_device() && !request.op.device_wide() {
123        return Err(ProError::bad(
124            "Choose a project: this view reads one project's data",
125        ));
126    }
127    Ok(request)
128}
129
130/// Collects the op's inputs into one bounded analysis request.
131pub(crate) fn prepare(
132    env: &Env<'_>,
133    request: &BridgeRequest,
134    guard_new: Option<Vec<GuardEvent>>,
135) -> ProResult<wire::Request> {
136    let collected = collect::inputs(request, env.scope, env.routes, env.now, guard_new)?;
137    let mut payload = wire::Request {
138        schema: wire::SCHEMA.to_string(),
139        op: request.op.name().to_string(),
140        now_utc: env.now.to_rfc3339(),
141        project: env.scope.project(),
142        query: request.query.clone(),
143        body: Value::Object(request.body.clone()),
144        inputs: collected.inputs,
145    };
146    let size = match request.op {
147        // Knowledge has no fact limit: a review reads the newest facts that
148        // fit; `facts_total` still counts them all.
149        Op::MemoryAnalyze | Op::MemorySnapshot => {
150            fit_newest(&mut payload, "facts", MAX_FRAME_BYTES)?
151        }
152        _ => encoded_len(&payload)?,
153    };
154    if size > MAX_FRAME_BYTES {
155        return Err(ProError::new(
156            413,
157            "This view needs more data than one analysis request can carry; narrow the period or filters, or consolidate memory first",
158        ));
159    }
160    Ok(payload)
161}
162
163fn encoded_len(payload: &wire::Request) -> ProResult<usize> {
164    serde_json::to_vec(payload)
165        .map(|bytes| bytes.len())
166        .map_err(|_| ProError::internal("The request could not be encoded"))
167}
168
169/// Drops the oldest tenth of the oldest-first list input `key` until the
170/// request fits `max` bytes. Returns the encoded size it ends with.
171pub(crate) fn fit_newest(payload: &mut wire::Request, key: &str, max: usize) -> ProResult<usize> {
172    let mut size = encoded_len(payload)?;
173    while size > max {
174        let Some(Value::Array(items)) = payload.inputs.get_mut(key) else {
175            break;
176        };
177        if items.is_empty() {
178            break;
179        }
180        let drop = items.len().div_ceil(10);
181        items.drain(..drop);
182        size = encoded_len(payload)?;
183    }
184    Ok(size)
185}
186
187/// Checks the cockpit's answer and executes its plans; returns `{body}`,
188/// `{file}` or `{applied, body}`.
189pub(crate) fn finish(
190    env: &Env<'_>,
191    request: &BridgeRequest,
192    response: wire::Response,
193) -> ProResult<Value> {
194    match response.status {
195        200 => {}
196        400 | 404 | 409 => {
197            let message = response
198                .error
199                .map(|e| e.message)
200                .filter(|m| super::plain_text(m, 500))
201                .unwrap_or_else(|| "The request was refused".into());
202            return Err(ProError::new(response.status, message));
203        }
204        _ => return Err(ProError::new(502, "The analysis answer is unreadable")),
205    }
206    if let Some(plans) = response.plans.filter(|p| !p.is_empty()) {
207        let plans = effects::parse_plans(request.op, plans)?;
208        let ctx = Ctx {
209            scope: env.scope,
210            request,
211            now: env.now,
212            notifier: env.notifier,
213        };
214        let mut applied = Vec::new();
215        for outcome in effects::apply(&ctx, plans)? {
216            match outcome {
217                Outcome::Applied(readback) => applied.push(readback),
218                Outcome::File(file) => return Ok(json!({ "file": file })),
219            }
220        }
221        return Ok(json!({ "applied": applied, "body": response.body }));
222    }
223    if request.op.is_write() {
224        // A write answered without a plan only reports (e.g. a preview).
225        return Ok(json!({ "body": response.body }));
226    }
227    if let Some(file) = response.file {
228        if !super::plain_text(&file.name, 120)
229            || !super::plain_text(&file.mime, 100)
230            || file.name.contains(['/', '\\'])
231        {
232            return Err(ProError::new(502, "The analysis file is refused"));
233        }
234        return Ok(json!({ "file": file }));
235    }
236    Ok(json!({ "body": response.body }))
237}
238
239/// The HTTP status line of a reply.
240pub fn status_line(status: u16) -> &'static str {
241    match status {
242        200 => "200 OK",
243        400 => "400 Bad Request",
244        404 => "404 Not Found",
245        405 => "405 Method Not Allowed",
246        409 => "409 Conflict",
247        413 => "413 Payload Too Large",
248        502 => "502 Bad Gateway",
249        _ => "500 Internal Server Error",
250    }
251}
lean_ctx/core/dashboard_pro/bridge.rs

lean_ctx/core/dashboard_pro/
bridge.rs
