use super::{
assistant_state::{Event, Pair, Subscription},
http::{Request, Response},
route::{event, sent},
signin::{Door, Issuer},
};
use crate::serve::{pass::now, storage::Failure};
use serde_json::json;
use std::{io::Write, path::PathBuf, sync::mpsc, time::Duration};
#[path = "assistant_cursor.rs"]
mod cursor_output;
use cursor_output::Output;
pub(super) fn events<I: Issuer>(
door: &Door<I>,
request: &Request,
pair: Pair,
socket: PathBuf,
timeout: Duration,
) -> Result<Response, Failure> {
let (conversation, after) = cursor(request)?;
let session = request
.cookie("haem_session")
.ok_or_else(invalid)?
.to_owned();
let hub = door.assistant()?;
let subscription = match hub.subscribe_replay(pair.clone(), conversation.clone()) {
Ok(subscription) => subscription,
Err(error) if error.code == "assistant_bound" => {
return Ok(Response::json(
429,
&json!({"code":error.code,"message":error.message}),
));
}
Err(error) => return Err(error),
};
let door = door.clone();
Ok(Response::stream_with_peer(
"text/event-stream",
move |output, peer| {
let monitor = peer
.map(|peer| {
let hub = std::sync::Arc::clone(&subscription.hub);
let id = subscription.id.clone();
crate::serve::follow::Monitor::start(peer.try_clone()?, move || {
if let Err(error) = hub.close(&id) {
eprintln!("haematite assistant_stream_close_failed: {}", error.code);
}
})
})
.transpose()?;
let selection = Selection {
person: &pair.0,
socket: &socket,
timeout,
conversation: conversation.as_deref(),
after,
};
let result = relay(&door, &session, &subscription, output, &selection);
drop(monitor);
match result {
Ok(()) => Ok(()),
Err(error) if error.code == "assistant_stream_closed" => Ok(()),
Err(error) => {
let kind = match error.code.as_str() {
"pass_expired" => "pass_expired",
"lys_unavailable" => "lys_unavailable",
_ => "refused",
};
sent(
output,
&event(
kind,
&json!({"code":error.code,"message":error.message}).to_string(),
),
)?;
Ok(())
}
}
},
))
}
fn cursor(request: &Request) -> Result<(Option<String>, u64), Failure> {
let named = request.parameter("conversation");
let (conversation, after) = if let Some(last) = request.header("last-event-id") {
let (id, seq) = last.rsplit_once(':').ok_or_else(invalid)?;
if named
.as_deref()
.is_some_and(|conversation| conversation != id)
{
return Err(invalid());
}
(
Some(id.to_owned()),
Some(seq.parse().map_err(|_| invalid())?),
)
} else {
let after = request
.parameter("after")
.map(|seq| seq.parse::<u64>().map_err(|_| invalid()))
.transpose()?;
(named, after)
};
if conversation
.as_deref()
.is_some_and(|id| !super::assistant_wire::identifier(id))
|| (after.is_some() && conversation.is_none())
{
return Err(invalid());
}
Ok((conversation, after.unwrap_or(0)))
}
struct LivePass {
token: String,
expires: u64,
valid: std::sync::Arc<std::sync::atomic::AtomicBool>,
}
fn pass<I: Issuer>(door: &Door<I>, session: &str, person: &str) -> Result<LivePass, Failure> {
let now = now();
let token = door.pass(session, now).map_err(|error| {
if matches!(
error.code.as_str(),
"lys_could_not_be_asked" | "lys_unavailable"
) {
Failure::new(
"lys_unavailable",
"Lys is temporarily unavailable; reconnect to retry this stream",
)
} else {
Failure::new(
"pass_expired",
"the person's pass could not be refreshed; sign in again",
)
}
})?;
let settings = door.session_at(session, now)?;
if settings["holder"] != person {
return Err(Failure::new(
"pass_expired",
"the session's person changed; sign in again",
));
}
let expires = settings["expires"].as_u64().ok_or_else(invalid)?;
Ok(LivePass {
token,
expires,
valid: door.session_validity(session, now)?,
})
}
struct Selection<'a> {
person: &'a str,
socket: &'a std::path::Path,
timeout: Duration,
conversation: Option<&'a str>,
after: u64,
}
fn relay<I: Issuer>(
door: &Door<I>,
session: &str,
subscription: &Subscription,
writer: &mut dyn Write,
selection: &Selection<'_>,
) -> Result<(), Failure> {
let Selection {
person,
socket,
timeout,
conversation,
after,
} = *selection;
let mut output = Output::new(
writer,
conversation,
after,
subscription.hub.config.events_after_bytes.get(),
);
let mut live = pass(door, session, person)?;
match subscription.input.recv() {
Ok(Event::Item(presence)) => output.item(&presence)?,
Ok(Event::Closed) | Err(_) => return Ok(()),
}
if let Some(conversation) = conversation {
let record = super::assistant_record::Record {
socket,
timeout,
pass: &live.token,
person,
bound: subscription.hub.config.events_after_bytes.get(),
};
let mut reader = record.open()?;
let mut from = after;
loop {
reader.replay(conversation, from, |item| {
if !live.valid.load(std::sync::atomic::Ordering::Acquire) {
return Err(Failure::new(
"pass_expired",
"the session ended; sign in again",
));
}
output.item(&item)
})?;
if let Some(held) = subscription.hub.finish_replay(&subscription.id)? {
for item in held {
output.item(&item)?;
}
break;
}
from = output.through(conversation);
}
drop(reader);
output.replayed(conversation)?;
}
loop {
let left = Duration::from_secs(live.expires.saturating_sub(now()));
match subscription.input.recv_timeout(left) {
Ok(Event::Item(item)) => {
if !live.valid.load(std::sync::atomic::Ordering::Acquire) {
return Err(Failure::new(
"pass_expired",
"the session ended; sign in again",
));
}
if now() >= live.expires {
live = pass(door, session, person)?;
}
output.item(&item)?;
}
Ok(Event::Closed) => return Ok(()),
Err(mpsc::RecvTimeoutError::Timeout) => {
live = pass(door, session, person)?;
}
Err(mpsc::RecvTimeoutError::Disconnected) => {
return Err(Failure::new(
"assistant_stream_lagged",
"the live item queue reached its max_per_store bound; reconnect from the last event id",
));
}
}
}
}
fn invalid() -> Failure {
Failure::new(
"assistant_cursor_invalid",
"use a conversation and an unsigned after cursor, or Last-Event-ID conversation:sequence",
)
}
#[cfg(test)]
#[path = "assistant_pass_revision_tests.rs"]
mod pass_revision;