gRPC and SSE status

.env.example

@@ -2,6 +2,12 @@
 VITE_OBP_API_HOST=http://127.0.0.1:8080
 VITE_OBP_API_VERSION=v5.1.0
 
+### OBP gRPC endpoint (used by the gRPC services browser) ###
+### Defaults to grpc.<VITE_OBP_API_HOST hostname> if not set — port 443 with TLS for an https
+### base URL, port 50051 without TLS for http (no grpc. prefix for localhost/IPs).
+### VITE_OBP_GRPC_TLS=true|false overrides the port-based TLS default.
+# VITE_OBP_GRPC_HOST=localhost:50051
+
 ### API Explorer Host ###
 VITE_OBP_API_EXPLORER_HOST=http://localhost:5173
 

server/routes/grpc.ts

@@ -29,10 +29,13 @@ import { Router } from 'express'
 import type { Request, Response } from 'express'
 import { credentials } from '@grpc/grpc-js'
 import { Client as ReflectionClient } from 'grpc-reflection-js'
+import { resolveGrpcTarget } from '../utils/grpcHost.js'
 
 const router = Router()
 
-const GRPC_HOST = process.env.VITE_OBP_GRPC_HOST || 'localhost:50051'
+const GRPC_TARGET = resolveGrpcTarget(process.env)
+const GRPC_HOST = GRPC_TARGET.host
+const GRPC_CREDENTIALS = GRPC_TARGET.tls ? credentials.createSsl() : credentials.createInsecure()
 
 const REFLECTION_SERVICE_NAMES = new Set([
   'grpc.reflection.v1.ServerReflection',
@@ -80,7 +83,7 @@ router.get('/grpc/services', async (_req: Request, res: Response) => {
   let client: ReflectionClient | null = null
   try {
     console.log(`gRPC: Reflecting against ${GRPC_HOST}`)
-    client = new ReflectionClient(GRPC_HOST, credentials.createInsecure())
+    client = new ReflectionClient(GRPC_HOST, GRPC_CREDENTIALS)
 
     const serviceNames = await client.listServices()
     const services: ServiceInfo[] = []

server/routes/status.ts

@@ -37,7 +37,8 @@ import {
   RESOURCE_DOCS_API_VERSION,
   MESSAGE_DOCS_API_VERSION,
   API_VERSIONS_LIST_API_VERSION,
-  V5_1_0
+  V5_1_0,
+  SSE_PROBE_SPACING_MS
 } from '../../src/shared-constants.js'
 
 const router = Router()
@@ -160,6 +161,26 @@ router.get('/health', (req: Request, res: Response) => {
   })
 })
 
+/**
+ * GET /status/stream
+ * SSE transport probe for the status page: emits two spaced events so the
+ * browser can tell real streaming from a proxy-buffered response. Uses the
+ * same headers as the real Opey SSE stream and no proxy opt-outs
+ * (e.g. X-Accel-Buffering), so it experiences the same proxy behavior.
+ * Carries no data, so no auth.
+ */
+router.get('/status/stream', (req: Request, res: Response) => {
+  res.setHeader('Content-Type', 'text/event-stream')
+  res.setHeader('Cache-Control', 'no-cache')
+  res.setHeader('Connection', 'keep-alive')
+  res.write(':ok\n\ndata: {"seq":1}\n\n')
+  const timer = setTimeout(() => {
+    res.write('data: {"seq":2}\n\n')
+    res.end()
+  }, SSE_PROBE_SPACING_MS)
+  req.on('close', () => clearTimeout(timer))
+})
+
 /**
  * GET /status
  * Get application status and health checks

server/test/grpcHost.test.ts

@@ -0,0 +1,66 @@
+import { describe, it, expect } from 'vitest'
+import { resolveGrpcTarget, defaultGrpcHost } from '../utils/grpcHost'
+
+describe('resolveGrpcTarget', () => {
+  it('prefers VITE_OBP_GRPC_HOST when set', () => {
+    expect(
+      resolveGrpcTarget({
+        VITE_OBP_GRPC_HOST: 'grpc.example.com:9999',
+        VITE_OBP_API_HOST: 'https://api.example.com'
+      })
+    ).toEqual({ host: 'grpc.example.com:9999', tls: false })
+  })
+
+  it('derives grpc.<host>:443 with TLS from an https VITE_OBP_API_HOST', () => {
+    expect(resolveGrpcTarget({ VITE_OBP_API_HOST: 'https://api.example.com' })).toEqual({
+      host: 'grpc.api.example.com:443',
+      tls: true
+    })
+  })
+
+  it('derives grpc.<host>:50051 without TLS from an http VITE_OBP_API_HOST', () => {
+    expect(resolveGrpcTarget({ VITE_OBP_API_HOST: 'http://obp.internal:8080' })).toEqual({
+      host: 'grpc.obp.internal:50051',
+      tls: false
+    })
+  })
+
+  it('turns on TLS for an explicit host on port 443', () => {
+    expect(resolveGrpcTarget({ VITE_OBP_GRPC_HOST: 'grpc.example.com:443' }).tls).toBe(true)
+  })
+
+  it('lets VITE_OBP_GRPC_TLS override the port-based default in both directions', () => {
+    expect(
+      resolveGrpcTarget({ VITE_OBP_GRPC_HOST: 'grpc.example.com:443', VITE_OBP_GRPC_TLS: 'false' })
+        .tls
+    ).toBe(false)
+    expect(
+      resolveGrpcTarget({ VITE_OBP_GRPC_HOST: 'grpc.example.com:50051', VITE_OBP_GRPC_TLS: 'true' })
+        .tls
+    ).toBe(true)
+  })
+
+  it('falls back to localhost:50051 without TLS when nothing is set', () => {
+    expect(resolveGrpcTarget({})).toEqual({ host: 'localhost:50051', tls: false })
+  })
+})
+
+describe('defaultGrpcHost', () => {
+  it('uses port 443 for https base URLs and 50051 for http ones', () => {
+    expect(defaultGrpcHost('https://api.example.com')).toBe('grpc.api.example.com:443')
+    expect(defaultGrpcHost('http://obp.internal:8080')).toBe('grpc.obp.internal:50051')
+  })
+
+  it('does not prefix grpc. onto localhost or IP literals', () => {
+    expect(defaultGrpcHost('http://localhost:8080')).toBe('localhost:50051')
+    expect(defaultGrpcHost('http://obp.localhost:8080')).toBe('obp.localhost:50051')
+    expect(defaultGrpcHost('http://127.0.0.1:8080')).toBe('127.0.0.1:50051')
+    expect(defaultGrpcHost('http://[::1]:8080')).toBe('[::1]:50051')
+  })
+
+  it('falls back to localhost when the base URL is unset or unparseable', () => {
+    expect(defaultGrpcHost(undefined)).toBe('localhost:50051')
+    expect(defaultGrpcHost('')).toBe('localhost:50051')
+    expect(defaultGrpcHost('not a url')).toBe('localhost:50051')
+  })
+})

server/utils/grpcHost.ts

@@ -0,0 +1,56 @@
+// OBP-API deployments conventionally expose gRPC on a `grpc.` subdomain of the
+// REST host. Behind a public (https) deployment that subdomain serves gRPC
+// through the ingress on port 443 with TLS — a raw high port is typically not
+// reachable there — so when VITE_OBP_GRPC_HOST is unset the default is
+// grpc.<VITE_OBP_API_HOST hostname>:443 with TLS for https deployments, and
+// port 50051 without TLS for http (dev) ones. localhost and IP literals get no
+// `grpc.` prefix (there is no subdomain to resolve there).
+
+export const DEFAULT_GRPC_PORT = 50051
+export const DEFAULT_GRPC_TLS_PORT = 443
+
+export interface GrpcTarget {
+  /** gRPC target as host:port (no scheme). */
+  host: string
+  /** Whether to dial with TLS channel credentials. */
+  tls: boolean
+}
+
+/**
+ * The gRPC target to connect to, resolved from an env-like record
+ * (pass `process.env`): VITE_OBP_GRPC_HOST when set, otherwise derived from
+ * VITE_OBP_API_HOST. TLS follows VITE_OBP_GRPC_TLS ("true"/"false") when set,
+ * otherwise the port: 443 means TLS.
+ */
+export function resolveGrpcTarget(env: Record<string, string | undefined>): GrpcTarget {
+  const host = env.VITE_OBP_GRPC_HOST || defaultGrpcHost(env.VITE_OBP_API_HOST)
+  const tls =
+    env.VITE_OBP_GRPC_TLS !== undefined
+      ? env.VITE_OBP_GRPC_TLS === 'true'
+      : host.endsWith(`:${DEFAULT_GRPC_TLS_PORT}`)
+  return { host, tls }
+}
+
+export function defaultGrpcHost(obpApiHost: string | undefined | null): string {
+  if (obpApiHost) {
+    try {
+      const url = new URL(obpApiHost)
+      if (grpcSubdomainApplies(url.hostname)) {
+        const port = url.protocol === 'https:' ? DEFAULT_GRPC_TLS_PORT : DEFAULT_GRPC_PORT
+        return `grpc.${url.hostname}:${port}`
+      }
+      return `${url.hostname}:${DEFAULT_GRPC_PORT}`
+    } catch {
+      // unparseable base URL — fall through to localhost
+    }
+  }
+  return `localhost:${DEFAULT_GRPC_PORT}`
+}
+
+function grpcSubdomainApplies(hostname: string): boolean {
+  if (hostname === 'localhost' || hostname.endsWith('.localhost')) {
+    return false
+  }
+  // IPv4 literal; IPv6 literals contain ':' (URL.hostname keeps their brackets)
+  return !/^\d{1,3}(\.\d{1,3}){3}$/.test(hostname) && !hostname.includes(':')
+}

src/obp/sseProbe.ts

@@ -0,0 +1,100 @@
+// Browser-side judge for the SSE transport probe served at SSE_PROBE_PATH.
+//
+// The Opey chat streams over SSE from the browser to the Express server; a
+// reverse proxy that buffers responses breaks that silently while every
+// request/response check stays green. The server emits two events a fixed
+// interval apart; if they arrive together instead of spaced, something between
+// the browser and the server is buffering the stream.
+
+import {
+  SSE_PROBE_PATH,
+  SSE_PROBE_EVENT_COUNT,
+  SSE_PROBE_SPACING_MS
+} from '../shared-constants.js'
+
+export interface SseProbeResult {
+  ok: boolean
+  /** ms from request start until the first event arrived */
+  timeToFirstEventMs?: number
+  /** ms between arrival of the first and last event — near zero means buffered */
+  eventSpreadMs?: number
+  buffered?: boolean
+  error?: string
+}
+
+export async function runSseProbe(
+  options: {
+    timeoutMs?: number
+    spacingMs?: number
+    path?: string
+    fetchFn?: typeof fetch
+  } = {}
+): Promise<SseProbeResult> {
+  const {
+    timeoutMs = 5000,
+    spacingMs = SSE_PROBE_SPACING_MS,
+    path = SSE_PROBE_PATH,
+    fetchFn = fetch
+  } = options
+
+  const start = performance.now()
+  const controller = new AbortController()
+  const timer = setTimeout(() => controller.abort('timeout'), timeoutMs)
+  try {
+    const res = await fetchFn(path, {
+      signal: controller.signal,
+      headers: { accept: 'text/event-stream' }
+    })
+    if (!res.ok) {
+      return { ok: false, error: `Unexpected status code: ${res.status}` }
+    }
+    if (!res.body) {
+      return { ok: false, error: 'Response has no body stream' }
+    }
+
+    const reader = res.body.getReader()
+    const decoder = new TextDecoder()
+    const eventArrivals: number[] = []
+    let pending = ''
+    for (;;) {
+      const { done, value } = await reader.read()
+      if (done) break
+      pending += decoder.decode(value, { stream: true })
+      const blocks = pending.split('\n\n')
+      pending = blocks.pop() ?? ''
+      const now = performance.now()
+      for (const block of blocks) {
+        if (block.split('\n').some((line) => line.startsWith('data:'))) {
+          eventArrivals.push(now)
+        }
+      }
+    }
+
+    if (eventArrivals.length < SSE_PROBE_EVENT_COUNT) {
+      return {
+        ok: false,
+        error: `Stream ended after ${eventArrivals.length} of ${SSE_PROBE_EVENT_COUNT} events`
+      }
+    }
+
+    const timeToFirstEventMs = Math.round(eventArrivals[0] - start)
+    const eventSpreadMs = Math.round(eventArrivals[eventArrivals.length - 1] - eventArrivals[0])
+    // The server spaced the events spacingMs apart; arriving in less than half
+    // that means they were held back and delivered together.
+    const buffered = eventSpreadMs < spacingMs / 2
+    return {
+      ok: !buffered,
+      timeToFirstEventMs,
+      eventSpreadMs,
+      buffered,
+      error: buffered
+        ? `Events arrived ${eventSpreadMs}ms apart though the server spaced them ${spacingMs}ms apart — a proxy between the browser and the server is buffering SSE responses, which breaks live streaming`
+        : undefined
+    }
+  } catch (err) {
+    const msg = err instanceof Error ? err.message : String(err)
+    return { ok: false, error: msg === 'timeout' ? 'Request timeout' : msg }
+  } finally {
+    clearTimeout(timer)
+  }
+}

src/shared-constants.ts

@@ -37,3 +37,13 @@ export const GLOSSARY_API_VERSION = 'v5.1.0'
  */
 export const V5_1_0 = 'v5.1.0'
 export const V6_0_0 = 'v6.0.0'
+
+/**
+ * Browser → Node SSE transport probe (see /api/status/stream and the status
+ * page's browser-side check). The server emits SSE_PROBE_EVENT_COUNT events
+ * SSE_PROBE_SPACING_MS apart; the browser judges the transport buffered when
+ * they arrive closer together than half that spacing.
+ */
+export const SSE_PROBE_PATH = '/api/status/stream'
+export const SSE_PROBE_EVENT_COUNT = 2
+export const SSE_PROBE_SPACING_MS = 700

src/test/sseProbe.test.ts

@@ -0,0 +1,76 @@
+import { describe, it, expect } from 'vitest'
+import { runSseProbe } from '../obp/sseProbe'
+
+// A genuinely streamed response: two events spacingMs apart.
+function streamedResponse(spacingMs: number): Response {
+  const encoder = new TextEncoder()
+  let timer: ReturnType<typeof setTimeout> | undefined
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.enqueue(encoder.encode(':ok\n\ndata: {"seq":1}\n\n'))
+      timer = setTimeout(() => {
+        controller.enqueue(encoder.encode('data: {"seq":2}\n\n'))
+        controller.close()
+      }, spacingMs)
+    },
+    cancel() {
+      clearTimeout(timer)
+    }
+  })
+  return new Response(stream, { headers: { 'Content-Type': 'text/event-stream' } })
+}
+
+// The same events delivered in a single chunk — what a buffering proxy produces.
+function bufferedResponse(): Response {
+  return new Response(':ok\n\ndata: {"seq":1}\n\ndata: {"seq":2}\n\n', {
+    headers: { 'Content-Type': 'text/event-stream' }
+  })
+}
+
+describe('runSseProbe', () => {
+  it('reports healthy for a genuinely streamed response', async () => {
+    const result = await runSseProbe({
+      spacingMs: 200,
+      fetchFn: async () => streamedResponse(200)
+    })
+    expect(result.ok).toBe(true)
+    expect(result.buffered).toBe(false)
+    expect(result.eventSpreadMs).toBeGreaterThanOrEqual(100)
+  })
+
+  it('detects a buffered stream', async () => {
+    const result = await runSseProbe({ fetchFn: async () => bufferedResponse() })
+    expect(result.ok).toBe(false)
+    expect(result.buffered).toBe(true)
+    expect(result.error).toContain('buffering')
+  })
+
+  it('reports an HTTP error status', async () => {
+    const result = await runSseProbe({
+      fetchFn: async () => new Response('nope', { status: 502 })
+    })
+    expect(result.ok).toBe(false)
+    expect(result.error).toContain('502')
+  })
+
+  it('reports a truncated stream', async () => {
+    const result = await runSseProbe({
+      fetchFn: async () =>
+        new Response('data: {"seq":1}\n\n', {
+          headers: { 'Content-Type': 'text/event-stream' }
+        })
+    })
+    expect(result.ok).toBe(false)
+    expect(result.error).toContain('1 of 2 events')
+  })
+
+  it('reports a network failure', async () => {
+    const result = await runSseProbe({
+      fetchFn: async () => {
+        throw new Error('Failed to fetch')
+      }
+    })
+    expect(result.ok).toBe(false)
+    expect(result.error).toBe('Failed to fetch')
+  })
+})