From f8309f4c771e629ca3911ff43fc914e0e8288058 Mon Sep 17 00:00:00 2001 From: styltsou Date: Thu, 23 Jul 2026 22:37:22 +0300 Subject: [PATCH 1/2] implement durable image processing --- client/src/api/collection/hooks.ts | 249 +- client/src/api/collection/types.ts | 7 + .../src/components/app-shell/app-sidebar.tsx | 16 +- .../components/board/board-upload-zone.tsx | 16 +- .../board/cards/image-asset-card.tsx | 13 +- .../board/cards/note-asset-card.tsx | 22 +- .../image-asset-viewer/image-metadata.tsx | 31 +- .../board/use-board-asset-actions.ts | 30 +- .../canvas/board-pointer-position.ts | 16 + client/src/components/canvas/canvas.tsx | 11 +- .../src/components/ui/progressive-image.tsx | 34 +- client/src/lib/asset-transform.ts | 1 + .../routes/$workspaceSlug/collections/$.tsx | 1 + client/src/sst-env.d.ts | 6 +- client/src/types/asset.ts | 2 + client/sst-env.d.ts | 4 +- .../migration.sql | 5 + .../snapshot.json | 3897 +++++++++++++++++ server/src/db/schema/app.ts | 13 + server/src/dto/collection.dto.ts | 2 + server/src/dto/upload.dto.test.ts | 12 +- server/src/dto/upload.dto.ts | 20 +- .../collection/collection-node-mappers.ts | 2 +- .../collection/collection-query.service.ts | 10 +- .../collection.service.integration.test.ts | 14 +- server/src/services/image-upload.service.ts | 245 +- .../image-upload/callback-state.test.ts | 8 +- .../services/image-upload/callback-state.ts | 11 +- .../image-upload/image-upload-finalizer.ts | 97 +- .../image-upload/pipeline-result.test.ts | 9 +- .../services/image-upload/pipeline-result.ts | 2 +- server/sst-env.d.ts | 4 +- services/image-pipeline/src/lambda.ts | 54 +- services/image-pipeline/src/processor.ts | 24 +- services/image-pipeline/sst-env.d.ts | 4 +- 35 files changed, 4540 insertions(+), 352 deletions(-) create mode 100644 client/src/components/canvas/board-pointer-position.ts create mode 100644 server/drizzle/20260723093146_friendly_black_queen/migration.sql create mode 100644 server/drizzle/20260723093146_friendly_black_queen/snapshot.json diff --git a/client/src/api/collection/hooks.ts b/client/src/api/collection/hooks.ts index e62fa4d..a0b792c 100644 --- a/client/src/api/collection/hooks.ts +++ b/client/src/api/collection/hooks.ts @@ -436,31 +436,23 @@ async function readImageDimensions(file: File): Promise<{ } } -async function makeOptimisticImageNode( +function makeOptimisticImageNode( file: File, index: number, -): Promise< - Extract & { - previewObjectUrl: string; - } -> { +): Extract & { + previewObjectUrl: string; +} { const previewObjectUrl = URL.createObjectURL(file); const id = `image-uploading-${Date.now()}-${index}-${Math.random().toString(36).slice(2)}`; - let dimensions: { width: number; height: number }; - - try { - dimensions = await readImageDimensions(file); - } catch { - // Keep the upload usable if the browser cannot decode a local preview. - dimensions = { width: 1, height: 1 }; - } return { id, type: "image", url: previewObjectUrl, - width: dimensions.width, - height: dimensions.height, + // Render immediately. The real dimensions replace this provisional square + // while the upload request is being prepared. + width: 1, + height: 1, title: file.name || null, alt: null, sourceLabel: null, @@ -485,12 +477,6 @@ function revokeOptimisticImageUrls( } } -function revokeOptimisticImageUrlsLater( - images: Array<{ previewObjectUrl?: string }> | undefined, -) { - window.setTimeout(() => revokeOptimisticImageUrls(images), 30_000); -} - function updateOptimisticImage( queryClient: ReturnType, workspaceSlug: string, @@ -1034,7 +1020,9 @@ export function useUploadLocalImages( collectionSlug, parentFolderPath, ); - await queryClient.cancelQueries({ queryKey: contentsKey }); + // The local preview is the completion point for this interaction. Do not + // wait for an in-flight refetch before placing it on the canvas. + void queryClient.cancelQueries({ queryKey: contentsKey }); const previousContents = queryClient.getQueryData(contentsKey); @@ -1045,9 +1033,7 @@ export function useUploadLocalImages( "workspace", workspaceSlug, ]); - const optimisticImages = await Promise.all( - files.map(makeOptimisticImageNode), - ); + const optimisticImages = files.map(makeOptimisticImageNode); const positions = reserveNodePositions( previousContents?.nodes ?? [], optimisticImages, @@ -1082,11 +1068,33 @@ export function useUploadLocalImages( optimisticImages.length, ); - const processingImages: Promise[] = []; + const imageDimensions = await Promise.all( + files.map(async (file) => { + try { + return await readImageDimensions(file); + } catch { + return { width: 1, height: 1 }; + } + }), + ); + for (const [index, dimensions] of imageDimensions.entries()) { + const optimisticImage = optimisticImages[index]!; + optimisticImage.width = dimensions.width; + optimisticImage.height = dimensions.height; + updateOptimisticImage( + queryClient, + workspaceSlug, + collectionSlug, + parentFolderPath, + optimisticImage.id, + dimensions, + ); + } + + const images: CollectionImageNode[] = []; const multiple = files.length > 1; const label = multiple ? `${files.length} images` : "1 image"; const toastId = toast.loading(`Uploading ${label}...`); - let cancelled = false; try { for (const [index, file] of files.entries()) { @@ -1098,6 +1106,8 @@ export function useUploadLocalImages( fileName: file.name || "clipboard-image.png", contentType: file.type, sizeBytes: file.size, + width: optimisticImage.width, + height: optimisticImage.height, parentFolderPath, position: optimisticImage.position ?? undefined, }, @@ -1122,81 +1132,51 @@ export function useUploadLocalImages( }, ); - updateOptimisticImage( - queryClient, - workspaceSlug, - collectionSlug, - parentFolderPath, - optimisticImage.id, - { uploadStatus: "processing", uploadProgress: 100 }, - ); - - processingImages.push( - waitForProcessedImage(() => - fetchImageUploadStatus( - workspaceSlug, - collectionSlug, - uploadData.upload.id, - ), - ).then((processedImage) => { - const image = { - ...processedImage, - clientId: optimisticImage.clientId, - position: optimisticImage.position, - }; - - if (cancelled) return image; - - queryClient.setQueryData( - contentsKey, - (current) => { - if (!current) return current; - - return { + const image = { + ...uploadData.upload.image, + clientId: optimisticImage.clientId, + position: optimisticImage.position, + localPreviewUrl: optimisticImage.previewObjectUrl, + }; + images.push(image); + queryClient.setQueryData( + contentsKey, + (current) => + !current + ? current + : { ...current, nodes: current.nodes.map((node) => node.id === optimisticImage.id ? image : node, ), - }; - }, - ); - - const preview: FolderChildPreview = { - assetId: image.id, - type: "image", - url: image.url, - blurDataURL: image.blurDataURL, - }; - addPreviewToCollection( - queryClient, - workspaceSlug, - collectionSlug, - preview, - ); - addPreviewToParentFolder( - queryClient, - workspaceSlug, - collectionSlug, - parentFolderPath, - preview, - 0, - ); - return image; - }), + }, + ); + const preview: FolderChildPreview = { + assetId: image.id, + type: "image", + url: image.url, + blurDataURL: image.blurDataURL, + }; + addPreviewToCollection( + queryClient, + workspaceSlug, + collectionSlug, + preview, + ); + addPreviewToParentFolder( + queryClient, + workspaceSlug, + collectionSlug, + parentFolderPath, + preview, + 0, ); } - const images = await Promise.all(processingImages); - - revokeOptimisticImageUrlsLater(optimisticImages); reconcileCollectionCaches(queryClient, workspaceSlug, collectionSlug); toast.success(`${label} uploaded`, { id: toastId }); return { images, parentFolderPath }; } catch (error) { - cancelled = true; - void Promise.allSettled(processingImages).then(() => { - reconcileCollectionCaches(queryClient, workspaceSlug, collectionSlug); - }); toast.error("Upload failed", { id: toastId }); revokeOptimisticImageUrls(optimisticImages); queryClient.setQueryData(contentsKey, previousContents); @@ -1228,15 +1208,15 @@ export function useUploadInboxImages(workspaceSlug: string) { mutationFn: async ({ files }: { files: File[] }) => { const inboxKey = collectionQueryKeys.inbox(workspaceSlug); const workspaceKey = ["workspace", workspaceSlug] as const; - await queryClient.cancelQueries({ queryKey: inboxKey }); + // Keep the modal-to-canvas handoff immediate; cancelling a refetch can + // happen in parallel with the optimistic cache update. + void queryClient.cancelQueries({ queryKey: inboxKey }); const previousInbox = queryClient.getQueryData(inboxKey); const previousWorkspace = queryClient.getQueryData(workspaceKey); - const optimisticImages = await Promise.all( - files.map(makeOptimisticImageNode), - ); + const optimisticImages = files.map(makeOptimisticImageNode); queryClient.setQueryData(inboxKey, (current) => { if (!current) return current; @@ -1252,11 +1232,31 @@ export function useUploadInboxImages(workspaceSlug: string) { (count) => count + files.length, ); - const processingImages: Promise[] = []; + const imageDimensions = await Promise.all( + files.map(async (file) => { + try { + return await readImageDimensions(file); + } catch { + return { width: 1, height: 1 }; + } + }), + ); + for (const [index, dimensions] of imageDimensions.entries()) { + const optimisticImage = optimisticImages[index]!; + optimisticImage.width = dimensions.width; + optimisticImage.height = dimensions.height; + updateOptimisticInboxImage( + queryClient, + workspaceSlug, + optimisticImage.id, + dimensions, + ); + } + + const images: CollectionImageNode[] = []; const multiple = files.length > 1; const label = multiple ? `${files.length} images` : "1 image"; const toastId = toast.loading(`Uploading ${label}...`); - let cancelled = false; try { for (const [index, file] of files.entries()) { @@ -1265,6 +1265,8 @@ export function useUploadInboxImages(workspaceSlug: string) { fileName: file.name || "clipboard-image.png", contentType: file.type, sizeBytes: file.size, + width: optimisticImage.width, + height: optimisticImage.height, }); await uploadFileToPresignedUrl( @@ -1283,50 +1285,27 @@ export function useUploadInboxImages(workspaceSlug: string) { ); }, ); - updateOptimisticInboxImage( - queryClient, - workspaceSlug, - optimisticImage.id, - { uploadStatus: "processing", uploadProgress: 100 }, - ); - - processingImages.push( - waitForProcessedImage(() => - fetchInboxImageUploadStatus(workspaceSlug, uploadData.upload.id), - ).then((processedImage) => { - const image = { - ...processedImage, - clientId: optimisticImage.clientId, - }; - - if (cancelled) return image; - - queryClient.setQueryData( - inboxKey, - (current) => { - if (!current) return current; - - return { - ...current, - nodes: current.nodes.map((node) => - node.id === optimisticImage.id ? image : node, - ), - }; + const image = { + ...uploadData.upload.image, + clientId: optimisticImage.clientId, + localPreviewUrl: optimisticImage.previewObjectUrl, + }; + images.push(image); + queryClient.setQueryData(inboxKey, (current) => + !current + ? current + : { + ...current, + nodes: current.nodes.map((node) => + node.id === optimisticImage.id ? image : node, + ), }, - ); - return image; - }), ); } - const images = await Promise.all(processingImages); - toast.success(`${label} uploaded`, { id: toastId }); - revokeOptimisticImageUrlsLater(optimisticImages); return { images }; } catch (error) { - cancelled = true; - void Promise.allSettled(processingImages); toast.error("Upload failed", { id: toastId }); revokeOptimisticImageUrls(optimisticImages); queryClient.setQueryData(inboxKey, previousInbox); diff --git a/client/src/api/collection/types.ts b/client/src/api/collection/types.ts index 0ebd7db..deb741c 100644 --- a/client/src/api/collection/types.ts +++ b/client/src/api/collection/types.ts @@ -96,6 +96,8 @@ export type CollectionImageNode = { id: string; type: "image"; url: string; + /** Browser-only preview retained while the uploaded original is decoded. */ + localPreviewUrl?: string; originalUrl?: string; originalWidth?: number; originalHeight?: number; @@ -108,6 +110,8 @@ export type CollectionImageNode = { isFavorite: boolean; blurDataURL?: string | null; dominantColors?: string[]; + variantStatus?: "processing" | "completed" | "failed"; + paletteStatus?: "processing" | "completed" | "failed"; uploadStatus?: "uploading" | "processing"; uploadProgress?: number; clientId?: string; @@ -136,6 +140,8 @@ export type CreateImageUploadInput = { fileName: string; contentType: string; sizeBytes: number; + width: number; + height: number; title?: string; alt?: string; parentFolderPath?: string; @@ -150,6 +156,7 @@ export type CreateImageUploadResponse = { headers: Record; expiresAt: string; maxSizeBytes: number; + image: CollectionImageNode; }; }; diff --git a/client/src/components/app-shell/app-sidebar.tsx b/client/src/components/app-shell/app-sidebar.tsx index 179655c..17f8e7b 100644 --- a/client/src/components/app-shell/app-sidebar.tsx +++ b/client/src/components/app-shell/app-sidebar.tsx @@ -4,6 +4,7 @@ import { NavProjects } from "@/components/app-shell/nav-projects"; import { NavSecondary } from "@/components/app-shell/nav-secondary"; import { NavUser } from "@/components/app-shell/nav-user"; import { WorkspaceSwitcher } from "@/components/app-shell/workspace-switcher"; +import { Skeleton } from "@/components/ui/skeleton"; import { Sidebar, SidebarContent, @@ -29,7 +30,7 @@ import { import { openSettings } from "@/lib/settings-dialog"; export function AppSidebar(props: React.ComponentProps) { - const { data: session } = useSession(); + const { data: session, isPending } = useSession(); const pathname = useRouterState({ select: (state) => state.location.pathname, }); @@ -169,7 +170,18 @@ export function AppSidebar(props: React.ComponentProps) { - {session?.user ? : null} + {isPending ? ( +
+ +
+ + +
+ +
+ ) : session?.user ? ( + + ) : null} ); diff --git a/client/src/components/board/board-upload-zone.tsx b/client/src/components/board/board-upload-zone.tsx index a4cd6f7..956ab50 100644 --- a/client/src/components/board/board-upload-zone.tsx +++ b/client/src/components/board/board-upload-zone.tsx @@ -1,6 +1,9 @@ -import React, { useState } from "react"; +import React, { useCallback, useState } from "react"; import { ImagePlusIcon, LoaderCircleIcon } from "lucide-react"; +import type { BoardInsertionPlacement } from "@/api/collection"; +import { getBoardPointerPosition } from "@/components/canvas/board-pointer-position"; import { SUPPORTED_IMAGE_MIME_TYPE_SET } from "@/constants"; +import { useTransientStore } from "@/store"; import { cn, parseHttpUrl } from "@/lib/utils"; import { useBoardAssetActions } from "./use-board-asset-actions"; @@ -8,13 +11,23 @@ export function BoardUploadZone({ workspaceSlug, collectionPath, target = "collection", + boardKey, children, }: { workspaceSlug: string; collectionPath: string; target?: "collection" | "inbox"; + boardKey?: string; children: React.ReactNode; }) { + const getPlacement = useCallback((): BoardInsertionPlacement | undefined => { + if (!boardKey) return undefined; + + const { boardVisibleBounds } = useTransientStore.getState(); + const position = getBoardPointerPosition(boardKey); + const visibleBounds = boardVisibleBounds[boardKey]; + return position || visibleBounds ? { position, visibleBounds } : undefined; + }, [boardKey]); const { createTextNote, importRemoteUrl, @@ -25,6 +38,7 @@ export function BoardUploadZone({ workspaceSlug, collectionPath, target, + getPlacement, }); const [isDraggingImage, setIsDraggingImage] = useState(false); diff --git a/client/src/components/board/cards/image-asset-card.tsx b/client/src/components/board/cards/image-asset-card.tsx index cb8e4e2..6e135ca 100644 --- a/client/src/components/board/cards/image-asset-card.tsx +++ b/client/src/components/board/cards/image-asset-card.tsx @@ -46,6 +46,7 @@ export function ImageAssetCard({ > -
- {asset.title && {asset.title}} +
+ {asset.title && ( + {asset.title} + )} {asset.sourceLabel && ( <> {asset.title && ( @@ -87,10 +90,10 @@ export function ImageAssetCard({ target="_blank" rel="noopener noreferrer" onClick={(event) => event.stopPropagation()} - className="inline-flex items-center gap-1 transition-colors duration-100 ease-[cubic-bezier(0.16,1,0.3,1)] hover:text-sidebar-foreground/70" + className="inline-flex min-w-0 items-center gap-1 transition-colors duration-100 ease-[cubic-bezier(0.16,1,0.3,1)] hover:text-sidebar-foreground/70" > - - {asset.sourceLabel} + + {asset.sourceLabel} )} diff --git a/client/src/components/board/cards/note-asset-card.tsx b/client/src/components/board/cards/note-asset-card.tsx index 7658603..c701d05 100644 --- a/client/src/components/board/cards/note-asset-card.tsx +++ b/client/src/components/board/cards/note-asset-card.tsx @@ -6,6 +6,15 @@ import { cn } from "@/lib/utils"; import { hasSelectionModifier } from "@/lib/selection"; import type { NoteAsset } from "@/types/asset"; +const BARE_URL_RE = /(^|[^(\[])(https?:\/\/[^\s<"'>)\]]+)/gi; + +function linkifyBareUrls(text: string): string { + return text.replace( + BARE_URL_RE, + (_, before, url) => `${before}[${url}](${url})`, + ); +} + const MD_COMPONENTS: Components = { h1: ({ className, ...props }) => (

( ), @@ -123,7 +134,9 @@ export function NoteMarkdown({ }) { return (
- {content} + + {linkifyBareUrls(content)} +
); } @@ -204,7 +217,10 @@ export function NoteAssetCard({ effectiveOnOpen(); }} > -
+
{hasOverflow ? ( diff --git a/client/src/components/board/image-asset-viewer/image-metadata.tsx b/client/src/components/board/image-asset-viewer/image-metadata.tsx index 0bf3ef6..3a039f1 100644 --- a/client/src/components/board/image-asset-viewer/image-metadata.tsx +++ b/client/src/components/board/image-asset-viewer/image-metadata.tsx @@ -1,5 +1,11 @@ import { Fragment } from "react"; +import { toast } from "sonner"; import type { ImageAsset } from "@/types/asset"; +import { + Tooltip, + TooltipContent, + TooltipTrigger, +} from "@/components/ui/tooltip"; function formatSize(bytes: number): string { if (bytes < 1024) return `${bytes} B`; @@ -62,13 +68,24 @@ export function ImageMetadata({ asset }: { asset: ImageAsset }) { {dominantColors.length > 0 ? (
{dominantColors.map((color) => ( - + + { + navigator.clipboard.writeText(color.toUpperCase()); + toast.success(`Copied ${color.toUpperCase()}`); + }} + aria-label={`Dominant color ${color}`} + className="cursor-pointer" + > + + + + {color.toUpperCase()} + + ))}
) : ( diff --git a/client/src/components/board/use-board-asset-actions.ts b/client/src/components/board/use-board-asset-actions.ts index f9c782a..f0dec36 100644 --- a/client/src/components/board/use-board-asset-actions.ts +++ b/client/src/components/board/use-board-asset-actions.ts @@ -21,11 +21,13 @@ export function useBoardAssetActions({ collectionPath, target = "collection", placement, + getPlacement, }: { workspaceSlug: string; collectionPath: string; target?: BoardAssetTarget; placement?: BoardInsertionPlacement; + getPlacement?: () => BoardInsertionPlacement | undefined; }) { const [collectionSlug = "", ...folderSegments] = collectionPath .split("/") @@ -71,6 +73,7 @@ export function useBoardAssetActions({ if (imageFiles.length === 0) return; try { + const insertionPlacement = getPlacement?.() ?? placement; if (target === "inbox") { await uploadInboxImages.mutateAsync({ files: imageFiles, @@ -79,7 +82,7 @@ export function useBoardAssetActions({ await uploadLocalImages.mutateAsync({ files: imageFiles, parentFolderPath, - placement, + placement: insertionPlacement, }); } toast.success( @@ -91,7 +94,14 @@ export function useBoardAssetActions({ ); } }, - [parentFolderPath, placement, target, uploadInboxImages, uploadLocalImages], + [ + getPlacement, + parentFolderPath, + placement, + target, + uploadInboxImages, + uploadLocalImages, + ], ); const importRemoteUrl = useCallback( @@ -100,6 +110,7 @@ export function useBoardAssetActions({ if (!url) return; try { + const insertionPlacement = getPlacement?.() ?? placement; if (target === "inbox") { await createInboxRemoteImage.mutateAsync({ url, @@ -108,7 +119,7 @@ export function useBoardAssetActions({ await createRemoteImage.mutateAsync({ url, parentFolderPath, - placement, + placement: insertionPlacement, }); } toast.success("Image imported"); @@ -121,6 +132,7 @@ export function useBoardAssetActions({ [ createInboxRemoteImage, createRemoteImage, + getPlacement, parentFolderPath, placement, target, @@ -132,6 +144,7 @@ export function useBoardAssetActions({ if (!content.trim()) return; try { + const insertionPlacement = getPlacement?.() ?? placement; if (target === "inbox") { await createInboxNote.mutateAsync({ content, @@ -140,7 +153,7 @@ export function useBoardAssetActions({ await createNote.mutateAsync({ content, parentFolderPath, - placement, + placement: insertionPlacement, }); } toast.success("Note created"); @@ -150,7 +163,14 @@ export function useBoardAssetActions({ ); } }, - [createInboxNote, createNote, parentFolderPath, placement, target], + [ + createInboxNote, + createNote, + getPlacement, + parentFolderPath, + placement, + target, + ], ); const addClipboardAsset = useCallback( diff --git a/client/src/components/canvas/board-pointer-position.ts b/client/src/components/canvas/board-pointer-position.ts new file mode 100644 index 0000000..57ac47c --- /dev/null +++ b/client/src/components/canvas/board-pointer-position.ts @@ -0,0 +1,16 @@ +import type { BoardPosition } from "@/api/collection"; + +// Pointer movement is high-frequency. Keep it outside React/Zustand state so +// paste can read the latest location without re-rendering the canvas. +const positions = new Map(); + +export function setBoardPointerPosition( + boardKey: string, + position: BoardPosition, +) { + positions.set(boardKey, position); +} + +export function getBoardPointerPosition(boardKey: string) { + return positions.get(boardKey); +} diff --git a/client/src/components/canvas/canvas.tsx b/client/src/components/canvas/canvas.tsx index 687c1f6..dc77f35 100644 --- a/client/src/components/canvas/canvas.tsx +++ b/client/src/components/canvas/canvas.tsx @@ -44,6 +44,7 @@ import { toast } from "sonner"; import { formatPlatformShortcut } from "@/lib/platform"; import { makeBoardKey } from "./canvas-key"; +import { setBoardPointerPosition } from "./board-pointer-position"; import { BOARD_CARD_WIDTH, arrangeNodesInGrid, @@ -723,7 +724,15 @@ function CanvasSurface({ ref={boardRef} className="relative h-full min-h-0 w-full bg-transparent" onPointerDownCapture={marquee.onPointerDownCapture} - onPointerMoveCapture={marquee.onPointerMoveCapture} + onPointerMoveCapture={(event) => { + marquee.onPointerMoveCapture(event); + setBoardPointerPosition( + boardKey, + roundPosition( + screenToFlowPosition({ x: event.clientX, y: event.clientY }), + ), + ); + }} onPointerUpCapture={marquee.onPointerUpCapture} onPointerCancelCapture={marquee.onPointerCancelCapture} onClickCapture={(event) => { diff --git a/client/src/components/ui/progressive-image.tsx b/client/src/components/ui/progressive-image.tsx index ad47c51..c7b8323 100644 --- a/client/src/components/ui/progressive-image.tsx +++ b/client/src/components/ui/progressive-image.tsx @@ -1,4 +1,4 @@ -import { useCallback, useState, type ComponentProps } from "react"; +import { useCallback, useEffect, useState, type ComponentProps } from "react"; import { motion, type MotionStyle } from "motion/react"; import { cn } from "@/lib/utils"; @@ -7,6 +7,8 @@ type ProgressiveImageProps = Omit< "src" | "style" > & { src: string; + /** A local Blob preview to keep visible while `src` finishes decoding. */ + fallbackSrc?: string; blurDataURL?: string | null; placeholderClassName?: string; style?: MotionStyle; @@ -26,6 +28,7 @@ function rememberDecodedSource(src: string) { export function ProgressiveImage({ src, + fallbackSrc, blurDataURL, alt = "", className, @@ -40,6 +43,15 @@ export function ProgressiveImage({ decodedSources.has(src) ? src : null, ); const isDecoded = decodedSrc === src || decodedSources.has(src); + const showFallback = Boolean(fallbackSrc) && !isDecoded; + + useEffect(() => { + if (!isDecoded || !fallbackSrc?.startsWith("blob:")) return; + + // Let the decoded remote image paint once before releasing the local Blob. + const frame = requestAnimationFrame(() => URL.revokeObjectURL(fallbackSrc)); + return () => cancelAnimationFrame(frame); + }, [fallbackSrc, isDecoded]); const handleLoad = useCallback>( (event) => { @@ -66,14 +78,15 @@ export function ProgressiveImage({ >( (event) => { onError?.(event); - setDecodedSrc(src); + // A failed signed URL should not hide the still-valid local preview. + if (!fallbackSrc) setDecodedSrc(src); }, - [onError, src], + [fallbackSrc, onError, src], ); return ( <> - {blurDataURL && !isDecoded ? ( + {blurDataURL && !showFallback && !isDecoded ? ( <> ) : null} + {fallbackSrc ? ( +

- {statusText ? ( -
- - {statusText} -
- ) : null} {isPending ? ( {statusText} diff --git a/client/src/components/board/image-asset-viewer/image-metadata.tsx b/client/src/components/board/image-asset-viewer/image-metadata.tsx index 3a039f1..c5a2746 100644 --- a/client/src/components/board/image-asset-viewer/image-metadata.tsx +++ b/client/src/components/board/image-asset-viewer/image-metadata.tsx @@ -78,7 +78,7 @@ export function ImageMetadata({ asset }: { asset: ImageAsset }) { className="cursor-pointer" > diff --git a/client/src/components/board/use-board-asset-actions.ts b/client/src/components/board/use-board-asset-actions.ts index f0dec36..d192a8b 100644 --- a/client/src/components/board/use-board-asset-actions.ts +++ b/client/src/components/board/use-board-asset-actions.ts @@ -85,9 +85,6 @@ export function useBoardAssetActions({ placement: insertionPlacement, }); } - toast.success( - imageFiles.length === 1 ? "Image uploaded" : "Images uploaded", - ); } catch (err) { toast.error( err instanceof Error ? err.message : "Unable to upload images.", diff --git a/client/src/components/ui/sonner.tsx b/client/src/components/ui/sonner.tsx index bdd3742..03e3b4d 100644 --- a/client/src/components/ui/sonner.tsx +++ b/client/src/components/ui/sonner.tsx @@ -30,6 +30,7 @@ const Toaster = ({ ...props }: ToasterProps) => { "--border-radius": "var(--radius)", } as React.CSSProperties } + position="top-right" toastOptions={{ classNames: { toast: "cn-toast", diff --git a/client/src/lib/auth-client.ts b/client/src/lib/auth-client.ts index ba51200..f6dd48b 100644 --- a/client/src/lib/auth-client.ts +++ b/client/src/lib/auth-client.ts @@ -6,6 +6,11 @@ const DEFAULT_SERVER_URL = "http://localhost:3000"; export const authClient = createAuthClient({ baseURL: import.meta.env.VITE_SERVER_URL ?? DEFAULT_SERVER_URL, plugins: [organizationClient()], + sessionOptions: { + // Route guards own session refreshes. Avoid network requests just because + // a user switches back to this browser tab. + refetchOnWindowFocus: false, + }, }); export const { signIn, signOut, signUp, useSession } = authClient; diff --git a/docs/server/assets-schema.md b/docs/server/assets-schema.md index 3aaeb2d..a9d9d68 100644 --- a/docs/server/assets-schema.md +++ b/docs/server/assets-schema.md @@ -34,7 +34,7 @@ while those URLs decode. `uploads` is the durable asynchronous ingestion workflow. It stores the target collection/folder, source metadata, original object key, lifecycle status, processing ETag, terminal error, and final asset ID. It is not an asset and is -created before an original is written to R2. +created before an original is written to S3. `image_colors` is the searchable palette table. It stores `organization_id`, the display-ready hex value, indexed OKLab coordinates, `coverage`, `salience`, diff --git a/docs/server/image-pipeline-reliability.md b/docs/server/image-pipeline-reliability.md index f3f8be2..d0c2666 100644 --- a/docs/server/image-pipeline-reliability.md +++ b/docs/server/image-pipeline-reliability.md @@ -1,160 +1,53 @@ -# Image Pipeline Reliability and Evolution +# Image Pipeline Reliability -This document records the reliability model for asynchronous image processing, -the tradeoffs in the current implementation, and the next architecture to adopt -when processing volume or API availability justifies the additional moving -parts. - -The Hono API is expected to remain an independently deployed server. The -pipeline therefore uses authenticated HTTPS callbacks rather than Cloudflare -Service Bindings. - -## Goals - -- Keep browser uploads independent of image-processing duration. -- Preserve work across Worker, API, and network failures. -- Make every state transition observable and recoverable. -- Ensure duplicates cannot create duplicate image assets. -- Avoid repeating CPU-heavy processing merely because the API is temporarily - unavailable once that cost becomes material. - -## Current Architecture +Image uploads use two independent, at-least-once SQS workflows. The API creates +the `assets`, `image_assets`, and `uploads` rows before the original reaches +S3, so neither processor depends on the other completing first. ```txt -R2 ingest/ event - -> image-processing Queue - -> pipeline Worker: variants + palette - -> authenticated HTTPS callback to Hono - -> Hono transaction: upload + asset + palette rows +browser or remote import -> S3 ingest/ object-created event + ├-> ImageVariantsQueue -> variants Lambda + └-> ImagePaletteQueue -> palette Lambda ``` -The pipeline Worker acknowledges its source Queue message only after Hono has -accepted the callback. Consequently, an API outage does not lose a completed -processing result: Cloudflare Queues redelivers the source event and the Worker -tries again. - -This is an **at-least-once** workflow. A source event, generated R2 objects, -and callback can each occur more than once. The design is safe because: - -- variant object keys are deterministic for a storage ID; -- rewriting a variant is harmless; -- Hono identifies the upload by original object key and R2 ETag; -- completed callbacks are idempotent and create no duplicate assets. - -Cloudflare Queues provides the durable delivery and retry mechanism. The -Worker's explicit delayed retries control per-message backoff and the point at -which Hono receives a terminal `failed` state. See the [Cloudflare Queues retry -and delay documentation](https://developers.cloudflare.com/queues/configuration/batching-retries/) -for the platform behavior. - -## Deliberate Current Tradeoff - -The simple topology has one important cost: if variants and palette extraction -succeed but the Hono callback cannot be delivered, a Queue retry runs the image -processor again. It is correct and safe, but may repeat decoding, resizing, and -palette extraction when only the control-plane request failed. - -This is an appropriate starting point because it has one Queue, one Worker, -and one idempotent API boundary. It keeps the failure model easy to reason -about while processing volume is low. It is not a loss of Queue reliability; -the Queue is precisely what prevents the callback failure from losing the job. - -The tradeoff becomes worth removing when repeated image computation, callback -outages, operational cost, or processing latency become significant. - -## Production-Grade Target - -At that point, split processing from API finalization with a second durable -Queue: - -```txt -R2 ingest/ event - -> image-processing Queue - -> processor Worker: variants + palette + completion manifest in R2 - -> image-finalization Queue - -> finalizer Worker: authenticated HTTPS callback to Hono - -> Hono transaction: upload + asset + palette rows -``` - -### Processor Stage - -1. Read the original from `ingest/` and generate deterministic variants under - `assets/{storageId}/`. -2. Write an idempotent completion manifest, for example - `assets/{storageId}/manifest.json`. It contains original key/ETag, image - metadata, variants, palette, extraction version, and blur data URL. -3. Publish a small finalization message containing the source identity and - manifest key to `image-finalization`. A terminal failure event instead - carries the source identity and bounded error message. -4. Acknowledge the source message only after that publish succeeds. - -If the processor is redelivered between any of these steps, it overwrites the -same deterministic objects or emits a duplicate finalization message. Both are -safe. The queue publish is the durability handoff that separates expensive -computation from delivery to Hono. - -### Finalizer Stage - -1. Read the manifest from R2. -2. Send the signed callback to Hono. -3. Acknowledge the finalization message only after Hono returns success. - -An API outage now retries only the finalizer, not Sharp processing. The -finalizer refreshes the callback timestamp and HMAC signature for every -delivery. Hono retains the same idempotency checks because duplicate messages -are expected. - -### Failure Policy - -- Keep a small, explicit retry budget for processing errors that are unlikely - to recover, then publish a terminal `failed` finalization event. -- Let the finalization Queue own API-delivery retries independently, with a - dead-letter queue configured for exhaustion. -- Treat transient responses, timeouts, and `429` responses as retryable. -- Treat invalid callback payloads or unexpected permanent `4xx` responses as - operational incidents: record them, move the message to the DLQ, and alert. -- Provide a replay command that can re-enqueue a finalization message from its - manifest without recomputing variants. - -## Operational Invariants +Both notifications are filtered to the `ingest/` prefix. The workers read the +same immutable original object identity (object key plus ETag) but own separate +effects: -The production target enforces these invariants; the current design already -enforces the storage, idempotency, and durable-handoff invariants: +- The variants worker writes deterministic display and preview WebP objects, + then marks `image_assets.variant_status` complete. +- The palette worker calculates and persists `image_colors`, + `dominant_colors`, and `image_assets.palette_status`. -| Invariant | Reason | -| -------------------------------------------------------------------------- | ------------------------------------------------------------- | -| Original key and ETag identify one processing generation. | Prevents an old event from finalizing a replaced source. | -| Variant and manifest keys are deterministic. | Makes retries overwrite-safe and easy to inspect. | -| Hono completion is idempotent. | Required for at-least-once Queue delivery. | -| A message is acknowledged only after its next durable handoff. | Prevents silent loss between stages. | -| Every log, metric, and callback includes the storage ID and original ETag. | Enables tracing and targeted replay. | -| DLQ messages are actionable and replayable. | Converts an exhausted retry budget into an operable incident. | +## Delivery and retries -## Reconciliation +Both consumers use a batch size of one and report partial batch failures. SQS +redelivers a failed message after the 180-second visibility timeout. The worker +retries processing for the first four receives; on receive five it sends the +matching terminal callback (`image.variants.failed` or +`image.palette.failed`). If that terminal callback cannot reach the API, the +message remains retryable for one more receive and is then retained in that +consumer's DLQ. -Even a well-designed at-least-once system needs repair tooling. Run a periodic -reconciliation job that finds uploads stuck in `uploaded` or `processing` past -an expected threshold. It should inspect the original and, when present, the -completion manifest, then either re-enqueue the appropriate source/finalization -message or mark the upload failed with an auditable reason. +This is at-least-once delivery. Duplicate S3 events, SQS deliveries, object +writes, and callbacks are expected. The system is safe because variant keys +are deterministic, palette writes replace the asset's existing colors, and API +callbacks are keyed by original object key and ETag. -Reconciliation is a safety net, not the normal path. Its existence makes the -workflow recoverable after configuration mistakes, accidental Queue deletion, -or manual operational intervention. +## Operational model -## Adoption Trigger +The two DLQs retain messages for 14 days. An item in either DLQ means the API +could not receive the terminal callback after image processing failed; it needs +investigation and, once understood, replay from the original S3 object. -Keep the current one-Queue design until measurements show repeated processing -caused by callback failures or the retry cost is meaningful. Introduce the -finalization Queue when at least one of these is true: +Pipeline logs include the worker name, source key, SQS receive count, and error +message. Monitor retries and DLQ depth separately for variants and palettes: +a palette incident must not delay rendering or create extra resize work. -- API availability is materially lower than pipeline availability. -- Image processing is expensive enough that duplicate work affects cost or - queue latency. -- The service needs a formal DLQ and operator replay workflow. -- Multiple downstream consumers need the completed-image event. +## Extending the pipeline -The migration is additive: first introduce manifests and idempotent finalizer -callbacks, then route new completions through the finalization Queue. Existing -source events can continue using the current callback path until the new stage -has been observed in production. +New work that only needs the original upload can receive its own S3-to-SQS +notification and run in parallel, just as palette extraction does. Introduce +an event bus only when several independent consumers make direct notification +rules hard to manage or when routing needs content-based rules. It is not +needed for the current two-worker topology. diff --git a/docs/server/image-upload-implementation-plan.md b/docs/server/image-upload-implementation-plan.md index dbc2d81..9c4eba2 100644 --- a/docs/server/image-upload-implementation-plan.md +++ b/docs/server/image-upload-implementation-plan.md @@ -5,20 +5,20 @@ the earlier synchronous server-side processing plan. ## Ownership -The browser owns transferring a local source file to S3. The image-pipeline -Lambda owns CPU-bound post-processing. The Hono API owns authorization, upload -records, and the transactional database write that makes an image visible in a -collection. +The browser owns transferring a local source file to S3. Two independent +image-pipeline Lambdas own CPU-bound post-processing: one generates variants +and the other extracts the palette. The Hono API owns authorization, upload +records, and the asset rows that are visible in a collection before enrichment +finishes. ```txt Browser or remote URL - -> Hono creates uploads row + -> Hono creates upload + image asset rows -> original written to S3 ingest/ - -> S3 ingest/ object-created event -> SQS - -> image-pipeline Lambda - -> S3 assets/ variants + signed Hono callback - -> asset, image_assets, image_colors, collection_nodes transaction - -> client polls upload status and renders the completed image + -> S3 ingest/ object-created event -> variants SQS + palette SQS + -> variants Lambda -> S3 assets/ variants + signed Hono callback + -> palette Lambda -> image_colors + signed Hono callback + -> client observes independent enrichment states ``` The main server must not decode images, generate variants, or extract colors. @@ -51,15 +51,15 @@ pending -> uploaded -> processing -> completed confirmed yet. - `uploaded`: the browser or remote import stored the original; it is awaiting the event consumer. -- `processing`: the Worker authenticated itself to Hono and began processing. -- `completed`: Hono persisted the image asset and optional collection node in - one transaction. -- `failed`: processing exhausted its application retries or storing a remote - source failed. `error_message` holds the recorded diagnostic. +- `processing`: the variants worker has begun rendering. +- `completed`: the variants worker completed; palette extraction may still be + processing, completed, or failed on the associated image asset. +- `failed`: variant processing exhausted its application retries or storing a + remote source failed. `error_message` holds the recorded diagnostic. -Optional title/alt metadata is supplied when the upload session is created, so -the Worker callback has everything needed to persist the asset. A successful -R2 PUT is the only event needed to start processing. +Optional title/alt metadata is supplied when the upload session is created, and +the associated image asset is persisted in the same transaction. A successful +S3 PUT is the only event needed to start both background jobs. ## API Contract @@ -71,13 +71,13 @@ GET /api/v1/workspace/:workspaceSlug/collections/:collectionSlug/images/uploads POST /api/v1/workspace/:workspaceSlug/collections/:collectionSlug/images/remote ``` -The create-direct response contains the server-generated R2 object key and a -presigned PUT URL. The client PUTs directly to R2, then polls the status +The create-direct response contains the server-generated S3 object key and a +presigned PUT URL. The client PUTs directly to S3, then polls the status endpoint; it does not make a second request to begin processing. Remote imports return `{ upload }` with a lifecycle status, not an image node. The status endpoint includes `image` only after completion. -The client reports exact R2 transfer progress from `0` through `100`. Once the +The client reports exact S3 transfer progress from `0` through `100`. Once the PUT succeeds, the tile changes to an indeterminate processing state while it polls every second for up to two minutes. This separates byte-transfer progress from asynchronous Worker time and allows later files to upload while earlier @@ -85,8 +85,9 @@ files process. ## Image pipeline callback -The image pipeline Lambda sends `processing`, `completed`, or terminal `failed` -callbacks to: +The variants Lambda sends `image.processing.started`, +`image.variants.completed`, or `image.variants.failed`. The palette Lambda +sends `image.palette.completed` or `image.palette.failed` independently to: ```txt POST /api/v1/internal/image-pipeline/callback @@ -103,13 +104,13 @@ Hono rejects missing, malformed, stale, or invalid signatures. Both functions use the same `IMAGE_PIPELINE_CALLBACK_SECRET` value. The callback identifies an upload by `originalObjectKey` and guards it with the -source S3 ETag. Completion is idempotent: duplicate callbacks for an already -completed upload succeed without creating another asset. Hono also validates -that received variants use the exact expected `assets/{storageId}/...` keys. +source S3 ETag. Completion is idempotent: duplicate callbacks safely overwrite +the same enrichment result. Hono also validates that received variants use the +exact expected `assets/{storageId}/...` keys. ## Variants and Rendering -The Worker generates these non-upscaled WebP variants: +The variants Lambda generates these non-upscaled WebP variants: | Role | Maximum width | Use | | --------- | ------------: | ----------------------------------- | @@ -148,7 +149,7 @@ with two complementary passes: Use OKLab distance as the primary search criterion. Use coverage and salience only as secondary ranking signals. `image_assets.dominant_colors` is a compact display cache sorted by salience; do not use it as the search source of truth. -The upload finalizer writes the upload's organization ID into every palette row +The palette callback writes the upload's organization ID into every palette row so the tenant-first GiST index can bound color candidate retrieval before the asset join. The endpoint contract, collection scopes, multi-color matching, and relevance @@ -157,30 +158,18 @@ cutoffs are specified in the ## Failure, Retry, and Observability -The Worker records structured lifecycle logs. It retries ordinary processing -failures with bounded backoff for the first two deliveries. On a processing -failure at the third delivery, it sends a signed `failed` callback and -acknowledges the Queue message only after the API accepts that terminal state. -This prevents clients from polling an upload that can no longer complete. - -This processing budget is intentionally separate from the Queue consumer's -`max_retries` setting, which must be higher. The remaining Queue deliveries are -reserved for a transient failure while delivering the terminal callback, such -as an API outage. Configure a dead-letter queue or an operational replay path -for the exceptional case where Queue delivery is exhausted before that callback -is accepted. - -The original and any already-written variants remain in R2 on failure for -debugging and replay. No `assets`, `image_assets`, `image_colors`, or -`collection_nodes` rows are created until the completed callback transaction -succeeds. - -The current one-Queue topology deliberately retries processing when the API -callback is unavailable. This is safe and simple, but can repeat expensive -work. [Image Pipeline Reliability and Evolution](./image-pipeline-reliability.md) -documents the delivery guarantees, operational invariants, and the two-Queue -architecture that isolates callback retries from image processing when scale -justifies it. +Each worker records structured lifecycle logs and retries only its own SQS +message. It retries processing for the first four receives and reports the +matching terminal `failed` callback on receive five. A sixth queue receive is +reserved for retrying a terminal callback that could not reach the API; then +the message is retained in that worker's DLQ for investigation and replay. + +The original and any already-written variants remain in S3 on failure for +debugging and replay. Assets and collection placement are created before the +upload; failures update their separate variant or palette enrichment state. + +[Image Pipeline Reliability](./image-pipeline-reliability.md) documents the +delivery guarantees and operational invariants. ## Configuration and Deployment diff --git a/docs/server/scaffold-recipes.md b/docs/server/scaffold-recipes.md index 7c2b9b9..075e66f 100644 --- a/docs/server/scaffold-recipes.md +++ b/docs/server/scaffold-recipes.md @@ -45,8 +45,9 @@ coding agents. 1. Keep the request-facing API responsible for authorization and durable upload state; do not move collection writes into the Worker. -2. Use distinct R2 prefixes for source objects and generated objects. -3. Configure the R2 event notification with the source prefix only. +2. Use distinct S3 prefixes for source objects and generated objects. +3. Configure S3 event notifications with the source prefix only, and give + independent processors separate queues when their retry behavior differs. 4. Authenticate Worker callbacks over the raw payload and make the persistence transaction idempotent. 5. Persist search-oriented extraction data independently from UI display caches. diff --git a/docs/server/schema-design-rationale.md b/docs/server/schema-design-rationale.md index 4a5cd13..3a334b1 100644 --- a/docs/server/schema-design-rationale.md +++ b/docs/server/schema-design-rationale.md @@ -47,13 +47,14 @@ Read services generate short-lived presigned URLs for the requested variants. ## Image Processing Is Asynchronous -An image becomes an asset only after the image pipeline Worker has generated its -variants and returned an authenticated callback to Hono. `uploads` records that -workflow separately from `assets`, which prevents partially processed images -from appearing in collections and lets the Worker retry safely. - -R2 originals and generated variants use separate namespaces (`ingest/` and -`assets/`). The R2 event rule matches only `ingest/`, preventing generated +The API creates an image asset and its `uploads` workflow row before the +original is written to S3. The asset can therefore appear immediately with +independent `variant_status` and `palette_status` enrichment states. Separate +SQS consumers generate variants and palette data from the original upload; +neither worker waits for the other. + +S3 originals and generated variants use separate namespaces (`ingest/` and +`assets/`). S3 notification rules match only `ingest/`, preventing generated variants from recursively scheduling more work. ## Image Colors Support More Than Display diff --git a/server/src/openapi.json b/server/src/openapi.json index 55bb04c..d2000e9 100644 --- a/server/src/openapi.json +++ b/server/src/openapi.json @@ -389,7 +389,7 @@ "operationId": "deleteCollectionNode", "tags": ["Collections"], "summary": "Delete a folder from a collection", - "description": "Recursively deletes the folder, all descendant folders, and all descendant assets (including their R2 storage objects). For deleting individual assets use `DELETE /workspace/{workspaceSlug}/assets/{assetId}`.", + "description": "Recursively deletes the folder, all descendant folders, and all descendant assets (including their S3 storage objects). For deleting individual assets use `DELETE /workspace/{workspaceSlug}/assets/{assetId}`.", "parameters": [ { "$ref": "#/components/parameters/WorkspaceSlug" @@ -757,7 +757,7 @@ "operationId": "deleteAsset", "tags": ["Assets"], "summary": "Permanently delete an asset", - "description": "Permanently deletes the asset, its metadata, and its associated R2 storage objects (original image, display variant, preview variant, and upload object). This action cannot be undone.\n\nTo only remove an asset from a collection without deleting it, use `DELETE /collections/{slug}/nodes/{nodeId}` instead.", + "description": "Permanently deletes the asset, its metadata, and its associated S3 storage objects (original image, display variant, preview variant, and upload object). This action cannot be undone.\n\nTo only remove an asset from a collection without deleting it, use `DELETE /collections/{slug}/nodes/{nodeId}` instead.", "parameters": [ { "$ref": "#/components/parameters/WorkspaceSlug" @@ -864,7 +864,7 @@ "operationId": "createDirectImageUpload", "tags": ["Image Uploads"], "summary": "Initiate a direct image upload to a collection", - "description": "Generates a presigned PUT URL for uploading an image file directly to R2 object storage scoped to a collection. R2 creation events start asynchronous processing; poll the upload status endpoint after the PUT succeeds.", + "description": "Generates a presigned PUT URL for uploading an image file directly to S3 object storage scoped to a collection. S3 creation events start asynchronous processing; poll the upload status endpoint after the PUT succeeds.", "parameters": [ { "$ref": "#/components/parameters/WorkspaceSlug" @@ -958,7 +958,7 @@ "operationId": "createImageFromRemoteUrl", "tags": ["Image Uploads"], "summary": "Import an image from a remote URL into a collection", - "description": "Downloads an image from a remote URL into R2 and returns an asynchronous upload status. The Worker generates variants and creates the image asset.", + "description": "Downloads an image from a remote URL into S3 and returns an asynchronous upload status. Independent workers generate variants and palette data for the pre-created image asset.", "parameters": [ { "$ref": "#/components/parameters/WorkspaceSlug" @@ -1008,7 +1008,7 @@ "operationId": "createInboxDirectImageUpload", "tags": ["Image Uploads"], "summary": "Initiate a direct image upload to the inbox", - "description": "Same as `createDirectImageUpload` but scoped to the inbox. R2 creation events start asynchronous processing, and the completed asset will appear in the inbox.", + "description": "Same as `createDirectImageUpload` but scoped to the inbox. S3 creation events start asynchronous processing, and the pre-created asset is enriched as processing completes.", "parameters": [ { "$ref": "#/components/parameters/WorkspaceSlug" @@ -2157,7 +2157,7 @@ "type": "string", "minLength": 1, "maxLength": 255, - "description": "Original file name (used for R2 key generation only; not retained as the asset title)." + "description": "Original file name (used for S3 key generation only; not retained as the asset title)." }, "contentType": { "type": "string", @@ -2215,7 +2215,7 @@ }, "objectKey": { "type": "string", - "description": "R2 object key for the uploaded file." + "description": "S3 object key for the uploaded file." }, "url": { "type": "string", diff --git a/server/sst-env.d.ts b/server/sst-env.d.ts index 3b8cffd..330c8c7 100644 --- a/server/sst-env.d.ts +++ b/server/sst-env.d.ts @@ -6,5 +6,5 @@ /// -import "sst" -export {} \ No newline at end of file +import "sst"; +export {}; diff --git a/services/image-pipeline/README.md b/services/image-pipeline/README.md index 69d8082..323a3f8 100644 --- a/services/image-pipeline/README.md +++ b/services/image-pipeline/README.md @@ -1,10 +1,20 @@ # Image pipeline -This is an AWS Lambda. Only S3 `ingest/` object-created events -are delivered directly to SQS; Lambda consumes the queue, creates WebP variants under -`assets/`, and sends a signed callback to the Hono API. - -The SQS retry policy and dead-letter queue are defined in the root -[`sst.config.ts`](../../sst.config.ts). See -[`SST_DEPLOYMENT.md`](../../SST_DEPLOYMENT.md) for local invocation and deploy -instructions. +The image pipeline consists of two independent AWS Lambda consumers: + +```txt +S3 ingest/ object-created event + ├─ ImageVariantsQueue -> variants Lambda -> assets/{storageId}/*.webp + └─ ImagePaletteQueue -> palette Lambda -> image_colors callback +``` + +Both queues receive the same original-object event and read the original from +S3. The image asset already exists before the upload begins, so either callback +may arrive first. The variants worker updates rendition metadata; the palette +worker updates palette data and its own enrichment status. + +Each queue has a separate 180-second visibility timeout, retry budget, and +dead-letter queue. A failure in palette extraction never re-runs variant +generation. The infrastructure is defined in the root +[`sst.config.ts`](../../sst.config.ts). Use `bun run dev` for a variants event +fixture and `bun run dev:palette` for a palette event fixture. diff --git a/services/image-pipeline/package.json b/services/image-pipeline/package.json index 22d6a11..3366c3a 100644 --- a/services/image-pipeline/package.json +++ b/services/image-pipeline/package.json @@ -4,7 +4,9 @@ "private": true, "scripts": { "dev": "tsx watch src/local.ts", + "dev:palette": "IMAGE_PIPELINE_HANDLER=palette tsx watch src/local.ts", "start": "tsx src/local.ts", + "start:palette": "IMAGE_PIPELINE_HANDLER=palette tsx src/local.ts", "lint": "oxlint .", "format": "oxfmt --check", "format:fix": "oxfmt --write", diff --git a/services/image-pipeline/src/lambda.ts b/services/image-pipeline/src/lambda.ts deleted file mode 100644 index c0b2d4c..0000000 --- a/services/image-pipeline/src/lambda.ts +++ /dev/null @@ -1,249 +0,0 @@ -import { - GetObjectCommand, - PutObjectCommand, - S3Client, -} from "@aws-sdk/client-s3"; -import type { SQSBatchResponse, SQSHandler, S3Event } from "aws-lambda"; -import { createHmac } from "node:crypto"; - -import { processImagePalette, processImageVariants } from "./processor"; - -const MAX_PROCESSING_ATTEMPTS = 3; -const MAX_SOURCE_BYTES = 20 * 1024 * 1024; -const client = new S3Client({}); - -type PipelineVariant = { - role: "display" | "preview"; - objectKey: string; - width: number; - height: number; - contentType: "image/webp"; - sizeBytes: number; -}; - -type PipelineCallback = - | { - event: "image.processing.started"; - originalObjectKey: string; - originalEtag: string; - } - | { - event: "image.variants.completed"; - originalObjectKey: string; - originalEtag: string; - width: number; - height: number; - format: string; - blurDataURL: string; - variants: PipelineVariant[]; - } - | { - event: "image.palette.completed"; - originalObjectKey: string; - originalEtag: string; - extractionVersion: number; - palette: unknown[]; - } - | { - event: "image.variants.failed" | "image.palette.failed"; - originalObjectKey: string; - originalEtag: string; - error: string; - }; - -const required = (name: string): string => { - const value = process.env[name]; - if (!value) throw new Error(`${name} is required`); - return value; -}; - -const decodeS3Key = (key: string) => - decodeURIComponent(key.replace(/\+/g, " ")); - -export const handler: SQSHandler = async (event): Promise => { - const batchItemFailures: SQSBatchResponse["batchItemFailures"] = []; - - for (const message of event.Records) { - try { - const s3Event = JSON.parse(message.body) as S3Event; - for (const record of s3Event.Records) { - await processRecord( - record.s3.bucket.name, - decodeS3Key(record.s3.object.key), - record.s3.object.eTag, - record.s3.object.size, - ); - } - } catch (error) { - const attempts = Number( - message.attributes.ApproximateReceiveCount ?? "1", - ); - const detail = - error instanceof Error - ? error.message - : "Unknown image processing error"; - console.error( - JSON.stringify({ - event: "image_pipeline.failed", - messageId: message.messageId, - attempts, - error: detail, - }), - ); - - // The third processing failure is terminal. We notify the API, then - // acknowledge the message. A failed callback remains retryable. - if (attempts >= MAX_PROCESSING_ATTEMPTS) { - try { - const s3Event = JSON.parse(message.body) as S3Event; - for (const record of s3Event.Records) { - await sendCallback({ - event: "image.variants.failed", - originalObjectKey: decodeS3Key(record.s3.object.key), - originalEtag: record.s3.object.eTag, - error: detail.slice(0, 1000), - }); - } - } catch (callbackError) { - console.error( - JSON.stringify({ - event: "image_pipeline.failure_callback_failed", - messageId: message.messageId, - error: String(callbackError), - }), - ); - batchItemFailures.push({ itemIdentifier: message.messageId }); - } - } else { - batchItemFailures.push({ itemIdentifier: message.messageId }); - } - } - } - - return { batchItemFailures }; -}; - -async function processRecord( - bucket: string, - objectKey: string, - originalEtag: string, - size?: number, -) { - if (!objectKey.startsWith("ingest/") || !originalEtag) return; - if ((size ?? 0) > MAX_SOURCE_BYTES) - throw new Error("Source image exceeds the 20 MiB processing limit"); - - await sendCallback({ - event: "image.processing.started", - originalObjectKey: objectKey, - originalEtag, - }); - const source = await client.send( - new GetObjectCommand({ Bucket: bucket, Key: objectKey }), - ); - if (!source.Body) throw new Error("Original object no longer exists"); - - const bytes = await source.Body.transformToByteArray(); - const result = await processImageVariants(bytes); - const storageId = storageIdFromOriginalKey(objectKey); - const variants = await Promise.all( - result.variants.map(async (variant) => { - const key = `assets/${storageId}/${variant.role}.webp`; - await client.send( - new PutObjectCommand({ - Bucket: bucket, - Key: key, - Body: variant.bytes, - ContentType: variant.contentType, - }), - ); - return { - role: variant.role, - objectKey: key, - width: variant.width, - height: variant.height, - contentType: variant.contentType, - sizeBytes: variant.sizeBytes, - }; - }), - ); - - await sendCallback({ - event: "image.variants.completed", - originalObjectKey: objectKey, - originalEtag, - width: result.width, - height: result.height, - format: result.format, - blurDataURL: result.blurDataURL, - variants, - }); - try { - const palette = await processImagePalette(bytes); - await sendCallback({ - event: "image.palette.completed", - originalObjectKey: objectKey, - originalEtag, - ...palette, - }); - } catch (error) { - await sendCallback({ - event: "image.palette.failed", - originalObjectKey: objectKey, - originalEtag, - error: - error instanceof Error - ? error.message.slice(0, 1000) - : "Palette extraction failed", - }); - console.error( - JSON.stringify({ - event: "image_pipeline.palette_failed", - objectKey, - error: String(error), - }), - ); - } - console.log( - JSON.stringify({ - event: "image_pipeline.completed", - objectKey, - variants: variants.length, - palette: "queued", - }), - ); -} - -async function sendCallback(payload: PipelineCallback) { - const body = JSON.stringify(payload); - const timestamp = Date.now().toString(); - const signature = createHmac( - "sha256", - required("IMAGE_PIPELINE_CALLBACK_SECRET"), - ) - .update(`${timestamp}.${body}`) - .digest("hex"); - const response = await fetch( - new URL( - "/api/v1/internal/image-pipeline/callback", - required("PIPELINE_API_BASE_URL"), - ), - { - method: "POST", - headers: { - "content-type": "application/json", - "x-aska-timestamp": timestamp, - "x-aska-signature": signature, - }, - body, - }, - ); - if (!response.ok) - throw new Error(`Pipeline callback failed with status ${response.status}`); -} - -function storageIdFromOriginalKey(objectKey: string): string { - const match = /^ingest\/([^/]+)\/original(?:\.[a-z0-9]+)?$/i.exec(objectKey); - if (!match) throw new Error("Invalid ingest object key"); - return match[1]!; -} diff --git a/services/image-pipeline/src/local.ts b/services/image-pipeline/src/local.ts index 4b7b1c3..d7bb836 100644 --- a/services/image-pipeline/src/local.ts +++ b/services/image-pipeline/src/local.ts @@ -2,7 +2,8 @@ import "dotenv/config"; import type { Context, SQSEvent } from "aws-lambda"; import { readFile } from "node:fs/promises"; -import { handler } from "./lambda"; +import { handler as paletteHandler } from "./palette-lambda"; +import { handler as variantsHandler } from "./variants-lambda"; const eventPath = process.env.SQS_EVENT_FILE; if (!eventPath) { @@ -12,4 +13,8 @@ if (!eventPath) { } const event = JSON.parse(await readFile(eventPath, "utf8")) as SQSEvent; +const handler = + process.env.IMAGE_PIPELINE_HANDLER === "palette" + ? paletteHandler + : variantsHandler; await handler(event, {} as Context, () => undefined); diff --git a/services/image-pipeline/src/palette-lambda.ts b/services/image-pipeline/src/palette-lambda.ts new file mode 100644 index 0000000..780aef3 --- /dev/null +++ b/services/image-pipeline/src/palette-lambda.ts @@ -0,0 +1,46 @@ +import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3"; + +import { sendCallback } from "./pipeline-callback"; +import { processImagePalette } from "./processor"; +import { createSqsHandler, type SourceImage } from "./sqs-handler"; + +const MAX_SOURCE_BYTES = 20 * 1024 * 1024; +const client = new S3Client({}); + +async function processPalette(source: SourceImage) { + if ((source.size ?? 0) > MAX_SOURCE_BYTES) + throw new Error("Source image exceeds the 20 MiB processing limit"); + + const original = await client.send( + new GetObjectCommand({ Bucket: source.bucket, Key: source.objectKey }), + ); + if (!original.Body) throw new Error("Original object no longer exists"); + + const bytes = await original.Body.transformToByteArray(); + const palette = await processImagePalette(bytes); + await sendCallback({ + event: "image.palette.completed", + originalObjectKey: source.objectKey, + originalEtag: source.originalEtag, + ...palette, + }); + console.log( + JSON.stringify({ + event: "image_palette.completed", + objectKey: source.objectKey, + colors: palette.palette.length, + }), + ); +} + +export const handler = createSqsHandler({ + pipeline: "palette", + process: processPalette, + reportTerminalFailure: (source, error) => + sendCallback({ + event: "image.palette.failed", + originalObjectKey: source.objectKey, + originalEtag: source.originalEtag, + error, + }), +}); diff --git a/services/image-pipeline/src/pipeline-callback.ts b/services/image-pipeline/src/pipeline-callback.ts new file mode 100644 index 0000000..db15f6a --- /dev/null +++ b/services/image-pipeline/src/pipeline-callback.ts @@ -0,0 +1,81 @@ +import { createHmac } from "node:crypto"; + +type PipelineVariant = { + role: "display" | "preview"; + objectKey: string; + width: number; + height: number; + contentType: "image/webp"; + sizeBytes: number; +}; + +export type PipelineCallback = + | { + event: "image.processing.started"; + originalObjectKey: string; + originalEtag: string; + } + | { + event: "image.variants.completed"; + originalObjectKey: string; + originalEtag: string; + width: number; + height: number; + format: string; + blurDataURL: string; + variants: PipelineVariant[]; + } + | { + event: "image.palette.completed"; + originalObjectKey: string; + originalEtag: string; + extractionVersion: number; + palette: unknown[]; + } + | { + event: "image.variants.failed" | "image.palette.failed"; + originalObjectKey: string; + originalEtag: string; + error: string; + }; + +const required = (name: string): string => { + const value = process.env[name]; + if (!value) throw new Error(`${name} is required`); + return value; +}; + +/** Sends an authenticated, idempotent processing result to the API. */ +export async function sendCallback(payload: PipelineCallback) { + const body = JSON.stringify(payload); + const timestamp = Date.now().toString(); + const signature = createHmac( + "sha256", + required("IMAGE_PIPELINE_CALLBACK_SECRET"), + ) + .update(`${timestamp}.${body}`) + .digest("hex"); + const response = await fetch( + new URL( + "/api/v1/internal/image-pipeline/callback", + required("PIPELINE_API_BASE_URL"), + ), + { + method: "POST", + headers: { + "content-type": "application/json", + "x-aska-timestamp": timestamp, + "x-aska-signature": signature, + }, + body, + }, + ); + if (!response.ok) + throw new Error(`Pipeline callback failed with status ${response.status}`); +} + +export function storageIdFromOriginalKey(objectKey: string): string { + const match = /^ingest\/([^/]+)\/original(?:\.[a-z0-9]+)?$/i.exec(objectKey); + if (!match) throw new Error("Invalid ingest object key"); + return match[1]!; +} diff --git a/services/image-pipeline/src/processor.ts b/services/image-pipeline/src/processor.ts index bf7e82e..19685bf 100644 --- a/services/image-pipeline/src/processor.ts +++ b/services/image-pipeline/src/processor.ts @@ -132,7 +132,7 @@ function evenlySample(samples: T[], limit: number): T[] { /** * Decodes an original image and produces its metadata, progressive placeholder, - * display variants, and search-oriented colour palette. + * and display variants. */ export async function processImageVariants( buffer: Uint8Array, diff --git a/services/image-pipeline/src/sqs-handler.test.ts b/services/image-pipeline/src/sqs-handler.test.ts new file mode 100644 index 0000000..83053f7 --- /dev/null +++ b/services/image-pipeline/src/sqs-handler.test.ts @@ -0,0 +1,126 @@ +import type { SQSEvent } from "aws-lambda"; +import { describe, expect, it, vi } from "vitest"; + +import { createSqsHandler } from "./sqs-handler"; + +const sourceEvent = { + Records: [ + { + s3: { + bucket: { name: "aska-dev" }, + object: { + key: "ingest%2Fupload-1%2Foriginal.jpg", + eTag: "etag-1", + size: 1234, + }, + }, + }, + ], +}; + +function sqsEvent(body: unknown, attempts = 1): SQSEvent { + return { + Records: [ + { + messageId: "message-1", + receiptHandle: "receipt", + body: JSON.stringify(body), + attributes: { + ApproximateReceiveCount: String(attempts), + ApproximateFirstReceiveTimestamp: "0", + SenderId: "s3", + SentTimestamp: "0", + }, + messageAttributes: {}, + md5OfBody: "", + eventSource: "aws:sqs", + eventSourceARN: "arn:aws:sqs:eu-central-1:123456789:queue", + awsRegion: "eu-central-1", + }, + ], + }; +} + +function handlerFor( + process = vi.fn(async () => undefined), + reportTerminalFailure = vi.fn(async () => undefined), +) { + return { + process, + reportTerminalFailure, + handler: createSqsHandler({ + pipeline: "palette", + process, + reportTerminalFailure, + }), + }; +} + +describe("SQS image processor retry contract", () => { + it("acknowledges S3 test notifications without processing", async () => { + const { handler, process } = handlerFor(); + + const result = await handler( + sqsEvent({ Event: "s3:TestEvent" }), + {} as never, + () => undefined, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(process).not.toHaveBeenCalled(); + }); + + it("retries ordinary processing failures before the terminal receive", async () => { + const process = vi.fn(async () => { + throw new Error("sharp failed"); + }); + const { handler, reportTerminalFailure } = handlerFor(process); + + const result = await handler( + sqsEvent(sourceEvent), + {} as never, + () => undefined, + ); + + expect(result).toEqual({ + batchItemFailures: [{ itemIdentifier: "message-1" }], + }); + expect(reportTerminalFailure).not.toHaveBeenCalled(); + }); + + it("reports the processor's own terminal failure on receive five", async () => { + const process = vi.fn(async () => { + throw new Error("sharp failed"); + }); + const { handler, reportTerminalFailure } = handlerFor(process); + + const result = await handler( + sqsEvent(sourceEvent, 5), + {} as never, + () => undefined, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(reportTerminalFailure).toHaveBeenCalledWith( + expect.objectContaining({ objectKey: "ingest/upload-1/original.jpg" }), + "sharp failed", + ); + }); + + it("uses the final receive only to retry a terminal callback", async () => { + const { handler, process, reportTerminalFailure } = handlerFor(); + + const result = await handler( + sqsEvent(sourceEvent, 6), + {} as never, + () => undefined, + ); + + expect(result).toEqual({ batchItemFailures: [] }); + expect(process).not.toHaveBeenCalled(); + expect(reportTerminalFailure).toHaveBeenCalledWith( + expect.objectContaining({ objectKey: "ingest/upload-1/original.jpg" }), + "Image processing failed before its terminal callback could be delivered", + ); + }); +}); diff --git a/services/image-pipeline/src/sqs-handler.ts b/services/image-pipeline/src/sqs-handler.ts new file mode 100644 index 0000000..41abd7f --- /dev/null +++ b/services/image-pipeline/src/sqs-handler.ts @@ -0,0 +1,123 @@ +import type { SQSBatchResponse, SQSHandler, S3Event } from "aws-lambda"; + +const MAX_PROCESSING_ATTEMPTS = 5; + +export type SourceImage = { + bucket: string; + objectKey: string; + originalEtag: string; + size?: number; +}; + +type HandlerOptions = { + pipeline: "variants" | "palette"; + process: (source: SourceImage) => Promise; + reportTerminalFailure: (source: SourceImage, error: string) => Promise; +}; + +const decodeS3Key = (key: string) => + decodeURIComponent(key.replace(/\+/g, " ")); + +function sourcesFromS3Event(event: S3Event): SourceImage[] { + // S3 sends an initial test notification without Records when a destination is + // configured. It is not an image-processing job. + if (!Array.isArray(event.Records)) return []; + return event.Records.flatMap((record) => { + const source = { + bucket: record.s3.bucket.name, + objectKey: decodeS3Key(record.s3.object.key), + originalEtag: record.s3.object.eTag, + size: record.s3.object.size, + }; + return source.objectKey.startsWith("ingest/") && source.originalEtag + ? [source] + : []; + }); +} + +/** + * Wraps an independent image processor in the shared SQS retry contract. + * + * A failed job is redelivered until its fifth receive. At that point the API + * receives the matching terminal status. If that terminal callback is down, + * the message remains failed so SQS can redeliver it and ultimately retain it + * in the queue's DLQ. + */ +export function createSqsHandler({ + pipeline, + process, + reportTerminalFailure, +}: HandlerOptions): SQSHandler { + return async (event): Promise => { + const batchItemFailures: SQSBatchResponse["batchItemFailures"] = []; + + for (const message of event.Records) { + const attempts = Number( + message.attributes.ApproximateReceiveCount ?? "1", + ); + if (attempts > MAX_PROCESSING_ATTEMPTS) { + try { + const s3Event = JSON.parse(message.body) as S3Event; + for (const source of sourcesFromS3Event(s3Event)) { + await reportTerminalFailure( + source, + "Image processing failed before its terminal callback could be delivered", + ); + } + } catch (callbackError) { + console.error( + JSON.stringify({ + event: `image_${pipeline}.failure_callback_failed`, + messageId: message.messageId, + attempts, + error: String(callbackError), + }), + ); + batchItemFailures.push({ itemIdentifier: message.messageId }); + } + continue; + } + + try { + const s3Event = JSON.parse(message.body) as S3Event; + for (const source of sourcesFromS3Event(s3Event)) await process(source); + } catch (error) { + const detail = + error instanceof Error + ? error.message + : "Unknown image processing error"; + console.error( + JSON.stringify({ + event: `image_${pipeline}.failed`, + messageId: message.messageId, + attempts, + error: detail, + }), + ); + + if (attempts < MAX_PROCESSING_ATTEMPTS) { + batchItemFailures.push({ itemIdentifier: message.messageId }); + continue; + } + + try { + const s3Event = JSON.parse(message.body) as S3Event; + for (const source of sourcesFromS3Event(s3Event)) { + await reportTerminalFailure(source, detail.slice(0, 1000)); + } + } catch (callbackError) { + console.error( + JSON.stringify({ + event: `image_${pipeline}.failure_callback_failed`, + messageId: message.messageId, + error: String(callbackError), + }), + ); + batchItemFailures.push({ itemIdentifier: message.messageId }); + } + } + } + + return { batchItemFailures }; + }; +} diff --git a/services/image-pipeline/src/variants-lambda.ts b/services/image-pipeline/src/variants-lambda.ts new file mode 100644 index 0000000..208494c --- /dev/null +++ b/services/image-pipeline/src/variants-lambda.ts @@ -0,0 +1,75 @@ +import { + GetObjectCommand, + PutObjectCommand, + S3Client, +} from "@aws-sdk/client-s3"; + +import { sendCallback, storageIdFromOriginalKey } from "./pipeline-callback"; +import { processImageVariants } from "./processor"; +import { createSqsHandler, type SourceImage } from "./sqs-handler"; + +const MAX_SOURCE_BYTES = 20 * 1024 * 1024; +const client = new S3Client({}); + +async function processVariants(source: SourceImage) { + if ((source.size ?? 0) > MAX_SOURCE_BYTES) + throw new Error("Source image exceeds the 20 MiB processing limit"); + + await sendCallback({ + event: "image.processing.started", + originalObjectKey: source.objectKey, + originalEtag: source.originalEtag, + }); + const original = await client.send( + new GetObjectCommand({ Bucket: source.bucket, Key: source.objectKey }), + ); + if (!original.Body) throw new Error("Original object no longer exists"); + + const bytes = await original.Body.transformToByteArray(); + const result = await processImageVariants(bytes); + const storageId = storageIdFromOriginalKey(source.objectKey); + const variants = await Promise.all( + result.variants.map(async (variant) => { + const objectKey = `assets/${storageId}/${variant.role}.webp`; + await client.send( + new PutObjectCommand({ + Bucket: source.bucket, + Key: objectKey, + Body: variant.bytes, + ContentType: variant.contentType, + }), + ); + return { ...variant, objectKey }; + }), + ); + + await sendCallback({ + event: "image.variants.completed", + originalObjectKey: source.objectKey, + originalEtag: source.originalEtag, + width: result.width, + height: result.height, + format: result.format, + blurDataURL: result.blurDataURL, + variants, + }); + console.log( + JSON.stringify({ + event: "image_variants.completed", + objectKey: source.objectKey, + variants: variants.length, + }), + ); +} + +export const handler = createSqsHandler({ + pipeline: "variants", + process: processVariants, + reportTerminalFailure: (source, error) => + sendCallback({ + event: "image.variants.failed", + originalObjectKey: source.objectKey, + originalEtag: source.originalEtag, + error, + }), +}); diff --git a/services/image-pipeline/sst-env.d.ts b/services/image-pipeline/sst-env.d.ts index 6444193..eec65b9 100644 --- a/services/image-pipeline/sst-env.d.ts +++ b/services/image-pipeline/sst-env.d.ts @@ -6,5 +6,5 @@ /// -import "sst" -export {} \ No newline at end of file +import "sst"; +export {}; diff --git a/sst-env.d.ts b/sst-env.d.ts index 7cf9287..f05092e 100644 --- a/sst-env.d.ts +++ b/sst-env.d.ts @@ -6,44 +6,52 @@ declare module "sst" { export interface Resource { - "Api": { - "type": "sst.aws.ApiGatewayV2" - "url": string - } - "Assets": { - "name": string - "type": "sst.aws.Bucket" - } - "BetterAuthSecret": { - "type": "sst.sst.Secret" - "value": string - } - "Client": { - "type": "sst.aws.StaticSite" - "url": string - } - "DatabaseUrl": { - "type": "sst.sst.Secret" - "value": string - } - "ImageDeadLetterQueue": { - "type": "sst.aws.Queue" - "url": string - } - "ImagePipelineCallbackSecret": { - "type": "sst.sst.Secret" - "value": string - } - "ImageQueue": { - "type": "sst.aws.Queue" - "url": string - } - "ResendApiKey": { - "type": "sst.sst.Secret" - "value": string - } + Api: { + type: "sst.aws.ApiGatewayV2"; + url: string; + }; + Assets: { + name: string; + type: "sst.aws.Bucket"; + }; + BetterAuthSecret: { + type: "sst.sst.Secret"; + value: string; + }; + Client: { + type: "sst.aws.StaticSite"; + url: string; + }; + DatabaseUrl: { + type: "sst.sst.Secret"; + value: string; + }; + ImagePaletteDeadLetterQueue: { + type: "sst.aws.Queue"; + url: string; + }; + ImagePipelineCallbackSecret: { + type: "sst.sst.Secret"; + value: string; + }; + ImagePaletteQueue: { + type: "sst.aws.Queue"; + url: string; + }; + ImageVariantsDeadLetterQueue: { + type: "sst.aws.Queue"; + url: string; + }; + ImageVariantsQueue: { + type: "sst.aws.Queue"; + url: string; + }; + ResendApiKey: { + type: "sst.sst.Secret"; + value: string; + }; } } -import "sst" -export {} \ No newline at end of file +import "sst"; +export {}; diff --git a/sst.config.ts b/sst.config.ts index 2946877..f3c710b 100644 --- a/sst.config.ts +++ b/sst.config.ts @@ -29,21 +29,32 @@ export default $config({ "ImagePipelineCallbackSecret", ); - const imageDeadLetterQueue = new sst.aws.Queue("ImageDeadLetterQueue", { - transform: { - queue: { - messageRetentionSeconds: 1_209_600, + const createImageQueue = (name: string, deadLetterQueueName: string) => { + const deadLetterQueue = new sst.aws.Queue(deadLetterQueueName, { + transform: { + queue: { + messageRetentionSeconds: 1_209_600, + }, }, - }, - }); + }); + const queue = new sst.aws.Queue(name, { + visibilityTimeout: "180 seconds", + // The worker reports a terminal image status on receive five. Keep one + // additional receive for a failed terminal callback before preserving + // the message in the DLQ. + dlq: { queue: deadLetterQueue.arn, retry: 6 }, + }); + return { queue, deadLetterQueue }; + }; - const imageQueue = new sst.aws.Queue("ImageQueue", { - visibilityTimeout: "180 seconds", - dlq: { - queue: imageDeadLetterQueue.arn, - retry: 5, - }, - }); + const { + queue: imageVariantsQueue, + deadLetterQueue: imageVariantsDeadLetterQueue, + } = createImageQueue("ImageVariantsQueue", "ImageVariantsDeadLetterQueue"); + const { + queue: imagePaletteQueue, + deadLetterQueue: imagePaletteDeadLetterQueue, + } = createImageQueue("ImagePaletteQueue", "ImagePaletteDeadLetterQueue"); const assets = new sst.aws.Bucket("Assets", { cors: { @@ -57,8 +68,14 @@ export default $config({ assets.notify({ notifications: [ { - name: "ProcessIngestedImage", - queue: imageQueue, + name: "GenerateImageVariants", + queue: imageVariantsQueue, + events: ["s3:ObjectCreated:*"], + filterPrefix: "ingest/", + }, + { + name: "ExtractImagePalette", + queue: imagePaletteQueue, events: ["s3:ObjectCreated:*"], filterPrefix: "ingest/", }, @@ -95,52 +112,69 @@ export default $config({ }, }); - imageQueue.subscribe( + const imageWorkerFiles = [ { - handler: "services/image-pipeline/src/lambda.handler", - runtime: "nodejs22.x", - memory: "2048 MB", - timeout: "120 seconds", - link: [assets], - nodejs: { - // Sharp is native code. Keep it external to esbuild and package the - // Linux runtime installed by Bun, rather than SST's npm-based - // `nodejs.install` helper. - esbuild: { - external: ["sharp"], - }, - }, - copyFiles: [ - { - from: "services/image-pipeline/node_modules/sharp", - to: "node_modules/sharp", - }, - { - from: "services/image-pipeline/node_modules/@img/colour", - to: "node_modules/@img/colour", - }, - { - from: "services/image-pipeline/node_modules/detect-libc", - to: "node_modules/detect-libc", - }, - { - from: "services/image-pipeline/node_modules/semver", - to: "node_modules/semver", - }, - { - from: "services/image-pipeline/node_modules/@img/sharp-linux-x64", - to: "node_modules/@img/sharp-linux-x64", - }, - { - from: "services/image-pipeline/node_modules/@img/sharp-libvips-linux-x64", - to: "node_modules/@img/sharp-libvips-linux-x64", - }, - ], - environment: { - PIPELINE_API_BASE_URL: api.url, - IMAGE_PIPELINE_CALLBACK_SECRET: imagePipelineCallbackSecret.value, + from: "services/image-pipeline/node_modules/sharp", + to: "node_modules/sharp", + }, + { + from: "services/image-pipeline/node_modules/@img/colour", + to: "node_modules/@img/colour", + }, + { + from: "services/image-pipeline/node_modules/detect-libc", + to: "node_modules/detect-libc", + }, + { + from: "services/image-pipeline/node_modules/semver", + to: "node_modules/semver", + }, + { + from: "services/image-pipeline/node_modules/@img/sharp-linux-x64", + to: "node_modules/@img/sharp-linux-x64", + }, + { + from: "services/image-pipeline/node_modules/@img/sharp-libvips-linux-x64", + to: "node_modules/@img/sharp-libvips-linux-x64", + }, + ]; + const imageWorkerEnvironment = { + PIPELINE_API_BASE_URL: api.url, + IMAGE_PIPELINE_CALLBACK_SECRET: imagePipelineCallbackSecret.value, + }; + const imageWorkerDefaults = { + runtime: "nodejs22.x", + memory: "2048 MB", + timeout: "120 seconds", + link: [assets], + nodejs: { + // Sharp is native code. Keep it external to esbuild and package the + // Linux runtime installed by Bun, rather than SST's npm-based + // `nodejs.install` helper. + esbuild: { external: ["sharp"] }, + }, + copyFiles: imageWorkerFiles, + environment: imageWorkerEnvironment, + }; + + imageVariantsQueue.subscribe( + { + handler: "services/image-pipeline/src/variants-lambda.handler", + ...imageWorkerDefaults, + }, + { + batch: { + size: 1, + partialResponses: true, }, }, + ); + + imagePaletteQueue.subscribe( + { + handler: "services/image-pipeline/src/palette-lambda.handler", + ...imageWorkerDefaults, + }, { batch: { size: 1, @@ -170,8 +204,10 @@ export default $config({ api: api.url, client: client.url, assetsBucket: assets.name, - imageQueue: imageQueue.url, - imageDeadLetterQueue: imageDeadLetterQueue.url, + imageVariantsQueue: imageVariantsQueue.url, + imageVariantsDeadLetterQueue: imageVariantsDeadLetterQueue.url, + imagePaletteQueue: imagePaletteQueue.url, + imagePaletteDeadLetterQueue: imagePaletteDeadLetterQueue.url, }; }, });