fix(digest): honor per-user cron schedules on the digest tick

Product claimed “your own schedule” while SlackDailyDigestTaskConfig only
ran the global slack.daily-digest.cron. Switch to a minute tick that
evaluates each enabled UserDigestPreference.cronExpression in that user’s
timezone (Spring CronExpression), skips already-logged fire windows, and
keeps legacy singleton broadcast when no enabled prefs exist.

Author: cursoragent

backend/src/main/java/com/dbaagent/model/UserDigestPreference.java

@@ -103,7 +103,7 @@ public class UserDigestPreference {
 
     /**
      * Optional timezone for schedule interpretation.
-     * When null, uses system default (typically UTC).
+     * When null or invalid, the digest tick evaluates the cron in UTC.
      * Format: IANA timezone ID (e.g., "America/New_York", "Europe/London").
      */
     @Column(name = "timezone", length = 64)

backend/src/main/java/com/dbaagent/repository/SlackDigestLogRepository.java

@@ -97,4 +97,23 @@ SELECT DISTINCT ON (recipient_username) *
      * Used for determining the window start for new digests.
      */
     Optional<SlackDigestLog> findTopByConnectionIdOrderBySentAtDesc(String connectionId);
+
+    /**
+     * Idempotency for per-preference scheduling: true if this preference already
+     * produced a digest log for the connection at or after the cron fire time.
+     */
+    boolean existsByPreferenceIdAndConnectionIdAndSentAtGreaterThanEqual(
+        Long preferenceId,
+        String connectionId,
+        LocalDateTime sentAt
+    );
+
+    /**
+     * Fallback idempotency when preferenceId is missing on older rows.
+     */
+    boolean existsByConnectionIdAndRecipientUsernameAndSentAtGreaterThanEqual(
+        String connectionId,
+        String recipientUsername,
+        LocalDateTime sentAt
+    );
 }

backend/src/main/java/com/dbaagent/service/SlackDailyDigestService.java

@@ -46,6 +46,7 @@
 import com.dbaagent.repository.SlackChannelBindingRepository;
 import com.dbaagent.repository.SlackDigestLogRepository;
 import com.dbaagent.repository.TableStatsHistoryRepository;
+import com.dbaagent.service.digest.DigestCronMatcher;
 import com.dbaagent.service.digest.DigestInsightAssemblerService;
 import com.slack.api.Slack;
 import com.slack.api.methods.MethodsClient;
@@ -60,6 +61,9 @@
 
 import java.io.IOException;
 import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.Instant;
+import java.time.Duration;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -89,6 +93,14 @@ public class SlackDailyDigestService {
     @Value("${slack.digest.admins-only:true}")
     private boolean digestAdminsOnly;
 
+    /**
+     * Global / legacy digest schedule. Also the default when a preference leaves
+     * {@code cronExpression} blank. Interpreted in UTC for legacy; per-user prefs
+     * use their own timezone.
+     */
+    @Value("${slack.daily-digest.cron:0 0 9 * * *}")
+    private String globalDigestCron;
+
     private final SlackRuntimeSettingsService slackRuntimeSettingsService;
     private final SlackChannelBindingRepository channelBindingRepository;
     private final CredentialService credentialService;
@@ -828,9 +840,149 @@ private LocalDateTime getLastDigestTime(String connectionId) {
             .orElse(LocalDateTime.now().minusHours(24));
     }
 
+    private static final Duration DIGEST_TICK_LOOKBACK = Duration.ofMinutes(1);
+
+    /**
+     * Minute-tick entry point used by {@code SlackDailyDigestTaskConfig}.
+     *
+     * <p>When no enabled preferences exist, runs the legacy channel broadcast
+     * only if the global {@code slack.daily-digest.cron} is due (UTC).
+     * When preferences exist, delivers only to preferences whose cron matches
+     * in that user's timezone and that have not already been logged for this
+     * fire window.
+     */
+    public void processDigestTick() {
+        processDigestTick(Instant.now());
+    }
+
+    /**
+     * Testable overload of {@link #processDigestTick()}.
+     */
+    public void processDigestTick(Instant now) {
+        if (now == null) {
+            now = Instant.now();
+        }
+
+        long enabledCount = userDigestPreferenceRepository.countByEnabledTrue();
+        if (enabledCount == 0) {
+            if (DigestCronMatcher.isDue(globalDigestCron, ZoneId.of("UTC"), now, DIGEST_TICK_LOOKBACK)) {
+                log.info("Digest tick: no enabled preferences; running legacy broadcast (global cron due)");
+                runLegacyBroadcastForAllConnections();
+            } else {
+                log.debug("Digest tick: no enabled preferences; global cron not due");
+            }
+            return;
+        }
+
+        String globalCron = (globalDigestCron == null || globalDigestCron.isBlank())
+            ? "0 0 9 * * *"
+            : globalDigestCron;
+
+        List<UserDigestPreference> enabledPrefs = userDigestPreferenceRepository
+            .findByEnabledTrueAndDeliveryMethod(DigestDeliveryMethod.SLACK_DM);
+
+        int dueCount = 0;
+        int delivered = 0;
+        for (UserDigestPreference pref : enabledPrefs) {
+            String cron = pref.getEffectiveCronExpression(globalCron);
+            ZoneId zone = DigestCronMatcher.resolveZone(pref.getTimezone());
+            var window = DigestCronMatcher.dueWindowStart(cron, zone, now, DIGEST_TICK_LOOKBACK);
+            if (window.isEmpty()) {
+                continue;
+            }
+            dueCount++;
+            Instant fireInstant = window.get();
+            List<String> connectionIds = resolvePreferenceConnections(pref);
+            for (String connectionId : connectionIds) {
+                if (alreadyDeliveredForWindow(pref, connectionId, fireInstant)) {
+                    log.debug("Skipping digest for {} / {} — already delivered this window",
+                        pref.getUsername(), connectionId);
+                    continue;
+                }
+                try {
+                    boolean slackEnabled = isSlackDeliveryEnabled();
+                    String connName = connectionName(connectionId);
+                    LocalDateTime since = getLastDigestTime(connectionId);
+                    PersonalizedDigestResult result = sendPersonalizedDigestToUser(
+                        connectionId, connName, pref, since, slackEnabled);
+                    if (result.generated() || result.sent()) {
+                        delivered++;
+                    }
+                } catch (Exception e) {
+                    log.error("Failed due-preference digest for user {} connection {}: {}",
+                        pref.getUsername(), connectionId, e.getMessage(), e);
+                }
+            }
+        }
+
+        log.info("Digest tick (per-user): {} due preference(s), {} delivery attempt(s)",
+            dueCount, delivered);
+
+        // Connections with no recipients still get legacy broadcast when the global cron fires.
+        if (DigestCronMatcher.isDue(globalCron, ZoneId.of("UTC"), now, DIGEST_TICK_LOOKBACK)) {
+            for (String connectionId : digestConnectionIds()) {
+                if (userDigestPreferenceRepository.findEnabledForConnection(connectionId).isEmpty()) {
+                    try {
+                        sendLegacyDigest(connectionId);
+                    } catch (Exception e) {
+                        log.error("Legacy fallback digest failed for {}: {}", connectionId, e.getMessage(), e);
+                    }
+                }
+            }
+        }
+    }
+
+    private void runLegacyBroadcastForAllConnections() {
+        List<String> connectionIds = digestConnectionIds();
+        if (connectionIds.isEmpty()) {
+            log.info("No connections available — skipping legacy digest");
+            return;
+        }
+        for (String connectionId : connectionIds) {
+            try {
+                sendLegacyDigest(connectionId);
+            } catch (Exception e) {
+                log.error("Failed legacy digest for connection {}: {}", connectionId, e.getMessage(), e);
+            }
+        }
+    }
+
+    private List<String> resolvePreferenceConnections(UserDigestPreference pref) {
+        if (pref.getConnectionId() != null && !pref.getConnectionId().isBlank()) {
+            return List.of(pref.getConnectionId());
+        }
+        return digestConnectionIds();
+    }
+
+    /**
+     * True when SlackDigestLog already has a row for this preference/connection
+     * at or after the cron fire time (idempotent minute-tick).
+     */
+    private boolean alreadyDeliveredForWindow(
+            UserDigestPreference pref,
+            String connectionId,
+            Instant fireInstant) {
+        // sentAt is LocalDateTime.now() (JVM default zone) — compare in that zone.
+        LocalDateTime since = LocalDateTime.ofInstant(fireInstant, ZoneId.systemDefault());
+        if (pref.getId() != null) {
+            if (digestLogRepository.existsByPreferenceIdAndConnectionIdAndSentAtGreaterThanEqual(
+                    pref.getId(), connectionId, since)) {
+                return true;
+            }
+        }
+        if (pref.getUsername() != null && !pref.getUsername().isBlank()) {
+            return digestLogRepository.existsByConnectionIdAndRecipientUsernameAndSentAtGreaterThanEqual(
+                connectionId, pref.getUsername(), since);
+        }
+        return false;
+    }
+
     /**
      * Entry point for the hybrid digest run: per-user when preferences exist,
      * legacy broadcast otherwise.
+     *
+     * <p>Used by manual/admin triggers. The scheduled tick uses
+     * {@link #processDigestTick()} so per-user crons are honored.
      */
     public void sendDailyDigestHybrid() {
         List<String> connectionIds = digestConnectionIds();

backend/src/main/java/com/dbaagent/service/digest/DigestCronMatcher.java

@@ -0,0 +1,84 @@
+package com.dbaagent.service.digest;
+
+import org.springframework.scheduling.support.CronExpression;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.Optional;
+
+/**
+ * Helpers for deciding whether a Spring 6-field cron expression is due
+ * within a recent lookback window, interpreted in a given timezone.
+ *
+ * <p>Used by the digest minute-tick scheduler so each
+ * {@code UserDigestPreference.cronExpression} fires in that user's timezone
+ * without registering a separate db-scheduler task per preference.
+ */
+public final class DigestCronMatcher {
+
+    private DigestCronMatcher() {}
+
+    /**
+     * Resolve an IANA timezone id; blank/invalid values fall back to UTC.
+     */
+    public static ZoneId resolveZone(String timezone) {
+        if (timezone == null || timezone.isBlank()) {
+            return ZoneId.of("UTC");
+        }
+        try {
+            return ZoneId.of(timezone.trim());
+        } catch (Exception e) {
+            return ZoneId.of("UTC");
+        }
+    }
+
+    /**
+     * Returns the fire instant for the cron if it falls in {@code (now - lookback, now]},
+     * interpreted in {@code zone}. Empty when not due or the cron is invalid.
+     *
+     * @param cronExpression Spring 6-field cron (sec min hour dom month dow)
+     * @param zone           timezone used to evaluate the cron
+     * @param now            current instant (typically clock.instant())
+     * @param lookback       how far back to search for a matching fire (e.g. 1 minute for a minute tick)
+     */
+    public static Optional<Instant> dueWindowStart(
+            String cronExpression,
+            ZoneId zone,
+            Instant now,
+            Duration lookback) {
+        if (cronExpression == null || cronExpression.isBlank() || zone == null || now == null) {
+            return Optional.empty();
+        }
+        if (lookback == null || lookback.isNegative() || lookback.isZero()) {
+            lookback = Duration.ofMinutes(1);
+        }
+
+        final CronExpression cron;
+        try {
+            cron = CronExpression.parse(cronExpression.trim());
+        } catch (IllegalArgumentException ex) {
+            return Optional.empty();
+        }
+
+        ZonedDateTime zonedNow = now.atZone(zone);
+        ZonedDateTime from = zonedNow.minus(lookback);
+        ZonedDateTime next = cron.next(from);
+        if (next != null && !next.isAfter(zonedNow)) {
+            return Optional.of(next.toInstant());
+        }
+        return Optional.empty();
+    }
+
+    /**
+     * Convenience: true when {@link #dueWindowStart} is present.
+     */
+    public static boolean isDue(
+            String cronExpression,
+            ZoneId zone,
+            Instant now,
+            Duration lookback) {
+        return dueWindowStart(cronExpression, zone, now, lookback).isPresent();
+    }
+}

backend/src/main/java/com/dbaagent/service/scheduler/SlackDailyDigestTaskConfig.java

@@ -11,18 +11,20 @@
 import org.springframework.context.annotation.Profile;
 
 /**
- * Scheduled task configuration for daily digest delivery.
+ * Scheduled task configuration for digest delivery.
  *
- * <p>Uses the hybrid mode: per-user personalized delivery when UserDigestPreference
- * rows exist; legacy channel-broadcast when none exist.
- *
- * <h3>PR3 Changes</h3>
+ * <p>Ticks frequently (default: every minute) and delegates to
+ * {@link SlackDailyDigestService#processDigestTick()} which:
  * <ul>
- *   <li>Now calls {@code sendDailyDigestHybrid()} which routes to per-user or legacy
- *       based on whether any preferences are configured</li>
- *   <li>Two users with different personas on the same connection get different digests</li>
- *   <li>Fallback to legacy broadcast if no per-user preferences exist for a connection</li>
+ *   <li>Honors each enabled {@code UserDigestPreference.cronExpression} in that
+ *       user's timezone when preferences exist</li>
+ *   <li>Falls back to the legacy singleton broadcast gated by
+ *       {@code slack.daily-digest.cron} when no enabled preferences exist</li>
  * </ul>
+ *
+ * <p>The global property {@code slack.daily-digest.cron} remains the default
+ * schedule for preferences that leave {@code cronExpression} blank, and the
+ * legacy broadcast schedule when the system is still in singleton mode.
  */
 @Configuration
 @Profile("!test")
@@ -32,12 +34,11 @@ public class SlackDailyDigestTaskConfig {
     @Bean
     Task<Void> slackDailyDigestTask(
             SlackDailyDigestService service,
-            @Value("${slack.daily-digest.cron:0 0 9 * * *}") String cron) {
-        return Tasks.recurring("slack-daily-digest", Schedules.cron(cron))
+            @Value("${slack.daily-digest.tick-cron:0 * * * * *}") String tickCron) {
+        return Tasks.recurring("slack-daily-digest", Schedules.cron(tickCron))
             .execute((inst, ctx) -> {
-                log.info("Starting scheduled daily digest delivery");
-                service.sendDailyDigestHybrid();
-                log.info("Completed scheduled daily digest delivery");
+                log.debug("Digest scheduler tick");
+                service.processDigestTick();
             });
     }
 }

backend/src/main/resources/application.properties

@@ -321,6 +321,9 @@ slack.bot-token=${SLACK_BOT_TOKEN:}
 slack.signing-secret=${SLACK_SIGNING_SECRET:}
 slack.deepsql-bot-username=${SLACK_DEEPSQL_BOT_USERNAME:}
 slack.daily-digest.cron=${SLACK_DAILY_DIGEST_CRON:0 0 9 * * *}
+# Minute tick that evaluates per-user UserDigestPreference.cronExpression (+ timezone).
+# Legacy/global cron above gates singleton broadcast when no enabled prefs exist.
+slack.daily-digest.tick-cron=${SLACK_DAILY_DIGEST_TICK_CRON:0 * * * * *}
 
 # Brain understanding defaults
 brain.profile.max-columns-per-table=40