fastify-ximagepipeline

Fastify plugin for async image upload processing — EXIF stripping, optional content moderation, WebP variant generation, blurhash placeholders, and Cloudflare R2/S3 storage. Uploads land in a staging prefix; a background worker polls a Prisma job queue and processes each image asynchronously.

Installation

npm install @xenterprises/fastify-ximagepipeline
# Peer deps
npm install @fastify/multipart fastify

Register @fastify/multipart before this plugin.

Quick Start

import Fastify from "fastify";
import multipart from "@fastify/multipart";
import xImagePipeline from "@xenterprises/fastify-ximagepipeline";

const fastify = Fastify({ logger: true });
await fastify.register(multipart);
await fastify.register(xImagePipeline, {
  r2: {
    endpoint: process.env.R2_ENDPOINT,
    accessKeyId: process.env.R2_ACCESS_KEY_ID,
    secretAccessKey: process.env.R2_SECRET_ACCESS_KEY,
    bucket: process.env.R2_BUCKET,
  },
  db: prisma, // Prisma client with MediaQueue + Media models
});

await fastify.listen({ port: 3000 });

# Upload an image
curl -F "file=@photo.jpg" -F "sourceType=avatar" -F "sourceId=user-123" \
  http://localhost:3000/image-pipeline/upload

Options

Plugin Options

r2 Config

worker Config

moderation Config

Default Variants

Default Source Types

Routes & Methods

Routes registered automatically at /image-pipeline/:

• POST /image-pipeline/upload — Upload a file for async processing; returns a jobId immediately.
• GET /image-pipeline/status/:jobId — Poll job status; returns variant URLs and blurhash on COMPLETE.

Decorator methods on fastify.xImagePipeline:

• getStatus(jobId) — Fetch a job record with its media relation from the database.
• deleteMedia(mediaId) — Delete a media record and all associated R2 objects.
• listMedia(sourceType, sourceId) — List all media records for a given source.
• getVariantPresets / getSourceTypes / getVariants — Inspect active variant and source type configuration.

Exported Utilities

@xenterprises/fastify-ximagepipeline/image

import { stripExif, generateVariants, generateBlurhash } from "@xenterprises/fastify-ximagepipeline/image";

@xenterprises/fastify-ximagepipeline/s3

import { uploadToS3, deleteFromS3 } from "@xenterprises/fastify-ximagepipeline/s3";

Error Reference

All errors use the [xImagePipeline] prefix.

Environment Variables

Database Schema

Requires two Prisma models (see SCHEMA.prisma in the package):

• MediaQueue — job queue: PENDING → PROCESSING → COMPLETE / REJECTED / FAILED
• Media — processed records with variant URLs, dimensions, blurhash, and source metadata

How It Works

Upload: Client sends multipart POST. The file is validated (type, size, sourceType) and uploaded to the R2 staging prefix. A MediaQueue row is created with PENDING status; the route returns a jobId immediately.

Processing: A background worker polls MediaQueue for PENDING jobs using pessimistic locking. For each job it downloads from staging, strips EXIF, runs content moderation (if a handler is configured), generates WebP variants per sourceType config, produces a blurhash placeholder, uploads variants and optional originals to R2, creates a Media record, marks the job COMPLETE, and deletes the staging file.

Retrieval: The client polls GET /image-pipeline/status/:jobId. On COMPLETE the response includes all variant URLs, blurhash, and dimensions.

Retry: Failed jobs are retried up to maxAttempts. Stale locks (worker crash) are automatically recovered after lockTimeout.

AI Context

package: "@xenterprises/fastify-ximagepipeline"
type: fastify-plugin
use-when: Async image upload pipeline — EXIF stripping, WebP variant generation, blurhash, content moderation, R2/S3 storage with background job queue
decorator: fastify.xImagePipeline (getStatus, deleteMedia, listMedia, getVariantPresets, getSourceTypes, getVariants)
routes: POST /image-pipeline/upload, GET /image-pipeline/status/:jobId
env: R2_ENDPOINT, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, R2_BUCKET, DATABASE_URL
requires: Prisma client with MediaQueue and Media models; @fastify/multipart registered before this plugin
utility-exports: "@xenterprises/fastify-ximagepipeline/image" (image processing), "@xenterprises/fastify-ximagepipeline/s3" (S3 helpers)