feat: implement organization-specific API keys and tenant-aware webhooks (Phase 2)

apps/api/src/routes/payments.ts

@@ -103,7 +103,12 @@ export async function stripeWebhookRoute(fastify: FastifyInstance) {
         }
     });
 
-    fastify.post('/webhook', async (request, reply) => {
+    // ── POST /webhook or /webhook/:organizationId ───────────────────────────
+    fastify.post('/webhook', async (request, reply) => handleWebhook(request, reply));
+    fastify.post('/webhook/:organizationId', async (request, reply) => handleWebhook(request, reply));
+
+    async function handleWebhook(request: any, reply: any) {
+        const { organizationId } = request.params as { organizationId?: string };
         const sig = request.headers['stripe-signature'];
 
         if (!sig || typeof sig !== 'string') {
@@ -115,10 +120,10 @@ export async function stripeWebhookRoute(fastify: FastifyInstance) {
         try {
             const rawBody = request.rawBody;
             if (!rawBody) throw new Error('Missing raw body');
-            event = stripeService.verifyWebhookSignature(rawBody, sig);
+            event = await stripeService.verifyWebhookSignature(rawBody, sig, organizationId);
         } catch (err: unknown) {
             const errorMsg = err instanceof Error ? err.message : String(err);
-            fastify.log.warn(`[Stripe Webhook] Signature verification failed: ${errorMsg}`);
+            fastify.log.warn(`[Stripe Webhook] Signature verification failed for Org ${organizationId || 'global'}: ${errorMsg}`);
             return reply.status(400).send(`Webhook Error: ${errorMsg}`);
         }
 

apps/api/src/routes/whatsapp.ts

@@ -3,6 +3,7 @@ import { FastifyInstance } from 'fastify';
 import crypto from 'crypto';
 import { z } from 'zod';
 import { getOrganizationByPhoneNumberId } from '../services/organization';
+import { whatsappQueue } from '../services/queue';
 
 // ─── Zod Schema for WhatsApp Webhook Payload ─────────────────────────────────
 const WhatsAppMessageSchema = z.object({
@@ -65,11 +66,7 @@ function verifyWebhookSignature(rawBody: Buffer, signature: string | undefined,
 
 // ─── Route Plugin ─────────────────────────────────────────────────────────────
 export async function whatsappRoutes(fastify: FastifyInstance) {
-    // ── Raw body capture for HMAC verification ──────────────────────────────
-    // We need the raw buffer BEFORE JSON.parse, so we override the content type parser
-    // for this specific route scope only.
     fastify.addContentTypeParser('application/json', { parseAs: 'buffer' }, (req, body, done) => {
-        // Store raw body on request for signature verification
         req.rawBody = body as Buffer;
         try {
             done(null, JSON.parse(body.toString('utf8')));
@@ -78,159 +75,82 @@ export async function whatsappRoutes(fastify: FastifyInstance) {
         }
     });
 
-    // ── GET /webhook — Meta verification handshake ──────────────────────────
     fastify.get('/webhook', async (request, reply) => {
         const query = request.query as Record<string, string>;
         const mode = query['hub.mode'];
         const token = query['hub.verify_token'];
         const challenge = query['hub.challenge'];
 
-        // Parameterless ping from HF proxy or Meta health probe
-        if (!mode && !token && !challenge) {
-            return reply.code(200).type('text/plain').send('ok');
-        }
+        if (!mode && !token && !challenge) return reply.code(200).type('text/plain').send('ok');
 
-        // Meta verification
         if (mode === 'subscribe' && token === process.env.WHATSAPP_VERIFY_TOKEN) {
-            request.log.info('[WEBHOOK] Meta verification successful');
             return reply.code(200).type('text/plain').send(challenge);
         }
-
-        request.log.warn('[WEBHOOK] Meta verification failed — token mismatch or wrong mode');
-        return reply.code(403).type('text/plain').send('Forbidden');
+        return reply.code(403).send('Forbidden');
     });
 
-    // ── POST /webhook — Incoming Meta events ────────────────────────────────
-    fastify.post('/webhook', async (request, reply) => {
-        // ── 1. HMAC Signature Verification ──────────────────────────────────
-        logger.info("[RAW-WHATSAPP-PAYLOAD]", JSON.stringify(request.body, null, 2));
-        const appSecret = process.env.WHATSAPP_APP_SECRET;
+    fastify.post('/webhook', async (request, reply) => handleIncoming(request, reply));
+    fastify.post('/webhook/:organizationId', async (request, reply) => handleIncoming(request, reply));
 
+    async function handleIncoming(request: any, reply: any) {
+        const { organizationId: urlOrgId } = request.params as { organizationId?: string };
+        
+        // 1. HMAC Verification
+        const appSecret = process.env.WHATSAPP_APP_SECRET;
         if (appSecret) {
             const signature = request.headers['x-hub-signature-256'] as string;
-            const rawBody = request.rawBody;
-
-            // Optional: verify webhook signature. Important for production on both HF and Railway.
-            if (!rawBody || !verifyWebhookSignature(rawBody, signature, appSecret)) {
-                request.log.warn('[WEBHOOK] Invalid HMAC signature — request rejected');
-                // Meta won't retry if we return 401/403 directly for invalid signatures. 
+            if (!request.rawBody || !verifyWebhookSignature(request.rawBody, signature, appSecret)) {
+                request.log.warn(`[WEBHOOK] Invalid HMAC for Org ${urlOrgId || 'global'}`);
                 return reply.code(403).send({ error: 'Invalid signature' });
             }
-        } else {
-            // Log a warning but allow in development (no secret set)
-            request.log.warn('[WEBHOOK] WHATSAPP_APP_SECRET not set — skipping signature verification');
         }
 
-        // ── 2. Forward to Railway Internal Worker (if configured as Gateway) ──
+        // 2. Gateway Forwarding
         const railwayInternalUrl = process.env.RAILWAY_INTERNAL_URL;
-
-        // Hardened Gateway Detection:
-        // We are a gateway if:
-        // 1. IS_GATEWAY is explicitly true
-        // 2. We are running on Hugging Face (HF_SPACE_ID exists)
-        // 3. We have a RAILWAY_INTERNAL_URL but NOT a RAILWAY_STATIC_URL (not on Railway)
-        const isGateway =
-            process.env.IS_GATEWAY === 'true' ||
-            process.env.HF_SPACE_ID !== undefined ||
-            (!!railwayInternalUrl && !process.env.RAILWAY_STATIC_URL);
+        const isGateway = process.env.IS_GATEWAY === 'true' || !!process.env.HF_SPACE_ID;
 
         if (railwayInternalUrl && isGateway) {
-            // Loop protection: don't forward to yourself
-            const myUrl = process.env.RAILWAY_PUBLIC_URL || '';
-            if (railwayInternalUrl.includes(myUrl) && myUrl !== '') {
-                request.log.warn('[WEBHOOK] Loop detected: RAILWAY_INTERNAL_URL matches self. Skipping forward.');
-            } else {
-                const targetUrl = `${railwayInternalUrl.replace(/\/$/, '')}/v1/internal/whatsapp/inbound`;
-                request.log.info(`[WEBHOOK] GATEWAY MODE: Forwarding payload to ${targetUrl}`);
-
-                try {
-                    // Fire and forget (don't await) to ensure fast 200 response to Meta
-                    fetch(targetUrl, {
-                        method: 'POST',
-                        headers: {
-                            'Content-Type': 'application/json',
-                            'Authorization': `Bearer ${process.env.ADMIN_API_KEY || ''}`,
-                            'X-Organization-Id': request.headers['x-organization-id'] as string || 'default-org-id'
-                        },
-                        body: request.body ? JSON.stringify(request.body) : ''
-                    }).then(res => {
-                        request.log.info(`[WEBHOOK] Gateway forward result: ${res.status} ${res.statusText}`);
-                    }).catch(err => {
-                        request.log.error(`[WEBHOOK] Forward to Railway failed: ${err instanceof Error ? err.message : String(err)}`);
-                    });
-
-                    // 🚨 CRITICAL: CRUCIAL EXIT POINT FOR GATEWAY
-                    return reply.code(200).send('EVENT_RECEIVED');
-                } catch (error: unknown) {
-                    request.log.error(`[WEBHOOK] Forward throwing error: ${error instanceof Error ? error.message : String(error)}`);
-                    // Still return 200 to Meta to avoid retries, even if gateway forward failed internally
-                    return reply.code(200).send('EVENT_RECEIVED_FW_ERR');
-                }
-            }
-        }
-
-        // ── 3. DETACH IMMEDIATELY ──
-        // Respond 200 OK right now to release the HF Gateway connection.
-        // We do this BEFORE any parsing or work.
-        if (!reply.sent) {
-            reply.code(200).send('EVENT_RECEIVED');
+            const targetUrl = `${railwayInternalUrl.replace(/\/$/, '')}/v1/internal/whatsapp/inbound`;
+            fetch(targetUrl, {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json',
+                    'Authorization': `Bearer ${process.env.ADMIN_API_KEY}`,
+                    'x-organization-id': urlOrgId || ''
+                },
+                body: JSON.stringify(request.body)
+            }).catch(err => logger.error('[WEBHOOK] Forwarding failed:', err));
+
+            return reply.code(200).send({ status: 'forwarded' });
         }
 
-        // ── 4. Background Processing (enqueue to Worker) ──
-        const body = request.body;
-        setImmediate(async () => {
-            try {
-                const parsed = WebhookPayloadSchema.safeParse(body);
-                if (!parsed.success) {
-                    fastify.log.warn(`[WEBHOOK] Failed to parse webhook payload: ${parsed.error.message}`);
-                    return;
-                }
-
-                const { scheduleInboundMessage } = await import('../services/queue');
-                const payload = parsed.data;
-
-                // 🏢 Multi-Tenant Routing (metadata is at change level)
-                const phoneNumberId = payload.entry?.[0]?.changes?.[0]?.value?.metadata?.phone_number_id || 'unknown';
-                const organizationId = await getOrganizationByPhoneNumberId(phoneNumberId);
-
-                // Use Shared Utility
-                const { extractWhatsAppPayload } = await import('@repo/shared-types');
-                const extractedMessages = extractWhatsAppPayload(body);
-
-                for (const msg of extractedMessages) {
-                    logger.info(`[WEBHOOK-TRACE] Processing message from ${msg.phone} (Org: ${organizationId})`);
-
-                    if (msg.text !== undefined) {
-                        await scheduleInboundMessage({ 
-                            phone: msg.phone, 
-                            text: msg.text, 
-                            messageId: msg.messageId, 
-                            organizationId 
-                        });
-                    } else if (msg.mediaId && msg.mediaType) {
-                        const accessToken = process.env.WHATSAPP_ACCESS_TOKEN || undefined;
-                        const { whatsappQueue: q } = await import('../services/queue');
-                        
-                        await q.add('download-media', {
-                            mediaId: msg.mediaId,
-                            mimeType: msg.mediaType === 'image' ? 'image/jpeg' : 'audio/ogg',
-                            phone: msg.phone,
-                            organizationId,
-                            caption: msg.caption,
-                            ...(accessToken ? { accessToken } : {})
-                        }, { priority: 1, attempts: 3, backoff: { type: 'exponential', delay: 2000 } });
-
-                        await q.add('send-message-direct', {
-                            phone: msg.phone,
-                            text: msg.mediaType === 'image' ? "⏳ J'analyse ton image..." : "⏳ J'analyse ton audio...",
-                            organizationId
-                        });
-                    }
+        // 3. Queue for Processing
+        const parsed = WebhookPayloadSchema.safeParse(request.body);
+        if (!parsed.success) return reply.code(200).send({ status: 'ignored' });
+
+        for (const entry of parsed.data.entry) {
+            for (const change of entry.changes) {
+                const phoneNumberId = change.value.metadata?.phone_number_id || 'unknown';
+                
+                // Use URL OrgId if present, otherwise resolve from phone number
+                const organizationId = urlOrgId || await getOrganizationByPhoneNumberId(phoneNumberId);
+
+                for (const message of change.value.messages || []) {
+                    await whatsappQueue.add('process-message', {
+                        message,
+                        organizationId,
+                        metadata: {
+                            phoneNumberId,
+                            displayPhoneNumber: change.value.metadata?.display_phone_number
+                        }
+                    }, {
+                        attempts: 3,
+                        backoff: { type: 'exponential', delay: 1000 }
+                    });
                 }
-            } catch (error) {
-                fastify.log.error(`[WEBHOOK] Detached processing error: ${String(error)}`);
             }
-        });
+        }
 
-    });
+        return reply.code(200).send({ status: 'received' });
+    }
 }

apps/api/src/services/ai/index.ts

@@ -15,6 +15,7 @@ import { getOrganizationId } from '@repo/database';
 import { prisma } from '../prisma';
 import { redis } from '../queue';
 import { ProviderRegistry, ProviderCapability } from './ProviderRegistry';
+import { getTenantSecrets } from '../organization';
 
 class AIService {
     private registry: ProviderRegistry;
@@ -75,22 +76,61 @@ class AIService {
         }
     }
 
+    private async getProvidersForTenant(capability: ProviderCapability, organizationId?: string) {
+        if (!organizationId) return this.registry.getProvidersFor(capability);
+
+        // Check if tenant has custom keys
+        const secrets = await getTenantSecrets(organizationId);
+        if (!secrets || (!secrets.openAiApiKey && !secrets.googleAiApiKey)) {
+            return this.registry.getProvidersFor(capability);
+        }
+
+        // Create temporary registry for this request with tenant keys
+        const tenantRegistry = new ProviderRegistry();
+        
+        if (secrets.googleAiApiKey) {
+            tenantRegistry.register('GEMINI_TENANT', new GeminiProvider(secrets.googleAiApiKey), 1000, [
+                ProviderCapability.TEXT,
+                ProviderCapability.VISION
+            ]);
+        }
+
+        if (secrets.openAiApiKey) {
+            tenantRegistry.register('OPENAI_TENANT', new OpenAIProvider(secrets.openAiApiKey), 500, [
+                ProviderCapability.TEXT,
+                ProviderCapability.VISION,
+                ProviderCapability.AUDIO_TRANSCRIPTION,
+                ProviderCapability.SPEECH_GENERATION,
+                ProviderCapability.IMAGE_GENERATION
+            ]);
+        }
+
+        // Add global providers as fallback
+        const globalProviders = this.registry.getProvidersFor(capability);
+        for (const p of globalProviders) {
+            tenantRegistry.register(p.name, p.instance, p.priority, p.capabilities);
+        }
+
+        return tenantRegistry.getProvidersFor(capability);
+    }
+
     private async callWithFailover<T>(
         prompt: string,
         schema: z.ZodSchema<T>,
         temperature?: number,
         imageUrl?: string
     ): Promise<{ data: T, source: string }> {
+        const organizationId = getOrganizationId();
         const capability = imageUrl ? ProviderCapability.VISION : ProviderCapability.TEXT;
-        const providers = this.registry.getProvidersFor(capability);
+        const providers = await this.getProvidersForTenant(capability, organizationId);
 
         for (const provider of providers) {
             try {
                 const data = await provider.instance.generateStructuredData(prompt, schema, temperature, imageUrl);
-                logger.info(`[AI_INFO] ${provider.name} used successfully. (Capability: ${capability})`);
+                logger.info(`[AI_INFO] ${provider.name} used successfully for Org: ${organizationId || 'global'}. (Capability: ${capability})`);
                 return { data, source: provider.name };
             } catch (err) {
-                logger.warn(`[AI_WARNING] ${provider.name} failed: ${(err as Error).message}. Trying next provider...`);
+                logger.warn(`[AI_WARNING] ${provider.name} failed for Org ${organizationId || 'global'}: ${(err as Error).message}. Trying next provider...`);
             }
         }
 

apps/api/src/services/organization.ts

@@ -73,5 +73,28 @@ export function encryptSecrets(data: any) {
 export function decryptSecrets(org: any) {
     if (org.systemUserToken) org.systemUserToken = decrypt(org.systemUserToken, ENCRYPTION_SECRET);
     if (org.webhookSecret) org.webhookSecret = decrypt(org.webhookSecret, ENCRYPTION_SECRET);
+    if (org.openAiApiKey) org.openAiApiKey = decrypt(org.openAiApiKey, ENCRYPTION_SECRET);
+    if (org.googleAiApiKey) org.googleAiApiKey = decrypt(org.googleAiApiKey, ENCRYPTION_SECRET);
+    if (org.stripeSecretKey) org.stripeSecretKey = decrypt(org.stripeSecretKey, ENCRYPTION_SECRET);
     return org;
 }
+
+/**
+ * Retrieves all secrets for a tenant, decrypted.
+ */
+export async function getTenantSecrets(organizationId: string) {
+    const org = await prisma.organization.findUnique({
+        where: { id: organizationId },
+        select: {
+            systemUserToken: true,
+            webhookSecret: true,
+            openAiApiKey: true,
+            googleAiApiKey: true,
+            stripeSecretKey: true,
+            stripeWebhookSecret: true
+        }
+    });
+    
+    if (!org) return null;
+    return decryptSecrets(org);
+}

apps/api/src/services/stripe.ts

@@ -1,12 +1,11 @@
 import { logger } from '../logger';
 import Stripe from 'stripe';
 import { PaymentProvider, CheckoutSessionParams } from './payments/types';
+import { getTenantSecrets } from './organization';
 
 export class StripeService implements PaymentProvider {
     public name = 'stripe';
-    private stripe: Stripe | null = null;
-    private webhookSecret: string | null = null;
-    private clientUrl: string;
+    private instances: Map<string, { stripe: Stripe, webhookSecret: string | null }> = new Map();
 
     constructor() {
         const secretKey = process.env.STRIPE_SECRET_KEY;
@@ -22,11 +21,37 @@ export class StripeService implements PaymentProvider {
         }
     }
 
+    private async getStripeInstance(organizationId?: string): Promise<{ stripe: Stripe | null, webhookSecret: string | null }> {
+        if (!organizationId) return { stripe: this.stripe, webhookSecret: this.webhookSecret };
+
+        // Check cache
+        if (this.instances.has(organizationId)) {
+            return this.instances.get(organizationId)!;
+        }
+
+        // Check DB for tenant secrets
+        const secrets = await getTenantSecrets(organizationId);
+        if (secrets?.stripeSecretKey) {
+            const instance = {
+                stripe: new Stripe(secrets.stripeSecretKey, {
+                    apiVersion: '2025-01-27.acacia' as any,
+                }),
+                webhookSecret: secrets.stripeWebhookSecret || null
+            };
+            this.instances.set(organizationId, instance);
+            return instance;
+        }
+
+        // Fallback to global
+        return { stripe: this.stripe, webhookSecret: this.webhookSecret };
+    }
+
     /**
      * Unified checkout session creator for the interface
      */
     async createCheckoutSession(params: CheckoutSessionParams): Promise<string> {
-        if (!this.stripe) throw new Error('[StripeService] STRIPE_SECRET_KEY is not configured');
+        const { stripe } = await this.getStripeInstance(params.organizationId);
+        if (!stripe) throw new Error('[StripeService] Stripe is not configured for this organization');
         
         // Decide mode based on params
         const isSubscription = !!params.organizationId && !params.trackId;
@@ -36,7 +61,7 @@ export class StripeService implements PaymentProvider {
         if (!priceId) throw new Error('[StripeService] Missing Price ID for checkout');
 
         try {
-            const session = await this.stripe.checkout.sessions.create({
+            const session = await stripe.checkout.sessions.create({
                 payment_method_types: ['card'],
                 line_items: [{ price: priceId, quantity: 1 }],
                 mode: mode as any,
@@ -75,19 +100,21 @@ export class StripeService implements PaymentProvider {
     /**
      * Verifies the signature of an incoming Stripe webhook.
      */
-    verifyWebhookSignature(payload: Buffer, signature: string | undefined): Stripe.Event {
-        if (!this.stripe || !this.webhookSecret) {
-            throw new Error('[StripeService] Stripe is not configured (missing keys)');
+    async verifyWebhookSignature(payload: Buffer, signature: string | undefined, organizationId?: string): Promise<Stripe.Event> {
+        const { stripe, webhookSecret } = await this.getStripeInstance(organizationId);
+        
+        if (!stripe || !webhookSecret) {
+            throw new Error('[StripeService] Stripe is not configured for this organization');
         }
         if (!signature) {
             throw new Error('Missing stripe-signature header');
         }
 
         try {
-            return this.stripe.webhooks.constructEvent(
+            return stripe.webhooks.constructEvent(
                 payload,
                 signature,
-                this.webhookSecret
+                webhookSecret
             );
         } catch (err: unknown) {
             throw new Error(`Webhook Error: ${(err instanceof Error ? err.message : String(err))}`);
@@ -97,11 +124,12 @@ export class StripeService implements PaymentProvider {
     /**
      * Creates a link to the Stripe Customer Portal for subscription management.
      */
-    async createCustomerPortalSession(customerId: string) {
-        if (!this.stripe) throw new Error('[StripeService] Stripe not configured');
+    async createCustomerPortalSession(customerId: string, organizationId?: string) {
+        const { stripe } = await this.getStripeInstance(organizationId);
+        if (!stripe) throw new Error('[StripeService] Stripe not configured for this organization');
         
         try {
-            const session = await this.stripe.billingPortal.sessions.create({
+            const session = await stripe.billingPortal.sessions.create({
                 customer: customerId,
                 return_url: `${this.clientUrl}/settings`,
             });