refactor(image-cleanup): refactor S3 and DB cleanup functionality

src/main/java/com/magamochi/common/config/RabbitConfig.java

@@ -61,6 +61,12 @@ public class RabbitConfig {
   @Value("${queues.manga-follow-update-chapter}")
   private String mangaFollowUpdateChapterQueue;
 
+  @Value("${queues.image-cleanup-s3-delete}")
+  private String imageCleanupS3DeleteQueue;
+
+  @Value("${queues.image-cleanup-db-delete}")
+  private String imageCleanupDbDeleteQueue;
+
   @Bean
   public TopicExchange imageUpdatesExchange() {
     return new TopicExchange(imageUpdatesTopic);
@@ -269,6 +275,34 @@ public class RabbitConfig {
     return QueueBuilder.durable(mangaFollowUpdateChapterQueue + ".dlq").quorum().build();
   }
 
+  @Bean
+  public Queue imageCleanupS3DeleteQueue() {
+    return QueueBuilder.durable(imageCleanupS3DeleteQueue)
+        .quorum()
+        .deadLetterExchange("")
+        .deadLetterRoutingKey(imageCleanupS3DeleteQueue + ".dlq")
+        .build();
+  }
+
+  @Bean
+  public Queue imageCleanupS3DeleteDlq() {
+    return QueueBuilder.durable(imageCleanupS3DeleteQueue + ".dlq").quorum().build();
+  }
+
+  @Bean
+  public Queue imageCleanupDbDeleteQueue() {
+    return QueueBuilder.durable(imageCleanupDbDeleteQueue)
+        .quorum()
+        .deadLetterExchange("")
+        .deadLetterRoutingKey(imageCleanupDbDeleteQueue + ".dlq")
+        .build();
+  }
+
+  @Bean
+  public Queue imageCleanupDbDeleteDlq() {
+    return QueueBuilder.durable(imageCleanupDbDeleteQueue + ".dlq").quorum().build();
+  }
+
   @Bean
   public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
     return new RabbitAdmin(connectionFactory);

src/main/java/com/magamochi/common/queue/command/ImageCleanupDbDeleteCommand.java

@@ -0,0 +1,6 @@
+package com.magamochi.common.queue.command;
+
+import java.util.Set;
+import java.util.UUID;
+
+public record ImageCleanupDbDeleteCommand(Set<UUID> imageIds) {}

src/main/java/com/magamochi/common/queue/command/ImageCleanupS3DeleteCommand.java

@@ -0,0 +1,5 @@
+package com.magamochi.common.queue.command;
+
+import java.util.Set;
+
+public record ImageCleanupS3DeleteCommand(Set<String> objectKeys) {}

src/main/java/com/magamochi/image/model/repository/ImageRepository.java

@@ -1,10 +1,17 @@
 package com.magamochi.image.model.repository;
 
 import com.magamochi.image.model.entity.Image;
+import java.util.Collection;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 
 public interface ImageRepository extends JpaRepository<Image, UUID> {
   Optional<Image> findByFileHash(String fileHash);
+
+  @Query("SELECT i.objectKey FROM Image i WHERE i.objectKey IN :keys")
+  Set<String> findExistingObjectKeysIn(@Param("keys") Collection<String> keys);
 }

src/main/java/com/magamochi/image/queue/consumer/ImageCleanupDbDeleteConsumer.java

@@ -0,0 +1,35 @@
+package com.magamochi.image.queue.consumer;
+
+import com.magamochi.common.queue.command.ImageCleanupDbDeleteCommand;
+import com.magamochi.image.service.ImageCleanupService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Service;
+
+@Log4j2
+@Service
+@RequiredArgsConstructor
+public class ImageCleanupDbDeleteConsumer {
+  private final ImageCleanupService imageCleanupService;
+
+  @RabbitListener(
+      queues = "${queues.image-cleanup-db-delete}",
+      concurrency = "${queues.image-cleanup-db-delete-concurrency:1-3}")
+  public void receive(ImageCleanupDbDeleteCommand command) {
+    var imageIds = command.imageIds();
+    log.info("Processing DB cleanup batch: {} image IDs", imageIds.size());
+
+    try {
+      imageCleanupService.deleteOrphanedImages(imageIds);
+      log.info("DB cleanup batch completed successfully: {} image IDs", imageIds.size());
+    } catch (Exception e) {
+      log.error(
+          "Failed to process DB cleanup batch ({} image IDs), will be retried. Error: {}",
+          imageIds.size(),
+          e.getMessage(),
+          e);
+      throw e;
+    }
+  }
+}

src/main/java/com/magamochi/image/queue/consumer/ImageCleanupS3DeleteConsumer.java

@@ -0,0 +1,35 @@
+package com.magamochi.image.queue.consumer;
+
+import com.magamochi.common.queue.command.ImageCleanupS3DeleteCommand;
+import com.magamochi.image.service.S3Service;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Service;
+
+@Log4j2
+@Service
+@RequiredArgsConstructor
+public class ImageCleanupS3DeleteConsumer {
+  private final S3Service s3Service;
+
+  @RabbitListener(
+      queues = "${queues.image-cleanup-s3-delete}",
+      concurrency = "${queues.image-cleanup-s3-delete-concurrency:3-5}")
+  public void receive(ImageCleanupS3DeleteCommand command) {
+    var keys = command.objectKeys();
+    log.info("Processing S3 delete batch: {} keys", keys.size());
+
+    try {
+      s3Service.deleteObjects(keys);
+      log.info("S3 delete batch completed successfully: {} keys", keys.size());
+    } catch (Exception e) {
+      log.error(
+          "Failed to delete S3 batch ({} keys), will be retried. Error: {}",
+          keys.size(),
+          e.getMessage(),
+          e);
+      throw e;
+    }
+  }
+}

src/main/java/com/magamochi/image/queue/producer/ImageCleanupDbDeleteProducer.java

@@ -0,0 +1,23 @@
+package com.magamochi.image.queue.producer;
+
+import com.magamochi.common.queue.command.ImageCleanupDbDeleteCommand;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Log4j2
+@Service
+@RequiredArgsConstructor
+public class ImageCleanupDbDeleteProducer {
+  private final RabbitTemplate rabbitTemplate;
+
+  @Value("${queues.image-cleanup-db-delete}")
+  private String queueName;
+
+  public void send(ImageCleanupDbDeleteCommand command) {
+    log.info("Enqueued DB cleanup batch: {} image IDs", command.imageIds().size());
+    rabbitTemplate.convertAndSend(queueName, command);
+  }
+}

src/main/java/com/magamochi/image/queue/producer/ImageCleanupS3DeleteProducer.java

@@ -0,0 +1,23 @@
+package com.magamochi.image.queue.producer;
+
+import com.magamochi.common.queue.command.ImageCleanupS3DeleteCommand;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+@Log4j2
+@Service
+@RequiredArgsConstructor
+public class ImageCleanupS3DeleteProducer {
+  private final RabbitTemplate rabbitTemplate;
+
+  @Value("${queues.image-cleanup-s3-delete}")
+  private String queueName;
+
+  public void send(ImageCleanupS3DeleteCommand command) {
+    log.info("Enqueued S3 delete batch: {} keys", command.objectKeys().size());
+    rabbitTemplate.convertAndSend(queueName, command);
+  }
+}

src/main/java/com/magamochi/image/service/ImageCleanupService.java

@@ -0,0 +1,105 @@
+package com.magamochi.image.service;
+
+import com.magamochi.catalog.model.repository.MangaRepository;
+import com.magamochi.content.model.repository.MangaContentImageRepository;
+import com.magamochi.image.model.repository.ImageRepository;
+import java.util.Set;
+import java.util.UUID;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+@Log4j2
+@Service
+@RequiredArgsConstructor
+public class ImageCleanupService {
+  private final ImageRepository imageRepository;
+  private final MangaRepository mangaRepository;
+  private final MangaContentImageRepository mangaContentImageRepository;
+
+  @Transactional
+  public void deleteOrphanedImages(Set<UUID> imageIds) {
+    log.info(
+        "Starting DB cleanup for {} orphaned images (S3 object no longer exists)", imageIds.size());
+
+    int deletedCount = 0;
+    int skippedCount = 0;
+    int errorCount = 0;
+
+    for (var imageId : imageIds) {
+      try {
+        if (deleteOrphanedImage(imageId)) {
+          deletedCount++;
+        } else {
+          skippedCount++;
+        }
+      } catch (Exception e) {
+        log.error("Failed to delete orphaned image {} from database", imageId, e);
+        errorCount++;
+      }
+    }
+
+    log.info(
+        "DB orphan cleanup finished: {} deleted, {} skipped (already gone), {} errors",
+        deletedCount,
+        skippedCount,
+        errorCount);
+  }
+
+  @Transactional(propagation = Propagation.REQUIRES_NEW)
+  public boolean deleteOrphanedImage(UUID imageId) {
+    var imageOpt = imageRepository.findById(imageId);
+    if (imageOpt.isEmpty()) {
+      log.info("Image {} already deleted, nothing to do", imageId);
+      return false;
+    }
+
+    var image = imageOpt.get();
+    log.info(
+        "Cleaning up orphaned DB record: imageId={}, objectKey={}, fileHash={}",
+        image.getId(),
+        image.getObjectKey(),
+        image.getFileHash());
+
+    var mangasWithCover = mangaRepository.findAllByCoverImage_Id(imageId);
+    if (!mangasWithCover.isEmpty()) {
+      log.info("Nullifying cover image reference from {} manga(s)", mangasWithCover.size());
+      mangasWithCover.forEach(
+          manga -> {
+            manga.setCoverImage(null);
+            mangaRepository.save(manga);
+          });
+    }
+
+    var contentImages = mangaContentImageRepository.findAllByImage_Id(imageId);
+    if (!contentImages.isEmpty()) {
+      log.info("Removing {} manga content image reference(s) and reindexing", contentImages.size());
+      contentImages.forEach(
+          mci -> {
+            var mangaContentId = mci.getMangaContent().getId();
+            mangaContentImageRepository.delete(mci);
+            reindexMangaContent(mangaContentId);
+          });
+    }
+
+    imageRepository.delete(image);
+    log.info(
+        "Orphaned image record deleted: id={}, objectKey={}", image.getId(), image.getObjectKey());
+    return true;
+  }
+
+  private void reindexMangaContent(Long mangaContentId) {
+    var remainingImages =
+        mangaContentImageRepository.findAllByMangaContent_IdOrderByPositionAsc(mangaContentId);
+
+    for (var i = 0; i < remainingImages.size(); i++) {
+      var img = remainingImages.get(i);
+      if (img.getPosition() != i) {
+        img.setPosition(i);
+        mangaContentImageRepository.save(img);
+      }
+    }
+  }
+}

src/main/java/com/magamochi/image/service/S3Service.java

@@ -7,6 +7,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Consumer;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
@@ -42,9 +43,12 @@ public class S3Service {
     return filename;
   }
 
-  public List<String> listAllObjectKeys() {
-    var keys = new ArrayList<String>();
+  public void forEachKeyPage(Consumer<List<String>> pageConsumer) {
+    log.info("Starting S3 key enumeration for bucket: {}", bucket);
+
     String continuationToken = null;
+    int pageCount = 0;
+    int totalKeys = 0;
 
     do {
       var requestBuilder = ListObjectsV2Request.builder().bucket(bucket).maxKeys(1000);
@@ -55,12 +59,25 @@ public class S3Service {
 
       var response = s3Client.listObjectsV2(requestBuilder.build());
 
-      response.contents().forEach(s3Object -> keys.add(s3Object.key()));
+      var keys = response.contents().stream().map(S3Object::key).toList();
+
+      pageCount++;
+      totalKeys += keys.size();
+      log.debug("S3 page {}: {} keys (running total: {})", pageCount, keys.size(), totalKeys);
+
+      pageConsumer.accept(keys);
 
       continuationToken = response.isTruncated() ? response.nextContinuationToken() : null;
 
     } while (nonNull(continuationToken));
 
+    log.info("S3 key enumeration complete: {} total keys across {} pages", totalKeys, pageCount);
+  }
+
+  @Deprecated
+  public List<String> listAllObjectKeys() {
+    var keys = new ArrayList<String>();
+    forEachKeyPage(keys::addAll);
     return keys;
   }
 

src/main/java/com/magamochi/image/task/ImageCleanupTask.java

@@ -1,12 +1,25 @@
 package com.magamochi.image.task;
 
-import com.magamochi.image.model.entity.Image;
-import com.magamochi.image.service.ImageService;
+import com.google.common.hash.BloomFilter;
+import com.google.common.hash.Funnels;
+import com.magamochi.common.queue.command.ImageCleanupDbDeleteCommand;
+import com.magamochi.common.queue.command.ImageCleanupS3DeleteCommand;
+import com.magamochi.image.model.repository.ImageRepository;
+import com.magamochi.image.queue.producer.ImageCleanupDbDeleteProducer;
+import com.magamochi.image.queue.producer.ImageCleanupS3DeleteProducer;
 import com.magamochi.image.service.S3Service;
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -14,41 +27,135 @@ import org.springframework.stereotype.Component;
 @Component
 @RequiredArgsConstructor
 public class ImageCleanupTask {
+  private static final int PAGE_SIZE = 1000;
+  private static final double BLOOM_FILTER_FPP = 0.001;
+  private static final int BLOOM_FILTER_EXPECTED_INSERTIONS = 1_000_000;
+
   @Value("${image-service.clean-up-enabled}")
   private Boolean cleanUpEnabled;
 
   private final S3Service s3Service;
-  private final ImageService imageService;
+  private final ImageRepository imageRepository;
+  private final ImageCleanupS3DeleteProducer s3DeleteProducer;
+  private final ImageCleanupDbDeleteProducer dbDeleteProducer;
+
+  private final AtomicBoolean running = new AtomicBoolean(false);
 
   @Scheduled(cron = "${image-service.cron-expression}")
   public void cleanUpImagesScheduled() {
     if (!cleanUpEnabled) {
-      log.info("S3 Image cleanup disabled.");
+      log.info("Image cleanup disabled (clean-up-enabled=false), skipping.");
       return;
     }
 
+    if (!running.compareAndSet(false, true)) {
+      log.warn("Previous image cleanup run is still in progress, skipping this execution.");
+      return;
+    }
+
+    try {
       cleanupImages();
+    } finally {
+      running.set(false);
+    }
   }
 
   public void cleanupImages() {
-    log.info("Getting unused S3 object keys to remove.");
+    log.info("=== Image cleanup started ===");
 
-    var imageKeys = s3Service.listAllObjectKeys();
+    var bloomFilter =
+        BloomFilter.create(
+            Funnels.stringFunnel(StandardCharsets.UTF_8),
+            BLOOM_FILTER_EXPECTED_INSERTIONS,
+            BLOOM_FILTER_FPP);
 
-    var existingImages =
-        imageService.findAll().parallelStream()
-            .map(Image::getObjectKey)
-            .collect(Collectors.toSet());
+    var s3Pages = new AtomicInteger(0);
+    var totalS3Orphans = new AtomicInteger(0);
+    var s3BatchCount = new AtomicInteger(0);
 
-    var keysToRemove =
-        imageKeys.parallelStream()
-            .filter(imageKey -> !existingImages.contains(imageKey))
-            .collect(Collectors.toSet());
+    log.info("Phase 1 START — scanning S3 for orphaned objects (bucket enumeration + DB lookup)");
 
-    log.info("Removing {} objects from S3 storage", keysToRemove.size());
+    s3Service.forEachKeyPage(
+        page -> {
+          s3Pages.incrementAndGet();
 
-    s3Service.deleteObjects(keysToRemove);
+          var existingKeys = imageRepository.findExistingObjectKeysIn(page);
 
-    log.info("Image cleanup finished.");
+          var orphanKeys =
+              page.stream().filter(key -> !existingKeys.contains(key)).collect(Collectors.toSet());
+
+          if (!orphanKeys.isEmpty()) {
+            s3DeleteProducer.send(new ImageCleanupS3DeleteCommand(orphanKeys));
+            totalS3Orphans.addAndGet(orphanKeys.size());
+            s3BatchCount.incrementAndGet();
+            log.info(
+                "Phase 1 — S3 page {}: {} keys, {} orphans enqueued",
+                s3Pages.get(),
+                page.size(),
+                orphanKeys.size());
+          }
+
+          page.forEach(bloomFilter::put);
+        });
+
+    log.info(
+        "Phase 1 complete: {} S3 pages, {} orphans enqueued across {} batches. BloomFilter FPP={}",
+        s3Pages.get(),
+        totalS3Orphans.get(),
+        s3BatchCount.get(),
+        BLOOM_FILTER_FPP);
+
+    log.info(
+        "Phase 2 START — scanning DB for orphaned records (DB pagination + BloomFilter lookup)");
+
+    int dbPages = 0;
+    int totalDbOrphans = 0;
+    int dbBatchCount = 0;
+    Pageable pageable = PageRequest.of(0, PAGE_SIZE);
+    var imagePage = imageRepository.findAll(pageable);
+
+    while (!imagePage.isEmpty()) {
+      dbPages++;
+      var pageOrphans = new HashSet<UUID>();
+
+      imagePage.forEach(
+          image -> {
+            if (!bloomFilter.mightContain(image.getObjectKey())) {
+              pageOrphans.add(image.getId());
+            }
+          });
+
+      if (!pageOrphans.isEmpty()) {
+        dbDeleteProducer.send(new ImageCleanupDbDeleteCommand(Set.copyOf(pageOrphans)));
+        totalDbOrphans += pageOrphans.size();
+        dbBatchCount++;
+        log.info(
+            "Phase 2 — DB page {}: {} images checked, {} orphans enqueued",
+            dbPages,
+            imagePage.getNumberOfElements(),
+            pageOrphans.size());
+      } else {
+        log.debug(
+            "Phase 2 — DB page {}: {} images checked, no orphans found",
+            dbPages,
+            imagePage.getNumberOfElements());
+      }
+
+      if (!imagePage.hasNext()) {
+        break;
+      }
+      pageable = imagePage.nextPageable();
+      imagePage = imageRepository.findAll(pageable);
+    }
+
+    log.info(
+        "Phase 2 complete: {} DB pages processed, {} orphan DB batches enqueued",
+        dbPages,
+        dbBatchCount);
+
+    log.info(
+        "=== Image cleanup finished. S3 orphans: {}, DB orphans: {} ===",
+        totalS3Orphans.get(),
+        totalDbOrphans);
   }
 }

src/main/resources/application.yml

@@ -76,6 +76,10 @@ queues:
   file-import: ${FILE_IMPORT_QUEUE:mangamochi.file.import}
   manga-content-download: ${MANGA_CONTENT_DOWNLOAD_QUEUE:mangamochi.manga.content.download}
   manga-follow-update-chapter: ${MANGA_FOLLOW_UPDATE_CHAPTER_QUEUE:mangamochi.follow.update}
+  image-cleanup-s3-delete: ${IMAGE_CLEANUP_S3_DELETE_QUEUE:mangamochi.image.cleanup.s3.delete}
+  image-cleanup-db-delete: ${IMAGE_CLEANUP_DB_DELETE_QUEUE:mangamochi.image.cleanup.db.delete}
+  image-cleanup-s3-delete-concurrency: 3-5
+  image-cleanup-db-delete-concurrency: 1-3
 
 routing-key:
   image-update: ${IMAGE_UPDATE_ROUTING_KEY:mangamochi.image.update}

src/main/resources/db/migration/V0017__ADD_INDEX_IMAGES_OBJECT_KEY.sql

@@ -0,0 +1 @@
+CREATE INDEX IF NOT EXISTS idx_images_object_key ON images(object_key);