@fastpix/fp-sync-engine

Mirror your FastPix video and live stream state into Postgres. Webhook-driven, portable, and idempotent.

Introduction

The FastPix Data Sync Engine synchronizes media, live streams, and uploads from your FastPix account to a PostgreSQL database. It processes webhook events, retrieves the latest resource from the FastPix API, and updates your local database.

It is a framework-free TypeScript library that works with any Postgres database and runs on both Node.js and Deno, with no platform lock-in. It is intended for use with Node.js 20 and above.

Features

• Synchronize FastPix media and live stream data to PostgreSQL.
• Verify FastPix webhook signatures.
• Automatically reconcile webhook events with the FastPix API.
• Backfill existing media and live streams.
• Support Node.js and Deno runtimes.
• Idempotent processing with automatic retries.

Prerequisites

Environment and Version Support

Requirement	Version	Description
Node.js	20+	Core runtime environment (also runs on Deno / edge runtimes)
Postgres	Any	Any provider — Neon, RDS, self-hosted, or any managed Postgres
npm / yarn / pnpm / bun	Latest	Package manager for dependencies
Internet	Required	FastPix API communication and authentication

Pro Tip: We recommend Node.js 20+ for optimal performance and the latest language features.

Before you begin

To get started, make sure you have the following:

• FastPix API token — a Token ID and Token Secret. The FastPix API is authenticated with HTTP Basic Auth, where the Token ID is the username and the Token Secret is the password. Follow the Authentication with Basic Auth guide to obtain them. These are used to re-fetch records from the API.
• A FastPix webhook signing secret — used to verify inbound webhooks (separate from the API token).
• A Postgres database from any provider (Neon, RDS, self-hosted, or any managed Postgres). You need its connection string, e.g. postgres://user:pass@host:5432/db.

Environment Variables (Optional)

Configure your credentials using environment variables for enhanced security and convenience:

# FastPix API token (used to re-fetch records from the API)
export FASTPIX_TOKEN_ID="your-token-id"
export FASTPIX_TOKEN_SECRET="your-token-secret"

# Webhook signing secret (used to verify inbound webhooks — separate from the API token)
export FASTPIX_WEBHOOK_SECRET="your-webhook-signing-secret"

# Your Postgres connection string
export DATABASE_URL="postgres://user:pass@host:5432/db"

Security Note: Never commit your credentials to version control. Use environment variables or a secure credential management system.

Table of Contents

• Setup
• Example Usage
• How it works
• Configuration options
• API reference
• Backfill and reconciliation
• Database schema
• Supported webhook events
• Deno and edge runtimes
• Migrations
• Runtime support
• Development

Setup

Installation

Install the sync engine using your preferred package manager:

npm install @fastpix/fp-sync-engine

PNPM

pnpm add @fastpix/fp-sync-engine

Bun

bun add @fastpix/fp-sync-engine

Yarn

yarn add @fastpix/fp-sync-engine

The postgres driver is included as a dependency, so that's all you need to run the engine. (If your own code also queries the mirror tables directly, add postgres to your app too.)

Imports

This package ships both ES modules and CommonJS, plus TypeScript type declarations.

ES Modules (Recommended)

import { FastPixSync, runMigrations, InvalidSignatureError } from "@fastpix/fp-sync-engine";

CommonJS

const { FastPixSync, runMigrations, InvalidSignatureError } = require("@fastpix/fp-sync-engine");

Initialization

Create the schema once, then initialize the engine with your credentials:

import { FastPixSync, runMigrations } from "@fastpix/fp-sync-engine";

// Create or upgrade the `fastpix` schema. Idempotent — safe to run on every boot.
await runMigrations({ databaseUrl: "postgres://user:password@localhost:5432/database" });

const sync = new FastPixSync({
  databaseUrl: "postgres://user:password@localhost:5432/database",
  fastpixTokenId: "your-fastpix-token-id",
  fastpixTokenSecret: "your-fastpix-token-secret",
  fastpixWebhookSecret: "your-webhook-secret",

  // Optional: re-fetch the full record from the FastPix API on each event (default: true).
  // Set false to write straight from the webhook payload (it falls back to a re-fetch if the
  // payload is incomplete).
  revalidateEntityViaFastPixApi: true,

  // Optional: also fetch entities REFERENCED by each backfilled/reconciled record (default: false).
  backfillRelatedEntities: false,

  // Optional: maximum number of Postgres connections (default: 10).
  maxPostgresConnections: 10,

  // Optional: custom logger (default: console). In production it must redact secrets.
  logger: console,
});

See Configuration options for the full list.

Example Usage

Receive a FastPix webhook in an Express app and sync it to Postgres:

import express from "express";
import { FastPixSync, runMigrations, InvalidSignatureError } from "@fastpix/fp-sync-engine";

await runMigrations({ databaseUrl: process.env.DATABASE_URL });

const sync = new FastPixSync({
  databaseUrl: process.env.DATABASE_URL,
  fastpixWebhookSecret: process.env.FASTPIX_WEBHOOK_SECRET,
  fastpixTokenId: process.env.FASTPIX_TOKEN_ID,
  fastpixTokenSecret: process.env.FASTPIX_TOKEN_SECRET,
});

const app = express();

// IMPORTANT: take the RAW body (express.text) — the signature is verified over the exact bytes.
app.post("/webhook", express.text({ type: "*/*" }), async (req, res) => {
  try {
    await sync.processWebhook(req.body, req.headers);
    res.sendStatus(202); // always 202 for a signature-valid event so FastPix stops retrying
  } catch (err) {
    res.sendStatus(err instanceof InvalidSignatureError ? 401 : 500);
  }
});

app.listen(3000, () => console.log("Listening on :3000/webhook"));

Acknowledge the webhook quickly with a 202. Because processing is idempotent, correctness doesn't depend on the response.

How it works

flowchart LR
  W([Webhook]) --> V{Verify signature}
  V -->|invalid| X([reject / 401])
  V -->|valid| S[(Store in webhook_events)]
  S --> R[Route + derive status from type]
  R --> Q{revalidate?}
  Q -->|true default| A[Re-fetch full record from API]
  Q -->|false| P[Use webhook payload]
  A --> U[(Upsert mirror table)]
  P --> U

Loading

1. Verify. The engine computes an HMAC-SHA256 hash over the raw body using the base64 secret and compares it in constant time, before parsing the body.
2. Store. The engine writes the raw event to webhook_events, keyed by event ID, and ignores duplicates (safe against FastPix's up-to-30 retries per event).
3. Resolve. The engine routes the event to its entity and derives the status from the event type. By default it re-fetches the full record from the FastPix API — treating the event as a trigger and the API as the source of truth.
4. Write. The engine upserts the record into its mirror table. Child collections fold into JSON columns, and *.deleted events hard-delete the record.

Split ingest and processing for queue-based deployments

const { eventId, isDuplicate } = await sync.ingestWebhook(rawBody, headers);
// Enqueue eventId, then in a worker:
await sync.processStoredEvent(eventId);

Configuration options

Pass the following options to the FastPixSync constructor:

Option	Type	Default	Description
databaseUrl	string	Required	Any Postgres connection string.
fastpixWebhookSecret	string	Required	The base64 webhook signing secret.
fastpixTokenId	string	None	The FastPix API token ID, used for re-fetches.
fastpixTokenSecret	string	None	The FastPix API token secret.
fastpixApiBaseUrl	string	https://api.fastpix.com/v1	The base URL of the FastPix API.
revalidateEntityViaFastPixApi	boolean	true	Re-fetch the full record for each event. When false, the engine writes from the webhook payload and falls back to a re-fetch if the payload is incomplete.
backfillRelatedEntities	boolean	false	Also fetch entities that each record references.
maxProcessAttempts	number	5	Attempts before the engine dead-letters an event.
maxPostgresConnections	number	10	Maximum size of the Postgres connection pool.
fastpixApiRetryBaseMs	number	250	Base backoff, in milliseconds, for re-fetch retries.
fastpixBurstWindowMs	number	500	Window, in milliseconds, in which the engine collapses re-fetches for the same resource.
fetch	typeof fetch	Global fetch	A custom fetch implementation, for tests or custom runtimes.
logger	Pick<Console, 'log' | 'warn' | 'error'>	console	A custom logger. In production, the logger must redact secrets.

API reference

Export	Signature	Description
FastPixSync	new (options)	The client.
.processWebhook	(rawBody, headers) => Promise<ProcessResult>	Verifies, stores, and syncs an event in one call.
.ingestWebhook	(rawBody, headers) => Promise<IngestResult>	Verifies and stores an event, for queue-based ingest.
.processStoredEvent	(eventId) => Promise<ProcessResult>	Processes a stored event, for the worker half of a queue.
.syncBackfill	({ object }) => Promise<void>	Paginates the API and imports existing records.
.reconcileRecent	({ sinceHours }) => Promise<{ media, liveStreams, deleted }>	Re-fetches recently active records; deletes rows that are gone (404) on FastPix.
.getMetrics	() => Promise<SyncMetrics>	Returns row and queue counts for monitoring.
.close	() => Promise<void>	Closes the Postgres pool.
runMigrations	({ databaseUrl }) => Promise<void>	Creates or upgrades the schema. Idempotent and self-healing.
verifySignature	(...) => Promise<boolean>	Verifies a signature independently of the client.
InvalidSignatureError	class	Thrown by processWebhook() and ingestWebhook() when a signature is invalid.

Error Handling

processWebhook() and ingestWebhook() throw InvalidSignatureError when a signature is invalid. When a re-fetch fails, the engine marks the event as failed rather than throwing to the caller or the queue.

Backfill and reconciliation

To import records that existed before you wired up webhooks, call syncBackfill(). The operation is resumable through a saved cursor:

// Backfill all data
await sync.syncBackfill({ object: "all" });

// Or backfill specific object types
await sync.syncBackfill({ object: "media" });
await sync.syncBackfill({ object: "live_streams" });

syncBackfill() resolves when the import finishes (it returns void); progress is checkpointed to sync_state so a crash resumes from the last page. Uploads have no list endpoint on the FastPix API, so they're synced from webhooks only.

To heal missed events, re-fetch recently active records. Run this on a nightly schedule:

const { media, liveStreams, deleted } = await sync.reconcileRecent({ sinceHours: 24 });
console.log(`Reconciled ${media} media and ${liveStreams} live streams`);
console.log(`Deleted ${deleted.media} media and ${deleted.liveStreams} live streams gone on FastPix`);

Reconcile makes the database a true mirror of FastPix, including deletions: a record that returns a definitive 404 (gone on FastPix) is deleted locally — the only way a live-stream deletion is detected, since FastPix sends no live_stream.deleted webhook. Any other error (5xx or network) skips the row and never deletes. Only rows updated within sinceHours are checked, so widen the window (for example sinceHours: 168) to reach older records.

Database schema

runMigrations() creates four mirror tables and a sync_state cursor under the fastpix schema. Child collections (playback IDs, tracks, AI results, simulcast responses, recorded VODs) fold into jsonb columns on their parent record — there are no child tables.

Column names mirror the FastPix API field names in camelCase (maxResolution, playbackIds, simulcastResponses), so double-quote them in SQL. The primary keys are media."mediaId" and live_streams."streamId". Boolean fields (e.g. trial, isAudioOnly) are stored as native boolean, even when FastPix sends them as the string "true"/"false".

Table	Contents
media	On-demand videos. Primary key mediaId. Includes playbackIds, tracks, generatedSubtitles, and the AI columns summary, chapters, namedEntities, moderation (all jsonb). streamId holds the source live stream on a VOD recording.
live_streams	Live streams. Primary key streamId. Includes playbackIds, simulcastResponses, and mediaIds (jsonb). Stores stream secrets.
uploads	Direct-upload sessions.
webhook_events	The raw event log, with idempotency keys and processing status.

Supported webhook events

The engine supports the following FastPix webhook events. Any event type it doesn't recognize is still stored in webhook_events (marked unknown), never dropped.

Media

• video.media.created
• video.media.ready
• video.media.failed
• video.media.updated
• video.media.deleted
• video.media.non_standard_format.detected
• video.media.live_stream.created
• video.media.track.*
• video.media.source.*
• video.playbackId.*

AI

• video.mediaAI.* — chapters, summary, named entities, moderation
• video.media.ai.* — advanced summary, attributed transcript

Live streams

• video.live_stream.created
• video.live_stream.preparing
• video.live_stream.active
• video.live_stream.idle
• video.live_stream.recording
• video.live_stream.disconnected
• video.live_stream.deleted
• video.live_stream.updated
• video.live_stream.simulcast_target.*

Uploads

• video.media.upload
• video.media.upload.cancelled
• video.upload.media_created

Deno and edge runtimes

The engine runs on any runtime with fetch and Postgres, including Deno and edge functions:

import { FastPixSync, InvalidSignatureError } from "npm:@fastpix/fp-sync-engine";

const sync = new FastPixSync({
  databaseUrl: Deno.env.get("DATABASE_URL"),
  fastpixWebhookSecret: Deno.env.get("FASTPIX_WEBHOOK_SECRET"),
  fastpixTokenId: Deno.env.get("FASTPIX_TOKEN_ID"),
  fastpixTokenSecret: Deno.env.get("FASTPIX_TOKEN_SECRET"),
});

Deno.serve(async (req) => {
  try {
    const body = await req.text();
    await sync.processWebhook(body, Object.fromEntries(req.headers));
    return new Response("OK", { status: 202 });
  } catch (e) {
    return new Response("Bad Request", { status: e instanceof InvalidSignatureError ? 401 : 400 });
  }
});

A companion @fastpix/supabase package provides an optional ready-made Supabase deployment (Edge Functions, a queue, and scheduled reconcile) built on this engine — use it only if you're on Supabase.

Migrations

runMigrations() is idempotent and self-healing, so it's safe to run on every boot. On an existing database, it:

• Adds new columns with ADD COLUMN IF NOT EXISTS.
• Renames legacy snake_case columns to their camelCase API names, including the primary keys, preserving data.
• Converts boolean fields stored as text to native boolean, and JSON columns left as text to jsonb.
• Drops obsolete child tables, whose contents now fold into JSON columns.

A database that an earlier version set up repairs itself the next time the engine starts.

Runtime support

The core uses only Web-standard APIs (fetch, Web Crypto), so the same build runs on:

• Node.js 20+ — your server, backfill/reconcile scripts, any consumer app.
• Deno — edge functions and edge runtimes, via npm:@fastpix/fp-sync-engine or a bundled copy.

It does not depend on Node-only modules, so nothing needs to change between runtimes.

Development

To build and type-check the library:

npm run build      # tsup -> dist/ (ESM + CJS + types)
npm run typecheck  # tsc --noEmit

We value community contributions and feedback. Feel free to open issues or submit pull requests.

License

MIT © FastPix, Inc.