fix(cleanup): clear an artist's attachments before the cascade delete (#3066)

`delete_artist_cascade` could abort partway through, and it aborted after
the irreversible half. Deleting an artist CASCADEs to Post (post.artist_id
is ondelete=CASCADE), which SET NULLs post_attachment.post_id — and
`uq_post_attachment_null_post_sha` is a partial UNIQUE on sha256 ALONE
WHERE post_id IS NULL. So any two of that artist's attachments sharing a
sha collapse onto one another and raise.

That shape is ordinary, not corrupt: `_capture_attachment` deliberately
writes one row per post over a single sha-addressed blob, so a creator who
attaches the same pdf to two posts already has two such rows. A
pre-existing filesystem-import row (post_id NULL) with the same sha
collides on its own.

The images and their on-disk files are deleted and committed in 500-row
batches BEFORE the artist row is touched, so the failure landed after
them: images gone, artist and posts alive, files unrecoverable.

Fix: delete the artist's post_attachment rows explicitly first, matched by
artist_id OR by the owning post's artist (artist_id is nullable, so neither
arm alone covers every row). `_repoint_post_links` already guards the
identical collision class in the reconcile path; this is its artist-cascade
counterpart. Migration 0043 reasoned only about upgrade-time safety and
never about this later SET NULL.

The sha-addressed blobs are deliberately left on disk: one blob backs many
rows, so unlinking needs a refcount pass, and this Tier-C op must not
delete bytes its own preview never disclosed.

Adds `attachments_deleted` to the summary, and two regression tests — the
same sha on two posts, and an unrelated NULL-post row that must survive.

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

backend/app/services/cleanup_service.py

@@ -277,6 +277,10 @@ def delete_artist_cascade(
     series_page / tag_suggestion_rejection from ImageRecord delete,
     and source / post / download_event / etc. from Artist delete
     (via Artist.sources cascade="all, delete-orphan").
+
+    The artist's post_attachment rows are cleared EXPLICITLY before the
+    artist row goes — see the comment at that step; leaving them to the
+    cascade aborts the whole delete on a unique violation.
     """
     artist = session.get(Artist, artist_id)
     if artist is None:
@@ -287,6 +291,7 @@ def delete_artist_cascade(
                 "files_deleted": 0,
                 "thumbs_deleted": 0,
                 "import_tasks_nulled": 0,
+                "attachments_deleted": 0,
                 "files_failed": 0,
             },
         }
@@ -323,6 +328,41 @@ def delete_artist_cascade(
     # source_path_prefix matching that's out of scope here.
     import_tasks_nulled = 0
 
+    # Clear the artist's attachments BEFORE the artist row, or the delete below
+    # aborts. Deleting an artist CASCADEs to Post (post.artist_id is
+    # ondelete=CASCADE), which SET NULLs post_attachment.post_id — and
+    # `uq_post_attachment_null_post_sha` is a partial UNIQUE on sha256 ALONE
+    # WHERE post_id IS NULL, so any two of this artist's attachments sharing a
+    # sha collapse onto one another and raise. That is an ORDINARY shape, not a
+    # corrupt one: _capture_attachment deliberately writes one row per post over
+    # a single sha-addressed blob (a creator who attaches the same pdf to two
+    # posts has two rows), and a pre-existing filesystem-import row with the same
+    # sha and a NULL post_id collides on its own. Migration 0043 reasoned only
+    # about upgrade-time safety and never about this later SET NULL.
+    # _repoint_post_links guards the identical collision class in the reconcile
+    # path; this is its artist-cascade counterpart.
+    #
+    # Matched by artist_id OR by the owning post's artist: artist_id is nullable
+    # and _capture_attachment leaves it NULL when no artist resolved, so neither
+    # predicate alone covers every row this cascade is about to strand.
+    #
+    # The sha-addressed BLOBS are deliberately left on disk. One blob backs many
+    # rows (attachment_store.store is sha-addressed + idempotent), so unlinking
+    # needs a refcount pass over the surviving rows — that belongs to the
+    # attachment-reclamation sweep, not here, and this Tier-C op must not delete
+    # bytes its own preview never disclosed.
+    attachments_deleted = session.execute(
+        delete(PostAttachment).where(
+            or_(
+                PostAttachment.artist_id == artist.id,
+                PostAttachment.post_id.in_(
+                    select(Post.id).where(Post.artist_id == artist.id)
+                ),
+            )
+        )
+    ).rowcount or 0
+    session.commit()
+
     session.delete(artist)
     session.commit()
 
@@ -333,6 +373,7 @@ def delete_artist_cascade(
             "files_deleted": files_deleted,
             "thumbs_deleted": thumbs_deleted,
             "import_tasks_nulled": import_tasks_nulled,
+            "attachments_deleted": attachments_deleted,
             "files_failed": files_failed,
         },
     }

tests/test_cleanup_service.py

@@ -294,6 +294,88 @@ def test_delete_artist_cascade_idempotent_on_missing(db_sync, tmp_path):
     assert result["summary"]["images_deleted"] == 0
 
 
+def test_delete_artist_cascade_survives_same_sha_on_two_posts(db_sync, tmp_path):
+    """Same file attached to two of the artist's posts must not abort the delete.
+
+    Left to the cascade this raises: artist delete CASCADEs to Post, which SET
+    NULLs post_attachment.post_id, and `uq_post_attachment_null_post_sha`
+    (sha256 alone, WHERE post_id IS NULL) then rejects the second row. That's an
+    ordinary shape — _capture_attachment writes one row per post over one
+    sha-addressed blob by design. Also covers the NULL-artist_id arm of the
+    delete predicate: the second row has no artist_id, only a post that does.
+    """
+    a = _make_artist(db_sync, slug="casatt")
+    p1 = Post(artist_id=a.id, external_post_id="att-p1")
+    p2 = Post(artist_id=a.id, external_post_id="att-p2")
+    db_sync.add_all([p1, p2])
+    db_sync.flush()
+
+    shared_sha = "ca5a".ljust(64, "0")
+    db_sync.add(PostAttachment(
+        post_id=p1.id, artist_id=a.id, sha256=shared_sha,
+        path="/store/ca5a/bundle.zip", original_filename="bundle.zip",
+        ext=".zip", size_bytes=7,
+    ))
+    db_sync.add(PostAttachment(
+        post_id=p2.id, artist_id=None, sha256=shared_sha,
+        path="/store/ca5a/bundle.zip", original_filename="bundle.zip",
+        ext=".zip", size_bytes=7,
+    ))
+    db_sync.commit()
+    artist_id = a.id
+
+    result = cleanup_service.delete_artist_cascade(
+        db_sync, artist_id=artist_id, images_root=tmp_path,
+    )
+
+    assert result["summary"]["attachments_deleted"] == 2
+    assert db_sync.execute(
+        select(func.count(Artist.id)).where(Artist.id == artist_id)
+    ).scalar_one() == 0
+    assert db_sync.execute(
+        select(func.count(PostAttachment.id))
+        .where(PostAttachment.sha256 == shared_sha)
+    ).scalar_one() == 0
+
+
+def test_delete_artist_cascade_keeps_unrelated_null_post_attachment(
+    db_sync, tmp_path,
+):
+    """A filesystem-import row (post_id NULL) sharing the sha is the other way
+    this collides — and it must SURVIVE: it belongs to no artist, so the
+    cascade has no claim on it."""
+    a = _make_artist(db_sync, slug="casorph")
+    p = Post(artist_id=a.id, external_post_id="orph-p1")
+    db_sync.add(p)
+    db_sync.flush()
+
+    sha = "0rfa".ljust(64, "0")
+    standalone = PostAttachment(
+        post_id=None, artist_id=None, sha256=sha,
+        path="/store/0rfa/manual.pdf", original_filename="manual.pdf",
+        ext=".pdf", size_bytes=3,
+    )
+    db_sync.add(standalone)
+    db_sync.add(PostAttachment(
+        post_id=p.id, artist_id=a.id, sha256=sha,
+        path="/store/0rfa/manual.pdf", original_filename="manual.pdf",
+        ext=".pdf", size_bytes=3,
+    ))
+    db_sync.commit()
+    artist_id, standalone_id = a.id, standalone.id
+
+    result = cleanup_service.delete_artist_cascade(
+        db_sync, artist_id=artist_id, images_root=tmp_path,
+    )
+
+    assert result["summary"]["attachments_deleted"] == 1
+    surviving = db_sync.execute(
+        select(PostAttachment.id, PostAttachment.post_id)
+        .where(PostAttachment.sha256 == sha)
+    ).all()
+    assert surviving == [(standalone_id, None)]
+
+
 # --- delete_images --------------------------------------------------