feat(stickers): implement sticker management features including creation, deletion, listing, and import/export functionality, along with database schema updates for user stickers

meshchatx/meshchat.py

@@ -25,6 +25,7 @@ import time
 import traceback
 import webbrowser
 import io
+import subprocess
 import zipfile
 import fnmatch
 from datetime import UTC, datetime, timedelta
@@ -103,6 +104,12 @@ from meshchatx.src.backend.persistent_log_handler import PersistentLogHandler
 from meshchatx.src.backend.recovery import CrashRecovery, HealthMonitor
 from meshchatx.src.backend.rnprobe_handler import RNProbeHandler
 from meshchatx.src.backend.sideband_commands import SidebandCommands
+from meshchatx.src.backend.sticker_utils import (
+    build_export_document,
+    mime_for_image_type,
+    sanitize_sticker_name,
+    validate_export_document,
+)
 from meshchatx.src.backend.telemetry_utils import Telemeter
 from meshchatx.src.backend.web_audio_bridge import WebAudioBridge
 from meshchatx.src.version import __version__ as app_version
@@ -4831,6 +4838,12 @@ class ReticulumMeshChat:
             self.database.misc.delete_all_user_icons()
             return web.json_response({"message": "All LXMF icons cleared"})
 
+        @routes.delete("/api/v1/maintenance/stickers")
+        async def maintenance_clear_stickers(request):
+            identity_hash = self.identity.hash.hex()
+            n = self.database.stickers.delete_all_for_identity(identity_hash)
+            return web.json_response({"message": "Stickers cleared", "deleted": n})
+
         # maintenance - export messages
         @routes.get("/api/v1/maintenance/messages/export")
         async def maintenance_export_messages(request):
@@ -9170,6 +9183,110 @@ class ReticulumMeshChat:
             self.database.map_drawings.update_drawing(drawing_id, name, drawing_data)
             return web.json_response({"message": "Drawing updated successfully"})
 
+        @routes.get("/api/v1/stickers")
+        async def stickers_list(request):
+            identity_hash = self.identity.hash.hex()
+            rows = self.database.stickers.list_for_identity(identity_hash)
+            return web.json_response({"stickers": [dict(r) for r in rows]})
+
+        @routes.post("/api/v1/stickers")
+        async def stickers_create(request):
+            identity_hash = self.identity.hash.hex()
+            try:
+                data = await request.json()
+            except (json.JSONDecodeError, ValueError):
+                return web.json_response({"error": "invalid_json"}, status=400)
+            image_b64 = data.get("image_bytes")
+            if not isinstance(image_b64, str) or not image_b64.strip():
+                return web.json_response({"error": "missing_image_bytes"}, status=400)
+            try:
+                raw = base64.b64decode(image_b64.strip(), validate=True)
+            except (ValueError, TypeError):
+                return web.json_response({"error": "invalid_base64"}, status=400)
+            name = sanitize_sticker_name(data.get("name"))
+            image_type = data.get("image_type")
+            src = data.get("source_message_hash")
+            src = src if isinstance(src, str) else None
+            try:
+                row = self.database.stickers.insert(
+                    identity_hash,
+                    name,
+                    image_type,
+                    raw,
+                    src,
+                )
+            except ValueError as e:
+                return web.json_response({"error": str(e)}, status=400)
+            if row is None:
+                return web.json_response({"error": "duplicate_sticker"}, status=409)
+            return web.json_response({"sticker": row})
+
+        @routes.delete("/api/v1/stickers/{sticker_id}")
+        async def stickers_delete(request):
+            identity_hash = self.identity.hash.hex()
+            sticker_id = int(request.match_info.get("sticker_id", "0"))
+            ok = self.database.stickers.delete(sticker_id, identity_hash)
+            if not ok:
+                return web.json_response({"error": "not_found"}, status=404)
+            return web.json_response({"message": "deleted"})
+
+        @routes.patch("/api/v1/stickers/{sticker_id}")
+        async def stickers_patch(request):
+            identity_hash = self.identity.hash.hex()
+            sticker_id = int(request.match_info.get("sticker_id", "0"))
+            try:
+                data = await request.json()
+            except (json.JSONDecodeError, ValueError):
+                return web.json_response({"error": "invalid_json"}, status=400)
+            if "name" not in data:
+                return web.json_response({"error": "missing_name"}, status=400)
+            name = sanitize_sticker_name(data.get("name"))
+            ok = self.database.stickers.update_name(sticker_id, identity_hash, name)
+            if not ok:
+                return web.json_response({"error": "not_found"}, status=404)
+            return web.json_response({"message": "updated"})
+
+        @routes.get("/api/v1/stickers/{sticker_id}/image")
+        async def stickers_get_image(request):
+            identity_hash = self.identity.hash.hex()
+            sticker_id = int(request.match_info.get("sticker_id", "0"))
+            row = self.database.stickers.get_row(sticker_id, identity_hash)
+            if row is None:
+                return web.json_response({"error": "not_found"}, status=404)
+            ct = mime_for_image_type(row["image_type"])
+            return web.Response(body=row["image_blob"], content_type=ct)
+
+        @routes.get("/api/v1/stickers/export")
+        async def stickers_export(request):
+            identity_hash = self.identity.hash.hex()
+            payloads = self.database.stickers.export_payloads_for_identity(
+                identity_hash
+            )
+            doc = build_export_document(
+                payloads,
+                datetime.now(UTC).isoformat(),
+            )
+            return web.json_response(doc)
+
+        @routes.post("/api/v1/stickers/import")
+        async def stickers_import(request):
+            identity_hash = self.identity.hash.hex()
+            try:
+                data = await request.json()
+            except (json.JSONDecodeError, ValueError):
+                return web.json_response({"error": "invalid_json"}, status=400)
+            replace = bool(data.get("replace_duplicates", False))
+            try:
+                items = validate_export_document(data)
+            except ValueError as e:
+                return web.json_response({"error": str(e)}, status=400)
+            result = self.database.stickers.import_payloads(
+                identity_hash,
+                items,
+                replace_duplicates=replace,
+            )
+            return web.json_response(result)
+
         # get latest telemetry for all peers
         @routes.get("/api/v1/telemetry/peers")
         async def get_all_latest_telemetry(request):
@@ -10264,6 +10381,11 @@ class ReticulumMeshChat:
                 data["message_failed_bubble_color"]
             )
 
+        if "message_waiting_bubble_color" in data:
+            self.config.message_waiting_bubble_color.set(
+                data["message_waiting_bubble_color"]
+            )
+
         # update desktop settings
         if "desktop_open_calls_in_separate_window" in data:
             self.config.desktop_open_calls_in_separate_window.set(
@@ -11429,6 +11551,7 @@ class ReticulumMeshChat:
             "message_outbound_bubble_color": ctx.config.message_outbound_bubble_color.get(),
             "message_inbound_bubble_color": ctx.config.message_inbound_bubble_color.get(),
             "message_failed_bubble_color": ctx.config.message_failed_bubble_color.get(),
+            "message_waiting_bubble_color": ctx.config.message_waiting_bubble_color.get(),
             "translator_enabled": ctx.config.translator_enabled.get(),
             "libretranslate_url": ctx.config.libretranslate_url.get(),
             "desktop_open_calls_in_separate_window": ctx.config.desktop_open_calls_in_separate_window.get(),
@@ -11778,6 +11901,47 @@ class ReticulumMeshChat:
         except Exception:
             return False
 
+    def _convert_webm_opus_to_ogg(self, audio_bytes: bytes) -> bytes:
+        """Convert WebM/Opus audio to OGG/Opus using ffmpeg.
+
+        Browser MediaRecorder outputs Opus in a WebM container, but LXMF
+        AM_OPUS_OGG expects an OGG container. If ffmpeg is unavailable or
+        the input is already OGG, the original bytes are returned as-is.
+        """
+        if audio_bytes[:4] == b"OggS":
+            return audio_bytes
+
+        ffmpeg_path = shutil.which("ffmpeg")
+        if ffmpeg_path is None:
+            return audio_bytes
+
+        try:
+            result = subprocess.run(  # noqa: S603
+                [
+                    ffmpeg_path,
+                    "-i",
+                    "pipe:0",
+                    "-c:a",
+                    "libopus",
+                    "-b:a",
+                    "24k",
+                    "-vbr",
+                    "on",
+                    "-f",
+                    "ogg",
+                    "pipe:1",
+                ],
+                input=audio_bytes,
+                capture_output=True,
+                timeout=30,
+            )
+            if result.returncode == 0 and len(result.stdout) > 0:
+                return result.stdout
+        except Exception as e:
+            print(f"WebM to OGG conversion failed: {e}")
+
+        return audio_bytes
+
     # check if a destination is blocked
     def is_destination_blocked(self, destination_hash: str, context=None) -> bool:
         ctx = context or self.current_context
@@ -12444,6 +12608,11 @@ class ReticulumMeshChat:
 
         lxmf_message.fields = {}
 
+        lxmf_message.fields[LXMF.FIELD_RENDERER] = LXMF.RENDERER_MARKDOWN
+
+        if self._is_contact(destination_hash, context=ctx):
+            lxmf_message.include_ticket = True
+
         # add file attachments field
         if file_attachments_field is not None:
             # create array of [[file_name, file_bytes], [file_name, file_bytes], ...]
@@ -12464,9 +12633,12 @@ class ReticulumMeshChat:
 
         # add audio field
         if audio_field is not None:
+            audio_bytes = audio_field.audio_bytes
+            if audio_field.audio_mode == LXMF.AM_OPUS_OGG:
+                audio_bytes = self._convert_webm_opus_to_ogg(audio_bytes)
             lxmf_message.fields[LXMF.FIELD_AUDIO] = [
                 audio_field.audio_mode,
-                audio_field.audio_bytes,
+                audio_bytes,
             ]
 
         # add telemetry field

meshchatx/src/backend/async_utils.py

@@ -62,11 +62,15 @@ class AsyncUtils:
         """
         if AsyncUtils.main_loop and AsyncUtils.main_loop.is_running():
             future = asyncio.run_coroutine_threadsafe(
-                coroutine, AsyncUtils.main_loop,
+                coroutine,
+                AsyncUtils.main_loop,
             )
             with AsyncUtils._futures_lock:
                 AsyncUtils._pending_futures.append(future)
-                if len(AsyncUtils._pending_futures) >= AsyncUtils._FUTURES_SWEEP_THRESHOLD:
+                if (
+                    len(AsyncUtils._pending_futures)
+                    >= AsyncUtils._FUTURES_SWEEP_THRESHOLD
+                ):
                     AsyncUtils._pending_futures = [
                         f for f in AsyncUtils._pending_futures if not f.done()
                     ]

meshchatx/src/backend/config_manager.py

@@ -317,6 +317,11 @@ class ConfigManager:
             "message_failed_bubble_color",
             "#ef4444",
         )
+        self.message_waiting_bubble_color = self.StringConfig(
+            self,
+            "message_waiting_bubble_color",
+            "#e5e7eb",
+        )
 
         # announce caps: max rows stored per aspect (oldest dropped). Default 1000.
         self.announce_max_stored_lxmf_delivery = self.IntConfig(

meshchatx/src/backend/database/__init__.py

@@ -18,6 +18,7 @@ from .misc import MiscDAO
 from .provider import DatabaseProvider
 from .ringtones import RingtoneDAO
 from .schema import DatabaseSchema
+from .stickers import UserStickersDAO
 from .telemetry import TelemetryDAO
 from .telephone import TelephoneDAO
 from .voicemails import VoicemailDAO
@@ -42,6 +43,7 @@ class Database:
         self.ringtones = RingtoneDAO(self.provider)
         self.contacts = ContactsDAO(self.provider)
         self.map_drawings = MapDrawingsDAO(self.provider)
+        self.stickers = UserStickersDAO(self.provider)
         self.debug_logs = DebugLogsDAO(self.provider)
         self.access_attempts = AccessAttemptsDAO(self.provider)
         self.crash_history = CrashHistoryDAO(self.provider)

meshchatx/src/backend/database/messages.py

@@ -59,9 +59,17 @@ class MessageDAO:
 
         self.provider.execute(query, params)
 
-    def update_lxmf_message_state(self, message_hash, state, progress,
-                                   delivery_attempts, next_delivery_attempt_at,
-                                   rssi=None, snr=None, quality=None):
+    def update_lxmf_message_state(
+        self,
+        message_hash,
+        state,
+        progress,
+        delivery_attempts,
+        next_delivery_attempt_at,
+        rssi=None,
+        snr=None,
+        quality=None,
+    ):
         """Lightweight update for delivery-state changes only.
 
         Avoids re-serializing the full message (including base64 attachment
@@ -73,8 +81,17 @@ class MessageDAO:
             "delivery_attempts = ?, next_delivery_attempt_at = ?, "
             "rssi = ?, snr = ?, quality = ?, updated_at = ? "
             "WHERE hash = ?",
-            (state, progress, delivery_attempts, next_delivery_attempt_at,
-             rssi, snr, quality, now, message_hash),
+            (
+                state,
+                progress,
+                delivery_attempts,
+                next_delivery_attempt_at,
+                rssi,
+                snr,
+                quality,
+                now,
+                message_hash,
+            ),
         )
 
     def get_lxmf_message_by_hash(self, message_hash):

meshchatx/src/backend/database/schema.py

@@ -13,7 +13,7 @@ def _validate_identifier(name: str, label: str = "identifier") -> str:
 
 
 class DatabaseSchema:
-    LATEST_VERSION = 43
+    LATEST_VERSION = 44
 
     def __init__(self, provider: DatabaseProvider):
         self.provider = provider
@@ -417,6 +417,20 @@ class DatabaseSchema:
                     updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
                 )
             """,
+            "user_stickers": """
+                CREATE TABLE IF NOT EXISTS user_stickers (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    identity_hash TEXT NOT NULL,
+                    name TEXT,
+                    image_type TEXT NOT NULL,
+                    image_blob BLOB NOT NULL,
+                    content_hash TEXT NOT NULL,
+                    source_message_hash TEXT,
+                    created_at REAL NOT NULL,
+                    updated_at REAL NOT NULL,
+                    UNIQUE(identity_hash, content_hash)
+                )
+            """,
             "lxmf_last_sent_icon_hashes": """
                 CREATE TABLE IF NOT EXISTS lxmf_last_sent_icon_hashes (
                     destination_hash TEXT PRIMARY KEY,
@@ -1119,6 +1133,28 @@ class DatabaseSchema:
                 "CREATE INDEX IF NOT EXISTS idx_lxmf_conversation_pins_pinned_at ON lxmf_conversation_pins(pinned_at)",
             )
 
+        if current_version < 44:
+            self._safe_execute("""
+                CREATE TABLE IF NOT EXISTS user_stickers (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    identity_hash TEXT NOT NULL,
+                    name TEXT,
+                    image_type TEXT NOT NULL,
+                    image_blob BLOB NOT NULL,
+                    content_hash TEXT NOT NULL,
+                    source_message_hash TEXT,
+                    created_at REAL NOT NULL,
+                    updated_at REAL NOT NULL,
+                    UNIQUE(identity_hash, content_hash)
+                )
+            """)
+            self._safe_execute(
+                "CREATE INDEX IF NOT EXISTS idx_user_stickers_identity ON user_stickers(identity_hash)",
+            )
+            self._safe_execute(
+                "CREATE INDEX IF NOT EXISTS idx_user_stickers_identity_updated ON user_stickers(identity_hash, updated_at)",
+            )
+
         # Update version in config
         self._safe_execute(
             """

meshchatx/src/backend/database/stickers.py

@@ -0,0 +1,229 @@
+import base64
+import sqlite3
+import time
+
+from meshchatx.src.backend import sticker_utils
+
+
+class UserStickersDAO:
+    def __init__(self, provider):
+        self.provider = provider
+
+    def count_for_identity(self, identity_hash: str) -> int:
+        row = self.provider.fetchone(
+            "SELECT COUNT(*) AS c FROM user_stickers WHERE identity_hash = ?",
+            (identity_hash,),
+        )
+        return int(row["c"]) if row else 0
+
+    def list_for_identity(self, identity_hash: str):
+        return self.provider.fetchall(
+            """
+            SELECT id, identity_hash, name, image_type, length(image_blob) AS image_size,
+                   content_hash, source_message_hash, created_at, updated_at
+            FROM user_stickers
+            WHERE identity_hash = ?
+            ORDER BY updated_at DESC, id DESC
+            """,
+            (identity_hash,),
+        )
+
+    def get_row(self, sticker_id: int, identity_hash: str):
+        return self.provider.fetchone(
+            """
+            SELECT id, identity_hash, name, image_type, image_blob, content_hash,
+                   source_message_hash, created_at, updated_at
+            FROM user_stickers
+            WHERE id = ? AND identity_hash = ?
+            """,
+            (sticker_id, identity_hash),
+        )
+
+    def delete(self, sticker_id: int, identity_hash: str) -> bool:
+        cur = self.provider.execute(
+            "DELETE FROM user_stickers WHERE id = ? AND identity_hash = ?",
+            (sticker_id, identity_hash),
+        )
+        return cur.rowcount > 0
+
+    def delete_all_for_identity(self, identity_hash: str) -> int:
+        cur = self.provider.execute(
+            "DELETE FROM user_stickers WHERE identity_hash = ?",
+            (identity_hash,),
+        )
+        return cur.rowcount
+
+    def update_name(
+        self, sticker_id: int, identity_hash: str, name: str | None
+    ) -> bool:
+        now = time.time()
+        cur = self.provider.execute(
+            """
+            UPDATE user_stickers
+            SET name = ?, updated_at = ?
+            WHERE id = ? AND identity_hash = ?
+            """,
+            (name, now, sticker_id, identity_hash),
+        )
+        return cur.rowcount > 0
+
+    def insert(
+        self,
+        identity_hash: str,
+        name: str | None,
+        image_type: str,
+        image_bytes: bytes,
+        source_message_hash: str | None = None,
+    ) -> dict | None:
+        """
+        Insert a sticker. Returns summary dict or None if duplicate (same content_hash).
+        """
+        if (
+            self.count_for_identity(identity_hash)
+            >= sticker_utils.MAX_STICKERS_PER_IDENTITY
+        ):
+            msg = "sticker_limit_reached"
+            raise ValueError(msg)
+
+        nt, ch = sticker_utils.validate_sticker_payload(image_bytes, image_type)
+        now = time.time()
+        try:
+            cur = self.provider.execute(
+                """
+                INSERT INTO user_stickers (
+                    identity_hash, name, image_type, image_blob, content_hash,
+                    source_message_hash, created_at, updated_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    identity_hash,
+                    name,
+                    nt,
+                    image_bytes,
+                    ch,
+                    source_message_hash,
+                    now,
+                    now,
+                ),
+            )
+        except sqlite3.IntegrityError:
+            return None
+
+        new_id = cur.lastrowid
+        row = self.provider.fetchone(
+            """
+            SELECT id, identity_hash, name, image_type, length(image_blob) AS image_size,
+                   content_hash, source_message_hash, created_at, updated_at
+            FROM user_stickers
+            WHERE id = ?
+            """,
+            (new_id,),
+        )
+        return dict(row) if row else None
+
+    def export_payloads_for_identity(self, identity_hash: str) -> list[dict]:
+        rows = self.provider.fetchall(
+            """
+            SELECT name, image_type, image_blob, source_message_hash
+            FROM user_stickers
+            WHERE identity_hash = ?
+            ORDER BY id ASC
+            """,
+            (identity_hash,),
+        )
+        out = []
+        for r in rows:
+            blob = r["image_blob"]
+            b64 = base64.b64encode(blob).decode("ascii")
+            out.append(
+                {
+                    "name": r["name"],
+                    "image_type": r["image_type"],
+                    "image_bytes": b64,
+                    "source_message_hash": r["source_message_hash"],
+                },
+            )
+        return out
+
+    def import_payloads(
+        self,
+        identity_hash: str,
+        items: list[dict],
+        *,
+        replace_duplicates: bool,
+    ) -> dict:
+        imported = 0
+        skipped_duplicates = 0
+        skipped_invalid = 0
+        errors: list[str] = []
+
+        for i, item in enumerate(items):
+            name = sticker_utils.sanitize_sticker_name(item.get("name"))
+            it = item.get("image_type")
+            b64 = item.get("image_bytes_b64")
+            src = item.get("source_message_hash")
+            try:
+                raw = base64.b64decode(b64, validate=False)
+            except (ValueError, TypeError):
+                skipped_invalid += 1
+                errors.append(f"decode_failed_at_{i}")
+                continue
+            try:
+                nt, ch = sticker_utils.validate_sticker_payload(raw, it)
+            except ValueError:
+                skipped_invalid += 1
+                errors.append(f"invalid_payload_at_{i}")
+                continue
+
+            existing = self.provider.fetchone(
+                "SELECT id FROM user_stickers WHERE identity_hash = ? AND content_hash = ?",
+                (identity_hash, ch),
+            )
+            if existing:
+                if not replace_duplicates:
+                    skipped_duplicates += 1
+                    continue
+                self.provider.execute(
+                    "DELETE FROM user_stickers WHERE identity_hash = ? AND content_hash = ?",
+                    (identity_hash, ch),
+                )
+
+            if (
+                self.count_for_identity(identity_hash)
+                >= sticker_utils.MAX_STICKERS_PER_IDENTITY
+            ):
+                errors.append("sticker_limit_reached")
+                break
+
+            now = time.time()
+            try:
+                self.provider.execute(
+                    """
+                    INSERT INTO user_stickers (
+                        identity_hash, name, image_type, image_blob, content_hash,
+                        source_message_hash, created_at, updated_at
+                    )
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (
+                        identity_hash,
+                        name,
+                        nt,
+                        raw,
+                        ch,
+                        src if isinstance(src, str) else None,
+                        now,
+                        now,
+                    ),
+                )
+                imported += 1
+            except sqlite3.IntegrityError:
+                skipped_duplicates += 1
+
+        return {
+            "imported": imported,
+            "skipped_duplicates": skipped_duplicates,
+            "skipped_invalid": skipped_invalid,
+            "errors": errors,
+        }

meshchatx/src/backend/nomadnet_downloader.py

@@ -35,7 +35,8 @@ def sweep_stale_links():
     """Evict all non-ACTIVE links from the global cache."""
     with _nomadnet_links_lock:
         stale = [
-            k for k, v in nomadnet_cached_links.items()
+            k
+            for k, v in nomadnet_cached_links.items()
             if v.status is not RNS.Link.ACTIVE
         ]
         for k in stale:

meshchatx/src/backend/sticker_utils.py

@@ -0,0 +1,179 @@
+"""Validation and hashing for user sticker images stored per identity."""
+
+from __future__ import annotations
+
+import hashlib
+import base64
+
+MAX_STICKER_BYTES = 512 * 1024
+MAX_STICKERS_PER_IDENTITY = 2000
+
+_ALLOWED_TYPES = frozenset({"png", "jpeg", "jpg", "gif", "webp", "bmp"})
+
+_TYPE_ALIASES = {
+    "jpeg": "jpeg",
+    "jpg": "jpeg",
+    "pjpeg": "jpeg",
+    "png": "png",
+    "gif": "gif",
+    "webp": "webp",
+    "bmp": "bmp",
+}
+
+
+def normalize_image_type(image_type: str | None) -> str | None:
+    if not image_type:
+        return None
+    t = str(image_type).strip().lower()
+    t = t.removeprefix("image/")
+    t = _TYPE_ALIASES.get(t, t)
+    return t if t in _ALLOWED_TYPES else None
+
+
+def content_hash_hex(image_bytes: bytes) -> str:
+    return hashlib.sha256(image_bytes).hexdigest()
+
+
+def detect_image_format_from_magic(image_bytes: bytes) -> str | None:
+    """
+    Identify image format from file signature (magic bytes). Returns normalized
+    type key (png, jpeg, gif, webp, bmp) or None if unknown / too short / not allowed.
+    """
+    if not isinstance(image_bytes, (bytes, bytearray)) or len(image_bytes) < 4:
+        return None
+    b = bytes(image_bytes)
+    if len(b) >= 8 and b[:8] == b"\x89PNG\r\n\x1a\n":
+        return "png"
+    if len(b) >= 3 and b[0:3] == b"\xff\xd8\xff":
+        return "jpeg"
+    if len(b) >= 6 and b[0:6] in (b"GIF87a", b"GIF89a"):
+        return "gif"
+    if len(b) >= 12 and b[0:4] == b"RIFF" and b[8:12] == b"WEBP":
+        return "webp"
+    if len(b) >= 2 and b[0:2] == b"BM":
+        return "bmp"
+    return None
+
+
+def validate_sticker_payload(
+    image_bytes: bytes,
+    image_type: str | None,
+) -> tuple[str, str]:
+    """
+    Returns (normalized_image_type, content_hash_hex).
+
+    Declared image_type must match the format detected from magic bytes; stored
+    type is the normalized detected format.
+
+    Raises ValueError with a short reason on invalid input.
+    """
+    if not isinstance(image_bytes, (bytes, bytearray)):
+        msg = "invalid_image_bytes"
+        raise ValueError(msg)
+    if len(image_bytes) == 0:
+        msg = "empty_image"
+        raise ValueError(msg)
+    if len(image_bytes) > MAX_STICKER_BYTES:
+        msg = "image_too_large"
+        raise ValueError(msg)
+
+    nt = normalize_image_type(image_type)
+    if not nt:
+        msg = "invalid_image_type"
+        raise ValueError(msg)
+
+    detected = detect_image_format_from_magic(image_bytes)
+    if not detected:
+        msg = "invalid_image_signature"
+        raise ValueError(msg)
+    if detected != nt:
+        msg = "magic_type_mismatch"
+        raise ValueError(msg)
+
+    h = content_hash_hex(bytes(image_bytes))
+    return detected, h
+
+
+_EXPORT_FORMAT = "meshchatx-stickers"
+_EXPORT_VERSION = 1
+
+
+def validate_export_document(data: object) -> list[dict]:
+    """
+    Parse and lightly validate an import JSON document.
+    Returns a list of sticker dicts with keys: name, image_type, image_bytes (str base64),
+    source_message_hash (optional).
+    """
+    if not isinstance(data, dict):
+        msg = "invalid_document"
+        raise ValueError(msg)
+    if data.get("format") != _EXPORT_FORMAT:
+        msg = "invalid_format"
+        raise ValueError(msg)
+    try:
+        ver = int(data["version"])
+    except (KeyError, TypeError, ValueError) as exc:
+        msg = "unsupported_version"
+        raise ValueError(msg) from exc
+    if ver != _EXPORT_VERSION:
+        msg = "unsupported_version"
+        raise ValueError(msg)
+    stickers = data.get("stickers")
+    if not isinstance(stickers, list):
+        msg = "invalid_stickers_array"
+        raise ValueError(msg)
+    out: list[dict] = []
+    for i, item in enumerate(stickers):
+        if not isinstance(item, dict):
+            msg = f"invalid_sticker_at_{i}"
+            raise ValueError(msg)
+        name = item.get("name")
+        image_type = item.get("image_type")
+        image_b64 = item.get("image_bytes")
+        if not isinstance(image_b64, str) or not image_b64.strip():
+            msg = f"missing_image_bytes_at_{i}"
+            raise ValueError(msg)
+        try:
+            base64.b64decode(image_b64.strip(), validate=False)
+        except (ValueError, TypeError) as exc:
+            msg = f"invalid_base64_at_{i}"
+            raise ValueError(msg) from exc
+        src = item.get("source_message_hash")
+        out.append(
+            {
+                "name": name if isinstance(name, str) else None,
+                "image_type": image_type,
+                "image_bytes_b64": image_b64.strip(),
+                "source_message_hash": src if isinstance(src, str) else None,
+            },
+        )
+    return out
+
+
+def build_export_document(stickers: list[dict], exported_at_iso: str) -> dict:
+    """stickers: rows with name, image_type, image_bytes (base64 str), source_message_hash."""
+    return {
+        "format": _EXPORT_FORMAT,
+        "version": _EXPORT_VERSION,
+        "exported_at": exported_at_iso,
+        "stickers": stickers,
+    }
+
+
+def mime_for_image_type(normalized_type: str) -> str:
+    return {
+        "jpeg": "image/jpeg",
+        "png": "image/png",
+        "gif": "image/gif",
+        "webp": "image/webp",
+        "bmp": "image/bmp",
+    }.get(normalized_type, "application/octet-stream")
+
+
+def sanitize_sticker_name(name: str | None) -> str | None:
+    if name is None:
+        return None
+    s = "".join(ch for ch in str(name).strip() if ch.isprintable())
+    if not s:
+        return None
+    return s[:128]