feat(firestore-bigquery-export): add the sync queue params

kits/firestore-bigquery-export/src/config.ts

@@ -20,9 +20,10 @@ import type {
   TimePartitioningGranularity,
 } from "@firebaseextensions/firestore-bigquery-change-tracker";
 import { LogLevel } from "@firebaseextensions/firestore-bigquery-change-tracker";
-import type { Expression } from "firebase-functions/params";
+import type { Expression, IntParam } from "firebase-functions/params";
 import {
   defineBoolean,
+  defineInt,
   defineString,
   projectID,
   select,
@@ -102,6 +103,7 @@ export interface ConfigExpressions {
   datasetId: ConfigExpression<string>;
   tableId: ConfigExpression<string>;
   database: ConfigExpression<string>;
+  maxDispatchesPerSecond: ConfigExpression<number>;
 }
 
 /**
@@ -290,6 +292,36 @@ const params = {
       "This (optional) parameter will allow you to specify a collection for which failed BigQuery updates will be written to.",
     default: "",
   }),
+  maxDispatchesPerSecond: defineInt("MAX_DISPATCHES_PER_SECOND", {
+    label: "Maximum number of synced documents per second",
+    description:
+      "This parameter will set the maximum number of synchronized documents per second with BQ. Please note, any other external updates to a Big Query table will be included within this quota. Ensure that you have set a low enough number to compensate. Defaults to 100.",
+
+    default: 100,
+    input: {
+      text: {
+        example: "100",
+
+        validationRegex: /^([1-9]|[1-9][0-9]|[1-4][0-9]{2}|500)$/,
+        validationErrorMessage: "Please select a number between 1 and 500",
+      },
+    },
+  }),
+  maxEnqueueAttempts: defineInt("MAX_ENQUEUE_ATTEMPTS", {
+    label: "Maximum number of enqueue attempts",
+    description:
+      "This parameter will set the maximum number of attempts to enqueue a document to cloud tasks for export to BigQuery.",
+
+    default: 3,
+    input: {
+      text: {
+        example: "3",
+
+        validationRegex: /^(10|[1-9])$/,
+        validationErrorMessage: "Please select an integer between 1 and 10",
+      },
+    },
+  }),
   transformFunction: defineString("TRANSFORM_FUNCTION", {
     label: "Transform function URL",
     description:
@@ -452,6 +484,7 @@ export const CONFIG_EXPRESSIONS: ConfigExpressions = {
   datasetId: params.datasetId,
   tableId: params.tableId,
   database: params.database,
+  maxDispatchesPerSecond: params.maxDispatchesPerSecond,
 };
 
 function timePartitioning(
@@ -604,6 +637,19 @@ function optional(value: string): string | undefined {
   return value.length > 0 ? value : undefined;
 }
 
+/**
+ * Reads an int param, reporting a missing or blank env var as `undefined`.
+ *
+ * `IntParam.value()` is `parseInt(env || "0", 10) || 0` and never consults the
+ * declared default, so an unset param has to reach `resolveExportConfig` as
+ * `undefined` for the documented default to apply. An explicit `0` is a real
+ * setting and is preserved.
+ */
+function optionalInt(param: IntParam): number | undefined {
+  const raw = process.env[param.name]?.trim();
+  return raw === undefined || raw === "" ? undefined : param.value();
+}
+
 /**
  * Resolves all deploy-time params into an {@link ExportConfig}.
  *
@@ -646,5 +692,7 @@ export function configFromEnv(): ExportConfig {
     transformFunction: optional(params.transformFunction.value()),
     kmsKeyName: optional(params.kmsKeyName.value()),
     logLevel: normalizeLogLevel(params.logLevel.value()),
+    maxDispatchesPerSecond: optionalInt(params.maxDispatchesPerSecond),
+    maxEnqueueAttempts: optionalInt(params.maxEnqueueAttempts),
   };
 }

kits/firestore-bigquery-export/src/export-config.ts

@@ -86,6 +86,17 @@ export interface ExportConfig {
 
   /** Log verbosity. Defaults to `info`. */
   logLevel?: ConfigValue<TrackerLogLevel | LogLevel>;
+
+  /**
+   * Cloud Tasks dispatch rate for the `syncBigQuery` queue, in tasks per
+   * second. Defaults to `100`.
+   */
+  maxDispatchesPerSecond?: ConfigValue<number>;
+  /**
+   * How many times the trigger tries to enqueue a failed write onto the
+   * `syncBigQuery` queue before giving up. Defaults to `3`.
+   */
+  maxEnqueueAttempts?: ConfigValue<number>;
 }
 
 /** {@link ExportConfig} with all defaults applied. */
@@ -109,6 +120,8 @@ export interface ResolvedExportConfig {
   transformFunction?: string;
   kmsKeyName?: string;
   logLevel: TrackerLogLevel;
+  maxDispatchesPerSecond: number;
+  maxEnqueueAttempts: number;
 }
 
 function isExpression<T extends string | number | boolean | string[]>(
@@ -165,6 +178,10 @@ export function resolveExportConfig(
     transformFunction: resolveOptionalConfigValue(config.transformFunction),
     kmsKeyName: resolveOptionalConfigValue(config.kmsKeyName),
     logLevel: (logLevel as TrackerLogLevel) ?? "info",
+    maxDispatchesPerSecond:
+      resolveOptionalConfigValue(config.maxDispatchesPerSecond) ?? 100,
+    maxEnqueueAttempts:
+      resolveOptionalConfigValue(config.maxEnqueueAttempts) ?? 3,
   };
 }
 

kits/firestore-bigquery-export/tests/config.test.ts

@@ -44,8 +44,11 @@ vi.mock("firebase-functions/params", () => ({
         : opts?.default?.value() ?? "",
     toString: () => `params.${_name}`,
   }),
+  // Mirrors the real IntParam: a missing or blank env var resolves to 0, and
+  // the declared default never reaches runtime.
   defineInt: (_name: string, opts?: { default?: number }) => ({
-    value: () => opts?.default ?? 0,
+    name: _name,
+    value: () => Number.parseInt(process.env[_name] || "0", 10) || 0,
     toString: () => `params.${_name}`,
   }),
   defineBoolean: (_name: string, opts?: { default?: boolean }) => ({
@@ -171,11 +174,67 @@ describe("configFromEnv", () => {
     expect(resolveExportConfig(config)).not.toHaveProperty("location");
   });
 
+  test("reports unset queue params as undefined so the documented defaults apply", () => {
+    // IntParam.value() resolves an unset var to 0, which would defeat the
+    // `?? 100` / `?? 3` fallbacks in resolveExportConfig.
+    const config = configFromEnv();
+    expect(config.maxDispatchesPerSecond).toBeUndefined();
+    expect(config.maxEnqueueAttempts).toBeUndefined();
+
+    const resolved = resolveExportConfig(config);
+    expect(resolved.maxDispatchesPerSecond).toBe(100);
+    expect(resolved.maxEnqueueAttempts).toBe(3);
+  });
+
+  test("reports a blank queue param as undefined", () => {
+    vi.stubEnv("MAX_ENQUEUE_ATTEMPTS", "  ");
+    expect(configFromEnv().maxEnqueueAttempts).toBeUndefined();
+    vi.unstubAllEnvs();
+  });
+
+  test("passes an explicit queue param through", () => {
+    vi.stubEnv("MAX_ENQUEUE_ATTEMPTS", "7");
+    vi.stubEnv("MAX_DISPATCHES_PER_SECOND", "250");
+    const config = configFromEnv();
+    expect(config.maxEnqueueAttempts).toBe(7);
+    expect(config.maxDispatchesPerSecond).toBe(250);
+    vi.unstubAllEnvs();
+  });
+
   test("exposes deploy-time expressions for trigger metadata", () => {
     expect(CONFIG_EXPRESSIONS.collectionPath.toString()).toBe(
       "params.COLLECTION_PATH"
     );
     expect(CONFIG_EXPRESSIONS.database.toString()).toBe("params.DATABASE");
+    expect(CONFIG_EXPRESSIONS.maxDispatchesPerSecond.toString()).toBe(
+      "params.MAX_DISPATCHES_PER_SECOND"
+    );
     expect(CONFIG_EXPRESSIONS).not.toHaveProperty("location");
   });
 });
+
+describe("resolveExportConfig queue defaults", () => {
+  test("applies the extension's queue defaults when unset", () => {
+    const resolved = resolveExportConfig({
+      collectionPath: "users",
+      datasetId: "ds",
+      tableId: "tbl",
+      projectId: "p",
+    });
+    expect(resolved.maxDispatchesPerSecond).toBe(100);
+    expect(resolved.maxEnqueueAttempts).toBe(3);
+  });
+
+  test("passes explicit queue values through", () => {
+    const resolved = resolveExportConfig({
+      collectionPath: "users",
+      datasetId: "ds",
+      tableId: "tbl",
+      projectId: "p",
+      maxDispatchesPerSecond: 250,
+      maxEnqueueAttempts: 5,
+    });
+    expect(resolved.maxDispatchesPerSecond).toBe(250);
+    expect(resolved.maxEnqueueAttempts).toBe(5);
+  });
+});

kits/firestore-bigquery-export/tests/export-config.test.ts

@@ -107,4 +107,14 @@ describe("toTrackerConfig", () => {
     );
     expect(tracker.bqProjectId).toBe("analytics-project");
   });
+
+  test("wires the backup collection into the tracker (queue exhaustion durability)", () => {
+    const tracker = toTrackerConfig(
+      resolveExportConfig({ ...base, backupCollectionId: "bq_failures" })
+    );
+    expect(tracker.backupTableId).toBe("bq_failures");
+
+    const withoutBackup = toTrackerConfig(resolveExportConfig(base));
+    expect(withoutBackup.backupTableId).toBeUndefined();
+  });
 });