@xenterprises/fastify-ximagepipeline

Fastify plugin for image uploads with EXIF stripping, WebP variant generation, blurhash placeholders, optional content moderation, and R2/S3 storage — processed asynchronously by a database-backed job queue with a polling worker. For Fastify 5 applications that need a production-grade upload pipeline with minimal consumer code.

Install

npm install @xenterprises/fastify-ximagepipeline fastify@5 @fastify/multipart

fastify@^5 and @fastify/multipart are peer dependencies. Register @fastify/multipart before this plugin.

Minimal example

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, // consumer owns env access
    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 (see SCHEMA.prisma)
});

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

Upload an image:

curl -F "[email protected]" -F "sourceType=avatar" -F "sourceId=user-123" \
  http://localhost:3000/image-pipeline/upload

Options

All configuration arrives via app.register(xImagePipeline, options). The plugin never reads process.env. Invalid options throw at registration time.

| Option | Type | Default | Required | Description | |--------|------|---------|----------|-------------| | r2 | object | — | Yes | R2/S3 connection config (see below) | | db | object | — | Yes | Prisma-like client exposing mediaQueue and media models | | s3Client | object | created from r2 | No | Pre-built S3-compatible client; must expose send(command). Created once at registration and reused | | moderation | object | null | No | Moderation config; handler: async (buffer, config) => ({ passed, flags, confidence }). Without a handler all images are approved | | variants | object | 6 sizes | No | Variant dimension specs (see Defaults) | | sourceTypes | object | 5 types | No | Per-source-type processing config; uploads are validated against these keys | | worker | object | enabled | No | Background worker settings (see below) | | stagingPath | string | "staging" | No | R2 prefix for staging uploads | | mediaPath | string | "media" | No | R2 prefix for processed variants | | originalsPath | string | "originals" | No | R2 prefix for compressed originals | | maxFileSize | number | 52428800 | No | Max upload size in bytes (50 MB) | | allowedMimeTypes | string[] | jpeg/png/webp/gif | No | Accepted upload MIME types |

r2

| Field | Type | Required | Description | |-------|------|----------|-------------| | endpoint | string | Yes | R2 or S3-compatible endpoint URL (also used to build public URLs) | | accessKeyId | string | Yes | Access key | | secretAccessKey | string | Yes | Secret key | | bucket | string | Yes | Bucket name | | region | string | No | Region (default "auto" for R2) |

worker

| Field | Type | Default | Description | |-------|------|---------|-------------| | enabled | boolean | true | Start the polling worker on registration | | pollInterval | number | 5000 | Queue poll interval (ms) | | maxAttempts | number | 3 | Attempts per job before it is marked FAILED | | lockTimeout | number | 300000 | Lock timeout (ms); stale locks are recovered on this cadence | | failOnError | boolean | true | Throw if the worker fails to start | | timers | object | globals | Timer injection { setInterval, clearInterval } (testing) |

Worker lifecycle: both intervals are unref'd (they never keep the process alive on their own) and are cleared in an onClose hook, which also awaits any in-flight job so the job row is left in a consistent state before the consumer's db client disconnects.

Default variants

| Name | Width | Height | Fit | |------|-------|--------|-----| | xs | 80 | 80 | cover | | sm | 200 | 200 | cover | | md | 600 | auto | inside | | lg | 1200 | auto | inside | | xl | 1920 | auto | inside | | 2xl | 2560 | auto | inside |

Default source types

| Type | Variants | Formats | Quality | Store Original | |------|----------|---------|---------|----------------| | avatar | xs, sm | webp | 85 | yes | | member_photo | xs, sm, md | webp | 85 | yes | | gallery | md, lg, xl | webp | 85 | no | | hero | lg, xl, 2xl | webp | 80 | no | | content | md, lg | webp | 85 | yes |

Decorators

One namespace: fastify.xImagePipeline.

| Method | Signature | Description | |--------|-----------|-------------| | getStatus | (jobId: string) => Promise<object\|null> | Job row with media relation, or null | | deleteMedia | (mediaId: string) => Promise<{ deleted: boolean, r2Deleted: number }> | Delete media record, its queue rows, and all R2 objects. Throws if the media id does not exist | | listMedia | (sourceType: string, sourceId: string) => Promise<object[]> | Media records for a source, newest first | | getVariantPresets | () => object | Source type → variant names mapping | | getSourceTypes | () => object | Effective source type configurations | | getVariants | () => object | Effective variant dimension specs |

Routes

POST /image-pipeline/upload

Multipart upload. Requires @fastify/multipart.

• Fields: file (binary, required), sourceType (string, required), sourceId (string, required)
• Validates MIME type, source type (must be a configured sourceTypes key), and file size; uploads to R2 staging; creates a PENDING queue job
• 202 { jobId, message, statusUrl } on success
• 400 no file / disallowed type / missing fields / unknown sourceType · 413 file too large · 500 storage or database failure (staging object is cleaned up on db failure)

GET /image-pipeline/status/:jobId

| Job status | HTTP | Extra fields | |------------|------|--------------| | PENDING, PROCESSING | 202 | — | | COMPLETE | 200 | media (id, urls, originalUrl, width, height, aspectRatio, blurhash, focalPoint) | | REJECTED | 400 | reason, moderationDetails | | FAILED | 500 | error, attempts | | unknown jobId | 404 | — |

Error behavior

• Registration throws real Errors whose messages name the plugin, the option, and an example, e.g. ximagepipeline: missing required option `r2` (object), e.g. `app.register(xImagePipeline, { r2: {...}, db: prisma })` . Missing/invalid r2, db, s3Client, moderation.handler, worker.*, maxFileSize, and allowedMimeTypes are all validated up front.
• Route errors use status-coded responses (see above); unexpected handler errors are logged and returned as 500.
• Job errors (download failures, corrupt images, moderation errors) are caught by the worker, recorded on the job (errorMsg, attempts), and retried until maxAttempts, then marked FAILED. Staging objects are deleted after the final failure.
• External SDK errors are re-thrown as Errors carrying error.message only — credentials and raw SDK error objects are never logged or propagated.

Exported utilities

Subpath exports for direct use:

• @xenterprises/fastify-ximagepipeline/image — stripExif, getImageMetadata, compressToJpeg, generateVariants, generateBlurhash, calculateFitDimensions, getAspectRatio, validateImage, processImage
• @xenterprises/fastify-ximagepipeline/s3 — initializeS3Client, uploadToS3, downloadFromS3, deleteFromS3, listFromS3, getSignedUrlForS3, getPublicUrl, batchDeleteFromS3

The default export also provides the named export getVariantPresets().

Database schema

The plugin requires two Prisma models, MediaQueue (job queue: PENDING → PROCESSING → COMPLETE/REJECTED/FAILED, with attempts/lockedAt/lockedBy for the worker) and Media (processed records with variant URLs, dimensions, blurhash). See SCHEMA.prisma in the package for the reference schema. The db client is consumer-provided via options.db — the plugin never constructs one.

How it works

1. Upload — file is validated and streamed to R2 staging; a PENDING job is created. The route returns 202 immediately.
2. Processing — the worker polls for PENDING jobs with pessimistic locking, then: downloads from staging → strips EXIF (orientation preserved) → reads metadata → optional moderation → generates WebP variants per source type → generates blurhash → uploads variants (and optional compressed original) → creates the Media record → marks the job COMPLETE and deletes the staging object.
3. Retry — failures increment attempts and re-queue the job until maxAttempts; stale locks (crashed worker) are recovered automatically after lockTimeout.

Requirements

• Node.js >= 20
• Fastify ^5
• @fastify/multipart registered before this plugin (for the upload route)
• A Prisma (or Prisma-shaped) client with the MediaQueue and Media models

License

Proprietary — All Rights Reserved X Enterprises. See LICENSE.