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 ca638a8..82896ae 100644 --- a/server/routes/oauth2.ts +++ b/server/routes/oauth2.ts @@ -254,6 +254,9 @@ router.get('/oauth2/callback', async (req: Request, res: Response) => { } // 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". // OBP-API validates the Bearer value as a JWT (issuer-based dispatch in OAuth2Login), // but some providers (e.g. Google) issue opaque access tokens (ya29...). For those, // send the id_token to OBP instead — OBP's Google branch expects it (applyIdTokenRules). @@ -265,6 +268,8 @@ router.get('/oauth2/callback', async (req: Request, res: Response) => { ) } session.clientConfig = { + baseUri: process.env.VITE_OBP_API_HOST, + version: process.env.VITE_OBP_API_VERSION, oauth2: { accessToken: obpAccessToken, tokenType: 'Bearer' 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 4b5af16..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,7 +137,10 @@ export default class OpeyClientService { // Get auth headers const authHeaders = await this.getConsentAuthHeaders(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 { const url = `${config.baseUri}${config.paths.stream}` @@ -140,10 +149,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) { @@ -185,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 @@ -193,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) { @@ -242,7 +255,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 +264,67 @@ 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). + */ + /** + * 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` + 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/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) diff --git a/src/obp/common-functions.ts b/src/obp/common-functions.ts index 3748269..a332cf6 100644 --- a/src/obp/common-functions.ts +++ b/src/obp/common-functions.ts @@ -103,6 +103,37 @@ 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: 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 + 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) { if ('caches' in window) { caches.delete(cacheName).then(function(success) { diff --git a/src/obp/message-docs.ts b/src/obp/message-docs.ts index bc187a2..4617c82 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'] @@ -40,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 @@ -154,17 +172,20 @@ 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) + 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) } - return group - }, Promise.resolve({})) + }) await cacheStorageOfMessageDocs.put('/', new Response(JSON.stringify(messageDocs))) return messageDocs } @@ -175,17 +196,20 @@ 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) + 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) } - return group - }, Promise.resolve({})) + }) await cacheStorageOfMessageDocsJsonSchema.put( '/', new Response(JSON.stringify(messageDocsJsonSchema)) @@ -199,8 +223,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 +243,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..6c291e9 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,42 @@ 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}` + const kind = isDynamicEntity ? 'dynamic resource docs' : 'resource docs' 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(`[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(`[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(`[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}` + `[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}`) + 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 +174,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) { @@ -213,8 +195,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) 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: { diff --git a/src/test/run-with-concurrency.test.ts b/src/test/run-with-concurrency.test.ts new file mode 100644 index 0000000..9870d98 --- /dev/null +++ b/src/test/run-with-concurrency.test.ts @@ -0,0 +1,81 @@ +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) + }) + + 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']) + }) +})