haqor_sync_server/
lib.rs

1use std::fs;
2use std::io::{BufRead, BufReader, Read, Write};
3use std::net::{SocketAddr, TcpListener, TcpStream};
4use std::path::Path;
5use std::time::{SystemTime, UNIX_EPOCH};
6
7use anyhow::{Context, Result, bail};
8
9const MAX_SNAPSHOT_BYTES: usize = 64 * 1024 * 1024;
10
11/// Run the deliberately small HTTP service used by Haqor apps on a trusted
12/// LAN. TLS termination, if wanted, belongs in the caller's reverse proxy;
13/// the built-in service always requires the supplied bearer token.
14pub fn serve_progress(bind: SocketAddr, progress: &Path, token: &str) -> Result<()> {
15    if token.len() < 16 {
16        bail!("--token must be at least 16 characters long");
17    }
18    if let Some(parent) = progress.parent() {
19        fs::create_dir_all(parent)
20            .with_context(|| format!("creating sync directory {}", parent.display()))?;
21    }
22    let listener =
23        TcpListener::bind(bind).with_context(|| format!("binding sync server at {bind}"))?;
24    println!("Haqor progress sync listening on http://{bind}/v1/progress");
25    println!("Keep the token private: anyone with it can read and change your learning progress.");
26    for stream in listener.incoming() {
27        match stream {
28            Ok(stream) => {
29                if let Err(error) = handle(stream, progress, token) {
30                    eprintln!("sync request failed: {error:#}");
31                }
32            }
33            Err(error) => eprintln!("sync connection failed: {error}"),
34        }
35    }
36    Ok(())
37}
38
39fn handle(stream: TcpStream, progress: &Path, token: &str) -> Result<()> {
40    let mut reader = BufReader::new(stream.try_clone()?);
41    let mut request_line = String::new();
42    reader.read_line(&mut request_line)?;
43    let mut authorization = None;
44    let mut content_length = None;
45    loop {
46        let mut line = String::new();
47        if reader.read_line(&mut line)? == 0 {
48            bail!("connection closed before request headers");
49        }
50        if line == "\r\n" {
51            break;
52        }
53        let Some((name, value)) = line.split_once(':') else {
54            respond(stream, "400 Bad Request", b"malformed header")?;
55            return Ok(());
56        };
57        match name.to_ascii_lowercase().as_str() {
58            "authorization" => authorization = Some(value.trim().to_owned()),
59            "content-length" => content_length = value.trim().parse::<usize>().ok(),
60            _ => {}
61        }
62    }
63    if authorization.as_deref() != Some(&format!("Bearer {token}")) {
64        respond(stream, "401 Unauthorized", b"unauthorized")?;
65        return Ok(());
66    }
67    let request = request_line.trim_end();
68    if request == "GET /health HTTP/1.1" || request == "GET /health HTTP/1.0" {
69        respond(stream, "200 OK", b"ok")?;
70        return Ok(());
71    }
72    if request == "GET /v1/progress HTTP/1.1" || request == "GET /v1/progress HTTP/1.0" {
73        let snapshot =
74            fs::read(progress).with_context(|| format!("reading {}", progress.display()))?;
75        respond_with_snapshot(stream, &snapshot)?;
76        return Ok(());
77    }
78    if request != "POST /v1/progress HTTP/1.1" && request != "POST /v1/progress HTTP/1.0" {
79        respond(stream, "404 Not Found", b"not found")?;
80        return Ok(());
81    }
82    let Some(length) = content_length else {
83        respond(stream, "411 Length Required", b"content-length required")?;
84        return Ok(());
85    };
86    if length > MAX_SNAPSHOT_BYTES {
87        respond(stream, "413 Payload Too Large", b"snapshot too large")?;
88        return Ok(());
89    }
90    let mut body = vec![0; length];
91    reader.read_exact(&mut body)?;
92    if !haqor_core::progress_sync::is_sqlite_snapshot(&body) {
93        respond(stream, "400 Bad Request", b"expected a SQLite snapshot")?;
94        return Ok(());
95    }
96    let nonce = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
97    let incoming =
98        std::env::temp_dir().join(format!("haqor-sync-{}-{nonce}.db", std::process::id()));
99    fs::write(&incoming, body)?;
100    let result = haqor_core::progress_sync::merge_progress_files(progress, &incoming)
101        .context("merging received progress snapshot");
102    let _ = fs::remove_file(&incoming);
103    result?;
104    let snapshot = fs::read(progress).with_context(|| format!("reading {}", progress.display()))?;
105    respond_with_snapshot(stream, &snapshot)
106}
107
108fn respond(mut stream: TcpStream, status: &str, body: &[u8]) -> Result<()> {
109    write!(
110        stream,
111        "HTTP/1.1 {status}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
112        body.len()
113    )?;
114    stream.write_all(body)?;
115    Ok(())
116}
117
118fn respond_with_snapshot(mut stream: TcpStream, snapshot: &[u8]) -> Result<()> {
119    write!(
120        stream,
121        "HTTP/1.1 200 OK\r\nContent-Type: application/vnd.sqlite3\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
122        snapshot.len()
123    )?;
124    stream.write_all(snapshot)?;
125    Ok(())
126}
haqor_sync_server/lib.rs

haqor_sync_server/
lib.rs
