Offline: newest 100 conversations; notifications no longer wait silently

Offline access keeps the newest 100 Inbox conversations (the list's first two
pages, saved as the app asks for them) plus recently opened ones, up to 130.

Turning notifications on in a browser that holds the permission request
without showing it (Opera, Chrome's quiet prompts) said nothing and never
finished. After a few seconds it now says how to allow notifications for the
site, and carries on by itself as soon as they are allowed.

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

backend/app/accounts/service.py

@@ -11,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.accounts.schemas import CreateAccountRequest
 from app.accounts.types import ServerConfig
 from app.indexer import blobstore
-from app.models import Account, Attachment, Message, Thread
+from app.models import Account, Attachment, ComposeAttachment, Message, Thread
 from app.security.crypto import encrypt_secret
 
 
@@ -127,6 +127,13 @@ async def delete_account(
                 .where(Attachment.user_id == user_id, Attachment.blob_sha256 == sha)
                 .limit(1)
             )
+        if still_used is None:
+            # A forwarded file waiting in a draft shares the hash: keep it for that.
+            still_used = await session.scalar(
+                select(ComposeAttachment.id)
+                .where(ComposeAttachment.user_id == user_id, ComposeAttachment.blob_sha256 == sha)
+                .limit(1)
+            )
         if still_used is None:
             blobstore.remove(user_id, sha)
     return True

backend/app/api/compose.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import uuid
 from datetime import UTC, datetime, timedelta
 from typing import Annotated
@@ -15,6 +16,7 @@ from app.api.schemas import AttachmentOut
 from app.config import get_settings
 from app.db import get_session
 from app.identity.deps import CurrentUser, require_csrf
+from app.indexer import attachments as att_store
 from app.indexer import blobstore
 from app.models import Attachment, BodyPart, ComposeAttachment, Draft, Mailbox, Message, Outbox
 from app.outbox.jobs import RETRY_WINDOW
@@ -207,6 +209,11 @@ async def adopt_remote_draft(message_id: uuid.UUID, user: CurrentUser, db: DbSes
     )
     attachment_ids: list[str] = []
     for a in att_rows:
+        # Kept only inside the message: give the compose attachment a blob of its own.
+        if not await asyncio.to_thread(
+            att_store.materialize, user.id, msg.blob_sha256, a.blob_sha256, a.part_index
+        ):
+            continue
         ca = ComposeAttachment(
             user_id=user.id,
             filename=a.filename,

backend/app/api/messages.py

@@ -20,6 +20,7 @@ from app.api.schemas import AttachmentOut
 from app.api.settings import add_allowed_image_sender, allowed_image_senders
 from app.db import get_session
 from app.identity.deps import CurrentUser, require_csrf
+from app.indexer import attachments as att_store
 from app.indexer import blobstore, thumbs
 from app.indexer.mime import parse_message
 from app.indexer.sanitize import SANITIZER_VERSION, sanitize_html
@@ -176,7 +177,7 @@ async def message_cid(
     )
     if att is None:
         raise_api_error(404, "not_found", "inline part not found")
-    data = blobstore.load(msg.user_id, att.blob_sha256)
+    data = await _attachment_bytes(msg, att)
     media_type = att.mime_type if att.mime_type in _INLINE_SAFE else "application/octet-stream"
     return Response(content=data, media_type=media_type)
 
@@ -190,6 +191,16 @@ def _content_disposition(disposition: str, filename: str) -> str:
     return f"{disposition}; filename=\"{fallback}\"; filename*=UTF-8''{quote(filename)}"
 
 
+async def _attachment_bytes(msg: Message, att: Attachment) -> bytes:
+    """An attachment's bytes, taken out of its message (404 if they cannot be)."""
+    try:
+        return await asyncio.to_thread(
+            att_store.load, msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index
+        )
+    except FileNotFoundError:
+        raise_api_error(404, "not_found", "attachment content missing")
+
+
 async def _real_attachments(db: AsyncSession, message_id: uuid.UUID) -> list[Attachment]:
     """Downloadable (non-inline) attachments for a message, in a stable order."""
     rows = await db.scalars(
@@ -221,9 +232,10 @@ async def forward_attachments(
     message_id: uuid.UUID, user: CurrentUser, db: DbSession
 ) -> list[AttachmentOut]:
     """Attach a message's files to a forward of it: one compose attachment per
-    real (non-inline) attachment, sharing the stored blob — nothing is copied.
+    real (non-inline) attachment. Each is taken out of the message into a blob
+    of its own (shared, by hash, with any other use of the same file).
     Returns them as the composer lists its attachments."""
-    await _owned_message(db, user.id, message_id)
+    msg = await _owned_message(db, user.id, message_id)
     rows = [
         ComposeAttachment(
             user_id=user.id,
@@ -233,7 +245,9 @@ async def forward_attachments(
             blob_sha256=a.blob_sha256,
         )
         for a in await _real_attachments(db, message_id)
-        if blobstore.exists(user.id, a.blob_sha256)
+        if await asyncio.to_thread(
+            att_store.materialize, user.id, msg.blob_sha256, a.blob_sha256, a.part_index
+        )
     ]
     db.add_all(rows)
     await db.commit()
@@ -276,7 +290,7 @@ async def download_attachments_zip(
                 name = f"{stem}-{seen[name]}{dot}{ext}" if dot else f"{name}-{seen[name]}"
             else:
                 seen[name] = 1
-            zf.writestr(name, blobstore.load(msg.user_id, att.blob_sha256))
+            zf.writestr(name, await _attachment_bytes(msg, att))
 
     return Response(
         content=buffer.getvalue(),
@@ -301,7 +315,14 @@ async def attachment_thumbnail(
     kind = thumb_type(att.mime_type, att.filename) if att is not None else None
     if att is None or kind is None:
         raise_api_error(404, "not_found", "no thumbnail")
-    data = await asyncio.to_thread(thumbs.thumbnail, msg.user_id, att.blob_sha256, kind)
+    data = await asyncio.to_thread(
+        thumbs.thumbnail,
+        msg.user_id,
+        att.blob_sha256,
+        kind,
+        att.size_bytes,
+        lambda: att_store.load(msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index),
+    )
     if data is None:
         raise_api_error(404, "not_found", "no thumbnail")
     # An attachment never changes under its id: the browser may keep this.
@@ -329,8 +350,13 @@ async def message_attachment(
     )
     if att is None:
         raise_api_error(404, "not_found", "attachment not found")
-    path = blobstore.blob_path(msg.user_id, att.blob_sha256)
-    if not path.exists():
+    try:
+        # Taken out of the message into a short-lived file (kept a day after
+        # its last use), so it can be streamed with HTTP Range like before.
+        path = await asyncio.to_thread(
+            att_store.file, msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index
+        )
+    except FileNotFoundError:
         raise_api_error(404, "not_found", "attachment content missing")
     inline_type = None if download else _inline_type(att.mime_type, att.filename)
     disposition = "inline" if inline_type else "attachment"

backend/app/indexer/attachments.py

@@ -0,0 +1,237 @@
+"""Attachments, kept once: inside their message (SPEC App. D §D.6 item 1).
+
+An attachment is not stored apart from its message any more — it was, which
+kept most of a mailbox's bytes twice (once base64 in the raw message, once
+decoded on its own). It is taken out of the raw message when wanted:
+
+* :func:`load` — its bytes (downloads, zip, inline images, forwarding, thumbnails);
+* :func:`file` — a file holding it, for streaming with HTTP Range (videos):
+  taken out into ``{user}/cache/`` and kept a day after its last use;
+* :func:`dedupe_existing` — the one-time pass removing the separate copies
+  stored before, each only after checking it can be taken out of its message.
+
+Blocking (parsing, disk): call these in a thread.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import logging
+import time
+import uuid
+from collections import defaultdict
+from pathlib import Path
+
+from app.indexer import blobstore
+from app.indexer.mime import load_message, part_payload
+
+logger = logging.getLogger(__name__)
+
+CACHE_KEEP_S = 24 * 3600  # a taken-out file is kept this long after its last use
+CACHE_MAX_BYTES = 1024**3  # per user; the least recently used go first beyond it
+# part_index of a row whose content is not in its message (kept as its own file).
+NOT_IN_MESSAGE = -1
+
+
+def _legacy(user_id: uuid.UUID, sha: str) -> Path | None:
+    """The separate copy stored before attachments were kept once, if still there."""
+    path = blobstore.blob_path(user_id, sha)
+    return path if path.exists() else None
+
+
+def load(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> bytes:
+    """An attachment's bytes. Raises FileNotFoundError when they cannot be had."""
+    legacy = _legacy(user_id, sha)
+    if legacy is not None:
+        return legacy.read_bytes()
+    raw = blobstore.blob_path(user_id, message_sha)
+    if part_index == NOT_IN_MESSAGE or not raw.exists():
+        raise FileNotFoundError(sha)
+    payload = part_payload(load_message(path=raw), part_index, sha)
+    if payload is None:
+        raise FileNotFoundError(sha)
+    return payload
+
+
+def materialize(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> bool:
+    """Make sure the attachment exists as a blob of its own — for a compose
+    attachment (forward, a draft reopened), which is read by its hash. False
+    when its bytes cannot be had."""
+    if blobstore.exists(user_id, sha):
+        return True
+    try:
+        blobstore.store(user_id, load(user_id, message_sha, sha, part_index))
+    except FileNotFoundError:
+        return False
+    return True
+
+
+def _cache_dir(user_id: uuid.UUID) -> Path:
+    return blobstore.blob_path(user_id, "00").parent.parent / "cache"
+
+
+def prune_cache(cache: Path) -> None:
+    """Drop streaming copies past their keep time, and the oldest beyond the cap."""
+    files = [(p, p.stat()) for p in cache.iterdir() if p.is_file()]
+    now = time.time()
+    total = 0
+    for path, st in sorted(files, key=lambda f: f[1].st_mtime, reverse=True):
+        total += st.st_size
+        if now - st.st_mtime > CACHE_KEEP_S or total > CACHE_MAX_BYTES:
+            path.unlink(missing_ok=True)
+
+
+def file(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> Path:
+    """A file holding the attachment, to stream from. Raises FileNotFoundError."""
+    legacy = _legacy(user_id, sha)
+    if legacy is not None:
+        return legacy
+    cache = _cache_dir(user_id)
+    path = cache / sha
+    if path.exists():
+        path.touch()  # used again: kept longer
+        return path
+    data = load(user_id, message_sha, sha, part_index)
+    cache.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(f".{uuid.uuid4().hex}.tmp")
+    tmp.write_bytes(data)
+    tmp.replace(path)
+    with contextlib.suppress(OSError):  # a file removed meanwhile by another request
+        prune_cache(cache)
+    return path
+
+
+def drop_cached(user_id: uuid.UUID) -> None:
+    """Forget every taken-out file of a user (account removed)."""
+    cache = _cache_dir(user_id)
+    if cache.exists():
+        for path in cache.iterdir():
+            path.unlink(missing_ok=True)
+
+
+# ---- removing the copies stored before ------------------------------------------------
+
+# A file this recent may belong to work in progress (a sent message waiting
+# for its Sent copy is referenced only by its job): left for the next pass.
+_MIN_AGE_S = 3600
+
+
+def _part_indexes(raw_path: Path) -> dict[str, int]:
+    """Content hash → MIME part position, for every leaf part of a message."""
+    out: dict[str, int] = {}
+    for index, part in enumerate(load_message(path=raw_path).walk()):
+        if part.is_multipart():
+            continue
+        payload = part.get_payload(decode=True)
+        if isinstance(payload, bytes):
+            out.setdefault(hashlib.sha256(payload).hexdigest(), index)
+    return out
+
+
+async def dedupe_existing(batch: int = 200, *, user_id: uuid.UUID | None = None) -> dict[str, int]:
+    """Record where each older attachment sits in its message, and delete its
+    separate copy once it is known to be there. Idempotent; resumable."""
+    import asyncio
+
+    from sqlalchemy import select, update
+
+    from app.db import session_scope
+    from app.models import Attachment, ComposeAttachment, Message
+
+    stats = {"rows": 0, "found": 0, "not_found": 0, "files_removed": 0, "bytes_freed": 0}
+    while True:
+        async with session_scope() as db:
+            rows = (
+                await db.execute(
+                    select(
+                        Attachment.id,
+                        Attachment.user_id,
+                        Attachment.blob_sha256,
+                        Message.blob_sha256,
+                    )
+                    .join(Message, Message.id == Attachment.message_id)
+                    .where(
+                        Attachment.part_index.is_(None),
+                        *([Attachment.user_id == user_id] if user_id is not None else []),
+                    )
+                    .order_by(Attachment.message_id)
+                    .limit(batch)
+                )
+            ).all()
+        if not rows:
+            return stats
+        by_message: dict[tuple[uuid.UUID, str], list[tuple[uuid.UUID, str]]] = defaultdict(list)
+        for att_id, owner, sha, msg_sha in rows:
+            by_message[(owner, msg_sha)].append((att_id, sha))
+
+        located: dict[uuid.UUID, int] = {}
+        for (owner, msg_sha), atts in by_message.items():
+            raw = blobstore.blob_path(owner, msg_sha)
+            try:
+                indexes = await asyncio.to_thread(_part_indexes, raw) if raw.exists() else {}
+            except Exception:  # noqa: BLE001 - unreadable message: keep the copies
+                logger.warning("attachments: cannot parse %s", raw, exc_info=True)
+                indexes = {}
+            for att_id, sha in atts:
+                located[att_id] = indexes.get(sha, NOT_IN_MESSAGE)
+
+        async with session_scope() as db:
+            for att_id, index in located.items():
+                await db.execute(
+                    update(Attachment).where(Attachment.id == att_id).values(part_index=index)
+                )
+        stats["rows"] += len(located)
+        stats["found"] += sum(1 for i in located.values() if i != NOT_IN_MESSAGE)
+        stats["not_found"] += sum(1 for i in located.values() if i == NOT_IN_MESSAGE)
+
+        # A separate copy goes only when every attachment row with that content
+        # is found in its message and nothing else uses the file.
+        candidates = {
+            (owner, sha)
+            for att_id, owner, sha, _ in rows
+            if located.get(att_id, NOT_IN_MESSAGE) != NOT_IN_MESSAGE
+        }
+        async with session_scope() as db:
+            for owner, sha in candidates:
+                path = blobstore.blob_path(owner, sha)
+                if not path.exists() or time.time() - path.stat().st_mtime < _MIN_AGE_S:
+                    continue
+                in_use = (
+                    await db.scalar(
+                        select(Message.id)
+                        .where(Message.user_id == owner, Message.blob_sha256 == sha)
+                        .limit(1)
+                    )
+                    or await db.scalar(
+                        select(ComposeAttachment.id)
+                        .where(
+                            ComposeAttachment.user_id == owner,
+                            ComposeAttachment.blob_sha256 == sha,
+                        )
+                        .limit(1)
+                    )
+                    or await db.scalar(
+                        select(Attachment.id)
+                        .where(
+                            Attachment.user_id == owner,
+                            Attachment.blob_sha256 == sha,
+                            (Attachment.part_index.is_(None))
+                            | (Attachment.part_index == NOT_IN_MESSAGE),
+                        )
+                        .limit(1)
+                    )
+                )
+                if in_use:
+                    continue
+                size = path.stat().st_size
+                path.unlink(missing_ok=True)
+                stats["files_removed"] += 1
+                stats["bytes_freed"] += size
+
+
+async def dedupe_existing_job(ctx: dict[str, object]) -> dict[str, int]:
+    """arq job (bulk queue): :func:`dedupe_existing`, run once after an update."""
+    stats = await dedupe_existing()
+    logger.info("attachments kept once: %s", stats)
+    return stats

backend/app/indexer/gc.py

@@ -0,0 +1,198 @@
+"""The nightly clean-up of stored files nothing uses any more.
+
+Messages leave in many ways — Delete forever, Empty Trash, the Trash/Spam
+retention purge, an expunge on the server, a folder re-synced — and none of
+them removes the message's file: files are stored by content, so one file can
+serve several rows (the same email in two folders, a forwarded file in a
+draft), and removing it is only safe once nothing refers to it. This pass does
+that for every user, whatever the way the rows went:
+
+* uploads (compose attachments) no draft has referred to for a week — left by
+  a sent message (its draft is gone) or a composer closed without saving;
+* message/attachment files no row refers to;
+* attachment previews (thumbnails) of attachments that are gone;
+* streaming copies past their keep time (app/indexer/attachments.py).
+
+A file younger than an hour is left alone: it may belong to work in progress
+(a message being stored, a sent message waiting for its Sent copy).
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import logging
+import re
+import time
+import uuid
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import delete, or_, select
+
+from app.config import get_settings
+from app.indexer import attachments as att_store
+
+logger = logging.getLogger(__name__)
+
+MIN_AGE_S = 3600
+UPLOAD_KEEP = dt.timedelta(days=7)
+_SHA = re.compile(r"^[0-9a-f]{64}$")
+
+
+# ---- files a job still needs ---------------------------------------------------------
+#
+# A sent message whose Sent copy could not be filed yet is referenced only by
+# its retrying job (app/outbox/jobs.py), for up to a day: held here meanwhile.
+# Entries older than HOLD_MAX are ignored, in case a job died without letting go.
+
+HOLD_MAX_S = 3 * 24 * 3600
+
+
+def _hold_key(user_id: uuid.UUID | str) -> str:
+    return f"blob:held:{user_id}"
+
+
+async def hold(user_id: uuid.UUID | str, sha: str) -> None:
+    from app.redis_client import get_redis
+
+    await get_redis().zadd(_hold_key(user_id), {sha: time.time()})
+
+
+async def release(user_id: uuid.UUID | str, sha: str) -> None:
+    from app.redis_client import get_redis
+
+    await get_redis().zrem(_hold_key(user_id), sha)
+
+
+async def _held(user_id: uuid.UUID) -> set[str]:
+    from app.redis_client import get_redis
+
+    redis = get_redis()
+    oldest = time.time() - HOLD_MAX_S
+    await redis.zremrangebyscore(_hold_key(user_id), "-inf", oldest)
+    return {
+        m.decode() if isinstance(m, bytes) else str(m)
+        for m in await redis.zrangebyscore(_hold_key(user_id), oldest, "+inf")
+    }
+
+
+def _draft_upload_ids(payloads: list[dict[str, Any]]) -> set[uuid.UUID]:
+    out: set[uuid.UUID] = set()
+    for payload in payloads:
+        for value in payload.get("attachment_ids") or []:
+            try:
+                out.add(uuid.UUID(str(value)))
+            except ValueError:
+                continue
+    return out
+
+
+def _old(path: Path, now: float) -> bool:
+    return now - path.stat().st_mtime >= MIN_AGE_S
+
+
+async def sweep_user(user_id: uuid.UUID) -> dict[str, int]:
+    """Clean one user's stored files; returns what was removed."""
+    from app.db import session_scope
+    from app.models import Attachment, ComposeAttachment, Draft, Message
+
+    stats = {"uploads": 0, "files": 0, "bytes": 0, "thumbs": 0}
+    async with session_scope() as db:
+        # Uploads no draft refers to, once a week old.
+        used = _draft_upload_ids(
+            list(await db.scalars(select(Draft.payload).where(Draft.user_id == user_id)))
+        )
+        stale = select(ComposeAttachment.id).where(
+            ComposeAttachment.user_id == user_id,
+            ComposeAttachment.created_at < dt.datetime.now(dt.UTC) - UPLOAD_KEEP,
+        )
+        if used:
+            stale = stale.where(ComposeAttachment.id.not_in(used))
+        gone = list(await db.scalars(stale))
+        if gone:
+            await db.execute(delete(ComposeAttachment).where(ComposeAttachment.id.in_(gone)))
+        stats["uploads"] = len(gone)
+
+    async with session_scope() as db:
+        # Every file still referred to. Attachments count only where their file
+        # is their only copy (not found in the message, or not yet looked for).
+        referenced = set(
+            await db.scalars(select(Message.blob_sha256).where(Message.user_id == user_id))
+        )
+        referenced |= set(
+            await db.scalars(
+                select(ComposeAttachment.blob_sha256).where(ComposeAttachment.user_id == user_id)
+            )
+        )
+        referenced |= set(
+            await db.scalars(
+                select(Attachment.blob_sha256).where(
+                    Attachment.user_id == user_id,
+                    or_(
+                        Attachment.part_index.is_(None),
+                        Attachment.part_index == att_store.NOT_IN_MESSAGE,
+                    ),
+                )
+            )
+        )
+        referenced |= await _held(user_id)
+        attachment_shas = set(
+            await db.scalars(select(Attachment.blob_sha256).where(Attachment.user_id == user_id))
+        )
+
+    root = Path(get_settings().blob_root) / str(user_id)
+    if not root.is_dir():
+        return stats
+    now = time.time()
+    for path in root.glob("??/*"):
+        if not path.is_file() or not _SHA.match(path.name) or path.name in referenced:
+            continue
+        if not _old(path, now):
+            continue
+        size = path.stat().st_size
+        path.unlink(missing_ok=True)
+        stats["files"] += 1
+        stats["bytes"] += size
+    thumbs = root / "thumbs"
+    if thumbs.is_dir():
+        for path in thumbs.iterdir():
+            sha = path.name.split("-", 1)[0]
+            if (
+                path.is_file()
+                and _SHA.match(sha)
+                and sha not in attachment_shas
+                and _old(path, now)
+            ):
+                path.unlink(missing_ok=True)
+                stats["thumbs"] += 1
+    cache = root / "cache"
+    if cache.is_dir():
+        att_store.prune_cache(cache)
+    return stats
+
+
+async def sweep(user_id: uuid.UUID | None = None) -> dict[str, int]:
+    """:func:`sweep_user` for one user, or every user."""
+    from app.db import session_scope
+    from app.models import User
+
+    async with session_scope() as db:
+        users = [user_id] if user_id is not None else list(await db.scalars(select(User.id)))
+    total = {"users": 0, "uploads": 0, "files": 0, "bytes": 0, "thumbs": 0}
+    for uid in users:
+        try:
+            stats = await sweep_user(uid)
+        except Exception:  # noqa: BLE001 - one user's trouble must not stop the rest
+            logger.warning("stored-file clean-up failed for user %s", uid, exc_info=True)
+            continue
+        total["users"] += 1
+        for key, value in stats.items():
+            total[key] += value
+    return total
+
+
+async def sweep_job(ctx: dict[str, Any]) -> dict[str, int]:
+    """arq cron job: the nightly :func:`sweep`."""
+    stats = await sweep()
+    logger.info("stored-file clean-up: %s", stats)
+    return stats

backend/app/indexer/indexer.py

@@ -216,8 +216,9 @@ async def persist_message(
             )
         )
 
+    # Attachments are not stored apart: they stay inside the raw blob, and are
+    # taken out of it when wanted (app/indexer/attachments.py).
     for att in parsed.attachments:
-        att_sha = blobstore.store(user_id, att.payload)
         session.add(
             Attachment(
                 message_id=message.id,
@@ -226,7 +227,8 @@ async def persist_message(
                 mime_type=att.mime_type,
                 size_bytes=len(att.payload),
                 content_id=att.content_id,
-                blob_sha256=att_sha,
+                blob_sha256=blobstore.sha256_hex(att.payload),
+                part_index=att.part_index,
             )
         )
 

backend/app/indexer/mime.py

@@ -41,6 +41,9 @@ class ParsedAttachment:
     mime_type: str
     content_id: str | None
     payload: bytes
+    # The part's position in ``msg.walk()``: where to find it again in the raw
+    # message, which is the only copy kept (attachments are not stored apart).
+    part_index: int = -1
 
 
 @dataclass
@@ -114,7 +117,7 @@ def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]:
     # or inline resource if it has a filename, a Content-ID, or attachment disposition;
     # the body text/plain and text/html parts have none of these and are skipped.
     out: list[ParsedAttachment] = []
-    for part in msg.walk():
+    for index, part in enumerate(msg.walk()):
         if part.is_multipart() or not isinstance(part, EmailMessage):
             continue
         cid = part.get("Content-ID")
@@ -137,13 +140,14 @@ def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]:
                 mime_type=part.get_content_type(),
                 content_id=cid.strip("<>") if (inline and cid is not None) else None,
                 payload=payload,
+                part_index=index,
             )
         )
     # An invitation sent only as the text/calendar alternative (Outlook, often) has
     # no file to show: keep it as invite.ics, so the event can be seen, answered
     # and downloaded like a Gmail one. Not when the message attaches its own .ics.
     if not any(a.filename.lower().endswith(".ics") for a in out):
-        for part in msg.walk():
+        for index, part in enumerate(msg.walk()):
             if part.get_content_type() != "text/calendar" or part.get_filename():
                 continue
             payload = part.get_payload(decode=True)
@@ -154,6 +158,7 @@ def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]:
                         mime_type="text/calendar",
                         content_id=None,
                         payload=payload,
+                        part_index=index,
                     )
                 )
                 break
@@ -215,3 +220,36 @@ def _parse_email(msg: EmailMessage) -> ParsedEmail:
         html_body=_scrub(html_body) if html_body is not None else None,
         attachments=_extract_attachments(msg),
     )
+
+
+def load_message(
+    raw: bytes | None = None, path: str | os.PathLike[str] | None = None
+) -> EmailMessage:
+    """The parsed tree of a raw message, from memory or straight from its file."""
+    if path is not None:
+        with open(path, "rb") as handle:
+            msg = email.message_from_binary_file(handle, policy=default_policy)
+    else:
+        msg = email.message_from_bytes(raw or b"", policy=default_policy)
+    assert isinstance(msg, EmailMessage)
+    return msg
+
+
+def part_payload(msg: EmailMessage, part_index: int | None, sha256: str) -> bytes | None:
+    """The decoded bytes of the attachment whose content hashes to ``sha256``:
+    the part at ``part_index`` when that is it, else whichever part is (rows
+    stored before the index was kept). None when the message has no such part."""
+    import hashlib
+
+    parts = list(msg.walk())
+    if part_index is not None and 0 <= part_index < len(parts):
+        payload = parts[part_index].get_payload(decode=True)
+        if isinstance(payload, bytes) and hashlib.sha256(payload).hexdigest() == sha256:
+            return payload
+    for part in parts:
+        if part.is_multipart():
+            continue
+        payload = part.get_payload(decode=True)
+        if isinstance(payload, bytes) and hashlib.sha256(payload).hexdigest() == sha256:
+            return payload
+    return None

backend/app/indexer/sanitize.py

@@ -17,7 +17,12 @@ import nh3
 
 # Bump when the sanitization policy changes; body_parts.sanitizer_v records it so a
 # render re-sanitizes stale content from the raw blob (SPEC §10.4).
-SANITIZER_VERSION = 4
+# v5: this server's own inline-image URLs quoted in a message load from here.
+SANITIZER_VERSION = 5
+
+# This server's inline-image URL, however it ended up written: "/api/…",
+# "https://host/api/…", or mangled to "http:///api/…" by an earlier copy.
+INTERNAL_CID = re.compile(r"/api/messages/([0-9a-fA-F-]{36})/cid/([^\"'?#\s<>]+)")
 
 _ALLOWED_TAGS: set[str] = {
     # structure
@@ -202,6 +207,13 @@ def sanitize_html(
             if lowered.startswith("cid:"):
                 content_id = value.strip()[4:]
                 return f"/api/messages/{message_id}/cid/{content_id}"
+            # This server's own inline-image URL, quoted from another message
+            # (by an earlier version, or another of the user's own mails) —
+            # possibly mangled to "http:///api/…", which a browser takes for a
+            # host named "api": load it from here, never look that host up.
+            internal = INTERNAL_CID.search(value)
+            if internal is not None and not lowered.startswith("data:"):
+                return f"/api/messages/{internal.group(1)}/cid/{internal.group(2)}"
             if lowered.startswith(("http:", "https:")):
                 state["has_remote"] = True
                 return value if allow_remote_images else None  # block by default (FR-23)

backend/app/indexer/thumbs.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 import io
 import logging
 import uuid
+from collections.abc import Callable
 from pathlib import Path
 
 from app.indexer import blobstore
@@ -86,8 +87,16 @@ def _render(data: bytes, mime_type: str) -> bytes | None:
     return out.getvalue()
 
 
-def thumbnail(user_id: uuid.UUID, sha: str, mime_type: str) -> bytes | None:
-    """The thumbnail of a stored attachment, or None if it cannot have one.
+def thumbnail(
+    user_id: uuid.UUID,
+    sha: str,
+    mime_type: str,
+    size: int,
+    load: Callable[[], bytes],
+) -> bytes | None:
+    """The thumbnail of an attachment (content hash ``sha``, ``size`` bytes,
+    read with ``load`` only when it is not made yet), or None if it cannot
+    have one.
 
     Blocking (decoding is CPU work): call it in a thread.
     """
@@ -97,11 +106,14 @@ def thumbnail(user_id: uuid.UUID, sha: str, mime_type: str) -> bytes | None:
     none = _path(user_id, sha, ".none")
     if none.exists() or not can_thumbnail(mime_type):
         return None
-    path = blobstore.blob_path(user_id, sha)
-    if not path.exists() or path.stat().st_size > _MAX_INPUT_BYTES:
+    if size > _MAX_INPUT_BYTES:
         return None
     try:
-        thumb = _render(path.read_bytes(), mime_type)
+        data = load()
+    except FileNotFoundError:
+        return None
+    try:
+        thumb = _render(data, mime_type)
     except Exception as exc:  # noqa: BLE001 - any undecodable file: no thumbnail
         logger.info("no thumbnail for blob %s (%s): %s", sha, mime_type, exc)
         thumb = None

backend/app/models.py

@@ -278,7 +278,11 @@ class Attachment(Base):
     mime_type: Mapped[str] = mapped_column(Text, nullable=False)
     size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
     content_id: Mapped[str | None] = mapped_column(Text)
+    # The content's hash (names its thumbnail, checks what is taken out). The
+    # bytes themselves live only inside the message's raw blob, at MIME part
+    # ``part_index`` (NULL for rows stored before that was recorded).
     blob_sha256: Mapped[str] = mapped_column(Text, nullable=False)
+    part_index: Mapped[int | None] = mapped_column(Integer)
 
 
 class ComposeAttachment(Base):

backend/app/outbox/draft_sync.py

@@ -17,6 +17,7 @@ from sqlalchemy import select
 
 from app.db import session_scope
 from app.models import Account, Draft, Mailbox, Message
+from app.outbox import inline_images
 from app.outbox.attachments import load_attachments
 from app.outbox.mime_build import build_mime
 from app.sync.provider_factory import build_provider
@@ -68,7 +69,15 @@ async def push_draft_to_imap_job(ctx: dict[str, Any], draft_id: str) -> None:
             else None
         )
         attachments = await load_attachments(session, user_id, draft)
-        _, sent_raw, _, _ = build_mime(account, draft, reply_to, attachments)
+        html, inline = await inline_images.embed(
+            session,
+            user_id,
+            str(draft.payload.get("html", "") or ""),
+            account.email.rsplit("@", 1)[-1],
+        )
+        _, sent_raw, _, _ = build_mime(
+            account, draft, reply_to, attachments, html=html, inline=inline
+        )
         drafts_mb = await session.scalar(
             select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts")
         )

backend/app/outbox/inline_images.py

@@ -0,0 +1,103 @@
+"""Inline images in an outgoing message's HTML.
+
+A reply quotes the original as the reading pane shows it, so an image embedded
+in the original (``cid:`` — a signature logo, a pasted screenshot) appears in
+the quote as this server's own URL, ``/api/messages/{id}/cid/{content-id}``.
+Sent as is, that URL means nothing to the recipient (and to anyone else,
+including ourselves once the original is gone). Before sending, each such image
+is put back into the message as an inline part with a new Content-ID — as any
+mail program does — and one that can no longer be had is left out.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import uuid
+from dataclasses import dataclass
+from email.utils import make_msgid
+from urllib.parse import unquote
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.indexer import attachments as att_store
+from app.indexer.sanitize import INTERNAL_CID
+from app.models import Attachment, Message
+
+_IMG = re.compile(r"<img\b[^>]*?\bsrc\s*=\s*([\"'])(.*?)\1[^>]*>", re.IGNORECASE | re.DOTALL)
+
+
+@dataclass(frozen=True)
+class InlinePart:
+    cid: str  # without angle brackets
+    mime_type: str
+    filename: str
+    data: bytes
+
+
+def internal_cid(src: str) -> tuple[uuid.UUID, str] | None:
+    """The (message id, content id) an image URL of this server points at."""
+    value = src.strip()
+    if value.lower().startswith(("cid:", "data:")):
+        return None
+    m = INTERNAL_CID.search(value)
+    if m is None:
+        return None
+    try:
+        return uuid.UUID(m.group(1)), unquote(m.group(2))
+    except ValueError:
+        return None
+
+
+async def embed(
+    session: AsyncSession, user_id: uuid.UUID, html: str, domain: str
+) -> tuple[str, list[InlinePart]]:
+    """``html`` with this server's image URLs turned into ``cid:`` references,
+    and the parts to send them in."""
+    parts: dict[tuple[uuid.UUID, str], InlinePart | None] = {}
+
+    for match in _IMG.finditer(html):
+        target = internal_cid(match.group(2))
+        if target is None or target in parts:
+            continue
+        message_id, content_id = target
+        found = (
+            await session.execute(
+                select(Attachment, Message.blob_sha256)
+                .join(Message, Message.id == Attachment.message_id)
+                .where(
+                    Attachment.message_id == message_id,
+                    Attachment.content_id == content_id,
+                    Message.user_id == user_id,
+                )
+                .limit(1)
+            )
+        ).first()
+        part: InlinePart | None = None
+        if found is not None:
+            att, message_sha = found
+            try:
+                data = await asyncio.to_thread(
+                    att_store.load, user_id, message_sha, att.blob_sha256, att.part_index
+                )
+                cid = make_msgid(domain=domain)[1:-1]
+                part = InlinePart(cid, att.mime_type, att.filename, data)
+            except FileNotFoundError:
+                part = None
+        parts[target] = part
+
+    def replace(match: re.Match[str]) -> str:
+        target = internal_cid(match.group(2))
+        if target is None:
+            return match.group(0)
+        part = parts.get(target)
+        if part is None:
+            return ""  # gone: no broken image for the recipient
+        quote = match.group(1)
+        return match.group(0).replace(
+            f"src={quote}{match.group(2)}{quote}", f"src={quote}cid:{part.cid}{quote}", 1
+        )
+
+    out = _IMG.sub(replace, html) if parts else html
+    return out, [p for p in parts.values() if p is not None]

backend/app/outbox/jobs.py

@@ -14,15 +14,17 @@ from datetime import UTC, datetime, timedelta
 from email.utils import getaddresses
 from typing import Any
 
+from arq import Retry
 from sqlalchemy import select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db import session_scope
 from app.events import get_event_bus
-from app.indexer import blobstore
+from app.indexer import blobstore, gc
 from app.indexer.indexer import persist_message
 from app.models import Account, Draft, KnownRecipient, Mailbox, Message, Outbox
+from app.outbox import inline_images
 from app.outbox.attachments import load_attachments
 from app.outbox.mime_build import build_mime
 from app.outbox.service import delete_mirrored_draft_row
@@ -171,7 +173,21 @@ async def _send_failed(
 
 async def file_sent_copy_job(ctx: dict[str, Any], account_id: str, user_id: str, sha: str) -> None:
     """File a sent message's copy in Sent, after the send itself succeeded but
-    the IMAP server could not be reached. Retried (arq Retry) for ~24 hours."""
+    the IMAP server could not be reached. Retried (arq Retry) for ~24 hours.
+
+    Until then the copy's file is referenced only by this job: it is held from
+    the nightly clean-up (app/indexer/gc.py) and let go once filed or given up."""
+    try:
+        await _file_sent_copy(ctx, account_id, user_id, sha)
+    except Retry:
+        raise  # tried again later: the file is still needed
+    except BaseException:
+        await gc.release(user_id, sha)
+        raise
+    await gc.release(user_id, sha)
+
+
+async def _file_sent_copy(ctx: dict[str, Any], account_id: str, user_id: str, sha: str) -> None:
     uid = uuid.UUID(user_id)
     async with session_scope() as session:
         account = await session.get(Account, uuid.UUID(account_id))
@@ -236,8 +252,16 @@ async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:
             return
         user_id = outbox.user_id
         attachments = await load_attachments(session, user_id, draft)
+        # Images quoted from the original travel inside the message, not as
+        # links to this server (app/outbox/inline_images.py).
+        html, inline = await inline_images.embed(
+            session,
+            user_id,
+            str(draft.payload.get("html", "") or ""),
+            account.email.rsplit("@", 1)[-1],
+        )
         transmit_raw, sent_raw, envelope_from, envelope_to = build_mime(
-            account, draft, reply_to, attachments
+            account, draft, reply_to, attachments, html=html, inline=inline
         )
         sent_path = await session.scalar(
             select(Mailbox.path).where(
@@ -271,6 +295,7 @@ async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:
             except Exception as exc:  # noqa: BLE001
                 logger.warning("outbox %s sent; filing it in Sent failed, will retry: %s", oid, exc)
                 sha = blobstore.store(user_id, sent_raw)
+                await gc.hold(user_id, sha)  # until filed: only the job refers to it
                 await enqueue_job("file_sent_copy_job", str(account_obj.id), str(user_id), sha)
         # Remove the mirrored Drafts-folder copy now that it has been sent.
         if drafts_path and draft_imap_uid:

backend/app/outbox/mime_build.py

@@ -14,6 +14,7 @@ from email.utils import format_datetime, make_msgid
 import nh3
 
 from app.models import Account, Draft, Message
+from app.outbox.inline_images import InlinePart
 
 
 def _html_to_text(html: str) -> str:
@@ -49,6 +50,9 @@ def build_mime(
     draft: Draft,
     reply_to: Message | None,
     attachments: list[tuple[str, str, bytes]] | None = None,
+    *,
+    html: str | None = None,
+    inline: list[InlinePart] | None = None,
 ) -> tuple[bytes, bytes, str, list[str]]:
     """Return ``(transmit_raw, sent_raw, envelope_from, envelope_to)`` for ``draft``.
 
@@ -59,13 +63,17 @@ def build_mime(
 
     ``attachments`` is a list of ``(filename, mime_type, data)`` uploaded via the
     compose Attach action; each is added as a file attachment part.
+
+    ``html`` replaces the draft's HTML when given (images already turned into
+    ``cid:`` references by app/outbox/inline_images.py), and ``inline`` are the
+    image parts those references name, sent alongside the HTML.
     """
     payload = draft.payload
     to = _addr_list(payload.get("to"))
     cc = _addr_list(payload.get("cc"))
     bcc = _addr_list(payload.get("bcc"))
     subject = str(payload.get("subject", ""))
-    html = str(payload.get("html", "") or "")
+    html = html if html is not None else str(payload.get("html", "") or "")
     text = str(payload.get("text", "") or "")
 
     msg = EmailMessage()
@@ -91,6 +99,19 @@ def build_mime(
     if html:
         msg.set_content(text or _html_to_text(html))
         msg.add_alternative(html, subtype="html")
+        if inline:
+            html_part = next(p for p in msg.walk() if p.get_content_type() == "text/html")
+            assert isinstance(html_part, EmailMessage)
+            for part in inline:
+                maintype, _, subtype = part.mime_type.partition("/")
+                html_part.add_related(
+                    part.data,
+                    maintype=maintype or "image",
+                    subtype=subtype or "png",
+                    cid=f"<{part.cid}>",
+                    filename=part.filename,
+                    disposition="inline",
+                )
     else:
         msg.set_content(text)
 

backend/app/queue.py

@@ -21,7 +21,7 @@ QUEUE_INTERACTIVE = "interactive"
 QUEUE_BULK = "bulk"
 
 # Jobs default to the interactive queue; only bulk backfills go to the bulk queue.
-_BULK_JOBS = frozenset({"sync_bootstrap_account"})
+_BULK_JOBS = frozenset({"sync_bootstrap_account", "dedupe_existing_job"})
 
 
 def queue_for(job_name: str) -> str:

backend/migrations/versions/0023_attachment_part_index.py

@@ -0,0 +1,36 @@
+"""attachments are kept once: inside their message
+
+An attachment used to be stored twice — inside the raw message blob, and again
+as a blob of its own. Now only the raw message is kept; ``part_index`` says
+which MIME part of it (its position in a walk of the message) holds the
+attachment, so it can be taken out when downloaded. ``blob_sha256`` stays: the
+attachment's content hash, which names its thumbnail and checks what is taken
+out. Rows stored before this have no index; they are found by that hash.
+
+Existing separate copies are removed by a background pass on the worker
+(app/indexer/attachments.py ``dedupe_existing``), only after checking each one
+can be taken out of its message again.
+
+Revision ID: 0023
+Revises: 0022
+Create Date: 2026-10-08
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "0023"
+down_revision: str | None = "0022"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute("ALTER TABLE attachments ADD COLUMN part_index integer")
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE attachments DROP COLUMN IF EXISTS part_index")

backend/tests/indexer/test_attachments_once.py

@@ -0,0 +1,250 @@
+"""Attachments are kept once — inside their message (SPEC App. D §D.6 item 1,
+acceptance §D.9.5): nothing is stored apart, every way of reading an
+attachment returns the exact bytes, and the clean-up pass removes the copies
+stored before only when they can be taken out of their message again."""
+
+from __future__ import annotations
+
+import io
+import os
+import shutil
+import time
+import uuid
+import zipfile
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from email.message import EmailMessage
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import pytest_asyncio
+from httpx import ASGITransport
+from PIL import Image
+from sqlalchemy import delete, select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+from app.identity.passwords import hash_password
+from app.identity.sessions import create_session
+from app.indexer import attachments as att_store
+from app.indexer import blobstore
+from app.indexer.indexer import persist_message
+from app.main import create_app
+from app.models import Account, Attachment, ComposeAttachment, Mailbox, Message, User
+from app.providers.base import RemoteMessage
+
+
+def _png(w: int, h: int) -> bytes:
+    out = io.BytesIO()
+    Image.new("RGB", (w, h), (200, 80, 40)).save(out, "PNG")
+    return out.getvalue()
+
+
+PNG = _png(640, 360)
+VIDEO = os.urandom(300_000)  # stands in for an .mp4: only the bytes matter here
+NOTES = "Zażółć gęślą jaźń\n".encode() * 50
+LOGO = _png(32, 32)
+
+
+def _raw() -> bytes:
+    m = EmailMessage()
+    m["Subject"] = "Files"
+    m["From"] = "Alice <alice@example.com>"
+    m["To"] = "me@example.com"
+    m["Message-ID"] = f"<{uuid.uuid4().hex}@x.com>"
+    m["Date"] = "Mon, 09 Mar 2026 08:36:37 +0100"
+    m.set_content("see attached")
+    m.add_alternative('<p>see <img src="cid:logo@x"> attached</p>', subtype="html")
+    m.get_payload()[1].add_related(LOGO, "image", "png", cid="<logo@x>")
+    m.add_attachment(PNG, "image", "png", filename="photo.png")
+    m.add_attachment(VIDEO, "video", "mp4", filename="clip.mp4")
+    m.add_attachment(NOTES, "text", "plain", filename="notatki.txt")
+    return m.as_bytes()
+
+
+@dataclass
+class Ctx:
+    client: httpx.AsyncClient
+    db: AsyncSession
+    user_id: uuid.UUID
+    message: Message
+    by_name: dict[str, Attachment]
+
+
+def _user_dir(user_id: uuid.UUID) -> Path:
+    return Path(get_settings().blob_root) / str(user_id)
+
+
+def _stored_bytes(user_id: uuid.UUID) -> int:
+    """Bytes in the user's blob store (not the thumbnail or take-out caches)."""
+    return sum(
+        p.stat().st_size
+        for p in _user_dir(user_id).rglob("*")
+        if p.is_file() and p.parent.name not in ("thumbs", "cache")
+    )
+
+
+@pytest_asyncio.fixture
+async def ctx() -> AsyncIterator[Ctx]:
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    db = async_sessionmaker(engine, expire_on_commit=False)()
+    user = User(handle=f"att{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
+    db.add(user)
+    await db.flush()
+    account = Account(
+        user_id=user.id,
+        email="me@example.com",
+        imap_config={},
+        smtp_config={},
+        secret_ct=b"",
+        discovery_src="manual",
+    )
+    db.add(account)
+    await db.flush()
+    inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
+    db.add(inbox)
+    await db.commit()
+    remote = RemoteMessage(uid=1, raw=_raw(), flags=frozenset(), internal_date=datetime.now(UTC))
+    result = await persist_message(
+        db,
+        user_id=user.id,
+        account_id=account.id,
+        mailbox_id=inbox.id,
+        mailbox_special_use="inbox",
+        remote=remote,
+    )
+    await db.commit()
+    assert result is not None
+    rows = (
+        await db.scalars(select(Attachment).where(Attachment.message_id == result.message.id))
+    ).all()
+    token = await create_session(user.id, "pytest")
+    client = httpx.AsyncClient(
+        transport=ASGITransport(app=create_app()),
+        base_url="http://testserver",
+        cookies={"session": token, "csrf": "t"},
+    )
+    try:
+        yield Ctx(client, db, user.id, result.message, {a.filename: a for a in rows})
+    finally:
+        await client.aclose()
+        await db.execute(delete(User).where(User.id == user.id))
+        await db.commit()
+        await db.close()
+        await engine.dispose()
+        shutil.rmtree(_user_dir(user.id), ignore_errors=True)
+
+
+def _url(c: Ctx, name: str, suffix: str = "") -> str:
+    return f"/api/messages/{c.message.id}/attachments/{c.by_name[name].id}{suffix}"
+
+
+async def test_only_the_message_is_stored(ctx: Ctx) -> None:
+    assert set(ctx.by_name) == {"photo.png", "clip.mp4", "notatki.txt", "attachment"}
+    for att in ctx.by_name.values():
+        assert att.part_index is not None and att.part_index >= 0
+        assert not blobstore.exists(ctx.user_id, att.blob_sha256)  # no copy of its own
+    assert blobstore.exists(ctx.user_id, ctx.message.blob_sha256)
+    # §D.9.5: at most 60% of what keeping the copies too took (base64 inside
+    # the message is ~1.37x the file; kept apart as well it was ~2.37x).
+    raw = _stored_bytes(ctx.user_id)
+    before = raw + len(PNG) + len(VIDEO) + len(NOTES) + len(LOGO)
+    assert raw <= 0.60 * before
+
+
+async def test_every_way_of_reading_gives_the_exact_bytes(ctx: Ctx) -> None:
+    for name, data in (("photo.png", PNG), ("clip.mp4", VIDEO), ("notatki.txt", NOTES)):
+        resp = await ctx.client.get(_url(ctx, name), params={"download": "true"})
+        assert resp.status_code == 200 and resp.content == data, name
+    ranged = await ctx.client.get(_url(ctx, "clip.mp4"), headers={"Range": "bytes=1000-1999"})
+    assert ranged.status_code == 206 and ranged.content == VIDEO[1000:2000]
+    inline = await ctx.client.get(f"/api/messages/{ctx.message.id}/cid/logo@x")
+    assert inline.status_code == 200 and inline.content == LOGO
+    zipped = await ctx.client.get(f"/api/messages/{ctx.message.id}/attachments.zip")
+    archive = zipfile.ZipFile(io.BytesIO(zipped.content))
+    assert archive.read("clip.mp4") == VIDEO and archive.read("notatki.txt") == NOTES
+    thumb = await ctx.client.get(_url(ctx, "photo.png", "/thumb"))
+    assert thumb.status_code == 200 and thumb.headers["content-type"] == "image/webp"
+    # The streamed copy is a short-lived cache, not a second store.
+    assert (_user_dir(ctx.user_id) / "cache" / ctx.by_name["clip.mp4"].blob_sha256).exists()
+    assert not blobstore.exists(ctx.user_id, ctx.by_name["clip.mp4"].blob_sha256)
+
+
+async def test_forwarding_gives_the_files_a_blob_of_their_own(ctx: Ctx) -> None:
+    from app.outbox.attachments import load_attachments
+
+    resp = await ctx.client.post(
+        f"/api/messages/{ctx.message.id}/forward-attachments", headers={"X-CSRF": "t"}
+    )
+    assert resp.status_code == 200
+    ids = [a["id"] for a in resp.json()]
+    assert sorted(a["filename"] for a in resp.json()) == ["clip.mp4", "notatki.txt", "photo.png"]
+    draft = SimpleNamespace(payload={"attachment_ids": ids})  # what a send reads
+    files = await load_attachments(ctx.db, ctx.user_id, draft)  # type: ignore[arg-type]
+    assert {name: data for name, _, data in files} == {
+        "photo.png": PNG,
+        "clip.mp4": VIDEO,
+        "notatki.txt": NOTES,
+    }
+
+
+async def test_old_copies_are_removed_only_when_safe(ctx: Ctx) -> None:
+    photo, notes, video = (ctx.by_name[n] for n in ("photo.png", "notatki.txt", "clip.mp4"))
+    # As stored before: copies of their own, and no part index.
+    old = time.time() - 2 * 3600
+    for att, data in ((photo, PNG), (notes, NOTES), (video, VIDEO)):
+        blobstore.store(ctx.user_id, data)
+        os.utime(blobstore.blob_path(ctx.user_id, att.blob_sha256), (old, old))
+    await ctx.db.execute(
+        update(Attachment).where(Attachment.message_id == ctx.message.id).values(part_index=None)
+    )
+    # A file that is not in its message (only its own copy has it) …
+    stray = blobstore.store(ctx.user_id, b"only stored apart")
+    os.utime(blobstore.blob_path(ctx.user_id, stray), (old, old))
+    ctx.db.add(
+        Attachment(
+            message_id=ctx.message.id,
+            user_id=ctx.user_id,
+            filename="stray.bin",
+            mime_type="application/octet-stream",
+            size_bytes=17,
+            blob_sha256=stray,
+        )
+    )
+    # … and one a draft is about to send (a compose attachment shares the copy).
+    ctx.db.add(
+        ComposeAttachment(
+            user_id=ctx.user_id,
+            filename="notatki.txt",
+            mime_type="text/plain",
+            size_bytes=len(NOTES),
+            blob_sha256=notes.blob_sha256,
+        )
+    )
+    await ctx.db.commit()
+
+    stats = await att_store.dedupe_existing(user_id=ctx.user_id)
+    assert stats["found"] == 4 and stats["not_found"] == 1
+    assert stats["files_removed"] == 2  # the photo's and the video's copies
+    assert not blobstore.exists(ctx.user_id, photo.blob_sha256)
+    assert not blobstore.exists(ctx.user_id, video.blob_sha256)
+    assert blobstore.exists(ctx.user_id, notes.blob_sha256)  # the draft still needs it
+    assert blobstore.exists(ctx.user_id, stray)  # its only copy
+    indexes = dict(
+        (
+            await ctx.db.execute(
+                select(Attachment.filename, Attachment.part_index).where(
+                    Attachment.message_id == ctx.message.id
+                )
+            )
+        ).all()
+    )
+    assert indexes["stray.bin"] == att_store.NOT_IN_MESSAGE
+    assert all(i is not None and i >= 0 for n, i in indexes.items() if n != "stray.bin")
+    # Everything still reads back exactly; a second pass has nothing to do.
+    for name, data in (("photo.png", PNG), ("clip.mp4", VIDEO), ("notatki.txt", NOTES)):
+        assert (await ctx.client.get(_url(ctx, name), params={"download": "true"})).content == data
+    assert (await att_store.dedupe_existing(user_id=ctx.user_id))["rows"] == 0

backend/tests/indexer/test_blob_gc.py

@@ -0,0 +1,175 @@
+"""The nightly clean-up of stored files (app/indexer/gc.py): what nothing uses
+goes, everything still used — or possibly about to be — stays."""
+
+from __future__ import annotations
+
+import datetime as dt
+import os
+import shutil
+import time
+import uuid
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest_asyncio
+from sqlalchemy import delete, select, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+from app.identity.passwords import hash_password
+from app.indexer import blobstore, gc
+from app.models import Account, Attachment, ComposeAttachment, Draft, Mailbox, Message, User
+
+OLD = time.time() - 2 * 3600
+
+
+@dataclass
+class Ctx:
+    db: AsyncSession
+    user_id: uuid.UUID
+    account_id: uuid.UUID
+    mailbox_id: uuid.UUID
+
+
+@pytest_asyncio.fixture
+async def ctx() -> AsyncIterator[Ctx]:
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    db = async_sessionmaker(engine, expire_on_commit=False)()
+    user = User(handle=f"gc{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
+    db.add(user)
+    await db.flush()
+    account = Account(
+        user_id=user.id,
+        email="me@example.com",
+        imap_config={},
+        smtp_config={},
+        secret_ct=b"",
+        discovery_src="manual",
+    )
+    db.add(account)
+    await db.flush()
+    inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
+    db.add(inbox)
+    await db.commit()
+    try:
+        yield Ctx(db, user.id, account.id, inbox.id)
+    finally:
+        await db.execute(delete(User).where(User.id == user.id))
+        await db.commit()
+        await db.close()
+        await engine.dispose()
+        shutil.rmtree(Path(get_settings().blob_root) / str(user.id), ignore_errors=True)
+
+
+def _file(c: Ctx, data: bytes, *, old: bool = True) -> str:
+    sha = blobstore.store(c.user_id, data)
+    if old:
+        os.utime(blobstore.blob_path(c.user_id, sha), (OLD, OLD))
+    return sha
+
+
+async def _message(c: Ctx, sha: str, uid: int) -> Message:
+    m = Message(
+        user_id=c.user_id,
+        account_id=c.account_id,
+        mailbox_id=c.mailbox_id,
+        uid=uid,
+        internal_date=dt.datetime.now(dt.UTC),
+        blob_sha256=sha,
+    )
+    c.db.add(m)
+    await c.db.commit()
+    return m
+
+
+async def test_unused_files_go_and_used_ones_stay(ctx: Ctx) -> None:
+    kept_message = _file(ctx, b"a message still here")
+    await _message(ctx, kept_message, 1)
+    deleted_message = _file(ctx, b"a message deleted forever")
+    fresh = _file(ctx, b"a message being stored right now", old=False)
+    legacy = _file(ctx, b"an attachment kept as its own file")
+    msg = await _message(ctx, _file(ctx, b"its message"), 2)
+    ctx.db.add(
+        Attachment(
+            message_id=msg.id,
+            user_id=ctx.user_id,
+            filename="x.bin",
+            mime_type="application/octet-stream",
+            size_bytes=1,
+            blob_sha256=legacy,
+            part_index=None,  # not looked for in its message yet
+        )
+    )
+    await ctx.db.commit()
+    # Previews: one of an attachment still here, one of a deleted one.
+    thumbs = blobstore.blob_path(ctx.user_id, "00").parent.parent / "thumbs"
+    thumbs.mkdir(parents=True, exist_ok=True)
+    live_thumb, dead_thumb = thumbs / f"{legacy}-480.webp", thumbs / f"{'f' * 64}-480.webp"
+    for t in (live_thumb, dead_thumb):
+        t.write_bytes(b"webp")
+        os.utime(t, (OLD, OLD))
+    # A sent message waiting for its Sent copy: only its job refers to it.
+    waiting = _file(ctx, b"sent, not yet filed in Sent")
+    await gc.hold(ctx.user_id, waiting)
+
+    stats = await gc.sweep(ctx.user_id)
+
+    assert stats["files"] == 1 and stats["thumbs"] == 1
+    assert not blobstore.exists(ctx.user_id, deleted_message)
+    for sha in (kept_message, fresh, legacy, waiting, msg.blob_sha256):
+        assert blobstore.exists(ctx.user_id, sha)
+    assert live_thumb.exists() and not dead_thumb.exists()
+
+    # Filed (or given up): no longer held, so it goes on the next night.
+    await gc.release(ctx.user_id, waiting)
+    assert (await gc.sweep(ctx.user_id))["files"] == 1
+    assert not blobstore.exists(ctx.user_id, waiting)
+
+
+async def test_abandoned_uploads_go_after_a_week(ctx: Ctx) -> None:
+    week_ago = dt.datetime.now(dt.UTC) - dt.timedelta(days=8)
+
+    def upload(data: bytes, created: dt.datetime) -> ComposeAttachment:
+        row = ComposeAttachment(
+            user_id=ctx.user_id,
+            filename="f.txt",
+            mime_type="text/plain",
+            size_bytes=len(data),
+            blob_sha256=_file(ctx, data),
+        )
+        ctx.db.add(row)
+        return row
+
+    abandoned = upload(b"closed without saving", week_ago)
+    in_draft = upload(b"in a draft still being written", week_ago)
+    recent = upload(b"uploaded a minute ago", dt.datetime.now(dt.UTC))
+    await ctx.db.flush()
+    await ctx.db.execute(
+        update(ComposeAttachment)
+        .where(ComposeAttachment.id.in_([abandoned.id, in_draft.id]))
+        .values(created_at=week_ago)
+    )
+    ctx.db.add(
+        Draft(
+            user_id=ctx.user_id,
+            account_id=ctx.account_id,
+            kind="new",
+            payload={"attachment_ids": [str(in_draft.id)]},
+        )
+    )
+    await ctx.db.commit()
+
+    stats = await gc.sweep(ctx.user_id)
+
+    assert stats["uploads"] == 1 and stats["files"] == 1
+    left = set(
+        await ctx.db.scalars(
+            select(ComposeAttachment.id).where(ComposeAttachment.user_id == ctx.user_id)
+        )
+    )
+    assert left == {in_draft.id, recent.id}
+    assert not blobstore.exists(ctx.user_id, abandoned.blob_sha256)
+    assert blobstore.exists(ctx.user_id, in_draft.blob_sha256)
+    assert blobstore.exists(ctx.user_id, recent.blob_sha256)

backend/tests/outbox/test_inline_images.py

@@ -0,0 +1,133 @@
+"""Images quoted from another message go out inside the message, not as links
+to this server; such links already sent load from here when shown."""
+
+from __future__ import annotations
+
+import email
+import io
+import shutil
+import uuid
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime
+from email import policy
+from email.message import EmailMessage
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest_asyncio
+from PIL import Image
+from sqlalchemy import delete
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+from app.identity.passwords import hash_password
+from app.indexer.indexer import persist_message
+from app.indexer.sanitize import sanitize_html
+from app.models import Account, Mailbox, User
+from app.outbox import inline_images
+from app.outbox.mime_build import build_mime
+from app.providers.base import RemoteMessage
+
+
+def _png() -> bytes:
+    out = io.BytesIO()
+    Image.new("RGB", (40, 20), (10, 120, 200)).save(out, "PNG")
+    return out.getvalue()
+
+
+LOGO = _png()
+
+
+@pytest_asyncio.fixture
+async def ctx() -> AsyncIterator[tuple[AsyncSession, uuid.UUID, uuid.UUID]]:
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    db = async_sessionmaker(engine, expire_on_commit=False)()
+    user = User(handle=f"inl{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
+    db.add(user)
+    await db.flush()
+    account = Account(
+        user_id=user.id,
+        email="me@example.com",
+        imap_config={},
+        smtp_config={},
+        secret_ct=b"",
+        discovery_src="manual",
+    )
+    db.add(account)
+    await db.flush()
+    inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox")
+    db.add(inbox)
+    await db.commit()
+    m = EmailMessage()
+    m["Subject"] = "Signed"
+    m["From"] = "Ben <ben@example.org>"
+    m["To"] = "me@example.com"
+    m["Message-ID"] = "<signed@x>"
+    m.set_content("hello")
+    m.add_alternative('<p>hello</p><img src="cid:logo@x">', subtype="html")
+    m.get_payload()[1].add_related(
+        LOGO, "image", "png", cid="<logo@x>", filename="logo.png", disposition="inline"
+    )
+    remote = RemoteMessage(
+        uid=1, raw=m.as_bytes(), flags=frozenset(), internal_date=datetime.now(UTC)
+    )
+    result = await persist_message(
+        db,
+        user_id=user.id,
+        account_id=account.id,
+        mailbox_id=inbox.id,
+        mailbox_special_use="inbox",
+        remote=remote,
+    )
+    await db.commit()
+    assert result is not None
+    try:
+        yield db, user.id, result.message.id
+    finally:
+        await db.execute(delete(User).where(User.id == user.id))
+        await db.commit()
+        await db.close()
+        await engine.dispose()
+        shutil.rmtree(Path(get_settings().blob_root) / str(user.id), ignore_errors=True)
+
+
+async def test_quoted_images_travel_inside_the_reply(ctx) -> None:  # type: ignore[no-untyped-def]
+    db, user_id, original = ctx
+    gone = uuid.uuid4()
+    html = (
+        "<p>Thanks!</p><blockquote>"
+        f'<img src="/api/messages/{original}/cid/logo@x" alt="logo">'
+        f'<img src="http:///api/messages/{original}/cid/logo@x">'  # same image, mangled
+        f'<img src="/api/messages/{gone}/cid/old@x">'  # its message is gone
+        '<img src="https://example.com/banner.png"></blockquote>'
+    )
+    out, parts = await inline_images.embed(db, user_id, html, "example.com")
+    assert len(parts) == 1 and parts[0].data == LOGO
+    assert out.count(f'src="cid:{parts[0].cid}"') == 2
+    assert "/api/messages/" not in out and 'alt="logo"' in out
+    assert "https://example.com/banner.png" in out  # other images untouched
+
+    account = SimpleNamespace(email="me@example.com", display_name="Me")
+    draft = SimpleNamespace(
+        payload={"to": ["ben@example.org"], "subject": "Re: Signed", "html": html}
+    )
+    raw, _, _, _ = build_mime(account, draft, None, [], html=out, inline=parts)  # type: ignore[arg-type]
+    sent = email.message_from_bytes(raw, policy=policy.default)
+    related = [p for p in sent.walk() if p.get_content_type() == "multipart/related"]
+    assert related, "the HTML and its image belong together"
+    image = next(p for p in sent.walk() if p.get_content_type() == "image/png")
+    assert image["Content-ID"] == f"<{parts[0].cid}>"
+    assert image.get_payload(decode=True) == LOGO
+
+
+def test_links_already_sent_load_from_here() -> None:
+    mid = uuid.uuid4()
+    for src in (
+        f"http:///api/messages/{mid}/cid/logo@x",
+        f"https://mail.example.com/api/messages/{mid}/cid/logo@x",
+        f"/api/messages/{mid}/cid/logo@x",
+    ):
+        result = sanitize_html(f'<img src="{src}">', uuid.uuid4(), allow_remote_images=True)
+        assert f'src="/api/messages/{mid}/cid/logo@x"' in result.html, src
+        assert not result.has_remote_images

backend/workers/run.py

@@ -25,14 +25,19 @@ from arq import cron
 from arq.worker import Worker
 
 from app.config import get_settings
+from app.indexer.gc import sweep_job
 from app.queue import QUEUE_BULK, QUEUE_INTERACTIVE, redis_settings
 from app.sync.actions import autopurge_trash_spam
 from app.sync.coordinator import SyncCoordinator
 from workers.worker import BULK_FUNCTIONS, INTERACTIVE_FUNCTIONS
 
 # Core scheduled jobs. Daily Trash/Spam retention purge at 03:17 (off-peak, and a
-# minute that won't collide with typical top-of-hour crons).
-_CORE_CRON_JOBS = [cron(autopurge_trash_spam, hour=3, minute=17)]
+# minute that won't collide with typical top-of-hour crons); then the clean-up of
+# stored files nothing uses any more (the purge's among them) at 03:47.
+_CORE_CRON_JOBS = [
+    cron(autopurge_trash_spam, hour=3, minute=17),
+    cron(sweep_job, hour=3, minute=47, timeout=3600),
+]
 
 logger = logging.getLogger(__name__)
 
@@ -95,6 +100,15 @@ async def _run() -> None:
         await coordinator.start()
     arq_tasks = [asyncio.create_task(w.async_run()) for w in build_arq_workers(role)]
     logger.info("worker starting: role=%s (%d arq worker(s))", role, len(arq_tasks))
+    if arq_tasks:
+        # Attachments stored apart before they were kept once: remove those
+        # copies in the background (quick when there are none; one at a time).
+        from app.queue import enqueue_job
+
+        try:
+            await enqueue_job("dedupe_existing_job", _job_id="dedupe-attachments")
+        except Exception:  # noqa: BLE001 - tried again on the next start
+            logger.warning("could not enqueue the attachment clean-up", exc_info=True)
 
     stop = asyncio.Event()
     loop = asyncio.get_running_loop()

backend/workers/worker.py

@@ -14,6 +14,7 @@ from typing import Any
 
 from arq import func
 
+from app.indexer.attachments import dedupe_existing_job
 from app.outbox.draft_sync import push_draft_to_imap_job, remove_imap_draft_job
 from app.outbox.jobs import file_sent_copy_job, send_job
 from app.push.service import push_new_mail_job
@@ -62,4 +63,8 @@ INTERACTIVE_FUNCTIONS: list[Any] = [
 ]
 
 # Bulk work (§6): long timeout, low concurrency, isolated from interactive jobs.
-BULK_FUNCTIONS = [sync_bootstrap_account]
+BULK_FUNCTIONS = [
+    sync_bootstrap_account,
+    # Resumable (each batch is saved), so a long first run may span timeouts.
+    func(dedupe_existing_job, timeout=6 * 3600),
+]

frontend/public/sw.js

@@ -33,7 +33,7 @@ const DATA = "wm-data-v1";
 const INDEX_KEY = "/__wm/offline-index";
 const APP_KEY = "/__wm/app";
 
-const MAX_THREADS = 80; // the newest 50 Inbox conversations + recently opened ones
+const MAX_THREADS = 130; // the newest 100 Inbox conversations + recently opened ones
 const MAX_ORPHANS = 50; // message parts whose conversation was never saved
 const MAX_LISTS = 40; // thread lists (one per folder / filter / page)
 const MAX_AVATARS = 300;

frontend/src/components/compose/Composer.tsx

@@ -1143,6 +1143,8 @@ export function Composer({
             <RichTextEditor
               ref={editorRef}
               initialHtml={initialHtml}
+              // A new reply: type straight away, above the quoted original.
+              focusOnStart={Boolean(init.quote) && !init.html && init.kind !== "forward"}
               onInput={(html) => {
                 markDirty();
                 setBodyHtml(html);

frontend/src/components/compose/RichTextEditor.tsx

@@ -193,9 +193,19 @@ export const RichTextEditor = forwardRef<
     /** This message's spell-check language from the compose window's More menu:
      *  "off", "auto" or a language code. Unset follows Settings. */
     spellLanguage?: string | null;
+    /** Start with the caret on the first line (a reply: above the quoted original). */
+    focusOnStart?: boolean;
   }
 >(function RichTextEditor(
-  { initialHtml, onInput, onReady, showToolbar = true, minHeight = "min-h-[250px]", spellLanguage: spellOverride },
+  {
+    initialHtml,
+    onInput,
+    onReady,
+    showToolbar = true,
+    minHeight = "min-h-[250px]",
+    spellLanguage: spellOverride,
+    focusOnStart = false,
+  },
   handleRef,
 ) {
   const t = useT();
@@ -247,6 +257,16 @@ export const RichTextEditor = forwardRef<
     el.innerHTML = initialHtml;
     addQuoteToggle(el, t("compose.showQuoted"));
     onReady?.(el);
+    if (focusOnStart) {
+      // The scroll is the composer's own business (it brings itself into view).
+      el.focus({ preventScroll: true });
+      const range = document.createRange();
+      range.setStart(el.firstElementChild ?? el, 0);
+      range.collapse(true);
+      const sel = window.getSelection();
+      sel?.removeAllRanges();
+      sel?.addRange(range);
+    }
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, []);
 

frontend/src/components/thread/MessageBody.tsx

@@ -118,9 +118,13 @@ export function MessageBody({
     // else: relative / cid: / javascript: / data: / tel: — blocked, no navigation.
   }, []);
 
-  const onLoad = () => {
-    const doc = frameRef.current?.contentDocument;
-    if (!doc) return;
+  // Set up the frame (sizing, link clicks, the quote toggle) as soon as its
+  // document is parsed — not at its load event, which waits for every image,
+  // remote ones included: until then the frame would stay at its first height.
+  const preparedRef = useRef<Document | null>(null);
+  const prepare = useCallback((doc: Document) => {
+    if (preparedRef.current === doc) return;
+    preparedRef.current = doc;
     // The frame is auto-sized to its content, so it must never scroll itself — the
     // thread panel is the scroll container. Hiding the document's own vertical
     // overflow removes any phantom scrollbar left by sub-pixel/line-box rounding
@@ -134,14 +138,42 @@ export function MessageBody({
     doc.addEventListener("click", handleFrameClick);
     // Re-measure as images finish loading (they change the content height).
     doc.querySelectorAll("img").forEach((img) => {
-      if (!img.complete) img.addEventListener("load", resize, { once: true });
+      if (!img.complete) {
+        img.addEventListener("load", resize, { once: true });
+        img.addEventListener("error", resize, { once: true });
+      }
     });
     observerRef.current?.disconnect();
     const ro = new ResizeObserver(resize);
     if (doc.body) ro.observe(doc.body);
     observerRef.current = ro;
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, []);
+
+  const onLoad = () => {
+    const doc = frameRef.current?.contentDocument;
+    if (!doc) return;
+    prepare(doc);
+    resize(); // everything has loaded: one last measure
   };
 
+  // Watch for the message's document to be parsed (the frame starts on an
+  // empty about:blank document, which does not count).
+  useEffect(() => {
+    if (data === undefined) return;
+    let frame = 0;
+    const check = () => {
+      const doc = frameRef.current?.contentDocument;
+      if (doc && doc.location.href === "about:srcdoc" && doc.readyState !== "loading" && doc.body) {
+        prepare(doc);
+        return;
+      }
+      frame = requestAnimationFrame(check);
+    };
+    frame = requestAnimationFrame(check);
+    return () => cancelAnimationFrame(frame);
+  }, [data, prepare]);
+
   useEffect(() => () => observerRef.current?.disconnect(), []);
 
   const loadImages = async (persist: boolean) => {

frontend/src/components/thread/ThreadView.tsx

@@ -5,6 +5,7 @@ import { useT } from "../../i18n";
 import { personHue } from "../../lib/avatarColor";
 import { api } from "../../lib/api";
 import { printMessages } from "../../lib/print";
+import { replyTargets } from "../../lib/replyTargets";
 import { useIsMobile } from "../../lib/useIsMobile";
 import { getDateLocale, getHour12 } from "../../lib/timeFormat";
 import type { AccountOut, MailboxOut, MessageEnvelope } from "../../lib/types";
@@ -60,9 +61,12 @@ function attributionFor(msg: MessageEnvelope): string {
   return `On ${date}, at ${time}, ${who} wrote:`;
 }
 
-function buildReplyInit(subject: string, msg: MessageEnvelope, kind: string): ComposerInit {
-  const from = msg.from_addr[0]?.email ?? "";
-  const toEmails = msg.to_addr.map((a) => a.email);
+function buildReplyInit(
+  subject: string,
+  msg: MessageEnvelope,
+  kind: string,
+  own: Set<string>,
+): ComposerInit {
   const norm = subject.replace(/^\s*(re|fwd?):\s*/i, "");
   const quote = { attribution: attributionFor(msg), messageId: msg.id };
   if (kind === "forward") {
@@ -76,8 +80,9 @@ function buildReplyInit(subject: string, msg: MessageEnvelope, kind: string): Co
       forwardFrom: msg.id,
     };
   }
-  const to = kind === "reply_all" ? [from, ...toEmails].filter(Boolean).join(", ") : from;
-  const cc = kind === "reply_all" ? msg.cc_addr.map((a) => a.email).join(", ") : "";
+  const targets = replyTargets(msg, own, kind);
+  const to = targets.to.join(", ");
+  const cc = targets.cc.join(", ");
   return {
     accountId: msg.account_id,
     threadId: msg.thread_id,
@@ -315,6 +320,7 @@ function MessageItem({
   messageLabels,
   onToggleLabel,
   onCreateLabel,
+  ownAddresses,
 }: {
   message: MessageEnvelope;
   expanded: boolean;
@@ -337,6 +343,7 @@ function MessageItem({
   messageLabels: string[];
   onToggleLabel: (label: string) => void;
   onCreateLabel: (name: string) => void;
+  ownAddresses: Set<string>;
 }) {
   const t = useT();
   const [showDetails, setShowDetails] = useState(false);
@@ -367,7 +374,9 @@ function MessageItem({
       {message.from_addr[0]?.name || message.from_addr[0]?.email || "(unknown)"}
     </span>
   );
-  const multiRecipient = message.to_addr.length + message.cc_addr.length > 1;
+  // Reply all only when it would reach more than one person (not counting you).
+  const allTargets = replyTargets(message, ownAddresses, "reply_all");
+  const multiRecipient = allTargets.to.length + allTargets.cc.length > 1;
 
   return (
     <div
@@ -631,6 +640,8 @@ export function ThreadView({
     for (const a of accounts ?? []) m.set(a.email.toLowerCase(), a.color);
     return m;
   }, [accounts]);
+  // Your own addresses (every account's): replies never go back to them.
+  const ownAddresses = useMemo(() => new Set(accountColor.keys()), [accountColor]);
   const accentFor = (msg: MessageEnvelope) =>
     accountColor.get((msg.from_addr[0]?.email ?? "").toLowerCase()) ?? null;
 
@@ -646,7 +657,7 @@ export function ThreadView({
   const openReply = (msg: MessageEnvelope, kind: string) => {
     if (specialUse.get(msg.mailbox_id) === "trash") return; // no reply/forward from Trash
     setExpanded((prev) => new Set(prev).add(msg.id));
-    setComposing({ anchorId: msg.id, init: buildReplyInit(subject, msg, kind) });
+    setComposing({ anchorId: msg.id, init: buildReplyInit(subject, msg, kind, ownAddresses) });
   };
 
   // A mailto: link in the message body opens a fresh composer to that address,
@@ -785,6 +796,7 @@ export function ThreadView({
                 messageLabels={msgLabels[message.id] ?? []}
                 onToggleLabel={(l) => toggleLabel(message.id, l)}
                 onCreateLabel={(name) => void createLabel(name, message.id)}
+                ownAddresses={ownAddresses}
               />
               {isNewest && (
                 <div data-testid="composer-anchor">

frontend/src/i18n/en.ts

@@ -591,7 +591,7 @@ export const en = {
   "offline.cantReach": "Can't reach the server.",
   "offline.settingTitle": "Offline access",
   "offline.settingHelp":
-    "Keep your newest 50 Inbox conversations, and the ones you open, on this device, so you can read them without a connection. They are stored unencrypted in this browser and removed when you sign out — leave this off on shared computers.",
+    "Keep your newest 100 Inbox conversations, and the ones you open, on this device, so you can read them without a connection. They are stored unencrypted in this browser and removed when you sign out — leave this off on shared computers.",
   "offline.settingStatus": "{count} conversations saved · {size} · updated {when}",
   "offline.settingStatusNoTime": "{count} conversations saved · {size}",
   "offline.settingSaving": "Saving recent mail…",
@@ -606,6 +606,8 @@ export const en = {
     "Notifications are blocked for this site in your browser. Allow them in the browser's site settings, then turn this on.",
   "push.unsupported": "Not available here: this browser can't show notifications from a website, or the app isn't opened over HTTPS.",
   "push.notAllowed": "Notifications weren't allowed.",
+  "push.waiting":
+    "Waiting for your browser's permission. If it didn't ask, click the icon at the left of the address bar and set Notifications to Allow — they turn on by themselves then.",
   "push.failed": "Couldn't turn notifications on.",
   "install.entry": "Install app",
   "install.title": "Install Ultimo",

frontend/src/lib/offlinePrefetch.ts

@@ -15,7 +15,8 @@ import type { UiManifest } from "../plugins/types";
 import type { MessageEnvelope, ThreadDetail, ThreadListOut, UserSettings } from "./types";
 
 const INBOX = "folder:inbox";
-const COUNT = 50; // the list's page size, so the first page is saved as shown
+const PAGE = 50; // the list's page size, so its pages are saved as the app asks for them
+const PAGES = 2; // the newest 100 conversations
 const PARALLEL = 3;
 const TIMEOUT_MS = 30_000;
 // The message protocol this page speaks with public/sw.js (its VERSION).
@@ -148,12 +149,17 @@ async function run(): Promise<void> {
   const order = settings.thread_order ?? "asc";
   // The list as the app shows it (newest first is the default), plus the other
   // order when that is what this device uses, so the list itself opens offline.
-  const list = await getJson<ThreadListOut>(threadsPath(INBOX, { limit: COUNT, order: "desc" }));
+  const threads: ThreadListOut["threads"] = [];
+  for (let page = 0; page < PAGES; page++) {
+    const list = await getJson<ThreadListOut>(threadsPath(INBOX, { offset: page * PAGE, limit: PAGE, order: "desc" }));
+    threads.push(...list.threads);
     if (localStorage.getItem("mail.listOrder") === "oldest") {
-    await viaWorker(`/api${threadsPath(INBOX, { limit: COUNT, order: "asc" })}`);
+      await viaWorker(`/api${threadsPath(INBOX, { offset: page * PAGE, limit: PAGE, order: "asc" })}`);
+    }
+    if (list.threads.length < PAGE) break; // no further page
   }
 
-  await inPool(list.threads, PARALLEL, async (summary) => {
+  await inPool(threads, PARALLEL, async (summary) => {
     for (const who of summary.participants) await saveAvatar(who.email);
     const detail = await getJson<ThreadDetail>(threadPath(summary.id, order, INBOX));
     for (const m of detail.messages) await saveMessage(m);

frontend/src/lib/push.ts

@@ -82,11 +82,45 @@ async function tellServer(sub: PushSubscription): Promise<void> {
   await apiFetch("/push/subscriptions", { method: "POST", body: JSON.stringify({ endpoint, keys }) });
 }
 
+/**
+ * Ask for permission to notify. Some browsers (Opera, Chrome's "quiet"
+ * prompts) hold the request without showing anything and never answer it, so
+ * also watch the site's permission: allowing notifications in the browser's
+ * site settings carries on from there. `onWaiting` runs when no answer has come
+ * after a few seconds, to say what to do.
+ */
+function askPermission(onWaiting?: () => void): Promise<NotificationPermission> {
+  if (Notification.permission !== "default") return Promise.resolve(Notification.permission);
+  return new Promise((resolve) => {
+    let watched: PermissionStatus | null = null;
+    let done = false;
+    const finish = (permission: NotificationPermission) => {
+      if (done) return;
+      done = true;
+      window.clearTimeout(timer);
+      if (watched) watched.onchange = null;
+      resolve(permission);
+    };
+    const timer = window.setTimeout(() => onWaiting?.(), 3000);
+    Notification.requestPermission().then(finish, () => finish(Notification.permission));
+    navigator.permissions
+      ?.query({ name: "notifications" })
+      .then((status) => {
+        watched = status;
+        status.onchange = () => {
+          if (status.state === "granted") finish("granted");
+          else if (status.state === "denied") finish("denied");
+        };
+      })
+      .catch(() => {});
+  });
+}
+
 /** Turn notifications on (must run from a click: browsers ask permission only then). */
-export async function enablePush(): Promise<PushState> {
+export async function enablePush(onWaiting?: () => void): Promise<PushState> {
   try {
     if (!supported()) return pushState();
-    const permission = await Notification.requestPermission();
+    const permission = await askPermission(onWaiting);
     if (permission !== "granted") return pushState();
     const reg = await registration();
     if (!reg) return "unsupported";

frontend/src/lib/replyTargets.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, it } from "vitest";
+
+import type { MessageEnvelope } from "./types";
+import { replyTargets } from "./replyTargets";
+
+const me = new Set(["me@work.com", "me@home.org"]);
+const addr = (email: string) => ({ name: "", email });
+const msg = (from: string, to: string[], cc: string[] = []) =>
+  ({ from_addr: [addr(from)], to_addr: to.map(addr), cc_addr: cc.map(addr) }) as unknown as MessageEnvelope;
+
+describe("replyTargets", () => {
+  it("answers the sender of someone else's message", () => {
+    expect(replyTargets(msg("ann@x.com", ["me@work.com"]), me, "reply")).toEqual({ to: ["ann@x.com"], cc: [] });
+  });
+
+  it("answers the people your own message went to, not yourself", () => {
+    expect(replyTargets(msg("me@work.com", ["ann@x.com"], ["bob@x.com"]), me, "reply")).toEqual({
+      to: ["ann@x.com"],
+      cc: [],
+    });
+    expect(replyTargets(msg("ME@work.com", ["ann@x.com"], ["bob@x.com"]), me, "reply_all")).toEqual({
+      to: ["ann@x.com"],
+      cc: ["bob@x.com"],
+    });
+  });
+
+  it("leaves every one of your addresses out of reply all", () => {
+    const m = msg("ann@x.com", ["me@work.com", "bob@x.com"], ["me@home.org", "cy@x.com", "Bob@x.com"]);
+    expect(replyTargets(m, me, "reply_all")).toEqual({ to: ["ann@x.com", "bob@x.com"], cc: ["cy@x.com"] });
+  });
+
+  it("still answers yourself for a note sent only to yourself", () => {
+    expect(replyTargets(msg("me@work.com", ["me@work.com"]), me, "reply")).toEqual({ to: ["me@work.com"], cc: [] });
+  });
+
+  it("uses Cc when your message had only Cc recipients", () => {
+    expect(replyTargets(msg("me@work.com", [], ["bob@x.com"]), me, "reply_all")).toEqual({ to: ["bob@x.com"], cc: [] });
+  });
+});

frontend/src/lib/replyTargets.ts

@@ -0,0 +1,40 @@
+import type { MessageEnvelope } from "./types";
+
+/**
+ * Who a reply goes to. To a message from someone else: its sender (reply), and
+ * everyone on it (reply all). To one of your own messages: the people it was
+ * sent to — not back to yourself. You are never among the recipients of a reply
+ * all, under any of your accounts' addresses; a note sent only to yourself
+ * still answers yourself.
+ */
+export function replyTargets(
+  msg: MessageEnvelope,
+  own: Set<string>,
+  kind: string,
+): { to: string[]; cc: string[] } {
+  const key = (e: string) => e.trim().toLowerCase();
+  const unique = (list: string[], skip: string[] = []) => {
+    const seen = new Set(skip.map(key));
+    return list.filter((e) => {
+      const k = key(e);
+      if (!k || seen.has(k)) return false;
+      seen.add(k);
+      return true;
+    });
+  };
+  const notMe = (e: string) => !own.has(key(e));
+  const from = msg.from_addr[0]?.email ?? "";
+  const toEmails = msg.to_addr.map((a) => a.email);
+  const ccEmails = msg.cc_addr.map((a) => a.email);
+  const mine = own.has(key(from));
+  if (mine) {
+    let to = unique(toEmails.filter(notMe));
+    let cc = kind === "reply_all" ? unique(ccEmails.filter(notMe), to) : [];
+    if (to.length === 0 && cc.length > 0) [to, cc] = [cc, []];
+    if (to.length === 0) to = unique(toEmails.length ? toEmails : [from]); // a note to yourself
+    return { to, cc };
+  }
+  if (kind !== "reply_all") return { to: unique([from]), cc: [] };
+  const to = unique([from, ...toEmails.filter(notMe)]);
+  return { to, cc: unique(ccEmails.filter(notMe), to) };
+}

frontend/src/routes/settings/Settings.tsx

@@ -377,6 +377,7 @@ function NotificationsSetting() {
   const state = usePushState();
   const [busy, setBusy] = useState(false);
   const [error, setError] = useState("");
+  const [waiting, setWaiting] = useState(false);
   const usable = state === "on" || state === "off";
 
   const toggle = async (on: boolean) => {
@@ -384,7 +385,7 @@ function NotificationsSetting() {
     setError("");
     try {
       if (on) {
-        const result = await enablePush();
+        const result = await enablePush(() => setWaiting(true));
         if (result === "off") setError(t("push.notAllowed"));
       } else {
         await disablePush();
@@ -393,6 +394,7 @@ function NotificationsSetting() {
       setError(e instanceof ApiError ? e.message : t("push.failed"));
     } finally {
       setBusy(false);
+      setWaiting(false);
     }
   };
 
@@ -424,6 +426,8 @@ function NotificationsSetting() {
         <p className="text-xs text-app-faint">{t("push.blocked")}</p>
       ) : state === "unsupported" ? (
         <p className="text-xs text-app-faint">{t("push.unsupported")}</p>
+      ) : waiting ? (
+        <p className="text-xs text-amber-600 dark:text-amber-400">{t("push.waiting")}</p>
       ) : error ? (
         <p className="text-xs text-red-500">{error}</p>
       ) : null}

frontend/src/test/setup.ts

@@ -1 +1,14 @@
 import "@testing-library/jest-dom/vitest";
+
+// jsdom has no Web Workers; the PDF preview (pdf.js) sets one up when its module
+// loads. A do-nothing stand-in lets components that import it render in tests.
+if (typeof globalThis.Worker === "undefined") {
+  class TestWorker {
+    onmessage: ((e: MessageEvent) => void) | null = null;
+    postMessage(): void {}
+    terminate(): void {}
+    addEventListener(): void {}
+    removeEventListener(): void {}
+  }
+  (globalThis as unknown as { Worker: unknown }).Worker = TestWorker;
+}

webmail-client-spec_v1.5.md

@@ -877,7 +877,7 @@ If phones/desktop clients may use the mailbox over IMAP/SMTP, a password alone w
 ### D.6 Storing mail once
 Today the indexer keeps every message in full (raw `.eml` blob + sanitized body in `body_parts` + search index), so external accounts are already a second copy, and with Stalwart both copies would sit on the same disk. Ideas, best first:
 
-1. **Attachments are stored twice inside webmail today.** `indexer.py` stores each attachment as its own blob *and* it is inside the raw blob (base64), ≈2.3× its size; attachments are most of mail's bytes. Fix: keep only the raw message, extract an attachment when downloaded, keep thumbnails cached. Roughly halves storage for **all** accounts, independent of Stalwart. A cleanup script frees existing duplicates.
+1. **Attachments are stored twice inside webmail today.** `indexer.py` stores each attachment as its own blob *and* it is inside the raw blob (base64), ≈2.3× its size; attachments are most of mail's bytes. Fix: keep only the raw message, extract an attachment when downloaded, keep thumbnails cached. Cuts attachment storage by about 40% (≈1.37× instead of ≈2.37× the file size) for **all** accounts, independent of Stalwart. A background pass frees existing duplicates. **Done (v1.3.34):** `app/indexer/attachments.py`, migration 0023.
 2. **Native accounts: no raw copy in webmail.** Stalwart is the master copy; webmail keeps metadata, sanitized body, snippet and search index (~5–10% of the size). Full message, attachments and "show original" are fetched on demand over the local network (IMAP `BODY[part]` or JMAP blob download), with a small cache for repeat opens and thumbnails.
 3. **External accounts (optional setting): age out old raw copies.** Keep raw blobs for N days or the newest X GB; refetch older ones over IMAP when needed. Readable bodies stay local; old attachments then need the external server reachable.
 4. **Filesystem:** at least lz4 compression on the ZFS data share (QuTS hero). ZFS dedup would merge identical copies but costs a lot of RAM and only hits uncompressed byte-identical data — last resort.
@@ -902,7 +902,7 @@ Today the indexer keeps every message in full (raw `.eml` blob + sanitized body
 2. The same handle + password signs in to IMAP (when §D.5 allows it); after a webmail password change or reset, the old password fails and the new one works with no other step; a blocked user cannot sign in to IMAP.
 3. The native account cannot be removed via API (409) or UI; deleting the user removes the Stalwart mailbox.
 4. Webmail stores no copy of the account's password (`secret_ct` unused for native accounts).
-5. After D1, an attachment-heavy fixture mailbox uses ≤ 55% of the pre-D1 blob storage, and every attachment still downloads byte-identical.
+5. After D1, an attachment-heavy fixture mailbox uses ≤ 60% of the pre-D1 blob storage, and every attachment still downloads byte-identical.
 6. After D4, a native message's raw blob is absent from `BLOB_ROOT`, yet opening, downloading attachments and "show original" all work; with Stalwart stopped, list and bodies still render from local data.
 7. Backup → restore on a fresh host brings back both webmail and Stalwart mail intact.