Some more refactoring. Push_events don't work at the moment

src/alpine_bits_python/db_setup.py

@@ -7,15 +7,18 @@ before the application starts accepting requests. It includes:
 """
 
 import asyncio
+from datetime import UTC, datetime
 from typing import Any
 
-from sqlalchemy import text
+from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker
+from sqlalchemy.orm import selectinload
 
-from .const import CONF_GOOGLE_ACCOUNT, CONF_HOTEL_ID, CONF_META_ACCOUNT
+from .const import CONF_GOOGLE_ACCOUNT, CONF_HOTEL_ID, CONF_META_ACCOUNT, WebhookStatus
 from .customer_service import CustomerService
-from .db import create_database_engine
+from .db import WebhookEndpoint, WebhookRequest, create_database_engine
 from .logging_config import get_logger
+from .webhook_processor import webhook_registry
 
 _LOGGER = get_logger(__name__)
 
@@ -236,6 +239,123 @@ async def backfill_acked_requests_username(
         )
 
 
+async def reprocess_stuck_webhooks(
+    sessionmaker: async_sessionmaker,
+    config: dict[str, Any] | None = None,
+) -> None:
+    """Reprocess webhooks that were stuck in 'processing' state.
+
+    Finds webhooks with status='processing' and reprocesses them.
+    These are webhooks that were not fully processed in the previous run,
+    likely due to a crash or unexpected shutdown.
+
+    Args:
+        sessionmaker: SQLAlchemy async sessionmaker
+        config: Application configuration dictionary
+    """
+    _LOGGER.info("Checking for stuck webhooks to reprocess...")
+
+    async with sessionmaker() as session:
+        # Find all webhooks stuck in 'processing' state
+        result = await session.execute(
+            select(WebhookRequest)
+            .where(WebhookRequest.status == WebhookStatus.PROCESSING)
+            .options(
+                selectinload(WebhookRequest.webhook_endpoint).selectinload(
+                    WebhookEndpoint.hotel
+                )
+            )
+        )
+        stuck_webhooks = result.scalars().all()
+
+        if not stuck_webhooks:
+            _LOGGER.info("No stuck webhooks found")
+            return
+
+        _LOGGER.info("Found %d stuck webhooks to reprocess", len(stuck_webhooks))
+
+        reprocessed_count = 0
+        failed_count = 0
+
+        for webhook_request in stuck_webhooks:
+            webhook_id = webhook_request.id
+            webhook_endpoint = webhook_request.webhook_endpoint
+
+            if not webhook_endpoint:
+                _LOGGER.error(
+                    "Webhook request %d has no webhook_endpoint, skipping", webhook_id
+                )
+                webhook_request.status = WebhookStatus.FAILED
+                webhook_request.last_error = (
+                    "No webhook endpoint found during startup reprocessing"
+                )
+                webhook_request.processing_completed_at = datetime.now(UTC)
+                failed_count += 1
+                continue
+
+            if not webhook_request.payload_json:
+                _LOGGER.error(
+                    "Webhook request %d has no payload (purged?), marking as failed",
+                    webhook_id,
+                )
+                webhook_request.status = WebhookStatus.FAILED
+                webhook_request.last_error = (
+                    "No payload available for reprocessing (purged)"
+                )
+                webhook_request.processing_completed_at = datetime.now(UTC)
+                failed_count += 1
+                continue
+
+            try:
+                _LOGGER.info(
+                    "Reprocessing webhook %d (hotel=%s, type=%s)",
+                    webhook_id,
+                    webhook_endpoint.hotel_id,
+                    webhook_endpoint.webhook_type,
+                )
+
+                # Get processor for webhook_type
+                processor = webhook_registry.get_processor(
+                    webhook_endpoint.webhook_type
+                )
+                if not processor:
+                    raise ValueError(
+                        f"No processor for type: {webhook_endpoint.webhook_type}"
+                    )
+
+                # Reprocess webhook with simplified interface
+                await processor.process(
+                    webhook_request=webhook_request,
+                    db_session=session,
+                    config=config,
+                )
+
+                # Update status to completed
+                webhook_request.status = WebhookStatus.COMPLETED
+                webhook_request.processing_completed_at = datetime.now(UTC)
+                reprocessed_count += 1
+
+                _LOGGER.info("Successfully reprocessed webhook %d", webhook_id)
+
+            except Exception as e:
+                _LOGGER.exception("Failed to reprocess webhook %d: %s", webhook_id, e)
+                webhook_request.status = WebhookStatus.FAILED
+                webhook_request.last_error = (
+                    f"Reprocessing failed during startup: {str(e)[:1950]}"
+                )
+                webhook_request.processing_completed_at = datetime.now(UTC)
+                failed_count += 1
+
+        # Commit all changes
+        await session.commit()
+
+        _LOGGER.info(
+            "Webhook reprocessing complete: %d successful, %d failed",
+            reprocessed_count,
+            failed_count,
+        )
+
+
 async def run_startup_tasks(
     sessionmaker: async_sessionmaker,
     config: dict[str, Any] | None = None,
@@ -284,3 +404,6 @@ async def run_startup_tasks(
             "No engine provided to run_startup_tasks, "
             "skipping config-based backfill tasks"
         )
+
+    # Reprocess stuck webhooks (those stuck in 'processing' state)
+    await reprocess_stuck_webhooks(sessionmaker, config)