diff --git a/src/main.py b/src/main.py index 4c71a9d..9cc76f4 100644 --- a/src/main.py +++ b/src/main.py @@ -3,29 +3,16 @@ import os from contextlib import asynccontextmanager import uvicorn -from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request -from service_auth import UserId +from fastapi import Depends, FastAPI 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, start_healthcheck_thread from handlers import register_handlers from metrics_server import metrics_api -from models import RepositoryType, RepositoryUpdate -from repositories import ( - fetch_latest, - reset_missing_checks, - unsubscribe_every_subscriber_from_repository, - validate_repository_combination_and_return_latest_tag, -) -from repository_errors import ( - ApiAuthenticationFailure, - RateLimitedError, - RepositoryApiError, - RepositoryNotFoundError, - RepositoryTagNotFoundError, -) +from routers.scheduler import scheduler_router +from routers.user import user_router from secret_manager import cleanup_secret_files -from send_notifications import NotificationClient, NotificationError, NotificationUnavailableError +from send_notifications import NotificationClient from simple_logger_handler import LOG_LEVEL, setup_logger logger = setup_logger(__name__) @@ -65,218 +52,6 @@ api = FastAPI( ) 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/{repository_type}") -def update_watched_repositories( - request: Request, - repository_type: RepositoryType, - db=Depends(get_db), - ): - results = {"checked": 0, "new": 0, "errors": 0} - with db.cursor() as cursor: - cursor.execute( - "SELECT id, org, repo, tag, latest_entry, missing_checks " - "FROM watched_repositories WHERE type = %s AND active = 1", - (repository_type.value,), - ) - repositories = cursor.fetchall() - logger.debug(f"[FastAPI] Found {len(repositories)} active {repository_type.value} repositories to check.") - - for repository_id, org, repo, tag, latest_entry, missing_checks in repositories: - results["checked"] += 1 - try: - online_version = fetch_latest(repository_type, org, repo, tag) - except RateLimitedError as e: - logger.warning(f"[FastAPI] Rate limited, stopping the {repository_type.value} check: {e}") - results["errors"] += 1 - break - except ApiAuthenticationFailure as e: - logger.error(f"[FastAPI] {e}") - results["errors"] += 1 - break - except (RepositoryNotFoundError, RepositoryTagNotFoundError) as e: - logger.warning(f"[FastAPI] {e}, deactivating repository ID {repository_id}") - unsubscribe_every_subscriber_from_repository(db, request.app.state.notifier, repository_id, org, repo, tag) - continue - except RepositoryApiError as e: - logger.warning(f"[FastAPI] Lookup failed for ID {repository_id}: {e}") - results["errors"] += 1 - continue - except Exception: - logger.exception(f"[FastAPI] Unexpected error for ID {repository_id} ({org}/{repo}:{tag})") - results["errors"] += 1 - continue - - if missing_checks: - reset_missing_checks(db, repository_id) - if latest_entry == online_version: - continue - - try: - - logger.info(f"[FastAPI] New version detected for org={org}, repo={repo}, tag={tag}") - - with db.cursor() as cursor: - cursor.execute( - "SELECT user_id, custom_name " - "FROM watched_repositories_subscriptions " - "WHERE repository_id = %s AND active = 1", - (repository_id,), - ) - subscribers = cursor.fetchall() - - sent = 0 - failed = 0 - for user_id, custom_name in subscribers: - try: - request.app.state.notifier.send( - title=custom_name or f"{org}/{repo}", - body=f"New version {online_version} for tag {tag}", - 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 repo={org}/{repo}" - ) - break - except NotificationError as e: - failed += 1 - logger.error( - f"[FastAPI] Failed to send notification to user_id={user_id} for repo={org}/{repo} " - f"status={getattr(e, 'status_code', None)}" - ) - - if subscribers and sent == 0: - results["errors"] += 1 - logger.error(f"[FastAPI] All notifications failed for ID:{repository_id}, will retry later") - continue - - with db.cursor() as cursor: - cursor.execute( - "UPDATE watched_repositories SET latest_entry = %s WHERE id = %s", - (online_version, repository_id), - ) - db.commit() - logger.debug(f"[FastAPI] Notification sent and database updated for repository ID {repository_id}") - results["new"] += 1 - - except Exception as e: - db.rollback() - results["errors"] += 1 - logger.error( - f"[FastAPI] Failed for ID {repository_id} (org={org}, repo={repo}, tag={tag}): {e}", - exc_info=True, - ) - - if results["errors"]: - logger.warning(f"[FastAPI] {repository_type.value} check finished with errors: {results}") - raise HTTPException(status_code=500, detail=results) - logger.info(f"[FastAPI] {repository_type.value} check complete: {results}") - return results - -@user_router.post("/add/repository") -def add_repository_subscription( - user_id: UserId, - repository: RepositoryUpdate, - db=Depends(get_db), -): - name = f"{repository.org}/{repository.repo}:{repository.tag}" - logger.info(f"[FastAPI] Subscribe request: {repository.type.value} {name}, user_id={user_id}") - - with db.cursor() as cursor: - cursor.execute( - "SELECT id, active FROM watched_repositories " - "WHERE type = %s AND org = %s AND repo = %s AND tag = %s", - (repository.type.value, repository.org, repository.repo, repository.tag), - ) - known = cursor.fetchone() - - latest_entry = None - if known is None or not known[1]: - latest_entry = validate_repository_combination_and_return_latest_tag(repository) - - try: - with db.cursor() as cursor: - cursor.execute( - "INSERT INTO watched_repositories (type, org, repo, tag, latest_entry) " - "VALUES (%s, %s, %s, %s, %s) " - "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, missing_checks = 0, " - "latest_entry = COALESCE(VALUES(latest_entry), latest_entry)", - (repository.type.value, repository.org, repository.repo, repository.tag, latest_entry), - ) - repository_id = cursor.lastrowid - - cursor.execute( - "INSERT INTO watched_repositories_subscriptions (user_id, repository_id, custom_name) " - "VALUES (%s, %s, %s) " - "ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)", - (user_id, repository_id, repository.custom_name), - ) - db.commit() - except Exception as e: - db.rollback() - logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {name}") - raise HTTPException(status_code=500) from e - - logger.info(f"[FastAPI] user_id={user_id} subscribed to repository ID {repository_id} ({name})") - return {"status": "ok", "detail": "Subscribed to repository", "name": name} - -@user_router.post("/delete/repository") -def delete_repository_subscription( - user_id: UserId, - repository: RepositoryUpdate, - db=Depends(get_db), -): - name = f"{repository.org}/{repository.repo}:{repository.tag}" - logger.info(f"[FastAPI] Unsubscribe request: {repository.type.value} {name}, user_id={user_id}") - - try: - with db.cursor() as cursor: - cursor.execute( - "SELECT id FROM watched_repositories " - "WHERE type = %s AND org = %s AND repo = %s AND tag = %s FOR UPDATE", - (repository.type.value, repository.org, repository.repo, repository.tag), - ) - known = cursor.fetchone() - - if known is None: - db.rollback() - logger.info(f"[FastAPI] Repository {repository.type.value} {name} is not tracked, nothing to do") - return {"status":"ok","detail":"Unsubscribed from repository","name":name} - - repository_id = known[0] - cursor.execute( - "UPDATE watched_repositories_subscriptions SET active = 0 " - "WHERE user_id = %s AND repository_id = %s", - (user_id, repository_id), - ) - - cursor.execute( - "SELECT 1 from watched_repositories_subscriptions " - "WHERE repository_id = %s AND active = 1 LIMIT 1 FOR SHARE", - (repository_id,), - ) - if cursor.fetchone() is None: - cursor.execute( - "UPDATE watched_repositories SET active = 0 WHERE id = %s", - (repository_id,), - ) - db.commit() - except Exception as e: - db.rollback() - logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {name}") - raise HTTPException(status_code=500) from e - - logger.info(f"[FastAPI] Unsubscribe request successful: {repository.type.value} {name}, user_id={user_id}") - return {"status":"ok","detail":"Unsubscribed from repository","name":name} - api.include_router(user_router) api.include_router(scheduler_router) diff --git a/src/models.py b/src/models.py index a27002b..248c898 100644 --- a/src/models.py +++ b/src/models.py @@ -1,7 +1,7 @@ import string from enum import StrEnum -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, Field, computed_field, field_validator from repository_errors import InvalidOrganisationError, InvalidRepositoryError, InvalidTagError @@ -10,6 +10,25 @@ _MAX_ORG = 39 _MAX_REPO = 255 _MAX_TAG = 128 +class SubscribedRepository(BaseModel): + type: str + repository_name: str + tag: str + latest_entry: str + +class Counts(BaseModel): + active: int + inactive: int + +class SubscribedRepositoryList(BaseModel): + active: list[SubscribedRepository] = Field(default_factory=list) + inactive: list[SubscribedRepository] = Field(default_factory=list) + + @computed_field + @property + def counts(self) -> Counts: + return Counts(active=len(self.active),inactive=len(self.inactive)) + class RepositoryType(StrEnum): GITHUB = "github" DOCKERHUB = "dockerhub" diff --git a/src/routers/__init__.py b/src/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/routers/scheduler.py b/src/routers/scheduler.py new file mode 100644 index 0000000..5754aff --- /dev/null +++ b/src/routers/scheduler.py @@ -0,0 +1,137 @@ + +from fastapi import APIRouter, Depends, HTTPException, Request + +from auth_config import Caller, auth +from db_module import get_db +from models import RepositoryType +from repositories import ( + fetch_latest, + reset_missing_checks, + unsubscribe_every_subscriber_from_repository, +) +from repository_errors import ( + ApiAuthenticationFailure, + RateLimitedError, + RepositoryApiError, + RepositoryNotFoundError, + RepositoryTagNotFoundError, +) +from send_notifications import NotificationError, NotificationUnavailableError +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + +scheduler_router = APIRouter(dependencies=[Depends(auth.require(Caller.SCHEDULER))]) + + +@scheduler_router.post("/update/{repository_type}") +def update_watched_repositories( + request: Request, + repository_type: RepositoryType, + db=Depends(get_db), + ): + results = {"checked": 0, "new": 0, "errors": 0} + with db.cursor() as cursor: + cursor.execute( + "SELECT id, org, repo, tag, latest_entry, missing_checks " + "FROM watched_repositories WHERE type = %s AND active = 1", + (repository_type.value,), + ) + repositories = cursor.fetchall() + logger.debug(f"[FastAPI] Found {len(repositories)} active {repository_type.value} repositories to check.") + + for repository_id, org, repo, tag, latest_entry, missing_checks in repositories: + results["checked"] += 1 + try: + online_version = fetch_latest(repository_type, org, repo, tag) + except RateLimitedError as e: + logger.warning(f"[FastAPI] Rate limited, stopping the {repository_type.value} check: {e}") + results["errors"] += 1 + break + except ApiAuthenticationFailure as e: + logger.error(f"[FastAPI] {e}") + results["errors"] += 1 + break + except (RepositoryNotFoundError, RepositoryTagNotFoundError) as e: + logger.warning(f"[FastAPI] {e}, deactivating repository ID {repository_id}") + unsubscribe_every_subscriber_from_repository(db, request.app.state.notifier, repository_id, org, repo, tag) + continue + except RepositoryApiError as e: + logger.warning(f"[FastAPI] Lookup failed for ID {repository_id}: {e}") + results["errors"] += 1 + continue + except Exception: + logger.exception(f"[FastAPI] Unexpected error for ID {repository_id} ({org}/{repo}:{tag})") + results["errors"] += 1 + continue + + if missing_checks: + reset_missing_checks(db, repository_id) + if latest_entry == online_version: + continue + + try: + + logger.info(f"[FastAPI] New version detected for org={org}, repo={repo}, tag={tag}") + + with db.cursor() as cursor: + cursor.execute( + "SELECT user_id, custom_name " + "FROM watched_repositories_subscriptions " + "WHERE repository_id = %s AND active = 1", + (repository_id,), + ) + subscribers = cursor.fetchall() + + sent = 0 + failed = 0 + for user_id, custom_name in subscribers: + try: + request.app.state.notifier.send( + title=custom_name or f"{org}/{repo}", + body=f"New version {online_version} for tag {tag}", + 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 repo={org}/{repo}" + ) + break + except NotificationError as e: + failed += 1 + logger.error( + f"[FastAPI] Failed to send notification to user_id={user_id} for repo={org}/{repo} " + f"status={getattr(e, 'status_code', None)}" + ) + + if subscribers and sent == 0: + results["errors"] += 1 + logger.error(f"[FastAPI] All notifications failed for ID:{repository_id}, will retry later") + continue + + with db.cursor() as cursor: + cursor.execute( + "UPDATE watched_repositories SET latest_entry = %s WHERE id = %s", + (online_version, repository_id), + ) + db.commit() + logger.debug(f"[FastAPI] Notification sent and database updated for repository ID {repository_id}") + results["new"] += 1 + + except Exception as e: + db.rollback() + results["errors"] += 1 + logger.error( + f"[FastAPI] Failed for ID {repository_id} (org={org}, repo={repo}, tag={tag}): {e}", + exc_info=True, + ) + + if results["errors"]: + logger.warning(f"[FastAPI] {repository_type.value} check finished with errors: {results}") + raise HTTPException(status_code=500, detail=results) + logger.info(f"[FastAPI] {repository_type.value} check complete: {results}") + return results diff --git a/src/routers/user.py b/src/routers/user.py new file mode 100644 index 0000000..5a15c77 --- /dev/null +++ b/src/routers/user.py @@ -0,0 +1,147 @@ + +from fastapi import APIRouter, Depends, HTTPException +from service_auth import UserId + +from auth_config import Caller, auth +from db_module import get_db +from models import RepositoryUpdate, SubscribedRepository, SubscribedRepositoryList +from repositories import ( + validate_repository_combination_and_return_latest_tag, +) +from simple_logger_handler import setup_logger + +logger = setup_logger(__name__) + +user_router = APIRouter(dependencies=[Depends(auth.require(Caller.GATEWAY))]) + + +@user_router.post("/repository/add") +def add_repository_subscription( + user_id: UserId, + repository: RepositoryUpdate, + db=Depends(get_db), +): + name = f"{repository.org}/{repository.repo}:{repository.tag}" + logger.info(f"[FastAPI] Subscribe request: {repository.type.value} {name}, user_id={user_id}") + + with db.cursor() as cursor: + cursor.execute( + "SELECT id, active FROM watched_repositories " + "WHERE type = %s AND org = %s AND repo = %s AND tag = %s", + (repository.type.value, repository.org, repository.repo, repository.tag), + ) + known = cursor.fetchone() + + latest_entry = None + if known is None or not known[1]: + latest_entry = validate_repository_combination_and_return_latest_tag(repository) + + try: + with db.cursor() as cursor: + cursor.execute( + "INSERT INTO watched_repositories (type, org, repo, tag, latest_entry) " + "VALUES (%s, %s, %s, %s, %s) " + "ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), active = 1, missing_checks = 0, " + "latest_entry = COALESCE(VALUES(latest_entry), latest_entry)", + (repository.type.value, repository.org, repository.repo, repository.tag, latest_entry), + ) + repository_id = cursor.lastrowid + + cursor.execute( + "INSERT INTO watched_repositories_subscriptions (user_id, repository_id, custom_name) " + "VALUES (%s, %s, %s) " + "ON DUPLICATE KEY UPDATE active = 1, custom_name = VALUES(custom_name)", + (user_id, repository_id, repository.custom_name), + ) + db.commit() + except Exception as e: + db.rollback() + logger.exception(f"[FastAPI] Failed to subscribe user_id={user_id} to {name}") + raise HTTPException(status_code=500) from e + + logger.info(f"[FastAPI] user_id={user_id} subscribed to repository ID {repository_id} ({name})") + return {"status": "ok", "detail": "Subscribed to repository", "name": name} + +@user_router.post("/repository/delete") +def delete_repository_subscription( + user_id: UserId, + repository: RepositoryUpdate, + db=Depends(get_db), +): + name = f"{repository.org}/{repository.repo}:{repository.tag}" + logger.info(f"[FastAPI] Unsubscribe request: {repository.type.value} {name}, user_id={user_id}") + + try: + with db.cursor() as cursor: + cursor.execute( + "SELECT id FROM watched_repositories " + "WHERE type = %s AND org = %s AND repo = %s AND tag = %s FOR UPDATE", + (repository.type.value, repository.org, repository.repo, repository.tag), + ) + known = cursor.fetchone() + + if known is None: + db.rollback() + logger.info(f"[FastAPI] Repository {repository.type.value} {name} is not tracked, nothing to do") + return {"status":"ok","detail":"Unsubscribed from repository","name":name} + + repository_id = known[0] + cursor.execute( + "UPDATE watched_repositories_subscriptions SET active = 0 " + "WHERE user_id = %s AND repository_id = %s", + (user_id, repository_id), + ) + + cursor.execute( + "SELECT 1 from watched_repositories_subscriptions " + "WHERE repository_id = %s AND active = 1 LIMIT 1 FOR SHARE", + (repository_id,), + ) + if cursor.fetchone() is None: + cursor.execute( + "UPDATE watched_repositories SET active = 0 WHERE id = %s", + (repository_id,), + ) + db.commit() + except Exception as e: + db.rollback() + logger.exception(f"[FastAPI] Failed to unsubscribe user_id={user_id} to {name}") + raise HTTPException(status_code=500) from e + + logger.info(f"[FastAPI] Unsubscribe request successful: {repository.type.value} {name}, user_id={user_id}") + return {"status":"ok","detail":"Unsubscribed from repository","name":name} + +@user_router.post("/repository/list") +def list_subscribed_repositories( + user_id: UserId, + db = Depends(get_db), +): + logger.info(f"[FastAPI] Received request to list repositories for user_id={user_id}") + try: + with db.cursor() as cursor: + cursor.execute( + "select w.type, w.org, w.repo, w.tag, w.latest_entry, s.custom_name, s.active from watched_repositories " + "AS w JOIN watched_repositories_subscriptions AS s ON w.id = s.repository_id " + "WHERE s.user_id = %s ORDER BY w.type", + (user_id,), + ) + rows = cursor.fetchall() + + repository_list = SubscribedRepositoryList() + for type, org, repo, tag, latest_entry, custom_name, is_active in rows: + entry = SubscribedRepository( + type = type, + repository_name = custom_name or f"{org}/{repo}", + tag = tag, + latest_entry = latest_entry, + ) + (repository_list.active if is_active else repository_list.inactive).append(entry) + + except Exception as e: + logger.exception("[FastAPI] Failed to list repository") + raise HTTPException(status_code=500) from e + + logger.info(f"[FastAPI] Listed all repositories for user_id={user_id}") + return repository_list + +