use crate::project::{self, ProjectBuildOptions};
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::{SystemTime, UNIX_EPOCH};
pub fn run(mut arguments: impl Iterator<Item = String>) -> Result<(), String> {
let operation = arguments
.next()
.ok_or("noxid queue requires `work [--queue name]` or `status [--queue name]`")?;
let mut queue = None;
while let Some(argument) = arguments.next() {
if argument != "--queue" || queue.is_some() {
return Err(format!("unknown noxid queue argument `{argument}`"));
}
queue = Some(arguments.next().ok_or("--queue requires a queue name")?);
}
if !matches!(operation.as_str(), "work" | "status") {
return Err(format!(
"unknown noxid queue operation `{operation}`; expected `work` or `status`"
));
}
let project = std::env::current_dir()
.map_err(|error| format!("cannot resolve the current project directory: {error}"))?;
run_in(&project, &operation, queue.as_deref())
}
pub(crate) fn run_in(
project_root: &Path,
operation: &str,
queue: Option<&str>,
) -> Result<(), String> {
let names = project::queue_names(project_root)?;
if let Some(name) = queue
&& !names.iter().any(|candidate| candidate == name)
{
return Err(format!(
"error[QUEUE_NOT_FOUND]: no durable queue `{name}` is declared under server/queues"
));
}
if names.is_empty() {
return Err(
"error[QUEUE_NOT_FOUND]: no durable queues are declared under server/queues".into(),
);
}
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_err(|error| error.to_string())?
.as_nanos();
let output = std::env::temp_dir().join(format!("noxid-queue-{}-{nonce}", std::process::id()));
let _cleanup = TemporaryBuild(output.clone());
project::build_project(
project_root,
&ProjectBuildOptions {
out_dir: output.clone(),
title: None,
development: true,
strict_npm: false,
},
)?;
let handler = output.join("server/handler.js");
if !handler.is_file() {
return Err(
"error[QUEUE_HANDLER_MISSING]: queue build did not emit server/handler.js".into(),
);
}
let script = if operation == "status" {
r#"const [handlerPath, queue] = process.argv.slice(1);
const { pathToFileURL } = await import("node:url");
const runtime = await import(pathToFileURL(handlerPath).href);
try { process.stdout.write(JSON.stringify(await runtime.queueStatus(queue || null)) + "\n"); }
finally { await runtime.closeQueueDatabase(); }
"#
} else {
r#"const [handlerPath, queue] = process.argv.slice(1);
const { pathToFileURL } = await import("node:url");
const runtime = await import(pathToFileURL(handlerPath).href);
if (process.env.NOXID_QUEUE_ONCE === "1") {
try { process.stdout.write(JSON.stringify(await runtime.workQueueOnce({ queue: queue || null })) + "\n"); }
finally { await runtime.closeQueueDatabase(); }
} else {
const worker = runtime.startQueueWorker({ queue: queue || null });
const stop = async () => { await worker.stop(); await runtime.closeQueueDatabase(); process.exit(0); };
process.once("SIGINT", stop); process.once("SIGTERM", stop);
}
"#
};
let result = Command::new("node")
.args(["--input-type=module", "-e", script])
.arg(&handler)
.arg(queue.unwrap_or(""))
.current_dir(project_root)
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit())
.status()
.map_err(|error| format!("cannot start Node.js queue runner: {error}"))?;
if !result.success() {
return Err(format!("queue {operation} exited with status {result}"));
}
Ok(())
}
struct TemporaryBuild(PathBuf);
impl Drop for TemporaryBuild {
fn drop(&mut self) {
if self.0.starts_with(std::env::temp_dir()) {
let _ = fs::remove_dir_all(&self.0);
}
}
}