use super::{Code, Coordinator, Request, Value, empty, wire};
use std::{
sync::{Arc, atomic::Ordering},
thread::JoinHandle,
time::{Duration, Instant},
};
pub(super) struct PendingAuthWork {
connection: String,
request: Request,
pub(super) worker: JoinHandle<Result<Value, Code>>,
}
impl Coordinator {
pub(super) fn start_auth_work(
&mut self,
connection: &str,
request: &Request,
now: Instant,
) -> Option<Result<Value, Code>> {
if self.auth_work.is_some() {
return Some(Err(Code::AuthBusy));
}
let legacy = request.legacy(&request.method);
let barrier = if request.method == "auth.logout" {
match self.execution.prepare_logout(&legacy) {
Ok(barrier) => barrier,
Err(code) => return Some(Err(code)),
}
} else {
if let Err(code) = empty(request) {
return Some(Err(code));
}
None
};
let runtime = Arc::clone(&self.runtime);
let status = request.method == "status";
let worker = match std::thread::Builder::new()
.name("magi-persistent-auth-read-write".into())
.spawn(move || {
if let Some(barrier) = barrier {
while !barrier.load(Ordering::Acquire) {
std::thread::sleep(Duration::from_millis(1));
}
}
if status {
runtime
.current_provider_auth_ready()
.map(Value::Bool)
.map_err(|_| Code::InternalError)
} else {
super::super::auth::ServiceAuthManager::new(runtime).operation(&legacy)
}
}) {
Ok(worker) => worker,
Err(_) => return Some(Err(Code::InternalError)),
};
self.auth_work = Some(PendingAuthWork {
connection: connection.to_owned(),
request: request.clone(),
worker,
});
if let Some(operation) = &request.operation_id {
self.operations
.update(operation, "accepted", Value::Null, Value::Null, now);
}
None
}
pub(super) fn finish_auth_work(&mut self, now: Instant) {
if !self
.auth_work
.as_ref()
.is_some_and(|work| work.worker.is_finished())
{
return;
}
let work = self.auth_work.take().expect("finished auth work");
let result = work.worker.join().unwrap_or(Err(Code::InternalError)).map(|payload| {
if work.request.method == "status" {
serde_json::json!({"service":"ready","provider_auth_ready":payload,"instance_id":self.instance,
"workspace_id":self.workspace,"state_root_id":self.state_root,"connected_clients":self.connections.len(),
"admitted_operations":self.operations.admitted()})
} else {
payload
}
});
self.settle_immediate(&work.request, &result, now);
self.queue(
&work.connection,
wire::response(&self.instance, &work.connection, &work.request, result),
false,
);
}
}