Harden bulk retries, parallelize guest uploads, and move guest map below gallery

apps/web/src/app/dashboard/events/[id]/bulk-photo-actions.tsx

@@ -15,12 +15,12 @@ export function BulkPhotoActions({ eventId, rows, selected, setSelected, canMode
   selected: string[]; setSelected: (ids: string[]) => void; canModerate: boolean; canDelete: boolean; canPrivate: boolean;
 }) {
   const utils = api.useUtils();
-  const [pending, setPending] = useState<{ eventId: string; photoIds: string[]; action: Action } | null>(null);
+  const [pending, setPending] = useState<{ eventId: string; photoIds: string[]; action: Action; requestId: string } | null>(null);
   const [report, setReport] = useState<{ photoId: string; status: string; reason?: string }[]>([]);
-  const preview = api.manager.previewBulkPhotos.useQuery(pending!, { enabled: !!pending, retry: false });
+  const preview = api.manager.previewBulkPhotos.useQuery(pending!, { enabled: !!pending, retry: false, refetchOnWindowFocus: false, staleTime: Infinity });
   const apply = api.manager.bulkPhotos.useMutation({
     onSuccess: async result => {
-      const allResults = [...result.results, ...(preview.data?.results.filter(row => row.status !== "eligible") ?? [])];
+      const allResults = result.results;
       setReport(allResults);
       setSelected(allResults.filter(row => row.status === "failed" || row.status === "skipped").map(row => row.photoId));
       setPending(null);
@@ -40,7 +40,7 @@ export function BulkPhotoActions({ eventId, rows, selected, setSelected, canMode
     </div>
     {selected.length ? <div className="flex flex-wrap gap-2">{actions.map(action => {
       const Icon = icons[action];
-      return <Button key={action} variant={action === "delete" ? "destructive" : "outline"} disabled={apply.isPending} onClick={() => setPending({ eventId, photoIds: [...selected], action })}><Icon data-icon="inline-start" />{labels[action]} selected</Button>;
+      return <Button key={action} variant={action === "delete" ? "destructive" : "outline"} disabled={apply.isPending} onClick={() => setPending({ eventId, photoIds: [...selected], action, requestId: crypto.randomUUID() })}><Icon data-icon="inline-start" />{labels[action]} selected</Button>;
     })}</div> : null}
     {report.length ? <details><summary className="cursor-pointer text-sm">Last bulk action · {report.length} results</summary><ul className="max-h-48 overflow-auto text-sm">{report.map(row => <li key={row.photoId}>{row.photoId.slice(0, 8)} · {row.status}{row.reason ? ` — ${row.reason}` : ""}</li>)}</ul></details> : null}
     <Dialog open={!!pending} onOpenChange={open => { if (!open && !apply.isPending) setPending(null); }}>
@@ -58,7 +58,7 @@ export function BulkPhotoActions({ eventId, rows, selected, setSelected, canMode
         </>}
         <DialogFooter><Button variant="outline" disabled={apply.isPending} onClick={() => setPending(null)}>Cancel</Button>
           <Button variant={pending?.action === "delete" ? "destructive" : "default"} disabled={apply.isPending || preview.isFetching || !preview.data?.affected || preview.isError} onClick={() => {
-            if (pending && preview.data) apply.mutate({ ...pending, photoIds: preview.data.results.filter(row => row.status === "eligible").map(row => row.photoId), confirm: true });
+            if (pending && preview.data) apply.mutate({ ...pending, confirm: true });
           }}>{apply.isPending ? "Applying…" : `Confirm ${pending ? labels[pending.action].toLowerCase() : "changes"}`}</Button></DialogFooter>
       </DialogContent>
     </Dialog>

apps/web/src/app/dashboard/events/[id]/gallery-upload.tsx

@@ -12,22 +12,30 @@ export function GalleryUpload({ eventId }: { eventId: string }) {
   const utils = api.useUtils();
   const create = api.manager.createGalleryPhotos.useMutation();
   const complete = api.manager.completeGalleryPhotos.useMutation();
+  const retry = api.manager.retryGalleryPhoto.useMutation();
+  const batch = useRef<{ files: File[]; requestId: string; uploads?: { photoId: string }[] } | null>(null);
   const [busy, setBusy] = useState(false);
   const [items, setItems] = useState<{ name: string; status: string }[]>([]);
-  async function upload(files: File[]) {
+  async function upload(files: File[], resume = false) {
     if (busy || !files.length) return;
-    const parsed = createGalleryPhotosInputSchema.safeParse({ eventId, files: files.map(file => ({ fileName: file.name, contentType: file.type, byteSize: file.size })) });
+    const requestId = resume && batch.current ? batch.current.requestId : crypto.randomUUID();
+    const parsed = createGalleryPhotosInputSchema.safeParse({ eventId, requestId, files: files.map(file => ({ fileName: file.name, contentType: file.type, byteSize: file.size })) });
     if (!parsed.success) { toast.error("Choose up to 25 supported images, each under 25 MB."); return; }
     setBusy(true);
-    setItems(files.map(file => ({ name: file.name, status: "Waiting" })));
+    if (!resume) { batch.current = { files, requestId }; setItems(files.map(file => ({ name: file.name, status: "Waiting" }))); }
     const status = (index: number, value: string) => setItems(current => current.map((item, n) => n === index ? { ...item, status: value } : item));
     try {
-      const uploads = await create.mutateAsync(parsed.data);
+      const uploads = batch.current?.uploads ?? await create.mutateAsync(parsed.data);
+      if (batch.current) batch.current.uploads = uploads;
       for (const [index, file] of files.entries()) {
+        if (resume && items[index]?.status === "Queued") continue;
         try {
           status(index, "Uploading");
-          const uploaded = await fetch(uploads[index]!.uploadUrl, { method: "PUT", headers: { "Content-Type": file.type }, body: file, signal: AbortSignal.timeout(120_000) });
-          if (!uploaded.ok) throw new Error("Upload failed");
+          const target = await retry.mutateAsync({ eventId, photoId: uploads[index]!.photoId });
+          if (target.uploadUrl) {
+            const uploaded = await fetch(target.uploadUrl, { method: "PUT", headers: { "Content-Type": file.type }, body: file, signal: AbortSignal.timeout(120_000) });
+            if (!uploaded.ok) throw new Error("Upload failed");
+          }
           const [result] = await complete.mutateAsync({ eventId, photoIds: [uploads[index]!.photoId] });
           status(index, result?.status === "processing" || result?.status === "ready" ? "Queued" : "Failed");
         } catch { status(index, "Failed"); }
@@ -45,6 +53,7 @@ export function GalleryUpload({ eventId }: { eventId: string }) {
     <input ref={input} className="sr-only" type="file" multiple accept="image/jpeg,image/png,image/webp,image/heic,image/heif" aria-label="Upload gallery photos" disabled={busy}
       onChange={event => { const files = Array.from(event.target.files ?? []); event.target.value = ""; void upload(files); }} />
     <Button className="self-start" variant="outline" disabled={busy} onClick={() => input.current?.click()}><UploadIcon data-icon="inline-start" />{busy ? "Uploading…" : "Upload photos"}</Button>
+    {items.some(item => item.status === "Failed") ? <Button className="self-start" variant="outline" disabled={busy} onClick={() => { if (batch.current) void upload(batch.current.files, true); }}>Retry failed files</Button> : null}
     {items.length ? <details><summary className="cursor-pointer text-sm">{Math.round(finished / items.length * 100)}% processed · {items.filter(item => item.status === "Queued").length}/{items.length} queued · {items.filter(item => item.status === "Failed").length} failed</summary>
       <ul className="max-h-40 overflow-auto text-sm">{items.map((item, index) => <li key={index}>{item.name} — {item.status}</li>)}</ul></details> : null}
   </div>;

apps/web/src/app/dashboard/events/[id]/moderation-grid.tsx

@@ -54,7 +54,11 @@ const processingOrder = {
   failed: 4,
 } as const;
 
-export function ModerationGrid({
+export function ModerationGrid(props: { eventId: string; canModerate: boolean; canDelete: boolean; canPrivate: boolean; canUpload: boolean }) {
+  return <>{props.canUpload ? <GalleryUpload eventId={props.eventId} /> : null}<GalleryGrid {...props} /></>;
+}
+
+function GalleryGrid({
   eventId,
   canModerate,
   canDelete,
@@ -126,7 +130,6 @@ export function ModerationGrid({
   if (rows.length === 0 && !filter && page === 0) {
     return (
       <Empty className="border">
-        {canUpload ? <GalleryUpload eventId={eventId} /> : null}
         <EmptyHeader>
           <EmptyTitle>No uploads yet</EmptyTitle>
           <EmptyDescription>
@@ -139,7 +142,6 @@ export function ModerationGrid({
 
   return (
     <>
-      {canUpload ? <GalleryUpload eventId={eventId} /> : null}
       <div className="mb-4 flex flex-wrap items-center gap-2">
         <DropdownMenu><DropdownMenuTrigger asChild><Button variant="outline">{filter ? visibilityLabels[filter] : "All photos"}<ChevronDownIcon data-icon="inline-end" /></Button></DropdownMenuTrigger>
           <DropdownMenuContent><DropdownMenuGroup>

apps/web/src/app/e/[slug]/guest-upload.tsx

@@ -19,6 +19,7 @@ import { Progress } from "@/components/ui/progress";
 import { Spinner } from "@/components/ui/spinner";
 import { Card, CardHeader, CardTitle, CardDescription, CardContent } from "@/components/ui/card";
 import { uploadProgress } from "@/lib/upload-progress";
+import { runUploadQueue } from "@/lib/upload-queue";
 
 type QueueItem = {
   file?: File;
@@ -106,7 +107,7 @@ export function GuestUpload({
     }
     if (!accepted.length) return;
     uploading.current = true;
-    const items: QueueItem[] = retryItems ?? accepted.map((file) => ({
+    const items: QueueItem[] = retryItems?.map((item) => ({ ...item })) ?? accepted.map((file) => ({
       file,
       id: crypto.randomUUID(),
       name: file.name,
@@ -131,13 +132,15 @@ export function GuestUpload({
         notifyWhenReady: Boolean(trimmedEmail) && notify,
         note: trimmedNote || undefined,
       });
-      const submission = await startSubmission.mutateAsync({ eventSlug: slug });
+      // Retrying known photos does not need another (empty) submission.
+      const submission = items.some((item) => !item.created)
+        ? await startSubmission.mutateAsync({ eventSlug: slug }) : null;
 
-      for (const [index, file] of accepted.entries()) {
+      await runUploadQueue(accepted, async (file, index) => {
         const item = items[index];
-        if (!item) continue;
+        if (!item) return;
         const contentType = imageContentType(file);
-        if (!contentType) continue;
+        if (!contentType) return;
         setQueue((current) =>
           current.map((entry) =>
             entry.id === item.id ? { ...entry, status: "uploading" } : entry,
@@ -146,7 +149,7 @@ export function GuestUpload({
         try {
           const created = item.created ?? await createPhoto.mutateAsync({
             eventSlug: slug,
-            submissionId: submission.submissionId,
+            submissionId: submission!.submissionId,
             contentType,
             fileName: file.name,
             byteSize: file.size,
@@ -181,8 +184,9 @@ export function GuestUpload({
             ),
           );
         }
-      }
-      await utils.event.gallery.invalidate(slug);
+      });
+      // Gallery refresh is not part of upload completion (nor is compression).
+      void utils.event.gallery.invalidate(slug).catch(() => undefined);
       router.refresh();
     } catch (error) {
       toast.error(error instanceof Error ? error.message : "Could not start upload");

apps/web/src/app/e/[slug]/page.tsx

@@ -70,12 +70,6 @@ export default async function EventPage({
             community.stats[key] !== null ? <div key={key}><dt className="text-sm text-muted-foreground">{label}</dt><dd className="text-2xl font-semibold">{community.stats[key]}</dd></div> : null)}
         </dl>
       ) : null}
-      {event.latitude !== null && event.longitude !== null ? (
-        <section aria-label="Event location" className="flex flex-col gap-3">
-          <h2 className="text-2xl font-semibold">Getting Here</h2>
-          <EventMap latitude={event.latitude} longitude={event.longitude} location={event.location ?? event.title} />
-        </section>
-      ) : null}
       <section id="upload" className="reveal-2 scroll-mt-24 flex flex-col gap-4">
         <h2 className="sr-only">Add photos or send a note</h2>
         <GuestUpload
@@ -104,6 +98,12 @@ export default async function EventPage({
           </p>
         )}
       </section></> : null}
+      {event.latitude !== null && event.longitude !== null ? (
+        <section aria-label="Event location" className="flex flex-col gap-3">
+          <h2 className="text-2xl font-semibold">Getting Here</h2>
+          <EventMap latitude={event.latitude} longitude={event.longitude} location={event.location ?? event.title} />
+        </section>
+      ) : null}
     </main>
   );
 }

apps/web/src/lib/upload-queue.test.ts

@@ -0,0 +1,63 @@
+import { expect, test } from "bun:test";
+import { runUploadQueue, UPLOAD_CONCURRENCY } from "./upload-queue";
+
+function gate() {
+  let release!: () => void;
+  const promise = new Promise<void>((resolve) => { release = resolve; });
+  return { promise, release };
+}
+
+test("three uploads start immediately and a free slot refills without waiting for a slow file", async () => {
+  const gates = Array.from({ length: 7 }, gate);
+  const fourthStarted = gate();
+  const started: number[] = [];
+  let active = 0;
+  let peak = 0;
+  let finished = false;
+  const batch = runUploadQueue(gates, async (item, index) => {
+    active++;
+    peak = Math.max(peak, active);
+    started.push(index);
+    if (index === 3) fourthStarted.release();
+    await item.promise;
+    active--;
+  }).then(() => { finished = true; });
+  expect(started).toEqual([0, 1, 2]);
+  expect(finished).toBe(false);
+  gates[1]!.release();
+  await fourthStarted.promise;
+  expect(started).toEqual([0, 1, 2, 3]);
+  expect(finished).toBe(false);
+  for (const item of gates) item.release();
+  await batch;
+  expect(started).toEqual([0, 1, 2, 3, 4, 5, 6]);
+  expect(peak).toBe(UPLOAD_CONCURRENCY);
+  expect(active).toBe(0);
+});
+
+test("a failed file does not stop the queue or release the batch before other uploads finish", async () => {
+  const slow = gate();
+  const lastStarted = gate();
+  const visited: number[] = [];
+  let settled = false;
+  const batch = runUploadQueue([0, 1, 2, 3, 4], async (item) => {
+    visited.push(item);
+    if (item === 0) throw new Error("Connection lost");
+    if (item === 1) await slow.promise;
+    if (item === 4) lastStarted.release();
+  }).catch((error: unknown) => { settled = true; return error; });
+  await lastStarted.promise;
+  expect(settled).toBe(false);
+  slow.release();
+  expect(await batch).toBeInstanceOf(AggregateError);
+  expect(visited.sort()).toEqual([0, 1, 2, 3, 4]);
+});
+
+test("empty and single-file selections complete without extra work", async () => {
+  const visited: string[] = [];
+  const upload = async (item: string) => { visited.push(item); };
+  await runUploadQueue([], upload);
+  expect(visited).toEqual([]);
+  await runUploadQueue(["photo"], upload);
+  expect(visited).toEqual(["photo"]);
+});

apps/web/src/lib/upload-queue.ts

@@ -0,0 +1,23 @@
+// Keep a few transfers in flight without flooding mobile connections or allocating
+// a promise/network request for every selected file at once.
+export const UPLOAD_CONCURRENCY = 3;
+
+export async function runUploadQueue<T>(
+  items: readonly T[],
+  upload: (item: T, index: number) => Promise<void>,
+) {
+  let next = 0;
+  const errors: unknown[] = [];
+  await Promise.all(Array.from({ length: Math.min(UPLOAD_CONCURRENCY, items.length) }, async () => {
+    while (next < items.length) {
+      const index = next++;
+      try {
+        await upload(items[index]!, index);
+      } catch (error) {
+        errors.push(error);
+      }
+    }
+  }));
+  // Never release the batch's busy guard while other transfers are still running.
+  if (errors.length) throw new AggregateError(errors, "Some uploads failed");
+}

apps/web/src/server/api/routers/manager.ts

@@ -1,6 +1,6 @@
 import { TRPCError } from "@trpc/server";
 import { bulkPhotos } from "@/server/bulk-photos";
-import { createGalleryPhotos, completeGalleryPhotos } from "@/server/gallery-uploads";
+import { createGalleryPhotos, completeGalleryPhotos, retryGalleryPhoto } from "@/server/gallery-uploads";
 import { customBannerUrl, eventBannerUrl } from "@/server/event-banner";
 import { searchLocations } from "@/server/location-search";
 import { notifyEventGuests } from "@/server/guest-notifications";
@@ -25,6 +25,7 @@ import {
 import {
   bulkPhotosInputSchema,
   createGalleryPhotosInputSchema,
+  retryGalleryPhotoInputSchema,
   completeGalleryPhotosInputSchema,
   applyBulkPhotosInputSchema,
   createEventInputSchema,
@@ -479,6 +480,7 @@ export const managerRouter = createTRPCRouter({
 
   previewBulkPhotos: protectedProcedure.input(bulkPhotosInputSchema).query(({ ctx, input }) => bulkPhotos(ctx.session.user.id, input, false)),
   createGalleryPhotos: protectedProcedure.input(createGalleryPhotosInputSchema).mutation(({ ctx, input }) => createGalleryPhotos(ctx.session.user.id, input)),
+  retryGalleryPhoto: protectedProcedure.input(retryGalleryPhotoInputSchema).mutation(({ ctx, input }) => retryGalleryPhoto(ctx.session.user.id, input)),
   completeGalleryPhotos: protectedProcedure.input(completeGalleryPhotosInputSchema).mutation(({ ctx, input }) => completeGalleryPhotos(ctx.session.user.id, input)),
   bulkPhotos: protectedProcedure.input(applyBulkPhotosInputSchema).mutation(({ ctx, input }) => bulkPhotos(ctx.session.user.id, input, true)),
 

apps/web/src/server/bulk-photos.ts

@@ -7,18 +7,24 @@ import { canTransitionVisibility } from "@/lib/photo-status";
 import { EVENT_PERMISSIONS } from "./permissions";
 import { loadEventAccess, requireEventPermission } from "./api/trpc";
 import { getPlatformRole } from "./roles";
+import { runOnce } from "./operation-receipts";
 
-export async function bulkPhotos(userId: string, input: z.infer<typeof bulkPhotosInputSchema>, apply: boolean) {
+export async function bulkPhotos(userId: string, input: z.infer<typeof bulkPhotosInputSchema> & { requestId?: string }, apply: boolean) {
   const { event, access } = await loadEventAccess(userId, input.eventId, await getPlatformRole(userId));
   requireEventPermission(access.permissions, EVENT_PERMISSIONS.PHOTOS_READ);
   requireEventPermission(access.permissions, input.action === "delete" ? EVENT_PERMISSIONS.PHOTOS_DELETE : EVENT_PERMISSIONS.PHOTOS_MODERATE);
   const canPrivate = access.permissions.includes(EVENT_PERMISSIONS.PHOTOS_PRIVATE_READ);
   if (input.action === "private") requireEventPermission(access.permissions, EVENT_PERMISSIONS.PHOTOS_PRIVATE_READ);
+  const identity = { userId, eventId: input.eventId, key: `bulk:${input.requestId}` };
+  if (apply) {
+    if (!input.requestId) throw new Error("Request ID required");
+    await runOnce(identity, { action: input.action, photoIds: [...input.photoIds].sort() }, async () => true);
+  }
   const results: { photoId: string; status: "eligible" | "updated" | "deleted" | "skipped" | "failed"; reason?: string }[] = [];
   // Each photo is its own transaction: failures never hide earlier successful work.
   for (const photoId of input.photoIds) {
     try {
-      results.push(await getDb().transaction(async tx => {
+      const execute = async (tx: Parameters<Parameters<ReturnType<typeof getDb>["transaction"]>[0]>[0]) => {
         const predicate = and(eq(photos.eventId, input.eventId), eq(photos.id, photoId));
         const [photo] = await tx.select().from(photos).where(predicate).for("update");
         const skip = (reason: string) => ({ photoId, status: "skipped" as const, reason });
@@ -36,7 +42,8 @@ export async function bulkPhotos(userId: string, input: z.infer<typeof bulkPhoto
           action: input.action === "delete" ? "photo.delete" : "photo.visibility", subjectType: "photo", subjectId: photoId,
           metadata: { bulk: true, "visibility.before": photo.visibility, "visibility.after": input.action } });
         return { photoId, status: input.action === "delete" ? "deleted" as const : "updated" as const };
-      }));
+      };
+      results.push(apply ? await runOnce({ ...identity, key: `${identity.key}:${photoId}` }, input.action, execute) : await getDb().transaction(execute));
     } catch {
       results.push({ photoId, status: "failed", reason: "Operation failed; refresh before retrying" });
     }

apps/web/src/server/gallery-uploads.ts

@@ -8,6 +8,7 @@ import { loadEventAccess, requireEventPermission } from "./api/trpc";
 import { EVENT_PERMISSIONS } from "./permissions";
 import { getPlatformRole } from "./roles";
 import { consumeRateLimit } from "./rate-limit";
+import { runOnce } from "./operation-receipts";
 
 async function authorize(userId: string, eventId: string) {
   const { event, access } = await loadEventAccess(userId, eventId, await getPlatformRole(userId));
@@ -19,19 +20,36 @@ export async function createGalleryPhotos(userId: string, input: z.infer<typeof
   const event = await authorize(userId, input.eventId);
   const rate = await consumeRateLimit({ namespace: `gallery-upload:${event.id}`, identifier: userId, limit: 10, windowMs: 600_000 });
   if (!rate.allowed) throw new TRPCError({ code: "TOO_MANY_REQUESTS", message: "Too many upload batches. Try again shortly." });
-  const uploads = await Promise.all(input.files.map(async file => {
-    const photoId = crypto.randomUUID();
-    const key = originalObjectKey(event.id, photoId);
-    return { photoId, key, uploadUrl: await createPresignedPutUrl({ key, contentType: file.contentType }) };
-  }));
-  await getDb().transaction(async tx => {
+  const uploads = await runOnce({ userId, eventId: input.eventId, key: `upload:${input.requestId}` }, input.files, async tx => {
+    const uploads = input.files.map(() => {
+      const photoId = crypto.randomUUID();
+      return { photoId, key: originalObjectKey(event.id, photoId) };
+    });
     const [guest] = await tx.insert(guests).values({ eventId: event.id, displayName: "Event organizer", tokenHash: crypto.randomUUID() }).returning();
     const [submission] = await tx.insert(submissions).values({ eventId: event.id, guestId: guest!.id }).returning();
     await tx.insert(photos).values(uploads.map((upload, index) => ({ id: upload.photoId, eventId: event.id, submissionId: submission!.id,
       originalKey: upload.key, contentType: input.files[index]!.contentType, byteSize: input.files[index]!.byteSize, processingStatus: "uploading" as const, visibility: "pending" as const })));
     await tx.insert(auditEvents).values(uploads.map(upload => ({ groupId: event.groupId, eventId: event.id, actorUserId: userId, action: "photo.create", subjectType: "photo", subjectId: upload.photoId })));
+    return uploads.map(({ photoId }) => ({ photoId }));
   });
-  return uploads.map(({ photoId, uploadUrl }) => ({ photoId, uploadUrl }));
+  return Promise.all(uploads.map(upload => prepareRetry(userId, event.id, upload.photoId)));
+}
+
+async function prepareRetry(userId: string, eventId: string, photoId: string) {
+  const [photo] = await getDb().select().from(photos).where(and(eq(photos.eventId, eventId), eq(photos.id, photoId)));
+  const [created] = await getDb().select({ id: auditEvents.id }).from(auditEvents).where(and(eq(auditEvents.eventId, eventId), eq(auditEvents.actorUserId, userId), eq(auditEvents.subjectId, photoId), eq(auditEvents.action, "photo.create"))).limit(1);
+  if (!photo || !created) throw new TRPCError({ code: "NOT_FOUND", message: "Your organizer upload was not found" });
+  if (photo.processingStatus !== "uploading") return { photoId, uploadUrl: null, status: photo.processingStatus };
+  // A PUT may have succeeded even when its response was lost. Never overwrite it.
+  const uploaded = await headObject(photo.originalKey);
+  return { photoId, status: "uploading", uploadUrl: uploaded ? null : await createPresignedPutUrl({ key: photo.originalKey, contentType: photo.contentType }) };
+}
+
+export async function retryGalleryPhoto(userId: string, input: { eventId: string; photoId: string }) {
+  await authorize(userId, input.eventId);
+  const rate = await consumeRateLimit({ namespace: `gallery-retry:${input.eventId}`, identifier: userId, limit: 100, windowMs: 600_000 });
+  if (!rate.allowed) throw new TRPCError({ code: "TOO_MANY_REQUESTS" });
+  return prepareRetry(userId, input.eventId, input.photoId);
 }
 
 export async function completeGalleryPhotos(userId: string, input: z.infer<typeof completeGalleryPhotosInputSchema>) {

apps/web/src/server/mcp/catalog.ts

@@ -36,6 +36,7 @@ export const assistantTools: ToolDefinition[] = [
   tool("manager.photos", "List event photos. Private photos require photos.private.read.", [E.PHOTOS_READ]),
   tool("manager.createGalleryPhotos", "Create up to 25 pending organizer photos and return presigned PUT URLs in file order. Upload original bytes directly to those URLs, then call completeGalleryPhotos. Does not publish photos.", [E.SETTINGS_MANAGE], true),
   tool("manager.completeGalleryPhotos", "Verify direct uploads and queue image processing for up to 25 photos in an event. Returns per-photo results.", [E.SETTINGS_MANAGE], true),
+  tool("manager.retryGalleryPhoto", "Resume your organizer upload using the same photo ID. A null uploadUrl means bytes already exist or processing started; do not upload again. Call completeGalleryPhotos if status is uploading.", [E.SETTINGS_MANAGE], true),
   tool("manager.previewBulkPhotos", "Preview up to 100 explicit photo IDs for a bulk action. Returns eligible and skipped IDs; permission checked for the chosen action.", [E.PHOTOS_READ]),
   tool("manager.bulkPhotos", "Apply approval, hiding, rejection, privacy, or deletion to up to 100 explicit photo IDs. Preview first; pass only eligible IDs and input.confirm=true. Returns per-photo results. Delete requires photos.delete; other actions require photos.moderate.", [E.PHOTOS_READ], true),
   tool("manager.notes", "List guest notes.", [E.NOTES_READ]),

apps/web/src/server/mcp/mcp.integration.test.ts

@@ -6,6 +6,8 @@ import { appRouter } from "../api/root";
 import { handleMcp } from "./server";
 import { assistantTools, permissionOptions } from "./catalog";
 import type { TrpcContext } from "../api/trpc";
+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
 
 test("MCP allowlist references real procedures and has unique names", () => {
   const procedures = appRouter._def.procedures as unknown as Record<string, unknown>;
@@ -39,6 +41,15 @@ test.skipIf(process.env.MCP_INTEGRATION !== "1")("MCP protocol, permission ceili
     otherGroupId = otherGroup!.id;
     const [otherEvent] = await db.insert(events).values({ groupId: otherGroupId, title: "Not accessible", slug: `${id}-other` }).returning();
     const read = await owner.create({ name: "Read only", permissions: [...permissionOptions], readOnly: true, days: 1 });
+    const http = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch: handleMcp });
+    const sdk = new Client({ name: "Manyangles compatibility test", version: "1" });
+    try {
+      await sdk.connect(new StreamableHTTPClientTransport(new URL(`http://127.0.0.1:${http.port}/api/mcp`), { requestInit: { headers: { Authorization: `Bearer ${read.token}` } } }));
+      expect((await sdk.listTools()).tools.some(tool => tool.name === "manager_stats")).toBe(true);
+      const result = await sdk.callTool({ name: "manager_stats", arguments: { input: { eventId: event!.id } } });
+      expect(result.isError).not.toBe(true);
+      expect((await sdk.callTool({ name: "manager_bulkPhotos", arguments: {} })).isError).toBe(true);
+    } finally { await sdk.close(); http.stop(true); }
     const write = await owner.create({ name: "Writer", permissions: ["group.manage", "group.read", "overview.read"], readOnly: false, days: 1 });
     const limited = await owner.create({ name: "Limited", permissions: ["group.read"], readOnly: false, days: 1 });
     const init = await request(read.token, "initialize", { protocolVersion: "2025-11-25", capabilities: {}, clientInfo: { name: "test", version: "1" } });

apps/web/src/server/operation-receipts.ts

@@ -0,0 +1,24 @@
+import { createHash } from "node:crypto";
+import { TRPCError } from "@trpc/server";
+import { and, eq, sql } from "drizzle-orm";
+import { getDb, operationReceipts } from "@album/database";
+
+type Transaction = Parameters<Parameters<ReturnType<typeof getDb>["transaction"]>[0]>[0];
+const hash = (value: string) => createHash("sha256").update(value).digest("hex");
+
+export async function runOnce<T>(identity: { userId: string; eventId: string; key: string }, payload: unknown, run: (tx: Transaction) => Promise<T>): Promise<T> {
+  const id = hash(JSON.stringify(identity));
+  const requestHash = hash(JSON.stringify(payload));
+  return getDb().transaction(async tx => {
+    // Serializes concurrent deliveries, including across web instances.
+    await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${id}, 0))`);
+    const [existing] = await tx.select().from(operationReceipts).where(and(eq(operationReceipts.id, id), eq(operationReceipts.eventId, identity.eventId), eq(operationReceipts.userId, identity.userId)));
+    if (existing) {
+      if (existing.requestHash !== requestHash) throw new TRPCError({ code: "CONFLICT", message: "This request ID was already used with different inputs. Use a new request ID for a new action." });
+      return existing.result as T;
+    }
+    const result = await run(tx);
+    await tx.insert(operationReceipts).values({ id, eventId: identity.eventId, userId: identity.userId, requestHash, result });
+    return result;
+  });
+}

apps/web/src/server/submission-groups.integration.test.ts

@@ -5,6 +5,7 @@ import { groupRouter } from "./api/routers/group";
 import { managerRouter } from "./api/routers/manager";
 import type { TrpcContext } from "./api/trpc";
 import { deletePrefix, photoObjectPrefix, headObject, originalObjectKey } from "@album/storage";
+import { runOnce } from "./operation-receipts";
 
 test.skipIf(process.env.POLISH_INTEGRATION !== "1")("group ownership and event-scoped submission approval", async () => {
   if (!["localhost", "127.0.0.1"].includes(new URL(process.env.DATABASE_URL!).hostname)) throw new Error("Local database required");
@@ -24,6 +25,13 @@ test.skipIf(process.env.POLISH_INTEGRATION !== "1")("group ownership and event-s
     await expect(groupRouter.createCaller(ctx(1)).rename({ groupId, name: "Denied" })).rejects.toThrow();
     const [event] = await db.insert(events).values({ groupId, title: "Workflow test", slug: id }).returning();
     await db.insert(eventMemberships).values({ eventId: event!.id, userId: people[0]!.id, role: "owner" });
+    const receipt = { userId: people[0]!.id, eventId: event!.id, key: `rollback-${id}` };
+    await expect(runOnce(receipt, "same", async tx => { await tx.update(events).set({ title: "Rolled back" }).where(eq(events.id, event!.id)); throw new Error("simulated interruption"); })).rejects.toThrow();
+    expect((await db.select().from(events).where(eq(events.id, event!.id)))[0]!.title).toBe("Workflow test");
+    let executions = 0;
+    const receipts = await Promise.all([0, 1, 2].map(() => runOnce(receipt, "same", async () => { executions++; return { ok: true }; })));
+    expect(executions).toBe(1);
+    expect(receipts).toEqual([{ ok: true }, { ok: true }, { ok: true }]);
     const [guest] = await db.insert(guests).values({ eventId: event!.id, tokenHash: id }).returning();
     const [submission] = await db.insert(submissions).values({ eventId: event!.id, guestId: guest!.id }).returning();
     await db.insert(photos).values((["ready", "processing"] as const).map(processingStatus => ({ eventId: event!.id, submissionId: submission!.id, originalKey: `test/${id}/${processingStatus}`, contentType: "image/jpeg", processingStatus })));
@@ -43,13 +51,18 @@ test.skipIf(process.env.POLISH_INTEGRATION !== "1")("group ownership and event-s
     const [preserved] = await db.select().from(photos).where(and(eq(photos.eventId, event!.id), eq(photos.id, privatePhoto!.id)));
     expect(preserved!.visibility).toBe("private");
     await expect(moderator.moderateSubmission({ ...input, visibility: "private" })).rejects.toThrow();
-    const bulk = { eventId: event!.id, photoIds: [...result.map(photo => photo.id), privatePhoto!.id], action: "public" as const };
+    const bulk = { eventId: event!.id, photoIds: [...result.map(photo => photo.id), privatePhoto!.id], action: "public" as const, requestId: crypto.randomUUID() };
     const preview = await moderator.previewBulkPhotos(bulk);
     expect(preview.affected).toBe(2);
     expect(preview.results.find(row => row.photoId === privatePhoto!.id)?.status).toBe("skipped");
     await expect(moderator.bulkPhotos({ ...bulk, confirm: false as true })).rejects.toThrow();
     expect((await moderator.bulkPhotos({ ...bulk, confirm: true })).affected).toBe(2);
-    expect((await moderator.bulkPhotos({ ...bulk, confirm: true })).affected).toBe(0);
+    expect((await moderator.bulkPhotos({ ...bulk, confirm: true })).affected).toBe(2);
+    await expect(moderator.bulkPhotos({ ...bulk, action: "hidden", confirm: true })).rejects.toThrow("different inputs");
+    // A replay cannot undo a later, intentional change.
+    await moderator.bulkPhotos({ ...bulk, requestId: crypto.randomUUID(), action: "hidden", confirm: true });
+    await moderator.bulkPhotos({ ...bulk, confirm: true });
+    expect((await db.select().from(photos).where(and(eq(photos.eventId, event!.id), eq(photos.id, result[0]!.id))))[0]!.visibility).toBe("hidden");
     await expect(moderator.bulkPhotos({ ...bulk, action: "private", confirm: true })).rejects.toThrow();
     const [otherEvent] = await db.insert(events).values({ groupId, title: "Other event", slug: `${id}-other` }).returning();
     await db.insert(eventMemberships).values({ eventId: otherEvent!.id, userId: people[0]!.id, role: "owner" });
@@ -58,21 +71,29 @@ test.skipIf(process.env.POLISH_INTEGRATION !== "1")("group ownership and event-s
     const deletion = await manager.previewBulkPhotos({ ...bulk, action: "delete" });
     expect(deletion.results.find(row => row.photoId === result.find(photo => photo.processingStatus === "processing")!.id)?.status).toBe("skipped");
     await expect(manager.bulkPhotos({ ...bulk, photoIds: [result[0]!.id, result[0]!.id], confirm: true })).rejects.toThrow();
-    await expect(moderator.createGalleryPhotos({ eventId: event!.id, files: [{ fileName: "test.png", contentType: "image/png", byteSize: 5 }] })).rejects.toThrow();
+    await expect(moderator.createGalleryPhotos({ eventId: event!.id, requestId: crypto.randomUUID(), files: [{ fileName: "test.png", contentType: "image/png", byteSize: 5 }] })).rejects.toThrow();
     expect((await moderator.photos({ eventId: event!.id, limit: 1, visibility: "private" })).length).toBe(0);
     expect((await manager.photos({ eventId: event!.id, limit: 1 })).length).toBe(1);
     if (process.env.GALLERY_STORAGE_INTEGRATION === "1") {
       if (!["localhost", "127.0.0.1"].includes(new URL(process.env.S3_ENDPOINT!).hostname)) throw new Error("Local object storage required");
       const bytes = Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aRZkAAAAASUVORK5CYII=", "base64");
-      const uploads = await manager.createGalleryPhotos({ eventId: event!.id, files: [{ fileName: "test.png", contentType: "image/png", byteSize: bytes.length }] });
+      const createInput = { eventId: event!.id, requestId: crypto.randomUUID(), files: [{ fileName: "test.png", contentType: "image/png" as const, byteSize: bytes.length }] };
+      const [uploads, replay] = await Promise.all([manager.createGalleryPhotos(createInput), manager.createGalleryPhotos(createInput)]);
+      expect(replay.map(row => row.photoId)).toEqual(uploads.map(row => row.photoId));
       for (const upload of uploads) cleanupUploads.push({ eventId: event!.id, photoId: upload.photoId });
-      expect((await fetch(uploads[0]!.uploadUrl, { method: "PUT", headers: { "Content-Type": "image/png" }, body: bytes })).ok).toBe(true);
-      const batch = { eventId: event!.id, photoIds: uploads.map(upload => upload.photoId) };
+      expect((await fetch(uploads[0]!.uploadUrl!, { method: "PUT", headers: { "Content-Type": "image/png" }, body: bytes })).ok).toBe(true);
+      expect((await manager.retryGalleryPhoto({ eventId: event!.id, photoId: uploads[0]!.photoId })).uploadUrl).toBeNull();
+      expect((await manager.createGalleryPhotos(createInput))[0]!.uploadUrl).toBeNull();
+      await expect(manager.createGalleryPhotos({ ...createInput, files: [{ ...createInput.files[0]!, byteSize: 1 }] })).rejects.toThrow("different inputs");
+      const batch = { eventId: event!.id, photoIds: uploads.map(upload => upload.photoId), requestId: crypto.randomUUID() };
       expect((await manager.completeGalleryPhotos(batch))[0]!.status).toBe("processing");
+      expect((await manager.completeGalleryPhotos(batch))[0]!.status).toBe("processing");
+      expect((await manager.retryGalleryPhoto({ eventId: event!.id, photoId: uploads[0]!.photoId })).uploadUrl).toBeNull();
       expect((await manager.previewBulkPhotos({ ...batch, action: "delete" })).affected).toBe(0);
       // Simulate the worker's terminal state; never modify a real upload.
       await db.update(photos).set({ processingStatus: "ready" }).where(and(eq(photos.eventId, event!.id), eq(photos.id, uploads[0]!.photoId)));
       expect((await manager.bulkPhotos({ ...batch, action: "delete", confirm: true })).affected).toBe(1);
+      expect((await manager.bulkPhotos({ ...batch, action: "delete", confirm: true })).affected).toBe(1);
       expect(await headObject(originalObjectKey(event!.id, uploads[0]!.photoId))).toBeNull();
     }
   } finally {