Merge pull request #443 from aibtcdev/duplicates

help ensure duplication doesnt happen

Author: human058382928

app/backend/supabase.py

@@ -665,12 +665,66 @@ def attempt_upload(attempt: int) -> Optional[str]:
     def create_queue_message(
         self, new_queue_message: "QueueMessageCreate"
     ) -> "QueueMessage":
-        payload = new_queue_message.model_dump(exclude_unset=True, mode="json")
-        response = self.client.table("queue").insert(payload).execute()
-        data = response.data or []
-        if not data:
-            raise ValueError("No data returned from Supabase insert for queue message.")
-        return QueueMessage(**data[0])
+        """Create a new queue message with deduplication logic to prevent 5x message multiplication."""
+
+        # Check for existing unprocessed messages with same content to prevent duplicates
+        if new_queue_message.dao_id and new_queue_message.message:
+            try:
+                # Use Supabase query to find existing unprocessed messages
+                query = (
+                    self.client.table("queue")
+                    .select("*")
+                    .eq("type", new_queue_message.type)
+                    .eq("dao_id", str(new_queue_message.dao_id))
+                    .eq("is_processed", False)
+                )
+
+                # Add wallet_id filter if present
+                if new_queue_message.wallet_id:
+                    query = query.eq("wallet_id", str(new_queue_message.wallet_id))
+
+                response = query.execute()
+                existing_data = response.data or []
+
+                # Check for duplicate content in existing messages
+                new_message_str = (
+                    str(new_queue_message.message) if new_queue_message.message else ""
+                )
+                for existing_row in existing_data:
+                    existing_message_str = str(existing_row.get("message", ""))
+                    if existing_message_str == new_message_str:
+                        existing_message = QueueMessage(**existing_row)
+                        logger.debug(
+                            f"Duplicate queue message detected for DAO {new_queue_message.dao_id}, "
+                            f"type {new_queue_message.type}, returning existing message {existing_message.id}"
+                        )
+                        return existing_message
+
+            except Exception as e:
+                # If deduplication check fails, log warning but continue with creation
+                logger.warning(
+                    f"Deduplication check failed: {str(e)}, proceeding with message creation"
+                )
+
+        # No duplicate found or deduplication skipped, create new message using Supabase
+        try:
+            payload = new_queue_message.model_dump(exclude_unset=True, mode="json")
+            response = self.client.table("queue").insert(payload).execute()
+            data = response.data or []
+            if not data:
+                raise ValueError(
+                    "No data returned from Supabase insert for queue message."
+                )
+
+            created_message = QueueMessage(**data[0])
+            logger.debug(
+                f"Created new queue message {created_message.id} for DAO {new_queue_message.dao_id}, type {new_queue_message.type}"
+            )
+            return created_message
+
+        except Exception as e:
+            logger.error(f"Failed to create queue message in Supabase: {str(e)}")
+            raise
 
     def get_queue_message(self, queue_message_id: UUID) -> Optional["QueueMessage"]:
         response = (

supabase/migrations/20250910000000_add_queue_deduplication_constraint.sql

@@ -0,0 +1,46 @@
+-- Add deduplication constraint to queue table to prevent 5x message multiplication
+-- This migration adds a unique constraint to prevent duplicate unprocessed messages
+
+-- Add a computed column for message content hash to enable efficient deduplication
+ALTER TABLE public.queue 
+ADD COLUMN IF NOT EXISTS message_hash TEXT 
+GENERATED ALWAYS AS (md5(message::text)) STORED;
+
+-- Create index for the message hash for performance
+CREATE INDEX IF NOT EXISTS idx_queue_message_hash ON public.queue(message_hash);
+
+-- Create unique constraint to prevent duplicate unprocessed messages
+-- This constraint ensures that for any combination of (type, dao_id, wallet_id, message_hash), 
+-- there can only be one unprocessed message at a time
+CREATE UNIQUE INDEX IF NOT EXISTS idx_queue_unique_unprocessed_message
+ON public.queue (type, dao_id, wallet_id, message_hash)
+WHERE is_processed = false;
+
+-- Add partial index for efficient queries on unprocessed messages
+CREATE INDEX IF NOT EXISTS idx_queue_unprocessed_by_type_dao 
+ON public.queue (type, dao_id, created_at)
+WHERE is_processed = false;
+
+-- Add comments for documentation
+COMMENT ON COLUMN public.queue.message_hash IS 'MD5 hash of the message content for deduplication';
+COMMENT ON INDEX public.idx_queue_unique_unprocessed_message IS 'Prevents duplicate unprocessed messages with same type, DAO, wallet, and content';
+
+-- Create function to handle duplicate message insertions gracefully
+CREATE OR REPLACE FUNCTION public.handle_duplicate_queue_message()
+RETURNS TRIGGER AS $$
+BEGIN
+    -- If a duplicate is detected (this would be caught by the unique constraint),
+    -- we could handle it here, but the constraint itself will prevent the insert
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Optional: Add logging for duplicate attempts (uncomment if you want to track these)
+-- CREATE OR REPLACE FUNCTION public.log_duplicate_queue_attempt()
+-- RETURNS TRIGGER AS $$
+-- BEGIN
+--     INSERT INTO public.queue_duplicate_log (attempted_type, attempted_dao_id, attempted_message_hash, attempted_at)
+--     VALUES (NEW.type, NEW.dao_id, NEW.message_hash, NOW());
+--     RETURN NULL; -- Don't actually insert the duplicate
+-- END;
+-- $$ LANGUAGE plpgsql;