Merge pull request #82 from oliver-oloughlin/feature/cron-jobs

added cron jobs and queue topics

Author: oliver-oloughlin

README.md

@@ -1,17 +1,19 @@
 # kvdex
 
-`kvdex` is an ORM-like wrapper for Deno KV with zero third-party dependencies.
-It's purpose is to enhance the experience of using Deno's KV store through
-additional features such as indexing and typed collections, while maintaining as
-much of the native functionality as possible, such as atomic operations.
+`kvdex` is a high level abstraction layer for Deno KV with zero third-party
+dependencies. It's purpose is to enhance the experience of using Deno's KV store
+through additional features such as indexing, and strongly typed collections,
+and cron jobs, while maintaining as much of the native functionality as
+possible, like atomic operations and queue listeners.
 
 ## Highlights
 
 - CRUD operations for selected and ranged documents with strong typing.
 - Primary (unique) and secondary (non-unique) indexing.
 - Segmented storage for large objects that exceed the native size limit.
 - Support for pagination and filtering.
-- Message queues at database and collection level.
+- Repeating cron jobs.
+- Message queues at database and collection level with topics.
 - Support for atomic operations.
 
 ## Table of Contents
@@ -53,6 +55,7 @@ much of the native functionality as possible, such as atomic operations.
     - [enqueue()](#enqueue-1)
     - [listenQueue()](#listenqueue-1)
     - [findUndelivered()](#findundelivered-1)
+    - [cron()](#cron)
     - [atomic()](#atomic)
   - [Atomic Operations](#atomic-operations)
     - [Without checking](#without-checking)
@@ -432,26 +435,28 @@ const count = await db.users.count({
 
 ### enqueue()
 
-Add data (of any type) to the collection queue to be delivered to the queue
-listener via `db.collection.listenQueue()`. The data will only be received by
-queue listeners on the specified collection. The method takes an optional
-options argument that can be used to set a delivery delay.
+Add data to the collection queue to be delivered to the queue listener via
+`db.collection.listenQueue()`. The data will only be received by queue listeners
+on the specified collection and topic. The method takes an optional options
+argument that can be used to set a delivery delay and topic.
 
 ```ts
 // Immediate delivery
 await db.users.enqueue("some data")
 
 // Delay of 2 seconds before delivery
-await db.users.enqueue("some data", {
+await db.users.enqueue("cake", {
   delay: 2_000,
+  topic: "food",
 })
 ```
 
 ### listenQueue()
 
 Listen for data from the collection queue that was enqueued with
 `db.collection.enqueue()`. Will only receive data that was enqueued to the
-specific collection queue. Takes a handler function as argument.
+specific collection queue and topic. Expects a handler function as argument, as
+well as optional options that can be used to set the topic.
 
 ```ts
 // Prints the data to console when recevied
@@ -463,11 +468,11 @@ db.users.listenQueue(async (data) => {
 
   const res = await fetch("...", {
     method: "POST",
-    body: dataBody,
+    body: data,
   })
 
   console.log("POSTED:", dataBody, res.ok)
-})
+}, { topic: "posts" })
 ```
 
 ## Indexable Collection Methods
@@ -594,26 +599,28 @@ await db.deleteAll()
 
 ### enqueue()
 
-Add data (of any type) to the database queue to be delivered to the queue
-listener via `db.listenQueue()`. The data will only be received by queue
-listeners on the database queue. The method takes an optional options argument
-that can be used to set a delivery delay.
+Add data to the database queue to be delivered to the queue listener via
+`db.listenQueue()`. The data will only be received by queue listeners on the
+database queue and specified topic. The method takes an optional options
+argument that can be used to set a delivery delay and topic.
 
 ```ts
 // Immediate delivery
 await db.enqueue("some data")
 
 // Delay of 2 seconds before delivery
-await db.enqueue("some data", {
+await db.enqueue("cake", {
   delay: 2_000,
+  topic: "food",
 })
 ```
 
 ### listenQueue()
 
 Listen for data from the database queue that was enqueued with `db.enqueue()`.
-Will only receive data that was enqueued to the database queue. Takes a handler
-function as argument.
+Will only receive data that was enqueued to the database queue and specified
+topic. Expects a handler function as argument, as well as optional options that
+can be used to set the topic.
 
 ```ts
 // Prints the data to console when recevied
@@ -625,11 +632,11 @@ db.listenQueue(async (data) => {
 
   const res = await fetch("...", {
     method: "POST",
-    body: dataBody,
+    body: data,
   })
 
   console.log("POSTED:", dataBody, res.ok)
-})
+}, { topic: "posts" })
 ```
 
 ### findUndelivered()
@@ -646,6 +653,26 @@ const doc2 = await db.findUndelivered("undelivered_id", {
 })
 ```
 
+### cron()
+
+Create a cron job that will run on interval, either indefinitely or until an
+exit condition is met. If no interval is set, the next job will run immediately
+after the previous has finished. Like with queue listeners, there can be
+multiple cron jobs defined.
+
+```ts
+// Will repeat indeefinitely without delay
+db.cron(() => console.log("Hello World!"))
+
+// First job starts with a 10 second delay, after that there is a 5 second delay between jobs
+// Will terminate after the 10th run (count starts at 0), or if the job returns n < 0.25
+db.cron(() => Math.random(), {
+  startDelay: 10_000,
+  interval: 5_000,
+  exit: ({ count, result }) => count >= 10 || result < 0.25,
+})
+```
+
 ### atomic()
 
 Initiate an atomic operation. The method takes a selection function as argument

src/atomic_builder.ts

@@ -14,6 +14,7 @@ import type {
   KvValue,
   Model,
   Operations,
+  QueueValue,
   Schema,
   SchemaDefinition,
 } from "./types.ts"
@@ -437,9 +438,14 @@ export class AtomicBuilder<
     return this
   }
 
-  enqueue(data: KvValue, options?: EnqueueOptions) {
+  enqueue(data: QueueValue, options?: EnqueueOptions) {
     // Prepare and add enqueue operation
-    const prep = prepareEnqueue(this.collection._keys.baseKey, data, options)
+    const prep = prepareEnqueue(
+      this.collection._keys.baseKey,
+      data,
+      options,
+    )
+
     this.operations.atomic.enqueue(prep.msg, prep.options)
 
     // Return current AtomicBuilder

src/collection.ts

@@ -17,21 +17,22 @@ import type {
   KvObject,
   KvValue,
   ListOptions,
+  QueueListenerOptions,
   QueueMessageHandler,
+  QueueValue,
   SetOptions,
   UpdateData,
   UpdateManyOptions,
 } from "./types.ts"
 import {
   allFulfilled,
+  createHandlerId,
   createListSelector,
   extendKey,
   generateId,
   getDocumentId,
   isKvObject,
-  keyEq,
   kvGetMany,
-  parseQueueMessage,
   prepareEnqueue,
 } from "./utils.ts"
 import { Document } from "./document.ts"
@@ -45,7 +46,11 @@ export class Collection<
   const T1 extends KvValue,
   const T2 extends CollectionOptions<T1>,
 > {
+  private queueHandlers: Map<string, QueueMessageHandler<QueueValue>[]>
+  private idempotentListener: () => void
+
   protected kv: Deno.Kv
+
   readonly _idGenerator: IdGenerator<KvValue>
   readonly _keys: CollectionKeys
 
@@ -60,7 +65,17 @@ export class Collection<
    *
    * @param options - Collection options.
    */
-  constructor(kv: Deno.Kv, key: KvKey, options?: T2) {
+  constructor(
+    kv: Deno.Kv,
+    key: KvKey,
+    queueHandlers: Map<string, QueueMessageHandler<QueueValue>[]>,
+    idempotentListener: () => void,
+    options?: T2,
+  ) {
+    // Set reference to queue handlers and idempotent listener
+    this.queueHandlers = queueHandlers
+    this.idempotentListener = idempotentListener
+
     // Set the KV instance
     this.kv = kv
 
@@ -535,9 +550,15 @@ export class Collection<
    * @param options - Enqueue options, optional.
    * @returns - Promise resolving to Deno.KvCommitResult.
    */
-  async enqueue(data: KvValue, options?: EnqueueOptions) {
-    // Prepare and perform enqueue operation
-    const prep = prepareEnqueue(this._keys.baseKey, data, options)
+  async enqueue(data: QueueValue, options?: EnqueueOptions) {
+    // Prepare message and options for enqueue
+    const prep = prepareEnqueue(
+      this._keys.baseKey,
+      data,
+      options,
+    )
+
+    // Enqueue message with options
     return await this.kv.enqueue(prep.msg, prep.options)
   }
 
@@ -565,32 +586,23 @@ export class Collection<
    * ```
    *
    * @param handler - Message handler function.
+   * @param options - Queue listener options.
+   * @returns void.
    */
-  async listenQueue<T extends KvValue = KvValue>(
+  listenQueue<T extends QueueValue = QueueValue>(
     handler: QueueMessageHandler<T>,
+    options?: QueueListenerOptions,
   ) {
-    // Listen for kv queue messages
-    await this.kv.listenQueue(async (msg) => {
-      // Parse queue message
-      const parsed = parseQueueMessage<T>(msg)
-
-      // If failed parse, ignore
-      if (!parsed.ok) {
-        return
-      }
+    // Create handler id
+    const handlerId = createHandlerId(this._keys.baseKey, options?.topic)
 
-      // Destruct queue message
-      const { collectionKey, data } = parsed.msg
+    // Add new handler to specified handlers
+    const handlers = this.queueHandlers.get(handlerId) ?? []
+    handlers.push(handler as QueueMessageHandler<QueueValue>)
+    this.queueHandlers.set(handlerId, handlers)
 
-      // Check that collection key is set and matches current collection context
-      if (
-        Array.isArray(collectionKey) &&
-        keyEq(collectionKey, this._keys.baseKey)
-      ) {
-        // Invoke data handler
-        await handler(data)
-      }
-    })
+    // Activate idempotent listener
+    this.idempotentListener()
   }
 
   /**

src/collection_builder.ts

@@ -9,6 +9,8 @@ import type {
   LargeCollectionOptions,
   LargeKvValue,
   Model,
+  QueueMessageHandler,
+  QueueValue,
 } from "./types.ts"
 
 /**
@@ -51,10 +53,17 @@ class CollectionBuilder<const T extends KvValue> {
   build(
     options?: CollectionOptions<T>,
   ) {
-    return (kv: Deno.Kv, key: KvKey) =>
+    return (
+      kv: Deno.Kv,
+      key: KvKey,
+      queueHandlers: Map<string, QueueMessageHandler<QueueValue>[]>,
+      idempotentListener: () => void,
+    ) =>
       new Collection<T, CollectionOptions<T>>(
         kv,
         key,
+        queueHandlers,
+        idempotentListener,
         options,
       )
   }
@@ -73,10 +82,17 @@ class IndexableCollectionBuilder<const T extends Model> {
   build<const T2 extends IndexableCollectionOptions<T>>(
     options: T2,
   ) {
-    return (kv: Deno.Kv, key: KvKey) =>
+    return (
+      kv: Deno.Kv,
+      key: KvKey,
+      queueHandlers: Map<string, QueueMessageHandler<QueueValue>[]>,
+      idempotentListener: () => void,
+    ) =>
       new IndexableCollection<T, T2>(
         kv,
         key,
+        queueHandlers,
+        idempotentListener,
         options,
       )
   }
@@ -93,10 +109,17 @@ class LargeCollectionBuilder<const T extends LargeKvValue> {
    * @returns A LargeCollection instance.
    */
   build(options?: LargeCollectionOptions<T>) {
-    return (kv: Deno.Kv, key: KvKey) =>
+    return (
+      kv: Deno.Kv,
+      key: KvKey,
+      queueHandlers: Map<string, QueueMessageHandler<QueueValue>[]>,
+      idempotentListener: () => void,
+    ) =>
       new LargeCollection<T, LargeCollectionOptions<T>>(
         kv,
         key,
+        queueHandlers,
+        idempotentListener,
         options,
       )
   }

src/indexable_collection.ts

@@ -18,6 +18,8 @@ import type {
   ListOptions,
   Model,
   PrimaryIndexKeys,
+  QueueMessageHandler,
+  QueueValue,
   SecondaryIndexKeys,
   SetOptions,
   UpdateData,
@@ -64,9 +66,15 @@ export class IndexableCollection<
    *
    * @param options - Indexable Collection options.
    */
-  constructor(kv: Deno.Kv, key: KvKey, options: T2) {
+  constructor(
+    kv: Deno.Kv,
+    key: KvKey,
+    queueHandlers: Map<string, QueueMessageHandler<QueueValue>[]>,
+    idempotentListener: () => void,
+    options: T2,
+  ) {
     // Invoke super constructor
-    super(kv, key, options)
+    super(kv, key, queueHandlers, idempotentListener, options)
 
     // Set indexable collection keys
     this._keys = {