Add Redis initialization to Kafka consumers and app startup

Author: nygrenh

backend/app.ts

@@ -5,6 +5,7 @@ import { ServerContext } from "./context"
 import prisma from "./prisma"
 import server from "./server"
 import knex from "./services/knex"
+import { initializeRedis } from "./services/redis"
 import { attachPrismaEvents } from "./util"
 
 require("sharp") // image library sharp seems to crash without this require
@@ -35,6 +36,9 @@ const startApp = async () => {
   attachPrismaEvents(ctx)
 
   if (!NEXUS_REFLECTION) {
+    initializeRedis().catch((err) => {
+      logger.warn("Redis initialization failed, continuing without cache", err)
+    })
     await ctx.prisma.$connect()
     httpServer.listen(4000, () => {
       console.log("server running on port 4000")

backend/bin/kafkaConsumer/exerciseConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -21,6 +22,10 @@ const logger = sentryLogger({ service: "kafka-consumer-exercise" })
 
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 consumer.connect()
 
 const context: KafkaContext = {

backend/bin/kafkaConsumer/userCoursePointsConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -20,6 +21,10 @@ const mutex = new Mutex()
 const logger = sentryLogger({ service: "kafka-consumer-user-course-points" })
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 consumer.connect()
 
 const context: KafkaContext = {

backend/bin/kafkaConsumer/userCoursePointsRealtimeConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -22,6 +23,10 @@ const logger = sentryLogger({
 })
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 consumer.connect()
 
 const context: KafkaContext = {

backend/bin/kafkaConsumer/userCourseProgressConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -22,6 +23,10 @@ const TOPIC_NAME = [config.user_course_progress_consumer.topic_name]
 const logger = sentryLogger({ service: "kafka-consumer-UserCourseProgress" })
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 const context: KafkaContext = {
   prisma,
   logger,

backend/bin/kafkaConsumer/userCourseProgressRealtimeConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -25,6 +26,10 @@ const logger = sentryLogger({
 })
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 consumer.connect()
 
 const context: KafkaContext = {

backend/bin/kafkaConsumer/userPointsConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -20,6 +21,10 @@ const mutex = new Mutex()
 const logger = sentryLogger({ service: "kafka-consumer-user-points" })
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 consumer.connect()
 
 const context: KafkaContext = {

backend/bin/kafkaConsumer/userPointsRealtimeConsumer/kafkaConsumer.ts

@@ -5,6 +5,7 @@ import { KafkaError } from "../../../lib/errors"
 import sentryLogger from "../../../lib/logger"
 import prisma from "../../../prisma"
 import knex from "../../../services/knex"
+import { initializeRedis } from "../../../services/redis"
 import { createKafkaConsumer } from "../common/createKafkaConsumer"
 import { handleMessage } from "../common/handleMessage"
 import { KafkaContext } from "../common/kafkaContext"
@@ -20,6 +21,10 @@ const mutex = new Mutex()
 const logger = sentryLogger({ service: "kafka-consumer-user-points-realtime" })
 const consumer = createKafkaConsumer({ logger, prisma })
 
+initializeRedis().catch((err) => {
+  logger.warn("Redis initialization failed, continuing without cache", err)
+})
+
 consumer.connect()
 
 const context: KafkaContext = {

backend/services/redis.ts

@@ -49,7 +49,7 @@ const getRedisClient = (): RedisClient | undefined => {
     return
   }
 
-  let url = (REDIS_URL && REDIS_URL.trim()) || "redis://127.0.0.1:6379"
+  let url = REDIS_URL?.trim() || "redis://127.0.0.1:6379"
   if (url && !url.startsWith("redis://") && !url.startsWith("rediss://")) {
     url = `redis://${url}`
   }
@@ -81,6 +81,23 @@ const getRedisClient = (): RedisClient | undefined => {
   return client
 }
 
+export const initializeRedis = async (): Promise<void> => {
+  _logger.info("Initializing Redis")
+  try {
+    const client = getRedisClient()
+    if (client && !client.isOpen) {
+      try {
+        await client.connect()
+        _logger.info("Redis connection established")
+      } catch (err: any) {
+        _logger.warn(`Redis connection failed, will retry automatically`, err)
+      }
+    }
+  } catch (err: any) {
+    _logger.warn(`Redis initialization error, continuing without cache`, err)
+  }
+}
+
 const getRedisSubscriberClient = (): RedisClient | undefined => {
   if (redisSubscriberClient) {
     return redisSubscriberClient
@@ -94,7 +111,7 @@ const getRedisSubscriberClient = (): RedisClient | undefined => {
     return
   }
 
-  let url = (REDIS_URL && REDIS_URL.trim()) || "redis://127.0.0.1:6379"
+  let url = REDIS_URL?.trim() || "redis://127.0.0.1:6379"
   if (url && !url.startsWith("redis://") && !url.startsWith("rediss://")) {
     url = `redis://${url}`
   }