use serde_json::{Value, json};
use std::collections::HashMap;
use std::fs::{File, OpenOptions};
use std::io::{self, BufRead, BufReader, Write};
use std::net::{Ipv4Addr, SocketAddr, TcpStream};
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::{Mutex, MutexGuard, OnceLock, PoisonError};
use std::time::Duration;
pub const SCHEMA: &str = include_str!("../fixtures/github-schema.docs.graphql");
const NESTING_LIMIT: usize = 200;
const PROTOCOL: u64 = 2;
const ATTEMPTS: usize = 4;
const ANSWER_WITHIN: Duration = Duration::from_secs(120);
fn workspace_root() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")
}
fn here() -> PathBuf {
let file = workspace_root().join(file!());
file.parent()
.expect("a source file has a parent directory")
.to_path_buf()
}
fn state() -> PathBuf {
workspace_root().join("target").join("github-schema")
}
fn modules() -> PathBuf {
state().join("node_modules")
}
fn shared_rendezvous() -> PathBuf {
state().join("server")
}
struct Connection {
reader: BufReader<TcpStream>,
writer: TcpStream,
greeted: bool,
}
impl Drop for Connection {
fn drop(&mut self) {
if self.greeted {
leave(&self.writer);
}
}
}
pub fn leave(stream: &TcpStream) {
let _ = (&*stream).write_all(b"{\"bye\":true}\n");
let _ = stream.shutdown(std::net::Shutdown::Both);
}
impl Connection {
fn open(port: u16) -> io::Result<(Self, String)> {
let address = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
let writer = TcpStream::connect_timeout(&address, Duration::from_secs(5))?;
writer.set_nodelay(true)?;
writer.set_read_timeout(Some(ANSWER_WITHIN))?;
let mut connection = Self {
reader: BufReader::new(writer.try_clone()?),
writer,
greeted: false,
};
let ready = connection.exchange(&json!({ "hello": PROTOCOL }).to_string())?;
match ready["schema_sha256"].as_str() {
Some(digest) if ready["ready"] == json!(true) => {
connection.greeted = true;
Ok((connection, digest.to_owned()))
}
_ => Err(io::Error::other(format!("the server answered {ready}"))),
}
}
fn exchange(&mut self, line: &str) -> io::Result<Value> {
self.writer.write_all(format!("{line}\n").as_bytes())?;
let mut answer = String::new();
if self.reader.read_line(&mut answer)? == 0 {
return Err(io::ErrorKind::UnexpectedEof.into());
}
serde_json::from_str(&answer).map_err(io::Error::other)
}
}
struct Validators {
rendezvous: PathBuf,
schema: PathBuf,
idle_for: Option<Duration>,
idle: Mutex<Vec<Connection>>,
schema_sha256: OnceLock<String>,
answers: Mutex<HashMap<String, Vec<Value>>>,
}
impl Validators {
fn new(rendezvous: PathBuf) -> Self {
Self {
rendezvous,
schema: here().join("../fixtures/github-schema.docs.graphql"),
idle_for: None,
idle: Mutex::new(Vec::new()),
schema_sha256: OnceLock::new(),
answers: Mutex::new(HashMap::new()),
}
}
fn rendezvous(&self) -> PathBuf {
std::fs::create_dir_all(&self.rendezvous).unwrap_or_else(|error| {
panic!(
"the GitHub schema check could not create {} ({error})",
self.rendezvous.display()
)
});
self.rendezvous.clone()
}
fn endpoint(&self) -> (u16, u64) {
let connection = self.connection();
self.idle().push(connection);
let endpoint: Value = serde_json::from_slice(
&std::fs::read(self.rendezvous().join("endpoint.json"))
.expect("a running server's endpoint"),
)
.expect("the endpoint is JSON");
(
endpoint["port"]
.as_u64()
.and_then(|port| u16::try_from(port).ok())
.expect("the endpoint names a port"),
endpoint["pid"].as_u64().expect("the endpoint names a pid"),
)
}
fn stop(&self) {
self.idle().clear();
let Ok(endpoint) = std::fs::read(self.rendezvous.join("endpoint.json")) else {
return;
};
let Some(port) = serde_json::from_slice::<Value>(&endpoint)
.ok()
.and_then(|endpoint| endpoint["port"].as_u64())
.and_then(|port| u16::try_from(port).ok())
else {
return;
};
let address = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
if let Ok(mut stream) = TcpStream::connect_timeout(&address, Duration::from_secs(5)) {
let _ = stream.write_all(b"{\"hello\":0}\n");
let _ = stream.set_read_timeout(Some(ANSWER_WITHIN));
let _ = BufReader::new(stream).read_line(&mut String::new());
}
let asked = std::time::Instant::now();
while TcpStream::connect_timeout(&address, Duration::from_secs(1)).is_ok()
&& asked.elapsed() < ANSWER_WITHIN
{
std::thread::sleep(Duration::from_millis(20));
}
}
fn kept(&self) -> MutexGuard<'_, HashMap<String, Vec<Value>>> {
self.answers.lock().unwrap_or_else(PoisonError::into_inner)
}
fn idle(&self) -> MutexGuard<'_, Vec<Connection>> {
self.idle.lock().unwrap_or_else(PoisonError::into_inner)
}
fn connection(&self) -> Connection {
if let Some(connection) = self.idle().pop() {
return connection;
}
let directory = self.rendezvous();
let mut tried = Vec::new();
for _ in 0..ATTEMPTS {
match self.running(&directory) {
Ok(connection) => return connection,
Err(error) => tried.push(error.to_string()),
}
let lock = File::create(directory.join("start.lock"))
.and_then(|lock| lock.lock().map(|()| lock))
.expect("the GitHub schema check's start lock can be taken");
let started = self.running(&directory).or_else(|_| self.start(&directory));
drop(lock);
match started {
Ok(connection) => return connection,
Err(error) => tried.push(error.to_string()),
}
}
panic!(
"the GitHub schema check could not be reached ({}); its server's standard error is \
in {}",
tried.join("; "),
directory.join("server.log").display()
)
}
fn running(&self, directory: &std::path::Path) -> io::Result<Connection> {
let endpoint: Value =
serde_json::from_slice(&std::fs::read(directory.join("endpoint.json"))?)
.map_err(io::Error::other)?;
let port = endpoint["port"]
.as_u64()
.and_then(|port| u16::try_from(port).ok())
.ok_or_else(|| io::Error::other(format!("endpoint.json holds {endpoint}")))?;
let (connection, digest) = Connection::open(port)?;
self.schema_sha256.get_or_init(|| digest);
Ok(connection)
}
fn start(&self, directory: &std::path::Path) -> io::Result<Connection> {
let here = here();
let log = OpenOptions::new()
.create(true)
.append(true)
.open(directory.join("server.log"))?;
let mut command = Command::new("bun");
command
.arg(here.join("validate.mjs"))
.arg(&self.schema)
.arg("--serve")
.arg(directory)
.current_dir(std::env::temp_dir())
.env("BUN_RUNTIME_TRANSPILER_CACHE_PATH", "0")
.env("GITHUB_SCHEMA_MODULES", modules())
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(log);
if let Some(idle_for) = self.idle_for {
command.env("GITHUB_SCHEMA_IDLE_MS", idle_for.as_millis().to_string());
}
detach(&mut command);
let mut child = spawn(&mut command).unwrap_or_else(|error| {
panic!(
"the GitHub schema check could not start `bun` ({error}); install the \
workspace's toolchain with `just bootstrap`"
)
});
let output = child.stdout.take().expect("a piped standard output");
std::thread::spawn(move || child.wait());
let mut ready = String::new();
BufReader::new(output).read_line(&mut ready)?;
let ready: Value = serde_json::from_str(&ready)
.map_err(|_| io::Error::other(format!("the server started by answering {ready:?}")))?;
if let Some(internal) = ready.get("internal") {
panic!("the GitHub schema check did not start: {internal}");
}
let port = ready["port"]
.as_u64()
.and_then(|port| u16::try_from(port).ok())
.ok_or_else(|| io::Error::other(format!("the server started with {ready}")))?;
let (connection, digest) = Connection::open(port)?;
self.schema_sha256.get_or_init(|| digest);
Ok(connection)
}
fn ask(&self, request: &str) -> Vec<Value> {
let mut failures = Vec::new();
for _ in 0..ATTEMPTS {
let mut connection = self.connection();
match connection.exchange(request) {
Ok(answer) => {
self.idle().push(connection);
if let Some(internal) = answer.get("internal") {
panic!("the GitHub schema check failed on {request}: {internal}");
}
return answer["errors"]
.as_array()
.cloned()
.expect("the GitHub schema check answers `errors`");
}
Err(error) => {
failures.push(error.to_string());
self.idle().clear();
}
}
}
panic!(
"the GitHub schema check did not answer {request}: {}",
failures.join("; ")
)
}
fn answer(&self, request: String) -> Vec<Value> {
if let Some(answer) = self.kept().get(&request) {
return answer.clone();
}
let answer = self.ask(&request);
let mut answers = self.kept();
if answers.len() >= ANSWERS_KEPT {
answers.clear();
}
answers.insert(request, answer.clone());
answer
}
}
#[cfg(unix)]
fn detach(command: &mut Command) {
use std::os::unix::process::CommandExt;
command.process_group(0);
}
#[cfg(windows)]
const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
#[cfg(windows)]
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
#[cfg(windows)]
const CREATE_BREAKAWAY_FROM_JOB: u32 = 0x0100_0000;
#[cfg(windows)]
fn detach(command: &mut Command) {
use std::os::windows::process::CommandExt;
command.creation_flags(CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW | CREATE_BREAKAWAY_FROM_JOB);
}
#[cfg(not(any(unix, windows)))]
fn detach(_command: &mut Command) {}
fn spawn(command: &mut Command) -> io::Result<std::process::Child> {
let spawned = command.spawn();
#[cfg(windows)]
if spawned.is_err() {
use std::os::windows::process::CommandExt;
return command
.creation_flags(CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW)
.spawn();
}
spawned
}
fn errors(
validators: &Validators,
document: &str,
operation_name: Option<&str>,
variables: Option<&Value>,
) -> Vec<Value> {
if variables.is_some_and(|variables| nesting(variables) > NESTING_LIMIT) {
return vec![
json!({"message": format!("Variables nest deeper than {NESTING_LIMIT} levels")}),
];
}
let request = json!({
"query": document,
"operationName": operation_name,
"variables": variables,
"coerce": variables.is_some(),
})
.to_string();
validators.answer(request)
}
const ANSWERS_KEPT: usize = 16_384;
#[must_use]
pub struct PrivateServer(Validators);
pub fn serve_privately() -> PrivateServer {
static TAKEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |since| since.as_nanos());
PrivateServer(Validators::new(std::env::temp_dir().join(format!(
"onetaskgraph-github-schema-{}-{nanos}-{}",
std::process::id(),
TAKEN.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
))))
}
pub fn serve_privately_from_a_copy(idle_for: Duration) -> PrivateServer {
let mut server = serve_privately();
let copy = server.0.rendezvous().join("github-schema.docs.graphql");
std::fs::copy(&server.0.schema, ©).expect("the schema is copied for a private server");
server.0.schema = copy;
server.0.idle_for = Some(idle_for);
server
}
impl PrivateServer {
#[must_use]
pub fn refuse(&self, body: &Value) -> Option<Value> {
refuse_by(&self.0, body)
}
pub fn server(&self) -> (u16, u64) {
self.0.endpoint()
}
pub fn schema(&self) -> &std::path::Path {
&self.0.schema
}
pub fn disconnect(&self) {
self.0.idle().clear();
}
}
impl Drop for PrivateServer {
fn drop(&mut self) {
self.0.stop();
let directory = &self.0.rendezvous;
let stopped = std::time::Instant::now();
while std::fs::remove_dir_all(directory).is_err() && directory.exists() {
if stopped.elapsed() > Duration::from_secs(10) {
eprintln!(
"the private GitHub schema check's directory {} could not be removed",
directory.display()
);
return;
}
std::thread::sleep(Duration::from_millis(50));
}
}
}
fn validators() -> &'static Validators {
static VALIDATORS: OnceLock<Validators> = OnceLock::new();
VALIDATORS.get_or_init(|| {
let validators = Validators::new(shared_rendezvous());
let connection = validators.connection();
validators.idle().push(connection);
validators
})
}
pub fn warm() {
validators();
}
pub fn schema_sha256() -> String {
validators()
.schema_sha256
.get()
.expect("a connected server reported the schema's digest")
.clone()
}
pub fn locations() -> (PathBuf, PathBuf, PathBuf) {
(here(), modules(), shared_rendezvous())
}
pub fn server() -> (u16, u64) {
validators().endpoint()
}
fn nesting(value: &Value) -> usize {
let mut deepest = 0;
let mut pending = vec![(value, 1)];
while let Some((value, depth)) = pending.pop() {
deepest = deepest.max(depth);
match value {
Value::Array(items) => pending.extend(items.iter().map(|item| (item, depth + 1))),
Value::Object(fields) => pending.extend(fields.values().map(|item| (item, depth + 1))),
_ => {}
}
}
deepest
}
fn messages(errors: Vec<Value>) -> Result<(), Vec<String>> {
if errors.is_empty() {
Ok(())
} else {
Err(errors
.iter()
.map(|error| error["message"].as_str().unwrap_or_default().to_owned())
.collect())
}
}
pub fn validate(document: &str, variables: &Value) -> Result<(), Vec<String>> {
messages(errors(validators(), document, None, Some(variables)))
}
pub fn validate_document(document: &str) -> Result<(), Vec<String>> {
messages(errors(validators(), document, None, None))
}
#[must_use]
pub fn refuse(body: &Value) -> Option<Value> {
refuse_by(validators(), body)
}
fn refuse_by(validators: &Validators, body: &Value) -> Option<Value> {
let document = body.get("query")?.as_str()?;
let operation_name = match body.get("operationName") {
None | Some(Value::Null) => None,
Some(Value::String(name)) => Some(name.as_str()),
Some(other) => {
return Some(json!({"errors": [{
"message": format!("operationName must be a string, not {other}")
}]}));
}
};
let empty = Value::Object(serde_json::Map::new());
let variables = body
.get("variables")
.filter(|value| !value.is_null())
.unwrap_or(&empty);
let errors = errors(validators, document, operation_name, Some(variables));
(!errors.is_empty()).then(|| json!({ "errors": errors }))
}