refactor(server): extract AudioEffectsMappingDB into lib/audio_effects_db.py (R3) (#831)

Move-only, same shape as the MetadataDB extraction. The core-owned song/tone ->
audio-effect-provider routing index leaves server.py for a flat lib/ module.
The class body is byte-identical; server.py reconstructs exactly from
origin/main minus the cut range plus the import-back and the call site.

server.py: 9,705 -> 9,433 lines.

The only non-verbatim change is the constructor seam: `__init__` takes
`config_dir` instead of reading the module-level CONFIG_DIR, so the module does
no IO at import (Principle V). The `audio_effect_mappings` singleton stays in
server.py -- no route, no test, and none of the `monkeypatch.setattr(server, ...)`
targets move. No import went dead.

Verified: pyflakes clean on the new module; no new undefined name in server.py;
pytest 2341 passed; eslint 0 errors; boot smoke drives the extracted DB
end-to-end (POST a mapping -> GET reads it back -> audio_effects.db lands in
CONFIG_DIR, proving the config_dir seam) and all three migrated plugins still
serve their src/ module graphs.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -8,6 +8,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Changed
+- **`AudioEffectsMappingDB` moved out of `server.py` into `lib/audio_effects_db.py`
+  (R3, move-only).** The core-owned song/tone → provider routing index follows
+  `MetadataDB` out of the host file, byte-identical apart from the same constructor
+  seam (`__init__` takes `config_dir`; `audio_effect_mappings = AudioEffectsMappingDB(CONFIG_DIR)`),
+  so the module does no IO at import. The singleton stays in `server.py`; no route,
+  no test and no `monkeypatch.setattr(server, …)` target moves. `server.py`:
+  **9,705 → 9,433 lines**.
 - **`MetadataDB` moved out of `server.py` into `lib/metadata_db.py` (R3, move-only).**
   The library metadata cache — the `MetadataDB` class (4,018 lines) plus the query
   helpers it owns (keyset paging cursors, the tuning grouping key, smart-arrangement

docs/size-exemptions.md

@@ -55,7 +55,8 @@ without a *signed* exemption" is unenforceable.
 ## Planned, NOT exempt (owned by split plans — listed so nothing falls between states)
 
 core `static/app.js` (11,852) · `static/highway.js` (4,168, whole file) · `server.py`
-(9,705 — was 14,037; ratcheted by the R3 `MetadataDB` extraction) ·
+(9,433 — was 14,037; ratcheted by the R3 `MetadataDB` + `AudioEffectsMappingDB`
+extractions) ·
 `lib/metadata_db.py` (4,373 — new in R3; the `MetadataDB` class alone is 4,018 lines
 and is a monolith in its own right, to be split per-table once the router train
 lands) · `static/v3/songs.js` (4,134) · `static/capabilities/audio-session.js`

lib/audio_effects_db.py

@@ -0,0 +1,287 @@
+"""Core-owned song/tone -> audio-effect-provider mapping index.
+
+Extracted verbatim from ``server.py`` (R3). ``server.py`` still owns the
+``audio_effect_mappings`` singleton; this module only supplies the class, so
+nothing here touches config paths at import time — the caller passes
+``config_dir`` in.
+"""
+
+import json
+import sqlite3
+import threading
+from pathlib import Path
+
+
+class AudioEffectsMappingDB:
+    """Core-owned public song/tone -> provider mapping index.
+
+    Providers own the preset/chain rows addressed by provider_ref. Core owns
+    the cross-provider routing index and the active mapping per song/tone.
+    """
+
+    def __init__(self, config_dir: Path):
+        config_dir.mkdir(parents=True, exist_ok=True)
+        self.db_path = str(config_dir / "audio_effects.db")
+        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        self.conn.execute("PRAGMA journal_mode=WAL")
+        self.conn.execute("PRAGMA foreign_keys=ON")
+        self.conn.execute("""
+            CREATE TABLE IF NOT EXISTS audio_effect_mappings (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                song_key TEXT NOT NULL,
+                filename TEXT NOT NULL DEFAULT '',
+                tone_key TEXT NOT NULL,
+                provider_id TEXT NOT NULL,
+                provider_ref TEXT NOT NULL,
+                label TEXT NOT NULL DEFAULT '',
+                source TEXT NOT NULL DEFAULT 'manual',
+                metadata_json TEXT NOT NULL DEFAULT '{}',
+                created_at TEXT NOT NULL DEFAULT (datetime('now')),
+                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+                UNIQUE(song_key, tone_key, provider_id)
+            )
+        """)
+        self.conn.execute("""
+            CREATE TABLE IF NOT EXISTS audio_effect_active_mappings (
+                song_key TEXT NOT NULL,
+                tone_key TEXT NOT NULL,
+                mapping_id INTEGER NOT NULL,
+                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+                PRIMARY KEY (song_key, tone_key),
+                FOREIGN KEY (mapping_id) REFERENCES audio_effect_mappings(id) ON DELETE CASCADE
+            )
+        """)
+        self.conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_audio_effect_mappings_provider "
+            "ON audio_effect_mappings(provider_id)"
+        )
+        self.conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_audio_effect_mappings_filename "
+            "ON audio_effect_mappings(filename)"
+        )
+        self.conn.commit()
+        self._lock = threading.Lock()
+
+    @staticmethod
+    def _text(value, *, field: str, limit: int, allow_empty: bool = False) -> str:
+        if value is None:
+            text = ""
+        elif not isinstance(value, str):
+            raise ValueError(f"{field} must be a string")
+        else:
+            text = value.strip()
+        if not text and not allow_empty:
+            raise ValueError(f"{field} is required")
+        if len(text) > limit:
+            raise ValueError(f"{field} is too long")
+        return text
+
+    @staticmethod
+    def _mapping_id(value) -> int | None:
+        # Bind only values SQLite can store as an INTEGER; an out-of-range id is a
+        # clean miss (404), not a 500 at bind time.
+        if isinstance(value, int) and not isinstance(value, bool) and -(2 ** 63) <= value < 2 ** 63:
+            return value
+        return None
+
+    @staticmethod
+    def _field(data: dict, *keys):
+        # Select the first present snake/camel alias by key, not by truthiness, so a
+        # falsey non-string value (false/0) still reaches _text() and is rejected
+        # instead of being silently swallowed by an `or` chain.
+        for key in keys:
+            if key in data:
+                return data[key]
+        return None
+
+    @staticmethod
+    def _metadata(value) -> str:
+        if value is None:
+            return "{}"
+        if not isinstance(value, dict):
+            raise ValueError("metadata must be an object")
+        encoded = json.dumps(value, ensure_ascii=True, sort_keys=True)
+        if len(encoded) > 8192:
+            raise ValueError("metadata is too large")
+        return encoded
+
+    @staticmethod
+    def _row(row) -> dict | None:
+        if row is None:
+            return None
+        metadata = {}
+        try:
+            metadata = json.loads(row[8]) if row[8] else {}
+        except Exception:
+            metadata = {}
+        return {
+            "id": int(row[0]),
+            "song_key": row[1],
+            "filename": row[2] or "",
+            "tone_key": row[3],
+            "provider_id": row[4],
+            "provider_ref": row[5],
+            "label": row[6] or "",
+            "source": row[7] or "manual",
+            "metadata": metadata if isinstance(metadata, dict) else {},
+            "created_at": row[9] or "",
+            "updated_at": row[10] or "",
+            "active": bool(row[11]),
+        }
+
+    def _select_sql(self) -> str:
+        return """
+            SELECT m.id, m.song_key, m.filename, m.tone_key, m.provider_id,
+                   m.provider_ref, m.label, m.source, m.metadata_json,
+                   m.created_at, m.updated_at,
+                   CASE WHEN a.mapping_id IS NULL THEN 0 ELSE 1 END AS active
+            FROM audio_effect_mappings m
+            LEFT JOIN audio_effect_active_mappings a
+              ON a.song_key = m.song_key AND a.tone_key = m.tone_key AND a.mapping_id = m.id
+        """
+
+    def list(self, *, song_key: str = "", filename: str = "", tone_key: str = "", provider_id: str = "") -> list[dict]:
+        clauses: list[str] = []
+        params: list[str] = []
+        song_key = self._text(song_key, field="song_key", limit=240, allow_empty=True)
+        filename = self._text(filename, field="filename", limit=500, allow_empty=True)
+        tone_key = self._text(tone_key, field="tone_key", limit=160, allow_empty=True)
+        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
+        if song_key and filename:
+            clauses.append("(m.song_key = ? OR m.filename = ?)")
+            params.extend([song_key, filename])
+        elif song_key:
+            clauses.append("m.song_key = ?")
+            params.append(song_key)
+        elif filename:
+            clauses.append("(m.song_key = ? OR m.filename = ?)")
+            params.extend([filename, filename])
+        if tone_key:
+            clauses.append("m.tone_key = ?")
+            params.append(tone_key)
+        if provider_id:
+            clauses.append("m.provider_id = ?")
+            params.append(provider_id)
+        sql = self._select_sql()
+        if clauses:
+            sql += " WHERE " + " AND ".join(clauses)
+        sql += " ORDER BY m.song_key COLLATE NOCASE, m.tone_key COLLATE NOCASE, m.provider_id COLLATE NOCASE"
+        with self._lock:
+            rows = self.conn.execute(sql, params).fetchall()
+        return [self._row(row) for row in rows]
+
+    def get(self, mapping_id: int) -> dict | None:
+        mapping_id = self._mapping_id(mapping_id)
+        if mapping_id is None:
+            return None
+        with self._lock:
+            row = self.conn.execute(self._select_sql() + " WHERE m.id = ?", (mapping_id,)).fetchone()
+        return self._row(row)
+
+    def upsert(self, data: dict) -> dict:
+        if not isinstance(data, dict):
+            raise ValueError("mapping body must be an object")
+        filename = self._text(data.get("filename", ""), field="filename", limit=500, allow_empty=True)
+        song_key_raw = self._field(data, "song_key", "songKey")
+        if song_key_raw is None or song_key_raw == "":
+            song_key_raw = filename
+        song_key = self._text(song_key_raw, field="song_key", limit=240)
+        tone_key = self._text(self._field(data, "tone_key", "toneKey"), field="tone_key", limit=160, allow_empty=True)
+        provider_id = self._text(self._field(data, "provider_id", "providerId"), field="provider_id", limit=96)
+        provider_ref = self._text(self._field(data, "provider_ref", "providerRef"), field="provider_ref", limit=240)
+        label = self._text(data.get("label", ""), field="label", limit=160, allow_empty=True)
+        source = self._text(data.get("source", "manual"), field="source", limit=40, allow_empty=True) or "manual"
+        metadata_json = self._metadata(data.get("metadata", {}))
+        with self._lock:
+            self.conn.execute(
+                """
+                INSERT INTO audio_effect_mappings
+                    (song_key, filename, tone_key, provider_id, provider_ref, label, source, metadata_json, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
+                ON CONFLICT(song_key, tone_key, provider_id) DO UPDATE SET
+                    -- Only overwrite filename when a non-empty one was supplied; an
+                    -- omitted/empty filename must preserve the stored value (it's an
+                    -- alternate lookup key for list(..., filename=...)).
+                    filename=CASE WHEN excluded.filename <> '' THEN excluded.filename ELSE audio_effect_mappings.filename END,
+                    provider_ref=excluded.provider_ref,
+                    label=excluded.label,
+                    source=excluded.source,
+                    metadata_json=excluded.metadata_json,
+                    updated_at=datetime('now')
+                """,
+                (song_key, filename, tone_key, provider_id, provider_ref, label, source, metadata_json),
+            )
+            row = self.conn.execute(
+                "SELECT id FROM audio_effect_mappings WHERE song_key = ? AND tone_key = ? AND provider_id = ?",
+                (song_key, tone_key, provider_id),
+            ).fetchone()
+            if row is None:
+                raise ValueError("failed to create audio-effects mapping")
+            mapping_id = int(row[0])
+            if data.get("active") is True:
+                self.conn.execute(
+                    """
+                    INSERT INTO audio_effect_active_mappings (song_key, tone_key, mapping_id, updated_at)
+                    VALUES (?, ?, ?, datetime('now'))
+                    ON CONFLICT(song_key, tone_key) DO UPDATE SET
+                        mapping_id=excluded.mapping_id,
+                        updated_at=datetime('now')
+                    """,
+                    (song_key, tone_key, mapping_id),
+                )
+            self.conn.commit()
+        return self.get(mapping_id)
+
+    def delete(self, mapping_id: int, *, provider_id: str = "") -> bool:
+        mapping_id = self._mapping_id(mapping_id)
+        if mapping_id is None:
+            return False
+        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
+        with self._lock:
+            if provider_id:
+                cur = self.conn.execute(
+                    "DELETE FROM audio_effect_mappings WHERE id = ? AND provider_id = ?",
+                    (mapping_id, provider_id),
+                )
+            else:
+                cur = self.conn.execute("DELETE FROM audio_effect_mappings WHERE id = ?", (mapping_id,))
+            self.conn.commit()
+            return cur.rowcount > 0
+
+    def activate(self, mapping_id: int, *, provider_id: str = "") -> dict | None:
+        mapping_id = self._mapping_id(mapping_id)
+        if mapping_id is None:
+            return None
+        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
+        with self._lock:
+            row = self.conn.execute(
+                self._select_sql() + " WHERE m.id = ?",
+                (mapping_id,),
+            ).fetchone()
+            mapping = self._row(row)
+            if not mapping or (provider_id and mapping["provider_id"] != provider_id):
+                return None
+            self.conn.execute(
+                """
+                INSERT INTO audio_effect_active_mappings (song_key, tone_key, mapping_id, updated_at)
+                VALUES (?, ?, ?, datetime('now'))
+                ON CONFLICT(song_key, tone_key) DO UPDATE SET
+                    mapping_id=excluded.mapping_id,
+                    updated_at=datetime('now')
+                """,
+                (mapping["song_key"], mapping["tone_key"], mapping_id),
+            )
+            self.conn.commit()
+            selected = self.conn.execute(self._select_sql() + " WHERE m.id = ?", (mapping_id,)).fetchone()
+        return self._row(selected)
+
+    def clear_active(self, *, song_key: str, tone_key: str) -> bool:
+        song_key = self._text(song_key, field="song_key", limit=240)
+        tone_key = self._text(tone_key, field="tone_key", limit=160, allow_empty=True)
+        with self._lock:
+            cur = self.conn.execute(
+                "DELETE FROM audio_effect_active_mappings WHERE song_key = ? AND tone_key = ?",
+                (song_key, tone_key),
+            )
+            self.conn.commit()
+            return cur.rowcount > 0

server.py

@@ -63,6 +63,9 @@ from metadata_db import (
     _tuning_group_key_sql,
     next_library_cursor,
 )
+# The audio-effect routing index. Same shape as metadata_db: the class lives in
+# its own module, the `audio_effect_mappings` singleton below stays here.
+from audio_effects_db import AudioEffectsMappingDB
 import sloppak as sloppak_mod
 import drums as drums_mod
 import notation as notation_mod
@@ -347,283 +350,8 @@ def _env_flag(name: str) -> bool:
 _TUNING_GROUP_KEY_SQL = _tuning_group_key_sql("songs")
 
 
-class AudioEffectsMappingDB:
-    """Core-owned public song/tone -> provider mapping index.
-
-    Providers own the preset/chain rows addressed by provider_ref. Core owns
-    the cross-provider routing index and the active mapping per song/tone.
-    """
-
-    def __init__(self):
-        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
-        self.db_path = str(CONFIG_DIR / "audio_effects.db")
-        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
-        self.conn.execute("PRAGMA journal_mode=WAL")
-        self.conn.execute("PRAGMA foreign_keys=ON")
-        self.conn.execute("""
-            CREATE TABLE IF NOT EXISTS audio_effect_mappings (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                song_key TEXT NOT NULL,
-                filename TEXT NOT NULL DEFAULT '',
-                tone_key TEXT NOT NULL,
-                provider_id TEXT NOT NULL,
-                provider_ref TEXT NOT NULL,
-                label TEXT NOT NULL DEFAULT '',
-                source TEXT NOT NULL DEFAULT 'manual',
-                metadata_json TEXT NOT NULL DEFAULT '{}',
-                created_at TEXT NOT NULL DEFAULT (datetime('now')),
-                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
-                UNIQUE(song_key, tone_key, provider_id)
-            )
-        """)
-        self.conn.execute("""
-            CREATE TABLE IF NOT EXISTS audio_effect_active_mappings (
-                song_key TEXT NOT NULL,
-                tone_key TEXT NOT NULL,
-                mapping_id INTEGER NOT NULL,
-                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
-                PRIMARY KEY (song_key, tone_key),
-                FOREIGN KEY (mapping_id) REFERENCES audio_effect_mappings(id) ON DELETE CASCADE
-            )
-        """)
-        self.conn.execute(
-            "CREATE INDEX IF NOT EXISTS idx_audio_effect_mappings_provider "
-            "ON audio_effect_mappings(provider_id)"
-        )
-        self.conn.execute(
-            "CREATE INDEX IF NOT EXISTS idx_audio_effect_mappings_filename "
-            "ON audio_effect_mappings(filename)"
-        )
-        self.conn.commit()
-        self._lock = threading.Lock()
-
-    @staticmethod
-    def _text(value, *, field: str, limit: int, allow_empty: bool = False) -> str:
-        if value is None:
-            text = ""
-        elif not isinstance(value, str):
-            raise ValueError(f"{field} must be a string")
-        else:
-            text = value.strip()
-        if not text and not allow_empty:
-            raise ValueError(f"{field} is required")
-        if len(text) > limit:
-            raise ValueError(f"{field} is too long")
-        return text
-
-    @staticmethod
-    def _mapping_id(value) -> int | None:
-        # Bind only values SQLite can store as an INTEGER; an out-of-range id is a
-        # clean miss (404), not a 500 at bind time.
-        if isinstance(value, int) and not isinstance(value, bool) and -(2 ** 63) <= value < 2 ** 63:
-            return value
-        return None
-
-    @staticmethod
-    def _field(data: dict, *keys):
-        # Select the first present snake/camel alias by key, not by truthiness, so a
-        # falsey non-string value (false/0) still reaches _text() and is rejected
-        # instead of being silently swallowed by an `or` chain.
-        for key in keys:
-            if key in data:
-                return data[key]
-        return None
-
-    @staticmethod
-    def _metadata(value) -> str:
-        if value is None:
-            return "{}"
-        if not isinstance(value, dict):
-            raise ValueError("metadata must be an object")
-        encoded = json.dumps(value, ensure_ascii=True, sort_keys=True)
-        if len(encoded) > 8192:
-            raise ValueError("metadata is too large")
-        return encoded
-
-    @staticmethod
-    def _row(row) -> dict | None:
-        if row is None:
-            return None
-        metadata = {}
-        try:
-            metadata = json.loads(row[8]) if row[8] else {}
-        except Exception:
-            metadata = {}
-        return {
-            "id": int(row[0]),
-            "song_key": row[1],
-            "filename": row[2] or "",
-            "tone_key": row[3],
-            "provider_id": row[4],
-            "provider_ref": row[5],
-            "label": row[6] or "",
-            "source": row[7] or "manual",
-            "metadata": metadata if isinstance(metadata, dict) else {},
-            "created_at": row[9] or "",
-            "updated_at": row[10] or "",
-            "active": bool(row[11]),
-        }
-
-    def _select_sql(self) -> str:
-        return """
-            SELECT m.id, m.song_key, m.filename, m.tone_key, m.provider_id,
-                   m.provider_ref, m.label, m.source, m.metadata_json,
-                   m.created_at, m.updated_at,
-                   CASE WHEN a.mapping_id IS NULL THEN 0 ELSE 1 END AS active
-            FROM audio_effect_mappings m
-            LEFT JOIN audio_effect_active_mappings a
-              ON a.song_key = m.song_key AND a.tone_key = m.tone_key AND a.mapping_id = m.id
-        """
-
-    def list(self, *, song_key: str = "", filename: str = "", tone_key: str = "", provider_id: str = "") -> list[dict]:
-        clauses: list[str] = []
-        params: list[str] = []
-        song_key = self._text(song_key, field="song_key", limit=240, allow_empty=True)
-        filename = self._text(filename, field="filename", limit=500, allow_empty=True)
-        tone_key = self._text(tone_key, field="tone_key", limit=160, allow_empty=True)
-        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
-        if song_key and filename:
-            clauses.append("(m.song_key = ? OR m.filename = ?)")
-            params.extend([song_key, filename])
-        elif song_key:
-            clauses.append("m.song_key = ?")
-            params.append(song_key)
-        elif filename:
-            clauses.append("(m.song_key = ? OR m.filename = ?)")
-            params.extend([filename, filename])
-        if tone_key:
-            clauses.append("m.tone_key = ?")
-            params.append(tone_key)
-        if provider_id:
-            clauses.append("m.provider_id = ?")
-            params.append(provider_id)
-        sql = self._select_sql()
-        if clauses:
-            sql += " WHERE " + " AND ".join(clauses)
-        sql += " ORDER BY m.song_key COLLATE NOCASE, m.tone_key COLLATE NOCASE, m.provider_id COLLATE NOCASE"
-        with self._lock:
-            rows = self.conn.execute(sql, params).fetchall()
-        return [self._row(row) for row in rows]
-
-    def get(self, mapping_id: int) -> dict | None:
-        mapping_id = self._mapping_id(mapping_id)
-        if mapping_id is None:
-            return None
-        with self._lock:
-            row = self.conn.execute(self._select_sql() + " WHERE m.id = ?", (mapping_id,)).fetchone()
-        return self._row(row)
-
-    def upsert(self, data: dict) -> dict:
-        if not isinstance(data, dict):
-            raise ValueError("mapping body must be an object")
-        filename = self._text(data.get("filename", ""), field="filename", limit=500, allow_empty=True)
-        song_key_raw = self._field(data, "song_key", "songKey")
-        if song_key_raw is None or song_key_raw == "":
-            song_key_raw = filename
-        song_key = self._text(song_key_raw, field="song_key", limit=240)
-        tone_key = self._text(self._field(data, "tone_key", "toneKey"), field="tone_key", limit=160, allow_empty=True)
-        provider_id = self._text(self._field(data, "provider_id", "providerId"), field="provider_id", limit=96)
-        provider_ref = self._text(self._field(data, "provider_ref", "providerRef"), field="provider_ref", limit=240)
-        label = self._text(data.get("label", ""), field="label", limit=160, allow_empty=True)
-        source = self._text(data.get("source", "manual"), field="source", limit=40, allow_empty=True) or "manual"
-        metadata_json = self._metadata(data.get("metadata", {}))
-        with self._lock:
-            self.conn.execute(
-                """
-                INSERT INTO audio_effect_mappings
-                    (song_key, filename, tone_key, provider_id, provider_ref, label, source, metadata_json, updated_at)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
-                ON CONFLICT(song_key, tone_key, provider_id) DO UPDATE SET
-                    -- Only overwrite filename when a non-empty one was supplied; an
-                    -- omitted/empty filename must preserve the stored value (it's an
-                    -- alternate lookup key for list(..., filename=...)).
-                    filename=CASE WHEN excluded.filename <> '' THEN excluded.filename ELSE audio_effect_mappings.filename END,
-                    provider_ref=excluded.provider_ref,
-                    label=excluded.label,
-                    source=excluded.source,
-                    metadata_json=excluded.metadata_json,
-                    updated_at=datetime('now')
-                """,
-                (song_key, filename, tone_key, provider_id, provider_ref, label, source, metadata_json),
-            )
-            row = self.conn.execute(
-                "SELECT id FROM audio_effect_mappings WHERE song_key = ? AND tone_key = ? AND provider_id = ?",
-                (song_key, tone_key, provider_id),
-            ).fetchone()
-            if row is None:
-                raise ValueError("failed to create audio-effects mapping")
-            mapping_id = int(row[0])
-            if data.get("active") is True:
-                self.conn.execute(
-                    """
-                    INSERT INTO audio_effect_active_mappings (song_key, tone_key, mapping_id, updated_at)
-                    VALUES (?, ?, ?, datetime('now'))
-                    ON CONFLICT(song_key, tone_key) DO UPDATE SET
-                        mapping_id=excluded.mapping_id,
-                        updated_at=datetime('now')
-                    """,
-                    (song_key, tone_key, mapping_id),
-                )
-            self.conn.commit()
-        return self.get(mapping_id)
-
-    def delete(self, mapping_id: int, *, provider_id: str = "") -> bool:
-        mapping_id = self._mapping_id(mapping_id)
-        if mapping_id is None:
-            return False
-        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
-        with self._lock:
-            if provider_id:
-                cur = self.conn.execute(
-                    "DELETE FROM audio_effect_mappings WHERE id = ? AND provider_id = ?",
-                    (mapping_id, provider_id),
-                )
-            else:
-                cur = self.conn.execute("DELETE FROM audio_effect_mappings WHERE id = ?", (mapping_id,))
-            self.conn.commit()
-            return cur.rowcount > 0
-
-    def activate(self, mapping_id: int, *, provider_id: str = "") -> dict | None:
-        mapping_id = self._mapping_id(mapping_id)
-        if mapping_id is None:
-            return None
-        provider_id = self._text(provider_id, field="provider_id", limit=96, allow_empty=True)
-        with self._lock:
-            row = self.conn.execute(
-                self._select_sql() + " WHERE m.id = ?",
-                (mapping_id,),
-            ).fetchone()
-            mapping = self._row(row)
-            if not mapping or (provider_id and mapping["provider_id"] != provider_id):
-                return None
-            self.conn.execute(
-                """
-                INSERT INTO audio_effect_active_mappings (song_key, tone_key, mapping_id, updated_at)
-                VALUES (?, ?, ?, datetime('now'))
-                ON CONFLICT(song_key, tone_key) DO UPDATE SET
-                    mapping_id=excluded.mapping_id,
-                    updated_at=datetime('now')
-                """,
-                (mapping["song_key"], mapping["tone_key"], mapping_id),
-            )
-            self.conn.commit()
-            selected = self.conn.execute(self._select_sql() + " WHERE m.id = ?", (mapping_id,)).fetchone()
-        return self._row(selected)
-
-    def clear_active(self, *, song_key: str, tone_key: str) -> bool:
-        song_key = self._text(song_key, field="song_key", limit=240)
-        tone_key = self._text(tone_key, field="tone_key", limit=160, allow_empty=True)
-        with self._lock:
-            cur = self.conn.execute(
-                "DELETE FROM audio_effect_active_mappings WHERE song_key = ? AND tone_key = ?",
-                (song_key, tone_key),
-            )
-            self.conn.commit()
-            return cur.rowcount > 0
-
-
 meta_db = MetadataDB(CONFIG_DIR)
-audio_effect_mappings = AudioEffectsMappingDB()
+audio_effect_mappings = AudioEffectsMappingDB(CONFIG_DIR)
 
 
 class LocalLibraryProvider: