use std::time::Duration; use axum::Json; use axum::extract::{Query, State}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::auth::middleware::AuthUser; use crate::error::AppError; use crate::services::config; use crate::state::AppState; #[derive(Deserialize)] pub struct FeedQuery { pub cursor: Option, pub limit: Option, pub hashtag: Option, } #[derive(Serialize)] pub struct FeedUpload { pub id: Uuid, pub user_id: Uuid, pub uploader_name: String, pub preview_url: Option, pub thumbnail_url: Option, /// Big-screen (~2048px) variant for the diashow. Absent until the derivative exists. pub display_url: Option, pub mime_type: String, pub caption: Option, pub like_count: i64, pub comment_count: i64, pub liked_by_me: bool, pub created_at: DateTime, } #[derive(Serialize)] pub struct FeedResponse { pub uploads: Vec, pub next_cursor: Option, } #[derive(sqlx::FromRow)] struct FeedRow { id: Uuid, user_id: Uuid, uploader_name: String, preview_path: Option, thumbnail_path: Option, display_path: Option, mime_type: String, caption: Option, like_count: i64, comment_count: i64, created_at: DateTime, } pub async fn feed( State(state): State, auth: AuthUser, Query(q): Query, ) -> Result, AppError> { let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; let feed_rate_on = config::get_bool(&state.config_cache, "feed_rate_enabled", true).await; if rate_limits_on && feed_rate_on { let rate_limit = config::get_usize(&state.config_cache, "feed_rate_per_min", 60).await; // Keyed per-user, exactly like `feed_delta` below: at a venue every guest shares // one public IP, so an IP key gave the whole party a single 60/min bucket and the // fastest scroller starved everyone else. if let Err(retry_after_secs) = state.rate_limiter.check_with_retry( format!("feed:{}", auth.user_id), rate_limit, Duration::from_secs(60), ) { return Err(AppError::TooManyRequests( "Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(), Some(retry_after_secs), )); } } let limit = q.limit.unwrap_or(20).min(100); // Resolve the cursor to a (created_at, id) position. The pair is compared as a // tuple so ties on created_at break on id — keyset pagination on created_at // alone would silently drop rows sharing a timestamp across a page boundary. let (cursor_time, cursor_id) = match q.cursor { Some(c) => match get_cursor_pos(&state.pool, c).await { Some((t, id)) => (Some(t), Some(id)), None => (None, None), }, None => (None, None), }; let rows = if let Some(hashtag) = &q.hashtag { let tag = hashtag.trim().trim_start_matches('#').to_lowercase(); sqlx::query_as::<_, FeedRow>( "SELECT v.id, v.user_id, v.uploader_name, v.preview_path, v.thumbnail_path, v.display_path, v.mime_type, v.caption, v.like_count, v.comment_count, v.created_at FROM v_feed v JOIN upload_hashtag uh ON uh.upload_id = v.id JOIN hashtag h ON h.id = uh.hashtag_id AND h.tag = $1 WHERE v.event_id = $2 AND ($3::timestamptz IS NULL OR (v.created_at, v.id) < ($3, $4)) ORDER BY v.created_at DESC, v.id DESC LIMIT $5", ) .bind(&tag) .bind(auth.event_id) .bind(cursor_time) .bind(cursor_id) .bind(limit + 1) .fetch_all(&state.pool) .await? } else { sqlx::query_as::<_, FeedRow>( "SELECT id, user_id, uploader_name, preview_path, thumbnail_path, display_path, mime_type, caption, like_count, comment_count, created_at FROM v_feed WHERE event_id = $1 AND ($2::timestamptz IS NULL OR (created_at, id) < ($2, $3)) ORDER BY created_at DESC, id DESC LIMIT $4", ) .bind(auth.event_id) .bind(cursor_time) .bind(cursor_id) .bind(limit + 1) .fetch_all(&state.pool) .await? }; let has_more = rows.len() as i64 > limit; let rows: Vec = rows.into_iter().take(limit as usize).collect(); let next_cursor = if has_more { rows.last().map(|r| r.id) } else { None }; // Batch check which uploads the current user has liked let upload_ids: Vec = rows.iter().map(|r| r.id).collect(); let liked_set = get_liked_set(&state.pool, auth.user_id, &upload_ids).await; let uploads = rows .into_iter() .map(|r| { // Gated media aliases (visibility-checked, direct /media blocked). Emit the // URL only when the variant actually exists — the URL is what signals the // client which variant to load. let preview_url = r .preview_path .as_ref() .map(|_| format!("/api/v1/upload/{}/preview", r.id)); let thumbnail_url = r .thumbnail_path .as_ref() .map(|_| format!("/api/v1/upload/{}/thumbnail", r.id)); let display_url = r .display_path .as_ref() .map(|_| format!("/api/v1/upload/{}/display", r.id)); FeedUpload { liked_by_me: liked_set.contains(&r.id), id: r.id, user_id: r.user_id, uploader_name: r.uploader_name, preview_url, thumbnail_url, display_url, mime_type: r.mime_type, caption: r.caption, like_count: r.like_count, comment_count: r.comment_count, created_at: r.created_at, } }) .collect(); Ok(Json(FeedResponse { uploads, next_cursor, })) } #[derive(Deserialize)] pub struct DeltaQuery { pub since: DateTime, } #[derive(Serialize)] pub struct DeltaResponse { pub uploads: Vec, pub deleted_ids: Vec, /// Users whose uploads became hidden (banned / uploads_hidden) since `since`. A ban is /// not a soft-delete, so it never appears in `deleted_ids`; without this, a client that /// missed the ephemeral `user-hidden` SSE (a reconnecting projector, most acutely) would /// keep displaying the banned user's already-loaded slides. The client evicts every /// upload from these users on receipt. pub hidden_user_ids: Vec, /// True when the upload query hit `DELTA_LIMIT`: the response carries only the /// newest slice of the gap, so the client must fall back to a full feed refresh /// rather than merging (the older missed uploads are absent and unrecoverable /// via a later delta, which advances `since` past them). pub truncated: bool, /// The server's clock at the moment this delta was computed. The client advances its /// reconnect cursor from THIS, never `new Date()` — a browser clock even seconds fast /// would otherwise silently skip uploads whose server `created_at` falls in the skew /// window (they'd never reappear without a hard refresh). pub server_time: DateTime, } pub async fn feed_delta( State(state): State, auth: AuthUser, Query(q): Query, ) -> Result, AppError> { // Rate-limit the delta the same way as the paginated feed. Without this, ~100 clients // reconnecting at once (post-outage, or a flapping network) each fire an unbounded // delta fetch — a reconnect stampede. Keyed per-user so one client can't starve others // behind a shared NAT. let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await; let feed_rate_on = config::get_bool(&state.config_cache, "feed_rate_enabled", true).await; if rate_limits_on && feed_rate_on { let rate_limit = config::get_usize(&state.config_cache, "feed_rate_per_min", 60).await; if let Err(retry_after_secs) = state.rate_limiter.check_with_retry( format!("feed_delta:{}", auth.user_id), rate_limit, Duration::from_secs(60), ) { return Err(AppError::TooManyRequests( "Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(), Some(retry_after_secs), )); } } // Anchor the next cursor to the DB clock, captured *before* the queries so an upload // committed during this handler is re-fetched next time rather than skipped (a // duplicate id merges idempotently on the client; a miss is unrecoverable). let server_time: DateTime = sqlx::query_scalar("SELECT NOW()") .fetch_one(&state.pool) .await?; // Bounded like the paginated feed: a stale `since` could otherwise pull the // entire event's uploads in one response. If a client hits the cap it should // fall back to a full feed refresh rather than another delta. const DELTA_LIMIT: i64 = 200; // `>= since` (not `>`): `created_at` is not unique, so a strict `>` anchored to the exact // timestamp of the last-seen upload silently drops a SECOND upload committed in the same // microsecond — an unrecoverable live-update miss. `>=` re-includes the boundary rows; // the client merges by id (see the feed-delta handler's `seen` set), so the duplicate is // harmless while the tied upload is no longer lost. The cursor still advances to the // response's `server_time`, so this doesn't re-fetch on every subsequent delta. let rows = sqlx::query_as::<_, FeedRow>( "SELECT id, user_id, uploader_name, preview_path, thumbnail_path, display_path, mime_type, caption, like_count, comment_count, created_at FROM v_feed WHERE event_id = $1 AND created_at >= $2 ORDER BY created_at DESC, id DESC LIMIT $3", ) .bind(auth.event_id) .bind(q.since) .bind(DELTA_LIMIT) .fetch_all(&state.pool) .await?; // Hit the cap => this is only the newest slice of a larger gap. Signal the // client to full-refresh instead of merging a partial delta. let truncated = rows.len() as i64 >= DELTA_LIMIT; // `>=` for the same tie-break reason as the uploads query above; re-signalling an // already-removed id is idempotent on the client (it filters its list by these ids). let deleted_ids: Vec<(Uuid,)> = sqlx::query_as( "SELECT id FROM upload WHERE event_id = $1 AND deleted_at IS NOT NULL AND deleted_at >= $2", ) .bind(auth.event_id) .bind(q.since) .fetch_all(&state.pool) .await?; // Users hidden since the cursor (ban / uploads_hidden). Uncapped like `deleted_ids` and // for the same reason — an eviction the client misses is unrecoverable via a later delta. let hidden_user_ids: Vec<(Uuid,)> = sqlx::query_as( "SELECT id FROM \"user\" WHERE event_id = $1 AND uploads_hidden = TRUE AND uploads_hidden_at IS NOT NULL AND uploads_hidden_at >= $2", ) .bind(auth.event_id) .bind(q.since) .fetch_all(&state.pool) .await?; let upload_ids: Vec = rows.iter().map(|r| r.id).collect(); let liked_set = get_liked_set(&state.pool, auth.user_id, &upload_ids).await; let uploads = rows .into_iter() .map(|r| FeedUpload { liked_by_me: liked_set.contains(&r.id), id: r.id, user_id: r.user_id, uploader_name: r.uploader_name, preview_url: r .preview_path .as_ref() .map(|_| format!("/api/v1/upload/{}/preview", r.id)), thumbnail_url: r .thumbnail_path .as_ref() .map(|_| format!("/api/v1/upload/{}/thumbnail", r.id)), display_url: r .display_path .as_ref() .map(|_| format!("/api/v1/upload/{}/display", r.id)), mime_type: r.mime_type, caption: r.caption, like_count: r.like_count, comment_count: r.comment_count, created_at: r.created_at, }) .collect(); Ok(Json(DeltaResponse { uploads, deleted_ids: deleted_ids.into_iter().map(|r| r.0).collect(), hidden_user_ids: hidden_user_ids.into_iter().map(|r| r.0).collect(), truncated, server_time, })) } #[derive(Serialize)] pub struct HashtagCount { pub tag: String, pub count: i64, } pub async fn hashtags( State(state): State, auth: AuthUser, ) -> Result>, AppError> { let rows: Vec<(String, i64)> = sqlx::query_as("SELECT tag, upload_count FROM v_hashtag_counts WHERE event_id = $1") .bind(auth.event_id) .fetch_all(&state.pool) .await?; Ok(Json( rows.into_iter() .map(|(tag, count)| HashtagCount { tag, count }) .collect(), )) } /// Resolve a cursor id to its `(created_at, id)` position. Both are needed: /// `created_at` alone isn't unique, so pagination must break ties on `id` to /// avoid silently dropping rows that share a timestamp across a page boundary. async fn get_cursor_pos(pool: &sqlx::PgPool, cursor_id: Uuid) -> Option<(DateTime, Uuid)> { let row: Option<(DateTime, Uuid)> = sqlx::query_as("SELECT created_at, id FROM upload WHERE id = $1") .bind(cursor_id) .fetch_optional(pool) .await .ok()?; row } async fn get_liked_set( pool: &sqlx::PgPool, user_id: Uuid, upload_ids: &[Uuid], ) -> std::collections::HashSet { if upload_ids.is_empty() { return std::collections::HashSet::new(); } let rows: Vec<(Uuid,)> = sqlx::query_as("SELECT upload_id FROM \"like\" WHERE user_id = $1 AND upload_id = ANY($2)") .bind(user_id) .bind(upload_ids) .fetch_all(pool) .await .unwrap_or_default(); rows.into_iter().map(|r| r.0).collect() }