@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

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.