matching: support REF-AUS ✅📝

Author: derhuerst

lib/match-with-schedule-trip.js

@@ -189,6 +189,7 @@ const createMatchGtfsRtTripUpdateWithScheduleStopTimes = async (cfg, opt = {}) =
 
 	const cache = await createCache({
 		noCaching: NO_CACHING,
+		prefix: 'match',
 		ttl: process.env.MATCHING_CACHING_TTL
 			? parseInt(process.env.MATCHING_CACHING_TTL) * 1000
 			: 24 * 60 * 60 * 1000, // 24 hours

lib/match.js

@@ -16,6 +16,9 @@ import {register} from './metrics.js'
 import {
 	jsonCodec as natsJson,
 } from './nats.js'
+import {
+	createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts,
+} from './merge-vdv-sollfahrts-istfahrts.js'
 import {
 	createMatchWithGtfs,
 } from './raw-match.js'
@@ -39,12 +42,17 @@ const isProgrammerError = (err) => {
 	return PROGRAMMER_ERRORS.some(Err => err instanceof Err)
 }
 
+const KIND_SOLLFAHRT = 'sollfahrt'
 const KIND_ISTFAHRT = 'istfahrt'
 
 // todo: DRY with OpenDataVBB/nats-consuming-gtfs-rt-server
+const NATS_JETSTREAM_REF_AUS_SOLLFAHRT_STREAM_NAME = `REF_AUS_SOLLFAHRT_${MAJOR_VERSION}`
 const NATS_JETSTREAM_AUS_ISTFAHRT_STREAM_NAME = `AUS_ISTFAHRT_${MAJOR_VERSION}`
 const NATS_JETSTREAM_GTFSRT_STREAM_NAME = `GTFS_RT_${MAJOR_VERSION}`
 
+const vdvMergingLogger = createLogger('vdv-merging', {
+	level: (process.env.LOG_LEVEL_VDV_MERGING || 'WARN').toLowerCase(),
+})
 const logger = createLogger('matching', {
 	level: (process.env.LOG_LEVEL_MATCHING || 'warn').toLowerCase(),
 })
@@ -184,13 +192,21 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 	})
 	const vdvFahrtHaltsTotal = new Summary({
 		name: 'vdv_fahrt_halts_total',
-		help: 'number of IstHalts per AUS IstFahrt',
+		help: 'number of SollHalts/IstHalts per REF-AUS SollFahrt/AUS IstFahrt',
 		registers: [register],
 		labelNames: [
 			'kind', // sollfahrt, istfahrt
 		],
 	})
 
+	const {
+		storeRefAusSollFahrt: storeVdvRefAusSollFahrtForLaterMerging,
+		storeAusIstFahrt: storeVdvAusIstFahrtForLaterMerging,
+		mergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts,
+	} = await createMergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts({
+		logger: vdvMergingLogger,
+	})
+
 	const {
 		matchVdvAusIstFahrtWithGtfs,
 		stop: stopMatching,
@@ -222,7 +238,7 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		}
 	}
 
-	const processVdvNatsMsg = async (msg, kind, kindTitle, haltsField) => {
+	const processVdvNatsMsg = async (msg, kind, kindTitle, haltsField, storeVdvFahrt) => {
 		const tReceived = Date.now()
 		const {
 			subject,
@@ -293,20 +309,24 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 				kind,
 			}, vdvFahrt[haltsField].length)
 
+			await storeVdvFahrt(vdvFahrt)
+			const mergedVdvFahrt = await mergeVdvFahrtWithRefAusSollFahrtAndAusIstFahrts(vdvFahrt)
+			// todo: trace-log?
+
 			const {
 				item: gtfsRtTripUpdate,
 				isMatched,
 				isCached,
 				matchingTime,
-			} = await matchVdvAusIstFahrtWithGtfs(vdvFahrt)
+			} = await matchVdvAusIstFahrtWithGtfs(mergedVdvFahrt)
 
 			if (isMatched || publishUnmatchedTripUpdates) {
 				publishGtfsRtTripUpdateToNats(gtfsRtTripUpdate, {
 					isMatched,
 					isCached,
 					matchingTime,
 					// todo: log just a slice?
-					[kind]: vdvFahrt,
+					[kind]: mergedVdvFahrt,
 					natsMsgSeq: msg.seq,
 				})
 			}
@@ -365,14 +385,70 @@ const runGtfsMatching = async (cfg, opt = {}) => {
 		}
 	}
 
+	const processRefAusSollFahrtMsg = async (msg) => {
+		await processVdvNatsMsg(
+			msg,
+			KIND_SOLLFAHRT,
+			'REF-AUS SollFahrt',
+			'SollHalts',
+			storeVdvRefAusSollFahrtForLaterMerging,
+		)
+	}
+
+	// subscribe to REF-AUS SollFahrt messages
+	{
+		{
+			// query details of the NATS JetStream stream for AUS IstFahrts
+			const stream = await natsJetstreamClient.streams.get(NATS_JETSTREAM_REF_AUS_SOLLFAHRT_STREAM_NAME)
+			const streamInfo = await stream.info()
+			serviceLogger.debug({
+				streamInfo,
+			}, 'using NATS JetStream stream for REF-AUS SollFahrts')
+			// todo: assert some properties?
+			// strictEqual(streamInfo.config.discard, 'old', `NATS JetStream's discard must be "old"`)
+		}
+
+		const sollFahrtsConsumer = await natsJetstreamClient.consumers.get(
+			NATS_JETSTREAM_REF_AUS_SOLLFAHRT_STREAM_NAME,
+			natsConsumerName,
+		)
+
+		{
+			// query details of the (externally created) NATS JetStream consumer
+			const consumerInfo = await sollFahrtsConsumer.info()
+			serviceLogger.debug({
+				stream: NATS_JETSTREAM_REF_AUS_SOLLFAHRT_STREAM_NAME,
+				consumerInfo,
+			}, 'using NATS JetStream consumer for REF-AUS SollFahrts')
+			// todo: assert some properties?
+			// strictEqual(consumerInfo.config.deliver_policy, 'new', `REF-AUS JetStream consumer's deliver_policy must be "new"`)
+			// strictEqual(consumerInfo.config.ack_policy, 'explicit', `REF-AUS JetStream consumer's ack_policy must be "explicit"`)
+		}
+
+		const sollFahrtsSub = await sollFahrtsConsumer.consume()
+		// We're not interested in the values, processRefAusSollFahrtMsg() publishes by itself.
+		asyncConsume(
+			mapConcurrently(
+				sollFahrtsSub[Symbol.asyncIterator](),
+				matchConcurrency,
+				processRefAusSollFahrtMsg,
+			),
+		).catch(abortWithError)
+
+		// todo: support SollUmlauf – would require adapting vdv-453-nats-adapter
+	}
+
 	const processAusIstFahrtMsg = async (msg) => {
 		await processVdvNatsMsg(
 			msg,
 			KIND_ISTFAHRT,
 			'AUS IstFahrt',
+			'IstHalts',
+			storeVdvAusIstFahrtForLaterMerging,
 		)
 	}
 
+	// subscribe to AUS IstFahrt messages
 	{
 		{
 			// query details of the NATS JetStream stream for AUS IstFahrts