mirror of https://github.com/Budibase/budibase.git
Michael Drury
4 years ago
committed by
GitHub
GitHub
17 changed files with 573 additions and 125 deletions
@ -1,5 +1,5 @@ |
|||||
module.exports = { |
module.exports = { |
||||
Client: require("./src/redis"), |
Client: require("./src/redis"), |
||||
utils: require("./src/redis/utils"), |
utils: require("./src/redis/utils"), |
||||
clients: require("./src/redis/authRedis"), |
clients: require("./src/redis/init"), |
||||
} |
} |
||||
|
|||||
@ -0,0 +1,92 @@ |
|||||
|
import { getTenantId } from "../../context" |
||||
|
import redis from "../../redis/init" |
||||
|
import RedisWrapper from "../../redis" |
||||
|
|
||||
|
function generateTenantKey(key: string) { |
||||
|
const tenantId = getTenantId() |
||||
|
return `${key}:${tenantId}` |
||||
|
} |
||||
|
|
||||
|
export = class BaseCache { |
||||
|
client: RedisWrapper | undefined |
||||
|
|
||||
|
constructor(client: RedisWrapper | undefined = undefined) { |
||||
|
this.client = client |
||||
|
} |
||||
|
|
||||
|
async getClient() { |
||||
|
return !this.client ? await redis.getCacheClient() : this.client |
||||
|
} |
||||
|
|
||||
|
async keys(pattern: string) { |
||||
|
const client = await this.getClient() |
||||
|
return client.keys(pattern) |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Read only from the cache. |
||||
|
*/ |
||||
|
async get(key: string, opts = { useTenancy: true }) { |
||||
|
key = opts.useTenancy ? generateTenantKey(key) : key |
||||
|
const client = await this.getClient() |
||||
|
return client.get(key) |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Write to the cache. |
||||
|
*/ |
||||
|
async store( |
||||
|
key: string, |
||||
|
value: any, |
||||
|
ttl: number | null = null, |
||||
|
opts = { useTenancy: true } |
||||
|
) { |
||||
|
key = opts.useTenancy ? generateTenantKey(key) : key |
||||
|
const client = await this.getClient() |
||||
|
await client.store(key, value, ttl) |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Remove from cache. |
||||
|
*/ |
||||
|
async delete(key: string, opts = { useTenancy: true }) { |
||||
|
key = opts.useTenancy ? generateTenantKey(key) : key |
||||
|
const client = await this.getClient() |
||||
|
return client.delete(key) |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Read from the cache. Write to the cache if not exists. |
||||
|
*/ |
||||
|
async withCache( |
||||
|
key: string, |
||||
|
ttl: number, |
||||
|
fetchFn: any, |
||||
|
opts = { useTenancy: true } |
||||
|
) { |
||||
|
const cachedValue = await this.get(key, opts) |
||||
|
if (cachedValue) { |
||||
|
return cachedValue |
||||
|
} |
||||
|
|
||||
|
try { |
||||
|
const fetchedValue = await fetchFn() |
||||
|
|
||||
|
await this.store(key, fetchedValue, ttl, opts) |
||||
|
return fetchedValue |
||||
|
} catch (err) { |
||||
|
console.error("Error fetching before cache - ", err) |
||||
|
throw err |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
async bustCache(key: string, opts = { client: null }) { |
||||
|
const client = await this.getClient() |
||||
|
try { |
||||
|
await client.delete(generateTenantKey(key)) |
||||
|
} catch (err) { |
||||
|
console.error("Error busting cache - ", err) |
||||
|
throw err |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,59 @@ |
|||||
|
require("../../../tests/utilities/TestConfiguration") |
||||
|
const { Writethrough } = require("../writethrough") |
||||
|
const { dangerousGetDB } = require("../../db") |
||||
|
const tk = require("timekeeper") |
||||
|
|
||||
|
const START_DATE = Date.now() |
||||
|
tk.freeze(START_DATE) |
||||
|
|
||||
|
const DELAY = 5000 |
||||
|
|
||||
|
const db = dangerousGetDB("test") |
||||
|
const db2 = dangerousGetDB("test2") |
||||
|
const writethrough = new Writethrough(db, DELAY), writethrough2 = new Writethrough(db2, DELAY) |
||||
|
|
||||
|
describe("writethrough", () => { |
||||
|
describe("put", () => { |
||||
|
let first |
||||
|
it("should be able to store, will go to DB", async () => { |
||||
|
const response = await writethrough.put({ _id: "test", value: 1 }) |
||||
|
const output = await db.get(response.id) |
||||
|
first = output |
||||
|
expect(output.value).toBe(1) |
||||
|
}) |
||||
|
|
||||
|
it("second put shouldn't update DB", async () => { |
||||
|
const response = await writethrough.put({ ...first, value: 2 }) |
||||
|
const output = await db.get(response.id) |
||||
|
expect(first._rev).toBe(output._rev) |
||||
|
expect(output.value).toBe(1) |
||||
|
}) |
||||
|
|
||||
|
it("should put it again after delay period", async () => { |
||||
|
tk.freeze(START_DATE + DELAY + 1) |
||||
|
const response = await writethrough.put({ ...first, value: 3 }) |
||||
|
const output = await db.get(response.id) |
||||
|
expect(response.rev).not.toBe(first._rev) |
||||
|
expect(output.value).toBe(3) |
||||
|
}) |
||||
|
}) |
||||
|
|
||||
|
describe("get", () => { |
||||
|
it("should be able to retrieve", async () => { |
||||
|
const response = await writethrough.get("test") |
||||
|
expect(response.value).toBe(3) |
||||
|
}) |
||||
|
}) |
||||
|
|
||||
|
describe("same doc, different databases (tenancy)", () => { |
||||
|
it("should be able to two different databases", async () => { |
||||
|
const resp1 = await writethrough.put({ _id: "db1", value: "first" }) |
||||
|
const resp2 = await writethrough2.put({ _id: "db1", value: "second" }) |
||||
|
expect(resp1.rev).toBeDefined() |
||||
|
expect(resp2.rev).toBeDefined() |
||||
|
expect((await db.get("db1")).value).toBe("first") |
||||
|
expect((await db2.get("db1")).value).toBe("second") |
||||
|
}) |
||||
|
}) |
||||
|
}) |
||||
|
|
||||
@ -0,0 +1,113 @@ |
|||||
|
import BaseCache from "./base" |
||||
|
import { getWritethroughClient } from "../redis/init" |
||||
|
|
||||
|
const DEFAULT_WRITE_RATE_MS = 10000 |
||||
|
let CACHE: BaseCache | null = null |
||||
|
|
||||
|
interface CacheItem { |
||||
|
doc: any |
||||
|
lastWrite: number |
||||
|
} |
||||
|
|
||||
|
async function getCache() { |
||||
|
if (!CACHE) { |
||||
|
const client = await getWritethroughClient() |
||||
|
CACHE = new BaseCache(client) |
||||
|
} |
||||
|
return CACHE |
||||
|
} |
||||
|
|
||||
|
function makeCacheKey(db: PouchDB.Database, key: string) { |
||||
|
return db.name + key |
||||
|
} |
||||
|
|
||||
|
function makeCacheItem(doc: any, lastWrite: number | null = null): CacheItem { |
||||
|
return { doc, lastWrite: lastWrite || Date.now() } |
||||
|
} |
||||
|
|
||||
|
export async function put( |
||||
|
db: PouchDB.Database, |
||||
|
doc: any, |
||||
|
writeRateMs: number = DEFAULT_WRITE_RATE_MS |
||||
|
) { |
||||
|
const cache = await getCache() |
||||
|
const key = doc._id |
||||
|
let cacheItem: CacheItem | undefined = await cache.get(makeCacheKey(db, key)) |
||||
|
const updateDb = !cacheItem || cacheItem.lastWrite < Date.now() - writeRateMs |
||||
|
let output = doc |
||||
|
if (updateDb) { |
||||
|
try { |
||||
|
// doc should contain the _id and _rev
|
||||
|
const response = await db.put(doc) |
||||
|
output = { |
||||
|
...doc, |
||||
|
_id: response.id, |
||||
|
_rev: response.rev, |
||||
|
} |
||||
|
} catch (err: any) { |
||||
|
// ignore 409s, some other high speed write has hit it first, just move straight to caching
|
||||
|
if (err.status !== 409) { |
||||
|
throw err |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
// if we are updating the DB then need to set the lastWrite to now
|
||||
|
cacheItem = makeCacheItem(output, updateDb ? null : cacheItem?.lastWrite) |
||||
|
await cache.store(makeCacheKey(db, key), cacheItem) |
||||
|
return { ok: true, id: output._id, rev: output._rev } |
||||
|
} |
||||
|
|
||||
|
export async function get(db: PouchDB.Database, id: string): Promise<any> { |
||||
|
const cache = await getCache() |
||||
|
const cacheKey = makeCacheKey(db, id) |
||||
|
let cacheItem: CacheItem = await cache.get(cacheKey) |
||||
|
if (!cacheItem) { |
||||
|
const doc = await db.get(id) |
||||
|
cacheItem = makeCacheItem(doc) |
||||
|
await cache.store(cacheKey, cacheItem) |
||||
|
} |
||||
|
return cacheItem.doc |
||||
|
} |
||||
|
|
||||
|
export async function remove( |
||||
|
db: PouchDB.Database, |
||||
|
docOrId: any, |
||||
|
rev?: any |
||||
|
): Promise<void> { |
||||
|
const cache = await getCache() |
||||
|
if (!docOrId) { |
||||
|
throw new Error("No ID/Rev provided.") |
||||
|
} |
||||
|
const id = typeof docOrId === "string" ? docOrId : docOrId._id |
||||
|
rev = typeof docOrId === "string" ? rev : docOrId._rev |
||||
|
try { |
||||
|
await cache.delete(makeCacheKey(db, id)) |
||||
|
} finally { |
||||
|
await db.remove(id, rev) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
export class Writethrough { |
||||
|
db: PouchDB.Database |
||||
|
writeRateMs: number |
||||
|
|
||||
|
constructor( |
||||
|
db: PouchDB.Database, |
||||
|
writeRateMs: number = DEFAULT_WRITE_RATE_MS |
||||
|
) { |
||||
|
this.db = db |
||||
|
this.writeRateMs = writeRateMs |
||||
|
} |
||||
|
|
||||
|
async put(doc: any) { |
||||
|
return put(this.db, doc, this.writeRateMs) |
||||
|
} |
||||
|
|
||||
|
async get(id: string) { |
||||
|
return get(this.db, id) |
||||
|
} |
||||
|
|
||||
|
async remove(docOrId: any, rev?: any) { |
||||
|
return remove(this.db, docOrId, rev) |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue