diff --git a/README.md b/README.md index 208369a2..eca44344 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,9 @@ python ComfyUI-Majoor-AssetsManager/scripts/install-requirements.py ``` ### Tkinter Note (Embedded Python) -The native folder picker requires built-in `tkinter`. On embedded Python builds (for example `python_embeded`), `tkinter` is often missing, so the UI falls back to manual path entry. +The native folder picker (`/mjr/sys/browse-folder`) requires built-in `tkinter`. +On embedded Python builds (for example `python_embeded` / `python_embedded`), `tkinter` is often missing. +In that case (or on headless Linux without `DISPLAY`), the backend returns a picker-unavailable error and the UI should use manual path entry. - The extension remains fully functional with manual path input. - `pip install tk` is not a replacement for built-in `tkinter`. @@ -257,10 +259,11 @@ Hotkeys are ignored while typing inside inputs. Settings are stored under the `mjrSettings` key in `localStorage` (browser-side). -Backend persistence is currently limited to `probeBackend.mode` (stored in the SQLite `metadata` table). It uses a version bump so changes propagate quickly even with multiple backend instances. Other settings are intentionally browser-local and won't sync across machines/browsers. +Backend persistence includes `probeBackend.mode`, metadata fallback toggles, output-directory override, and security preferences (stored in SQLite `metadata`). UI layout/visual preferences remain browser-local (`localStorage`) and are not synced across machines/browsers. -Remote access & API Token (new): -- The UI now exposes **API token** settings allowing administrators to configure a token and enable remote access to the Assets Manager. Access is disabled by default and the token is stored in the backend settings; ensure you secure your network if enabling remote access. See the user guide for configuration details. +Remote access & API Token: +- The UI exposes an **API token** field for client-side header injection. The authoritative backend token still comes from environment variables (`MAJOOR_API_TOKEN` / `MJR_API_TOKEN`) at startup. +- Remote write behavior is controlled by backend security settings/env vars (`MAJOOR_REQUIRE_AUTH`, `MAJOOR_ALLOW_REMOTE_WRITE`, and security prefs). - Page size (assets per request) @@ -311,35 +314,28 @@ Canonical namespace is now `MJR_AM_...` to avoid collisions with ComfyUI/global Compatibility aliases are still accepted (`MAJOOR_...` and some `MJR_...` names), but new deployments should prefer `MJR_AM_...`. -- `MAJOOR_OUTPUT_DIRECTORY` - override output directory -- `MAJOOR_EXIFTOOL_PATH` / `MAJOOR_EXIFTOOL_BIN` - ExifTool path -- `MAJOOR_FFPROBE_PATH` / `MAJOOR_FFPROBE_BIN` - FFprobe path -- `MAJOOR_MEDIA_PROBE_BACKEND` - `auto|exiftool|ffprobe|both` -- `MAJOOR_EXIFTOOL_MIN_VERSION` - minimum ExifTool version required (optional, e.g. `12.40`) -- `MAJOOR_FFPROBE_MIN_VERSION` - minimum FFprobe version required (optional, e.g. `6.0.0`) -- `MAJOOR_ENABLE_FILE_WATCHER` - auto reindexing (default `false`) -- `MAJOOR_WATCHER_INTERVAL` / `MAJOOR_WATCHER_JOIN_TIMEOUT` / `MAJOOR_WATCHER_PATHS` - - `MJR_WATCHER_DEBOUNCE_MS` - debounce delay (ms) for watcher batches (default `500`) - - `MJR_WATCHER_DEDUPE_TTL_MS` - dup suppression window (ms) before a file can be reprocessed (default `3000`) - - `MJR_WATCHER_PENDING_MAX` - max outstanding watcher events queued before new ones are dropped (default `5000`) - - `MJR_WATCHER_MAX_FLUSH_CONCURRENCY` - max concurrent watcher flush jobs (default `2`) - - `MJR_DB_CONSISTENCY_SAMPLE` - how many random DB entries to poke during periodic consistency sweeps (default `32`) - - `MJR_DB_CONSISTENCY_COOLDOWN_SECONDS` - minimum seconds between consistency sweeps (default `3600`) -- `MAJOOR_DB_TIMEOUT` / `MAJOOR_DB_MAX_CONNECTIONS` / `MAJOOR_DB_QUERY_TIMEOUT` -- `MAJOOR_TO_THREAD_TIMEOUT` - timeout (seconds) for `asyncio.to_thread(...)` work in HTTP handlers (default `30`) -- `MAJOOR_MAX_METADATA_JSON_BYTES` - max metadata JSON size stored in DB/cache (default `2097152`) +- `MJR_AM_OUTPUT_DIRECTORY` (alias: `MAJOOR_OUTPUT_DIRECTORY`) - override output directory +- `MJR_AM_EXIFTOOL_PATH` / `MJR_AM_FFPROBE_PATH` (aliases: `MAJOOR_*`) - tool executable paths +- `MJR_AM_MEDIA_PROBE_BACKEND` (alias: `MAJOOR_MEDIA_PROBE_BACKEND`) - `auto|exiftool|ffprobe|both` +- `MJR_AM_EXIFTOOL_MIN_VERSION` / `MJR_AM_FFPROBE_MIN_VERSION` (aliases: `MAJOOR_*`) +- `MJR_AM_ENABLE_WATCHER` (alias: `MJR_ENABLE_WATCHER`) - enable file watcher +- `MJR_AM_WATCHER_DEBOUNCE_MS` / `MJR_AM_WATCHER_DEDUPE_TTL_MS` (aliases: `MJR_WATCHER_*`) +- `MJR_AM_WATCHER_PENDING_MAX` / `MJR_AM_WATCHER_MAX_FLUSH_CONCURRENCY` (aliases: `MJR_WATCHER_*`) +- `MJR_AM_WATCHER_MIN_FILE_SIZE_BYTES` / `MJR_AM_WATCHER_MAX_FILE_SIZE_BYTES` (aliases include legacy `MAJOOR_WATCHER_*`) +- `MAJOOR_SCAN_IOPS_LIMIT` - optional scan I/O pacing (ops/sec) for directory walk on slow NAS/SMB (`0` disables) +- `MJR_DB_CONSISTENCY_SAMPLE` / `MJR_DB_CONSISTENCY_COOLDOWN_SECONDS` - periodic consistency sweep tuning +- `MJR_AM_DB_TIMEOUT` / `MJR_AM_DB_MAX_CONNECTIONS` / `MJR_AM_DB_QUERY_TIMEOUT` (aliases: `MAJOOR_*`) +- `MJR_AM_TO_THREAD_TIMEOUT` (alias: `MAJOOR_TO_THREAD_TIMEOUT`) - timeout for `asyncio.to_thread(...)` work in HTTP handlers +- `MJR_AM_MAX_METADATA_JSON_BYTES` (alias: `MAJOOR_MAX_METADATA_JSON_BYTES`) - `MJR_AM_CLEANUP_RESERVED_NAMES` - set to `1` to enable Windows reserved-name cleanup in `custom_nodes` (disabled by default) - `MJR_COLLECTION_MAX_ITEMS` - max items per collection JSON (default `50000`) - `MJR_ALLOW_SYMLINKS` - allow symlink/junction custom roots (default `off`) - `MAJOOR_TRUSTED_PROXIES` - comma-separated IPs/CIDRs allowed to supply `X-Forwarded-For`/`X-Real-IP` (default `127.0.0.1,::1`) - `MAJOOR_API_TOKEN` (or `MJR_API_TOKEN`) - optional token to authorize write operations (send via `X-MJR-Token` or `Authorization: Bearer `) -- `MAJOOR_REQUIRE_AUTH` - set to `1` to require `MAJOOR_API_TOKEN` even for localhost -- `MAJOOR_ALLOW_REMOTE_WRITE` - set to `1` to allow write operations from non-local clients without a token (**unsafe**) -- `MAJOOR_SAFE_MODE` - set to `0` to disable Safe Mode (default enabled) -- `MAJOOR_ALLOW_WRITE` - set to `1` to allow rating/tags writes while Safe Mode is enabled -- `MAJOOR_ALLOW_DELETE` - set to `1` to enable asset deletion (disabled by default) -- `MAJOOR_ALLOW_RENAME` - set to `1` to enable asset renaming (disabled by default) -- `MAJOOR_ALLOW_OPEN_IN_FOLDER` - set to `1` to enable `/mjr/am/open-in-folder` (disabled by default) +- `MAJOOR_REQUIRE_AUTH` - require token auth even for localhost +- `MAJOOR_ALLOW_REMOTE_WRITE` - allow non-local write operations without token (**unsafe**) +- `MAJOOR_SAFE_MODE` - safe-mode gate for write operations +- `MAJOOR_ALLOW_WRITE` / `MAJOOR_ALLOW_DELETE` / `MAJOOR_ALLOW_RENAME` / `MAJOOR_ALLOW_OPEN_IN_FOLDER` / `MAJOOR_ALLOW_RESET_INDEX` - operation gates (also configurable via persisted security preferences) ### API Token Setup @@ -375,10 +371,10 @@ Once set, include the token in your API requests via one of these headers: ### Security Summary -- Without token: write operations are localhost-only. +- Without token: write operations are localhost-only (unless `MAJOOR_ALLOW_REMOTE_WRITE=1`). - With token: write operations require authentication. -- Safe Mode is enabled by default. -- Destructive operations (delete/rename/open-in-folder) are disabled by default and must be explicitly enabled. +- Operation gates for delete/rename/open-in-folder/reset-index are enforced via security prefs and corresponding `MAJOOR_ALLOW_*` env vars. +- Safe Mode behavior is controlled by `MAJOOR_SAFE_MODE` and persisted security preferences. You can persist the same secret in ComfyUI's Settings modal under **Security -> Majoor: API Token**. The Assets Manager stores it in browser storage and automatically sends both `X-MJR-Token` and `Authorization: Bearer ...` on behalf of the UI, so remote clients only need to match the shared secret between the backend environment and the settings field. The backend still requires the environment variable (`MAJOOR_API_TOKEN`/`MJR_API_TOKEN`) when ComfyUI starts; the UI control just remembers it for you so the headers stay in sync. @@ -387,8 +383,8 @@ The backend still requires the environment variable (`MAJOOR_API_TOKEN`/`MJR_API - **CSRF**: state-changing endpoints require `X-Requested-With: XMLHttpRequest` (or `X-CSRF-Token`) and validate `Origin` vs `Host` when present. - **Rate limiting**: in-memory per-client limits exist on expensive endpoints (search/scan/metadata/batch-zip). Client identity is based on IP; `X-Forwarded-For` is only honored when the connection comes from `MAJOOR_TRUSTED_PROXIES`. - **Path safety**: file operations validate root containment for output/input/custom roots and reject paths outside allowed roots (symlink/junction handling is opt-in). -- **Write access**: destructive actions are loopback-only by default unless a token is configured. -- **Safe Mode**: write operations are disabled by default unless explicitly enabled via environment variables. +- **Write access**: loopback-only by default when no token is configured (can be relaxed by `MAJOOR_ALLOW_REMOTE_WRITE=1`). +- **Operation gating**: delete/rename/open-in-folder/reset-index and write behavior are enforced by security prefs + `MAJOOR_ALLOW_*` / `MAJOOR_SAFE_MODE`. - **Batch ZIP**: ZIP building streams from an open file handle (avoids TOCTOU rename/replace races). ## UI Preview diff --git a/js/features/panel/AssetsManagerPanel.js b/js/features/panel/AssetsManagerPanel.js index 98e9c222..34767021 100644 --- a/js/features/panel/AssetsManagerPanel.js +++ b/js/features/panel/AssetsManagerPanel.js @@ -234,7 +234,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } updateSummaryBar?.({ state, gridContainer }); } catch {} }; - gridContainer.addEventListener("mjr:grid-stats", handler); + gridContainer.addEventListener("mjr:grid-stats", handler, { signal: panelLifecycleAC?.signal }); registerSummaryDispose(() => { try { gridContainer.removeEventListener("mjr:grid-stats", handler); @@ -628,7 +628,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } gridController.reloadGrid(); } catch {} }; - gridContainer.addEventListener("mjr:reload-grid", _reloadGridHandler); + gridContainer.addEventListener("mjr:reload-grid", _reloadGridHandler, { signal: panelLifecycleAC?.signal }); const onCustomSubfolderChanged = async (e) => { const next = String(e?.detail?.subfolder || "").trim().replaceAll("\\", "/"); const prev = currentFolderPath(); @@ -642,7 +642,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } } catch {} } }; - gridContainer.addEventListener("mjr:custom-subfolder-changed", onCustomSubfolderChanged); + gridContainer.addEventListener("mjr:custom-subfolder-changed", onCustomSubfolderChanged, { signal: panelLifecycleAC?.signal }); try { gridContainer._mjrHasCustomSubfolderHandler = true; } catch {} @@ -670,7 +670,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } } catch {} } catch {} }; - gridContainer.addEventListener("mjr:badge-duplicates-focus", onDuplicateBadgeFocus); + gridContainer.addEventListener("mjr:badge-duplicates-focus", onDuplicateBadgeFocus, { signal: panelLifecycleAC?.signal }); registerSummaryDispose(() => { try { if (_reloadGridHandler) gridContainer.removeEventListener("mjr:reload-grid", _reloadGridHandler); @@ -868,11 +868,11 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } btn.addEventListener("mouseenter", () => { btn.style.borderColor = "rgba(145,205,255,0.4)"; btn.style.background = "linear-gradient(135deg, rgba(80,140,255,0.18), rgba(32,100,200,0.14))"; - }); + }, { signal: panelLifecycleAC?.signal }); btn.addEventListener("mouseleave", () => { btn.style.borderColor = "rgba(120,190,255,0.18)"; btn.style.background = "linear-gradient(135deg, rgba(255,255,255,0.03), rgba(255,255,255,0.01))"; - }); + }, { signal: panelLifecycleAC?.signal }); return btn; }; @@ -930,7 +930,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } } catch {} popovers.close(pinnedFoldersPopover); await gridController.reloadGrid(); - }); + }, { signal: panelLifecycleAC?.signal }); const unpinBtn = document.createElement("button"); unpinBtn.type = "button"; @@ -963,7 +963,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } } catch {} await renderPinnedFoldersMenu(); await gridController.reloadGrid(); - }); + }, { signal: panelLifecycleAC?.signal }); row.appendChild(openBtn); row.appendChild(unpinBtn); @@ -1012,7 +1012,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } }); } catch {} - bindFilters({ + const disposeFilters = bindFilters({ state, kindSelect, wfCheckbox, @@ -1026,6 +1026,7 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } dateRangeSelect, dateExactInput, reloadGrid: gridController.reloadGrid, + lifecycleSignal: panelLifecycleAC?.signal || null, onFiltersChanged: () => { try { agendaCalendar?.refresh?.(); @@ -1118,9 +1119,9 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } } catch {} notifyContextChanged(); }; - gridContainer.addEventListener("mjr:grid-stats", onStats); - gridContainer.addEventListener("mjr:selection-changed", onSelectionChanged); - document.addEventListener?.("mjr-settings-changed", onStats); + gridContainer.addEventListener("mjr:grid-stats", onStats, { signal: panelLifecycleAC?.signal }); + gridContainer.addEventListener("mjr:selection-changed", onSelectionChanged, { signal: panelLifecycleAC?.signal }); + document.addEventListener?.("mjr-settings-changed", onStats, { signal: panelLifecycleAC?.signal }); // Observe DOM changes for card count / selection class changes. let raf = null; @@ -1227,6 +1228,12 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } container._mjrVersionUpdateCleanup?.(); } catch {} container._mjrVersionUpdateCleanup = null; + try { + container._mjrPanelState?._mjrDispose?.(); + } catch {} + try { + container._mjrPanelState = null; + } catch {} try { if (_scrollTimer) clearTimeout(_scrollTimer); _scrollTimer = null; @@ -1245,6 +1252,9 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } try { agendaCalendar?.dispose?.(); } catch {} + try { + disposeFilters?.(); + } catch {} try { if (searchTimeout) clearTimeout(searchTimeout); searchTimeout = null; @@ -1266,6 +1276,9 @@ export async function renderAssetsManager(container, { useComfyThemeUI = true } try { gridContainer?._mjrSummaryBarDispose?.(); } catch {} + try { + gridContainer?._mjrSummaryBarObserver?.disconnect?.(); + } catch {} try { gridContainer?._mjrGridContextMenuUnbind?.(); } catch {} diff --git a/js/features/panel/controllers/filtersController.js b/js/features/panel/controllers/filtersController.js index 96433e62..f15ab7b5 100644 --- a/js/features/panel/controllers/filtersController.js +++ b/js/features/panel/controllers/filtersController.js @@ -14,8 +14,33 @@ export function bindFilters({ dateRangeSelect, dateExactInput, reloadGrid, - onFiltersChanged = null + onFiltersChanged = null, + lifecycleSignal = null }) { + const disposers = []; + const addManagedListener = (target, event, handler, options) => { + if (!target?.addEventListener || typeof handler !== "function") return; + try { + target.addEventListener(event, handler, options); + } catch { + try { + target.addEventListener(event, handler); + } catch { + return; + } + } + if (lifecycleSignal) return; + disposers.push(() => { + try { + target.removeEventListener(event, handler, options); + } catch { + try { + target.removeEventListener(event, handler); + } catch {} + } + }); + }; + // Cleanup previous window listener to prevent accumulation if (_lastAgendaHandler) { try { @@ -24,27 +49,27 @@ export function bindFilters({ _lastAgendaHandler = null; } - kindSelect.addEventListener("change", async () => { + addManagedListener(kindSelect, "change", async () => { state.kindFilter = kindSelect.value || ""; try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); - wfCheckbox.addEventListener("change", async () => { + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); + addManagedListener(wfCheckbox, "change", async () => { state.workflowOnly = Boolean(wfCheckbox.checked); try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); - ratingSelect.addEventListener("change", async () => { + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); + addManagedListener(ratingSelect, "change", async () => { state.minRating = Number(ratingSelect.value || 0) || 0; try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); if (workflowTypeSelect) { - workflowTypeSelect.addEventListener("change", async () => { + addManagedListener(workflowTypeSelect, "change", async () => { state.workflowType = String(workflowTypeSelect.value || "").trim().toUpperCase(); try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); } const applySizeFilter = async () => { @@ -59,8 +84,8 @@ export function bindFilters({ try { onFiltersChanged?.(); } catch {} await reloadGrid(); }; - minSizeInput?.addEventListener("change", applySizeFilter); - maxSizeInput?.addEventListener("change", applySizeFilter); + addManagedListener(minSizeInput, "change", applySizeFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined); + addManagedListener(maxSizeInput, "change", applySizeFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined); const applyResolutionFilter = async () => { const minW = Number(minWidthInput?.value || 0); @@ -73,11 +98,11 @@ export function bindFilters({ try { onFiltersChanged?.(); } catch {} await reloadGrid(); }; - minWidthInput?.addEventListener("change", applyResolutionFilter); - minHeightInput?.addEventListener("change", applyResolutionFilter); + addManagedListener(minWidthInput, "change", applyResolutionFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined); + addManagedListener(minHeightInput, "change", applyResolutionFilter, lifecycleSignal ? { signal: lifecycleSignal } : undefined); if (resolutionPresetSelect) { - resolutionPresetSelect.addEventListener("change", async () => { + addManagedListener(resolutionPresetSelect, "change", async () => { const preset = String(resolutionPresetSelect.value || ""); const map = { hd: [1280, 720], @@ -92,10 +117,10 @@ export function bindFilters({ if (minHeightInput) minHeightInput.value = state.minHeight > 0 ? String(state.minHeight) : ""; try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); } if (dateRangeSelect) { - dateRangeSelect.addEventListener("change", async () => { + addManagedListener(dateRangeSelect, "change", async () => { state.dateRangeFilter = dateRangeSelect.value || ""; if (state.dateRangeFilter && state.dateExactFilter) { state.dateExactFilter = ""; @@ -105,14 +130,14 @@ export function bindFilters({ } try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); } if (dateExactInput) { const applyAgendaStyle = (status) => { dateExactInput.classList.toggle("mjr-agenda-filled", status === "filled"); dateExactInput.classList.toggle("mjr-agenda-empty", status === "empty"); }; - dateExactInput.addEventListener("change", async () => { + addManagedListener(dateExactInput, "change", async () => { state.dateExactFilter = dateExactInput.value ? String(dateExactInput.value) : ""; if (state.dateExactFilter && state.dateRangeFilter) { state.dateRangeFilter = ""; @@ -125,7 +150,7 @@ export function bindFilters({ } try { onFiltersChanged?.(); } catch {} await reloadGrid(); - }); + }, lifecycleSignal ? { signal: lifecycleSignal } : undefined); const handleAgendaStatus = (event) => { if (!event?.detail) return; const { date, hasResults } = event.detail; @@ -137,9 +162,35 @@ export function bindFilters({ applyAgendaStyle(hasResults ? "filled" : "empty"); }; try { - window?.addEventListener?.("MJR:AgendaStatus", handleAgendaStatus, { passive: true }); + const opts = lifecycleSignal ? { passive: true, signal: lifecycleSignal } : { passive: true }; + window?.addEventListener?.("MJR:AgendaStatus", handleAgendaStatus, opts); _lastAgendaHandler = handleAgendaStatus; + if (!lifecycleSignal) { + disposers.push(() => { + try { + window.removeEventListener("MJR:AgendaStatus", handleAgendaStatus); + } catch {} + }); + } else { + lifecycleSignal.addEventListener("abort", () => { + if (_lastAgendaHandler === handleAgendaStatus) _lastAgendaHandler = null; + }, { once: true }); + } } catch {} } + + return () => { + for (const dispose of disposers.splice(0)) { + try { + dispose(); + } catch {} + } + if (_lastAgendaHandler) { + try { + window.removeEventListener("MJR:AgendaStatus", _lastAgendaHandler); + } catch {} + _lastAgendaHandler = null; + } + }; } diff --git a/mjr_am_backend/adapters/db/schema.py b/mjr_am_backend/adapters/db/schema.py index 2bb3c77a..ab110bc1 100644 --- a/mjr_am_backend/adapters/db/schema.py +++ b/mjr_am_backend/adapters/db/schema.py @@ -404,7 +404,7 @@ def _schema_fingerprint() -> str: async def _ensure_schema_fingerprint(db) -> Result[bool]: - if not db.has_table("metadata"): + if not await db.ahas_table("metadata"): return Result.Ok(True) try: diff --git a/mjr_am_backend/adapters/db/sqlite.py b/mjr_am_backend/adapters/db/sqlite.py index f6104366..d15ad422 100644 --- a/mjr_am_backend/adapters/db/sqlite.py +++ b/mjr_am_backend/adapters/db/sqlite.py @@ -1,9 +1,9 @@ """ SQLite database connection manager. -Implementation note (Option A migration): +Implementation note (async-first migration): - This adapter uses `aiosqlite` internally. -- The public API remains synchronous for compatibility with the existing codebase. +- The public API exposes async methods as primary surface. Why this shape: - Most of the backend logic (scanner/searcher/updater/schema) is synchronous today. @@ -27,7 +27,8 @@ import time import uuid import os import json -from contextlib import contextmanager, asynccontextmanager +import ctypes +from contextlib import asynccontextmanager from pathlib import Path from queue import Empty, Queue from typing import Any, Dict, List, Optional, Tuple @@ -381,8 +382,6 @@ class Sqlite: self._write_lock: Optional[asyncio.Lock] = None self._tx_write_lock_tokens: set[str] = set() self._tx_state_lock = threading.Lock() - # Thread-local storage for sync transactions (caller thread). - self._tx_local = threading.local() self._asset_locks: Dict[str, Dict[str, Any]] = {} self._asset_locks_lock = threading.Lock() self._malformed_recovery_lock = threading.Lock() @@ -817,11 +816,6 @@ class Sqlite: raise def _tx_token(self) -> Optional[str]: - # Thread-local token for sync transaction() remains primary for back-compat. - # Fallback to ContextVar to support sync helper calls accidentally invoked from async flow. - tok = getattr(self._tx_local, "token", None) if hasattr(self, "_tx_local") else None - if tok: - return str(tok) try: ctx_tok = _TX_TOKEN.get() except Exception: @@ -1008,20 +1002,6 @@ class Sqlite: except Exception: raise - def execute(self, query: str, params: Optional[tuple] = None, fetch: bool = False) -> Result[Any]: - """Execute SQL on the DB loop thread (sync).""" - try: - # Sync callers may be inside `transaction()` (token lives on the caller thread). - # Pass it explicitly because DB work runs on the loop thread. - token = self._tx_token() - return self._loop_thread.run(self._execute_async(query, params, fetch, tx_token=token)) - except Exception as exc: - return Result.Err(ErrorCode.DB_ERROR, str(exc)) - - def query(self, sql: str, params: Optional[tuple] = None) -> Result[List[Dict[str, Any]]]: - """Execute a SELECT query and return rows (sync).""" - return self.execute(sql, params, fetch=True) - async def aexecute(self, query: str, params: Optional[tuple] = None, fetch: bool = False) -> Result[Any]: """Execute SQL on the DB loop thread (async).""" token = _TX_TOKEN.get() @@ -1032,44 +1012,6 @@ class Sqlite: """Execute a SELECT query and return rows (async).""" return await self.aexecute(sql, params, fetch=True) - def query_in( - self, - base_query: str, - column: str, - values: List[Any], - additional_params: Optional[tuple] = None, - ) -> Result[List[Dict[str, Any]]]: - """ - Execute a query with an IN clause safely. - - `base_query` must contain the `{IN_CLAUSE}` placeholder, which will be replaced - with ` IN (?, ?, ...)` using parameter binding for values. - """ - if not values: - return Result.Ok([]) - - if not isinstance(values, (list, tuple)): - return Result.Err(ErrorCode.INVALID_INPUT, "values must be a list or tuple") - - ok_col, safe_col = _validate_and_repair_column_name(column) - if not ok_col or not safe_col: - return Result.Err(ErrorCode.INVALID_INPUT, f"Invalid column name: {column}") - - ok_tpl, why = _validate_in_base_query(base_query) - if not ok_tpl: - return Result.Err(ErrorCode.INVALID_INPUT, why or "Invalid base_query template") - - ok_q, query_or_err, _ = _build_in_query(str(base_query), safe_col, len(values)) - if not ok_q: - return Result.Err(ErrorCode.INVALID_INPUT, str(query_or_err)) - query = query_or_err - - params = tuple(values) - if additional_params: - params = params + tuple(additional_params) - - return self.query(query, params) - async def aquery_in( self, base_query: str, @@ -1185,14 +1127,6 @@ class Sqlite: except Exception: raise - def executemany(self, query: str, params_list: List[Tuple]) -> Result[int]: - """Execute a parameterized statement over multiple param tuples (sync).""" - try: - token = self._tx_token() - return self._loop_thread.run(self._executemany_async(query, params_list, tx_token=token)) - except Exception as exc: - return Result.Err(ErrorCode.DB_ERROR, str(exc)) - async def aexecutemany(self, query: str, params_list: List[Tuple]) -> Result[int]: """Execute a parameterized statement over multiple param tuples (async).""" token = _TX_TOKEN.get() @@ -1243,14 +1177,6 @@ class Sqlite: return Result.Err(ErrorCode.DB_ERROR, str(exc)) return await self._with_query_timeout(_execute_inner()) - def executescript(self, script: str) -> Result[bool]: - """Execute a multi-statement SQL script (sync).""" - try: - token = self._tx_token() - return self._loop_thread.run(self._executescript_async(script, tx_token=token)) - except Exception as exc: - return Result.Err(ErrorCode.DB_ERROR, str(exc)) - async def aexecutescript(self, script: str) -> Result[bool]: """Execute a multi-statement SQL script (async).""" token = _TX_TOKEN.get() @@ -1271,27 +1197,11 @@ class Sqlite: finally: await self._release_connection_async(conn) - def vacuum(self) -> Result[bool]: - """Run SQLite VACUUM (sync).""" - try: - return self._loop_thread.run(self._vacuum_async()) - except Exception as exc: - return Result.Err(ErrorCode.DB_ERROR, str(exc)) - async def avacuum(self) -> Result[bool]: """Run SQLite VACUUM (async).""" fut = self._loop_thread.submit(self._vacuum_async()) return await asyncio.wrap_future(fut) - def has_table(self, table_name: str) -> bool: - """Return True if `table_name` exists in sqlite_master.""" - result = self.execute( - "SELECT name FROM sqlite_master WHERE type='table' AND name = ?", - (table_name,), - fetch=True, - ) - return bool(result.ok and result.data and len(result.data) > 0) - async def ahas_table(self, table_name: str) -> bool: """Return True if `table_name` exists in sqlite_master (async).""" result = await self.aexecute( @@ -1301,23 +1211,6 @@ class Sqlite: ) return bool(result.ok and result.data and len(result.data) > 0) - def get_schema_version(self) -> int: - """Get the schema version from the `metadata` table (0 if missing).""" - if not self.has_table("metadata"): - return 0 - - result = self.execute( - "SELECT value FROM metadata WHERE key = 'schema_version'", - fetch=True, - ) - if result.ok and result.data and len(result.data) > 0: - try: - return int(result.data[0]["value"]) - except (ValueError, KeyError, TypeError): - logger.warning("Invalid schema_version value in database") - return 0 - return 0 - async def aget_schema_version(self) -> int: """Get the schema version from the `metadata` table (0 if missing) (async).""" if not await self.ahas_table("metadata"): @@ -1335,13 +1228,6 @@ class Sqlite: return 0 return 0 - def set_schema_version(self, version: int) -> Result[bool]: - """Set the schema version in the `metadata` table.""" - return self.execute( - "INSERT OR REPLACE INTO metadata (key, value) VALUES ('schema_version', ?)", - (str(version),), - ) - async def aset_schema_version(self, version: int) -> Result[bool]: """Set the schema version in the `metadata` table (async).""" return await self.aexecute( @@ -1470,9 +1356,6 @@ class Sqlite: except Exception: pass - # NOTE: `transaction()` is defined later (sync back-compat). Keeping a single - # implementation avoids subtle bugs (e.g. missing `_tx_local`). - async def _close_all_async(self): # 1. Close transaction connections first. tokens = list(self._tx_conns.keys()) @@ -1509,15 +1392,6 @@ class Sqlite: # Reset counters/semaphores as everything is closed self._async_sem = None - def close(self): - """Close connections and stop the DB loop thread (sync).""" - try: - self._loop_thread.run(self._close_all_async()) - except Exception as exc: - logger.debug("DB close failed: %s", exc) - finally: - self._loop_thread.stop() - async def aclose(self): """Close connections and stop the DB loop thread (async).""" fut = self._loop_thread.submit(self._close_all_async()) @@ -1530,6 +1404,13 @@ class Sqlite: """Prepare for reset: lock new connections, wait for active ones, close all.""" self._resetting = True + # Best effort: force WAL checkpoint while connections are still alive. + # This reduces the chance of carrying stale WAL state into a reset on Windows. + try: + await self._checkpoint_wal_before_reset_async() + except Exception as exc: + logger.debug("DB Reset: WAL checkpoint pre-drain failed: %s", exc) + # Wait deterministically for in-flight checkouts to drain (max 5s), then force close. drained = await asyncio.to_thread(self._active_conns_idle.wait, 5.0) if not drained and self._active_conns: @@ -1537,6 +1418,47 @@ class Sqlite: await self._close_all_async() + async def _checkpoint_wal_before_reset_async(self) -> None: + """ + Best-effort checkpoint prior to hard reset. + Must be called on DB loop thread while connections are still valid. + """ + await self._ensure_initialized_async() + conn = await self._acquire_connection_async() + try: + for attempt in range(self._lock_retry_attempts + 1): + try: + cur = await conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + try: + await cur.fetchall() + finally: + try: + await cur.close() + except Exception: + pass + break + except sqlite3.OperationalError as exc: + if self._is_locked_error(exc) and attempt < self._lock_retry_attempts: + await self._sleep_backoff(attempt) + continue + raise + finally: + await self._release_connection_async(conn) + + @staticmethod + def _schedule_delete_on_reboot(path: Path) -> bool: + """ + Windows fallback: request deletion on next reboot for locked files. + """ + if os.name != "nt": + return False + try: + MOVEFILE_DELAY_UNTIL_REBOOT = 0x4 + ok = ctypes.windll.kernel32.MoveFileExW(str(path), None, MOVEFILE_DELAY_UNTIL_REBOOT) + return bool(ok) + except Exception: + return False + async def areset(self) -> Result[bool]: """ Aggressive reset: @@ -1558,7 +1480,7 @@ class Sqlite: # 3. Delete files with RETRY logic base = str(self.db_path) - for ext in ["", "-wal", "-shm"]: + for ext in ["", "-wal", "-shm", "-journal"]: p = Path(base + ext) if not p.exists(): continue @@ -1587,8 +1509,18 @@ class Sqlite: logger.info(f"Renamed locked DB file to {trash}") deleted = True except Exception as rename_err: - # If main file fails, it's fatal. WAL/SHM failures are also fatal here - # because the user explicitly requested a hard reset. + # Final fallback (Windows): schedule delete at reboot. + # We still fail the reset to avoid reinitializing over potentially + # stale sidecar files in the current process/session. + scheduled = self._schedule_delete_on_reboot(p) + with self._lock: + self._resetting = False + if scheduled: + return Result.Err( + "DELETE_PENDING_REBOOT", + f"Could not delete {p} now; scheduled for deletion at reboot. " + f"Delete error: {last_error} | Rename failed: {rename_err}", + ) return Result.Err( "DELETE_FAILED", f"Could not delete {p}: {last_error} | Rename failed: {rename_err}", @@ -1630,39 +1562,6 @@ class Sqlite: fut = self._loop_thread.submit(self._rollback_tx_async(token)) return await asyncio.wrap_future(fut) - @contextmanager - def transaction(self, mode: str = "immediate"): - """Context manager for a DB transaction (sync).""" - # Back-compat sync transaction: routes token via thread-local. - tx_state = Result.Ok(True) - begin_res = self._loop_thread.run(self._begin_tx_async(mode)) - if not begin_res.ok or not begin_res.data: - tx_state = Result.Err(ErrorCode.DB_ERROR, str(begin_res.error or "Failed to begin transaction")) - yield tx_state - return - - token = str(begin_res.data) - self._tx_local.token = token - try: - yield tx_state - commit_res = self._loop_thread.run(self._commit_tx_async(token)) - if not commit_res.ok: - tx_state.ok = False - tx_state.code = str(getattr(commit_res, "code", None) or ErrorCode.DB_ERROR) - tx_state.error = str(commit_res.error or "Commit failed") - except Exception: - try: - self._loop_thread.run(self._rollback_tx_async(token)) - except Exception: - pass - raise - finally: - try: - if hasattr(self._tx_local, "token"): - del self._tx_local.token - except Exception: - pass - @asynccontextmanager async def atransaction(self, mode: str = "immediate"): """Async context manager for a DB transaction.""" diff --git a/mjr_am_backend/adapters/tools/ffprobe.py b/mjr_am_backend/adapters/tools/ffprobe.py index d6c8a627..16448d25 100644 --- a/mjr_am_backend/adapters/tools/ffprobe.py +++ b/mjr_am_backend/adapters/tools/ffprobe.py @@ -5,6 +5,7 @@ import subprocess import json import shutil import os +import asyncio from pathlib import Path from typing import Optional, List, Dict @@ -184,6 +185,98 @@ class FFProbe: quality="degraded" ) + async def aread(self, path: str) -> Result[dict]: + """ + Async variant of read() using native asyncio subprocess execution. + """ + if not self._available: + return Result.Err( + ErrorCode.TOOL_MISSING, + "ffprobe not found in PATH", + quality="none" + ) + + try: + cmd = [ + self._resolved_bin or self.bin, + "-v", "error", + "-print_format", "json", + "-show_format", + "-show_streams", + path, + ] + + process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + close_fds=os.name != "nt", + ) + + try: + stdout_b, stderr_b = await asyncio.wait_for(process.communicate(), timeout=self.timeout) + except asyncio.TimeoutError: + try: + process.kill() + except Exception: + pass + logger.error(f"ffprobe timeout for {path}") + return Result.Err( + ErrorCode.TIMEOUT, + f"ffprobe timeout after {self.timeout}s", + quality="degraded" + ) + + stdout = (stdout_b or b"").decode("utf-8", errors="replace") + stderr = (stderr_b or b"").decode("utf-8", errors="replace") + + if process.returncode != 0: + stderr_msg = stderr.strip() + logger.warning(f"ffprobe error for {path}: {stderr_msg}") + return Result.Err( + ErrorCode.FFPROBE_ERROR, + stderr_msg or "ffprobe command failed", + quality="degraded" + ) + + if not stdout.strip(): + logger.warning(f"ffprobe returned empty output for {path}") + return Result.Err( + ErrorCode.FFPROBE_ERROR, + "No ffprobe output", + quality="degraded" + ) + + data = json.loads(stdout) + if not isinstance(data, dict): + return Result.Err( + ErrorCode.PARSE_ERROR, + "Invalid ffprobe output format", + quality="degraded" + ) + + result = { + "format": data.get("format", {}), + "streams": data.get("streams", []), + "video_stream": self._find_video_stream(data.get("streams", [])), + "audio_stream": self._find_audio_stream(data.get("streams", [])), + } + return Result.Ok(result, quality="full") + except json.JSONDecodeError as e: + logger.error(f"ffprobe JSON parse error: {e}") + return Result.Err( + ErrorCode.PARSE_ERROR, + f"Failed to parse ffprobe output: {e}", + quality="degraded" + ) + except Exception as e: + logger.error(f"ffprobe unexpected error: {e}") + return Result.Err( + ErrorCode.FFPROBE_ERROR, + str(e), + quality="degraded" + ) + def read_batch(self, paths: List[str]) -> Dict[str, Result[dict]]: """ Read metadata from multiple video files. @@ -223,6 +316,31 @@ class FFProbe: return results + async def aread_batch(self, paths: List[str]) -> Dict[str, Result[dict]]: + """ + Async batch variant using bounded asyncio concurrency. + """ + if not self._available: + err: Result[dict] = Result.Err(ErrorCode.TOOL_MISSING, "ffprobe not found in PATH", quality="none") + return {str(p): err for p in paths} + if not paths: + return {} + + max_workers = min(self._max_workers, len(paths)) + sem = asyncio.Semaphore(max_workers) + results: Dict[str, Result[dict]] = {} + + async def _one(path: str): + async with sem: + try: + results[str(path)] = await self.aread(path) + except Exception as e: + logger.error(f"FFProbe async batch error for {path}: {e}") + results[str(path)] = Result.Err(ErrorCode.FFPROBE_ERROR, str(e), quality="degraded") + + await asyncio.gather(*[_one(p) for p in paths]) + return results + def _find_video_stream(self, streams: list) -> dict: """Find first video stream in streams list.""" for stream in streams: diff --git a/mjr_am_backend/features/index/scanner.py b/mjr_am_backend/features/index/scanner.py index 01b8a61f..932c592a 100644 --- a/mjr_am_backend/features/index/scanner.py +++ b/mjr_am_backend/features/index/scanner.py @@ -44,6 +44,11 @@ MAX_TRANSACTION_BATCH_SIZE = 500 MAX_SCAN_JOURNAL_LOOKUP = 5000 STAT_RETRY_COUNT = 3 STAT_RETRY_BASE_DELAY_S = 0.15 +try: + # 0 disables scan I/O throttling (default: current behavior). + SCAN_IOPS_LIMIT = float(os.getenv("MAJOOR_SCAN_IOPS_LIMIT", "0") or 0.0) +except Exception: + SCAN_IOPS_LIMIT = 0.0 # Extensions explicitly excluded from indexing _EXCLUDED_EXTENSIONS: set = {".psd", ".json", ".txt", ".csv", ".db", ".sqlite", ".log"} @@ -97,6 +102,25 @@ class IndexScanner: self._current_scan_id: Optional[str] = None self._batch_fallback_count = 0 self._batch_fallback_lock = threading.Lock() + # Throttle only directory walk I/O operations (scandir entry processing). + self._scan_iops_limit = max(0.0, float(SCAN_IOPS_LIMIT)) + self._scan_iops_next_ts = 0.0 + + def _scan_iops_wait(self) -> None: + """ + Best-effort I/O pacing for directory scans. + Runs in the walk producer thread to avoid blocking the event loop. + """ + limit = self._scan_iops_limit + if limit <= 0.0: + return + now = time.perf_counter() + next_ts = self._scan_iops_next_ts + if next_ts > now: + time.sleep(next_ts - now) + now = time.perf_counter() + step = 1.0 / limit if limit > 0.0 else 0.0 + self._scan_iops_next_ts = max(next_ts, now) + step @staticmethod def _drain_walk_queue(q: "Queue[Optional[Path]]", max_items: int) -> list[Optional[Path]]: @@ -122,6 +146,8 @@ class IndexScanner: def _walk_and_enqueue(self, dir_path: Path, recursive: bool, stop_event: threading.Event, q: "Queue[Optional[Path]]") -> None: """Producer running on executor: walks filesystem and pushes paths into queue.""" + # Reset pacing window for each full walk. + self._scan_iops_next_ts = 0.0 try: for fp in self._iter_files(dir_path, recursive): if stop_event.is_set(): @@ -1270,6 +1296,7 @@ class IndexScanner: try: with os.scandir(current) as it: for entry in it: + self._scan_iops_wait() try: if entry.is_dir(follow_symlinks=False): stack.append(Path(entry.path)) @@ -1294,6 +1321,7 @@ class IndexScanner: continue else: for item in directory.iterdir(): + self._scan_iops_wait() if item.is_file() and _is_supported(item): yield item @@ -1312,7 +1340,10 @@ class IndexScanner: fallback_count = int(self._batch_fallback_count) except Exception: fallback_count = 0 - return {"batch_fallbacks_total": fallback_count} + return { + "batch_fallbacks_total": fallback_count, + "scan_iops_limit": float(self._scan_iops_limit), + } def _chunk_file_batches(self, files: List[Path]) -> Iterable[List[Path]]: """Yield batches of files for bounded transactions.""" diff --git a/mjr_am_backend/features/metadata/service.py b/mjr_am_backend/features/metadata/service.py index f9ac6436..dc8c86eb 100644 --- a/mjr_am_backend/features/metadata/service.py +++ b/mjr_am_backend/features/metadata/service.py @@ -206,6 +206,24 @@ class MetadataService: max_concurrency = 1 self._extract_sem = asyncio.Semaphore(max_concurrency) + async def _exif_read(self, file_path: str) -> Result[Dict[str, Any]]: + aread = getattr(self.exiftool, "aread", None) + if callable(aread): + return await aread(file_path) + return await asyncio.to_thread(self.exiftool.read, file_path) + + async def _ffprobe_read(self, file_path: str) -> Result[Dict[str, Any]]: + aread = getattr(self.ffprobe, "aread", None) + if callable(aread): + return await aread(file_path) + return await asyncio.to_thread(self.ffprobe.read, file_path) + + async def _ffprobe_read_batch(self, paths: list[str]) -> Dict[str, Result[Dict[str, Any]]]: + aread_batch = getattr(self.ffprobe, "aread_batch", None) + if callable(aread_batch): + return await aread_batch(paths) + return await asyncio.to_thread(self.ffprobe.read_batch, paths) + async def _enrich_with_geninfo_async(self, combined: Dict[str, Any]) -> None: """Helper to parse geninfo from prompt/workflow in combined metadata (Worker Thread).""" prompt_graph = combined.get("prompt") @@ -355,7 +373,7 @@ class MetadataService: ext = os.path.splitext(file_path)[1].lower() exif_start = time.perf_counter() - exif_result = await asyncio.to_thread(self.exiftool.read, file_path) + exif_result = await self._exif_read(file_path) exif_duration = time.perf_counter() - exif_start exif_data = exif_result.data if exif_result.ok else None image_fallback_enabled, _ = await self._resolve_fallback_prefs() @@ -411,7 +429,7 @@ class MetadataService: image_fallback_enabled, _ = await self._resolve_fallback_prefs() if allow_exif: exif_start = time.perf_counter() - exif_result = await asyncio.to_thread(self.exiftool.read, file_path) + exif_result = await self._exif_read(file_path) exif_duration = time.perf_counter() - exif_start exif_data = exif_result.data if exif_result.ok else None if not exif_result.ok: @@ -479,7 +497,7 @@ class MetadataService: _, media_fallback_enabled = await self._resolve_fallback_prefs() if allow_exif: exif_start = time.perf_counter() - exif_result = await asyncio.to_thread(self.exiftool.read, file_path) + exif_result = await self._exif_read(file_path) exif_duration = time.perf_counter() - exif_start exif_data = exif_result.data if exif_result.ok else None if not exif_result.ok: @@ -500,7 +518,7 @@ class MetadataService: ffprobe_duration = 0.0 if allow_ffprobe: ffprobe_start = time.perf_counter() - ffprobe_result = await asyncio.to_thread(self.ffprobe.read, file_path) + ffprobe_result = await self._ffprobe_read(file_path) ffprobe_duration = time.perf_counter() - ffprobe_start if not ffprobe_result.ok: self._log_metadata_issue( @@ -593,7 +611,7 @@ class MetadataService: if allow_exif: exif_start = time.perf_counter() - exif_result = await asyncio.to_thread(self.exiftool.read, file_path) + exif_result = await self._exif_read(file_path) exif_duration = time.perf_counter() - exif_start exif_data = exif_result.data if exif_result.ok else None if not exif_result.ok: @@ -609,7 +627,7 @@ class MetadataService: if allow_ffprobe: ffprobe_start = time.perf_counter() - ffprobe_result = await asyncio.to_thread(self.ffprobe.read, file_path) + ffprobe_result = await self._ffprobe_read(file_path) ffprobe_duration = time.perf_counter() - ffprobe_start ffprobe_data = ffprobe_result.data if ffprobe_result.ok else None if not ffprobe_result.ok: @@ -736,7 +754,7 @@ class MetadataService: await self.exiftool.aread_batch(exif_targets) if exif_targets else {} ) ffprobe_results: Dict[str, Result[Dict[str, Any]]] = ( - await asyncio.to_thread(self.ffprobe.read_batch, ffprobe_targets) if ffprobe_targets else {} + await self._ffprobe_read_batch(ffprobe_targets) if ffprobe_targets else {} ) def _exif_for(path: str) -> Optional[Dict[str, Any]]: diff --git a/mjr_am_backend/routes/handlers/assets.py b/mjr_am_backend/routes/handlers/assets.py index faad24cb..14671587 100644 --- a/mjr_am_backend/routes/handlers/assets.py +++ b/mjr_am_backend/routes/handlers/assets.py @@ -87,6 +87,15 @@ def _validate_filename(name: str) -> tuple[bool, str]: return True, "" +def _is_resolved_path_allowed(path: Path) -> bool: + """ + Enforce allowed roots check on canonical paths. + """ + try: + return bool(_is_path_allowed(path) or _is_path_allowed_custom(path)) + except Exception: + return False + def _delete_file_best_effort(path: Path) -> Result[bool]: @@ -564,6 +573,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: resolved = candidate.resolve(strict=True) except Exception: return _json_response(Result.Err("NOT_FOUND", "File does not exist")) + if not _is_resolved_path_allowed(resolved): + return _json_response(Result.Err("FORBIDDEN", "Path is not within allowed roots")) def _execute_command(command: list[str]) -> None: if not shutil.which(command[0]): @@ -745,6 +756,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: resolved = candidate.resolve(strict=True) except Exception: return _json_response(Result.Err("NOT_FOUND", "File does not exist")) + if not _is_resolved_path_allowed(resolved): + return _json_response(Result.Err("FORBIDDEN", "Path is not within allowed roots")) if not (resolved.exists() and resolved.is_file()): return _json_response(Result.Err("NOT_FOUND", "File does not exist")) @@ -908,6 +921,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: current_resolved = current_path.resolve(strict=True) except Exception: return _json_response(Result.Err("NOT_FOUND", "Current file does not exist")) + if not _is_resolved_path_allowed(current_resolved): + return _json_response(Result.Err("FORBIDDEN", "Path is not within allowed roots")) if not current_resolved.exists() or not current_resolved.is_file(): return _json_response(Result.Err("NOT_FOUND", "Current file does not exist")) @@ -930,13 +945,22 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: ) ) + async def _rollback_physical_rename() -> None: + try: + if new_path.exists() and not current_resolved.exists(): + new_path.rename(current_resolved) + except Exception as rollback_exc: + logger.error("Failed to rollback rename for asset %s: %s", asset_id, rollback_exc) + # Update database record try: try: mtime = int(new_path.stat().st_mtime) except FileNotFoundError: + await _rollback_physical_rename() return _json_response(Result.Err("NOT_FOUND", "Renamed file does not exist")) except Exception as exc: + await _rollback_physical_rename() return _json_response( Result.Err( "FS_ERROR", @@ -947,7 +971,7 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: async with svc["db"].atransaction(mode="immediate"): defer_fk = await svc["db"].aexecute("PRAGMA defer_foreign_keys = ON") if not defer_fk.ok: - return _json_response(Result.Err("DB_ERROR", defer_fk.error or "Failed to defer foreign key checks")) + raise RuntimeError(defer_fk.error or "Failed to defer foreign key checks") if asset_id is not None: update_res = await svc["db"].aexecute( @@ -955,12 +979,14 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: (new_name, str(new_path), mtime, asset_id) ) if not update_res.ok: - return _json_response(Result.Err("DB_ERROR", update_res.error or "Failed to update assets filepath")) + raise RuntimeError(update_res.error or "Failed to update assets filepath") else: - await svc["db"].aexecute( + up2 = await svc["db"].aexecute( "UPDATE assets SET filename = ?, filepath = ?, mtime = ? WHERE filepath = ?", (new_name, str(new_path), mtime, str(current_resolved)), ) + if not up2.ok: + raise RuntimeError(up2.error or "Failed to update assets filepath") # Keep FK-linked tables in sync with renamed filepath. sj_res = await svc["db"].aexecute( @@ -968,20 +994,16 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: (str(new_path), str(current_resolved)), ) if not sj_res.ok: - return _json_response(Result.Err("DB_ERROR", sj_res.error or "Failed to update scan_journal filepath")) + raise RuntimeError(sj_res.error or "Failed to update scan_journal filepath") mc_res = await svc["db"].aexecute( "UPDATE metadata_cache SET filepath = ? WHERE filepath = ?", (str(new_path), str(current_resolved)), ) if not mc_res.ok: - return _json_response(Result.Err("DB_ERROR", mc_res.error or "Failed to update metadata_cache filepath")) + raise RuntimeError(mc_res.error or "Failed to update metadata_cache filepath") except Exception as exc: - try: - if new_path.exists() and not current_resolved.exists(): - new_path.rename(current_resolved) - except Exception as rollback_exc: - logger.error("Failed to rollback rename for asset %s: %s", asset_id, rollback_exc) + await _rollback_physical_rename() return _json_response( Result.Err( "DB_ERROR", @@ -1091,6 +1113,8 @@ def register_asset_routes(routes: web.RouteTableDef) -> None: resolved = candidate.resolve(strict=True) except Exception: resolved = None + if resolved is not None and not _is_resolved_path_allowed(resolved): + return _json_response(Result.Err("FORBIDDEN", f"Path for asset ID {asset_id} is not within allowed roots")) validated_assets.append({"id": int(asset_id), "filepath": str(raw_path), "resolved": resolved}) @@ -1202,6 +1226,8 @@ async def download_asset(request: web.Request) -> web.StreamResponse: resolved = candidate.resolve(strict=True) except (OSError, RuntimeError, ValueError): return web.Response(status=404, text="File not found") + if not _is_resolved_path_allowed(resolved): + return web.Response(status=403, text="Path is not within allowed roots") if not resolved.is_file(): return web.Response(status=404, text="File not found") diff --git a/mjr_am_backend/routes/handlers/db_maintenance.py b/mjr_am_backend/routes/handlers/db_maintenance.py index df3011b0..a1576d77 100644 --- a/mjr_am_backend/routes/handlers/db_maintenance.py +++ b/mjr_am_backend/routes/handlers/db_maintenance.py @@ -313,7 +313,7 @@ def register_db_maintenance_routes(routes: web.RouteTableDef) -> None: logger.warning("DB adapter areset() failed (%s), falling back to manual file delete", exc) # Close whatever we can before manual delete try: - db.close() + await db.aclose() except Exception: pass diff --git a/mjr_am_backend/routes/handlers/scan.py b/mjr_am_backend/routes/handlers/scan.py index 254d224f..c81b4da3 100644 --- a/mjr_am_backend/routes/handlers/scan.py +++ b/mjr_am_backend/routes/handlers/scan.py @@ -1327,7 +1327,7 @@ def register_scan_routes(routes: web.RouteTableDef) -> None: if scope == "all" and (clear_scan_journal or clear_metadata_cache): # 1. Vacuum DB to reclaim space and rebuild file try: - await asyncio.to_thread(db.execute, "VACUUM") + await db.avacuum() logger.info("Database VACUUM completed during index reset") except Exception as exc: logger.warning(f"Failed to VACUUM database: {exc}") diff --git a/tests/README.md b/tests/README.md index 2443eb6b..fb334acb 100644 --- a/tests/README.md +++ b/tests/README.md @@ -170,11 +170,12 @@ pytest tests/ --cov=backend --cov-report=html Each test should be independent using in-memory databases: ```python -def test_example(): +@pytest.mark.asyncio +async def test_example(): db = Sqlite(":memory:") - migrate_schema(db) + await migrate_schema(db) # ... test logic - db.close() + await db.aclose() ``` ### Result pattern diff --git a/tests/core/test_integration.py b/tests/core/test_integration.py index 09813f09..9ad82435 100644 --- a/tests/core/test_integration.py +++ b/tests/core/test_integration.py @@ -137,7 +137,7 @@ def main(): logger.info("=" * 60) # Cleanup - services["db"].close() + asyncio.run(services["db"].aclose()) test_db = Path("test_integration.db") if test_db.exists(): test_db.unlink() diff --git a/tests/regressions/test_scan_batching.py b/tests/regressions/test_scan_batching.py index 5ba7aae1..167adfbd 100644 --- a/tests/regressions/test_scan_batching.py +++ b/tests/regressions/test_scan_batching.py @@ -8,7 +8,7 @@ async def test_scan_uses_batched_transactions(tmp_path: Path): """ Large scans should not BEGIN/COMMIT per file. - We count `db.transaction()` invocations during a scan and ensure it's bounded + We count `db.atransaction()` invocations during a scan and ensure it's bounded by the batch size logic (<< number of files). """ from mjr_am_backend.deps import build_services @@ -30,15 +30,15 @@ async def test_scan_uses_batched_transactions(tmp_path: Path): files.append(p) txn_calls = {"count": 0} - original_txn = db.transaction + original_txn = db.atransaction - @contextlib.contextmanager - def counted_transaction(*args, **kwargs): + @contextlib.asynccontextmanager + async def counted_transaction(*args, **kwargs): txn_calls["count"] += 1 - with original_txn(*args, **kwargs): - yield + async with original_txn(*args, **kwargs) as tx: + yield tx - db.transaction = counted_transaction # type: ignore[assignment] + db.atransaction = counted_transaction # type: ignore[assignment] try: res = await index.scan_directory( str(scan_dir), @@ -51,7 +51,7 @@ async def test_scan_uses_batched_transactions(tmp_path: Path): ) assert res.ok, res.error finally: - db.transaction = original_txn # type: ignore[assignment] + db.atransaction = original_txn # type: ignore[assignment] # With default batching for ~120 files we expect ~3 transactions (50-sized batches), # but allow a small buffer if batching logic changes slightly.