fix(concept-mapping): use the dataset's persisted cache id

Concept mapping opened trex connections against a cacheId derived at
runtime from getDatabaseCredentials(): it looked up the credential whose
code matched databaseCode, then recomputed the alias from the dataset
UUID with a HANA special case. That derivation is not equivalent to the
persisted portal.dataset.cache_id and diverges for several populations:

  - legacy rows the migration backfilled with database_code
  - snapshots, which inherit the SOURCE dataset's cacheId, not their own
  - operator-supplied custom cache ids

It also cannot tell apart two datasets that share a databaseCode but use
different schemas, so mappings could be read from or written to the
wrong cache catalog.

Read portal.dataset.cache_id instead and treat it as the source of
truth. The route now accepts an optional datasetId; when supplied, the
dataset record is fetched with the caller's token and resolution falls
back to the record's databaseCode only when cacheId is genuinely absent.
When a datasetId is given but cannot be resolved we surface 502 rather
than guessing an alias, since a wrong alias silently targets the wrong
database. Callers that omit datasetId keep the existing databaseCode
path, so the endpoint stays backward compatible.

The dataset lookup is injected into ConceptMappingRouter so the router
remains importable without the env-dependent PortalAPI module.

Adds regression coverage for the divergent populations (legacy/migrated,
custom, cloned, and two datasets sharing a databaseCode), for concept
mapping save/load over HTTP, and for the API error contract, including
that a failed lookup opens no connection and writes nothing. Adds the
test task the plugin was missing.

Author: ohdsi-trex

plugins/functions/concept-mapping/deno.json

@@ -2,6 +2,9 @@
   "workspace": [],
   "nodeModulesDir": "auto",
   "unstable": ["sloppy-imports"],
+  "tasks": {
+    "test": "deno test --allow-read --allow-env --allow-net --no-check"
+  },
   "compilerOptions": {
     "lib": [
       "deno.window"
@@ -19,6 +22,8 @@
     "pg-promise": "npm:pg-promise@^10.10.1",
     "axios": "npm:axios@^1.7.7",
     "@sap/textbundle": "npm:@sap/textbundle@4.2.0",
+    "@std/assert": "jsr:@std/assert@1",
+    "@std/testing": "jsr:@std/testing@1",
     "./src/env": "./src/env.ts",
     "./src/types": "./src/types.ts",
     "./src/constants": "./src/constants.ts",

plugins/functions/concept-mapping/deno.lock

@@ -1,5 +1,6 @@
 import express, { Application } from "express";
 import { ConceptMappingRouter } from "./src/concept-mapping/routes";
+import { PortalAPI } from "./src/api/PortalAPI";
 
 export class App {
   private app: Application;
@@ -11,7 +12,12 @@ export class App {
 
   async start() {
     this.app.use(express.json());
-    this.app.use("/concept-mapping", new ConceptMappingRouter().router);
+    this.app.use(
+      "/concept-mapping",
+      new ConceptMappingRouter(
+        (token) => (datasetId) => new PortalAPI(token).getDataset(datasetId),
+      ).router,
+    );
     this.app.listen(8000);
     this.logger.info(`Concept Mapping service started successfully!`);
   }

plugins/functions/concept-mapping/index.ts

@@ -17,6 +17,7 @@ export class PortalAPI {
     }
     if (env.SERVICE_ROUTES.portalServer) {
       this.baseURL = env.SERVICE_ROUTES.portalServer;
+      // @ts-ignore Cannot find name 'Trex'
       this.channel = Trex.tokioChannel("d2e-functions/portal");
       // this.httpsAgent = new https.Agent({
       //   rejectUnauthorized: true,

plugins/functions/concept-mapping/src/api/PortalAPI.ts

@@ -9,6 +9,12 @@ export const GetConceptMappingDto = () => [
     .isString()
     .notEmpty()
     .withMessage("schemaName is required"),
+  // Optional: when supplied, the dataset's persisted cacheId is authoritative.
+  // Omitted by pre-dataset / infra callers, which keep the databaseCode path.
+  query("datasetId")
+    .optional()
+    .isUUID()
+    .withMessage("datasetId must be a UUID"),
 ];
 
 export const ConceptMappingDto = () => [

plugins/functions/concept-mapping/src/concept-mapping/middleware.ts

@@ -0,0 +1,276 @@
+import { describe, it } from "@std/testing/bdd";
+import { assertEquals, assertStringIncludes } from "@std/assert";
+import pako from "pako";
+import { encodeBase64 } from "base64";
+
+/** Mirrors the encoding the UI performs before POSTing (pako.deflate + base64). */
+function encodePayload(mappings: unknown[]): string {
+  return encodeBase64(pako.deflate(JSON.stringify(mappings)));
+}
+
+// Records the cacheId trex is asked to open a connection against.
+let lastGetConnectionArgs: unknown[] = [];
+// Records the statement trex was asked to run, so save tests can assert the
+// INSERT actually targets the resolved cache/schema.
+let lastExecute: { sql: string; params: unknown[] } | null = null;
+let executeImpl: (sql: string, params: unknown[]) => unknown[] = () => [];
+
+// deno-lint-ignore no-explicit-any
+(globalThis as any).Trex = {
+  databaseManager: () => ({
+    getConnection: (...args: unknown[]) => {
+      lastGetConnectionArgs = args;
+      return {
+        execute: (
+          sql: string,
+          params: unknown[],
+          cb: (err: unknown, res: unknown) => void,
+        ) => {
+          lastExecute = { sql, params: params.map((p) => (p as { value: unknown }).value) };
+          try {
+            cb(null, executeImpl(sql, params));
+          } catch (e) {
+            cb(e, null);
+          }
+        },
+        close: () => {},
+      };
+    },
+  }),
+};
+
+const express = (await import("express")).default;
+const { ConceptMappingRouter } = await import("./routes.ts");
+
+const DS = "3f2a1c44-7777-4aaa-9bbb-000000000007";
+const EMPTY_PAYLOAD = "eJyLjgUAARUAuQ=="; // zlib+base64 of []
+
+// deno-lint-ignore no-explicit-any
+async function withServer(fetcherFactory: any, fn: (base: string) => Promise<void>) {
+  const app = express();
+  app.use(express.json());
+  app.use("/concept-mapping", new ConceptMappingRouter(fetcherFactory).router);
+
+  const server = app.listen(0);
+  await new Promise((resolve) => server.once("listening", resolve));
+  const { port } = server.address() as { port: number };
+  try {
+    await fn(`http://127.0.0.1:${port}/concept-mapping`);
+  } finally {
+    await new Promise((resolve) => server.close(resolve));
+  }
+}
+
+describe("GET /concept-mapping validation", () => {
+  it("returns 400 when schemaName is missing", async () => {
+    await withServer(undefined, async (base) => {
+      const res = await fetch(`${base}/?databaseCode=PG_MAIN`);
+      assertEquals(res.status, 400);
+      await res.body?.cancel();
+    });
+  });
+
+  it("returns 400 when datasetId is not a UUID", async () => {
+    await withServer(undefined, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=PG_MAIN&schemaName=cdm&datasetId=not-a-uuid`,
+      );
+      assertEquals(res.status, 400);
+      await res.body?.cancel();
+    });
+  });
+});
+
+describe("cacheId resolution over HTTP", () => {
+  it("opens the connection with the dataset's persisted cacheId", async () => {
+    lastGetConnectionArgs = [];
+    executeImpl = () => [{ source_code: "X1" }];
+
+    const factory = () => (_id: string) =>
+      Promise.resolve({ cacheId: "_stored_cache", databaseCode: "PG_SHARE" });
+
+    await withServer(factory, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=PG_SHARE&schemaName=cdm&datasetId=${DS}`,
+      );
+      assertEquals(res.status, 200);
+      assertEquals(await res.json(), [{ source_code: "X1" }]);
+    });
+
+    // Authoritative cacheId, NOT the request's databaseCode.
+    assertEquals(lastGetConnectionArgs[0], "_stored_cache");
+  });
+
+  it("loads a legacy dataset via the record's databaseCode when cacheId is absent", async () => {
+    lastGetConnectionArgs = [];
+    executeImpl = () => [{ source_code: "LEGACY" }];
+
+    // Migrated row the cache_id backfill left unset -> guarded fallback.
+    const factory = () => (_id: string) =>
+      Promise.resolve({ cacheId: null, databaseCode: "PG_OLDER" });
+
+    await withServer(factory, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=REQ_CODE&schemaName=cdm&datasetId=${DS}`,
+      );
+      assertEquals(res.status, 200);
+      assertEquals(await res.json(), [{ source_code: "LEGACY" }]);
+    });
+
+    // The record's databaseCode wins over the request's, which is never trusted
+    // once a datasetId is in play.
+    assertEquals(lastGetConnectionArgs[0], "PG_OLDER");
+  });
+
+  it("uses the databaseCode when no datasetId is supplied (infra path)", async () => {
+    lastGetConnectionArgs = [];
+    executeImpl = () => [];
+
+    await withServer(undefined, async (base) => {
+      const res = await fetch(`${base}/?databaseCode=PG_MAIN&schemaName=cdm`);
+      assertEquals(res.status, 200);
+      assertEquals(await res.json(), []);
+    });
+
+    assertEquals(lastGetConnectionArgs[0], "PG_MAIN");
+  });
+
+  it("returns 502 when the dataset lookup fails, never a guessed cache", async () => {
+    const factory = () => (_id: string) => Promise.reject(new Error("portal down"));
+
+    await withServer(factory, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=PG_SHARE&schemaName=cdm&datasetId=${DS}`,
+      );
+      assertEquals(res.status, 502);
+      assertStringIncludes(await res.text(), "cache id");
+    });
+  });
+});
+
+describe("POST /concept-mapping save path", () => {
+  it("writes the mappings to the dataset's persisted cacheId", async () => {
+    lastGetConnectionArgs = [];
+    lastExecute = null;
+    // rowCount is derived from the driver result length.
+    executeImpl = () => [{}, {}];
+
+    const factory = () => (_id: string) =>
+      Promise.resolve({ cacheId: "_stored_cache", databaseCode: "PG_SHARE" });
+
+    const payload = encodePayload([
+      { source_code: "A1", target_concept_id: 111 },
+      { source_code: "B2", target_concept_id: 222 },
+    ]);
+
+    await withServer(factory, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=PG_SHARE&schemaName=cdm&datasetId=${DS}`,
+        {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            sourceVocabularyId: "MY_VOCAB",
+            conceptMappings: payload,
+          }),
+        },
+      );
+      assertEquals(res.status, 200);
+      assertStringIncludes(await res.text(), "Inserted 2 rows");
+    });
+
+    // The write targets the authoritative cacheId, NOT the request databaseCode.
+    assertEquals(lastGetConnectionArgs[0], "_stored_cache");
+    assertStringIncludes(lastExecute!.sql, "INSERT INTO cdm.source_to_concept_map");
+    // sourceVocabularyId is stamped onto every row.
+    assertEquals(
+      lastExecute!.params.filter((p) => p === "MY_VOCAB").length,
+      2,
+    );
+    assertEquals(lastExecute!.params.includes("A1"), true);
+    assertEquals(lastExecute!.params.includes("B2"), true);
+  });
+
+  it("saves against the databaseCode when no datasetId is supplied (infra path)", async () => {
+    lastGetConnectionArgs = [];
+    executeImpl = () => [{}];
+
+    await withServer(undefined, async (base) => {
+      const res = await fetch(`${base}/?databaseCode=PG_MAIN&schemaName=cdm`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          sourceVocabularyId: "V",
+          conceptMappings: encodePayload([{ source_code: "A1" }]),
+        }),
+      });
+      assertEquals(res.status, 200);
+      await res.body?.cancel();
+    });
+
+    assertEquals(lastGetConnectionArgs[0], "PG_MAIN");
+  });
+
+  it("returns 502 without writing when the dataset lookup fails", async () => {
+    lastGetConnectionArgs = [];
+    lastExecute = null;
+    executeImpl = () => [{}];
+
+    const factory = () => (_id: string) => Promise.reject(new Error("portal down"));
+
+    await withServer(factory, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=PG_SHARE&schemaName=cdm&datasetId=${DS}`,
+        {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            sourceVocabularyId: "V",
+            conceptMappings: encodePayload([{ source_code: "A1" }]),
+          }),
+        },
+      );
+      assertEquals(res.status, 502);
+      assertStringIncludes(await res.text(), "cache id");
+    });
+
+    // Critically: no connection opened and nothing written to a guessed cache.
+    assertEquals(lastGetConnectionArgs.length, 0);
+    assertEquals(lastExecute, null);
+  });
+});
+
+describe("POST /concept-mapping error contract", () => {
+  it("returns 400 for an empty mappings payload even when portal is down", async () => {
+    const factory = () => (_id: string) => Promise.reject(new Error("portal down"));
+
+    await withServer(factory, async (base) => {
+      const res = await fetch(
+        `${base}/?databaseCode=PG_SHARE&schemaName=cdm&datasetId=${DS}`,
+        {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            sourceVocabularyId: "vocab",
+            conceptMappings: EMPTY_PAYLOAD,
+          }),
+        },
+      );
+      // Empty payload is rejected before cacheId resolution -> 400, not 502.
+      assertEquals(res.status, 400);
+      assertStringIncludes(await res.text(), "No concept mappings to save");
+    });
+  });
+
+  it("returns 400 when conceptMappings is an empty string", async () => {
+    await withServer(undefined, async (base) => {
+      const res = await fetch(`${base}/?databaseCode=PG_MAIN&schemaName=cdm`, {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ sourceVocabularyId: "v", conceptMappings: "" }),
+      });
+      assertEquals(res.status, 400);
+      await res.body?.cancel();
+    });
+  });
+});