From 4d23e93fbaabfce42b960b9e04eeea36335341b5 Mon Sep 17 00:00:00 2001 From: jbiskur Date: Fri, 3 Jul 2026 16:35:22 +0100 Subject: [PATCH] feat: add encrypted pathway payload envelopes --- src/pathways/builder.ts | 97 +++++- src/pathways/encryption.ts | 114 +++++++ src/pathways/index.ts | 1 + src/pathways/types.ts | 8 + tests/pathway-encryption.test.ts | 523 +++++++++++++++++++++++++++++++ 5 files changed, 740 insertions(+), 3 deletions(-) create mode 100644 src/pathways/encryption.ts create mode 100644 tests/pathway-encryption.test.ts diff --git a/src/pathways/builder.ts b/src/pathways/builder.ts index 005c6ea..b3bae7c 100644 --- a/src/pathways/builder.ts +++ b/src/pathways/builder.ts @@ -86,6 +86,16 @@ import { import { FileEventSchema, FileInputSchema } from "./types.ts" import type { Buffer } from "node:buffer" import process from "node:process" +import { + createPathwayEncryptionProvider, + decryptPayloadEnvelope, + encryptPayloadEnvelope, + PATHWAY_ENCRYPTED_METADATA_KEY, + PATHWAY_ENCRYPTION_SCHEME, + PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY, + type PathwayEncryptionConfig, + type PathwayEncryptionProvider, +} from "./encryption.ts" /** * Default timeout for pathway processing in milliseconds (10 seconds) @@ -247,6 +257,15 @@ export interface PathwaysBuilderConfig { */ provisionFailure?: ProvisionFailureMode | ProvisionFailureConfig managedConfig?: ManagedPathwayConfig + /** + * Optional symmetric encryption for pathways marked with `encrypted`. + * + * When omitted, `mode: "none"`, or `mode: "symmetric"` without a key, encrypted + * pathway payloads are left unchanged. When a symmetric key is supplied, the + * full payload is AES-256-GCM encrypted on write and decrypted before + * process-time schema validation and handlers. + */ + encryption?: PathwayEncryptionConfig } /** @@ -386,6 +405,7 @@ export class PathwaysBuilder< private readonly writable: Record = {} as Record private readonly subscribed: Record = {} as Record private readonly pumpGroups: Record = {} as Record + private readonly encryptedPathways: Record = {} as Record private readonly timeouts: Record = {} as Record private readonly maxRetries: Record = {} as Record private readonly retryDelays: Record = {} as Record @@ -403,6 +423,7 @@ export class PathwaysBuilder< // Logger instance (but not using it yet due to TypeScript errors) private readonly logger: Logger + private readonly encryptionProvider: PathwayEncryptionProvider | null // Configuration values needed for cloning private readonly baseUrl: string @@ -481,6 +502,7 @@ export class PathwaysBuilder< defaultAutoProvision, provisionFailure, managedConfig, + encryption, }: PathwaysBuilderConfig) { // Initialize logger (use NoopLogger if none provided) this.logger = logger ?? new NoopLogger() @@ -505,6 +527,7 @@ export class PathwaysBuilder< this.dataCoreAccessControl = dataCoreAccessControl ?? "private" this.dataCoreDeleteProtection = dataCoreDeleteProtection ?? false this.provisionFailure = provisionFailure + this.encryptionProvider = createPathwayEncryptionProvider(encryption) // Store virtual pathway auto-provisioning config this.pathwayName = pathwayName @@ -662,6 +685,8 @@ export class PathwaysBuilder< throw new Error(error) } + data.payload = this.decryptPathwayPayload(pathway, data.payload, data.metadata) + // Validate event payload against schema if available if (this.schemas[pathway]) { const parsedPayload = this.schemas[pathway].safeParse(data.payload) @@ -845,6 +870,7 @@ export class PathwaysBuilder< maxRetries?: number retryDelayMs?: number isFilePathway?: FP + encrypted?: boolean }, ): PathwaysBuilder< & TPathway @@ -857,6 +883,10 @@ export class PathwaysBuilder< const path = `${contract.flowType}/${contract.eventType}` as PathwayKey const writable = contract.writable ?? true const subscribe = contract.subscribe ?? true + const encrypted = contract.encrypted === true + if (contract.isFilePathway && encrypted) { + throw new Error(`Pathway ${path} is a file pathway and cannot be encrypted`) + } if (contract.pumpGroup !== undefined && contract.pumpGroup.trim() === "") { throw new Error( `Pathway ${path} has an empty pumpGroup — pumpGroup must be a non-empty string when set`, @@ -871,6 +901,7 @@ export class PathwaysBuilder< writable, subscribe, pumpGroup, + encrypted, isFilePathway: contract.isFilePathway, timeoutMs: contract.timeoutMs, maxRetries: contract.maxRetries, @@ -921,6 +952,7 @@ export class PathwaysBuilder< this.writable[path] = writable this.subscribed[path] = subscribe this.pumpGroups[path] = pumpGroup + this.encryptedPathways[path] = encrypted // Store provisioning descriptions if (contract.description !== undefined) { @@ -937,6 +969,7 @@ export class PathwaysBuilder< writable, subscribe, pumpGroup, + encrypted, isFilePathway: contract.isFilePathway, }) @@ -1208,11 +1241,17 @@ export class PathwaysBuilder< finalMetadata[AUDIT_SESSION_ID] = options.sessionId } + const { data: eventData, encrypted } = this.encryptPathwayPayload(path, data, Boolean(batch)) + if (encrypted) { + finalMetadata[PATHWAY_ENCRYPTED_METADATA_KEY] = "true" + finalMetadata[PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY] = PATHWAY_ENCRYPTION_SCHEME + } + let eventIds: string | string[] = [] this.logger.debug(`Writing webhook data to pathway`, { pathway: pathStr, batch }) if (batch) { eventIds = await (this.batchWriters[path] as SendWebhookBatch)( - data as unknown as TPathway[TPath]["output"][], + eventData as unknown as TPathway[TPath]["output"][], finalMetadata, options, ).catch((error) => { @@ -1223,7 +1262,7 @@ export class PathwaysBuilder< throw error }) } else if (this.filePathways.has(path)) { - const { fileId, fileName, fileContent, ...additionalProperties } = data as z.infer + const { fileId, fileName, fileContent, ...additionalProperties } = eventData as z.infer const fileType = await fileTypeFromBuffer(fileContent as Buffer) process.env.DEBUG?.includes("pathways") && console.log("additionalProperties", additionalProperties) eventIds = await (this.fileWriters[path] as SendFilehook)( @@ -1246,7 +1285,7 @@ export class PathwaysBuilder< throw error }) } else { - eventIds = await (this.writers[path] as SendWebhook)(data, finalMetadata, options) + eventIds = await (this.writers[path] as SendWebhook)(eventData, finalMetadata, options) .catch((error) => { this.logger.error(`Error writing webhook to pathway`, { pathway: pathStr, @@ -1278,6 +1317,58 @@ export class PathwaysBuilder< return eventIds } + private encryptPathwayPayload( + path: TPath, + data: unknown, + batch: boolean, + ): { data: unknown; encrypted: boolean } { + if (!this.encryptedPathways[path] || !this.encryptionProvider) { + return { data, encrypted: false } + } + + if (batch) { + if (!Array.isArray(data)) { + return { data, encrypted: false } + } + return { + data: data.map((item) => encryptPayloadEnvelope(item, this.encryptionProvider!)), + encrypted: true, + } + } + + return { + data: encryptPayloadEnvelope(data, this.encryptionProvider), + encrypted: true, + } + } + + private decryptPathwayPayload( + path: TPath, + payload: unknown, + metadata: unknown, + ): unknown { + if (!this.encryptedPathways[path] || !this.hasEncryptedPayloadMetadata(metadata)) { + return payload + } + + if (!this.encryptionProvider) { + throw new Error( + `Pathway ${String(path)} received encrypted payload but no symmetric encryption key is configured`, + ) + } + + return decryptPayloadEnvelope(payload, this.encryptionProvider) + } + + private hasEncryptedPayloadMetadata(metadata: unknown): boolean { + if (!metadata || typeof metadata !== "object") { + return false + } + + const value = (metadata as Record)[PATHWAY_ENCRYPTED_METADATA_KEY] + return value === true || value === "true" + } + /** * Waits for a specific event to be processed * diff --git a/src/pathways/encryption.ts b/src/pathways/encryption.ts new file mode 100644 index 0000000..4558512 --- /dev/null +++ b/src/pathways/encryption.ts @@ -0,0 +1,114 @@ +import { createCipheriv, createDecipheriv, createHash, randomBytes } from "node:crypto" +import { Buffer } from "node:buffer" + +const ALGORITHM = "aes-256-gcm" +const IV_LENGTH = 12 +const TAG_LENGTH = 16 +const MIN_KEY_LENGTH = 32 + +export const ENCRYPTED_PAYLOAD_FIELD = "encryptedPayload" +export const PATHWAY_ENCRYPTED_METADATA_KEY = "pathways/encrypted" +export const PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY = "pathways/encryption-scheme" +export const PATHWAY_ENCRYPTION_SCHEME = "aes-256-gcm-sha256-v1" + +export type PathwayEncryptionMode = "none" | "symmetric" + +export interface PathwayEncryptionConfig { + mode?: PathwayEncryptionMode + key?: string +} + +export interface PathwayEncryptionProvider { + encrypt(plaintext: string): string + decrypt(payload: string): string +} + +export function deriveEncryptionKey(secret: string): Buffer { + return createHash("sha256").update(secret).digest() +} + +export function aesGcmEncrypt(plaintext: string, key: Buffer): string { + const iv = randomBytes(IV_LENGTH) + const cipher = createCipheriv(ALGORITHM, key, iv, { authTagLength: TAG_LENGTH }) + const ciphertext = Buffer.concat([cipher.update(plaintext, "utf8"), cipher.final()]) + const authTag = cipher.getAuthTag() + return `${iv.toString("base64")}.${ciphertext.toString("base64")}.${authTag.toString("base64")}` +} + +export function aesGcmDecrypt(payload: string, key: Buffer): string { + const parts = payload.split(".") + if (parts.length !== 3) { + throw new Error("Invalid AES-256-GCM payload format") + } + + const [ivB64, ciphertextB64, authTagB64] = parts + try { + const decipher = createDecipheriv(ALGORITHM, key, Buffer.from(ivB64, "base64"), { + authTagLength: TAG_LENGTH, + }) + decipher.setAuthTag(Buffer.from(authTagB64, "base64")) + return Buffer.concat([ + decipher.update(Buffer.from(ciphertextB64, "base64")), + decipher.final(), + ]).toString("utf8") + } catch { + throw new Error("AES-256-GCM decryption failed: auth tag mismatch (wrong key or tampered payload)") + } +} + +export function createPathwayEncryptionProvider( + config?: PathwayEncryptionConfig, +): PathwayEncryptionProvider | null { + const mode = config?.mode ?? (config?.key ? "symmetric" : "none") + if (mode === "none") { + return null + } + + if (mode !== "symmetric") { + throw new Error(`Unknown encryption mode: ${String(mode)}`) + } + + const secret = config?.key + if (!secret) { + return null + } + + if (secret.length < MIN_KEY_LENGTH) { + throw new Error( + "Pathways symmetric encryption key must be at least 32 characters (generate with: openssl rand -hex 32)", + ) + } + + const key = deriveEncryptionKey(secret) + return { + encrypt: (plaintext: string) => aesGcmEncrypt(plaintext, key), + decrypt: (payload: string) => aesGcmDecrypt(payload, key), + } +} + +export function encryptPayloadEnvelope(payload: unknown, provider: PathwayEncryptionProvider): Record { + return { + [ENCRYPTED_PAYLOAD_FIELD]: provider.encrypt(JSON.stringify(payload)), + } +} + +export function decryptPayloadEnvelope(payload: unknown, provider: PathwayEncryptionProvider): unknown { + const encryptedPayload = typeof payload === "string" + ? payload + : payload && typeof payload === "object" && !Array.isArray(payload) + ? (payload as Record)[ENCRYPTED_PAYLOAD_FIELD] + : undefined + + if (typeof encryptedPayload !== "string") { + throw new Error(`Encrypted pathway payload must be a string or an object with ${ENCRYPTED_PAYLOAD_FIELD}`) + } + + try { + return JSON.parse(provider.decrypt(encryptedPayload)) + } catch (error) { + if (error instanceof SyntaxError) { + throw new Error("Encrypted pathway payload decrypted to invalid JSON") + } + throw error + } +} diff --git a/src/pathways/index.ts b/src/pathways/index.ts index accd572..1aed24d 100644 --- a/src/pathways/index.ts +++ b/src/pathways/index.ts @@ -21,6 +21,7 @@ export type { ProvisionFailureMode, } from "./provisioner.ts" export * from "./constants.ts" +export * from "./encryption.ts" export * from "./internal-pathway.state.ts" export * from "./kv/kv-adapter.ts" export * from "./logger.ts" diff --git a/src/pathways/types.ts b/src/pathways/types.ts index 6f1b0ee..dfa35d4 100644 --- a/src/pathways/types.ts +++ b/src/pathways/types.ts @@ -85,6 +85,14 @@ export interface PathwayContract, key = ENCRYPTION_KEY): unknown { + return JSON.parse(aesGcmDecrypt(envelope[ENCRYPTED_PAYLOAD_FIELD] as string, deriveEncryptionKey(key))) +} + +function decodeMetadata(header: string): Record { + return JSON.parse(atob(header)) +} + +function createEvent(payload: unknown, metadata: Record = {}): FlowcoreEvent { + return { + eventId: crypto.randomUUID(), + timeBucket: "202607030000", + tenant: "test-tenant", + dataCoreId: "test-data-core", + flowType: "encrypted-flow", + eventType: "created", + metadata, + validTime: new Date().toISOString(), + payload, + } +} + +Deno.test("pathway AES-GCM helpers roundtrip and reject invalid payloads", async () => { + const key = deriveEncryptionKey(ENCRYPTION_KEY) + + const first = aesGcmEncrypt(JSON.stringify({ value: "hello" }), key) + const second = aesGcmEncrypt(JSON.stringify({ value: "hello" }), key) + assertNotEquals(first, second) + assertEquals(JSON.parse(aesGcmDecrypt(first, key)), { value: "hello" }) + assertEquals(JSON.parse(aesGcmDecrypt(aesGcmEncrypt("{}", key), key)), {}) + assertEquals(JSON.parse(aesGcmDecrypt(aesGcmEncrypt(JSON.stringify({ text: "føroyskt 🔐" }), key), key)), { + text: "føroyskt 🔐", + }) + + await assertRejects( + async () => aesGcmDecrypt("not-a-valid-payload", key), + Error, + "Invalid AES-256-GCM payload format", + ) + + await assertRejects( + async () => aesGcmDecrypt(first, deriveEncryptionKey(OTHER_KEY)), + Error, + "AES-256-GCM decryption failed", + ) +}) + +Deno.test({ + name: "encrypted pathway writes Flowcore-compatible envelope and marker metadata", + sanitizeResources: false, + sanitizeOps: false, + fn: async () => { + const server = createTestServer(8010) + await server.start() + try { + const builder = new PathwaysBuilder({ + baseUrl: `http://localhost:${server.port}`, + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + const plaintext = { + id: "fragment-1", + title: "Top Secret", + content: "Sensitive content", + tags: ["confidential", "leaf"], + workspaceId: "workspace-1", + } + + const eventId = await pathway.write("encrypted-flow/created", { + data: plaintext, + metadata: { source: "test" }, + options: { fireAndForget: true, sessionId: "session-1" }, + }) + + const stored = server.storedEvents.get(eventId as string) as { + body: Record + headers: Record + } + assertExists(stored) + assertEquals(Object.keys(stored.body), [ENCRYPTED_PAYLOAD_FIELD]) + assertEquals(typeof stored.body[ENCRYPTED_PAYLOAD_FIELD], "string") + assertNotEquals(stored.body[ENCRYPTED_PAYLOAD_FIELD], JSON.stringify(plaintext)) + assertEquals(decryptEnvelope(stored.body), plaintext) + + const metadata = decodeMetadata(stored.headers["x-flowcore-metadata-json"]) + assertEquals(metadata.source, "test") + assertEquals(metadata["audit/session-id"], "session-1") + assertEquals(metadata[PATHWAY_ENCRYPTED_METADATA_KEY], "true") + assertEquals(metadata[PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY], PATHWAY_ENCRYPTION_SCHEME) + } finally { + await server.stop() + } + }, +}) + +Deno.test("encrypted pathway without configured key stays plaintext and unmarked", async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8011", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric" }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let capturedPayload: Record | undefined + let capturedMetadata: Record | undefined + pathway["writers"]["encrypted-flow/created"] = async (payload, metadata) => { + capturedPayload = payload + capturedMetadata = metadata + return "event-1" + } + + await pathway.write("encrypted-flow/created", { + data: { + id: "fragment-1", + title: "Plain Title", + content: "Plain content", + workspaceId: "workspace-1", + }, + options: { fireAndForget: true }, + }) + + assertEquals(capturedPayload?.title, "Plain Title") + assertEquals(capturedMetadata?.[PATHWAY_ENCRYPTED_METADATA_KEY], undefined) +}) + +Deno.test("encryption config defaults to symmetric when only key is provided", async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8017", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let captured: Record | undefined + pathway["writers"]["encrypted-flow/created"] = async (payload) => { + captured = payload + return "event-1" + } + + const plaintext = { + id: "fragment-1", + title: "Key Only", + content: "Encrypted content", + workspaceId: "workspace-1", + } + await pathway.write("encrypted-flow/created", { + data: plaintext, + options: { fireAndForget: true }, + }) + + assertExists(captured) + assertEquals(Object.keys(captured), [ENCRYPTED_PAYLOAD_FIELD]) + assertEquals(decryptEnvelope(captured), plaintext) +}) + +Deno.test({ + name: "encrypted pathway process decrypts marked envelope before schema validation and handler", + sanitizeResources: false, + sanitizeOps: false, + fn: async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8012", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let handledPayload: Record | undefined + pathway.handle("encrypted-flow/created", (event) => { + handledPayload = event.payload as Record + }) + + const plaintext = { + id: "fragment-1", + title: "Secret Title", + content: "Secret content", + tags: ["leaf"], + workspaceId: "workspace-1", + } + + await pathway.process( + "encrypted-flow/created", + createEvent( + { [ENCRYPTED_PAYLOAD_FIELD]: encryptedPayload(plaintext) }, + { [PATHWAY_ENCRYPTED_METADATA_KEY]: "true" }, + ), + ) + + assertEquals(handledPayload, plaintext) + }, +}) + +Deno.test({ + name: "encrypted pathway process accepts raw encrypted string when marker is present", + sanitizeResources: false, + sanitizeOps: false, + fn: async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8018", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let handledPayload: Record | undefined + pathway.handle("encrypted-flow/created", (event) => { + handledPayload = event.payload as Record + }) + + const plaintext = { + id: "fragment-1", + title: "Manual Secret", + content: "Manual content", + workspaceId: "workspace-1", + } + + await pathway.process( + "encrypted-flow/created", + createEvent(encryptedPayload(plaintext), { [PATHWAY_ENCRYPTED_METADATA_KEY]: true }), + ) + + assertEquals(handledPayload, plaintext) + }, +}) + +Deno.test({ + name: "encrypted pathway process leaves markerless legacy plaintext unchanged", + sanitizeResources: false, + sanitizeOps: false, + fn: async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8019", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let handledPayload: Record | undefined + pathway.handle("encrypted-flow/created", (event) => { + handledPayload = event.payload as Record + }) + + const plaintext = { + id: "fragment-1", + title: "Legacy Plain", + content: "Legacy content", + workspaceId: "workspace-1", + } + await pathway.process("encrypted-flow/created", createEvent(plaintext)) + + assertEquals(handledPayload, plaintext) + }, +}) + +Deno.test("encrypted pathway process throws on wrong key ciphertext", async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8013", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + await assertRejects( + async () => + pathway.process( + "encrypted-flow/created", + createEvent( + { + [ENCRYPTED_PAYLOAD_FIELD]: encryptedPayload({ + id: "fragment-1", + title: "Foreign title", + content: "Foreign content", + workspaceId: "workspace-1", + }, OTHER_KEY), + }, + { [PATHWAY_ENCRYPTED_METADATA_KEY]: "true" }, + ), + ), + Error, + "AES-256-GCM decryption failed", + ) +}) + +Deno.test("encrypted pathway batch write encrypts each item independently", async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8014", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let captured: Record[] | undefined + let capturedMetadata: Record | undefined + pathway["batchWriters"]["encrypted-flow/created"] = async (payload, metadata) => { + captured = payload + capturedMetadata = metadata + return ["event-1", "event-2"] + } + + const first = { id: "fragment-1", title: "One", content: "First", workspaceId: "workspace-1" } + const second = { id: "fragment-2", title: "Two", content: "Second", workspaceId: "workspace-1" } + await pathway.write("encrypted-flow/created", { + batch: true, + data: [first, second], + options: { fireAndForget: true }, + }) + + assertExists(captured) + assertEquals(captured.length, 2) + assertEquals(Object.keys(captured[0]), [ENCRYPTED_PAYLOAD_FIELD]) + assertEquals(Object.keys(captured[1]), [ENCRYPTED_PAYLOAD_FIELD]) + assertNotEquals(captured[0][ENCRYPTED_PAYLOAD_FIELD], captured[1][ENCRYPTED_PAYLOAD_FIELD]) + assertEquals(decryptEnvelope(captured[0]), first) + assertEquals(decryptEnvelope(captured[1]), second) + assertEquals(capturedMetadata?.[PATHWAY_ENCRYPTED_METADATA_KEY], "true") +}) + +Deno.test({ + name: "router path decrypts marked encrypted envelope through PathwaysBuilder.process", + sanitizeResources: false, + sanitizeOps: false, + fn: async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8015", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const pathway = builder.register({ + flowType: "encrypted-flow", + eventType: "created", + schema: eventSchema, + encrypted: true, + }) + + let handledPayload: Record | undefined + pathway.handle("encrypted-flow/created", (event) => { + handledPayload = event.payload as Record + }) + + const plaintext = { + id: "fragment-1", + title: "Router Secret", + content: "Router content", + workspaceId: "workspace-1", + } + + const router = new PathwayRouter(pathway, "secret") + await router.processEvent( + createEvent( + { [ENCRYPTED_PAYLOAD_FIELD]: encryptedPayload(plaintext) }, + { [PATHWAY_ENCRYPTED_METADATA_KEY]: "true" }, + ), + "secret", + ) + + assertEquals(handledPayload, plaintext) + }, +}) + +Deno.test("non-encrypted pathway unchanged and encrypted file pathways are rejected", async () => { + const builder = new PathwaysBuilder({ + baseUrl: "http://localhost:8016", + tenant: "test-tenant", + dataCore: "test-data-core", + apiKey: "test-api-key", + encryption: { mode: "symmetric", key: ENCRYPTION_KEY }, + }) + + const plain = builder.register({ + flowType: "plain-flow", + eventType: "created", + schema: eventSchema, + }) + + let captured: Record | undefined + plain["writers"]["plain-flow/created"] = async (payload) => { + captured = payload + return "event-1" + } + await plain.write("plain-flow/created", { + data: { id: "fragment-1", title: "Plain", content: "Plain content", workspaceId: "workspace-1" }, + options: { fireAndForget: true }, + }) + assertEquals(captured?.title, "Plain") + + assertRejects( + async () => + plain.register({ + flowType: "file-flow", + eventType: "uploaded", + schema: z.object({}), + isFilePathway: true, + encrypted: true, + }), + Error, + "file pathway and cannot be encrypted", + ) + + const file = plain.register({ + flowType: "file-flow", + eventType: "uploaded", + schema: z.object({}), + isFilePathway: true, + }) + + await assertRejects( + async () => + file.write("file-flow/uploaded", { + batch: true, + data: [ + { + fileId: "file-1", + fileName: "test.txt", + fileContent: Buffer.from("hello"), + }, + ], + }), + Error, + "Batch is not possible for file pathways", + ) +})