firestore-bigquery-changelog

SDK to log Firestore document changelog directly to BigQuery. This package helps you track every change in your Firestore collections and sync them to BigQuery for analysis.

Features

• Single-place registry — declare every Firestore → BigQuery mapping once via registerV1() / registerV2().
• Direct BigQuery writes — no API proxy. Firebase Functions V1 & V2 both supported.
• Multiple destinations per collection — fan out each change to several tables with independent pickKeys, upsertKeys, and transformRow.
• Upsert (MERGE) mode — composite keys, field aliases, in-process mutex against races.
• Zero-config tables — auto-create + schema migration (adds missing columns), optional time partitioning.

Installation

npm install firestore-bigquery-changelog

Quick Start (Firebase Functions V2)

Declare every collection in one place — the SDK handles the Cloud Function wiring.

1. Initialize

// config/changelog.ts
import { createChangelogTrigger, TRIGGER_TYPE } from 'firestore-bigquery-changelog';

export const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  credentials: process.env.CHANGELOG_CREDENTIALS,
});

export const changelogTriggers = changelog.registerV2({
  collections: [
    { collectionId: 'shops', trigger: TRIGGER_TYPE.ON_CREATE }, // onDocumentCreated('shops/{docId}', ...)
    { collectionId: 'subscriptions' },                          // onDocumentWritten (default)
    { collectionId: 'settings' },
    { collectionId: 'integrations' },
  ],
});

Only appId is required on createChangelogTrigger. Everything else has sensible defaults — see Configuration Reference.

2. Export the triggers

// index.ts
export { changelogTriggers } from './config/changelog';

Firebase auto-deploys each field of changelogTriggers as its own Cloud Function, named <exportName>-<key>:

• changelogTriggers-shops
• changelogTriggers-subscriptions
• changelogTriggers-settings
• changelogTriggers-integrations

That's it. No per-collection handler file, no manual onDocumentWritten(...) calls.

Config reference for registerV2

| Option | Level | Type | Default | Description | | :--- | :--- | :--- | :--- | :--- | | collectionId | per-collection | string | — | Required. Firestore collection name. Used as the return object key. | | trigger | per-collection | TRIGGER_TYPE.* | ON_WRITE | Trigger type: ON_WRITE, ON_CREATE, ON_UPDATE, ON_DELETE (or raw strings). | | document | per-collection | string | {collectionId}/{docId} | Override only for nested paths (e.g. 'shops/{shopId}/orders/{orderId}'). | | destinations | per-collection | DestinationConfig[] | single default changelog table | See Multiple Destination Tables. | | region | top-level or per-collection | string \| string[] | Firebase default | Cloud Function region(s). | | memory | top-level or per-collection | string | '256MiB' | Cloud Function memory. | | timeoutSeconds, concurrency, cpu, minInstances, maxInstances | top-level or per-collection | see Firebase docs | Firebase defaults | Passthrough to Firebase Functions v2 options. |

Per-collection overrides win over top-level defaults.

export const changelogTriggers = changelog.registerV2({
  memory: '512MiB',                                 // default for all triggers
  collections: [
    {
      collectionId: 'shops',
      trigger: TRIGGER_TYPE.ON_CREATE,
      region: 'asia-southeast1',                    // overrides default
      destinations: [
        { tableName: 'avada_customers', upsertKeys: ['shopifyDomain'] },
      ],
    },
    {
      collectionId: 'orders',
      document: 'shops/{shopId}/orders/{orderId}',  // nested path
    },
  ],
});

Raw handlers (for composition)

When a collection needs extra logic alongside the changelog (e.g. sending a welcome email, writing to additional BigQuery tables, calling an external API), compose the raw handler inside your own function instead of using registerV2.

// handlers/onCreateShop.ts
import { changelog } from './config/changelog';
import * as avadaioService from './services/avadaioService';

const onCreateShopChangelog = changelog.onCreate({ collectionId: 'shops' });

export default async function onCreateShop(shopSnap, context) {
  await Promise.allSettled([
    avadaioService.installApp({ recipient: shopSnap.data().email, ... }),
    onCreateShopChangelog(shopSnap, context),
  ]);
}

Then wire your handler to Firebase manually:

// index.ts
import { onDocumentCreated } from 'firebase-functions/v2/firestore';
import onCreateShop from './handlers/onCreateShop';

export const onCreateShopV2 = onDocumentCreated(
  'shops/{shopId}',
  event => onCreateShop(event.data, { timestamp: event.time, eventId: event.id })
);

Available raw handlers

| Handler | Trigger | Input | When to use | | :--- | :--- | :--- | :--- | | changelog.onWrite | V1 onWrite | (change, context) | Firebase Functions V1. | | changelog.onCreate | V1 onCreate | (snapshot, context) | V1, single-snapshot triggers. | | changelog.onUpdate | V1 onUpdate | (change, context) | V1. Alias of onWrite. | | changelog.onDelete | V1 onDelete | (snapshot, context) | V1, single-snapshot triggers. | | changelog.onWriteV2 | V2 onDocumentWritten | (event) or (change, context) | V2, with composition. | | changelog.onCreateV2 | V2 onDocumentCreated | (event) | V2, with composition. | | changelog.onUpdateV2 | V2 onDocumentUpdated | (event) or (change, context) | V2, with composition. | | changelog.onDeleteV2 | V2 onDocumentDeleted | (event) | V2, with composition. |

Note: onUpdate / onUpdateV2 are aliases of onWrite / onWriteV2. onCreate and onDelete handlers take a single snapshot and build the change pair internally.

Firebase Functions V1

Same ergonomics as V2 — use registerV1() for bulk registration, with V1-specific option types (memory is '256MB' etc., not '256MiB').

// config/changelog.ts
import * as functions from 'firebase-functions';
import { createChangelogTrigger, TRIGGER_TYPE } from 'firestore-bigquery-changelog';

export const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  credentials: functions.config().changelog.credentials,
});

export const changelogTriggers = changelog.registerV1({
  collections: [
    { collectionId: 'shops', trigger: TRIGGER_TYPE.ON_CREATE },
    { collectionId: 'subscriptions' },
    { collectionId: 'settings' },
    { collectionId: 'integrations' },
  ],
});

// index.ts
export { changelogTriggers } from './config/changelog';
// Deploys as: changelogTriggers-shops, changelogTriggers-subscriptions, ...

Config differences vs V2:

| Option | V1 type | V2 type | | :--- | :--- | :--- | | memory | '128MB' \| '256MB' \| '512MB' \| '1GB' \| '2GB' \| '4GB' \| '8GB' | '256MiB', '512MiB', '1GiB', ... | | region | string \| string[] — chained via .region(...) | string \| string[] | | timeoutSeconds, minInstances, maxInstances | supported (via .runWith(...)) | supported |

For composition, use raw handlers the same way as V2:

const onCreateShopChangelog = changelog.onCreate({ collectionId: 'shops' });

export const onShopCreate = functions.firestore
  .document('shops/{shopId}')
  .onCreate(onCreateShopChangelog);

Credentials

The credentials field on createChangelogTrigger accepts three formats (auto-detected):

• JSON string — e.g. from functions.config() (V1) or process.env (V2)
• Base64-encoded string — base64 of the JSON above
• JSON object — e.g. require('./service-account.json')

When omitted, the SDK uses new BigQuery() which picks up the default service account automatically (e.g. the Firebase project's service account).

Setting credentials via functions.config() (V1)

Firebase V1 uses firebase functions:config:set to store secrets. Because a service account JSON contains newlines inside private_key, the easiest way is base64:

# Encode + set in one command (Mac/Linux)
firebase functions:config:set changelog.credentials="$(base64 -i service-account.json)"

# Verify
firebase functions:config:get changelog.credentials

Then read it in code with the matching key path:

credentials: functions.config().changelog.credentials,  // ← key path matches `changelog.credentials`

The SDK auto-detects base64 and decodes it. A raw JSON string works too but is harder to quote correctly from the shell.

Local emulator: pull the live config into .runtimeconfig.json before starting the emulator:

firebase functions:config:get > .runtimeconfig.json
firebase emulators:start --only functions

Deploy: config changes only apply on the next deploy, not to already-running functions:

firebase deploy --only functions

Setting credentials via env var (V2)

V2 uses environment variables instead of functions.config(). Store the service account as a base64 string in .env or via Secret Manager:

# .env (committed as .env.example, real values in .env.local or Secret Manager)
CHANGELOG_CREDENTIALS=<base64-encoded-service-account-json>

credentials: process.env.CHANGELOG_CREDENTIALS,

Advanced Configuration

The destinations, upsertConfig, transformRow options work the same way whether you pass them inside registerV2 or to a raw handler (changelog.onWrite({...})). Examples below show the raw-handler form for brevity — drop the same destinations array into a registerV2 collection entry when using the bulk API.

Multiple Destination Tables

Use destinations to write a single collection's changes to multiple BigQuery tables, each with its own pickKeys, upsertKeys, and transformRow.

// Inside registerV2:
collections: [
  {
    collectionId: 'shops',
    destinations: [
      { tableName: 'shops_changelog', pickKeys: ['name', 'status'] },
      { tableName: 'avada_customers', upsertKeys: ['shopifyDomain'], pickKeys: ['shopifyDomain', 'email', 'plan'] },
    ],
  },
]

// Or via raw handler:
changelog.onWrite({
  collectionId: 'shops',
  destinations: [
    { tableName: 'shops_changelog', pickKeys: ['name', 'status'] },
    { tableName: 'avada_customers', upsertKeys: ['shopifyDomain'], pickKeys: ['shopifyDomain', 'email', 'plan'] },
  ],
})

Upsert Mode (MERGE)

Use upsertKeys on a destination to enable upsert mode. Instead of appending a new row for every change, the SDK will MERGE (insert or update) based on the specified keys.

This is useful for maintaining a single row per entity (e.g., a CRM table with one row per shop).

changelog.onWriteV2({
  collectionId: 'shops',
  destinations: [
    {
      tableName: 'avada_customers',
      upsertKeys: ['shopifyDomain'],
    },
  ],
})

When upsertKeys is set:

• MERGE uses ON condition with all upsert keys (auto-converted to snake_case).
• DELETE operations are skipped (no data to merge).
• An updated_at timestamp is automatically added.
• Fields not present in the document are left unchanged in BigQuery.

Upsert with Field Picking and Aliases

For more control over upsert behavior, use upsertConfig to specify which fields to pick from the document data JSON, and define field aliases for flexible matching.

changelog.onWrite({
  collectionId: 'shops',
  destinations: [
    {
      tableName: 'avada_customers',
      upsertConfig: {
        upsertKeys: ['shopifyDomain'],
        pickKeys: ['shopifyDomain', 'email', 'name', 'country', 'planDisplayName'],
        fieldAliases: {
          shopifyDomain: ['myshopifyDomain'],
        },
      },
    },
  ],
})

Custom Data Transformation

Use transformRow to modify the data before writing to BigQuery. This is useful for formatting dates, calculating fields, or cleaning up data.

changelog.onWrite({
  collectionId: 'users',
  destinations: [
    {
      tableName: 'users',
      transformRow: (row) => ({
        ...row,
        full_name: `${row.first_name} ${row.last_name}`,
        processed_at: new Date().toISOString()
      }),
    },
  ],
})

Time Partitioning

BigQuery time partitioning is enabled by default (DAY partition on timestamp field) for better query performance and lower costs on large tables.

// Default — already partitioned by `timestamp` (DAY), no config needed
const changelog = createChangelogTrigger({
  appId: 'orderLimit',
});

// Custom — partition by MONTH with 90-day expiration
const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  timePartitioning: {
    type: 'MONTH',
    field: 'timestamp',
    expirationMs: '7776000000', // 90 days
  },
});

// Disable partitioning
const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  timePartitioning: false,
});

| Option | Type | Default | Description | | :--- | :--- | :--- | :--- | | type | 'DAY' \| 'HOUR' \| 'MONTH' \| 'YEAR' | 'DAY' | Partition granularity. | | field | string | 'timestamp' | TIMESTAMP field to partition on. | | expirationMs | string | — | Auto-delete partitions older than this (in milliseconds). |

Note: Partitioning is only applied when creating new tables. Existing tables are not affected — you must drop and recreate them to change partitioning.

Clustering by shop_id

Every row carries a shop_id column and tables are clustered by shop_id by default, so queries filtering WHERE shop_id = '...' scan far fewer bytes (lower cost). Combined with the DAY partition on timestamp, a query scoped to one shop on one day reads almost nothing.

Zero-config for most apps: shop_id is read from the document's shopId field by default, which the common collections (subscriptions, shopInfos, settings, …) already have — so just bumping the SDK version works. The shops collection is excluded from clustering by default (the document is the shop, so clustering adds nothing there), so it needs no config either.

// Defaults — nothing below is required
const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  clustering: ['shop_id'],                  // default
  shopIdField: 'shopId',                    // default
  clusteringExcludeCollections: ['shops'],  // default
});

changelog.registerV2({
  collections: [
    { collectionId: 'subscriptions' },  // clustered by shop_id (← data.shopId)
    { collectionId: 'shops' },          // not clustered (excluded by default)
  ],
});

• Missing field → shop_id is written as null (no error; nulls cluster into one group).
• Only scalar values (string/number) become shop_id; objects/arrays → null.
• shopIdField is resolved per-collection, falling back to the trigger-level value, then 'shopId'. Set shopIdField: 'document_id' for any collection where the document id is the shop.
• Clustering fields not present in a table's schema are skipped automatically (e.g. custom upsert tables without shop_id).

Note: Existing tables self-heal — on the first write to a table per function instance, the SDK adds any missing columns (e.g. shop_id) and sets clustering, then caches the table so later inserts skip the check. So just bumping the SDK version is enough for pre-existing tables too; no migration script is required for schema/clustering. The optional script below only matters if you also want to backfill historical shop_id values.

Migrating existing tables

Schema + clustering migrate automatically on the next write (see the self-heal note above). What self-heal does not do is fill shop_id on historical rows — clustering only affects newly written rows. To also backfill old rows (from data.shopId) so they land clustered, run:

cd scripts
node addShopIdClustering.js --dry-run     # preview (skips *_shops_changelog)
node addShopIdClustering.js               # add shop_id column + set clustering
node addShopIdClustering.js --backfill    # also fill historical shop_id (extra cost)

The script is idempotent and skips *_shops_changelog tables (shops is excluded from clustering). --backfill skips rows still in the streaming buffer (BigQuery can't UPDATE those) and bounds the UPDATE to settled rows — buffered/recent rows get shop_id from the SDK once deployed, or on a later re-run.

Logger

Pass a logger to createChangelogTrigger to enable debug logging. Any object with info and error methods works (e.g. console, functions.logger).

// V1
import * as functions from 'firebase-functions';

const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  credentials: functions.config().changelog_credentials,
  logger: functions.logger,
});

// V2
const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  credentials: process.env.CHANGELOG_CREDENTIALS,
  logger: console,
});

API Reference

createChangelogTrigger(config)

| Option | Type | Description | | :--- | :--- | :--- | | appId | string | Required. App identifier (e.g. 'orderLimit'). Used as the table prefix and app_id column value. | | appPrefix | string | Optional. Defaults to appId value. Table name = {appPrefix}_{collectionId}_changelog. | | datasetId | string | Optional. BigQuery dataset ID (default: 'avada_product_raw'). | | credentials | object \| string | Optional. Service account credentials. Omit to use default credentials (new BigQuery()). Accepts: JSON object, JSON string, or base64-encoded string. Auto-detected. | | projectId | string | Optional. Firebase project ID (default: 'avada-crm'). | | changelogSchema | SchemaField[] | Optional. Custom schema for changelog tables. | | timePartitioning | boolean \| TimePartitioning | Optional. Default: true (DAY partition on timestamp). Pass object to customize, or false to disable. | | clustering | string[] | Optional. Default: ['shop_id']. BigQuery clustering fields. Fields absent from the table schema are skipped. Pass [] to disable. | | clusteringExcludeCollections | string[] | Optional. Default: ['shops']. Collections that are NOT clustered. Pass [] to cluster every collection. | | shopIdField | string | Optional. Default: 'shopId'. Document field used for the shop_id column. Sentinel 'document_id' → use the Firestore document id. Overridable per-collection. | | logger | Logger | Optional. Logger instance for debugging (must have info and error methods). |

CollectionConfig

| Option | Type | Description | | :--- | :--- | :--- | | collectionId | string | Required. Firestore collection name. | | shopIdField | string | Optional. Per-collection override of the trigger-level shopIdField. Use 'document_id' when the document id is the shop. | | destinations | DestinationConfig[] | Optional. Array of destination tables. Defaults to a single append-only changelog table. |

DestinationConfig

| Option | Type | Description | | :--- | :--- | :--- | | tableName | string | Optional. BigQuery destination table name. Defaults to {appPrefix}_{collectionId}_changelog. | | pickKeys | string[] | Optional. Fields to extract from the document as extra columns (auto snake_case). | | upsertKeys | string[] | Optional. camelCase field names for MERGE mode. When set, SDK will upsert instead of insert. | | upsertConfig | UpsertConfig | Optional. Advanced upsert config with pickKeys, fieldAliases, and upsertKeys. | | transformRow | function | Optional. Async/sync function to modify the row before writing. |

UpsertConfig

| Option | Type | Description | | :--- | :--- | :--- | | upsertKeys | string[] | Required. camelCase field names used as MERGE keys. | | pickKeys | string[] | Required. Fields to pick from the data JSON for the upsert row. | | fieldAliases | Record<string, string[]> | Optional. Alternative field names to match in the document data. |

registerV2(options)

Top-level options (applied as defaults to every collection):

| Option | Type | Description | | :--- | :--- | :--- | | collections | RegisterV2CollectionConfig[] | Required. Array of collection configs. | | region | string \| string[] | Optional. Default region for all triggers. | | memory | string | Optional. Default memory for all triggers. | | timeoutSeconds | number | Optional. Default timeout for all triggers. |

Per-collection (RegisterV2CollectionConfig — extends CollectionConfig):

| Option | Type | Description | | :--- | :--- | :--- | | collectionId | string | Required. Firestore collection name — also the key in the return object. | | trigger | TriggerType | Optional. TRIGGER_TYPE.ON_WRITE (default), ON_CREATE, ON_UPDATE, or ON_DELETE. | | document | string | Optional. Override default path {collectionId}/{docId} for nested collections. | | destinations | DestinationConfig[] | Optional. See DestinationConfig. | | region, memory, timeoutSeconds, concurrency, cpu, minInstances, maxInstances | Firebase types | Optional. Overrides the top-level defaults. |

Returns Record<string, CloudFunction> keyed by collectionId. Re-export the whole object (export { changelogTriggers }) for grouped deploy, or pick individual fields if you need custom export names.

registerV1(options)

Same shape as registerV2, with V1-specific option types. Top-level options:

| Option | Type | Description | | :--- | :--- | :--- | | collections | RegisterV1CollectionConfig[] | Required. Array of collection configs. | | region | string \| string[] | Optional. Default region for all triggers. | | memory | '128MB' \| '256MB' \| '512MB' \| '1GB' \| '2GB' \| '4GB' \| '8GB' | Optional. Default memory (V1 unit — no 'MiB'). | | timeoutSeconds | number | Optional. Default timeout for all triggers. |

Per-collection (RegisterV1CollectionConfig):

| Option | Type | Description | | :--- | :--- | :--- | | collectionId, trigger, document, destinations | same as V2 | See registerV2. TRIGGER_TYPE values work for both. | | region, memory, timeoutSeconds, minInstances, maxInstances | V1 types | Optional. Overrides top-level defaults. |

Raw handler return values

All raw handlers return async functions that resolve to DestinationResult[], where each result is { table, skipped?, error? }. See the Raw handlers table for signatures.

App Integration Guide

Each app should only sync the collections it actually needs — every handler adds a Cloud Function, so only push data that serves a purpose (e.g. CRM, analytics).

Below is an example using Order Limit. Other apps should follow the same pattern — define which collections to sync, pick the right trigger for each, and keep the list minimal.

Example: Order Limit (appId: 'orderLimit')

1. Collections & Triggers

| Collection | Trigger | Composed? | Reason | | :--- | :--- | :--- | :--- | | shops | onCreate | ❌ | Only track new shop installs | | subscriptions | onWrite | ❌ | Track all plan changes | | settings | onWrite | ❌ | Track all config changes | | integrations | onWrite | ❌ | Track all integration changes | | orderLimits | onWrite | ✅ | Track all rule changes + writes to secondary BigQuery tables | | purchaseActivities | onWrite | ✅ | Track all purchase events + writes to secondary tables |

2. Code

// config/changelog.ts
import { createChangelogTrigger, TRIGGER_TYPE } from 'firestore-bigquery-changelog';

export const changelog = createChangelogTrigger({
  appId: 'orderLimit',
  credentials: process.env.CHANGELOG_CREDENTIALS,
});

// Pure changelog — one shot via registerV2
export const changelogTriggers = changelog.registerV2({
  collections: [
    { collectionId: 'shops', trigger: TRIGGER_TYPE.ON_CREATE },
    { collectionId: 'subscriptions' },
    { collectionId: 'settings' },
    { collectionId: 'integrations' },
  ],
});

// handlers/trigger/orderLimits.ts — composed: multi-table writes
import { changelog } from '../../config/changelog';

const onWriteOrderLimitChangelog = changelog.onWrite({ collectionId: 'orderLimits' });

export const onWriteOrderLimitHandler = async (change, context) => {
  await Promise.allSettled([
    insertBigQueryTable(row, 'orderLimits_changelog'),
    insertBigQueryTable(row, 'orderLimits_changelog_partitioned'),
    onWriteOrderLimitChangelog(change, context),
  ]);
};

// index.ts
import { onDocumentWritten } from 'firebase-functions/v2/firestore';
import { onWriteOrderLimitHandler } from './handlers/trigger/orderLimits';

export { changelogTriggers } from './config/changelog';   // pure triggers (grouped)

export const onWriteOrderLimitsV2 = onDocumentWritten(    // composed
  'orderLimits/{docId}',
  event => onWriteOrderLimitHandler(event.data, { timestamp: event.time, eventId: event.id })
);
// ... same pattern for purchaseActivities

Total: 6 Cloud Functions (4 from changelogTriggers, 2 composed). Default memory 256MiB is sufficient.

3. BigQuery Output

Each row written to BigQuery has this structure (table name: orderLimit_{collectionId}_changelog):

| Column | Type | Example | | :--- | :--- | :--- | | timestamp | TIMESTAMP | 2026-04-16T08:30:00.000Z | | event_id | STRING | abc123-def456 | | document_name | STRING | projects/avada-crm/databases/(default)/documents/shops/shop_001 | | operation | STRING | CREATE, UPDATE, or DELETE | | data | STRING (JSON) | {"shopifyDomain":"mystore.myshopify.com","email":"[email protected]","plan":"premium"} | | old_data | STRING (JSON) | null (on CREATE) or previous document state | | document_id | STRING | shop_001 | | app_id | STRING | orderLimit | | shop_id | STRING | shop_001 (clustering key — from the shopId field; null for excluded collections like shops) |

Example query:

SELECT
  timestamp,
  operation,
  document_id,
  JSON_VALUE(data, '$.shopifyDomain') AS domain,
  JSON_VALUE(data, '$.plan') AS plan
FROM `avada-crm.avada_product_raw.orderLimit_shops_changelog`
WHERE operation = 'CREATE'
ORDER BY timestamp DESC
LIMIT 10;

Do NOT add handlers for collections that are not in this list (e.g. products is not needed for CRM).

Project Structure

src/
  index.ts                  — Public exports
  config.ts                 — Default changelog schema
  types.ts                  — All TypeScript interfaces
  utils.ts                  — Helpers (toSnakeCase, getWriteType, generateDefaultRow, pickTriggerData)
  createChangelogTrigger.ts — Main trigger factory
  bigquery/
    index.ts                — Barrel export
    credentials.ts          — Credential parsing & BigQuery client construction
    operations.ts           — Table schema management, insertRow, upsertRow
    upsertLock.ts           — In-process mutex for upsert serialization
    destinationProcessor.ts — Destination routing & processing

Development

Building the project

npm run build

Publishing to NPM

1. Update the version in package.json.
2. Publish:

   npm publish --access public

   The prepublishOnly script will automatically run npm run build before publishing.

License

MIT