Merge pull request #2664 from ankaboot-source/feat/add-fetch-mining-source-edge-function

feat: add central fetch-mining-source edge function

Author: baderdean

AGENTS.md

@@ -84,6 +84,34 @@ npm run build            # Compile TypeScript
 npm run dev:supabase-functions  # Serve functions locally
 ```
 
+#### Creating New Edge Functions
+
+**Always use the Supabase CLI to create new edge functions:**
+
+```bash
+# Create new edge function (auto-generates proper structure)
+npx supabase functions new <function-name>
+```
+
+**Why?** The CLI:
+
+- Creates proper directory structure
+- Sets up deno.json with correct permissions
+- Ensures function is registered with Supabase
+
+**Example workflow:**
+
+```bash
+# 1. Create the function scaffold
+npx supabase functions new my-new-function
+
+# 2. Implement your code in the generated index.ts
+# (copy your implementation into the created file)
+
+# 3. Test locally
+npm run dev:supabase-functions
+```
+
 ## Code Style Guidelines
 
 ### TypeScript Configuration

backend/.env.dev

@@ -22,6 +22,7 @@ LEADMINER_API_LOG_LEVEL = debug   # Logging level (debug, info, notice, warning.
 # ==============| API - SUPABASE |============= #
 SUPABASE_PROJECT_URL = http://127.0.0.1:54321   # ( REQUIRED ) Supabase project URL (e.g., https://db.yourdomain.com for self-hosted/prod)
 SUPABASE_SECRET_PROJECT_TOKEN = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU # ( REQUIRED ) Supabase project token
+SUPABASE_SERVICE_ROLE_KEY = eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU # ( REQUIRED ) Supabase service role key
 PG_CONNECTION_STRING = postgresql://postgres:postgres@127.0.0.1:54322/postgres  # ( REQUIRED ) Postgres connection string
 
 

backend/src/config/schema.ts

@@ -45,6 +45,7 @@ const schema = z.object({
   /* SUPABASE + POSTGRES */
   SUPABASE_PROJECT_URL: z.string().url(),
   SUPABASE_SECRET_PROJECT_TOKEN: z.string().min(1),
+  SUPABASE_SERVICE_ROLE_KEY: z.string().min(1),
   PG_CONNECTION_STRING: z.string().url(),
 
   /* SENTRY */

backend/src/controllers/imap.controller.ts

@@ -9,6 +9,7 @@ import azureOAuth2Client from '../services/OAuth2/azure';
 import googleOAuth2Client from '../services/OAuth2/google';
 import ImapBoxesFetcher from '../services/imap/ImapBoxesFetcher';
 import ImapConnectionProvider from '../services/imap/ImapConnectionProvider';
+import { miningSourceService } from '../services/MiningSourceService';
 import { ImapAuthError } from '../utils/errors';
 import hashEmail from '../utils/helpers/hashHelpers';
 import validateType from '../utils/helpers/validation';
@@ -93,20 +94,34 @@ export default function initializeImapController(miningSources: MiningSources) {
             });
 
           if (token.expired(1000)) {
-            const newToken = (await token.refresh()).token as NewToken;
-
-            await upsertMiningSource(
-              miningSources,
+            const { sources } = await miningSourceService.getSourcesForUser(
               userId,
-              {
-                ...newToken,
-                refresh_token: newToken.refresh_token ?? refreshToken
-              },
-              provider,
               data.email
             );
 
-            data.accessToken = newToken.access_token;
+            const refreshedSource = sources.find((s) => s.email === data.email);
+            if (
+              refreshedSource &&
+              'accessToken' in refreshedSource.credentials
+            ) {
+              data.accessToken = refreshedSource.credentials
+                .accessToken as string;
+            } else {
+              const newToken = (await token.refresh()).token as NewToken;
+
+              await upsertMiningSource(
+                miningSources,
+                userId,
+                {
+                  ...newToken,
+                  refresh_token: newToken.refresh_token ?? refreshToken
+                },
+                provider,
+                data.email
+              );
+
+              data.accessToken = newToken.access_token;
+            }
           }
         }
 

backend/src/services/MiningSourceService.ts

@@ -0,0 +1,48 @@
+import ENV from '../config';
+
+export interface MiningSourceWithCredentials {
+  email: string;
+  type: string;
+  credentials: Record<string, unknown>;
+}
+
+export class MiningSourceService {
+  private supabaseUrl: string;
+
+  constructor() {
+    this.supabaseUrl = ENV.SUPABASE_PROJECT_URL;
+  }
+
+  async getSourcesForUser(
+    userId: string,
+    email?: string
+  ): Promise<{
+    sources: MiningSourceWithCredentials[];
+    refreshed: string[];
+  }> {
+    const response = await fetch(
+      `${this.supabaseUrl}/functions/v1/fetch-mining-source`,
+      {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          'x-service-key': ENV.SUPABASE_SERVICE_ROLE_KEY
+        },
+        body: JSON.stringify({
+          email: email || 'all',
+          mode: 'service',
+          user_id: userId
+        })
+      }
+    );
+
+    if (!response.ok) {
+      const error = await response.text();
+      throw new Error(`Failed to fetch mining sources: ${error}`);
+    }
+
+    return response.json();
+  }
+}
+
+export const miningSourceService = new MiningSourceService();

supabase/config.toml

@@ -337,3 +337,14 @@ entrypoint = "./functions/delete-mining-source/index.ts"
 # Specifies static files to be bundled with the function. Supports glob patterns.
 # For example, if you want to serve static HTML pages in your function:
 # static_files = [ "./functions/delete-mining-source/*.html" ]
+
+[functions.fetch-mining-source]
+enabled = true
+verify_jwt = true
+import_map = "./functions/fetch-mining-source/deno.json"
+# Uncomment to specify a custom file path to the entrypoint.
+# Supported file extensions are: .ts, .js, .mjs, .jsx, .tsx
+entrypoint = "./functions/fetch-mining-source/index.ts"
+# Specifies static files to be bundled with the function. Supports glob patterns.
+# For example, if you want to serve static HTML pages in your function:
+# static_files = [ "./functions/fetch-mining-source/*.html" ]

supabase/functions/email-campaigns/index.ts

@@ -526,22 +526,30 @@ async function getAuthenticatedUser(c: Context) {
 }
 
 async function getUserMiningSources(authorization: string) {
-  const supabase = createSupabaseClient(authorization);
-  const { data, error } = await supabase
-    .schema("private")
-    .rpc("get_user_mining_source_credentials", {
-      _encryption_key: LEADMINER_API_HASH_SECRET,
-    });
+  const response = await fetch(`${SUPABASE_URL}/functions/v1/fetch-mining-source`, {
+    method: "POST",
+    headers: {
+      "Content-Type": "application/json",
+      "Authorization": authorization,
+    },
+    body: JSON.stringify({ email: "all" }),
+  });
 
-  if (error) {
-    throw new Error(`Unable to fetch mining credentials: ${error.message}`);
+  if (!response.ok) {
+    const errorText = await response.text();
+    throw new Error(`Failed to fetch mining sources: ${response.status} ${errorText}`);
   }
 
-  return (data ?? []) as {
-    email: string;
-    type: string;
-    credentials: Record<string, unknown>;
-  }[];
+  const result = await response.json() as {
+    sources: { email: string; type: string; credentials: Record<string, unknown> }[];
+    refreshed: string[];
+  };
+
+  if (result.refreshed.length > 0) {
+    logger.info("Tokens refreshed via central function", { emails: result.refreshed });
+  }
+
+  return result.sources;
 }
 
 async function guessCustomSmtpHost(email: string) {
@@ -635,43 +643,11 @@ async function resolveSenderOptions(authorization: string, userEmail: string) {
   const transportBySender: Record<string, Transport | null> = {
     [fallbackSenderEmail]: null,
   };
-  const supabaseAdmin = createSupabaseAdmin();
 
   const sources = listUniqueSenderSources(
     await getUserMiningSources(authorization),
   );
 
-  // Refresh expired OAuth tokens
-  for (let i = 0; i < sources.length; i++) {
-    const source = sources[i];
-    const credentialIssue = getSenderCredentialIssue(source);
-
-    if (credentialIssue?.includes("expired")) {
-      try {
-        const refreshed = await refreshOAuthToken(source);
-        if (refreshed) {
-          const updated = await updateMiningSourceCredentials(
-            supabaseAdmin,
-            source.email,
-            refreshed.credentials,
-          );
-          if (updated) {
-            sources[i] = refreshed;
-          }
-        } else {
-          logger.warn("Could not refresh token, source will remain unavailable", {
-            email: source.email,
-          });
-        }
-      } catch (error) {
-        logger.error("Failed to refresh token for source", {
-          email: source.email,
-          error: error instanceof Error ? error.message : String(error),
-        });
-      }
-    }
-  }
-
   for (const source of sources) {
     const nowMs = Date.now();
     const expired = isTokenExpired(source.credentials, nowMs);
@@ -2068,9 +2044,9 @@ app.get("/unsubscribe/:token", async (c: Context) => {
     status: 302,
     headers: {
       ...corsHeaders,
-      Location: success  });
-});Url,
+      Location: successUrl,
     },
+  });
 
 
 app.get("/track/open/:token", async (c: Context) => {

supabase/functions/fetch-mining-source/.npmrc

@@ -0,0 +1,3 @@
+# Configuration for private npm package dependencies
+# For more information on using private registries with Edge Functions, see:
+# https://supabase.com/docs/guides/functions/import-maps#importing-from-private-registries

supabase/functions/fetch-mining-source/deno.json

@@ -0,0 +1,3 @@
+{
+  "imports": {}
+}