Record what may be shared, ahead of federation

The bot-side half of TODO #4. Nothing federates yet; this establishes
what would be eligible to, and proves the boundary before there is a
hub that could leak across it.

- Schema v5 adds visibility, defaulting to private. Upgrading an
  existing library therefore opts nothing into sharing, which a test
  on the legacy upgrade path pins down.
- list_syncable() is the only function that will ever feed a hub, and
  it requires both public and approved. Tests force rows into states
  set_visibility refuses, so the filter is what protects the boundary
  rather than the caller being careful.
- set_visibility refuses anything not approved: media still in the
  moderation queue cannot be marked shared.
- /share lists memes with their current state and asks before making
  one public, since un-sharing stops future copies but cannot recall
  those already taken. The wording describes what sharing will mean
  once a hub exists, because by then warning is too late.
- /list shows the same markers.

Vault posts have no path into the sync set at all; being in a separate
table is what guarantees it, and there is a test saying so.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.env.example

@@ -1,3 +1,17 @@
 BOT_TOKEN=your_telegram_bot_token_here
 ALLOWED_USERS=123456789,987654321
 ADMIN_USERS=123456789
+
+# Let non-admin users submit media for moderator approval (default: off)
+ACCEPT_SUBMISSIONS=false
+
+# Open the bot to anyone who meets the membership requirement below,
+# instead of using the ALLOWED_USERS list (default: off)
+PUBLIC_MODE=false
+
+# Chats a user must belong to. @username or numeric id, comma-separated.
+# The bot must be an admin in each one. Leave empty for a fully open bot.
+REQUIRED_CHATS=
+
+# all = must be in every chat listed above, any = one is enough
+REQUIRE_MODE=all

.gitignore

@@ -3,3 +3,4 @@ bot_db/
 __pycache__/
 *.pyc
 push.sh
+TODO

PLAN.md

@@ -0,0 +1,233 @@
+# Implementation plan — TODO #2–#6
+
+Progress on branch `develop` (not pushed):
+
+- [x] **#1** develop branch
+- [x] **Step 0** migration harness, per-owner uniqueness — `b55cee1`
+- [x] **#6** full media coverage — `efc2be5`
+- [x] **#3** moderation queue — `0a8c81d`
+- [x] **#2** public mode + membership gate — `82b5da4`
+- [x] **#5** save vs. extract (vault) — forward-based
+- [~] **#4** federation — bot-side visibility done; hub, channel and sync client outstanding
+
+Test suites: `test_db.py`, `test_bot.py`, `test_access.py`, `test_vault.py` — 34 cases, all passing.
+
+---
+
+## Step 0 — Prerequisites (blocking for #2, #3, #4, #5)
+
+Two things in the current schema block almost everything else, so they come first.
+
+**0a. There is no migration mechanism.** `db.init_db()` (`db.py:7-21`) only runs `CREATE TABLE IF NOT EXISTS`, so it can never alter an existing deployment's table. Every item below adds columns. Add a `schema_version` table and run ordered migration steps:
+
+```python
+async def migrate(db):
+    (v,) = await fetch_one("PRAGMA user_version")
+    if v < 2: ...  # ALTER TABLE media ADD COLUMN ...
+    await db.execute(f"PRAGMA user_version = {TARGET}")
+```
+
+SQLite's `PRAGMA user_version` is enough here — no extra table needed.
+
+**0b. `file_unique_id` is globally UNIQUE** (`db.py:14`). That is correct for a single-owner bot and wrong for every multi-user mode in this TODO: two users cannot each save the same meme. It also makes `INSERT OR REPLACE` (`db.py:28`) silently *delete and reinsert* a colliding row, changing its `id` (AUTOINCREMENT + REPLACE), which quietly breaks any `/delete <id>` the user had noted.
+
+Change to `UNIQUE(owner_id, file_unique_id)`. SQLite can't drop a column constraint in place — this needs the create-new-table / copy / rename dance inside migration v2.
+
+**0c. Related, worth fixing while in there:** `search_media()` (`db.py:34`) never filters by `owner_id`, so today every allowed user sees every user's media. Harmless for a personal bot, unacceptable once #2 opens the bot up. Search needs an explicit scope argument from the start.
+
+---
+
+## #6 — Full media coverage *(do this first: small, independent, immediate value)*
+
+Currently handled (`bot.py:157-190`): photo, animation, video, voice, sticker.
+Missing: **audio (mp3), document (file), video_note (round video)**.
+
+**Refactor.** The five handlers are near-identical. Replace with one handler and an extractor table:
+
+```python
+EXTRACTORS = {
+    "photo":     lambda m: m.photo[-1],
+    "gif":       lambda m: m.animation,
+    "video":     lambda m: m.video,
+    "voice":     lambda m: m.voice,
+    "sticker":   lambda m: m.sticker,
+    "audio":     lambda m: m.audio,
+    "document":  lambda m: m.document,
+    "video_note":lambda m: m.video_note,
+}
+```
+
+One `@dp.message(F.photo | F.animation | ... , StateFilter(None))` that walks the table. This also fixes the duplicated filter list at `bot.py:225`, which currently misses any newly added type.
+
+**Inline results** (`bot.py:395-416`) need matching branches: `InlineQueryResultCachedAudio` and `InlineQueryResultCachedDocument` (the latter requires a non-empty `title`).
+
+**Constraint to accept:** Telegram has **no cached inline result type for `video_note`**. Round videos can be stored and listed, but cannot be returned via inline search. Options: store them as `video` (loses the round presentation), or accept that they're `/list`-only. Recommend the latter and say so in the README.
+
+Also note `F.animation` catches most GIFs, but Telegram sometimes delivers a GIF as `document` with an image mime-type — the document branch will pick those up once it exists.
+
+---
+
+## #3 — Moderation queue *(before #2, since public mode depends on it)*
+
+**Schema (migration v3):** add to `media`:
+`status TEXT NOT NULL DEFAULT 'approved'` ∈ {`pending`, `approved`, `rejected`}, plus `submitted_by INTEGER`, `reviewed_by INTEGER`, `reviewed_at TIMESTAMP`.
+
+Existing rows default to `approved`, so current deployments are unaffected. Every read path (`search_media`, `list_media`, `count_media`) gains `WHERE status = 'approved'`.
+
+**Review flow — `/review` (moderators only):** pulls the *oldest pending* row, sends the media itself followed by its keywords and an inline keyboard:
+
+`[✅ Approve] [❌ Reject] [✏️ Fix keywords] [⏸ Stop]`
+
+**The "continue later if they are too much" requirement falls out for free**, and this is the key design choice: the queue position is not stored in FSM state — each button press just re-queries for the next `pending` row. A moderator can stop mid-queue, restart days later, or have several moderators working simultaneously, with no state to resume. Include a `(14 left)` counter in each message.
+
+Guard against two moderators grabbing the same item with a conditional update — `UPDATE media SET status=? WHERE id=? AND status='pending'` and check `rowcount`.
+
+---
+
+## #2 — Public mode with membership conditions
+
+**Config:** `PUBLIC_MODE=true`, `REQUIRED_CHATS=@channel,-100123456`, `REQUIRE_MODE=all|any`.
+
+**Gate:** `getChatMember(chat_id, user_id)`, accepting status ∈ {`creator`, `administrator`, `member`, and `restricted` with `is_member=true`}. The bot must be an admin in each required chat for this call to work — that's a setup requirement for the README.
+
+**Caching is not optional here.** Inline queries fire on *every keystroke*. An uncached membership check means several Telegram API calls per typed character, which will hit rate limits fast. Cache per `(user_id, chat_id)` with a ~5 minute TTL in memory.
+
+**Rejection UX:** for inline queries, don't return an empty list — use `answerInlineQuery`'s `button` parameter (`InlineQueryResultsButton`, supported in aiogram 3.15) to show a tappable "Join @channel to use this bot" bar above the results. In direct chat, reply with the channel links.
+
+With `PUBLIC_MODE=true`, non-admin submissions land as `status='pending'` from #3 rather than being ignored.
+
+---
+
+## #5 — Save vs. extract on forward — **implemented**
+
+Built on forwarding rather than the file_id copy originally planned here, which
+removed most of the work.
+
+**The bot already holds the post.** When you forward something, Telegram puts a
+copy in your private chat with the bot, so the bot can forward *that* back later
+by `message_id`. It never needs to reach the original channel, store media, or
+re-upload anything.
+
+**`forwardMessages` (plural) preserves albums.** It takes up to 100 message ids
+and returns them as one intact album, so the caption + N media case that made
+this item large is handled by Telegram itself. Ids must be strictly ascending.
+
+**One table, not two:**
+
+```sql
+vault_items(id, owner_id, chat_id, message_ids TEXT, caption, keywords, created_at)
+```
+
+Every vault query takes an `owner_id` and filters on it, so a guessed id cannot
+reach another user's post. Vault rows live in their own table and are never
+searched by inline mode, never moderated, and never federated.
+
+**Flow.** A forwarded message prompts *Save whole post* / *Extract media*
+(extract only appears when there is media). Save collects keywords and stores
+the pointer; extract runs the existing add path, now generalised to a list so a
+whole album shares one set of keywords. Albums are buffered by `media_group_id`
+with a 1s debounce, since Telegram gives no "album complete" signal.
+
+**Recall is inside the bot only**, as decided: `/vault [search]` lists matching
+posts with a retrieve button and a delete button. Matching covers keywords and
+the original caption. The save confirmation tells the user plainly that vault
+posts do not appear in inline search.
+
+**The tradeoff accepted:** a `message_id` is a pointer into the bot's chat with
+the owner, not self-contained like a `file_id`. If that history is cleared, the
+pointer breaks; retrieval catches the failure and tells the user to forward the
+post again.
+
+---
+
+## #4 — Federation
+
+### The constraint that decides the design
+
+**A `file_id` is only valid for the bot that received it.** A different bot cannot send, download, or resolve another bot's `file_id`. `file_unique_id` *is* stable across bots, but cannot be used to send anything — it's an identity key only.
+
+So "sync the public database every hour" **cannot work by copying rows between instances.** Each peer bot must independently obtain its *own* `file_id` for every shared meme. Any design that just ships database rows around will produce a table full of file_ids that throw `wrong file identifier` on use. This is the single most important thing to get right before writing code.
+
+### Recommended architecture: shared channel + metadata hub
+
+Split media transport from metadata sync:
+
+**Media → a private Telegram "federation channel" that you own.** Every participating bot is added as an admin. When an instance marks media public, it copies that media into the channel. Peers, being admins, receive the `channel_post` update and extract a `file_id` *valid for them*, storing it against the shared `file_unique_id`. Telegram stores the bytes once; nobody hosts files; no media ever transits your server.
+
+**Metadata → a small hub API** (`POST /sync` / `GET /since?ts=`) carrying only `file_unique_id`, `media_type`, `keywords`, `channel_message_id`, `origin_instance`. Rows keyed by `file_unique_id`, which is the one identifier that means the same thing everywhere.
+
+**Backfill for a newly joined instance:** a bot can't read channel history, so it won't have seen older posts. It resolves them on demand — `forwardMessage(from_chat_id=<fed_channel>, message_id=<from hub metadata>, chat_id=<own log chat>)` returns a full `Message` containing a `file_id` valid for that bot. Note this must be `forwardMessage`, not `copyMessage`: `copyMessage` returns only a `MessageId` with no file data. Do this lazily on first use, not as an hour-long startup storm.
+
+### Privacy and safety
+
+- `visibility TEXT NOT NULL DEFAULT 'private'` ∈ {`private`, `public`} on `media` (migration v5). The sync job reads **only** `visibility='public' AND status='approved'`. Private rows have no code path to the hub — assert this in a test.
+- Marking public is explicit and per-item, and should be **irreversible in practice**: once shared, peers have copies. Say so in the confirmation prompt.
+- Each instance gets an `INSTANCE_ID` + bearer token you issue; HTTPS only; an instance may only write rows it owns. You keep revocation and a takedown endpoint that tombstones a `file_unique_id` so peers delete on next sync.
+- Rate-limit per instance. A compromised peer token should be able to spam, at worst, its own rows.
+- Practical note: a federated pool of user-submitted media needs a takedown path for illegal content, and as the hub operator that lands on you. Worth deciding the policy before launching, not after.
+
+### Status
+
+Done in this repo: the `visibility` column (private by default), `/share` to
+change it, and `list_syncable()` — the single function that defines what is
+ever allowed to leave this instance. It is guarded by tests that force rows
+into states the writer would refuse, so the filter is proven rather than
+assumed. Nothing calls it yet.
+
+Outstanding, and all of it depends on decisions only the owner can make:
+the federation channel, the hub deployment, the instance tokens, and the sync
+client that pushes to and pulls from the hub.
+
+### Open question
+
+The hub is a separate deployable (small FastAPI + SQLite would do). Do you want it in this repo as `hub/`, or as its own repository? It has a different release cadence and a different threat model, which argues for separate.
+
+---
+
+## Suggested order
+
+| # | Item | Depends on | Size |
+|---|---|---|---|
+| 0 | Migration harness + unique-constraint fix | — | S |
+| 6 | Full media coverage | 0 (only for the refactor) | S |
+| 3 | Moderation queue | 0 | M |
+| 2 | Public mode + membership gate | 0, 3 | M |
+| 5 | Save vs. extract + vault | 0 | M (was L) |
+| 4 | Federation | 0, 3, hub built | XL |
+
+#6 is shippable on its own and touches nothing else — good first commit on `develop`. #4 should stay last: it's the only item that's hard to reverse once other people are running instances against your hub.
+
+## Decisions — settled 2026-09-01
+
+1. **#5 recall UX** — settled further during implementation: the vault is **forward-based and retrievable only inside the bot**. Inline search does not surface vault posts at all; the save confirmation says so.
+2. **#4 hub location** — **separate repository.** Channel name and hub hosting are the owner's call, not decided here.
+3. **#6 video_note** — **store as `video`**, accepting the loss of round presentation, so round videos stay inline-searchable. See the risk note below.
+4. **#4 architecture** — **shared federation channel accepted.** Media rides the channel, metadata rides the hub API.
+
+### Risk carried by decision 3
+
+A Telegram `file_id` encodes its own file type. Storing a `video_note` file_id and later returning it inside an `InlineQueryResultCachedVideo` may be rejected with `wrong file identifier`, because the cached-video slot expects a file_id minted as a video. This cannot be verified without a live bot token — **send the bot a round video and try to recall it inline before trusting this path.** If Telegram rejects it, the fallback is to treat video notes as `/list`-only (the option not chosen), which needs a one-line change to the extractor table.
+
+### Process
+
+Each step below lands as its own git commit on `develop`, so any step can be reverted independently.
+
+---
+
+## Verification
+
+The API claims above were checked against a real `aiogram==3.15.0` install (the version pinned in `requirements.txt`), not asserted from memory:
+
+| Claim | Result |
+|---|---|
+| `InlineQueryResultsButton` exists (used by #2) | ✅ present |
+| `answerInlineQuery` accepts `button` | ✅ present |
+| `InlineQueryResultCachedAudio` / `...CachedDocument` exist (#6) | ✅ both present |
+| `InlineQueryResultCachedVideoNote` exists | ❌ **does not exist** — confirms the video_note limitation |
+| `CachedDocument` requires a `title` | ✅ required fields are `id`, `title`, `document_file_id` |
+| `copyMessage` returns file data (#4 backfill) | ❌ returns bare `MessageId` |
+| `forwardMessage` returns file data (#4 backfill) | ✅ returns full `Message` |
+
+The two ❌ rows are why the plan routes around `video_note` inline results and specifies `forwardMessage` for federation backfill.
+
+Not verified here (needs a live bot token and a real chat): `getChatMember` behavior for #2, and the album `media_group_id` debounce timing for #5.

README.fa.md

@@ -1,6 +1,6 @@
 # Memes Bot
 
-ربات شخصی تلگرام برای ذخیره و جستجوی میم (عکس، گیف، ویدیو، صدا) با کلیدواژه.
+ربات شخصی تلگرام برای ذخیره و جستجوی میم (عکس، گیف، ویدیو، صدا، موزیک، استیکر، فایل) با کلیدواژه.
 
 [English](README.md)
 
@@ -50,7 +50,7 @@ docker compose up -d --build
 
 ### نحوه استفاده
 
-**افزودن میم** — یک عکس، گیف، ویدیو یا صدا برای ربات بفرستید. ربات از شما کلیدواژه می‌خواهد. کلیدواژه‌ها را یکی‌یکی یا با کاما جدا بفرستید، سپس `/done` بزنید.
+**افزودن میم** — یک عکس، گیف، ویدیو، صدا، موزیک، استیکر یا هر فایلی برای ربات بفرستید. ربات از شما کلیدواژه می‌خواهد. کلیدواژه‌ها را یکی‌یکی یا با کاما جدا بفرستید، سپس `/done` بزنید.
 
 **جستجوی اینلاین** — در هر چتی بنویسید `@نام_ربات کلیدواژه` تا نتایج نمایش داده شود.
 
@@ -59,9 +59,68 @@ docker compose up -d --build
 | دستور | توضیح |
 |---|---|
 | `/list` | نمایش همه رسانه‌های ذخیره‌شده |
+| `/vault [search]` | بازیابی پستی که کامل ذخیره کرده‌اید |
+| `/review` | بررسی یک‌به‌یک ارسال‌های در انتظار تأیید (فقط ادمین) |
+| `/share [search]` | مشخص‌کردن اینکه یک میم به اشتراک گذاشته شود یا خصوصی بماند (فقط ادمین) |
 | `/delete <id>` | حذف یک مورد (ID را از `/list` بگیرید) |
 | `/cancel` | لغو عملیات جاری |
 
+### گاوصندوق — ذخیره کامل یک پست
+
+اگر رسانه‌ای را مستقیم برای ربات بفرستید، مثل همیشه رفتار می‌کند: رسانه را جدا می‌کند و کلیدواژه می‌خواهد.
+
+اما اگر پستی را **فوروارد** کنید، ربات می‌پرسد چه می‌خواهید:
+
+- **ذخیره کل پست** — پست را دست‌نخورده با کپشن در گاوصندوق نگه می‌دارد. آلبوم هم آلبوم می‌ماند.
+- **جدا کردن رسانه** — همان رفتار قبلی؛ رسانه به کتابخانه قابل جستجو اضافه می‌شود.
+
+با `/vault [search]` پست‌های ذخیره‌شده را پیدا کنید؛ جستجو هم روی کلیدواژه‌ها و هم روی متن کپشن اصلی انجام می‌شود. کنار هر نتیجه یک دکمه برای فرستادن دوباره پست و یک دکمه برای حذف آن هست.
+
+> **پست‌های گاوصندوق فقط داخل خود ربات برمی‌گردند.** حالت inline تلگرام نمی‌تواند پیام موجود را فوروارد کند، پس `@yourbotusername keyword` به گاوصندوق دسترسی ندارد. پست را داخل ربات بگیرید و بعد هرجا خواستید فوروارد کنید.
+
+گاوصندوق برای هر کاربر جداگانه و خصوصی است: در جدولی جدا از کتابخانه میم‌ها ذخیره می‌شود، هر جستجو محدود به صاحب آن است، و هیچ‌چیزِ آن بررسی یا به اشتراک گذاشته نمی‌شود.
+
+ذخیره‌سازی به‌جای کپی رسانه، اشاره‌ای به پیام موجود در چت شما با ربات را نگه می‌دارد. اگر تاریخچه چتتان با ربات را پاک کنید، آن پست‌ها دیگر قابل بازیابی نیستند — ربات در این حالت به شما اطلاع می‌دهد تا دوباره فوروارد کنید.
+
+### عمومی‌کردن ربات
+
+با تنظیم `PUBLIC_MODE=true` ربات به‌جای فهرست `ALLOWED_USERS`، به هر کسی که شرط عضویت را داشته باشد اجازه دسترسی می‌دهد:
+
+```env
+PUBLIC_MODE=true
+ADMIN_USERS=123456789
+REQUIRED_CHATS=@mychannel,@mygroup
+REQUIRE_MODE=all
+```
+
+در `REQUIRED_CHATS` می‌توانید `@username` یا شناسه عددی بنویسید. **ربات باید در هر کدام ادمین باشد**، وگرنه نمی‌تواند عضویت را بررسی کند و همه رد می‌شوند. ربات هنگام راه‌اندازی این را بررسی می‌کند و نام هر چتی را که نمی‌تواند بخواند در لاگ می‌نویسد. مقدار `all` یعنی عضویت در همه چت‌ها لازم است و `any` یعنی یکی کافی است. اگر `REQUIRED_CHATS` خالی باشد ربات برای همه باز است.
+
+کاربری که شرط را ندارد پیامی می‌بیند که فهرست چت‌های لازم را نشان می‌دهد. در جستجوی inline هم به‌جای نتیجه خالی و بدون توضیح، یک نوار قابل لمس می‌بیند که ربات را باز می‌کند و همین توضیح را می‌دهد.
+
+> **قبل از روشن‌کردن این گزینه حتماً `ADMIN_USERS` را تنظیم کنید.** وقتی `ADMIN_USERS` خالی باشد، ربات خصوصی کاربرانش را ادمین حساب می‌کند؛ این برای ربات تک‌نفره راحت است اما در حالت عمومی یعنی همه ادمین می‌شوند. به همین دلیل در حالت عمومی تا وقتی `ADMIN_USERS` تنظیم نشده باشد هیچ‌کس ادمین نیست و ربات هنگام راه‌اندازی خطا ثبت می‌کند.
+
+پاسخ بررسی عضویت پنج دقیقه کش می‌شود، چون جستجوی inline در غیر این صورت با هر بار فشردن کلید از تلگرام سؤال می‌کرد. رد شدن فقط سی ثانیه کش می‌شود تا کاربری که تازه عضو شده تقریباً بلافاصله دسترسی پیدا کند.
+
+### پذیرش ارسال کاربران
+
+اگر در فایل `.env` مقدار `ACCEPT_SUBMISSIONS=true` را تنظیم کنید، کاربران غیرادمین هم می‌توانند رسانه بفرستند. این موارد ذخیره می‌شوند اما تا وقتی یک ناظر تأییدشان نکند در جستجوی inline و `/list` دیده نمی‌شوند. رسانه‌ای که خود ادمین‌ها می‌فرستند مثل قبل مستقیم ذخیره می‌شود.
+
+با دستور `/review` می‌توانید صف را بررسی کنید. هر مورد همراه با کلیدواژه‌هایش و چهار دکمه نمایش داده می‌شود: تأیید، رد، اصلاح کلیدواژه‌ها، و توقف.
+
+صف هیچ نشانگری ذخیره نمی‌کند و `/review` همیشه قدیمی‌ترین مورد در انتظار را می‌آورد. برای همین می‌توانید بعد از چند مورد متوقف شوید و فردا ادامه دهید، و چند ناظر هم می‌توانند هم‌زمان کار کنند — اگر دو نفر یک مورد را باز کنند، تصمیم اولی ثبت می‌شود و به دومی گفته می‌شود که قبلاً بررسی شده است.
+
+> **تا وقتی مطمئن نیستید این گزینه را خاموش بگذارید.** وقتی `ALLOWED_USERS` خالی باشد همه کاربرِ مجاز حساب می‌شوند، پس روشن‌کردن این گزینه یعنی پذیرفتن رسانه از هر کسی که ربات را پیدا کند.
+
+### اشتراک‌گذاری (مقدمات)
+
+هر میم به‌صورت پیش‌فرض **خصوصی** است. با `/share [search]` فهرست میم‌ها را با نشان 🔒 یا 🌐 می‌بینید و می‌توانید وضعیت هرکدام را عوض کنید. در `/list` هم همین نشان‌ها دیده می‌شود.
+
+> **هنوز هیچ‌چیزی به اشتراک گذاشته نمی‌شود.** این فقط خواستهٔ شما را ثبت می‌کند. فدراسیون — جایی که ربات‌هایی که جدا اجرا می‌شوند میم‌های عمومی‌شان را روی هم می‌گذارند — هنوز ساخته نشده و تا آن‌موقع هیچ میمی از این ربات بیرون نمی‌رود. متن تأیید، معنای اشتراک‌گذاری را در آینده توضیح می‌دهد، چون بعد از راه‌افتادن هاب دیگر برای هشدار دادن دیر است.
+
+دو قاعده در همان تک‌کوئری‌ای که قرار است به هاب داده شود اعمال می‌شود: فقط رسانه‌ای که هم **تأییدشده** و هم **عمومی** باشد واجد شرایط است. رسانه‌ای که هنوز در انتظار بررسی است اصلاً قابل اشتراک‌گذاری نیست، و پست‌های گاوصندوق در جدولی جدا هستند و هیچ مسیری به آنجا ندارند.
+
+خصوصی‌کردن دوباره، جلوی کپی‌های بعدی را می‌گیرد اما کپی‌هایی که قبلاً گرفته شده را برنمی‌گرداند؛ به همین دلیل قبل از عمومی‌کردن تأیید گرفته می‌شود، نه بعد از آن.
+
 ### انتقال به سرور دیگر
 
 فقط کافی است این دو فایل را منتقل کنید:

README.md

@@ -1,6 +1,6 @@
 # Memes Bot
 
-A personal Telegram inline bot to store and search media (photos, GIFs, videos, voice) by keywords.
+A personal Telegram inline bot to store and search media (photos, GIFs, videos, voice, music, stickers, files) by keywords.
 
 [فارسی](README.fa.md)
 
@@ -50,7 +50,9 @@ The database will be saved in `./bot_db/media.db`.
 
 ### Usage
 
-**Adding media** — send the bot a photo, GIF, video, or voice message, then type keywords when prompted (comma-separated or one per message), then `/done`.
+**Adding media** — send the bot a photo, GIF, video, voice message, music file, sticker, or any document, then type keywords when prompted (comma-separated or one per message), then `/done`.
+
+Round video messages are stored as regular videos: Telegram has no inline result type for video notes, so keeping them as video is what makes them searchable.
 
 **Searching inline** — in any chat type `@yourbotusername keyword` to search and send.
 
@@ -63,10 +65,69 @@ Telegram suggests these commands as you type `/`, with a short description besid
 | `/start` | Start the bot and show usage instructions |
 | `/menu` | Show the available commands |
 | `/list` | Show all saved media |
+| `/vault [search]` | Retrieve a whole post you saved |
+| `/review` | Review pending submissions one by one (admin only) |
 | `/edit [search]` | Select a meme and edit its keywords (admin only) |
+| `/share [search]` | Mark a meme as shared or private (admin only) |
 | `/delete <id>` | Remove an entry (get ID from `/list`; admin only) |
 | `/cancel` | Cancel current add operation |
 
+### The vault — saving whole posts
+
+Send the bot media directly and it behaves as it always has: it extracts the media and asks for keywords.
+
+**Forward** a post to it instead and it asks what you want:
+
+- **Save whole post** — keeps the post intact in your vault, caption and all. An album stays an album.
+- **Extract media** — the existing behaviour, adding the media to your searchable library.
+
+Retrieve saved posts with `/vault [search]`, which matches your keywords and the original caption. Each result has a button to forward it back to you, and one to remove it.
+
+> **Vault posts only come back inside this bot.** Telegram's inline mode cannot forward an existing message, so `@yourbotusername keyword` does not reach the vault. Retrieve the post in the bot chat, then forward it wherever you want.
+
+The vault is per-user and private: it is stored separately from the meme library, every lookup is scoped to its owner, and nothing in it is ever moderated or shared.
+
+Saving stores a pointer to the message in your chat with the bot rather than a copy of the media. If you clear your chat history with the bot, saved posts can no longer be retrieved — the bot tells you when that happens so you can forward them again.
+
+### Making the bot public
+
+Set `PUBLIC_MODE=true` to open the bot to anyone who meets a membership requirement, instead of the `ALLOWED_USERS` list:
+
+```env
+PUBLIC_MODE=true
+ADMIN_USERS=123456789
+REQUIRED_CHATS=@mychannel,@mygroup
+REQUIRE_MODE=all
+```
+
+`REQUIRED_CHATS` accepts `@username` or numeric ids. **The bot must be an admin in each one**, otherwise it cannot check membership and everyone is refused. The bot verifies this at startup and logs an error naming any chat it cannot read. `REQUIRE_MODE=all` demands every chat; `any` accepts one. With `REQUIRED_CHATS` empty the bot is open to everyone.
+
+Users who don't qualify get a message listing what to join. In inline search they see a tappable bar that opens the bot and explains the same thing, rather than empty results with no reason given.
+
+> **Set `ADMIN_USERS` before turning this on.** With no `ADMIN_USERS`, a private bot treats its users as admins — convenient for a one-person bot, but in public mode that would mean everyone. Public mode therefore grants admin to nobody until `ADMIN_USERS` is set, and logs an error at startup if you forget.
+
+Membership answers are cached for five minutes, since inline search would otherwise query Telegram on every keystroke. Refusals are cached for only thirty seconds, so a user who has just joined gets in almost immediately.
+
+### Accepting submissions
+
+Set `ACCEPT_SUBMISSIONS=true` in `.env` to let non-admin users send media too. Their submissions are stored but stay invisible — they don't appear in inline search or `/list` until a moderator approves them. Admins' own media is still saved directly.
+
+Run `/review` to work through the queue. Each item is shown with its keywords and four buttons: approve, reject, fix keywords, or stop.
+
+The queue keeps no cursor. `/review` always fetches the oldest pending item, so you can stop after three and come back tomorrow, and several moderators can review at the same time without stepping on each other — if two open the same item, the first decision wins and the second is told it was already handled.
+
+> **Leave this off unless you mean it.** With `ALLOWED_USERS` empty, everyone is an allowed user, so turning submissions on would accept media from anyone who finds the bot.
+
+### Sharing (groundwork)
+
+Every meme is **private** by default. `/share [search]` lists your memes with 🔒 or 🌐 and lets you flip one. `/list` shows the same markers.
+
+> **Nothing is shared yet.** This only records your intent. Federation — where independently run bots pool their shared memes — is not built, and no meme leaves this bot until it is. The wording in the confirmation describes what sharing *will* mean, because once a hub exists it is too late to start warning.
+
+Two rules are enforced in the one query that will ever feed a hub: only **approved** and **public** media is eligible. Media awaiting moderation cannot be marked shared at all, and vault posts are in a separate table with no path there whatsoever.
+
+Un-sharing stops future copies but cannot recall copies already taken, which is why the confirmation asks before making something public rather than after.
+
 ### Moving to another server
 
 The only files you need to transfer are:

VERSION

@@ -1 +1 @@
-1.0.0
+1.5.0

db.py

@@ -3,11 +3,24 @@ import os
 
 DB_PATH = os.getenv("DB_PATH", "/data/media.db")
 
+# Bump this and add a matching _migrate_to_vN() when the schema changes.
+SCHEMA_VERSION = 5
 
-async def init_db():
-    async with aiosqlite.connect(DB_PATH) as db:
+
+# ── schema / migrations ──────────────────────────────────────────────────────
+
+async def _table_exists(db, name: str) -> bool:
+    cursor = await db.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
+    )
+    return await cursor.fetchone() is not None
+
+
+async def _create_v1(db) -> None:
+    """The original schema. Fresh installs start here and migrate forward, so
+    the migration path is exercised on every new deployment, not just old ones."""
     await db.execute("""
-            CREATE TABLE IF NOT EXISTS media (
+        CREATE TABLE media (
             id INTEGER PRIMARY KEY AUTOINCREMENT,
             owner_id INTEGER NOT NULL,
             file_id TEXT NOT NULL,
@@ -18,83 +31,123 @@ async def init_db():
         )
     """)
     await db.execute("CREATE INDEX IF NOT EXISTS idx_owner ON media(owner_id)")
+
+
+async def _migrate_to_v2(db) -> None:
+    """Scope uniqueness to the owner so two users can each save the same file.
+
+    SQLite cannot drop a column constraint in place, so the table is rebuilt.
+    Row ids are carried over, which keeps any /delete <id> the user noted valid.
+    """
+    await db.execute("""
+        CREATE TABLE media_v2 (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            owner_id INTEGER NOT NULL,
+            file_id TEXT NOT NULL,
+            file_unique_id TEXT NOT NULL,
+            media_type TEXT NOT NULL,
+            keywords TEXT NOT NULL,
+            added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE(owner_id, file_unique_id)
+        )
+    """)
+    await db.execute("""
+        INSERT INTO media_v2 (id, owner_id, file_id, file_unique_id, media_type, keywords, added_at)
+        SELECT id, owner_id, file_id, file_unique_id, media_type, keywords, added_at FROM media
+    """)
+    await db.execute("DROP TABLE media")
+    await db.execute("ALTER TABLE media_v2 RENAME TO media")
+    await db.execute("CREATE INDEX IF NOT EXISTS idx_owner ON media(owner_id)")
+
+
+async def _migrate_to_v3(db) -> None:
+    """Add the moderation queue. Existing rows default to approved, so an
+    upgraded deployment behaves exactly as it did before."""
+    await db.execute("ALTER TABLE media ADD COLUMN status TEXT NOT NULL DEFAULT 'approved'")
+    await db.execute("ALTER TABLE media ADD COLUMN reviewed_by INTEGER")
+    await db.execute("ALTER TABLE media ADD COLUMN reviewed_at TIMESTAMP")
+    await db.execute("CREATE INDEX IF NOT EXISTS idx_status ON media(status)")
+
+
+async def _migrate_to_v4(db) -> None:
+    """The vault: posts kept whole, as pointers to the messages the bot already
+    holds in its private chat with the owner. Nothing here is ever shared, so it
+    lives in its own table rather than as a flag on media."""
+    await db.execute("""
+        CREATE TABLE IF NOT EXISTS vault_items (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            owner_id INTEGER NOT NULL,
+            chat_id INTEGER NOT NULL,
+            message_ids TEXT NOT NULL,
+            caption TEXT NOT NULL DEFAULT '',
+            keywords TEXT NOT NULL DEFAULT '',
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    """)
+    await db.execute("CREATE INDEX IF NOT EXISTS idx_vault_owner ON vault_items(owner_id)")
+
+
+async def _migrate_to_v5(db) -> None:
+    """Mark what may leave this instance. Everything defaults to private, so an
+    upgrade never opts an existing library into sharing."""
+    await db.execute("ALTER TABLE media ADD COLUMN visibility TEXT NOT NULL DEFAULT 'private'")
+    await db.execute("CREATE INDEX IF NOT EXISTS idx_visibility ON media(visibility)")
+
+
+MIGRATIONS = {
+    2: _migrate_to_v2,
+    3: _migrate_to_v3,
+    4: _migrate_to_v4,
+    5: _migrate_to_v5,
+}
+
+
+async def init_db():
+    async with aiosqlite.connect(DB_PATH) as db:
+        cursor = await db.execute("PRAGMA user_version")
+        version = (await cursor.fetchone())[0]
+
+        # A pre-versioning deployment also reports 0, so tell the two apart by
+        # whether the table is already there.
+        if version == 0:
+            if not await _table_exists(db, "media"):
+                await _create_v1(db)
+            version = 1
+            await db.execute(f"PRAGMA user_version = {version}")
+
+        for target in range(version + 1, SCHEMA_VERSION + 1):
+            migrate = MIGRATIONS.get(target)
+            if migrate is None:
+                continue
+            await migrate(db)
+            await db.execute(f"PRAGMA user_version = {target}")
+
         await db.commit()
 
 
-async def add_media(owner_id: int, file_id: str, file_unique_id: str, media_type: str, keywords: list[str]):
+# ── writes ───────────────────────────────────────────────────────────────────
+
+async def add_media(owner_id: int, file_id: str, file_unique_id: str, media_type: str,
+                    keywords: list[str], status: str = "approved"):
     kw = ",".join(k.strip().lower() for k in keywords if k.strip())
     async with aiosqlite.connect(DB_PATH) as db:
+        # Upsert rather than INSERT OR REPLACE: replacing would delete and
+        # reinsert the row, handing it a new id and breaking /delete <id>.
         await db.execute(
-            "INSERT OR REPLACE INTO media (owner_id, file_id, file_unique_id, media_type, keywords) VALUES (?,?,?,?,?)",
-            (owner_id, file_id, file_unique_id, media_type, kw),
+            """
+            INSERT INTO media (owner_id, file_id, file_unique_id, media_type, keywords, status)
+            VALUES (?,?,?,?,?,?)
+            ON CONFLICT(owner_id, file_unique_id) DO UPDATE SET
+                file_id = excluded.file_id,
+                media_type = excluded.media_type,
+                keywords = excluded.keywords,
+                status = excluded.status
+            """,
+            (owner_id, file_id, file_unique_id, media_type, kw, status),
         )
         await db.commit()
 
 
-async def search_media(query: str) -> list[dict]:
-    query = query.strip().lower()
-    async with aiosqlite.connect(DB_PATH) as db:
-        db.row_factory = aiosqlite.Row
-        if query:
-            cursor = await db.execute(
-                """
-                SELECT id, file_id, file_unique_id, media_type, keywords
-                FROM media
-                WHERE (',' || keywords || ',') LIKE ?
-                ORDER BY added_at DESC
-                LIMIT 50
-                """,
-                (f"%,{query}%",),
-            )
-        else:
-            cursor = await db.execute(
-                """
-                SELECT id, file_id, file_unique_id, media_type, keywords
-                FROM media
-                ORDER BY added_at DESC
-                LIMIT 50
-                """,
-            )
-        rows = await cursor.fetchall()
-        return [dict(r) for r in rows]
-
-
-async def list_media(page: int, page_size: int) -> list[dict]:
-    """Return one page of saved media, newest first."""
-    offset = page * page_size
-    async with aiosqlite.connect(DB_PATH) as db:
-        db.row_factory = aiosqlite.Row
-        cursor = await db.execute(
-            """
-            SELECT id, file_id, file_unique_id, media_type, keywords
-            FROM media
-            ORDER BY added_at DESC, id DESC
-            LIMIT ? OFFSET ?
-            """,
-            (page_size, offset),
-        )
-        rows = await cursor.fetchall()
-        return [dict(r) for r in rows]
-
-
-async def count_media() -> int:
-    async with aiosqlite.connect(DB_PATH) as db:
-        cursor = await db.execute("SELECT COUNT(*) FROM media")
-        row = await cursor.fetchone()
-        return row[0]
-
-
-async def get_media_by_id(media_id: int) -> dict | None:
-    async with aiosqlite.connect(DB_PATH) as db:
-        db.row_factory = aiosqlite.Row
-        cursor = await db.execute(
-            "SELECT id, file_id, file_unique_id, media_type, keywords FROM media WHERE id = ?",
-            (media_id,),
-        )
-        row = await cursor.fetchone()
-        return dict(row) if row else None
-
-
 async def update_media_keywords(media_id: int, keywords: list[str]) -> bool:
     kw = ",".join(k.strip().lower() for k in keywords if k.strip())
     async with aiosqlite.connect(DB_PATH) as db:
@@ -111,3 +164,241 @@ async def delete_media(media_id: int) -> bool:
         cur = await db.execute("DELETE FROM media WHERE id = ?", (media_id,))
         await db.commit()
         return cur.rowcount > 0
+
+
+# ── reads ────────────────────────────────────────────────────────────────────
+
+COLUMNS = "id, owner_id, file_id, file_unique_id, media_type, keywords, status, visibility"
+
+
+async def search_media(query: str, owner_id: int | None = None,
+                       status: str | None = "approved") -> list[dict]:
+    """Search by keyword prefix. owner_id=None searches every owner, which is
+    the current shared-library behaviour; public mode will pass a scope.
+    status=None includes items still awaiting moderation."""
+    query = query.strip().lower()
+    where = []
+    params: list = []
+    if status is not None:
+        where.append("status = ?")
+        params.append(status)
+    if query:
+        where.append("(',' || keywords || ',') LIKE ?")
+        params.append(f"%,{query}%")
+    if owner_id is not None:
+        where.append("owner_id = ?")
+        params.append(owner_id)
+    clause = f"WHERE {' AND '.join(where)}" if where else ""
+
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"SELECT {COLUMNS} FROM media {clause} ORDER BY added_at DESC LIMIT 50",
+            params,
+        )
+        rows = await cursor.fetchall()
+        return [dict(r) for r in rows]
+
+
+async def list_media(page: int, page_size: int, status: str | None = "approved") -> list[dict]:
+    """Return one page of saved media, newest first."""
+    offset = page * page_size
+    clause = "WHERE status = ?" if status is not None else ""
+    params = ([status] if status is not None else []) + [page_size, offset]
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"""
+            SELECT {COLUMNS}
+            FROM media
+            {clause}
+            ORDER BY added_at DESC, id DESC
+            LIMIT ? OFFSET ?
+            """,
+            params,
+        )
+        rows = await cursor.fetchall()
+        return [dict(r) for r in rows]
+
+
+async def count_media(status: str | None = "approved") -> int:
+    clause = "WHERE status = ?" if status is not None else ""
+    params = [status] if status is not None else []
+    async with aiosqlite.connect(DB_PATH) as db:
+        cursor = await db.execute(f"SELECT COUNT(*) FROM media {clause}", params)
+        row = await cursor.fetchone()
+        return row[0]
+
+
+async def get_media_by_id(media_id: int) -> dict | None:
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"SELECT {COLUMNS} FROM media WHERE id = ?",
+            (media_id,),
+        )
+        row = await cursor.fetchone()
+        return dict(row) if row else None
+
+
+# ── moderation queue ─────────────────────────────────────────────────────────
+
+async def next_pending() -> dict | None:
+    """Oldest item awaiting review. The queue keeps no cursor: each call just
+    asks for the next one, so a moderator can stop and resume at any time and
+    several moderators can work at once."""
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"SELECT {COLUMNS} FROM media WHERE status = 'pending' ORDER BY added_at ASC, id ASC LIMIT 1"
+        )
+        row = await cursor.fetchone()
+        return dict(row) if row else None
+
+
+async def count_pending() -> int:
+    return await count_media(status="pending")
+
+
+async def set_status(media_id: int, status: str, reviewer_id: int) -> bool:
+    """Approve or reject a pending item. Returns False if it was not pending
+    any more, which is how two moderators reviewing at once are kept apart."""
+    async with aiosqlite.connect(DB_PATH) as db:
+        cur = await db.execute(
+            """
+            UPDATE media SET status = ?, reviewed_by = ?, reviewed_at = CURRENT_TIMESTAMP
+            WHERE id = ? AND status = 'pending'
+            """,
+            (status, reviewer_id, media_id),
+        )
+        await db.commit()
+        return cur.rowcount > 0
+
+
+# ── visibility / federation ──────────────────────────────────────────────────
+
+async def set_visibility(media_id: int, visibility: str) -> bool:
+    """Only approved media can be shared; something still awaiting moderation
+    or already rejected must never become publishable."""
+    async with aiosqlite.connect(DB_PATH) as db:
+        cur = await db.execute(
+            "UPDATE media SET visibility = ? WHERE id = ? AND status = 'approved'",
+            (visibility, media_id),
+        )
+        await db.commit()
+        return cur.rowcount > 0
+
+
+async def list_syncable(limit: int = 500) -> list[dict]:
+    """The only rows permitted to leave this instance.
+
+    Both conditions matter: private media is nobody else's business, and
+    unreviewed or rejected media must not be pushed to peers. Everything that
+    ever talks to the hub goes through this function.
+    """
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"""
+            SELECT {COLUMNS} FROM media
+            WHERE visibility = 'public' AND status = 'approved'
+            ORDER BY added_at ASC, id ASC
+            LIMIT ?
+            """,
+            (limit,),
+        )
+        return [dict(r) for r in await cursor.fetchall()]
+
+
+async def count_public() -> int:
+    async with aiosqlite.connect(DB_PATH) as db:
+        cursor = await db.execute(
+            "SELECT COUNT(*) FROM media WHERE visibility = 'public' AND status = 'approved'"
+        )
+        return (await cursor.fetchone())[0]
+
+
+# ── vault ────────────────────────────────────────────────────────────────────
+#
+# Every query here takes an owner_id and filters on it. A vault item belongs to
+# one person and is never searchable, never moderated and never federated; the
+# separate table is what makes that structural rather than a flag to remember.
+
+VAULT_COLUMNS = "id, owner_id, chat_id, message_ids, caption, keywords, created_at"
+
+
+def pack_message_ids(message_ids: list[int]) -> str:
+    return ",".join(str(m) for m in message_ids)
+
+
+def unpack_message_ids(packed: str) -> list[int]:
+    return [int(m) for m in packed.split(",") if m.strip()]
+
+
+async def add_vault_item(owner_id: int, chat_id: int, message_ids: list[int],
+                         caption: str, keywords: list[str]) -> int:
+    kw = ",".join(k.strip().lower() for k in keywords if k.strip())
+    async with aiosqlite.connect(DB_PATH) as db:
+        cur = await db.execute(
+            """
+            INSERT INTO vault_items (owner_id, chat_id, message_ids, caption, keywords)
+            VALUES (?,?,?,?,?)
+            """,
+            (owner_id, chat_id, pack_message_ids(message_ids), caption or "", kw),
+        )
+        await db.commit()
+        return cur.lastrowid
+
+
+async def search_vault(owner_id: int, query: str, limit: int = 20) -> list[dict]:
+    """Match keywords by prefix or the original caption anywhere in its text —
+    the point of a vault is finding a post you never got round to tagging."""
+    query = query.strip().lower()
+    params: list = [owner_id]
+    clause = ""
+    if query:
+        clause = "AND ((',' || keywords || ',') LIKE ? OR LOWER(caption) LIKE ?)"
+        params += [f"%,{query}%", f"%{query}%"]
+    params.append(limit)
+
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"""
+            SELECT {VAULT_COLUMNS} FROM vault_items
+            WHERE owner_id = ? {clause}
+            ORDER BY created_at DESC, id DESC
+            LIMIT ?
+            """,
+            params,
+        )
+        return [dict(r) for r in await cursor.fetchall()]
+
+
+async def get_vault_item(owner_id: int, item_id: int) -> dict | None:
+    """Scoped by owner so one user cannot fetch another user's saved post."""
+    async with aiosqlite.connect(DB_PATH) as db:
+        db.row_factory = aiosqlite.Row
+        cursor = await db.execute(
+            f"SELECT {VAULT_COLUMNS} FROM vault_items WHERE id = ? AND owner_id = ?",
+            (item_id, owner_id),
+        )
+        row = await cursor.fetchone()
+        return dict(row) if row else None
+
+
+async def delete_vault_item(owner_id: int, item_id: int) -> bool:
+    async with aiosqlite.connect(DB_PATH) as db:
+        cur = await db.execute(
+            "DELETE FROM vault_items WHERE id = ? AND owner_id = ?", (item_id, owner_id)
+        )
+        await db.commit()
+        return cur.rowcount > 0
+
+
+async def count_vault(owner_id: int) -> int:
+    async with aiosqlite.connect(DB_PATH) as db:
+        cursor = await db.execute(
+            "SELECT COUNT(*) FROM vault_items WHERE owner_id = ?", (owner_id,)
+        )
+        return (await cursor.fetchone())[0]

test_access.py

@@ -0,0 +1,165 @@
+"""Access-control tests. Run with: python test_access.py
+
+Covers who gets into the bot in public mode, and the membership cache that
+keeps inline search from hammering the Telegram API on every keystroke.
+"""
+import asyncio
+import importlib
+import logging
+import os
+import sys
+import time
+
+os.environ.setdefault("BOT_TOKEN", "123456789:AAHtestTokenPlaceholder_ABCDEFGHIJKLMNOP")
+os.environ.setdefault("DB_PATH", ":memory:")
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from aiogram.exceptions import TelegramAPIError
+
+# Several cases deliberately make the API fail; the bot logs those with a
+# traceback, which is correct in production and only noise here.
+logging.disable(logging.CRITICAL)
+
+CONFIG_KEYS = ("PUBLIC_MODE", "REQUIRED_CHATS", "REQUIRE_MODE",
+               "ALLOWED_USERS", "ADMIN_USERS", "ACCEPT_SUBMISSIONS")
+
+
+class FakeMember:
+    def __init__(self, status: str, is_member: bool | None = None):
+        self.status = status
+        if is_member is not None:
+            self.is_member = is_member
+
+
+def load_bot(**env):
+    """Reload the bot module with a fresh environment; config is read at import."""
+    for key in CONFIG_KEYS:
+        os.environ.pop(key, None)
+    os.environ.update(env)
+    import bot
+    return importlib.reload(bot)
+
+
+def stub_membership(B, members: dict, calls: list | None = None):
+    async def get_chat_member(chat, user_id):
+        if calls is not None:
+            calls.append((chat, user_id))
+        if (chat, user_id) not in members:
+            raise TelegramAPIError(method=None, message="user not found")
+        return members[(chat, user_id)]
+    B.bot.get_chat_member = get_chat_member
+
+
+async def test_member_statuses():
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan")
+    allowed = ["creator", "administrator", "member"]
+    denied = ["left", "kicked"]
+
+    for status in allowed:
+        B._membership_cache.clear()
+        stub_membership(B, {("@chan", 1): FakeMember(status)})
+        assert await B.has_access(1) is True, f"{status} should be allowed"
+    for status in denied:
+        B._membership_cache.clear()
+        stub_membership(B, {("@chan", 1): FakeMember(status)})
+        assert await B.has_access(1) is False, f"{status} should be denied"
+
+    # A restricted user is still a member unless they have actually left.
+    B._membership_cache.clear()
+    stub_membership(B, {("@chan", 1): FakeMember("restricted", is_member=True)})
+    assert await B.has_access(1) is True, "restricted but still in the chat"
+    B._membership_cache.clear()
+    stub_membership(B, {("@chan", 1): FakeMember("restricted", is_member=False)})
+    assert await B.has_access(1) is False, "restricted and gone"
+    print("  member statuses are classified correctly, including restricted")
+
+
+async def test_require_all_and_any():
+    members = {("@a", 1): FakeMember("member")}  # in @a, not in @b
+
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@a,@b", REQUIRE_MODE="all")
+    stub_membership(B, members)
+    assert await B.has_access(1) is False, "all mode needs every chat"
+
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@a,@b", REQUIRE_MODE="any")
+    stub_membership(B, members)
+    assert await B.has_access(1) is True, "any mode needs just one"
+    print("  REQUIRE_MODE all and any behave differently as configured")
+
+
+async def test_cache_absorbs_repeat_checks():
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan")
+    calls: list = []
+    stub_membership(B, {("@chan", 1): FakeMember("member")}, calls)
+
+    for _ in range(10):
+        assert await B.has_access(1) is True
+    assert len(calls) == 1, f"10 inline keystrokes caused {len(calls)} API calls"
+    print("  repeated checks hit the API once, not once per keystroke")
+
+
+async def test_denials_expire_sooner_than_grants():
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan")
+    assert B.MEMBERSHIP_TTL_DENIED < B.MEMBERSHIP_TTL_ALLOWED
+
+    stub_membership(B, {("@chan", 1): FakeMember("member")})
+    await B.has_access(1)
+    await B.has_access(2)  # unknown user -> denied
+
+    now = time.monotonic()
+    granted = B._membership_cache[(1, "@chan")][1] - now
+    denied = B._membership_cache[(2, "@chan")][1] - now
+    assert denied < granted, "a user who just joined should not wait the full window"
+    print("  denials expire quickly so joining takes effect promptly")
+
+
+async def test_admin_bypasses_membership():
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan", ADMIN_USERS="111")
+    calls: list = []
+    stub_membership(B, {}, calls)  # nobody is a member
+
+    assert await B.has_access(111) is True, "admins must not be locked out"
+    assert calls == [], "admin check should not call the API at all"
+    assert await B.has_access(222) is False
+    print("  admins bypass the membership check without an API call")
+
+
+async def test_private_mode_still_uses_the_allowlist():
+    B = load_bot(ALLOWED_USERS="111,222")
+    calls: list = []
+    stub_membership(B, {}, calls)
+
+    assert await B.has_access(111) is True
+    assert await B.has_access(333) is False
+    assert calls == [], "a private bot should never call getChatMember"
+    print("  with PUBLIC_MODE off, the allowlist behaves exactly as before")
+
+
+async def test_api_failure_fails_closed():
+    B = load_bot(PUBLIC_MODE="true", REQUIRED_CHATS="@chan")
+
+    async def always_fails(chat, user_id):
+        raise TelegramAPIError(method=None, message="bot is not an admin there")
+    B.bot.get_chat_member = always_fails
+
+    assert await B.has_access(1) is False, "an unreadable chat must not open the gate"
+    print("  a failing membership check denies access rather than opening up")
+
+
+async def test_open_public_bot():
+    B = load_bot(PUBLIC_MODE="true")  # no REQUIRED_CHATS
+    assert await B.has_access(999) is True
+    print("  public mode with no required chats is open to everyone")
+
+
+async def main():
+    for test in (test_member_statuses, test_require_all_and_any,
+                 test_cache_absorbs_repeat_checks, test_denials_expire_sooner_than_grants,
+                 test_admin_bypasses_membership, test_private_mode_still_uses_the_allowlist,
+                 test_api_failure_fails_closed, test_open_public_bot):
+        await test()
+    print("\nAll access tests passed.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test_bot.py

@@ -0,0 +1,103 @@
+"""Media handling tests. Run with: python test_bot.py
+
+Covers the dispatch table that decides how an incoming message is stored and
+how it comes back through inline search.
+"""
+import datetime
+import os
+import sys
+
+os.environ.setdefault("BOT_TOKEN", "123456789:AAHtestTokenPlaceholder_ABCDEFGHIJKLMNOP")
+os.environ.setdefault("DB_PATH", ":memory:")
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from aiogram.types import (Animation, Audio, Chat, Document, Message, PhotoSize,
+                           Sticker, Video, VideoNote, Voice)
+
+import bot as B
+
+CHAT = Chat(id=1, type="private")
+NOW = datetime.datetime.now()
+
+
+def msg(**kw) -> Message:
+    return Message(message_id=1, date=NOW, chat=CHAT, **kw)
+
+
+def sample_messages():
+    photo = PhotoSize(file_id="F_photo", file_unique_id="U_photo", width=1, height=1)
+    anim = Animation(file_id="F_gif", file_unique_id="U_gif", width=1, height=1, duration=1)
+    sticker = Sticker(file_id="F_st", file_unique_id="U_st", width=1, height=1,
+                      type="regular", is_animated=False, is_video=False)
+    return [
+        ("photo", msg(photo=[photo])),
+        # Telegram sets document alongside animation for backward compatibility.
+        ("gif", msg(animation=anim, document=Document(file_id="F_gif", file_unique_id="U_gif"))),
+        ("sticker", msg(sticker=sticker)),
+        ("video", msg(video=Video(file_id="F_v", file_unique_id="U_v", width=1, height=1, duration=1))),
+        ("video", msg(video_note=VideoNote(file_id="F_vn", file_unique_id="U_vn", length=1, duration=1))),
+        ("voice", msg(voice=Voice(file_id="F_vo", file_unique_id="U_vo", duration=1))),
+        ("audio", msg(audio=Audio(file_id="F_a", file_unique_id="U_a", duration=1))),
+        ("document", msg(document=Document(file_id="F_doc", file_unique_id="U_doc"))),
+    ]
+
+
+def test_every_type_extracts():
+    for expected, message in sample_messages():
+        found = B.extract_media(message)
+        assert found is not None, f"{expected}: nothing extracted"
+        assert found[0] == expected, f"expected {expected}, got {found[0]}"
+    print("  every supported media type extracts to the right media_type")
+
+
+def test_animation_beats_document():
+    gif = sample_messages()[1][1]
+    assert gif.document is not None, "test fixture should mimic Telegram's dual field"
+    assert B.extract_media(gif)[0] == "gif", "a GIF must not be filed as a document"
+    print("  animation wins over the backward-compat document field")
+
+
+def test_video_note_stored_as_video():
+    note = sample_messages()[4][1]
+    assert B.extract_media(note) == ("video", "F_vn", "U_vn")
+    print("  video notes are stored as video so inline search can return them")
+
+
+def test_non_media_ignored():
+    assert B.extract_media(msg(text="hello")) is None
+    print("  plain text messages are ignored")
+
+
+def test_every_stored_type_renders_inline():
+    stored = {t for t, _ in B.MEDIA_EXTRACTORS}
+    missing = stored - set(B.INLINE_BUILDERS)
+    assert not missing, f"stored but not renderable inline: {sorted(missing)}"
+    print("  every stored media_type has a matching inline builder")
+
+
+def test_every_stored_type_can_be_sent_back():
+    stored = {t for t, _ in B.MEDIA_EXTRACTORS}
+    missing = stored - set(B.MEDIA_SENDERS)
+    assert not missing, f"stored but not sendable for review: {sorted(missing)}"
+    print("  every stored media_type can be sent back for review")
+
+
+def test_builders_construct():
+    for media_type, build in B.INLINE_BUILDERS.items():
+        build("rid", f"FID_{media_type}", "cat, funny")
+        # Documents and videos require a non-empty title; keywords can be blank.
+        build("rid", f"FID_{media_type}", "")
+    print("  inline builders construct, including with empty keywords")
+
+
+def main():
+    for test in (test_every_type_extracts, test_animation_beats_document,
+                 test_video_note_stored_as_video, test_non_media_ignored,
+                 test_every_stored_type_renders_inline,
+                 test_every_stored_type_can_be_sent_back, test_builders_construct):
+        test()
+    print("\nAll media tests passed.")
+
+
+if __name__ == "__main__":
+    main()

test_db.py

@@ -0,0 +1,263 @@
+"""Migration tests. Run with: python test_db.py
+
+The v1->v2 migration rebuilds the media table, so it drops the original.
+These cases cover the paths that rebuild touches before it ever runs against
+a real bot_db/media.db.
+"""
+import asyncio
+import importlib
+import os
+import sqlite3
+import sys
+import tempfile
+
+V1_SCHEMA = """
+    CREATE TABLE media (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        owner_id INTEGER NOT NULL,
+        file_id TEXT NOT NULL,
+        file_unique_id TEXT NOT NULL UNIQUE,
+        media_type TEXT NOT NULL,
+        keywords TEXT NOT NULL,
+        added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+    );
+"""
+
+
+def user_version(path: str) -> int:
+    with sqlite3.connect(path) as c:
+        return c.execute("PRAGMA user_version").fetchone()[0]
+
+
+def load_db(path: str):
+    os.environ["DB_PATH"] = path
+    import db
+    return importlib.reload(db)
+
+
+async def test_legacy_upgrade(tmp: str):
+    path = os.path.join(tmp, "legacy.db")
+    with sqlite3.connect(path) as c:
+        c.executescript(V1_SCHEMA)
+        c.execute("INSERT INTO media (id,owner_id,file_id,file_unique_id,media_type,keywords)"
+                  " VALUES (7,111,'FID_A','UNI_A','photo','cat,funny')")
+        c.execute("INSERT INTO media (id,owner_id,file_id,file_unique_id,media_type,keywords)"
+                  " VALUES (9,111,'FID_B','UNI_B','gif','dog')")
+
+    db = load_db(path)
+    await db.init_db()
+
+    # Pinned to the module so adding a migration does not require editing this.
+    assert user_version(path) == db.SCHEMA_VERSION
+    rows = sorted((r["id"], r["keywords"]) for r in await db.search_media(""))
+    assert rows == [(7, "cat,funny"), (9, "dog")], f"rows not preserved: {rows}"
+    with sqlite3.connect(path) as c:
+        statuses = {r[0] for r in c.execute("SELECT status FROM media")}
+        tables = {r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table'")}
+    assert statuses == {"approved"}, f"upgraded rows must stay live: {statuses}"
+    assert "vault_items" in tables, "upgrading should create the vault table too"
+    with sqlite3.connect(path) as c:
+        visibilities = {r[0] for r in c.execute("SELECT visibility FROM media")}
+    assert visibilities == {"private"}, \
+        f"upgrading must not opt an existing library into sharing: {visibilities}"
+    print("  legacy v1 database upgrades with ids, keywords and visibility intact")
+
+
+async def test_per_owner_uniqueness(tmp: str):
+    path = os.path.join(tmp, "owners.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "FID_A", "UNI_A", "photo", ["cat"])
+    await db.add_media(222, "FID_A2", "UNI_A", "photo", ["cat"])
+    assert len(await db.search_media("cat")) == 2, "two owners must each keep their own row"
+    assert len(await db.search_media("", owner_id=222)) == 1, "owner scoping failed"
+    print("  two owners can save the same file; owner scoping filters correctly")
+
+
+async def test_readd_keeps_id(tmp: str):
+    path = os.path.join(tmp, "readd.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "FID_A", "UNI_A", "photo", ["cat"])
+    original = (await db.search_media("cat"))[0]["id"]
+    await db.add_media(111, "FID_NEW", "UNI_A", "photo", ["kitten"])
+    row = await db.get_media_by_id(original)
+
+    assert row is not None, "re-adding the same file orphaned its id"
+    assert row["keywords"] == "kitten" and row["file_id"] == "FID_NEW"
+    print("  re-adding a file updates in place and keeps its id")
+
+
+async def test_fresh_and_idempotent(tmp: str):
+    path = os.path.join(tmp, "fresh.db")
+    db = load_db(path)
+    await db.init_db()
+
+    with sqlite3.connect(path) as c:
+        sql = c.execute("SELECT sql FROM sqlite_master WHERE name='media'").fetchone()[0]
+    assert "UNIQUE(owner_id, file_unique_id)" in sql
+    assert user_version(path) == db.SCHEMA_VERSION
+
+    await db.init_db()
+    await db.init_db()
+    print("  fresh install lands on the current version; init_db is idempotent")
+
+
+async def test_pending_is_hidden(tmp: str):
+    path = os.path.join(tmp, "pending.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["live"])
+    await db.add_media(222, "F2", "U2", "photo", ["waiting"], status="pending")
+
+    assert len(await db.search_media("")) == 1, "pending items must not be searchable"
+    assert await db.count_media() == 1
+    assert len(await db.list_media(0, 30)) == 1, "pending items must not appear in /list"
+    assert len(await db.search_media("", status=None)) == 2, "status=None should see everything"
+    assert await db.count_pending() == 1
+    print("  pending items stay out of search, /list and counts until approved")
+
+
+async def test_queue_is_oldest_first(tmp: str):
+    path = os.path.join(tmp, "queue.db")
+    db = load_db(path)
+    await db.init_db()
+
+    for i in range(3):
+        await db.add_media(100 + i, f"F{i}", f"U{i}", "photo", [f"kw{i}"], status="pending")
+
+    first = await db.next_pending()
+    assert first["keywords"] == "kw0", f"queue must be oldest-first, got {first['keywords']}"
+
+    # Approving removes it from the queue and the next one steps up.
+    assert await db.set_status(first["id"], "approved", 999) is True
+    second = await db.next_pending()
+    assert second["keywords"] == "kw1"
+    assert await db.count_pending() == 2
+    print("  queue serves oldest first and advances as items are handled")
+
+
+async def test_two_moderators_cannot_double_review(tmp: str):
+    path = os.path.join(tmp, "race.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["kw"], status="pending")
+    item = await db.next_pending()
+
+    assert await db.set_status(item["id"], "approved", 111) is True
+    assert await db.set_status(item["id"], "rejected", 222) is False, \
+        "a second moderator must not overwrite a completed review"
+
+    row = await db.get_media_by_id(item["id"])
+    assert row["status"] == "approved", "first review must win"
+    print("  a second moderator cannot overturn an already-reviewed item")
+
+
+async def test_rejected_stays_hidden(tmp: str):
+    path = os.path.join(tmp, "rejected.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["nope"], status="pending")
+    item = await db.next_pending()
+    await db.set_status(item["id"], "rejected", 999)
+
+    assert await db.search_media("nope") == [], "rejected items must never surface"
+    assert await db.count_pending() == 0
+    assert await db.next_pending() is None
+    print("  rejected items leave the queue and never surface in search")
+
+
+async def test_new_media_is_private(tmp: str):
+    path = os.path.join(tmp, "visibility.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["kw"])
+    assert (await db.search_media("kw"))[0]["visibility"] == "private"
+    assert await db.list_syncable() == [], "nothing should be shared by default"
+    print("  media is private unless explicitly shared")
+
+
+async def test_only_public_and_approved_can_sync(tmp: str):
+    path = os.path.join(tmp, "syncable.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["private_approved"])
+    await db.add_media(111, "F2", "U2", "photo", ["public_approved"])
+    await db.add_media(111, "F3", "U3", "photo", ["public_pending"], status="pending")
+    await db.add_media(111, "F4", "U4", "photo", ["public_rejected"], status="rejected")
+
+    # Force the states set_visibility would refuse, to prove the query itself
+    # filters rather than relying on the writer being careful.
+    with sqlite3.connect(path) as c:
+        c.execute("UPDATE media SET visibility='public' WHERE file_unique_id IN ('U2','U3','U4')")
+
+    syncable = [r["keywords"] for r in await db.list_syncable()]
+    assert syncable == ["public_approved"], f"leaked: {syncable}"
+    assert await db.count_public() == 1
+    print("  only approved AND public media is ever syncable")
+
+
+async def test_sharing_requires_approval(tmp: str):
+    path = os.path.join(tmp, "share_guard.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["waiting"], status="pending")
+    pending = (await db.search_media("", status="pending"))[0]
+    assert await db.set_visibility(pending["id"], "public") is False, \
+        "unreviewed media must not be shareable"
+    assert await db.list_syncable() == []
+    print("  media awaiting moderation cannot be marked shared")
+
+
+async def test_unsharing_stops_future_syncs(tmp: str):
+    path = os.path.join(tmp, "unshare.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_media(111, "F1", "U1", "photo", ["kw"])
+    item = (await db.search_media("kw"))[0]
+    assert await db.set_visibility(item["id"], "public") is True
+    assert len(await db.list_syncable()) == 1
+
+    assert await db.set_visibility(item["id"], "private") is True
+    assert await db.list_syncable() == [], "unsharing should remove it from the sync set"
+    print("  unsharing takes an item back out of the sync set")
+
+
+async def test_vault_never_syncs(tmp: str):
+    path = os.path.join(tmp, "vault_sync.db")
+    db = load_db(path)
+    await db.init_db()
+
+    await db.add_vault_item(111, 500, [1, 2], "a private post", ["personal"])
+    # The vault lives in its own table, so the sync query cannot reach it at all.
+    assert await db.list_syncable() == []
+    assert await db.count_public() == 0
+    print("  vault posts have no path into the sync set")
+
+
+async def main():
+    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+    with tempfile.TemporaryDirectory() as tmp:
+        for test in (test_legacy_upgrade, test_per_owner_uniqueness,
+                     test_readd_keeps_id, test_fresh_and_idempotent,
+                     test_pending_is_hidden, test_queue_is_oldest_first,
+                     test_two_moderators_cannot_double_review,
+                     test_rejected_stays_hidden, test_new_media_is_private,
+                     test_only_public_and_approved_can_sync,
+                     test_sharing_requires_approval,
+                     test_unsharing_stops_future_syncs, test_vault_never_syncs):
+            await test(tmp)
+    print("\nAll database tests passed.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test_vault.py

@@ -0,0 +1,176 @@
+"""Vault tests. Run with: python test_vault.py
+
+The vault stores pointers to messages the bot already holds, so the things
+worth pinning down are that albums are grouped into one post, that retrieval
+asks Telegram for the right message ids, and that one user can never reach
+another user's saved posts.
+"""
+import asyncio
+import datetime
+import importlib
+import logging
+import os
+import sys
+import tempfile
+
+os.environ.setdefault("BOT_TOKEN", "123456789:AAHtestTokenPlaceholder_ABCDEFGHIJKLMNOP")
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from aiogram.types import Chat, Message, PhotoSize, User
+
+logging.disable(logging.CRITICAL)
+
+CHAT = Chat(id=500, type="private")
+USER = User(id=111, is_bot=False, first_name="Owner")
+NOW = datetime.datetime.now()
+
+
+def load(path: str, **env):
+    for key in ("PUBLIC_MODE", "REQUIRED_CHATS", "ALLOWED_USERS", "ADMIN_USERS"):
+        os.environ.pop(key, None)
+    os.environ["DB_PATH"] = path
+    os.environ.update(env)
+    import db, bot
+    importlib.reload(db)
+    B = importlib.reload(bot)
+    B.db = db
+    return db, B
+
+
+def photo_msg(message_id: int, caption: str | None = None, group: str | None = None) -> Message:
+    photo = PhotoSize(file_id=f"F{message_id}", file_unique_id=f"U{message_id}", width=1, height=1)
+    return Message(message_id=message_id, date=NOW, chat=CHAT, from_user=USER,
+                   photo=[photo], caption=caption, media_group_id=group)
+
+
+# ── database ─────────────────────────────────────────────────────────────────
+
+async def test_roundtrip(db):
+    item_id = await db.add_vault_item(111, 500, [12, 10, 11], "A trip to the sea", ["sea"])
+    item = await db.get_vault_item(111, item_id)
+    assert db.unpack_message_ids(item["message_ids"]) == [12, 10, 11]
+    assert item["caption"] == "A trip to the sea"
+    print("  a saved post round-trips with its message ids and caption")
+
+
+async def test_search_by_keyword_and_caption(db):
+    await db.add_vault_item(111, 500, [20], "Recipe for lasagne", ["food"])
+    assert len(await db.search_vault(111, "food")) == 1, "keyword search"
+    assert len(await db.search_vault(111, "lasagne")) == 1, "caption search"
+    assert len(await db.search_vault(111, "lasagne")) == 1
+    assert await db.search_vault(111, "nonsense") == []
+    print("  posts are findable by keyword and by words in the original caption")
+
+
+async def test_vault_is_private_to_its_owner(db):
+    mine = await db.add_vault_item(111, 500, [30], "private note", ["secret"])
+
+    assert await db.get_vault_item(222, mine) is None, "another user fetched my post"
+    assert await db.search_vault(222, "secret") == [], "another user searched my vault"
+    assert await db.delete_vault_item(222, mine) is False, "another user deleted my post"
+    assert await db.get_vault_item(111, mine) is not None, "owner lost their own post"
+    print("  one user cannot read, search or delete another user's vault")
+
+
+async def test_delete(db):
+    item_id = await db.add_vault_item(111, 500, [40], "", ["bye"])
+    assert await db.delete_vault_item(111, item_id) is True
+    assert await db.get_vault_item(111, item_id) is None
+    assert await db.delete_vault_item(111, item_id) is False
+    print("  deleting removes the post and is safe to repeat")
+
+
+# ── album grouping ───────────────────────────────────────────────────────────
+
+async def test_album_becomes_one_post(B):
+    B.ALBUM_DEBOUNCE = 0.05
+    B._album_buffers.clear()
+    B._pending_forwards.clear()
+    sent: list = []
+
+    async def send_message(chat_id, text, **kw):
+        sent.append(text)
+    B.bot.send_message = send_message
+
+    # Telegram delivers an album as separate messages; only one has the caption.
+    for msg in (photo_msg(10, caption="beach day", group="g1"),
+                photo_msg(11, group="g1"),
+                photo_msg(12, group="g1")):
+        await B.on_forwarded(msg, None)
+    await asyncio.sleep(0.3)
+
+    assert len(sent) == 1, f"an album should prompt once, got {len(sent)} prompts"
+    assert len(B._pending_forwards) == 1
+    record = next(iter(B._pending_forwards.values()))
+    assert record["message_ids"] == [10, 11, 12]
+    assert record["caption"] == "beach day", "the one caption in the album was lost"
+    assert len(record["media"]) == 3, "every item in the album should be extractable"
+    print("  an album is buffered into one post, keeping its caption and all media")
+
+
+async def test_single_forward_prompts_immediately(B):
+    B._album_buffers.clear()
+    B._pending_forwards.clear()
+    sent: list = []
+
+    async def send_message(chat_id, text, **kw):
+        sent.append(text)
+    B.bot.send_message = send_message
+
+    await B.on_forwarded(photo_msg(20, caption="just one"), None)
+    assert len(sent) == 1, "a lone forward should not wait for an album"
+    record = next(iter(B._pending_forwards.values()))
+    assert record["message_ids"] == [20]
+    print("  a single forwarded post prompts straight away")
+
+
+async def test_pending_prompts_are_bounded(B):
+    B._pending_forwards.clear()
+    B.bot.send_message = lambda *a, **kw: asyncio.sleep(0)
+
+    for i in range(B.PENDING_FORWARD_LIMIT + 25):
+        await B.prompt_forward_choice(
+            {"user_id": 1, "chat_id": 500, "message_ids": [i], "caption": "", "media": []}
+        )
+    assert len(B._pending_forwards) <= B.PENDING_FORWARD_LIMIT
+    print("  abandoned prompts cannot grow without bound")
+
+
+async def test_retrieval_forwards_ascending_ids(db, B):
+    item_id = await db.add_vault_item(111, 500, [12, 10, 11], "out of order", ["x"])
+    forwarded: dict = {}
+
+    async def forward_messages(chat_id, from_chat_id, message_ids):
+        forwarded.update(chat_id=chat_id, from_chat_id=from_chat_id, message_ids=message_ids)
+    B.bot.forward_messages = forward_messages
+
+    class Cb:
+        data = f"vault:get:{item_id}"
+        from_user = type("U", (), {"id": 111})()
+        message = type("M", (), {"chat": CHAT})()
+        async def answer(self, *a, **kw): pass
+
+    await B.on_vault_action(Cb())
+    # Telegram requires strictly increasing ids, and the album must stay whole.
+    assert forwarded["message_ids"] == [10, 11, 12], forwarded
+    assert forwarded["from_chat_id"] == 500
+    print("  retrieval forwards the whole post with ids in ascending order")
+
+
+async def main():
+    with tempfile.TemporaryDirectory() as tmp:
+        db, B = load(os.path.join(tmp, "vault.db"), ADMIN_USERS="111")
+        await db.init_db()
+
+        for test in (test_roundtrip, test_search_by_keyword_and_caption,
+                     test_vault_is_private_to_its_owner, test_delete):
+            await test(db)
+        for test in (test_album_becomes_one_post, test_single_forward_prompts_immediately,
+                     test_pending_prompts_are_bounded):
+            await test(B)
+        await test_retrieval_forwards_ascending_ids(db, B)
+    print("\nAll vault tests passed.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())