[KV Feedback]: Jobs using KV and Queue with Limits

[CodingXD]

🔍

• Did you search for existing issues?

Type of feedback

Feature request

Description

Is there a way to send items to the queue while they only get processed based on a limit (like 10 or 15 at a time or concurrently)?

This is what I'm currently doing, but is there a better way or a native way to do this?

import { Hono } from "https://deno.land/x/hono/mod.ts";
import { z } from "https://deno.land/x/zod/mod.ts";
import { zValidator } from "npm:@hono/zod-validator";

const app = new Hono();
const kv = await Deno.openKv();
const MAX_JOBS = 3;

app.post(
  "/",
  zValidator(
    "json",
    z.object({
      id: z.number(),
      name: z.string(),
    }),
  ),
  async (c) => {
    const body = c.req.valid('json');
    await kv.enqueue(body);
    return c.json({ message: "Added to queue" });
  },
);

Deno.serve(app.fetch);

const stream = kv.watch([["activeJobs"]]);
for await (const entries of stream) {
  // check if we're already processing more than 3 jobs
  const activeJobs = (entries[0].value as number) || 0;
  if (activeJobs <= MAX_JOBS) {
    // if not, listen to queue for more jobs
    console.log("listening for jobs, active jobs:", activeJobs);
    kv.listenQueue(async (msg) => {
      // once you get a job, increment the activeJobs by 1
      console.log("msg:", msg);
      await kv.atomic().sum(["activeJobs"], 1n).commit();

      // will run some logic and once done, reduce the activejobs count
      // to accept another job from queue
      setTimeout(async () => {
        await kv.atomic().sum(["activeJobs"], -1n).commit();
      }, 4000);
    });
  }
}

Steps to reproduce (if applicable)

No response

Expected behavior (if applicable)

No response

Possible solution (if applicable)

No response

Additional context

No response