refactor: remove event bus component and related files

castle-api/src/castle_api/bus.py

@@ -1,121 +0,0 @@
-"""Event bus core — in-memory subscription table and HTTP fan-out delivery."""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-
-import httpx
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class Subscription:
-    """A subscription to a topic."""
-
-    topic: str
-    callback_url: str
-    subscriber: str = ""  # optional label for debugging
-
-
-@dataclass
-class EventBus:
-    """In-memory event bus with HTTP fan-out delivery."""
-
-    subscriptions: dict[str, list[Subscription]] = field(default_factory=dict)
-    _client: httpx.AsyncClient | None = field(default=None, repr=False)
-
-    async def start(self) -> None:
-        """Initialize the HTTP client."""
-        self._client = httpx.AsyncClient(timeout=10.0)
-
-    async def stop(self) -> None:
-        """Close the HTTP client."""
-        if self._client:
-            await self._client.aclose()
-            self._client = None
-
-    def subscribe(self, topic: str, callback_url: str, subscriber: str = "") -> None:
-        """Register a subscription to a topic."""
-        if topic not in self.subscriptions:
-            self.subscriptions[topic] = []
-
-        # Don't duplicate
-        for sub in self.subscriptions[topic]:
-            if sub.callback_url == callback_url:
-                return
-
-        self.subscriptions[topic].append(
-            Subscription(topic=topic, callback_url=callback_url, subscriber=subscriber)
-        )
-        logger.info("Subscribed %s to topic '%s' -> %s", subscriber, topic, callback_url)
-
-    def unsubscribe(self, topic: str, callback_url: str) -> bool:
-        """Remove a subscription. Returns True if found and removed."""
-        if topic not in self.subscriptions:
-            return False
-
-        before = len(self.subscriptions[topic])
-        self.subscriptions[topic] = [
-            s for s in self.subscriptions[topic] if s.callback_url != callback_url
-        ]
-        removed = len(self.subscriptions[topic]) < before
-
-        if not self.subscriptions[topic]:
-            del self.subscriptions[topic]
-
-        return removed
-
-    async def publish(self, topic: str, payload: dict) -> int:
-        """Publish an event to all subscribers. Returns number of subscribers notified."""
-        subscribers = self.subscriptions.get(topic, [])
-        if not subscribers:
-            return 0
-
-        event = {
-            "topic": topic,
-            "payload": payload,
-            "published_at": datetime.now(timezone.utc).isoformat(),
-        }
-
-        # Fan out to all subscribers concurrently, fire-and-forget style
-        tasks = [self._deliver(sub, event) for sub in subscribers]
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-
-        delivered = sum(1 for r in results if r is True)
-        return delivered
-
-    async def _deliver(self, sub: Subscription, event: dict) -> bool:
-        """Deliver an event to a single subscriber."""
-        if not self._client:
-            logger.error("HTTP client not initialized")
-            return False
-
-        try:
-            response = await self._client.post(sub.callback_url, json=event)
-            if response.status_code < 300:
-                return True
-            logger.warning(
-                "Delivery to %s returned %d", sub.callback_url, response.status_code
-            )
-            return False
-        except Exception:
-            logger.warning("Delivery to %s failed", sub.callback_url, exc_info=True)
-            return False
-
-    def list_topics(self) -> dict[str, list[dict]]:
-        """List all topics and their subscribers."""
-        return {
-            topic: [
-                {"callback_url": s.callback_url, "subscriber": s.subscriber}
-                for s in subs
-            ]
-            for topic, subs in self.subscriptions.items()
-        }
-
-
-# Singleton instance
-bus = EventBus()

castle-api/src/castle_api/events.py

@@ -1,61 +0,0 @@
-"""Event bus API routes — subscribe, publish, unsubscribe, list topics."""
-
-from __future__ import annotations
-
-from fastapi import APIRouter
-from pydantic import BaseModel
-
-from castle_api.bus import bus
-
-router = APIRouter(prefix="/events", tags=["events"])
-
-
-class PublishRequest(BaseModel):
-    topic: str
-    payload: dict
-
-
-class SubscribeRequest(BaseModel):
-    topic: str
-    callback_url: str
-    subscriber: str = ""
-
-
-class UnsubscribeRequest(BaseModel):
-    topic: str
-    callback_url: str
-
-
-@router.post("/publish")
-async def publish(request: PublishRequest) -> dict:
-    """Publish an event to a topic."""
-    delivered = await bus.publish(request.topic, request.payload)
-    return {"topic": request.topic, "subscribers_notified": delivered}
-
-
-@router.post("/subscribe")
-async def subscribe(request: SubscribeRequest) -> dict:
-    """Subscribe to a topic with a webhook callback URL."""
-    bus.subscribe(request.topic, request.callback_url, request.subscriber)
-    return {
-        "topic": request.topic,
-        "callback_url": request.callback_url,
-        "status": "subscribed",
-    }
-
-
-@router.post("/unsubscribe")
-async def unsubscribe(request: UnsubscribeRequest) -> dict:
-    """Unsubscribe from a topic."""
-    removed = bus.unsubscribe(request.topic, request.callback_url)
-    return {
-        "topic": request.topic,
-        "callback_url": request.callback_url,
-        "status": "unsubscribed" if removed else "not_found",
-    }
-
-
-@router.get("/topics")
-async def list_topics() -> dict:
-    """List all topics and their subscribers."""
-    return {"topics": bus.list_topics()}

castle-api/src/castle_api/main.py

@@ -1,4 +1,4 @@
-"""Castle API — dashboard data, health aggregation, event bus, service management."""
+"""Castle API — dashboard data, health aggregation, and service management."""
 
 from __future__ import annotations
 
@@ -11,10 +11,8 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from starlette.responses import StreamingResponse
 
-from castle_api.bus import bus
 from castle_api.config import settings
 from castle_api.config_editor import router as config_router
-from castle_api.events import router as events_router
 from castle_api.logs import router as logs_router
 from castle_api.routes import router as dashboard_router
 from castle_api.secrets import router as secrets_router
@@ -41,7 +39,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     global _shutting_down
     _shutting_down = False
 
-    await bus.start()
     poll_task = asyncio.create_task(health_poll_loop())
 
     yield
@@ -49,12 +46,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     _shutting_down = True
     poll_task.cancel()
     close_all_subscribers()
-    await bus.stop()
 
 
 app = FastAPI(
     title="castle-api",
-    description="Castle API, event bus, and service management",
+    description="Castle API and service management",
     version="0.1.0",
     lifespan=lifespan,
 )
@@ -68,7 +64,6 @@ app.add_middleware(
 
 app.include_router(config_router)
 app.include_router(dashboard_router)
-app.include_router(events_router)
 app.include_router(logs_router)
 app.include_router(secrets_router)
 app.include_router(services_router)