From 95cc2df3755b10878d28214b558eeb4435c25fd6 Mon Sep 17 00:00:00 2001 From: 4cuck <40539855+4cuck@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:59:32 +0000 Subject: [PATCH] Route GraphQL through session-pinned SOCKS5 and serve repeats from cache. Search and profiles were paying a new tunnel plus extra X hops on every hit; keep-alive, first-page SWR, and the example config document a reverse-proxied instance without committing live exits. --- .gitignore | 9 + README.md | 85 +++++- config/teapot.example.toml | 13 +- proxies.example.txt | 5 + public/css/clean.css | 6 + public/css/general.css | 3 +- public/js/infiniteScroll.js | 66 +++-- src/api/auth.rs | 10 + src/api/budget.rs | 14 +- src/api/client.rs | 67 +++-- src/api/client_endpoints.rs | 117 ++++++-- src/api/filters.rs | 12 +- src/api/http.rs | 520 ++++++++++++++++++++++++++++++------ src/api/mod.rs | 2 + src/api/proxy_pool.rs | 303 +++++++++++++++++++++ src/cache/keys.rs | 18 ++ src/cache/mod.rs | 5 +- src/cache/store.rs | 103 ++++++- src/cache/ttl.rs | 9 + src/config.rs | 6 + src/error.rs | 7 +- src/main.rs | 7 +- src/routes/helpers.rs | 218 +++++++++++++-- src/routes/search.rs | 200 ++++++++++++-- src/routes/status.rs | 5 + src/routes/timeline.rs | 91 +++++-- src/views/embed_tests.rs | 2 + src/views/layout.rs | 4 +- 28 files changed, 1652 insertions(+), 255 deletions(-) create mode 100644 proxies.example.txt create mode 100644 src/api/proxy_pool.rs diff --git a/.gitignore b/.gitignore index e8346d3..4f68786 100644 --- a/.gitignore +++ b/.gitignore @@ -28,6 +28,15 @@ # Live instance files stay ignored even if allowlists are widened later # sessions.jsonl, config/teapot.toml, session-limits.json, ecosystem.config.cjs, *.log +# proxies.txt, session-proxies.json +proxies.txt +session-proxies.json +config/teapot.toml +sessions.jsonl +session-limits.json + +# Documented dummy list only — never a real vendor file +!proxies.example.txt # Source !src/ diff --git a/README.md b/README.md index 1485c2b..90b5f35 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,10 @@ Public instance: [nitter.cf](https://nitter.cf) (also [xitter.cf](https://xitter - **Privacy-focused**: No third-party JavaScript, tracking, ads, or remote fonts - **RSS feeds**: Subscribe to any user's tweets +- **SOCKS5 exits for GraphQL**: Pin each cookie session to one proxy port so + its apparent IP stays stable +- **In-process cache**: First-page search, profiles, and stale-while-revalidate + so repeats do not wait on X - **FxEmbed-style Discord embeds**: - Multiple images displayed in carousel on mobile - Videos play directly in the embed @@ -20,6 +24,7 @@ Public instance: [nitter.cf](https://nitter.cf) (also [xitter.cf](https://xitter - Rust 1.96.1+ - Twitter/X session tokens for API access +- Optional: SOCKS5 endpoints if X blocks the server's own IP ## Building @@ -27,6 +32,9 @@ Public instance: [nitter.cf](https://nitter.cf) (also [xitter.cf](https://xitter cargo build --release ``` +The crate and binary stay named `teapot` so existing configs and process +managers keep working. + ## Configuration 1. Copy the example config: @@ -35,10 +43,14 @@ cargo build --release cp config/teapot.example.toml config/teapot.toml ``` -1. Edit `config/teapot.toml` with your settings. Set `title = "teapawt"` (or - whatever you want shown in the navbar). +2. Edit `config/teapot.toml`. Set `title = "teapawt"` (or whatever you want + shown in the navbar). -2. Create a sessions file with your Twitter/X credentials: +3. Behind nginx or another TLS terminator, set `https = true` and + `publicPort = 443` so prefs bookmarks, RSS, and `og:image` links omit the + listen port (`8471` or whatever `port` is). + +4. Create a sessions file with your Twitter/X credentials: ```bash cp sessions.example.jsonl sessions.jsonl @@ -47,7 +59,57 @@ cp sessions.example.jsonl sessions.jsonl Cookie sessions use `auth_token` and `ct0`. OAuth sessions use `oauth_token` and `oauth_secret`. See `sessions.example.jsonl` for the JSONL -shape. Never commit a real session or generated `teapot.toml`. +shape. + +**Never commit** a real `sessions.jsonl`, `config/teapot.toml`, `proxies.txt`, +or `session-proxies.json`. Those paths are gitignored. + +### Proxies + +X GraphQL (search, profiles, tweets) can go out through SOCKS5. Media (`/pic`, +`/video`, `/gif`) stays on the optional HTTP `proxy` and does not use the +SOCKS list. + +```bash +cp proxies.example.txt proxies.txt +# one host:port:user:pass per line +``` + +```toml +# config/teapot.toml +socksProxiesFile = "proxies.txt" +sessionProxiesFile = "session-proxies.json" +``` + +All lines must share one hostname (different ports are the rotating exits). +On startup teapawt DNS-probes that host, keeps the lowest-ping A record, and +pins each session id to one port. Pins are rewritten to `sessionProxiesFile` +as `{ "sessionId": port }` only — credentials stay in `proxies.txt`. + +`apiProxy` is an HTTP/HTTPS proxy for GraphQL when you are not using SOCKS. +It is ignored when `socksProxiesFile` is set. + +Tunnels to `x.com` are HTTP/1.1 over TLS with keep-alive, `TCP_NODELAY`, and +rustls session resumption. Idle tunnels are reused for a short window so a +second search does not pay a new SOCKS + TLS handshake. + +### Rate limits and cache + +- `maxConcurrentReqs` is **per cookie session**, not global. +- `clientBudget` caps how fast one visitor IP can spend uncached X calls. + Cache hits, media, and static files are free. SearchTimeline costs 2 tokens. +- First-page search is cached for 60 seconds, then served stale for 4 more + minutes while a refresh runs. +- Profiles are cached for 5 minutes, then served stale for 20 more minutes. +- Search result pages seed user ids and prefetch a few profiles so a click + often hits cache. + +Empty HTTP 404s from SearchTimeline are treated as a brief X/WAF blip +(`TransientUpstream`) and retried on another session. Dead cookies are 401, +or 403 with code 326 — not those empty 404s. + +Some queries fail with X error 214 (`Query is denylisted in Search Content +Control tool`). That is X's platform denylist, not a teapawt setting. ## Running @@ -59,9 +121,8 @@ cargo run ./target/release/teapot ``` -The crate and binary stay named `teapot` so existing configs and process -managers keep working. The server will start on `http://localhost:8080` by -default. +The server listens on `http://localhost:8080` by default (`address` / `port` +in the config). ## Getting Twitter Sessions @@ -81,14 +142,16 @@ teapawt/ │ ├── config.rs # Configuration │ ├── error.rs # Error types │ ├── types/ # Data structures (User, Tweet, Timeline, etc.) -│ ├── api/ # Twitter API client, OAuth, parsing -│ ├── cache/ # In-process caching +│ ├── api/ # Twitter API client, SOCKS5 pool, OAuth, parsing +│ ├── cache/ # In-process cache (TTL + stale-while-revalidate) │ ├── routes/ # HTTP route handlers │ ├── views/ # Maud HTML templates │ └── utils/ # Utilities (HMAC, formatters) ├── public/ # Static assets (CSS, JS, fonts, logo) -├── config/ # Configuration files -└── Cargo.toml # Dependencies +├── config/ # teapot.example.toml only +├── proxies.example.txt # Dummy SOCKS5 list +├── sessions.example.jsonl +└── Cargo.toml ``` ## Embed Improvements diff --git a/config/teapot.example.toml b/config/teapot.example.toml index bd06aab..05a187c 100644 --- a/config/teapot.example.toml +++ b/config/teapot.example.toml @@ -7,6 +7,10 @@ hostname = "localhost" httpMaxConnections = 100 https = false # Set to true when using HTTPS port = 8080 +# Port used in generated links (prefs bookmark, RSS, og:image, …). +# Leave unset to use `port`. Set to 443 (HTTPS) or 80 (HTTP) when teapot +# sits behind a reverse proxy so URLs omit the listen port. +# publicPort = 443 staticDir = "./public" title = "teapawt" @@ -50,13 +54,20 @@ enableDebug = false # Generate a unique value with at least 32 characters. debugToken = "" -# HTTP/HTTPS proxy (SOCKS proxies not supported) +# HTTP/HTTPS proxy for media fetches. X API calls use socksProxiesFile when set. proxy = "" proxyAuth = "" # HTTP/HTTPS proxy URL used only for X API requests, e.g. http://localhost:7000 +# Ignored when socksProxiesFile is set. apiProxy = "" +# SOCKS5 list (`host:port:user:pass`, one per line). Each cookie session is +# pinned to one port so its exit IP stays stable. Pins are stored in +# sessionProxiesFile. +socksProxiesFile = "" +sessionProxiesFile = "session-proxies.json" + # Enable this if cookie-based auth is failing disableTid = false diff --git a/proxies.example.txt b/proxies.example.txt new file mode 100644 index 0000000..d7f3e2e --- /dev/null +++ b/proxies.example.txt @@ -0,0 +1,5 @@ +# SOCKS5 endpoints for X API calls. One `host:port:user:pass` per line. +# Copy to proxies.txt and point socksProxiesFile at it. Never commit the copy. +# +# socks.example.com:1080:username:password +# socks.example.com:1081:username:password diff --git a/public/css/clean.css b/public/css/clean.css index 722ab83..e127418 100644 --- a/public/css/clean.css +++ b/public/css/clean.css @@ -111,6 +111,12 @@ main .container { display: block; } +/* The nitter search bar sits up into the old banner. Error pages have + no banner, so that pull-up hid the message under the fixed nav. */ +.error-panel { + margin-top: 0; +} + .timeline-container { width: min(100%, var(--feed-width)); max-width: var(--feed-width); diff --git a/public/css/general.css b/public/css/general.css index b3a5837..a7a02dd 100644 --- a/public/css/general.css +++ b/public/css/general.css @@ -22,8 +22,7 @@ background: var(--tp-error-container); color: var(--tp-on-error-container); box-shadow: var(--elevation-1); - margin: auto; - margin-top: -50px; + margin: 0 auto; text-align: center; font-weight: 500; } diff --git a/public/js/infiniteScroll.js b/public/js/infiniteScroll.js index 7bbca05..ac4f5e1 100644 --- a/public/js/infiniteScroll.js +++ b/public/js/infiniteScroll.js @@ -1,9 +1,5 @@ // @license http://www.gnu.org/licenses/agpl-3.0.html AGPL-3.0 // SPDX-License-Identifier: AGPL-3.0-only -function insertBeforeLast(node, elem) { - node.insertBefore(elem, node.childNodes[node.childNodes.length - 2]); -} - function getLoadMore(doc) { return doc.querySelector(".show-more:not(.timeline-item)"); } @@ -15,6 +11,12 @@ function isDuplicate(item, itemClass) { return document.querySelector(itemClass + " .tweet-link[href='" + href + "']") != null; } +function setLoadMoreLabel(loadMore, text) { + if (loadMore && loadMore.children[0]) { + loadMore.children[0].text = text; + } +} + window.onload = function () { const url = window.location.pathname; const isEngagement = /\/status\/\d+\/(retweets|quotes)/.test(url); @@ -30,45 +32,69 @@ window.onload = function () { if (loading) return; if (html.scrollTop + html.clientHeight >= html.scrollHeight - 3000) { - loading = true; var loadMore = getLoadMore(document); if (loadMore == null) return; - loadMore.children[0].text = "Loading..."; + loading = true; + setLoadMoreLabel(loadMore, "Loading..."); var url = new URL(loadMore.children[0].href); url.searchParams.append("scroll", "true"); fetch(url.toString()).then(function (response) { - if (response.status > 299) throw "error"; - + if (response.status === 204) { + loadMore.remove(); + loading = false; + return null; + } + if (!response.ok) throw "error"; return response.text(); }).then(function (html) { + if (html == null) return; + var parser = new DOMParser(); var doc = parser.parseFromString(html, "text/html"); - loadMore.remove(); + if (doc.querySelector(".error-panel")) throw "error"; + var items = []; for (var item of doc.querySelectorAll(itemClass)) { if (item.className == "timeline-item show-more") continue; if (isDuplicate(item, itemClass)) continue; - if (isTweet || isEngagement) container.appendChild(item); - else insertBeforeLast(container, item); + items.push(item); } + var newLoadMore = getLoadMore(doc); + if (items.length === 0 && newLoadMore == null) { + if (doc.querySelector(".timeline-end, .search-empty")) { + loadMore.remove(); + loading = false; + return; + } + throw "empty"; + } + + for (var item of items) { + container.insertBefore(item, loadMore); + } + if (newLoadMore == null) { + loadMore.remove(); + } else { + loadMore.replaceWith(newLoadMore); + } loading = false; - const newLoadMore = getLoadMore(doc); - if (newLoadMore == null) return; - if (isTweet || isEngagement) container.appendChild(newLoadMore); - else insertBeforeLast(container, newLoadMore); }).catch(function (err) { console.warn("Something went wrong.", err); - if (failed > 3) { - loadMore.children[0].text = "Error"; + var next = (failed || 0) + 1; + if (next > 3) { + setLoadMoreLabel(loadMore, "Error"); + loading = false; return; } - - loading = false; - handleScroll((failed || 0) + 1); + setLoadMoreLabel(loadMore, "Load more"); + setTimeout(function () { + loading = false; + handleScroll(next); + }, 1500); }); } } diff --git a/src/api/auth.rs b/src/api/auth.rs index aa2d4dc..d0a4e9b 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -470,6 +470,16 @@ impl SessionPool { self.sessions.len() } + /// Session ids in load order, used to pin SOCKS5 exits. + #[must_use] + pub fn session_ids(&self) -> Vec { + self + .sessions + .iter() + .map(|slot| slot.credentials.id) + .collect() + } + /// Check if pool is empty. pub const fn is_empty(&self) -> bool { self.sessions.is_empty() diff --git a/src/api/budget.rs b/src/api/budget.rs index 22aa341..03e5042 100644 --- a/src/api/budget.rs +++ b/src/api/budget.rs @@ -27,17 +27,17 @@ pub struct Tier { /// Widening horizons. The tightest tier decides. const TIERS: [Tier; 3] = [ + Tier { + refill_per_sec: 5.0, + capacity: 100.0, + }, Tier { refill_per_sec: 1.0, - capacity: 20.0, + capacity: 300.0, }, Tier { - refill_per_sec: 0.2, - capacity: 60.0, - }, - Tier { - refill_per_sec: 0.05, - capacity: 200.0, + refill_per_sec: 0.25, + capacity: 1000.0, }, ]; diff --git a/src/api/client.rs b/src/api/client.rs index f33b927..a576503 100644 --- a/src/api/client.rs +++ b/src/api/client.rs @@ -1,16 +1,14 @@ -use std::time::Duration; +use std::sync::Arc; use axum::http::header; use serde::{ Deserialize, de::DeserializeOwned, }; -use tokio::time::{ - sleep, - timeout, -}; +use tokio::time::timeout; use super::{ + ProxyPool, SessionLease, SessionPool, TidClient, @@ -19,7 +17,10 @@ use super::{ ClientBudget, }, endpoints, - http::HttpClient, + http::{ + HttpClient, + ProxyConfig, + }, parser, }; use crate::{ @@ -69,10 +70,13 @@ use crate::{ /// A search spends a `SearchTimeline` call and is never served from cache. const SEARCH_COST: f64 = 2.0; -/// First try plus this many replacements. Empty GraphQL 404s often hit several -/// sessions in a row before X answers. -const GRAPHQL_REPLACEMENTS: usize = 4; -const TRANSIENT_RETRY_WAIT: Duration = Duration::from_millis(200); +/// First try plus this many replacements for a locked session or empty 404. +/// Each attempt spends SearchTimeline quota (50/15m per account), so this +/// must stay small. +const GRAPHQL_REPLACEMENTS: usize = 2; +/// Per-account 429 (JSON / `x-rate-limit-*`). HTML 429s are the WAF and must +/// not walk the pool — that used to zero every cookie for 15 minutes. +const RATE_LIMIT_REPLACEMENTS: usize = 5; /// Community-cache strings are written by third parties, so cap them before /// they reach a page. @@ -280,13 +284,14 @@ fn twitter_error_code(body: &str) -> Option { pub struct ApiClient { pub(crate) client: HttpClient, pub(crate) sessions: SessionPool, + proxies: Option>, tid: TidClient, budget: ClientBudget, tid_enabled: bool, } impl ApiClient { - pub fn new(config: &Config, sessions: SessionPool) -> Self { + pub fn new(config: &Config, sessions: SessionPool, proxies: Option) -> Self { let mut headers = header::HeaderMap::new(); headers.insert( header::USER_AGENT, @@ -310,20 +315,30 @@ impl ApiClient { } else { &config.config.api_proxy }; - let client = + let mut client = HttpClient::new(api_proxy, &config.config.proxy_auth).with_default_headers(headers); + if api_proxy.is_empty() + && let Some(ref pool) = proxies + { + client = client.with_default_proxy(pool.first()); + } let tid = TidClient::new(client.clone(), sessions.clone()); Self { client, sessions, + proxies: proxies.map(Arc::new), tid, budget: ClientBudget::new(config.config.client_budget), tid_enabled: !config.config.disable_tid, } } + pub(crate) fn proxy_for(&self, session: &SessionLease) -> Option { + self.proxies.as_ref().map(|pool| pool.for_session(session.id)) + } + pub(crate) async fn bearer_and_tid(&self, api_path: &str) -> (&'static str, Option) { if !self.tid_enabled { return (endpoints::BEARER_TOKEN_NO_TID, None); @@ -455,13 +470,13 @@ impl ApiClient { .graphql_request_inner(&session, endpoint, variables, features, field_toggles) .await; - for _ in 0..GRAPHQL_REPLACEMENTS { + for replacement in 0..GRAPHQL_REPLACEMENTS { let session_id = session.id; let retry = match &last { Err(Error::SessionRejected(_)) | Err(Error::NotFound(_)) | Err(Error::TransientUpstream) => true, - Err(Error::RateLimited) => { + Err(Error::RateLimited) if replacement < RATE_LIMIT_REPLACEMENTS => { self .sessions .has_unlimited(endpoint, retry_kind, session_id) @@ -472,12 +487,6 @@ impl ApiClient { if !retry { return last; } - if matches!( - last, - Err(Error::NotFound(_)) | Err(Error::TransientUpstream) - ) { - sleep(TRANSIENT_RETRY_WAIT).await; - } tracing::warn!( session_id, @@ -549,7 +558,8 @@ impl ApiClient { } } - /// A search spends a `SearchTimeline` call and is never served from cache. + /// SearchTimeline is billed higher because first-page cache still misses on + /// a new query and each page spends the tight 50/15m search budget. fn cost_of(endpoint: &str) -> f64 { if endpoint.contains("SearchTimeline") { SEARCH_COST @@ -613,7 +623,10 @@ impl ApiClient { ) .await?; - let response = self.client.get_with_headers(&url, &headers).await?; + let response = self + .client + .get_on(&url, &headers, self.proxy_for(session).as_ref()) + .await?; let (bytes, limit_recorded) = self.account_response(session, endpoint, response).await?; // Check for API errors before full deserialization. @@ -682,6 +695,11 @@ impl ApiClient { session_user = %session.username, "API request failed: {status} - {body}" ); + let body_trim = body.trim_start(); + let ip_throttle = body_trim.is_empty() || body_trim.starts_with('<'); + if ip_throttle { + return Err(Error::IpThrottled); + } if !limit_recorded { self .sessions @@ -900,7 +918,10 @@ impl ApiClient { headers.insert("x-client-transaction-id", value); } - let response = self.client.get_with_headers(url, &headers).await?; + let response = self + .client + .get_on(url, &headers, self.proxy_for(&session).as_ref()) + .await?; let (bytes, _) = self .account_response(&session, session_key, response) .await?; diff --git a/src/api/client_endpoints.rs b/src/api/client_endpoints.rs index 216ef55..c47b803 100644 --- a/src/api/client_endpoints.rs +++ b/src/api/client_endpoints.rs @@ -1,4 +1,7 @@ -use std::collections::BTreeMap; +use std::{ + collections::BTreeMap, + time::Duration, +}; use super::{ AboutAccountData, @@ -11,7 +14,6 @@ use super::{ Conversation, ConversationData, Deserialize, - Duration, EditHistory, EditHistoryData, Error, @@ -47,6 +49,22 @@ use super::{ timeout, }; +fn profile_from_parts(user: User, tweets: Timeline, photo_rail: Vec) -> Profile { + let pinned = tweets + .content + .iter() + .flatten() + .find(|tweet| user.pinned_tweet > 0 && tweet.id == user.pinned_tweet) + .cloned(); + + Profile { + user, + photo_rail, + pinned, + tweets, + } +} + fn search_page_is_stuck(timeline: &Timeline) -> bool { let mut ids = timeline.content.iter().flatten().map(Tweet::original_id); let Some(first) = ids.next() else { @@ -414,10 +432,69 @@ impl ApiClient { /// Get user's profile with tweets. pub async fn get_profile(&self, screen_name: &str, cursor: Option<&str>) -> Result { - // First get user info - let user = self.get_user(screen_name).await?; + self.get_profile_hinted(screen_name, cursor, None).await + } - // Protected/suspended accounts don't expose tweets + /// Same as [`get_profile`], but a known user (or id stub) lets tweets and + /// the photo rail start in the same wave as `UserByScreenName`. + pub async fn get_profile_hinted( + &self, + screen_name: &str, + cursor: Option<&str>, + known: Option<&User>, + ) -> Result { + let known_id = known + .map(|user| user.id.as_str()) + .filter(|id| !id.is_empty()); + + if let Some(id) = known_id { + if cursor.is_none() { + let (user_res, tweets_res, rail_res) = tokio::join!( + self.get_user(screen_name), + self.get_user_tweets(id, None), + self.get_photo_rail(id) + ); + let user = match user_res { + Ok(user) => user, + Err(err) => { + known + .filter(|user| !user.fullname.is_empty()) + .cloned() + .ok_or(err)? + }, + }; + if user.protected || user.suspended { + return Ok(Profile { + user, + ..Profile::default() + }); + } + return Ok(profile_from_parts( + user, + tweets_res?, + rail_res.unwrap_or_default(), + )); + } + + let user = if let Some(user) = known.filter(|user| !user.fullname.is_empty()) { + user.clone() + } else { + self.get_user(screen_name).await? + }; + if user.protected || user.suspended { + return Ok(Profile { + user, + ..Profile::default() + }); + } + return Ok(profile_from_parts( + user, + self.get_user_tweets(id, cursor).await?, + Vec::new(), + )); + } + + let user = self.get_user(screen_name).await?; if user.protected || user.suspended { return Ok(Profile { user, @@ -425,34 +502,15 @@ impl ApiClient { }); } - // Fetch tweets and photo rail in parallel (only for first page) let (tweets, photo_rail) = if cursor.is_none() { - let tweets_future = self.get_user_tweets(&user.id, None); - let photo_rail_future = self.get_photo_rail(&user.id); - let (tweets_result, photo_rail_result) = tokio::join!(tweets_future, photo_rail_future); + let (tweets_result, photo_rail_result) = + tokio::join!(self.get_user_tweets(&user.id, None), self.get_photo_rail(&user.id)); (tweets_result?, photo_rail_result.unwrap_or_default()) } else { (self.get_user_tweets(&user.id, cursor).await?, Vec::new()) }; - // Get pinned tweet if present - let pinned = if user.pinned_tweet > 0 { - tweets - .content - .iter() - .flatten() - .find(|tweet| tweet.id == user.pinned_tweet) - .cloned() - } else { - None - }; - - Ok(Profile { - user, - photo_rail, - pinned, - tweets, - }) + Ok(profile_from_parts(user, tweets, photo_rail)) } /// Search tweets. @@ -713,7 +771,10 @@ impl ApiClient { headers.insert("x-client-transaction-id", val); } - let response = self.client.get_with_headers(&url, &headers).await?; + let response = self + .client + .get_on(&url, &headers, self.proxy_for(&session).as_ref()) + .await?; let (bytes, _) = self .account_response(&session, endpoints::STRATO_TRANSLATE, response) .await?; diff --git a/src/api/filters.rs b/src/api/filters.rs index 4c21883..6e977b7 100644 --- a/src/api/filters.rs +++ b/src/api/filters.rs @@ -94,7 +94,11 @@ impl ApiClient { .await?; let response = self .client - .get_with_headers(endpoints::VERIFY_CREDENTIALS_URL, &headers) + .get_on( + endpoints::VERIFY_CREDENTIALS_URL, + &headers, + self.proxy_for(session).as_ref(), + ) .await?; let (bytes, _) = self .account_response(session, endpoints::VERIFY_CREDENTIALS, response) @@ -123,10 +127,11 @@ impl ApiClient { ); let response = self .client - .post_with_headers( + .post_on( endpoints::ACCOUNT_SETTINGS_URL, &headers, Bytes::from_static(DISPLAY_SENSITIVE_BODY.as_bytes()), + self.proxy_for(session).as_ref(), ) .await?; let (bytes, _) = self @@ -159,10 +164,11 @@ impl ApiClient { ); let response = self .client - .post_with_headers( + .post_on( &url, &headers, Bytes::from_static(SEARCH_SAFETY_BODY.as_bytes()), + self.proxy_for(session).as_ref(), ) .await?; let _ = self diff --git a/src/api/http.rs b/src/api/http.rs index 3e9e634..3ad5417 100644 --- a/src/api/http.rs +++ b/src/api/http.rs @@ -1,4 +1,8 @@ use std::{ + collections::{ + HashMap, + VecDeque, + }, fmt::Write as _, future::Future as _, io::{ @@ -9,12 +13,18 @@ use std::{ pin::Pin, result::Result as StdResult, str, - sync::Arc, + sync::{ + Arc, + Mutex, + }, task::{ Context, Poll, }, - time::Duration, + time::{ + Duration, + Instant as StdInstant, + }, }; use axum::http::{ @@ -75,13 +85,78 @@ const DEFAULT_BODY_LIMIT: usize = 32 * 1024 * 1024; // 32 MiB const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); const BODY_TIMEOUT: Duration = Duration::from_secs(60); const BODY_IDLE_TIMEOUT: Duration = Duration::from_secs(20); +const TUNNEL_IDLE: Duration = Duration::from_secs(45); +const TUNNEL_PER_KEY: usize = 2; + +type TunnelSender = http1::SendRequest>; + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct TunnelKey { + proxy_host: String, + proxy_port: u16, + origin_host: String, + origin_port: u16, +} + +struct TunnelPool { + idle: Mutex>>, +} + +impl TunnelPool { + fn new() -> Arc { + Arc::new(Self { + idle: Mutex::new(HashMap::new()), + }) + } + + fn take(&self, key: &TunnelKey) -> Option { + let mut map = self.idle.lock().ok()?; + let queue = map.get_mut(key)?; + let now = StdInstant::now(); + while let Some((sender, since)) = queue.pop_front() { + if sender.is_closed() || now.saturating_duration_since(since) > TUNNEL_IDLE { + continue; + } + return Some(sender); + } + None + } + + fn put(&self, key: TunnelKey, sender: TunnelSender) { + if sender.is_closed() { + return; + } + let Ok(mut map) = self.idle.lock() else { + return; + }; + let queue = map.entry(key).or_default(); + while queue.len() >= TUNNEL_PER_KEY { + queue.pop_front(); + } + queue.push_back((sender, StdInstant::now())); + } +} + +enum ResponseBody { + Incoming(hyper_body::Incoming), + Buffered(Bytes), +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ProxyKind { + Http, + Socks5, +} /// Parsed proxy configuration. #[derive(Clone)] -struct ProxyConfig { - host: String, - port: u16, - auth: Option, +pub struct ProxyConfig { + pub host: String, + pub port: u16, + pub kind: ProxyKind, + pub auth: Option, + pub socks_user: Option, + pub socks_pass: Option, } /// Lightweight HTTP client wrapping hyper-util's connection-pooling client. @@ -97,13 +172,14 @@ pub struct HttpClient { default_headers: HeaderMap, proxy: Option, tls: Arc, + tunnels: Arc, } /// Response wrapper providing convenience methods. pub struct Response { status: StatusCode, headers: HeaderMap, - body: hyper_body::Incoming, + body: ResponseBody, } /// Response body with both an overall deadline and an idle-chunk deadline. @@ -187,11 +263,12 @@ impl HttpClient { store }); - let tls = Arc::new( - rustls::ClientConfig::builder() - .with_root_certificates(roots.clone()) - .with_no_client_auth(), - ); + let mut tls_config = rustls::ClientConfig::builder() + .with_root_certificates(roots.clone()) + .with_no_client_auth(); + tls_config.alpn_protocols = vec![b"http/1.1".to_vec()]; + tls_config.resumption = rustls::client::Resumption::in_memory_sessions(512); + let tls = Arc::new(tls_config); let connector = HttpsConnectorBuilder::new() .with_tls_config( @@ -218,6 +295,7 @@ impl HttpClient { default_headers: HeaderMap::new(), proxy, tls, + tunnels: TunnelPool::new(), } } @@ -227,6 +305,13 @@ impl HttpClient { self } + /// Override the default proxy for this client. + #[must_use] + pub fn with_default_proxy(mut self, proxy: ProxyConfig) -> Self { + self.proxy = Some(proxy); + self + } + /// Send a request with a given method, optional extra headers, and body. async fn send( &self, @@ -234,9 +319,22 @@ impl HttpClient { uri: &str, extra_headers: &HeaderMap, body: Bytes, + ) -> Result { + self + .send_on(method, uri, extra_headers, body, None) + .await + } + + async fn send_on( + &self, + method: Method, + uri: &str, + extra_headers: &HeaderMap, + body: Bytes, + via: Option<&ProxyConfig>, ) -> Result { timeout(REQUEST_TIMEOUT, async { - if let Some(ref proxy) = self.proxy { + if let Some(proxy) = via.or(self.proxy.as_ref()) { self .send_via_proxy(proxy, method, uri, extra_headers, body) .await @@ -280,9 +378,9 @@ impl HttpClient { let (parts, body) = resp.into_parts(); Ok(Response { - status: parts.status, + status: parts.status, headers: parts.headers, - body, + body: ResponseBody::Incoming(body), }) } @@ -310,11 +408,51 @@ impl HttpClient { } }); let is_https = parsed.scheme_str() == Some("https"); + let path_and_query = parsed.path_and_query().map_or("/", PathAndQuery::as_str); + let key = TunnelKey { + proxy_host: proxy.host.clone(), + proxy_port: proxy.port, + origin_host: target_host.to_owned(), + origin_port: target_port, + }; + + if (proxy.kind == ProxyKind::Socks5 || is_https) + && let Some(sender) = self.tunnels.take(&key) + { + match self + .proxy_http1( + sender, + method.clone(), + path_and_query, + target_host, + extra_headers, + body.clone(), + None, + ) + .await + { + Ok((resp, sender)) => { + self.tunnels.put(key, sender); + return Ok(resp); + }, + Err(err) => { + tracing::debug!("idle proxy tunnel dropped: {err}"); + }, + } + } // TCP connect to proxy let mut stream = TcpStream::connect((&*proxy.host, proxy.port)) .await .map_err(|err| Error::Internal(format!("proxy connect: {err}")))?; + let _ = stream.set_nodelay(true); + + if proxy.kind == ProxyKind::Socks5 { + socks5_connect(&mut stream, proxy, target_host, target_port).await?; + return self + .send_origin_request(stream, is_https, &parsed, method, extra_headers, body, key) + .await; + } if is_https { // CONNECT handshake @@ -361,56 +499,11 @@ impl HttpClient { ))); } - // TLS handshake over the tunnel - let server_name = rustls::pki_types::ServerName::try_from(target_host.to_owned()) - .map_err(|err| Error::Internal(format!("invalid server name: {err}")))?; - - let tls_connector = tokio_rustls::TlsConnector::from(Arc::clone(&self.tls)); - let tls_stream = tls_connector - .connect(server_name, stream) + self + .send_origin_request(stream, true, &parsed, method, extra_headers, body, key) .await - .map_err(|err| Error::Internal(format!("proxy TLS handshake: {err}")))?; - - // HTTP/1.1 over the TLS tunnel - let (mut sender, conn) = http1::handshake(TokioIo::new(tls_stream)) - .await - .map_err(|err| Error::Internal(format!("proxy HTTP handshake: {err}")))?; - - tokio::spawn(async move { - if let Err(err) = conn.await { - tracing::debug!("proxy connection closed: {err}"); - } - }); - - let path_and_query = parsed.path_and_query().map_or("/", PathAndQuery::as_str); - let mut builder = hyper::Request::builder() - .method(method) - .uri(path_and_query) - .header(header::HOST, target_host); - - for (key, value) in &self.default_headers { - builder = builder.header(key, value); - } - for (key, value) in extra_headers { - builder = builder.header(key, value); - } - - let request = builder - .body(Full::new(body)) - .map_err(|err| Error::Internal(format!("build proxied request: {err}")))?; - - let resp = sender - .send_request(request) - .await - .map_err(|err| Error::Internal(format!("proxied request failed: {err}")))?; - - let (parts, body) = resp.into_parts(); - Ok(Response { - status: parts.status, - headers: parts.headers, - body, - }) } else { + let _ = key; // Send plain HTTP proxy requests with an absolute URI let (mut sender, conn) = http1::handshake(TokioIo::new(stream)) .await @@ -448,13 +541,127 @@ impl HttpClient { let (parts, body) = resp.into_parts(); Ok(Response { - status: parts.status, + status: parts.status, headers: parts.headers, - body, + body: ResponseBody::Incoming(body), }) } } + /// HTTP/1.1 on an already-tunneled stream (CONNECT or SOCKS5). + async fn send_origin_request( + &self, + stream: TcpStream, + is_https: bool, + parsed: &Uri, + method: Method, + extra_headers: &HeaderMap, + body: Bytes, + key: TunnelKey, + ) -> Result { + let target_host = parsed + .host() + .ok_or_else(|| Error::Internal("no host in URI".into()))?; + let path_and_query = parsed.path_and_query().map_or("/", PathAndQuery::as_str); + + let sender = if is_https { + let server_name = rustls::pki_types::ServerName::try_from(target_host.to_owned()) + .map_err(|err| Error::Internal(format!("invalid server name: {err}")))?; + let tls_connector = tokio_rustls::TlsConnector::from(Arc::clone(&self.tls)); + let tls_stream = tls_connector + .connect(server_name, stream) + .await + .map_err(|err| Error::Internal(format!("proxy TLS handshake: {err}")))?; + Self::http1_handshake(TokioIo::new(tls_stream)).await? + } else { + Self::http1_handshake(TokioIo::new(stream)).await? + }; + let (resp, sender) = self + .proxy_http1( + sender, + method, + path_and_query, + target_host, + extra_headers, + body, + None, + ) + .await?; + self.tunnels.put(key, sender); + Ok(resp) + } + + async fn http1_handshake(io: TokioIo) -> Result + where + IO: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static, + { + let (sender, conn) = http1::handshake(io) + .await + .map_err(|err| Error::Internal(format!("proxy HTTP handshake: {err}")))?; + tokio::spawn(async move { + if let Err(err) = conn.await { + tracing::debug!("proxy connection closed: {err}"); + } + }); + Ok(sender) + } + + #[expect( + clippy::too_many_arguments, + reason = "mirrors the origin HTTP/1.1 request" + )] + async fn proxy_http1( + &self, + mut sender: TunnelSender, + method: Method, + uri: &str, + host: &str, + extra_headers: &HeaderMap, + body: Bytes, + proxy_auth: Option<&str>, + ) -> Result<(Response, TunnelSender)> { + let mut builder = hyper::Request::builder() + .method(method) + .uri(uri) + .header(header::HOST, host) + .header(header::CONNECTION, "keep-alive"); + if let Some(auth) = proxy_auth { + builder = builder.header("Proxy-Authorization", format!("Basic {auth}")); + } + for (key, value) in &self.default_headers { + builder = builder.header(key, value); + } + for (key, value) in extra_headers { + builder = builder.header(key, value); + } + let request = builder + .body(Full::new(body)) + .map_err(|err| Error::Internal(format!("build proxied request: {err}")))?; + let resp = sender + .send_request(request) + .await + .map_err(|err| Error::Internal(format!("proxied request failed: {err}")))?; + let (parts, body) = resp.into_parts(); + let collected = timeout(BODY_TIMEOUT, body.collect()) + .await + .map_err(|_| Error::Internal("response body deadline exceeded".into()))? + .map_err(|err| Error::Internal(format!("read body: {err}")))?; + let buf = collected.to_bytes(); + if buf.len() > DEFAULT_BODY_LIMIT { + return Err(Error::Internal(format!( + "response body exceeded {DEFAULT_BODY_LIMIT} bytes" + ))); + } + Ok(( + Response { + status: parts.status, + headers: parts.headers, + body: ResponseBody::Buffered(buf), + }, + sender, + )) + } + /// Send a GET request. pub async fn get(&self, uri: &str) -> Result { self @@ -469,6 +676,18 @@ impl HttpClient { .await } + /// GET through an explicit proxy (session-pinned SOCKS5). + pub async fn get_on( + &self, + uri: &str, + extra_headers: &HeaderMap, + via: Option<&ProxyConfig>, + ) -> Result { + self + .send_on(Method::GET, uri, extra_headers, Bytes::new(), via) + .await + } + /// Send a POST request with additional headers and a body. pub async fn post_with_headers( &self, @@ -479,6 +698,19 @@ impl HttpClient { self.send(Method::POST, uri, extra_headers, body).await } + /// POST through an explicit proxy (session-pinned SOCKS5). + pub async fn post_on( + &self, + uri: &str, + extra_headers: &HeaderMap, + body: Bytes, + via: Option<&ProxyConfig>, + ) -> Result { + self + .send_on(Method::POST, uri, extra_headers, body, via) + .await + } + /// Send a HEAD request. pub async fn head(&self, uri: &str) -> Result { self @@ -487,6 +719,106 @@ impl HttpClient { } } +/// SOCKS5 CONNECT with username/password auth (RFC 1928 + 1929). +async fn socks5_connect( + stream: &mut TcpStream, + proxy: &ProxyConfig, + target_host: &str, + target_port: u16, +) -> Result<()> { + let user = proxy.socks_user.as_deref().unwrap_or(""); + let pass = proxy.socks_pass.as_deref().unwrap_or(""); + if user.len() > 255 || pass.len() > 255 { + return Err(Error::Internal("SOCKS5 credentials too long".into())); + } + + stream + .write_all(&[0x05, 0x01, 0x02]) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 greeting: {err}")))?; + let mut method = [0_u8; 2]; + stream + .read_exact(&mut method) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 greeting read: {err}")))?; + if method[0] != 0x05 || method[1] != 0x02 { + return Err(Error::Internal(format!( + "SOCKS5 auth method rejected: {method:?}" + ))); + } + + let mut auth = Vec::with_capacity(3 + user.len() + pass.len()); + auth.push(0x01); + auth.push(u8::try_from(user.len()).unwrap_or(0)); + auth.extend_from_slice(user.as_bytes()); + auth.push(u8::try_from(pass.len()).unwrap_or(0)); + auth.extend_from_slice(pass.as_bytes()); + stream + .write_all(&auth) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 auth: {err}")))?; + let mut auth_reply = [0_u8; 2]; + stream + .read_exact(&mut auth_reply) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 auth read: {err}")))?; + if auth_reply[1] != 0 { + return Err(Error::Internal(format!( + "SOCKS5 authentication failed: {}", + auth_reply[1] + ))); + } + + let host = target_host.as_bytes(); + if host.len() > 255 { + return Err(Error::Internal("SOCKS5 target hostname too long".into())); + } + let mut req = Vec::with_capacity(7 + host.len()); + req.extend_from_slice(&[0x05, 0x01, 0x00, 0x03]); + req.push(u8::try_from(host.len()).unwrap_or(0)); + req.extend_from_slice(host); + req.extend_from_slice(&target_port.to_be_bytes()); + stream + .write_all(&req) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 connect: {err}")))?; + + let mut hdr = [0_u8; 4]; + stream + .read_exact(&mut hdr) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 reply: {err}")))?; + if hdr[0] != 0x05 || hdr[1] != 0 { + return Err(Error::Internal(format!( + "SOCKS5 connect failed: status {}", + hdr[1] + ))); + } + let skip = match hdr[3] { + 1 => 4 + 2, + 4 => 16 + 2, + 3 => { + let mut len = [0_u8; 1]; + stream + .read_exact(&mut len) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 bind addr: {err}")))?; + usize::from(len[0]) + 2 + }, + atyp => { + return Err(Error::Internal(format!( + "SOCKS5 unknown address type {atyp}" + ))); + }, + }; + let mut rest = vec![0_u8; skip]; + stream + .read_exact(&mut rest) + .await + .map_err(|err| Error::Internal(format!("SOCKS5 bind addr: {err}")))?; + Ok(()) +} + /// Parse proxy URL (e.g. `http://host:port`) and optional `user:pass` auth. fn parse_proxy(url: &str, auth: &str) -> ProxyConfig { let stripped = url @@ -503,7 +835,14 @@ fn parse_proxy(url: &str, auth: &str) -> ProxyConfig { } else { Some(data_encoding::BASE64.encode(auth.as_bytes())) }; - ProxyConfig { host, port, auth } + ProxyConfig { + host, + port, + kind: ProxyKind::Http, + auth, + socks_user: None, + socks_pass: None, + } } impl Response { @@ -516,7 +855,12 @@ impl Response { } pub fn into_body(self) -> TimedBody { - TimedBody::new(self.body) + match self.body { + ResponseBody::Incoming(body) => TimedBody::new(body), + ResponseBody::Buffered(_) => { + panic!("into_body is for streamed media responses, not proxied API calls") + }, + } } /// Collect the response body as bytes, decompressing gzip if needed. @@ -532,23 +876,35 @@ impl Response { .and_then(|val| val.to_str().ok()) .is_some_and(|val| val.contains("gzip")); - let mut body = TimedBody::new(self.body); - let mut collected = Vec::new(); - while let Some(frame) = body.frame().await { - let frame = frame.map_err(|err| Error::Internal(format!("read body: {err}")))?; - let Ok(chunk) = frame.into_data() else { - continue; - }; - if collected.len().saturating_add(chunk.len()) > max_bytes { - return Err(Error::Internal(format!( - "response body exceeded {max_bytes} bytes" - ))); - } - collected.extend_from_slice(&chunk); + let collected = match self.body { + ResponseBody::Buffered(bytes) => bytes, + ResponseBody::Incoming(incoming) => { + let mut body = TimedBody::new(incoming); + let mut collected = Vec::new(); + while let Some(frame) = body.frame().await { + let frame = frame.map_err(|err| Error::Internal(format!("read body: {err}")))?; + let Ok(chunk) = frame.into_data() else { + continue; + }; + if collected.len().saturating_add(chunk.len()) > max_bytes { + return Err(Error::Internal(format!( + "response body exceeded {max_bytes} bytes" + ))); + } + collected.extend_from_slice(&chunk); + } + Bytes::from(collected) + }, + }; + + if collected.len() > max_bytes { + return Err(Error::Internal(format!( + "response body exceeded {max_bytes} bytes" + ))); } if is_gzip { - let gz = GzDecoder::new(&*collected); + let gz = GzDecoder::new(collected.as_ref()); let mut decoded = Vec::new(); let mut limited = gz.take(max_bytes.saturating_add(1) as u64); limited @@ -561,7 +917,7 @@ impl Response { } Ok(Bytes::from(decoded)) } else { - Ok(Bytes::from(collected)) + Ok(collected) } } diff --git a/src/api/mod.rs b/src/api/mod.rs index 53b216a..1c8f8ff 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -5,10 +5,12 @@ mod endpoints; mod filters; pub mod http; mod parser; +mod proxy_pool; pub mod schema; mod tid; pub use auth::*; pub use client::*; pub use http::HttpClient; +pub use proxy_pool::ProxyPool; pub use tid::*; diff --git a/src/api/proxy_pool.rs b/src/api/proxy_pool.rs new file mode 100644 index 0000000..0dbc5cb --- /dev/null +++ b/src/api/proxy_pool.rs @@ -0,0 +1,303 @@ +//! SOCKS5 exit pool for X API calls. +//! +//! Each cookie session is pinned to one `host:port` so its apparent IP stays +//! stable across restarts. The pin file stores only session id → port; secrets +//! stay in the proxy list. + +use std::{ + collections::BTreeMap, + net::IpAddr, + path::Path, + time::Duration, +}; + +use tokio::{ + fs, + net::{ + TcpStream, + lookup_host, + }, + process::Command, + time::{ + Instant, + timeout, + }, +}; + +use super::http::{ + ProxyConfig, + ProxyKind, +}; +use crate::{ + config::Config, + error::{ + Error, + Result, + }, +}; + +struct Endpoint { + port: u16, + username: String, + password: String, +} + +/// Resolved SOCKS5 endpoints plus the session→port map. +pub struct ProxyPool { + connect_host: String, + endpoints: Vec, + pins: BTreeMap, +} + +impl ProxyPool { + /// Load `host:port:user:pass` lines, pick the lowest-latency A record, and + /// restore pins. Returns [`None`] when the file is unset or empty. + pub async fn load(config: &Config) -> Result> { + let path = config.config.socks_proxies_file.trim(); + if path.is_empty() { + return Ok(None); + } + let content = fs::read_to_string(path).await.map_err(|err| { + Error::InvalidConfig(format!("socksProxiesFile {path:?} unreadable: {err}")) + })?; + let mut endpoints = Vec::new(); + let mut hostname = None; + for (idx, line) in content.lines().enumerate() { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + let (host, port, username, password) = parse_proxy_line(line).ok_or_else(|| { + Error::InvalidConfig(format!( + "socksProxiesFile line {}: expected host:port:user:pass", + idx + 1 + )) + })?; + match hostname { + None => hostname = Some(host), + Some(ref existing) if existing == &host => {}, + Some(ref existing) => { + return Err(Error::InvalidConfig(format!( + "socksProxiesFile mixes hosts {existing} and {host}; pin one hostname" + ))); + }, + } + endpoints.push(Endpoint { + port, + username, + password, + }); + } + if endpoints.is_empty() { + return Ok(None); + } + let hostname = hostname.unwrap_or_default(); + let probe_port = endpoints[0].port; + let connect_host = pick_fastest_host(&hostname, probe_port).await; + tracing::info!( + host = %hostname, + connect = %connect_host, + endpoints = endpoints.len(), + "SOCKS5 pool ready" + ); + + let pin_path = session_proxies_path(config); + let pins = load_pins(&pin_path).await; + Ok(Some(Self { + connect_host, + endpoints, + pins, + })) + } + + /// Assign a stable port to every session and rewrite the pin file. + pub async fn bind_sessions(mut self, session_ids: &[i64], config: &Config) -> Result { + let valid: std::collections::HashSet = + self.endpoints.iter().map(|endpoint| endpoint.port).collect(); + self.pins.retain(|_, port| valid.contains(port)); + + for id in session_ids { + if self.pins.contains_key(id) { + continue; + } + let port = least_used_port(&self.endpoints, &self.pins); + self.pins.insert(*id, port); + } + + let pin_path = session_proxies_path(config); + if !pin_path.is_empty() { + let body = serde_json::to_string_pretty(&self.pins) + .map_err(|err| Error::Internal(format!("session proxy pins: {err}")))?; + let tmp = format!("{pin_path}.tmp"); + fs::write(&tmp, body).await?; + fs::rename(&tmp, &pin_path).await?; + } + tracing::info!( + sessions = session_ids.len(), + pins = self.pins.len(), + "SOCKS5 session pins saved" + ); + Ok(self) + } + + /// SOCKS5 config for this session, always the same port once pinned. + #[must_use] + pub fn for_session(&self, session_id: i64) -> ProxyConfig { + let port = self.pins.get(&session_id).copied().unwrap_or_else(|| { + let idx = (session_id.unsigned_abs() as usize) % self.endpoints.len(); + self.endpoints[idx].port + }); + self.at_port(port) + } + + /// First endpoint, used for TID bootstrap and other unscoped API fetches. + #[must_use] + pub fn first(&self) -> ProxyConfig { + self.at_port(self.endpoints[0].port) + } + + fn at_port(&self, port: u16) -> ProxyConfig { + let endpoint = self + .endpoints + .iter() + .find(|endpoint| endpoint.port == port) + .unwrap_or(&self.endpoints[0]); + ProxyConfig { + host: self.connect_host.clone(), + port: endpoint.port, + kind: ProxyKind::Socks5, + auth: None, + socks_user: Some(endpoint.username.clone()), + socks_pass: Some(endpoint.password.clone()), + } + } +} + +fn session_proxies_path(config: &Config) -> String { + let path = config.config.session_proxies_file.trim(); + if path.is_empty() { + "session-proxies.json".to_owned() + } else { + path.to_owned() + } +} + +fn parse_proxy_line(line: &str) -> Option<(String, u16, String, String)> { + let mut parts = line.splitn(4, ':'); + let host = parts.next()?.to_owned(); + let port = parts.next()?.parse().ok()?; + let username = parts.next()?.to_owned(); + let password = parts.next()?.to_owned(); + if host.is_empty() || username.is_empty() || password.is_empty() || port == 0 { + return None; + } + Some((host, port, username, password)) +} + +async fn load_pins(path: &str) -> BTreeMap { + if path.is_empty() || !Path::new(path).exists() { + return BTreeMap::new(); + } + match fs::read_to_string(path).await { + Ok(content) => { + serde_json::from_str(&content).unwrap_or_else(|err| { + tracing::warn!("Ignoring unreadable session proxy pins: {err}"); + BTreeMap::new() + }) + }, + Err(_) => BTreeMap::new(), + } +} + +fn least_used_port(endpoints: &[Endpoint], pins: &BTreeMap) -> u16 { + let mut counts: BTreeMap = endpoints.iter().map(|ep| (ep.port, 0)).collect(); + for port in pins.values() { + if let Some(count) = counts.get_mut(port) { + *count += 1; + } + } + counts + .into_iter() + .min_by_key(|(port, count)| (*count, *port)) + .map(|(port, _)| port) + .unwrap_or(endpoints[0].port) +} + +/// Resolve `hostname` and pick the A record with the lowest ping. +async fn pick_fastest_host(hostname: &str, probe_port: u16) -> String { + let lookup = format!("{hostname}:{probe_port}"); + let addrs = match lookup_host(&lookup).await { + Ok(addrs) => addrs, + Err(err) => { + tracing::warn!("SOCKS5 DNS for {hostname} failed ({err}); using hostname"); + return hostname.to_owned(); + }, + }; + let mut ips: Vec = addrs.map(|addr| addr.ip()).collect(); + ips.sort(); + ips.dedup(); + if ips.is_empty() { + return hostname.to_owned(); + } + + let mut best: Option<(IpAddr, f64)> = None; + for ip in ips { + let rtt = probe_ip(ip, probe_port).await; + tracing::info!(%ip, rtt_ms = ?rtt, "SOCKS5 probe"); + let Some(ms) = rtt else { + continue; + }; + match best { + Some((_, best_ms)) if ms >= best_ms => {}, + _ => best = Some((ip, ms)), + } + } + match best { + Some((ip, ms)) => { + tracing::info!(%ip, rtt_ms = ms, "SOCKS5 using lowest-ping address"); + ip.to_string() + }, + None => hostname.to_owned(), + } +} + +async fn probe_ip(ip: IpAddr, port: u16) -> Option { + if let Some(ms) = icmp_rtt_ms(ip).await { + return Some(ms); + } + tcp_rtt_ms(ip, port).await +} + +async fn icmp_rtt_ms(ip: IpAddr) -> Option { + let output = timeout( + Duration::from_secs(6), + Command::new("ping") + .args(["-c", "3", "-W", "1", "-n", "-q", &ip.to_string()]) + .output(), + ) + .await + .ok()? + .ok()?; + if !output.status.success() { + return None; + } + let stdout = String::from_utf8_lossy(&output.stdout); + for line in stdout.lines() { + if !line.contains("min/avg/max") { + continue; + } + let stats = line.split('=').nth(1)?; + let avg = stats.split('/').nth(1)?.trim(); + return avg.parse().ok(); + } + None +} + +async fn tcp_rtt_ms(ip: IpAddr, port: u16) -> Option { + let start = Instant::now(); + timeout(Duration::from_secs(2), TcpStream::connect((ip, port))) + .await + .ok()? + .ok()?; + Some(start.elapsed().as_secs_f64() * 1000.0) +} diff --git a/src/cache/keys.rs b/src/cache/keys.rs index cdf29da..68a2d8a 100644 --- a/src/cache/keys.rs +++ b/src/cache/keys.rs @@ -43,6 +43,11 @@ pub fn user_id(id: &str) -> String { format!("uid:{id}") } +/// Username to user ID mapping, so a later profile can start tweets immediately. +pub fn username_id(username: &str) -> String { + format!("unid:{}", username.to_lowercase()) +} + pub fn rss_user(username: &str) -> String { rss(&format!("user:{}", username.to_lowercase())) } @@ -59,6 +64,19 @@ pub fn rss_search(query: &str) -> String { rss(&format!("search:{}", query.to_lowercase())) } +pub fn search_timeline(query: &str, product: &str, cursor: Option<&str>) -> String { + format!( + "search:{}:{}:{}", + query.to_lowercase(), + product, + cursor.unwrap_or("") + ) +} + +pub fn search_users(query: &str, cursor: Option<&str>) -> String { + format!("search-users:{}:{}", query.to_lowercase(), cursor.unwrap_or("")) +} + pub fn rss_user_search(username: &str, query: &str) -> String { rss(&format!( "usersearch:{}:{}", diff --git a/src/cache/mod.rs b/src/cache/mod.rs index b8ae241..7bb6362 100644 --- a/src/cache/mod.rs +++ b/src/cache/mod.rs @@ -7,4 +7,7 @@ mod store; /// Cache TTL constants (in seconds). pub mod ttl; pub use gif_cache::GifCache; -pub use store::Cache; +pub use store::{ + Cache, + Hit, +}; diff --git a/src/cache/store.rs b/src/cache/store.rs index f58f04c..0877384 100644 --- a/src/cache/store.rs +++ b/src/cache/store.rs @@ -2,9 +2,13 @@ use std::{ any::Any, - collections::HashMap, + collections::{ + HashMap, + HashSet, + }, sync::{ Arc, + Mutex, RwLock, }, time::{ @@ -15,8 +19,27 @@ use std::{ /// Cache entry holding a type-erased value and its expiry. struct Entry { - value: Arc, - expires: Instant, + value: Arc, + fresh_until: Instant, + stale_until: Instant, +} + +/// A cache lookup that may still be served after its fresh TTL. +pub enum Hit { + Fresh(T), + Stale(T), +} + +/// Drops the in-flight refresh mark when the refresher finishes or panics. +pub struct RefreshGuard { + cache: Cache, + key: String, +} + +impl Drop for RefreshGuard { + fn drop(&mut self) { + self.cache.finish_refresh(&self.key); + } } /// In-process cache with TTL-based expiry and a hard entry-count cap. @@ -28,6 +51,7 @@ struct Entry { #[derive(Clone)] pub struct Cache { inner: Arc>>, + inflight: Arc>>, max_entries: usize, } @@ -35,23 +59,42 @@ impl Cache { pub fn new(max_entries: usize) -> Self { Self { inner: Arc::new(RwLock::new(HashMap::new())), + inflight: Arc::new(Mutex::new(HashSet::new())), max_entries: max_entries.max(16), } } /// Get a value from cache, returning [`None`] if missing, expired, or /// type-mismatched. - #[expect(clippy::significant_drop_tightening, reason = "entry borrows from map")] pub fn get(&self, key: &str) -> Option + where + T: Any + Send + Sync + Clone, + { + match self.lookup(key) { + Some(Hit::Fresh(value)) => Some(value), + _ => None, + } + } + + /// Fresh or stale value. Stale is only returned for keys stored with + /// [`set_swr`](Self::set_swr). + #[expect(clippy::significant_drop_tightening, reason = "entry borrows from map")] + pub fn lookup(&self, key: &str) -> Option> where T: Any + Send + Sync + Clone, { let map = self.inner.read().ok()?; let entry = map.get(key)?; - if entry.expires <= Instant::now() { + let now = Instant::now(); + if entry.stale_until <= now { return None; } - entry.value.downcast_ref::().cloned() + let value = entry.value.downcast_ref::()?.clone(); + if entry.fresh_until > now { + Some(Hit::Fresh(value)) + } else { + Some(Hit::Stale(value)) + } } /// Set a value in cache with TTL in seconds. @@ -59,23 +102,61 @@ impl Cache { where T: Any + Send + Sync + Clone, { + self.insert(cache_key, value, ttl_seconds, 0); + } + + /// Fresh for `fresh_seconds`, then serve stale for `stale_seconds` more. + pub fn set_swr(&self, cache_key: &str, value: &T, fresh_seconds: u64, stale_seconds: u64) + where + T: Any + Send + Sync + Clone, + { + self.insert(cache_key, value, fresh_seconds, stale_seconds); + } + + /// Mark `key` as refreshing. The guard clears the mark on drop. + pub fn start_refresh(&self, key: &str) -> Option { + let inserted = self + .inflight + .lock() + .ok()? + .insert(key.to_owned()); + inserted.then(|| { + RefreshGuard { + cache: self.clone(), + key: key.to_owned(), + } + }) + } + + fn finish_refresh(&self, key: &str) { + if let Ok(mut inflight) = self.inflight.lock() { + inflight.remove(key); + } + } + + fn insert(&self, cache_key: &str, value: &T, fresh_seconds: u64, stale_seconds: u64) + where + T: Any + Send + Sync + Clone, + { + let now = Instant::now(); + let fresh_until = now + Duration::from_secs(fresh_seconds); let cache_entry = Entry { - value: Arc::new(value.clone()), - expires: Instant::now() + Duration::from_secs(ttl_seconds), + value: Arc::new(value.clone()), + fresh_until, + stale_until: fresh_until + Duration::from_secs(stale_seconds), }; if let Ok(mut map) = self.inner.write() { map.insert(cache_key.to_owned(), cache_entry); if map.len() > self.max_entries { - let now = Instant::now(); - map.retain(|_, cached| cached.expires > now); + map.retain(|_, cached| cached.stale_until > now); if map.len() > self.max_entries { let target = self.max_entries * 3 / 4; let drop_n = map.len() - target; let mut by_expiry: Vec<(Instant, String)> = map .iter() - .map(|(stored_key, stored_entry)| (stored_entry.expires, stored_key.clone())) + .map(|(stored_key, stored_entry)| (stored_entry.stale_until, stored_key.clone())) .collect(); by_expiry.select_nth_unstable_by_key(drop_n, |&(expiry, _)| expiry); for (_, eviction_key) in by_expiry.into_iter().take(drop_n) { diff --git a/src/cache/ttl.rs b/src/cache/ttl.rs index 54617a1..125ffe1 100644 --- a/src/cache/ttl.rs +++ b/src/cache/ttl.rs @@ -1,6 +1,15 @@ /// Standard TTL for user/profile/tweet data. pub const DEFAULT: u64 = 5 * 60; // 5 minutes +/// Serve a stale profile this long after [`DEFAULT`] while a refresh runs. +pub const DEFAULT_STALE: u64 = 20 * 60; + +/// First-page search. Short so results stay fresh but repeats skip X. +pub const SEARCH: u64 = 60; + +/// Serve a stale first search page this long after [`SEARCH`]. +pub const SEARCH_STALE: u64 = 4 * 60; + /// Long TTL for immutable mappings like user ID -> username. pub const USER_ID_MAPPING: u64 = 24 * 60 * 60; // 1 day diff --git a/src/config.rs b/src/config.rs index 44d992e..e42b0ed 100644 --- a/src/config.rs +++ b/src/config.rs @@ -138,6 +138,12 @@ pub struct AppConfig { pub proxy_auth: String, #[serde(default, rename = "apiProxy")] pub api_proxy: String, + /// `host:port:user:pass` SOCKS5 list used for X API calls. + #[serde(default, rename = "socksProxiesFile")] + pub socks_proxies_file: String, + /// session id → SOCKS5 port pins so each account keeps the same exit IP. + #[serde(default, rename = "sessionProxiesFile")] + pub session_proxies_file: String, #[serde(default, rename = "disableTid")] pub disable_tid: bool, /// Read about-account data from X-Posed's public cache before asking X. diff --git a/src/error.rs b/src/error.rs index 65ba604..b44f54a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -116,6 +116,11 @@ pub enum Error { #[error("Rate limited")] RateLimited, + /// Cloudflare/WAF 429 against this IP. Other cookie sessions share the + /// path, so retrying them only marks the whole pool as exhausted. + #[error("Upstream throttled this IP")] + IpThrottled, + #[error("Session rejected: {0}")] SessionRejected(String), @@ -202,7 +207,7 @@ impl Error { client_budget_message(), ) }, - Self::RateLimited => { + Self::RateLimited | Self::IpThrottled => { ( StatusCode::TOO_MANY_REQUESTS, "Rate limited", diff --git a/src/main.rs b/src/main.rs index 1e052c6..ee9674d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -48,6 +48,7 @@ use crate::{ api::{ ApiClient, HttpClient, + ProxyPool, SessionPool, }, cache::Cache, @@ -108,9 +109,13 @@ async fn main() -> eyre::Result<()> { let sessions_path = env::var("TEAPOT_SESSIONS_FILE").unwrap_or_else(|_| "sessions.jsonl".to_owned()); let sessions = SessionPool::load(&sessions_path, config.config.max_concurrent_reqs).await?; + let mut proxies = ProxyPool::load(&config).await?; + if let Some(pool) = proxies.take() { + proxies = Some(pool.bind_sessions(&sessions.session_ids(), &config).await?); + } // Initialize API client - let api = ApiClient::new(&config, sessions); + let api = ApiClient::new(&config, sessions, proxies); api.spawn_filter_sync(); // Initialize GIF transcoder if local mode diff --git a/src/routes/helpers.rs b/src/routes/helpers.rs index 99a60f4..fedc51c 100644 --- a/src/routes/helpers.rs +++ b/src/routes/helpers.rs @@ -1,7 +1,10 @@ use std::{ + any::Any, collections::BTreeMap, + future::Future, iter, slice, + sync::LazyLock, }; use axum::{ @@ -16,9 +19,12 @@ use axum::{ }, }; +use tokio::sync::Semaphore; + use crate::{ AppState, cache::{ + Hit, keys as cache_keys, ttl, }, @@ -30,6 +36,7 @@ use crate::{ types::{ AccountContext, Conversation, + Profile, Timeline, Translation, Tweet, @@ -39,56 +46,207 @@ use crate::{ views::layout, }; +/// Caps background profile prefetches so they cannot crowd out live pages. +static PROFILE_PREFETCH: LazyLock = LazyLock::new(|| Semaphore::new(3)); +const PREFETCH_PROFILES: usize = 4; + #[derive(Clone)] struct CachedRss { body: String, min_id: Option, } +/// Fresh cache hit, or a stale value plus a single background refresh. +pub fn swr_take(state: &AppState, key: &str, refresh: F) -> Option +where + T: Any + Clone + Send + Sync, + F: FnOnce(AppState) -> Fut + Send + 'static, + Fut: Future + Send + 'static, +{ + match state.cache.lookup(key) { + Some(Hit::Fresh(value)) => Some(value), + Some(Hit::Stale(value)) => { + if let Some(guard) = state.cache.start_refresh(key) { + let state = state.clone(); + tokio::spawn(async move { + let _guard = guard; + refresh(state).await; + }); + } + Some(value) + }, + None => None, + } +} + +/// Cached user, or a stub that only has `id` so tweets can start immediately. +pub fn user_hint(state: &AppState, username: &str) -> Option { + let key = cache_keys::user(username); + if let Some(Hit::Fresh(user) | Hit::Stale(user)) = state.cache.lookup(&key) { + return Some(user); + } + let id = state.cache.get::(&cache_keys::username_id(username))?; + Some(User { + id, + username: username.to_owned(), + ..User::default() + }) +} + +/// Persist the bidirectional id mapping. Does not overwrite a full user card. +pub fn remember_user_id(state: &AppState, user: &User) { + if user.id.is_empty() || user.username.is_empty() { + return; + } + state + .cache + .set(&cache_keys::user_id(&user.id), &user.username, ttl::USER_ID_MAPPING); + state.cache.set( + &cache_keys::username_id(&user.username), + &user.id, + ttl::USER_ID_MAPPING, + ); +} + +/// Store a full user record plus id mappings. +pub fn store_user(state: &AppState, user: &User) { + remember_user_id(state, user); + if user.id.is_empty() || user.username.is_empty() { + return; + } + state + .cache + .set_swr(&cache_keys::user(&user.username), user, ttl::DEFAULT, ttl::DEFAULT_STALE); +} + +/// Store a first-page profile and its user. +pub fn store_profile(state: &AppState, profile: &Profile) { + store_user(state, &profile.user); + state.cache.set_swr( + &cache_keys::profile(&profile.user.username), + profile, + ttl::DEFAULT, + ttl::DEFAULT_STALE, + ); +} + +/// Seed id mappings from tweets already on the page so a later profile click +/// can fetch tweets in the same wave as `UserByScreenName`. +pub fn remember_tweets(state: &AppState, tweets: &[Tweet]) { + for tweet in tweets { + for user in tweet_users(tweet) { + remember_user_id(state, user); + } + } +} + +/// Warm a few uncached profiles in the background after search/timeline HTML. +pub fn prefetch_profiles(state: &AppState, tweets: &[Tweet]) { + let mut names = Vec::new(); + for tweet in tweets { + let name = tweet.user.username.to_lowercase(); + if name.is_empty() || names.iter().any(|seen: &String| seen == &name) { + continue; + } + if state.cache.get::(&cache_keys::profile(&name)).is_some() { + continue; + } + names.push(name); + if names.len() >= PREFETCH_PROFILES { + break; + } + } + if names.is_empty() { + return; + } + + let state = state.clone(); + tokio::spawn(async move { + for name in names { + let Ok(permit) = PROFILE_PREFETCH.try_acquire() else { + break; + }; + let hint = user_hint(&state, &name); + if let Ok(profile) = state.api.get_profile_hinted(&name, None, hint.as_ref()).await { + store_profile(&state, &profile); + } + drop(permit); + } + }); +} + /// Fetch a user, using cache when available. pub async fn get_cached_user(state: &AppState, username: &str) -> Result { let cache_key = cache_keys::user(username); - if let Some(cached) = state.cache.get::(&cache_key) { + if let Some(cached) = swr_take(state, &cache_key, { + let username = username.to_owned(); + move |state| async move { + if let Ok(user) = state.api.get_user(&username).await { + store_user(&state, &user); + } + } + }) { return Ok(cached); } let fetched = state.api.get_user(username).await?; - state.cache.set(&cache_key, &fetched, ttl::DEFAULT); + store_user(state, &fetched); Ok(fetched) } -/// Decorate one user in place, spending an `AboutAccountQuery` only when both -/// caches miss. -pub async fn apply_account_context(state: &AppState, user: &mut User) { +pub fn apply_context_to_user(user: &mut User, context: &AccountContext) { + user.account_based_in.clone_from(&context.account_based_in); + user.connection_source.clone_from(&context.connection_source); + user.location_accurate = context.location_accurate; +} + +/// Apply About Account from cache only. A miss is filled in the background. +pub fn apply_cached_account_context(state: &AppState, user: &mut User) { let username = user.username.to_lowercase(); if username.is_empty() { return; } let key = cache_keys::account_context(&username); - - let context = if let Some(cached) = state.cache.get::(&key) { - cached - } else { - let community = community_contexts(state, slice::from_ref(&username)).await; - let resolved = match community.into_values().next() { - Some(context) => context, - None => { - state - .api - .get_account_context(&username) - .await - .unwrap_or_else(|err| { - tracing::debug!(username, "About Account unavailable: {err}"); - AccountContext::default() - }) - }, - }; - state.cache.set(&key, &resolved, ttl::ACCOUNT_CONTEXT); - resolved + if let Some(context) = state.cache.get::(&key) { + apply_context_to_user(user, &context); + return; + } + let Some(guard) = state.cache.start_refresh(&key) else { + return; }; + let state = state.clone(); + tokio::spawn(async move { + let _guard = guard; + let _ = load_account_context(&state, &username).await; + }); +} - user.account_based_in = context.account_based_in; - user.connection_source = context.connection_source; - user.location_accurate = context.location_accurate; +/// Load About Account, overlapping community cache and X when needed. +pub async fn load_account_context(state: &AppState, username: &str) -> Option { + let username = username.to_lowercase(); + if username.is_empty() { + return None; + } + let key = cache_keys::account_context(&username); + if let Some(cached) = state.cache.get::(&key) { + return (!cached.is_empty()).then_some(cached); + } + + let community = community_contexts(state, slice::from_ref(&username)).await; + let resolved = match community.into_values().next() { + Some(context) => context, + None => { + state + .api + .get_account_context(&username) + .await + .unwrap_or_else(|err| { + tracing::debug!(username, "About Account unavailable: {err}"); + AccountContext::default() + }) + }, + }; + state.cache.set(&key, &resolved, ttl::ACCOUNT_CONTEXT); + (!resolved.is_empty()).then_some(resolved) } /// Resolve about-account data for many users at once without spending X @@ -347,7 +505,9 @@ pub fn api_error(config: &Config, err: &Error) -> Response { pub fn api_error_titled(config: &Config, err: &Error, generic: &str) -> Response { let (status, title, message) = classify_error(err, generic); - if status.is_server_error() { + if matches!(err, Error::TransientUpstream) { + tracing::warn!(error = ?err, "upstream empty"); + } else if status.is_server_error() { tracing::error!(error = ?err, "request failed"); } else if status == StatusCode::TOO_MANY_REQUESTS { tracing::debug!(error = ?err, "request refused"); diff --git a/src/routes/search.rs b/src/routes/search.rs index 9b89fb8..b866448 100644 --- a/src/routes/search.rs +++ b/src/routes/search.rs @@ -8,7 +8,14 @@ use axum::{ RawQuery, State, }, - http::header::CONTENT_TYPE, + http::{ + StatusCode, + header::{ + CACHE_CONTROL, + CONTENT_TYPE, + REFRESH, + }, + }, response::{ Html, IntoResponse as _, @@ -18,17 +25,29 @@ use axum::{ routing::get, }; use axum_extra::extract::CookieJar; +use maud::html; use serde::Deserialize; use super::helpers; use crate::{ AppState, - error::Result, + cache::{ + keys as cache_keys, + ttl, + }, + config::Config, + error::{ + Error, + Result, + }, types::{ + PaginatedResult, Prefs, Query, QueryKind, + Timeline, Tweet, + User, }, views::{ layout, @@ -88,6 +107,11 @@ pub struct SearchQuery { pub e_replies: Option, #[serde(rename = "e-retweets")] pub e_retweets: Option, + /// Browser auto-retry count after an empty SearchTimeline 404. + #[serde(default)] + pub retry: u8, + /// AJAX infinite-scroll request — return a timeline fragment, not a page. + pub scroll: Option, } impl SearchQuery { @@ -181,6 +205,83 @@ impl SearchQuery { } } +const SEARCH_AUTO_RETRIES: u8 = 2; +const SEARCH_RETRY_WAIT_SECS: u8 = 2; + +fn with_retry_param(raw_qs: &str, retry: u8) -> String { + let rest = raw_qs + .split('&') + .filter(|pair| { + !pair.is_empty() && !pair.starts_with("retry=") && !pair.starts_with("scroll=") + }) + .collect::>(); + if rest.is_empty() { + format!("/search?retry={retry}") + } else { + format!("/search?{}&retry={retry}", rest.join("&")) + } +} + +fn is_scroll_request(params: &SearchQuery) -> bool { + params.scroll.as_deref() == Some("true") +} + +fn search_error( + config: &Config, + prefs: &Prefs, + raw_qs: Option<&str>, + params: &SearchQuery, + err: &Error, +) -> Response { + if is_scroll_request(params) { + // A 200 "Trying again…" page looks like success to infiniteScroll.js, + // which then drops the Load more sentinel and cannot page further. + return helpers::api_error_titled(config, err, "Search Error"); + } + search_upstream_error(config, prefs, raw_qs, params.retry, err) +} + +fn search_upstream_error( + config: &Config, + prefs: &Prefs, + raw_qs: Option<&str>, + attempt: u8, + err: &Error, +) -> Response { + if matches!(err, Error::TransientUpstream) && attempt < SEARCH_AUTO_RETRIES { + let url = with_retry_param(raw_qs.unwrap_or(""), attempt + 1); + let wait = SEARCH_RETRY_WAIT_SECS; + tracing::warn!(attempt, %url, "search empty, auto-retrying in the browser"); + let refresh = html! { + meta http-equiv="refresh" content=(format!("{wait};url={url}")); + }; + let content = html! { + div class="panel-container" { + div class="error-panel" { + span { "X did not return a result. Trying again…" } + " " + a href=(&url) { "Retry now" } + } + } + }; + let markup = layout::PageLayout::new(config, "Try again", content) + .prefs(prefs) + .head_extra(&refresh) + .description("X did not return a result. Trying again…") + .render(); + return ( + StatusCode::OK, + [ + (REFRESH, format!("{wait};url={url}")), + (CACHE_CONTROL, "no-store".to_owned()), + ], + Html(markup.into_string()), + ) + .into_response(); + } + helpers::api_error_titled(config, err, "Search Error") +} + pub fn router() -> Router { Router::new() .route("/search", get(search)) @@ -212,6 +313,7 @@ async fn search( // Extract prefs from cookies let prefs = Prefs::from_cookies(&jar, &state.config); + let is_scroll = is_scroll_request(¶ms); let raw_q = params.query.clone().unwrap_or_default(); @@ -271,14 +373,41 @@ async fn search( } if is_user_search { - // User search - let search_result = state - .api - .search_users(&raw_q, params.cursor.as_deref()) - .await; + let search_result = if params.cursor.is_none() { + let cache_key = cache_keys::search_users(&raw_q, None); + if let Some(cached) = helpers::swr_take::, _, _>(&state, &cache_key, { + let raw_q = raw_q.clone(); + let cache_key = cache_key.clone(); + move |state| async move { + if let Ok(data) = state.api.search_users(&raw_q, None).await { + state + .cache + .set_swr(&cache_key, &data, ttl::SEARCH, ttl::SEARCH_STALE); + } + } + }) { + Ok(cached) + } else { + let result = state.api.search_users(&raw_q, None).await; + if let Ok(ref data) = result { + state + .cache + .set_swr(&cache_key, data, ttl::SEARCH, ttl::SEARCH_STALE); + } + result + } + } else { + state + .api + .search_users(&raw_q, params.cursor.as_deref()) + .await + }; match search_result { Ok(result) => { + for user in &result.content { + helpers::remember_user_id(&state, user); + } let cursor = result.bottom.as_deref(); let newer_url = params.cursor.is_some().then(|| { @@ -304,6 +433,9 @@ async fn search( "/search?q={}&f=users", percent_encoding::utf8_percent_encode(&raw_q, percent_encoding::NON_ALPHANUMERIC) ); + if is_scroll { + return Ok(Html(content.into_string()).into_response()); + } let markup = layout::PageLayout::new(&state.config, &title, content) .prefs(&prefs) .canonical(&canonical) @@ -312,10 +444,12 @@ async fn search( Ok(Html(markup.into_string()).into_response()) }, Err(err) => { - Ok(helpers::api_error_titled( + Ok(search_error( &state.config, + &prefs, + raw_qs.as_deref(), + ¶ms, &err, - "Search Error", )) }, } @@ -339,16 +473,42 @@ async fn search( _ => "tweets", }; - // Execute search - let search_result = state - .api - .search(&api_query, params.cursor.as_deref(), product) - .await; + let search_result = if params.cursor.is_none() { + let cache_key = cache_keys::search_timeline(&api_query, product, None); + if let Some(cached) = helpers::swr_take::(&state, &cache_key, { + let api_query = api_query.clone(); + let cache_key = cache_key.clone(); + move |state| async move { + if let Ok(data) = state.api.search(&api_query, None, product).await { + state + .cache + .set_swr(&cache_key, &data, ttl::SEARCH, ttl::SEARCH_STALE); + } + } + }) { + Ok(cached) + } else { + let result = state.api.search(&api_query, None, product).await; + if let Ok(ref data) = result { + state + .cache + .set_swr(&cache_key, data, ttl::SEARCH, ttl::SEARCH_STALE); + } + result + } + } else { + state + .api + .search(&api_query, params.cursor.as_deref(), product) + .await + }; match search_result { Ok(timeline) => { let mut tweets = dedup_search_tweets(timeline.content.into_iter().flatten().collect()); helpers::enrich_tweet_groups(&state, slice::from_mut(&mut tweets)).await; + helpers::remember_tweets(&state, &tweets); + helpers::prefetch_profiles(&state, &tweets); let cursor = timeline.bottom.as_deref(); // Display the original user query, not the API query @@ -388,6 +548,9 @@ async fn search( "/search?q={}", percent_encoding::utf8_percent_encode(&raw_q, percent_encoding::NON_ALPHANUMERIC) ); + if is_scroll { + return Ok(Html(content.into_string()).into_response()); + } let markup = layout::PageLayout::new(&state.config, &title, content) .prefs(&prefs) .rss(&rss_url) @@ -397,10 +560,12 @@ async fn search( Ok(Html(markup.into_string()).into_response()) }, Err(err) => { - Ok(helpers::api_error_titled( + Ok(search_error( &state.config, + &prefs, + raw_qs.as_deref(), + ¶ms, &err, - "Search Error", )) }, } @@ -421,9 +586,10 @@ async fn hashtag( jar, RawQuery(None), AxumQuery(SearchQuery { - query: Some(hashtag_query), + query: Some(hashtag_query), filter: query.filter, cursor: query.cursor, + scroll: query.scroll, ..Default::default() }), ) diff --git a/src/routes/status.rs b/src/routes/status.rs index 1b5d65d..e1b1887 100644 --- a/src/routes/status.rs +++ b/src/routes/status.rs @@ -607,6 +607,11 @@ async fn user_by_id(State(state): State, Path(id): Path) -> Re state .cache .set(&cache_key, &user.username, ttl::USER_ID_MAPPING); + state.cache.set( + &cache_keys::username_id(&user.username), + &user.id, + ttl::USER_ID_MAPPING, + ); Ok(Redirect::to(&format!("/{}", user.username)).into_response()) }, Err(Error::UserNotFound(msg)) => { diff --git a/src/routes/timeline.rs b/src/routes/timeline.rs index 447dce6..0eec6fe 100644 --- a/src/routes/timeline.rs +++ b/src/routes/timeline.rs @@ -52,6 +52,40 @@ use crate::{ }, }; +async fn load_first_profile(state: &AppState, username: &str) -> Result { + let cache_key = cache_keys::profile(username); + if let Some(cached) = helpers::swr_take(state, &cache_key, { + let username = username.to_owned(); + move |state| async move { + let hint = helpers::user_hint(&state, &username); + if let Ok(profile) = state + .api + .get_profile_hinted(&username, None, hint.as_ref()) + .await + { + helpers::store_profile(&state, &profile); + } + } + }) { + tracing::debug!("Cache hit for profile: {username}"); + return Ok(cached); + } + + let hint = helpers::user_hint(state, username); + let (profile_res, about) = tokio::join!( + state + .api + .get_profile_hinted(username, None, hint.as_ref()), + helpers::load_account_context(state, username), + ); + let mut profile = profile_res?; + if let Some(context) = about { + helpers::apply_context_to_user(&mut profile.user, &context); + } + helpers::store_profile(state, &profile); + Ok(profile) +} + /// Reserved path names that cannot be usernames. const RESERVED_PATHS: &[&str] = &[ "pic", @@ -162,33 +196,26 @@ async fn user_timeline( // Extract prefs from cookies let prefs = Prefs::from_cookies(&jar, &state.config); - // Check cache first (only for initial page load without cursor) - let cache_key = cache_keys::profile(&username); let profile_result = if query.cursor.is_none() { - if let Some(cached) = state.cache.get::(&cache_key) { - tracing::debug!("Cache hit for profile: {username}"); - Ok(cached) - } else { - let result = state.api.get_profile(&username, None).await; - if let Ok(ref profile_data) = result { - state.cache.set(&cache_key, profile_data, ttl::DEFAULT); - } - result - } + load_first_profile(&state, &username).await } else { + let hint = helpers::user_hint(&state, &username); state .api - .get_profile(&username, query.cursor.as_deref()) + .get_profile_hinted(&username, query.cursor.as_deref(), hint.as_ref()) .await }; match profile_result { Ok(mut profile_data) => { - helpers::apply_account_context(&state, &mut profile_data.user).await; + helpers::apply_cached_account_context(&state, &mut profile_data.user); // For AJAX scroll requests, return only the tweets HTML if is_scroll { let (mut tweets, cursor) = extract_timeline(profile_data.tweets); helpers::enrich_tweet_groups(&state, &mut tweets).await; + for group in &tweets { + helpers::remember_tweets(&state, group); + } let base_url = format!("/{}", profile_data.user.username); let content = timeline::render_timeline_with_prefs( &tweets, @@ -221,6 +248,13 @@ async fn user_timeline( .cursor .is_some() .then(|| format!("/{}", profile_data.user.username)); + for group in &profile_data.tweets.content { + helpers::remember_tweets(&state, group); + } + if let Some(first) = profile_data.tweets.content.first() { + helpers::prefetch_profiles(&state, first); + } + let content = profile::render_profile_with_prefs( &profile_data, &state.config, @@ -265,7 +299,24 @@ async fn user_tab_handler( let fetch_timeline = async { let cache_key = cache_keys::timeline(username, cache_kind); if cursor.is_none() { - if let Some(cached) = state.cache.get::(&cache_key) { + if let Some(cached) = helpers::swr_take::(state, &cache_key, { + let user_id = user.id.clone(); + let cache_key = cache_key.clone(); + move |state| async move { + let result = match tab { + TimelineKind::Replies => { + state.api.get_user_tweets_and_replies(&user_id, None).await + }, + TimelineKind::Media => state.api.get_user_media(&user_id, None).await, + _ => return, + }; + if let Ok(data) = result { + state + .cache + .set_swr(&cache_key, &data, ttl::DEFAULT, ttl::DEFAULT_STALE); + } + } + }) { return Ok(cached); } let result = match tab { @@ -274,7 +325,9 @@ async fn user_tab_handler( _ => unreachable!(), }; if let Ok(ref data) = result { - state.cache.set(&cache_key, data, ttl::DEFAULT); + state + .cache + .set_swr(&cache_key, data, ttl::DEFAULT, ttl::DEFAULT_STALE); } result } else { @@ -304,6 +357,12 @@ async fn user_tab_handler( Ok(data) => { let (mut tweets, next_cursor) = extract_timeline(data); helpers::enrich_tweet_groups(state, &mut tweets).await; + for group in &tweets { + helpers::remember_tweets(state, group); + } + if let Some(first) = tweets.first() { + helpers::prefetch_profiles(state, first); + } let base_url = format!("/{username}/{tab_str}"); let newer = has_request_cursor.then_some(base_url.as_str()); diff --git a/src/views/embed_tests.rs b/src/views/embed_tests.rs index ac81ece..f3d246c 100644 --- a/src/views/embed_tests.rs +++ b/src/views/embed_tests.rs @@ -49,6 +49,8 @@ fn test_config() -> Config { proxy: String::new(), proxy_auth: String::new(), api_proxy: String::new(), + socks_proxies_file: String::new(), + session_proxies_file: String::new(), disable_tid: false, x_posed_community_cache: false, max_concurrent_reqs: 2, diff --git a/src/views/layout.rs b/src/views/layout.rs index d2d8694..04af378 100644 --- a/src/views/layout.rs +++ b/src/views/layout.rs @@ -14,7 +14,7 @@ use crate::{ }; pub const STYLE_CSS: &str = "/css/style.css"; -pub const CLEAN_CSS: &str = "/css/clean.css?teapawt"; +pub const CLEAN_CSS: &str = "/css/clean.css?teapawt2"; pub const FONTELLO_CSS: &str = "/css/fontello.css"; /// Builder for rendering a full page layout. @@ -240,7 +240,7 @@ impl<'a> PageLayout<'a> { script src="/js/threadContext.js" defer="" {} script src="/js/imageLightbox.js" defer="" {} @if infinite_scroll { - script src="/js/infiniteScroll.js" defer="" {} + script src="/js/infiniteScroll.js?teapawt" defer="" {} } // Font preload