[TRI-1275] Add `io.waitForNextEvent` that resumes once a matching event is found

[ericallam]

It would be nice to support the following io task in a job run:

const event = await io.waitForNextEvent("✋", { name: "opt-in", filter: { userId: [payload.userId] })

We could also add a continueAfter parameter that would return undefined if the event isn't received in time:

const event = await io.waitForNextEvent("✋", {
  name: "opt-in",
  filter: { userId: [payload.userId] },
  continueAfter: 60 * 10,
});

Type Safety

Currently the event payload type in eventTrigger is inferred via an optionally passed in schema:

eventTrigger({
    name: "foo.event",
    schema: z.object({
      userId: z.string(),
      delay: z.number(),
    }),
  })

We could add support for a schema to waitForNextEvent as well:

const event = await io.waitForNextEvent("✋", {
  name: "opt-in",
  filter: { userId: [payload.userId] },
  continueAfter: 60 * 10,
  schema: z.object({ userId: z.string() })
});

Once we add support for an Event Catalog (#251) the event payload type would be automatically inferred based on it's name

_TRI-1275

[Chigala]

I thought we were going to wait until there was a batch polling system up before this would be scalable? Or would the continueAfter have a very short time limit to prevent a lot of job executions?

[ericallam]

The Batch polling stuff is necessary for the waitUntil task but not this one. Don't have time to go into all the details about how it would work but basically when the "wait for next event" task is created in RunTaskService we'd create an EventDispatcher with a dispatchable of task instead of a job version. Then we'd update the InvokeDispatcherService to handle dispatching to a task and resuming the run with the output being the payload of the event, and then removing the EventDispatcher record. If the continueAfter param is included we'd also schedule a graphile job to run and it would check if the EventDispatcher still existed (meaning it hadn't be invoked). Then it would delete the event dispatcher and resume the run with a task output of null.

[ericallam]

I actually had this working at some point but removed it. You might be able to find it in the history

[Chigala]

Thanks for the explanation. I'll also check the history.