waitForEvent with custom events

[psteinroe]

Add first-class waitForEvent support for custom events, using the same event matching path as event-triggered tasks.

Design:

• Store emitted events in conductor_events.
• Process events through the internal queue in batches.
• Represent task triggers and pending waits as event subscriptions.
• Define filterable fields on event definitions.
• Derive filter value types from the event schema; do not duplicate field types in filterable.
• Use filterable only as an allowlist for top-level fields, or as aliases for nested schema paths.
• Allow filters only on declared filterable fields.
• Compile subscription filters into an indexed event_subscription_constraints table.
• Do not persist extracted event filter values initially; extract event values on the fly for the claimed batch.
• Match event batches to subscriptions set-wise in SQL, then bulk insert task executions or resume waiting executions.
• Keep DB trigger when as a DB-native optimization / escape hatch, not the universal filter language.

Example API:

const invoicePaid = conductor.defineEvent("invoice.paid", {
  schema: z.object({
    invoiceId: z.string(),
    accountId: z.string(),
    currency: z.enum(["EUR", "USD"]),
    total: z.number(),
    customer: z.object({
      country: z.string(),
    }),
  }),

  // Types are derived from the schema.
  filterable: ["invoiceId", "accountId", "currency", "total"],

  // Optional later extension for nested fields:
  // filterable: {
  //   invoiceId: "invoiceId",
  //   accountId: "accountId",
  //   country: "customer.country",
  //   total: "total",
  // },
});

await ctx.waitForEvent("wait-for-payment", {
  event: invoicePaid,
  filter: {
    invoiceId: [event.payload.invoiceId],
  },
  timeout: "3d",
});

Implementation outline:

1. Add durable event waits.
2. Add a unified event subscription table for task triggers and waits.
3. Add compiled subscription constraints with indexes.
4. Batch-process pending events.
5. Extract only declared filterable values from the claimed batch.
6. Match event batches against subscription constraints.
7. Insert matched task executions or resume matched waits transactionally.