feat(core): accept Fusor JSON webhooks

docs/webhooks.mdx.vel

@@ -15,11 +15,11 @@ import { TypeTooltip } from "/snippets/type-tooltip.mdx";
 
 | | Native Spectrum webhook | Fusor webhook |
 |---|---|---|
-| Body | HMAC-signed, normalized JSON | Protobuf envelope (raw provider request) |
+| Body | HMAC-signed, normalized JSON | Versioned JSON envelope (original provider request included) |
 | Auth | HMAC over body, verified with `webhookSecret` | Platform's own signature via provider `verify()` |
 | Requires a Fusor provider | No | Yes |
 
-Detection is by payload shape (JSON vs protobuf), not headers. Your handler receives the same `(space, message)` pair either way.
+Fusor deliveries are selected by the CloudEvents header `ce-type: dev.spctrm.fusor.delivery`; every other request follows the native signed-webhook path. Your handler receives the same `(space, message)` pair either way.
 
 ## Configuring a webhook secret
 
@@ -79,6 +79,19 @@ The handler is invoked **fire-and-forget** — it runs after the HTTP response i
 Pass the raw body bytes. The HMAC is computed over the exact bytes on the wire. If your framework parses the body to JSON and you re-stringify it, the bytes change and verification fails.
 </Warning>
 
+Fusor's schema-version `1` envelope is plain JSON. It exposes the original request's `method`, path (including query string), lower-case headers, a normalized `body` arm, and `rawBodyBase64`. The SDK validates the envelope and always passes the bytes decoded from `rawBodyBase64` to the provider's `verify()` function, so signatures are checked against the exact provider payload rather than reserialized JSON.
+
+`request.bodyEncoding` describes the JSON-friendly `request.body` value:
+
+| `bodyEncoding` | `request.body` |
+|---|---|
+| `json` | Any JSON value |
+| `form` | An object of strings, with repeated form keys represented as ordered string arrays |
+| `text` | A UTF-8 string |
+| `base64` | A base64 string, identical to `rawBodyBase64` |
+
+Low-code consumers can work directly with `request.body`. Signature-aware provider code should use the bytes decoded from `rawBodyBase64`; the SDK does this automatically before calling `verify()`.
+
 ## Framework adapters
 
 First-party adapters mount the endpoint for you and handle raw-body parsing correctly. Install the adapter package and its framework only when you use it.
@@ -205,7 +218,7 @@ First-party adapters mount the endpoint for you and handle raw-body parsing corr
 - **Signature verification.** Native webhooks are verified with `HMAC-SHA256` over `v0:<timestamp>:<rawBody>`, with a 5-minute replay window. Bad signature returns `401`, missing headers return `400`.
 - **Payload deserialization.** Native webhook JSON is deserialized into normal <TypeTooltip name="Message" type={`{{ message.signature }}`} /> and <TypeTooltip name="Space" type={`{{ space.signature }}`} /> objects, including reactions and grouped items.
 - **Attachment rehydration.** Native webhooks carry attachment metadata only. `read()` and `stream()` fetch the bytes lazily via the platform.
-- **Format detection.** Native vs Fusor is detected per request by payload shape — JSON for native, protobuf for Fusor.
+- **Format detection.** `ce-type: dev.spctrm.fusor.delivery` selects the Fusor v1 JSON path; all other requests use native HMAC verification.
 
 ## Delivery semantics
 

packages/core/src/fusor/core.ts

@@ -38,6 +38,11 @@ export interface RegisteredFusorHandler<TPayload = unknown> {
   verify: FusorVerify<TPayload>;
 }
 
+export interface FusorEventMetadata {
+  eventId: string;
+  platform: string;
+}
+
 function toReplyBytes(body: string | Uint8Array | undefined): Uint8Array {
   if (body === undefined) {
     return new Uint8Array(0);
@@ -160,7 +165,7 @@ function runHandlerOnce<TPayload>(
 
 export interface FusorCoreOptions {
   // Optional: only the streaming transport (start) needs cloud credentials to
-  // mint a token. The webhook path (processEvent) routes registered handlers
+  // mint a token. The webhook path (processRequest) routes registered handlers
   // without them, so a webhook-only Spectrum can construct a core with
   // neither set.
   projectId?: string;
@@ -316,40 +321,53 @@ export class FusorCore {
     }
   }
 
-  // Transport-independent event processing: route by platform, parse the wire
-  // request, run every registered handler (verify → messages), and combine the
-  // results into a single InboundReply. Returns the reply instead of writing it
-  // anywhere, so both the streaming session (sendReply) and the synchronous
-  // webhook path can drive it. `deliver` controls where produced records go:
-  // the streaming path defaults to each handler's pushMessage (the per-platform
-  // queue feeding spectrum.messages); the webhook path collects them for the
-  // request instead.
+  private noHandlerReply(event: FusorEventMetadata): InboundReply {
+    // Reply shape stays wire-compatible; only the local log gets the install
+    // hint (since v5 the official providers are separate packages, so "no
+    // handler" is usually a missing install, not a routing bug).
+    const hint = officialProviderInstallHint(event.platform);
+    log.warn(
+      hint
+        ? `fusor: no handler for platform — ${hint}`
+        : "fusor: no handler for platform",
+      {
+        "spectrum.fusor.platform": event.platform,
+        "spectrum.fusor.event_id": event.eventId,
+      }
+    );
+    return {
+      eventId: event.eventId,
+      errorReason: `no handler for platform ${event.platform}`,
+      status: 0,
+      headers: {},
+      body: new Uint8Array(0),
+    };
+  }
+
+  private async processParsedRequest(
+    event: FusorEventMetadata,
+    parsedRequest: ParsedHttpRequest,
+    handlers: RegisteredFusorHandler[],
+    deliver?: (record: ProviderMessageRecord) => void
+  ): Promise<InboundReply> {
+    const outcomes = await Promise.all(
+      handlers.map((handler) => runHandlerOnce(handler, parsedRequest, deliver))
+    );
+
+    const combined = combineReplies(outcomes);
+    combined.eventId = event.eventId;
+    return combined;
+  }
+
+  // WebSocket events still carry protobuf/raw HTTP. Parse that transport shape,
+  // then hand the resulting request to the shared provider pipeline.
   async processEvent(
     event: RawInboundEvent,
     deliver?: (record: ProviderMessageRecord) => void
   ): Promise<InboundReply> {
     const handlers = this.handlers.get(event.platform) ?? [];
     if (handlers.length === 0) {
-      // Reply shape stays wire-compatible; only the local log gets the
-      // install hint (since v5 the official providers are separate packages,
-      // so "no handler" is usually a missing install, not a routing bug).
-      const hint = officialProviderInstallHint(event.platform);
-      log.warn(
-        hint
-          ? `fusor: no handler for platform — ${hint}`
-          : "fusor: no handler for platform",
-        {
-          "spectrum.fusor.platform": event.platform,
-          "spectrum.fusor.event_id": event.eventId,
-        }
-      );
-      return {
-        eventId: event.eventId,
-        errorReason: `no handler for platform ${event.platform}`,
-        status: 0,
-        headers: {},
-        body: new Uint8Array(0),
-      };
+      return this.noHandlerReply(event);
     }
 
     let parsedRequest: ParsedHttpRequest;
@@ -371,13 +389,22 @@ export class FusorCore {
       };
     }
 
-    const outcomes = await Promise.all(
-      handlers.map((handler) => runHandlerOnce(handler, parsedRequest, deliver))
-    );
+    return this.processParsedRequest(event, parsedRequest, handlers, deliver);
+  }
 
-    const combined = combineReplies(outcomes);
-    combined.eventId = event.eventId;
-    return combined;
+  // HTTP JSON deliveries already provide method/path/headers plus the exact
+  // original body bytes, so they enter after raw HTTP parsing. Keeping this
+  // seam shared means WebSocket protobuf behavior remains unchanged.
+  async processRequest(
+    event: FusorEventMetadata,
+    parsedRequest: ParsedHttpRequest,
+    deliver?: (record: ProviderMessageRecord) => void
+  ): Promise<InboundReply> {
+    const handlers = this.handlers.get(event.platform) ?? [];
+    if (handlers.length === 0) {
+      return this.noHandlerReply(event);
+    }
+    return this.processParsedRequest(event, parsedRequest, handlers, deliver);
   }
 
   async close(): Promise<void> {

packages/core/src/fusor/parse.ts

@@ -8,6 +8,17 @@ export interface ParsedHttpRequest {
 const CR = 0x0d;
 const LF = 0x0a;
 
+export function mergeHeaderValue(
+  headers: Record<string, string>,
+  name: string,
+  value: string
+): void {
+  const lowerName = name.toLowerCase();
+  headers[lowerName] = Object.hasOwn(headers, lowerName)
+    ? `${headers[lowerName]}, ${value}`
+    : value;
+}
+
 function findHeaderEnd(bytes: Uint8Array): number {
   for (let i = 0; i + 3 < bytes.length; i++) {
     if (
@@ -23,9 +34,9 @@ function findHeaderEnd(bytes: Uint8Array): number {
 }
 
 /**
- * Parses an HTTP/1.1 wire-format request out of `raw_request` from
- * `RawInboundEvent`. Headers are lowercased. Multiple header values with the
- * same name are joined with ", " (RFC 7230 §3.2.2).
+ * Parses the HTTP/1.1 wire-format request carried by the WebSocket protobuf
+ * transport. HTTP JSON deliveries already contain these parsed fields. Headers
+ * are lowercased; repeated values are joined with ", " (RFC 7230 §3.2.2).
  */
 export function parseHttpRequest(bytes: Uint8Array): ParsedHttpRequest {
   const headerEnd = findHeaderEnd(bytes);
@@ -61,13 +72,12 @@ export function parseHttpRequest(bytes: Uint8Array): ParsedHttpRequest {
     if (colon < 0) {
       continue;
     }
-    const key = line.slice(0, colon).trim().toLowerCase();
+    const key = line.slice(0, colon).trim();
     const value = line.slice(colon + 1).trim();
     if (!key) {
       continue;
     }
-    const existing = headers[key];
-    headers[key] = existing ? `${existing}, ${value}` : value;
+    mergeHeaderValue(headers, key, value);
   }
 
   return { method, path, headers, rawBody };

packages/core/src/fusor/types.ts

@@ -84,15 +84,16 @@ export type WebhookHandler = (
 /**
  * Raw webhook input for HTTP servers without Web `Request`/`Response` (Express,
  * raw Node). `body` MUST be the exact bytes POSTed — never a re-encoded
- * JSON/text body — so both the protobuf decode (fusor) and the HMAC
- * verification (native Spectrum webhook) work.
+ * JSON/text body — native Spectrum HMAC verification is over those bytes, and
+ * Fusor validates its versioned JSON envelope from them.
  *
- * `headers` ARE read for **native Spectrum webhooks**: `X-Spectrum-Signature` /
- * `X-Spectrum-Timestamp` carry the HMAC verified against
- * `Spectrum({ webhookSecret })`, and the signature header also selects the
- * native path. For **fusor** envelopes they are ignored (authenticity is the
- * per-platform `verify()` reading the inner reconstructed request). The natural
- * `{ headers: req.headers, body: req.body }` shape works for both.
+ * `headers` is optional on this raw-input type; omitted headers are treated as
+ * an empty record. Routing and verification still depend on them: Fusor
+ * deliveries carry `ce-type: dev.spctrm.fusor.delivery`; every other request is
+ * treated as a native Spectrum webhook, whose `X-Spectrum-Signature` /
+ * `X-Spectrum-Timestamp` are verified against `Spectrum({ webhookSecret })` and
+ * rejected when missing. The natural `{ headers: req.headers, body: req.body }`
+ * shape works for both.
  */
 export interface WebhookRawRequest {
   body: Uint8Array | ArrayBuffer;

packages/core/src/fusor/webhook.ts

@@ -0,0 +1,123 @@
+import z from "zod";
+import { mergeHeaderValue, type ParsedHttpRequest } from "./parse";
+
+export const FUSOR_DELIVERY_CE_TYPE = "dev.spctrm.fusor.delivery";
+
+const CANONICAL_BASE64 =
+  /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/;
+
+const canonicalBase64Schema = z.string().refine((value) => {
+  if (!CANONICAL_BASE64.test(value)) {
+    return false;
+  }
+  try {
+    const binary = atob(value);
+    return btoa(binary) === value;
+  } catch {
+    return false;
+  }
+}, "expected canonical padded base64");
+
+const requestFields = {
+  method: z.string().min(1),
+  path: z.string().min(1),
+  headers: z.record(z.string(), z.string()),
+  rawBodyBase64: canonicalBase64Schema,
+};
+
+const fusorWebhookRequestSchema = z.discriminatedUnion("bodyEncoding", [
+  z.looseObject({
+    ...requestFields,
+    bodyEncoding: z.literal("json"),
+    body: z.json(),
+  }),
+  z.looseObject({
+    ...requestFields,
+    bodyEncoding: z.literal("form"),
+    body: z.record(z.string(), z.union([z.string(), z.array(z.string())])),
+  }),
+  z.looseObject({
+    ...requestFields,
+    bodyEncoding: z.literal("text"),
+    body: z.string(),
+  }),
+  z.looseObject({
+    ...requestFields,
+    bodyEncoding: z.literal("base64"),
+    body: canonicalBase64Schema,
+  }),
+]);
+
+const fusorWebhookEnvelopeSchema = z
+  .looseObject({
+    schemaVersion: z.literal(1),
+    eventId: z.string().min(1),
+    projectId: z.string().min(1),
+    platform: z.string().min(1),
+    receivedAt: z.iso.datetime({ offset: true }).optional(),
+    sourceId: z.string().min(1).optional(),
+    prevSubjectSeq: z.number().int().nonnegative().safe(),
+    request: fusorWebhookRequestSchema,
+  })
+  .superRefine((envelope, context) => {
+    if (
+      envelope.request.bodyEncoding === "base64" &&
+      envelope.request.body !== envelope.request.rawBodyBase64
+    ) {
+      context.addIssue({
+        code: "custom",
+        path: ["request", "body"],
+        message: "base64 body must equal rawBodyBase64",
+      });
+    }
+  });
+
+export interface FusorWebhookEvent {
+  eventId: string;
+  platform: string;
+  request: ParsedHttpRequest;
+}
+
+const decodeCanonicalBase64 = (value: string): Uint8Array => {
+  const binary = atob(value);
+  return Uint8Array.from(binary, (character) => character.charCodeAt(0));
+};
+
+const normalizeHeaders = (
+  input: Record<string, string>
+): Record<string, string> => {
+  const headers: Record<string, string> = Object.create(null) as Record<
+    string,
+    string
+  >;
+  for (const [name, value] of Object.entries(input)) {
+    mergeHeaderValue(headers, name, value);
+  }
+  return headers;
+};
+
+/**
+ * Parses the versioned JSON envelope delivered by Fusor over HTTP. The
+ * normalized `body` arm is validation/debugging data; provider verification
+ * always receives the exact original bytes from `rawBodyBase64`.
+ */
+export const decodeFusorWebhookEvent = (
+  bodyBytes: Uint8Array
+): FusorWebhookEvent | null => {
+  try {
+    const json = new TextDecoder("utf-8", { fatal: true }).decode(bodyBytes);
+    const envelope = fusorWebhookEnvelopeSchema.parse(JSON.parse(json));
+    return {
+      eventId: envelope.eventId,
+      platform: envelope.platform,
+      request: {
+        method: envelope.request.method,
+        path: envelope.request.path,
+        headers: normalizeHeaders(envelope.request.headers),
+        rawBody: decodeCanonicalBase64(envelope.request.rawBodyBase64),
+      },
+    };
+  } catch {
+    return null;
+  }
+};