rotor: bring back redis store when mongo is unavailable

libs/core-functions/src/functions/lib/store.ts

@@ -44,11 +44,19 @@ export const createOldStore = (namespace: string, redisClient: Redis): Store =>
   },
 });
 
-export const createTtlStore = (namespace: string, redisClient: Redis, defaultTtlSec: number): Store => ({
+export const createTtlStore = (namespace: string, redisClient: Redis, defaultTtlSec: number): TTLStore => ({
   get: async (key: string) => {
     const res = await redisClient.get(`store:${namespace}:${key}`);
     return res ? JSON.parse(res) : undefined;
   },
+  getWithTTL: async (key: string) => {
+    const res = await redisClient.get(`store:${namespace}:${key}`);
+    if (!res) {
+      return undefined;
+    }
+    const ttl = await redisClient.ttl(`store:${namespace}:${key}`);
+    return { value: JSON.parse(res), ttl };
+  },
   set: async (key: string, obj: any, opts?: SetOpts) => {
     const ttl = getTtlSec(opts);
     if (ttl >= 0) {

services/rotor/src/lib/message-handler.ts

@@ -14,9 +14,11 @@ import {
   mongoAnonymousEventsStore,
   parseUserAgent,
   SystemContext,
+  createTtlStore,
 } from "@jitsu/core-functions";
 import { redisLogger } from "./redis-logger";
 import { buildFunctionChain, checkError, runChain } from "./functions-chain";
+import { redis } from "@jitsu-internal/console/lib/server/redis";
 export const log = getLog("rotor");
 
 const anonymousEventsStore = mongoAnonymousEventsStore();
@@ -88,11 +90,9 @@ export async function rotorMessageHandler(
     connectionId: connection.id,
     retries,
   };
-  const store = createMongoStore(
-    connection.workspaceId,
-    mongodb(),
-    fastStoreWorskpaceId.includes(connection.workspaceId)
-  );
+  const store = process.env.MONGODB_URL
+    ? createMongoStore(connection.workspaceId, mongodb(), fastStoreWorskpaceId.includes(connection.workspaceId))
+    : createTtlStore(connection.workspaceId, redis(), defaultTTL);
   //system context for builtin functions only
   const systemContext: SystemContext = {
     $system: {