fix(bigquery-firestore-export): notify this instance's topic when linking a config

kits/bigquery-firestore-export/README.md

@@ -152,11 +152,40 @@ query on first deploy, then reconciles supported query, table, schedule,
 dataset, partitioning, and topic changes on later deploys. It retries transient
 failures up to five times with at least 30 seconds of backoff.
 
-Set `TRANSFER_CONFIG_NAME` to link an existing scheduled-query config without
-changing it. Otherwise the kit stores `extInstanceId` on the Firestore config
-document and uses that value to find the config on later deploys. BigQuery DTS
-does not support clearing a partitioning field once set; create a new transfer
-config to remove partitioning.
+Without `TRANSFER_CONFIG_NAME` the kit stores `extInstanceId` on the Firestore
+config document and uses that value to find its own config on later deploys.
+BigQuery DTS does not support clearing a partitioning field once set; create a
+new transfer config to remove partitioning.
+
+### Linking an existing scheduled query
+
+Set `TRANSFER_CONFIG_NAME` to link an existing scheduled-query config. The kit
+needs that config's completion notifications to read its run output, so it
+repoints them at this instance's topic. The update is masked to
+`notification_pubsub_topic` alone, so the query, schedule, destination, and
+display name stay exactly as the config's owner set them.
+
+Repointing the topic is a takeover, though. A config notifies one topic, so
+whatever consumed the previous one stops receiving runs: no error on either
+side, and nothing in the config's run history looks different. Removing this
+instance does not put the old topic back either. The deploy that changes it logs
+the previous value at warn, which is the only record of what to restore. Two
+instances linking the same config fail the same way, with the last deploy
+winning and the earlier instance going quiet.
+
+Linking also assumes the config is shaped the way the kit builds its own:
+
+- Its destination table template is `<name>_{run_time|"%H%M%S"}`. `processMessages`
+  substitutes that exact placeholder to work out which table a finished run
+  wrote to.
+- It lives in this project. The results query is built from `PROJECT_ID`, not
+  from the run notification.
+- Its destination dataset is in `BIGQUERY_DATASET_LOCATION`.
+
+A config outside that shape surfaces as a BigQuery table-not-found inside
+`processMessages` rather than as anything naming the mismatch, and
+`processMessages` retries, so it repeats instead of failing once. That shape is
+the only supported one.
 
 ## Firestore layout
 

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

@@ -65,6 +65,12 @@ function stringField(value: string | undefined): { stringValue: string } {
   return { stringValue: value ?? "" };
 }
 
+function notificationTopicName(
+  config: ResolvedBigqueryFirestoreExportConfig
+): string {
+  return `projects/${config.projectId}/topics/${config.pubSubTopic}`;
+}
+
 /** Creates the protobuf-shaped request used for a scheduled query. */
 export function createTransferConfigRequest(
   config: ResolvedBigqueryFirestoreExportConfig,
@@ -87,7 +93,7 @@ export function createTransferConfigRequest(
         },
       },
       schedule: config.schedule,
-      notificationPubsubTopic: `projects/${config.projectId}/topics/${config.pubSubTopic}`,
+      notificationPubsubTopic: notificationTopicName(config),
       ...(serviceAccountEmail
         ? { serviceAccountName: serviceAccountEmail }
         : {}),
@@ -129,6 +135,52 @@ export async function createTransferConfig(
   return created;
 }
 
+/**
+ * Points an adopted config at this deployment's notification topic. The mask
+ * covers only the topic: everything else, including the query, schedule and
+ * destination, stays as the config's owner set it.
+ *
+ * A config notifies exactly one topic, so this claims it. The previous value is
+ * logged at warn because nothing else records it, and it is the only way back.
+ */
+export async function updateNotificationTopic(
+  client: DataTransferClient,
+  transferConfig: TransferConfig,
+  config: ResolvedBigqueryFirestoreExportConfig
+): Promise<TransferConfig> {
+  const expectedTopic = notificationTopicName(config);
+  if (transferConfig.notificationPubsubTopic === expectedTopic) {
+    return transferConfig;
+  }
+  if (!transferConfig.name) {
+    throw new Error("BigQuery transfer config is missing its resource name");
+  }
+
+  const previousTopic = transferConfig.notificationPubsubTopic;
+  if (previousTopic) {
+    logs.notificationTopicTakeover(
+      transferConfig.name,
+      previousTopic,
+      expectedTopic
+    );
+  }
+
+  logs.updateNotificationTopic(transferConfig.name, expectedTopic);
+  const converted =
+    bigqueryDataTransfer.protos.google.cloud.bigquery.datatransfer.v1.UpdateTransferConfigRequest.fromObject(
+      {
+        transferConfig: {
+          name: transferConfig.name,
+          notificationPubsubTopic: expectedTopic,
+        },
+        updateMask: { paths: ["notification_pubsub_topic"] },
+      }
+    );
+  const [updated] = await client.updateTransferConfig(converted);
+  logs.notificationTopicUpdated(transferConfig.name, expectedTopic);
+  return updated;
+}
+
 /** Builds a minimal update mask while retaining unsupported immutable fields. */
 export async function constructUpdateTransferConfigRequest(
   client: DataTransferClient,
@@ -181,7 +233,7 @@ export async function constructUpdateTransferConfigRequest(
     updatedConfig.schedule = config.schedule;
   }
 
-  const expectedTopic = `projects/${config.projectId}/topics/${config.pubSubTopic}`;
+  const expectedTopic = notificationTopicName(config);
   if (expectedTopic !== transferConfig.notificationPubsubTopic) {
     updateMask.push("notification_pubsub_topic");
     updatedConfig.notificationPubsubTopic = expectedTopic;

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

@@ -23,6 +23,7 @@ import {
   createTransferConfig,
   type DataTransferClient,
   getTransferConfig,
+  updateNotificationTopic,
   updateTransferConfig,
 } from "./dts";
 import type { ResolvedBigqueryFirestoreExportConfig } from "./export-config";
@@ -109,7 +110,12 @@ export async function handleUpsertTransferConfig(
         `Transfer config not found: ${ctx.config.transferConfigName}`
       );
     }
-    await storeTransferConfig(ctx, linked);
+    const notifying = await updateNotificationTopic(
+      ctx.dataTransfer,
+      linked,
+      ctx.config
+    );
+    await storeTransferConfig(ctx, notifying);
     return;
   }
 

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

@@ -28,6 +28,7 @@ export {
   PARTITIONING_FIELD_REMOVAL_ERROR,
   PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX,
   type TransferConfig,
+  updateNotificationTopic,
   updateTransferConfig,
 } from "./dts";
 export {

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

@@ -122,6 +122,33 @@ export function transferConfigUpdated(name: string): void {
   logger.info("Updated BigQuery Data Transfer config", { name });
 }
 
+export function notificationTopicTakeover(
+  name: string,
+  previousTopic: string,
+  topic: string
+): void {
+  logger.warn(
+    "Taking over the notification topic of a linked transfer config. A config " +
+      "notifies one topic, so anything consuming the previous topic stops " +
+      "receiving runs, and removing this instance does not restore it.",
+    { name, previousTopic, topic }
+  );
+}
+
+export function updateNotificationTopic(name: string, topic: string): void {
+  logger.info("Pointing a linked transfer config at this instance's topic", {
+    name,
+    topic,
+  });
+}
+
+export function notificationTopicUpdated(name: string, topic: string): void {
+  logger.info("Updated the notification topic of a linked transfer config", {
+    name,
+    topic,
+  });
+}
+
 export function transferConfigNotFound(name: string): void {
   logger.error("BigQuery Data Transfer config not found", { name });
 }

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

@@ -14,12 +14,14 @@
  * limitations under the License.
  */
 
-import { describe, expect, test, vi } from "vitest";
+import { afterEach, describe, expect, test, vi } from "vitest";
+import { logger } from "firebase-functions";
 import {
   constructUpdateTransferConfigRequest,
   createTransferConfigRequest,
   type DataTransferClient,
   PARTITIONING_FIELD_REMOVAL_ERROR,
+  updateNotificationTopic,
 } from "../src/dts";
 import { resolveConfig } from "../src/export-config";
 
@@ -122,3 +124,93 @@ describe("constructUpdateTransferConfigRequest", () => {
     ).rejects.toThrow(PARTITIONING_FIELD_REMOVAL_ERROR);
   });
 });
+
+describe("updateNotificationTopic", () => {
+  const linked = {
+    name: "projects/p/locations/us/transferConfigs/c",
+    destinationDatasetId: "adopted_dataset",
+    schedule: "every 6 hours",
+    params: {
+      fields: {
+        query: { stringValue: "SELECT * FROM adopted.rows" },
+        destination_table_name_template: {
+          stringValue: 'adopted_{run_time|"%H%M%S"}',
+        },
+      },
+    },
+  };
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  test("updates only the notification topic when it differs", async () => {
+    const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
+    const updateTransferConfig = vi
+      .fn()
+      .mockResolvedValue([{ ...linked, notificationPubsubTopic: "updated" }]);
+    const client = {
+      updateTransferConfig,
+    } as unknown as DataTransferClient;
+
+    const updated = await updateNotificationTopic(
+      client,
+      { ...linked, notificationPubsubTopic: "projects/p/topics/ext-old-topic" },
+      config
+    );
+
+    expect(warn).toHaveBeenCalledWith(expect.any(String), {
+      name: linked.name,
+      previousTopic: "projects/p/topics/ext-old-topic",
+      topic: "projects/test-project/topics/kit-users-export-processMessages",
+    });
+    const request = updateTransferConfig.mock.calls[0][0];
+    expect(request.updateMask.paths).toEqual(["notification_pubsub_topic"]);
+    expect(request.transferConfig.notificationPubsubTopic).toBe(
+      "projects/test-project/topics/kit-users-export-processMessages"
+    );
+    expect(request.transferConfig.name).toBe(linked.name);
+    expect(request.transferConfig.destinationDatasetId).toBeFalsy();
+    expect(request.transferConfig.schedule).toBeFalsy();
+    expect(request.transferConfig.params).toBeFalsy();
+    expect(updated.notificationPubsubTopic).toBe("updated");
+  });
+
+  test("leaves a config already notifying this instance untouched", async () => {
+    const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
+    const updateTransferConfig = vi.fn();
+    const client = {
+      updateTransferConfig,
+    } as unknown as DataTransferClient;
+    const alreadyLinked = {
+      ...linked,
+      notificationPubsubTopic:
+        "projects/test-project/topics/kit-users-export-processMessages",
+    };
+
+    const returned = await updateNotificationTopic(
+      client,
+      alreadyLinked,
+      config
+    );
+
+    expect(updateTransferConfig).not.toHaveBeenCalled();
+    expect(returned).toBe(alreadyLinked);
+    expect(warn).not.toHaveBeenCalled();
+  });
+
+  test("does not warn about a takeover when the config had no topic", async () => {
+    const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined);
+    const updateTransferConfig = vi
+      .fn()
+      .mockResolvedValue([{ ...linked, notificationPubsubTopic: "updated" }]);
+    const client = {
+      updateTransferConfig,
+    } as unknown as DataTransferClient;
+
+    await updateNotificationTopic(client, linked, config);
+
+    expect(updateTransferConfig).toHaveBeenCalledTimes(1);
+    expect(warn).not.toHaveBeenCalled();
+  });
+});

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

@@ -21,13 +21,15 @@ import { resolveConfig } from "../src/export-config";
 const mocks = vi.hoisted(() => ({
   createTransferConfig: vi.fn(),
   getTransferConfig: vi.fn(),
+  updateNotificationTopic: vi.fn(),
   updateTransferConfig: vi.fn(),
   handleTransferRunMessage: vi.fn(),
 }));
 
 vi.mock("../src/dts", () => ({
   createTransferConfig: mocks.createTransferConfig,
   getTransferConfig: mocks.getTransferConfig,
+  updateNotificationTopic: mocks.updateNotificationTopic,
   updateTransferConfig: mocks.updateTransferConfig,
 }));
 
@@ -146,11 +148,18 @@ describe("handleUpsertTransferConfig", () => {
     });
   });
 
-  test("links an explicitly named transfer config without updating it", async () => {
+  test("links an explicitly named transfer config and repoints its topic", async () => {
     const linked = {
       name: "projects/p/locations/us/transferConfigs/config-2",
+      notificationPubsubTopic: "projects/p/topics/ext-old-topic",
+    };
+    const notifying = {
+      ...linked,
+      notificationPubsubTopic:
+        "projects/test-project/topics/kit-users-export-processMessages",
     };
     mocks.getTransferConfig.mockResolvedValue(linked);
+    mocks.updateNotificationTopic.mockResolvedValue(notifying);
     const { ctx, set } = makeContext({
       transferConfigName: linked.name,
     });
@@ -161,10 +170,15 @@ describe("handleUpsertTransferConfig", () => {
       ctx.dataTransfer,
       linked.name
     );
+    expect(mocks.updateNotificationTopic).toHaveBeenCalledWith(
+      ctx.dataTransfer,
+      linked,
+      ctx.config
+    );
     expect(mocks.updateTransferConfig).not.toHaveBeenCalled();
     expect(set).toHaveBeenCalledWith({
       extInstanceId: "users-export",
-      ...linked,
+      ...notifying,
     });
   });
 });