Merge pull request 'fix(cache): канонизация ключа канала (Durov/durov/@name -> один ключ) + one-shot миграция' (#47) from fix/24-channel-key into main

Reviewed-on: #47

api_server.py

@@ -44,6 +44,8 @@ from file_io import (DB_PATH, init_db_sync, get_all_media_file_ids_sync,
                      remove_media_file_ids_sync,
                      get_mime_type_sync, set_mime_type_sync)
 from tg_cache import cleanup_legacy_cache_files, sweep_tgcache
+from channel_key import canonical_channel_key
+from migrate_channel_keys import migrate_channel_keys_sync
 
 # Global python-magic instance for MIME type detection
 magic_mime = magic.Magic(mime=True)
@@ -270,6 +272,12 @@ async def lifespan(_: FastAPI):
     # Initialize SQLite database (creates table if not present)
     await asyncio.to_thread(init_db_sync, DB_PATH)
 
+    # One-shot migration of existing cache dirs + DB rows to the canonical channel key
+    # (case-insensitive username collapse). Runs AFTER init_db_sync and BEFORE client.start()
+    # and the background tasks. Wrapped in to_thread because it is a blocking FS-rename +
+    # SQLite routine that would otherwise stall the event loop. Idempotent: a re-run is a no-op.
+    await asyncio.to_thread(migrate_channel_keys_sync, DB_PATH, MEDIA_CACHE_DIR)
+
     # One-shot startup maintenance of the history/chatinfo cache: drop legacy pickle files
     # (which are now always a miss), age-sweep stale tgcache entries, and remove the
     # pre-SQLite data/media_file_ids.json legacy dump (no longer read by any code).
@@ -1311,30 +1319,32 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re
         #else:
         #    logger.info(f"Valid digest for media {url}: {digest}")   
             
-        # Convert numeric channel ID to int if needed
-        channel_id: Union[str, int] = channel
-        if isinstance(channel, str) and channel.startswith('-100'):
-            channel_id = int(channel)
-            
+        # Canonical filesystem/DB/API key. Computed AFTER the digest check (which must run
+        # against the ORIGINAL url string) so 'Durov', 'durov' and '@durov' collapse to one
+        # on-disk tree, one _access_updates/media_key identity and one download identity. The
+        # canonical form is API-safe: usernames are case-insensitive on Telegram's side and
+        # numeric '-100...' ids are preserved verbatim (download_media_file re-ints them).
+        fs_channel = canonical_channel_key(channel)
+
         try: # Wrap the download and prepare call
             import time as _time
 
             # Pre-semaphore cache check: serve already-cached files without acquiring the semaphore
-            cache_path = media_cache_path(str(channel), post_id, file_unique_id)
+            cache_path = media_cache_path(fs_channel, post_id, file_unique_id)
             if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0:
                 # File is already in cache — skip semaphore and serve directly
-                logger.info(f"pre_semaphore_cache_hit: {channel}/{post_id}/{file_unique_id}")
+                logger.info(f"pre_semaphore_cache_hit: {fs_channel}/{post_id}/{file_unique_id}")
                 # Record the access time into the accumulator instead of firing a per-hit
-                # SQLite write. Key channel as str(channel) — see _access_updates.
-                _access_updates[(str(channel), post_id, file_unique_id)] = datetime.now().timestamp()
+                # SQLite write. Key channel as the canonical fs_channel — see _access_updates.
+                _access_updates[(fs_channel, post_id, file_unique_id)] = datetime.now().timestamp()
                 return await prepare_file_response(cache_path, request=request,
-                                                   media_key=(str(channel), post_id, file_unique_id))
+                                                   media_key=(fs_channel, post_id, file_unique_id))
 
             # A file that recently kept failing is in backoff: fast-reject instead of
             # occupying a scarce download slot (and a Pyrogram transmission permit) on a
             # request that will very likely hang again. Cached files already returned above,
             # so this only guards the live-download path.
-            backoff_remaining = _download_backoff_remaining((str(channel), post_id, file_unique_id))
+            backoff_remaining = _download_backoff_remaining((fs_channel, post_id, file_unique_id))
             if backoff_remaining > 0:
                 logger.info(f"media_backoff_skip: {channel}/{post_id}/{file_unique_id} in backoff {backoff_remaining:.0f}s")
                 return Response(status_code=503, content="Media temporarily unavailable, retry later",
@@ -1369,7 +1379,7 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re
                 if _sem_wait > 0.5:
                     logger.warning(f"diag_semaphore_wait: {channel}/{post_id}/{file_unique_id} waited {_sem_wait:.3f}s for HTTP_DOWNLOAD_SEMAPHORE")
                 _dl_start = _time.monotonic()
-                file_path, delete_after = await _download_deduped(channel_id, post_id, file_unique_id)
+                file_path, delete_after = await _download_deduped(fs_channel, post_id, file_unique_id)
                 _dl_elapsed = _time.monotonic() - _dl_start
                 logger.info(f"diag_download_timing: {channel}/{post_id}/{file_unique_id} download_media_file took {_dl_elapsed:.3f}s (semaphore_wait={_sem_wait:.3f}s)")
             finally:
@@ -1378,7 +1388,7 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re
                 raise HTTPException(status_code=404, detail="File not found")
             if file_path:
                 return await prepare_file_response(file_path, request=request, delete_after=delete_after,
-                                                   media_key=(str(channel), post_id, file_unique_id))
+                                                   media_key=(fs_channel, post_id, file_unique_id))
         except ZeroSizeFileError as e: # Catch zero-size file errors
             logger.warning(f"zero_size_file_encountered: {str(e)}. Instructing client to retry.")
             return Response(

channel_key.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# flake8: noqa
+# pylint: disable=missing-function-docstring
+
+"""Canonical channel key.
+
+Dependency-free helper shared by tg_cache, post_parser and api_server. Kept free of
+any project imports so it can be imported by all three layers without introducing a
+circular import.
+"""
+
+from typing import Union
+
+
+def canonical_channel_key(channel: Union[str, int]) -> str:
+    """Canonical cache/DB key for a channel.
+
+    Telegram usernames are case-insensitive -> lowercase them.
+    Numeric '-100...' ids keep their exact string form. The '@' prefix is stripped.
+    The canonical form is also SAFE for Telegram API calls (usernames case-insensitive
+    on the API side; numeric unchanged) -- callers may thread one value through both
+    filesystem paths and API calls.
+    """
+    s = str(channel).strip().lstrip('@')
+    if s.startswith('-100') and s[4:].isdigit():
+        return s
+    return s.lower()

migrate_channel_keys.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# flake8: noqa
+# pylint: disable=missing-function-docstring, broad-exception-caught, logging-fstring-interpolation
+
+"""One-shot migration of existing cache data to the canonical channel key.
+
+Collapses non-canonical (mixed-case) channel names in both the on-disk media cache
+(data/cache/<channel>/...) and the SQLite media_file_ids table onto their canonical,
+lowercase form. Numeric '-100...' ids are already canonical and are never touched.
+
+Unit of work = a WHOLE channel, processed FS FIRST then SQL. The order is essential:
+if the FS rename/merge fails, the channel's SQL rows are left old-cased so the old tree
+is still reachable by the media sweeper via its DB rows (no eternal orphan directory).
+
+The migration is idempotent: a second run finds nothing to do and is a no-op.
+"""
+
+import os
+import shutil
+import sqlite3
+import logging
+
+from channel_key import canonical_channel_key
+
+logger = logging.getLogger(__name__)
+
+
+def _is_safe_channel_segment(name: str) -> bool:
+    """True iff ``name`` is a plain single path segment safe to join under cache_dir.
+
+    Defends the destructive FS ops below (os.rename / _merge_dir_tree / shutil.rmtree)
+    against a dirty channel string that reached the DB or a crafted dir on disk,
+    regardless of any upstream route-traversal. A dirty name (containing '/', '\\' or a
+    '..' component, or that is not a bare basename, or empty/'.') would let a rename or
+    rmtree escape cache_dir, so it is rejected and the channel is left un-migrated.
+    """
+    if not name or name == '.':
+        return False
+    if '/' in name or '\\' in name:
+        return False
+    if '..' in name.replace('\\', '/').split('/'):
+        return False
+    if os.path.basename(name) != name:
+        return False
+    return True
+
+
+def _merge_dir_tree(src: str, dst: str) -> None:
+    """Merge every file under ``src`` into ``dst`` where an existing ``dst`` file WINS.
+
+    Both trees share the same filesystem (siblings under the media cache root), so files
+    are moved with os.rename. Any file that already exists in ``dst`` is kept and the
+    ``src`` copy is discarded. Emptied ``src`` remnants are removed at the end.
+    """
+    for root, _dirs, files in os.walk(src):
+        rel = os.path.relpath(root, src)
+        target_root = dst if rel == '.' else os.path.join(dst, rel)
+        os.makedirs(target_root, exist_ok=True)
+        for name in files:
+            s = os.path.join(root, name)
+            t = os.path.join(target_root, name)
+            if os.path.exists(t):
+                os.remove(s)  # existing file in dst wins
+            else:
+                os.rename(s, t)
+    shutil.rmtree(src, ignore_errors=True)
+
+
+def migrate_channel_keys_sync(db_path: str, cache_dir: str) -> None:
+    """Migrate mixed-case channel cache dirs and DB rows onto the canonical key.
+
+    Blocking (FS renames + SQLite). Call via asyncio.to_thread from an async context.
+    """
+    rows_merged = 0
+    dirs_renamed = 0
+    samefile_noops = 0
+    failures = 0
+
+    conn = sqlite3.connect(db_path)
+    conn.execute("PRAGMA busy_timeout = 5000")
+    try:
+        # ------------------------------------------------------------------ #
+        # Candidate channels: mixed-case first-level cache dirs UNION mixed-case DB rows.
+        # ------------------------------------------------------------------ #
+        candidates: set[str] = set()
+        try:
+            for entry in os.listdir(cache_dir):
+                full = os.path.join(cache_dir, entry)
+                if not os.path.isdir(full):
+                    continue
+                if entry.startswith('-100'):
+                    continue
+                if entry != entry.lower():
+                    candidates.add(entry)
+        except FileNotFoundError:
+            pass  # No cache dir yet — nothing to migrate on the FS side.
+
+        try:
+            cur = conn.execute(
+                "SELECT DISTINCT channel FROM media_file_ids "
+                "WHERE channel != lower(channel) AND channel NOT LIKE '-100%'"
+            )
+            for (ch,) in cur.fetchall():
+                candidates.add(ch)
+        except sqlite3.Error as e:
+            logger.error(f"migrate_channel_keys: DB candidate query failed: {e}")
+
+        for name in sorted(candidates):
+            # Traversal guard: reject any dirty channel name BEFORE it reaches the
+            # destructive FS ops (rename/merge/rmtree). Applies to both candidate sources
+            # (DB rows and the dir listing). A rejected channel is left as-is (its old-cased
+            # rows/dirs survive until the 20-day sweep, same as any un-migrated channel).
+            if not _is_safe_channel_segment(name):
+                logger.warning(
+                    f"migrate_channel_keys: unsafe channel name {name!r} rejected "
+                    f"(would escape cache_dir), skipping"
+                )
+                failures += 1
+                continue
+            canonical = canonical_channel_key(name)
+            if canonical == name:
+                # Nothing to change (already canonical) — defensive, candidates shouldn't hit this.
+                continue
+            try:
+                # ---------------------------- FS step ---------------------------- #
+                src = os.path.join(cache_dir, name)
+                dst = os.path.join(cache_dir, canonical)
+                if os.path.isdir(src):
+                    try:
+                        if os.path.exists(dst) and os.path.samefile(src, dst):
+                            # Case-insensitive FS (macOS/APFS, Docker Desktop for Mac):
+                            # src and dst are ONE directory. Renaming/merging would destroy
+                            # the channel's entire cache — treat as a pure no-op.
+                            samefile_noops += 1
+                        elif not os.path.exists(dst):
+                            os.rename(src, dst)
+                            dirs_renamed += 1
+                        else:
+                            # Genuinely different dirs (case-sensitive FS, both forms present):
+                            # per-file merge with the existing dst file winning.
+                            _merge_dir_tree(src, dst)
+                            dirs_renamed += 1
+                    except OSError as e:
+                        # FS step failed (EACCES etc.): mark orphan candidate and SKIP the SQL
+                        # step so the old-cased DB rows keep the old tree visible to the sweeper.
+                        logger.error(
+                            f"migrate_channel_keys: FS step failed for {name!r} -> {canonical!r} "
+                            f"(orphan candidate, SQL skipped): {e}"
+                        )
+                        failures += 1
+                        continue
+
+                # ---------------------------- SQL step --------------------------- #
+                old_rows = conn.execute(
+                    "SELECT post_id, file_unique_id, added, mime_type "
+                    "FROM media_file_ids WHERE channel = ?",
+                    (name,),
+                ).fetchall()
+                for (post_id, file_unique_id, added, mime_type) in old_rows:
+                    twin = conn.execute(
+                        "SELECT added, mime_type FROM media_file_ids "
+                        "WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+                        (canonical, post_id, file_unique_id),
+                    ).fetchone()
+                    if twin is None:
+                        # No lowercase twin: a plain re-key is safe (no PK conflict).
+                        conn.execute(
+                            "UPDATE media_file_ids SET channel = ? "
+                            "WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+                            (canonical, name, post_id, file_unique_id),
+                        )
+                    else:
+                        # Twin exists: merge (added = max, prefer non-NULL mime_type) into the
+                        # canonical row, then delete the old-cased row. A bare
+                        # UPDATE ... SET channel=lower(channel) is FORBIDDEN (PK conflict).
+                        t_added, t_mime = twin
+                        new_added = max(added, t_added)
+                        new_mime = t_mime if t_mime is not None else mime_type
+                        conn.execute(
+                            "UPDATE media_file_ids SET added = ?, mime_type = ? "
+                            "WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+                            (new_added, new_mime, canonical, post_id, file_unique_id),
+                        )
+                        conn.execute(
+                            "DELETE FROM media_file_ids "
+                            "WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
+                            (name, post_id, file_unique_id),
+                        )
+                        rows_merged += 1
+                conn.commit()
+            except Exception as e:
+                # Never crash startup: log and move on. A re-run is an idempotent no-op.
+                try:
+                    conn.rollback()
+                except Exception:
+                    pass
+                logger.error(f"migrate_channel_keys: channel {name!r} migration failed: {e}")
+                failures += 1
+    finally:
+        conn.close()
+
+    logger.info(
+        f"migration_summary: rows merged {rows_merged}, dirs renamed {dirs_renamed}, "
+        f"samefile no-ops {samefile_noops}, failures {failures}"
+    )

post_parser.py

@@ -1416,8 +1416,8 @@ class PostParser:
         if hasattr(chat, 'usernames') and chat.usernames: # Check many usernames
             active_usernames = [u.username for u in chat.usernames if u.active]
             if active_usernames:
-                return active_usernames[0]
-        if hasattr(chat, 'username') and chat.username: return chat.username # Check single username
+                return active_usernames[0].lower() # Canonicalize: usernames are case-insensitive
+        if hasattr(chat, 'username') and chat.username: return chat.username.lower() # Check single username (canonicalized)
         
         # Return numeric ID only if no username found
         if isinstance(chat.id, int) and str(chat.id).startswith('-100'): return str(chat.id)

tests/test_channel_key.py

@@ -0,0 +1,350 @@
+# flake8: noqa
+# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
+# pylint: disable=redefined-outer-name, line-too-long
+# pylance: disable=reportMissingImports, reportMissingModuleSource
+"""Tests for the canonical channel key: helper, the three wiring layers, and the
+one-shot migration (issue #24, tasks 8-11)."""
+
+import logging
+import os
+import shutil
+import sqlite3
+from types import SimpleNamespace
+
+import pytest
+
+import tg_cache
+from channel_key import canonical_channel_key
+from migrate_channel_keys import migrate_channel_keys_sync
+from post_parser import PostParser
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 — unit: canonical_channel_key.
+# --------------------------------------------------------------------------- #
+def test_canonical_lowercases_username():
+    assert canonical_channel_key('Durov') == 'durov'
+
+
+def test_canonical_strips_at_prefix():
+    assert canonical_channel_key('@durov') == 'durov'
+    assert canonical_channel_key('@Durov') == 'durov'
+
+
+def test_canonical_numeric_id_str_preserved():
+    assert canonical_channel_key('-1001234567890') == '-1001234567890'
+
+
+def test_canonical_numeric_id_int_preserved():
+    assert canonical_channel_key(-1001234567890) == '-1001234567890'
+
+
+def test_canonical_already_lowercase_idempotent():
+    assert canonical_channel_key('durov') == 'durov'
+    assert canonical_channel_key(canonical_channel_key('Durov')) == 'durov'
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 — tg_cache path is identical for the two casings.
+# --------------------------------------------------------------------------- #
+def test_tgcache_path_identical_for_casings(tmp_path, monkeypatch):
+    monkeypatch.setattr(tg_cache, "CACHE_DIR", str(tmp_path))
+    p_upper = tg_cache._cache_file_path('Durov', 'history.json')
+    p_lower = tg_cache._cache_file_path('durov', 'history.json')
+    p_at = tg_cache._cache_file_path('@DUROV', 'history.json')
+    assert p_upper == p_lower == p_at
+    assert os.path.basename(p_upper) == 'durov.history.json'
+
+
+def test_tgcache_path_numeric_preserved(tmp_path, monkeypatch):
+    monkeypatch.setattr(tg_cache, "CACHE_DIR", str(tmp_path))
+    p = tg_cache._cache_file_path('-1001234567890', 'chatinfo.json')
+    assert os.path.basename(p) == '-1001234567890.chatinfo.json'
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 — get_channel_username lowercases both branches.
+# --------------------------------------------------------------------------- #
+def _parser():
+    return PostParser(SimpleNamespace())
+
+
+def test_get_channel_username_single_lowercased():
+    parser = _parser()
+    msg = SimpleNamespace(chat=SimpleNamespace(username='MixedCase', id=1))
+    assert parser.get_channel_username(msg) == 'mixedcase'
+
+
+def test_get_channel_username_usernames_list_lowercased():
+    parser = _parser()
+    chat = SimpleNamespace(
+        usernames=[SimpleNamespace(username='MixedCase', active=True)],
+        username=None,
+        id=1,
+    )
+    assert parser.get_channel_username(SimpleNamespace(chat=chat)) == 'mixedcase'
+
+
+def test_get_channel_username_numeric_id_unchanged():
+    parser = _parser()
+    chat = SimpleNamespace(usernames=None, username=None, id=-1001234567890)
+    assert parser.get_channel_username(SimpleNamespace(chat=chat)) == '-1001234567890'
+
+
+# --------------------------------------------------------------------------- #
+# Migration helpers.
+# --------------------------------------------------------------------------- #
+def _make_db(path):
+    conn = sqlite3.connect(path)
+    conn.execute(
+        """CREATE TABLE media_file_ids (
+               channel        TEXT    NOT NULL,
+               post_id        INTEGER NOT NULL,
+               file_unique_id TEXT    NOT NULL,
+               added          REAL    NOT NULL,
+               mime_type      TEXT,
+               PRIMARY KEY (channel, post_id, file_unique_id)
+           )"""
+    )
+    conn.commit()
+    conn.close()
+
+
+def _rows(path):
+    conn = sqlite3.connect(path)
+    conn.row_factory = sqlite3.Row
+    rows = [dict(r) for r in conn.execute(
+        "SELECT channel, post_id, file_unique_id, added, mime_type FROM media_file_ids "
+        "ORDER BY channel, post_id, file_unique_id")]
+    conn.close()
+    return rows
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 (a) — SQL merge of both forms → one row, max(added), non-NULL mime_type.
+# --------------------------------------------------------------------------- #
+def test_migration_sql_merge(tmp_path):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    conn = sqlite3.connect(db)
+    # Old-cased row (higher added, NULL mime) and lowercase twin (lower added, has mime).
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 200.0, None))
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('durov', 5, 'fid', 100.0, 'image/jpeg'))
+    conn.commit()
+    conn.close()
+
+    migrate_channel_keys_sync(db, str(cache))
+
+    rows = _rows(db)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r['channel'] == 'durov'
+    assert r['added'] == 200.0            # max(added)
+    assert r['mime_type'] == 'image/jpeg'  # non-NULL preferred
+
+
+def test_migration_sql_rekey_no_twin(tmp_path):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    conn = sqlite3.connect(db)
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 7, 'fid', 50.0, 'video/mp4'))
+    conn.commit()
+    conn.close()
+
+    migrate_channel_keys_sync(db, str(cache))
+
+    rows = _rows(db)
+    assert len(rows) == 1
+    assert rows[0]['channel'] == 'durov'
+    assert rows[0]['mime_type'] == 'video/mp4'
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 (b) — samefile guard → no-op, data intact (case-insensitive FS simulated).
+# --------------------------------------------------------------------------- #
+def test_migration_samefile_guard_no_data_loss(tmp_path, monkeypatch):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    # Single dir 'Durov' with a file; on a case-insensitive FS 'durov' would be the same dir.
+    src = cache / "Durov" / "5"
+    src.mkdir(parents=True)
+    (src / "fid").write_bytes(b"payload")
+
+    # Simulate a case-insensitive FS: any existence check on the lowercase twin resolves,
+    # and samefile reports src == dst so the FS step must be a pure no-op.
+    real_exists = os.path.exists
+
+    def fake_exists(p):
+        if os.path.basename(p) == 'durov':
+            return True
+        return real_exists(p)
+
+    monkeypatch.setattr(os.path, "exists", fake_exists)
+    monkeypatch.setattr(os.path, "samefile", lambda a, b: True)
+
+    migrate_channel_keys_sync(db, str(cache))
+
+    # Data intact: the original file must still be present and unchanged.
+    assert (cache / "Durov" / "5" / "fid").read_bytes() == b"payload"
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 (c) — merge of two genuinely different dirs → combined, target wins.
+# --------------------------------------------------------------------------- #
+def test_migration_fs_merge_different_dirs(tmp_path):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+
+    # Old-cased tree.
+    (cache / "Durov" / "5").mkdir(parents=True)
+    (cache / "Durov" / "5" / "shared").write_bytes(b"OLD")   # collides -> dst wins
+    (cache / "Durov" / "5" / "only_old").write_bytes(b"OLD_ONLY")
+    # Canonical tree.
+    (cache / "durov" / "5").mkdir(parents=True)
+    (cache / "durov" / "5" / "shared").write_bytes(b"NEW")   # existing dst wins
+    (cache / "durov" / "6").mkdir(parents=True)
+    (cache / "durov" / "6" / "only_new").write_bytes(b"NEW_ONLY")
+
+    migrate_channel_keys_sync(db, str(cache))
+
+    # Old dir gone, everything merged under 'durov'.
+    assert not (cache / "Durov").exists()
+    assert (cache / "durov" / "5" / "shared").read_bytes() == b"NEW"        # target wins
+    assert (cache / "durov" / "5" / "only_old").read_bytes() == b"OLD_ONLY"  # brought over
+    assert (cache / "durov" / "6" / "only_new").read_bytes() == b"NEW_ONLY"  # untouched
+
+
+def test_migration_fs_rename_when_dst_missing(tmp_path):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    (cache / "Durov" / "5").mkdir(parents=True)
+    (cache / "Durov" / "5" / "fid").write_bytes(b"data")
+
+    migrate_channel_keys_sync(db, str(cache))
+
+    assert not (cache / "Durov").exists()
+    assert (cache / "durov" / "5" / "fid").read_bytes() == b"data"
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 (d) — re-run is a no-op.
+# --------------------------------------------------------------------------- #
+def test_migration_rerun_noop(tmp_path):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    conn = sqlite3.connect(db)
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 200.0, None))
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('durov', 5, 'fid', 100.0, 'image/jpeg'))
+    conn.commit()
+    conn.close()
+    (cache / "Durov" / "5").mkdir(parents=True)
+    (cache / "Durov" / "5" / "fid").write_bytes(b"data")
+
+    migrate_channel_keys_sync(db, str(cache))
+    first = _rows(db)
+    # Second run: nothing left to migrate.
+    migrate_channel_keys_sync(db, str(cache))
+    second = _rows(db)
+
+    assert first == second
+    assert all(r['channel'] == r['channel'].lower() for r in second)
+    assert (cache / "durov" / "5" / "fid").read_bytes() == b"data"
+
+
+# --------------------------------------------------------------------------- #
+# Data-loss guard — an FS-step failure must SKIP the SQL step (rows stay old-cased).
+# Has teeth: fails if the `continue` after the FS OSError is removed.
+# --------------------------------------------------------------------------- #
+def test_migration_fs_failure_skips_sql(tmp_path, monkeypatch, caplog):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    conn = sqlite3.connect(db)
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 100.0, 'image/jpeg'))
+    conn.commit()
+    conn.close()
+    # Old-cased dir with no lowercase twin → the migration takes the os.rename path.
+    (cache / "Durov" / "5").mkdir(parents=True)
+    (cache / "Durov" / "5" / "fid").write_bytes(b"data")
+
+    def boom(*_a, **_k):
+        raise OSError("disk on fire")
+
+    monkeypatch.setattr(os, "rename", boom)
+
+    with caplog.at_level(logging.INFO):
+        migrate_channel_keys_sync(db, str(cache))  # (a) must NOT crash startup
+
+    # (b) SQL step skipped: the row is still OLD-cased.
+    rows = _rows(db)
+    assert len(rows) == 1
+    assert rows[0]['channel'] == 'Durov'
+    # (c) failures counter increased (surfaced in the summary log).
+    assert "failures 1" in caplog.text
+
+
+# --------------------------------------------------------------------------- #
+# Reverse-direction max(added): lowercase twin has the LARGER `added` → it wins.
+# Proves max() is not one-directional ("old always wins").
+# --------------------------------------------------------------------------- #
+def test_migration_sql_merge_reverse_max(tmp_path):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    conn = sqlite3.connect(db)
+    # OLD-cased added=100 (SMALLER), lowercase twin added=300 (LARGER).
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('Durov', 5, 'fid', 100.0, 'image/jpeg'))
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('durov', 5, 'fid', 300.0, None))
+    conn.commit()
+    conn.close()
+
+    migrate_channel_keys_sync(db, str(cache))
+
+    rows = _rows(db)
+    assert len(rows) == 1
+    assert rows[0]['channel'] == 'durov'
+    assert rows[0]['added'] == 300.0          # max(added): the twin's LARGER value survives
+    assert rows[0]['mime_type'] == 'image/jpeg'  # non-NULL preferred (from the old row)
+
+
+# --------------------------------------------------------------------------- #
+# Traversal guard — a DB channel name with '/' or '..' is SKIPPED before any FS op.
+# Has teeth: fails if the _is_safe_channel_segment guard is removed.
+# --------------------------------------------------------------------------- #
+def test_migration_traversal_guard_skips_dirty_channel(tmp_path, monkeypatch):
+    db = str(tmp_path / "m.db")
+    cache = tmp_path / "cache"
+    cache.mkdir()
+    _make_db(db)
+    conn = sqlite3.connect(db)
+    # Dirty mixed-case channel names → become migration candidates via the DB query.
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('../Evil', 5, 'fid', 100.0, None))
+    conn.execute("INSERT INTO media_file_ids VALUES (?,?,?,?,?)", ('a/B', 6, 'fid', 100.0, None))
+    conn.commit()
+    conn.close()
+
+    # Spy: no destructive FS op may run (all dirty candidates rejected before the FS step).
+    called = []
+    monkeypatch.setattr(os, "rename", lambda *a, **k: called.append(('rename', a)))
+    monkeypatch.setattr(shutil, "rmtree", lambda *a, **k: called.append(('rmtree', a)))
+
+    migrate_channel_keys_sync(db, str(cache))  # must NOT crash
+
+    assert called == []  # nothing renamed/removed → no op escaped cache_dir
+    # Rejected channels are left un-migrated (rows keep their original dirty names).
+    channels = {r['channel'] for r in _rows(db)}
+    assert channels == {'../Evil', 'a/B'}

tg_cache.py

@@ -27,6 +27,7 @@ from message_snapshot import (
     snapshot_messages,
     restore_messages,
 )
+from channel_key import canonical_channel_key
 
 logger = logging.getLogger(__name__)
 
@@ -51,11 +52,15 @@ def _safe_key(key: Union[str, int]) -> str:
 
 
 def _cache_file_path(key: Union[str, int], suffix: str) -> str:
-    """Return the cache file path <safe_key>.<suffix> (e.g. 'history.json' / 'chatinfo.json')."""
+    """Return the cache file path <safe_key>.<suffix> (e.g. 'history.json' / 'chatinfo.json').
+
+    The key is first canonicalized so that 'Durov', 'durov' and '@durov' collapse to a
+    single cache file (Telegram usernames are case-insensitive).
+    """
     if not os.path.exists(CACHE_DIR):
         os.makedirs(CACHE_DIR, exist_ok=True)
         logger.info(f"cache_dir_created: path {CACHE_DIR}")
-    return os.path.join(CACHE_DIR, f"{_safe_key(key)}.{suffix}")
+    return os.path.join(CACHE_DIR, f"{_safe_key(canonical_channel_key(key))}.{suffix}")
 
 
 # --------------------------------------------------------------------------- #