diff --git a/platforms/dreamsync-api/src/context/OperationContext.ts b/platforms/dreamsync-api/src/context/OperationContext.ts new file mode 100644 index 000000000..b6a7d6a27 --- /dev/null +++ b/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 = 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 { + return new Map(this.contexts); + } +} + +export const operationContextManager = OperationContextManager.getInstance(); + +/** + * Decorator to wrap database operations with context + */ +export function withOperationContext( + serviceName: string, + operationId: string, + operation: () => Promise +): Promise { + 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); +} diff --git a/platforms/dreamsync-api/src/services/ConsentService.ts b/platforms/dreamsync-api/src/services/ConsentService.ts index 25ed1cfc2..00066f35a 100644 --- a/platforms/dreamsync-api/src/services/ConsentService.ts +++ b/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 { - 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); + } + }); } /** diff --git a/platforms/dreamsync-api/src/web3adapter/watchers/subscriber.ts b/platforms/dreamsync-api/src/web3adapter/watchers/subscriber.ts index 7568346a9..d8a9d5484 100644 --- a/platforms/dreamsync-api/src/web3adapter/watchers/subscriber.ts +++ b/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 { 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 { 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}`);