fix: a Discord message that re-posts a file is grouped, and joins the drop it repeats

Grouping read a message's images through primary_post_id alone, which only
the first message imported with a file holds. The backfill runs newest-first,
so the original message usually owned nothing: 101 of Yellowroom's messages
(mostly 2018-2020) could never be grouped. Every image they carried also sat
in another message (296 links, measured on the live instance).

_message_images unions ownership with provenance for the candidate query,
the drop seed, the member image links, and the merge pass. Nothing is
re-owned. A later drop carrying the very file an earlier one carries now
merges by a new same_image route, which nearest-neighbour could not see
because it skips the image itself.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/services/discord_grouping.py

@@ -63,7 +63,7 @@ from collections import Counter
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 
-from sqlalchemy import Select, delete, func, select, update
+from sqlalchemy import Select, delete, func, select, union, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -126,6 +126,31 @@ def cosine_distance(a, b) -> float:
     return 1.0 - (dot / (na * nb))
 
 
+def _message_images(posts):
+    """(post_id, image_id) for every image a message carries — owned AND re-posted.
+
+    A message owns an image through `primary_post_id`, but only the FIRST
+    message imported with a given file does. The same file posted again is a
+    provenance link, and the backfill runs newest-first, so it is usually the
+    ORIGINAL message that ends up owning nothing. Reading ownership alone left
+    101 of Yellowroom's messages (2018–2020 mostly) ungroupable: every image
+    they carried also sat in another message.
+
+    `posts` is a list of ids or a select of them; filtering both branches by it
+    keeps the union to the messages in hand rather than the whole library.
+    Callers pass MESSAGE posts only — a drop's own provenance rows would read
+    as images it carries.
+    """
+    owned = select(
+        ImageRecord.primary_post_id.label("post_id"), ImageRecord.id.label("image_id"),
+    ).where(ImageRecord.primary_post_id.in_(posts))
+    reposted = select(
+        ImageProvenance.post_id.label("post_id"),
+        ImageProvenance.image_record_id.label("image_id"),
+    ).where(ImageProvenance.post_id.in_(posts))
+    return union(owned, reposted).subquery()
+
+
 def _candidate_stmt(source_id: int, *, not_after: datetime) -> Select:
     """Ungrouped Discord message-posts, one representative image each, OLDEST
     FIRST — which is the order `build_groups` requires.
@@ -143,13 +168,15 @@ def _candidate_stmt(source_id: int, *, not_after: datetime) -> Select:
     take the OLDEST candidates instead of the lowest-numbered ones.
     """
     sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
+    carried = _message_images(select(Post.id).where(Post.source_id == source_id))
     inner = (
         select(
             Post.id.label("post_id"),
             sort_key.label("occurred_at"),
             ImageRecord.siglip_embedding.label("embedding"),
         )
-        .join(ImageRecord, ImageRecord.primary_post_id == Post.id)
+        .join(carried, carried.c.post_id == Post.id)
+        .join(ImageRecord, ImageRecord.id == carried.c.image_id)
         .where(
             Post.source_id == source_id,
             # Never absorb a post FC wrote, and never re-absorb one already
@@ -295,9 +322,10 @@ async def _link_member_images(
     """
     if not member_ids:
         return 0
-    image_rows = (await session.execute(
+    carried = _message_images(member_ids)
-        select(ImageRecord.id).where(ImageRecord.primary_post_id.in_(member_ids))
+    image_rows = sorted(set((await session.execute(
-    )).scalars().all()
+        select(carried.c.image_id)
+    )).scalars().all()))
     if not image_rows:
         return 0
     await session.execute(
@@ -421,9 +449,11 @@ async def _group_seed(session: AsyncSession, post_id: int) -> list[float] | None
     free to disagree; this way there is one.
     """
     sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
+    carried = _message_images(select(Post.id).where(Post.absorbed_by_post_id == post_id))
     return (await session.execute(
         select(ImageRecord.siglip_embedding)
-        .join(Post, ImageRecord.primary_post_id == Post.id)
+        .join(carried, carried.c.image_id == ImageRecord.id)
+        .join(Post, Post.id == carried.c.post_id)
         .where(
             Post.absorbed_by_post_id == post_id,
             ImageRecord.siglip_embedding.is_not(None),
@@ -725,12 +755,15 @@ async def _load_drops(session: AsyncSession, source: Source) -> list[_Drop]:
     names: dict[int, set[str]] = {pid: set() for pid in by_id}
     images: dict[int, list[int]] = {pid: [] for pid in by_id}
     if owner_of:
-        for iid, primary, path in (await session.execute(
+        carried = _message_images(list(owner_of))
-            select(ImageRecord.id, ImageRecord.primary_post_id, ImageRecord.path)
+        for iid, message, path in (await session.execute(
-            .where(ImageRecord.primary_post_id.in_(list(owner_of)))
+            select(ImageRecord.id, carried.c.post_id, ImageRecord.path)
+            .join(carried, carried.c.image_id == ImageRecord.id)
             .order_by(ImageRecord.id)
         )).all():
-            drop = owner_of[primary]
+            drop = owner_of[message]
+            if iid in images[drop]:
+                continue  # one file carried by two of the drop's messages
             images[drop].append(iid)
             if (name := leading_name(path)) is not None:
                 names[drop].add(name)
@@ -844,7 +877,12 @@ async def merge_trickles(
             if drop.first_at - earlier.last_at > gap:
                 continue
             shared = mine & gated(earlier.names)
-            if shared:
+            if set(drop.images) & set(earlier.images):
+                # The creator posted the very same file again — the strongest
+                # evidence there is, and one nearest-neighbour cannot see: it
+                # skips the image itself, which is the one they share.
+                targets[earlier.post.id] = (earlier, "same_image")
+            elif shared:
                 targets[earlier.post.id] = (earlier, f"name:{min(shared)}")
             elif drop.nearest & earlier.members or (earlier.nearest or set()) & drop.members:
                 targets[earlier.post.id] = (earlier, "nearest")

tests/test_discord_trickles.py

@@ -20,6 +20,7 @@ from sqlalchemy import select
 
 from backend.app.models import (
     Artist,
+    ImageProvenance,
     ImageRecord,
     Post,
     PostAssociation,
@@ -251,3 +252,63 @@ async def test_a_checked_drop_is_not_examined_again(db):
     for drop in await _drops(db, source):
         assert drop.synthesis_details["trickle_checked"] is True
         assert "nearest_message_ids" in drop.synthesis_details
+
+
+# --- the same file posted twice -----------------------------------------------
+# 101 of Yellowroom's messages were never grouped: every image they carried also
+# sat in another message, which the newest-first backfill had made its owner.
+
+
+async def _repost(db, artist, source, *, at, of, text=None):
+    """A message carrying a file another message already owns."""
+    n = next(_n)
+    post = Post(source_id=source.id, artist_id=artist.id,
+                external_post_id=f"msg-{n}", post_date=at, description=text)
+    db.add(post)
+    await db.flush()
+    image_id = (await db.execute(
+        select(ImageRecord.id).where(ImageRecord.primary_post_id == of.id)
+    )).scalar_one()
+    db.add(ImageProvenance(image_record_id=image_id, post_id=post.id, source_id=source.id))
+    await db.flush()
+    return post
+
+
+@pytest.mark.asyncio
+async def test_a_message_that_only_reposts_an_image_is_still_grouped(db):
+    artist, source = await _seed(db, "repost-artist")
+    later = await _message(db, artist, source, at=T0, angle=0.0)
+    original = await _repost(db, artist, source, at=T0 - timedelta(days=400), of=later,
+                             text="first posted here")
+    await db.commit()
+
+    await group_source(db, source, max_distance=0.10, window_minutes=60)
+    await db.commit()
+
+    await db.refresh(original)
+    assert original.absorbed_by_post_id is not None
+    # Grouping links, it never re-owns: the image keeps its primary post.
+    owner = (await db.execute(
+        select(ImageRecord.primary_post_id).join(
+            ImageProvenance, ImageProvenance.image_record_id == ImageRecord.id,
+        ).where(ImageProvenance.post_id == original.absorbed_by_post_id)
+    )).scalar_one()
+    assert owner == later.id
+
+
+@pytest.mark.asyncio
+async def test_the_same_file_posted_a_day_apart_is_one_drop(db):
+    """The live pair: 32554 (Aug 25) and 32553 (Aug 26) carry one file.
+    Nearest-neighbour cannot see it — it skips the image itself."""
+    artist, source = await _seed(db, "twice-artist")
+    second = await _message(db, artist, source, at=T0 + timedelta(days=1), angle=0.0)
+    first = await _repost(db, artist, source, at=T0, of=second)
+    await db.commit()
+
+    await _group_then_merge(db, source)
+
+    (drop,) = await _drops(db, source)
+    assert set(drop.synthesis_details["member_post_ids"]) | {
+        m for r in drop.synthesis_details.get("merged", []) for m in r["message_ids"]
+    } >= {first.id, second.id}
+    assert drop.synthesis_details["merged"][0]["route"] == "same_image"