Merge branch 'main' into dependabot/uv/main/dev-dependencies-ba2befb238

Author: human058382928

app/api/tools/dao.py

@@ -118,23 +118,29 @@ async def _validate_airdrop_recipients(recipients: List[str]) -> Dict[str, bool]
     if not recipients:
         return {}
 
-    # Query wallets table for all recipients at once
-    wallet_filter = WalletFilterN(
-        mainnet_addresses=recipients, testnet_addresses=recipients
-    )
+    # Determine which network to check based on config
+    from app.config import config
+
+    use_mainnet = config.network.network == "mainnet"
+
+    # Query wallets table for all recipients at once, filtering by the appropriate network
+    if use_mainnet:
+        wallet_filter = WalletFilterN(mainnet_addresses=recipients)
+    else:
+        wallet_filter = WalletFilterN(testnet_addresses=recipients)
+
     wallets = backend.list_wallets_n(filters=wallet_filter)
 
-    # Create sets of valid addresses for efficient lookup
-    valid_mainnet_addresses = {w.mainnet_address for w in wallets if w.mainnet_address}
-    valid_testnet_addresses = {w.testnet_address for w in wallets if w.testnet_address}
+    # Create set of valid addresses for efficient lookup
+    if use_mainnet:
+        valid_addresses = {w.mainnet_address for w in wallets if w.mainnet_address}
+    else:
+        valid_addresses = {w.testnet_address for w in wallets if w.testnet_address}
 
     # Check each recipient
     validation_results = {}
     for recipient in recipients:
-        is_valid = (
-            recipient in valid_mainnet_addresses or recipient in valid_testnet_addresses
-        )
-        validation_results[recipient] = is_valid
+        validation_results[recipient] = recipient in valid_addresses
 
     return validation_results
 

app/backend/supabase.py

@@ -2265,12 +2265,11 @@ def get_lottery_result_by_proposal(
             self.client.table("lottery_results")
             .select("*")
             .eq("proposal_id", str(proposal_id))
-            .single()
             .execute()
         )
         if not response.data:
             return None
-        return LotteryResult(**response.data)
+        return LotteryResult(**response.data[0])
 
     def list_lottery_results(
         self, filters: Optional["LotteryResultFilter"] = None

app/lib/logger.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 from typing import Optional
@@ -12,6 +13,51 @@
 }
 
 
+class JSONFormatter(logging.Formatter):
+    """JSON formatter that outputs single-line structured logs."""
+
+    def format(self, record):
+        log_entry = {
+            "timestamp": self.formatTime(record, self.datefmt),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+        }
+
+        # Add exception info if present
+        if record.exc_info:
+            log_entry["exception"] = self.formatException(record.exc_info)
+
+        # Add extra fields if present
+        for key, value in record.__dict__.items():
+            if key not in [
+                "name",
+                "msg",
+                "args",
+                "levelname",
+                "levelno",
+                "pathname",
+                "filename",
+                "module",
+                "exc_info",
+                "exc_text",
+                "stack_info",
+                "lineno",
+                "funcName",
+                "created",
+                "msecs",
+                "relativeCreated",
+                "thread",
+                "threadName",
+                "processName",
+                "process",
+                "message",
+            ]:
+                log_entry[key] = value
+
+        return json.dumps(log_entry, ensure_ascii=False)
+
+
 def configure_logger(name: Optional[str] = None) -> logging.Logger:
     """
     Configure and return a logger instance with consistent formatting and level.
@@ -34,37 +80,37 @@ def configure_logger(name: Optional[str] = None) -> logging.Logger:
     if not logger.handlers:
         console_handler = logging.StreamHandler()
         console_handler.setLevel(log_level)
-        formatter = logging.Formatter("%(levelname)s:     %(message)s")
+        formatter = JSONFormatter()
         console_handler.setFormatter(formatter)
         logger.addHandler(console_handler)
 
     return logger
 
 
 def setup_uvicorn_logging():
-    """Configure uvicorn and other loggers to use normal formatting."""
+    """Configure uvicorn and other loggers to use JSON formatting."""
     # Disable uvicorn access logging since we handle it with middleware
     logging.getLogger("uvicorn.access").disabled = True
 
-    # Create a standard formatter
-    standard_formatter = logging.Formatter("%(levelname)s:     %(message)s")
+    # Create a JSON formatter
+    json_formatter = JSONFormatter()
 
     # Get all existing loggers and configure them
     for logger_name in ["uvicorn", "uvicorn.error", "fastapi"]:
         logger = logging.getLogger(logger_name)
         for handler in logger.handlers:
-            handler.setFormatter(standard_formatter)
+            handler.setFormatter(json_formatter)
 
     # Configure root logger
     root_logger = logging.getLogger()
     for handler in root_logger.handlers:
-        handler.setFormatter(standard_formatter)
+        handler.setFormatter(json_formatter)
 
     # Set up a hook to catch any new handlers that get added later
     original_add_handler = logging.Logger.addHandler
 
     def patched_add_handler(self, hdlr):
-        hdlr.setFormatter(standard_formatter)
+        hdlr.setFormatter(json_formatter)
         return original_add_handler(self, hdlr)
 
     logging.Logger.addHandler = patched_add_handler

app/middleware/logging.py

@@ -37,9 +37,14 @@ async def dispatch(self, request: Request, call_next):
             "process_time_ms": round(process_time * 1000, 2),
         }
 
-        # Log the request/response
+        # Log the request/response with structured data
         logger.info(
-            f"{request_info['method']} {request_info['path']} - {response_info['status_code']} - {response_info['process_time_ms']}ms"
+            f"HTTP {request_info['method']} {request_info['path']} - {response_info['status_code']}",
+            extra={
+                "request": request_info,
+                "response": response_info,
+                "event_type": "http_request",
+            },
         )
 
         return response

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
 
@@ -122,6 +123,7 @@ def _get_job_interval(self, job_type, metadata: JobMetadata) -> int:
 
     async def _execute_job_via_executor(self, job_type: str) -> None:
         """Execute a job through the enhanced executor system with proper concurrency control."""
+        logger.info(f"🚀 Scheduled execution triggered for job type: {job_type}")
         try:
             from app.backend.models import QueueMessage, QueueMessageType
 
@@ -130,13 +132,38 @@ 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"Converted job type '{job_type}' to enum: {job_type_enum}")
+
+            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(
@@ -147,21 +174,22 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
                     "triggered_at": str(datetime.now()),
                 },
                 dao_id=None,
-                tweet_id=None,
-                conversation_id=None,
+                wallet_id=None,
                 is_processed=False,
                 result=None,
                 created_at=datetime.now(),
-                updated_at=datetime.now(),
             )
 
             # Enqueue the synthetic message with the job's priority
+            logger.debug(
+                f"Enqueuing job {job_type} to executor with priority {metadata.priority}"
+            )
             job_id = await self._executor.priority_queue.enqueue(
                 synthetic_message, metadata.priority
             )
 
-            logger.debug(
-                f"Enqueued scheduled job {job_type} with ID {job_id} (priority: {metadata.priority})"
+            logger.info(
+                f"✅ Enqueued scheduled job '{job_type}' with ID {job_id} (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/infrastructure/startup_service.py

@@ -91,19 +91,34 @@ async def start_enhanced_job_system(self) -> None:
             raise RuntimeError("Job system initialization failed")
 
         # Schedule jobs with the scheduler
+        logger.info("Attempting to schedule jobs with the job manager...")
         any_jobs_scheduled = self.job_manager.schedule_jobs(self.scheduler)
         if any_jobs_scheduled:
             # Start the scheduler
+            logger.info("Starting APScheduler with scheduled jobs...")
             self.scheduler.start()
             logger.info("Job scheduler started successfully")
+
+            # Log scheduler status
+            jobs = self.scheduler.get_jobs()
+            logger.info(
+                f"APScheduler is running with {len(jobs)} active job schedules:"
+            )
+            for job in jobs:
+                logger.info(f"  - Job ID: {job.id}, next run: {job.next_run_time}")
         else:
-            logger.warning("No jobs were scheduled")
+            logger.warning("No jobs were scheduled - scheduler will not be started")
 
         # Start the job executor
+        logger.info("Starting job executor for background job processing...")
         await self.job_manager.start_executor()
-        logger.info("Enhanced job manager executor started successfully")
+        executor_stats = self.job_manager.get_executor_stats()
+        logger.info(
+            f"Enhanced job executor started with {executor_stats.get('worker_count', 0)} workers"
+        )
 
         # Start system metrics collection
+        logger.info("Starting system metrics monitoring...")
         await system_metrics.start_monitoring()
         logger.info("System metrics monitoring started")