feat(replays): Let the deletion script select which steps it runs

src/sentry/replays/scripts/delete_replays.py

@@ -46,6 +46,10 @@ def delete_replays(
     tags: list[str],
     start_utc: datetime,
     end_utc: datetime,
+    *,
+    archive: bool = True,
+    delete_seer_data: bool = True,
+    delete_blobs: bool = True,
 ) -> None:
     """Delete a set of replays from a query."""
     search_filters = translate_cli_tags_param_to_snuba_tag_param(tags)
@@ -89,6 +93,9 @@ def delete_replays(
                     "window_end": window_end,
                     "has_more": has_more,
                     "total_replays": total_replays,
+                    "archive": archive,
+                    "delete_seer_data": delete_seer_data,
+                    "delete_blobs": delete_blobs,
                 }
                 if dry_run:
                     logger.info(
@@ -101,6 +108,9 @@ def delete_replays(
                         rows=replays,
                         has_seer_data=has_seer_data,
                         total_replays=total_replays,
+                        archive=archive,
+                        delete_seer_data=delete_seer_data,
+                        delete_blobs=delete_blobs,
                     )
 
 
@@ -114,48 +124,65 @@ def delete_replay_ids(
     rows: list[tuple[int, str, int]],
     has_seer_data: bool,
     total_replays: int,
+    *,
+    archive: bool = True,
+    delete_seer_data: bool = True,
+    delete_blobs: bool = True,
 ) -> None:
     """Delete a set of replay-ids for a specific project."""
     logging_context = {
         "project_id": project_id,
         "num_replays": len(rows),
         "total_replays": total_replays,
+        "archive": archive,
+        "delete_seer_data": delete_seer_data,
+        "delete_blobs": delete_blobs,
     }
-    logger.info("Archiving %d replays.", len(rows), extra=logging_context)
-
-    # Bulk produce archived replay rows to the ingest-replay-events topic before flushing.
-    #
-    # This operation is fast enough that it can be performed synchronously. Archiving
-    # synchronously gives the script runner the feeling that the script executed to completion
-    # and will allow them to immediately spot-check before moving on with their day. In a
-    # purely asynchronous world the script runner will have to continually worry if their tasks
-    # executed.
-    #
-    # This also gives us reasonable assurances that if the script ran to completion the customer
-    # will not be able to access their deleted data even if the actual deletion takes place some
-    # time later
-    for _, replay_id, _ in rows:
-        archive_replay(project_id, replay_id)
-
-    if has_seer_data:
+
+    if archive:
+        logger.info("Archiving %d Replays.", len(rows), extra=logging_context)
+
+        # Bulk produce archived replay rows to the ingest-replay-events topic before flushing.
+        #
+        # This operation is fast enough that it can be performed synchronously. Archiving
+        # synchronously gives the script runner the feeling that the script executed to completion
+        # and will allow them to immediately spot-check before moving on with their day. In a
+        # purely asynchronous world the script runner will have to continually worry if their tasks
+        # executed.
+        #
+        # This also gives us reasonable assurances that if the script ran to completion the customer
+        # will not be able to access their deleted data even if the actual deletion takes place some
+        # time later
+        for _, replay_id, _ in rows:
+            archive_replay(project_id, replay_id)
+    else:
+        # Archiving is the only step the customer can observe. Without it they keep their Replays.
+        logger.warning(
+            "Not archiving %d Replays; they remain visible to the customer.",
+            len(rows),
+            extra=logging_context,
+        )
+
+    if has_seer_data and delete_seer_data:
         # The finder strips dashes from `replay_id`; Seer keys on the dashed UUID
+        logger.info("Deleting Seer data for %d Replays.", len(rows), extra=logging_context)
         replay_ids = [str(uuid.UUID(replay_id)) for _, replay_id, _ in rows]
         delete_seer_replay_data(organization_id, project_id, replay_ids)
 
-    logger.info("Scheduling %d replays for deletion.", len(rows), extra=logging_context)
+    if delete_blobs:
+        logger.info("Scheduling %d Replays for blob deletion.", len(rows), extra=logging_context)
 
-    # Asynchronously delete RRWeb recording data.
-    #
-    # Because this operation could involve millions of requests to the blob storage provider we
-    # schedule the tasks to run on a cluster of workers. This allows us to parallelize the work
-    # and complete the task as quickly as possible.
-    for retention_days, replay_id, max_segment_id in rows:
-        delete_replays_script_async.delay(retention_days, project_id, replay_id, max_segment_id)
+        # Asynchronously delete RRWeb recording data.
+        #
+        # Because this operation could involve millions of requests to the blob storage provider we
+        # schedule the tasks to run on a cluster of workers. This allows us to parallelize the work
+        # and complete the task as quickly as possible.
+        for retention_days, replay_id, max_segment_id in rows:
+            delete_replays_script_async.delay(retention_days, project_id, replay_id, max_segment_id)
 
-    logger.info("%d replays were successfully deleted.", len(rows), extra=logging_context)
     logger.info(
-        "The customer will no longer have access to the replays passed to this function. Deletion "
-        "of RRWeb data will complete asynchronously.",
+        "Finished processing %d Replays.",
+        len(rows),
         extra=logging_context,
     )