Feature/user id support #6

Merged
florian merged 2 commits from feature/user-id-support into main 2026-10-30 15:39:28 +00:00
3 changed files with 157 additions and 105 deletions
Showing only changes of commit 79fc87c9c5 - Show all commits

No files matched your search

+18 -8
View File
@@ -1,15 +1,25 @@
CREATE TABLE `rss_feeds` ( CREATE TABLE `rss_feeds` (
`id` int(10) unsigned NOT NULL AUTO_INCREMENT, `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
`user_id` bigint(20) NOT NULL, `type` enum('royalroad','gitea') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
`type` enum('royalroad','gitea') COLLATE utf8mb4_unicode_ci NOT NULL, `rss_feed` varchar(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
`feed_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL, `last_entry` varchar(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL,
`rss_feed` varchar(512) COLLATE utf8mb4_unicode_ci NOT NULL,
`last_entry` varchar(512) COLLATE utf8mb4_unicode_ci DEFAULT NULL,
`active` tinyint(1) NOT NULL DEFAULT '1', `active` tinyint(1) NOT NULL DEFAULT '1',
`created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`), PRIMARY KEY (`id`),
UNIQUE KEY `uq_rss_feed` (`rss_feed`), UNIQUE KEY `uq_rss_feed` (`rss_feed`),
UNIQUE KEY `uq_user_rss_feed` (`user_id`,`rss_feed`), KEY `idx_active_type` (`active`,`type`)
KEY `idx_active_type` (`active`,`type`), )
CONSTRAINT `fk_rss_feeds_user` FOREIGN KEY (`user_id`) REFERENCES `users` (`user_id`) ON DELETE CASCADE
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
) )
+4 -2
View File
@@ -1,9 +1,11 @@
import time import time
from dataclasses import dataclass
from typing import Tuple from typing import Tuple
from pydantic import BaseModel
import feedparser import feedparser
import requests import requests
from dataclasses import dataclass from pydantic import BaseModel
from simple_logger_handler import setup_logger from simple_logger_handler import setup_logger
logger = setup_logger(__name__) logger = setup_logger(__name__)
+135 -95
View File
@@ -1,18 +1,21 @@
import asyncio import asyncio
from contextlib import asynccontextmanager
import os import os
from contextlib import asynccontextmanager
import uvicorn import uvicorn
from fastapi import Depends, FastAPI, HTTPException, Request, APIRouter 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 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 feed_handler import (
build_gitea_runner,
grab_gitea_release_information,
grab_royal_road_chapter_information,
)
from metrics_server import REQUEST_COUNTER, metrics_api from metrics_server import REQUEST_COUNTER, metrics_api
from secret_manager import cleanup_secret_files, get_secret from secret_manager import cleanup_secret_files, get_secret
from send_notifications import NotificationClient, NotificationUnavailableError, NotificationError from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError
from simple_logger_handler import LOG_LEVEL, setup_logger from simple_logger_handler import LOG_LEVEL, setup_logger
from auth_config import Caller, auth
from service_auth import UserId
logger = setup_logger(__name__) logger = setup_logger(__name__)
@@ -43,8 +46,8 @@ async def lifespan(app: FastAPI):
logger.info("[FastAPI] Shutdown complete.") logger.info("[FastAPI] Shutdown complete.")
api = FastAPI( api = FastAPI(
title="Docker Repository Query", title="Service Rss Feeds",
description="Queries Dockerhub and GHCR for new docker images", description="Queries RSS feeds for updates",
version="1.0.0", version="1.0.0",
lifespan=lifespan, lifespan=lifespan,
dependencies=[Depends(auth.authenticate)], dependencies=[Depends(auth.authenticate)],
@@ -70,7 +73,7 @@ async def prometheus_middleware(request: Request, call_next):
user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))])
scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))])
@scheduler_router.get("/gitea") @scheduler_router.post("/gitea")
def get_release_information(request: Request, db = Depends(get_db)): def get_release_information(request: Request, db = Depends(get_db)):
""" """
Checks for new Gitea releases and updates the database accordingly. Checks for new Gitea releases and updates the database accordingly.
@@ -90,48 +93,79 @@ def get_release_information(request: Request, db = Depends(get_db)):
results = {"checked": 0, "new": 0, "errors": 0} results = {"checked": 0, "new": 0, "errors": 0}
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute( 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", "FROM rss_feeds WHERE active = 1 AND type = %s",
("gitea",), ("gitea",),
) )
feeds = cursor.fetchall() feeds = cursor.fetchall()
logger.debug(f"[FastAPI] Found {len(feeds)} active Gitea feeds to check.") logger.debug(f"[FastAPI] Found {len(feeds)} active Gitea feeds to check.")
for feed_id, rss_feed, feed_name, last_entry in feeds: for feed_id, rss_feed, last_entry in feeds:
results["checked"] += 1 results["checked"] += 1
try: try:
release_title, link = grab_gitea_release_information(rss_feed) 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}") logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={link}")
if link == last_entry: if link == last_entry:
continue continue
logger.info(f"[FastAPI] Release change for ID {feed_id}: {release_title}") logger.info(f"[FastAPI] Release change for ID {feed_id}: {release_title}")
request.app.state.notifier.send(
title=feed_name, with db.cursor() as cursor:
body=release_title, cursor.execute(
link=link, "SELECT user_id, feed_name "
category="utility" "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( cursor.execute(
"UPDATE rss_feeds SET last_entry = %s WHERE id = %s", "UPDATE rss_feeds SET last_entry = %s WHERE id = %s",
(link, feed_id), (link, feed_id),
) )
db.commit() db.commit()
logger.debug(f"[FastAPI] Notification sent and database updated for Gitea feed ID {feed_id}") logger.debug(f"[FastAPI] Notification sent and database updated for Gitea feed ID {feed_id}")
results["new"] += 1 results["new"] += 1
if link.startswith("https://gitea.com/gitea/runner/releases/tag/v"): if link.startswith("https://gitea.com/gitea/runner/releases/tag/v"):
gitea_build_success = build_gitea_runner(release_title, request.app.state.gitea_token) gitea_build_success = build_gitea_runner(release_title, request.app.state.gitea_token)
logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}") logger.info(f"[FastAPI] Gitea build ran successfully:{gitea_build_success}")
except Exception as e: except Exception as e:
db.rollback() db.rollback()
results["errors"] += 1 results["errors"] += 1
logger.error( logger.error(
f"[FastAPI] Failed for feed ID {feed_id} ({rss_feed}): {e}", f"[FastAPI] Failed for feed ID {feed_id} ({rss_feed}): {e}",
exc_info=True, exc_info=True,
) )
if results["errors"]: if results["errors"]:
logger.warning(f"[FastAPI] Gitea check finished with errors: {results}") logger.warning(f"[FastAPI] Gitea check finished with errors: {results}")
@@ -140,7 +174,7 @@ def get_release_information(request: Request, db = Depends(get_db)):
logger.info(f"[FastAPI] Gitea check complete: {results}") logger.info(f"[FastAPI] Gitea check complete: {results}")
return results return results
@scheduler_router.get("/royalroad") @scheduler_router.post("/royalroad")
def get_chapters(request: Request, db = Depends(get_db)): def get_chapters(request: Request, db = Depends(get_db)):
""" """
Checks for new Royalroad chapters and updates the database accordingly. Checks for new Royalroad chapters and updates the database accordingly.
@@ -167,67 +201,66 @@ def get_chapters(request: Request, db = Depends(get_db)):
feeds = cursor.fetchall() feeds = cursor.fetchall()
logger.debug(f"[FastAPI] Found {len(feeds)} active stories to check.") 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:
for feed_id, rss_feed, last_entry in feeds: results["checked"] += 1
results["checked"] += 1 try:
try: story = grab_royal_road_chapter_information(rss_feed)
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}")
logger.debug(f"[FastAPI] ID:{feed_id},RSS Feed {rss_feed}: db={last_entry}, online={story.chapter_id}")
if story.chapter_id == last_entry: if story.chapter_id == last_entry:
continue continue
logger.info(f"[FastAPI] New chapter detected for story ID {feed_id}: {story.story_title}") logger.info(f"[FastAPI] New chapter detected for story ID {feed_id}: {story.story_title}")
with db.cursor() as cursor: with db.cursor() as cursor:
cursor.execute ( cursor.execute (
"SELECT user_id, feed_name " "SELECT user_id, feed_name "
"FROM rss_feed_subscriptions " "FROM rss_feed_subscriptions "
"WHERE rss_feed_id = %s AND active = 1", "WHERE rss_feed_id = %s AND active = 1",
(feed_id,), (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,
) )
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",
(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"]: if results["errors"]:
logger.warning(f"[FastAPI] Royalroad check finished with errors: {results}") logger.warning(f"[FastAPI] Royalroad check finished with errors: {results}")
@@ -236,9 +269,16 @@ def get_chapters(request: Request, db = Depends(get_db)):
logger.info(f"[FastAPI] Royalroad check complete: {results}") logger.info(f"[FastAPI] Royalroad check complete: {results}")
return results return results
api.include_router(user_router)
api.include_router(scheduler_router)
async def start_servers(): async def start_servers():
logger.info("[Server] Starting main API (port 5000) and metrics server (port 9000)...") 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")}") 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_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) config_metrics = uvicorn.Config(metrics_api, host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(), log_config=None)