Adds secret service for plugin config loading

apps/api/src/env.ts

@@ -24,6 +24,13 @@ export const env = createEnv({
     XPOSTBOUNTY1_RSS_API_SECRET: z.string().optional(),
     AFROBEATS_RSS_API_SECRET: z.string().optional(),
     AFRICA_RSS_API_SECRET: z.string().optional(),
+
+    SECRET_SERVICE_URL: z
+      .string()
+      .url({ message: "SECRET_SERVICE_URL must be a valid URL" }),
+    SECRET_SERVICE_INTERNAL_API_KEY: z
+      .string()
+      .min(1, { message: "SECRET_SERVICE_INTERNAL_API_KEY must be defined" }),
   },
 
   runtimeEnv: process.env,

apps/api/src/services/distribution.service.ts

@@ -2,9 +2,9 @@ import { DistributorConfig, RichSubmission } from "@curatedotfun/shared-db";
 import type { ActionArgs } from "@curatedotfun/types";
 import { PluginError, PluginErrorCode } from "@curatedotfun/utils";
 import type { Logger } from "pino";
-import { isStaging } from "./config.service";
 import { logPluginError } from "../utils/error";
 import { sanitizeJson } from "../utils/sanitize";
+import { isStaging } from "./config.service";
 import type { IBaseService } from "./interfaces/base-service.interface";
 import { PluginService } from "./plugin.service";
 
@@ -21,6 +21,7 @@ export class DistributionService implements IBaseService {
   async distributeContent<T = RichSubmission>(
     distributor: DistributorConfig,
     input: T,
+    feedId: string,
   ): Promise<void> {
     const sanitizedInput = sanitizeJson(input) as T;
 
@@ -32,6 +33,7 @@ export class DistributionService implements IBaseService {
           type: "distributor",
           config: pluginConfig || {},
         },
+        feedId,
       );
 
       try {

apps/api/src/services/feed.service.ts

@@ -68,10 +68,10 @@ export class FeedService implements IBaseService {
         { feedId },
         "FeedService: processFeed - Feed not found",
       );
-      throw new Error(`Feed not found: ${feedId}`); // Or a custom NotFoundError
+      throw new Error(`Feed not found: ${feedId}`);
     }
 
-    const feedConfig = await this.feedRepository.getFeedConfig(feedId); // Get config from DB
+    const feedConfig = await this.feedRepository.getFeedConfig(feedId);
     if (!feedConfig) {
       this.logger.error(
         { feedId },
@@ -154,14 +154,13 @@ export class FeedService implements IBaseService {
           }
         }
 
-        await this.processorService.process(submission, streamConfig);
+        await this.processorService.process(submission, streamConfig, feedId);
         processedCount++;
       } catch (error) {
         this.logger.error(
           { error, submissionId: submission.tweetId, feedId },
           `Error processing submission ${submission.tweetId} for feed ${feedId}`,
         );
-        // Decide if one error should stop all processing or just skip this one
       }
     }
 

apps/api/src/services/plugin.service.ts

@@ -12,15 +12,14 @@ import { PluginError, PluginErrorCode } from "@curatedotfun/utils";
 import { performReload } from "@module-federation/node/utils";
 import { init, loadRemote } from "@module-federation/runtime";
 import type { Logger } from "pino";
-import Mustache from "mustache";
 import { PluginConfig } from "types/config";
-import { env } from "../env";
 import { db } from "../db";
 import { logPluginError } from "../utils/error";
 import { logger } from "../utils/logger";
 import { createPluginInstanceKey } from "../utils/plugin";
 import { isProduction } from "./config.service";
 import { IBaseService } from "./interfaces/base-service.interface";
+import { SecretServiceApiClient } from "./secret-service-client";
 
 /**
  * Cache entry for a loaded plugin
@@ -64,8 +63,8 @@ type PluginContainer<
   TConfig extends Record<string, unknown> = Record<string, unknown>,
 > =
   | {
-      default?: new () => PluginTypeMap<TInput, TOutput, TConfig>[T];
-    }
+    default?: new () => PluginTypeMap<TInput, TOutput, TConfig>[T];
+  }
   | (new () => PluginTypeMap<TInput, TOutput, TConfig>[T]);
 
 /**
@@ -76,6 +75,7 @@ export class PluginService implements IBaseService {
   private remotes: Map<string, RemoteState> = new Map();
   private instances: Map<string, InstanceState<PluginType>> = new Map();
   private pluginRepository: PluginRepository;
+  private secretServiceApiClient: SecretServiceApiClient;
 
   // Time in milliseconds before cached items are considered stale
   private readonly instanceCacheTimeout: number = 7 * 24 * 60 * 60 * 1000; // 7 days (instance of a plugin with config)
@@ -86,9 +86,10 @@ export class PluginService implements IBaseService {
   private readonly retryDelays: number[] = [1000, 5000]; // Delays between retries in ms
 
   public readonly logger: Logger;
-  constructor(logger: Logger) {
+  constructor(logger: Logger, secretServiceApiClient: SecretServiceApiClient) {
     this.logger = logger;
     this.pluginRepository = new PluginRepository(db);
+    this.secretServiceApiClient = secretServiceApiClient;
   }
 
   /**
@@ -102,9 +103,9 @@ export class PluginService implements IBaseService {
   >(
     name: string,
     pluginConfig: { type: T; config: TConfig },
+    feedId: string,
   ): Promise<PluginTypeMap<TInput, TOutput, TConfig>[T]> {
     try {
-      // Get plugin metadata from database
       const registeredPlugin =
         await this.pluginRepository.getPluginByName(name);
 
@@ -144,7 +145,6 @@ export class PluginService implements IBaseService {
         throw error;
       }
 
-      // Create full config with URL from database
       const config: PluginConfig<T> = {
         type: pluginConfig.type,
         url: registeredPlugin.entryPoint,
@@ -190,7 +190,6 @@ export class PluginService implements IBaseService {
         this.remotes.set(normalizedName, remote);
       }
 
-      // Create and initialize instance with retries
       let lastError: Error | null = null;
       for (let attempt = 0; attempt <= this.retryDelays.length; attempt++) {
         try {
@@ -215,24 +214,62 @@ export class PluginService implements IBaseService {
             TConfig
           >[T];
 
-          // Hydrate config with environment variables
-          const stringifiedConfig = JSON.stringify(config.config);
-          const populatedConfigString = Mustache.render(stringifiedConfig, env); // TODO: Whitelist values
-          const hydratedConfig = JSON.parse(populatedConfigString) as TConfig;
+          this.logger.debug(
+            `PluginService: Preparing to hydrate config for plugin '${name}', feedId '${feedId}'.`,
+          );
+
+          let processedConfig: TConfig = JSON.parse(
+            JSON.stringify(config.config),
+          );
+
+          // --- Secret Hydration Logic ---
+          for (const key in processedConfig) {
+            if (Object.prototype.hasOwnProperty.call(processedConfig, key)) {
+              const value = processedConfig[key];
+
+              if (
+                typeof value === "string" &&
+                value.startsWith("{{") &&
+                value.endsWith("}}")
+              ) {
+                const varName = value.substring(2, value.length - 2).trim(); // e.g., "OPENROUTER_API_KEY"
+
+                this.logger.debug(
+                  `PluginService: Found placeholder '${value}' for key '${key}' in plugin '${name}', feedId '${feedId}'. Attempting to resolve.`,
+                );
+
+                const secret =
+                  await this.secretServiceApiClient.getPlaintextSecret(
+                    feedId,
+                    varName,
+                  );
+
+                if (secret !== null) {
+                  (processedConfig as any)[key] = secret;
+                  this.logger.info(
+                    `PluginService: Hydrated config key '${key}' for plugin '${name}' (feedId '${feedId}') from secret-service for varName '${varName}'.`,
+                  );
+                } else {
+                  this.logger.warn(
+                    `PluginService: Config key '${key}' for plugin '${name}' (feedId '${feedId}') references unknown secret/env var '${varName}'. Placeholder '${value}' remains.`,
+                  );
+                  // Optionally, throw an error if a required secret is missing:
+                  throw new PluginError(`Required secret '${varName}' not found for plugin '${name}'`,
+                    {
+                      pluginName: name, operation: "hydrate"
+                    }, PluginErrorCode.PLUGIN_INITIALIZATION_FAILED
+                  );
+                }
+              }
+            }
+          }
 
-          await newInstance.initialize(hydratedConfig);
+          this.logger.debug(
+            `PluginService: Config hydration complete for plugin '${name}', feedId '${feedId}'.`,
+          );
 
-          // // Validate instance implements required interface
-          // if (!this.validatePluginInterface<T, TInput, TOutput, TConfig>(newInstance, config.type)) {
-          //   throw new PluginInitError(
-          //     name,
-          //     new Error(
-          //       `Plugin does not implement required ${config.type} interface`,
-          //     ),
-          //   );
-          // }
+          await newInstance.initialize(processedConfig);
 
-          // Cache successful instance
           const instanceState: InstanceState<T> = {
             instance: newInstance as PluginTypeMap<
               unknown,
@@ -302,17 +339,17 @@ export class PluginService implements IBaseService {
       throw error instanceof PluginError
         ? error
         : new PluginError(
-            `Unexpected error with plugin ${name}`,
-            {
-              pluginName: name,
-              operation: "unknown",
-            },
-            PluginErrorCode.UNKNOWN_PLUGIN_ERROR,
-            false,
-            {
-              cause: error instanceof Error ? error : undefined,
-            },
-          );
+          `Unexpected error with plugin ${name}`,
+          {
+            pluginName: name,
+            operation: "unknown",
+          },
+          PluginErrorCode.UNKNOWN_PLUGIN_ERROR,
+          false,
+          {
+            cause: error instanceof Error ? error : undefined,
+          },
+        );
     }
   }
 

apps/api/src/services/processor.service.ts

@@ -9,9 +9,9 @@ import { logger } from "../utils/logger";
 import { sanitizeJson } from "../utils/sanitize";
 import { DistributionService } from "./distribution.service";
 import { IBaseService } from "./interfaces/base-service.interface";
-import { TransformationService } from "./transformation.service";
+import { TransformationService } from "./transformation.service.js";
 
-interface ProcessConfig {
+export interface ProcessConfig {
   enabled?: boolean;
   transform?: TransformConfig[];
   distribute?: DistributorConfig[];
@@ -30,9 +30,12 @@ export class ProcessorService implements IBaseService {
 
   /**
    * Process content through transformation pipeline and distribute
-   * Can be used for both individual submissions and bulk content (like recaps)
    */
-  async process(content: RichSubmission, config: ProcessConfig) {
+  async process(
+    content: RichSubmission,
+    config: ProcessConfig,
+    feedId: string,
+  ) {
     try {
       // Apply global transforms if any
       let processed = content;
@@ -42,6 +45,7 @@ export class ProcessorService implements IBaseService {
             processed,
             config.transform,
             "global",
+            feedId,
           );
 
           processed = sanitizeJson(processed);
@@ -75,6 +79,7 @@ export class ProcessorService implements IBaseService {
                   distributorContent,
                   distributor.transform,
                   "distributor",
+                  feedId,
                 );
               distributorContent = sanitizeJson(distributorContent);
             } catch (error) {
@@ -95,6 +100,7 @@ export class ProcessorService implements IBaseService {
           await this.distributionService.distributeContent(
             distributor,
             distributorContent,
+            feedId,
           );
         } catch (error) {
           // Collect errors but continue with other distributors
@@ -132,8 +138,9 @@ export class ProcessorService implements IBaseService {
   async processBatch(
     items: any[],
     config: ProcessConfig & {
-      batchTransform?: TransformConfig[]; // Optional transforms to apply to collected results
+      batchTransform?: TransformConfig[];
     },
+    feedId: string,
   ) {
     try {
       // Process each item through global and distributor transforms
@@ -147,6 +154,7 @@ export class ProcessorService implements IBaseService {
                 processed,
                 config.transform,
                 "global",
+                feedId,
               );
 
               processed = sanitizeJson(processed);
@@ -168,6 +176,7 @@ export class ProcessorService implements IBaseService {
             results,
             config.batchTransform,
             "batch",
+            feedId,
           );
 
           batchResult = sanitizeJson(batchResult);
@@ -197,6 +206,7 @@ export class ProcessorService implements IBaseService {
                 distributorContent,
                 distributor.transform,
                 "distributor",
+                feedId,
               );
 
             distributorContent = sanitizeJson(distributorContent);
@@ -206,6 +216,7 @@ export class ProcessorService implements IBaseService {
           await this.distributionService.distributeContent(
             distributor,
             distributorContent,
+            feedId,
           );
         } catch (error) {
           errors.push(