import asyncio from contextlib import asynccontextmanager import os import uvicorn from fastapi import Depends, FastAPI, HTTPException, Request, APIRouter from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread from feed_handler import grab_gitea_release_information, grab_royal_road_chapter_information, build_gitea_runner, RoyalRoadStory from metrics_server import REQUEST_COUNTER, metrics_api from secret_manager import cleanup_secret_files, get_secret from send_notifications import NotificationClient, NotificationUnavailableError, NotificationError from simple_logger_handler import LOG_LEVEL, setup_logger from auth_config import Caller, auth from service_auth import UserId logger = setup_logger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("[FastAPI] Starting application...") app.state.notifier = NotificationClient() app.state.gitea_token = get_secret("GITEA_TOKEN","/etc/secrets/gitea_token") logger.info("[FastAPI] Creating MySQL connection pool...") create_connection_pool("royalRoadPool") start_healthcheck_thread() logger.info("[FastAPI] MySQL healthcheck thread started.") SECRET_PATHS = frozenset({ "/etc/secrets/gitea_token", "/etc/secrets/api_key", "/etc/secrets/db_username", "/etc/secrets/db_password" }) cleanup_secret_files(SECRET_PATHS) yield logger.info("[FastAPI] Closing MySQL connection pool...") close_connection_pool() logger.info("[FastAPI] Shutdown complete.") api = FastAPI( title="Docker Repository Query", description="Queries Dockerhub and GHCR for new docker images", version="1.0.0", lifespan=lifespan, dependencies=[Depends(auth.authenticate)], ) @api.middleware("http") async def prometheus_middleware(request: Request, call_next): logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}") status = 500 try: response = await call_next(request) status = response.status_code logger.debug(f"[FastAPI] Request processed with status {status}") except Exception as e: logger.error(f"[FastAPI] Exception occurred: {e}", exc_info=True) raise finally: REQUEST_COUNTER.labels(request.method, request.url.path, status).inc() logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]") return response user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) @scheduler_router.get("/gitea") def get_release_information(request: Request, db = Depends(get_db)): """ Checks for new Gitea releases and updates the database accordingly. Behaviour: - Fetches all active Gitea feeds and their last known release link. - Queries each feed for its latest release. - Updates `last_entry` if the latest release link differs from the stored one. - Sends a notification for each changed release. - A failure on one feed is logged and the remaining feeds are still checked. Returns: A JSON object with checked/new/error counts. If any feed failed, a 500 is returned with the same counts as the detail. """ logger.info("[FastAPI] Checking for new Gitea releases") results = {"checked": 0, "new": 0, "errors": 0} with db.cursor() as cursor: cursor.execute( "SELECT id, rss_feed, feed_name, last_entry " "FROM rss_feeds WHERE active = 1 AND type = %s", ("gitea",), ) feeds = cursor.fetchall() logger.debug(f"[FastAPI] Found {len(feeds)} active Gitea feeds to check.") for feed_id, rss_feed, feed_name, last_entry in feeds: results["checked"] += 1 try: release_title, link = grab_gitea_release_information(rss_feed) logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={link}") if link == last_entry: continue logger.info(f"[FastAPI] Release change for ID {feed_id}: {release_title}") request.app.state.notifier.send( title=feed_name, body=release_title, link=link, category="utility" ) cursor.execute( "UPDATE rss_feeds SET last_entry = %s WHERE id = %s", (link, feed_id), ) db.commit() logger.debug(f"[FastAPI] Notification sent and database updated for Gitea feed ID {feed_id}") results["new"] += 1 if link.startswith("https://gitea.com/gitea/runner/releases/tag/v"): gitea_build_success = build_gitea_runner(release_title, request.app.state.gitea_token) logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}") except Exception as e: db.rollback() results["errors"] += 1 logger.error( f"[FastAPI] Failed for feed ID {feed_id} ({rss_feed}): {e}", exc_info=True, ) if results["errors"]: logger.warning(f"[FastAPI] Gitea check finished with errors: {results}") raise HTTPException(status_code=500, detail=results) logger.info(f"[FastAPI] Gitea check complete: {results}") return results @scheduler_router.get("/royalroad") def get_chapters(request: Request, db = Depends(get_db)): """ Checks for new Royalroad chapters and updates the database accordingly. Behaviour: - Fetches all active stories and their last known chapter ID. - Queries Royalroad for each story's latest chapter. - Updates `last_entry` if a new chapter is found. - Sends a notification for any newly discovered chapters. - A failure on one story is logged and the remaining stories are still checked. Returns: A JSON object with checked/new/error counts. If any story failed, a 500 is returned with the same counts as the detail. """ logger.info("[FastAPI] Checking for new Royalroad chapters") results = {"checked": 0, "new": 0, "errors": 0} with db.cursor() as cursor: cursor.execute( "SELECT id, rss_feed, last_entry " "FROM rss_feeds WHERE active = 1 AND type = %s", ("royalroad",), ) feeds = cursor.fetchall() logger.debug(f"[FastAPI] Found {len(feeds)} active stories to check.") cache: dict[str, str | None] = {} for feed_id, rss_feed, last_entry in feeds: results["checked"] += 1 try: story = grab_royal_road_chapter_information(rss_feed) logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={story.chapter_id}") if story.chapter_id == last_entry: continue logger.info(f"[FastAPI] New chapter detected for story ID {feed_id}: {story.story_title}") with db.cursor() as cursor: cursor.execute ( "SELECT user_id, feed_name " "FROM rss_feed_subscriptions " "WHERE rss_feed_id = %s AND active = 1", (feed_id,), ) subscribers = cursor.fetchall() sent = 0 failed = 0 for user_id, feed_name in subscribers: try: request.app.state.notifier.send( title=feed_name or story.story_title, body=story.chapter_title, link=story.link, category="royal-road", user_id=user_id, ) send += 1 except NotificationUnavailableError as e: failed = len(subscribers) - sent logger.error (f"[FastAPI] Notifications backend unavailable, skipping" f"{failed} remaining user(s) for story={story.story_title}") break except NotificationError as e: failed += 1 logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for story={story.story_title}" f"status={getattr(e, 'status_code',None)}") if subscribers and sent == 0: results["errors"]+=failed logger.error(f"[FastAPI] All notifications failed for Feed ID:{feed_id}, will retry later") continue with db.cursor() as cursor: cursor.execute( "UPDATE rss_feeds SET last_entry = %s WHERE id = %s", (story.chapter_id, feed_id), ) db.commit() logger.debug(f"[FastAPI] Notification sent and database updated for story ID {feed_id}") results["new"] += 1 except Exception as e: db.rollback() results["errors"] += 1 logger.error( f"[Royalroad] Failed for feed ID {feed_id} ({rss_feed}): {e}", exc_info=True, ) if results["errors"]: logger.warning(f"[FastAPI] Royalroad check finished with errors: {results}") raise HTTPException(status_code=500, detail=results) logger.info(f"[FastAPI] Royalroad check complete: {results}") return results async def start_servers(): logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...") logger.info(f"[Server] Build commit id={os.getenv("COMMIT_ID","unknown")}") config_main = uvicorn.Config(api, host="0.0.0.0", port=5000, log_level=LOG_LEVEL.lower(), log_config=None) config_metrics = uvicorn.Config(metrics_api, host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(), log_config=None) server_main = uvicorn.Server(config_main) server_metrics = uvicorn.Server(config_metrics) await asyncio.gather(server_main.serve(), server_metrics.serve()) logger.info("[Server] Both servers have stopped.") if __name__ == "__main__": asyncio.run(start_servers())