fix(tool): modify list schema to use SchemaRegistryClient to be accessible for CP (#37)

* feat(schema): Enhance ListSchemasHandler with improved error handling and logging

- Integrated SchemaRegistryClient for fetching schema subjects and metadata.
- Updated argument validation to include a boolean for deleted schemas.
- Enhanced logging to provide detailed debug information during schema retrieval.
- Improved error handling for fetching schema metadata and versions, ensuring clearer error messages in responses.

* refactor(schema): Remove baseUrl from ListSchemasHandler and update required environment variables

- Eliminated the baseUrl argument from the ListSchemasHandler, simplifying the argument structure.
- Updated the getRequiredEnvVars method to include SCHEMA_REGISTRY_ENDPOINT as a necessary environment variable, enhancing configuration management.

Author: eddyv

src/confluent/tools/handlers/schema/list-schemas-handler.ts

@@ -1,3 +1,4 @@
+import { SchemaRegistryClient } from "@confluentinc/schemaregistry";
 import { ClientManager } from "@src/confluent/client-manager.js";
 import { CallToolResult } from "@src/confluent/schema.js";
 import {
@@ -6,17 +7,10 @@ import {
 } from "@src/confluent/tools/base-tools.js";
 import { ToolName } from "@src/confluent/tools/tool-name.js";
 import { EnvVar } from "@src/env-schema.js";
-import env from "@src/env.js";
-import { wrapAsPathBasedClient } from "openapi-fetch";
+import { logger } from "@src/logger.js";
 import { z } from "zod";
 
 const listSchemasArguments = z.object({
-  baseUrl: z
-    .string()
-    .describe("The base URL of the Schema Registry REST API.")
-    .url()
-    .default(() => env.SCHEMA_REGISTRY_ENDPOINT ?? "")
-    .optional(),
   latestOnly: z
     .boolean()
     .describe("If true, only return the latest version of each schema.")
@@ -26,44 +20,140 @@ const listSchemasArguments = z.object({
     .string()
     .describe("The prefix of the subject to list schemas for.")
     .optional(),
-  deleted: z.string().describe("List deleted schemas.").optional(),
+  deleted: z
+    .boolean()
+    .describe("List deleted schemas. (Only used if latestOnly is false)")
+    .default(false)
+    .optional(),
 });
 
 export class ListSchemasHandler extends BaseToolHandler {
   async handle(
     clientManager: ClientManager,
     toolArguments: Record<string, unknown>,
   ): Promise<CallToolResult> {
-    const { baseUrl, latestOnly, subjectPrefix, deleted } =
+    const { latestOnly, subjectPrefix, deleted } =
       listSchemasArguments.parse(toolArguments);
 
-    if (baseUrl !== undefined && baseUrl !== "") {
-      clientManager.setConfluentCloudSchemaRegistryEndpoint(baseUrl);
-    }
-
-    const pathBasedClient = wrapAsPathBasedClient(
-      clientManager.getConfluentCloudSchemaRegistryRestClient(),
+    logger.debug(
+      {
+        latestOnly,
+        subjectPrefix,
+        deleted,
+      },
+      "ListSchemasHandler.handle called with arguments",
     );
 
-    // First get all schemas
-    const { data: response, error: error } = await pathBasedClient[
-      "/schemas"
-    ].GET({
-      query: {
-        ...(latestOnly ? { latestOnly: true } : {}),
-        ...(subjectPrefix ? { subjectPrefix: subjectPrefix } : {}),
-        ...(deleted ? { deleted: true } : {}),
-      },
-    });
+    const registry: SchemaRegistryClient =
+      clientManager.getSchemaRegistryClient();
+
+    try {
+      let subjects: string[] = await registry.getAllSubjects();
+      logger.debug(
+        { subjectsCount: subjects.length },
+        "Fetched all subjects from registry",
+      );
+      if (subjectPrefix) {
+        subjects = subjects.filter((s) => s.startsWith(subjectPrefix));
+        logger.debug(
+          { filteredSubjectsCount: subjects.length, subjectPrefix },
+          "Filtered subjects by prefix",
+        );
+      }
 
-    if (error) {
+      const result: Record<string, unknown> = {};
+      for (const subject of subjects) {
+        if (latestOnly) {
+          try {
+            const latest = await registry.getLatestSchemaMetadata(subject);
+            logger.debug({ subject, latest }, "Fetched latest schema metadata");
+            result[subject] = {
+              version: latest.version,
+              id: latest.id,
+              schemaType: latest.schemaType,
+              schema: latest.schema,
+            };
+          } catch (err) {
+            logger.warn(
+              {
+                subject,
+                error: err instanceof Error ? err.message : String(err),
+              },
+              "Failed to fetch latest schema metadata",
+            );
+            result[subject] = {
+              error: err instanceof Error ? err.message : String(err),
+            };
+          }
+        } else {
+          try {
+            const versions: number[] = await registry.getAllVersions(subject);
+            logger.debug({ subject, versions }, "Fetched all schema versions");
+            result[subject] = [];
+            const versionPromises = versions.map(async (version) => {
+              try {
+                const schema = await registry.getSchemaMetadata(
+                  subject,
+                  version,
+                  deleted,
+                );
+                logger.debug(
+                  { subject, version, schema },
+                  "Fetched schema metadata for version",
+                );
+                (result[subject] as unknown[]).push({
+                  version: schema.version,
+                  id: schema.id,
+                  schemaType: schema.schemaType,
+                  schema: schema.schema,
+                });
+              } catch (err) {
+                logger.warn(
+                  {
+                    subject,
+                    version,
+                    error: err instanceof Error ? err.message : String(err),
+                  },
+                  "Failed to fetch schema metadata for version",
+                );
+                (result[subject] as unknown[]).push({
+                  version,
+                  error: err instanceof Error ? err.message : String(err),
+                });
+              }
+            });
+            await Promise.all(versionPromises);
+          } catch (err) {
+            logger.warn(
+              {
+                subject,
+                error: err instanceof Error ? err.message : String(err),
+              },
+              "Failed to fetch all versions for subject",
+            );
+            result[subject] = {
+              error: err instanceof Error ? err.message : String(err),
+            };
+          }
+        }
+      }
+      logger.info(
+        { subjects: Object.keys(result).length },
+        "Returning schema listing result",
+      );
+      return this.createResponse(JSON.stringify(result));
+    } catch (error) {
+      logger.error(
+        {
+          error: error instanceof Error ? error.message : JSON.stringify(error),
+        },
+        "Failed to list schemas",
+      );
       return this.createResponse(
-        `Failed to list schemas: ${JSON.stringify(error)}`,
+        `Failed to list schemas: ${error instanceof Error ? error.message : JSON.stringify(error)}`,
         true,
       );
     }
-
-    return this.createResponse(`${JSON.stringify(response)}`);
   }
 
   getToolConfig(): ToolConfig {
@@ -75,6 +165,10 @@ export class ListSchemasHandler extends BaseToolHandler {
   }
 
   getRequiredEnvVars(): EnvVar[] {
-    return ["SCHEMA_REGISTRY_API_KEY", "SCHEMA_REGISTRY_API_SECRET"];
+    return [
+      "SCHEMA_REGISTRY_ENDPOINT",
+      "SCHEMA_REGISTRY_API_KEY",
+      "SCHEMA_REGISTRY_API_SECRET",
+    ];
   }
 }