Add real-time file synchronization with watchdog, SSE notifications, and dynamic vault management API
This commit is contained in:
+252
-3
@@ -14,7 +14,7 @@ import frontmatter
|
||||
import mistune
|
||||
from fastapi import FastAPI, HTTPException, Query, Body
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.responses import HTMLResponse, FileResponse, Response
|
||||
from fastapi.responses import HTMLResponse, FileResponse, Response, StreamingResponse
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from backend.indexer import (
|
||||
@@ -22,12 +22,18 @@ from backend.indexer import (
|
||||
reload_index,
|
||||
index,
|
||||
path_index,
|
||||
vault_config,
|
||||
get_vault_data,
|
||||
get_vault_names,
|
||||
find_file_in_index,
|
||||
parse_markdown_file,
|
||||
_extract_tags,
|
||||
SUPPORTED_EXTENSIONS,
|
||||
update_single_file,
|
||||
remove_single_file,
|
||||
handle_file_move,
|
||||
remove_vault_from_index,
|
||||
add_vault_to_index,
|
||||
)
|
||||
from backend.search import search, get_all_tags, advanced_search, suggest_titles, suggest_tags
|
||||
from backend.image_processor import preprocess_images
|
||||
@@ -214,29 +220,146 @@ class HealthResponse(BaseModel):
|
||||
total_files: int
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SSE Manager — Server-Sent Events for real-time notifications
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class SSEManager:
|
||||
"""Manages SSE client connections and broadcasts events."""
|
||||
|
||||
def __init__(self):
|
||||
self._clients: List[asyncio.Queue] = []
|
||||
|
||||
async def connect(self) -> asyncio.Queue:
|
||||
"""Register a new SSE client and return its message queue."""
|
||||
queue: asyncio.Queue = asyncio.Queue()
|
||||
self._clients.append(queue)
|
||||
logger.debug(f"SSE client connected (total: {len(self._clients)})")
|
||||
return queue
|
||||
|
||||
def disconnect(self, queue: asyncio.Queue):
|
||||
"""Remove a disconnected SSE client."""
|
||||
if queue in self._clients:
|
||||
self._clients.remove(queue)
|
||||
logger.debug(f"SSE client disconnected (total: {len(self._clients)})")
|
||||
|
||||
async def broadcast(self, event_type: str, data: dict):
|
||||
"""Send an event to all connected SSE clients."""
|
||||
message = _json.dumps(data, ensure_ascii=False)
|
||||
dead: List[asyncio.Queue] = []
|
||||
for q in self._clients:
|
||||
try:
|
||||
q.put_nowait({"event": event_type, "data": message})
|
||||
except asyncio.QueueFull:
|
||||
dead.append(q)
|
||||
for q in dead:
|
||||
self.disconnect(q)
|
||||
|
||||
@property
|
||||
def client_count(self) -> int:
|
||||
return len(self._clients)
|
||||
|
||||
|
||||
sse_manager = SSEManager()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Application lifespan (replaces deprecated on_event)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
from backend.watcher import VaultWatcher
|
||||
|
||||
# Thread pool for offloading CPU-bound search from the event loop.
|
||||
# Sized to 2 workers so concurrent searches don't starve other requests.
|
||||
_search_executor: Optional[ThreadPoolExecutor] = None
|
||||
_vault_watcher: Optional[VaultWatcher] = None
|
||||
|
||||
|
||||
async def _on_vault_change(events: list):
|
||||
"""Callback invoked by VaultWatcher when files change in watched vaults.
|
||||
|
||||
Processes each event (create/modify/delete/move) and updates the index
|
||||
incrementally, then broadcasts SSE notifications.
|
||||
"""
|
||||
updated_vaults = set()
|
||||
changes = []
|
||||
|
||||
for event in events:
|
||||
vault_name = event["vault"]
|
||||
event_type = event["type"]
|
||||
src = event["src"]
|
||||
dest = event.get("dest")
|
||||
|
||||
try:
|
||||
if event_type in ("created", "modified"):
|
||||
result = await update_single_file(vault_name, src)
|
||||
if result:
|
||||
changes.append({"action": "updated", "vault": vault_name, "path": result["path"]})
|
||||
updated_vaults.add(vault_name)
|
||||
|
||||
elif event_type == "deleted":
|
||||
result = await remove_single_file(vault_name, src)
|
||||
if result:
|
||||
changes.append({"action": "deleted", "vault": vault_name, "path": result["path"]})
|
||||
updated_vaults.add(vault_name)
|
||||
|
||||
elif event_type == "moved":
|
||||
result = await handle_file_move(vault_name, src, dest)
|
||||
if result:
|
||||
changes.append({"action": "moved", "vault": vault_name, "path": result["path"]})
|
||||
updated_vaults.add(vault_name)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing {event_type} event for {src}: {e}")
|
||||
|
||||
if changes:
|
||||
await sse_manager.broadcast("index_updated", {
|
||||
"vaults": list(updated_vaults),
|
||||
"changes": changes,
|
||||
"total_changes": len(changes),
|
||||
})
|
||||
logger.info(f"Hot-reload: {len(changes)} change(s) in {list(updated_vaults)}")
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""Application lifespan: build index on startup, cleanup on shutdown."""
|
||||
global _search_executor
|
||||
global _search_executor, _vault_watcher
|
||||
_search_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="search")
|
||||
logger.info("ObsiGate starting \u2014 building index...")
|
||||
await build_index()
|
||||
|
||||
# Start file watcher
|
||||
config = _load_config()
|
||||
watcher_enabled = config.get("watcher_enabled", True)
|
||||
if watcher_enabled:
|
||||
use_polling = config.get("watcher_use_polling", False)
|
||||
polling_interval = config.get("watcher_polling_interval", 5.0)
|
||||
debounce = config.get("watcher_debounce", 2.0)
|
||||
_vault_watcher = VaultWatcher(
|
||||
on_file_change=_on_vault_change,
|
||||
debounce_seconds=debounce,
|
||||
use_polling=use_polling,
|
||||
polling_interval=polling_interval,
|
||||
)
|
||||
vaults_to_watch = {name: cfg["path"] for name, cfg in vault_config.items()}
|
||||
await _vault_watcher.start(vaults_to_watch)
|
||||
logger.info("File watcher started.")
|
||||
else:
|
||||
logger.info("File watcher disabled by configuration.")
|
||||
|
||||
logger.info("ObsiGate ready.")
|
||||
yield
|
||||
|
||||
# Shutdown
|
||||
if _vault_watcher:
|
||||
await _vault_watcher.stop()
|
||||
_vault_watcher = None
|
||||
_search_executor.shutdown(wait=False)
|
||||
_search_executor = None
|
||||
|
||||
|
||||
app = FastAPI(title="ObsiGate", version="1.2.0", lifespan=lifespan)
|
||||
app = FastAPI(title="ObsiGate", version="1.3.0", lifespan=lifespan)
|
||||
|
||||
# Resolve frontend path relative to this file
|
||||
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
||||
@@ -886,9 +1009,131 @@ async def api_reload():
|
||||
``ReloadResponse`` with per-vault file and tag counts.
|
||||
"""
|
||||
stats = await reload_index()
|
||||
await sse_manager.broadcast("index_reloaded", {
|
||||
"vaults": list(stats.keys()),
|
||||
"stats": stats,
|
||||
})
|
||||
return {"status": "ok", "vaults": stats}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SSE endpoint — Server-Sent Events stream
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.get("/api/events")
|
||||
async def api_events():
|
||||
"""SSE stream for real-time index update notifications.
|
||||
|
||||
Sends keepalive comments every 30s. Events:
|
||||
- ``index_updated``: partial index change (file create/modify/delete/move)
|
||||
- ``index_reloaded``: full re-index completed
|
||||
- ``vault_added``: new vault added dynamically
|
||||
- ``vault_removed``: vault removed dynamically
|
||||
"""
|
||||
queue = await sse_manager.connect()
|
||||
|
||||
async def event_generator():
|
||||
try:
|
||||
# Send initial connection event
|
||||
yield f"event: connected\ndata: {_json.dumps({'sse_clients': sse_manager.client_count})}\n\n"
|
||||
while True:
|
||||
try:
|
||||
msg = await asyncio.wait_for(queue.get(), timeout=30.0)
|
||||
yield f"event: {msg['event']}\ndata: {msg['data']}\n\n"
|
||||
except asyncio.TimeoutError:
|
||||
# Keepalive comment
|
||||
yield ": keepalive\n\n"
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
finally:
|
||||
sse_manager.disconnect(queue)
|
||||
|
||||
return StreamingResponse(
|
||||
event_generator(),
|
||||
media_type="text/event-stream",
|
||||
headers={
|
||||
"Cache-Control": "no-cache",
|
||||
"Connection": "keep-alive",
|
||||
"X-Accel-Buffering": "no",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dynamic vault management endpoints
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.post("/api/vaults/add")
|
||||
async def api_add_vault(body: dict = Body(...)):
|
||||
"""Add a new vault dynamically without restarting.
|
||||
|
||||
Body:
|
||||
name: Display name for the vault.
|
||||
path: Absolute filesystem path to the vault directory.
|
||||
"""
|
||||
name = body.get("name", "").strip()
|
||||
vault_path = body.get("path", "").strip()
|
||||
|
||||
if not name or not vault_path:
|
||||
raise HTTPException(status_code=400, detail="Both 'name' and 'path' are required")
|
||||
|
||||
if name in index:
|
||||
raise HTTPException(status_code=409, detail=f"Vault '{name}' already exists")
|
||||
|
||||
if not Path(vault_path).exists():
|
||||
raise HTTPException(status_code=400, detail=f"Path does not exist: {vault_path}")
|
||||
|
||||
stats = await add_vault_to_index(name, vault_path)
|
||||
|
||||
# Start watching the new vault
|
||||
if _vault_watcher:
|
||||
await _vault_watcher.add_vault(name, vault_path)
|
||||
|
||||
await sse_manager.broadcast("vault_added", {"vault": name, "stats": stats})
|
||||
return {"status": "ok", "vault": name, "stats": stats}
|
||||
|
||||
|
||||
@app.delete("/api/vaults/{vault_name}")
|
||||
async def api_remove_vault(vault_name: str):
|
||||
"""Remove a vault from the index and stop watching it.
|
||||
|
||||
Args:
|
||||
vault_name: Name of the vault to remove.
|
||||
"""
|
||||
if vault_name not in index:
|
||||
raise HTTPException(status_code=404, detail=f"Vault '{vault_name}' not found")
|
||||
|
||||
# Stop watching
|
||||
if _vault_watcher:
|
||||
await _vault_watcher.remove_vault(vault_name)
|
||||
|
||||
await remove_vault_from_index(vault_name)
|
||||
await sse_manager.broadcast("vault_removed", {"vault": vault_name})
|
||||
return {"status": "ok", "vault": vault_name}
|
||||
|
||||
|
||||
@app.get("/api/vaults/status")
|
||||
async def api_vaults_status():
|
||||
"""Detailed status of all vaults including watcher state.
|
||||
|
||||
Returns per-vault: file count, tag count, watching status, vault path.
|
||||
"""
|
||||
statuses = {}
|
||||
for vname, vdata in index.items():
|
||||
watching = _vault_watcher is not None and vname in _vault_watcher.observers
|
||||
statuses[vname] = {
|
||||
"file_count": len(vdata.get("files", [])),
|
||||
"tag_count": len(vdata.get("tags", {})),
|
||||
"path": vdata.get("path", ""),
|
||||
"watching": watching,
|
||||
}
|
||||
return {
|
||||
"vaults": statuses,
|
||||
"watcher_active": _vault_watcher is not None,
|
||||
"sse_clients": sse_manager.client_count,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/image/{vault_name}")
|
||||
async def api_image(vault_name: str, path: str = Query(..., description="Relative path to image")):
|
||||
"""Serve an image file with proper MIME type.
|
||||
@@ -979,6 +1224,10 @@ _DEFAULT_CONFIG = {
|
||||
"max_snippet_highlights": 5,
|
||||
"title_boost": 3.0,
|
||||
"path_boost": 1.5,
|
||||
"watcher_enabled": True,
|
||||
"watcher_use_polling": False,
|
||||
"watcher_polling_interval": 5.0,
|
||||
"watcher_debounce": 2.0,
|
||||
"tag_boost": 2.0,
|
||||
"prefix_max_expansions": 50,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user