Skip to content

Commit 6f54ae5

Browse files
Bill LeoutsakosBill Leoutsakos
authored andcommitted
feat(oci-streaming): add native streaming integration
1 parent 3fa59e7 commit 6f54ae5

54 files changed

Lines changed: 5334 additions & 7 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -482,6 +482,7 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
482482
notion: NotionIcon,
483483
notion_v2: NotionIcon,
484484
obsidian: ObsidianIcon,
485+
oci_streaming: NetSuiteIcon,
485486
okta: OktaIcon,
486487
onedrive: MicrosoftOneDriveIcon,
487488
onepassword: OnePasswordIcon,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,7 @@
185185
"notion",
186186
"notion-service-account",
187187
"obsidian",
188+
"oci_streaming",
188189
"okta",
189190
"onedrive",
190191
"onepassword",

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

Lines changed: 818 additions & 0 deletions
Large diffs are not rendered by default.

apps/sim/blocks/blocks/oci_streaming.ts

Lines changed: 520 additions & 0 deletions
Large diffs are not rendered by default.

apps/sim/blocks/registry-maps.ts

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -249,6 +249,7 @@ import {
249249
NotionV2BlockMeta,
250250
} from '@/blocks/blocks/notion'
251251
import { ObsidianBlock, ObsidianBlockMeta } from '@/blocks/blocks/obsidian'
252+
import { OciStreamingBlock, OciStreamingBlockMeta } from '@/blocks/blocks/oci_streaming'
252253
import { OktaBlock, OktaBlockMeta } from '@/blocks/blocks/okta'
253254
import { OneDriveBlock, OneDriveBlockMeta } from '@/blocks/blocks/onedrive'
254255
import { OnePasswordBlock, OnePasswordBlockMeta } from '@/blocks/blocks/onepassword'
@@ -595,6 +596,7 @@ export const BLOCK_REGISTRY: Record<string, BlockConfig> = {
595596
notion: NotionBlock,
596597
notion_v2: NotionV2Block,
597598
obsidian: ObsidianBlock,
599+
oci_streaming: OciStreamingBlock,
598600
okta: OktaBlock,
599601
onedrive: OneDriveBlock,
600602
onepassword: OnePasswordBlock,
@@ -920,6 +922,7 @@ export const BLOCK_META_REGISTRY: Record<string, BlockMeta> = {
920922
notion: NotionBlockMeta,
921923
notion_v2: NotionV2BlockMeta,
922924
obsidian: ObsidianBlockMeta,
925+
oci_streaming: OciStreamingBlockMeta,
923926
okta: OktaBlockMeta,
924927
onedrive: OneDriveBlockMeta,
925928
onepassword: OnePasswordBlockMeta,

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -243,6 +243,7 @@ export const DOCS_MANIFEST: readonly string[] = [
243243
'integrations/notion-service-account.mdx',
244244
'integrations/notion.mdx',
245245
'integrations/obsidian.mdx',
246+
'integrations/oci_streaming.mdx',
246247
'integrations/okta.mdx',
247248
'integrations/onedrive.mdx',
248249
'integrations/onepassword.mdx',

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -465,6 +465,7 @@ export const blockTypeToIconMap: Record<string, IconComponent> = {
465465
notion: NotionIcon,
466466
notion_v2: NotionIcon,
467467
obsidian: ObsidianIcon,
468+
oci_streaming: NetSuiteIcon,
468469
okta: OktaIcon,
469470
onedrive: MicrosoftOneDriveIcon,
470471
onepassword: OnePasswordIcon,
Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
/**
2+
* @vitest-environment node
3+
*/
4+
import { beforeEach, describe, expect, it, vi } from 'vitest'
5+
6+
const mocks = vi.hoisted(() => ({ authorize: vi.fn(), create: vi.fn(), prepare: vi.fn(), request: vi.fn() }))
7+
vi.mock('@/lib/auth/credential-access', () => ({ authorizeCredentialUseForAuth: mocks.authorize }))
8+
vi.mock('@/lib/internal/oci/client.server', () => ({ createOciClient: mocks.create }))
9+
10+
import { AuthType } from '@/lib/auth/hybrid'
11+
import { executeOciStreamingTool } from '@/lib/internal/oci-streaming/execute-tool'
12+
import type { InternalToolOperationCall } from '@/lib/internal/tool-operations/types'
13+
14+
function call(overrides: Partial<InternalToolOperationCall> = {}): InternalToolOperationCall {
15+
return {
16+
toolId: 'oci_streaming_list_streams',
17+
input: { operation: 'list_streams', ociCredential: 'supplied-reference', compartmentId: 'compartment-1', ociRegion: 'us-ashburn-1' },
18+
context: { userId: 'actor-1', workspaceId: 'workspace-1', workflowId: 'workflow-1' },
19+
headers: new Headers(), requestId: 'request-1', ...overrides,
20+
}
21+
}
22+
23+
describe('OCI Streaming trusted credential boundary', () => {
24+
beforeEach(() => {
25+
vi.clearAllMocks()
26+
mocks.authorize.mockResolvedValue({ ok: true, credentialType: 'service_account', resolvedCredentialId: 'resolved-credential', workspaceId: 'workspace-1' })
27+
mocks.create.mockResolvedValue({ prepareStaticEndpoint: mocks.prepare, request: mocks.request })
28+
mocks.prepare.mockResolvedValue({ origin: 'https://streaming.us-ashburn-1.oci.oraclecloud.com' })
29+
mocks.request.mockResolvedValue({ status: 200, body: new TextEncoder().encode('[]'), headers: {} })
30+
})
31+
32+
it('passes only the resolved credential ID and trusted workspace to the foundation', async () => {
33+
const response = await executeOciStreamingTool(call())
34+
expect(response.status).toBe(200)
35+
expect(mocks.authorize).toHaveBeenCalledWith(
36+
{ success: true, userId: 'actor-1', authType: AuthType.INTERNAL_JWT },
37+
{ credentialId: 'supplied-reference', workspaceId: 'workspace-1', workflowId: 'workflow-1', callerUserId: 'actor-1' }
38+
)
39+
expect(mocks.create).toHaveBeenCalledWith({ credentialId: 'resolved-credential', workspaceId: 'workspace-1', serviceId: 'oci-streaming', region: 'us-ashburn-1' })
40+
})
41+
42+
it.each([
43+
{ ok: false },
44+
{ ok: true, credentialType: 'oauth', resolvedCredentialId: 'resolved', workspaceId: 'workspace-1' },
45+
{ ok: true, credentialType: 'service_account', workspaceId: 'workspace-1' },
46+
{ ok: true, credentialType: 'service_account', resolvedCredentialId: 'resolved', workspaceId: 'other-workspace' },
47+
])('starts no provider work for unauthorized credential resolution', async (access) => {
48+
mocks.authorize.mockResolvedValue(access)
49+
expect((await executeOciStreamingTool(call())).status).toBe(403)
50+
expect(mocks.create).not.toHaveBeenCalled()
51+
expect(mocks.request).not.toHaveBeenCalled()
52+
})
53+
54+
it.each([
55+
{ freeformTags: Object.fromEntries(Array.from({ length: 11 }, (_, i) => [String(i), 'value'])) },
56+
{ definedTags: { namespace: Object.fromEntries(Array.from({ length: 65 }, (_, i) => [String(i), 'value'])) } },
57+
{ freeformTags: { key: 'x'.repeat(257) } },
58+
{ definedTags: { namespace: Object.fromEntries(Array.from({ length: 20 }, (_, i) => [String(i), 'é'.repeat(256)])) } },
59+
])('rejects oversized administrative tags before authorization or provider work', async (tags) => {
60+
const response = await executeOciStreamingTool(call({
61+
toolId: 'oci_streaming_update_stream',
62+
input: { operation: 'update_stream', ociCredential: 'credential', streamId: 'stream', ...tags },
63+
}))
64+
expect(response.status).toBe(400)
65+
expect(mocks.authorize).not.toHaveBeenCalled()
66+
expect(mocks.create).not.toHaveBeenCalled()
67+
expect(mocks.request).not.toHaveBeenCalled()
68+
})
69+
70+
it('rejects forged context in operation input before authorization', async () => {
71+
const request = call()
72+
request.input = { operation: 'list_streams', ociCredential: 'credential', compartmentId: 'compartment', workspaceId: 'forged', _context: { userId: 'forged' } }
73+
expect((await executeOciStreamingTool(request)).status).toBe(400)
74+
expect(mocks.authorize).not.toHaveBeenCalled()
75+
expect(mocks.create).not.toHaveBeenCalled()
76+
})
77+
78+
it('rejects missing trusted identity and mismatched operation IDs', async () => {
79+
expect((await executeOciStreamingTool(call({ context: { workflowId: '' } }))).status).toBe(401)
80+
expect((await executeOciStreamingTool(call({ toolId: 'oci_streaming_delete_stream' }))).status).toBe(400)
81+
expect(mocks.authorize).not.toHaveBeenCalled()
82+
})
83+
84+
it('does not load credentials after cancellation while authorization is pending', async () => {
85+
const controller = new AbortController()
86+
let finish: ((value: unknown) => void) | undefined
87+
mocks.authorize.mockImplementation(() => new Promise((resolve) => { finish = resolve }))
88+
const pending = executeOciStreamingTool(call({ signal: controller.signal }))
89+
controller.abort(new DOMException('Canceled', 'AbortError'))
90+
await pending
91+
finish?.({ ok: true, credentialType: 'service_account', resolvedCredentialId: 'resolved', workspaceId: 'workspace-1' })
92+
await Promise.resolve()
93+
expect(mocks.create).not.toHaveBeenCalled()
94+
})
95+
})
Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
import { getErrorMessage } from '@sim/utils/errors'
2+
import { authorizeCredentialUseForAuth } from '@/lib/auth/credential-access'
3+
import { AuthType } from '@/lib/auth/hybrid'
4+
import { createOciClient } from '@/lib/internal/oci/client.server'
5+
import { OciClientError } from '@/lib/internal/oci/errors'
6+
import {
7+
awaitOciStreaming,
8+
executeOciStreamingOperation,
9+
OCI_STREAMING_ADMIN_ENDPOINT,
10+
OCI_STREAMING_SERVICE_ID,
11+
withOciStreamingBudget,
12+
} from '@/lib/internal/oci-streaming/operations'
13+
import { ociStreamingInputSchema } from '@/lib/internal/oci-streaming/schema'
14+
import type { InternalToolOperationHandler } from '@/lib/internal/tool-operations/types'
15+
16+
export const executeOciStreamingTool: InternalToolOperationHandler = async (request) => {
17+
const parsed = ociStreamingInputSchema.safeParse(request.input)
18+
if (!parsed.success) {
19+
return Response.json(
20+
{ success: false, error: parsed.error.issues.map((issue) => issue.message).join('; ') },
21+
{ status: 400 }
22+
)
23+
}
24+
const input = parsed.data
25+
if (request.toolId !== `oci_streaming_${input.operation}`) {
26+
return Response.json({ success: false, error: 'OCI Streaming operation does not match tool' }, { status: 400 })
27+
}
28+
const { userId, workspaceId, workflowId } = request.context
29+
if (!userId || !workspaceId) {
30+
return Response.json({ success: false, error: 'Trusted user and workspace context are required' }, { status: 401 })
31+
}
32+
try {
33+
return await withOciStreamingBudget(async (budget) => {
34+
const access = await awaitOciStreaming(authorizeCredentialUseForAuth(
35+
{ success: true, userId, authType: AuthType.INTERNAL_JWT },
36+
{ credentialId: input.ociCredential, workspaceId, workflowId: workflowId || undefined, callerUserId: userId }
37+
), budget.signal)
38+
if (!access.ok || access.credentialType !== 'service_account' || !access.resolvedCredentialId || access.workspaceId !== workspaceId) {
39+
return Response.json({ success: false, error: 'OCI service account is not accessible in this workspace' }, { status: 403 })
40+
}
41+
budget.signal.throwIfAborted()
42+
const client = await awaitOciStreaming(createOciClient({
43+
credentialId: access.resolvedCredentialId,
44+
workspaceId,
45+
serviceId: OCI_STREAMING_SERVICE_ID,
46+
region: input.ociRegion,
47+
}), budget.signal)
48+
budget.signal.throwIfAborted()
49+
const endpoint = await awaitOciStreaming(client.prepareStaticEndpoint(OCI_STREAMING_ADMIN_ENDPOINT), budget.signal)
50+
budget.signal.throwIfAborted()
51+
const result = await executeOciStreamingOperation(input, { client, endpoint }, budget)
52+
return Response.json(result)
53+
}, request.signal)
54+
} catch (error) {
55+
if (error instanceof OciClientError) {
56+
return Response.json({
57+
success: false,
58+
error: error.message,
59+
output: { status: error.status ?? null, requestId: error.opcRequestId ?? null, code: error.code },
60+
}, { status: error.status && error.status >= 400 ? error.status : 502 })
61+
}
62+
return Response.json({ success: false, error: getErrorMessage(error, 'OCI Streaming operation failed') }, { status: 400 })
63+
}
64+
}

0 commit comments

Comments
 (0)