Test with spies

packages/backend-core/src/cache/docWritethrough.ts

@@ -1,11 +1,11 @@
+import { AnyDocument, Database, LockName, LockType } from "@budibase/types"
 import BaseCache from "./base"
 import { getDocWritethroughClient } from "../redis/init"
-import { AnyDocument, Database, LockName, LockType } from "@budibase/types"
 
 import { JobQueue, createQueue } from "../queue"
 import * as dbUtils from "../db"
 import { Duration, newid } from "../utils"
-import { context, locks } from ".."
+import { locks } from ".."
 
 let CACHE: BaseCache | null = null
 async function getCache() {
@@ -36,26 +36,12 @@ export const docWritethroughProcessorQueue = createQueue<ProcessDocMessage>(
   }
 )
 
-docWritethroughProcessorQueue.process(async message => {
-  const { cacheKeyPrefix, messageId } = message.data
+class DocWritethroughProcessor {
+  init() {
+    docWritethroughProcessorQueue.process(async message => {
+      const { cacheKeyPrefix, messageId } = message.data
 
-  const cache = await getCache()
-  const latestMessageId = await cache.get(
-    REDIS_KEYS(cacheKeyPrefix).LATEST_MESSAGE_ID
-  )
-  if (messageId !== latestMessageId) {
-    // Nothing to do, another message overrode it
-    return
-  }
-
-  const lockResponse = await locks.doWithLock(
-    {
-      type: LockType.TRY_TWICE,
-      name: LockName.PERSIST_DOC_WRITETHROUGH,
-      resource: cacheKeyPrefix,
-      ttl: Duration.fromSeconds(60).toMs(),
-    },
-    async () => {
+      const cache = await getCache()
       const latestMessageId = await cache.get(
         REDIS_KEYS(cacheKeyPrefix).LATEST_MESSAGE_ID
       )
@@ -64,56 +50,77 @@ docWritethroughProcessorQueue.process(async message => {
         return
       }
 
-      await persistToDb(cache, message.data)
-      console.log("DocWritethrough persisted", { data: message.data })
+      const lockResponse = await locks.doWithLock(
+        {
+          type: LockType.TRY_TWICE,
+          name: LockName.PERSIST_DOC_WRITETHROUGH,
+          resource: cacheKeyPrefix,
+          ttl: Duration.fromSeconds(60).toMs(),
+        },
+        async () => {
+          const latestMessageId = await cache.get(
+            REDIS_KEYS(cacheKeyPrefix).LATEST_MESSAGE_ID
+          )
+          if (messageId !== latestMessageId) {
+            // Nothing to do, another message overrode it
+            return
+          }
 
-      await cache.deleteIfValue(
-        REDIS_KEYS(cacheKeyPrefix).LATEST_MESSAGE_ID,
-        latestMessageId
+          await this.persistToDb(cache, message.data)
+          console.log("DocWritethrough persisted", { data: message.data })
+
+          await cache.deleteIfValue(
+            REDIS_KEYS(cacheKeyPrefix).LATEST_MESSAGE_ID,
+            latestMessageId
+          )
+        }
       )
+
+      if (!lockResponse.executed) {
+        throw new Error(`Ignoring redlock conflict in write-through cache`)
+      }
+    })
+    return this
+  }
+
+  private async persistToDb(
+    cache: BaseCache,
+    {
+      dbName,
+      docId,
+      cacheKeyPrefix,
+    }: {
+      dbName: string
+      docId: string
+      cacheKeyPrefix: string
+    }
+  ) {
+    const db = dbUtils.getDB(dbName)
+    let doc: AnyDocument | undefined
+    try {
+      doc = await db.get(docId)
+    } catch {
+      doc = { _id: docId }
     }
-  )
 
-  if (!lockResponse.executed) {
-    throw new Error(`Ignoring redlock conflict in write-through cache`)
-  }
-})
+    const keysToPersist = await cache.keys(
+      REDIS_KEYS(cacheKeyPrefix).DATA.GET_ALL
+    )
+    for (const key of keysToPersist) {
+      const data = await cache.get(key, { useTenancy: false })
+      doc[data.key] = data.value
+    }
 
-export async function persistToDb(
-  cache: BaseCache,
-  {
-    dbName,
-    docId,
-    cacheKeyPrefix,
-  }: {
-    dbName: string
-    docId: string
-    cacheKeyPrefix: string
-  }
-) {
-  const db = dbUtils.getDB(dbName)
-  let doc: AnyDocument | undefined
-  try {
-    doc = await db.get(docId)
-  } catch {
-    doc = { _id: docId }
-  }
+    await db.put(doc)
 
-  const keysToPersist = await cache.keys(
-    REDIS_KEYS(cacheKeyPrefix).DATA.GET_ALL
-  )
-  for (const key of keysToPersist) {
-    const data = await cache.get(key, { useTenancy: false })
-    doc[data.key] = data.value
-  }
-
-  await db.put(doc)
-
-  for (const key of keysToPersist) {
-    await cache.delete(key, { useTenancy: false })
+    for (const key of keysToPersist) {
+      await cache.delete(key, { useTenancy: false })
+    }
   }
 }
 
+export const processor = new DocWritethroughProcessor().init()
+
 export class DocWritethrough {
   private db: Database
   private _docId: string

packages/backend-core/src/cache/tests/docWritethrough.spec.ts

@@ -1,12 +1,11 @@
+import _ from "lodash"
 import { DBTestConfiguration, generator, structures } from "../../../tests"
 import { getDB } from "../../db"
-import _ from "lodash"
 
-import {
-  DocWritethrough,
-  docWritethroughProcessorQueue,
-} from "../docWritethrough"
+import { DocWritethrough, processor } from "../docWritethrough"
+
 import InMemoryQueue from "../../queue/inMemoryQueue"
+import { docWritethroughProcessorQueue } from "../docWritethrough"
 
 const WRITE_RATE_MS = 1000
 
@@ -240,12 +239,13 @@ describe("docWritethrough", () => {
           )
         )
       }
-
+      const persistToDbSpy = jest.spyOn(processor as any, "persistToDb")
       const storeToCacheSpy = jest.spyOn(docWritethrough as any, "storeToCache")
 
       await config.doInTenant(async () => {
         await parallelPatch(5)
         expect(storeToCacheSpy).toBeCalledTimes(5)
+        expect(persistToDbSpy).not.toBeCalled()
         expect(await db.exists(documentId)).toBe(false)
 
         await travelForward(WRITE_RATE_MS)
@@ -253,7 +253,7 @@ describe("docWritethrough", () => {
         await parallelPatch(40)
 
         expect(storeToCacheSpy).toBeCalledTimes(45)
-
+        expect(persistToDbSpy).toBeCalledTimes(1)
         // Ideally we want to spy on persistToDb from ./docWritethrough, but due our barrel files configuration required quite of a complex setup.
         // We are relying on the document being stored only once (otherwise we would have _rev updated)
         expect(await db.get(documentId)).toEqual(