Feat/idempotency logic on dreamsync adapter

platforms/dreamsync-api/src/context/OperationContext.ts

@@ -0,0 +1,123 @@
+/**
+ * Request-scoped context system to track database operation origins
+ * This ensures we only trigger webhooks for operations from specific services
+ */
+
+interface OperationContext {
+    serviceName: string;
+    operationId: string;
+    timestamp: number;
+}
+
+class OperationContextManager {
+    private static instance: OperationContextManager;
+    private contexts: Map<string, OperationContext> = new Map();
+    private readonly CONTEXT_EXPIRY_MS = 30_000; // 30 seconds
+
+    static getInstance(): OperationContextManager {
+        if (!OperationContextManager.instance) {
+            OperationContextManager.instance = new OperationContextManager();
+        }
+        return OperationContextManager.instance;
+    }
+
+    /**
+     * Create a new operation context for a specific service
+     */
+    createContext(serviceName: string, operationId: string): string {
+        const contextKey = `${serviceName}:${operationId}:${Date.now()}`;
+        const context: OperationContext = {
+            serviceName,
+            operationId,
+            timestamp: Date.now()
+        };
+
+        this.contexts.set(contextKey, context);
+
+        // Clean up expired contexts
+        this.cleanupExpiredContexts();
+
+        console.log(`🔧 Created operation context: ${contextKey}`);
+        return contextKey;
+    }
+
+    /**
+     * Check if an operation should be processed based on context
+     * Only applies protection to groups and messages entities
+     */
+    shouldProcessOperation(entityId: string, tableName: string): boolean {
+        // Only apply context protection to groups and messages
+        const protectedEntities = ['groups', 'messages'];
+        if (!protectedEntities.includes(tableName.toLowerCase())) {
+            console.log(`✅ Entity ${tableName} is not protected, allowing operation for ${entityId}`);
+            return true;
+        }
+
+        // Look for any active ConsentService context
+        for (const [contextKey, context] of this.contexts.entries()) {
+            if (context.serviceName === 'ConsentService') {
+                console.log(`✅ Found ConsentService context: ${contextKey}, processing protected operation for ${tableName}:${entityId}`);
+                return true;
+            }
+        }
+
+        console.log(`❌ No ConsentService context found, skipping protected operation for ${tableName}:${entityId}`);
+        return false;
+    }
+
+    /**
+     * Remove a specific context
+     */
+    removeContext(contextKey: string): void {
+        if (this.contexts.delete(contextKey)) {
+            console.log(`🗑️ Removed operation context: ${contextKey}`);
+        }
+    }
+
+    /**
+     * Clean up expired contexts
+     */
+    private cleanupExpiredContexts(): void {
+        const now = Date.now();
+        for (const [contextKey, context] of this.contexts.entries()) {
+            if (now - context.timestamp > this.CONTEXT_EXPIRY_MS) {
+                this.contexts.delete(contextKey);
+                console.log(`🧹 Cleaned up expired context: ${contextKey}`);
+            }
+        }
+    }
+
+    /**
+     * Get all active contexts (for debugging)
+     */
+    getActiveContexts(): Map<string, OperationContext> {
+        return new Map(this.contexts);
+    }
+}
+
+export const operationContextManager = OperationContextManager.getInstance();
+
+/**
+ * Decorator to wrap database operations with context
+ */
+export function withOperationContext<T>(
+    serviceName: string, 
+    operationId: string, 
+    operation: () => Promise<T>
+): Promise<T> {
+    const contextKey = operationContextManager.createContext(serviceName, operationId);
+
+    return operation().finally(() => {
+        // Remove context after operation completes
+        setTimeout(() => {
+            operationContextManager.removeContext(contextKey);
+        }, 1000); // Small delay to ensure all database operations complete
+    });
+}
+
+/**
+ * Check if current operation should be processed by webhooks
+ */
+export function shouldProcessWebhook(entityId: string, tableName: string): boolean {
+    return operationContextManager.shouldProcessOperation(entityId, tableName);
+}

platforms/dreamsync-api/src/services/ConsentService.ts

@@ -5,6 +5,7 @@ import { Group } from "../database/entities/Group";
 import { Message } from "../database/entities/Message";
 import { UserService } from "./UserService";
 import { GroupService } from "./GroupService";
+import { withOperationContext } from "../context/OperationContext";
 
 export class ConsentService {
     private userService: UserService;
@@ -20,7 +21,11 @@ export class ConsentService {
      * Process a consent response to a match notification (by Match ID)
      */
     async processConsentResponse(messageText: string, senderId: string, groupId: string): Promise<void> {
-        try {
+        // Generate unique operation ID for this consent processing
+        const operationId = `consent-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
+
+        return withOperationContext('ConsentService', operationId, async () => {
+            try {
             // Check if the message text itself is a UUID (Match ID)
             const uuidRegex = /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i;
             const isDirectMatchId = uuidRegex.test(messageText.trim());
@@ -150,9 +155,10 @@ export class ConsentService {
                 }
             }
 
-        } catch (error) {
-            console.error("Error processing consent response:", error);
-        }
+            } catch (error) {
+                console.error("Error processing consent response:", error);
+            }
+        });
     }
 
     /**

platforms/dreamsync-api/src/web3adapter/watchers/subscriber.ts

@@ -10,6 +10,7 @@ import { Web3Adapter } from "web3-adapter";
 import path from "path";
 import dotenv from "dotenv";
 import { AppDataSource } from "../../database/data-source";
+import { shouldProcessWebhook } from "../../context/OperationContext";
 
 dotenv.config({ path: path.resolve(__dirname, "../../../../../.env") });
 export const adapter = new Web3Adapter({
@@ -253,6 +254,12 @@ export class PostgresSubscriber implements EntitySubscriberInterface {
     private async handleChange(entity: any, tableName: string): Promise<void> {
         console.log(`🔍 handleChange called for: ${tableName}, entityId: ${entity?.id}`);
 
+        // Check if this operation should be processed (only ConsentService operations for groups/messages)
+        if (!shouldProcessWebhook(entity?.id, tableName)) {
+            console.log(`⏭️ Skipping webhook for ${tableName}:${entity?.id} - not from ConsentService (protected entity)`);
+            return;
+        }
+
         // Handle junction table changes - DON'T IGNORE group_participants!
         // @ts-ignore
         const junctionInfo = JUNCTION_TABLE_MAP[tableName];
@@ -344,6 +351,12 @@ export class PostgresSubscriber implements EntitySubscriberInterface {
                 return;
             }
 
+            // Check if this junction table change should be processed (only ConsentService operations for groups)
+            if (!shouldProcessWebhook(parentId, junctionInfo.entity)) {
+                console.log(`⏭️ Skipping junction table webhook for ${junctionInfo.entity}:${parentId} - not from ConsentService (protected entity)`);
+                return;
+            }
+
             const repository = AppDataSource.getRepository(junctionInfo.entity);
             const parentEntity = await repository.findOne({
                 where: { id: parentId },
@@ -408,6 +421,12 @@ export class PostgresSubscriber implements EntitySubscriberInterface {
      */
     private async sendGroupWebhook(data: any): Promise<void> {
         try {
+            // Check if this group webhook should be processed (only ConsentService operations for groups)
+            if (!shouldProcessWebhook(data.id, "groups")) {
+                console.log(`⏭️ Skipping group webhook for ${data.id} - not from ConsentService (protected entity)`);
+                return;
+            }
+
             // Skip groups with Match ID in description (already processed)
             if (data.description && typeof data.description === 'string' && data.description.startsWith('Match ID:')) {
                 console.log(`🔍 Skipping group webhook - has Match ID in description: ${data.description}`);