feat: enable multiple user accounts for DTS plugin

CHANGELOG.md

@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.5.0] - 2023-01-26
+
+- feat: enable multiple user accounts for DTS plugin
+- refactor: move DTS keys from GCP Secrets to env
+
 ## [1.4.1] - 2022-12-23
 
 - chore: update `jsonwebtoken` to mitigate CVE-2022-23529

config/dtsKeys.js

@@ -0,0 +1,6 @@
+// fetch and convert env
+const radiohubKeys = Buffer.from(process.env.DTS_KEYS, 'base64').toString('utf8')
+const decoded = JSON.parse(radiohubKeys)
+
+// export content
+module.exports = decoded

config/index.js

@@ -12,10 +12,6 @@ const coreIdPrefixes = require('./coreIdPrefixes.json')
 // load winston logger
 const logger = require('../src/utils/logger')
 
-// read env vars
-const stage = process.env.STAGE.toLowerCase()
-const port = process.env.PORT || 8080
-
 const exitWithError = (message) => {
 	logger.log({
 		level: 'error',
@@ -26,55 +22,45 @@ const exitWithError = (message) => {
 }
 
 // check env vars
-if (!process.env.SERVICE_NAME) exitWithError('SERVICE_NAME not found')
-if (!process.env.GCP_PROJECT_ID) exitWithError('GCP_PROJECT_ID not found')
+if (!process.env.DTS_KEYS) exitWithError('DTS_KEYS not found')
 if (!process.env.FIREBASE_API_KEY) exitWithError('FIREBASE_API_KEY not found')
+if (!process.env.GCP_PROJECT_ID) exitWithError('GCP_PROJECT_ID not found')
 if (!process.env.GOOGLE_APPLICATION_CREDENTIALS) exitWithError('GOOGLE_APPLICATION_CREDENTIALS not found')
 if (!process.env.PUBSUB_SERVICE_ACCOUNT_EMAIL_INTERNAL) exitWithError('PUBSUB_SERVICE_ACCOUNT_EMAIL_INTERNAL not found')
+if (!process.env.SERVICE_NAME) exitWithError('SERVICE_NAME not found')
 
-// set protocol, hostname and hostUrl
+// set static envs
+const stage = process.env.STAGE.toLowerCase()
 const protocol = stage === 'dev' ? 'http' : 'https'
 const hostname = stage === 'dev' ? 'localhost' : `eventhub-ingest.ard.de`
-const serviceUrl = `${protocol}://${hostname}:${port}`
+const port = process.env.PORT || 8080
+const serviceName = process.env.SERVICE_NAME
 
 // set config
-const serviceName = process.env.SERVICE_NAME
-const baseConfig = {
+const config = {
+	// load core data
 	coreIdPrefixes,
+
+	// set pub sub config
 	pubSubPrefix: `de.ard.eventhub.${stage}.`,
 	pubSubTopicSelf: `de.ard.eventhub.${stage}.internal`,
+
+	// set service config
+	serviceName,
 	stage,
 	port,
-	userAgent: `${serviceName}/${version}`,
 	version,
-	serviceUrl,
-	isDebug: process.env.DEBUG === 'true',
-	isDev: process.env.STAGE === 'dev',
-}
+	serviceUrl: `${protocol}://${hostname}:${port}`,
 
-// set config based on stages
-const config = {
-	dev: {
-		...baseConfig,
-		serviceName: `${serviceName}-dev`,
-	},
-	test: {
-		...baseConfig,
-		serviceName: `${serviceName}-test`,
-	},
-	prod: {
-		...baseConfig,
-		serviceName,
-	},
-}
+	// custom user agent for headers
+	userAgent: `${serviceName}/${version}`,
 
-// check stage and config
-if (!stage || !config[stage]) {
-	console.error('STAGE not found >', stage)
-	process.exit(1)
+	// set service flags
+	isDebug: process.env.DEBUG === 'true',
+	isDev: process.env.STAGE === 'dev',
 }
 
 // update user agent env for undici-wrapper
 process.env.USER_AGENT = config.userAgent
 
-module.exports = config[stage]
+module.exports = config

keys/secret.example.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: Secret
+type: Opaque
+
+metadata:
+  name: eventhub-ingest-secrets
+
+data:
+  DTS_KEYS: BASE_64_OF_DTS_KEYS
+  FIREBASE_API_KEY: BASE_64_OF_FIREBASE_KEY
+  GCP_SA_KEY: BASE_64_OF_GCP_JSON_KEY

openapi.json

@@ -11,7 +11,7 @@
 			"name": "European Union Public License 1.2",
 			"url": "https://spdx.org/licenses/EUPL-1.2.html"
 		},
-		"version": "1.4.1"
+		"version": "1.5.0"
 	},
 	"externalDocs": {
 		"description": "ARD-Eventhub Documentation",

openapi.yaml

@@ -10,7 +10,7 @@ info:
   license:
     name: European Union Public License 1.2
     url: "https://spdx.org/licenses/EUPL-1.2.html"
-  version: 1.4.1
+  version: 1.5.0
 externalDocs:
   description: ARD-Eventhub Documentation
   url: "https://swrlab.github.io/ard-eventhub/"

package.json

@@ -1,6 +1,6 @@
 {
 	"name": "ard-eventhub",
-	"version": "1.4.1",
+	"version": "1.5.0",
 	"description": "ARD system to distribute real-time (live) metadata for primarily radio broadcasts.",
 	"main": "./src/ingest/index.js",
 	"engines": {
@@ -29,16 +29,16 @@
 	"author": "SWR Audio Lab <[email protected]>",
 	"license": "EUPL-1.2",
 	"dependencies": {
-		"@google-cloud/datastore": "^7.0.0",
-		"@google-cloud/pubsub": "^3.2.1",
+		"@google-cloud/datastore": "^7.1.0",
+		"@google-cloud/pubsub": "^3.3.0",
 		"@google-cloud/secret-manager": "^4.2.0",
 		"@swrlab/utils": "^1.1.0",
 		"compression": "^1.7.4",
-		"dd-trace": "^3.9.3",
+		"dd-trace": "^3.12.1",
 		"dotenv": "^16.0.3",
 		"express": "4.18.2",
-		"express-openapi-validator": "^5.0.0",
-		"firebase-admin": "^11.4.1",
+		"express-openapi-validator": "^5.0.1",
+		"firebase-admin": "^11.5.0",
 		"google-auth-library": "^8.7.0",
 		"jsonwebtoken": "^9.0.0",
 		"moment": "^2.29.4",
@@ -53,12 +53,12 @@
 		"chai": "^4.3.7",
 		"chai-http": "^4.3.0",
 		"docsify-cli": "^4.4.4",
-		"eslint": "^8.30.0",
+		"eslint": "^8.32.0",
 		"eslint-plugin-chai-friendly": "^0.7.2",
 		"license-compliance": "^1.2.5",
 		"mocha": "^10.2.0",
 		"nodemon": "^2.0.20",
-		"prettier": "^2.8.1",
+		"prettier": "^2.8.3",
 		"typescript": "^4.9.4"
 	},
 	"resolutions": {

src/ingest/events/post.js

@@ -262,6 +262,7 @@ module.exports = async (req, res) => {
 					action: `plugins.${plugin.type}.event`,
 					event: message,
 					plugin,
+					institutionId: user.institutionId,
 				}
 
 				// try sending message

src/utils/logger/index.js

@@ -6,11 +6,14 @@
 */
 
 // load node utils
+const os = require('os')
 const { createLogger, config, format, transports } = require('winston')
 
 // get version
 const { version } = require('../../../package.json')
 
+const hostName = os.hostname()
+
 // set error formatter
 const convertError = format((event) => {
 	if (event?.error instanceof Error) {
@@ -25,7 +28,8 @@ const convertError = format((event) => {
 
 // set converter for globals
 const convertGlobals = format((event) => {
-	event.serviceName = 'eventhub-ingest'
+	event.host = process.env.K_REVISION || hostName
+	event.serviceName = process.env.SERVICE_NAME
 	event.stage = process.env.STAGE
 	event.version = version
 	event.nodeVersion = process.version

src/utils/plugins/dts/event.js

@@ -1,4 +1,3 @@
-/* eslint-disable no-nested-ternary */
 /*
 
 	ard-eventhub
@@ -8,23 +7,27 @@
 
 // load utils
 const logger = require('../../logger')
-const secrets = require('../../secrets')
 const undici = require('../../undici')
 
 // load config
 const config = require('../../../../config')
 
+// load keys
+const dtsKeys = require('../../../../config/dtsKeys')
+
+const { credentials, endpoints, integrationName, permittedExcludedFields } = dtsKeys
+
 const source = 'utils/plugins/dts/event'
 
-const defaultHeaders = { Accept: 'application/json' }
+const defaultHeaders = { Accept: 'application/json', 'Content-Type': 'application/json' }
 
 const checkIfArrayHasContent = (thisArray) => {
 	return Array.isArray(thisArray) && thisArray.length > 0
 }
 
 module.exports = async (job) => {
 	// remap input
-	const { event, messageId, plugin } = job
+	const { event, messageId, plugin, institutionId } = job
 
 	// only process now playing events
 	if (event.name !== 'de.ard.eventhub.v1.radio.track.playing') {
@@ -37,30 +40,21 @@ module.exports = async (job) => {
 		return Promise.resolve()
 	}
 
-	// fetch secrets config
-	const { json: pluginSecrets } = await secrets.get(`plugins-dts-${config.stage}`)
-	const { credentials, endpoints, integrationName, permittedExcludedFields } = pluginSecrets
-
 	// collect ARD Core ids
-	const coreIds = event.services.map((service) => service?.topic?.id)
+	const coreIds = event.services.map((service) => service.topic.id)
+
+	// set up dashboard API fetching
+	const lookupConfig = { timeout: 7e3, headers: { ...defaultHeaders, Authorization: credentials.dashboardToken } }
 
 	// fetch all externally mapped ids
-	const lookupConfig = {
-		url: endpoints.listIntegrationRecords.replace('{integrationName}', integrationName),
-		method: 'GET',
-		timeout: 7e3,
-		headers: {
-			...defaultHeaders,
-			Authorization: credentials.dashboard,
-		},
-	}
-	const integrationsList = await undici(lookupConfig.url, lookupConfig)
+	const integrationsEndpointUrl = endpoints.listIntegrationRecords.replace('{integrationName}', integrationName)
+	const integrationsList = await undici(integrationsEndpointUrl, lookupConfig)
 
 	// end processing if no integrations were found
 	if (!integrationsList.ok) {
 		logger.log({
 			level: 'error',
-			message: `failed loading DTS integrations (err 1)`,
+			message: `failed loading DTS integrations`,
 			source,
 			data: { messageId, job, coreIds, response: integrationsList.string },
 		})
@@ -71,7 +65,7 @@ module.exports = async (job) => {
 	if (!checkIfArrayHasContent(integrationsList.json)) {
 		logger.log({
 			level: 'error',
-			message: `failed loading DTS integrations (err 2)`,
+			message: `failed loading DTS integrations`,
 			source,
 			data: { messageId, job, coreIds },
 		})
@@ -91,22 +85,22 @@ module.exports = async (job) => {
 	if (!checkIfArrayHasContent(contentIds)) {
 		logger.log({
 			level: 'notice',
-			message: `DTS BID mapping missing for coreIds (err 3)`,
+			message: `DTS BID mapping missing for coreIds`,
 			source,
 			data: { messageId, job, coreIds },
 		})
 		return Promise.resolve()
 	}
 
 	// fetch all matching broadcasts
-	lookupConfig.url = endpoints.searchBroadcasts.replace('{contentQuery}', contentIds.join(','))
-	const broadcasts = await undici(lookupConfig.url, lookupConfig)
+	const searchBroadcastsUrl = endpoints.searchBroadcasts.replace('{contentQuery}', contentIds.join(','))
+	const broadcasts = await undici(searchBroadcastsUrl, lookupConfig)
 
 	// end processing if no integrations were found
 	if (!broadcasts.ok) {
 		logger.log({
 			level: 'error',
-			message: `failed loading DTS broadcasts for coreIds (err 4)`,
+			message: `failed loading DTS broadcasts for coreIds`,
 			source,
 			data: { messageId, job, coreIds, response: broadcasts.string },
 		})
@@ -117,7 +111,7 @@ module.exports = async (job) => {
 	if (!checkIfArrayHasContent(broadcasts.json)) {
 		logger.log({
 			level: 'notice',
-			message: `failed finding DTS broadcasts for coreIds (err 5)`,
+			message: `failed finding DTS broadcasts for coreIds`,
 			source,
 			data: { messageId, job, coreIds, contentIds, response: broadcasts.json },
 		})
@@ -127,12 +121,17 @@ module.exports = async (job) => {
 	// remap broadcast IDs
 	const linkedBroadcastIds = broadcasts.json.map((broadcast) => broadcast.broadcast_id)
 
+	// remap playing type
+	let type = 'other'
+	if (event.type === 'music') type = event.type
+	if (event.type === 'advertisement') type = 'ad'
+
 	// remap Eventhub variables to external ones
 	const liveRadioEvent = {
 		broadcastId: null,
 		linkedBroadcastIds,
 
-		type: event.type === 'music' ? event.type : event.type === 'advertisement' ? 'ad' : 'other',
+		type,
 		status: 'playing',
 
 		client: config.serviceName,
@@ -177,19 +176,29 @@ module.exports = async (job) => {
 		}
 	}
 
+	// set event host and auth
+	const liveradioUrl = endpoints.liveRadioEvent[config.stage]
+	const liveradioUser = credentials.liveradio.find((user) => user.coreId === institutionId)
+	const liveradioLogin = `${liveradioUser?.username}:${liveradioUser?.password}`
+	const liveradioToken = `Basic ${Buffer.from(liveradioLogin, 'utf8').toString('base64')}`
+	if (!liveradioUrl || !liveradioUser || !liveradioLogin) {
+		logger.log({
+			level: 'error',
+			message: `failed loading DTS user for liveradio API`,
+			source,
+			data: { messageId, job, coreIds },
+		})
+		return Promise.resolve()
+	}
+
 	// post event
 	const postConfig = {
-		url: endpoints.liveRadioEvent[config.stage],
 		method: 'POST',
 		body: JSON.stringify([liveRadioEvent]),
 		timeout: 7e3,
-		headers: {
-			...defaultHeaders,
-			Authorization: credentials.liveradio,
-			'Content-Type': 'application/json',
-		},
+		headers: { ...defaultHeaders, Authorization: liveradioToken },
 	}
-	const posted = await undici(postConfig.url, postConfig)
+	const posted = await undici(liveradioUrl, postConfig)
 
 	// log result
 	logger.log({