Introduce a deprecation reminder service

service/src/main/kotlin/app/cash/backfila/api/ConfigureServiceAction.kt

@@ -68,13 +68,15 @@ class ConfigureServiceAction @Inject constructor(
           request.connector_extra_data,
           request.slack_channel,
           variant,
+          clock.instant(),
         )
         session.save(dbService)
       } else {
         dbService.connector = request.connector_type
         dbService.connector_extra_data = request.connector_extra_data
         dbService.slack_channel = request.slack_channel
         dbService.variant = variant
+        dbService.last_registered_at = clock.instant()
       }
 
       // Add any missing backfills, update modified ones, and mark missing ones as deleted.

service/src/main/kotlin/app/cash/backfila/service/deletion/DeprecationNotificationHelper.kt

@@ -0,0 +1,210 @@
+package app.cash.backfila.service.deletion
+
+import app.cash.backfila.service.listener.SlackHelper
+import app.cash.backfila.service.persistence.BackfilaDb
+import app.cash.backfila.service.persistence.BackfillRunQuery
+import app.cash.backfila.service.persistence.BackfillState
+import app.cash.backfila.service.persistence.DbDeprecationReminder
+import app.cash.backfila.service.persistence.DbRegisteredBackfill
+import app.cash.backfila.service.persistence.DeprecationReminderQuery
+import app.cash.backfila.service.persistence.RegisteredBackfillQuery
+import java.time.Clock
+import java.time.DayOfWeek
+import java.time.Duration
+import java.time.Instant
+import java.time.ZoneOffset
+import javax.inject.Inject
+import javax.inject.Singleton
+import misk.hibernate.Query
+import misk.hibernate.Transacter
+import misk.hibernate.newQuery
+
+@Singleton
+class DeprecationNotificationHelper @Inject constructor(
+  @BackfilaDb private val transacter: Transacter,
+  private val queryFactory: Query.Factory,
+  private val slackHelper: SlackHelper,
+  private val notificationProvider: DeprecationNotificationProvider,
+  private val clock: Clock,
+) {
+  private val daysInMonth = 30L
+
+  fun getRegisteredBackfillsForNotification(): List<DbRegisteredBackfill> {
+    return transacter.transaction { session ->
+      queryFactory.newQuery<RegisteredBackfillQuery>()
+        .active()
+        .list(session)
+    }
+  }
+
+  fun notifyRegisteredBackfill(registeredBackfill: DbRegisteredBackfill): NotificationDecision? {
+    return transacter.transaction { session ->
+      val now = clock.instant()
+      val config = notificationProvider.getNotificationConfig()
+
+      // If service has not been registered for 90 days, skip notification
+      val service = registeredBackfill.service // This should be loaded due to JPA relationship
+      val lastRegisteredAt = service.last_registered_at
+      if (lastRegisteredAt != null &&
+        Duration.between(lastRegisteredAt, now) > Duration.ofDays(daysInMonth * 3)
+      ) {
+        return@transaction null
+      }
+
+      if (!isBusinessHours(registeredBackfill)) {
+        return@transaction null
+      }
+
+      val deleteByDates = mutableListOf<Pair<NotificationDecision, Instant>>()
+
+      // Explicit delete-by date
+      registeredBackfill.delete_by?.let { deleteBy ->
+        deleteByDates.add(NotificationDecision.EXPLICIT_DELETE_BY to deleteBy)
+      }
+
+      // Default creation-based delete-by
+      val defaultDeleteBy = registeredBackfill.created_at
+        .plus(config.defaultDelayDays[NotificationDecision.DEFAULT_CREATION]!!)
+      deleteByDates.add(NotificationDecision.DEFAULT_CREATION to defaultDeleteBy)
+
+      // Get the latest run and its status
+      val latestRun = queryFactory.newQuery<BackfillRunQuery>()
+        .registeredBackfillId(registeredBackfill.id)
+        .orderByUpdatedAtDesc()
+        .apply {
+          maxRows = 1
+        }
+        .list(session)
+        .firstOrNull()
+
+      // Add run-based delete_by dates to deleteByDates
+      latestRun?.let {
+        val runDate = it.created_at
+        when (it.state) {
+          BackfillState.COMPLETE -> deleteByDates.add(
+            NotificationDecision.COMPLETE_RUN to runDate.plus(config.defaultDelayDays[NotificationDecision.COMPLETE_RUN]!!),
+          )
+          BackfillState.PAUSED -> deleteByDates.add(
+            NotificationDecision.PAUSED_RUN to runDate.plus(config.defaultDelayDays[NotificationDecision.PAUSED_RUN]!!),
+          )
+          else -> null
+        }
+      }
+
+      // Find the latest delete-by date and its associated decision
+      val (effectiveDecision, effectiveDeleteBy) = deleteByDates.maxByOrNull { it.second }
+        ?: return@transaction null
+
+      // Don't send notifications before the delete_by date
+      val timeUntilDeletion = Duration.between(now, effectiveDeleteBy)
+      if (!timeUntilDeletion.isNegative) {
+        return@transaction null
+      }
+
+      // We're past the delete_by date, determine notification frequency
+      val timeSinceDeletion = Duration.between(effectiveDeleteBy, now)
+      val notifications = config.notifications[effectiveDecision] ?: emptyList()
+
+      // Find the appropriate notification based on timeSinceDeletion
+      val appropriateNotification = notifications
+        .sortedBy { it.delay } // Sort by delay to get earliest first
+        .findLast { notification ->
+          // Find the last notification whose delay is less than or equal to timeSinceDeletion
+          timeSinceDeletion >= notification.delay
+        } ?: return@transaction null
+
+      // Get the last notification sent
+      val lastReminder = queryFactory.newQuery<DeprecationReminderQuery>()
+        .registeredBackfillId(registeredBackfill.id)
+        .orderByCreatedAtDesc()
+        .apply { maxRows = 1 }
+        .list(session)
+        .firstOrNull()
+
+      // Check if we should send this notification
+      val shouldSendNotification = when {
+        lastReminder == null -> {
+          // No previous reminder, should send
+          true
+        }
+        appropriateNotification.repeated -> {
+          // For repeated notifications, check if enough time has passed since last reminder
+          Duration.between(lastReminder.created_at, now) >= appropriateNotification.delay
+        }
+        else -> {
+          // For non-repeated notifications, check if this specific type hasn't been sent
+          !(lastReminder.message_last_user == appropriateNotification.messageLastUser && !lastReminder.repeated)
+        }
+      }
+
+      if (shouldSendNotification) {
+        val message = generateNotificationMessage(appropriateNotification, registeredBackfill)
+        sendNotification(message, registeredBackfill.service.slack_channel)
+        // Log to deprecation reminder table
+        val reminder = DbDeprecationReminder(
+          registeredBackfill.id,
+          appropriateNotification.messageLastUser, appropriateNotification.repeated,
+          now,
+        )
+        session.save(reminder)
+        return@transaction effectiveDecision
+      }
+
+      return@transaction null
+    }
+  }
+
+  // Separate message generation function that doesn't need DB access
+  private fun generateNotificationMessage(
+    notification: DeprecationMessage,
+    registeredBackfill: DbRegisteredBackfill,
+  ): String {
+    // Use the configured message directly
+    val baseMessage = notification.message
+
+    // Add metadata about the backfill
+    val metadata = buildString {
+      appendLine()
+      appendLine("*Additional Information:*")
+      appendLine("• Backfill: `${registeredBackfill.name}`")
+
+      // Add action items
+      appendLine()
+      appendLine("*Actions Required:*")
+      appendLine("1. Review if this backfill is still needed")
+      appendLine("2. Update the `@DeleteBy` annotation with a new date")
+      appendLine("3. Or run the backfill again to extend its lifetime")
+    }
+
+    return baseMessage + metadata
+  }
+
+  // Update send notification to use the new message generation
+  fun sendNotification(
+    message: String,
+    channel: String?,
+  ) {
+    if (channel == null) {
+      // logger.warn { "No Slack channel specified for notification. Skipping." }
+      return
+    }
+    // Send to Slack
+    slackHelper.sendDeletionNotification(message, channel)
+  }
+
+  private fun isBusinessHours(registeredBackfill: DbRegisteredBackfill): Boolean {
+    val creationTime = registeredBackfill.created_at
+    val currentTime = clock.instant()
+
+    // Avoid weekends in UTC
+    if (currentTime.atZone(ZoneOffset.UTC).dayOfWeek !in listOf(DayOfWeek.SATURDAY, DayOfWeek.SUNDAY)) {
+      return false
+    }
+
+    // Keep the same hour of day as when the backfill was created
+    val creationHour = creationTime.atZone(ZoneOffset.UTC).hour
+    val currentHour = currentTime.atZone(ZoneOffset.UTC).hour
+
+    return creationHour == currentHour
+  }
+}

service/src/main/kotlin/app/cash/backfila/service/deletion/DeprecationNotificationProvider.kt

@@ -0,0 +1,26 @@
+package app.cash.backfila.service.deletion
+
+import java.time.Duration
+
+enum class NotificationDecision {
+  EXPLICIT_DELETE_BY,
+  COMPLETE_RUN,
+  PAUSED_RUN,
+  DEFAULT_CREATION,
+}
+
+data class DeprecationMessage(
+  val delay: Duration,
+  val message: String,
+  val messageLastUser: Boolean = false,
+  val repeated: Boolean = false,
+)
+
+interface DeprecationMessageBuilder {
+  val notifications: Map<NotificationDecision, List<DeprecationMessage>>
+  val defaultDelayDays: Map<NotificationDecision, Duration>
+}
+
+interface DeprecationNotificationProvider {
+  fun getNotificationConfig(): DeprecationMessageBuilder
+}

service/src/main/kotlin/app/cash/backfila/service/deletion/DeprecationNotificationService.kt

@@ -0,0 +1,55 @@
+package app.cash.backfila.service.deletion
+
+import com.google.common.util.concurrent.AbstractExecutionThreadService
+import java.time.Clock
+import java.util.Random
+import javax.inject.Inject
+import javax.inject.Singleton
+import wisp.logging.getLogger
+
+/**
+ * Service that periodically checks for backfills approaching their delete_by date
+ * and sends notifications according to the configured schedule.
+ */
+@Singleton
+class DeprecationNotificationService @Inject constructor(
+  private val notificationHelper: DeprecationNotificationHelper,
+  private val clock: Clock,
+) : AbstractExecutionThreadService() {
+  @Volatile private var running = false
+  private val random = Random()
+
+  override fun startUp() {
+    running = true
+    logger.info { "Starting DeleteByNotificationService" }
+  }
+
+  override fun run() {
+    while (running) {
+      try {
+        checkBackfills()
+      } catch (e: Exception) {
+        logger.error(e) { "Error checking backfills for deletion notifications" }
+      }
+
+      // Sleep for an hour plus random jitter to avoid clustering
+      Thread.sleep(FOUR_HOURS_IN_MILLIS + random.nextInt(JITTER_RANGE_MILLIS))
+    }
+  }
+
+  override fun triggerShutdown() {
+    running = false
+  }
+
+  private fun checkBackfills() {
+    notificationHelper.getRegisteredBackfillsForNotification().forEach { registeredBackfill ->
+      notificationHelper.notifyRegisteredBackfill(registeredBackfill)
+    }
+  }
+
+  companion object {
+    private val logger = getLogger<DeprecationNotificationService>()
+    private const val FOUR_HOURS_IN_MILLIS = 14_400_000L // 4 hours
+    private const val JITTER_RANGE_MILLIS = 300_000 // 5 minutes
+  }
+}

service/src/main/kotlin/app/cash/backfila/service/listener/SlackHelper.kt

@@ -76,4 +76,17 @@ class SlackHelper @Inject constructor(
     val url = "${backfilaConfig.web_url_root}backfills/$id"
     return "<$url|$id>"
   }
+
+  /**
+   * Sends a deletion notification message to a specified Slack channel.
+   * This method is used by the DeleteByNotificationService to send scheduled deletion reminders.
+   */
+  fun sendDeletionNotification(message: String, channel: String) {
+    slackClient.postMessage(
+      username = "Backfila Deletion Reminder",
+      iconEmoji = ":backfila_cleanup:",
+      message = message,
+      channel = channel,
+    )
+  }
 }

service/src/main/kotlin/app/cash/backfila/service/persistence/BackfillRunQuery.kt

@@ -13,6 +13,9 @@ interface BackfillRunQuery : Query<DbBackfillRun> {
   @Constraint("service_id")
   fun serviceId(serviceId: Id<DbService>): BackfillRunQuery
 
+  @Constraint("registered_backfill_id")
+  fun registeredBackfillId(registeredBackfillId: Id<DbRegisteredBackfill>): BackfillRunQuery
+
   @Constraint("service_id", Operator.IN)
   fun serviceIdIn(serviceIds: Collection<Id<DbService>>): BackfillRunQuery
 

service/src/main/kotlin/app/cash/backfila/service/persistence/DbDeprecationReminder.kt

@@ -0,0 +1,30 @@
+package app.cash.backfila.service.persistence
+
+import java.time.Instant
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.GeneratedValue
+import javax.persistence.Table
+import misk.hibernate.DbUnsharded
+import misk.hibernate.Id
+
+@Entity
+@Table(name = "deprecation_reminders")
+class DbDeprecationReminder(
+  @Column(nullable = false)
+  var registered_backfill_id: Id<DbRegisteredBackfill>,
+
+  @Column(nullable = false)
+  var message_last_user: Boolean = false,
+
+  @Column(nullable = false)
+  var repeated: Boolean = false,
+
+  @Column(nullable = false)
+  var created_at: Instant,
+) : DbUnsharded<DbDeprecationReminder> {
+
+  @javax.persistence.Id
+  @GeneratedValue
+  override lateinit var id: Id<DbDeprecationReminder>
+}

service/src/main/kotlin/app/cash/backfila/service/persistence/DbService.kt

@@ -37,17 +37,22 @@ class DbService() : DbUnsharded<DbService>, DbTimestampedEntity {
   @Column
   override lateinit var updated_at: Instant
 
+  @Column(nullable = true)
+  var last_registered_at: Instant? = null
+
   constructor(
     registry_name: String,
     connector: String,
     connector_extra_data: String?,
     slack_channel: String?,
     variant: String,
+    last_registered_at: Instant?,
   ) : this() {
     this.registry_name = registry_name
     this.connector = connector
     this.connector_extra_data = connector_extra_data
     this.slack_channel = slack_channel
     this.variant = variant
+    this.last_registered_at = last_registered_at
   }
 }