Event webhooks

import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";

export default defineSchema({
  oaoEvents: defineTable({
    eventId: v.string(),
    projectPosition: v.number(),
    kind: v.string(),
    aggregateType: v.string(),
    aggregateId: v.string(),
    publicPayload: v.any(),
    occurredAt: v.string(),
  })
    .index("by_event_id", ["eventId"])
    .index("by_aggregate", ["aggregateType", "aggregateId"]),
  oaoMessages: defineTable({
    messageId: v.string(),
    runId: v.string(),
    role: v.string(),
    content: v.string(),
  })
    .index("by_message_id", ["messageId"])
    .index("by_run", ["runId"]),
});

import { v } from "convex/values";
import { internalMutation } from "./_generated/server";

export const ingest = internalMutation({
  // Events keep OAO's public shape; see the delivery format above.
  args: { events: v.array(v.any()) },
  handler: async (ctx, { events }) => {
    for (const event of events) {
      const seen = await ctx.db
        .query("oaoEvents")
        .withIndex("by_event_id", (q) => q.eq("eventId", event.id))
        .unique();
      // OAO delivers at least once; a retried batch repeats events.
      if (seen) continue;
      await ctx.db.insert("oaoEvents", {
        eventId: event.id,
        projectPosition: Number(event.projectPosition),
        kind: event.kind,
        aggregateType: event.aggregateType,
        aggregateId: event.aggregateId,
        publicPayload: event.publicPayload,
        occurredAt: event.occurredAt,
      });
      if (event.message) {
        await ctx.db.insert("oaoMessages", {
          messageId: event.message.id,
          runId: event.message.runId,
          role: event.message.role,
          content: event.message.content,
        });
      }
    }
  },
});

import { httpRouter } from "convex/server";
import { internal } from "./_generated/api";
import { httpAction } from "./_generated/server";
import { verifyOaoSignature } from "./oaoSignature";

const http = httpRouter();

http.route({
  path: "/oao/events",
  method: "POST",
  handler: httpAction(async (ctx, request) => {
    const body = await request.text();
    const valid = await verifyOaoSignature({
      body,
      webhookId: request.headers.get("webhook-id"),
      webhookTimestamp: request.headers.get("webhook-timestamp"),
      webhookSignature: request.headers.get("webhook-signature"),
      secret: process.env.OAO_WEBHOOK_SECRET!,
    });
    if (!valid) return new Response("Invalid signature", { status: 401 });
    const batch = JSON.parse(body);
    // A thrown error returns 500, so OAO retries the same batch.
    await ctx.runMutation(internal.oao.ingest, { events: batch.data.events });
    return new Response(null, { status: 204 });
  }),
});

export default http;