Any way to define a step as a queue?

[piccinnigius]

Hi @jumski , glad to here be again!

I'm tring to move my single worker code to pgflow.

type Input = {
  ingestionJobID: string;
};

export default new Flow<Input>({
  slug: "embeddings-flow",
})
  .step({ slug: "analyze" }, (input) =>
    analyzeDocument(input.run.ingestionJobID),
  )
  .step({ slug: "embed", dependsOn: ["analyze"] }, (input) =>
    embedDocument(input.analyze.chunks),
  );

In the "old" implementation I've managed concurrency in the following way, by chunking first in my API and then with a trigger on the table that pushes the single chunk ID in the embeddings-tasksqueue:

EdgeWorker.start(handler, {
  queueName: "embeddings-tasks",
  maxConcurrent: 5,
  maxPollSeconds: 5,
  pollIntervalMs: 200,
  retryDelay: 60,
  retryLimit: 5,
});

II saw that the complete flow runs inside a single worker (as it should be), my question is: how can I have a step that replicates the previous behavior? The embed step must take n chunks from the db (with the ids coming from analyze) and reprocess them in groups. I imagine that if there are many, the worker will incur in wall time and fails, do you think it makes sense to use this mechanism to continue processing until they are all completed? Are there better solutions that I can use?

Thanks again for your work.

[jumski]

Hey again!

Currently pgflow does not support processing arrays like this, but this is very wanted and planned feature.

I recommend keeping your handlers reusable (check this guide) so it will be easy to migrate when i implement this feature (it is coming!).

[piccinnigius]

I can't wait!  🚀