alopex_server/http/
session.rs

1use std::sync::Arc;
2
3use axum::extract::{Extension, Path};
4use axum::response::Response;
5use serde::Serialize;
6
7use crate::error::{Result, ServerError};
8use crate::http::sql::sync_catalog_to_store;
9use crate::http::{error_response, json_response, RequestContext};
10use crate::ops::distributed_read::ReadExecutionOwner;
11use crate::server::ServerState;
12use crate::session::SessionId;
13
14#[derive(Serialize)]
15struct SessionBeginResponse {
16    session_id: String,
17    expires_at: String,
18}
19
20#[derive(Serialize)]
21struct SessionActionResponse {
22    success: bool,
23}
24
25pub async fn begin(
26    Extension(state): Extension<Arc<ServerState>>,
27    Extension(ctx): Extension<RequestContext>,
28) -> Response {
29    match begin_session(state.clone()).await {
30        Ok(resp) => json_response(resp, state.config.max_response_size, &ctx),
31        Err(err) => error_response(err, &ctx),
32    }
33}
34
35pub async fn commit(
36    Extension(state): Extension<Arc<ServerState>>,
37    Extension(ctx): Extension<RequestContext>,
38    Path(id): Path<String>,
39) -> Response {
40    match session_action(state.clone(), &id, Action::Commit).await {
41        Ok(resp) => json_response(resp, state.config.max_response_size, &ctx),
42        Err(err) => error_response(err, &ctx),
43    }
44}
45
46pub async fn rollback(
47    Extension(state): Extension<Arc<ServerState>>,
48    Extension(ctx): Extension<RequestContext>,
49    Path(id): Path<String>,
50) -> Response {
51    match session_action(state.clone(), &id, Action::Rollback).await {
52        Ok(resp) => json_response(resp, state.config.max_response_size, &ctx),
53        Err(err) => error_response(err, &ctx),
54    }
55}
56
57/// Bind a distributed-read registration to the authenticated HTTP profile and
58/// (when supplied) to a live SQL session. The session is correlation state;
59/// the profile remains the authorization authority for later stream and cancel
60/// requests, so a guessed session ID cannot broaden access.
61pub async fn distributed_read_owner(
62    state: &ServerState,
63    ctx: &RequestContext,
64    session_id: Option<&str>,
65) -> Result<ReadExecutionOwner> {
66    let profile = ctx.actor.clone().ok_or_else(|| {
67        ServerError::Unauthorized("distributed read requires an authenticated profile".into())
68    })?;
69    let session_id = match session_id {
70        Some(id) => {
71            let session_id = id
72                .parse::<SessionId>()
73                .map_err(|_| ServerError::BadRequest("invalid session id".into()))?;
74            // Preserve the existing session expiry/not-found classification.
75            state.session_manager.get_session(&session_id).await?;
76            Some(session_id)
77        }
78        None => None,
79    };
80    ReadExecutionOwner::new(profile, session_id)
81}
82
83async fn begin_session(state: Arc<ServerState>) -> Result<SessionBeginResponse> {
84    let session_id = state.session_manager.create_session().await?;
85    state.session_manager.begin_transaction(&session_id).await?;
86    let snapshot = state.session_manager.get_session(&session_id).await?;
87    let expires_at = chrono::DateTime::<chrono::Utc>::from(snapshot.expires_at);
88    Ok(SessionBeginResponse {
89        session_id: session_id.to_string(),
90        expires_at: expires_at.to_rfc3339(),
91    })
92}
93
94enum Action {
95    Commit,
96    Rollback,
97}
98
99async fn session_action(
100    state: Arc<ServerState>,
101    id: &str,
102    action: Action,
103) -> Result<SessionActionResponse> {
104    let session_id = id
105        .parse::<SessionId>()
106        .map_err(|_| ServerError::BadRequest("invalid session id".into()))?;
107    match action {
108        Action::Commit => {
109            let effects = state.session_manager.commit(&session_id).await?;
110            if !effects.is_empty() {
111                state.apply_table_lifecycle_effects(effects)?;
112                sync_catalog_to_store(&state)?;
113            }
114        }
115        Action::Rollback => {
116            let effects = state.session_manager.rollback(&session_id).await?;
117            state.apply_catalog_rollback_effects(effects)?;
118        }
119    }
120    Ok(SessionActionResponse { success: true })
121}
alopex_server/http/session.rs

alopex_server/http/
session.rs
