diff --git a/.env.example b/.env.example index 9340d5b..4160fa1 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,17 @@ RUNGU_SECURE_COOKIE=false # Set to * to allow all (dev only — NOT safe for production). RUNGU_CORS_ORIGINS= +# Rate limiting — max requests per minute per IP. 0 disables a limiter. +# API is broad (board browsing); auth is strict (OAuth login/callback abuse). +# Defaults: API 300/min, auth 30/min. +# RUNGU_RATE_LIMIT_PER_MIN=300 +# RUNGU_AUTH_RATE_LIMIT_PER_MIN=30 + +# Trust X-Forwarded-For for rate-limit client IPs. Enable ONLY behind a +# reverse proxy that overwrites the header (nginx/Caddy/Cloudflare). +# Default: false (use the socket address). +# RUNGU_TRUST_PROXY=false + # Admin emails — comma-separated. Users with these emails get admin role on login. # Example: ADMIN_EMAILS=admin@example.com,root@example.com ADMIN_EMAILS= diff --git a/CHANGELOG.md b/CHANGELOG.md index e370541..5e812cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,29 @@ ## [Unreleased] +### Added + +- **Rate limiting** (in-memory, per-IP fixed window): two independent + limiters — strict on `/auth/*` (default 30/min) and broader on `/api/*` + (default 300/min). Configurable via `RUNGU_AUTH_RATE_LIMIT_PER_MIN` / + `RUNGU_RATE_LIMIT_PER_MIN`; `0` disables. Previously deferred (#52) due + to an MSRV-blocked `governor` dependency — reimplemented without external + crates. Closes #43. + +### Changed + +- **`list_posts` now populates `user_voted`** for the requesting user via a + single batched lookup. Previously the board list always reported + `user_voted: false`, so it could not show which posts the current user had + voted on (only `GET /posts/{id}` set it). Replaces the latent N+1 with one + indexed query over the `votes` primary key. +- **PostgreSQL full-text search**: the Postgres backend now uses a generated + `tsvector` column + GIN index (`search_tsv @@ plainto_tsquery(?)`) instead + of an unindexed `LOWER(...) LIKE` full scan. SQLite still uses FTS5. +- **`get_project_by_slug` is cached** (5-minute TTL, in-memory `moka`) — it + was hit on nearly every board/vote/comment request. Create/update/delete + invalidate the cache. + ## [0.2.1] - 2026-06-27 ### Added diff --git a/Cargo.lock b/Cargo.lock index abaad20..a95bc7a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -97,6 +97,17 @@ dependencies = [ "derive_arbitrary", ] +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -412,6 +423,24 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.12" @@ -569,6 +598,16 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "fastrand" version = "2.4.1" @@ -661,6 +700,17 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -681,6 +731,7 @@ checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", @@ -1261,6 +1312,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "moka" +version = "0.12.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046" +dependencies = [ + "async-lock", + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "event-listener", + "futures-util", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + [[package]] name = "multer" version = "3.1.0" @@ -1474,6 +1545,12 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" +[[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + [[package]] name = "potential_utf" version = "0.1.5" @@ -1825,6 +1902,7 @@ dependencies = [ "anyhow", "async-trait", "chrono", + "moka", "rungu_proto", "serde", "serde_json", @@ -2516,6 +2594,12 @@ dependencies = [ "syn", ] +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "tempfile" version = "3.27.0" diff --git a/Cargo.toml b/Cargo.toml index 0bca29a..375e0ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -86,6 +86,9 @@ clap = { version = "4", features = ["derive"] } # Time chrono = { version = "0.4", features = ["serde"] } +# In-memory cache (project-by-slug lookup cache) +moka = { version = "0.12", features = ["future"] } + # UUID uuid = { version = "1", features = ["v4", "serde"] } diff --git a/crates/rungu-api/src/post_routes.rs b/crates/rungu-api/src/post_routes.rs index c4aaf0e..ad4aaa8 100644 --- a/crates/rungu-api/src/post_routes.rs +++ b/crates/rungu-api/src/post_routes.rs @@ -59,6 +59,7 @@ pub async fn list_posts( State(state): State, Path(slug): Path, Query(query): Query, + user: rungu_auth::OptionalCurrentUser, ) -> Result { let project = state.store.get_project_by_slug(&slug).await?.ok_or_else(|| ApiError::not_found("Project not found"))?; @@ -86,6 +87,7 @@ pub async fn list_posts( category, query: query.q.as_deref(), since: None, + user_id: user.user.as_ref().map(|cu| cu.id.as_str()), offset, limit: per_page, }; @@ -336,6 +338,7 @@ async fn fetch_bucket( category: None, query: None, since: None, + user_id: None, offset: 0, limit, }; @@ -407,6 +410,7 @@ pub async fn get_project_changelog( category: None, query: None, since, + user_id: None, offset, limit: per_page, }; diff --git a/crates/rungu-core/Cargo.toml b/crates/rungu-core/Cargo.toml index 312dfcf..1f16d04 100644 --- a/crates/rungu-core/Cargo.toml +++ b/crates/rungu-core/Cargo.toml @@ -18,6 +18,7 @@ tracing = { workspace = true } thiserror = { workspace = true } anyhow = { workspace = true } async-trait = "0.1" +moka = { workspace = true } [dev-dependencies] tokio = { workspace = true } diff --git a/crates/rungu-core/migrations/postgres/001_initial.sql b/crates/rungu-core/migrations/postgres/001_initial.sql index 149e36a..4291557 100644 --- a/crates/rungu-core/migrations/postgres/001_initial.sql +++ b/crates/rungu-core/migrations/postgres/001_initial.sql @@ -60,6 +60,18 @@ CREATE TABLE IF NOT EXISTS votes ( PRIMARY KEY (user_id, post_id) ); +-- Full-text search: generated tsvector + GIN index. +-- SQLite uses the FTS5 virtual table; on PostgreSQL we derive a `tsvector` +-- from title+description and index it with GIN so `@@ plainto_tsquery(?)` +-- is an indexed scan instead of a full-table `LIKE`. +-- Added via ALTER ... ADD COLUMN IF NOT EXISTS so the migration stays +-- idempotent (the whole file is re-run on every startup). +ALTER TABLE posts ADD COLUMN IF NOT EXISTS search_tsv tsvector + GENERATED ALWAYS AS ( + to_tsvector('english', coalesce(title, '') || ' ' || coalesce(description, '')) + ) STORED; +CREATE INDEX IF NOT EXISTS idx_posts_search_tsv ON posts USING GIN (search_tsv); + -- Comments (threaded) CREATE TABLE IF NOT EXISTS comments ( id TEXT PRIMARY KEY, diff --git a/crates/rungu-core/src/store.rs b/crates/rungu-core/src/store.rs index db7c323..d0fc62b 100644 --- a/crates/rungu-core/src/store.rs +++ b/crates/rungu-core/src/store.rs @@ -159,6 +159,20 @@ fn sanitize_fts_query(input: &str) -> String { .join(" ") } +/// Which full-text search path `list_posts` should use, if any. +/// +/// Kept as a small enum (rather than two `Option`s) so the JOIN, WHERE +/// fragment, and bind value all branch on a single decision. +enum Search { + /// SQLite FTS5: bind a sanitized prefix-token MATCH expression. + Fts5(String), + /// PostgreSQL: bind the raw query to `plainto_tsquery` against the + /// generated `search_tsv` column. + PgTsv(String), + /// No search filter applied. + None, +} + /// Parse PostStatus from SQLite TEXT column. fn parse_status(s: &str) -> PostStatus { match s { @@ -184,15 +198,28 @@ fn parse_category(s: &str) -> PostCategory { #[derive(Clone)] pub struct Store { pool: AnyPool, - /// True when the underlying database is SQLite. Drives FTS5 vs LIKE - /// search-path selection (only SQLite has the `posts_fts` virtual table - /// populated by triggers; PostgreSQL FTS is deferred to v0.3). + /// True when the underlying database is SQLite. Drives FTS5 vs PostgreSQL + /// `tsvector` search-path selection (SQLite has the `posts_fts` virtual + /// table populated by triggers; PostgreSQL uses a generated `tsvector` + /// column + GIN index). is_sqlite: bool, + /// Short-TTL cache for `get_project_by_slug`. Project slugs are resolved + /// on nearly every board/vote/comment request, but rarely change, so a + /// small cache eliminates most of those round-trips. Mutations to projects + /// (`create`/`update`/`delete`) invalidate the whole cache — projects are + /// few and the simplicity beats per-key tracking. + project_cache: moka::future::Cache, } +/// How long a cached `Project` row stays fresh before re-hitting the DB. +const PROJECT_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(5 * 60); +/// Upper bound on cached projects. Projects are few in practice; this just +/// caps memory if a caller enumerates many slugs (including misses). +const PROJECT_CACHE_MAX_ENTRIES: u64 = 256; + impl Store { pub fn new(pool: AnyPool) -> Self { - Self { pool, is_sqlite: false } + Self::build(pool, false) } /// Construct a Store with explicit backend knowledge. @@ -202,7 +229,15 @@ impl Store { /// URL starts with `sqlite:` — callers should pass that detection result /// here so the store can pick the right query dialect. pub fn new_with_kind(pool: AnyPool, is_sqlite: bool) -> Self { - Self { pool, is_sqlite } + Self::build(pool, is_sqlite) + } + + fn build(pool: AnyPool, is_sqlite: bool) -> Self { + let project_cache = moka::future::Cache::builder() + .time_to_live(PROJECT_CACHE_TTL) + .max_capacity(PROJECT_CACHE_MAX_ENTRIES) + .build(); + Self { pool, is_sqlite, project_cache } } /// Get a reference to the pool. @@ -222,13 +257,25 @@ impl Store { } /// Get a project by slug. + /// + /// Serves from a short-TTL in-memory cache; misses fall through to the DB + /// and populate the cache. Negative results (slug not found) are **not** + /// cached — a subsequent create with the same slug should be visible + /// immediately. pub async fn get_project_by_slug(&self, slug: &str) -> Result> { + if let Some(cached) = self.project_cache.get(slug).await { + return Ok(Some(cached)); + } let row = sqlx::query("SELECT id, slug, name, description, created_at FROM projects WHERE slug = ?") .bind(slug) .fetch_optional(&self.pool) .await .context("Failed to get project")?; - Ok(row.as_ref().map(map_project)) + let project = row.as_ref().map(map_project); + if let Some(ref p) = project { + self.project_cache.insert(slug.to_string(), p.clone()).await; + } + Ok(project) } /// Get a project by ID. @@ -254,6 +301,12 @@ impl Store { .execute(&self.pool) .await .context("Failed to create project")?; + + // Defensive cache invalidation: a negative (miss) result is never + // cached, so this is strictly unnecessary today — but keep the cache + // correct regardless of future caching-policy changes. + self.project_cache.invalidate_all(); + Ok(Project { id, slug: slug.to_string(), @@ -296,6 +349,11 @@ impl Store { .context("Failed to update project description")?; } + // Invalidate the project cache — name/description changed. We clear + // the whole cache rather than one key because `update_project` is + // called by id (the cached key is the slug). + self.project_cache.invalidate_all(); + // Fetch and return the updated row self.get_project_by_id(project_id).await?.context("Project disappeared after update") } @@ -309,6 +367,9 @@ impl Store { .execute(&self.pool) .await .context("Failed to delete project")?; + // Invalidate the project cache so a re-create with the same slug is + // visible immediately and a stale row isn't served post-delete. + self.project_cache.invalidate_all(); Ok(()) } @@ -320,22 +381,25 @@ impl Store { /// The search query uses `LIKE ?` with the pattern passed as a bind parameter. pub async fn list_posts(&self, params: ListPostsParams<'_>) -> Result<(Vec, i64)> { // Search path selection: - // - SQLite: use the FTS5 virtual table `posts_fts` (maintained by triggers) - // via `posts_fts MATCH ?`. Faster and relevance-ranked. - // - Other backends (PostgreSQL — v0.3): fall back to LIKE. PostgreSQL - // will get its own `tsvector` index in a later release. - // Sanitize once up front so both the WHERE-builder and the binder see - // the same decision (a token that's all punctuation would otherwise - // produce an empty MATCH query that errors in FTS5). - let (fts_term, search_pattern): (Option, Option) = if self.is_sqlite { - let sanitized = params.query.map(sanitize_fts_query).filter(|s| !s.is_empty()); - // SQLite: punctuation-only queries sanitize to "" → treat as no filter - // (no meaningful search term). The FTS path is used when a real token survives. - (sanitized, None) - } else { - (None, params.query.map(|q| format!("%{q}%"))) + // - SQLite: FTS5 virtual table `posts_fts` (maintained by triggers) via + // `posts_fts MATCH ?`. Faster and relevance-ranked. User input is + // sanitized into prefix tokens so FTS5 query syntax can't error out + // or do expensive work. + // - PostgreSQL: generated `search_tsv` tsvector column + GIN index via + // `p.search_tsv @@ plainto_tsquery(?)`. `plainto_tsquery` already + // rejects unsafe syntax, so we bind the raw (trimmed) query. + // - Empty / punctuation-only input → no search filter. + let search = match (self.is_sqlite, params.query) { + (true, Some(q)) => { + let sanitized = sanitize_fts_query(q); + if sanitized.is_empty() { Search::None } else { Search::Fts5(sanitized) } + } + (false, Some(q)) => { + let trimmed = q.trim(); + if trimmed.is_empty() { Search::None } else { Search::PgTsv(trimmed.to_string()) } + } + _ => Search::None, }; - let use_fts = fts_term.is_some(); // Build WHERE clause fragments — only add conditions for filters that are present. // Each `?` is a positional placeholder bound later via .bind(). @@ -347,10 +411,6 @@ impl Store { if params.category.is_some() { conditions.push("p.category = ?".to_string()); } - if !use_fts && search_pattern.is_some() { - // LIKE fallback for non-SQLite backends. - conditions.push("(LOWER(p.title) LIKE LOWER(?) OR LOWER(p.description) LIKE LOWER(?))".to_string()); - } if params.since.is_some() { // Incremental-pull lower bound on `updated_at` (used by changelog). conditions.push("p.updated_at >= ?".to_string()); @@ -369,24 +429,23 @@ impl Store { PostSort::RecentlyUpdated => "p.updated_at DESC", }; - // FTS JOIN clause — only emitted when the FTS path is active. - let fts_join = if use_fts { "JOIN posts_fts ON posts_fts.rowid = p.rowid" } else { "" }; - // Extra WHERE fragment for the FTS path. We bind a sanitized MATCH term. - let fts_where = if use_fts { " AND posts_fts MATCH ?" } else { "" }; + // Search JOIN + WHERE fragment, emitted only when a search path is active. + // SQLite joins the FTS5 table; PostgreSQL filters on its generated tsvector. + let (search_join, search_where): (&str, &str) = match &search { + Search::Fts5(_) => ("JOIN posts_fts ON posts_fts.rowid = p.rowid", " AND posts_fts MATCH ?"), + Search::PgTsv(_) => ("", " AND p.search_tsv @@ plainto_tsquery(?)"), + Search::None => ("", ""), + }; - // Helper: bind optional filter values in order (project_id already first) + // Helper: bind optional filter values in order (project_id is first). macro_rules! bind_filters { ($query:expr) => {{ let q = $query.bind(params.project_id); let q = if let Some(ref s) = params.status { q.bind(status_to_str(*s)) } else { q }; let q = if let Some(ref c) = params.category { q.bind(category_to_str(*c)) } else { q }; - let q = if use_fts { - // Bind the FTS5 MATCH term — sanitized for FTS syntax. - q.bind(fts_term.clone().unwrap_or_default()) - } else if let Some(ref pat) = search_pattern { - q.bind(pat.clone()).bind(pat.clone()) - } else { - q + let q = match &search { + Search::Fts5(t) | Search::PgTsv(t) => q.bind(t.clone()), + Search::None => q, }; if let Some(ts) = params.since { // Bind the `updated_at >= ?` lower bound as an RFC3339 string. @@ -398,7 +457,7 @@ impl Store { } // Count query (same WHERE, no LIMIT/OFFSET) - let count_sql = format!("SELECT COUNT(*) FROM posts p {fts_join} WHERE {where_sql}{fts_where}"); + let count_sql = format!("SELECT COUNT(*) FROM posts p {search_join} WHERE {where_sql}{search_where}"); let total: i64 = bind_filters!(sqlx::query_scalar::<_, i64>(&count_sql)) .fetch_one(&self.pool) .await @@ -408,9 +467,9 @@ impl Store { let sql = format!( "SELECT p.*, u.id as user_id, u.email as user_email, u.name as user_name, u.avatar_url as user_avatar \ FROM posts p \ - {fts_join} + {search_join} LEFT JOIN users u ON p.created_by = u.id \ - WHERE {where_sql}{fts_where} \ + WHERE {where_sql}{search_where} \ ORDER BY {order} \ LIMIT ? OFFSET ?" ); @@ -418,8 +477,29 @@ impl Store { let query = bind_filters!(sqlx::query(&sql)).bind(params.limit).bind(params.offset); let rows = query.fetch_all(&self.pool).await.context("Failed to list posts")?; + let mut posts: Vec = rows.iter().map(map_post_detail).collect(); + + // Batch-populate `user_voted` for the requesting user. Previously this + // was always `false` in list views (only `get_post` set it per row), + // which left the board unable to show which posts the current user had + // voted on. A naive per-row lookup would be N+1; instead we do one + // indexed query over the `votes` (user_id, post_id) primary key. + if let Some(uid) = params.user_id { + if !posts.is_empty() { + let placeholders = std::iter::repeat_n("?", posts.len()).collect::>().join(","); + let sql = format!("SELECT post_id FROM votes WHERE user_id = ? AND post_id IN ({placeholders})"); + let mut q = sqlx::query_scalar::<_, String>(&sql).bind(uid); + for p in &posts { + q = q.bind(&p.post.id); + } + let voted_ids: std::collections::HashSet = + q.fetch_all(&self.pool).await.context("Failed to look up votes")?.into_iter().collect(); + for p in posts.iter_mut() { + p.user_voted = voted_ids.contains(&p.post.id); + } + } + } - let posts = rows.iter().map(map_post_detail).collect(); Ok((posts, total)) } diff --git a/crates/rungu-core/tests/store_test.rs b/crates/rungu-core/tests/store_test.rs index a4f694e..4e12c27 100644 --- a/crates/rungu-core/tests/store_test.rs +++ b/crates/rungu-core/tests/store_test.rs @@ -112,6 +112,7 @@ async fn test_list_posts_with_filters() { category: None, query: None, since: None, + user_id: None, offset: 0, limit: 20, }) @@ -129,6 +130,7 @@ async fn test_list_posts_with_filters() { category: Some(PostCategory::Bug), query: None, since: None, + user_id: None, offset: 0, limit: 20, }) @@ -146,6 +148,7 @@ async fn test_list_posts_with_filters() { category: None, query: None, since: None, + user_id: None, offset: 0, limit: 20, }) @@ -163,6 +166,7 @@ async fn test_list_posts_with_filters() { category: None, query: Some("dark"), since: None, + user_id: None, offset: 0, limit: 20, }) @@ -191,6 +195,7 @@ async fn test_list_posts_search_with_sql_injection_chars() { category: None, query: Some("'; DROP TABLE posts; --"), since: None, + user_id: None, offset: 0, limit: 20, }) @@ -208,6 +213,7 @@ async fn test_list_posts_search_with_sql_injection_chars() { category: None, query: None, since: None, + user_id: None, offset: 0, limit: 20, }) @@ -257,6 +263,7 @@ async fn test_list_posts_search_fts5_multitoken() { category: None, query: Some("dark mode"), since: None, + user_id: None, offset: 0, limit: 20, }) @@ -285,6 +292,7 @@ async fn test_list_posts_search_punctuation_only_drops_to_noop() { category: None, query: Some("!!!"), since: None, + user_id: None, offset: 0, limit: 20, }) @@ -315,6 +323,7 @@ async fn test_list_posts_pagination() { category: None, query: None, since: None, + user_id: None, offset: 0, limit: 2, }) @@ -332,6 +341,7 @@ async fn test_list_posts_pagination() { category: None, query: None, since: None, + user_id: None, offset: 4, limit: 2, }) @@ -457,3 +467,89 @@ async fn test_get_current_user() { // Non-existent user assert!(store.get_current_user("nonexistent").await.is_err()); } + +// ── user_voted batch population (list_posts) ─────────────────────────── + +#[tokio::test] +async fn test_list_posts_populates_user_voted_batch() { + // Regression: `list_posts` previously left `user_voted` always `false`. + // It should now batch-populate it via a single indexed lookup. + let store = setup().await; + let project = store.create_project("App", "app", "").await.unwrap(); + let alice = store.find_or_create_user("alice@t.com", None, None, &[]).await.unwrap(); + let bob = store.find_or_create_user("bob@t.com", None, None, &[]).await.unwrap(); + + let p1 = store.create_post(&project.id, "Post 1", "", PostCategory::Feedback, &alice.id).await.unwrap(); + let p2 = store.create_post(&project.id, "Post 2", "", PostCategory::Feature, &bob.id).await.unwrap(); + + // Alice votes on p1 only. + store.toggle_vote(&alice.id, &p1.id).await.unwrap(); + + let (posts, _total) = store + .list_posts(ListPostsParams { + project_id: &project.id, + sort: PostSort::Newest, + status: None, + category: None, + query: None, + since: None, + user_id: Some(&alice.id), + offset: 0, + limit: 20, + }) + .await + .unwrap(); + + let by_id: std::collections::HashMap<&str, bool> = + posts.iter().map(|pd| (pd.post.id.as_str(), pd.user_voted)).collect(); + assert_eq!(by_id.get(p1.id.as_str()), Some(&true), "alice voted on p1"); + assert_eq!(by_id.get(p2.id.as_str()), Some(&false), "alice did not vote on p2"); +} + +#[tokio::test] +async fn test_list_posts_user_voted_none_when_anonymous() { + // Without `user_id`, every entry must keep `user_voted == false`. + let store = setup().await; + let project = store.create_project("App", "app", "").await.unwrap(); + let user = store.find_or_create_user("u@t.com", None, None, &[]).await.unwrap(); + let post = store.create_post(&project.id, "P", "", PostCategory::Feedback, &user.id).await.unwrap(); + store.toggle_vote(&user.id, &post.id).await.unwrap(); + + let (posts, _total) = store + .list_posts(ListPostsParams { + project_id: &project.id, + sort: PostSort::Newest, + status: None, + category: None, + query: None, + since: None, + user_id: None, + offset: 0, + limit: 20, + }) + .await + .unwrap(); + assert!(posts.iter().all(|p| !p.user_voted)); +} + +// ── project-by-slug cache ────────────────────────────────────────────── + +#[tokio::test] +async fn test_project_cache_invalidated_on_update_and_delete() { + // The cache must not serve stale rows after a project is renamed or deleted. + let store = setup().await; + let project = store.create_project("Original", "slug-a", "").await.unwrap(); + + // Prime the cache. + let _ = store.get_project_by_slug("slug-a").await.unwrap().unwrap(); + + // Rename → description change goes through update_project, which clears + // the cache; the next read must reflect the new value. + store.update_project(&project.id, Some("Renamed"), None).await.unwrap(); + let after = store.get_project_by_slug("slug-a").await.unwrap().unwrap(); + assert_eq!(after.name, "Renamed"); + + // Delete → cached reads must observe the deletion immediately. + store.delete_project(&project.id).await.unwrap(); + assert!(store.get_project_by_slug("slug-a").await.unwrap().is_none()); +} diff --git a/crates/rungu-mcp/src/lib.rs b/crates/rungu-mcp/src/lib.rs index 68a521c..c2a745a 100644 --- a/crates/rungu-mcp/src/lib.rs +++ b/crates/rungu-mcp/src/lib.rs @@ -186,6 +186,7 @@ async fn list_posts(params: &Value, store: &Store) -> Result { category, query, since: None, + user_id: None, offset: 0, limit, }) @@ -276,6 +277,7 @@ async fn search_posts(params: &Value, store: &Store) -> Result { category: None, query: Some(query), since: None, + user_id: None, offset: 0, limit, }) @@ -306,6 +308,7 @@ async fn get_changelog(params: &Value, store: &Store) -> Result { category: None, query: None, since: None, + user_id: None, offset: 0, limit, }) @@ -356,6 +359,7 @@ async fn get_stats(params: &Value, store: &Store) -> Result { category: None, query: None, since: None, + user_id: None, offset: 0, limit: 1000, }) @@ -410,6 +414,7 @@ async fn get_trending(params: &Value, store: &Store) -> Result { category: None, query: None, since: None, + user_id: None, offset: 0, limit, }) diff --git a/crates/rungu-proto/src/lib.rs b/crates/rungu-proto/src/lib.rs index 14614c7..30490e0 100644 --- a/crates/rungu-proto/src/lib.rs +++ b/crates/rungu-proto/src/lib.rs @@ -258,6 +258,10 @@ pub struct ListPostsParams<'a> { /// endpoint for incremental pulls ("what shipped since my last sync?"). /// When `None`, no `updated_at` filter is applied. pub since: Option>, + /// When set, each returned [`PostDetail::user_voted`] is populated for this + /// user via a single batched lookup (avoids N+1). When `None`, all entries + /// are left `false` (anonymous / public endpoints). + pub user_id: Option<&'a str>, pub offset: i64, pub limit: i64, } diff --git a/crates/rungud/src/config.rs b/crates/rungud/src/config.rs index 7937d8a..745bc77 100644 --- a/crates/rungud/src/config.rs +++ b/crates/rungud/src/config.rs @@ -11,6 +11,18 @@ pub struct Config { pub cors_origins: Vec, /// Sentry DSN. None = Sentry disabled (no events sent). pub sentry_dsn: Option, + /// Max requests per minute per IP on `/api/*` routes. `0` disables the + /// API rate limiter entirely. + pub rate_limit_per_min: u32, + /// Max requests per minute per IP on `/auth/*` routes (OAuth + /// login/callback). Stricter than the API limiter to blunt credential / + /// OAuth abuse. `0` disables it. + pub auth_rate_limit_per_min: u32, + /// Honor `X-Forwarded-For` when resolving client IPs for rate limiting. + /// Default `false` (use the socket address) so a directly-exposed Rungu + /// can't be spoofed. Enable only behind a trusted reverse proxy that + /// **overwrites** the header. + pub trust_proxy: bool, } impl Config { @@ -32,6 +44,9 @@ impl Config { auth: AuthConfig::from_env(), cors_origins, sentry_dsn: std::env::var("SENTRY_DSN").ok(), + rate_limit_per_min: parse_per_min("RUNGU_RATE_LIMIT_PER_MIN", 300), + auth_rate_limit_per_min: parse_per_min("RUNGU_AUTH_RATE_LIMIT_PER_MIN", 30), + trust_proxy: parse_bool("RUNGU_TRUST_PROXY", false), } } @@ -60,3 +75,19 @@ impl Config { Ok(()) } } + +/// Parse a `requests per minute` env var into a `u32`. Falls back to +/// `default` when unset or unparseable. `0` is a valid value meaning +/// "disable this limiter". +fn parse_per_min(var: &str, default: u32) -> u32 { + std::env::var(var).ok().and_then(|s| s.parse().ok()).unwrap_or(default) +} + +/// Parse a boolean env var. Accepts true/1/yes/on (case-insensitive); +/// everything else (including unset) falls back to `default`. +fn parse_bool(var: &str, default: bool) -> bool { + match std::env::var(var).ok().and_then(|s| s.to_lowercase().parse().ok()) { + Some(v) => v, + None => default, + } +} diff --git a/crates/rungud/src/main.rs b/crates/rungud/src/main.rs index 0bfcd1e..8a16a0b 100644 --- a/crates/rungud/src/main.rs +++ b/crates/rungud/src/main.rs @@ -3,6 +3,7 @@ //! Main binary: CLI subcommands + HTTP server. pub mod config; +pub mod ratelimit; pub mod server; pub mod spa; diff --git a/crates/rungud/src/ratelimit.rs b/crates/rungud/src/ratelimit.rs new file mode 100644 index 0000000..740359e --- /dev/null +++ b/crates/rungud/src/ratelimit.rs @@ -0,0 +1,187 @@ +//! In-memory per-IP rate limiting (fixed-window counter). +//! +//! No external crate — `governor`/`axum-governor` were dropped due to an MSRV +//! constraint (see `Cargo.toml`). State is shared via `Arc` behind a +//! `tokio::sync::Mutex`; a background task periodically clears buckets so the +//! map can't grow unbounded under spoofed-IP flooding. +//! +//! Two independent limiters are wired up in [`crate::server`]: +//! - **auth routes** (`/auth/*`) — stricter, defends OAuth login/callback abuse +//! - **API routes** (`/api/*`) — looser, protects general flooding +//! +//! Set the corresponding `_PER_MIN` env var to `0` to disable a limiter. +//! +//! ## Trust boundary +//! +//! Client IP defaults to the connected socket address. `X-Forwarded-For` +//! (first hop) is honored only when `RUNGU_TRUST_PROXY=true` — enable that +//! solely behind a reverse proxy that *overwrites* the header, since XFF is +//! otherwise client-controlled and trivially spoofed. + +use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use axum::body::Body; +use axum::extract::{ConnectInfo, State}; +use axum::http::{HeaderMap, Request, StatusCode}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use tokio::sync::Mutex; + +/// Per-IP request counter for one window. +#[derive(Copy, Clone)] +struct Bucket { + window_start: Instant, + count: u32, +} + +/// Shared rate-limiter state. Cloning is cheap (inner is `Arc`). +#[derive(Clone)] +pub struct RateLimiter { + inner: Arc>>, + max_requests: u32, + window: Duration, + /// When `false` (default), ignore `X-Forwarded-For` and always use the + /// connected socket address. Enable only when Rungu sits behind a + /// trusted reverse proxy that **overwrites** XFF — otherwise a client + /// can spoof the header to dodge the limit. + trust_proxy: bool, +} + +impl RateLimiter { + /// Build a limiter that allows `max_requests` per `window` per IP. + /// `trust_proxy` toggles honoring `X-Forwarded-For` (see field docs). + pub fn new(max_requests: u32, window: Duration, trust_proxy: bool) -> Self { + Self { inner: Arc::new(Mutex::new(HashMap::new())), max_requests, window, trust_proxy } + } + + /// Returns `Ok(remaining)` when allowed, `Err(retry_after_secs)` when the + /// IP is over the limit for the remainder of the current window. + pub async fn check(&self, ip: IpAddr) -> Result { + let now = Instant::now(); + let mut map = self.inner.lock().await; + let entry = map.entry(ip).or_insert(Bucket { window_start: now, count: 0 }); + + // Reset the bucket once the window has fully elapsed. + if now.duration_since(entry.window_start) >= self.window { + entry.window_start = now; + entry.count = 0; + } + + if entry.count >= self.max_requests { + let elapsed = now.duration_since(entry.window_start); + let remaining = self.window.saturating_sub(elapsed); + Err(remaining.as_secs().max(1)) + } else { + entry.count += 1; + Ok(self.max_requests - entry.count) + } + } + + /// Spawn a background task that evicts buckets whose window has elapsed, + /// so IPs that sent a few requests (or none) don't accumulate forever. + /// + /// Eviction is **per-bucket** (not a global clear): a global clear would + /// reset every IP's window on the same cadence and hand a synchronized + /// burst budget to coordinated clients. Evicting only expired buckets + /// preserves each IP's independent fixed window. + pub fn spawn_pruner(&self) { + let window = self.window; + let inner = self.inner.clone(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(window); + loop { + ticker.tick().await; + let now = Instant::now(); + let mut map = inner.lock().await; + map.retain(|_, bucket| now.duration_since(bucket.window_start) < window); + } + }); + } +} + +/// Parse the first hop of an `X-Forwarded-For` header into an `IpAddr`. +fn xff_ip(headers: &HeaderMap) -> Option { + let xff = headers.get("x-forwarded-for")?.to_str().ok()?; + let first = xff.split(',').next()?.trim(); + first.parse::().ok() +} + +/// Axum middleware entry point. Rejects with `429 Too Many Requests` +/// (and a `Retry-After` header) once an IP exceeds the limiter's budget. +/// +/// Expects the app to be served with +/// [`axum::serve`](axum::serve) via +/// `into_make_service_with_connect_info::()` so that +/// [`ConnectInfo`] is available. +pub async fn rate_limit_middleware( + State(limiter): State, + ConnectInfo(addr): ConnectInfo, + headers: HeaderMap, + req: Request, + next: Next, +) -> Response { + let ip = if limiter.trust_proxy { xff_ip(&headers).unwrap_or_else(|| addr.ip()) } else { addr.ip() }; + + match limiter.check(ip).await { + Ok(remaining) => { + let mut resp = next.run(req).await; + if let Ok(v) = remaining.to_string().parse() { + resp.headers_mut().insert("x-ratelimit-remaining", v); + } + resp + } + Err(retry_after) => { + let mut resp = (StatusCode::TOO_MANY_REQUESTS, "Too Many Requests").into_response(); + if let Ok(v) = retry_after.to_string().parse() { + resp.headers_mut().insert("retry-after", v); + } + resp + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ip(s: &str) -> IpAddr { + s.parse().unwrap() + } + + fn rl(max: u32) -> RateLimiter { + RateLimiter::new(max, Duration::from_secs(60), false) + } + + #[tokio::test] + async fn allows_up_to_limit_then_blocks() { + let rl = rl(3); + assert!(rl.check(ip("10.0.0.1")).await.is_ok()); + assert!(rl.check(ip("10.0.0.1")).await.is_ok()); + assert!(rl.check(ip("10.0.0.1")).await.is_ok()); + // 4th within the same window → blocked, retry-after reported. + let err = rl.check(ip("10.0.0.1")).await.unwrap_err(); + assert!(err >= 1); + } + + #[tokio::test] + async fn tracks_ips_independently() { + let rl = rl(1); + assert!(rl.check(ip("10.0.0.1")).await.is_ok()); + assert!(rl.check(ip("10.0.0.2")).await.is_ok()); + assert!(rl.check(ip("10.0.0.1")).await.is_err()); + assert!(rl.check(ip("10.0.0.2")).await.is_err()); + } + + #[tokio::test] + async fn resets_after_window_elapses() { + let rl = RateLimiter::new(1, Duration::from_millis(20), false); + assert!(rl.check(ip("10.0.0.1")).await.is_ok()); + assert!(rl.check(ip("10.0.0.1")).await.is_err()); + tokio::time::sleep(Duration::from_millis(30)).await; + // Window has elapsed → counter resets. + assert!(rl.check(ip("10.0.0.1")).await.is_ok()); + } +} diff --git a/crates/rungud/src/server.rs b/crates/rungud/src/server.rs index 9aee0ed..a4050a8 100644 --- a/crates/rungud/src/server.rs +++ b/crates/rungud/src/server.rs @@ -1,11 +1,12 @@ //! HTTP server — Axum router, API routes, SPA handler, Swagger UI. +use axum::middleware::from_fn_with_state; use axum::{Router, routing::get}; use rungu_api::AppState; use rungu_api::openapi::ApiDoc; use rungu_api::{api_routes, auth_routes}; -// Rate limiting removed — external crates require rustc >1.88. -// Will implement simple in-memory limiter in a future release. +// In-memory rate limiting — no external crate (governor was MSRV-incompatible). +use crate::ratelimit::{RateLimiter, rate_limit_middleware}; use tower_http::cors::Any; use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; @@ -13,6 +14,9 @@ use tracing::info; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; +use std::net::SocketAddr; +use std::time::Duration; + use crate::config::Config; use crate::spa::spa_handler; @@ -57,9 +61,34 @@ pub async fn serve(config: Config, pool: sqlx::AnyPool, is_sqlite: bool, listen: .allow_headers(Any) }; + // ── Rate limiters ─────────────────────────────────────────────────── + // Two independent per-IP limiters: a strict one for `/auth/*` (OAuth + // login/callback abuse) and a looser one for `/api/*`. Both are `0`-able + // via env to disable. Each limiter also spawns a pruner task so the IP + // map can't grow unbounded. + let api_routes = if config.rate_limit_per_min > 0 { + let limiter = RateLimiter::new(config.rate_limit_per_min, Duration::from_secs(60), config.trust_proxy); + limiter.spawn_pruner(); + tracing::info!("API rate limit: {} req/min per IP", config.rate_limit_per_min); + api_routes().layer(from_fn_with_state(limiter, rate_limit_middleware)) + } else { + tracing::info!("API rate limit: disabled (RUNGU_RATE_LIMIT_PER_MIN=0)"); + api_routes() + }; + + let auth_routes = if config.auth_rate_limit_per_min > 0 { + let limiter = RateLimiter::new(config.auth_rate_limit_per_min, Duration::from_secs(60), config.trust_proxy); + limiter.spawn_pruner(); + tracing::info!("Auth rate limit: {} req/min per IP", config.auth_rate_limit_per_min); + auth_routes().layer(from_fn_with_state(limiter, rate_limit_middleware)) + } else { + tracing::info!("Auth rate limit: disabled (RUNGU_AUTH_RATE_LIMIT_PER_MIN=0)"); + auth_routes() + }; + let app = Router::new() - .nest("/api", api_routes()) - .merge(auth_routes()) + .nest("/api", api_routes) + .merge(auth_routes) .merge(SwaggerUi::new("/swagger-ui").url("/api-docs/openapi.json", ApiDoc::openapi())) .route("/health", get(health_check)) .fallback(spa_handler) @@ -74,7 +103,9 @@ pub async fn serve(config: Config, pool: sqlx::AnyPool, is_sqlite: bool, listen: info!("Rungu listening on {listen}"); info!("Swagger UI: http://{listen}/swagger-ui"); info!("OpenAPI spec: http://{listen}/api-docs/openapi.json"); - axum::serve(listener, app).await?; + // `into_make_service_with_connect_info` exposes the peer `SocketAddr` to + // the rate-limit middleware via `ConnectInfo`. + axum::serve(listener, app.into_make_service_with_connect_info::()).await?; Ok(()) } diff --git a/docs/configuration.md b/docs/configuration.md index d35532b..71f4f62 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -10,6 +10,9 @@ All configuration is done via environment variables. | `RUNGU_DB` | `rungu.db` | SQLite database path | | `DATABASE_URL` | _(unset)_ | Override the database connection. When set, takes precedence over `RUNGU_DB`. Format: `sqlite:path.db` or `postgres://user:pass@host/db`. | | `RUNGU_CORS_ORIGINS` | _(APP_URL only)_ | Comma-separated CORS origins. Default: only `APP_URL`. Set to `*` to allow all (dev only). | +| `RUNGU_RATE_LIMIT_PER_MIN` | `300` | Max `/api/*` requests per minute per client IP (fixed window). `0` disables the limiter. Client IP is taken from `X-Forwarded-For` (first hop) when present, else the socket address. | +| `RUNGU_AUTH_RATE_LIMIT_PER_MIN` | `30` | Max `/auth/*` requests per minute per client IP. Stricter than the API limiter to blunt OAuth/login abuse. `0` disables it. | +| `RUNGU_TRUST_PROXY` | `false` | Honor `X-Forwarded-For` when resolving rate-limit client IPs. Enable only behind a trusted reverse proxy that overwrites the header; otherwise clients can spoof it. When `false`, the socket address is used. | | `RUNGU_SECURE_COOKIE` | `true` | Set `false` for HTTP (no Secure flag on cookies). Accepts (case-insensitive): `true\|1\|yes\|on`, `false\|0\|no\|off`. Any other value exits with a fatal error — see [Security](#security). | | `RUST_LOG` | `rungu=info` | Log level (trace, debug, info, warn, error). Supports `tracing_subscriber`'s [`EnvFilter`](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html) syntax. |