use super::{DaemonCommand, queue, store};
use crate::{
ClientFrame, ServerFrame,
relay::{Operation, RelayAcknowledgement, RelayRequest},
};
use anyhow::{Context, Result, bail};
use axum::{
Json, Router,
extract::{DefaultBodyLimit, Path, State},
http::{HeaderMap, StatusCode},
response::{IntoResponse, Response},
routing::{get, post},
};
use fs2::FileExt;
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::{
collections::{BTreeMap, HashMap},
sync::Arc,
time::Duration,
};
use tokio::{
sync::{Mutex, OwnedSemaphorePermit, Semaphore, mpsc, watch},
task::{JoinHandle, JoinSet},
};
use tokio_tungstenite::tungstenite::Message;
use uuid::Uuid;
type Channels = Arc<Mutex<BTreeMap<String, mpsc::Sender<ClientFrame>>>>;
type Statuses = Arc<Mutex<BTreeMap<String, Value>>>;
#[derive(Clone)]
struct App {
context: RuntimeContext,
token: String,
shutdown: watch::Sender<bool>,
statuses: Statuses,
}
#[derive(Clone)]
struct RuntimeContext {
store: store::Store,
queue: queue::Queue,
payload_budget: Arc<Semaphore>,
}
const PAYLOAD_BUDGET_MIB: u32 = 128;
fn reserve_payload(context: &RuntimeContext, bytes: u32) -> Option<OwnedSemaphorePermit> {
let permits = bytes.div_ceil(1024 * 1024).clamp(1, PAYLOAD_BUDGET_MIB);
context
.payload_budget
.clone()
.try_acquire_many_owned(permits)
.ok()
}
struct AbortOnDrop(JoinHandle<()>);
impl Drop for AbortOnDrop {
fn drop(&mut self) {
self.0.abort();
}
}
struct ShutdownOnDrop(watch::Sender<bool>);
impl Drop for ShutdownOnDrop {
fn drop(&mut self) {
let _ = self.0.send(true);
}
}
pub async fn start(
state: &store::Store,
launch: &DaemonCommand,
port: Option<u16>,
) -> Result<Value> {
let config = state.load()?;
if let Ok(Ok(status)) =
tokio::time::timeout(Duration::from_secs(1), store::relay(&config)?.status()).await
{
return Ok(status);
}
if let Some(port) = port {
if port == 0 {
bail!("relay port must be greater than zero")
}
state.update(|c| {
c.relay_port = port;
Ok(())
})?;
}
let mut log = store::secure_open(&state.directory().join("daemon.log"))?;
use std::io::Seek;
log.seek(std::io::SeekFrom::End(0))?;
let executable = &launch.executable;
let mut command = std::process::Command::new(executable);
command
.args(&launch.arguments)
.env("SILICON_DM_HOME", state.directory())
.stdin(std::process::Stdio::null())
.stdout(log.try_clone()?)
.stderr(log);
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
unsafe {
command.pre_exec(|| {
if libc::setsid() < 0 {
Err(std::io::Error::last_os_error())
} else {
Ok(())
}
});
}
}
let mut child = tokio::process::Command::from(command)
.spawn()
.context("could not start relay daemon")?;
let (exited, mut exit_status) = tokio::sync::oneshot::channel();
tokio::spawn(async move {
let _ = exited.send(child.wait().await);
});
let readiness = async {
loop {
tokio::time::sleep(Duration::from_millis(100)).await;
if let Ok(status) = store::relay(&state.load()?)?.status().await {
return Ok::<Value, anyhow::Error>(status);
}
}
};
tokio::select! {
result = tokio::time::timeout(Duration::from_secs(5), readiness) => {
if let Ok(result) = result { return result; }
}
_ = &mut exit_status => {
bail!("relay exited before becoming ready; inspect {}", state.directory().join("daemon.log").display());
}
}
bail!(
"relay did not become ready; inspect {}",
state.directory().join("daemon.log").display()
)
}
pub async fn run(state: store::Store) -> Result<()> {
let lock = store::secure_open(&state.directory().join("daemon.lock"))?;
lock.try_lock_exclusive()
.context("a DM relay daemon is already running")?;
let context = RuntimeContext {
queue: queue::Queue::open(&state)?,
store: state,
payload_budget: Arc::new(Semaphore::new(PAYLOAD_BUDGET_MIB as usize)),
};
context.queue.expire_transient_commands()?;
let config = context.store.load()?;
let listener =
tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, config.relay_port)).await?;
let (shutdown, mut shutdown_rx) = watch::channel(false);
let _shutdown_guard = ShutdownOnDrop(shutdown.clone());
let channels: Channels = Arc::default();
let statuses: Statuses = Arc::default();
let app = App {
context: context.clone(),
token: config.relay_token.clone(),
shutdown: shutdown.clone(),
statuses: statuses.clone(),
};
let router = Router::new()
.route("/status", get(status))
.route("/requests", post(submit))
.route("/requests/{id}", get(request_result))
.route("/requests/{id}/status", get(request_status))
.route("/shutdown", post(stop))
.layer(DefaultBodyLimit::max(128 * 1024 * 1024))
.with_state(app);
let mut supervisor = AbortOnDrop(tokio::spawn(supervise(
context.clone(),
channels.clone(),
statuses,
)));
let mut outgoing = AbortOnDrop(tokio::spawn(outbox(context.clone(), channels)));
let mut callbacks = AbortOnDrop(tokio::spawn(webhooks(context.clone())));
let server = axum::serve(listener, router).with_graceful_shutdown(async move {
let _ = shutdown_rx.changed().await;
});
eprintln!(
"DM relay listening at http://dm.localhost:{} (127.0.0.1); durable state in {}",
config.relay_port,
context.store.directory().display()
);
tokio::select! {result=server=>{result?;},_=tokio::signal::ctrl_c()=>{let _=shutdown.send(true);}}
supervisor.0.abort();
outgoing.0.abort();
callbacks.0.abort();
let _ = tokio::join!(&mut supervisor.0, &mut outgoing.0, &mut callbacks.0);
FileExt::unlock(&lock)?;
Ok(())
}
fn authorized(headers: &HeaderMap, app: &App) -> bool {
if *app.shutdown.borrow() || headers.contains_key("origin") {
return false;
}
let expected = blake3::hash(format!("Bearer {}", app.token).as_bytes());
headers
.get("authorization")
.is_some_and(|value| blake3::hash(value.as_bytes()) == expected)
}
fn error(status: StatusCode, code: &str, message: &str) -> Response {
(
status,
Json(json!({"error":{"code":code,"message":message}})),
)
.into_response()
}
async fn status(State(app): State<App>, headers: HeaderMap) -> Response {
let context = &app.context;
if !authorized(&headers, &app) {
return error(
StatusCode::UNAUTHORIZED,
"unauthorized",
"local relay bearer token required",
);
}
match context.queue.stats(){Ok(queues)=>Json(json!({"running":true,"pid":std::process::id(),"version":env!("CARGO_PKG_VERSION"),"profiles":*app.statuses.lock().await,"queues":queues})).into_response(),Err(_)=>error(StatusCode::INTERNAL_SERVER_ERROR,"storage_error","local database unavailable")}
}
async fn submit(
State(app): State<App>,
headers: HeaderMap,
Json(original): Json<Value>,
) -> Response {
let context = &app.context;
if !authorized(&headers, &app) {
return error(
StatusCode::UNAUTHORIZED,
"unauthorized",
"local relay bearer token required",
);
}
let request = match RelayRequest::deserialize(&original) {
Ok(r) => r,
Err(_) => {
return error(
StatusCode::BAD_REQUEST,
"invalid_request",
"expected request_id, profile, optional testing_environment_id, and typed request; run dm relay submit --help",
);
}
};
let config = match context.store.load() {
Ok(c) => c,
Err(_) => {
return error(
StatusCode::INTERNAL_SERVER_ERROR,
"storage_error",
"cannot load profiles",
);
}
};
if store::profile(&config, &request.profile, request.testing_environment_id).is_err() {
return error(
StatusCode::BAD_REQUEST,
"profile_missing",
"log in to this profile and testing environment first",
);
}
let queued = context.queue.enqueue(&request, &original);
let request_id = request.request_id;
drop(request);
match queued {
Ok(()) => (
StatusCode::ACCEPTED,
Json(RelayAcknowledgement {
acknowledged: true,
request_id,
request: original,
}),
)
.into_response(),
Err(e) => error(StatusCode::CONFLICT, "queue_conflict", &e.to_string()),
}
}
async fn request_status(
State(app): State<App>,
headers: HeaderMap,
Path(id): Path<Uuid>,
) -> Response {
if !authorized(&headers, &app) {
return error(
StatusCode::UNAUTHORIZED,
"unauthorized",
"local relay bearer token required",
);
}
match app.context.queue.request_status(id) {
Ok(Some(status)) => Json(status).into_response(),
Ok(None) => error(StatusCode::NOT_FOUND, "not_found", "unknown request ID"),
Err(_) => error(
StatusCode::INTERNAL_SERVER_ERROR,
"storage_error",
"cannot read request state",
),
}
}
async fn request_result(
State(app): State<App>,
headers: HeaderMap,
Path(id): Path<Uuid>,
) -> Response {
let context = &app.context;
if !authorized(&headers, &app) {
return error(
StatusCode::UNAUTHORIZED,
"unauthorized",
"local relay bearer token required",
);
}
match context.queue.result(id) {
Ok(Some(result)) => Json(result).into_response(),
Ok(None) => error(StatusCode::NOT_FOUND, "not_found", "unknown request ID"),
Err(_) => error(
StatusCode::INTERNAL_SERVER_ERROR,
"storage_error",
"cannot read request state",
),
}
}
async fn stop(State(app): State<App>, headers: HeaderMap) -> Response {
if !authorized(&headers, &app) {
return error(
StatusCode::UNAUTHORIZED,
"unauthorized",
"local relay bearer token required",
);
}
let _ = app.shutdown.send(true);
StatusCode::NO_CONTENT.into_response()
}
async fn supervise(context: RuntimeContext, channels: Channels, statuses: Statuses) {
let mut jobs: BTreeMap<String, (String, AbortOnDrop)> = BTreeMap::new();
loop {
if let Ok(config) = context.store.load() {
let active: BTreeMap<_, _> = config
.profiles
.iter()
.filter(|(_, p)| p.enabled)
.map(|(key, p)| {
(
key.clone(),
format!(
"{}:{}:{}:{:?}",
p.base_url, p.tokens.actor.id, p.webhook_url, p.testing_environment_id
),
)
})
.collect();
let remove: Vec<_> = jobs
.iter()
.filter(|(key, (fingerprint, handle))| {
handle.0.is_finished() || active.get(*key) != Some(fingerprint)
})
.map(|(key, _)| key.clone())
.collect();
for key in remove {
if let Some((_, job)) = jobs.remove(&key) {
job.0.abort()
}
channels.lock().await.remove(&key);
statuses.lock().await.remove(&key);
}
for (key, fingerprint) in active {
if let std::collections::btree_map::Entry::Vacant(entry) = jobs.entry(key.clone()) {
let task = AbortOnDrop(tokio::spawn(connection_loop(
context.clone(),
key.clone(),
channels.clone(),
statuses.clone(),
)));
entry.insert((fingerprint, task));
}
}
}
tokio::time::sleep(Duration::from_secs(2)).await;
}
}
async fn connection_loop(
context: RuntimeContext,
key: String,
channels: Channels,
statuses: Statuses,
) {
let mut delay = 1;
loop {
statuses
.lock()
.await
.insert(key.clone(), json!({"state":"connecting"}));
match connected(&context, &key, &channels, &statuses).await {
Ok(()) => delay = 1,
Err(error) => {
let detail = connection_error(&error);
let requires_login = detail["stage"].as_str() == Some("load_profile")
&& matches!(detail["status"].as_u64(), Some(401 | 403));
let mut status = json!({
"state": if requires_login { "authentication_required" } else { "reconnecting" },
"retry_in_seconds": delay,
"last_error": detail,
});
if requires_login {
status["recovery"] = json!(
"Log in again with a fresh IAM short-lived token for this profile; queued work is retained."
);
}
statuses.lock().await.insert(key.clone(), status);
}
}
channels.lock().await.remove(&key);
tokio::time::sleep(Duration::from_secs(delay)).await;
delay = (delay * 2).min(30);
}
}
async fn connected(
context: &RuntimeContext,
key: &str,
channels: &Channels,
statuses: &Statuses,
) -> Result<()> {
let (config, profile) = context
.store
.fresh_profile(key)
.await
.context("load_profile")?;
let client = store::client(&config, &profile).context("configure_client")?;
let known_generation = context.queue.generation(key)?;
let mut stream_session = queue::stream_session(key, known_generation);
let socket = client
.connect_with_generation(
std::slice::from_ref(&profile.tokens.actor.id),
&profile.device_id,
known_generation,
)
.await;
if matches!(
&socket,
Err(crate::Error::WebSocket(tokio_tungstenite::tungstenite::Error::Http(response)))
if response.status() == StatusCode::UNAUTHORIZED
) {
context
.store
.update(|config| {
if let Some(current) = config.profiles.get_mut(key)
&& current.tokens.access_token == profile.tokens.access_token
{
current.expires_at = 0;
}
Ok(())
})
.context("load_profile")?;
}
let mut socket = socket.context("open_socket")?;
let (tx, mut rx) = mpsc::channel::<ClientFrame>(64);
let refresh_in = profile.expires_at.saturating_sub(store::now() + 45).max(1);
let expiry = tokio::time::sleep(Duration::from_secs(refresh_in));
tokio::pin!(expiry);
let mut timeout = tokio::time::interval(Duration::from_secs(30));
let mut last_received = tokio::time::Instant::now();
loop {
tokio::select! {
_=&mut expiry=>{socket.close(None).await?;return Ok(())},
_=timeout.tick()=>{if last_received.elapsed()>Duration::from_secs(120){bail!("server heartbeat expired")}},
command=rx.recv()=>{if let Some(command)=command{socket.send(Message::Text(serde_json::to_string(&command)?.into())).await?;}},
incoming=socket.next()=>{
match incoming.context("WebSocket ended")?? {
Message::Text(text)=>{
last_received=tokio::time::Instant::now();
let frame:ServerFrame=serde_json::from_str(&text).context("decode_frame")?;
match &frame {
ServerFrame::Ping{ping_id}=>socket.send(Message::Text(serde_json::to_string(&ClientFrame::Pong{ping_id:ping_id.clone()})?.into())).await?,
ServerFrame::Ready{protocol_version,actors,testing_generation,..}=>{
if *protocol_version!=2{bail!("unsupported protocol version")}
context.queue.adopt_generation(key,*testing_generation)?;
stream_session=queue::stream_session(key,*testing_generation);
channels.lock().await.insert(key.to_owned(), tx.clone());
for actor in actors {
let after=context.queue.cursor(&stream_session,actor)?;
socket.send(Message::Text(serde_json::to_string(&ClientFrame::Resume{actor_id:actor.clone(),after_sequence:after})?.into())).await?;
if after>0{socket.send(Message::Text(serde_json::to_string(&ClientFrame::Ack{actor_id:actor.clone(),through_sequence:after})?.into())).await?;}
}
statuses.lock().await.insert(key.to_owned(),json!({"state":"connected","actor_id":profile.tokens.actor.id,"testing_environment_id":profile.testing_environment_id}));
},
ServerFrame::Error{recoverable:false,..}=>bail!("server closed authorization"),
_=>{
if let Some((_,actor,sequence))=frame.delivery_position(){
let through=context.queue.receive(&stream_session,&frame,&text).context("persist_delivery")?;
if through>0{socket.send(Message::Text(serde_json::to_string(&ClientFrame::Ack{actor_id:actor.into(),through_sequence:through})?.into())).await?;}
if through<sequence{socket.send(Message::Text(serde_json::to_string(&ClientFrame::Resume{actor_id:actor.into(),after_sequence:through})?.into())).await?;}
}
}
}
},
Message::Ping(data)=>socket.send(Message::Pong(data)).await?,
Message::Close(_)=>return Ok(()),
_=>{}
}
}
}
}
}
fn connection_error(error: &anyhow::Error) -> Value {
let stage = error.to_string();
let stage = match stage.as_str() {
"load_profile" | "configure_client" | "open_socket" | "decode_frame"
| "persist_delivery" => stage.as_str(),
_ => "realtime_connection",
};
if let Some(crate::Error::Api { status, code, .. }) = error.downcast_ref::<crate::Error>() {
return json!({"stage":stage,"code":code,"status":status});
}
if let Some(crate::Error::WebSocket(tokio_tungstenite::tungstenite::Error::Http(response))) =
error.downcast_ref::<crate::Error>()
{
return json!({"stage":stage,"code":"websocket_handshake_rejected","status":response.status().as_u16()});
}
if let Some(error) = error.downcast_ref::<serde_json::Error>() {
return json!({"stage":stage,"code":"invalid_json_shape","line":error.line(),"column":error.column()});
}
json!({"stage":stage,"code":"connection_failed"})
}
fn client_error(error: &crate::Error) -> Value {
match error {
crate::Error::Api {
status,
code,
message,
body,
request_id,
retry_after,
} => {
json!({"status":status,"code":code,"message":message,"body":body,"request_id":request_id,"retry_after":retry_after})
}
_ => {
json!({"code":"transport_error","message":"DM request did not complete; retry preserves its idempotency key"})
}
}
}
async fn outbox(context: RuntimeContext, channels: Channels) {
let mut workers = JoinSet::new();
let mut running = HashMap::new();
let mut polling = tokio::time::interval(Duration::from_millis(300));
loop {
tokio::select! {
finished = workers.join_next_with_id(), if !workers.is_empty() => {
match finished {
Some(Ok((id, ()))) => { running.remove(&id); }
Some(Err(error)) => { running.remove(&error.id()); }
None => {}
}
}
_ = polling.tick() => {
if let Ok(candidates) = context.queue.request_candidates() {
for candidate in candidates {
if running.len() >= 16 { break; }
if running.values().any(|active| active == &candidate.session) { continue; }
let Some(permit) = reserve_payload(&context, candidate.payload_bytes) else { continue; };
let Ok(Some(request)) = context.queue.load_request(candidate.request_id) else { continue; };
let context = context.clone();
let channels = channels.clone();
let task = workers.spawn(async move {
let _permit = permit;
execute_request(context, request, channels).await;
});
running.insert(task.id(), candidate.session);
}
}
}
}
}
}
async fn execute_request(context: RuntimeContext, request: RelayRequest, channels: Channels) {
let session = store::session_key(&request.profile, request.testing_environment_id);
if request.testing_environment_id.is_some()
&& request.request.is_mutation()
&& (request.testing_generation.is_none() || !channels.lock().await.contains_key(&session))
{
let _ = context.queue.defer_until_ready(
request.request_id,
&json!({"code":"awaiting_testing_generation","message":"waiting for the selected sandbox's authenticated realtime generation before writing"}),
);
return;
}
let (config, profile) = match context.store.fresh_profile(&session).await {
Ok(p) => p,
Err(_) => {
let _ = context.queue.retry(
request.request_id,
&json!({"code":"authentication_required","message":"log in again to resume this profile's pending requests"}),
);
return;
}
};
if let Operation::SetPresence { activity } = request.request {
let result = if let Some(sender) = channels.lock().await.get(&session).cloned() {
sender
.send(ClientFrame::Presence {
actor_id: profile.tokens.actor.id.clone(),
activity,
})
.await
.is_ok()
} else {
false
};
let _ = if result {
context.queue.finish(
request.request_id,
Some(&json!({"queued_to_socket":true,"transient":true})),
None,
)
} else {
context.queue.finish(
request.request_id,
None,
Some(
&json!({"code":"not_connected","message":"presence requires a connected daemon; check dm daemon status"}),
),
)
};
return;
}
let mut client = match store::client(&config, &profile) {
Ok(c) => c,
Err(_) => return,
};
if let Some(generation) = request.testing_generation {
match client.clone().with_testing_generation(generation) {
Ok(bound) => client = bound,
Err(error) => {
let _ = context
.queue
.finish(request.request_id, None, Some(&client_error(&error)));
return;
}
}
}
match request.request.execute(&client).await {
Ok(value) => {
let _ = context.queue.finish(request.request_id, Some(&value), None);
}
Err(error) => {
if error.unauthorized() {
let _ = context.store.update(|c| {
if let Some(p) = c.profiles.get_mut(&session) {
p.expires_at = 0
}
Ok(())
});
}
let detail = client_error(&error);
let _ = if (request.request.is_mutation() && error.retryable()) || error.unauthorized()
{
context.queue.retry(request.request_id, &detail)
} else {
context
.queue
.finish(request.request_id, None, Some(&detail))
};
}
}
}
async fn webhooks(context: RuntimeContext) {
let http = match reqwest::Client::builder()
.timeout(Duration::from_secs(15))
.redirect(reqwest::redirect::Policy::none())
.build()
{
Ok(c) => c,
Err(_) => return,
};
let mut workers = JoinSet::new();
let mut running = HashMap::new();
let mut polling = tokio::time::interval(Duration::from_millis(300));
loop {
tokio::select! {
finished = workers.join_next_with_id(), if !workers.is_empty() => {
match finished {
Some(Ok((id, ()))) => { running.remove(&id); }
Some(Err(error)) => { running.remove(&error.id()); }
None => {}
}
}
_ = polling.tick() => {
if let Ok(candidates) = context.queue.webhook_candidates() {
for candidate in candidates {
if running.len() >= 16 { break; }
if running.values().any(|active| active == &candidate.session) { continue; }
let Some(permit) = reserve_payload(&context, candidate.payload_bytes) else { continue; };
let Ok(Some(item)) = context.queue.load_webhook(&candidate.session, &candidate.delivery_id) else { continue; };
let context = context.clone();
let http = http.clone();
let task = workers.spawn(async move {
let _permit = permit;
deliver_webhook(context, item, http).await;
});
running.insert(task.id(), candidate.session);
}
}
}
}
}
}
#[derive(Serialize)]
struct CallbackPayload<'a> {
delivery_id: &'a str,
actor_id: &'a str,
profile: &'a str,
testing_environment_id: Option<Uuid>,
event: &'a Value,
}
async fn deliver_webhook(context: RuntimeContext, item: queue::WebhookWork, http: reqwest::Client) {
let config = match context.store.load() {
Ok(c) => c,
Err(_) => return,
};
let profile_key = item.session.split('#').next().unwrap_or(&item.session);
let Some(profile) = config.profiles.get(profile_key).filter(|p| p.enabled) else {
return;
};
if !matches!(
context
.queue
.webhook_is_pending(&item.session, &item.delivery_id),
Ok(true)
) {
return;
}
let payload = CallbackPayload {
delivery_id: &item.delivery_id,
actor_id: &profile.tokens.actor.id,
profile: &profile.name,
testing_environment_id: profile.testing_environment_id,
event: &item.frame,
};
let success = match http
.post(&profile.webhook_url)
.header("Idempotency-Key", &item.delivery_id)
.json(&payload)
.send()
.await
{
Ok(response) if response.status().is_success() => {
callback_acknowledged(response, &item.delivery_id).await
}
_ => false,
};
let receipt = if success {
delivery_receipt(&item, profile)
} else {
None
};
let _ = context
.queue
.webhook_done(&item.session, &item.delivery_id, success, receipt.as_ref());
}
async fn callback_acknowledged(mut response: reqwest::Response, delivery_id: &str) -> bool {
const MAX_ACK_BYTES: usize = 16 * 1024;
if response
.content_length()
.is_some_and(|size| size > MAX_ACK_BYTES as u64)
{
return false;
}
let mut body = Vec::new();
loop {
match response.chunk().await {
Ok(Some(chunk)) if chunk.len() <= MAX_ACK_BYTES.saturating_sub(body.len()) => {
body.extend_from_slice(&chunk)
}
Ok(Some(_)) | Err(_) => return false,
Ok(None) => break,
}
}
#[derive(Deserialize)]
struct Acknowledgement {
acknowledged: bool,
delivery_id: String,
}
serde_json::from_slice::<Acknowledgement>(&body)
.is_ok_and(|ack| ack.acknowledged && ack.delivery_id == delivery_id)
}
fn delivery_receipt(item: &queue::WebhookWork, profile: &store::Profile) -> Option<RelayRequest> {
if item.frame.get("type")?.as_str()? != "message" {
return None;
}
let message = item.frame.get("message")?;
if message.pointer("/sender/id")?.as_str()? == profile.tokens.actor.id
|| message.get("deleted_at").is_some_and(|v| !v.is_null())
{
return None;
}
let conversation_id = message.get("conversation_id")?.as_str()?.parse().ok()?;
let message_id = message.get("id")?.as_str()?.parse().ok()?;
let digest =
blake3::hash(format!("delivered:{}:{}", item.session, item.delivery_id).as_bytes());
let mut bytes = [0_u8; 16];
bytes.copy_from_slice(&digest.as_bytes()[..16]);
Some(RelayRequest {
request_id: Uuid::from_bytes(bytes),
profile: profile.name.clone(),
testing_environment_id: profile.testing_environment_id,
testing_generation: item.session.split('#').nth(1).and_then(|n| n.parse().ok()),
request: Operation::Receipt {
conversation_id,
message_id,
status: crate::ReceiptStatus::Delivered,
device_id: profile.device_id.clone(),
},
})
}