Guide needed on using queues

[waptik]

So i haven't used the native queuing system(Deno.enqueue, Deno.listenQueue) before. But i'd like to see a realistic example of its usage both at the model(db.users.enqueue, db.users.listenQueue, db.users.deleteUndelivered, db.users.findUndelivered) and database(db.enqueue, db.listenQueue) levels in a live server such as Deno.serve or oakjs.
In the readme file, the model structure for users wasn't defined when it came to using queue related ops

If an example is not possible, a guide as to where to use what would be very much appreciated.

[waptik]

I tried out enqueue/listenQueue at the model level and found out that the data has nothing to do with the model's schema(no type error or runtime error).

[waptik]

It seems something has changed again with using type in v0.29.1(had to upgrade because kvdex was returning any in v0.28.3 so i wasn't getting type definitions for db.*).

Also, using interface by extending the KvObject type, vscode is screaming at me

[oliver-oloughlin]

Have you remembered to cache your dependencies after updating the version?

The interface issue I would guess is because any included interfaces must also extend KvObject, meaning User must extend KvObject.

[waptik]

Jeez, i forgot to do that but it didn’t help either. I initially started
with “deno cache —reload dev.ts” then moved on to using the command
palette to do the same thing but it didn’t seem to fix the problem.
Oh okay, too much trouble when using interfaces 🤦🏾‍♂️. I’ll probably have
to stick using interface for now even though there’ll be some extra work.

Update: after extending User interface with KvObject, it seems to have also fixed the issue with "type".

[oliver-oloughlin]

Re-caching like you described should have worked. Another alternative that I like to use is setting this in your /.vscode/settings.json:

{
  "deno.cacheOnSave": true,
}

If you are still having issues with the DB types, would you mind showing how the database schema is defined?

I pretty much always use types because in my opinion they are better in almost every use case, and can be used for everything that interfaces are used for. The issue you were facing is my main reason for avoiding interfaces: TypeScript doesn't recognize an interface as the correct type before it actually extends the specific type, even if the interface itself conforms to the type. Meanwhile TypeScript uses duck-typing for types, meaning if the signature is correct then it sees it as a matching type.

[waptik]

So i used to have cacheOnSave set to true but i had to disable it because i've been facing this issue
Oh okay, that's some nice understanding of things. There is little solid information about the differences between interfaces and types and most seem to say types and interfaces can be used interchangeably.

Here's a portion of the client definition and that of some collections

//------------------------------ collection schema defintions inside db/models.ts ------------------------------//
import { z } from "zod";

// @see https://stackoverflow.com/a/73825370
const withDefautTimestamps = z.object({
  created_at: z
    .string()
    .datetime()
    .default(() => new Date().toISOString()),
  updated_at: z
    .string()
    .datetime()
    .default(() => new Date().toISOString()),
});

export type IDBAudioVoice = z.infer<typeof AudioVoiceSchema>;

....

export const AudioVoiceSchema = z.object({
  // information from telegram
 ...
  file_unique_id: z.string().describe("unique"),

  key: z.string().describe("unique"),
  ...

  uploaded_by: z.number(),
  message_id: z.number(),
}).merge(withDefautTimestamps);
....

//------------------------------ db client definition in db/client.ts ------------------------------//
const dbClient = kvdex(kv, {
  users: collection(zodModel(UserSchema), {
    indices: {
      telegram_user_id: "primary", // unique
      id: "secondary", // non-unique
    },
    // history: true,
  }),
// some omitted
  audio_voices: collection(zodModel(AudioVoiceSchema), {
    indices: {
      file_unique_id: "primary", // unique
      file_id: "secondary", // non-unique
      key: "secondary", // non-unique
    },
    // history: true,
  }),
});

[waptik]

Hi again,

So i've come to face another issue regarding processing queues with .collection.listenQueue inside a fresh app. I'm not sure if it's a bug from fresh or kvdex so i'm also referencing the opened issue from fresh's repo here as well.
Ref: freshframework/fresh#2186

[oliver-oloughlin]

I see. I am happy to help anyway I can, and have a few thoughts:

1. I would start by trying to use the native kv.listenQueue() to see if there's any difference, just to identify if it is a kvdex specific issue or not.
2. Make sure you are invoking listenQueue() and enqueue() on the same collection (collection queues are treated as their own topics).
3. When are you running enqueue() compared to listenQueue()? I experienced an issue a little while back where if I fired an enqueue() very fast after listenQueue(), the queue lsitener had not been registered/created yet, and therefore didn't catch the message.
4. Make sure the data being queued is not undefined, this will lead to it not being delivered as expecetd due to how kvdex parses queue messages to deliver them to the correct topics.

[waptik]

1. As mentioned in the referenced issue, it wasn't specific to kvdex because the kv.listenQueue had no effect.
2. In the screenshots provided on the other side, you could see that they were being invoked on the same collection
3. enqueue() is being fired on-demand so it's after listenQueue() as it's defined in entrypoint file of fresh (main.ts)
4. I can assure you that the data passed to enqueue() is always defined as i made sure not to queue them if they are undefined
   I'm going to upload the same pics here again

• kvdex .collection.enqueue() and .collection.listenQueue()
  
  

• Native kv.enqueue() and kv.listenQueue()
  
  

[waptik]

I just deployed the example repo to deploy and it's working as it should. AS you can see from the screenshot below, listenQueue from both native kv and kvdex are able to receive data effectively so i think this is a fresh issue in local development with the dev.ts file

Thanks for your help!

[oliver-oloughlin]

Yeah, seems like it is actually an issue with fresh itself. Good job investigating it. I'll follow any developments in the issue you posted. I can't really help much with issues that come from fresh, but I'll keep an eye out for anything that needs to be changed with kvdex.

[waptik]

Sure, thanks again.

[ryanleecode]

I still don't understand why would we need queues for a specific model if the callback from listenQueue does not include a value from the model itself?

[oliver-oloughlin]

Really, it's just a matter of logical structuring. Collection-specific queues can be seen as their own topic.
If you were for example to have a lot of listeners doing tasks specific to a collection, it might be simpler to use collection queues instead of many top-level db queues with custom-set topics which require you to give the correct topic when enqueueing messages as well.