""" Webhook Processor - Handles queued webhook events asynchronously. Fetches order details from ML API and updates the database. """ import logging import threading import time from queue import Empty from datetime import datetime logger = logging.getLogger(__name__) class WebhookProcessor: """ Background worker that processes webhook events from the queue. """ def __init__(self, webhook_queue, db_session_factory, meli_service_factory): """ Args: webhook_queue: Queue instance with webhook events db_session_factory: Callable that returns a new DB session meli_service_factory: Callable that returns a MeliApiService instance """ self.queue = webhook_queue self.db_session_factory = db_session_factory self.meli_service_factory = meli_service_factory self.running = False self.thread = None self.processed_count = 0 self.error_count = 0 self.last_processed_at = None # SSE subscribers for real-time updates self.subscribers = [] def start(self): """Start the background processor thread.""" if self.running: logger.warning("[WEBHOOK_PROCESSOR] Already running") return self.running = True self.thread = threading.Thread(target=self._process_loop, daemon=True) self.thread.start() logger.info("[WEBHOOK_PROCESSOR] Started") def stop(self): """Stop the background processor.""" self.running = False if self.thread: self.thread.join(timeout=5) logger.info("[WEBHOOK_PROCESSOR] Stopped") def _process_loop(self): """Main processing loop.""" while self.running: try: # Wait for an event with timeout event = self.queue.get(timeout=1) self._process_event(event) except Empty: # No events, continue waiting continue except Exception as e: logger.error(f"[WEBHOOK_PROCESSOR] Error in loop: {e}") self.error_count += 1 time.sleep(1) # Backoff on error def _process_event(self, event): """Process a single webhook event.""" topic = event.get('topic') resource = event.get('resource', '') logger.info(f"[WEBHOOK_PROCESSOR] Processing: {topic} -> {resource}") try: if topic == 'orders_v2': self._handle_order_event(resource) elif topic == 'items': self._handle_item_event(resource) elif topic == 'payments': self._handle_payment_event(resource) else: logger.info(f"[WEBHOOK_PROCESSOR] Ignored topic: {topic}") self.processed_count += 1 self.last_processed_at = datetime.utcnow() # Broadcast to SSE clients try: from app.api.endpoints.sse import broadcast_event broadcast_event('webhook_processed', { 'topic': topic, 'resource': resource, 'timestamp': self.last_processed_at.isoformat() }) except Exception as sse_err: logger.warning(f"[WEBHOOK_PROCESSOR] SSE broadcast failed: {sse_err}") # Sync visits to update real-time metrics (visits, conversion) try: self._sync_visits_for_update() except Exception as visits_err: logger.warning(f"[WEBHOOK_PROCESSOR] Visits sync failed: {visits_err}") # Notify internal subscribers (legacy) self._notify_subscribers({ 'type': 'webhook_processed', 'topic': topic, 'resource': resource, 'timestamp': self.last_processed_at.isoformat() }) except Exception as e: logger.error(f"[WEBHOOK_PROCESSOR] Error processing {topic}: {e}") self.error_count += 1 def _handle_order_event(self, resource): """ Handle orders_v2 topic. Resource format: /orders/{order_id} """ # Extract order ID from resource order_id = resource.replace('/orders/', '').strip() if not order_id: logger.warning("[WEBHOOK_PROCESSOR] Invalid order resource") return logger.info(f"[WEBHOOK_PROCESSOR] Fetching order: {order_id}") # Get DB session and ML service db = self.db_session_factory() meli = self.meli_service_factory(db) try: # Fetch order from ML API using the correct method order_data = meli.get_order(order_id) if order_data: # Use V2 InitialLoadService for consistent upsert logic (matches Manual Sync) from app.services.sync_v2.initial_load import InitialLoadService loader = InitialLoadService(db, meli_client=meli) # _upsert_order parses and updates MlOrder and Items status = loader._upsert_order(order_data) db.commit() logger.info(f"[WEBHOOK_PROCESSOR] Order {order_id} synced successfully ({status})") # ONLY broadcast NOVA VENDA celebration for NEW orders (created), NOT updates # status from _upsert_order: 'created' = new, 'updated' = existing order changed if status == 'created': try: from app.api.endpoints.sse import broadcast_event order_items = order_data.get('order_items', []) first_item = order_items[0] if order_items else {} product_title = first_item.get('item', {}).get('title', 'Novo Pedido') broadcast_event('order_update', { 'order_id': order_id, 'title': product_title, 'status': status, 'timestamp': datetime.utcnow().isoformat() }) logger.info(f"[WEBHOOK_PROCESSOR] 🎉 SSE NOVA VENDA sent: {product_title}") except Exception as sse_err: logger.warning(f"[WEBHOOK_PROCESSOR] SSE order_update failed: {sse_err}") else: logger.info(f"[WEBHOOK_PROCESSOR] Order {order_id} was UPDATE, no celebration") # Trigger specific Visits Sync for items in this order try: order_items = order_data.get('order_items', []) item_ids = [item.get('item', {}).get('id') for item in order_items if item.get('item', {}).get('id')] if item_ids: logger.info(f"[WEBHOOK_PROCESSOR] Triggering visits sync for sold items: {item_ids}") # Note: We must call self._sync_visits_for_update, but it creates own DB session. # This matches pattern. self._sync_visits_for_update(item_ids=item_ids) except Exception as v_err: logger.warning(f"[WEBHOOK_PROCESSOR] Failed to trigger visits update for order {order_id}: {v_err}") else: logger.warning(f"[WEBHOOK_PROCESSOR] Order {order_id} not found in API") except Exception as e: logger.error(f"[WEBHOOK_PROCESSOR] Error syncing order {order_id}: {e}") raise finally: db.close() def _handle_item_event(self, resource): """ Handle items topic. Resource format: /items/{item_id} """ item_id = resource.replace('/items/', '').strip() if not item_id: return logger.info(f"[WEBHOOK_PROCESSOR] Item update: {item_id}") # For items, we can trigger a targeted ad sync # This is lower priority than orders db = self.db_session_factory() meli = self.meli_service_factory(db) try: # Use the request method with GET item_data = meli.request('GET', f'/items/{item_id}') if item_data: from app.services.sync_engine import SyncEngine engine = SyncEngine() # Creates its own DB session engine._upsert_ad(item_data) engine.db.commit() engine.db.close() logger.info(f"[WEBHOOK_PROCESSOR] Item {item_id} synced") # Trigger Visits Sync for this item self._sync_visits_for_update(item_ids=[item_id]) except Exception as e: logger.error(f"[WEBHOOK_PROCESSOR] Error syncing item {item_id}: {e}") finally: db.close() def _handle_payment_event(self, resource): """ Handle payments topic. Resource format: /collections/{payment_id} or /payments/{payment_id} """ # Payments can trigger order status updates # We extract the payment ID and find the associated order logger.info(f"[WEBHOOK_PROCESSOR] Payment update: {resource}") # For now, log only - full implementation would fetch payment and update order def _notify_subscribers(self, data): """Notify all SSE subscribers of an update.""" for subscriber in self.subscribers[:]: try: subscriber(data) except Exception: self.subscribers.remove(subscriber) def add_subscriber(self, callback): """Add an SSE subscriber.""" self.subscribers.append(callback) def remove_subscriber(self, callback): """Remove an SSE subscriber.""" if callback in self.subscribers: self.subscribers.remove(callback) def _sync_visits_for_update(self, item_ids: list = None): """ Sync visits data from ML API to update real-time metrics. If item_ids provided, syncs only those items. Otherwise, syncs a batch of active items. """ from datetime import date, timedelta from app.models.ml_metrics_daily import MlMetricsDaily from app.models.ad import Ad import os logger.info(f"[WEBHOOK_PROCESSOR] Syncing visits for real-time update (Specific Items: {item_ids})...") db = self.db_session_factory() meli = self.meli_service_factory(db) try: user_id = os.getenv('MELI_USER_ID') # If item_ids passed, we might not need user_id depending on MeliApiService method, # but usually visits endpoint needs user? No, items/{id}/visits/time_window works with item ID. items = [] # 1. Priority: Specific Items (e.g. from Order) if item_ids: items = db.query(Ad).filter(Ad.id.in_(item_ids)).all() # 2. Background: Sync a batch of active ads to keep "Total Visits" fresh # We want to rotate these, maybe by 'visits_last_updated' asc? # For now, let's just grab 50 active ads to ensure the global counter moves. # Using random or just standard query limit? standard is fine if we assume rotation happens elsewhere or just brute force top active. # Ideally: Order by visits_last_updated nulls first, then oldest. background_limit = 50 background_items = db.query(Ad).filter( Ad.status == 'active', Ad.id.notin_([i.id for i in items]) ).order_by(Ad.visits_last_updated.asc()).limit(background_limit).all() items.extend(background_items) logger.info(f"[WEBHOOK_PROCESSOR] Syncing visits for {len(items)} items (Specific: {len(item_ids) if item_ids else 0}, Background: {len(background_items)})") today = date.today() yesterday = today - timedelta(days=1) for item in items: try: # Fetch visits from ML API visits_data = meli.request( 'GET', f'/items/{item.id}/visits/time_window', params={'last': 7, 'unit': 'day'} ) if visits_data and isinstance(visits_data, list): total_visits_sum = 0 for visit_entry in visits_data: visit_date = visit_entry.get('date') if visit_date: visit_date_obj = date.fromisoformat(visit_date[:10]) total_visits = visit_entry.get('total', 0) total_visits_sum += total_visits # Upsert to ml_metrics_daily existing = db.query(MlMetricsDaily).filter( MlMetricsDaily.item_id == item.id, MlMetricsDaily.date == visit_date_obj ).first() if existing: existing.visits = total_visits else: new_metric = MlMetricsDaily( item_id=item.id, date=visit_date_obj, visits=total_visits, sales_qty=0 ) db.add(new_metric) # Update Ad.total_visits with the sum from the time window # This keeps the total fresh, matching how sales are updated item.visits_30d = total_visits_sum item.total_visits = total_visits_sum # Keep total_visits fresh for UI item.visits_last_updated = datetime.now() except Exception as item_err: logger.debug(f"[WEBHOOK_PROCESSOR] Visits sync error for {item.id}: {item_err}") continue db.commit() logger.info("[WEBHOOK_PROCESSOR] Visits sync completed") except Exception as e: logger.error(f"[WEBHOOK_PROCESSOR] Visits sync error: {e}") db.rollback() finally: db.close() def get_status(self): """Get processor status.""" return { 'running': self.running, 'queue_size': self.queue.qsize(), 'processed_count': self.processed_count, 'error_count': self.error_count, 'last_processed_at': self.last_processed_at.isoformat() if self.last_processed_at else None } # Global processor instance (initialized in run_web.py) _processor = None def get_processor(): """Get the global webhook processor instance.""" return _processor def init_processor(webhook_queue, db_session_factory, meli_service_factory): """Initialize the global webhook processor.""" global _processor _processor = WebhookProcessor(webhook_queue, db_session_factory, meli_service_factory) return _processor