codex_sync/sync/
http.rs

1//! HTTP client adapter over the transport-independent snapshot core.
2
3use super::core::{self, FileInfo, Manifest, PullReport, SnapshotSource};
4use crate::config::Config;
5use anyhow::{Result, bail};
6use reqwest::header::AUTHORIZATION;
7
8struct HttpSource {
9    client: reqwest::Client,
10    base: String,
11    authorization: String,
12}
13
14impl HttpSource {
15    fn new(peer: &str, shared_key: &str) -> Self {
16        let base = if peer.starts_with("http") {
17            peer.trim_end_matches('/').to_owned()
18        } else {
19            format!("http://{}", peer.trim_end_matches('/'))
20        };
21        Self {
22            client: reqwest::Client::new(),
23            base,
24            authorization: format!("Bearer {shared_key}"),
25        }
26    }
27}
28
29impl SnapshotSource for HttpSource {
30    async fn manifest(&self) -> Result<Manifest> {
31        Ok(self
32            .client
33            .get(format!("{}/api/manifest", self.base))
34            .header(AUTHORIZATION, &self.authorization)
35            .send()
36            .await?
37            .error_for_status()?
38            .json()
39            .await?)
40    }
41
42    async fn read(&self, file: &FileInfo) -> Result<Vec<u8>> {
43        let url = format!(
44            "{}/api/file/{}",
45            self.base,
46            file.path
47                .split('/')
48                .map(url_encode)
49                .collect::<Vec<_>>()
50                .join("/")
51        );
52        let mut response = self
53            .client
54            .get(url)
55            .header(AUTHORIZATION, &self.authorization)
56            .send()
57            .await?
58            .error_for_status()?;
59        if response
60            .content_length()
61            .is_some_and(|size| size != file.size)
62        {
63            bail!("远端文件大小不符合清单：{}", file.path);
64        }
65        let mut bytes = Vec::with_capacity(file.size as usize);
66        while let Some(chunk) = response.chunk().await? {
67            if bytes.len().saturating_add(chunk.len()) > file.size as usize {
68                bail!("远端文件超过清单大小：{}", file.path);
69            }
70            bytes.extend_from_slice(&chunk);
71        }
72        Ok(bytes)
73    }
74}
75
76pub async fn pull(cfg: &Config, peer: &str, overwrite: bool) -> Result<PullReport> {
77    core::import_from(cfg, &HttpSource::new(peer, &cfg.shared_key), overwrite).await
78}
79
80fn url_encode(segment: &str) -> String {
81    segment
82        .bytes()
83        .map(|byte| {
84            if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
85                (byte as char).to_string()
86            } else {
87                format!("%{byte:02X}")
88            }
89        })
90        .collect()
91}
codex_sync/sync/http.rs

codex_sync/sync/
http.rs
