fix(eq2db): apply all 10 code-review findings from the catalogue arc

backend/eq2db/_catalogue.py

@@ -44,11 +44,21 @@
 from pathlib import Path
 from typing import ClassVar
 
+from backend.db_helpers import like_escape
 from backend.eq2db import _meta as _meta_db
 
 _log = logging.getLogger(__name__)
 
 
+def _is_unbuilt_schema(exc: sqlite3.OperationalError) -> bool:
+    """True for the "DB file exists but the schema hasn't been created yet"
+    class of errors the read helpers degrade gracefully on (fresh volume,
+    pre-seeded stub file). Anything else — locked database, disk I/O, SQL
+    syntax — is a real fault and must propagate."""
+    msg = str(exc)
+    return "no such table" in msg or "no such column" in msg
+
+
 class BaseCatalogue:
     """Read (and build) access to one eq2db SQLite file."""
 
@@ -138,51 +148,79 @@ def _fetchall(self, sql: str, params: Sequence | Mapping = ()) -> list[sqlite3.R
         """Run one read query with Row factory; the connection is opened and
         closed per call. Returns [] when the DB file is missing or the table
         isn't built yet — eq2db read paths degrade gracefully on an
-        unprovisioned DB rather than 500."""
+        unprovisioned DB rather than 500. Other OperationalErrors (locked DB,
+        disk I/O) propagate: a transient failure must surface as an error,
+        not be served — and possibly cached — as an empty result."""
         if not self.path.exists():
             return []
         conn = sqlite3.connect(self.path)
         conn.row_factory = sqlite3.Row
         try:
             return conn.execute(sql, params).fetchall()
-        except sqlite3.OperationalError:
-            _log.exception("[eq2db] read failed on %r (unbuilt db?)", self)
+        except sqlite3.OperationalError as exc:
+            if not _is_unbuilt_schema(exc):
+                raise
+            _log.warning("[eq2db] read on unbuilt db %r: %s", self, exc)
             return []
         finally:
             conn.close()
 
     def _fetchone(self, sql: str, params: Sequence | Mapping = ()) -> sqlite3.Row | None:
         """Single-row variant of :meth:`_fetchall`. None on missing DB,
-        unbuilt table, or no match."""
+        unbuilt table, or no match; other OperationalErrors propagate."""
         if not self.path.exists():
             return None
         conn = sqlite3.connect(self.path)
         conn.row_factory = sqlite3.Row
         try:
             return conn.execute(sql, params).fetchone()
-        except sqlite3.OperationalError:
-            _log.exception("[eq2db] read failed on %r (unbuilt db?)", self)
+        except sqlite3.OperationalError as exc:
+            if not _is_unbuilt_schema(exc):
+                raise
+            _log.warning("[eq2db] read on unbuilt db %r: %s", self, exc)
             return None
         finally:
             conn.close()
 
+    def _find_exact_then_like(self, exact_sql: str, like_sql: str, name: str) -> list[sqlite3.Row]:
+        """The shared name-search protocol: exact lowercased match first, then
+        a LIKE fallback with user wildcards escaped (BE-006) — both queries on
+        ONE connection. ``exact_sql`` takes the lowercased name; ``like_sql``
+        takes the escaped %-wrapped pattern with ``ESCAPE '\\'`` semantics."""
+        if not self.path.exists():
+            return []
+        conn = sqlite3.connect(self.path)
+        conn.row_factory = sqlite3.Row
+        try:
+            rows = conn.execute(exact_sql, (name.lower(),)).fetchall()
+            if not rows:
+                rows = conn.execute(like_sql, (f"%{like_escape(name.lower())}%",)).fetchall()
+            return rows
+        except sqlite3.OperationalError as exc:
+            if not _is_unbuilt_schema(exc):
+                raise
+            _log.warning("[eq2db] read on unbuilt db %r: %s", self, exc)
+            return []
+        finally:
+            conn.close()
+
     # ── DB lifecycle ─────────────────────────────────────────────────────────
 
     def init_db(self) -> sqlite3.Connection:
         """Create tables/indexes if missing. Returns an open connection.
 
         Template method: the connection preamble and commit live here;
         the module-specific schema comes from ``_create_schema`` and any
-        post-commit backfills from ``_post_init``. ``:memory:`` is
-        supported for tests (skips mkdir + WAL).
+        post-commit backfills from ``_post_init``.
+
+        ``:memory:`` is deliberately NOT supported: every read helper opens
+        its own connection against ``self.path``, so a memory DB would be a
+        fresh empty database per read — tests use a tmp_path file instead.
         """
         _log.debug("[eq2db] init_db %r", self)
-        if str(self.path) == ":memory:":
-            conn = sqlite3.connect(":memory:")
-        else:
-            self.path.parent.mkdir(parents=True, exist_ok=True)
-            conn = sqlite3.connect(self.path)
-            conn.execute("PRAGMA journal_mode = WAL;")
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        conn = sqlite3.connect(self.path)
+        conn.execute("PRAGMA journal_mode = WAL;")
         conn.execute("PRAGMA synchronous = NORMAL;")
         if self.FOREIGN_KEYS:
             conn.execute("PRAGMA foreign_keys = ON;")

backend/eq2db/items.py

@@ -22,6 +22,8 @@
 
 import aiosqlite
 
+from backend.census._coerce import coerce_int as _coerce_int
+from backend.census._coerce import coerce_str_or_none as _coerce_str
 from backend.census.item_level import compute_ilvl
 from backend.db_helpers import like_escape, resolve_db_path
 from backend.eq2db._catalogue import BaseCatalogue
@@ -135,30 +137,18 @@ def _flag(flags: dict, key: str) -> int:
 
 
 def _str_field(item: dict, key: str) -> str | None:
-    v = item.get(key)
-    if v is None or isinstance(v, dict):
-        return None
-    s = str(v).strip()
-    return s if s else None
+    """Stripped string or None — shared Census coercion semantics
+    (backend.census._coerce), applied to a dict field."""
+    return _coerce_str(item.get(key))
 
 
 def _int_field(v: Any) -> int | None:
-    if v is None:
-        return None
-    try:
-        return int(v) or None  # treat 0 as NULL for quest IDs etc.
-    except (ValueError, TypeError):
-        return None
+    """coerce_int with 0 treated as NULL (quest IDs etc.)."""
+    return _coerce_int(v) or None
 
 
-def _int_field_zero(v: Any) -> int | None:
-    """Like _int_field but keeps 0."""
-    if v is None:
-        return None
-    try:
-        return int(v)
-    except (ValueError, TypeError):
-        return None
+# Keeps 0 — exactly the shared coercer.
+_int_field_zero = _coerce_int
 
 
 class GearRow(NamedTuple):
@@ -505,7 +495,7 @@ def _backfill_effect_stats(conn: sqlite3.Connection) -> None:
             try:
                 raw = json.loads(raw_json_str)
             except Exception as exc:
-                _log.warning("[db] Failed to parse effect_stats JSON for item_id=%s: %s", item_id, exc)
+                _log.warning("[items-db] Failed to parse effect_stats JSON for item_id=%s: %s", item_id, exc)
                 continue
             for stat_name, value in ItemCatalogue.extract_effect_stats(raw).items():
                 stat_rows.append((item_id, stat_name, value))

backend/eq2db/recipes.py

@@ -36,7 +36,7 @@
 from typing import TypedDict, cast
 
 from backend.census._coerce import coerce_int as _int
-from backend.db_helpers import like_escape, resolve_db_path
+from backend.db_helpers import resolve_db_path
 from backend.eq2db._catalogue import BaseCatalogue
 from backend.sql_loader import load_sql
 
@@ -133,7 +133,7 @@ def _row_to_dict(row: sqlite3.Row) -> RecipeRow:
     try:
         d["secondary_comps"] = json.loads(d.get("secondary_comps") or "[]")
     except Exception as exc:
-        _log.warning("[recipes_db] Failed to parse secondary_comps for recipe id=%s: %s", d.get("id"), exc)
+        _log.warning("[recipes-db] Failed to parse secondary_comps for recipe id=%s: %s", d.get("id"), exc)
         d["secondary_comps"] = []
     return cast(RecipeRow, d)
 
@@ -293,13 +293,11 @@ def find_by_id(self, recipe_id: int) -> RecipeRow | None:
 
     def find_by_name(self, name: str) -> list[RecipeRow]:
         """Return recipes whose name matches (exact then LIKE), ordered by name."""
-        rows = self._fetchall(_SQL["find_by_name_exact"].format(cols=_SELECT_COLS), (name.lower(),))
-        if not rows:
-            # LIKE fallback — escape user wildcards (BE-006).
-            rows = self._fetchall(
-                _SQL["find_by_name_like"].format(cols=_SELECT_COLS),
-                (f"%{like_escape(name.lower())}%",),
-            )
+        rows = self._find_exact_then_like(
+            _SQL["find_by_name_exact"].format(cols=_SELECT_COLS),
+            _SQL["find_by_name_like"].format(cols=_SELECT_COLS),
+            name,
+        )
         return [_row_to_dict(r) for r in rows]
 
     def find_by_spell(self, spell_name: str, tier: str) -> list[RecipeRow]:

backend/eq2db/spells.py

@@ -34,7 +34,7 @@
 from backend.census._coerce import coerce_float as _float
 from backend.census._coerce import coerce_int as _int
 from backend.census._coerce import coerce_str_or_none as _str
-from backend.db_helpers import like_escape, resolve_db_path
+from backend.db_helpers import resolve_db_path
 from backend.eq2db._catalogue import BaseCatalogue
 from backend.sql_loader import load_sql
 
@@ -162,6 +162,11 @@ class SpellCatalogue(BaseCatalogue):
     changed; stale CRC lookups would lie).
     """
 
+    # Cache bound — parity with the pre-catalogue @lru_cache(maxsize=4096).
+    # The route feeding this (GET /aa/spell/{crc}?tier=N) takes arbitrary
+    # client ints, so an unbounded dict would grow until process restart.
+    _CRC_CACHE_MAX = 4096
+
     def __init__(self, path: Path = DB_PATH) -> None:
         super().__init__(path)
         self._crc_cache: dict[tuple[int, int | None], SpellRow | None] = {}
@@ -252,7 +257,7 @@ def load_blocklist(path: Path = _BLOCKLIST_PATH) -> Blocklist:
 
             return Blocklist(frozenset(exact), patterns)
         except Exception as exc:
-            _log.warning("[spells_db] Failed to load blocklist: %s", exc)
+            _log.warning("[spells-db] Failed to load blocklist: %s", exc)
             return Blocklist(frozenset(), [])
 
     @staticmethod
@@ -282,7 +287,7 @@ def _parse_effects(spell: dict) -> str:
             return "[]"
         if not isinstance(raw, list):
             _log.warning(
-                "[spells_db] effect_list for spell %s has unexpected shape %s — returning empty",
+                "[spells-db] effect_list for spell %s has unexpected shape %s — returning empty",
                 spell.get("id"),
                 type(raw).__name__,
             )
@@ -414,18 +419,20 @@ def find_by_crc(self, crc: int, tier: int | None = None) -> SpellRow | None:
             # Fallback: highest available tier
             row = self._fetchone(_SQL["find_by_crc_highest_tier"].format(cols=_SELECT_COLS), (crc,))
         result = _row_to_dict(row) if row else None
+        if len(self._crc_cache) >= self._CRC_CACHE_MAX:
+            # FIFO eviction (dicts preserve insertion order) — cheap and
+            # good enough for a cache that upsert_spells fully clears anyway.
+            self._crc_cache.pop(next(iter(self._crc_cache)))
         self._crc_cache[key] = result
         return result
 
     def find_by_name(self, name: str) -> list[SpellRow]:
         """Return all spell rows whose name matches (exact, then LIKE). Ordered by level."""
-        rows = self._fetchall(_SQL["find_by_name_exact"].format(cols=_SELECT_COLS), (name.lower(),))
-        if not rows:
-            # LIKE fallback — escape user wildcards (BE-006).
-            rows = self._fetchall(
-                _SQL["find_by_name_like"].format(cols=_SELECT_COLS),
-                (f"%{like_escape(name.lower())}%",),
-            )
+        rows = self._find_exact_then_like(
+            _SQL["find_by_name_exact"].format(cols=_SELECT_COLS),
+            _SQL["find_by_name_like"].format(cols=_SELECT_COLS),
+            name,
+        )
         return [_row_to_dict(r) for r in rows]
 
     def character_upgradeable_spells(self, spell_ids: list[int]) -> list[SpellRow]: