Merge branch 'main' into check-net

Author: human058382928

app/services/infrastructure/job_management/executor.py

@@ -253,7 +253,9 @@ async def _worker(self, worker_name: str) -> None:
                 # Get next job from priority queue
                 execution = await self.priority_queue.get_next_job()
                 if not execution:
-                    await asyncio.sleep(0.1)  # Brief pause if no jobs
+                    await asyncio.sleep(
+                        0.5
+                    )  # Increased pause if no jobs to reduce CPU usage
                     continue
 
                 # Check if we can acquire a slot for this job type

app/services/infrastructure/job_management/job_manager.py

@@ -4,6 +4,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Dict, List, Optional
+from app.backend.factory import backend
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
@@ -131,12 +132,36 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
             # Convert job_type string to JobType enum
             job_type_enum = JobType.get_or_create(job_type)
 
+            logger.debug(f"🔍 Checking if {job_type} has work to do...")
+
             # Get job metadata to check if it should run
             metadata = JobRegistry.get_metadata(job_type_enum)
             if not metadata:
                 logger.error(f"No metadata found for job type: {job_type}")
                 return
 
+            # For jobs that process messages, check if there are pending messages first
+            # This prevents unnecessary job executions when there's no work
+            if job_type in ["tweet", "discord", "stx_transfer"]:
+                from app.backend.models import QueueMessageFilter
+
+                pending_messages = backend.list_queue_messages(
+                    filters=QueueMessageFilter(
+                        type=QueueMessageType.get_or_create(job_type),
+                        is_processed=False,
+                    )
+                )
+
+                if not pending_messages:
+                    logger.debug(
+                        f"⏭️ Skipping {job_type} execution - no pending messages"
+                    )
+                    return
+
+                logger.debug(
+                    f"📝 Found {len(pending_messages)} pending {job_type} messages"
+                )
+
             # Create a synthetic queue message for scheduled execution
             # This allows the job to go through the proper executor pipeline with concurrency control
             synthetic_message = QueueMessage(
@@ -160,8 +185,8 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
                 synthetic_message, metadata.priority
             )
 
-            logger.debug(
-                f"Enqueued scheduled job {job_type} with ID {job_id} (priority: {metadata.priority})"
+            logger.info(
+                f"✅ Queued {job_type} job {job_id} for execution (priority: {metadata.priority})"
             )
 
         except Exception as e:

app/services/infrastructure/job_management/tasks/tweet_task.py

@@ -43,12 +43,12 @@ class TweetProcessingResult(RunnerResult):
     job_type="tweet",
     name="Tweet Processor",
     description="Processes and sends tweets for DAOs with automatic retry and error handling",
-    interval_seconds=5,
-    priority=JobPriority.HIGH,
+    interval_seconds=30,  # Reduced frequency from 5s to 30s
+    priority=JobPriority.NORMAL,  # Changed from HIGH to NORMAL to not dominate queue
     max_retries=3,
     retry_delay_seconds=60,
     timeout_seconds=300,
-    max_concurrent=1,
+    max_concurrent=2,  # Increased from 1 to 2 to allow parallel processing
     requires_twitter=True,
     batch_size=5,
     enable_dead_letter_queue=True,
@@ -200,7 +200,7 @@ async def _validate_prerequisites(self, context: JobContext) -> bool:
     async def _validate_task_specific(self, context: JobContext) -> bool:
         """Validate task-specific conditions."""
         if not self._pending_messages:
-            logger.debug("No pending tweet messages found")
+            logger.debug("No pending tweet messages found - skipping execution")
             return False
 
         # Validate each message before processing

app/services/integrations/webhooks/chainhook/handlers/airdrop_stx_handler.py

@@ -53,52 +53,73 @@ def can_handle_transaction(self, transaction: TransactionWithReceipt) -> bool:
         tx_data_content = tx_data["tx_data"]
         tx_metadata = tx_data["tx_metadata"]
 
+        # Add debug logging
+        tx_hash = transaction.transaction_identifier.hash
+        self.logger.debug(f"AirdropSTXHandler checking transaction: {tx_hash}")
+
         # Only handle ContractCall type transactions
         if not isinstance(tx_kind, dict):
-            self.logger.debug(f"Skipping: tx_kind is not a dict: {type(tx_kind)}")
+            self.logger.debug(
+                f"[{tx_hash}] Rejecting: tx_kind is not a dict: {type(tx_kind)}"
+            )
             return False
 
         tx_kind_type = tx_kind.get("type")
+        self.logger.debug(f"[{tx_hash}] tx_kind_type: {tx_kind_type}")
 
         if not isinstance(tx_data_content, dict):
             self.logger.debug(
-                f"Skipping: tx_data_content is not a dict: {type(tx_data_content)}"
+                f"[{tx_hash}] Rejecting: tx_data_content is not a dict: {type(tx_data_content)}"
             )
             return False
 
         # Check if the method name is exactly "send-many"
         tx_method = tx_data_content.get("method", "")
         is_send_many = tx_method == "send-many"
+        self.logger.debug(
+            f"[{tx_hash}] method: '{tx_method}', is_send_many: {is_send_many}"
+        )
 
         # Check if this is the specific send-many contract
         contract_identifier = tx_data_content.get("contract_identifier", "")
         is_target_contract = contract_identifier == self.TARGET_CONTRACT
+        self.logger.debug(
+            f"[{tx_hash}] contract: '{contract_identifier}', is_target_contract: {is_target_contract}"
+        )
+        self.logger.debug(f"[{tx_hash}] TARGET_CONTRACT: '{self.TARGET_CONTRACT}'")
 
         # Access success from TransactionMetadata
         tx_success = tx_metadata.success
+        self.logger.debug(
+            f"[{tx_hash}] tx_success: {tx_success} (type: {type(tx_success)})"
+        )
 
         # Check if there are any STX transfers to agent accounts in the events
         events = tx_metadata.receipt.events if hasattr(tx_metadata, "receipt") else []
-        has_agent_account_transfers = self._has_agent_account_stx_transfers(events)
+        self.logger.debug(f"[{tx_hash}] events count: {len(events)}")
 
-        if (
-            is_send_many
-            and is_target_contract
-            and tx_success
-            and has_agent_account_transfers
-        ):
-            self.logger.debug(
-                f"Found send-many STX funding to agent accounts: {tx_method} on contract: {contract_identifier}"
-            )
+        has_agent_account_transfers = self._has_agent_account_stx_transfers(events)
+        self.logger.debug(
+            f"[{tx_hash}] has_agent_account_transfers: {has_agent_account_transfers}"
+        )
 
-        return (
+        final_result = (
             tx_kind_type == "ContractCall"
             and is_send_many
             and is_target_contract
             and tx_success is True
             and has_agent_account_transfers
         )
 
+        self.logger.debug(f"[{tx_hash}] Final result: {final_result}")
+
+        if final_result:
+            self.logger.info(f"AirdropSTXHandler CLAIMING transaction: {tx_hash}")
+        else:
+            self.logger.info(f"AirdropSTXHandler REJECTING transaction: {tx_hash}")
+
+        return final_result
+
     def _has_agent_account_stx_transfers(self, events: List[Event]) -> bool:
         """Check if there are any STX transfers.