improve(loaders): add getClusterForTopic() helper

src/commands/kafkaClusters.test.ts

@@ -452,7 +452,7 @@ describe("commands/kafkaClusters.ts", () => {
 
     it("should fire the topicChanged event with change='deleted' after successful deletion", async () => {
       showInputBoxStub.resolves(TEST_CCLOUD_KAFKA_TOPIC.name);
-      stubbedLoader.getKafkaClustersForEnvironmentId.resolves([TEST_CCLOUD_KAFKA_CLUSTER]);
+      stubbedLoader.getClusterForTopic.resolves(TEST_CCLOUD_KAFKA_CLUSTER);
 
       await deleteTopicCommand(TEST_CCLOUD_KAFKA_TOPIC);
 

src/commands/kafkaClusters.ts

@@ -152,8 +152,7 @@ export async function deleteTopicCommand(topic: KafkaTopic) {
 
         // look up the parent Kafka cluster in order to fire the topicChanged event
         const loader = ResourceLoader.getInstance(topic.connectionId);
-        const clusters = await loader.getKafkaClustersForEnvironmentId(topic.environmentId);
-        const cluster = clusters.find((c) => c.id === topic.clusterId);
+        const cluster = await loader.getClusterForTopic(topic);
         if (cluster) {
           topicChanged.fire({ change: "deleted", cluster });
         }

src/commands/scaffold.test.ts

@@ -7,6 +7,7 @@ import {
   TEST_CCLOUD_SCHEMA_REGISTRY,
 } from "../../tests/unit/testResources";
 import { TEST_CCLOUD_FLINK_COMPUTE_POOL } from "../../tests/unit/testResources/flinkComputePool";
+import * as errors from "../errors";
 import type { CCloudResourceLoader } from "../loaders";
 import { CCloudEnvironment } from "../models/environment";
 import { KafkaTopic } from "../models/topic";
@@ -102,32 +103,37 @@ describe("commands/scaffold.ts", () => {
       );
     });
 
-    it("should show an error notification if environment for topic is not found", async () => {
+    it("should show an error notification and log to Sentry if no cluster is found for the topic", async () => {
       const showErrorNotificationWithButtonsStub = sandbox.stub(
         notifications,
         "showErrorNotificationWithButtons",
       );
+      const logErrorStub = sandbox.stub(errors, "logError");
 
-      // make a new topic with an environment ID that won't be found
-      const topicWithMissingEnv = new KafkaTopic({
+      // topic whose cluster can't be resolved within its environment
+      const topicWithMissingCluster = new KafkaTopic({
         ...TEST_CCLOUD_KAFKA_TOPIC,
         clusterId: "missing-cluster-id",
       });
+      stubbedResourceLoader.getClusterForTopic
+        .withArgs(topicWithMissingCluster)
+        .resolves(undefined);
 
-      await resourceScaffoldProjectCommand(topicWithMissingEnv);
+      await resourceScaffoldProjectCommand(topicWithMissingCluster);
 
       sinon.assert.notCalled(scaffoldProjectRequestStub);
       sinon.assert.calledOnce(showErrorNotificationWithButtonsStub);
       sinon.assert.calledWithExactly(
         showErrorNotificationWithButtonsStub,
         `Unable to find Kafka cluster for topic "${TEST_CCLOUD_KAFKA_TOPIC.name}".`,
       );
+      sinon.assert.calledOnce(logErrorStub);
     });
 
     it("should call scaffoldProjectRequest with correct parameters for KafkaTopic", async () => {
-      stubbedResourceLoader.getKafkaClustersForEnvironmentId
-        .withArgs(TEST_CCLOUD_KAFKA_TOPIC.environmentId)
-        .resolves([TEST_CCLOUD_KAFKA_CLUSTER]);
+      stubbedResourceLoader.getClusterForTopic
+        .withArgs(TEST_CCLOUD_KAFKA_TOPIC)
+        .resolves(TEST_CCLOUD_KAFKA_CLUSTER);
 
       await resourceScaffoldProjectCommand(TEST_CCLOUD_KAFKA_TOPIC);
 

src/commands/scaffold.ts

@@ -2,6 +2,7 @@
 
 import type * as vscode from "vscode";
 import { registerCommandWithLogging } from ".";
+import { logError } from "../errors";
 import { CCloudResourceLoader, ResourceLoader } from "../loaders";
 import type { CCloudFlinkComputePool } from "../models/flinkComputePool";
 import { KafkaCluster } from "../models/kafkaCluster";
@@ -52,9 +53,15 @@ export async function resourceScaffoldProjectCommand(
       telemetrySource = "cluster";
     } else {
       // KafkaTopic
-      const clusters = environment.kafkaClusters;
-      const cluster = clusters.find((c) => c.id === item.clusterId);
+      const cluster = await loader.getClusterForTopic(item);
       if (!cluster) {
+        // a topic should always resolve to a cluster in its environment; log to Sentry so we
+        // learn if this unexpected condition happens in the wild.
+        logError(
+          new Error("Unable to find Kafka cluster for topic during project scaffolding"),
+          "Kafka cluster not found for topic during project scaffolding",
+          { extra: { clusterId: item.clusterId, environmentId: item.environmentId } },
+        );
         void showErrorNotificationWithButtons(
           `Unable to find Kafka cluster for topic "${item.name}".`,
         );

src/loaders/resourceLoader.test.ts

@@ -471,6 +471,53 @@ describe("ResourceLoader::getTopicsForCluster()", () => {
   });
 });
 
+describe("ResourceLoader::getClusterForTopic()", () => {
+  let loaderInstance: ResourceLoader;
+  let sandbox: sinon.SinonSandbox;
+  let getKafkaClustersForEnvironmentIdStub: sinon.SinonStub;
+
+  beforeEach(() => {
+    sandbox = sinon.createSandbox();
+    loaderInstance = LocalResourceLoader.getInstance();
+    getKafkaClustersForEnvironmentIdStub = sandbox.stub(
+      loaderInstance,
+      "getKafkaClustersForEnvironmentId",
+    );
+  });
+
+  afterEach(() => {
+    sandbox.restore();
+  });
+
+  it("Raises error for a topic from a mismatched connection", async () => {
+    await assert.rejects(loaderInstance.getClusterForTopic(TEST_CCLOUD_KAFKA_TOPIC), (err) => {
+      return (err as Error).message.startsWith(`Mismatched connectionId ${LOCAL_CONNECTION_ID}`);
+    });
+
+    sinon.assert.notCalled(getKafkaClustersForEnvironmentIdStub);
+  });
+
+  it("Returns the cluster matching the topic's clusterId", async () => {
+    getKafkaClustersForEnvironmentIdStub
+      .withArgs(TEST_LOCAL_KAFKA_TOPIC.environmentId)
+      .resolves([TEST_LOCAL_KAFKA_CLUSTER]);
+
+    const cluster = await loaderInstance.getClusterForTopic(TEST_LOCAL_KAFKA_TOPIC);
+
+    assert.strictEqual(cluster, TEST_LOCAL_KAFKA_CLUSTER);
+  });
+
+  it("Returns undefined when no cluster in the environment matches the topic's clusterId", async () => {
+    getKafkaClustersForEnvironmentIdStub
+      .withArgs(TEST_LOCAL_KAFKA_TOPIC.environmentId)
+      .resolves([]);
+
+    const cluster = await loaderInstance.getClusterForTopic(TEST_LOCAL_KAFKA_TOPIC);
+
+    assert.strictEqual(cluster, undefined);
+  });
+});
+
 describe("ResourceLoader::getConsumerGroupsForCluster()", () => {
   let loaderInstance: LocalResourceLoader;
   let sandbox: sinon.SinonSandbox;

src/loaders/resourceLoader.ts

@@ -145,6 +145,28 @@ export abstract class ResourceLoader extends DisposableCollection implements IRe
     forceRefresh?: boolean,
   ): Promise<ConsumerGroup[]>;
 
+  /**
+   * Get the {@link KafkaCluster} that a given {@link KafkaTopic} belongs to.
+   *
+   * Connection-aware (implemented atop {@link getKafkaClustersForEnvironmentId}), so unlike a
+   * raw cluster-id lookup it resolves correctly for direct-connection topics.
+   *
+   * @param topic The topic to resolve. Must belong to the same connection as this loader.
+   * @returns The topic's {@link KafkaCluster}, or `undefined` if no cluster in the topic's
+   *   environment matches its {@linkcode KafkaTopic.clusterId}.
+   * @throws Error if the topic is not from the same connection as this loader.
+   */
+  public async getClusterForTopic(topic: KafkaTopic): Promise<KafkaCluster | undefined> {
+    if (topic.connectionId !== this.connectionId) {
+      throw new Error(
+        `Mismatched connectionId ${this.connectionId} for topic ${topic.name} (${topic.connectionId})`,
+      );
+    }
+
+    const clusters = await this.getKafkaClustersForEnvironmentId(topic.environmentId);
+    return clusters.find((cluster) => cluster.id === topic.clusterId);
+  }
+
   // Schema registry methods.
 
   /**