FRE-4517, FRE-4499: Complete SpamShield implementation and billing updates

- SpamFeedback table migration with timestamp index
- Real-time interception engine completion
- Billing service enhancements
- Classifier and rule engine updates

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/shared-billing/src/services/billing.service.ts

@@ -3,7 +3,19 @@ import { loadBillingConfig, SubscriptionTier } from '../config/billing.config';
 import type { Subscription, SubscriptionCreateSchema, SubscriptionUpdateSchema } from '../models/subscription.model';
 
 const config = loadBillingConfig();
-const stripe = new Stripe(config.stripe.apiKey, { apiVersion: '2024-04-10' });
+const stripe = new Stripe(config.stripe.apiKey, { apiVersion: '2023-10-16' });
+
+const processedEvents = new Map<string, number>();
+const IDEMPOTENCY_TTL_MS = 24 * 60 * 60 * 1000;
+
+function cleanupOldEvents(): void {
+  const now = Date.now();
+  for (const [eventId, timestamp] of processedEvents.entries()) {
+    if (now - timestamp > IDEMPOTENCY_TTL_MS) {
+      processedEvents.delete(eventId);
+    }
+  }
+}
 
 export class BillingService {
   private static instance: BillingService;
@@ -34,6 +46,51 @@ export class BillingService {
     }
   }
 
+  private async verifyCustomerOwnership(
+    customerId: string,
+    userId: string
+  ): Promise<void> {
+    const customer = await stripe.customers.retrieve(customerId);
+    const customerUserId = (customer as Stripe.Customer).metadata?.userId;
+
+    if (customerUserId !== userId) {
+      throw new Error(
+        `Customer ${customerId} does not belong to user ${userId}`
+      );
+    }
+  }
+
+  async getUserTier(userId: string): Promise<SubscriptionTier | null> {
+    try {
+      const customers = await stripe.customers.list({
+        limit: 100,
+        expand: ['data.subscriptions'],
+      });
+
+      const customer = customers.data.find(
+        (c: Stripe.Customer) => 
+          c.metadata?.userId === userId && c.subscriptions?.data.length && c.subscriptions.data.length > 0
+      );
+
+      if (!customer || !customer.subscriptions) {
+        return null;
+      }
+
+      const activeSubscription = customer.subscriptions.data.find(
+        (sub: Stripe.Subscription) => sub.status === 'active'
+      );
+
+      if (!activeSubscription) {
+        return null;
+      }
+
+      const tier = activeSubscription.metadata?.tier as SubscriptionTier;
+      return tier || null;
+    } catch {
+      return null;
+    }
+  }
+
   async createSubscription(
     userId: string,
     tier: SubscriptionTier,
@@ -41,21 +98,36 @@ export class BillingService {
   ): Promise<{ subscription: Stripe.Subscription; customer: Stripe.Customer }> {
     const tierConfig = config.tiers[tier];
 
+    const customer = await this.getCustomer(customerId);
+    if (!customer) {
+      throw new Error(`Customer ${customerId} not found`);
+    }
+
+    await this.verifyCustomerOwnership(customerId, userId);
+
     const subscription = await stripe.subscriptions.create({
       customer: customerId,
       items: [{ price: tierConfig.priceId }],
       metadata: { userId, tier },
     });
 
-    const customer = await this.getCustomer(customerId);
-
-    return { subscription, customer: customer! };
+    return { subscription, customer };
   }
 
   async cancelSubscription(
     subscriptionId: string,
+    userId: string,
     cancelAtPeriodEnd: boolean = false
   ): Promise<Stripe.Subscription> {
+    const subscription = await stripe.subscriptions.retrieve(subscriptionId);
+    const customerUserId = subscription.metadata?.userId;
+
+    if (customerUserId !== userId) {
+      throw new Error(
+        `Subscription ${subscriptionId} does not belong to user ${userId}`
+      );
+    }
+
     if (cancelAtPeriodEnd) {
       return await stripe.subscriptions.update(subscriptionId, {
         cancel_at_period_end: true,
@@ -66,11 +138,19 @@ export class BillingService {
 
   async updateSubscription(
     subscriptionId: string,
+    userId: string,
     newTier: SubscriptionTier
   ): Promise<Stripe.Subscription> {
-    const newTierConfig = config.tiers[newTier];
-    
     const subscription = await stripe.subscriptions.retrieve(subscriptionId);
+    const customerUserId = subscription.metadata?.userId;
+
+    if (customerUserId !== userId) {
+      throw new Error(
+        `Subscription ${subscriptionId} does not belong to user ${userId}`
+      );
+    }
+
+    const newTierConfig = config.tiers[newTier];
     
     const updated = await stripe.subscriptions.update(subscriptionId, {
       proration_behavior: 'create_prorations',
@@ -104,6 +184,29 @@ export class BillingService {
     }
   }
 
+  async getUserSubscription(userId: string): Promise<Stripe.Subscription | null> {
+    try {
+      const customers = await stripe.customers.list({
+        limit: 100,
+        expand: ['data.subscriptions'],
+      });
+
+      for (const customer of customers.data) {
+        if (customer.metadata?.userId === userId && customer.subscriptions) {
+          const activeSub = customer.subscriptions.data.find(
+            (sub: Stripe.Subscription) => sub.status === 'active'
+          );
+          if (activeSub) {
+            return activeSub;
+          }
+        }
+      }
+      return null;
+    } catch {
+      return null;
+    }
+  }
+
   async getTierLimits(tier: SubscriptionTier) {
     return config.tiers[tier];
   }
@@ -130,24 +233,41 @@ export class BillingService {
     description: string,
     metadata?: Record<string, string>
   ): Promise<Stripe.Invoice> {
-    return await stripe.invoices.create({
+    const invoice = await stripe.invoices.create({
       customer: customerId,
-      line_items: [
-        {
-          amount_data: { currency: 'usd', unit_amount: amount },
-          description: description,
-          quantity: 1,
-        },
-      ],
       metadata: metadata,
     });
+    
+    await stripe.invoiceItems.create({
+      invoice: invoice.id,
+      customer: customerId,
+      price_data: {
+        currency: 'usd',
+        unit_amount: amount,
+        product: 'default_product',
+      },
+      description: description,
+      quantity: 1,
+    });
+    
+    return await stripe.invoices.retrieve(invoice.id);
   }
 
   async handleWebhook(
     sig: string,
     body: Buffer
-  ): Promise<Stripe.Event> {
-    return stripe.webhooks.constructEvent(body, sig, config.stripe.webhookSecret);
+  ): Promise<Stripe.Event | null> {
+    const event = stripe.webhooks.constructEvent(body, sig, config.stripe.webhookSecret);
+    
+    cleanupOldEvents();
+
+    if (processedEvents.has(event.id)) {
+      return null;
+    }
+
+    processedEvents.set(event.id, Date.now());
+
+    return event;
   }
 
   async getInvoiceHistory(customerId: string): Promise<Stripe.ApiList<Stripe.Invoice>> {