diff --git a/docs/contributing/index.md b/docs/contributing/index.md index 0aa16af56d..d0096ee6a9 100644 --- a/docs/contributing/index.md +++ b/docs/contributing/index.md @@ -10,6 +10,7 @@ style, tests, MCP capabilities, and runtime architecture. - [Setup](./setup.md), [code style](./code-style.md), [testing](./testing-principles.md) - [Packages and manifests](./packages-and-manifests.md) +- [External package invocation API](./package-invocation-api.md) - [Adding capabilities](./adding-capabilities.md) - [Architecture](./architecture/index.md) - [MCP server patterns](./mcp-server-patterns.md) (reference for server design) diff --git a/docs/contributing/package-invocation-api.md b/docs/contributing/package-invocation-api.md new file mode 100644 index 0000000000..2836a18c2d --- /dev/null +++ b/docs/contributing/package-invocation-api.md @@ -0,0 +1,213 @@ +# External package invocation API + +Use this API when a trusted external service should invoke a saved package +export inside Kody without going through the interactive MCP transport. + +The first intended caller is a stable Discord Gateway proxy: + +1. the external proxy owns the long-lived Discord websocket +2. the proxy normalizes gateway events +3. the proxy calls Kody's package invocation API +4. Kody executes the saved package export with package context, user context, + package storage, and normal secret/capability rules + +Kody remains the package runtime and storage brain. The external service owns +the socket lifecycle. + +## Endpoint + +`POST /api/package-invocations/:packageIdOrKodyId/:exportName` + +Examples: + +- `/api/package-invocations/pkg_123/dispatch-message-created` +- `/api/package-invocations/discord-gateway/dispatch-message-created` + +The path accepts either: + +- the saved package id +- the package `kody.id` + +The export name is normalized to package export form, so +`dispatch-message-created` resolves as `./dispatch-message-created`. + +## Authentication + +Authentication uses a private bearer token stored in Kody's database-backed +package invocation token table. + +Kody stores only the token hash for request-time lookup. The raw bearer token is +generated or collected by the token-management surface, shown to the operator +once, and then sent by the trusted external service as: + +```http +Authorization: Bearer +``` + +Each token row includes: + +- token id and human-readable name +- owning `user_id`, `email`, and `display_name` +- package scope (`package_ids_json` and/or `package_kody_ids_json`) +- allowed package exports (`export_names_json`) +- allowed request sources (`sources_json`) +- `last_used_at` +- `revoked_at` + +The token is not a global backdoor: + +- package lookup is still user-owned +- package access is scoped by the token row +- export access requires an explicit allowlist +- `source` metadata is checked against the allowlist when provided +- tokens can be revoked without deploys +- execution still uses normal package runtime machinery + +## Request body + +```json +{ + "params": { + "messageId": "123", + "content": "hello" + }, + "idempotencyKey": "discord:message-create:123", + "source": "discord-gateway", + "topic": "discord.message.created" +} +``` + +Fields: + +- `params` — JSON object passed to the package export as runtime params +- `idempotencyKey` — required stable key for replay protection +- `source` — optional source label for auditing and token scoping; when present, + it must match the token's `sources_json` allowlist +- `topic` — optional event topic label for downstream logic and logs + +## Idempotency + +Kody persists package invocation idempotency in D1. + +The identity key is: + +- user +- token id +- package id +- export name +- idempotency key + +Behavior: + +- same request + same idempotency key => stored response replayed +- same idempotency key + different payload => `409 idempotency_mismatch` +- duplicate while first invocation is still active => + `409 invocation_in_progress` + +This makes duplicate Discord gateway deliveries safe when the proxy retries. + +## Response shape + +Success: + +```json +{ + "ok": true, + "package": { + "id": "pkg_123", + "kodyId": "discord-gateway" + }, + "exportName": "./dispatch-message-created", + "source": "discord-gateway", + "topic": "discord.message.created", + "idempotency": { + "key": "discord:message-create:123", + "replayed": false + }, + "result": { + "reply": "handled" + }, + "logs": [] +} +``` + +Replay responses return the same stored body with `idempotency.replayed: true`. + +Failures return: + +```json +{ + "ok": false, + "error": { + "code": "package_not_found", + "message": "Saved package \"discord-gateway\" was not found for this user." + } +} +``` + +Execution failures return sanitized structured errors and logs. The route does +not expose Worker secrets directly. + +## Runtime behavior + +The API reuses the existing published package bundle path: + +- resolve the saved package for the configured user +- resolve the requested package export +- load the published `module` artifact if present +- rebuild and persist the module artifact on cache miss +- execute through `runBundledModuleWithRegistry` + +Execution includes: + +- package context +- repo context when a published source exists +- writable package storage bound to `package:` +- user context from the scoped token config + +## Rate limiting and auditing + +The endpoint is shaped for standard Cloudflare edge rate limiting: + +- path is stable and narrow +- caller metadata includes `source` and `topic` +- each request is audit-logged with hashed email/IP metadata + +Prefer Cloudflare WAF/rate limiting rules in front of this path rather than +adding bespoke in-Worker rate limiting first. + +## Discord gateway proxy pattern + +Recommended flow: + +1. keep the Discord Gateway websocket in a stable external process +2. receive `MESSAGE_CREATE` +3. derive a stable idempotency key from the gateway event +4. call the Kody endpoint +5. use Kody's structured result to decide what to post back to Discord + +Example request: + +```bash +curl --fail --silent \ + -X POST \ + -H "Authorization: Bearer $PACKAGE_INVOCATION_TOKEN" \ + -H "Content-Type: application/json" \ + "https://kody.example.com/api/package-invocations/discord-gateway/dispatch-message-created" \ + -d '{ + "params": { + "messageId": "123", + "channelId": "456", + "content": "hello" + }, + "idempotencyKey": "discord:message-create:123", + "source": "discord-gateway", + "topic": "discord.message.created" + }' +``` + +## Related + +- [Packages and manifests](./packages-and-manifests.md) +- [Environment variables](./environment-variables.md) +- [Setup manifest](./setup-manifest.md) diff --git a/packages/worker/.env.example b/packages/worker/.env.example index 56bf7e0923..49e04d4f39 100644 --- a/packages/worker/.env.example +++ b/packages/worker/.env.example @@ -41,3 +41,4 @@ CLOUDFLARE_EMAIL_FROM=reset@kody.dev # Authorization: Bearer repopulates Vectorize after deploys. CI can set # GitHub secret CAPABILITY_REINDEX_SECRET and sync it via deploy workflow. # CAPABILITY_REINDEX_SECRET= + diff --git a/packages/worker/migrations/0029-package-invocations.sql b/packages/worker/migrations/0029-package-invocations.sql new file mode 100644 index 0000000000..42fa67e22e --- /dev/null +++ b/packages/worker/migrations/0029-package-invocations.sql @@ -0,0 +1,51 @@ +CREATE TABLE IF NOT EXISTS package_invocation_tokens ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + name TEXT NOT NULL, + token_hash TEXT NOT NULL UNIQUE, + email TEXT NOT NULL, + display_name TEXT NOT NULL, + package_ids_json TEXT NOT NULL DEFAULT '[]', + package_kody_ids_json TEXT NOT NULL DEFAULT '[]', + export_names_json TEXT NOT NULL DEFAULT '[]', + sources_json TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + last_used_at TEXT, + revoked_at TEXT +); + +CREATE INDEX IF NOT EXISTS idx_package_invocation_tokens_user_id +ON package_invocation_tokens(user_id); + +CREATE INDEX IF NOT EXISTS idx_package_invocation_tokens_token_hash +ON package_invocation_tokens(token_hash); + +CREATE TABLE IF NOT EXISTS package_invocations ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + token_id TEXT NOT NULL, + package_id TEXT NOT NULL, + package_kody_id TEXT NOT NULL, + export_name TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + request_hash TEXT NOT NULL, + source TEXT, + topic TEXT, + status TEXT NOT NULL CHECK (status IN ('in_progress', 'completed', 'failed')), + response_json TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_package_invocations_identity +ON package_invocations( + user_id, + token_id, + package_id, + export_name, + idempotency_key +); + +CREATE INDEX IF NOT EXISTS idx_package_invocations_user_created_at +ON package_invocations(user_id, created_at); diff --git a/packages/worker/src/index.ts b/packages/worker/src/index.ts index bd27a3e638..d070f6fb66 100644 --- a/packages/worker/src/index.ts +++ b/packages/worker/src/index.ts @@ -33,6 +33,10 @@ import { handleGeneratedUiApiRequest, isGeneratedUiApiRequest, } from './mcp/generated-ui-api.ts' +import { + handlePackageInvocationApiRequest, + isPackageInvocationApiRequest, +} from './package-invocations/http.ts' import { withCors } from './utils.ts' import { handleCapabilityReindexRequest } from './capability-maintenance.ts' import { handleJobReindexRequest } from './job-maintenance.ts' @@ -263,6 +267,9 @@ function addOAuthDiscoveryCorsHeaders( const workerHandler = { fetch(request: Request, env: Env, ctx: ExecutionContext) { const url = new URL(request.url) + if (isPackageInvocationApiRequest(url.pathname)) { + return handlePackageInvocationApiRequest(request, env, ctx) + } // OAuthProvider serves this URL first and defaults `resource` to the origin only. // MCP clients must use `/mcp` as the resource (RFC 8707) to match our diff --git a/packages/worker/src/package-invocations/http.ts b/packages/worker/src/package-invocations/http.ts new file mode 100644 index 0000000000..12259c5439 --- /dev/null +++ b/packages/worker/src/package-invocations/http.ts @@ -0,0 +1,334 @@ +import { getRequestIp, logAuditEvent } from '#app/audit-log.ts' +import { getAppBaseUrl } from '#app/app-base-url.ts' +import { + invokePackageExport, + type PackageInvocationTokenScope, +} from './service.ts' +import { + getActivePackageInvocationTokenByHash, + hashPackageInvocationBearerToken, + updatePackageInvocationTokenLastUsed, +} from './repo.ts' + +const packageInvocationPathPrefix = '/api/package-invocations/' + +type PackageInvocationRequestBody = { + params?: Record + idempotencyKey?: string + source?: string | null + topic?: string | null +} + +function jsonResponse(data: unknown, init?: ResponseInit) { + const headers = new Headers(init?.headers) + headers.set('Content-Type', 'application/json') + headers.set('Cache-Control', 'no-store') + return new Response(JSON.stringify(data), { + ...init, + headers, + }) +} + +function buildAuthenticateHeader() { + return 'Bearer realm="package-invocations"' +} + +function unauthorizedResponse(message = 'Unauthorized') { + return jsonResponse( + { + ok: false, + error: { + code: 'unauthorized', + message, + }, + }, + { + status: 401, + headers: { + 'WWW-Authenticate': buildAuthenticateHeader(), + }, + }, + ) +} + +function notFoundResponse() { + return jsonResponse( + { + ok: false, + error: { + code: 'not_found', + message: 'Not found.', + }, + }, + { status: 404 }, + ) +} + +function readBearerToken(request: Request) { + const authHeader = request.headers.get('Authorization') + if (!authHeader || !authHeader.startsWith('Bearer ')) { + return null + } + const token = authHeader.slice('Bearer '.length).trim() + return token.length > 0 ? token : null +} + +function decodePathComponent(value: string) { + try { + return decodeURIComponent(value) + } catch { + return value + } +} + +function parsePackageInvocationPath(pathname: string) { + if (!pathname.startsWith(packageInvocationPathPrefix)) { + return null + } + const rest = pathname.slice(packageInvocationPathPrefix.length) + const parts = rest.split('/') + if (parts.length < 2) { + return null + } + const packageIdOrKodyId = decodePathComponent(parts[0] ?? '') + const exportName = decodePathComponent(parts.slice(1).join('/')) + if (!packageIdOrKodyId || !exportName) { + return null + } + return { packageIdOrKodyId, exportName } +} + +async function resolveTokenScope(input: { + env: Env + bearerToken: string +}): Promise { + const tokenHash = await hashPackageInvocationBearerToken(input.bearerToken) + const record = await getActivePackageInvocationTokenByHash({ + db: input.env.APP_DB, + tokenHash, + }).catch(() => null) + if (!record) return null + const touched = await updatePackageInvocationTokenLastUsed({ + db: input.env.APP_DB, + id: record.id, + }).catch(() => false) + if (!touched) return null + return { + tokenId: record.id, + userId: record.user_id, + email: record.email, + displayName: record.display_name, + packageIds: record.packageIds, + packageKodyIds: record.packageKodyIds, + exportNames: record.exportNames, + sources: record.sources, + } +} + +async function readRequestBody( + request: Request, +): Promise< + | { ok: true; body: PackageInvocationRequestBody } + | { ok: false; response: Response } +> { + let body: unknown + try { + body = (await request.json()) as unknown + } catch { + return { + ok: false, + response: jsonResponse( + { + ok: false, + error: { + code: 'invalid_json', + message: 'Request body must be valid JSON.', + }, + }, + { status: 400 }, + ), + } + } + if (!body || typeof body !== 'object' || Array.isArray(body)) { + return { + ok: false, + response: jsonResponse( + { + ok: false, + error: { + code: 'invalid_body', + message: 'Request body must be a JSON object.', + }, + }, + { status: 400 }, + ), + } + } + return { ok: true, body: body as PackageInvocationRequestBody } +} + +export function isPackageInvocationApiRequest(pathname: string) { + return pathname.startsWith(packageInvocationPathPrefix) +} + +export async function handlePackageInvocationApiRequest( + request: Request, + env: Env, + ctx?: ExecutionContext, +) { + const route = parsePackageInvocationPath(new URL(request.url).pathname) + if (!route) { + return notFoundResponse() + } + if (request.method !== 'POST') { + return jsonResponse( + { + ok: false, + error: { + code: 'method_not_allowed', + message: 'Method not allowed.', + }, + }, + { status: 405, headers: { Allow: 'POST' } }, + ) + } + const requestIp = getRequestIp(request) ?? undefined + const bearerToken = readBearerToken(request) + if (!bearerToken) { + logPackageInvocationAudit(ctx, { + category: 'oauth', + action: 'package_invoke', + result: 'failure', + ip: requestIp, + path: new URL(request.url).pathname, + reason: 'missing_bearer_token', + }) + return unauthorizedResponse() + } + const tokenScope = await resolveTokenScope({ + env, + bearerToken, + }) + if (!tokenScope) { + logPackageInvocationAudit(ctx, { + category: 'oauth', + action: 'package_invoke', + result: 'failure', + ip: requestIp, + path: new URL(request.url).pathname, + reason: 'invalid_private_token', + }) + return unauthorizedResponse('Invalid package invocation token.') + } + const parsedBody = await readRequestBody(request) + if (!parsedBody.ok) { + return parsedBody.response + } + const body = parsedBody.body + const source = typeof body.source === 'string' ? body.source : null + const topic = typeof body.topic === 'string' ? body.topic : null + if ( + body.params !== undefined && + (!body.params || + typeof body.params !== 'object' || + Array.isArray(body.params)) + ) { + return jsonResponse( + { + ok: false, + error: { + code: 'invalid_params', + message: 'params must be a JSON object when provided.', + }, + }, + { status: 400 }, + ) + } + if (typeof body.idempotencyKey !== 'string' || !body.idempotencyKey.trim()) { + return jsonResponse( + { + ok: false, + error: { + code: 'missing_idempotency_key', + message: 'Request body must include a non-empty idempotencyKey.', + }, + }, + { status: 400 }, + ) + } + if (body.source != null && typeof body.source !== 'string') { + return jsonResponse( + { + ok: false, + error: { + code: 'invalid_source', + message: 'source must be a string when provided.', + }, + }, + { status: 400 }, + ) + } + if (body.topic != null && typeof body.topic !== 'string') { + return jsonResponse( + { + ok: false, + error: { + code: 'invalid_topic', + message: 'topic must be a string when provided.', + }, + }, + { status: 400 }, + ) + } + + const response = await invokePackageExport({ + env, + baseUrl: getAppBaseUrl({ + env, + requestUrl: request.url, + }), + token: tokenScope, + request: { + packageIdOrKodyId: route.packageIdOrKodyId, + exportName: route.exportName, + params: body.params, + idempotencyKey: body.idempotencyKey, + source, + topic, + }, + }) + const result = + response.status >= 200 && response.status < 400 ? 'success' : 'failure' + const reason = + response.status >= 400 + ? String( + (response.body['error'] as Record | undefined)?.[ + 'code' + ] ?? 'request_failed', + ) + : undefined + logPackageInvocationAudit(ctx, { + category: 'oauth', + action: 'package_invoke', + result, + email: tokenScope.email, + ip: requestIp, + path: new URL(request.url).pathname, + reason, + }) + return jsonResponse(response.body, { status: response.status }) +} + +type PackageInvocationAuditEvent = Parameters[0] + +function logPackageInvocationAudit( + ctx: ExecutionContext | undefined, + event: PackageInvocationAuditEvent, +) { + const promise = logAuditEvent(event) + if (ctx) { + ctx.waitUntil(promise) + } else { + void promise + } +} diff --git a/packages/worker/src/package-invocations/http.workers.test.ts b/packages/worker/src/package-invocations/http.workers.test.ts new file mode 100644 index 0000000000..0387dd0667 --- /dev/null +++ b/packages/worker/src/package-invocations/http.workers.test.ts @@ -0,0 +1,358 @@ +import { expect, test, vi } from 'vitest' +import type * as PackageInvocationServiceModule from './service.ts' +import { + handlePackageInvocationApiRequest, + isPackageInvocationApiRequest, +} from './http.ts' +import { hashPackageInvocationBearerToken } from './repo.ts' + +const invocationMockModule = vi.hoisted(() => ({ + invokePackageExport: vi.fn(), +})) + +vi.mock('./service.ts', async () => { + const actual = + await vi.importActual('./service.ts') + return { + ...actual, + invokePackageExport: (...args: Array) => + invocationMockModule.invokePackageExport(...args), + } +}) + +async function createEnv( + options: { + tokenRow?: { + package_ids_json?: string + package_kody_ids_json?: string + export_names_json?: string + sources_json?: string + revoked_at?: string | null + } + touchChanges?: number + touchError?: Error + } = {}, +) { + const tokenRows = [ + { + id: 'discord-gateway', + user_id: 'user-123', + token_hash: await hashPackageInvocationBearerToken('private-token-123'), + name: 'Discord gateway', + email: 'me@example.com', + display_name: 'me', + package_ids_json: options.tokenRow?.package_ids_json ?? '[]', + package_kody_ids_json: + options.tokenRow?.package_kody_ids_json ?? + JSON.stringify(['discord-gateway']), + export_names_json: + options.tokenRow?.export_names_json ?? + JSON.stringify(['./dispatch-message-created']), + sources_json: + options.tokenRow?.sources_json ?? JSON.stringify(['discord-gateway']), + created_at: '2026-04-27T00:00:00.000Z', + updated_at: '2026-04-27T00:00:00.000Z', + last_used_at: null, + revoked_at: options.tokenRow?.revoked_at ?? null, + }, + ] + return { + APP_DB: { + prepare(query: string) { + return { + bind(...params: Array) { + return { + async first>() { + if (query.includes('FROM package_invocation_tokens')) { + const tokenHash = String(params[0] ?? '') + return (tokenRows.find( + (row) => + row.token_hash === tokenHash && row.revoked_at === null, + ) ?? null) as T | null + } + return null + }, + async run() { + if (query.includes('UPDATE package_invocation_tokens')) { + if (options.touchError) { + throw options.touchError + } + const id = String(params[1] ?? '') + const row = tokenRows.find( + (entry) => entry.id === id && entry.revoked_at === null, + ) + if (!row) { + return { meta: { changes: 0, last_row_id: 0 } } + } + if (options.touchChanges !== undefined) { + return { + meta: { + changes: options.touchChanges, + last_row_id: 0, + }, + } + } + row.last_used_at = String(params[0]) + return { meta: { changes: 1, last_row_id: 0 } } + } + return { meta: { changes: 0, last_row_id: 0 } } + }, + } + }, + } + }, + } as unknown as D1Database, + BUNDLE_ARTIFACTS_KV: { + get: async () => null, + put: async () => undefined, + delete: async () => undefined, + }, + COOKIE_SECRET: 'test-cookie-secret-0123456789abcdef0123456789', + JOB_MANAGER: { + idFromName: () => ({ toString: () => 'job-manager-id' }), + get: () => ({}) as DurableObjectStub, + } as DurableObjectNamespace, + STORAGE_RUNNER: { + idFromName: () => ({ toString: () => 'storage-runner-id' }), + get: () => ({}) as DurableObjectStub, + } as DurableObjectNamespace, + PACKAGE_REALTIME_SESSION: { + idFromName: () => ({ toString: () => 'package-realtime-id' }), + get: () => ({}) as DurableObjectStub, + } as DurableObjectNamespace, + PACKAGE_SERVICE_INSTANCE: { + idFromName: () => ({ toString: () => 'package-service-id' }), + get: () => ({}) as DurableObjectStub, + } as DurableObjectNamespace, + } as unknown as Env +} + +function createContext() { + return { + waitUntil: vi.fn(), + passThroughOnException: vi.fn(), + } as unknown as ExecutionContext +} + +test('isPackageInvocationApiRequest matches the external package invocation route', () => { + expect( + isPackageInvocationApiRequest( + '/api/package-invocations/discord-gateway/dispatch-message-created', + ), + ).toBe(true) + expect(isPackageInvocationApiRequest('/api/me')).toBe(false) +}) + +test('package invocation API returns 401 when bearer token is missing', async () => { + const response = await handlePackageInvocationApiRequest( + new Request( + 'https://example.com/api/package-invocations/discord-gateway/dispatch-message-created', + { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ idempotencyKey: 'evt-1' }), + }, + ), + await createEnv(), + createContext(), + ) + + expect(response.status).toBe(401) + expect(response.headers.get('WWW-Authenticate')).toBe( + 'Bearer realm="package-invocations"', + ) + await expect(response.json()).resolves.toEqual({ + ok: false, + error: { + code: 'unauthorized', + message: 'Unauthorized', + }, + }) +}) + +test('package invocation API returns 401 for invalid private tokens', async () => { + const response = await handlePackageInvocationApiRequest( + new Request( + 'https://example.com/api/package-invocations/discord-gateway/dispatch-message-created', + { + method: 'POST', + headers: { + Authorization: 'Bearer wrong-token', + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ idempotencyKey: 'evt-1' }), + }, + ), + await createEnv(), + createContext(), + ) + + expect(response.status).toBe(401) + await expect(response.json()).resolves.toEqual({ + ok: false, + error: { + code: 'unauthorized', + message: 'Invalid package invocation token.', + }, + }) +}) + +test('package invocation API fails closed when token touch loses revocation race', async () => { + const response = await handlePackageInvocationApiRequest( + new Request( + 'https://example.com/api/package-invocations/discord-gateway/dispatch-message-created', + { + method: 'POST', + headers: { + Authorization: 'Bearer private-token-123', + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ idempotencyKey: 'evt-1' }), + }, + ), + await createEnv({ touchChanges: 0 }), + createContext(), + ) + + expect(response.status).toBe(401) + await expect(response.json()).resolves.toEqual({ + ok: false, + error: { + code: 'unauthorized', + message: 'Invalid package invocation token.', + }, + }) + expect(invocationMockModule.invokePackageExport).not.toHaveBeenCalled() +}) + +test('package invocation API fails closed when token scope JSON is malformed', async () => { + const response = await handlePackageInvocationApiRequest( + new Request( + 'https://example.com/api/package-invocations/discord-gateway/dispatch-message-created', + { + method: 'POST', + headers: { + Authorization: 'Bearer private-token-123', + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ idempotencyKey: 'evt-1' }), + }, + ), + await createEnv({ + tokenRow: { + package_kody_ids_json: '{bad json', + }, + }), + createContext(), + ) + + expect(response.status).toBe(401) + await expect(response.json()).resolves.toEqual({ + ok: false, + error: { + code: 'unauthorized', + message: 'Invalid package invocation token.', + }, + }) + expect(invocationMockModule.invokePackageExport).not.toHaveBeenCalled() +}) + +test('package invocation API validates the JSON body shape', async () => { + const response = await handlePackageInvocationApiRequest( + new Request( + 'https://example.com/api/package-invocations/discord-gateway/dispatch-message-created', + { + method: 'POST', + headers: { + Authorization: 'Bearer private-token-123', + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ params: [] }), + }, + ), + await createEnv(), + createContext(), + ) + + expect(response.status).toBe(400) + await expect(response.json()).resolves.toEqual({ + ok: false, + error: { + code: 'invalid_params', + message: 'params must be a JSON object when provided.', + }, + }) +}) + +test('package invocation API invokes the package export with the scoped token context', async () => { + invocationMockModule.invokePackageExport.mockResolvedValue({ + status: 200, + body: { + ok: true, + exportName: './dispatch-message-created', + idempotency: { + key: 'evt-1', + replayed: false, + }, + result: { reply: 'hello discord' }, + logs: ['ran'], + }, + }) + + const ctx = createContext() + const response = await handlePackageInvocationApiRequest( + new Request( + 'https://example.com/api/package-invocations/discord-gateway/dispatch-message-created', + { + method: 'POST', + headers: { + Authorization: 'Bearer private-token-123', + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + params: { content: 'hi' }, + idempotencyKey: 'evt-1', + source: 'discord-gateway', + topic: 'discord.message.created', + }), + }, + ), + await createEnv(), + ctx, + ) + + expect(invocationMockModule.invokePackageExport).toHaveBeenCalledWith({ + env: expect.any(Object), + baseUrl: 'https://example.com', + token: { + tokenId: 'discord-gateway', + userId: 'user-123', + email: 'me@example.com', + displayName: 'me', + packageIds: [], + packageKodyIds: ['discord-gateway'], + exportNames: ['./dispatch-message-created'], + sources: ['discord-gateway'], + }, + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-1', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + expect(response.status).toBe(200) + expect(ctx.waitUntil).toHaveBeenCalled() + await expect(response.json()).resolves.toEqual({ + ok: true, + exportName: './dispatch-message-created', + idempotency: { + key: 'evt-1', + replayed: false, + }, + result: { reply: 'hello discord' }, + logs: ['ran'], + }) +}) diff --git a/packages/worker/src/package-invocations/repo.ts b/packages/worker/src/package-invocations/repo.ts new file mode 100644 index 0000000000..7dd2dc3471 --- /dev/null +++ b/packages/worker/src/package-invocations/repo.ts @@ -0,0 +1,317 @@ +import { + nullable, + object, + parseSafe, + string, + type InferOutput, +} from 'remix/data-schema' +import { toHex } from '@kody-internal/shared/hex.ts' + +const packageInvocationRowSchema = object({ + id: string(), + user_id: string(), + token_id: string(), + package_id: string(), + package_kody_id: string(), + export_name: string(), + idempotency_key: string(), + request_hash: string(), + source: nullable(string()), + topic: nullable(string()), + status: string(), + response_json: nullable(string()), + created_at: string(), + updated_at: string(), +}) + +const packageInvocationTokenRowSchema = object({ + id: string(), + user_id: string(), + token_hash: string(), + name: string(), + email: string(), + display_name: string(), + package_ids_json: string(), + package_kody_ids_json: string(), + export_names_json: string(), + sources_json: string(), + created_at: string(), + updated_at: string(), + last_used_at: nullable(string()), + revoked_at: nullable(string()), +}) + +export type PackageInvocationStoredResponse = { + status: number + body: Record +} + +export type PackageInvocationRecord = InferOutput< + typeof packageInvocationRowSchema +> & { + storedResponse: PackageInvocationStoredResponse | null +} + +export type PackageInvocationTokenRecord = InferOutput< + typeof packageInvocationTokenRowSchema +> & { + packageIds: Array + packageKodyIds: Array + exportNames: Array + sources: Array +} + +export async function hashPackageInvocationBearerToken(token: string) { + const digest = await crypto.subtle.digest( + 'SHA-256', + new TextEncoder().encode(token), + ) + return toHex(new Uint8Array(digest)) +} + +function parseStringArrayJson(input: { value: string; field: string }) { + let parsed: unknown + try { + parsed = JSON.parse(input.value) as unknown + } catch { + throw new Error( + `Invalid package invocation token record: ${input.field} must be valid JSON.`, + ) + } + if (!Array.isArray(parsed)) { + throw new Error( + `Invalid package invocation token record: ${input.field} must be a JSON array.`, + ) + } + return parsed.map((entry) => { + if (typeof entry !== 'string' || !entry.trim()) { + throw new Error( + `Invalid package invocation token record: ${input.field} must contain only non-empty strings.`, + ) + } + return entry.trim() + }) +} + +function parseStoredResponse( + value: string | null, +): PackageInvocationStoredResponse | null { + if (!value) return null + try { + const parsed = JSON.parse(value) as unknown + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + return null + } + const record = parsed as Record + const status = record['status'] + const body = record['body'] + if ( + typeof status !== 'number' || + !Number.isInteger(status) || + !body || + typeof body !== 'object' || + Array.isArray(body) + ) { + return null + } + return { + status, + body: body as Record, + } + } catch { + return null + } +} + +function mapTokenRow( + row: Record, +): PackageInvocationTokenRecord { + const parsed = parseSafe(packageInvocationTokenRowSchema, row) + if (!parsed.success) { + const message = parsed.issues.map((issue) => issue.message).join(', ') + throw new Error(`Invalid package invocation token record: ${message}`) + } + return { + ...parsed.value, + packageIds: parseStringArrayJson({ + value: parsed.value.package_ids_json, + field: 'package_ids_json', + }), + packageKodyIds: parseStringArrayJson({ + value: parsed.value.package_kody_ids_json, + field: 'package_kody_ids_json', + }), + exportNames: parseStringArrayJson({ + value: parsed.value.export_names_json, + field: 'export_names_json', + }), + sources: parseStringArrayJson({ + value: parsed.value.sources_json, + field: 'sources_json', + }), + } +} + +function mapRow(row: Record): PackageInvocationRecord { + const parsed = parseSafe(packageInvocationRowSchema, row) + if (!parsed.success) { + const message = parsed.issues.map((issue) => issue.message).join(', ') + throw new Error(`Invalid package invocation record: ${message}`) + } + return { + ...parsed.value, + storedResponse: parseStoredResponse(parsed.value.response_json), + } +} + +export async function getActivePackageInvocationTokenByHash(input: { + db: D1Database + tokenHash: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM package_invocation_tokens + WHERE token_hash = ? + AND revoked_at IS NULL + LIMIT 1`, + ) + .bind(input.tokenHash) + .first>() + return row ? mapTokenRow(row) : null +} + +export async function updatePackageInvocationTokenLastUsed(input: { + db: D1Database + id: string +}) { + const result = await input.db + .prepare( + `UPDATE package_invocation_tokens + SET last_used_at = ? + WHERE id = ? AND revoked_at IS NULL`, + ) + .bind(new Date().toISOString(), input.id) + .run() + return (result.meta.changes ?? 0) > 0 +} + +export async function insertPackageInvocationRow(input: { + db: D1Database + row: { + id: string + userId: string + tokenId: string + packageId: string + packageKodyId: string + exportName: string + idempotencyKey: string + requestHash: string + source?: string | null + topic?: string | null + status: 'in_progress' | 'completed' | 'failed' + response?: PackageInvocationStoredResponse | null + } +}) { + const now = new Date().toISOString() + const responseJson = input.row.response + ? JSON.stringify({ + status: input.row.response.status, + body: input.row.response.body, + }) + : null + const result = await input.db + .prepare( + `INSERT OR IGNORE INTO package_invocations ( + id, + user_id, + token_id, + package_id, + package_kody_id, + export_name, + idempotency_key, + request_hash, + source, + topic, + status, + response_json, + created_at, + updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + input.row.id, + input.row.userId, + input.row.tokenId, + input.row.packageId, + input.row.packageKodyId, + input.row.exportName, + input.row.idempotencyKey, + input.row.requestHash, + input.row.source ?? null, + input.row.topic ?? null, + input.row.status, + responseJson, + now, + now, + ) + .run() + return (result.meta.changes ?? 0) > 0 +} + +export async function getPackageInvocationByKey(input: { + db: D1Database + userId: string + tokenId: string + packageId: string + exportName: string + idempotencyKey: string +}) { + const row = await input.db + .prepare( + `SELECT * + FROM package_invocations + WHERE user_id = ? + AND token_id = ? + AND package_id = ? + AND export_name = ? + AND idempotency_key = ?`, + ) + .bind( + input.userId, + input.tokenId, + input.packageId, + input.exportName, + input.idempotencyKey, + ) + .first>() + return row ? mapRow(row) : null +} + +export async function updatePackageInvocationResult(input: { + db: D1Database + id: string + userId: string + status: 'completed' | 'failed' + response: PackageInvocationStoredResponse +}) { + const responseJson = JSON.stringify({ + status: input.response.status, + body: input.response.body, + }) + const result = await input.db + .prepare( + `UPDATE package_invocations + SET status = ?, response_json = ?, updated_at = ? + WHERE id = ? AND user_id = ?`, + ) + .bind( + input.status, + responseJson, + new Date().toISOString(), + input.id, + input.userId, + ) + .run() + return (result.meta.changes ?? 0) > 0 +} diff --git a/packages/worker/src/package-invocations/service.node.test.ts b/packages/worker/src/package-invocations/service.node.test.ts new file mode 100644 index 0000000000..68a881dbf2 --- /dev/null +++ b/packages/worker/src/package-invocations/service.node.test.ts @@ -0,0 +1,626 @@ +import { expect, test, vi } from 'vitest' +import { invokePackageExport } from './service.ts' + +const repoMockModule = vi.hoisted(() => ({ + getSavedPackageById: vi.fn(), + getSavedPackageByKodyId: vi.fn(), + loadPackageSourceBySourceId: vi.fn(), + getEntitySourceById: vi.fn(), + loadPublishedBundleArtifactByIdentity: vi.fn(), + persistPublishedBundleArtifact: vi.fn(), + typecheckPackageEntrypointsFromSourceFiles: vi.fn(), + runBundledModuleWithRegistry: vi.fn(), +})) + +vi.mock('#worker/package-registry/repo.ts', () => ({ + getSavedPackageById: (...args: Array) => + repoMockModule.getSavedPackageById(...args), + getSavedPackageByKodyId: (...args: Array) => + repoMockModule.getSavedPackageByKodyId(...args), +})) + +vi.mock('#worker/package-registry/source.ts', () => ({ + loadPackageSourceBySourceId: (...args: Array) => + repoMockModule.loadPackageSourceBySourceId(...args), +})) + +vi.mock('#worker/repo/entity-sources.ts', () => ({ + getEntitySourceById: (...args: Array) => + repoMockModule.getEntitySourceById(...args), +})) + +vi.mock('#worker/package-runtime/published-bundle-artifacts.ts', () => ({ + loadPublishedBundleArtifactByIdentity: (...args: Array) => + repoMockModule.loadPublishedBundleArtifactByIdentity(...args), + persistPublishedBundleArtifact: (...args: Array) => + repoMockModule.persistPublishedBundleArtifact(...args), +})) + +vi.mock('#worker/repo/checks.ts', () => ({ + typecheckPackageEntrypointsFromSourceFiles: (...args: Array) => + repoMockModule.typecheckPackageEntrypointsFromSourceFiles(...args), +})) + +vi.mock('#mcp/run-codemode-registry.ts', () => ({ + runBundledModuleWithRegistry: (...args: Array) => + repoMockModule.runBundledModuleWithRegistry(...args), +})) + +function createDatabase(options: { failInsert?: boolean } = {}) { + const tables = new Map>>([ + ['package_invocations', []], + ]) + + const clone = (value: T): T => structuredClone(value) + + function getTable(name: string) { + const table = tables.get(name) + if (!table) { + throw new Error(`Unknown table ${name}`) + } + return table + } + + function selectOne( + tableName: string, + predicate: (row: Record) => boolean, + ) { + return clone(getTable(tableName).find(predicate) ?? null) + } + + const db = { + prepare(query: string) { + return { + bind(...params: Array) { + return { + async first>() { + if ( + query.includes('FROM package_invocations') && + query.includes('idempotency_key = ?') + ) { + return selectOne( + 'package_invocations', + (row) => + row['user_id'] === params[0] && + row['token_id'] === params[1] && + row['package_id'] === params[2] && + row['export_name'] === params[3] && + row['idempotency_key'] === params[4], + ) as T | null + } + return null + }, + async run() { + if (query.includes('INTO package_invocations')) { + if (options.failInsert) { + throw new Error('D1 unavailable') + } + const table = getTable('package_invocations') + const existing = table.find( + (row) => + row['user_id'] === params[1] && + row['token_id'] === params[2] && + row['package_id'] === params[3] && + row['export_name'] === params[5] && + row['idempotency_key'] === params[6], + ) + if (existing) { + return { meta: { changes: 0, last_row_id: 0 } } + } + table.push( + clone({ + id: params[0], + user_id: params[1], + token_id: params[2], + package_id: params[3], + package_kody_id: params[4], + export_name: params[5], + idempotency_key: params[6], + request_hash: params[7], + source: params[8], + topic: params[9], + status: params[10], + response_json: params[11], + created_at: params[12], + updated_at: params[13], + }), + ) + return { meta: { changes: 1, last_row_id: 1 } } + } + if (query.includes('UPDATE package_invocations')) { + const table = getTable('package_invocations') + const existing = table.find( + (row) => + row['id'] === params[3] && row['user_id'] === params[4], + ) + if (!existing) { + return { meta: { changes: 0, last_row_id: 0 } } + } + existing['status'] = params[0] + existing['response_json'] = params[1] + existing['updated_at'] = params[2] + return { meta: { changes: 1, last_row_id: 0 } } + } + throw new Error(`Unhandled query: ${query}`) + }, + } + }, + } + }, + corruptStoredResponses() { + for (const row of getTable('package_invocations')) { + row['response_json'] = '{"status":200,"body":null}' + } + }, + } as unknown as D1Database & { corruptStoredResponses(): void } + return db +} + +function createEnv(db: D1Database) { + return { + APP_DB: db, + BUNDLE_ARTIFACTS_KV: { + get: async () => null, + put: async () => undefined, + delete: async () => undefined, + }, + } as unknown as Env +} + +function createToken( + overrides: Partial<{ + exportNames: Array + sources: Array + }> = {}, +) { + return { + tokenId: 'discord-gateway', + userId: 'user-123', + email: 'me@example.com', + displayName: 'me', + packageKodyIds: ['discord-gateway'], + exportNames: overrides.exportNames ?? ['./dispatch-message-created'], + sources: overrides.sources ?? ['discord-gateway'], + } as const +} + +function seedPackageResolution() { + repoMockModule.getSavedPackageById.mockResolvedValue(null) + repoMockModule.getSavedPackageByKodyId.mockResolvedValue({ + id: 'pkg-1', + userId: 'user-123', + name: '@kentcdodds/discord-gateway', + kodyId: 'discord-gateway', + description: 'Discord gateway helpers', + tags: [], + searchText: null, + sourceId: 'source-1', + hasApp: true, + createdAt: '2026-04-27T00:00:00.000Z', + updatedAt: '2026-04-27T00:00:00.000Z', + }) + repoMockModule.loadPackageSourceBySourceId.mockResolvedValue({ + source: { + id: 'source-1', + user_id: 'user-123', + entity_kind: 'package', + entity_id: 'pkg-1', + repo_id: 'repo-1', + published_commit: 'commit-1', + indexed_commit: null, + manifest_path: 'package.json', + source_root: '/', + created_at: '2026-04-27T00:00:00.000Z', + updated_at: '2026-04-27T00:00:00.000Z', + }, + manifest: { + name: '@kentcdodds/discord-gateway', + exports: { + './dispatch-message-created': './src/dispatch-message-created.ts', + }, + kody: { + id: 'discord-gateway', + description: 'Discord gateway helpers', + app: { + entry: './src/app.ts', + }, + }, + }, + files: { + 'package.json': JSON.stringify({ + name: '@kentcdodds/discord-gateway', + exports: { + './dispatch-message-created': './src/dispatch-message-created.ts', + }, + kody: { + id: 'discord-gateway', + description: 'Discord gateway helpers', + app: { + entry: './src/app.ts', + }, + }, + }), + 'src/dispatch-message-created.ts': + 'export default async function run(){ return { ok: true } }', + }, + }) + repoMockModule.getEntitySourceById.mockResolvedValue({ + id: 'source-1', + user_id: 'user-123', + entity_kind: 'package', + entity_id: 'pkg-1', + repo_id: 'repo-1', + published_commit: 'commit-1', + indexed_commit: null, + manifest_path: 'package.json', + source_root: '/', + created_at: '2026-04-27T00:00:00.000Z', + updated_at: '2026-04-27T00:00:00.000Z', + }) + repoMockModule.loadPublishedBundleArtifactByIdentity.mockResolvedValue({ + row: { + id: 'artifact-1', + }, + artifact: { + version: 1, + kind: 'module', + artifactName: './dispatch-message-created', + sourceId: 'source-1', + publishedCommit: 'commit-1', + entryPoint: 'src/dispatch-message-created.ts', + mainModule: 'dist/index.js', + modules: { + 'dist/index.js': + 'export default async function run(){ return { ok: true } }', + }, + dependencies: [], + packageContext: { + packageId: 'pkg-1', + kodyId: 'discord-gateway', + sourceId: 'source-1', + }, + serviceContext: null, + createdAt: '2026-04-27T00:00:00.000Z', + }, + }) + repoMockModule.typecheckPackageEntrypointsFromSourceFiles.mockResolvedValue({ + ok: true, + message: 'ok', + }) + repoMockModule.persistPublishedBundleArtifact.mockResolvedValue('kv:key') +} + +test('invokePackageExport executes a scoped package export successfully', async () => { + const db = createDatabase() + seedPackageResolution() + repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ + result: { reply: 'hello discord' }, + logs: ['dispatched'], + }) + + const response = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-1', + topic: 'discord.message.created', + }, + }) + + expect(response.status).toBe(200) + expect(response.body).toMatchObject({ + ok: true, + exportName: './dispatch-message-created', + idempotency: { + key: 'evt-1', + replayed: false, + }, + result: { reply: 'hello discord' }, + logs: ['dispatched'], + }) + expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) +}) + +test('invokePackageExport replays the stored response for a duplicate idempotency key', async () => { + const db = createDatabase() + seedPackageResolution() + repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ + result: { reply: 'hello discord' }, + logs: ['dispatched'], + }) + + const first = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-1', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + const second = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-1', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + + expect(first.status).toBe(200) + expect(second.status).toBe(200) + expect(second.body).toMatchObject({ + ok: true, + idempotency: { + key: 'evt-1', + replayed: true, + }, + }) + expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) +}) + +test('invokePackageExport does not re-execute terminal rows with unusable stored responses', async () => { + const db = createDatabase() + seedPackageResolution() + repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ + result: { reply: 'hello discord' }, + logs: ['dispatched'], + }) + + const first = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-corrupt', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + ;( + db as unknown as { corruptStoredResponses(): void } + ).corruptStoredResponses() + const second = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-corrupt', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + + expect(first.status).toBe(200) + expect(second.status).toBe(409) + expect(second.body).toMatchObject({ + ok: false, + error: { + code: 'idempotency_response_unavailable', + }, + idempotency: { + key: 'evt-corrupt', + replayed: false, + }, + }) + expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) +}) + +test('invokePackageExport returns structured errors when idempotency insert fails', async () => { + const db = createDatabase({ failInsert: true }) + seedPackageResolution() + + const response = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-insert-failure', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + + expect(response.status).toBe(500) + expect(response.body).toMatchObject({ + ok: false, + error: { + code: 'idempotency_persistence_failed', + }, + idempotency: { + key: 'evt-insert-failure', + replayed: false, + }, + }) + expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() +}) + +test('invokePackageExport rejects explicitly disallowed source values', async () => { + const db = createDatabase() + seedPackageResolution() + + const response = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-bad-source', + source: 'other-gateway', + }, + }) + + expect(response.status).toBe(403) + expect(response.body).toMatchObject({ + ok: false, + error: { + code: 'source_not_allowed', + }, + }) + expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() +}) + +test('invokePackageExport rejects reusing an idempotency key for a different payload', async () => { + const db = createDatabase() + seedPackageResolution() + repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ + result: { reply: 'hello discord' }, + logs: ['dispatched'], + }) + + await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-1', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + const mismatch = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'different' }, + idempotencyKey: 'evt-1', + source: 'discord-gateway', + topic: 'discord.message.created', + }, + }) + + expect(mismatch.status).toBe(409) + expect(mismatch.body).toEqual({ + ok: false, + error: { + code: 'idempotency_mismatch', + message: + 'This idempotency key has already been used for a different package invocation request.', + }, + idempotency: { + key: 'evt-1', + replayed: false, + }, + }) + expect(repoMockModule.runBundledModuleWithRegistry).toHaveBeenCalledTimes(1) +}) + +test('invokePackageExport serializes execution failures without exposing thrown objects', async () => { + const db = createDatabase() + seedPackageResolution() + repoMockModule.runBundledModuleWithRegistry.mockResolvedValue({ + error: new Error('Discord downstream failed'), + logs: ['before-error'], + }) + + const response = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken(), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'dispatch-message-created', + params: { content: 'hi' }, + idempotencyKey: 'evt-2', + source: 'discord-gateway', + }, + }) + + expect(response.status).toBe(500) + expect(response.body).toMatchObject({ + ok: false, + error: { + code: 'execution_failed', + message: 'Discord downstream failed', + }, + logs: ['before-error'], + }) +}) + +test('invokePackageExport stores export-not-found responses as terminal failures', async () => { + const db = createDatabase() + seedPackageResolution() + + const first = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken({ + exportNames: ['./missing-export'], + }), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'missing-export', + params: { content: 'hi' }, + idempotencyKey: 'evt-missing-export', + source: 'discord-gateway', + }, + }) + const second = await invokePackageExport({ + env: createEnv(db), + baseUrl: 'https://kody.dev', + token: createToken({ + exportNames: ['./missing-export'], + }), + request: { + packageIdOrKodyId: 'discord-gateway', + exportName: 'missing-export', + params: { content: 'hi' }, + idempotencyKey: 'evt-missing-export', + source: 'discord-gateway', + }, + }) + + expect(first.status).toBe(404) + expect(first.body).toMatchObject({ + ok: false, + error: { + code: 'export_not_found', + }, + idempotency: { + key: 'evt-missing-export', + replayed: false, + }, + }) + expect(second.status).toBe(404) + expect(second.body).toMatchObject({ + ok: false, + error: { + code: 'export_not_found', + }, + idempotency: { + key: 'evt-missing-export', + replayed: true, + }, + }) + expect(repoMockModule.runBundledModuleWithRegistry).not.toHaveBeenCalled() +}) diff --git a/packages/worker/src/package-invocations/service.ts b/packages/worker/src/package-invocations/service.ts new file mode 100644 index 0000000000..c96d325051 --- /dev/null +++ b/packages/worker/src/package-invocations/service.ts @@ -0,0 +1,701 @@ +import { type ContentBlock } from '@modelcontextprotocol/sdk/types.js' +import { toHex } from '@kody-internal/shared/hex.ts' +import { extractRawContent, getExecutionErrorDetails } from '#mcp/executor.ts' +import { createMcpCallerContext } from '#mcp/context.ts' +import { runBundledModuleWithRegistry } from '#mcp/run-codemode-registry.ts' +import { + getSavedPackageById, + getSavedPackageByKodyId, +} from '#worker/package-registry/repo.ts' +import { loadPackageSourceBySourceId } from '#worker/package-registry/source.ts' +import { type SavedPackageRecord } from '#worker/package-registry/types.ts' +import { resolvePackageExportPath } from '#worker/package-registry/manifest.ts' +import { typecheckPackageEntrypointsFromSourceFiles } from '#worker/repo/checks.ts' +import { + loadPublishedBundleArtifactByIdentity, + persistPublishedBundleArtifact, +} from '#worker/package-runtime/published-bundle-artifacts.ts' +import { getEntitySourceById } from '#worker/repo/entity-sources.ts' +import { type EntitySourceRow } from '#worker/repo/types.ts' +import { + getPackageInvocationByKey, + insertPackageInvocationRow, + updatePackageInvocationResult, + type PackageInvocationStoredResponse, +} from './repo.ts' + +export type PackageInvocationTokenScope = { + tokenId: string + userId: string + email: string + displayName: string + packageIds?: Array + packageKodyIds?: Array + exportNames?: Array + sources?: Array +} + +export type PackageInvocationRequest = { + packageIdOrKodyId: string + exportName: string + params?: Record + idempotencyKey: string + source?: string | null + topic?: string | null +} + +export type PackageInvocationResponse = PackageInvocationStoredResponse + +function normalizeExportName(exportName: string) { + const trimmed = exportName.trim() + if (!trimmed) { + throw new Error('Package export name must not be empty.') + } + if (trimmed === '.' || trimmed === './') { + return '.' + } + return trimmed.startsWith('./') ? trimmed : `./${trimmed}` +} + +function normalizeNullableString(value: string | null | undefined) { + const trimmed = value?.trim() + return trimmed && trimmed.length > 0 ? trimmed : null +} + +function buildPackageInvocationStorageId(packageId: string) { + return `package:${encodeURIComponent(packageId)}` +} + +function createRepoContext(source: EntitySourceRow) { + return { + sourceId: source.id, + repoId: source.repo_id, + sessionId: null, + sessionRepoId: null, + baseCommit: source.published_commit, + manifestPath: source.manifest_path, + sourceRoot: source.source_root, + publishedCommit: source.published_commit, + entityKind: source.entity_kind, + entityId: source.entity_id, + } +} + +function toJsonSafeValue(value: unknown): unknown { + try { + return JSON.parse(JSON.stringify(value)) as unknown + } catch { + return value instanceof Error ? value.message : String(value) + } +} + +function markStoredResponseAsReplayed( + response: PackageInvocationStoredResponse, +) { + const body = structuredClone(response.body) + const record = body as Record + const existingIdempotency = record['idempotency'] + if (existingIdempotency && typeof existingIdempotency === 'object') { + record['idempotency'] = { + ...(existingIdempotency as Record), + replayed: true, + } + } else { + record['idempotency'] = { replayed: true } + } + return { + status: response.status, + body, + } satisfies PackageInvocationStoredResponse +} + +async function createRequestHash(input: { + packageId: string + exportName: string + params?: Record + source: string | null + topic: string | null +}) { + const canonical = canonicalJsonStringify(input) + const digest = await crypto.subtle.digest( + 'SHA-256', + new TextEncoder().encode(canonical), + ) + return toHex(new Uint8Array(digest)) +} + +function canonicalJsonStringify(value: unknown): string { + return JSON.stringify(canonicalizeJsonValue(value)) +} + +function canonicalizeJsonValue(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map((entry) => canonicalizeJsonValue(entry)) + } + if (value && typeof value === 'object') { + const record = value as Record + return Object.fromEntries( + Object.keys(record) + .sort((left, right) => left.localeCompare(right)) + .map((key) => [key, canonicalizeJsonValue(record[key])]), + ) + } + return value +} + +async function resolveSavedPackage(input: { + db: D1Database + userId: string + packageIdOrKodyId: string +}): Promise { + return ( + (await getSavedPackageById(input.db, { + userId: input.userId, + packageId: input.packageIdOrKodyId, + })) ?? + (await getSavedPackageByKodyId(input.db, { + userId: input.userId, + kodyId: input.packageIdOrKodyId, + })) + ) +} + +function tokenAllowsPackage(input: { + token: PackageInvocationTokenScope + savedPackage: SavedPackageRecord +}) { + const allowsPackageId = + input.token.packageIds?.includes(input.savedPackage.id) ?? false + const allowsKodyId = + input.token.packageKodyIds?.includes(input.savedPackage.kodyId) ?? false + return allowsPackageId || allowsKodyId +} + +function tokenAllowsExport(input: { + token: PackageInvocationTokenScope + exportName: string +}) { + const exportNames = input.token.exportNames ?? [] + return exportNames + .map((entry) => normalizeExportName(entry)) + .includes(input.exportName) +} + +function tokenAllowsSource(input: { + token: PackageInvocationTokenScope + source: string | null +}) { + if (!input.source) return true + const sources = input.token.sources ?? [] + return sources.includes(input.source) +} + +function isMissingPackageExportError(error: unknown) { + return ( + error instanceof Error && + (error.message.includes('does not define export') || + error.message.includes('does not define a runtime target')) + ) +} + +async function ensureModuleArtifact(input: { + env: Env + baseUrl: string + savedPackage: SavedPackageRecord + exportName: string + userId: string +}) { + const packageSource = await loadPackageSourceBySourceId({ + env: input.env, + baseUrl: input.baseUrl, + userId: input.userId, + sourceId: input.savedPackage.sourceId, + }) + const entryPoint = resolvePackageExportPath({ + manifest: packageSource.manifest, + exportName: input.exportName, + }) + const loaded = await loadPublishedBundleArtifactByIdentity({ + env: input.env, + userId: input.userId, + sourceId: input.savedPackage.sourceId, + kind: 'module', + artifactName: input.exportName, + entryPoint, + }) + if (loaded?.artifact) { + return { + artifact: loaded.artifact, + source: packageSource.source, + entryPoint, + } + } + const typecheckResult = await typecheckPackageEntrypointsFromSourceFiles({ + sourceFiles: packageSource.files, + entryPoints: [{ path: entryPoint, includeStorage: true }], + }) + if (!typecheckResult.ok) { + throw new Error(typecheckResult.message) + } + const { buildKodyModuleBundle } = + await import('#worker/package-runtime/module-graph.ts') + const bundle = await buildKodyModuleBundle({ + env: input.env, + baseUrl: input.baseUrl, + userId: input.userId, + sourceFiles: packageSource.files, + entryPoint, + }) + await persistPublishedBundleArtifact({ + env: input.env, + userId: input.userId, + source: packageSource.source, + kind: 'module', + artifactName: input.exportName, + entryPoint, + mainModule: bundle.mainModule, + modules: bundle.modules, + dependencies: bundle.dependencies, + packageContext: { + packageId: input.savedPackage.id, + kodyId: input.savedPackage.kodyId, + sourceId: input.savedPackage.sourceId, + }, + }) + const rebuilt = await loadPublishedBundleArtifactByIdentity({ + env: input.env, + userId: input.userId, + sourceId: input.savedPackage.sourceId, + kind: 'module', + artifactName: input.exportName, + entryPoint, + }) + if (!rebuilt?.artifact) { + throw new Error( + `Published bundle artifact for export "${input.exportName}" could not be loaded after rebuild.`, + ) + } + return { + artifact: rebuilt.artifact, + source: packageSource.source, + entryPoint, + } +} + +function buildExecutionSuccessResponse(input: { + savedPackage: SavedPackageRecord + exportName: string + idempotencyKey: string + source: string | null + topic: string | null + result: unknown + logs: Array + rawContent: Array | null +}): PackageInvocationStoredResponse { + return { + status: 200, + body: { + ok: true, + package: { + id: input.savedPackage.id, + kodyId: input.savedPackage.kodyId, + }, + exportName: input.exportName, + source: input.source, + topic: input.topic, + idempotency: { + key: input.idempotencyKey, + replayed: false, + }, + result: toJsonSafeValue(input.result), + logs: input.logs, + ...(input.rawContent + ? { rawContent: toJsonSafeValue(input.rawContent) } + : {}), + }, + } +} + +function buildExecutionErrorResponse(input: { + savedPackage: SavedPackageRecord + exportName: string + idempotencyKey: string + source: string | null + topic: string | null + error: unknown + logs: Array +}): PackageInvocationStoredResponse { + const message = + input.error instanceof Error ? input.error.message : String(input.error) + return { + status: 500, + body: { + ok: false, + package: { + id: input.savedPackage.id, + kodyId: input.savedPackage.kodyId, + }, + exportName: input.exportName, + source: input.source, + topic: input.topic, + idempotency: { + key: input.idempotencyKey, + replayed: false, + }, + error: { + code: 'execution_failed', + message, + details: toJsonSafeValue(getExecutionErrorDetails(input.error)), + }, + logs: input.logs, + }, + } +} + +function buildJsonErrorResponse(input: { + status: number + code: string + message: string + idempotencyKey?: string + replayed?: boolean +}) { + return { + status: input.status, + body: { + ok: false, + error: { + code: input.code, + message: input.message, + }, + ...(input.idempotencyKey + ? { + idempotency: { + key: input.idempotencyKey, + replayed: input.replayed ?? false, + }, + } + : {}), + }, + } satisfies PackageInvocationStoredResponse +} + +function buildIdempotencyResponseUnavailable(input: { + idempotencyKey: string +}) { + return buildJsonErrorResponse({ + status: 409, + code: 'idempotency_response_unavailable', + message: + 'This idempotency key already has a terminal invocation record, but its stored response could not be replayed.', + idempotencyKey: input.idempotencyKey, + }) +} + +function resolveExistingInvocation(input: { + record: NonNullable>> + requestHash: string + idempotencyKey: string +}): PackageInvocationStoredResponse { + if (input.record.request_hash !== input.requestHash) { + return buildJsonErrorResponse({ + status: 409, + code: 'idempotency_mismatch', + message: + 'This idempotency key has already been used for a different package invocation request.', + idempotencyKey: input.idempotencyKey, + }) + } + if (input.record.status === 'in_progress') { + return buildJsonErrorResponse({ + status: 409, + code: 'invocation_in_progress', + message: + 'This idempotency key is already processing for the requested package export.', + idempotencyKey: input.idempotencyKey, + }) + } + if (input.record.storedResponse) { + return markStoredResponseAsReplayed(input.record.storedResponse) + } + return buildIdempotencyResponseUnavailable({ + idempotencyKey: input.idempotencyKey, + }) +} + +export async function invokePackageExport(input: { + env: Env + baseUrl: string + token: PackageInvocationTokenScope + request: PackageInvocationRequest +}): Promise { + const packageIdOrKodyId = input.request.packageIdOrKodyId.trim() + if (!packageIdOrKodyId) { + return buildJsonErrorResponse({ + status: 400, + code: 'invalid_package', + message: 'Package id or kody id is required.', + }) + } + const exportName = normalizeExportName(input.request.exportName) + const idempotencyKey = input.request.idempotencyKey.trim() + if (!idempotencyKey) { + return buildJsonErrorResponse({ + status: 400, + code: 'missing_idempotency_key', + message: 'Package invocations require a non-empty idempotencyKey.', + }) + } + const source = normalizeNullableString(input.request.source) + const topic = normalizeNullableString(input.request.topic) + if (!tokenAllowsSource({ token: input.token, source })) { + return buildJsonErrorResponse({ + status: 403, + code: 'source_not_allowed', + message: 'This token is not allowed to invoke the requested source.', + idempotencyKey, + }) + } + const savedPackage = await resolveSavedPackage({ + db: input.env.APP_DB, + userId: input.token.userId, + packageIdOrKodyId, + }) + if (!savedPackage) { + return buildJsonErrorResponse({ + status: 404, + code: 'package_not_found', + message: `Saved package "${packageIdOrKodyId}" was not found for this user.`, + idempotencyKey, + }) + } + if (!tokenAllowsPackage({ token: input.token, savedPackage })) { + return buildJsonErrorResponse({ + status: 403, + code: 'package_not_allowed', + message: 'This token is not allowed to invoke the requested package.', + idempotencyKey, + }) + } + if (!tokenAllowsExport({ token: input.token, exportName })) { + return buildJsonErrorResponse({ + status: 403, + code: 'export_not_allowed', + message: `This token is not allowed to invoke export "${exportName}".`, + idempotencyKey, + }) + } + + const requestHash = await createRequestHash({ + packageId: savedPackage.id, + exportName, + params: input.request.params, + source, + topic, + }) + let existing: Awaited> + try { + existing = await getPackageInvocationByKey({ + db: input.env.APP_DB, + userId: input.token.userId, + tokenId: input.token.tokenId, + packageId: savedPackage.id, + exportName, + idempotencyKey, + }) + } catch (error) { + console.error('package invocation idempotency lookup failed', error) + return buildJsonErrorResponse({ + status: 500, + code: 'idempotency_lookup_failed', + message: + 'Unable to look up the package invocation idempotency record. Please retry.', + idempotencyKey, + }) + } + if (existing) { + return resolveExistingInvocation({ + record: existing, + requestHash, + idempotencyKey, + }) + } + + const invocationId = crypto.randomUUID() + let inserted: boolean + try { + inserted = await insertPackageInvocationRow({ + db: input.env.APP_DB, + row: { + id: invocationId, + userId: input.token.userId, + tokenId: input.token.tokenId, + packageId: savedPackage.id, + packageKodyId: savedPackage.kodyId, + exportName, + idempotencyKey, + requestHash, + source, + topic, + status: 'in_progress', + }, + }) + } catch (error) { + console.error('package invocation idempotency persistence failed', error) + return buildJsonErrorResponse({ + status: 500, + code: 'idempotency_persistence_failed', + message: + 'Unable to persist the package invocation idempotency record. Please retry.', + idempotencyKey, + }) + } + if (!inserted) { + let current: Awaited> + try { + current = await getPackageInvocationByKey({ + db: input.env.APP_DB, + userId: input.token.userId, + tokenId: input.token.tokenId, + packageId: savedPackage.id, + exportName, + idempotencyKey, + }) + } catch (error) { + console.error('package invocation idempotency lookup failed', error) + return buildJsonErrorResponse({ + status: 500, + code: 'idempotency_lookup_failed', + message: + 'Unable to look up the package invocation idempotency record. Please retry.', + idempotencyKey, + }) + } + if (!current) { + return buildJsonErrorResponse({ + status: 500, + code: 'idempotency_conflict_unresolved', + message: + 'Package invocation idempotency insert conflicted but no existing row was found.', + idempotencyKey, + }) + } + return resolveExistingInvocation({ + record: current, + requestHash, + idempotencyKey, + }) + } + + try { + const { artifact, source: sourceRow } = await ensureModuleArtifact({ + env: input.env, + baseUrl: input.baseUrl, + savedPackage, + exportName, + userId: input.token.userId, + }) + const repoSource = + artifact.packageContext?.sourceId != null + ? await getEntitySourceById( + input.env.APP_DB, + artifact.packageContext.sourceId, + ) + : sourceRow + const callerContext = createMcpCallerContext({ + baseUrl: input.baseUrl, + user: { + userId: input.token.userId, + email: input.token.email, + displayName: input.token.displayName, + }, + storageContext: { + sessionId: null, + appId: savedPackage.id, + storageId: buildPackageInvocationStorageId(savedPackage.id), + }, + repoContext: repoSource ? createRepoContext(repoSource) : null, + }) + const executionResult = await runBundledModuleWithRegistry( + input.env, + callerContext, + { + mainModule: artifact.mainModule, + modules: artifact.modules, + }, + input.request.params, + { + storageTools: { + userId: input.token.userId, + storageId: buildPackageInvocationStorageId(savedPackage.id), + writable: true, + }, + ...(artifact.packageContext + ? { packageContext: artifact.packageContext } + : {}), + }, + ) + const response = executionResult.error + ? buildExecutionErrorResponse({ + savedPackage, + exportName, + idempotencyKey, + source, + topic, + error: executionResult.error, + logs: executionResult.logs ?? [], + }) + : buildExecutionSuccessResponse({ + savedPackage, + exportName, + idempotencyKey, + source, + topic, + result: executionResult.result, + logs: executionResult.logs ?? [], + rawContent: extractRawContent(executionResult.result), + }) + await updatePackageInvocationResult({ + db: input.env.APP_DB, + id: invocationId, + userId: input.token.userId, + status: executionResult.error ? 'failed' : 'completed', + response, + }) + return response + } catch (error) { + if (isMissingPackageExportError(error)) { + const response = buildJsonErrorResponse({ + status: 404, + code: 'export_not_found', + message: error instanceof Error ? error.message : String(error), + idempotencyKey, + }) + await updatePackageInvocationResult({ + db: input.env.APP_DB, + id: invocationId, + userId: input.token.userId, + status: 'failed', + response, + }).catch(() => { + // Best effort; preserve the original invocation error. + }) + return response + } + const response = buildJsonErrorResponse({ + status: 500, + code: 'invocation_failed', + message: error instanceof Error ? error.message : String(error), + idempotencyKey, + }) + await updatePackageInvocationResult({ + db: input.env.APP_DB, + id: invocationId, + userId: input.token.userId, + status: 'failed', + response, + }).catch(() => { + // Best effort; preserve the original invocation error. + }) + return response + } +}