Two smaller operational items. POSTGRES 512M -> 1G. DATABASE_MAX_CONNECTIONS is 30 for a ~100-guest event (feed polling + SSE + uploads at once), and 30 backends plus Postgres 16's default shared_buffers leaves very little headroom at 512M. An OOM here doesn't degrade one feature -- every request path touches the database, so it takes the event down. Memory is the cheaper knob than shrinking the pool back and reintroducing the queueing it was raised to fix. .env.example now names the pairing explicitly, the way it already does for COMPRESSION_WORKER_CONCURRENCY. SOCIAL WRITES WERE UNTHROTTLED. toggle_like, add_comment and delete_comment were the only mutating endpoints in the app with no limit at all -- upload, join, recover, export and admin login all carry one. Asymmetric coverage rather than a deliberate decision. Low severity, and honestly so: a like fans an SSE broadcast to every client, but the export regeneration a comment deletion triggers is contained (REGEN_DEBOUNCE 20s, workers born with their epoch, superseded ones inert). So the ceiling is 120/min -- far above anything a real guest produces. This bounds a script, not an enthusiastic double-tapper. ONE bucket across all three actions: separate buckets would let a caller triple the aggregate write rate by alternating between them. Keyed per USER, matching the feed and upload limits -- at a venue every guest is behind one NAT, and an IP key is what made the /join and /feed limits turn guests away in the first place. Migration 020 seeds both keys, and both are wired into the admin allowlist, the config UI and the e2e reseed -- the step two earlier per-area toggles missed, which left switches that existed in code and could never be flipped. Tests: 4 e2e, including that the shared bucket really is shared (the part most likely to be lost in a refactor) and that one guest hitting the ceiling doesn't block another behind the same IP. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
286 lines
11 KiB
Rust
286 lines
11 KiB
Rust
use axum::Json;
|
|
use axum::extract::{Path, Query, State};
|
|
use axum::http::StatusCode;
|
|
use chrono::{DateTime, Utc};
|
|
use serde::{Deserialize, Serialize};
|
|
use uuid::Uuid;
|
|
|
|
use crate::auth::middleware::AuthUser;
|
|
use crate::error::AppError;
|
|
use crate::models::comment::{Comment, CommentDto};
|
|
use crate::models::hashtag::{self, Hashtag};
|
|
use crate::models::upload::Upload;
|
|
use crate::services::config;
|
|
use crate::state::AppState;
|
|
|
|
/// Throttle a social write. Keyed PER USER, like the feed and upload limits and for the same
|
|
/// reason: at a venue every guest sits behind one NAT, so an IP key hands the whole party a
|
|
/// single bucket and the most active guest starves everyone else.
|
|
///
|
|
/// These were the only mutating endpoints in the app with no limit at all — the coverage was
|
|
/// asymmetric, not deliberately open. The ceiling is set well above anything a real guest
|
|
/// produces; this bounds a script, not an enthusiastic double-tapper.
|
|
async fn check_social_rate(state: &AppState, user_id: Uuid) -> Result<(), AppError> {
|
|
let rate_limits_on = config::get_bool(&state.config_cache, "rate_limits_enabled", true).await;
|
|
let social_rate_on = config::get_bool(&state.config_cache, "social_rate_enabled", true).await;
|
|
if !(rate_limits_on && social_rate_on) {
|
|
return Ok(());
|
|
}
|
|
let rate_limit = config::get_usize(&state.config_cache, "social_rate_per_min", 120).await;
|
|
// ONE bucket across likes, comments and comment deletions. Separate buckets would let a
|
|
// caller triple the aggregate write rate just by alternating between them.
|
|
state
|
|
.rate_limiter
|
|
.check_with_retry(
|
|
format!("social:{user_id}"),
|
|
rate_limit,
|
|
std::time::Duration::from_secs(60),
|
|
)
|
|
.map_err(|retry_after_secs| {
|
|
AppError::TooManyRequests(
|
|
"Zu viele Aktionen. Bitte warte kurz und versuche es erneut.".into(),
|
|
Some(retry_after_secs),
|
|
)
|
|
})
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct LikeResponse {
|
|
/// The caller's like state *after* this toggle. The client sets `liked_by_me` from
|
|
/// this rather than blind-inverting local state — otherwise a second device (same
|
|
/// recovered user) drifts, since the `like-update` broadcast only carries `like_count`.
|
|
pub liked: bool,
|
|
/// Fresh like count, or `null` if the (best-effort) count query hiccuped. The client
|
|
/// keeps its current count when this is null rather than adopting a wrong number.
|
|
pub like_count: Option<i64>,
|
|
}
|
|
|
|
pub async fn toggle_like(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
Path(upload_id): Path<Uuid>,
|
|
) -> Result<Json<LikeResponse>, AppError> {
|
|
// Check if user is banned
|
|
let user = crate::models::user::User::find_by_id(&state.pool, auth.user_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;
|
|
if user.is_banned {
|
|
return Err(AppError::Forbidden("Du bist gesperrt.".into()));
|
|
}
|
|
check_social_rate(&state, auth.user_id).await?;
|
|
|
|
// Event-scope: the upload must belong to the caller's event (404 otherwise),
|
|
// matching the host handlers' find_by_id_and_event pattern.
|
|
Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
|
|
// NOTE: liking is intentionally allowed while the event is locked. Locking
|
|
// ("Event schließen") freezes *new uploads* only — likes, comments and
|
|
// browsing stay open (USER_JOURNEYS §9.3, FEATURES capability matrix).
|
|
|
|
// Try to insert; if conflict, delete (toggle). `liked` = the caller's state afterwards.
|
|
let result = sqlx::query(
|
|
"INSERT INTO \"like\" (upload_id, user_id) VALUES ($1, $2)
|
|
ON CONFLICT (upload_id, user_id) DO NOTHING",
|
|
)
|
|
.bind(upload_id)
|
|
.bind(auth.user_id)
|
|
.execute(&state.pool)
|
|
.await?;
|
|
|
|
let liked = result.rows_affected() > 0;
|
|
if !liked {
|
|
// Already liked — remove
|
|
sqlx::query("DELETE FROM \"like\" WHERE upload_id = $1 AND user_id = $2")
|
|
.bind(upload_id)
|
|
.bind(auth.user_id)
|
|
.execute(&state.pool)
|
|
.await?;
|
|
}
|
|
|
|
// Fresh count so feed clients can patch the single card in place instead of
|
|
// refetching page 1 (mirrors v_feed.like_count = COUNT(DISTINCT user_id)). The like
|
|
// itself is already committed, so a failed count must not fail the request — but we
|
|
// also must NOT broadcast/return a bogus 0 (that would push like_count: 0 to every
|
|
// client until the next event). On error we skip the broadcast and return null.
|
|
let like_count = sqlx::query_scalar::<_, i64>(
|
|
"SELECT COUNT(DISTINCT user_id) FROM \"like\" WHERE upload_id = $1",
|
|
)
|
|
.bind(upload_id)
|
|
.fetch_one(&state.pool)
|
|
.await
|
|
.ok();
|
|
|
|
if let Some(count) = like_count {
|
|
// Broadcast the new count so other clients patch their card. Only `like_count` is
|
|
// shared — each client's own `liked_by_me` only changes via its own toggle (which
|
|
// now reads it straight from this response).
|
|
let _ = state.sse_tx.send(crate::state::SseEvent {
|
|
event_type: "like-update".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id, "like_count": count }).to_string(),
|
|
});
|
|
}
|
|
|
|
Ok(Json(LikeResponse { liked, like_count }))
|
|
}
|
|
|
|
#[derive(Deserialize, Default)]
|
|
pub struct ListCommentsQuery {
|
|
/// RFC3339 timestamp — return only comments older than this. Pass the
|
|
/// `created_at` of the oldest currently-loaded comment to fetch the next
|
|
/// older page.
|
|
pub before: Option<DateTime<Utc>>,
|
|
}
|
|
|
|
const COMMENT_PAGE_SIZE: i64 = 50;
|
|
|
|
pub async fn list_comments(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
Path(upload_id): Path<Uuid>,
|
|
Query(q): Query<ListCommentsQuery>,
|
|
) -> Result<Json<Vec<CommentDto>>, AppError> {
|
|
// Event-scope: only list comments for an upload in the caller's event.
|
|
Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
|
|
let comments =
|
|
Comment::list_for_upload(&state.pool, upload_id, q.before, COMMENT_PAGE_SIZE).await?;
|
|
Ok(Json(comments))
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
pub struct AddCommentRequest {
|
|
pub body: String,
|
|
}
|
|
|
|
pub async fn add_comment(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
Path(upload_id): Path<Uuid>,
|
|
Json(body): Json<AddCommentRequest>,
|
|
) -> Result<(StatusCode, Json<CommentDto>), AppError> {
|
|
// Comments can be disabled instance-wide (env COMMENTS_ENABLED). The frontend hides
|
|
// the UI, but gate the API too so a stale client or direct call can't slip one in.
|
|
if !state.config.comments_enabled {
|
|
return Err(AppError::Forbidden("Kommentare sind deaktiviert.".into()));
|
|
}
|
|
|
|
let user = crate::models::user::User::find_by_id(&state.pool, auth.user_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Benutzer nicht gefunden.".into()))?;
|
|
if user.is_banned {
|
|
return Err(AppError::Forbidden("Du bist gesperrt.".into()));
|
|
}
|
|
check_social_rate(&state, auth.user_id).await?;
|
|
|
|
// Event-scope: only comment on an upload that belongs to the caller's event.
|
|
Upload::find_by_id_and_event(&state.pool, upload_id, auth.event_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Upload nicht gefunden.".into()))?;
|
|
|
|
// NOTE: commenting is intentionally allowed while the event is locked. Locking
|
|
// freezes *new uploads* only — likes, comments and browsing stay open
|
|
// (USER_JOURNEYS §9.3, FEATURES capability matrix).
|
|
|
|
let text = body.body.trim();
|
|
let text_chars = text.chars().count();
|
|
if text_chars == 0 || text_chars > 500 {
|
|
return Err(AppError::BadRequest(
|
|
"Kommentar muss zwischen 1 und 500 Zeichen lang sein.".into(),
|
|
));
|
|
}
|
|
|
|
// Insert the comment and link its hashtags atomically, so a crash mid-loop
|
|
// can't leave a committed comment with only some of its tags indexed.
|
|
let tags = hashtag::extract_hashtags(text);
|
|
let mut tx = state.pool.begin().await?;
|
|
let comment = Comment::create(&mut *tx, upload_id, auth.user_id, text).await?;
|
|
for tag in &tags {
|
|
let h = Hashtag::upsert(&mut *tx, auth.event_id, tag).await?;
|
|
sqlx::query(
|
|
"INSERT INTO comment_hashtag (comment_id, hashtag_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
|
|
)
|
|
.bind(comment.id)
|
|
.bind(h.id)
|
|
.execute(&mut *tx)
|
|
.await?;
|
|
}
|
|
tx.commit().await?;
|
|
|
|
// Fresh count so feed clients can patch the single card in place instead of
|
|
// refetching page 1 (mirrors v_feed.comment_count = COUNT(DISTINCT c.id); COUNT(*)
|
|
// over the same deleted_at filter is identical since comment.id is the PK). The
|
|
// count + broadcast are a UI optimisation — the comment is already committed, so a
|
|
// failure here must not fail the request. Swallow the error and skip the broadcast.
|
|
if let Ok(comment_count) = sqlx::query_scalar::<_, i64>(
|
|
"SELECT COUNT(*) FROM comment WHERE upload_id = $1 AND deleted_at IS NULL",
|
|
)
|
|
.bind(upload_id)
|
|
.fetch_one(&state.pool)
|
|
.await
|
|
{
|
|
let _ = state.sse_tx.send(crate::state::SseEvent {
|
|
event_type: "new-comment".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id, "comment_count": comment_count })
|
|
.to_string(),
|
|
});
|
|
}
|
|
|
|
let dto = CommentDto {
|
|
id: comment.id,
|
|
upload_id,
|
|
user_id: auth.user_id,
|
|
uploader_name: user.display_name,
|
|
body: comment.body,
|
|
created_at: comment.created_at,
|
|
};
|
|
|
|
Ok((StatusCode::CREATED, Json(dto)))
|
|
}
|
|
|
|
pub async fn delete_comment(
|
|
State(state): State<AppState>,
|
|
auth: AuthUser,
|
|
Path(comment_id): Path<Uuid>,
|
|
) -> Result<StatusCode, AppError> {
|
|
// Banned users keep read access but cannot mutate (USER_JOURNEYS §10).
|
|
if auth.is_banned {
|
|
return Err(AppError::Forbidden("Du bist gesperrt.".into()));
|
|
}
|
|
check_social_rate(&state, auth.user_id).await?;
|
|
let comment = Comment::find_by_id(&state.pool, comment_id)
|
|
.await?
|
|
.ok_or_else(|| AppError::NotFound("Kommentar nicht gefunden.".into()))?;
|
|
|
|
if comment.user_id != auth.user_id {
|
|
return Err(AppError::Forbidden("Nur eigene Kommentare löschen.".into()));
|
|
}
|
|
|
|
// Event-scope: soft_delete_in_event only matches comments whose upload is in
|
|
// the caller's event, so a cross-event comment_id resolves to a 404 here.
|
|
let mut tx = state.pool.begin().await?;
|
|
let deleted = Comment::soft_delete_in_event(&mut tx, comment_id, auth.event_id).await?;
|
|
if !deleted {
|
|
return Err(AppError::NotFound("Kommentar nicht gefunden.".into()));
|
|
}
|
|
// Comments live only in the HTML viewer, so the ZIP is carried forward, not rebuilt.
|
|
let regen = crate::services::export::invalidate_and_arm(
|
|
&mut tx,
|
|
&state.config.event_slug,
|
|
crate::services::export::Affects::ViewerOnly,
|
|
)
|
|
.await?;
|
|
tx.commit().await?;
|
|
if let Some(r) = regen {
|
|
crate::handlers::host::start_regen(&state, r);
|
|
}
|
|
let _ = state.sse_tx.send(crate::state::SseEvent::new(
|
|
"comment-deleted",
|
|
serde_json::json!({ "comment_id": comment_id, "upload_id": comment.upload_id }).to_string(),
|
|
));
|
|
Ok(StatusCode::NO_CONTENT)
|
|
}
|