From b534a496fd50b0c6c0bcf19a9f15a2aa0b4d248e Mon Sep 17 00:00:00 2001 From: retoor Date: Thu, 23 Jul 2026 01:15:04 +0200 Subject: [PATCH] update --- devplacepy/services/game/store/treasury.py | 158 +++++ devplacepy/templates/_game_grant.html | 8 + devplacepy_services/__init__.py | 1 + devplacepy_services/audit/SERVICE.md | 1 + devplacepy_services/audit/__init__.py | 0 devplacepy_services/audit/main.py | 23 + devplacepy_services/backup/SERVICE.md | 1 + devplacepy_services/backup/__init__.py | 0 devplacepy_services/backup/main.py | 23 + devplacepy_services/base/__init__.py | 5 + devplacepy_services/base/auth.py | 61 ++ devplacepy_services/base/bootstrap.py | 4 + devplacepy_services/base/cache.py | 67 ++ devplacepy_services/base/compound.py | 153 +++++ devplacepy_services/base/config.py | 70 ++ devplacepy_services/base/db_client.py | 236 +++++++ devplacepy_services/base/db_codec.py | 126 ++++ devplacepy_services/base/errors.py | 63 ++ devplacepy_services/base/health.py | 57 ++ devplacepy_services/base/http.py | 77 +++ devplacepy_services/base/manifest.py | 215 ++++++ devplacepy_services/base/middleware.py | 121 ++++ devplacepy_services/base/proxy.py | 37 ++ devplacepy_services/base/pubsub_client.py | 13 + devplacepy_services/base/schemas.py | 29 + devplacepy_services/base/service.py | 134 ++++ devplacepy_services/base/sqlite_broker.py | 110 ++++ devplacepy_services/base/stats.py | 26 + devplacepy_services/base/testing.py | 40 ++ devplacepy_services/bot/SERVICE.md | 1 + devplacepy_services/bot/__init__.py | 0 devplacepy_services/bot/main.py | 23 + devplacepy_services/containers/SERVICE.md | 1 + devplacepy_services/containers/__init__.py | 0 devplacepy_services/containers/main.py | 24 + devplacepy_services/database/SERVICE.md | 1 + devplacepy_services/database/__init__.py | 0 devplacepy_services/database/broker_setup.py | 52 ++ .../database/compounds_page.py | 135 ++++ .../database/compounds_primitive.py | 138 ++++ devplacepy_services/database/db_patch.py | 119 ++++ .../database/invoke_registry.py | 40 ++ devplacepy_services/database/main.py | 52 ++ devplacepy_services/database/routes.py | 421 ++++++++++++ devplacepy_services/devii/SERVICE.md | 1 + devplacepy_services/devii/__init__.py | 0 devplacepy_services/devii/main.py | 40 ++ devplacepy_services/devii/store_brokers.py | 27 + devplacepy_services/devii/store_routes.py | 63 ++ devplacepy_services/email/SERVICE.md | 1 + devplacepy_services/email/__init__.py | 0 devplacepy_services/email/main.py | 20 + devplacepy_services/email/routes.py | 9 + devplacepy_services/gateway/SERVICE.md | 1 + devplacepy_services/gateway/__init__.py | 0 devplacepy_services/gateway/main.py | 24 + devplacepy_services/gitea/SERVICE.md | 1 + devplacepy_services/gitea/__init__.py | 0 devplacepy_services/gitea/main.py | 23 + devplacepy_services/jobs/SERVICE.md | 1 + devplacepy_services/jobs/__init__.py | 0 devplacepy_services/jobs/main.py | 49 ++ devplacepy_services/news/SERVICE.md | 1 + devplacepy_services/news/__init__.py | 0 devplacepy_services/news/main.py | 23 + devplacepy_services/pubsub/SERVICE.md | 1 + devplacepy_services/pubsub/__init__.py | 0 devplacepy_services/pubsub/main.py | 31 + devplacepy_services/telegram/SERVICE.md | 1 + devplacepy_services/telegram/__init__.py | 0 devplacepy_services/telegram/main.py | 24 + devplacepy_services/web/SERVICE.md | 1 + devplacepy_services/web/__init__.py | 0 devplacepy_services/web/factory.py | 616 ++++++++++++++++++ devplacepy_services/web/ingress.py | 216 ++++++ devplacepy_services/web/main.py | 21 + devplacepy_services/xmlrpc/SERVICE.md | 1 + devplacepy_services/xmlrpc/__init__.py | 0 devplacepy_services/xmlrpc/main.py | 24 + 79 files changed, 4086 insertions(+) create mode 100644 devplacepy/services/game/store/treasury.py create mode 100644 devplacepy/templates/_game_grant.html create mode 100644 devplacepy_services/__init__.py create mode 100644 devplacepy_services/audit/SERVICE.md create mode 100644 devplacepy_services/audit/__init__.py create mode 100644 devplacepy_services/audit/main.py create mode 100644 devplacepy_services/backup/SERVICE.md create mode 100644 devplacepy_services/backup/__init__.py create mode 100644 devplacepy_services/backup/main.py create mode 100644 devplacepy_services/base/__init__.py create mode 100644 devplacepy_services/base/auth.py create mode 100644 devplacepy_services/base/bootstrap.py create mode 100644 devplacepy_services/base/cache.py create mode 100644 devplacepy_services/base/compound.py create mode 100644 devplacepy_services/base/config.py create mode 100644 devplacepy_services/base/db_client.py create mode 100644 devplacepy_services/base/db_codec.py create mode 100644 devplacepy_services/base/errors.py create mode 100644 devplacepy_services/base/health.py create mode 100644 devplacepy_services/base/http.py create mode 100644 devplacepy_services/base/manifest.py create mode 100644 devplacepy_services/base/middleware.py create mode 100644 devplacepy_services/base/proxy.py create mode 100644 devplacepy_services/base/pubsub_client.py create mode 100644 devplacepy_services/base/schemas.py create mode 100644 devplacepy_services/base/service.py create mode 100644 devplacepy_services/base/sqlite_broker.py create mode 100644 devplacepy_services/base/stats.py create mode 100644 devplacepy_services/base/testing.py create mode 100644 devplacepy_services/bot/SERVICE.md create mode 100644 devplacepy_services/bot/__init__.py create mode 100644 devplacepy_services/bot/main.py create mode 100644 devplacepy_services/containers/SERVICE.md create mode 100644 devplacepy_services/containers/__init__.py create mode 100644 devplacepy_services/containers/main.py create mode 100644 devplacepy_services/database/SERVICE.md create mode 100644 devplacepy_services/database/__init__.py create mode 100644 devplacepy_services/database/broker_setup.py create mode 100644 devplacepy_services/database/compounds_page.py create mode 100644 devplacepy_services/database/compounds_primitive.py create mode 100644 devplacepy_services/database/db_patch.py create mode 100644 devplacepy_services/database/invoke_registry.py create mode 100644 devplacepy_services/database/main.py create mode 100644 devplacepy_services/database/routes.py create mode 100644 devplacepy_services/devii/SERVICE.md create mode 100644 devplacepy_services/devii/__init__.py create mode 100644 devplacepy_services/devii/main.py create mode 100644 devplacepy_services/devii/store_brokers.py create mode 100644 devplacepy_services/devii/store_routes.py create mode 100644 devplacepy_services/email/SERVICE.md create mode 100644 devplacepy_services/email/__init__.py create mode 100644 devplacepy_services/email/main.py create mode 100644 devplacepy_services/email/routes.py create mode 100644 devplacepy_services/gateway/SERVICE.md create mode 100644 devplacepy_services/gateway/__init__.py create mode 100644 devplacepy_services/gateway/main.py create mode 100644 devplacepy_services/gitea/SERVICE.md create mode 100644 devplacepy_services/gitea/__init__.py create mode 100644 devplacepy_services/gitea/main.py create mode 100644 devplacepy_services/jobs/SERVICE.md create mode 100644 devplacepy_services/jobs/__init__.py create mode 100644 devplacepy_services/jobs/main.py create mode 100644 devplacepy_services/news/SERVICE.md create mode 100644 devplacepy_services/news/__init__.py create mode 100644 devplacepy_services/news/main.py create mode 100644 devplacepy_services/pubsub/SERVICE.md create mode 100644 devplacepy_services/pubsub/__init__.py create mode 100644 devplacepy_services/pubsub/main.py create mode 100644 devplacepy_services/telegram/SERVICE.md create mode 100644 devplacepy_services/telegram/__init__.py create mode 100644 devplacepy_services/telegram/main.py create mode 100644 devplacepy_services/web/SERVICE.md create mode 100644 devplacepy_services/web/__init__.py create mode 100644 devplacepy_services/web/factory.py create mode 100644 devplacepy_services/web/ingress.py create mode 100644 devplacepy_services/web/main.py create mode 100644 devplacepy_services/xmlrpc/SERVICE.md create mode 100644 devplacepy_services/xmlrpc/__init__.py create mode 100644 devplacepy_services/xmlrpc/main.py diff --git a/devplacepy/services/game/store/treasury.py b/devplacepy/services/game/store/treasury.py new file mode 100644 index 00000000..1f46afe9 --- /dev/null +++ b/devplacepy/services/game/store/treasury.py @@ -0,0 +1,158 @@ +# retoor + +from __future__ import annotations + +from datetime import datetime + +from devplacepy.database import get_table + +from .. import economy +from .common import GameError, _iso, _iso_week, _lvl, _now, conditional_update_farm +from .farm import ensure_farm, get_farm + +TREASURY_UID = "treasury-main" + + +def _treasury(): + return get_table("game_treasury") + + +def ensure_treasury() -> dict: + row = _treasury().find_one(uid=TREASURY_UID) + if row: + return row + _treasury().insert( + { + "uid": TREASURY_UID, + "balance": 0, + "collected_total": 0, + "granted_total": 0, + "updated_at": _iso(_now()), + } + ) + return _treasury().find_one(uid=TREASURY_UID) + + +def treasury_balance() -> int: + row = _treasury().find_one(uid=TREASURY_UID) + return int(row.get("balance") or 0) if row else 0 + + +def _treasury_update(set_clause: str, where_clause: str, params: dict) -> int: + from sqlalchemy import text + + from devplacepy.database import db + + sql = ( + f"UPDATE game_treasury SET {set_clause}, updated_at = :updated_at " + f"WHERE uid = :treasury_uid AND ({where_clause})" + ) + bind = {**params, "updated_at": _iso(_now()), "treasury_uid": TREASURY_UID} + with db: + result = db.executable.execute(text(sql), bind) + return result.rowcount + + +def credit_treasury(amount: int) -> None: + if amount <= 0: + return + ensure_treasury() + _treasury_update( + set_clause=( + "balance = COALESCE(balance, 0) + :amount, " + "collected_total = COALESCE(collected_total, 0) + :amount" + ), + where_clause="1 = 1", + params={"amount": amount}, + ) + + +def _debit_treasury(amount: int) -> bool: + ensure_treasury() + rows = _treasury_update( + set_clause=( + "balance = COALESCE(balance, 0) - :amount, " + "granted_total = COALESCE(granted_total, 0) + :amount" + ), + where_clause="COALESCE(balance, 0) >= :amount", + params={"amount": amount}, + ) + return rows > 0 + + +def _refund_treasury(amount: int) -> None: + if amount <= 0: + return + _treasury_update( + set_clause=( + "balance = COALESCE(balance, 0) + :amount, " + "granted_total = COALESCE(granted_total, 0) - :amount" + ), + where_clause="1 = 1", + params={"amount": amount}, + ) + + +def grant_status(farm: dict, now: datetime) -> dict: + week = _iso_week(now) + if farm.get("last_grant_week") == week: + return {"available": False, "amount": 0, "reason": "Grant already claimed this week."} + if _lvl(farm, "prestige") > economy.GRANT_MAX_PRESTIGE: + return { + "available": False, + "amount": 0, + "reason": f"Grants support farms up to prestige {economy.GRANT_MAX_PRESTIGE}.", + } + if int(farm.get("coins", 0)) >= economy.GRANT_WEALTH_CEILING: + return { + "available": False, + "amount": 0, + "reason": f"Grants support farms below {economy.GRANT_WEALTH_CEILING} coins.", + } + if _lvl(farm, "harvests_week") < economy.GRANT_MIN_WEEK_HARVESTS: + return { + "available": False, + "amount": 0, + "reason": ( + f"Harvest at least {economy.GRANT_MIN_WEEK_HARVESTS} builds this week to qualify." + ), + } + amount = economy.grant_amount(treasury_balance()) + if amount <= 0: + return { + "available": False, + "amount": 0, + "reason": "The treasury is empty - refactor fees fill it.", + } + return {"available": True, "amount": amount, "reason": ""} + + +def claim_grant(user: dict) -> dict: + farm = ensure_farm(user["uid"]) + now = _now() + status = grant_status(farm, now) + if not status["available"]: + raise GameError(status["reason"]) + amount = status["amount"] + week = _iso_week(now) + if not _debit_treasury(amount): + raise GameError("The treasury is empty - refactor fees fill it.") + rows = conditional_update_farm( + farm["uid"], + set_clause=( + "coins = coins + :amount, " + "last_grant_week = :week, " + "lifetime_coins_earned = COALESCE(lifetime_coins_earned, 0) + :amount" + ), + where_clause=( + "(last_grant_week IS NULL OR last_grant_week != :week) AND coins < :ceiling" + ), + params={"amount": amount, "week": week, "ceiling": economy.GRANT_WEALTH_CEILING}, + ) + if rows == 0: + _refund_treasury(amount) + farm = get_farm(user["uid"]) + if farm and farm.get("last_grant_week") == week: + raise GameError("Grant already claimed this week.") + raise GameError("Your farm changed - refresh and try again.") + return {"amount": amount, "week": week} diff --git a/devplacepy/templates/_game_grant.html b/devplacepy/templates/_game_grant.html new file mode 100644 index 00000000..a0833add --- /dev/null +++ b/devplacepy/templates/_game_grant.html @@ -0,0 +1,8 @@ +

Treasury: {{ farm.treasury_balance }}c collected from refactor fees

+{% if farm.grant_available %} +
+ +
+{% else %} +{{ farm.grant_reason }} +{% endif %} diff --git a/devplacepy_services/__init__.py b/devplacepy_services/__init__.py new file mode 100644 index 00000000..b8f73fd5 --- /dev/null +++ b/devplacepy_services/__init__.py @@ -0,0 +1 @@ +# retoor \ No newline at end of file diff --git a/devplacepy_services/audit/SERVICE.md b/devplacepy_services/audit/SERVICE.md new file mode 100644 index 00000000..0e2f1d67 --- /dev/null +++ b/devplacepy_services/audit/SERVICE.md @@ -0,0 +1 @@ +# Audit Service (stub) \ No newline at end of file diff --git a/devplacepy_services/audit/__init__.py b/devplacepy_services/audit/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/audit/main.py b/devplacepy_services/audit/main.py new file mode 100644 index 00000000..5d4377c3 --- /dev/null +++ b/devplacepy_services/audit/main.py @@ -0,0 +1,23 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.services.audit import AuditService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class AuditMicroservice(BaseMicroservice): + name = "audit" + title = "Audit" + default_port = 10639 + workers = 1 + stateful = True + depends_on = ["database"] + managed_services = [AuditService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self) + + +_service = AuditMicroservice() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/backup/SERVICE.md b/devplacepy_services/backup/SERVICE.md new file mode 100644 index 00000000..9091af55 --- /dev/null +++ b/devplacepy_services/backup/SERVICE.md @@ -0,0 +1 @@ +# Backup Service (stub) \ No newline at end of file diff --git a/devplacepy_services/backup/__init__.py b/devplacepy_services/backup/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/backup/main.py b/devplacepy_services/backup/main.py new file mode 100644 index 00000000..c4887186 --- /dev/null +++ b/devplacepy_services/backup/main.py @@ -0,0 +1,23 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.services.backup.service import BackupService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class BackupMicroservice(BaseMicroservice): + name = "backup" + title = "Backup" + default_port = 10633 + workers = 1 + stateful = True + depends_on = ["database"] + managed_services = [BackupService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self) + + +_service = BackupMicroservice() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/base/__init__.py b/devplacepy_services/base/__init__.py new file mode 100644 index 00000000..4f9f2712 --- /dev/null +++ b/devplacepy_services/base/__init__.py @@ -0,0 +1,5 @@ +# retoor + +from devplacepy_services.base.service import BaseMicroservice + +__all__ = ["BaseMicroservice"] \ No newline at end of file diff --git a/devplacepy_services/base/auth.py b/devplacepy_services/base/auth.py new file mode 100644 index 00000000..d96d5150 --- /dev/null +++ b/devplacepy_services/base/auth.py @@ -0,0 +1,61 @@ +# retoor + +from __future__ import annotations + +import os +from typing import Annotated + +from fastapi import Header, Request + +from devplacepy_services.base.errors import http_error +from devplacepy_services.base.schemas import InternalUser + +_KEY_CACHE = "" + + +def internal_gateway_key() -> str: + global _KEY_CACHE + if _KEY_CACHE: + return _KEY_CACHE + env_key = os.environ.get("DEVPLACE_GATEWAY_INTERNAL_KEY", "").strip() + if env_key: + _KEY_CACHE = env_key + return env_key + try: + from devplacepy.db_client import internal_gateway_key as load_key + + loaded = load_key().strip() + if loaded: + _KEY_CACHE = loaded + return loaded + except ImportError: + return "" + + +def set_internal_gateway_key(value: str) -> None: + global _KEY_CACHE + _KEY_CACHE = value.strip() + + +def validate_internal_key(presented: str | None) -> bool: + expected = internal_gateway_key() + if not expected: + return True + if not presented: + return False + return presented.strip() == expected + + +def require_internal_key(request: Request) -> None: + if validate_internal_key(request.headers.get("X-Internal-Key")): + return + raise http_error(401, "Unauthorized", "unauthorized") + + +async def internal_user( + request: Request, + x_authenticated_user: Annotated[str | None, Header()] = None, +) -> InternalUser: + require_internal_key(request) + uid = x_authenticated_user.strip() if x_authenticated_user else None + return InternalUser(uid=uid or None) \ No newline at end of file diff --git a/devplacepy_services/base/bootstrap.py b/devplacepy_services/base/bootstrap.py new file mode 100644 index 00000000..d88f16d4 --- /dev/null +++ b/devplacepy_services/base/bootstrap.py @@ -0,0 +1,4 @@ +import os + +os.environ["DEVPLACE_REMOTE_DB"] = "1" +import devplacepy.db_client \ No newline at end of file diff --git a/devplacepy_services/base/cache.py b/devplacepy_services/base/cache.py new file mode 100644 index 00000000..9e29d9aa --- /dev/null +++ b/devplacepy_services/base/cache.py @@ -0,0 +1,67 @@ +# retoor + +from __future__ import annotations + +import time +from collections import OrderedDict + + +class TTLCache: + def __init__(self, ttl: int, max_size: int = 0): + self.ttl = ttl + self.max_size = max_size + self._store: OrderedDict[str, tuple[object, float]] = OrderedDict() + + def get(self, key: str): + entry = self._store.get(key) + if entry is None: + return None + value, expiry = entry + if time.time() >= expiry: + self._store.pop(key, None) + return None + self._store.move_to_end(key) + return value + + def set(self, key: str, value) -> None: + self._store[key] = (value, time.time() + self.ttl) + self._store.move_to_end(key) + if self.max_size and len(self._store) > self.max_size: + self._store.popitem(last=False) + + def pop(self, key: str) -> None: + self._store.pop(key, None) + + def clear(self) -> None: + self._store.clear() + + +class CacheStateGate: + def __init__(self, *, ttl: int = 60, max_staleness: float = 5.0): + self.ttl = ttl + self.max_staleness = max_staleness + self._cache = TTLCache(ttl=ttl) + self._versions: dict[str, int] = {} + self._bumped_at: dict[str, float] = {} + + def get_version(self, name: str) -> int: + return int(self._versions.get(name, 0)) + + def bump(self, name: str) -> int: + next_version = self.get_version(name) + 1 + self._versions[name] = next_version + self._bumped_at[name] = time.time() + self._cache.clear() + return next_version + + def get(self, key: str, version_name: str, loader): + bumped_at = self._bumped_at.get(version_name, 0.0) + if bumped_at and (time.time() - bumped_at) < self.max_staleness: + self._cache.pop(key) + version_key = f"{version_name}:{self.get_version(version_name)}" + cached = self._cache.get(version_key) + if cached is not None: + return cached + value = loader() + self._cache.set(version_key, value) + return value \ No newline at end of file diff --git a/devplacepy_services/base/compound.py b/devplacepy_services/base/compound.py new file mode 100644 index 00000000..54c5b6f4 --- /dev/null +++ b/devplacepy_services/base/compound.py @@ -0,0 +1,153 @@ +# retoor + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field + + +class InvokeIn(BaseModel): + fn: str + args: list[Any] = Field(default_factory=list) + kwargs: dict[str, Any] = Field(default_factory=dict) + write: bool = False + + +class CacheBumpIn(BaseModel): + name: str + + +class UsersByUidsIn(BaseModel): + uids: list[str] + + +class CommentCountsIn(BaseModel): + post_uids: list[str] + + +class VoteCountsIn(BaseModel): + target_uids: list[str] + + +class ReactionsIn(BaseModel): + target_type: str + target_uids: list[str] + user: dict[str, Any] | None = None + + +class BookmarksIn(BaseModel): + user_uid: str + target_type: str + target_uids: list[str] + + +class PollsIn(BaseModel): + post_uids: list[str] + user: dict[str, Any] | None = None + + +class RecentCommentsIn(BaseModel): + post_uids: list[str] + limit: int = 3 + user: dict[str, Any] | None = None + + +class CommentsIn(BaseModel): + target_type: str + target_uid: str + user: dict[str, Any] | None = None + + +class FollowBundleIn(BaseModel): + user_uid: str + target_uids: list[str] | None = None + + +class RelationsIn(BaseModel): + viewer_uid: str | None = None + + +class OnlineUsersIn(BaseModel): + cutoff_iso: str + limit: int = 30 + + +class NotificationPrefsIn(BaseModel): + user_uid: str + + +class LeaderboardBundleIn(BaseModel): + limit: int = 50 + offset: int = 0 + viewer_uid: str | None = None + + +class SiteSidebarIn(BaseModel): + authors_limit: int = 5 + + +class AwardsBundleIn(BaseModel): + receiver_uid: str + page: int = 1 + per_page: int = 12 + profile_user: dict[str, Any] | None = None + + +class AttachmentsIn(BaseModel): + resource_type: str + resource_uids: list[str] + + +class UserMediaIn(BaseModel): + user_uid: str + page: int = 1 + per_page: int = 24 + + +class SeoMetaIn(BaseModel): + target_type: str + uids: list[str] + + +class FeedPageIn(BaseModel): + user: dict[str, Any] | None = None + tab: str = "all" + topic: str | None = None + search: str = "" + before: str | None = None + + +class PostDetailIn(BaseModel): + post_uid: str + user: dict[str, Any] | None = None + + +class ProfileBundleIn(BaseModel): + profile_uid: str + viewer: dict[str, Any] | None = None + + +class MessagesPageIn(BaseModel): + user_uid: str + + +class NotificationsPageIn(BaseModel): + user_uid: str + + +class LeaderboardPageIn(BaseModel): + viewer_uid: str | None = None + + +class ProjectDetailIn(BaseModel): + project_uid: str + + +class GistDetailIn(BaseModel): + gist_uid: str + user: dict[str, Any] | None = None + + +class GameStateIn(BaseModel): + user_uid: str | None = None \ No newline at end of file diff --git a/devplacepy_services/base/config.py b/devplacepy_services/base/config.py new file mode 100644 index 00000000..4c5ffc89 --- /dev/null +++ b/devplacepy_services/base/config.py @@ -0,0 +1,70 @@ +# retoor + +from __future__ import annotations + +import os +from pathlib import Path + +from devplacepy_services.base.manifest import ( + FORBIDDEN_GAPS, + PORT_HINTS, + SERVICE_URL_ENV, + SERVICES, + build_profiles, +) + +PROFILES = build_profiles() + +_BROKER_URL_ENV = { + "main": "DEVPLACE_DB_SERVICE_URL", + "devii_tasks": "DEVPLACE_DEVII_TASKS_STORE_URL", + "devii_lessons": "DEVPLACE_DEVII_LESSONS_STORE_URL", +} + + +def port_profile() -> str: + return os.environ.get("DEVPLACE_PORT_PROFILE", "micro-dev") + + +def data_dir() -> Path: + root = Path(__file__).resolve().parents[2] + return Path(os.environ.get("DEVPLACE_DATA_DIR", str(root / "data"))) + + +def service_url(name: str) -> str: + env_key = SERVICE_URL_ENV.get(name) + if env_key: + override = os.environ.get(env_key, "").strip().rstrip("/") + if override: + return override + profile = port_profile() + host = os.environ.get("DEVPLACE_SERVICE_HOST", "127.0.0.1") + port = PROFILES[profile][name] + return f"http://{host}:{port}" + + +def broker_url(name: str) -> str: + env_key = _BROKER_URL_ENV.get(name) + if env_key: + override = os.environ.get(env_key, "").strip().rstrip("/") + if override: + return override + if name == "main": + return service_url("database") + if name == "devii_tasks": + return f"{service_url('devii')}/internal/store/devii_tasks" + if name == "devii_lessons": + return f"{service_url('devii')}/internal/store/devii_lessons" + raise KeyError(name) + + +def sqlite_read_pool_size() -> int: + raw = os.environ.get("DEVPLACE_SQLITE_READ_POOL", "8").strip() + try: + return max(1, int(raw)) + except ValueError: + return 8 + + +def service_spec(name: str): + return SERVICES[name] \ No newline at end of file diff --git a/devplacepy_services/base/db_client.py b/devplacepy_services/base/db_client.py new file mode 100644 index 00000000..4c643765 --- /dev/null +++ b/devplacepy_services/base/db_client.py @@ -0,0 +1,236 @@ +# retoor + +from __future__ import annotations + +from typing import Any + +from devplacepy_services.base.compound import ( + AttachmentsIn, + AwardsBundleIn, + BookmarksIn, + CacheBumpIn, + CommentCountsIn, + CommentsIn, + FeedPageIn, + FollowBundleIn, + LeaderboardBundleIn, + NotificationPrefsIn, + OnlineUsersIn, + PollsIn, + ReactionsIn, + RecentCommentsIn, + RelationsIn, + SeoMetaIn, + SiteSidebarIn, + UserMediaIn, + UsersByUidsIn, + VoteCountsIn, +) +from devplacepy_services.base.errors import http_error +from devplacepy_services.base.http import internal_request + + +class BrokerClient: + def __init__(self, name: str, url: str | None = None) -> None: + self.name = name + if url is not None: + self.url = url.rstrip("/") + else: + from devplacepy_services.base.config import broker_url + + self.url = broker_url(name).rstrip("/") + + async def post(self, path: str, body: dict[str, Any] | None = None) -> Any: + target = f"{self.url}/{path.lstrip('/')}" + response = await internal_request("POST", target, json=body or {}) + if response.status_code >= 400: + payload = response.json() if response.content else {} + message = payload.get("error", "Broker request failed") + code = payload.get("code", "broker_error") + raise http_error(response.status_code, message, code) + if not response.content: + return None + return response.json() + + async def get(self, path: str, *, params: dict[str, Any] | None = None) -> Any: + target = f"{self.url}/{path.lstrip('/')}" + response = await internal_request("GET", target, params=params) + if response.status_code >= 400: + payload = response.json() if response.content else {} + message = payload.get("error", "Broker request failed") + code = payload.get("code", "broker_error") + raise http_error(response.status_code, message, code) + if not response.content: + return None + return response.json() + + +main = BrokerClient("main") +devii_tasks = BrokerClient("devii_tasks") +devii_lessons = BrokerClient("devii_lessons") + + +async def get_setting(key: str, default: str = "") -> str: + payload = await main.get(f"settings/{key}", params={"default": default}) + return payload.get("value", default) if isinstance(payload, dict) else default + + +async def get_int_setting(key: str, default: int) -> int: + raw = await get_setting(key, str(default)) + try: + return int(raw) + except (TypeError, ValueError): + return default + + +async def bump_cache_version(name: str) -> None: + await main.post("/cache/bump", CacheBumpIn(name=name).model_dump()) + + +async def get_users_by_uids(uids: list[str]) -> dict: + return await main.post("/compound/users-by-uids", UsersByUidsIn(uids=uids).model_dump()) + + +async def get_comment_counts_by_post_uids(post_uids: list[str]) -> dict: + return await main.post( + "/compound/comment-counts", CommentCountsIn(post_uids=post_uids).model_dump() + ) + + +async def get_vote_counts(target_uids: list[str]) -> tuple[dict, dict]: + payload = await main.post( + "/compound/vote-counts", VoteCountsIn(target_uids=target_uids).model_dump() + ) + return payload.get("ups", {}), payload.get("downs", {}) + + +async def get_reactions_by_targets( + target_type: str, target_uids: list[str], user: dict | None = None +) -> dict: + return await main.post( + "/compound/reactions", + ReactionsIn(target_type=target_type, target_uids=target_uids, user=user).model_dump(), + ) + + +async def get_user_bookmarks( + user_uid: str, target_type: str, target_uids: list[str] +) -> set[str]: + payload = await main.post( + "/compound/bookmarks", + BookmarksIn( + user_uid=user_uid, target_type=target_type, target_uids=target_uids + ).model_dump(), + ) + return set(payload or []) + + +async def get_polls_by_post_uids(post_uids: list[str], user: dict | None = None) -> dict: + return await main.post( + "/compound/polls", PollsIn(post_uids=post_uids, user=user).model_dump() + ) + + +async def get_recent_comments_by_post_uids( + post_uids: list[str], limit: int = 3, user: dict | None = None +) -> dict: + return await main.post( + "/compound/recent-comments", + RecentCommentsIn(post_uids=post_uids, limit=limit, user=user).model_dump(), + ) + + +async def load_comments( + target_type: str, target_uid: str, user: dict | None = None +) -> list: + return await main.post( + "/compound/comments", + CommentsIn(target_type=target_type, target_uid=target_uid, user=user).model_dump(), + ) + + +async def get_follow_bundle( + user_uid: str, target_uids: list[str] | None = None +) -> dict: + return await main.post( + "/compound/follow-bundle", + FollowBundleIn(user_uid=user_uid, target_uids=target_uids).model_dump(), + ) + + +async def get_user_relations_bundle(viewer_uid: str | None = None) -> dict: + return await main.post( + "/compound/relations", RelationsIn(viewer_uid=viewer_uid).model_dump() + ) + + +async def get_online_users(cutoff_iso: str, limit: int = 30) -> list: + return await main.post( + "/compound/online-users", + OnlineUsersIn(cutoff_iso=cutoff_iso, limit=limit).model_dump(), + ) + + +async def get_notification_prefs(user_uid: str) -> list: + return await main.post( + "/compound/notification-prefs", + NotificationPrefsIn(user_uid=user_uid).model_dump(), + ) + + +async def get_leaderboard_bundle( + limit: int = 50, offset: int = 0, viewer_uid: str | None = None +) -> dict: + return await main.post( + "/compound/leaderboard", + LeaderboardBundleIn(limit=limit, offset=offset, viewer_uid=viewer_uid).model_dump(), + ) + + +async def get_site_sidebar(authors_limit: int = 5) -> dict: + return await main.post( + "/compound/site-sidebar", SiteSidebarIn(authors_limit=authors_limit).model_dump() + ) + + +async def get_awards_bundle( + receiver_uid: str, + page: int = 1, + per_page: int = 12, + profile_user: dict | None = None, +) -> dict: + return await main.post( + "/compound/awards", + AwardsBundleIn( + receiver_uid=receiver_uid, + page=page, + per_page=per_page, + profile_user=profile_user, + ).model_dump(), + ) + + +async def get_attachments_batch(resource_type: str, resource_uids: list[str]) -> dict: + return await main.post( + "/compound/attachments", + AttachmentsIn(resource_type=resource_type, resource_uids=resource_uids).model_dump(), + ) + + +async def get_user_media( + user_uid: str, page: int = 1, per_page: int = 24 +) -> dict: + return await main.post( + "/compound/user-media", + UserMediaIn(user_uid=user_uid, page=page, per_page=per_page).model_dump(), + ) + + +async def get_seo_metadata_batch(target_type: str, uids: list[str]) -> dict: + return await main.post( + "/compound/seo-meta", SeoMetaIn(target_type=target_type, uids=uids).model_dump() + ) + + +async def build_feed_page(**kwargs) -> dict: + return await main.post("/compound/feed-page", FeedPageIn(**kwargs).model_dump()) \ No newline at end of file diff --git a/devplacepy_services/base/db_codec.py b/devplacepy_services/base/db_codec.py new file mode 100644 index 00000000..a7788185 --- /dev/null +++ b/devplacepy_services/base/db_codec.py @@ -0,0 +1,126 @@ +# retoor + +from __future__ import annotations + +from datetime import date, datetime +from pathlib import Path +from typing import Any + +WRITE_PREFIXES = ( + "set_", + "add_", + "delete_", + "create_", + "update_", + "upsert_", + "record_", + "migrate_", + "backfill_", + "mark_", + "invalidate_", + "ensure_", + "restore", + "purge", + "revoke_", + "recompute_", + "soft_delete", + "bump_", + "init_", + "clear_settings", +) + +WRITE_EXACT = frozenset( + { + "soft_delete_in", + "restore_event", + "purge_event", + "delete_engagement", + "soft_delete_engagement", + "delete_fork_relations", + "soft_delete_fork_relations", + "delete_attachments", + "delete_attachment_record", + "_delete_attachment_file", + "_index", + "_drop_index", + "_uid_index", + "_ensure_cache_state", + "_refresh_query_planner_stats", + "_backfill_gamification", + } +) + +REMOTE_TABLE_MARKER = "__remote_table__" + + +def is_write(name: str) -> bool: + if name in WRITE_EXACT: + return True + return any(name.startswith(prefix) for prefix in WRITE_PREFIXES) + + +SQL_WRITE_KEYWORDS = frozenset( + {"INSERT", "UPDATE", "DELETE", "REPLACE", "CREATE", "ALTER", "DROP"} +) + + +def is_write_sql(sql: str) -> bool: + first_word = sql.strip().split(None, 1)[0].upper() if sql.strip() else "" + return first_word in SQL_WRITE_KEYWORDS + + +def encode_value(value: Any) -> Any: + if value is None or isinstance(value, (bool, int, float, str)): + return value + if type(value).__name__ == "RemoteTable": + return {REMOTE_TABLE_MARKER: value._name} + if isinstance(value, (datetime, date)): + return value.isoformat() + if isinstance(value, Path): + return str(value) + if isinstance(value, frozenset): + return [encode_value(item) for item in value] + if isinstance(value, set): + return [encode_value(item) for item in value] + if isinstance(value, tuple): + return [encode_value(item) for item in value] + if isinstance(value, list): + return [encode_value(item) for item in value] + if isinstance(value, dict): + return {str(key): encode_value(item) for key, item in value.items()} + if hasattr(value, "items") and callable(value.items): + try: + return {str(key): encode_value(item) for key, item in value.items()} + except TypeError: + pass + return str(value) + + +def encode_args(args: tuple | list, kwargs: dict) -> tuple[list, dict]: + return [encode_value(item) for item in args], { + str(key): encode_value(value) for key, value in kwargs.items() + } + + +def decode_value(value: Any) -> Any: + if isinstance(value, list): + return [decode_value(item) for item in value] + if isinstance(value, dict): + return {key: decode_value(item) for key, item in value.items()} + return value + + +def decode_arg(value: Any) -> Any: + if isinstance(value, list): + return [decode_arg(item) for item in value] + if isinstance(value, dict): + if set(value) == {REMOTE_TABLE_MARKER}: + from devplacepy.db_client import get_table + + return get_table(value[REMOTE_TABLE_MARKER]) + return {key: decode_arg(item) for key, item in value.items()} + return value + + +def encode_result(value: Any) -> Any: + return encode_value(value) \ No newline at end of file diff --git a/devplacepy_services/base/errors.py b/devplacepy_services/base/errors.py new file mode 100644 index 00000000..c16f527e --- /dev/null +++ b/devplacepy_services/base/errors.py @@ -0,0 +1,63 @@ +# retoor + +from fastapi import HTTPException +from fastapi.responses import JSONResponse + +from devplacepy_services.base.schemas import ErrorOut + + +def error_response(status_code: int, message: str, code: str) -> JSONResponse: + return JSONResponse( + ErrorOut(error=message, code=code).model_dump(), + status_code=status_code, + ) + + +def http_error(status_code: int, message: str, code: str) -> HTTPException: + return HTTPException( + status_code=status_code, + detail=ErrorOut(error=message, code=code).model_dump(), + ) + + +def error_code_for_status(status_code: int) -> str: + if status_code == 404: + return "not_found" + if status_code == 401: + return "unauthorized" + if status_code == 403: + return "forbidden" + if status_code == 422: + return "validation_error" + if status_code >= 500: + return "internal_error" + return "error" + + +def error_message_for_status(status_code: int, detail: str | None = None) -> str: + if status_code == 404: + return "Not found" + if status_code == 401: + return "Unauthorized" + if status_code == 403: + return "Forbidden" + if status_code >= 500: + return "Internal server error" + if detail: + return detail + return "Request failed" + + +def sanitize_detail(detail: object, status_code: int = 400) -> tuple[str, str]: + if isinstance(detail, dict): + if "error" in detail and "code" in detail: + return str(detail["error"]), str(detail["code"]) + if "message" in detail: + return str(detail["message"]), error_code_for_status(status_code) + if isinstance(detail, list): + return "Validation failed", "validation_error" + if isinstance(detail, str): + return error_message_for_status(status_code, detail), error_code_for_status(status_code) + if detail is None: + return error_message_for_status(status_code), error_code_for_status(status_code) + return error_message_for_status(status_code, str(detail)), error_code_for_status(status_code) \ No newline at end of file diff --git a/devplacepy_services/base/health.py b/devplacepy_services/base/health.py new file mode 100644 index 00000000..e9c18625 --- /dev/null +++ b/devplacepy_services/base/health.py @@ -0,0 +1,57 @@ +# retoor + +from __future__ import annotations + +import asyncio +import time + +from fastapi import APIRouter + +from devplacepy_services.base.config import service_url +from devplacepy_services.base.http import internal_request +from devplacepy_services.base.schemas import HealthOut, HealthStatsOut +from devplacepy_services.base.service import BaseMicroservice +from devplacepy_services.base.stats import COUNTERS + + +async def _dep_status(name: str) -> str: + url = f"{service_url(name)}/health" + try: + response = await internal_request("GET", url, timeout=2.0) + if response.status_code == 200: + payload = response.json() + if payload.get("status") == "ok": + return "ok" + return "degraded" + return "down" + except Exception: + return "down" + + +def health_router(service: BaseMicroservice) -> APIRouter: + router = APIRouter() + + @router.get("/health", response_model=HealthOut) + async def health() -> HealthOut: + deps: dict[str, str] = {} + if service.depends_on: + results = await asyncio.gather( + *[_dep_status(dep) for dep in service.depends_on] + ) + for dep_name, status in zip(service.depends_on, results, strict=True): + deps[dep_name] = status + overall = "ok" + if deps and any(status != "ok" for status in deps.values()): + overall = "degraded" + uptime_s = int(time.monotonic() - service.started_at) + stats = COUNTERS.snapshot() + return HealthOut( + service=service.name, + status=overall, + uptime_s=uptime_s, + version=service.version, + deps=deps, + stats=HealthStatsOut(requests=stats["requests"], errors=stats["errors"]), + ) + + return router \ No newline at end of file diff --git a/devplacepy_services/base/http.py b/devplacepy_services/base/http.py new file mode 100644 index 00000000..7462bc90 --- /dev/null +++ b/devplacepy_services/base/http.py @@ -0,0 +1,77 @@ +# retoor + +from __future__ import annotations + +import contextvars +import os +from typing import Any + +import httpx + +from devplacepy import stealth + +INTERNAL_CALLS: contextvars.ContextVar[int] = contextvars.ContextVar( + "internal_calls", default=0 +) +REQUEST_ID: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="") + +_POOL: httpx.AsyncClient | None = None + + +def bump_internal_calls() -> None: + INTERNAL_CALLS.set(INTERNAL_CALLS.get() + 1) + + +def current_internal_calls() -> int: + return INTERNAL_CALLS.get() + + +def current_request_id() -> str: + return REQUEST_ID.get() + + +async def startup_pool() -> None: + global _POOL + if _POOL is None: + _POOL = httpx.AsyncClient( + limits=httpx.Limits(max_connections=32, max_keepalive_connections=32), + timeout=httpx.Timeout(30.0), + http2=False, + ) + + +async def shutdown_pool() -> None: + global _POOL + if _POOL is not None: + await _POOL.aclose() + _POOL = None + + +def stealth_async_client(**kwargs: Any) -> httpx.AsyncClient: + return stealth.stealth_async_client(**kwargs) + + +async def internal_request( + method: str, + url: str, + *, + json: dict | None = None, + params: dict | None = None, + headers: dict[str, str] | None = None, + timeout: float = 30.0, +) -> httpx.Response: + await startup_pool() + assert _POOL is not None + bump_internal_calls() + merged: dict[str, str] = {} + request_id = current_request_id() + if request_id: + merged["X-Request-Id"] = request_id + internal_key = os.environ.get("DEVPLACE_GATEWAY_INTERNAL_KEY", "").strip() + if internal_key: + merged["X-Internal-Key"] = internal_key + if headers: + merged.update(headers) + return await _POOL.request( + method, url, json=json, params=params, headers=merged, timeout=timeout + ) \ No newline at end of file diff --git a/devplacepy_services/base/manifest.py b/devplacepy_services/base/manifest.py new file mode 100644 index 00000000..ed02e786 --- /dev/null +++ b/devplacepy_services/base/manifest.py @@ -0,0 +1,215 @@ +# retoor + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +WorkersSpec = int | Literal["auto"] + + +@dataclass(frozen=True) +class ServiceSpec: + name: str + module: str + port: int + tier: int + workers: WorkersSpec + stateful: bool + depends_on: tuple[str, ...] + health_client_timeout: float = 2.0 + + +@dataclass(frozen=True) +class IngressRoute: + prefix: str + service: str + websocket: bool = False + + +SERVICE_SPECS: tuple[ServiceSpec, ...] = ( + ServiceSpec("database", "devplacepy_services.database.main:app", 10601, 1, 1, True, ()), + ServiceSpec("pubsub", "devplacepy_services.pubsub.main:app", 10602, 1, 1, True, ("database",)), + ServiceSpec( + "web", + "devplacepy_services.web.main:app", + 10500, + 2, + "auto", + False, + ("database", "pubsub"), + health_client_timeout=4.0, + ), + ServiceSpec( + "gateway", + "devplacepy_services.gateway.main:app", + 10620, + 2, + "auto", + False, + ("database",), + health_client_timeout=4.0, + ), + ServiceSpec( + "jobs", + "devplacepy_services.jobs.main:app", + 10630, + 3, + 1, + True, + ("database", "pubsub", "gateway"), + health_client_timeout=8.0, + ), + ServiceSpec( + "devii", + "devplacepy_services.devii.main:app", + 10631, + 3, + 2, + True, + ("database", "pubsub", "gateway"), + health_client_timeout=4.0, + ), + ServiceSpec("bot", "devplacepy_services.bot.main:app", 10632, 3, 1, True, ("database", "gateway")), + ServiceSpec("backup", "devplacepy_services.backup.main:app", 10633, 3, 1, True, ("database",)), + ServiceSpec( + "containers", + "devplacepy_services.containers.main:app", + 10634, + 3, + 1, + True, + ("database",), + health_client_timeout=4.0, + ), + ServiceSpec("telegram", "devplacepy_services.telegram.main:app", 10635, 3, 1, True, ("database", "pubsub")), + ServiceSpec("email", "devplacepy_services.email.main:app", 10636, 3, 1, True, ("database",)), + ServiceSpec( + "news", + "devplacepy_services.news.main:app", + 10637, + 3, + 1, + True, + ("database", "gateway"), + ), + ServiceSpec("gitea", "devplacepy_services.gitea.main:app", 10638, 3, 1, True, ("database", "gateway")), + ServiceSpec("audit", "devplacepy_services.audit.main:app", 10639, 3, 1, True, ("database",)), + ServiceSpec("xmlrpc", "devplacepy_services.xmlrpc.main:app", 10649, 3, 1, True, ("database",)), +) + +SERVICES: dict[str, ServiceSpec] = {spec.name: spec for spec in SERVICE_SPECS} + +TIER_ORDER: dict[int, tuple[str, ...]] = { + 1: ("database", "pubsub"), + 2: ("web", "gateway"), + 3: ( + "jobs", + "devii", + "bot", + "backup", + "containers", + "telegram", + "email", + "news", + "gitea", + "audit", + "xmlrpc", + ), +} + +SERVICE_URL_ENV: dict[str, str] = { + "web": "DEVPLACE_WEB_URL", + "database": "DEVPLACE_DB_SERVICE_URL", + "pubsub": "DEVPLACE_PUBSUB_URL", + "gateway": "DEVPLACE_GATEWAY_URL", + "jobs": "DEVPLACE_JOBS_URL", + "devii": "DEVPLACE_DEVII_URL", + "bot": "DEVPLACE_BOT_URL", + "backup": "DEVPLACE_BACKUP_URL", + "containers": "DEVPLACE_CONTAINERS_URL", + "telegram": "DEVPLACE_TELEGRAM_URL", + "email": "DEVPLACE_EMAIL_URL", + "news": "DEVPLACE_NEWS_URL", + "gitea": "DEVPLACE_GITEA_URL", + "audit": "DEVPLACE_AUDIT_URL", + "xmlrpc": "DEVPLACE_XMLRPC_URL", +} + +INGRESS_ROUTES: tuple[IngressRoute, ...] = ( + IngressRoute("/openai", "gateway"), + IngressRoute("/devii", "devii", websocket=True), + IngressRoute("/pubsub", "pubsub", websocket=True), + IngressRoute("/zips", "jobs"), + IngressRoute("/forks", "jobs"), + IngressRoute("/tools", "jobs", websocket=True), + IngressRoute("/xmlrpc", "xmlrpc"), +) + +INGRESS_WS_PREFIXES: tuple[tuple[str, str], ...] = ( + ( + "/projects/", + "containers", + ), +) + +FORBIDDEN_GAPS: tuple[tuple[int, int], ...] = ( + (10520, 10599), + (10610, 10619), + (10650, 10699), + (10750, 10799), + (20550, 20599), + (20650, 20699), +) + +PORT_HINTS: dict[int, str] = { + 10500: "web service", + 20500: "test orchestrator", + 10502: "locust", +} + +_MICRO_DEV_PORTS: dict[str, int] = {spec.name: spec.port for spec in SERVICE_SPECS} + +XMLRPC_RAW_SOCKET_PORT_DEV = 10648 + + +def xmlrpc_raw_socket_port(profile: str) -> int: + if profile == "micro-test": + return XMLRPC_RAW_SOCKET_PORT_DEV + 10000 + if profile == "micro-prod": + return XMLRPC_RAW_SOCKET_PORT_DEV + 100 + return XMLRPC_RAW_SOCKET_PORT_DEV + + +def build_profiles() -> dict[str, dict[str, int]]: + return { + "micro-dev": dict(_MICRO_DEV_PORTS), + "micro-test": { + k: (20500 if k == "web" else v + 10000) for k, v in _MICRO_DEV_PORTS.items() + }, + "micro-prod": { + k: (10500 if k == "web" else v + 100) for k, v in _MICRO_DEV_PORTS.items() + }, + "locust": {"app": 10502, "ui": 10503}, + } + + +def orchestrator_services_dict() -> dict[str, dict]: + return { + spec.name: { + "module": spec.module, + "port": spec.port, + "tier": spec.tier, + "workers": spec.workers, + "stateful": spec.stateful, + "depends_on": list(spec.depends_on), + } + for spec in SERVICE_SPECS + } + + +def health_client_timeout(name: str) -> float: + spec = SERVICES.get(name) + if spec is None: + return 2.0 + return spec.health_client_timeout \ No newline at end of file diff --git a/devplacepy_services/base/middleware.py b/devplacepy_services/base/middleware.py new file mode 100644 index 00000000..0ab66c06 --- /dev/null +++ b/devplacepy_services/base/middleware.py @@ -0,0 +1,121 @@ +# retoor + +from __future__ import annotations + +import json +import logging +import time +import uuid_utils +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request +from starlette.responses import Response + +from devplacepy_services.base.auth import validate_internal_key +from devplacepy_services.base.errors import error_response, sanitize_detail +from devplacepy_services.base.http import INTERNAL_CALLS, REQUEST_ID, current_internal_calls +from devplacepy_services.base.stats import COUNTERS + +logger = logging.getLogger(__name__) + +PUBLIC_PATHS = {"/health"} + + +class RequestStatsMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + COUNTERS.record_request() + response = await call_next(request) + if response.status_code >= 500: + COUNTERS.record_error() + return response + + +class InternalCallCounterMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + token_calls = INTERNAL_CALLS.set(0) + response = await call_next(request) + service_name = getattr(request.app.state, "service_name", "") + if service_name in {"web", "gateway"}: + response.headers["X-Internal-Calls"] = str(current_internal_calls()) + INTERNAL_CALLS.reset(token_calls) + return response + + +class InternalAuthMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + path = request.url.path + if path in PUBLIC_PATHS: + return await call_next(request) + if not validate_internal_key(request.headers.get("X-Internal-Key")): + return error_response(401, "Unauthorized", "unauthorized") + return await call_next(request) + + +class SanitizeErrorsMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + request_id = request.headers.get("X-Request-Id") or uuid_utils.uuid7().hex + token_id = REQUEST_ID.set(request_id) + started = time.perf_counter() + try: + response = await call_next(request) + except Exception: + COUNTERS.record_error() + logger.exception( + "unhandled service error", + extra={ + "request_id": request_id, + "service": getattr(request.app.state, "service_name", ""), + "path": request.url.path, + }, + ) + response = error_response(500, "Internal server error", "internal_error") + finally: + REQUEST_ID.reset(token_id) + duration_ms = int((time.perf_counter() - started) * 1000) + service_name = getattr(request.app.state, "service_name", "") + logger.info( + "request", + extra={ + "request_id": request_id, + "service": service_name, + "path": request.url.path, + "status": response.status_code, + "duration_ms": duration_ms, + "internal_calls": current_internal_calls(), + }, + ) + if response.status_code < 400: + return response + if not hasattr(response, "body_iterator"): + return response + content_type = response.headers.get("content-type", "") + if "application/json" not in content_type: + if response.status_code == 404: + return error_response(404, "Not found", "not_found") + if response.status_code == 401: + return error_response(401, "Unauthorized", "unauthorized") + if response.status_code >= 500: + return error_response(500, "Internal server error", "internal_error") + return error_response(response.status_code, "Request failed", "error") + body = b"" + async for chunk in response.body_iterator: + body += chunk + try: + parsed = json.loads(body) + except json.JSONDecodeError: + return error_response(response.status_code, "Request failed", "error") + if isinstance(parsed, dict) and "error" in parsed and "code" in parsed: + return Response( + content=body, + status_code=response.status_code, + media_type="application/json", + ) + if isinstance(parsed, dict) and isinstance(parsed.get("error"), dict) and "message" in parsed["error"]: + return Response( + content=body, + status_code=response.status_code, + media_type="application/json", + ) + if isinstance(parsed, dict) and "detail" in parsed: + message, code = sanitize_detail(parsed["detail"], response.status_code) + return error_response(response.status_code, message, code) + return error_response(response.status_code, "Request failed", "error") \ No newline at end of file diff --git a/devplacepy_services/base/proxy.py b/devplacepy_services/base/proxy.py new file mode 100644 index 00000000..14a7c6c4 --- /dev/null +++ b/devplacepy_services/base/proxy.py @@ -0,0 +1,37 @@ +# retoor + +from __future__ import annotations + +from starlette.requests import Request + +HOP_HEADERS = frozenset( + { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + "host", + "content-length", + "content-encoding", + } +) + + +def forward_headers(request: Request, prefix: str) -> dict[str, str]: + headers = { + k: v for k, v in request.headers.items() if k.lower() not in HOP_HEADERS + } + headers["X-Forwarded-Prefix"] = prefix + headers["X-Script-Name"] = prefix + headers["X-Forwarded-Host"] = request.headers.get( + "host", request.url.hostname or "" + ) + headers["X-Forwarded-Proto"] = request.headers.get( + "x-forwarded-proto", request.url.scheme + ) + headers["Accept-Encoding"] = "identity" + return headers \ No newline at end of file diff --git a/devplacepy_services/base/pubsub_client.py b/devplacepy_services/base/pubsub_client.py new file mode 100644 index 00000000..ce9abeda --- /dev/null +++ b/devplacepy_services/base/pubsub_client.py @@ -0,0 +1,13 @@ +# retoor + +from __future__ import annotations + +from typing import Any + +from devplacepy_services.base.config import service_url +from devplacepy_services.base.http import internal_request + + +async def publish(topic: str, data: dict[str, Any]) -> None: + url = f"{service_url('pubsub')}/internal/publish" + await internal_request("POST", url, json={"topic": topic, "data": data}) \ No newline at end of file diff --git a/devplacepy_services/base/schemas.py b/devplacepy_services/base/schemas.py new file mode 100644 index 00000000..0f34f03d --- /dev/null +++ b/devplacepy_services/base/schemas.py @@ -0,0 +1,29 @@ +# retoor + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict + + +class ErrorOut(BaseModel): + error: str + code: str + + +class HealthStatsOut(BaseModel): + requests: int + errors: int + + +class HealthOut(BaseModel): + service: str + status: str + uptime_s: int + version: str + deps: dict[str, str] + stats: HealthStatsOut + + +class InternalUser(BaseModel): + model_config = ConfigDict(frozen=True) + uid: str | None = None \ No newline at end of file diff --git a/devplacepy_services/base/service.py b/devplacepy_services/base/service.py new file mode 100644 index 00000000..c5fdc00b --- /dev/null +++ b/devplacepy_services/base/service.py @@ -0,0 +1,134 @@ +# retoor + +from __future__ import annotations + +import os +import subprocess +import time +from contextlib import asynccontextmanager +from typing import Literal + +from fastapi import APIRouter, FastAPI +from fastapi.exceptions import HTTPException + +from devplacepy_services.base.config import PROFILES, port_profile +from devplacepy_services.base.errors import error_response, sanitize_detail +from devplacepy_services.base.http import shutdown_pool, startup_pool +from devplacepy_services.base.middleware import ( + InternalAuthMiddleware, + InternalCallCounterMiddleware, + RequestStatsMiddleware, + SanitizeErrorsMiddleware, +) + + +class BaseMicroservice: + name: str = "" + title: str = "" + default_port: int = 0 + workers: int | Literal["auto"] = 1 + stateful: bool = True + depends_on: list[str] = [] + managed_services: list = [] + use_background: bool = False + run_supervisor: bool = False + + def __init__(self) -> None: + self.started_at = time.monotonic() + self.version = self._resolve_version() + + def _resolve_version(self) -> str: + env_version = os.environ.get("DEVPLACE_VERSION", "").strip() + if env_version: + return env_version + try: + result = subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + if result.returncode == 0: + return result.stdout.strip() or "unknown" + except (OSError, subprocess.TimeoutExpired): + pass + return "unknown" + + def resolved_port(self) -> int: + if self.default_port: + return self.default_port + profile = port_profile() + return PROFILES[profile][self.name] + + def apply_base_middleware(self, app: FastAPI, *, internal_auth: bool = False) -> None: + app.add_middleware(SanitizeErrorsMiddleware) + if internal_auth: + app.add_middleware(InternalAuthMiddleware) + app.add_middleware(InternalCallCounterMiddleware) + app.add_middleware(RequestStatsMiddleware) + + @asynccontextmanager + async def lifespan(self, app: FastAPI): + app.state.service_name = self.name + self.started_at = time.monotonic() + await startup_pool() + background = None + if self.use_background: + from devplacepy.services.background import background + + await background.start() + background = background + if self.managed_services: + from devplacepy.services.manager import service_manager + + for svc in self.managed_services: + service_manager.register(svc) + service_manager.set_lock_owner(True) + if self.run_supervisor: + import asyncio + + asyncio.get_running_loop().call_soon(service_manager.supervise) + yield + if self.managed_services: + from devplacepy.services.manager import service_manager + + await service_manager.shutdown_all() + if background is not None: + await background.stop() + await shutdown_pool() + + def create_app(self) -> FastAPI: + app = FastAPI(title=self.title, lifespan=self.lifespan) + app.state.service_name = self.name + + @app.exception_handler(HTTPException) + async def http_exception_handler(_request, exc: HTTPException): + if isinstance(exc.detail, dict) and "error" in exc.detail and "code" in exc.detail: + return error_response(exc.status_code, exc.detail["error"], exc.detail["code"]) + message, code = sanitize_detail(exc.detail, exc.status_code) + return error_response(exc.status_code, message, code) + + return app + + def build_app(self) -> FastAPI: + raise NotImplementedError + + +def build_standard_app( + service: BaseMicroservice, + *, + routers: list[tuple[APIRouter, str] | tuple[APIRouter]] | None = None, + internal_auth: bool = False, +) -> FastAPI: + from devplacepy_services.base.health import health_router + + app = service.create_app() + service.apply_base_middleware(app, internal_auth=internal_auth) + app.include_router(health_router(service)) + for entry in routers or []: + if len(entry) == 2: + app.include_router(entry[0], prefix=entry[1]) + else: + app.include_router(entry[0]) + return app \ No newline at end of file diff --git a/devplacepy_services/base/sqlite_broker.py b/devplacepy_services/base/sqlite_broker.py new file mode 100644 index 00000000..c41030aa --- /dev/null +++ b/devplacepy_services/base/sqlite_broker.py @@ -0,0 +1,110 @@ +# retoor + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from pathlib import Path +from typing import Any, TypeVar + +import dataset + +from devplacepy_services.base.config import sqlite_read_pool_size + +T = TypeVar("T") + +PRAGMAS = [ + "PRAGMA journal_mode=WAL", + "PRAGMA synchronous=NORMAL", + "PRAGMA busy_timeout=30000", + "PRAGMA cache_size=-8000", + "PRAGMA temp_store=MEMORY", + "PRAGMA mmap_size=268435456", +] + + +class SQLiteFileBroker: + name: str + path: Path + read_pool_size: int + + def __init__(self, name: str, path: Path, read_pool_size: int | None = None) -> None: + self.name = name + self.path = Path(path) + self.read_pool_size = read_pool_size or sqlite_read_pool_size() + self._write_db: Any = None + self._read_pool: asyncio.Queue[Any] | None = None + self._write_queue: asyncio.Queue[Any] | None = None + self._write_worker_task: asyncio.Task | None = None + + def _connect_rw(self): + return dataset.connect( + f"sqlite:///{self.path}", + engine_kwargs={ + "connect_args": { + "timeout": 30, + "check_same_thread": False, + }, + }, + on_connect_statements=PRAGMAS, + ) + + def _connect_ro(self): + abs_path = self.path.resolve().as_posix() + return dataset.connect( + f"sqlite:///file:{abs_path}?uri=true&mode=ro", + engine_kwargs={ + "connect_args": { + "timeout": 30, + "check_same_thread": False, + "uri": True, + }, + }, + on_connect_statements=PRAGMAS, + ) + + async def startup(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + self._write_db = self._connect_rw() + self._read_pool = asyncio.Queue() + for _ in range(self.read_pool_size): + await self._read_pool.put(self._connect_ro()) + self._write_queue = asyncio.Queue() + self._write_worker_task = asyncio.create_task(self._write_worker()) + + async def shutdown(self) -> None: + if self._write_worker_task is not None: + self._write_worker_task.cancel() + try: + await self._write_worker_task + except asyncio.CancelledError: + pass + self._write_worker_task = None + + async def read(self, fn: Callable[..., T], *args, **kwargs) -> T: + assert self._read_pool is not None + db = await self._read_pool.get() + try: + return await asyncio.to_thread(fn, db, *args, **kwargs) + finally: + await self._read_pool.put(db) + + async def write(self, fn: Callable[..., T], *args, **kwargs) -> T: + assert self._write_queue is not None + loop = asyncio.get_running_loop() + future = loop.create_future() + await self._write_queue.put((fn, args, kwargs, future)) + return await future + + async def _write_worker(self) -> None: + assert self._write_queue is not None + while True: + fn, args, kwargs, future = await self._write_queue.get() + try: + result = fn(self._write_db, *args, **kwargs) + future.set_result(result) + except Exception as exc: + future.set_exception(exc) + + async def read_after_write(self, fn: Callable[..., T], *args, **kwargs) -> T: + return await asyncio.to_thread(fn, self._write_db, *args, **kwargs) \ No newline at end of file diff --git a/devplacepy_services/base/stats.py b/devplacepy_services/base/stats.py new file mode 100644 index 00000000..f4020e0c --- /dev/null +++ b/devplacepy_services/base/stats.py @@ -0,0 +1,26 @@ +# retoor + +from dataclasses import dataclass, field +import threading + + +@dataclass +class RequestCounters: + requests: int = 0 + errors: int = 0 + _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + def record_request(self) -> None: + with self._lock: + self.requests += 1 + + def record_error(self) -> None: + with self._lock: + self.errors += 1 + + def snapshot(self) -> dict[str, int]: + with self._lock: + return {"requests": self.requests, "errors": self.errors} + + +COUNTERS = RequestCounters() \ No newline at end of file diff --git a/devplacepy_services/base/testing.py b/devplacepy_services/base/testing.py new file mode 100644 index 00000000..5b253d93 --- /dev/null +++ b/devplacepy_services/base/testing.py @@ -0,0 +1,40 @@ +# retoor + +from __future__ import annotations + +from typing import Any + +from fastapi import FastAPI +from starlette.testclient import TestClient + +from devplacepy_services.base.auth import set_internal_gateway_key + + +class ServiceTestClient: + def __init__( + self, + app: FastAPI, + *, + internal_key: str = "test-internal-key", + authenticated_user: str | None = None, + ) -> None: + set_internal_gateway_key(internal_key) + headers: dict[str, str] = {"X-Internal-Key": internal_key} + if authenticated_user: + headers["X-Authenticated-User"] = authenticated_user + self._client = TestClient(app, headers=headers) + + def get(self, path: str, **kwargs: Any): + return self._client.get(path, **kwargs) + + def post(self, path: str, **kwargs: Any): + return self._client.post(path, **kwargs) + + def put(self, path: str, **kwargs: Any): + return self._client.put(path, **kwargs) + + def delete(self, path: str, **kwargs: Any): + return self._client.delete(path, **kwargs) + + def close(self) -> None: + self._client.close() \ No newline at end of file diff --git a/devplacepy_services/bot/SERVICE.md b/devplacepy_services/bot/SERVICE.md new file mode 100644 index 00000000..ceaf1e3a --- /dev/null +++ b/devplacepy_services/bot/SERVICE.md @@ -0,0 +1 @@ +# Bot Service (stub) \ No newline at end of file diff --git a/devplacepy_services/bot/__init__.py b/devplacepy_services/bot/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/bot/main.py b/devplacepy_services/bot/main.py new file mode 100644 index 00000000..0f44e24f --- /dev/null +++ b/devplacepy_services/bot/main.py @@ -0,0 +1,23 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.services.bot.service import BotsService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class BotService(BaseMicroservice): + name = "bot" + title = "Bot" + default_port = 10632 + workers = 1 + stateful = True + depends_on = ["database", "gateway"] + managed_services = [BotsService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self) + + +_service = BotService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/containers/SERVICE.md b/devplacepy_services/containers/SERVICE.md new file mode 100644 index 00000000..cb64b7e1 --- /dev/null +++ b/devplacepy_services/containers/SERVICE.md @@ -0,0 +1 @@ +# Containers Service (stub) \ No newline at end of file diff --git a/devplacepy_services/containers/__init__.py b/devplacepy_services/containers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/containers/main.py b/devplacepy_services/containers/main.py new file mode 100644 index 00000000..2749c85b --- /dev/null +++ b/devplacepy_services/containers/main.py @@ -0,0 +1,24 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.routers.projects import containers +from devplacepy.services.containers.service import ContainerService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class ContainersService(BaseMicroservice): + name = "containers" + title = "Containers" + default_port = 10634 + workers = 1 + stateful = True + depends_on = ["database"] + managed_services = [ContainerService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self, routers=[(containers.router, "/projects")]) + + +_service = ContainersService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/database/SERVICE.md b/devplacepy_services/database/SERVICE.md new file mode 100644 index 00000000..48f53008 --- /dev/null +++ b/devplacepy_services/database/SERVICE.md @@ -0,0 +1 @@ +# Database Service (stub) \ No newline at end of file diff --git a/devplacepy_services/database/__init__.py b/devplacepy_services/database/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/database/broker_setup.py b/devplacepy_services/database/broker_setup.py new file mode 100644 index 00000000..a88a83cb --- /dev/null +++ b/devplacepy_services/database/broker_setup.py @@ -0,0 +1,52 @@ +# retoor + +from __future__ import annotations + +import os +from pathlib import Path + +from devplacepy.config import DATA_DIR, DATABASE_URL +from devplacepy_services.base.sqlite_broker import SQLiteFileBroker + + +def _db_path() -> Path: + # DEVPLACE_DATABASE_URL is the documented override (tests use a dedicated + # tempfile for isolation - see tests/conftest.py). DATABASE_URL already + # falls back to DATA_DIR / "devplace.db" when the env var is unset, so + # deriving the broker's path from it (instead of hardcoding DATA_DIR + # directly) keeps the broker and every other DEVPLACE_DATABASE_URL + # consumer (devplacepy/database/core.py) pointed at the same file. + if DATABASE_URL.startswith("sqlite:///"): + raw = DATABASE_URL[len("sqlite:///") :] + if raw and raw != ":memory:": + return Path(raw) + return DATA_DIR / "devplace.db" + + +_broker = SQLiteFileBroker("main", _db_path()) + + +def get_broker() -> SQLiteFileBroker: + return _broker + + +async def startup() -> None: + await _broker.startup() + from devplacepy_services.database.db_patch import patch_all_db, set_fallback_db + + patch_all_db() + set_fallback_db(_broker._write_db) + os.environ["DEVPLACE_DB_SERVICE"] = "1" + from devplacepy.database import init_db + + init_db() + from devplacepy.database.settings import internal_gateway_key + + key = (internal_gateway_key() or "").strip() + if key: + os.environ["DEVPLACE_GATEWAY_INTERNAL_KEY"] = key + (DATA_DIR / ".internal_key").write_text(key + "\n") + + +async def shutdown() -> None: + await _broker.shutdown() \ No newline at end of file diff --git a/devplacepy_services/database/compounds_page.py b/devplacepy_services/database/compounds_page.py new file mode 100644 index 00000000..b420de35 --- /dev/null +++ b/devplacepy_services/database/compounds_page.py @@ -0,0 +1,135 @@ +# retoor + +from __future__ import annotations + +from devplacepy.database import ( + get_daily_topic, + get_polls_by_post_uids, + get_reactions_by_targets, + get_recent_comments_by_post_uids, + get_site_stats, + get_top_authors, + get_user_bookmarks, + get_users_by_uids, +) +from devplacepy.database.attachments_data import get_attachments_by_type +from devplacepy.routers.feed import get_feed_posts + + +def build_feed_page( + user: dict | None = None, + tab: str = "all", + topic: str | None = None, + search: str = "", + before: str | None = None, +) -> dict: + posts, next_cursor = get_feed_posts(user, tab, topic, search, before) + stats = get_site_stats() + top_authors = get_top_authors(5) + daily_topic = get_daily_topic() + post_uids = [item["post"]["uid"] for item in posts] + attachments_map = get_attachments_by_type("post", post_uids) + recent_comments = get_recent_comments_by_post_uids(post_uids, 3, user) + reactions_map = get_reactions_by_targets("post", post_uids, user) + bookmark_set = ( + get_user_bookmarks(user["uid"], "post", post_uids) if user else set() + ) + polls_map = get_polls_by_post_uids(post_uids, user) + for item in posts: + uid = item["post"]["uid"] + item["attachments"] = attachments_map.get(uid, []) + item["recent_comments"] = recent_comments.get(uid, []) + item["reactions"] = reactions_map.get(uid, {"counts": {}, "mine": []}) + item["bookmarked"] = uid in bookmark_set + item["poll"] = polls_map.get(uid) + return { + "posts": posts, + "next_cursor": next_cursor, + "stats": stats, + "top_authors": top_authors, + "daily_topic": daily_topic, + "online_users": [], + "current_tab": tab, + "current_topic": topic, + "search": search, + } + + +def build_post_detail(post_uid: str, user: dict | None = None) -> dict: + from devplacepy.database import get_table, load_comments + + post = get_table("posts").find_one(uid=post_uid, deleted_at=None) + if not post: + return {"post": None, "author": None, "comments": [], "attachments": []} + author = get_users_by_uids([post["user_uid"]]).get(post["user_uid"]) + comments = load_comments("post", post_uid, user) + attachments = get_attachments_by_type("post", [post_uid]).get(post_uid, []) + return { + "post": post, + "author": author, + "comments": comments, + "attachments": attachments, + } + + +def build_profile_bundle(profile_uid: str, viewer: dict | None = None) -> dict: + from devplacepy.database import get_table + from devplacepy.database.follows import get_follow_counts + + user = get_table("users").find_one(uid=profile_uid) + if not user: + return {"user": None, "follow_counts": {}, "awards": [], "posts_count": 0} + follow_counts = get_follow_counts(profile_uid) + return { + "user": user, + "follow_counts": follow_counts, + "awards": [], + "posts_count": 0, + "viewer_relation": {}, + } + + +def build_messages_page(user_uid: str) -> dict: + return {"threads": [], "unread": 0, "partners": {}} + + +def build_notifications_page(user_uid: str) -> dict: + return {"notifications": [], "actors": {}, "unread_count": 0} + + +def build_leaderboard_page(viewer_uid: str | None = None) -> dict: + from devplacepy.database import get_leaderboard, get_user_rank + + return { + "leaderboard": get_leaderboard(), + "viewer_rank": get_user_rank(viewer_uid) if viewer_uid else None, + } + + +def build_project_detail(project_uid: str) -> dict: + from devplacepy.database import get_table + + project = get_table("projects").find_one(uid=project_uid, deleted_at=None) + owner = None + if project: + owner = get_users_by_uids([project["user_uid"]]).get(project["user_uid"]) + return {"project": project, "owner": owner, "files": [], "containers": []} + + +def build_gist_detail(gist_uid: str, user: dict | None = None) -> dict: + from devplacepy.database import get_table, load_comments + + gist = get_table("gists").find_one(uid=gist_uid, deleted_at=None) + author = None + if gist: + author = get_users_by_uids([gist["user_uid"]]).get(gist["user_uid"]) + comments = load_comments("gist", gist_uid, user) if gist else [] + return {"gist": gist, "author": author, "comments": comments} + + +def build_game_state(user_uid: str | None = None) -> dict: + return {"store": {}, "leaderboard": []} + + +def build_admin_dashboard() -> dict: + return {"stats": get_site_stats(), "service_states": []} \ No newline at end of file diff --git a/devplacepy_services/database/compounds_primitive.py b/devplacepy_services/database/compounds_primitive.py new file mode 100644 index 00000000..570268ce --- /dev/null +++ b/devplacepy_services/database/compounds_primitive.py @@ -0,0 +1,138 @@ +# retoor + +from __future__ import annotations + +from devplacepy.database import ( + get_attachments_by_type, + get_comment_counts_by_post_uids, + get_follow_counts, + get_following_among, + get_leaderboard, + get_muted_uids, + get_blocked_uids, + get_user_relations, + get_notification_prefs, + get_online_users, + get_polls_by_post_uids, + get_prominent_award, + get_reactions_by_targets, + get_recent_comments_by_post_uids, + get_seo_metadata_batch, + get_site_stats, + get_top_authors, + get_user_awards, + get_user_bookmarks, + get_user_media, + get_user_rank, + get_users_by_uids, + get_vote_counts, + load_comments, +) + + +def users_by_uids(uids: list[str]) -> dict: + return get_users_by_uids(uids) + + +def comment_counts(post_uids: list[str]) -> dict: + return get_comment_counts_by_post_uids(post_uids) + + +def vote_counts(target_uids: list[str]) -> dict: + ups, downs = get_vote_counts(target_uids) + return {"ups": ups, "downs": downs} + + +def reactions(target_type: str, target_uids: list[str], user: dict | None = None) -> dict: + return get_reactions_by_targets(target_type, target_uids, user) + + +def bookmarks(user_uid: str, target_type: str, target_uids: list[str]) -> list[str]: + result = get_user_bookmarks(user_uid, target_type, target_uids) + return sorted(result) + + +def polls(post_uids: list[str], user: dict | None = None) -> dict: + return get_polls_by_post_uids(post_uids, user) + + +def recent_comments( + post_uids: list[str], limit: int = 3, user: dict | None = None +) -> dict: + return get_recent_comments_by_post_uids(post_uids, limit, user) + + +def comments(target_type: str, target_uid: str, user: dict | None = None) -> list: + return load_comments(target_type, target_uid, user) + + +def follow_bundle(user_uid: str, target_uids: list[str] | None = None) -> dict: + payload = { + "counts": get_follow_counts(user_uid), + "following_among": sorted(get_following_among(user_uid, target_uids or [])), + } + return payload + + +def relations_bundle(viewer_uid: str | None) -> dict: + relations = get_user_relations(viewer_uid) + return { + "relations": { + "block": sorted(relations["block"]), + "mute": sorted(relations["mute"]), + }, + "blocked_uids": sorted(get_blocked_uids(viewer_uid)), + "muted_uids": sorted(get_muted_uids(viewer_uid)), + } + + +def online_users_bundle(cutoff_iso: str, limit: int = 30) -> list: + return get_online_users(cutoff_iso, limit) + + +def notification_prefs_bundle(user_uid: str) -> list: + return get_notification_prefs(user_uid) + + +def leaderboard_bundle( + limit: int = 50, offset: int = 0, viewer_uid: str | None = None +) -> dict: + return { + "leaderboard": get_leaderboard(limit, offset), + "viewer_rank": get_user_rank(viewer_uid) if viewer_uid else None, + } + + +def site_sidebar(authors_limit: int = 5) -> dict: + return { + "stats": get_site_stats(), + "top_authors": get_top_authors(authors_limit), + } + + +def awards_bundle( + receiver_uid: str, + page: int = 1, + per_page: int = 12, + profile_user: dict | None = None, +) -> dict: + items, pagination = get_user_awards(receiver_uid, page, per_page) + prominent = get_prominent_award(profile_user) if profile_user else None + return { + "items": items, + "pagination": pagination, + "prominent": prominent, + } + + +def attachments_batch(resource_type: str, resource_uids: list[str]) -> dict: + return get_attachments_by_type(resource_type, resource_uids) + + +def user_media_bundle(user_uid: str, page: int = 1, per_page: int = 24) -> dict: + items, pagination = get_user_media(user_uid, page, per_page) + return {"items": items, "pagination": pagination} + + +def seo_meta_batch(target_type: str, uids: list[str]) -> dict: + return get_seo_metadata_batch(target_type, uids) \ No newline at end of file diff --git a/devplacepy_services/database/db_patch.py b/devplacepy_services/database/db_patch.py new file mode 100644 index 00000000..337b97b2 --- /dev/null +++ b/devplacepy_services/database/db_patch.py @@ -0,0 +1,119 @@ +# retoor + +from __future__ import annotations + +import contextvars +import importlib +from typing import Any + +_DB_MODULES = ( + "schema", + "content", + "pagination", + "stats", + "forks", + "awards", + "customization", + "soft_delete", + "usage", + "email", + "follows", + "settings", + "seo_meta", + "deepsearch", + "activity", + "attachments_data", + "users", + "notifications", + "comments", + "ranking", + "engagement", + "relations", +) + +_EXTERNAL_DB_MODULES = ( + "devplacepy.services.audit.store", + "devplacepy.services.backup.store", + "devplacepy.services.openai_gateway.routing", + "devplacepy.services.openai_gateway.usage", + "devplacepy.services.openai_gateway.analytics", + "devplacepy.services.base", + "devplacepy.services.jobs.queue", + "devplacepy.services.containers.store", + "devplacepy.services.devii.store", + "devplacepy.services.devii.hub", + "devplacepy.attachments", + "devplacepy.project_files", +) + +# The read pool runs on asyncio.to_thread worker threads while the write +# lane runs on the main event loop task - a plain module-global "current +# connection" swapped in/out per call is a data race between them (a +# concurrent read could reassign the global to a read-only connection +# while a write is still mid-flight, raising "attempt to write a readonly +# database"). asyncio.to_thread propagates a COPY of the calling task's +# contextvars.Context into the new thread, so storing the active +# connection in a ContextVar gives each call its own isolated view with +# no cross-talk, while every db-layer module keeps a single shared proxy +# object as its permanent `db` binding. +_CURRENT_DB: contextvars.ContextVar[Any] = contextvars.ContextVar("current_db", default=None) + +# Fallback for any code path that touches the db outside an explicit +# use_db()-scoped call (e.g. startup's own init_db()) - write-once at +# broker startup, read-only afterward, so it carries no race of its own. +_fallback_db: Any = None + + +def _resolve() -> Any: + db = _CURRENT_DB.get() + if db is not None: + return db + if _fallback_db is not None: + return _fallback_db + raise RuntimeError("no database connection configured for this context") + + +class _ContextDb: + def __getattr__(self, name: str) -> Any: + return getattr(_resolve(), name) + + def __getitem__(self, name: str) -> Any: + return _resolve()[name] + + def __enter__(self): + return _resolve().__enter__() + + def __exit__(self, exc_type, exc, tb): + return _resolve().__exit__(exc_type, exc, tb) + + +_CONTEXT_DB = _ContextDb() + + +def patch_all_db() -> None: + import devplacepy.database as database_pkg + import devplacepy.database.core as core + + core.db = _CONTEXT_DB + database_pkg.db = _CONTEXT_DB + for name in _DB_MODULES: + mod = importlib.import_module(f"devplacepy.database.{name}") + if hasattr(mod, "db"): + mod.db = _CONTEXT_DB + for name in _EXTERNAL_DB_MODULES: + mod = importlib.import_module(name) + if hasattr(mod, "db"): + mod.db = _CONTEXT_DB + + +def set_fallback_db(db_conn) -> None: + global _fallback_db + _fallback_db = db_conn + + +def use_db(db_conn) -> contextvars.Token: + return _CURRENT_DB.set(db_conn) + + +def reset_db(token: contextvars.Token) -> None: + _CURRENT_DB.reset(token) diff --git a/devplacepy_services/database/invoke_registry.py b/devplacepy_services/database/invoke_registry.py new file mode 100644 index 00000000..ea44c9af --- /dev/null +++ b/devplacepy_services/database/invoke_registry.py @@ -0,0 +1,40 @@ +# retoor + +from __future__ import annotations + +import inspect +from typing import Any + +import devplacepy.database as db_module + +from devplacepy_services.base.db_codec import decode_arg, encode_result, is_write + + +def build_registry() -> dict[str, Any]: + registry: dict[str, Any] = {} + for name in db_module.__all__: + target = getattr(db_module, name, None) + if target is None or not callable(target): + continue + if inspect.isclass(target): + continue + registry[name] = target + return registry + + +REGISTRY = build_registry() + + +_encode_result = encode_result + + +def run_with_db(db_conn, fn, args, kwargs): + from devplacepy_services.database.db_patch import use_db, reset_db + + token = use_db(db_conn) + try: + decoded_args = decode_arg(args) + decoded_kwargs = decode_arg(kwargs) + return fn(*decoded_args, **decoded_kwargs) + finally: + reset_db(token) \ No newline at end of file diff --git a/devplacepy_services/database/main.py b/devplacepy_services/database/main.py new file mode 100644 index 00000000..e85d6417 --- /dev/null +++ b/devplacepy_services/database/main.py @@ -0,0 +1,52 @@ +# retoor + +from __future__ import annotations + +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from devplacepy_services.base.health import health_router +from devplacepy_services.base.http import shutdown_pool, startup_pool +from devplacepy_services.base.middleware import ( + InternalAuthMiddleware, + InternalCallCounterMiddleware, + RequestStatsMiddleware, + SanitizeErrorsMiddleware, +) +from devplacepy_services.base.service import BaseMicroservice +from devplacepy_services.database import broker_setup +from devplacepy_services.database.routes import router as database_router + + +class DatabaseService(BaseMicroservice): + name = "database" + title = "Database" + default_port = 10601 + workers = 1 + stateful = True + depends_on = [] + + @asynccontextmanager + async def lifespan(self, app: FastAPI): + app.state.service_name = self.name + self.started_at = __import__("time").monotonic() + await startup_pool() + await broker_setup.startup() + yield + await broker_setup.shutdown() + await shutdown_pool() + + def build_app(self): + app = self.create_app() + app.add_middleware(SanitizeErrorsMiddleware) + app.add_middleware(InternalAuthMiddleware) + app.add_middleware(InternalCallCounterMiddleware) + app.add_middleware(RequestStatsMiddleware) + app.include_router(health_router(self)) + app.include_router(database_router) + return app + + +_service = DatabaseService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/database/routes.py b/devplacepy_services/database/routes.py new file mode 100644 index 00000000..3729811c --- /dev/null +++ b/devplacepy_services/database/routes.py @@ -0,0 +1,421 @@ +# retoor + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends +from pydantic import BaseModel, Field + +from devplacepy.database import bump_cache_version, get_int_setting, get_setting +from devplacepy_services.base.auth import internal_user +from devplacepy_services.base.compound import ( + AttachmentsIn, + AwardsBundleIn, + BookmarksIn, + CacheBumpIn, + CommentCountsIn, + CommentsIn, + FeedPageIn, + FollowBundleIn, + GameStateIn, + GistDetailIn, + InvokeIn, + LeaderboardBundleIn, + LeaderboardPageIn, + MessagesPageIn, + NotificationPrefsIn, + NotificationsPageIn, + OnlineUsersIn, + PollsIn, + PostDetailIn, + ProfileBundleIn, + ProjectDetailIn, + ReactionsIn, + RecentCommentsIn, + RelationsIn, + SeoMetaIn, + SiteSidebarIn, + UserMediaIn, + UsersByUidsIn, + VoteCountsIn, +) +from devplacepy_services.base.db_codec import is_write_sql +from devplacepy_services.base.schemas import InternalUser +from devplacepy_services.database import compounds_page, compounds_primitive +from devplacepy_services.database.broker_setup import get_broker +from devplacepy_services.database.invoke_registry import ( + REGISTRY, + _encode_result, + is_write, + run_with_db, +) + +router = APIRouter() + + +class DbOpIn(BaseModel): + op: str + table: str | None = None + method: str | None = None + args: list[Any] = Field(default_factory=list) + kwargs: dict[str, Any] = Field(default_factory=dict) + write: bool = False + + +def _db_tables(db_conn): + return list(db_conn.tables) + + +def _db_query(db_conn, sql: str, **params): + return list(db_conn.query(sql, **params)) + + +def _db_table_op(db_conn, table: str, method: str, args: list, kwargs: dict): + target = db_conn[table] + fn = getattr(target, method) + result = fn(*args, **kwargs) + if hasattr(result, "__iter__") and not isinstance(result, (str, bytes, dict)): + try: + return [_encode_result(row) for row in result] + except TypeError: + pass + return _encode_result(result) + + +@router.post("/internal/invoke") +async def internal_invoke(body: InvokeIn, _: InternalUser = Depends(internal_user)): + fn = REGISTRY.get(body.fn) + if fn is None: + return {"error": f"Unknown function: {body.fn}", "code": "unknown_function"} + write = body.write or is_write(body.fn) + broker = get_broker() + if write: + result = await broker.write(run_with_db, fn, body.args, body.kwargs) + else: + result = await broker.read(run_with_db, fn, body.args, body.kwargs) + return {"result": _encode_result(result)} + + +@router.post("/internal/db-op") +async def internal_db_op(body: DbOpIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + if body.op == "tables": + tables = await broker.read(_db_tables) + return tables + if body.op == "query": + sql = body.args[0] if body.args else "" + write = body.write or is_write_sql(sql) + if write: + rows = await broker.write(_db_query, sql, **body.kwargs) + else: + rows = await broker.read(_db_query, sql, **body.kwargs) + return [_encode_result(row) for row in rows] + if body.op == "table_op": + if not body.table or not body.method: + return {"error": "table and method required", "code": "invalid_request"} + write_methods = {"insert", "update", "delete", "create_column_by_example"} + write = body.write or body.method in write_methods + if write: + result = await broker.write( + _db_table_op, body.table, body.method, body.args, body.kwargs + ) + else: + result = await broker.read( + _db_table_op, body.table, body.method, body.args, body.kwargs + ) + return result + return {"error": f"Unknown op: {body.op}", "code": "unknown_op"} + + +@router.get("/settings/{key}") +async def settings_get(key: str, default: str = "", _: InternalUser = Depends(internal_user)): + broker = get_broker() + value = await broker.read(run_with_db, get_setting, [key, default], {}) + return {"key": key, "value": value} + + +@router.post("/cache/bump") +async def cache_bump(body: CacheBumpIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + await broker.write(run_with_db, bump_cache_version, [body.name], {}) + return {"ok": True, "name": body.name} + + +@router.post("/compound/users-by-uids") +async def compound_users_by_uids(body: UsersByUidsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read(run_with_db, compounds_primitive.users_by_uids, [body.uids], {}) + + +@router.post("/compound/comment-counts") +async def compound_comment_counts(body: CommentCountsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_primitive.comment_counts, [body.post_uids], {} + ) + + +@router.post("/compound/vote-counts") +async def compound_vote_counts(body: VoteCountsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_primitive.vote_counts, [body.target_uids], {} + ) + + +@router.post("/compound/reactions") +async def compound_reactions(body: ReactionsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.reactions, + [body.target_type, body.target_uids, body.user], + {}, + ) + + +@router.post("/compound/bookmarks") +async def compound_bookmarks(body: BookmarksIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.bookmarks, + [body.user_uid, body.target_type, body.target_uids], + {}, + ) + + +@router.post("/compound/polls") +async def compound_polls(body: PollsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_primitive.polls, [body.post_uids, body.user], {} + ) + + +@router.post("/compound/recent-comments") +async def compound_recent_comments( + body: RecentCommentsIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.recent_comments, + [body.post_uids, body.limit, body.user], + {}, + ) + + +@router.post("/compound/comments") +async def compound_comments(body: CommentsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.comments, + [body.target_type, body.target_uid, body.user], + {}, + ) + + +@router.post("/compound/follow-bundle") +async def compound_follow_bundle(body: FollowBundleIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.follow_bundle, + [body.user_uid, body.target_uids], + {}, + ) + + +@router.post("/compound/relations") +async def compound_relations(body: RelationsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_primitive.relations_bundle, [body.viewer_uid], {} + ) + + +@router.post("/compound/online-users") +async def compound_online_users(body: OnlineUsersIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.online_users_bundle, + [body.cutoff_iso, body.limit], + {}, + ) + + +@router.post("/compound/notification-prefs") +async def compound_notification_prefs( + body: NotificationPrefsIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.notification_prefs_bundle, + [body.user_uid], + {}, + ) + + +@router.post("/compound/leaderboard") +async def compound_leaderboard( + body: LeaderboardBundleIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.leaderboard_bundle, + [body.limit, body.offset, body.viewer_uid], + {}, + ) + + +@router.post("/compound/site-sidebar") +async def compound_site_sidebar(body: SiteSidebarIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_primitive.site_sidebar, [body.authors_limit], {} + ) + + +@router.post("/compound/awards") +async def compound_awards(body: AwardsBundleIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.awards_bundle, + [body.receiver_uid, body.page, body.per_page, body.profile_user], + {}, + ) + + +@router.post("/compound/attachments") +async def compound_attachments(body: AttachmentsIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.attachments_batch, + [body.resource_type, body.resource_uids], + {}, + ) + + +@router.post("/compound/user-media") +async def compound_user_media(body: UserMediaIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.user_media_bundle, + [body.user_uid, body.page, body.per_page], + {}, + ) + + +@router.post("/compound/seo-meta") +async def compound_seo_meta(body: SeoMetaIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_primitive.seo_meta_batch, + [body.target_type, body.uids], + {}, + ) + + +@router.post("/compound/feed-page") +async def compound_feed_page(body: FeedPageIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_page.build_feed_page, [], body.model_dump() + ) + + +@router.post("/compound/post-detail") +async def compound_post_detail(body: PostDetailIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_page.build_post_detail, + [body.post_uid, body.user], + {}, + ) + + +@router.post("/compound/profile-bundle") +async def compound_profile_bundle( + body: ProfileBundleIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_page.build_profile_bundle, + [body.profile_uid, body.viewer], + {}, + ) + + +@router.post("/compound/messages-page") +async def compound_messages_page(body: MessagesPageIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_page.build_messages_page, [body.user_uid], {} + ) + + +@router.post("/compound/notifications-page") +async def compound_notifications_page( + body: NotificationsPageIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, compounds_page.build_notifications_page, [body.user_uid], {} + ) + + +@router.post("/compound/leaderboard-page") +async def compound_leaderboard_page( + body: LeaderboardPageIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, compounds_page.build_leaderboard_page, [body.viewer_uid], {} + ) + + +@router.post("/compound/project-detail") +async def compound_project_detail( + body: ProjectDetailIn, _: InternalUser = Depends(internal_user) +): + broker = get_broker() + return await broker.read( + run_with_db, compounds_page.build_project_detail, [body.project_uid], {} + ) + + +@router.post("/compound/gist-detail") +async def compound_gist_detail(body: GistDetailIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, + compounds_page.build_gist_detail, + [body.gist_uid, body.user], + {}, + ) + + +@router.post("/compound/game-state") +async def compound_game_state(body: GameStateIn, _: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read( + run_with_db, compounds_page.build_game_state, [body.user_uid], {} + ) + + +@router.post("/compound/admin-dashboard") +async def compound_admin_dashboard(_: InternalUser = Depends(internal_user)): + broker = get_broker() + return await broker.read(run_with_db, compounds_page.build_admin_dashboard, [], {}) \ No newline at end of file diff --git a/devplacepy_services/devii/SERVICE.md b/devplacepy_services/devii/SERVICE.md new file mode 100644 index 00000000..b3780e96 --- /dev/null +++ b/devplacepy_services/devii/SERVICE.md @@ -0,0 +1 @@ +# Devii Service (stub) \ No newline at end of file diff --git a/devplacepy_services/devii/__init__.py b/devplacepy_services/devii/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/devii/main.py b/devplacepy_services/devii/main.py new file mode 100644 index 00000000..a381ce41 --- /dev/null +++ b/devplacepy_services/devii/main.py @@ -0,0 +1,40 @@ +import devplacepy_services.base.bootstrap + +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from devplacepy.routers import devii +from devplacepy.services.devii import DeviiService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app +from devplacepy_services.devii import store_brokers +from devplacepy_services.devii.store_routes import router as store_router + + +class DeviiMicroservice(BaseMicroservice): + name = "devii" + title = "Devii" + default_port = 10631 + workers = 2 + stateful = True + depends_on = ["database", "pubsub", "gateway"] + managed_services = [DeviiService()] + use_background = True + run_supervisor = True + + @asynccontextmanager + async def lifespan(self, app: FastAPI): + await store_brokers.startup() + async with super(DeviiMicroservice, self).lifespan(app): + yield + await store_brokers.shutdown() + + def build_app(self): + return build_standard_app( + self, + routers=[(store_router,), (devii.router, "/devii")], + ) + + +_service = DeviiMicroservice() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/devii/store_brokers.py b/devplacepy_services/devii/store_brokers.py new file mode 100644 index 00000000..4dc36f6b --- /dev/null +++ b/devplacepy_services/devii/store_brokers.py @@ -0,0 +1,27 @@ +# retoor + +from __future__ import annotations + +from devplacepy.config import DEVII_LESSONS_DB, DEVII_TASKS_DB +from devplacepy_services.base.sqlite_broker import SQLiteFileBroker + +_tasks = SQLiteFileBroker("devii_tasks", DEVII_TASKS_DB) +_lessons = SQLiteFileBroker("devii_lessons", DEVII_LESSONS_DB) + + +def broker_for(name: str) -> SQLiteFileBroker: + if name == "devii_tasks": + return _tasks + if name == "devii_lessons": + return _lessons + raise KeyError(name) + + +async def startup() -> None: + await _tasks.startup() + await _lessons.startup() + + +async def shutdown() -> None: + await _tasks.shutdown() + await _lessons.shutdown() \ No newline at end of file diff --git a/devplacepy_services/devii/store_routes.py b/devplacepy_services/devii/store_routes.py new file mode 100644 index 00000000..8efa0a11 --- /dev/null +++ b/devplacepy_services/devii/store_routes.py @@ -0,0 +1,63 @@ +# retoor + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends +from pydantic import BaseModel, Field + +from devplacepy_services.base.auth import internal_user +from devplacepy_services.base.schemas import InternalUser +from devplacepy_services.database.invoke_registry import _encode_result +from devplacepy_services.database.routes import DbOpIn, _db_query, _db_table_op, _db_tables +from devplacepy_services.devii.store_brokers import broker_for + +router = APIRouter() + + +class StoreDbOpIn(BaseModel): + op: str + table: str | None = None + method: str | None = None + args: list[Any] = Field(default_factory=list) + kwargs: dict[str, Any] = Field(default_factory=dict) + write: bool = False + + +def _make_handler(store_name: str): + async def handler(body: StoreDbOpIn, _: InternalUser = Depends(internal_user)): + broker = broker_for(store_name) + if body.op == "tables": + return await broker.read(_db_tables) + if body.op == "query": + sql = body.args[0] if body.args else "" + rows = await broker.read(_db_query, sql, **body.kwargs) + return [_encode_result(row) for row in rows] + if body.op == "table_op": + if not body.table or not body.method: + return {"error": "table and method required", "code": "invalid_request"} + write_methods = {"insert", "update", "delete", "create_column_by_example"} + write = body.write or body.method in write_methods + if write: + return await broker.write( + _db_table_op, body.table, body.method, body.args, body.kwargs + ) + return await broker.read( + _db_table_op, body.table, body.method, body.args, body.kwargs + ) + return {"error": f"Unknown op: {body.op}", "code": "unknown_op"} + + return handler + + +router.add_api_route( + "/internal/store/devii_tasks/db-op", + _make_handler("devii_tasks"), + methods=["POST"], +) +router.add_api_route( + "/internal/store/devii_lessons/db-op", + _make_handler("devii_lessons"), + methods=["POST"], +) \ No newline at end of file diff --git a/devplacepy_services/email/SERVICE.md b/devplacepy_services/email/SERVICE.md new file mode 100644 index 00000000..78c586a4 --- /dev/null +++ b/devplacepy_services/email/SERVICE.md @@ -0,0 +1 @@ +# Email Service (stub) \ No newline at end of file diff --git a/devplacepy_services/email/__init__.py b/devplacepy_services/email/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/email/main.py b/devplacepy_services/email/main.py new file mode 100644 index 00000000..a311af82 --- /dev/null +++ b/devplacepy_services/email/main.py @@ -0,0 +1,20 @@ +import devplacepy_services.base.bootstrap + +from devplacepy_services.base.service import BaseMicroservice, build_standard_app +from devplacepy_services.email.routes import router + + +class EmailService(BaseMicroservice): + name = "email" + title = "Email" + default_port = 10636 + workers = 1 + stateful = True + depends_on = ["database"] + + def build_app(self): + return build_standard_app(self, routers=[(router,)]) + + +_service = EmailService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/email/routes.py b/devplacepy_services/email/routes.py new file mode 100644 index 00000000..ca92024a --- /dev/null +++ b/devplacepy_services/email/routes.py @@ -0,0 +1,9 @@ +from fastapi import APIRouter +from fastapi.responses import JSONResponse + +router = APIRouter() + + +@router.get("/status") +async def email_status(): + return JSONResponse({"service": "email", "status": "stub"}) \ No newline at end of file diff --git a/devplacepy_services/gateway/SERVICE.md b/devplacepy_services/gateway/SERVICE.md new file mode 100644 index 00000000..828bf831 --- /dev/null +++ b/devplacepy_services/gateway/SERVICE.md @@ -0,0 +1 @@ +# Gateway Service (stub) \ No newline at end of file diff --git a/devplacepy_services/gateway/__init__.py b/devplacepy_services/gateway/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/gateway/main.py b/devplacepy_services/gateway/main.py new file mode 100644 index 00000000..40bfaac3 --- /dev/null +++ b/devplacepy_services/gateway/main.py @@ -0,0 +1,24 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.routers import openai_gateway +from devplacepy.services.openai_gateway import GatewayService as OpenAIGatewayService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class GatewayService(BaseMicroservice): + name = "gateway" + title = "Gateway" + default_port = 10620 + workers = "auto" + stateful = False + depends_on = ["database"] + managed_services = [OpenAIGatewayService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self, routers=[(openai_gateway.router, "/openai")]) + + +_service = GatewayService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/gitea/SERVICE.md b/devplacepy_services/gitea/SERVICE.md new file mode 100644 index 00000000..e44045a4 --- /dev/null +++ b/devplacepy_services/gitea/SERVICE.md @@ -0,0 +1 @@ +# Gitea Service (stub) \ No newline at end of file diff --git a/devplacepy_services/gitea/__init__.py b/devplacepy_services/gitea/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/gitea/main.py b/devplacepy_services/gitea/main.py new file mode 100644 index 00000000..7d5647b9 --- /dev/null +++ b/devplacepy_services/gitea/main.py @@ -0,0 +1,23 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.services.gitea.service import IssueTrackerService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class GiteaService(BaseMicroservice): + name = "gitea" + title = "Gitea" + default_port = 10638 + workers = 1 + stateful = True + depends_on = ["database", "gateway"] + managed_services = [IssueTrackerService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self) + + +_service = GiteaService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/jobs/SERVICE.md b/devplacepy_services/jobs/SERVICE.md new file mode 100644 index 00000000..ff34829e --- /dev/null +++ b/devplacepy_services/jobs/SERVICE.md @@ -0,0 +1 @@ +# Jobs Service (stub) \ No newline at end of file diff --git a/devplacepy_services/jobs/__init__.py b/devplacepy_services/jobs/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/jobs/main.py b/devplacepy_services/jobs/main.py new file mode 100644 index 00000000..44738559 --- /dev/null +++ b/devplacepy_services/jobs/main.py @@ -0,0 +1,49 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.routers import forks, tools, zips +from devplacepy.services.jobs.award_service import AwardService +from devplacepy.services.jobs.deepsearch.service import DeepsearchService +from devplacepy.services.jobs.fork_service import ForkService +from devplacepy.services.jobs.issue_create_service import IssueCreateService +from devplacepy.services.jobs.isslop.service import IsslopService +from devplacepy.services.jobs.planning_service import PlanningReportService +from devplacepy.services.jobs.seo.service import SeoService +from devplacepy.services.jobs.seo_meta_service import SeoMetaService +from devplacepy.services.jobs.zip_service import ZipService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class JobsService(BaseMicroservice): + name = "jobs" + title = "Jobs" + default_port = 10630 + workers = 1 + stateful = True + depends_on = ["database", "pubsub", "gateway"] + managed_services = [ + ZipService(), + ForkService(), + SeoService(), + SeoMetaService(), + AwardService(), + DeepsearchService(), + IsslopService(), + IssueCreateService(), + PlanningReportService(), + ] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app( + self, + routers=[ + (zips.router, "/zips"), + (forks.router, "/forks"), + (tools.router, "/tools"), + ], + ) + + +_service = JobsService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/news/SERVICE.md b/devplacepy_services/news/SERVICE.md new file mode 100644 index 00000000..ef5f758b --- /dev/null +++ b/devplacepy_services/news/SERVICE.md @@ -0,0 +1 @@ +# News Service (stub) \ No newline at end of file diff --git a/devplacepy_services/news/__init__.py b/devplacepy_services/news/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/news/main.py b/devplacepy_services/news/main.py new file mode 100644 index 00000000..343eb883 --- /dev/null +++ b/devplacepy_services/news/main.py @@ -0,0 +1,23 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.services.news import NewsService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class NewsMicroservice(BaseMicroservice): + name = "news" + title = "News" + default_port = 10637 + workers = 1 + stateful = True + depends_on = ["database", "gateway"] + managed_services = [NewsService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self) + + +_service = NewsMicroservice() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/pubsub/SERVICE.md b/devplacepy_services/pubsub/SERVICE.md new file mode 100644 index 00000000..9aac94d2 --- /dev/null +++ b/devplacepy_services/pubsub/SERVICE.md @@ -0,0 +1 @@ +# PubSub Service (stub) \ No newline at end of file diff --git a/devplacepy_services/pubsub/__init__.py b/devplacepy_services/pubsub/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/pubsub/main.py b/devplacepy_services/pubsub/main.py new file mode 100644 index 00000000..e54407a2 --- /dev/null +++ b/devplacepy_services/pubsub/main.py @@ -0,0 +1,31 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.routers import pubsub +from devplacepy.services.live_view_relay import LiveViewRelayService +from devplacepy.services.notification_relay import NotificationRelayService +from devplacepy.services.presence_relay import PresenceRelayService +from devplacepy.services.pubsub import PubSubService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class PubsubService(BaseMicroservice): + name = "pubsub" + title = "PubSub" + default_port = 10602 + workers = 1 + stateful = True + depends_on = ["database"] + managed_services = [ + PubSubService(), + PresenceRelayService(), + NotificationRelayService(), + LiveViewRelayService(), + ] + run_supervisor = True + + def build_app(self): + return build_standard_app(self, routers=[(pubsub.router, "/pubsub")]) + + +_service = PubsubService() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/telegram/SERVICE.md b/devplacepy_services/telegram/SERVICE.md new file mode 100644 index 00000000..511cd06d --- /dev/null +++ b/devplacepy_services/telegram/SERVICE.md @@ -0,0 +1 @@ +# Telegram Service (stub) \ No newline at end of file diff --git a/devplacepy_services/telegram/__init__.py b/devplacepy_services/telegram/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/telegram/main.py b/devplacepy_services/telegram/main.py new file mode 100644 index 00000000..d46c6681 --- /dev/null +++ b/devplacepy_services/telegram/main.py @@ -0,0 +1,24 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.services.telegram.outbox_service import TelegramOutboxService +from devplacepy.services.telegram.service import TelegramService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class TelegramMicroservice(BaseMicroservice): + name = "telegram" + title = "Telegram" + default_port = 10635 + workers = 1 + stateful = True + depends_on = ["database", "pubsub"] + managed_services = [TelegramService(), TelegramOutboxService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self) + + +_service = TelegramMicroservice() +app = _service.build_app() \ No newline at end of file diff --git a/devplacepy_services/web/SERVICE.md b/devplacepy_services/web/SERVICE.md new file mode 100644 index 00000000..6381d08a --- /dev/null +++ b/devplacepy_services/web/SERVICE.md @@ -0,0 +1 @@ +# Web Service (stub) \ No newline at end of file diff --git a/devplacepy_services/web/__init__.py b/devplacepy_services/web/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/web/factory.py b/devplacepy_services/web/factory.py new file mode 100644 index 00000000..f1d416c1 --- /dev/null +++ b/devplacepy_services/web/factory.py @@ -0,0 +1,616 @@ +import os + +import devplacepy.db_client + +import asyncio +import logging +import time +from collections import defaultdict +from contextlib import asynccontextmanager +from pathlib import Path + +from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse +from fastapi.staticfiles import StaticFiles +from starlette.middleware.gzip import GZipMiddleware + +from devplacepy.cache import TTLCache +from devplacepy.config import ( + PORT, + STATIC_DIR, + STATIC_VERSION, + UPLOADS_DIR, + ensure_data_dirs, +) +from devplacepy.db_client import ( + db, + get_blocked_uids, + get_comment_counts_by_post_uids, + get_int_setting, + get_news_images_by_uids, + get_setting, + get_table, + get_user_post_count, + get_user_stars, + get_users_by_uids, + get_vote_counts, + interleave_by_author, +) +from devplacepy.responses import json_error, respond, wants_json +from devplacepy.routers import ( + admin, + auth, + avatar, + awards, + bookmarks, + comments, + dbapi, + devrant, + docs, + feed, + follow, + game, + gists, + issues, + leaderboard, + media, + messages, + news, + notifications, + polls, + posts, + profile, + projects, + proxy, + push, + reactions, + relations, + seo, + uploads, + votes, +) +from devplacepy.schemas import LandingOut, ValidationErrorOut +from devplacepy.seo import base_seo_context, site_url, website_schema +from devplacepy.services import presence +from devplacepy.services.audit import record as audit +from devplacepy.services.correction import PENDING_SCOPE_KEY +from devplacepy.templating import templates +from devplacepy.utils import client_ip, get_current_user, safe_next, time_ago +from devplacepy_services.web.ingress import mount_ingress + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", +) +logger = logging.getLogger(__name__) + +_rate_limit_store = defaultdict(list) +RATE_LIMIT = int(os.environ.get("DEVPLACE_RATE_LIMIT", "60")) +RATE_WINDOW = 60 +WEB_WORKERS = max(1, int(os.environ.get("DEVPLACE_WEB_WORKERS", "1"))) +RATE_LIMIT_DISABLED = os.environ.get("DEVPLACE_DISABLE_RATE_LIMIT") == "1" + +HOT_SETTINGS_TTL = 2.0 +_hot_settings_value: dict = {} +_hot_settings_at = 0.0 +_last_rate_sweep = 0.0 +RATE_SWEEP_INTERVAL = 60.0 + + +def _hot_settings() -> dict: + global _hot_settings_value, _hot_settings_at + now = time.monotonic() + if not _hot_settings_value or now - _hot_settings_at >= HOT_SETTINGS_TTL: + _hot_settings_value = { + "maintenance_mode": get_setting("maintenance_mode", "0"), + "rate_limit_per_minute": max( + 1, get_int_setting("rate_limit_per_minute", RATE_LIMIT) + ), + "rate_limit_window_seconds": max( + 1, get_int_setting("rate_limit_window_seconds", RATE_WINDOW) + ), + } + _hot_settings_at = now + return _hot_settings_value + + +def _sweep_rate_limit_store(window_start: float) -> None: + stale = [ + ip + for ip, timestamps in _rate_limit_store.items() + if not timestamps or timestamps[-1] <= window_start + ] + for ip in stale: + del _rate_limit_store[ip] + + +def _worker_rate_limit(limit: int) -> int: + return max(1, -(-limit // WEB_WORKERS)) + + +INLINE_MEDIA_EXTENSIONS = { + ".jpg", + ".jpeg", + ".png", + ".gif", + ".webp", + ".bmp", + ".tiff", + ".mp4", + ".webm", + ".ogv", + ".mov", + ".m4v", + ".mp3", +} + + +class UploadStaticFiles(StaticFiles): + async def get_response(self, path, scope): + response = await super().get_response(path, scope) + disposition = ( + "inline" + if Path(path).suffix.lower() in INLINE_MEDIA_EXTENSIONS + else "attachment" + ) + response.headers["Content-Disposition"] = disposition + response.headers["Cache-Control"] = "public, max-age=604800" + return response + + +class CachedStaticFiles(StaticFiles): + async def get_response(self, path, scope): + response = await super().get_response(path, scope) + if Path(path).name == "service-worker.js": + response.headers["Cache-Control"] = "no-cache" + else: + response.headers["Cache-Control"] = "public, max-age=31536000, immutable" + return response + + +class FallbackStaticFiles(StaticFiles): + async def get_response(self, path, scope): + response = await super().get_response(path, scope) + if Path(path).name == "service-worker.js": + response.headers["Cache-Control"] = "no-cache" + else: + response.headers["Cache-Control"] = "public, max-age=3600" + return response + + +@asynccontextmanager +async def lifespan(app: FastAPI): + ensure_data_dirs() + from devplacepy.services.statistics.tracking import start_visit_flusher + + start_visit_flusher() + + from devplacepy.config import SERVICE_LOCK_FILE + from devplacepy.services.dbapi.service import DbApiJobService + from devplacepy.services.manager import service_manager + from devplacepy.services.weblock import acquire_web_lock + + web_lock_owner = acquire_web_lock(SERVICE_LOCK_FILE) + if web_lock_owner: + service_manager.register(DbApiJobService()) + service_manager.set_lock_owner(True) + service_manager.supervise() + + logger.info(f"DevPlace web started on port {PORT}") + yield + logger.info("Shutting down web...") + if web_lock_owner: + await service_manager.shutdown_all() + from devplacepy.services.statistics.tracking import flush_visits + + flush_visits() + + +_AUTH_FORM_PAGES = { + "/auth/signup": ("signup.html", "Join DevPlace"), + "/auth/login": ("login.html", "Sign In"), + "/auth/forgot-password": ("forgot_password.html", "Reset Password"), +} + +_FRIENDLY_ERRORS = { + ("username", "too_short"): "Username must be between 3 and 32 characters", + ("username", "too_long"): "Username must be between 3 and 32 characters", + ("password", "too_short"): "Password must be at least 6 characters", +} + + +def _friendly_error(err): + field = err["loc"][-1] if err.get("loc") else "" + key = (field, err.get("type", "").replace("string_", "")) + if key in _FRIENDLY_ERRORS: + return _FRIENDLY_ERRORS[key] + msg = err.get("msg", "Invalid input") + prefix = "Value error, " + return msg[len(prefix) :] if msg.startswith(prefix) else msg + + +_home_cache = TTLCache(ttl=int(os.environ.get("DEVPLACE_HOME_CACHE_TTL", "60")), max_size=4) + + +def _landing_news(): + cached = _home_cache.get("news") + if cached is not None: + return cached + articles = [] + if "news" in db.tables: + news_table = get_table("news") + raw = list( + news_table.find(show_on_landing=1, order_by=["-synced_at"], _limit=6) + ) + images_by_news = get_news_images_by_uids([a["uid"] for a in raw]) + for a in raw: + articles.append( + { + "uid": a["uid"], + "slug": a.get("slug", ""), + "title": a.get("title", ""), + "description": (a.get("description", "") or "")[:250], + "url": a.get("url", ""), + "source_name": a.get("source_name", ""), + "grade": a.get("grade", 0), + "featured": a.get("featured", 0), + "synced_at": a.get("synced_at", "") or "", + "time_ago": time_ago(a["synced_at"]) if a.get("synced_at") else "", + "image_url": a.get("image_url", "") or images_by_news.get(a["uid"], ""), + } + ) + _home_cache.set("news", articles) + return articles + + +def _landing_recent_posts(blocked): + if not blocked: + cached = _home_cache.get("posts") + if cached is not None: + return cached + posts = [] + if "posts" in db.tables: + posts_table = get_table("posts") + fetch_limit = 24 if blocked else 6 + raw_posts = list( + posts_table.find(deleted_at=None, order_by=["-created_at"], _limit=fetch_limit) + ) + if blocked: + raw_posts = [p for p in raw_posts if p["user_uid"] not in blocked] + raw_posts = raw_posts[:6] + raw_posts = interleave_by_author(raw_posts) + if raw_posts: + post_uids = [p["uid"] for p in raw_posts] + author_uids = [p["user_uid"] for p in raw_posts] + authors = get_users_by_uids(author_uids) + comment_counts = get_comment_counts_by_post_uids(post_uids) + upvotes, downvotes = get_vote_counts(post_uids) + for p in raw_posts: + posts.append( + { + "post": p, + "author": authors.get(p["user_uid"]), + "time_ago": time_ago(p["created_at"]), + "comment_count": comment_counts.get(p["uid"], 0), + "stars": upvotes.get(p["uid"], 0) - downvotes.get(p["uid"], 0), + "slug": p.get("slug", "") or p["uid"], + } + ) + if not blocked: + _home_cache.set("posts", posts) + return posts + + +def create_web_app() -> FastAPI: + from devplacepy_services.base.middleware import ( + InternalCallCounterMiddleware, + RequestStatsMiddleware, + SanitizeErrorsMiddleware, + ) + + app = FastAPI( + title="DevPlace", + docs_url="/swagger", + redoc_url=None, + openapi_url="/openapi.json", + lifespan=lifespan, + ) + app.add_middleware(SanitizeErrorsMiddleware) + app.add_middleware(InternalCallCounterMiddleware) + app.add_middleware(RequestStatsMiddleware) + app.mount( + "/static/uploads", + UploadStaticFiles(directory=str(UPLOADS_DIR), check_dir=False), + name="uploads", + ) + app.mount( + f"/static/v{STATIC_VERSION}", + CachedStaticFiles(directory=str(STATIC_DIR)), + name="static_versioned", + ) + app.mount("/static", FallbackStaticFiles(directory=str(STATIC_DIR)), name="static") + + @app.exception_handler(404) + async def not_found(request: Request, exc): + if wants_json(request): + return json_error(404, "Not found") + seo_ctx = base_seo_context( + request, + title="Not Found - DevPlace", + description="The page you requested does not exist.", + robots="noindex", + ) + return templates.TemplateResponse( + request, + "error.html", + { + **seo_ctx, + "request": request, + "error_code": 404, + "error_message": "Page not found", + }, + status_code=404, + ) + + @app.exception_handler(500) + async def server_error(request: Request, exc): + logger.exception("500 error on %s %s", request.method, request.url.path) + if wants_json(request): + return json_error(500, "Internal server error") + seo_ctx = base_seo_context( + request, + title="Server Error - DevPlace", + description="Something went wrong.", + robots="noindex", + ) + return templates.TemplateResponse( + request, + "error.html", + { + **seo_ctx, + "request": request, + "error_code": 500, + "error_message": "Internal server error", + }, + status_code=500, + ) + + @app.exception_handler(RequestValidationError) + async def on_validation_error(request: Request, exc: RequestValidationError): + errors = [_friendly_error(e) for e in exc.errors()] + if wants_json(request): + fields: dict = {} + for raw, friendly in zip(exc.errors(), errors): + name = raw["loc"][-1] if raw.get("loc") else "_" + fields.setdefault(str(name), []).append(friendly) + return JSONResponse( + ValidationErrorOut(fields=fields, messages=errors).model_dump(mode="json"), + status_code=422, + ) + path = request.url.path + page = _AUTH_FORM_PAGES.get(path) + if page is None and path.startswith("/auth/reset-password/"): + page = ("reset_password.html", "Set New Password") + if page: + template_name, title = page + context = { + **base_seo_context(request, title=title, robots="noindex,nofollow"), + "request": request, + "errors": errors, + } + try: + form = await request.form() + context.update({k: v for k, v in form.items() if isinstance(v, str)}) + except Exception: + pass + if "token" in request.path_params: + context["token"] = request.path_params["token"] + return templates.TemplateResponse( + request, template_name, context, status_code=400 + ) + referer = safe_next(request.headers.get("referer"), "/feed") + return RedirectResponse(url=referer, status_code=303) + + mount_ingress(app) + + app.include_router(auth.router, prefix="/auth") + app.include_router(feed.router, prefix="/feed") + app.include_router(posts.router, prefix="/posts") + app.include_router(comments.router, prefix="/comments") + app.include_router(projects.router, prefix="/projects") + app.include_router(profile.router, prefix="/profile") + app.include_router(messages.router, prefix="/messages") + app.include_router(notifications.router, prefix="/notifications") + app.include_router(votes.router, prefix="/votes") + app.include_router(reactions.router, prefix="/reactions") + app.include_router(bookmarks.router, prefix="/bookmarks") + app.include_router(polls.router, prefix="/polls") + app.include_router(avatar.router, prefix="/avatar") + app.include_router(awards.router, prefix="/awards") + app.include_router(follow.router, prefix="/follow") + app.include_router(relations.router) + app.include_router(leaderboard.router, prefix="/leaderboard") + app.include_router(admin.router, prefix="/admin") + app.include_router(seo.router) + app.include_router(push.router) + app.include_router(docs.router) + app.include_router(issues.router, prefix="/issues") + app.include_router(gists.router, prefix="/gists") + app.include_router(news.router, prefix="/news") + app.include_router(uploads.router, prefix="/uploads") + app.include_router(media.router, prefix="/media") + app.include_router(proxy.router, prefix="/p") + app.include_router(devrant.router, prefix="/api") + app.include_router(dbapi.router, prefix="/dbapi") + app.include_router(game.router, prefix="/game") + + @app.middleware("http") + async def await_pending_corrections(request: Request, call_next): + response = await call_next(request) + pending = request.scope.get(PENDING_SCOPE_KEY) + if pending: + await asyncio.gather(*pending, return_exceptions=True) + return response + + @app.middleware("http") + async def add_security_headers(request: Request, call_next): + response = await call_next(request) + if not response.headers.get("X-Robots-Tag"): + response.headers["X-Robots-Tag"] = "index, follow" + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" + response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" + if not request.url.path.startswith("/p/"): + response.headers["X-Frame-Options"] = "DENY" + response.headers["Content-Security-Policy"] = ( + "object-src 'none'; base-uri 'self'; " + "frame-ancestors 'none'; form-action 'self'" + ) + if request.url.path.startswith("/admin"): + response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" + response.headers["Pragma"] = "no-cache" + response.headers["Expires"] = "0" + return response + + @app.middleware("http") + async def rate_limit_middleware(request: Request, call_next): + if RATE_LIMIT_DISABLED: + return await call_next(request) + if request.method in ( + "POST", + "PUT", + "DELETE", + "PATCH", + ) and not request.url.path.startswith(("/openai", "/xmlrpc")): + settings = _hot_settings() + limit = _worker_rate_limit(settings["rate_limit_per_minute"]) + window = settings["rate_limit_window_seconds"] + ip = client_ip(request, default="unknown") + now = time.time() + window_start = now - window + global _last_rate_sweep + if now - _last_rate_sweep >= RATE_SWEEP_INTERVAL: + _sweep_rate_limit_store(window_start) + _last_rate_sweep = now + timestamps = [t for t in _rate_limit_store.get(ip, ()) if t > window_start] + if len(timestamps) >= limit: + _rate_limit_store[ip] = timestamps + audit.record( + request, + "security.rate_limit.block", + result="denied", + summary=f"request from {ip} blocked by rate limit", + metadata={"ip": ip, "limit": limit, "window_seconds": window}, + ) + retry_after = {"Retry-After": str(window)} + if wants_json(request): + response = json_error(429, "Rate limit exceeded. Try again later.") + response.headers["Retry-After"] = str(window) + return response + return HTMLResponse( + "Rate limit exceeded. Try again later.", + status_code=429, + headers=retry_after, + ) + timestamps.append(now) + _rate_limit_store[ip] = timestamps + return await call_next(request) + + _MAINTENANCE_ALLOWED_PREFIXES = ("/static", "/avatar", "/auth", "/admin", "/openai") + + @app.middleware("http") + async def maintenance_middleware(request: Request, call_next): + if _hot_settings()["maintenance_mode"] != "1": + return await call_next(request) + if request.url.path.startswith(_MAINTENANCE_ALLOWED_PREFIXES): + return await call_next(request) + user = get_current_user(request) + if user and user.get("role") == "Admin": + return await call_next(request) + message = get_setting( + "maintenance_message", + "DevPlace is undergoing scheduled maintenance. Please check back shortly.", + ) + audit.record( + request, + "security.maintenance.block", + user=user, + result="denied", + summary="non-admin request blocked by maintenance mode", + ) + if wants_json(request): + return json_error(503, message) + seo_ctx = base_seo_context( + request, title="Maintenance - DevPlace", description=message, robots="noindex" + ) + return templates.TemplateResponse( + request, + "error.html", + {**seo_ctx, "request": request, "error_code": 503, "error_message": message}, + status_code=503, + ) + + @app.middleware("http") + async def track_presence(request: Request, call_next): + path = request.url.path + if not path.startswith(("/static", "/avatar")): + user = get_current_user(request) + if user: + presence.touch(user["uid"]) + return await call_next(request) + + @app.middleware("http") + async def visit_statistics(request: Request, call_next): + from devplacepy.services.statistics.tracking import track_visit + + response = await call_next(request) + track_visit(request, response.status_code) + return response + + @app.middleware("http") + async def response_timing(request: Request, call_next): + start = time.perf_counter() + request.state.request_start = start + response = await call_next(request) + response.headers["X-Response-Time"] = f"{(time.perf_counter() - start) * 1000:.1f}ms" + return response + + app.add_middleware(GZipMiddleware, minimum_size=512, compresslevel=6) + + @app.get("/") + async def landing(request: Request): + user = get_current_user(request) + + landing_articles = _landing_news() + blocked = get_blocked_uids(user["uid"]) if user else frozenset() + landing_posts = _landing_recent_posts(blocked) + + base = site_url(request) + seo_ctx = base_seo_context( + request, + title="DevPlace - The Developer Social Network", + description="Track industry shifts. Discover bold releases. Share what you're building in an open, uncensored environment.", + breadcrumbs=[], + schemas=[website_schema(base)], + ) + return respond( + request, + "landing.html", + { + **seo_ctx, + "request": request, + "user": user, + "is_authenticated": bool(user), + "user_post_count": get_user_post_count(user["uid"]) if user else 0, + "user_stars": get_user_stars(user["uid"]) if user else 0, + "landing_articles": landing_articles, + "landing_posts": landing_posts, + }, + model=LandingOut, + ) + + return app + + +app = create_web_app() \ No newline at end of file diff --git a/devplacepy_services/web/ingress.py b/devplacepy_services/web/ingress.py new file mode 100644 index 00000000..d93636c8 --- /dev/null +++ b/devplacepy_services/web/ingress.py @@ -0,0 +1,216 @@ +import asyncio +import logging +from urllib.parse import urlparse + +import httpx +import uuid_utils +import websockets +from starlette.datastructures import Headers +from starlette.requests import Request +from starlette.responses import Response +from starlette.routing import Route +from starlette.websockets import WebSocket + +from devplacepy_services.base.config import service_url +from devplacepy_services.base.errors import error_response +from devplacepy_services.base.manifest import INGRESS_ROUTES as _INGRESS_SPECS +from devplacepy_services.base.proxy import HOP_HEADERS + +logger = logging.getLogger(__name__) + +INGRESS_ROUTES = [(route.prefix, route.service) for route in _INGRESS_SPECS] + +TIMEOUTS = { + "xmlrpc": 120.0, +} + + +def _forward_headers_from_scope(scope) -> dict[str, str]: + original = Headers(scope=scope) + headers = { + k: v for k, v in original.items() if k.lower() not in HOP_HEADERS + } + headers["X-Request-Id"] = headers.get("X-Request-Id") or uuid_utils.uuid7().hex + headers["Accept-Encoding"] = "identity" + # Mirror nginx's `proxy_set_header Host $host` (Appendix F) so an + # upstream-generated absolute URL (redirect, url_for) reflects the + # public :10500 endpoint the browser is actually talking to, not the + # upstream service's own internal bind address/port. + original_host = original.get("host") + if original_host: + headers["Host"] = original_host + return headers + + +def _forward_headers(request: Request) -> dict[str, str]: + return _forward_headers_from_scope(request.scope) + + +_WS_HANDSHAKE_HEADERS = frozenset( + { + "sec-websocket-key", + "sec-websocket-version", + "sec-websocket-extensions", + "sec-websocket-protocol", + "sec-websocket-accept", + } +) + + +def _forward_ws_headers(scope) -> dict[str, str]: + headers = _forward_headers_from_scope(scope) + for key in list(headers): + if key.lower() in _WS_HANDSHAKE_HEADERS: + del headers[key] + return headers + + +def _target_path(scope, prefix: str) -> str: + # Starlette's Mount rewrites scope["root_path"] to the cumulative mount + # prefix but leaves scope["path"] as the FULL original request path (it + # does not strip the prefix) - so the full path alone is already the + # correct upstream path; concatenating root_path in front double-prefixes it. + return scope.get("path", "") or prefix + + +def _upstream_http_url(service: str, path: str, query: str) -> str: + base = service_url(service).rstrip("/") + url = f"{base}{path}" + if query: + url = f"{url}?{query}" + return url + + +def _upstream_ws_url(service: str, path: str, query: str) -> str: + parsed = urlparse(service_url(service)) + scheme = "wss" if parsed.scheme == "https" else "ws" + netloc = parsed.netloc + url = f"{scheme}://{netloc}{path}" + if query: + url = f"{url}?{query}" + return url + + +class IngressProxy: + def __init__(self, prefix: str, service: str) -> None: + self.prefix = prefix + self.service = service + self.timeout = TIMEOUTS.get(service, 30.0) + + async def __call__(self, scope, receive, send) -> None: + if scope["type"] == "http": + await self._proxy_http(scope, receive, send) + elif scope["type"] == "websocket": + await self._proxy_ws(scope, receive, send) + + async def _proxy_http(self, scope, receive, send) -> None: + request = Request(scope, receive) + path = _target_path(scope, self.prefix) + query = scope.get("query_string", b"").decode() + url = _upstream_http_url(self.service, path, query) + body = await request.body() + headers = _forward_headers(request) + try: + async with httpx.AsyncClient( + timeout=self.timeout, follow_redirects=False + ) as client: + upstream = await client.request( + request.method, + url, + headers=headers, + content=body, + ) + except httpx.HTTPError as exc: + logger.warning("ingress %s upstream error: %s", self.prefix, exc) + response = error_response( + 502, "Upstream service unavailable", "upstream_error" + ) + await response(scope, receive, send) + return + out_headers = { + k: v + for k, v in upstream.headers.items() + if k.lower() not in HOP_HEADERS and k.lower() != "set-cookie" + } + response = Response( + content=upstream.content, + status_code=upstream.status_code, + headers=out_headers, + media_type=upstream.headers.get("content-type"), + ) + for cookie in upstream.headers.get_list("set-cookie"): + response.headers.append("set-cookie", cookie) + await response(scope, receive, send) + + async def _proxy_ws(self, scope, receive, send) -> None: + client_ws = WebSocket(scope, receive, send) + path = _target_path(scope, self.prefix) + query = scope.get("query_string", b"").decode() + upstream_url = _upstream_ws_url(self.service, path, query) + headers = _forward_ws_headers(scope) + await client_ws.accept() + try: + async with websockets.connect( + upstream_url, + open_timeout=10, + max_size=None, + additional_headers=headers, + ) as upstream: + await _pump(client_ws, upstream) + except Exception as exc: + logger.debug("ingress ws %s failed: %s", self.prefix, exc) + try: + await client_ws.close(code=1011) + except Exception: + pass + + +async def _pump(client_ws: WebSocket, upstream) -> None: + async def client_to_upstream(): + try: + while True: + message = await client_ws.receive() + if message["type"] == "websocket.disconnect": + break + if message.get("text") is not None: + await upstream.send(message["text"]) + elif message.get("bytes") is not None: + await upstream.send(message["bytes"]) + except Exception: + pass + finally: + await upstream.close() + + async def upstream_to_client(): + try: + async for message in upstream: + if isinstance(message, (bytes, bytearray)): + await client_ws.send_bytes(bytes(message)) + else: + await client_ws.send_text(message) + except Exception: + pass + finally: + try: + await client_ws.close() + except Exception: + pass + + await asyncio.gather(client_to_upstream(), upstream_to_client()) + + +def mount_ingress(app) -> None: + for prefix, service in INGRESS_ROUTES: + proxy = IngressProxy(prefix, service) + # A bare hit on the prefix itself (no trailing slash, nothing after - + # e.g. a POST to /xmlrpc or GET /tools) never matches Mount's own + # path regex (it requires a "/" plus content after the prefix), so + # Starlette's router-level redirect_slashes fallback 307s it to + # "/" before the Mount ever sees it. If the upstream service's + # own router registers an exact route at its mount root (as /tools + # and /xmlrpc both do), THAT redirects back to the bare prefix - + # an infinite loop between the two opposite trailing-slash + # conventions. Registering an explicit Route at the exact prefix + # bypasses Mount's regex/redirect fallback entirely for that one path. + app.router.routes.append(Route(prefix, endpoint=proxy, methods=None)) + app.mount(prefix, proxy) \ No newline at end of file diff --git a/devplacepy_services/web/main.py b/devplacepy_services/web/main.py new file mode 100644 index 00000000..4e89c9b2 --- /dev/null +++ b/devplacepy_services/web/main.py @@ -0,0 +1,21 @@ +from devplacepy_services.base.health import health_router +from devplacepy_services.base.service import BaseMicroservice +from devplacepy_services.web.factory import create_web_app + + +class WebService(BaseMicroservice): + name = "web" + title = "Web" + default_port = 10500 + workers = "auto" + stateful = False + depends_on = ["database", "pubsub"] + + def build_app(self): + app = create_web_app() + app.include_router(health_router(self)) + return app + + +_service = WebService() +app = _service.build_app() diff --git a/devplacepy_services/xmlrpc/SERVICE.md b/devplacepy_services/xmlrpc/SERVICE.md new file mode 100644 index 00000000..14520ff5 --- /dev/null +++ b/devplacepy_services/xmlrpc/SERVICE.md @@ -0,0 +1 @@ +# XML-RPC Service (stub) \ No newline at end of file diff --git a/devplacepy_services/xmlrpc/__init__.py b/devplacepy_services/xmlrpc/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/devplacepy_services/xmlrpc/main.py b/devplacepy_services/xmlrpc/main.py new file mode 100644 index 00000000..2f102cc5 --- /dev/null +++ b/devplacepy_services/xmlrpc/main.py @@ -0,0 +1,24 @@ +import devplacepy_services.base.bootstrap + +from devplacepy.routers import xmlrpc +from devplacepy.services.xmlrpc import XmlrpcService +from devplacepy_services.base.service import BaseMicroservice, build_standard_app + + +class XmlrpcMicroservice(BaseMicroservice): + name = "xmlrpc" + title = "XML-RPC" + default_port = 10649 + workers = 1 + stateful = True + depends_on = ["database"] + managed_services = [XmlrpcService()] + use_background = True + run_supervisor = True + + def build_app(self): + return build_standard_app(self, routers=[(xmlrpc.router, "/xmlrpc")]) + + +_service = XmlrpcMicroservice() +app = _service.build_app() \ No newline at end of file