service: fix draft access on deleted published record

invenio_rdm_records/services/services.py

@@ -17,7 +17,12 @@
 from invenio_db import db
 from invenio_drafts_resources.services.records import RecordService
 from invenio_records_resources.services import LinksTemplate, ServiceSchemaWrapper
-from invenio_records_resources.services.uow import RecordCommitOp, unit_of_work
+from invenio_records_resources.services.uow import (
+    RecordCommitOp,
+    RecordIndexDeleteOp,
+    RecordIndexOp,
+    unit_of_work,
+)
 from invenio_requests.services.results import EntityResolverExpandableField
 from invenio_search.engine import dsl
 from sqlalchemy.exc import NoResultFound
@@ -197,6 +202,13 @@ def delete_record(
         # Commit and reindex record
         uow.register(RecordCommitOp(record, indexer=self.indexer))
 
+        # delete associated draft from index
+        try:
+            draft = self.draft_cls.pid.resolve(id_)
+            uow.register(RecordIndexDeleteOp(draft, indexer=self.draft_indexer))
+        except NoResultFound:
+            pass
+
         # Commit and reindex new latest record
         if new_record_latest_version:
             uow.register(
@@ -289,6 +301,13 @@ def restore_record(self, identity, id_, expand=False, uow=None):
         # Commit and reindex record
         uow.register(RecordCommitOp(record, indexer=self.indexer))
 
+        # reindex associated draft
+        try:
+            draft = self.draft_cls.pid.resolve(id_)
+            uow.register(RecordIndexOp(draft, indexer=self.draft_indexer))
+        except NoResultFound:
+            pass
+
         # commit and reindex the old latest record
         if latest_record_version and record.id != latest_record_version.id:
             uow.register(RecordCommitOp(latest_record_version, indexer=self.indexer))
@@ -516,6 +535,22 @@ def read(self, identity, id_, expand=False, include_deleted=False):
 
         return result
 
+    def read_draft(self, identity, id_, expand=False):
+        """Retrieve a draft of a record.
+
+        If the draft has a "deleted" published record then we return 410.
+        """
+        result = super().read_draft(identity, id_, expand=expand)
+        # check that if there is a published deleted record then return 410
+        draft = result._record
+        if draft.is_published:
+            record = self.record_cls.pid.resolve(id_)
+            if record.deletion_status.is_deleted:
+                result = super().read(identity, id_, expand=expand)
+                raise RecordDeletedException(record, result_item=result)
+
+        return result
+
     #
     # Record file quota handling
     #