Merge pull request 'Feature/user id support' (#6) from feature/user-id-support into main

Reviewed-on: #6

database.sql

@@ -1,5 +1,25 @@
-CREATE TABLE stories (id INT AUTO_INCREMENT PRIMARY KEY,
-	royalroadId INT NOT NULL,
-	lastChapter INT DEFAULT 0,
-	active BOOLEAN NOT NULL DEFAULT TRUE
- );
+CREATE TABLE `rss_feeds` (
+  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
+  `type` enum('royalroad','gitea') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
+  `rss_feed` varchar(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
+  `last_entry` varchar(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL,
+  `active` tinyint(1) NOT NULL DEFAULT '1',
+  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uq_rss_feed` (`rss_feed`),
+  KEY `idx_active_type` (`active`,`type`)
+)
+
+CREATE TABLE `rss_feed_subscriptions` (
+  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
+  `user_id` bigint(20) NOT NULL,
+  `rss_feed_id` int(10) unsigned NOT NULL,
+  `feed_name` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
+  `active` tinyint(1) NOT NULL DEFAULT '1',
+  `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uq_user_feed` (`user_id`,`rss_feed_id`),
+  KEY `idx_feed_active` (`rss_feed_id`,`active`),
+  CONSTRAINT `fk_sub_feed` FOREIGN KEY (`rss_feed_id`) REFERENCES `rss_feeds` (`id`) ON DELETE CASCADE,
+  CONSTRAINT `fk_sub_user` FOREIGN KEY (`user_id`) REFERENCES `users` (`user_id`) ON DELETE CASCADE
+)

pyproject.toml

@@ -13,7 +13,7 @@ ignore = ["E501"]
 
 [tool.ruff.lint.isort]
 known-first-party = ["db_module", "metrics_server", "rabbitmq_handler", "secret_manager",
-					"simple_logger_handler", "uvicorn_logger_config", "validator","send_notifications"]
+					"simple_logger_handler", "service_auth", "send_notifications"]
 
 [tool.ruff.format]
 indent-style = "tab"

requirements.txt

@@ -13,6 +13,7 @@ idna==3.20
 lib-db-module==0.3.0
 lib-secret-manager==0.3.1
 lib-send-notifications==0.1.0
+lib-service-auth==0.1.0
 mysql-connector-python==9.4.0
 opentelemetry-api==1.45.1
 prometheus_client==0.26.0

scripts/filter_requirements.py

@@ -12,6 +12,7 @@ SKIP = {
 	"lib-secret-manager",
 	"lib-send-notifications",
 	"simple-logger-handler",
+	"lib-service-auth"
 }
 
 NAME_RE = re.compile(r"^\s*([A-Za-z0-9][A-Za-z0-9._-]*)")

src/auth_config.py

@@ -0,0 +1,13 @@
+from enum import StrEnum
+
+from service_auth import KeySource, ServiceAuth
+
+
+class Caller(StrEnum):
+	GATEWAY = "gateway"
+	SCHEDULER = "scheduler"
+
+auth = ServiceAuth({
+	Caller.GATEWAY: KeySource("GATEWAY_KEY", "/etc/secrets/gateway_key"),
+	Caller.SCHEDULER: KeySource("SCHEDULER_KEY", "/etc/secrets/scheduler_key"),
+})

src/feed_handler.py

@@ -1,8 +1,10 @@
 import time
+from dataclasses import dataclass
 from typing import Tuple
 
 import feedparser
 import requests
+from pydantic import BaseModel
 
 from simple_logger_handler import setup_logger
 
@@ -11,6 +13,15 @@ logger = setup_logger(__name__)
 _GITEA_INSTANCE_BASE_URL="https://git.gansejunge.com"
 _GITEA_REPOSITORY_OWNER="utility"
 
+
+@dataclass(frozen=True)
+class RoyalRoadStory(BaseModel):
+	chapter_title: str
+	link: str
+	story_title: str
+	chapter_id: str
+
+
 def get_rss_feed_data(rss_feed_url: str, max_retries: int = 3) -> feedparser.FeedParserDict:
 	"""
 	Fetches information from any RSS feed, with retries on network-related errors.
@@ -70,7 +81,7 @@ def grab_gitea_release_information(rss_feed_url: str) -> Tuple[str, str]:
 	logger.debug(f"[Feed] Latest release for feed '{rss_feed_url}': {title} -> {link}")
 	return title, link
 
-def grab_royal_road_chapter_information(rss_feed_url: str) -> Tuple[str, str, str, str]:
+def grab_royal_road_chapter_information(rss_feed_url: str) -> RoyalRoadStory :
 	"""
 	Fetches the latest chapter information from a Royalroad RSS feed.
 
@@ -85,15 +96,20 @@ def grab_royal_road_chapter_information(rss_feed_url: str) -> Tuple[str, str, st
 	"""
 	feed = get_rss_feed_data(rss_feed_url)
 	latest_chapter_data = feed.entries[0]
-	chapter_link = latest_chapter_data["link"]
-	story_title = feed.feed.title
 	chapter_title_with_story_title = latest_chapter_data["title"]
-	chapter_title = chapter_title_with_story_title.removeprefix(story_title).removeprefix(" - ")
-	chapter_id = str(latest_chapter_data["id"])  # str so it compares equal to the DB value
+	chapter_title = chapter_title_with_story_title.removeprefix(feed.feed.title).removeprefix(" - ")
 
-	logger.info(f"[Feed] Latest chapter for story '{story_title}': {chapter_title}")
-	logger.debug(f"[Feed] Latest chapter for story '{story_title}' (RSS URL {rss_feed_url}): {chapter_title} (ID {chapter_id}) -> {chapter_link}")
-	return chapter_title, chapter_link, story_title, chapter_id
+	royal_road_story =  RoyalRoadStory (
+		chapter_title = chapter_title,
+		link = latest_chapter_data["link"],
+		story_title = feed.feed.title,
+		chapter_id = str(latest_chapter_data["id"])
+	)
+
+	logger.info(f"[Feed] Latest chapter for story '{royal_road_story.story_title}': {royal_road_story.chapter_title}")
+	logger.debug(f"[Feed] Latest chapter for story '{royal_road_story.story_title}' (RSS URL {rss_feed_url}):"
+				f"{royal_road_story.chapter_title} (ID {royal_road_story.chapter_id}) -> {royal_road_story.link}")
+	return royal_road_story
 
 def build_gitea_runner(version: str, gitea_token: str) -> bool:
 	url = (

src/main.py

@@ -1,15 +1,20 @@
 import asyncio
-from contextlib import asynccontextmanager
 import os
+from contextlib import asynccontextmanager
 
 import uvicorn
-from fastapi import Depends, FastAPI, HTTPException, Request
+from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
 
+from auth_config import Caller, auth
 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
+from feed_handler import (
+	build_gitea_runner,
+	grab_gitea_release_information,
+	grab_royal_road_chapter_information,
+)
 from metrics_server import REQUEST_COUNTER, metrics_api
 from secret_manager import cleanup_secret_files, get_secret
-from send_notifications import NotificationClient
+from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError
 from simple_logger_handler import LOG_LEVEL, setup_logger
 
 logger = setup_logger(__name__)
@@ -19,7 +24,7 @@ logger = setup_logger(__name__)
 async def lifespan(app: FastAPI):
 	logger.info("[FastAPI] Starting application...")
 	app.state.notifier = NotificationClient()
-	app.state.gitea_token = get_secret(os.getenv("GITEA_TOKEN",""),"/etc/secrets/gitea_token")
+	app.state.gitea_token = get_secret("GITEA_TOKEN","/etc/secrets/gitea_token")
 
 	logger.info("[FastAPI] Creating MySQL connection pool...")
 	create_connection_pool("royalRoadPool")
@@ -41,10 +46,11 @@ async def lifespan(app: FastAPI):
 	logger.info("[FastAPI] Shutdown complete.")
 
 api = FastAPI(
-		title="Docker Repository Query",
-		description="Queries Dockerhub and GHCR for new docker images",
+		title="Service Rss Feeds",
+		description="Queries RSS feeds for updates",
 		version="1.0.0",
-		lifespan=lifespan
+		lifespan=lifespan,
+		dependencies=[Depends(auth.authenticate)],
 )
 
 @api.middleware("http")
@@ -63,7 +69,11 @@ async def prometheus_middleware(request: Request, call_next):
 		logger.debug(f"[FastAPI] Counter incremented for {request.method} {request.url.path} [{status}]")
 	return response
 
-@api.get("/gitea")
+
+user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
+scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
+
+@scheduler_router.post("/gitea")
 def get_release_information(request: Request, db = Depends(get_db)):
 	"""
 	Checks for new Gitea releases and updates the database accordingly.
@@ -83,48 +93,79 @@ def get_release_information(request: Request, db = Depends(get_db)):
 	results = {"checked": 0, "new": 0, "errors": 0}
 	with db.cursor() as cursor:
 		cursor.execute(
-			"SELECT id, rss_feed, feed_name, last_entry "
+			"SELECT id, rss_feed, 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}")
+	for feed_id, rss_feed, 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
+			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"
+			logger.info(f"[FastAPI] Release change for ID {feed_id}: {release_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:
+					repo = link.split("/releases")[0].removeprefix("https://")
+					request.app.state.notifier.send(
+						title=feed_name or repo,
+						body=release_title,
+						link=link,
+						category="utility",
+						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_id={feed_id}")
+					break
+				except NotificationError as e:
+					failed += 1
+					logger.error(f"[FastAPI] Failed to send notification to user_id={user_id} for feed_id={feed_id} "
+								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 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
+			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}")
+			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,
-				)
+		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}")
@@ -133,7 +174,7 @@ def get_release_information(request: Request, db = Depends(get_db)):
 	logger.info(f"[FastAPI] Gitea check complete: {results}")
 	return results
 
-@api.get("/royalroad")
+@scheduler_router.post("/royalroad")
 def get_chapters(request: Request, db = Depends(get_db)):
 	"""
 	Checks for new Royalroad chapters and updates the database accordingly.
@@ -160,37 +201,66 @@ def get_chapters(request: Request, db = Depends(get_db)):
 		feeds = cursor.fetchall()
 		logger.debug(f"[FastAPI] Found {len(feeds)} active stories to check.")
 
-		for feed_id, rss_feed, last_entry in feeds:
-			results["checked"] += 1
-			try:
-				chapter_title, link, story_title, online_latest_entry = grab_royal_road_chapter_information(rss_feed)
-				logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={online_latest_entry}")
+	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 online_latest_entry == last_entry:
-					continue
+			if story.chapter_id == last_entry:
+				continue
 
-				logger.info(f"[FastAPI] New chapter detected for story ID {feed_id}: {story_title}")
-				request.app.state.notifier.send(
-					title=story_title,
-					body=chapter_title,
-					link=link,
-					category="royal-road"
+			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,
+					)
+					sent += 1
+				except NotificationUnavailableError:
+					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"] += 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 WHERE id = %s",
-					(online_latest_entry, feed_id),
+					(story.chapter_id, feed_id),
 				)
-				db.commit()
-				logger.debug(f"[FastAPI] Notification sent and database updated for story ID {feed_id}")
-				results["new"] += 1
+			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,
-				)
+		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}")
@@ -199,9 +269,16 @@ def get_chapters(request: Request, db = Depends(get_db)):
 	logger.info(f"[FastAPI] Royalroad check complete: {results}")
 	return results
 
+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)