chore: format everything
This commit is contained in:
+51
-48
@@ -1,7 +1,15 @@
|
||||
import NodeCache from '@cacheable/node-cache'
|
||||
import { randomBytes } from 'crypto'
|
||||
import { DEFAULT_CACHE_TTLS } from '../Defaults'
|
||||
import type { AuthenticationCreds, CacheStore, SignalDataSet, SignalDataTypeMap, SignalKeyStore, SignalKeyStoreWithTransaction, TransactionCapabilityOptions } from '../Types'
|
||||
import type {
|
||||
AuthenticationCreds,
|
||||
CacheStore,
|
||||
SignalDataSet,
|
||||
SignalDataTypeMap,
|
||||
SignalKeyStore,
|
||||
SignalKeyStoreWithTransaction,
|
||||
TransactionCapabilityOptions
|
||||
} from '../Types'
|
||||
import { Curve, signedKeyPair } from './crypto'
|
||||
import { delay, generateRegistrationId } from './generics'
|
||||
import { ILogger } from './logger'
|
||||
@@ -17,11 +25,13 @@ export function makeCacheableSignalKeyStore(
|
||||
logger?: ILogger,
|
||||
_cache?: CacheStore
|
||||
): SignalKeyStore {
|
||||
const cache = _cache || new NodeCache({
|
||||
stdTTL: DEFAULT_CACHE_TTLS.SIGNAL_STORE, // 5 minutes
|
||||
useClones: false,
|
||||
deleteOnExpire: true,
|
||||
})
|
||||
const cache =
|
||||
_cache ||
|
||||
new NodeCache({
|
||||
stdTTL: DEFAULT_CACHE_TTLS.SIGNAL_STORE, // 5 minutes
|
||||
useClones: false,
|
||||
deleteOnExpire: true
|
||||
})
|
||||
|
||||
function getUniqueId(type: string, id: string) {
|
||||
return `${type}.${id}`
|
||||
@@ -29,23 +39,23 @@ export function makeCacheableSignalKeyStore(
|
||||
|
||||
return {
|
||||
async get(type, ids) {
|
||||
const data: { [_: string]: SignalDataTypeMap[typeof type] } = { }
|
||||
const data: { [_: string]: SignalDataTypeMap[typeof type] } = {}
|
||||
const idsToFetch: string[] = []
|
||||
for(const id of ids) {
|
||||
for (const id of ids) {
|
||||
const item = cache.get<SignalDataTypeMap[typeof type]>(getUniqueId(type, id))
|
||||
if(typeof item !== 'undefined') {
|
||||
if (typeof item !== 'undefined') {
|
||||
data[id] = item
|
||||
} else {
|
||||
idsToFetch.push(id)
|
||||
}
|
||||
}
|
||||
|
||||
if(idsToFetch.length) {
|
||||
if (idsToFetch.length) {
|
||||
logger?.trace({ items: idsToFetch.length }, 'loading from store')
|
||||
const fetched = await store.get(type, idsToFetch)
|
||||
for(const id of idsToFetch) {
|
||||
for (const id of idsToFetch) {
|
||||
const item = fetched[id]
|
||||
if(item) {
|
||||
if (item) {
|
||||
data[id] = item
|
||||
cache.set(getUniqueId(type, id), item)
|
||||
}
|
||||
@@ -56,8 +66,8 @@ export function makeCacheableSignalKeyStore(
|
||||
},
|
||||
async set(data) {
|
||||
let keys = 0
|
||||
for(const type in data) {
|
||||
for(const id in data[type]) {
|
||||
for (const type in data) {
|
||||
for (const id in data[type]) {
|
||||
cache.set(getUniqueId(type, id), data[type][id])
|
||||
keys += 1
|
||||
}
|
||||
@@ -89,52 +99,45 @@ export const addTransactionCapability = (
|
||||
// number of queries made to the DB during the transaction
|
||||
// only there for logging purposes
|
||||
let dbQueriesInTransaction = 0
|
||||
let transactionCache: SignalDataSet = { }
|
||||
let mutations: SignalDataSet = { }
|
||||
let transactionCache: SignalDataSet = {}
|
||||
let mutations: SignalDataSet = {}
|
||||
|
||||
let transactionsInProgress = 0
|
||||
|
||||
return {
|
||||
get: async(type, ids) => {
|
||||
if(isInTransaction()) {
|
||||
get: async (type, ids) => {
|
||||
if (isInTransaction()) {
|
||||
const dict = transactionCache[type]
|
||||
const idsRequiringFetch = dict
|
||||
? ids.filter(item => typeof dict[item] === 'undefined')
|
||||
: ids
|
||||
const idsRequiringFetch = dict ? ids.filter(item => typeof dict[item] === 'undefined') : ids
|
||||
// only fetch if there are any items to fetch
|
||||
if(idsRequiringFetch.length) {
|
||||
if (idsRequiringFetch.length) {
|
||||
dbQueriesInTransaction += 1
|
||||
const result = await state.get(type, idsRequiringFetch)
|
||||
|
||||
transactionCache[type] ||= {}
|
||||
Object.assign(
|
||||
transactionCache[type]!,
|
||||
result
|
||||
)
|
||||
Object.assign(transactionCache[type]!, result)
|
||||
}
|
||||
|
||||
return ids.reduce(
|
||||
(dict, id) => {
|
||||
const value = transactionCache[type]?.[id]
|
||||
if(value) {
|
||||
dict[id] = value
|
||||
}
|
||||
return ids.reduce((dict, id) => {
|
||||
const value = transactionCache[type]?.[id]
|
||||
if (value) {
|
||||
dict[id] = value
|
||||
}
|
||||
|
||||
return dict
|
||||
}, { }
|
||||
)
|
||||
return dict
|
||||
}, {})
|
||||
} else {
|
||||
return state.get(type, ids)
|
||||
}
|
||||
},
|
||||
set: data => {
|
||||
if(isInTransaction()) {
|
||||
if (isInTransaction()) {
|
||||
logger.trace({ types: Object.keys(data) }, 'caching in transaction')
|
||||
for(const key in data) {
|
||||
transactionCache[key] = transactionCache[key] || { }
|
||||
for (const key in data) {
|
||||
transactionCache[key] = transactionCache[key] || {}
|
||||
Object.assign(transactionCache[key], data[key])
|
||||
|
||||
mutations[key] = mutations[key] || { }
|
||||
mutations[key] = mutations[key] || {}
|
||||
Object.assign(mutations[key], data[key])
|
||||
}
|
||||
} else {
|
||||
@@ -145,27 +148,27 @@ export const addTransactionCapability = (
|
||||
async transaction(work) {
|
||||
let result: Awaited<ReturnType<typeof work>>
|
||||
transactionsInProgress += 1
|
||||
if(transactionsInProgress === 1) {
|
||||
if (transactionsInProgress === 1) {
|
||||
logger.trace('entering transaction')
|
||||
}
|
||||
|
||||
try {
|
||||
result = await work()
|
||||
// commit if this is the outermost transaction
|
||||
if(transactionsInProgress === 1) {
|
||||
if(Object.keys(mutations).length) {
|
||||
if (transactionsInProgress === 1) {
|
||||
if (Object.keys(mutations).length) {
|
||||
logger.trace('committing transaction')
|
||||
// retry mechanism to ensure we've some recovery
|
||||
// in case a transaction fails in the first attempt
|
||||
let tries = maxCommitRetries
|
||||
while(tries) {
|
||||
while (tries) {
|
||||
tries -= 1
|
||||
//eslint-disable-next-line max-depth
|
||||
try {
|
||||
await state.set(mutations)
|
||||
logger.trace({ dbQueriesInTransaction }, 'committed transaction')
|
||||
break
|
||||
} catch(error) {
|
||||
} catch (error) {
|
||||
logger.warn(`failed to commit ${Object.keys(mutations).length} mutations, tries left=${tries}`)
|
||||
await delay(delayBetweenTriesMs)
|
||||
}
|
||||
@@ -176,9 +179,9 @@ export const addTransactionCapability = (
|
||||
}
|
||||
} finally {
|
||||
transactionsInProgress -= 1
|
||||
if(transactionsInProgress === 0) {
|
||||
transactionCache = { }
|
||||
mutations = { }
|
||||
if (transactionsInProgress === 0) {
|
||||
transactionCache = {}
|
||||
mutations = {}
|
||||
dbQueriesInTransaction = 0
|
||||
}
|
||||
}
|
||||
@@ -211,6 +214,6 @@ export const initAuthCreds = (): AuthenticationCreds => {
|
||||
registered: false,
|
||||
pairingCode: undefined,
|
||||
lastPropHash: undefined,
|
||||
routingInfo: undefined,
|
||||
routingInfo: undefined
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user