Deterministic Hook Ids only work once, is this by design?

[brad-decker]

Scenario:

1. When a webhook is called, first call resumeHook with a deterministic ID to determine if we are currently waiting for responses.
2. If we receive null back from this, then start the workflow (which waits for the hook, and sleeps 30 seconds)
3. Subsequent calls to the webhook call the resumeHook method, which adds to the batch running the workflow.
4. After thirty seconds collect all the responses.

This works as written; however, after this workflow is successful, subsequent calls to resumeHook return the hook created by the successfully completed workflow and then are skipped.

Is this by design?

My use case is that I have a webhook for analytics events. Each analytics event has an anonymousId (generated by third party, guaranteed uniqueness) and a userId if the user is known by our system. It also includes a number of other fields. I want to be able to achieve the following:

1. Batch events by user and process them together in a workflow within a window to avoid excessive write attempts to our Salesforce instance.
2. When an anonymous event is received, I want to wait for a separate identity event and then process however many payloads were received for that anonymous user at once.

I think I have achieved what I want for the first instance of each anonymous id or user id (used in the token name) but once the first flow completes they'll never run again.

Also in the docs it looks like webhooks can be passed to steps for use, but when I try to pass a hook to a step for use it errors out and you can't create hooks in steps so it feels like I'm forced to put most of my logic for these queues into the main workflow. Is that accurate?

[brad-decker]

I should add that I have only tested locally. After reviewing the code for world-local, it definitely returns the first hook it finds that matches the token. I am unfamiliar with the hooks api for Vercel, so I am unsure what the world-vercel getHookByToken function would return. I am guessing that the order of filenames predetermines that the first hook with the token will be the one always returned.

If it is not the expected behavior that deterministic token names might be reused by subsequent workflow runs for the same user/channel/or whatever that determines the token name then is there another flow that might be suggested for my use cases? It's also plausible to add an option to the resumeHook method that prefers hooks attached to currently running workflows over those that have already completed.

[brad-decker]

I think I understand a little bit more. getHookByToken, nor resumeHook, perform checks to determine whether the attached workflow run has completed. It queues it as it should, and code elsewhere sees the workflow this hook is attached to is done, and so skips it.

I think fundamentally the issue at the core of this is my desire to batch requests in short periods together. If I were to process each and every webhook received from my analytics provider, then having a workflow per action payload wouldn't be a holdup. However, there are a number of operations we take in post-processing that I need to think about.

Though I heard on your RFCs batching is a common request you're hearing so maybe I'll wait for some insight on that topic before I dive further.

[pranaygp]

@brad-decker thanks for your patience here

Right, so the idea with hook and webhook is that the tokens are globally unique. You can resume the same hook/webhook multiple times, and each one will get added to the event log. The workflow itself can process all events hitting a hook/webhook because that API is just an AsyncIterator (docs link with example)

It seems to me you've gotten that far and have been able to start collecting requests and batching to send into salesforce (fwiw, the idea for webhooks did spawn from me facing a very similar use case to what you're trying to build) - however it seems like there's som discrepancy between the local and vercel world implementations of how resumeHook should work. The ideal spec is that resumeHook works while the workflow is in the running state. once it's complete, that hook ID should no longer work (the webhook version automatically returns "HTTP 410 Gone" for instance)

With that in mind, do you mind sharing a little code example of what you're trying to do and are struggling with. That might help us repro and see if there's a bug in the implementation or if it's usage/documentation, or if maybe the spec itself needs to be changed here

[brad-decker]

however it seems like there's som discrepancy between the local and vercel world implementations of how resumeHook should work.

I am not sure, I think I just didn't understand the nature of hooks. If every token is intended to be unique then what I was trying to do is flawed. I'll share some more details starting with my rough flowchart of what I am trying to build (or rather refactor from its current state using redis queues outside of vercel. We are a new customer and working on migrating)

Problem

Essentially, Rudderstack (A Segment competitor) sends payloads to our webhook for every analytics event fired on our site. Each of these events has several properties we care about, including the name of the action and a score indicating how valuable it is. In addition, there is a campaign ID that points to a campaign in Salesforce.

While it is theoretically plausible to simply process each request in its own flow, that would result in about three writes to Salesforce per event coming in (1 to write contact updates, 1 to write the action, and 1 to write a campaign member for purposes of ranking on individual campaigns). Presently, we use Redis queues and workers that run every second, taking the first user out of the queue along with all of their associated actions, and then write the batch of actions in a bulk write.

Additionally, some events are take by anonymous users who we cannot yet write to Salesforce as we don't know enough about them. In this event they stay in the queue until we get a subsequent identify payload from Rudderstack. When that happens we move their actions into the appropriate queue in redis and put them in the user queue behind the last known user inserted.

Attempted Solution

I was attempting to take advantage of the fact that resumeHook returns null when there isn't a hook associated with the token. The idea was to run resumeHook first, if the return value is null **then and only then start the workflow. The idea being that we'd create or find a contact once, and then collect events as they come in within a given window to write a series of actions in batch. This approach I see now is fundamentally flawed given that tokens must be unique. For my solution to work we'd need to be able to reuse a deterministic id so that once the workflow is complete we can start another one and create a new hook for waiting for batch timeouts to complete.

Trial Code

I set out to just tinker with stuff to learn more, so this code is just test code but in the interest of hoping to find a better way to tackle the problem here is my monster:

test.ts

import { createHook, sleep } from 'workflow';

export async function testWorkflow(payload: {
  userId?: string;
  anonymousId: string;
  message: string;
}) {
  'use workflow';

  let payloads = [payload];

  if (payload.userId) {
    const userHook = createHook<{
      userId?: string;
      anonymousId: string;
      message: string;
    }>({
      token: `user_message_hook:${payload.userId}`,
    });

    let stop = false;

    await sleep('10 seconds').then(() => {
      console.log('Stopping hook after timeout');
      stop = true;
      console.log('stop set to true', stop);
    });

    for await (const userPayload of userHook) {
      console.log('received event', stop);
      // Keep the hook alive
      payloads.push(userPayload);
      if (stop) {
        console.log('STOP');
        break;
      }
    }

    console.log('payloads', payloads);
  } else {
    const userHook = createHook<{
      userId?: string;
      anonymousId: string;
      message: string;
    }>({
      token: `user_message_hook:${payload.anonymousId}`,
    });

    const anonHook = createHook<{
      message: string;
      userId?: string;
      anonymousId: string;
    }>({
      token: `anon_message_hook:${payload.anonymousId}`,
    });
    let done = false;
    let knownUserId: string;

    anonHook.then((msg) => {
      if (msg.userId) {
        knownUserId = msg.userId;
        done = true;
      }
    });

    for await (const hookPayload of userHook) {
      console.log('Received from anonymous user:', hookPayload.message);
      payloads.push(hookPayload);
      if (done) break;
    }

    payloads = payloads.map((p) => ({
      ...p,
      userId: knownUserId || p.userId,
    }));
  }

  payloads.forEach((p) => {
    console.log(`Message from user ${p.userId || 'anonymous'}: ${p.message}`);
  });

  return 'done';
}

/api/ea/v2/route.ts

import { getRun, resumeHook, start } from 'workflow/api';
import { testWorkflow } from '@/workflows/test';
import { NextResponse } from 'next/server';

async function startWorkFlowAndPostMessage(payload: {
  anonymousId: string;
  message: string;
  userId?: string;
}) {
  await start(testWorkflow, [payload]);
  const res = await resumeHook<{
    message: string;
    userId?: string;
    anonymousId: string;
  }>(`user_message_hook:${payload.userId ?? payload.anonymousId}`, payload);
  console.log(res);
}

export async function POST(request: Request) {
  const { anonymousId, message, userId } = await request.json();

  const payload = { anonymousId, message, userId };

  if (anonymousId && userId) {
    await resumeHook<{ message: string; userId: string }>(
      `anon_message_hook:${anonymousId}`,
      { message, userId },
    );
  }

  const hookResponse = await resumeHook<{
    message: string;
    userId?: string;
    anonymousId: string;
  }>(`user_message_hook:${userId ?? anonymousId}`, payload);

  if (hookResponse === null) {
    console.log('Starting workflow for user signup');
    // Executes asynchronously and doesn't block your app
    await startWorkFlowAndPostMessage(payload);
  } else {
    const run = await getRun(hookResponse.runId);
    if ((await run.status) === 'completed') {
      await startWorkFlowAndPostMessage(payload);
    }
  }

  return NextResponse.json({
    message: 'User signup workflow started',
  });
}

thanks for your patience here

Thank you for your patience with me! This solution is mind-blowing in its potential for our team, so I am privileged to be getting familiar with it at this stage in development. Thank you for your time and resposnes.