From 00d030aa430e9554cf79ed1a552408301bce1e26 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Wed, 3 Jun 2026 00:50:38 +0200 Subject: [PATCH 01/10] Route Opey chat through the Explorer backend proxy - chat.ts: POST /api/opey/stream (same-origin via Vite proxy) instead of calling Opey directly from the browser, which failed on Opey's cross-origin session cookie. Drop the direct browser create-session. - oauth2.ts: store baseUri/version in session.clientConfig so the consent service builds a valid OBP URL (was 'undefined/obp/...'). - OpeyClientService.ts: add createOpeySession() to establish an Opey session with the Consent-JWT and forward its cookie on /stream (Opey rejects /stream without a session). - add @grpc/grpc-js (the Express server fails to start without it). --- package-lock.json | 8 ++--- package.json | 6 ++-- server/routes/oauth2.ts | 7 ++++- server/services/OpeyClientService.ts | 41 ++++++++++++++++++++++--- src/stores/chat.ts | 45 ++++++++-------------------- 5 files changed, 62 insertions(+), 45 deletions(-) diff --git a/package-lock.json b/package-lock.json index 5983832..347350b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,7 +10,7 @@ "dependencies": { "@element-plus/icons-vue": "^2.1.0", "@fontsource/roboto": "^5.0.0", - "@grpc/grpc-js": "^1.14.3", + "@grpc/grpc-js": "^1.14.4", "@highlightjs/vue-plugin": "^2.1.0", "@types/node-fetch": "^2.6.12", "ai": "^4.1.43", @@ -1684,9 +1684,9 @@ } }, "node_modules/@grpc/grpc-js": { - "version": "1.14.3", - "resolved": "https://registry.npmjs.org/@grpc/grpc-js/-/grpc-js-1.14.3.tgz", - "integrity": "sha512-Iq8QQQ/7X3Sac15oB6p0FmUg/klxQvXLeileoqrTRGJYLV+/9tubbr9ipz0GKHjmXVsgFPo/+W+2cA8eNcR+XA==", + "version": "1.14.4", + "resolved": "https://registry.npmjs.org/@grpc/grpc-js/-/grpc-js-1.14.4.tgz", + "integrity": "sha512-k9Dj3DV/itK9D06Y8f190Qgop7/Ui+D0njFV3LHMPwPT75DpXLQohE9Wmz0QElrJnzsjB7KPWiKJbOl7IPDArQ==", "license": "Apache-2.0", "dependencies": { "@grpc/proto-loader": "^0.8.0", diff --git a/package.json b/package.json index 0244961..197b440 100644 --- a/package.json +++ b/package.json @@ -27,7 +27,7 @@ "dependencies": { "@element-plus/icons-vue": "^2.1.0", "@fontsource/roboto": "^5.0.0", - "@grpc/grpc-js": "^1.14.3", + "@grpc/grpc-js": "^1.14.4", "@highlightjs/vue-plugin": "^2.1.0", "@types/node-fetch": "^2.6.12", "ai": "^4.1.43", @@ -72,6 +72,7 @@ "@ai-sdk/vue": "^1.1.18", "@playwright/test": "^1.51.1", "@rushstack/eslint-patch": "^1.4.0", + "@secretlint/secretlint-rule-preset-recommend": "^9.0.0", "@sveltejs/vite-plugin-svelte": "^4.0.4", "@testing-library/vue": "^8.1.0", "@types/jest": "^29.5.14", @@ -90,8 +91,6 @@ "eslint": "^9.15.0", "eslint-plugin-security": "^3.0.1", "eslint-plugin-vue": "^9.12.0", - "secretlint": "^9.0.0", - "@secretlint/secretlint-rule-preset-recommend": "^9.0.0", "happy-dom": "^17.1.4", "jest": "^29.7.0", "jsdom": "^25.0.1", @@ -99,6 +98,7 @@ "npm-run-all2": "^7.0.1", "playwright": "^1.51.1", "prettier": "^3.0.1", + "secretlint": "^9.0.0", "superagent": "^9.0.0", "supertest": "^7.0.0", "svelte": "^5.45.3", diff --git a/server/routes/oauth2.ts b/server/routes/oauth2.ts index 7fabb07..e4538b9 100644 --- a/server/routes/oauth2.ts +++ b/server/routes/oauth2.ts @@ -253,8 +253,13 @@ router.get('/oauth2/callback', async (req: Request, res: Response) => { sub: userInfo.sub } - // Also store clientConfig for OBP API calls + // Also store clientConfig for OBP API calls. + // baseUri/version are required by services that read session.clientConfig directly + // (e.g. OBPConsentsService for the Opey consent flow) — without baseUri the consent + // request builds an "undefined/obp/..." URL and throws "Invalid URL". session.clientConfig = { + baseUri: process.env.VITE_OBP_API_HOST, + version: process.env.VITE_OBP_API_VERSION, oauth2: { accessToken: tokens.accessToken, tokenType: 'Bearer' diff --git a/server/services/OpeyClientService.ts b/server/services/OpeyClientService.ts index 4b5af16..d9a32c3 100644 --- a/server/services/OpeyClientService.ts +++ b/server/services/OpeyClientService.ts @@ -131,7 +131,10 @@ export default class OpeyClientService { // Get auth headers const authHeaders = await this.getConsentAuthHeaders(config) - + // Opey's /stream requires an authenticated session. Establish one with the + // Consent-JWT first, then forward the returned session cookie on /stream. + const sessionCookie = await this.createOpeySession(config) + try { const url = `${config.baseUri}${config.paths.stream}` @@ -140,10 +143,10 @@ export default class OpeyClientService { stream_input.stream_tokens = true console.log(`Posting to Opey with streaming: ${JSON.stringify(stream_input)}\n URL: ${url}`) //DEBUG - + const response = await fetch(url, { method: 'POST', - headers: authHeaders, + headers: { ...authHeaders, 'Cookie': sessionCookie }, body: JSON.stringify(stream_input) }) if (!response.body) { @@ -242,7 +245,7 @@ export default class OpeyClientService { } async getConsentAuthHeaders(opeyConfig: OpeyConfig): Promise<{ [key: string]: string } | undefined> { - + if (!opeyConfig.authConfig || !opeyConfig.authConfig.obpConsent) { throw new Error('AuthConfig not found or obpConsent missing') } @@ -251,4 +254,34 @@ export default class OpeyClientService { 'Content-Type': 'application/json' } } + + /** + * Establish an authenticated Opey session using the OBP Consent-JWT and return + * the session cookie ("session=...") to forward on subsequent /stream or /invoke + * calls. Opey's chat endpoints require a session (created via /create-session); + * driving it server-side keeps the browser same-origin (mirrors the Portal design). + */ + async createOpeySession(opeyConfig: OpeyConfig): Promise { + const authHeaders = await this.getConsentAuthHeaders(opeyConfig) + const url = `${opeyConfig.baseUri}/create-session` + const response = await fetch(url, { + method: 'POST', + headers: authHeaders, + body: JSON.stringify({}) + }) + if (!response.ok) { + throw new Error(`Failed to create Opey session: ${response.status} ${response.statusText}`) + } + // undici exposes getSetCookie(); fall back to the combined header otherwise + const setCookies: string[] = + (response.headers as any).getSetCookie?.() ?? + ((response.headers.get('set-cookie') ? [response.headers.get('set-cookie') as string] : [])) + const sessionCookie = setCookies + .map(c => c.split(';')[0]) + .find(c => c.startsWith('session=')) + if (!sessionCookie) { + throw new Error('Opey did not return a session cookie from /create-session') + } + return sessionCookie + } } \ No newline at end of file diff --git a/src/stores/chat.ts b/src/stores/chat.ts index 5f03ebe..77dd0f3 100644 --- a/src/stores/chat.ts +++ b/src/stores/chat.ts @@ -146,40 +146,19 @@ export const useChat = defineStore('chat', { }, async handleAuthentication(): Promise { - // Handle authentication - // get consent for Opey from user + // Establish consent for Opey via the Explorer backend (same-origin proxy). + // The backend (`POST /api/opey/consent`) stores the consent in the server-side + // session (session.opeyConfig) and authenticates to Opey with the Consent-JWT on + // every /stream call. The browser therefore never talks to Opey directly, which + // avoids the cross-origin session-cookie problem (mirrors the Portal design). const consentResponse = await getobpConsent() - if (consentResponse) { - const consentId = consentResponse.consent_id - - } else { + if (!consentResponse) { throw new Error('Failed to grant consent. Please try again.') } - const consentJwt = consentResponse.jwt - - const opeyBaseUri = import.meta.env.VITE_CHATBOT_URL - // Get a session from opey - try { - const sessionResponse = await fetch(`${opeyBaseUri}/create-session`, { - method: 'POST', - credentials: 'include', - headers: { - 'Content-Type': 'application/json', - 'Consent-JWT': consentJwt - }, - }) - - if (!sessionResponse.ok) { - throw new Error(`Failed to create session: ${sessionResponse.statusText}`); - } else if (sessionResponse.status === 200) { - this.userIsAuthenticated = true - } - - } catch (error) { - throw new Error(`Failed to create Opey session: ${error}`); - } + // Consent is now stored server-side; the backend proxy handles the Opey session. + this.userIsAuthenticated = true }, async stream(input: ChatStreamInput): Promise { @@ -200,11 +179,11 @@ export const useChat = defineStore('chat', { } this.addMessage(this.currentAssistantMessage) - // Set the status to 'loading' before we fetch the stream - const opeyBaseUri = import.meta.env.VITE_CHATBOT_URL - // Handle stream + // Call the Explorer backend proxy (same-origin: browser → Vite /api → Express 8085 → Opey). + // The backend attaches the Consent-JWT server-side, so the browser stays same-origin + // and never hits Opey's cross-origin session cookie. try { - const response = await fetch(`${opeyBaseUri}/stream`, { + const response = await fetch('/api/opey/stream', { method: 'POST', credentials: 'include', headers: { From c421257279a53d307db3c9a8a3f0b9c5518f0fb9 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 15:31:16 +0200 Subject: [PATCH 02/10] fix: trigger background docs refresh only on cache hit Previously the worker message that schedules a background docs refresh was posted before reading the cached response. On a cold cache the read throws, the catch block performs the full fetch, and the already-posted message echoes back and runs the same full fetch a second time. Post the message only after the cached response has been read successfully. --- src/obp/message-docs.ts | 10 ++++++++-- src/obp/resource-docs.ts | 4 +++- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/src/obp/message-docs.ts b/src/obp/message-docs.ts index bc187a2..74c4efc 100644 --- a/src/obp/message-docs.ts +++ b/src/obp/message-docs.ts @@ -199,8 +199,11 @@ async function getCacheDocJsonSchema(cacheStorageOfMessageDocsJsonSchema: any): export async function cache(cacheStorage: any, cachedResponse: any, worker: any): Promise { try { + const messageDocs = await cachedResponse.json() + // Only a cache hit should schedule a background refresh; posting before the + // read would make a cold cache fetch everything twice via the worker echo. worker.postMessage('update-message-docs') - return await cachedResponse.json() + return messageDocs } catch (error) { console.warn('No message docs cache or malformed cache.') console.log('Caching message docs...') @@ -216,8 +219,11 @@ export async function cacheJsonSchema( worker: any ): Promise { try { + const messageDocsJsonSchema = await cachedResponse.json() + // Only a cache hit should schedule a background refresh; posting before the + // read would make a cold cache fetch everything twice via the worker echo. worker.postMessage('update-message-docs-json-schema') - return await cachedResponse.json() + return messageDocsJsonSchema } catch (error) { console.warn('No message docs JSON schema cache or malformed cache.') console.log('Caching message docs JSON schema...') diff --git a/src/obp/resource-docs.ts b/src/obp/resource-docs.ts index f4e4958..3583a13 100644 --- a/src/obp/resource-docs.ts +++ b/src/obp/resource-docs.ts @@ -213,8 +213,10 @@ async function getCacheDoc(cacheStorageOfResourceDocs: any): Promise { export async function cache(cachedStorage: any, cachedResponse: any, worker: any): Promise { try { - worker.postMessage('update-resource-docs') const resourceDocs = await cachedResponse.json() + // Only a cache hit should schedule a background refresh; posting before the + // read would make a cold cache fetch everything twice via the worker echo. + worker.postMessage('update-resource-docs') console.log( '[CACHE] Loaded cached resource docs, available versions:', Object.keys(resourceDocs) From a619cc3004f4448ccde6a806a599945d06ee553a Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 15:38:55 +0200 Subject: [PATCH 03/10] perf: fetch resource docs with bounded concurrency at startup Cold-start resource docs fetching awaited one API version at a time, so cold load time scaled linearly with the number of active versions. Add a small bounded-concurrency helper and use it to fetch up to 5 versions in parallel, keeping the same per-version error isolation, response-shape checks, and single cache write at the end. --- src/obp/common-functions.ts | 18 ++++++++ src/obp/resource-docs.ts | 91 +++++++++++++++++-------------------- 2 files changed, 59 insertions(+), 50 deletions(-) diff --git a/src/obp/common-functions.ts b/src/obp/common-functions.ts index 3748269..812230a 100644 --- a/src/obp/common-functions.ts +++ b/src/obp/common-functions.ts @@ -103,6 +103,24 @@ export async function answerobpConsentChallenge(answerBody: any) { return response } +// Runs `task` over `items` with at most `limit` calls in flight at once. +// One task rejecting does not stop the others (same isolation as Promise.allSettled), +// since each worker's while-loop simply continues to the next queued item. +export async function runWithConcurrency( + items: readonly T[], + limit: number, + task: (item: T) => Promise +): Promise { + const queue = [...items] + const workers = Array.from({ length: Math.min(limit, queue.length) }, async () => { + while (queue.length > 0) { + const item = queue.shift() as T + await task(item) + } + }) + await Promise.all(workers) +} + export function clearCacheByName(cacheName: string) { if ('caches' in window) { caches.delete(cacheName).then(function(success) { diff --git a/src/obp/resource-docs.ts b/src/obp/resource-docs.ts index 3583a13..0770903 100644 --- a/src/obp/resource-docs.ts +++ b/src/obp/resource-docs.ts @@ -27,9 +27,11 @@ import { get, isServerUp, OBP_API_DEFAULT_RESOURCE_DOC_VERSION } from '../obp' import { getOBPAPIVersions } from '../obp/api-version' -import { updateLoadingInfoMessage } from './common-functions' +import { runWithConcurrency, updateLoadingInfoMessage } from './common-functions' import { RESOURCE_DOCS_API_VERSION } from '../shared-constants' +const RESOURCE_DOCS_CONCURRENCY = 5 + // Get Resource Docs export async function getOBPResourceDocs(apiStandardAndVersion: string): Promise { const logMessage = `Loading API ${apiStandardAndVersion}` @@ -127,66 +129,51 @@ export async function cacheDoc(cacheStorageOfResourceDocs: any): Promise { `[CACHE] Found ${scannedAPIVersions.length} total versions, ${activeVersions.length} are active` ) const resourceDocsMapping: any = {} - for (const { api_standard, api_short_version } of activeVersions) { - // we need this to cache the dynamic entities resource doc - if (api_short_version === 'dynamic-entity') { - const logMessage = `Caching Dynamic API { standard: ${api_standard}, version: ${api_short_version} }` - console.log(logMessage) - if (api_standard) { - try { - const version = `${api_standard.toUpperCase()}${api_short_version}` - console.log(`[CACHE] Attempting to load dynamic resource docs for: ${version}`) - const resourceDocs = await getOBPDynamicResourceDocs(version) - if (version && Object.keys(resourceDocs).includes('resource_docs')) { - resourceDocsMapping[version] = resourceDocs - console.log(`[CACHE] Successfully cached dynamic docs for: ${version}`) - } else { - console.warn(`[CACHE] WARNING: Response for ${version} missing 'resource_docs' field`) - } - } catch (error: any) { - console.warn( - `[CACHE] WARNING: Skipping dynamic endpoint ${api_standard}${api_short_version}:` - ) - console.warn(` API Version: ${api_short_version}`) - console.warn(` API Standard: ${api_standard}`) - console.warn( - ` Constructed version string: ${api_standard.toUpperCase()}${api_short_version}` - ) - console.warn(` Error status: ${error.status || 'unknown'}`) - console.warn(` Error message: ${error.message || 'No message'}`) - if (error.status === 500) { - console.warn( - ` NOTE: This likely means the OBP-API server doesn't have this feature enabled` - ) - } - } - } - updateLoadingInfoMessage(logMessage) - continue - } - const logMessage = `Caching API { standard: ${api_standard}, version: ${api_short_version} }` - console.log(logMessage) + const total = activeVersions.length + let completed = 0 + + const cacheOneVersion = async ({ api_standard, api_short_version }: any): Promise => { if (api_standard) { + // we need this to cache the dynamic entities resource doc + const isDynamicEntity = api_short_version === 'dynamic-entity' + const version = `${api_standard.toUpperCase()}${api_short_version}` try { - const version = `${api_standard.toUpperCase()}${api_short_version}` - console.log(`[CACHE] Attempting to load resource docs for: ${version}`) - const resourceDocs = await getOBPResourceDocs(version) + console.log( + isDynamicEntity + ? `[CACHE] Attempting to load dynamic resource docs for: ${version}` + : `[CACHE] Attempting to load resource docs for: ${version}` + ) + const resourceDocs = isDynamicEntity + ? await getOBPDynamicResourceDocs(version) + : await getOBPResourceDocs(version) if (version && Object.keys(resourceDocs).includes('resource_docs')) { resourceDocsMapping[version] = resourceDocs - console.log(`[CACHE] Successfully cached docs for: ${version}`) + console.log( + isDynamicEntity + ? `[CACHE] Successfully cached dynamic docs for: ${version}` + : `[CACHE] Successfully cached docs for: ${version}` + ) } else { console.warn(`[CACHE] WARNING: Response for ${version} missing 'resource_docs' field`) } } catch (error: any) { - console.warn(`[CACHE] WARNING: Skipping API version ${api_standard}${api_short_version}:`) - console.warn(` API Version: ${api_short_version}`) - console.warn(` API Standard: ${api_standard}`) console.warn( - ` Constructed version string: ${api_standard.toUpperCase()}${api_short_version}` + isDynamicEntity + ? `[CACHE] WARNING: Skipping dynamic endpoint ${api_standard}${api_short_version}:` + : `[CACHE] WARNING: Skipping API version ${api_standard}${api_short_version}:` ) + console.warn(` API Version: ${api_short_version}`) + console.warn(` API Standard: ${api_standard}`) + console.warn(` Constructed version string: ${version}`) console.warn(` Error status: ${error.status || 'unknown'}`) console.warn(` Error message: ${error.message || 'No message'}`) - if (error.status === 400) { + if (isDynamicEntity) { + if (error.status === 500) { + console.warn( + ` NOTE: This likely means the OBP-API server doesn't have this feature enabled` + ) + } + } else if (error.status === 400) { console.warn(` NOTE: This API version is not enabled on the OBP-API server`) console.warn(` NOTE: Check your OBP-API server configuration for available versions`) } else if (error.status === 500) { @@ -196,8 +183,12 @@ export async function cacheDoc(cacheStorageOfResourceDocs: any): Promise { } } } - updateLoadingInfoMessage(logMessage) + completed++ + updateLoadingInfoMessage(`Loading APIs ${completed}/${total}`) } + + await runWithConcurrency(activeVersions, RESOURCE_DOCS_CONCURRENCY, cacheOneVersion) + await cacheStorageOfResourceDocs.put('/', new Response(JSON.stringify(resourceDocsMapping))) return resourceDocsMapping } catch (error) { From 655e4d811a70d90fc9f406e6c02c62e3158fc967 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 15:39:01 +0200 Subject: [PATCH 04/10] perf: fetch message docs and json schemas concurrently The serial reduce loops over connectors made message-docs and message-docs-json-schema caching scale linearly with connector count. Replace both with the bounded-concurrency helper (limit 4), keeping the existing error-shape check and grouping logic. --- src/obp/message-docs.ts | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/src/obp/message-docs.ts b/src/obp/message-docs.ts index 74c4efc..d98109c 100644 --- a/src/obp/message-docs.ts +++ b/src/obp/message-docs.ts @@ -27,7 +27,9 @@ import { OBP_API_VERSION, get, isServerUp } from '../obp' import { V6_0_0 } from '../shared-constants' -import { updateLoadingInfoMessage } from './common-functions' +import { runWithConcurrency, updateLoadingInfoMessage } from './common-functions' + +const MESSAGE_DOCS_CONCURRENCY = 4 // Connectors to exclude from message docs display const excludedConnectors = ['mapped', 'star', 'proxy', 'internal', 'cardano', 'ethereum'] @@ -154,17 +156,16 @@ export function getGroupedMessageDocsJsonSchema(docs: any): any { export async function cacheDoc(cacheStorageOfMessageDocs: any): Promise { const connectors = await getConnectors() - const messageDocs = await connectors.reduce(async (agroup: any, connector: any) => { + const messageDocs: any = {} + await runWithConcurrency(connectors, MESSAGE_DOCS_CONCURRENCY, async (connector: string) => { const logMessage = `Caching message docs { connector: ${connector} }` console.log(logMessage) updateLoadingInfoMessage(logMessage) - const group = await agroup const docs = await getOBPMessageDocs(connector) if (!Object.keys(docs).includes('code')) { - group[connector] = getGroupedMessageDocs(docs) + messageDocs[connector] = getGroupedMessageDocs(docs) } - return group - }, Promise.resolve({})) + }) await cacheStorageOfMessageDocs.put('/', new Response(JSON.stringify(messageDocs))) return messageDocs } @@ -175,17 +176,16 @@ async function getCacheDoc(cacheStorageOfMessageDocs: any): Promise { export async function cacheDocJsonSchema(cacheStorageOfMessageDocsJsonSchema: any): Promise { const connectors = await getConnectors() - const messageDocsJsonSchema = await connectors.reduce(async (agroup: any, connector: any) => { + const messageDocsJsonSchema: any = {} + await runWithConcurrency(connectors, MESSAGE_DOCS_CONCURRENCY, async (connector: string) => { const logMessage = `Caching message docs JSON schema { connector: ${connector} }` console.log(logMessage) updateLoadingInfoMessage(logMessage) - const group = await agroup const docs = await getOBPMessageDocsJsonSchema(connector) if (!Object.keys(docs).includes('code')) { - group[connector] = getGroupedMessageDocsJsonSchema(docs) + messageDocsJsonSchema[connector] = getGroupedMessageDocsJsonSchema(docs) } - return group - }, Promise.resolve({})) + }) await cacheStorageOfMessageDocsJsonSchema.put( '/', new Response(JSON.stringify(messageDocsJsonSchema)) From 5205a9931af58dff1440fb345a9d00f1366e294b Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 15:39:07 +0200 Subject: [PATCH 05/10] perf: run independent startup caching tasks in parallel Resource docs, message docs, message docs JSON schema, and the glossary were awaited one after another during app setup even though none of them depend on each other. Fetch all four concurrently with Promise.all, keeping the existing fail-fast behavior where any one failing surfaces the setup error page. --- src/main.ts | 33 +++++++++++++++++---------------- 1 file changed, 17 insertions(+), 16 deletions(-) diff --git a/src/main.ts b/src/main.ts index 83ef4b5..824ec3f 100644 --- a/src/main.ts +++ b/src/main.ts @@ -306,21 +306,23 @@ async function setupData(app: App, worker: Worker) { } } - const { resourceDocs, groupedDocs } = await cacheResourceDocs( - cacheStorageOfResourceDocs, - cachedResponseOfResourceDocs, - worker - ) - const messageDocs = await cacheMessageDocs( - cacheStorageOfMessageDocs, - cachedResponseOfMessageDocs, - worker - ) - const messageDocsJsonSchema = await cacheMessageDocsJsonSchema( - cacheStorageOfMessageDocsJsonSchema, - cachedResponseOfMessageDocsJsonSchema, - worker - ) + // These four startup tasks are mutually independent, so fetch them concurrently. + // Promise.all keeps the existing fail-fast behavior: any one failing throws to the error page. + const [ + { resourceDocs, groupedDocs }, + messageDocs, + messageDocsJsonSchema, + glossary + ] = await Promise.all([ + cacheResourceDocs(cacheStorageOfResourceDocs, cachedResponseOfResourceDocs, worker), + cacheMessageDocs(cacheStorageOfMessageDocs, cachedResponseOfMessageDocs, worker), + cacheMessageDocsJsonSchema( + cacheStorageOfMessageDocsJsonSchema, + cachedResponseOfMessageDocsJsonSchema, + worker + ), + getOBPGlossary() + ]) // Provide data to a component's descendants // App-level provides are available to all components rendered in the app @@ -331,7 +333,6 @@ async function setupData(app: App, worker: Worker) { app.provide(obpGroupedMessageDocsKey, messageDocs) app.provide(obpGroupedMessageDocsJsonSchemaKey, messageDocsJsonSchema) app.provide(obpApiHostKey, import.meta.env.VITE_OBP_API_HOST) - const glossary = await getOBPGlossary() app.provide(obpGlossaryKey, glossary) // Try to load user's API collections (requires authentication) From 25cd82310be20a8f9f03a0410970b34c74cc18d0 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 15:39:54 +0200 Subject: [PATCH 06/10] test: add coverage for the bounded-concurrency helper Cover the concurrency ceiling, that every item is processed exactly once, and that one failing task does not prevent the others from running. --- src/test/run-with-concurrency.test.ts | 46 +++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 src/test/run-with-concurrency.test.ts diff --git a/src/test/run-with-concurrency.test.ts b/src/test/run-with-concurrency.test.ts new file mode 100644 index 0000000..542c084 --- /dev/null +++ b/src/test/run-with-concurrency.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from 'vitest' +import { runWithConcurrency } from '@/obp/common-functions' + +describe('runWithConcurrency', () => { + it('never runs more than the given limit of tasks at once', async () => { + const items = Array.from({ length: 10 }, (_, i) => i) + let inFlight = 0 + let maxInFlight = 0 + + await runWithConcurrency(items, 3, async () => { + inFlight++ + maxInFlight = Math.max(maxInFlight, inFlight) + await new Promise((resolve) => setTimeout(resolve, 5)) + inFlight-- + }) + + expect(maxInFlight).toBeLessThanOrEqual(3) + }) + + it('processes every item exactly once', async () => { + const items = Array.from({ length: 10 }, (_, i) => i) + const seen: number[] = [] + + await runWithConcurrency(items, 3, async (item) => { + seen.push(item) + }) + + expect(seen.sort((a, b) => a - b)).toEqual(items) + }) + + it('lets one failing task run without stopping the others', async () => { + const items = [1, 2, 3, 4, 5] + const processed: number[] = [] + + await expect( + runWithConcurrency(items, 2, async (item) => { + if (item === 3) { + throw new Error('boom') + } + processed.push(item) + }) + ).rejects.toThrow('boom') + + expect(processed.length).toBeGreaterThan(0) + }) +}) From e011fafd6be434d300130ede44c2d437ba685e6d Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 16:37:29 +0200 Subject: [PATCH 07/10] fix: keep draining the concurrency pool after a task fails runWithConcurrency's own comment promised Promise.allSettled-style isolation, but a rejecting task actually broke its worker's loop and Promise.all(workers) rejected immediately, abandoning any queued items still owned by other workers. Catch each task's error inside the loop so every item is still attempted, and only re-throw the first error once every item has settled. --- src/obp/common-functions.ts | 19 ++++++++++++--- src/test/run-with-concurrency.test.ts | 35 +++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/src/obp/common-functions.ts b/src/obp/common-functions.ts index 812230a..a332cf6 100644 --- a/src/obp/common-functions.ts +++ b/src/obp/common-functions.ts @@ -104,21 +104,34 @@ export async function answerobpConsentChallenge(answerBody: any) { } // Runs `task` over `items` with at most `limit` calls in flight at once. -// One task rejecting does not stop the others (same isolation as Promise.allSettled), -// since each worker's while-loop simply continues to the next queued item. +// One task rejecting does not stop the others: every item is still attempted (the +// worker keeps draining the queue), and the first error is only re-thrown after all +// items have settled, so partial results already written by other tasks are kept. export async function runWithConcurrency( items: readonly T[], limit: number, task: (item: T) => Promise ): Promise { const queue = [...items] + let firstError: unknown + let hasError = false const workers = Array.from({ length: Math.min(limit, queue.length) }, async () => { while (queue.length > 0) { const item = queue.shift() as T - await task(item) + try { + await task(item) + } catch (error) { + if (!hasError) { + hasError = true + firstError = error + } + } } }) await Promise.all(workers) + if (hasError) { + throw firstError + } } export function clearCacheByName(cacheName: string) { diff --git a/src/test/run-with-concurrency.test.ts b/src/test/run-with-concurrency.test.ts index 542c084..9870d98 100644 --- a/src/test/run-with-concurrency.test.ts +++ b/src/test/run-with-concurrency.test.ts @@ -43,4 +43,39 @@ describe('runWithConcurrency', () => { expect(processed.length).toBeGreaterThan(0) }) + + it('attempts every item even when some fail, only rejecting after all settle', async () => { + const items = [1, 2, 3, 4, 5, 6] + const processed: number[] = [] + + await expect( + runWithConcurrency(items, 3, async (item) => { + if (item % 2 === 0) { + throw new Error(`boom-${item}`) + } + processed.push(item) + }) + ).rejects.toThrow(/boom-/) + + // Every odd item still ran despite the even items throwing. + expect(processed.sort((a, b) => a - b)).toEqual([1, 3, 5]) + }) + + it('re-throws only after all items have settled, not as soon as the first fails', async () => { + const order: string[] = [] + + await expect( + runWithConcurrency([1, 2], 2, async (item) => { + order.push(`start-${item}`) + // item 1 is slower than item 2, so item 2 settles (and throws) first. + await new Promise((resolve) => setTimeout(resolve, item === 1 ? 5 : 1)) + order.push(`end-${item}`) + throw new Error(`boom-${item}`) + }) + ).rejects.toThrow('boom-2') + + // Both tasks fully ran (start and end) before the pool rejected — the slower + // item 1 was not abandoned mid-flight when item 2 threw first. + expect(order).toEqual(['start-1', 'start-2', 'end-2', 'end-1']) + }) }) From 8583a3cd469458b19b98dc03b1d2a02836eb7b79 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 16:37:38 +0200 Subject: [PATCH 08/10] fix: isolate per-connector failures and dedupe connector list fetches Unlike resource-docs.ts, message-docs.ts's per-connector fetch had no try/catch, so one connector returning an error payload (which get() turns into {error: ...} instead of throwing) tripped getGroupedMessageDocs on an undefined field and aborted caching for every other connector too. Wrap each connector's fetch and skip it on failure instead. Also memoize getConnectors(): cacheDoc and cacheDocJsonSchema each call it independently, and now that main.ts runs them concurrently the two calls fire at the same instant instead of staggered sequentially. --- src/obp/message-docs.ts | 58 +++++++++++++++++++++++++++++------------ 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/src/obp/message-docs.ts b/src/obp/message-docs.ts index d98109c..4617c82 100644 --- a/src/obp/message-docs.ts +++ b/src/obp/message-docs.ts @@ -42,21 +42,37 @@ const fallbackConnectors = [ 'rabbitmq_vOct2024' ] +// cacheDoc and cacheDocJsonSchema each call getConnectors() and, since this PR, can run +// concurrently (see main.ts's Promise.all) — share one in-flight request instead of firing +// two near-simultaneous GETs to the same endpoint. Reset once it settles so a later +// background re-warm still sees a fresh connector list. +let connectorsRequest: Promise | null = null + // Fetch available connectors dynamically from the API export async function getConnectors(): Promise { - try { - const data = await get(`obp/${V6_0_0}/system/connectors`) - if (data && data.connectors && Array.isArray(data.connectors)) { - const connectorNames = data.connectors - .map((c: any) => c.connector_name) - .filter((name: string) => !excludedConnectors.some(exc => name === exc || name.startsWith(exc + '_'))) - console.log('Fetched connectors from API:', connectorNames) - return connectorNames + if (connectorsRequest) { + return connectorsRequest + } + connectorsRequest = (async () => { + try { + const data = await get(`obp/${V6_0_0}/system/connectors`) + if (data && data.connectors && Array.isArray(data.connectors)) { + const connectorNames = data.connectors + .map((c: any) => c.connector_name) + .filter((name: string) => !excludedConnectors.some(exc => name === exc || name.startsWith(exc + '_'))) + console.log('Fetched connectors from API:', connectorNames) + return connectorNames + } + } catch (error) { + console.warn('Failed to fetch connectors from API, using fallback list:', error) } - } catch (error) { - console.warn('Failed to fetch connectors from API, using fallback list:', error) + return fallbackConnectors + })() + try { + return await connectorsRequest + } finally { + connectorsRequest = null } - return fallbackConnectors } // Get Message Docs @@ -161,9 +177,13 @@ export async function cacheDoc(cacheStorageOfMessageDocs: any): Promise { const logMessage = `Caching message docs { connector: ${connector} }` console.log(logMessage) updateLoadingInfoMessage(logMessage) - const docs = await getOBPMessageDocs(connector) - if (!Object.keys(docs).includes('code')) { - messageDocs[connector] = getGroupedMessageDocs(docs) + try { + const docs = await getOBPMessageDocs(connector) + if (!Object.keys(docs).includes('code')) { + messageDocs[connector] = getGroupedMessageDocs(docs) + } + } catch (error: any) { + console.warn(`[CACHE] WARNING: Skipping message docs for connector ${connector}:`, error.message || error) } }) await cacheStorageOfMessageDocs.put('/', new Response(JSON.stringify(messageDocs))) @@ -181,9 +201,13 @@ export async function cacheDocJsonSchema(cacheStorageOfMessageDocsJsonSchema: an const logMessage = `Caching message docs JSON schema { connector: ${connector} }` console.log(logMessage) updateLoadingInfoMessage(logMessage) - const docs = await getOBPMessageDocsJsonSchema(connector) - if (!Object.keys(docs).includes('code')) { - messageDocsJsonSchema[connector] = getGroupedMessageDocsJsonSchema(docs) + try { + const docs = await getOBPMessageDocsJsonSchema(connector) + if (!Object.keys(docs).includes('code')) { + messageDocsJsonSchema[connector] = getGroupedMessageDocsJsonSchema(docs) + } + } catch (error: any) { + console.warn(`[CACHE] WARNING: Skipping message docs JSON schema for connector ${connector}:`, error.message || error) } }) await cacheStorageOfMessageDocsJsonSchema.put( From 23b975e0f7bcd74518cdccd22048bb045ace668c Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 16:37:46 +0200 Subject: [PATCH 09/10] refactor: reduce repeated isDynamicEntity branching in resource docs caching cacheOneVersion branched on isDynamicEntity separately for each log line. Compute the 'kind' label once and reuse it, cutting the ternary count without splitting the shared try/catch/concurrency skeleton into two functions. --- src/obp/resource-docs.ts | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/src/obp/resource-docs.ts b/src/obp/resource-docs.ts index 0770903..6c291e9 100644 --- a/src/obp/resource-docs.ts +++ b/src/obp/resource-docs.ts @@ -137,30 +137,21 @@ export async function cacheDoc(cacheStorageOfResourceDocs: any): Promise { // we need this to cache the dynamic entities resource doc const isDynamicEntity = api_short_version === 'dynamic-entity' const version = `${api_standard.toUpperCase()}${api_short_version}` + const kind = isDynamicEntity ? 'dynamic resource docs' : 'resource docs' try { - console.log( - isDynamicEntity - ? `[CACHE] Attempting to load dynamic resource docs for: ${version}` - : `[CACHE] Attempting to load resource docs for: ${version}` - ) + console.log(`[CACHE] Attempting to load ${kind} for: ${version}`) const resourceDocs = isDynamicEntity ? await getOBPDynamicResourceDocs(version) : await getOBPResourceDocs(version) if (version && Object.keys(resourceDocs).includes('resource_docs')) { resourceDocsMapping[version] = resourceDocs - console.log( - isDynamicEntity - ? `[CACHE] Successfully cached dynamic docs for: ${version}` - : `[CACHE] Successfully cached docs for: ${version}` - ) + console.log(`[CACHE] Successfully cached ${kind} for: ${version}`) } else { console.warn(`[CACHE] WARNING: Response for ${version} missing 'resource_docs' field`) } } catch (error: any) { console.warn( - isDynamicEntity - ? `[CACHE] WARNING: Skipping dynamic endpoint ${api_standard}${api_short_version}:` - : `[CACHE] WARNING: Skipping API version ${api_standard}${api_short_version}:` + `[CACHE] WARNING: Skipping ${isDynamicEntity ? 'dynamic endpoint' : 'API version'} ${api_standard}${api_short_version}:` ) console.warn(` API Version: ${api_short_version}`) console.warn(` API Standard: ${api_standard}`) From 307f83af4abe73f7219ad5cdcd99bba58bbb7739 Mon Sep 17 00:00:00 2001 From: Hongwei Date: Tue, 7 Jul 2026 16:37:57 +0200 Subject: [PATCH 10/10] fix: establish and reuse an Opey session for both stream and invoke Two gaps in the Opey session handling added when chat was routed through the backend proxy: - invoke() sent no session cookie at all, even though createOpeySession's own doc comment says both /stream and /invoke need one - only stream() was updated to establish and forward it. - stream() created a brand new Opey session on every single chat message instead of reusing one for the conversation, doubling backend requests and adding a round trip of latency per message. - handleAuthentication() on the client used to prove the session actually worked (by creating it directly) before marking the user authenticated; once that moved server-side, a truthy consent response alone was enough to flip userIsAuthenticated, so a broken session only surfaced later as a generic error on the first chat message. Add getOrCreateSessionCookie(), which reuses a cookie already cached on the config or creates and persists one, and call it from both stream() and invoke(). Call it eagerly from /opey/consent via a new establishSession() so a rejected session fails the request immediately instead of being deferred to the first message. --- server/routes/opey.ts | 8 ++++ server/schema/OpeySchema.ts | 3 ++ server/services/OpeyClientService.ts | 55 +++++++++++++++++++++++++--- 3 files changed, 60 insertions(+), 6 deletions(-) diff --git a/server/routes/opey.ts b/server/routes/opey.ts index 58bca65..d4dfbd7 100644 --- a/server/routes/opey.ts +++ b/server/routes/opey.ts @@ -256,6 +256,10 @@ router.post('/opey/consent', async (req: Request, res: Response) => { // If we have a consent id, we can get the consent from OBP const consent = await obpConsentsService.getConsentByConsentId(session, consentId) + // Establish (or confirm) the Opey session now, so a consent that OBP accepts but + // Opey rejects surfaces here instead of failing silently on the first chat message. + await opeyClientService.establishSession(session.opeyConfig) + return res.status(200).json({ consent_id: consent.consent_id, jwt: consent.jwt }) } else { console.log('No existing consent ID found') @@ -265,6 +269,10 @@ router.post('/opey/consent', async (req: Request, res: Response) => { console.log('Consent at controller: ', session.opeyConfig) + // Establish the Opey session now (see comment above) rather than deferring it to + // the first stream()/invoke() call. + await opeyClientService.establishSession(session.opeyConfig) + const authConfig = session.opeyConfig?.authConfig res.status(200).json({ diff --git a/server/schema/OpeySchema.ts b/server/schema/OpeySchema.ts index 7572e85..72c963c 100644 --- a/server/schema/OpeySchema.ts +++ b/server/schema/OpeySchema.ts @@ -29,6 +29,9 @@ export interface OpeyConfig { baseUri: string, paths: OpeyPaths, authConfig?: AuthConfig, + // Opey session cookie ("session=..."), established once via /create-session and + // reused for subsequent /stream and /invoke calls in the same conversation. + sessionCookie?: string, } export interface ConsentRequestResponse { diff --git a/server/services/OpeyClientService.ts b/server/services/OpeyClientService.ts index d9a32c3..c4af5ec 100644 --- a/server/services/OpeyClientService.ts +++ b/server/services/OpeyClientService.ts @@ -63,7 +63,13 @@ export default class OpeyClientService { }; } } - + + // Merge the Opey session cookie if provided, so a session established on an + // earlier call is picked up instead of creating a new one every time. + if (partialConfig.sessionCookie) { + mergedConfig.sessionCookie = partialConfig.sessionCookie; + } + return mergedConfig; } @@ -131,9 +137,9 @@ export default class OpeyClientService { // Get auth headers const authHeaders = await this.getConsentAuthHeaders(config) - // Opey's /stream requires an authenticated session. Establish one with the - // Consent-JWT first, then forward the returned session cookie on /stream. - const sessionCookie = await this.createOpeySession(config) + // Opey's /stream requires an authenticated session. Reuse one already + // established (e.g. by /opey/consent) or create one now, then forward it. + const sessionCookie = await this.getOrCreateSessionCookie(config, opeyConfig) try { @@ -188,7 +194,11 @@ export default class OpeyClientService { // Get auth headers const authHeaders = await this.getConsentAuthHeaders(config) - + + // Opey's /invoke requires the same authenticated session as /stream (see + // createOpeySession's doc comment) — reuse one already established or create it. + const sessionCookie = await this.getOrCreateSessionCookie(config, opeyConfig) + const url = `${config.baseUri}${config.paths.invoke}` console.log(`Posting to Opey, STREAMING OFF: ${JSON.stringify(user_input)}\n URL: ${url}`) //DEBUG @@ -196,7 +206,7 @@ export default class OpeyClientService { try { const response = await fetch(url, { method: 'POST', - headers: authHeaders, + headers: { ...authHeaders, 'Cookie': sessionCookie }, body: JSON.stringify(user_input) }) if (response.status === 200) { @@ -261,6 +271,39 @@ export default class OpeyClientService { * calls. Opey's chat endpoints require a session (created via /create-session); * driving it server-side keeps the browser same-origin (mirrors the Portal design). */ + /** + * Returns an Opey session cookie, reusing `config.sessionCookie` if one was already + * established, otherwise creating one via createOpeySession(). A freshly created cookie + * is written onto `persistTo` (the caller's own config object, e.g. an Express session's + * opeyConfig) so later calls in the same conversation reuse it instead of establishing a + * new session per message. + */ + async getOrCreateSessionCookie(config: OpeyConfig, persistTo?: Partial): Promise { + if (config.sessionCookie) { + return config.sessionCookie + } + const sessionCookie = await this.createOpeySession(config) + if (persistTo) { + persistTo.sessionCookie = sessionCookie + } + return sessionCookie + } + + /** + * Validates the auth config and eagerly establishes (or reuses) the Opey session, + * persisting the cookie onto `opeyConfig`. Intended to be called right after consent + * is granted so a broken/rejected consent surfaces immediately instead of only on the + * user's first chat message. + */ + async establishSession(opeyConfig: Partial): Promise { + const config = await this.getOpeyConfig(opeyConfig) + const auth = await this.checkAuthConfig(config) + if (!auth.valid) { + throw new Error(`AuthConfig not valid: ${auth.reason}`) + } + return await this.getOrCreateSessionCookie(config, opeyConfig) + } + async createOpeySession(opeyConfig: OpeyConfig): Promise { const authHeaders = await this.getConsentAuthHeaders(opeyConfig) const url = `${opeyConfig.baseUri}/create-session`