matching: SollFahrt/IstFahrt merging: use Redis Hashes 🐛✅📝

The previous approach (plain keys + SCAN + MGET)
- Failed to SCAN to completion sometimes, and
- scaled badly.

Using Redis Hashes seems much more elegant in general.

Author: derhuerst

lib/caching.js

@@ -111,5 +111,6 @@ const createCache = async (opt = {}) => {
 }
 
 export {
+	PREFIX,
 	createCache,
 }

lib/merge-vdv-sollfahrts-istfahrts.js

@@ -5,11 +5,14 @@ import {
 } from 'node:assert/strict'
 import omit from 'lodash/omit.js'
 import maxBy from 'lodash/maxBy.js'
+import {
+	PREFIX as REDIS_PREFIX,
+} from './caching.js'
 import {
     mergeButPreferNonNull,
 	unixTimestampFromIso8601,
 } from './util.js'
-import {createCache} from './caching.js'
+import {connectToRedis} from './redis.js'
 
 // The logic in this file uses three sources of information:
 // 1. In almost all cases, a REF-AUS SollFahrt is sent ahead of time.
@@ -20,9 +23,14 @@ import {createCache} from './caching.js'
 
 // ---
 
+const VDV_STORAGE_TTL = process.env.VDV_STORAGE_TTL
+	? parseInt(process.env.VDV_STORAGE_TTL) * 1000
+	: 32 * 60 * 60 * 1000 // 32 hours
+
 const KIND_SOLLFAHRT = Symbol('REF-AUS SollFahrt')
 const KIND_ISTFAHRT = Symbol('AUS IstFahrt')
 
+const STORAGE_KEY_PREFIX = REDIS_PREFIX + 'vdv:'
 const STORAGE_KEY_REF_AUS_SOLLFAHRT = 'ref_aus_soll'
 const STORAGE_KEY_AUS_ISTFAHRT_KOMPLETTFAHRT = 'aus_komplett'
 const STORAGE_KEY_AUS_ISTFAHRT_PARTIAL = 'aus_partial'
@@ -214,9 +222,12 @@ const createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts = async (cfg) => {
 		logger,
 	} = cfg
 
-	const storage = await createCache({
-		prefix: 'vdv:',
-	})
+	const storage = await connectToRedis()
+	// As of [email protected], it doesn't support the HSETEX command, so we patch our client.
+	// https://redis.io/docs/latest/commands/?group=hash
+	if (typeof storage.hsetex !== 'function') {
+		storage.addBuiltinCommand('hsetex')
+	}
 
 	const _storeVdvFahrt = async (vdvFahrt, kind) => {
 		if (kind !== KIND_SOLLFAHRT && kind !== KIND_ISTFAHRT) {
@@ -231,10 +242,11 @@ const createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts = async (cfg) => {
 		if (!fahrtId) {
 			return null
 		}
+		const storageKey = STORAGE_KEY_PREFIX + fahrtId
 
 		const isKomplettfahrt = vdvFahrt.Komplettfahrt === 'true'
 		if (kind === KIND_ISTFAHRT && !isKomplettfahrt) {
-			const entries = halts.map((istHalt, i) => {
+			const fieldsArgs = halts.flatMap((istHalt, i) => {
 				ok(istHalt.HaltID, `vdvFahrt.${haltsKey}[${i}].HaltID must not be missing`)
 				let depOrArrPrefix = null
 				let depOrArr = null
@@ -255,32 +267,41 @@ const createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts = async (cfg) => {
 
 				// todo: what if there are >1 IstHalts stopping at the same HaltID within the same minute?
 				// todo: what if an IstHalt is first seen only with an Ankunftszeit and later with an Abfahrtszeit? – it will be stored & read twice!
-				const storageKey = [
-					fahrtId,
+				const hashField = [
 					STORAGE_KEY_AUS_ISTFAHRT_PARTIAL,
 					istHalt.HaltID,
 					depOrArrPrefix,
 					depOrArr,
 				].join(':')
 				return [
-					storageKey,
-					{
+					hashField,
+					JSON.stringify({
 						...istHalt,
 						IstFahrt: sparseIstFahrt,
-					},
+					}),
 				]
 			})
-			await storage.putMany(entries)
+
+			// todo: support Valkey once they have a "Hash set + expiration" command
+			await storage.hsetex(
+				storageKey, // key of Redis Hash
+				'PX', VDV_STORAGE_TTL, // expiration time in milliseconds
+				'FIELDS', fieldsArgs.length / 2, // number of Hash fields to set
+				...fieldsArgs,
+			)
 		} else {
-			const storageKeySuffix = kind === KIND_SOLLFAHRT
+			const field = kind === KIND_SOLLFAHRT
 				? STORAGE_KEY_REF_AUS_SOLLFAHRT
 				: STORAGE_KEY_AUS_ISTFAHRT_KOMPLETTFAHRT
-			const storageKey = [
-				fahrtId,
-				storageKeySuffix,
-			].join(':')
 
-			await storage.put(storageKey, vdvFahrt)
+			// todo: support Valkey once they have a "Hash set + expiration" command
+			await storage.hsetex(
+				storageKey, // key of Redis Hash
+				'PX', VDV_STORAGE_TTL, // expiration time in milliseconds
+				'FIELDS', 1, // number of Hash fields to set
+				field,
+				JSON.stringify(vdvFahrt), // value
+			)
 		}
 	}
 	const storeRefAusSollFahrt = async (sollFahrt) => {
@@ -300,7 +321,7 @@ const createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts = async (cfg) => {
 		if (!fahrtId) {
 			return []
 		}
-		const storageKeyPrefix = fahrtId + ':'
+		const storageKey = STORAGE_KEY_PREFIX + fahrtId
 
 		const res = {
 			refAusSollFahrt: null,
@@ -309,8 +330,10 @@ const createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts = async (cfg) => {
 		}
 
 		// load from storage
-		for (const [key, item] of await storage.getMany(storageKeyPrefix)) {
-			const [kindPart] = key.slice(storageKeyPrefix.length).split(':')
+		const hash = await storage.hgetall(storageKey)
+		for (const [key, val] of Object.entries(hash)) {
+			const [kindPart] = key.split(':')
+			const item = JSON.parse(val)
 			if (kindPart === STORAGE_KEY_REF_AUS_SOLLFAHRT) {
 				res.refAusSollFahrt = item
 			} else if (kindPart === STORAGE_KEY_AUS_ISTFAHRT_KOMPLETTFAHRT) {
@@ -430,7 +453,7 @@ const createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts = async (cfg) => {
 	}
 
 	const stop = async () => {
-		await storage.stop()
+		await storage.quit()
 	}
 
 	const res = {

readme.md

@@ -187,7 +187,7 @@ cd postgis-gtfs-importer && npm install --omit dev
 
 - a [NATS message queue](https://docs.nats.io) with [JetStream](https://docs.nats.io/nats-concepts/jetstream) enabled
 - a [PostgreSQL database server](https://postgresql.org), with the permission to dynamically create new databases (see [postgis-gtfs-importer](https://github.com/mobidata-bw/postgis-gtfs-importer)'s readme)
-- a [Redis in-memory cache](https://redis.io/docs/latest/)
+- a [Redis in-memory cache](https://redis.io/docs/latest/), at least 8.0.0 is required (Valkey currently doesn't support the `HSETEX` command)
 
 #### configure access to PostgreSQL
 

test/vdv-merging.js

@@ -29,7 +29,7 @@ const {
 	})
 })
 beforeEach(async () => {
-	await _vdvMergingStorage.redis.flushdb()
+	await _vdvMergingStorage.flushdb()
 })
 after(async () => {
 	await stopVdvMerging()