fix(upload): stop a late retry from undoing a host takedown

Migration 026 freed the idempotency key as soon as deleted_at was set, so a
retry after a delete uploads afresh instead of 409ing forever. That rationale
only considered the GUEST deleting. deleted_at is also set by
host_delete_upload, and there the same rule reverses a moderation decision:

  1. Guest uploads; the row commits and the photo appears, but the response
     is lost on the way back — the flaky-wifi case the key exists for — so
     the phone keeps the queue item.
  2. The host takes the photo down. Epoch bumped, keepsake rebuilt without it.
  3. The phone reconnects ten minutes later and retries. The key is free, the
     INSERT succeeds, and the photo is back — in the feed and in the next
     keepsake, under a NEW uuid that matches nothing in the host's moderation
     history, with nothing logged to say a takedown was reversed.

Migration 031 keeps the key claimed for a host takedown and releases it only
for a guest's own delete, so the retry resolves to the duplicate path and is
refused. The refusal now says why ("von den Gastgebern entfernt") rather than
"already processed", which invites another try.

The index predicate and the ON CONFLICT arbiter are changed in lockstep;
these queries are not compile-checked, so a drift between them is a 500 on
exactly the retries the index exists to serve. Verified against a real
Postgres: live retry suppressed, host takedown holds the key, guest delete
releases it. The integration test's copy of the insert is updated too — it is
verbatim by design, and a stale copy would have kept passing.

backend/src/handlers/host.rs

@@ -705,7 +705,9 @@ pub async fn host_delete_upload(
     // invalidation didn't, the taken-down photo would stay downloadable forever and nothing would
     // notice (the keepsake still looks complete, and the host can no longer find the upload to retry).
     let mut tx = state.pool.begin().await?;
-    let deleted = Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id).await?;
+    // `by_host: true` — the takedown holds the uploader's idempotency key so a late retry from
+    // their queue cannot resurrect the photo. See migration 031.
+    let deleted = Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id, true).await?;
     if !deleted {
         return Err(AppError::NotFound("Upload nicht gefunden.".into()));
     }

backend/src/handlers/upload.rs

@@ -770,11 +770,28 @@ pub async fn upload(
                     .map_err(AppError::from)?,
                 None => None,
             };
-            // If the winning row has vanished between the conflict and this lookup (deleted in
-            // the intervening milliseconds), there is nothing to replay — report the conflict.
-            let existing = existing.ok_or_else(|| {
-                AppError::Conflict("Dieser Upload wurde bereits verarbeitet.".into())
-            })?;
+            // No live row behind the key. Two very different causes, and the guest deserves to be
+            // told which: either the winning row vanished in the intervening milliseconds, or the
+            // key is still held by a photo the HOST took down (migration 031), in which case the
+            // refusal is the whole point and re-sending will never work.
+            let existing = match existing {
+                Some(e) => e,
+                None => {
+                    let taken_down = match client_upload_id {
+                        Some(cid) => {
+                            Upload::taken_down_by_client_upload_id(&state.pool, auth.user_id, cid)
+                                .await
+                                .unwrap_or(false)
+                        }
+                        None => false,
+                    };
+                    return Err(AppError::Conflict(if taken_down {
+                        "Dieses Foto wurde von den Gastgebern entfernt.".into()
+                    } else {
+                        "Dieser Upload wurde bereits verarbeitet.".to_string()
+                    }));
+                }
+            };
             tracing::info!(
                 upload_id = %existing.id,
                 "concurrent duplicate upload resolved; replaying the stored row"
@@ -978,7 +995,9 @@ pub async fn delete_upload(
     // Atomic with the keepsake invalidation: a guest removing their own photo must have it removed
     // from the downloadable archive too, and a half-applied delete would leave it there forever.
     let mut tx = state.pool.begin().await?;
-    Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id).await?;
+    // `by_host: false` — the guest deleted their own photo, so the idempotency key is released and
+    // a later retry of the same queue item uploads afresh rather than 409ing forever.
+    Upload::soft_delete_in_event(&mut tx, upload_id, auth.event_id, false).await?;
     let regen = crate::services::export::invalidate_and_arm(
         &mut tx,
         &state.config.event_slug,

backend/src/models/upload.rs

@@ -78,18 +78,20 @@ impl Upload {
         // The conflict target repeats the index's `WHERE` clause because it is a partial index;
         // without it Postgres cannot prove which index to use and rejects the statement.
         //
-        // KEEP THIS IN LOCKSTEP WITH `upload_client_upload_id_key` (migration 026). The predicate
-        // here must match the index's, or the arbiter cannot be inferred and every upload that
-        // carries a `client_upload_id` fails as a runtime 500 — queries in this codebase are not
-        // compile-time checked, so nothing catches a drift between the two at build time.
+        // KEEP THIS IN LOCKSTEP WITH `upload_client_upload_id_key` (migrations 026 and 031). The
+        // predicate here must match the index's, or the arbiter cannot be inferred and every
+        // upload that carries a `client_upload_id` fails as a runtime 500 — queries in this
+        // codebase are not compile-time checked, so nothing catches a drift at build time.
         //
-        // `AND deleted_at IS NULL` is what makes a retry-after-delete work instead of 409ing
-        // forever: the key is claimed only while a LIVE row holds it, which is what
-        // `find_by_client_upload_id` below has always assumed.
+        // `deleted_at IS NULL` is what makes a retry-after-delete work instead of 409ing forever:
+        // the key is claimed only while a LIVE row holds it, which is what
+        // `find_by_client_upload_id` below has always assumed. `OR taken_down_by_host` carves the
+        // moderation case back out — see migration 031: releasing the key for a HOST takedown let
+        // a late retry resurrect a photo the host had deliberately removed.
         sqlx::query_as::<_, Self>(
             "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes, caption, client_upload_id)
              VALUES ($1, $2, $3, $4, $5, $6, $7)
-             ON CONFLICT (client_upload_id) WHERE client_upload_id IS NOT NULL AND deleted_at IS NULL DO NOTHING
+             ON CONFLICT (client_upload_id) WHERE client_upload_id IS NOT NULL AND (deleted_at IS NULL OR taken_down_by_host) DO NOTHING
              RETURNING *",
         )
         .bind(event_id)
@@ -124,6 +126,29 @@ impl Upload {
         .await
     }
 
+    /// Was this key claimed by a row the HOST took down?
+    ///
+    /// Only used to answer a refused retry honestly. Without it the guest's queue shows
+    /// "Dieser Upload wurde bereits verarbeitet." for a photo that was in fact removed by the
+    /// hosts — technically true, actively misleading, and it invites them to try again.
+    pub async fn taken_down_by_client_upload_id(
+        pool: &sqlx::PgPool,
+        user_id: Uuid,
+        client_upload_id: Uuid,
+    ) -> Result<bool, sqlx::Error> {
+        sqlx::query_scalar::<_, bool>(
+            "SELECT EXISTS (
+                 SELECT 1 FROM upload
+                 WHERE client_upload_id = $1 AND user_id = $2
+                   AND deleted_at IS NOT NULL AND taken_down_by_host
+             )",
+        )
+        .bind(client_upload_id)
+        .bind(user_id)
+        .fetch_one(pool)
+        .await
+    }
+
     /// Lean lookup for the public media aliases (`get_original`/`get_preview`/
     /// `get_thumbnail`): returns ONLY the file paths + mime for a visible upload —
     /// excluding soft-deleted rows, hidden owners (`uploads_hidden`), and banned owners
@@ -289,20 +314,27 @@ impl Upload {
     /// dropped handler future, a failed second tx), the taken-down photo stays in the downloadable
     /// archive forever, and recovery can't tell — the keepsake still looks complete at the current
     /// epoch, and the host can no longer even find the upload to retry.
+    ///
+    /// `by_host` records WHO removed it, which decides whether the row keeps holding its
+    /// idempotency key — see migration 031. A host takedown holds it, so a late retry from the
+    /// uploader's queue cannot bring the photo back; a guest deleting their own photo releases it,
+    /// so their next upload of the same queue item succeeds.
     pub async fn soft_delete_in_event(
         conn: &mut sqlx::PgConnection,
         id: Uuid,
         event_id: Uuid,
+        by_host: bool,
     ) -> Result<bool, sqlx::Error> {
         let tx = conn;
         let row: Option<(Uuid, i64)> = sqlx::query_as(
             "UPDATE upload
-             SET deleted_at = NOW()
+             SET deleted_at = NOW(), taken_down_by_host = $3
              WHERE id = $1 AND event_id = $2 AND deleted_at IS NULL
              RETURNING user_id, original_size_bytes",
         )
         .bind(id)
         .bind(event_id)
+        .bind(by_host)
         .fetch_optional(&mut *tx)
         .await?;
         let deleted = if let Some((user_id, bytes)) = row {