import asyncio import os from contextlib import asynccontextmanager import uvicorn from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request from requests.exceptions import RequestException from auth_config import Caller, auth from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread from feed_errors import FeedCheckFailedError from feed_handler import ( build_gitea_runner, ensure_feed_exists, fetch_latest, ) from handlers import register_handlers from metrics_server import metrics_api from models import FeedEntry, FeedEntryType, SubscribedFeed, SubscribedFeedList from secret_manager import cleanup_secret_files, get_secret from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError from service_auth import UserId from simple_logger_handler import LOG_LEVEL, setup_logger 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="Service Rss Feeds", description="Queries RSS feeds for updates", version="1.0.0", lifespan=lifespan, dependencies=[Depends(auth.authenticate)], ) register_handlers(api) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) @scheduler_router.post("/update/{rss_feed_type}") def update_rss_feeds( request: Request, rss_feed_type: FeedEntryType, db=Depends(get_db), ): results = {"checked": 0, "new": 0, "errors": 0} with db.cursor() as cursor: cursor.execute( "SELECT id, rss_feed, last_entry " "FROM rss_feeds WHERE type = %s AND active = 1", (rss_feed_type.value,), ) rss_feeds = cursor.fetchall() logger.debug(f"[FastAPI] Found {len(rss_feeds)} active {rss_feed_type.value} RSS feeds to check.") for feed_id, rss_feed, last_entry in rss_feeds: results["checked"] += 1 try: online_information = fetch_latest(rss_feed_type, rss_feed) logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={online_information.entry_id}") if online_information.entry_id == last_entry: continue 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 online_information.feed_title, body=online_information.body_title, link=online_information.link, category=online_information.notification_type, user_id=user_id, ) sent += 1 except NotificationUnavailableError: failed = len(subscribers) - sent logger.error (f"[FastAPI] Notifications backend unavailable, skipping " f"{failed} remaining user(s) for feed={online_information.feed_title}") break except NotificationError as e: failed += 1 logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for feed={online_information.feed_title} " f"status={getattr(e, 'status_code',None)}") if subscribers and sent == 0: results["errors"] += 1 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, feed_name = %s WHERE id = %s", (online_information.entry_id, online_information.feed_title, feed_id), ) db.commit() logger.debug(f"[FastAPI] Notification sent and database updated for feed ID {feed_id}") results["new"] += 1 if rss_feed_type is FeedEntryType.GITEA and (online_information.link or "").startswith("https://gitea.com/gitea/runner/releases/tag/v"): gitea_build_success = build_gitea_runner(online_information.body_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] {rss_feed_type.value} check finished with errors: {results}") raise HTTPException(status_code=500, detail=results) logger.info(f"[FastAPI] {rss_feed_type.value} check complete: {results}") return results @user_router.post("/feed/add") def add_feed( user_id: UserId, feed_entry: FeedEntry, db = Depends(get_db) ): url = feed_entry.rss_feed_url logger.info(f"[FastAPI] Subscribe request: RSS URL={url}, user_id={user_id}") with db.cursor() as cursor: cursor.execute("SELECT id FROM rss_feeds WHERE rss_feed = %s", (url,),) known = cursor.fetchone() last_entry_id = None feed_name = None if known is None: ensure_feed_exists(url) logger.info(f"[FastApi] Adding new RSS feed, type={feed_entry.type}, rss_feed_url={url}") try: update = fetch_latest(feed_entry.type, url) last_entry_id, feed_name = update.entry_id, update.feed_title except (ValueError, RequestException) as e: logger.warning(f"[FastAPI] Could not read latest entry for {url}") raise FeedCheckFailedError(url) from e try: with db.cursor() as cursor: cursor.execute( "INSERT INTO rss_feeds (type, rss_feed, feed_name, last_entry) VALUES (%s, %s, %s, %s) " "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, feed_name = COALESCE(VALUES(feed_name), feed_name)", (feed_entry.type.value, url, feed_name, last_entry_id,), ) feed_id = cursor.lastrowid cursor.execute ( "INSERT INTO rss_feed_subscriptions (user_id, rss_feed_id, feed_name) " "VALUES (%s, %s, %s) " "ON DUPLICATE KEY UPDATE active = 1, feed_name = VALUES(feed_name)", (user_id, feed_id, feed_entry.custom_name), ) db.commit() except Exception as e: db.rollback() logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {url}") raise HTTPException(status_code=500) from e logger.info(f"[FastAPI] user_id={user_id} subscribed to feed_id={feed_id}, url={url}") return {"status":"ok","detail":"Subscribed to feed","feed_url":url} @user_router.post("/feed/delete") def delete_feed( user_id: UserId, feed_entry: FeedEntry, db = Depends(get_db), ): url = feed_entry.rss_feed_url logger.info(f"[FastAPI] Unsubscribe request: RSS URL={url}, user_id={user_id}") try: with db.cursor() as cursor: cursor.execute( "SELECT id FROM rss_feeds WHERE rss_feed = %s FOR UPDATE", (url,), ) known = cursor.fetchone() if known is None: db.rollback() return {"status":"ok","detail":"Unsubscribed to feed","feed_url":url} feed_id = known[0] cursor.execute( "UPDATE rss_feed_subscriptions SET active = 0 " "WHERE user_id = %s AND rss_feed_id = %s", (user_id, feed_id), ) cursor.execute( "SELECT 1 from rss_feed_subscriptions " "WHERE rss_feed_id = %s AND active = 1 LIMIT 1 FOR SHARE", (feed_id,), ) if cursor.fetchone() is None: cursor.execute( "UPDATE rss_feeds SET active = 0 WHERE id = %s", (feed_id,), ) db.commit() except Exception as e: db.rollback() logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {url}") raise HTTPException(status_code=500) from e return {"status":"ok","detail":"Unsubscribed to feed","feed_url":url} @user_router.get("/feed/list") def list_subscribed_feeds(user_id: UserId, db=Depends(get_db)): logger.info(f"[FastAPI] Received request to list feeds for User ID = {user_id}") try: with db.cursor() as cursor: cursor.execute( "SELECT s.feed_name as custom_name, r.feed_name as feed_name, r.type, r.rss_feed, s.active from rss_feed_subscriptions as s " "JOIN rss_feeds as r on s.rss_feed_id = r.id " "WHERE s.user_id = %s ORDER BY r.type", (user_id,), ) rows = cursor.fetchall() feed_list = SubscribedFeedList() for custom_name, feed_name, feed_type, rss_feed, is_active in rows: entry = SubscribedFeed( feed_name = custom_name or feed_name, type = feed_type, rss_feed = rss_feed, ) (feed_list.active if is_active else feed_list.inactive).append(entry) except Exception as e: logger.exception("[FastAPI] Failed to list feeds") raise HTTPException(status_code=500) from e return feed_list api.include_router(user_router) api.include_router(scheduler_router) 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")}") auth.load_keys(api, Caller.GATEWAY, Caller.SCHEDULER) cleanup_secret_files(auth.key_paths(Caller.GATEWAY, Caller.SCHEDULER)) 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())