Demote Signal receipt/typing/sync spam from INFO to DEBUG

signalbot logs every received envelope at INFO via `[Raw Message] {json}`. On a
busy account the delivery/read receipts, typing indicators, and read-sync
messages flood journald hard enough to rotate the journal down to ~1 minute,
burying real traffic and making failures impossible to debug after the fact.

Add a logging filter on the `signalbot` logger that suppresses content-free
envelopes (anything without a `dataMessage`) at INFO, while still emitting them
if the logger is explicitly set to DEBUG. Real message content stays at INFO.

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

.gitignore

@@ -8,6 +8,10 @@ signal-cli-data.bak.*
 # Local service file (contains your phone number) — use signal-bot.service.example
 signal-bot.service
 
+# Runtime state (SQLite dedup cache) — created next to bot.py at runtime
+bot-state.db
+bot-state.db-*
+
 # Python
 venv/
 __pycache__/

README.md

@@ -50,7 +50,11 @@ Signal  ──►  signal-cli-rest-api (json-rpc, :8080)  ──►  bot.py (sig
    python3 -m venv venv
    venv/bin/pip install -r requirements.txt
    ```
-   YouTube also needs a system `node` on PATH (for `yt-dlp-ejs`).
+   YouTube also needs **Node >= 22** on PATH (for `yt-dlp-ejs`'s challenge
+   solver). If your distro only ships an older Node, install a standalone Node 22+
+   and point `bot.py`'s `_NODE22` at it. Note: YouTube cookies are deliberately
+   withheld (`_cookie_args`) — sending account cookies forces yt-dlp onto a player
+   client whose media URLs 403; cookieless YouTube requests work.
 
 3. **Cookies:** copy `cookies.txt.example` to `cookies.txt` and fill in real
    exported cookies for the sites you want auth'd. (gitignored)

bot.py

@@ -5,9 +5,14 @@ import json
 import logging
 import os
 import re
+import secrets
+import shlex
+import shutil
 import subprocess
 import tempfile
 import time
+from datetime import datetime
+from urllib.parse import quote, urlsplit
 
 from signalbot import Command, Context, SignalBot
 from signalbot.command import regex_triggered, triggered
@@ -18,7 +23,11 @@ from signalbot.message import MessageType
 # (?:/[^/\s]+)* allows multi-segment forms like x.com/i/web/status/<id> while
 # still refusing whitespace (so two links can't merge).
 TWITTER_URL_PATTERN = r"https?://(?:www\.)?(?:twitter\.com|x\.com|fxtwitter\.com|vxtwitter\.com|fixupx\.com)/[^/\s]+(?:/[^/\s]+)*/status/\d+"
-INSTAGRAM_URL_PATTERN = r"https?://(?:www\.)?instagram\.com/(?:reel|p)/[\w-]+"
+# Instagram shares come in several path shapes: /p/ (posts), /reel/ (singular),
+# /reels/ (plural — the form the app's share sheet now emits), /tv/ (IGTV), and a
+# /share/ prefix on any of them. Match all of them or the link is silently dropped
+# (regex never matches -> no download, no error reply).
+INSTAGRAM_URL_PATTERN = r"https?://(?:www\.)?instagram\.com/(?:share/)?(?:reels?|p|tv)/[\w-]+"
 YOUTUBE_URL_PATTERN = r"https?://(?:www\.)?(?:youtube\.com/(?:watch\?v=|shorts/)|youtu\.be/)[\w-]+"
 TIKTOK_URL_PATTERN = r"https?://(?:(?:www|m)\.tiktok\.com/(?:@[\w.-]+/video/\d+|t/\w+|v/\d+)|(?:vm|vt)\.tiktok\.com/\w+)"
 VIDEO_URL_PATTERN = rf"(?:{TWITTER_URL_PATTERN}|{INSTAGRAM_URL_PATTERN}|{YOUTUBE_URL_PATTERN}|{TIKTOK_URL_PATTERN})"
@@ -30,23 +39,100 @@ MAX_CLIP_DURATION = 600  # ceiling for a user-supplied /clip override
 MAX_CONCURRENT_JOBS = 2
 YTDLP = os.path.join(os.path.dirname(os.path.abspath(__file__)), "venv", "bin", "yt-dlp")
 COOKIES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cookies.txt")
+STATE_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bot-state.db")
 ADMIN_NUMBERS = {n.strip() for n in os.environ.get("BOT_ADMINS", "").split(",") if n.strip()}
 
+# YouTube's JS "n-sig" challenge solver (yt-dlp-ejs) requires Node >= 22, but
+# Debian 13 only ships Node 20. We install a standalone Node 22 at /opt/node22 and
+# point yt-dlp at it explicitly via `--js-runtimes node:<path>`; fall back to PATH
+# `node` if that binary is ever missing.
+_NODE22 = "/opt/node22/bin/node"
+JS_RUNTIME = f"node:{_NODE22}" if os.path.exists(_NODE22) else "node"
+
+# YouTube is the one host where we must NOT send cookies: with account cookies
+# present, yt-dlp skips the cookieless `android_vr`/`android` clients (whose media
+# URLs work) and falls back to the `tv` client, whose URLs now 403 on the data
+# download. A cookieless YouTube request just works. Cookies are still needed for
+# Instagram/TikTok/X, so we withhold them per-URL rather than dropping the file.
+_YOUTUBE_HOST_RE = re.compile(r"(?:^|\.)(?:youtube\.com|youtu\.be|youtube-nocookie\.com)$", re.I)
+
+def _cookie_args(url: str) -> list[str]:
+    """yt-dlp --cookies args for this URL, withheld for YouTube (see note above)."""
+    if not os.path.exists(COOKIES):
+        return []
+    host = (urlsplit(url).hostname or "").lower()
+    if _YOUTUBE_HOST_RE.search(host):
+        return []
+    return ["--cookies", COOKIES]
+
+# --- /force: host an oversized video as a temporary download link ---
+FORCE_DURATION_THRESHOLD = 3600   # offer /force for videos longer than this (1 hr)
+FORCE_OFFER_TTL = 600             # a /force offer stays valid for 10 min
+FORCE_TIMEOUT = 1800             # max seconds for a forced full download (30 min)
+FORCE_MAX_FILESIZE = "10G"        # yt-dlp aborts a single download larger than this
+RSYNC_TIMEOUT = 1800             # max seconds to upload to the web box
+FORCE_STAGE_ROOT = "/var/tmp/signal-bot-force"  # disk-backed staging (/tmp is tmpfs/RAM)
+MEDIA_TTL = 24 * 3600             # delete hosted files after 24 h
+MEDIA_MAX_BYTES = 10 * 1024 * 1024 * 1024  # evict oldest once the hosted folder exceeds 10 GB
+TOKEN_RE = re.compile(r"^[0-9a-f]{24}$")   # cleanup only ever touches dirs we created
+
+# /force hosting target — configured via env so the published code carries no
+# infra details. The systemd unit supplies the real values; unset = feature off.
+FORCE_REMOTE = os.environ.get("FORCE_REMOTE", "")             # ssh target, e.g. "user@host"
+FORCE_REMOTE_DIR = os.environ.get("FORCE_REMOTE_DIR", "")     # web docroot on that host
+FORCE_BASE_URL = os.environ.get("FORCE_BASE_URL", "").rstrip("/")  # public URL for that dir
+FORCE_SSH_PORT = os.environ.get("FORCE_SSH_PORT", "22")
+FORCE_SSH_KEY = os.environ.get("FORCE_SSH_KEY", "")          # ssh identity file (optional)
+FORCE_ENABLED = bool(FORCE_REMOTE and FORCE_REMOTE_DIR and FORCE_BASE_URL)
+_SSH_OPTS = ["-p", FORCE_SSH_PORT, "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=accept-new",
+             *(["-i", FORCE_SSH_KEY] if FORCE_SSH_KEY else [])]
+_RSYNC_RSH = "ssh " + " ".join(shlex.quote(o) for o in _SSH_OPTS)
+
+# /force is known to be big and slow, so run its subprocesses (yt-dlp + the node
+# n-sig solver and ffmpeg merge it spawns, plus rsync) at lowest CPU and idle IO
+# priority. They yield to everything else under load but still run full-speed when
+# the box is idle; child processes inherit the niceness. Normal short downloads
+# are left at full priority for responsiveness.
+_NICE = (["nice", "-n", "19"] if shutil.which("nice") else []) + \
+        (["ionice", "-c", "3"] if shutil.which("ionice") else [])
+
 logging.basicConfig(
     level=logging.INFO,
     format="%(asctime)s [%(levelname)s] %(message)s",
 )
 log = logging.getLogger("signal-bot")
 
+
+class _RawReceiptNoiseFilter(logging.Filter):
+    """signalbot logs every received envelope at INFO via `[Raw Message] {json}`.
+    On a busy account that's a flood of delivery/read receipts, typing indicators,
+    and read-sync messages that rotates the journal down to ~1 minute and buries
+    real traffic. Suppress the content-free envelopes (anything without a
+    `dataMessage`) unless the signalbot logger is explicitly set to DEBUG — so they
+    stay capturable when you actually want to debug raw traffic, but don't spam INFO."""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        msg = record.getMessage()
+        if msg.startswith("[Raw Message] ") and '"dataMessage"' not in msg:
+            return logging.getLogger("signalbot").isEnabledFor(logging.DEBUG)
+        return True
+
+
+logging.getLogger("signalbot").addFilter(_RawReceiptNoiseFilter())
+
 # (group_id, sender) -> {"b64": ..., "time": monotonic}. Keyed per sender so a
 # stranger's later video doesn't silently become the target of your /speed or
 # /rev; _get_video falls back to the group's latest video when you have none.
+# Intentionally in-memory ONLY: each value is a base64 video up to ~133 MB, so
+# persisting these to SQLite would bloat the DB. Only the dedup map below persists.
 last_video = {}
 VIDEO_TTL = 3600  # 1 hour
 
-# (group_id, url) -> monotonic time the URL was successfully handled.
+# (group_id, url) -> wall-clock epoch seconds the URL was successfully handled.
 # When a user edits a message, signal-cli redelivers it as MessageType.EDIT_MESSAGE
 # with the same text — without this guard the bot re-downloads and re-posts the video.
+# Persisted to SQLite so a restart doesn't forget recent dedup; tiny and TTL-bounded.
+# Wall-clock (not monotonic) so the stored timestamps stay meaningful across restarts.
 recent_urls = {}
 RECENT_URL_TTL = 600  # 10 min
 
@@ -57,6 +143,24 @@ _inflight = set()
 # Lazily created so it binds to the bot's running event loop, not import time.
 _job_semaphore = None
 
+# signalbot SQLite key/value store (set in main()); persists recent_urls only.
+_storage = None
+_RECENT_URLS_KEY = "recent_urls"
+_RECENT_SEP = "\x1f"  # joins (group, url) into one JSON-safe storage key
+
+# (group_id, sender) -> {"url","title","duration","time"}: a pending /force offer.
+pending_force = {}
+
+# Separate 1-at-a-time slot for forced downloads so a long one can't starve clips.
+_force_semaphore = None
+
+
+def _get_force_semaphore() -> asyncio.Semaphore:
+    global _force_semaphore
+    if _force_semaphore is None:
+        _force_semaphore = asyncio.Semaphore(1)
+    return _force_semaphore
+
 
 def _get_job_semaphore() -> asyncio.Semaphore:
     global _job_semaphore
@@ -92,11 +196,43 @@ def _get_video(group_id, sender):
 
 
 def _sweep_recent_urls() -> None:
-    now = time.monotonic()
+    now = time.time()
     for key in [k for k, t in recent_urls.items() if now - t > RECENT_URL_TTL]:
         del recent_urls[key]
 
 
+def _load_recent_urls() -> None:
+    """Restore the (pruned) dedup map from SQLite on startup."""
+    if _storage is None:
+        return
+    try:
+        if not _storage.exists(_RECENT_URLS_KEY):
+            return
+        data = _storage.read(_RECENT_URLS_KEY) or {}
+    except Exception as e:  # noqa: BLE001
+        log.warning("Could not load dedup state: %s", e)
+        return
+    now = time.time()
+    for joined, ts in data.items():
+        if not isinstance(ts, (int, float)) or now - ts > RECENT_URL_TTL:
+            continue
+        group_id, _, url = joined.partition(_RECENT_SEP)
+        recent_urls[(group_id, url)] = ts
+    log.info("Restored %d dedup entries from %s", len(recent_urls), STATE_DB)
+
+
+def _persist_recent_urls() -> None:
+    if _storage is None:
+        return
+    try:
+        _storage.save(
+            _RECENT_URLS_KEY,
+            {f"{g}{_RECENT_SEP}{u}": ts for (g, u), ts in recent_urls.items()},
+        )
+    except Exception as e:  # noqa: BLE001
+        log.warning("Could not persist dedup state: %s", e)
+
+
 def _url_recently_handled(group_id, url) -> bool:
     _sweep_recent_urls()
     return (group_id, url) in recent_urls
@@ -108,7 +244,26 @@ def _url_busy(group_id, url) -> bool:
 
 def _mark_url_handled(group_id, url) -> None:
     _sweep_recent_urls()
-    recent_urls[(group_id, url)] = time.monotonic()
+    recent_urls[(group_id, url)] = time.time()
+    _persist_recent_urls()
+
+
+def _sweep_pending_force() -> None:
+    now = time.monotonic()
+    for k in [k for k, v in pending_force.items() if now - v["time"] > FORCE_OFFER_TTL]:
+        del pending_force[k]
+
+
+def _set_pending_force(group_id, sender, url, title, duration) -> None:
+    _sweep_pending_force()
+    pending_force[(group_id, sender)] = {
+        "url": url, "title": title, "duration": duration, "time": time.monotonic(),
+    }
+
+
+def _pop_pending_force(group_id, sender):
+    _sweep_pending_force()
+    return pending_force.pop((group_id, sender), None)
 
 
 async def _safe_reply(c: Context, text: str) -> None:
@@ -303,17 +458,53 @@ class VideoCommand(Command):
         key = (group, url)
         _inflight.add(key)
         try:
+            # Clip path: a bounded window, just grab it (no oversize concern).
             if clip is not None:
                 log.info("Clipping %s to window %d-%ds", url, clip[0], clip[1])
                 async with _get_job_semaphore():
                     b64, err, silent = await asyncio.to_thread(_produce_video, url, clip)
+                await self._deliver(c, group, sender, url, b64, err, silent,
+                                    title=None, duration=None, allow_force=False)
+                return
 
+            # Full-video path. Only when /force hosting is enabled do we probe
+            # duration first (to fast-path an over-long video to a /force offer);
+            # otherwise behave like before and just download.
+            async with _get_job_semaphore():
+                if FORCE_ENABLED:
+                    duration, title = await asyncio.to_thread(_probe_metadata, url)
+                    too_long = duration is not None and duration > FORCE_DURATION_THRESHOLD
+                else:
+                    duration, title, too_long = None, None, False
+                if too_long:
+                    b64, err, silent = None, "", False
+                else:
+                    b64, err, silent = await asyncio.to_thread(_produce_video, url, clip)
+
+            if too_long:
+                _mark_url_handled(group, url)
+                await _too_big_response(c, group, sender, url, title, duration)
+                return
+            await self._deliver(c, group, sender, url, b64, err, silent,
+                                title=title, duration=duration, allow_force=True)
+        except Exception as e:  # noqa: BLE001
+            log.exception("Unexpected error handling %s: %s", url, e)
+            await _safe_reply(c, "Something went wrong handling that video.")
+        finally:
+            _inflight.discard(key)
+
+    async def _deliver(self, c: Context, group, sender, url, b64, err, silent,
+                       *, title, duration, allow_force) -> None:
         if silent:
             # A genuine no-media determination: record it so an edit replay
             # doesn't pointlessly re-run yt-dlp.
             _mark_url_handled(group, url)
             return
         if err:
+            if allow_force and _is_too_big_error(err):
+                _mark_url_handled(group, url)
+                await _too_big_response(c, group, sender, url, title, duration)
+            else:
                 # Hard error: do NOT mark handled, so a corrective edit can retry.
                 await _safe_reply(c, f"Couldn't grab that video: {err}")
             return
@@ -321,11 +512,6 @@ class VideoCommand(Command):
         _mark_url_handled(group, url)
         _set_video(group, sender, b64)
         await _safe_send_video(c, b64)
-        except Exception as e:  # noqa: BLE001
-            log.exception("Unexpected error handling %s: %s", url, e)
-            await _safe_reply(c, "Something went wrong handling that video.")
-        finally:
-            _inflight.discard(key)
 
 
 def _run_ytdlp(url: str, outpath: str, tmpdir: str,
@@ -339,7 +525,7 @@ def _run_ytdlp(url: str, outpath: str, tmpdir: str,
         "--no-playlist",
         # YouTube wraps URLs in a JS "n-sig" challenge; node solves it
         # via yt-dlp-ejs. Without this, only image/thumb formats resolve.
-        "--js-runtimes", "node",
+        "--js-runtimes", JS_RUNTIME,
     ]
     if clip is not None:
         start, end = clip
@@ -362,7 +548,7 @@ def _run_ytdlp(url: str, outpath: str, tmpdir: str,
         ]
     cmd += [
         "--merge-output-format", "mp4",
-        *(["--cookies", COOKIES] if os.path.exists(COOKIES) else []),
+        *_cookie_args(url),
         "-o", outpath,
         "--", url,
     ]
@@ -438,6 +624,190 @@ def _produce_video(url: str, clip: tuple[int, int] | None) -> tuple[str | None,
             return base64.b64encode(f.read()).decode("utf-8"), None, False
 
 
+def _probe_metadata(url: str) -> tuple[int | None, str | None]:
+    """Fetch (duration_seconds, title) without downloading. Blocking — use to_thread."""
+    cmd = [
+        YTDLP, "--no-playlist", "--quiet", "--no-warnings",
+        "--js-runtimes", JS_RUNTIME,
+        "--print", "%(duration)s\n%(title)s",
+        *_cookie_args(url),
+        "--", url,
+    ]
+    try:
+        r = subprocess.run(cmd, capture_output=True, text=True, timeout=90)
+    except subprocess.TimeoutExpired:
+        return None, None
+    if r.returncode != 0:
+        return None, None
+    lines = r.stdout.splitlines()
+    duration = None
+    if lines:
+        try:
+            duration = int(float(lines[0]))
+        except ValueError:
+            duration = None
+    title = lines[1].strip() if len(lines) > 1 and lines[1].strip() not in ("", "NA") else None
+    return duration, title
+
+
+def _fmt_duration(seconds: int | None) -> str:
+    if not seconds:
+        return ""
+    h, rem = divmod(int(seconds), 3600)
+    m, s = divmod(rem, 60)
+    return f"{h}h{m:02d}m" if h else f"{m}m{s:02d}s"
+
+
+def _too_big_offer(title: str | None, duration: int | None) -> str:
+    name = f' "{title}"' if title else ""
+    dur = f" ({_fmt_duration(duration)})" if duration else ""
+    return (
+        f"That video{name}{dur} is too big to post here. "
+        f"Reply `/force` within 10 min and I'll download it and give you a temporary "
+        f"link (auto-deleted in 24h)."
+    )
+
+
+# Bot-generated oversize/slow sentinels only — NOT yt-dlp's transient network
+# "read operation timed out", which should stay a retryable hard error.
+_TOO_BIG_PATTERNS = ("too large", "after re-encode", "yt-dlp timed out after", "ffmpeg timed out")
+
+
+def _is_too_big_error(err: str) -> bool:
+    e = (err or "").lower()
+    return any(p in e for p in _TOO_BIG_PATTERNS)
+
+
+async def _too_big_response(c: Context, group, sender, url, title, duration) -> None:
+    """Reply to an oversized video: offer /force if hosting is set up, else just note it."""
+    if FORCE_ENABLED:
+        _set_pending_force(group, sender, url, title, duration)
+        await _safe_reply(c, _too_big_offer(title, duration))
+    else:
+        name = f' "{title}"' if title else ""
+        dur = f" ({_fmt_duration(duration)})" if duration else ""
+        await _safe_reply(c, f"That video{name}{dur} is too big to post here.")
+
+
+def _force_download_and_publish(url: str, token: str) -> tuple[str | None, int | None, str | None]:
+    """Blocking: download the full video, upload it to the web box, return
+    (public_url, size_mb, error). Stages on disk (/tmp is RAM) and cleans up."""
+    os.makedirs(FORCE_STAGE_ROOT, exist_ok=True)
+    # Self-heal: drop staging dirs orphaned by a hard crash (only one force at a time).
+    for old in os.listdir(FORCE_STAGE_ROOT):
+        if old.startswith("force-"):
+            shutil.rmtree(os.path.join(FORCE_STAGE_ROOT, old), ignore_errors=True)
+    stage = tempfile.mkdtemp(dir=FORCE_STAGE_ROOT, prefix="force-")
+    try:
+        tokdir = os.path.join(stage, token)
+        os.makedirs(tokdir)
+        cmd = [
+            *_NICE,
+            YTDLP, "--no-playlist", "--no-warnings", "--js-runtimes", JS_RUNTIME,
+            "-f", "best[ext=mp4]/best", "--merge-output-format", "mp4",
+            "--max-filesize", FORCE_MAX_FILESIZE,
+            *_cookie_args(url),
+            "-o", os.path.join(tokdir, "video.%(ext)s"),
+            "--", url,
+        ]
+        try:
+            r = subprocess.run(cmd, capture_output=True, text=True, timeout=FORCE_TIMEOUT)
+        except subprocess.TimeoutExpired:
+            return None, None, f"timed out after {FORCE_TIMEOUT // 60} min"
+
+        files = [f for f in os.listdir(tokdir) if os.path.isfile(os.path.join(tokdir, f))]
+        if r.returncode != 0 or not files:
+            # yt-dlp prints the max-filesize abort to stdout (and exits 0), so check both.
+            blob = (r.stdout + r.stderr).lower()
+            if "larger than" in blob or "max-filesize" in blob:
+                return None, None, f"video is larger than the {FORCE_MAX_FILESIZE}B hosting limit"
+            return None, None, _summarize_ytdlp_error(r.stderr)
+
+        # Pick the actual video (largest file), not an arbitrary listdir entry.
+        fname = max(files, key=lambda f: os.path.getsize(os.path.join(tokdir, f)))
+        size_mb = os.path.getsize(os.path.join(tokdir, fname)) // (1024 * 1024)
+
+        up = subprocess.run(
+            [*_NICE, "rsync", "-a", "--chmod=F644,D755", "-e", _RSYNC_RSH,
+             tokdir, f"{FORCE_REMOTE}:{FORCE_REMOTE_DIR}/"],
+            capture_output=True, text=True, timeout=RSYNC_TIMEOUT,
+        )
+        if up.returncode != 0:
+            return None, None, f"upload failed ({up.stderr.strip()[-160:]})"
+
+        return f"{FORCE_BASE_URL}/{token}/{quote(fname)}", size_mb, None
+    except Exception as e:  # noqa: BLE001
+        log.exception("forced download failed for %s", url)
+        return None, None, f"unexpected error ({e})"
+    finally:
+        shutil.rmtree(stage, ignore_errors=True)
+
+
+def _cleanup_media() -> None:
+    """Delete hosted videos older than MEDIA_TTL and evict the oldest past
+    MEDIA_MAX_BYTES. Only ever touches token dirs we created. Blocking."""
+    ssh_base = ["ssh", *_SSH_OPTS, FORCE_REMOTE]
+    # Emit "age_seconds<TAB>bytes<TAB>name" using the REMOTE clock, so the TTL
+    # decision doesn't depend on this host's clock matching the web box's.
+    list_script = (
+        f"cd {shlex.quote(FORCE_REMOTE_DIR)} 2>/dev/null || exit 0; now=$(date +%s); "
+        f'for d in */; do [ -d "$d" ] || continue; n="${{d%/}}"; '
+        f'printf "%s\\t%s\\t%s\\n" "$((now - $(stat -c %Y "$d")))" '
+        f'"$(du -sb "$d" 2>/dev/null | cut -f1)" "$n"; done'
+    )
+    try:
+        r = subprocess.run([*ssh_base, list_script], capture_output=True, text=True, timeout=120)
+    except subprocess.TimeoutExpired:
+        log.warning("media cleanup: listing timed out")
+        return
+    if r.returncode != 0:
+        log.warning("media cleanup: listing failed (%s)", r.stderr.strip()[-160:])
+        return
+
+    entries = []  # (age_seconds, bytes, name)
+    for line in r.stdout.splitlines():
+        parts = line.split("\t")
+        if len(parts) != 3 or not TOKEN_RE.match(parts[2]):
+            continue
+        try:
+            age = int(parts[0])
+        except ValueError:
+            continue  # no usable age -> can't make a TTL decision, skip this round
+        # A du failure leaves the size field empty; treat as 0 so TTL can still fire.
+        nbytes = int(parts[1]) if parts[1].isdigit() else 0
+        entries.append((age, nbytes, parts[2]))
+
+    doomed = {name for age, _b, name in entries if age > MEDIA_TTL}
+    fresh = sorted(e for e in entries if e[2] not in doomed)  # ascending age = newest first
+    total = 0
+    for _age, nbytes, name in fresh:
+        total += nbytes
+        if total > MEDIA_MAX_BYTES:
+            doomed.add(name)
+
+    doomed = [n for n in doomed if TOKEN_RE.match(n)]
+    if not doomed:
+        return
+    paths = " ".join(shlex.quote(f"{FORCE_REMOTE_DIR}/{n}") for n in doomed)
+    try:
+        rm = subprocess.run([*ssh_base, f"rm -rf -- {paths}"],
+                            capture_output=True, text=True, timeout=120)
+    except subprocess.TimeoutExpired:
+        log.warning("media cleanup: rm timed out")
+        return
+    if rm.returncode == 0:
+        log.info("media cleanup: removed %d hosted video(s)", len(doomed))
+    else:
+        log.warning("media cleanup: rm failed (%s)", rm.stderr.strip()[-160:])
+
+
+async def _cleanup_media_job() -> None:
+    try:
+        await asyncio.to_thread(_cleanup_media)
+    except Exception as e:  # noqa: BLE001
+        log.warning("media cleanup job error: %s", e)
+
+
 def _reencode(input_file: str, tmpdir: str) -> tuple[str | None, str]:
     """Re-encode video with ffmpeg to fit under MAX_FILE_SIZE.
 
@@ -700,6 +1070,43 @@ class ReverseCommand(Command):
         await _safe_send_video(c, out)
 
 
+class ForceCommand(Command):
+    """`/force` downloads a video the bot refused as too big and hosts it as a
+    temporary link (auto-deleted in 24h)."""
+    async def handle(self, c: Context) -> None:
+        if not c.message.is_group():
+            return
+        if (c.message.text or "").strip().lower() != "/force":
+            return
+        if not FORCE_ENABLED:
+            await _safe_reply(c, "Big-video hosting isn't set up, sorry.")
+            return
+
+        group = c.message.group
+        sender = _sender_number(c.message)
+        pend = _pop_pending_force(group, sender)
+        if not pend:
+            await _safe_reply(c, "Nothing to force (offers expire after 10 min). Share the video link again and I'll re-offer `/force` if it's too big.")
+            return
+
+        url = pend["url"]
+        title = pend.get("title")
+        duration = pend.get("duration")
+        nm = f' "{title}"' if title else ""
+        dur = f" ({_fmt_duration(duration)})" if duration else ""
+        await _safe_reply(c, f"Downloading{nm}{dur}… this may take a while.")
+
+        token = secrets.token_hex(12)
+        async with _get_force_semaphore():
+            public_url, size_mb, err = await asyncio.to_thread(_force_download_and_publish, url, token)
+        if err:
+            # Re-arm the offer so a transient failure (network/rsync) can be retried.
+            _set_pending_force(group, sender, url, title, duration)
+            await _safe_reply(c, f"Couldn't download that video: {err}. Reply `/force` to try again.")
+            return
+        await _safe_reply(c, f"{public_url}\n({size_mb} MB — will be deleted in 24 hours)")
+
+
 HELP_TEXT = f"""🎬 Video bot — what I can do
 
 Share a video link (X/Twitter, Instagram, YouTube, TikTok) and I'll post the video back to the group.
@@ -717,6 +1124,8 @@ A YouTube link with a timestamp → I post a {CLIP_DURATION}s clip starting at t
 
 /rev — reverse the last video.
 
+/force — if a video's too big to post (e.g. over an hour), I'll offer this; reply `/force` and I'll download it and post a temporary link (auto-deleted in 24h).
+
 /help — show this message.
 
 (In a DM, admins can run /cookies to refresh Instagram login cookies.)"""
@@ -796,6 +1205,8 @@ class CookiesCommand(Command):
 
 
 def main():
+    global _storage
+
     phone_number = os.environ.get("SIGNAL_PHONE_NUMBER")
     signal_service = os.environ.get("SIGNAL_SERVICE", "127.0.0.1:8080")
 
@@ -810,13 +1221,28 @@ def main():
         # Don't let the library download+base64 every attachment of every message
         # on the producer loop. VideoTracker fetches only video attachments lazily.
         "download_attachments": False,
+        # Tiny SQLite KV (a few KB): persists only the URL dedup map across
+        # restarts. The big last-video blobs stay in memory by design.
+        "storage": {"type": "sqlite", "sqlite_db": STATE_DB},
     })
+    _storage = bot.storage
+    _load_recent_urls()
     bot.register(VideoTracker(), contacts=False, groups=True)
     bot.register(VideoCommand(), contacts=False, groups=True)
     bot.register(ReverseCommand(), contacts=False, groups=True)
     bot.register(SpeedCommand(), contacts=False, groups=True)
+    bot.register(ForceCommand(), contacts=False, groups=True)
     bot.register(HelpCommand(), contacts=False, groups=True)
     bot.register(CookiesCommand(), contacts=True, groups=False)
+
+    # Sweep hosted /force videos hourly (delete >24h old, evict oldest past 10 GB),
+    # starting shortly after launch. Only when /force hosting is configured.
+    if FORCE_ENABLED:
+        bot.scheduler.add_job(_cleanup_media_job, "interval", hours=1, next_run_time=datetime.now())
+        log.info("/force hosting enabled -> %s", FORCE_BASE_URL)
+    else:
+        log.info("/force hosting disabled (FORCE_REMOTE/FORCE_REMOTE_DIR/FORCE_BASE_URL unset)")
+
     log.info("Starting Signal video bot...")
     bot.start()
 

requirements.txt

@@ -1,6 +1,9 @@
 signalbot==1.1.0
 # yt-dlp is kept current for site/extractor changes; bump the floor deliberately.
-yt-dlp>=2026.3.17
+# 2026.6.9 fixes a YouTube 403-on-data-download regression; it also requires the
+# JS challenge to run on Node >= 22 (see below).
+yt-dlp>=2026.6.9
 # Needed for YouTube: yt-dlp wraps URLs in a JS "n-sig" challenge that a JS
-# runtime must solve. Requires a system `node` on PATH plus this package.
+# runtime must solve. Requires Node >= 22 on PATH (bot.py points at a standalone
+# /opt/node22 since Debian 13 only ships Node 20) plus this package.
 yt-dlp-ejs>=0.8.0

signal-bot.service.example

@@ -12,6 +12,15 @@ Environment=SIGNAL_PHONE_NUMBER=+15551234567
 Environment=SIGNAL_SERVICE=127.0.0.1:8080
 # Optional: comma-separated numbers allowed to run /cookies (admin command).
 # Environment=BOT_ADMINS=+15551234567
+# Optional: enable /force, which hosts oversized videos as temporary links.
+# The bot rsyncs the file (over SSH) into a web-served directory and posts the URL;
+# an hourly job deletes anything older than 24h and evicts oldest past 10 GB.
+# Leave unset to disable /force. All three of REMOTE/REMOTE_DIR/BASE_URL are required.
+# Environment=FORCE_REMOTE=user@webhost            # ssh target that can write the docroot
+# Environment=FORCE_REMOTE_DIR=/var/www/site/media # docroot on that host
+# Environment=FORCE_BASE_URL=https://site/media    # public URL serving that docroot
+# Environment=FORCE_SSH_PORT=22                     # ssh port (default 22)
+# Environment=FORCE_SSH_KEY=/home/YOUR_USER/.ssh/id_ed25519  # ssh identity (optional)
 ExecStart=/home/YOUR_USER/signal-bot/venv/bin/python bot.py
 # always (not on-failure): the bot's library swallows exceptions in its own run
 # loop and never exits non-zero even when wedged, so on-failure would never bounce it.