feat: CommonKeyValueDaoMemoCacheCfg.ttl

Author: kirillgroshkov

src/commondao/common.dao.ts

@@ -18,6 +18,7 @@ import {
   ErrorMode,
   JsonSchemaObject,
   JsonSchemaRootObject,
+  nowUnix,
   ObjectWithId,
   pMap,
   SKIP,
@@ -606,7 +607,7 @@ export class CommonDao<BM extends BaseDBEntity, DBM extends BaseDBEntity = BM> {
    * "Returns", just to have a type of "Saved"
    */
   assignIdCreatedUpdated<T extends BaseDBEntity>(obj: Partial<T>, opt: CommonDaoOptions = {}): T {
-    const now = Math.floor(Date.now() / 1000)
+    const now = nowUnix()
 
     if (this.cfg.useCreatedProperty) {
       obj.created ||= obj.updated || now

src/kv/commonKeyValueDao.ts

@@ -2,7 +2,11 @@ import { AppError, CommonLogger, KeyValueTuple, pMap } from '@naturalcycles/js-l
 import { deflateString, inflateToString, ReadableTyped } from '@naturalcycles/nodejs-lib'
 import { CommonDaoLogLevel } from '../commondao/common.dao.model'
 import { CommonDBCreateOptions } from '../db.model'
-import { CommonKeyValueDB, KeyValueDBTuple } from './commonKeyValueDB'
+import {
+  CommonKeyValueDB,
+  CommonKeyValueDBSaveBatchOptions,
+  KeyValueDBTuple,
+} from './commonKeyValueDB'
 
 export interface CommonKeyValueDaoCfg<T> {
   db: CommonKeyValueDB
@@ -44,6 +48,8 @@ export interface CommonKeyValueDaoCfg<T> {
   deflatedJsonValue?: boolean
 }
 
+export type CommonKeyValueDaoSaveOptions = CommonKeyValueDBSaveBatchOptions
+
 // todo: logging
 // todo: readonly
 
@@ -133,13 +139,13 @@ export class CommonKeyValueDao<T> {
     } as T
   }
 
-  async patch(id: string, patch: Partial<T>): Promise<T> {
+  async patch(id: string, patch: Partial<T>, opt?: CommonKeyValueDaoSaveOptions): Promise<T> {
     const v: T = {
       ...(await this.getByIdOrEmpty(id)),
       ...patch,
     }
 
-    await this.save(id, v)
+    await this.save(id, v, opt)
 
     return v
   }
@@ -158,31 +164,35 @@ export class CommonKeyValueDao<T> {
     return await this.cfg.db.getByIds(this.cfg.table, ids)
   }
 
-  async save(id: string, value: T): Promise<void> {
-    await this.saveBatch([[id, value]])
+  async save(id: string, value: T, opt?: CommonKeyValueDaoSaveOptions): Promise<void> {
+    await this.saveBatch([[id, value]], opt)
   }
 
-  async saveAsBuffer(id: string, value: Buffer): Promise<void> {
-    await this.cfg.db.saveBatch(this.cfg.table, [[id, value]])
+  async saveAsBuffer(id: string, value: Buffer, opt?: CommonKeyValueDaoSaveOptions): Promise<void> {
+    await this.cfg.db.saveBatch(this.cfg.table, [[id, value]], opt)
   }
 
-  async saveBatch(entries: KeyValueTuple<string, T>[]): Promise<void> {
+  async saveBatch(
+    entries: KeyValueTuple<string, T>[],
+    opt?: CommonKeyValueDaoSaveOptions,
+  ): Promise<void> {
+    const { mapValueToBuffer } = this.cfg.hooks
     let bufferEntries: KeyValueDBTuple[]
 
-    if (!this.cfg.hooks.mapValueToBuffer) {
+    if (!mapValueToBuffer) {
       bufferEntries = entries as any
     } else {
-      bufferEntries = await pMap(entries, async ([id, v]) => [
-        id,
-        await this.cfg.hooks.mapValueToBuffer!(v),
-      ])
+      bufferEntries = await pMap(entries, async ([id, v]) => [id, await mapValueToBuffer(v)])
     }
 
-    await this.cfg.db.saveBatch(this.cfg.table, bufferEntries)
+    await this.cfg.db.saveBatch(this.cfg.table, bufferEntries, opt)
   }
 
-  async saveBatchAsBuffer(entries: KeyValueDBTuple[]): Promise<void> {
-    await this.cfg.db.saveBatch(this.cfg.table, entries)
+  async saveBatchAsBuffer(
+    entries: KeyValueDBTuple[],
+    opt?: CommonKeyValueDaoSaveOptions,
+  ): Promise<void> {
+    await this.cfg.db.saveBatch(this.cfg.table, entries, opt)
   }
 
   async deleteByIds(ids: string[]): Promise<void> {
@@ -204,19 +214,19 @@ export class CommonKeyValueDao<T> {
       return this.cfg.db.streamValues(this.cfg.table, limit) as ReadableTyped<T>
     }
 
-    const stream: ReadableTyped<T> = this.cfg.db
-      .streamValues(this.cfg.table, limit)
-      // .on('error', err => stream.emit('error', err))
-      .flatMap(async buf => {
+    return this.cfg.db.streamValues(this.cfg.table, limit).flatMap(
+      async buf => {
         try {
           return [await mapBufferToValue(buf)]
         } catch (err) {
           this.cfg.logger.error(err)
           return [] // SKIP
         }
-      })
-
-    return stream
+      },
+      {
+        concurrency: 16,
+      },
+    )
   }
 
   streamEntries(limit?: number): ReadableTyped<KeyValueTuple<string, T>> {
@@ -228,18 +238,18 @@ export class CommonKeyValueDao<T> {
       >
     }
 
-    const stream: ReadableTyped<KeyValueTuple<string, T>> = this.cfg.db
-      .streamEntries(this.cfg.table, limit)
-      // .on('error', err => stream.emit('error', err))
-      .flatMap(async ([id, buf]) => {
+    return this.cfg.db.streamEntries(this.cfg.table, limit).flatMap(
+      async ([id, buf]) => {
         try {
           return [[id, await mapBufferToValue(buf)]]
         } catch (err) {
           this.cfg.logger.error(err)
           return [] // SKIP
         }
-      })
-
-    return stream
+      },
+      {
+        concurrency: 16,
+      },
+    )
   }
 }

src/kv/commonKeyValueDaoMemoCache.ts

@@ -1,6 +1,15 @@
-import { AsyncMemoCache, MISS } from '@naturalcycles/js-lib'
+import { AsyncMemoCache, MISS, nowUnix, NumberOfSeconds } from '@naturalcycles/js-lib'
 import { CommonKeyValueDao } from './commonKeyValueDao'
 
+export interface CommonKeyValueDaoMemoCacheCfg<VALUE> {
+  dao: CommonKeyValueDao<VALUE>
+
+  /**
+   * If set, every `set()` will set `expireAt` (TTL) option.
+   */
+  ttl?: NumberOfSeconds
+}
+
 /**
  * AsyncMemoCache implementation, backed by CommonKeyValueDao.
  *
@@ -10,14 +19,16 @@ import { CommonKeyValueDao } from './commonKeyValueDao'
  * clear the whole table/cache.
  */
 export class CommonKeyValueDaoMemoCache<VALUE = any> implements AsyncMemoCache<string, VALUE> {
-  constructor(private dao: CommonKeyValueDao<VALUE>) {}
+  constructor(private cfg: CommonKeyValueDaoMemoCacheCfg<VALUE>) {}
 
   async get(k: string): Promise<VALUE | typeof MISS> {
-    return (await this.dao.getById(k)) || MISS
+    return (await this.cfg.dao.getById(k)) || MISS
   }
 
   async set(k: string, v: VALUE): Promise<void> {
-    await this.dao.save(k, v)
+    const opt = this.cfg.ttl ? { expireAt: nowUnix() + this.cfg.ttl } : undefined
+
+    await this.cfg.dao.save(k, v, opt)
   }
 
   async clear(): Promise<void> {

src/model.util.ts

@@ -1,10 +1,10 @@
-import { CreatedUpdated, CreatedUpdatedId } from '@naturalcycles/js-lib'
+import { CreatedUpdated, CreatedUpdatedId, nowUnix } from '@naturalcycles/js-lib'
 import { stringId } from '@naturalcycles/nodejs-lib'
 
 export function createdUpdatedFields(
   existingObject?: Partial<CreatedUpdated> | null,
 ): CreatedUpdated {
-  const now = Math.floor(Date.now() / 1000)
+  const now = nowUnix()
   return {
     created: existingObject?.created || now,
     updated: now,
@@ -14,7 +14,7 @@ export function createdUpdatedFields(
 export function createdUpdatedIdFields(
   existingObject?: Partial<CreatedUpdatedId> | null,
 ): CreatedUpdatedId {
-  const now = Math.floor(Date.now() / 1000)
+  const now = nowUnix()
   return {
     created: existingObject?.created || now,
     id: existingObject?.id || stringId(),

yarn.lock

@@ -836,9 +836,9 @@
     typescript "^5.0.2"
 
 "@naturalcycles/dev-lib@^13.0.0":
-  version "13.51.1"
-  resolved "https://registry.yarnpkg.com/@naturalcycles/dev-lib/-/dev-lib-13.51.1.tgz#0b96023d5cdccff5f844296ed03edbd0c7eb3d14"
-  integrity sha512-kKul83X8hysRzeyx9KbjmoStYYFG7+zbBebEyb1WEFTdFRjx6W509pVCrqxmXSS7z7Zit0O6dVT1u5EmqQwTvA==
+  version "13.51.2"
+  resolved "https://registry.yarnpkg.com/@naturalcycles/dev-lib/-/dev-lib-13.51.2.tgz#7f8ea46a936849051527a958a6b17d3e970287b0"
+  integrity sha512-N1TUk/F7iFlrtHqlNdABD0bJ3j2D9+EutH1jb7xFKg6b0R2SOEGmUr+OKG8GL75FwkJ2DEBtPXfdIbxwfRqnBw==
   dependencies:
     "@commitlint/cli" "^19.0.0"
     "@commitlint/config-conventional" "^19.0.0"
@@ -871,9 +871,9 @@
     yargs "^17.0.0"
 
 "@naturalcycles/js-lib@^14.0.0", "@naturalcycles/js-lib@^14.116.0":
-  version "14.219.2"
-  resolved "https://registry.yarnpkg.com/@naturalcycles/js-lib/-/js-lib-14.219.2.tgz#95cc0e875e3befba83f541856ecb84b0d94667a5"
-  integrity sha512-m68SFGyDT6s/j9kOxLsQs/5kfy4xWdJyyw/RIdR8m2svstA0bkZsgS4xYCaIX+onXDQWyEPKKH8EgqfUXPhHIA==
+  version "14.221.0"
+  resolved "https://registry.yarnpkg.com/@naturalcycles/js-lib/-/js-lib-14.221.0.tgz#cfd70eec1c2678f202f0982da18ff34bf7abaf6e"
+  integrity sha512-jF00+CgTxIjwgQNGFjoZKsjrnhs5sz4CTsH0RTQfODRWfvHwld7XqY1Tzc7Rf/Patb4hTH2IwtHvGGczhPj6Zw==
   dependencies:
     tslib "^2.0.0"
     zod "^3.20.2"
@@ -2377,9 +2377,9 @@ eslint-plugin-import@^2.22.1:
     tsconfig-paths "^3.15.0"
 
 eslint-plugin-jest@^28.0.0:
-  version "28.0.0"
-  resolved "https://registry.yarnpkg.com/eslint-plugin-jest/-/eslint-plugin-jest-28.0.0.tgz#b18f22977c3c216de928eeae6643c231c5b47316"
-  integrity sha512-FHiVI/nMYy48juLJKIt34MWPemvZyl0XT8JC3HTiUu/jgKJzoGgrNTCsyq4DzMlEjPZfmXKc0ogIzfrm6DJEuQ==
+  version "28.2.0"
+  resolved "https://registry.yarnpkg.com/eslint-plugin-jest/-/eslint-plugin-jest-28.2.0.tgz#863e2b2bda95eb41981ba9bcf4c44f57dce40a73"
+  integrity sha512-yRDti/a+f+SMSmNTiT9/M/MzXGkitl8CfzUxnpoQcTyfq8gUrXMriVcWU36W1X6BZSUoyUCJrDAWWUA2N4hE5g==
   dependencies:
     "@typescript-eslint/utils" "^6.0.0"