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.
This commit is contained in:
4cuck committed 2026-09-02 19:59:32 +00:00
1 parent b41745925c
commit 95cc2df375
28 files changed
+1652 -255

No files matched your search

+9
View File
@@ -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/
+74 -11
View File
@@ -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
+12 -1
View File
@@ -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
+5
View File
@@ -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
+6
View File
@@ -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);
+1 -2
View File
@@ -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;
}
+46 -20
View File
@@ -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);
});
}
}
+10
View File
@@ -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<i64> {
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()
+7 -7
View File
@@ -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,
},
];
+44 -23
View File
@@ -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<i64> {
pub struct ApiClient {
pub(crate) client: HttpClient,
pub(crate) sessions: SessionPool,
proxies: Option<Arc<ProxyPool>>,
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<ProxyPool>) -> 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<ProxyConfig> {
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<String>) {
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?;
+89 -28
View File
@@ -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<GalleryPhoto>) -> 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<Profile> {
// 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<Profile> {
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?;
+9 -3
View File
@@ -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
+438 -82
View File
@@ -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<Full<Bytes>>;
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct TunnelKey {
proxy_host: String,
proxy_port: u16,
origin_host: String,
origin_port: u16,
}
struct TunnelPool {
idle: Mutex<HashMap<TunnelKey, VecDeque<(TunnelSender, StdInstant)>>>,
}
impl TunnelPool {
fn new() -> Arc<Self> {
Arc::new(Self {
idle: Mutex::new(HashMap::new()),
})
}
fn take(&self, key: &TunnelKey) -> Option<TunnelSender> {
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<String>,
pub struct ProxyConfig {
pub host: String,
pub port: u16,
pub kind: ProxyKind,
pub auth: Option<String>,
pub socks_user: Option<String>,
pub socks_pass: Option<String>,
}
/// Lightweight HTTP client wrapping hyper-util's connection-pooling client.
@@ -97,13 +172,14 @@ pub struct HttpClient {
default_headers: HeaderMap,
proxy: Option<ProxyConfig>,
tls: Arc<rustls::ClientConfig>,
tunnels: Arc<TunnelPool>,
}
/// 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<Response> {
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<Response> {
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<Response> {
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>(io: TokioIo<IO>) -> Result<TunnelSender>
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<Response> {
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<Response> {
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<Response> {
self
.send_on(Method::POST, uri, extra_headers, body, via)
.await
}
/// Send a HEAD request.
pub async fn head(&self, uri: &str) -> Result<Response> {
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)
}
}
+2
View File
@@ -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::*;
+303
View File
@@ -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<Endpoint>,
pins: BTreeMap<i64, u16>,
}
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<Option<Self>> {
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<Self> {
let valid: std::collections::HashSet<u16> =
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<i64, u16> {
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<i64, u16>) -> u16 {
let mut counts: BTreeMap<u16, usize> = 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<IpAddr> = 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<f64> {
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<f64> {
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<f64> {
let start = Instant::now();
timeout(Duration::from_secs(2), TcpStream::connect((ip, port)))
.await
.ok()?
.ok()?;
Some(start.elapsed().as_secs_f64() * 1000.0)
}
+18
View File
@@ -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:{}:{}",
+4 -1
View File
@@ -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,
};
+92 -11
View File
@@ -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<dyn Any + Send + Sync>,
expires: Instant,
value: Arc<dyn Any + Send + Sync>,
fresh_until: Instant,
stale_until: Instant,
}
/// A cache lookup that may still be served after its fresh TTL.
pub enum Hit<T> {
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<RwLock<HashMap<String, Entry>>>,
inflight: Arc<Mutex<HashSet<String>>>,
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<T>(&self, key: &str) -> Option<T>
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<T>(&self, key: &str) -> Option<Hit<T>>
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::<T>().cloned()
let value = entry.value.downcast_ref::<T>()?.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<T>(&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<RefreshGuard> {
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<T>(&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) {
+9
View File
@@ -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
+6
View File
@@ -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.
+6 -1
View File
@@ -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",
+6 -1
View File
@@ -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
+189 -29
View File
@@ -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<Semaphore> = LazyLock::new(|| Semaphore::new(3));
const PREFETCH_PROFILES: usize = 4;
#[derive(Clone)]
struct CachedRss {
body: String,
min_id: Option<i64>,
}
/// Fresh cache hit, or a stale value plus a single background refresh.
pub fn swr_take<T, F, Fut>(state: &AppState, key: &str, refresh: F) -> Option<T>
where
T: Any + Clone + Send + Sync,
F: FnOnce(AppState) -> Fut + Send + 'static,
Fut: Future<Output = ()> + 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<User> {
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::<String>(&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::<Profile>(&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<User> {
let cache_key = cache_keys::user(username);
if let Some(cached) = state.cache.get::<User>(&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::<AccountContext>(&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::<AccountContext>(&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<AccountContext> {
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::<AccountContext>(&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");
+183 -17
View File
@@ -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<String>,
#[serde(rename = "e-retweets")]
pub e_retweets: Option<String>,
/// 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<String>,
}
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::<Vec<_>>();
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<AppState> {
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(&params);
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::<PaginatedResult<User>, _, _>(&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(),
&params,
&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::<Timeline, _, _>(&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(),
&params,
&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()
}),
)
+5
View File
@@ -607,6 +607,11 @@ async fn user_by_id(State(state): State<AppState>, Path(id): Path<String>) -> 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)) => {
+75 -16
View File
@@ -52,6 +52,40 @@ use crate::{
},
};
async fn load_first_profile(state: &AppState, username: &str) -> Result<Profile> {
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::<Profile>(&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::<Timeline>(&cache_key) {
if let Some(cached) = helpers::swr_take::<Timeline, _, _>(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());
+2
View File
@@ -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,
+2 -2
View File
@@ -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