feat(oci-streaming): add native streaming integration

Author: Bill Leoutsakos

apps/docs/components/ui/icon-mapping.ts

@@ -482,6 +482,7 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
   notion: NotionIcon,
   notion_v2: NotionIcon,
   obsidian: ObsidianIcon,
+  oci_streaming: NetSuiteIcon,
   okta: OktaIcon,
   onedrive: MicrosoftOneDriveIcon,
   onepassword: OnePasswordIcon,

apps/docs/content/docs/integrations/meta.json

@@ -185,6 +185,7 @@
     "notion",
     "notion-service-account",
     "obsidian",
+    "oci_streaming",
     "okta",
     "onedrive",
     "onepassword",

apps/docs/content/docs/integrations/oci_streaming.mdx

@@ -249,6 +249,7 @@ import {
   NotionV2BlockMeta,
 } from '@/blocks/blocks/notion'
 import { ObsidianBlock, ObsidianBlockMeta } from '@/blocks/blocks/obsidian'
+import { OciStreamingBlock, OciStreamingBlockMeta } from '@/blocks/blocks/oci_streaming'
 import { OktaBlock, OktaBlockMeta } from '@/blocks/blocks/okta'
 import { OneDriveBlock, OneDriveBlockMeta } from '@/blocks/blocks/onedrive'
 import { OnePasswordBlock, OnePasswordBlockMeta } from '@/blocks/blocks/onepassword'
@@ -595,6 +596,7 @@ export const BLOCK_REGISTRY: Record<string, BlockConfig> = {
   notion: NotionBlock,
   notion_v2: NotionV2Block,
   obsidian: ObsidianBlock,
+  oci_streaming: OciStreamingBlock,
   okta: OktaBlock,
   onedrive: OneDriveBlock,
   onepassword: OnePasswordBlock,
@@ -920,6 +922,7 @@ export const BLOCK_META_REGISTRY: Record<string, BlockMeta> = {
   notion: NotionBlockMeta,
   notion_v2: NotionV2BlockMeta,
   obsidian: ObsidianBlockMeta,
+  oci_streaming: OciStreamingBlockMeta,
   okta: OktaBlockMeta,
   onedrive: OneDriveBlockMeta,
   onepassword: OnePasswordBlockMeta,

apps/sim/blocks/blocks/oci_streaming.ts

@@ -243,6 +243,7 @@ export const DOCS_MANIFEST: readonly string[] = [
   'integrations/notion-service-account.mdx',
   'integrations/notion.mdx',
   'integrations/obsidian.mdx',
+  'integrations/oci_streaming.mdx',
   'integrations/okta.mdx',
   'integrations/onedrive.mdx',
   'integrations/onepassword.mdx',

apps/sim/blocks/registry-maps.ts

@@ -465,6 +465,7 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
   notion: NotionIcon,
   notion_v2: NotionIcon,
   obsidian: ObsidianIcon,
+  oci_streaming: NetSuiteIcon,
   okta: OktaIcon,
   onedrive: MicrosoftOneDriveIcon,
   onepassword: OnePasswordIcon,

apps/sim/lib/copilot/generated/docs-manifest.ts

@@ -0,0 +1,166 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({
+  authorize: vi.fn(),
+  create: vi.fn(),
+  prepare: vi.fn(),
+  request: vi.fn(),
+}))
+vi.mock('@/lib/auth/credential-access', () => ({ authorizeCredentialUseForAuth: mocks.authorize }))
+vi.mock('@/lib/internal/oci/client.server', () => ({ createOciClient: mocks.create }))
+
+import { AuthType } from '@/lib/auth/hybrid'
+import { executeOciStreamingTool } from '@/lib/internal/oci-streaming/execute-tool'
+import type { InternalToolOperationCall } from '@/lib/internal/tool-operations/types'
+
+function call(overrides: Partial<InternalToolOperationCall> = {}): InternalToolOperationCall {
+  return {
+    toolId: 'oci_streaming_list_streams',
+    input: {
+      operation: 'list_streams',
+      ociCredential: 'supplied-reference',
+      compartmentId: 'compartment-1',
+      ociRegion: 'us-ashburn-1',
+    },
+    context: { userId: 'actor-1', workspaceId: 'workspace-1', workflowId: 'workflow-1' },
+    headers: new Headers(),
+    requestId: 'request-1',
+    ...overrides,
+  }
+}
+
+describe('OCI Streaming trusted credential boundary', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mocks.authorize.mockResolvedValue({
+      ok: true,
+      credentialType: 'service_account',
+      resolvedCredentialId: 'resolved-credential',
+      workspaceId: 'workspace-1',
+    })
+    mocks.create.mockResolvedValue({ prepareStaticEndpoint: mocks.prepare, request: mocks.request })
+    mocks.prepare.mockResolvedValue({
+      origin: 'https://streaming.us-ashburn-1.oci.oraclecloud.com',
+    })
+    mocks.request.mockResolvedValue({
+      status: 200,
+      body: new TextEncoder().encode('[]'),
+      headers: {},
+    })
+  })
+
+  it('passes only the resolved credential ID and trusted workspace to the foundation', async () => {
+    const response = await executeOciStreamingTool(call())
+    expect(response.status).toBe(200)
+    expect(mocks.authorize).toHaveBeenCalledWith(
+      { success: true, userId: 'actor-1', authType: AuthType.INTERNAL_JWT },
+      {
+        credentialId: 'supplied-reference',
+        workspaceId: 'workspace-1',
+        workflowId: 'workflow-1',
+        callerUserId: 'actor-1',
+      }
+    )
+    expect(mocks.create).toHaveBeenCalledWith({
+      credentialId: 'resolved-credential',
+      workspaceId: 'workspace-1',
+      serviceId: 'oci-streaming',
+      region: 'us-ashburn-1',
+    })
+  })
+
+  it.each([
+    { ok: false },
+    {
+      ok: true,
+      credentialType: 'oauth',
+      resolvedCredentialId: 'resolved',
+      workspaceId: 'workspace-1',
+    },
+    { ok: true, credentialType: 'service_account', workspaceId: 'workspace-1' },
+    {
+      ok: true,
+      credentialType: 'service_account',
+      resolvedCredentialId: 'resolved',
+      workspaceId: 'other-workspace',
+    },
+  ])('starts no provider work for unauthorized credential resolution', async (access) => {
+    mocks.authorize.mockResolvedValue(access)
+    expect((await executeOciStreamingTool(call())).status).toBe(403)
+    expect(mocks.create).not.toHaveBeenCalled()
+    expect(mocks.request).not.toHaveBeenCalled()
+  })
+
+  it.each([
+    {
+      freeformTags: Object.fromEntries(Array.from({ length: 11 }, (_, i) => [String(i), 'value'])),
+    },
+    {
+      definedTags: {
+        namespace: Object.fromEntries(Array.from({ length: 65 }, (_, i) => [String(i), 'value'])),
+      },
+    },
+    { freeformTags: { key: 'x'.repeat(257) } },
+    {
+      definedTags: {
+        namespace: Object.fromEntries(
+          Array.from({ length: 20 }, (_, i) => [String(i), 'é'.repeat(256)])
+        ),
+      },
+    },
+  ])(
+    'rejects oversized administrative tags before authorization or provider work',
+    async (tags) => {
+      const response = await executeOciStreamingTool(
+        call({
+          toolId: 'oci_streaming_update_stream',
+          input: {
+            operation: 'update_stream',
+            ociCredential: 'credential',
+            streamId: 'stream',
+            ...tags,
+          },
+        })
+      )
+      expect(response.status).toBe(400)
+      expect(mocks.authorize).not.toHaveBeenCalled()
+      expect(mocks.create).not.toHaveBeenCalled()
+      expect(mocks.request).not.toHaveBeenCalled()
+    }
+  )
+
+  it('rejects forged context in operation input before authorization', async () => {
+    const request = call()
+    request.input = {
+      operation: 'list_streams',
+      ociCredential: 'credential',
+      compartmentId: 'compartment',
+      workspaceId: 'forged',
+      _context: { userId: 'forged' },
+    }
+    expect((await executeOciStreamingTool(request)).status).toBe(400)
+    expect(mocks.authorize).not.toHaveBeenCalled()
+    expect(mocks.create).not.toHaveBeenCalled()
+  })
+
+  it('rejects missing trusted identity and mismatched operation IDs', async () => {
+    expect((await executeOciStreamingTool(call({ context: { workflowId: '' } }))).status).toBe(401)
+    expect((await executeOciStreamingTool(call({ toolId: 'oci_streaming_delete_stream' }))).status).toBe(400)
+    expect(mocks.authorize).not.toHaveBeenCalled()
+  })
+
+  it('does not load credentials after cancellation while authorization is pending', async () => {
+    const controller = new AbortController()
+    let finish: ((value: unknown) => void) | undefined
+    mocks.authorize.mockImplementation(() => new Promise((resolve) => { finish = resolve }))
+    const pending = executeOciStreamingTool(call({ signal: controller.signal }))
+    controller.abort(new DOMException('Canceled', 'AbortError'))
+    await pending
+    finish?.({ ok: true, credentialType: 'service_account', resolvedCredentialId: 'resolved', workspaceId: 'workspace-1' })
+    await Promise.resolve()
+    expect(mocks.create).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/integrations/icon-mapping.ts

@@ -0,0 +1,102 @@
+import { getErrorMessage } from '@sim/utils/errors'
+import { authorizeCredentialUseForAuth } from '@/lib/auth/credential-access'
+import { AuthType } from '@/lib/auth/hybrid'
+import { createOciClient } from '@/lib/internal/oci/client.server'
+import { OciClientError } from '@/lib/internal/oci/errors'
+import {
+  awaitOciStreaming,
+  executeOciStreamingOperation,
+  OCI_STREAMING_ADMIN_ENDPOINT,
+  OCI_STREAMING_SERVICE_ID,
+  withOciStreamingBudget,
+} from '@/lib/internal/oci-streaming/operations'
+import { ociStreamingInputSchema } from '@/lib/internal/oci-streaming/schema'
+import type { InternalToolOperationHandler } from '@/lib/internal/tool-operations/types'
+
+export const executeOciStreamingTool: InternalToolOperationHandler = async (request) => {
+  const parsed = ociStreamingInputSchema.safeParse(request.input)
+  if (!parsed.success) {
+    return Response.json(
+      { success: false, error: parsed.error.issues.map((issue) => issue.message).join('; ') },
+      { status: 400 }
+    )
+  }
+  const input = parsed.data
+  if (request.toolId !== `oci_streaming_${input.operation}`) {
+    return Response.json(
+      { success: false, error: 'OCI Streaming operation does not match tool' },
+      { status: 400 }
+    )
+  }
+  const { userId, workspaceId, workflowId } = request.context
+  if (!userId || !workspaceId) {
+    return Response.json(
+      { success: false, error: 'Trusted user and workspace context are required' },
+      { status: 401 }
+    )
+  }
+  try {
+    return await withOciStreamingBudget(async (budget) => {
+      const access = await awaitOciStreaming(
+        authorizeCredentialUseForAuth(
+          { success: true, userId, authType: AuthType.INTERNAL_JWT },
+          {
+            credentialId: input.ociCredential,
+            workspaceId,
+            workflowId: workflowId || undefined,
+            callerUserId: userId,
+          }
+        ),
+        budget.signal
+      )
+      if (
+        !access.ok ||
+        access.credentialType !== 'service_account' ||
+        !access.resolvedCredentialId ||
+        access.workspaceId !== workspaceId
+      ) {
+        return Response.json(
+          { success: false, error: 'OCI service account is not accessible in this workspace' },
+          { status: 403 }
+        )
+      }
+      budget.signal.throwIfAborted()
+      const client = await awaitOciStreaming(
+        createOciClient({
+          credentialId: access.resolvedCredentialId,
+          workspaceId,
+          serviceId: OCI_STREAMING_SERVICE_ID,
+          region: input.ociRegion,
+        }),
+        budget.signal
+      )
+      budget.signal.throwIfAborted()
+      const endpoint = await awaitOciStreaming(
+        client.prepareStaticEndpoint(OCI_STREAMING_ADMIN_ENDPOINT),
+        budget.signal
+      )
+      budget.signal.throwIfAborted()
+      const result = await executeOciStreamingOperation(input, { client, endpoint }, budget)
+      return Response.json(result)
+    }, request.signal)
+  } catch (error) {
+    if (error instanceof OciClientError) {
+      return Response.json(
+        {
+          success: false,
+          error: error.message,
+          output: {
+            status: error.status ?? null,
+            requestId: error.opcRequestId ?? null,
+            code: error.code,
+          },
+        },
+        { status: error.status && error.status >= 400 ? error.status : 502 }
+      )
+    }
+    return Response.json(
+      { success: false, error: getErrorMessage(error, 'OCI Streaming operation failed') },
+      { status: 400 }
+    )
+  }
+}