make mongodb optional again

absorbb · absorbb · commit b9330e11de31 · 2024-12-06T13:05:13.000+04:00
diff --git a/libs/core-functions/src/functions/lib/store.ts b/libs/core-functions/src/functions/lib/store.ts
@@ -3,6 +3,7 @@ import type { Redis } from "ioredis";
 import parse from "parse-duration";
 import { MongoClient, ReadPreference, Collection } from "mongodb";
 import { RetryError } from "@jitsu/functions-lib";
+import { Singleton } from "juava";
 
 export const defaultTTL = 60 * 60 * 24 * 31; // 31 days
 export const maxAllowedTTL = 2147483647; // max allowed value for ttl in redis (68years)
@@ -64,7 +65,7 @@ const MongoCreatedCollections: Record<string, Collection<StoreValue>> = {};
 
 export const createMongoStore = (
   namespace: string,
-  mongo: MongoClient,
+  mongo: Singleton<MongoClient>,
   useLocalCache: boolean,
   fast: boolean
 ): TTLStore => {
@@ -87,7 +88,7 @@ export const createMongoStore = (
       return collection;
     }
     try {
-      const db = mongo.db(dbName);
+      const db = mongo().db(dbName);
 
       const col = db.collection<StoreValue>(namespace);
       const collStatus = await col
diff --git a/services/profiles/src/lib/functions-chain.ts b/services/profiles/src/lib/functions-chain.ts
@@ -62,7 +62,7 @@ export function buildFunctionChain(
   fetchTimeoutMs: number = 2000
 ): FuncChain {
   const pbLongId = `${profileBuilder.workspaceId}-${profileBuilder.id}-v${profileBuilder.version}`;
-  const store = createMongoStore(profileBuilder.workspaceId, mongodb(), false, true);
+  const store = createMongoStore(profileBuilder.workspaceId, mongodb, false, true);
 
   const chainCtx: FunctionChainContext = {
     fetch: makeFetch(profileBuilder.id, eventsLogger, "info", fetchTimeoutMs),
diff --git a/services/rotor/src/http/profiles-udf.ts b/services/rotor/src/http/profiles-udf.ts
@@ -6,7 +6,7 @@ const log = getLog("profile-udf-run");
 export const ProfileUDFRunHandler = async (req, res) => {
   const body = req.body as ProfileUDFTestRequest;
   log.atInfo().log(`Running profile func: ${body?.id} workspace: ${body?.workspaceId}`, JSON.stringify(body));
-  body.store = createMongoStore(body?.workspaceId, mongodb(), true, false);
+  body.store = createMongoStore(body?.workspaceId, mongodb, true, false);
   const result = await ProfileUDFTestRun(body);
   if (result.error) {
     log
diff --git a/services/rotor/src/http/udf.ts b/services/rotor/src/http/udf.ts
@@ -6,7 +6,7 @@ const log = getLog("udf_run");
 export const UDFRunHandler = async (req, res) => {
   const body = req.body as UDFTestRequest;
   log.atInfo().log(`Running function: ${body?.functionId} workspace: ${body?.workspaceId}`, JSON.stringify(body));
-  body.store = createMongoStore(body?.workspaceId, mongodb(), true, false);
+  body.store = createMongoStore(body?.workspaceId, mongodb, true, false);
   const result = await UDFTestRun(body);
   if (result.error) {
     log
diff --git a/services/rotor/src/index.ts b/services/rotor/src/index.ts
@@ -56,7 +56,11 @@ async function main() {
   let redisClient: Redis | undefined;
   try {
     Prometheus.collectDefaultMetrics();
-    await mongodb.waitInit();
+    try {
+      await mongodb.waitInit();
+    } catch (e: any) {
+      log.atWarn().log("Failed to connect to mongodb. Functions Persistent Store won't work: " + e.message);
+    }
     if (process.env.CLICKHOUSE_HOST || process.env.CLICKHOUSE_URL) {
       eventsLogger = createClickhouseLogger();
     } else {
diff --git a/services/rotor/src/lib/functions-chain.ts b/services/rotor/src/lib/functions-chain.ts
@@ -137,7 +137,7 @@ export function buildFunctionChain(
   if (!store) {
     store = createMongoStore(
       connection.workspaceId,
-      mongodb(),
+      mongodb,
       false,
       fastStoreWorkspaceId.includes(connection.workspaceId)
     );
