diff --git a/alembic/versions/20260610_1200_add_node_is_observer.py b/alembic/versions/20260610_1200_add_node_is_observer.py new file mode 100644 index 0000000..cc1aa63 --- /dev/null +++ b/alembic/versions/20260610_1200_add_node_is_observer.py @@ -0,0 +1,52 @@ +"""add is_observer flag to nodes + +Revision ID: 20260610_1200 +Revises: 20260604_1200 +Create Date: 2026-06-10 12:00:00.000000+00:00 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = "20260610_1200" +down_revision: Union[str, None] = "20260604_1200" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +# Backfill the flag from every event source, matching the union the API used to +# evaluate at query time. Portable across SQLite and PostgreSQL. +_BACKFILL_SQL = """ +UPDATE nodes SET is_observer = true WHERE id IN ( + SELECT observer_node_id FROM advertisements WHERE observer_node_id IS NOT NULL + UNION SELECT observer_node_id FROM messages WHERE observer_node_id IS NOT NULL + UNION SELECT observer_node_id FROM telemetry WHERE observer_node_id IS NOT NULL + UNION SELECT observer_node_id FROM trace_paths WHERE observer_node_id IS NOT NULL + UNION SELECT observer_node_id FROM event_observers +) +""" + + +def upgrade() -> None: + with op.batch_alter_table("nodes", schema=None) as batch_op: + batch_op.add_column( + sa.Column( + "is_observer", + sa.Boolean(), + nullable=False, + server_default=sa.false(), + ) + ) + batch_op.create_index("ix_nodes_is_observer", ["is_observer"]) + + op.execute(_BACKFILL_SQL) + + +def downgrade() -> None: + with op.batch_alter_table("nodes", schema=None) as batch_op: + batch_op.drop_index("ix_nodes_is_observer") + batch_op.drop_column("is_observer") diff --git a/src/meshcore_hub/api/app.py b/src/meshcore_hub/api/app.py index 997aca1..e2c02b8 100644 --- a/src/meshcore_hub/api/app.py +++ b/src/meshcore_hub/api/app.py @@ -186,12 +186,12 @@ def create_app( # Health check endpoints @app.get("/health", tags=["Health"]) - async def health() -> dict: + def health() -> dict: """Basic health check.""" return {"status": "healthy", "version": __version__} @app.get("/health/ready", tags=["Health"]) - async def health_ready() -> dict: + def health_ready() -> dict: """Readiness check including database and optional Redis.""" try: db = get_db_manager() diff --git a/src/meshcore_hub/api/cache.py b/src/meshcore_hub/api/cache.py index d74ca54..285bf2a 100644 --- a/src/meshcore_hub/api/cache.py +++ b/src/meshcore_hub/api/cache.py @@ -1,6 +1,7 @@ """Cache decorator for API endpoints.""" import functools +import inspect import json import logging from typing import Any, Callable, Optional @@ -28,6 +29,53 @@ def _find_request(kwargs: dict[str, Any]) -> Request: raise TypeError("No Request parameter found in handler arguments") +# Sentinel distinguishing "cache miss" from "cache hit holding a JSON null". +_MISS = object() + + +def _build_cache_key( + request: Request, + endpoint_name: str, + key_builder: Optional[Callable[[Request], str]], +) -> str: + """Build the cache key for a request.""" + if key_builder is not None: + return key_builder(request) + return f"{endpoint_name}:{sorted_query_string(request)}" + + +def _lookup(cache: Any, cache_key: str, request: Request) -> Any: + """Return the cached value, or _MISS, and record the cache status.""" + try: + cached_value = cache.get(cache_key) + except Exception as e: + logger.warning("Redis GET error for %s: %s", cache_key, e) + cached_value = None + + if cached_value is not None: + logger.debug("Cache HIT: %s", cache_key) + request.state.cache_status = "HIT" + return json.loads(cached_value) + + logger.debug("Cache MISS: %s", cache_key) + request.state.cache_status = "MISS" + return _MISS + + +def _store(cache: Any, cache_key: str, result: Any, ttl: int) -> None: + """Serialize and store a handler result in the cache.""" + try: + if hasattr(result, "model_dump"): + serialized = json.dumps(result.model_dump(mode="json")) + elif isinstance(result, dict): + serialized = json.dumps(result) + else: + serialized = json.dumps(result, default=str) + cache.set(cache_key, serialized, ttl) + except Exception as e: + logger.warning("Cache store error for %s: %s", cache_key, e) + + def cached( endpoint_name: str, ttl_setting: str = "redis_cache_ttl", @@ -43,49 +91,49 @@ def cached( """ def decorator(func: Callable[..., Any]) -> Callable[..., Any]: + # Async handlers keep an async wrapper (runs on the event loop); sync + # handlers get a sync wrapper so FastAPI runs them in its threadpool, + # keeping blocking DB/Redis calls off the event loop. + if inspect.iscoroutinefunction(func): + + @functools.wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + request = _find_request(kwargs) + cache = getattr(request.app.state, "redis_cache", None) + if cache is None: + return await func(*args, **kwargs) + + ttl = getattr(request.app.state, ttl_setting, 30) + cache_key = _build_cache_key(request, endpoint_name, key_builder) + + cached = _lookup(cache, cache_key, request) + if cached is not _MISS: + return cached + + result = await func(*args, **kwargs) + _store(cache, cache_key, result, ttl) + return result + + return async_wrapper + @functools.wraps(func) - async def wrapper(*args: Any, **kwargs: Any) -> Any: + def sync_wrapper(*args: Any, **kwargs: Any) -> Any: request = _find_request(kwargs) cache = getattr(request.app.state, "redis_cache", None) if cache is None: - return await func(*args, **kwargs) + return func(*args, **kwargs) ttl = getattr(request.app.state, ttl_setting, 30) + cache_key = _build_cache_key(request, endpoint_name, key_builder) - if key_builder is not None: - cache_key = key_builder(request) - else: - cache_key = f"{endpoint_name}:{sorted_query_string(request)}" - - try: - cached_value = cache.get(cache_key) - except Exception as e: - logger.warning("Redis GET error for %s: %s", cache_key, e) - cached_value = None - - if cached_value is not None: - logger.debug("Cache HIT: %s", cache_key) - request.state.cache_status = "HIT" - return json.loads(cached_value) - - logger.debug("Cache MISS: %s", cache_key) - request.state.cache_status = "MISS" - - result = await func(*args, **kwargs) - - try: - if hasattr(result, "model_dump"): - serialized = json.dumps(result.model_dump(mode="json")) - elif isinstance(result, dict): - serialized = json.dumps(result) - else: - serialized = json.dumps(result, default=str) - cache.set(cache_key, serialized, ttl) - except Exception as e: - logger.warning("Cache store error for %s: %s", cache_key, e) + cached = _lookup(cache, cache_key, request) + if cached is not _MISS: + return cached + result = func(*args, **kwargs) + _store(cache, cache_key, result, ttl) return result - return wrapper + return sync_wrapper return decorator diff --git a/src/meshcore_hub/api/routes/adoptions.py b/src/meshcore_hub/api/routes/adoptions.py index 65336f3..aa7393f 100644 --- a/src/meshcore_hub/api/routes/adoptions.py +++ b/src/meshcore_hub/api/routes/adoptions.py @@ -18,7 +18,7 @@ router = APIRouter() @router.post("", response_model=AdoptedNodeRead, status_code=201) -async def adopt_node( +def adopt_node( adopt_request: NodeAdoptRequest, caller_info: RequireOperatorOrAdmin, session: DbSession, @@ -74,7 +74,7 @@ async def adopt_node( @router.delete("/{public_key}", status_code=204) -async def release_node( +def release_node( public_key: str, caller_info: RequireOperatorOrAdmin, request: Request, diff --git a/src/meshcore_hub/api/routes/advertisements.py b/src/meshcore_hub/api/routes/advertisements.py index ccaf1d7..aa78183 100644 --- a/src/meshcore_hub/api/routes/advertisements.py +++ b/src/meshcore_hub/api/routes/advertisements.py @@ -46,7 +46,7 @@ def _get_tag_description(node: Optional[Node]) -> Optional[str]: @router.get("", response_model=AdvertisementList) @cached("advertisements") -async def list_advertisements( +def list_advertisements( _: RequireRead, session: DbSession, request: Request, @@ -243,7 +243,7 @@ async def list_advertisements( @router.get("/{advertisement_id}", response_model=AdvertisementRead) -async def get_advertisement( +def get_advertisement( _: RequireRead, session: DbSession, advertisement_id: str, diff --git a/src/meshcore_hub/api/routes/channels.py b/src/meshcore_hub/api/routes/channels.py index ebfffb3..e10265d 100644 --- a/src/meshcore_hub/api/routes/channels.py +++ b/src/meshcore_hub/api/routes/channels.py @@ -45,7 +45,7 @@ def _channel_to_read(channel: Channel, include_key: bool = False) -> ChannelRead @router.get("", response_model=ChannelList) @cached("channels", key_builder=_channels_key_builder) -async def list_channels( +def list_channels( _: RequireRead, session: DbSession, request: Request, @@ -71,7 +71,7 @@ async def list_channels( @router.post("", response_model=ChannelRead, status_code=201) -async def create_channel( +def create_channel( __: RequireAdmin, session: DbSession, body: ChannelCreate, @@ -110,7 +110,7 @@ async def create_channel( @router.put("/{channel_id}", response_model=ChannelRead) -async def update_channel( +def update_channel( __: RequireAdmin, session: DbSession, channel_id: str, @@ -149,7 +149,7 @@ async def update_channel( @router.delete("/{channel_id}", status_code=204) -async def delete_channel( +def delete_channel( __: RequireAdmin, session: DbSession, channel_id: str, diff --git a/src/meshcore_hub/api/routes/dashboard.py b/src/meshcore_hub/api/routes/dashboard.py index 8694824..d0c88cb 100644 --- a/src/meshcore_hub/api/routes/dashboard.py +++ b/src/meshcore_hub/api/routes/dashboard.py @@ -65,7 +65,7 @@ def _flood_only_filter( ttl_setting="redis_cache_ttl_dashboard", key_builder=_dashboard_stats_key_builder, ) -async def get_stats( +def get_stats( _: RequireRead, session: DbSession, request: Request, @@ -323,7 +323,7 @@ async def get_stats( @router.get("/activity", response_model=DailyActivity) @cached("dashboard/activity", ttl_setting="redis_cache_ttl_dashboard") -async def get_activity( +def get_activity( _: RequireRead, session: DbSession, request: Request, @@ -383,7 +383,7 @@ async def get_activity( ttl_setting="redis_cache_ttl_dashboard", key_builder=_dashboard_msg_activity_key_builder, ) -async def get_message_activity( +def get_message_activity( _: RequireRead, session: DbSession, request: Request, @@ -443,7 +443,7 @@ async def get_message_activity( @router.get("/node-count", response_model=NodeCountHistory) @cached("dashboard/node-count", ttl_setting="redis_cache_ttl_dashboard") -async def get_node_count_history( +def get_node_count_history( _: RequireRead, session: DbSession, request: Request, diff --git a/src/meshcore_hub/api/routes/messages.py b/src/meshcore_hub/api/routes/messages.py index 726b46c..2b25b48 100644 --- a/src/meshcore_hub/api/routes/messages.py +++ b/src/meshcore_hub/api/routes/messages.py @@ -41,7 +41,7 @@ def _get_tag_name(node: Optional[Node]) -> Optional[str]: @router.get("", response_model=MessageList) @cached("messages", key_builder=_messages_key_builder) -async def list_messages( +def list_messages( _: RequireRead, session: DbSession, request: Request, @@ -233,7 +233,7 @@ async def list_messages( @router.get("/{message_id}", response_model=MessageRead) -async def get_message( +def get_message( _: RequireRead, session: DbSession, request: Request, diff --git a/src/meshcore_hub/api/routes/node_tags.py b/src/meshcore_hub/api/routes/node_tags.py index 797e50e..d80f406 100644 --- a/src/meshcore_hub/api/routes/node_tags.py +++ b/src/meshcore_hub/api/routes/node_tags.py @@ -52,7 +52,7 @@ def _check_tag_access( @router.get("/nodes/{public_key}/tags", response_model=list[NodeTagRead]) -async def list_node_tags( +def list_node_tags( _: RequireRead, session: DbSession, public_key: str, @@ -68,7 +68,7 @@ async def list_node_tags( @router.post("/nodes/{public_key}/tags", response_model=NodeTagRead, status_code=201) -async def create_node_tag( +def create_node_tag( caller_info: RequireOperatorOrAdmin, session: DbSession, request: Request, @@ -115,7 +115,7 @@ async def create_node_tag( @router.put("/nodes/{public_key}/tags/{key}", response_model=NodeTagRead) -async def update_node_tag( +def update_node_tag( caller_info: RequireOperatorOrAdmin, session: DbSession, request: Request, @@ -171,7 +171,7 @@ async def update_node_tag( @router.delete("/nodes/{public_key}/tags/{key}", status_code=204) -async def delete_node_tag( +def delete_node_tag( caller_info: RequireOperatorOrAdmin, session: DbSession, request: Request, diff --git a/src/meshcore_hub/api/routes/nodes.py b/src/meshcore_hub/api/routes/nodes.py index 81da398..27e0deb 100644 --- a/src/meshcore_hub/api/routes/nodes.py +++ b/src/meshcore_hub/api/routes/nodes.py @@ -10,13 +10,8 @@ from meshcore_hub.api.auth import RequireRead from meshcore_hub.api.cache import cached from meshcore_hub.api.dependencies import DbSession from meshcore_hub.common.models import ( - Advertisement, - EventObserver, - Message, Node, NodeTag, - Telemetry, - TracePath, UserProfileNode, ) from meshcore_hub.common.schemas.nodes import AdoptedByUser, NodeList, NodeRead @@ -49,7 +44,7 @@ VALID_NODE_SORT_COLUMNS = {"name", "public_key", "last_seen"} @router.get("", response_model=NodeList) @cached("nodes") -async def list_nodes( +def list_nodes( _: RequireRead, session: DbSession, request: Request, @@ -149,56 +144,9 @@ async def list_nodes( ) if observer is not None: - if observer: - query = query.where( - or_( - Node.id.in_( - select(Advertisement.observer_node_id).where( - Advertisement.observer_node_id.is_not(None) - ) - ), - Node.id.in_( - select(Message.observer_node_id).where( - Message.observer_node_id.is_not(None) - ) - ), - Node.id.in_( - select(Telemetry.observer_node_id).where( - Telemetry.observer_node_id.is_not(None) - ) - ), - Node.id.in_( - select(TracePath.observer_node_id).where( - TracePath.observer_node_id.is_not(None) - ) - ), - Node.id.in_(select(EventObserver.observer_node_id)), - ) - ) - else: - query = query.where( - ~Node.id.in_( - select(Advertisement.observer_node_id).where( - Advertisement.observer_node_id.is_not(None) - ) - ), - ~Node.id.in_( - select(Message.observer_node_id).where( - Message.observer_node_id.is_not(None) - ) - ), - ~Node.id.in_( - select(Telemetry.observer_node_id).where( - Telemetry.observer_node_id.is_not(None) - ) - ), - ~Node.id.in_( - select(TracePath.observer_node_id).where( - TracePath.observer_node_id.is_not(None) - ) - ), - ~Node.id.in_(select(EventObserver.observer_node_id)), - ) + # Uses the precomputed, indexed nodes.is_observer flag (maintained by the + # collector + cleanup job) instead of scanning the event tables. + query = query.where(Node.is_observer.is_(bool(observer))) # Get total count count_query = select(func.count()).select_from(query.subquery()) @@ -245,7 +193,7 @@ async def list_nodes( @router.get("/prefix/{prefix}", response_model=NodeRead) -async def get_node_by_prefix( +def get_node_by_prefix( _: RequireRead, session: DbSession, prefix: str = Path(description="Public key prefix to search for"), @@ -275,7 +223,7 @@ async def get_node_by_prefix( @router.get("/{public_key}", response_model=NodeRead) -async def get_node( +def get_node( _: RequireRead, session: DbSession, public_key: str = Path(description="Full 64-character public key"), diff --git a/src/meshcore_hub/api/routes/telemetry.py b/src/meshcore_hub/api/routes/telemetry.py index 002678e..b2db2cd 100644 --- a/src/meshcore_hub/api/routes/telemetry.py +++ b/src/meshcore_hub/api/routes/telemetry.py @@ -17,7 +17,7 @@ router = APIRouter() @router.get("", response_model=TelemetryList) -async def list_telemetry( +def list_telemetry( _: RequireRead, session: DbSession, node_public_key: Optional[str] = Query(None, description="Filter by node"), @@ -91,7 +91,7 @@ async def list_telemetry( @router.get("/{telemetry_id}", response_model=TelemetryRead) -async def get_telemetry( +def get_telemetry( _: RequireRead, session: DbSession, telemetry_id: str, diff --git a/src/meshcore_hub/api/routes/trace_paths.py b/src/meshcore_hub/api/routes/trace_paths.py index b592157..eac386a 100644 --- a/src/meshcore_hub/api/routes/trace_paths.py +++ b/src/meshcore_hub/api/routes/trace_paths.py @@ -17,7 +17,7 @@ router = APIRouter() @router.get("", response_model=TracePathList) -async def list_trace_paths( +def list_trace_paths( _: RequireRead, session: DbSession, observed_by: Optional[str] = Query( @@ -91,7 +91,7 @@ async def list_trace_paths( @router.get("/{trace_path_id}", response_model=TracePathRead) -async def get_trace_path( +def get_trace_path( _: RequireRead, session: DbSession, trace_path_id: str, diff --git a/src/meshcore_hub/api/routes/user_profiles.py b/src/meshcore_hub/api/routes/user_profiles.py index 7d313c0..1de01ab 100644 --- a/src/meshcore_hub/api/routes/user_profiles.py +++ b/src/meshcore_hub/api/routes/user_profiles.py @@ -53,7 +53,7 @@ def _build_adopted_nodes(profile: UserProfile) -> list[AdoptedNodeRead]: @router.get("/profiles", response_model=UserProfileList) @cached("profiles") -async def list_profiles( +def list_profiles( _: RequireRead, session: DbSession, request: Request, @@ -116,7 +116,7 @@ async def list_profiles( @router.get("/profile/me", response_model=UserProfileWithNodes) -async def get_my_profile( +def get_my_profile( request: Request, session: DbSession, ) -> UserProfileWithNodes: @@ -148,7 +148,7 @@ async def get_my_profile( @router.get("/profile/{profile_id}") -async def get_profile( +def get_profile( profile_id: str, request: Request, session: DbSession, @@ -201,7 +201,7 @@ async def get_profile( @router.put("/profile/{profile_id}", response_model=UserProfileRead) -async def update_profile( +def update_profile( profile_id: str, profile_update: UserProfileUpdate, caller_id: RequireUserOwner, diff --git a/src/meshcore_hub/collector/cleanup.py b/src/meshcore_hub/collector/cleanup.py index c0ffe73..fdcd05c 100644 --- a/src/meshcore_hub/collector/cleanup.py +++ b/src/meshcore_hub/collector/cleanup.py @@ -7,7 +7,7 @@ based on configured retention policies. import logging from datetime import datetime, timedelta, timezone -from sqlalchemy import delete, func, select +from sqlalchemy import CompoundSelect, delete, func, select, union, update from sqlalchemy.ext.asyncio import AsyncSession from meshcore_hub.common.models import ( @@ -35,6 +35,7 @@ class CleanupStats: self.trace_paths_deleted: int = 0 self.event_logs_deleted: int = 0 self.nodes_deleted: int = 0 + self.observers_cleared: int = 0 self.total_deleted: int = 0 def __repr__(self) -> str: @@ -45,7 +46,8 @@ class CleanupStats: f"telemetry={self.telemetry_deleted}, " f"trace_paths={self.trace_paths_deleted}, " f"event_logs={self.event_logs_deleted}, " - f"nodes={self.nodes_deleted})" + f"nodes={self.nodes_deleted}, " + f"observers_cleared={self.observers_cleared})" ) @@ -107,6 +109,9 @@ async def cleanup_old_data( + stats.event_logs_deleted ) + # Clear the is_observer flag for nodes whose events were all pruned above. + stats.observers_cleared = await recompute_observer_flags(db, dry_run) + if not dry_run: await db.commit() logger.info("Cleanup completed: %s", stats) @@ -116,6 +121,67 @@ async def cleanup_old_data( return stats +def _observer_node_id_union() -> CompoundSelect: + """Union of observer_node_id across every event source (excluding NULLs).""" + return union( + select(Advertisement.observer_node_id).where( + Advertisement.observer_node_id.is_not(None) + ), + select(Message.observer_node_id).where(Message.observer_node_id.is_not(None)), + select(Telemetry.observer_node_id).where( + Telemetry.observer_node_id.is_not(None) + ), + select(TracePath.observer_node_id).where( + TracePath.observer_node_id.is_not(None) + ), + select(EventObserver.observer_node_id), + ) + + +async def recompute_observer_flags(db: AsyncSession, dry_run: bool = False) -> int: + """Clear nodes.is_observer for nodes that no longer have any events. + + The collector sets the flag when a node observes an event; this clears it + again once retention has pruned all of a node's events, so stale observers + drop off the API's ``observer=true`` listing. + + Args: + db: Database session + dry_run: If True, only count nodes without modifying them + + Returns: + Number of nodes whose is_observer flag was (or would be) cleared. + """ + observer_union = _observer_node_id_union().subquery() + stale = Node.id.not_in(select(observer_union.c.observer_node_id)) + + if dry_run: + count_stmt = ( + select(func.count()) + .select_from(Node) + .where(Node.is_observer.is_(True), stale) + ) + result = await db.execute(count_stmt) + count = result.scalar() or 0 + else: + upd = ( + update(Node) + .where(Node.is_observer.is_(True), stale) + .values(is_observer=False) + ) + result = await db.execute(upd) + count = result.rowcount or 0 # type: ignore[attr-defined] + + if count > 0: + logger.info( + "is_observer: %s %d stale observer node(s)", + "would clear" if dry_run else "cleared", + count, + ) + + return count + + async def _cleanup_table( db: AsyncSession, model: type, diff --git a/src/meshcore_hub/common/database.py b/src/meshcore_hub/common/database.py index bb5cb73..4b0427b 100644 --- a/src/meshcore_hub/common/database.py +++ b/src/meshcore_hub/common/database.py @@ -25,24 +25,42 @@ def create_database_engine( SQLAlchemy Engine instance """ connect_args = {} + engine_kwargs: dict[str, Any] = {} # SQLite-specific configuration if database_url.startswith("sqlite"): connect_args["check_same_thread"] = False + # Size the pool above the default Starlette threadpool (~40 threads) so + # concurrent request handlers don't block waiting for a connection. Applies + # to file-based SQLite and networked backends (e.g. a future Postgres). + # In-memory SQLite uses a non-overflow pool, so skip these args there. + is_memory_sqlite = database_url in ("sqlite://", "sqlite:///:memory:") + if not is_memory_sqlite: + engine_kwargs["pool_size"] = 20 + engine_kwargs["max_overflow"] = 30 + engine = create_engine( database_url, echo=echo, connect_args=connect_args, pool_pre_ping=True, + **engine_kwargs, ) - # Enable foreign keys for SQLite + # Apply SQLite pragmas on every new connection if database_url.startswith("sqlite"): @event.listens_for(engine, "connect") def set_sqlite_pragma(dbapi_connection, connection_record): # type: ignore cursor = dbapi_connection.cursor() + # WAL lets readers run concurrently with a single writer (the + # collector), and busy_timeout waits instead of immediately raising + # "database is locked" under contention. synchronous=NORMAL is safe + # under WAL and faster. + cursor.execute("PRAGMA journal_mode=WAL") + cursor.execute("PRAGMA busy_timeout=5000") + cursor.execute("PRAGMA synchronous=NORMAL") cursor.execute("PRAGMA foreign_keys=ON") cursor.close() @@ -127,7 +145,8 @@ class DatabaseManager: async_url = self.database_url.replace("sqlite://", "sqlite+aiosqlite://") self._async_engine = create_async_engine(async_url, echo=self._echo) - # Enable foreign keys for async SQLite engine + # Apply the same SQLite pragmas as the sync engine (see + # create_database_engine) for the async engine's connections. if self.database_url.startswith("sqlite"): @event.listens_for(self._async_engine.sync_engine, "connect") @@ -135,6 +154,9 @@ class DatabaseManager: dbapi_connection: object, connection_record: object ) -> None: cursor = dbapi_connection.cursor() # type: ignore[attr-defined] + cursor.execute("PRAGMA journal_mode=WAL") + cursor.execute("PRAGMA busy_timeout=5000") + cursor.execute("PRAGMA synchronous=NORMAL") cursor.execute("PRAGMA foreign_keys=ON") cursor.close() diff --git a/src/meshcore_hub/common/models/event_observer.py b/src/meshcore_hub/common/models/event_observer.py index 14442d7..a3478cd 100644 --- a/src/meshcore_hub/common/models/event_observer.py +++ b/src/meshcore_hub/common/models/event_observer.py @@ -12,6 +12,7 @@ from sqlalchemy import ( Index, String, UniqueConstraint, + update, ) from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.orm import Mapped, Session, mapped_column, relationship @@ -138,4 +139,16 @@ def add_event_observer( ) result = session.execute(stmt) rowcount = getattr(result, "rowcount", 0) + + # Mark the observing node as an observer. Guarded on is_observer == False so + # this only writes on the first observation; the indexed flag lets the API's + # observer filter avoid scanning the (large) event tables. + from meshcore_hub.common.models.node import Node + + session.execute( + update(Node) + .where(Node.id == observer_node_id, Node.is_observer.is_(False)) + .values(is_observer=True) + ) + return bool(rowcount and rowcount > 0) diff --git a/src/meshcore_hub/common/models/node.py b/src/meshcore_hub/common/models/node.py index a920aa1..97f1754 100644 --- a/src/meshcore_hub/common/models/node.py +++ b/src/meshcore_hub/common/models/node.py @@ -3,7 +3,7 @@ from datetime import datetime from typing import TYPE_CHECKING, Optional -from sqlalchemy import DateTime, Float, Index, Integer, String +from sqlalchemy import Boolean, DateTime, Float, Index, Integer, String from sqlalchemy.orm import Mapped, mapped_column, relationship from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now @@ -26,6 +26,7 @@ class Node(Base, UUIDMixin, TimestampMixin): last_seen: Timestamp of most recent activity lat: GPS latitude coordinate (if available) lon: GPS longitude coordinate (if available) + is_observer: True if this node has observed at least one event created_at: Record creation timestamp updated_at: Record update timestamp """ @@ -74,6 +75,12 @@ class Node(Base, UUIDMixin, TimestampMixin): Float, nullable=True, ) + is_observer: Mapped[bool] = mapped_column( + Boolean, + default=False, + server_default="0", + nullable=False, + ) # Relationships tags: Mapped[list["NodeTag"]] = relationship( @@ -92,6 +99,7 @@ class Node(Base, UUIDMixin, TimestampMixin): __table_args__ = ( Index("ix_nodes_last_seen", "last_seen"), Index("ix_nodes_adv_type", "adv_type"), + Index("ix_nodes_is_observer", "is_observer"), ) def __repr__(self) -> str: diff --git a/src/meshcore_hub/common/schemas/nodes.py b/src/meshcore_hub/common/schemas/nodes.py index b706e9f..1238e0e 100644 --- a/src/meshcore_hub/common/schemas/nodes.py +++ b/src/meshcore_hub/common/schemas/nodes.py @@ -116,6 +116,9 @@ class NodeRead(BaseModel): ) lat: Optional[float] = Field(default=None, description="GPS latitude coordinate") lon: Optional[float] = Field(default=None, description="GPS longitude coordinate") + is_observer: bool = Field( + default=False, description="Whether this node has observed at least one event" + ) created_at: datetime = Field(..., description="Record creation timestamp") updated_at: datetime = Field(..., description="Record update timestamp") tags: list[NodeTagRead] = Field(default_factory=list, description="Node tags") diff --git a/tests/test_api/test_nodes.py b/tests/test_api/test_nodes.py index e9cf304..30fd9dd 100644 --- a/tests/test_api/test_nodes.py +++ b/tests/test_api/test_nodes.py @@ -171,27 +171,10 @@ class TestListNodesFilters: def test_filter_by_observer_true( self, client_no_auth, api_db_session, receiver_node ): - """Test filtering nodes that have observed events.""" - from datetime import datetime, timezone - - from meshcore_hub.common.models import Advertisement, Message - - # This node has observed an ad and a message - advert = Advertisement( - public_key="obsflt1obsflt1obsflt1obsflt1ob", - name="ObservedAd", - adv_type="CLIENT", - received_at=datetime.now(timezone.utc), - observer_node_id=receiver_node.id, - ) - msg = Message( - message_type="channel", - channel_idx=1, - text="Observed msg", - received_at=datetime.now(timezone.utc), - observer_node_id=receiver_node.id, - ) - api_db_session.add_all([advert, msg]) + """Test filtering nodes by the precomputed is_observer flag.""" + # The collector sets this flag when a node observes an event. + receiver_node.is_observer = True + api_db_session.add(receiver_node) api_db_session.commit() response = client_no_auth.get("/api/v1/nodes?observer=true") diff --git a/tests/test_collector/test_cleanup.py b/tests/test_collector/test_cleanup.py index 9ba1b3e..6ca08ec 100644 --- a/tests/test_collector/test_cleanup.py +++ b/tests/test_collector/test_cleanup.py @@ -417,3 +417,52 @@ async def test_cleanup_orphaned_node_relations_dry_run( await async_db_session.scalar(select(func.count()).select_from(UserProfileNode)) == 1 ) + + +@pytest.mark.asyncio +async def test_cleanup_clears_stale_observer_flags( + async_db_session: AsyncSession, +) -> None: + """A node whose only events are pruned gets is_observer cleared; an active + observer keeps its flag.""" + old_date = datetime.now(timezone.utc) - timedelta(days=60) + recent_date = datetime.now(timezone.utc) - timedelta(days=5) + + stale = Node(public_key="s" * 64, name="Stale Observer", is_observer=True) + active = Node(public_key="b" * 64, name="Active Observer", is_observer=True) + async_db_session.add_all([stale, active]) + await async_db_session.flush() + + # Stale observer's only event is old and will be pruned + async_db_session.add( + Advertisement( + public_key=stale.public_key, + observer_node_id=stale.id, + created_at=old_date, + updated_at=old_date, + ) + ) + # Active observer has a recent event that survives cleanup + async_db_session.add( + Advertisement( + public_key=active.public_key, + observer_node_id=active.id, + created_at=recent_date, + updated_at=recent_date, + ) + ) + await async_db_session.commit() + + stats = await cleanup_old_data(async_db_session, retention_days=30, dry_run=False) + + assert stats.observers_cleared == 1 + + await async_db_session.rollback() # Refresh from DB + stale_flag = await async_db_session.scalar( + select(Node.is_observer).where(Node.id == stale.id) + ) + active_flag = await async_db_session.scalar( + select(Node.is_observer).where(Node.id == active.id) + ) + assert stale_flag is False + assert active_flag is True diff --git a/tests/test_common/test_database.py b/tests/test_common/test_database.py new file mode 100644 index 0000000..6cc6284 --- /dev/null +++ b/tests/test_common/test_database.py @@ -0,0 +1,36 @@ +"""Tests for database engine configuration.""" + +from pathlib import Path + +from sqlalchemy import text + +from meshcore_hub.common.database import create_database_engine + + +class TestSqlitePragmas: + """Verify concurrency-related SQLite pragmas are applied on connect.""" + + def test_wal_and_busy_timeout_enabled(self, tmp_path: Path) -> None: + """File-based SQLite engines should run in WAL mode with a busy timeout.""" + db_path = tmp_path / "pragma.db" + engine = create_database_engine(f"sqlite:///{db_path}") + try: + with engine.connect() as conn: + journal_mode = conn.execute(text("PRAGMA journal_mode")).scalar() + busy_timeout = conn.execute(text("PRAGMA busy_timeout")).scalar() + foreign_keys = conn.execute(text("PRAGMA foreign_keys")).scalar() + + assert str(journal_mode).lower() == "wal" + assert busy_timeout is not None and int(busy_timeout) >= 5000 + assert foreign_keys is not None and int(foreign_keys) == 1 + finally: + engine.dispose() + + def test_in_memory_engine_builds(self) -> None: + """In-memory SQLite must still build (no overflow-pool kwargs).""" + engine = create_database_engine("sqlite:///:memory:") + try: + with engine.connect() as conn: + assert conn.execute(text("SELECT 1")).scalar() == 1 + finally: + engine.dispose() diff --git a/tests/test_common/test_models.py b/tests/test_common/test_models.py index dff3a6e..50b11fd 100644 --- a/tests/test_common/test_models.py +++ b/tests/test_common/test_models.py @@ -260,3 +260,21 @@ class TestEventObserverModel: observer = db_session.execute(select(EventObserver)).scalar_one() assert observer.path_len is None assert observer.snr is None + + def test_add_event_observer_sets_is_observer_flag(self, db_session) -> None: + """Observing an event marks the observer node with is_observer=True.""" + node = Node(public_key="c" * 64, name="Observer3") + db_session.add(node) + db_session.commit() + assert node.is_observer is False + + add_event_observer( + session=db_session, + event_type="advertisement", + event_hash="deadbeef", + observer_node_id=node.id, + ) + db_session.commit() + + db_session.refresh(node) + assert node.is_observer is True