Generalize "is idle?" check in idle device notification scheduler

service/src/main/java/org/whispersystems/textsecuregcm/push/IdleDeviceNotificationScheduler.java

@@ -26,10 +26,7 @@ public class IdleDeviceNotificationScheduler extends JobScheduler {
   private final Clock clock;
 
   @VisibleForTesting
-  static final Duration MIN_IDLE_DURATION = Duration.ofDays(14);
-
-  @VisibleForTesting
-  record AccountAndDeviceIdentifier(UUID accountIdentifier, byte deviceId) {}
+  record JobDescriptor(UUID accountIdentifier, byte deviceId, long lastSeen) {}
 
   public IdleDeviceNotificationScheduler(final AccountsManager accountsManager,
       final PushNotificationManager pushNotificationManager,
@@ -52,24 +49,24 @@ public String getSchedulerName() {
 
   @Override
   protected CompletableFuture<String> processJob(@Nullable final byte[] jobData) {
-    final AccountAndDeviceIdentifier accountAndDeviceIdentifier;
+    final JobDescriptor jobDescriptor;
 
     try {
-      accountAndDeviceIdentifier = SystemMapper.jsonMapper().readValue(jobData, AccountAndDeviceIdentifier.class);
+      jobDescriptor = SystemMapper.jsonMapper().readValue(jobData, JobDescriptor.class);
     } catch (final IOException e) {
       return CompletableFuture.failedFuture(e);
     }
 
-    return accountsManager.getByAccountIdentifierAsync(accountAndDeviceIdentifier.accountIdentifier())
+    return accountsManager.getByAccountIdentifierAsync(jobDescriptor.accountIdentifier())
         .thenCompose(maybeAccount -> maybeAccount.map(account ->
-                account.getDevice(accountAndDeviceIdentifier.deviceId()).map(device -> {
-                      if (!isIdle(device)) {
+                account.getDevice(jobDescriptor.deviceId()).map(device -> {
+                      if (jobDescriptor.lastSeen() != device.getLastSeen()) {
                         return CompletableFuture.completedFuture("deviceSeenRecently");
                       }
 
                       try {
                         return pushNotificationManager
-                            .sendNewMessageNotification(account, accountAndDeviceIdentifier.deviceId(), true)
+                            .sendNewMessageNotification(account, jobDescriptor.deviceId(), true)
                             .thenApply(ignored -> "sent");
                       } catch (final NotPushRegisteredException e) {
                         return CompletableFuture.completedFuture("deviceTokenDeleted");
@@ -79,18 +76,12 @@ protected CompletableFuture<String> processJob(@Nullable final byte[] jobData) {
             .orElse(CompletableFuture.completedFuture("accountDeleted")));
   }
 
-  public boolean isIdle(final Device device) {
-    final Duration idleDuration = Duration.between(Instant.ofEpochMilli(device.getLastSeen()), clock.instant());
-
-    return idleDuration.compareTo(MIN_IDLE_DURATION) >= 0;
-  }
-
-  public CompletableFuture<Void> scheduleNotification(final Account account, final byte deviceId, final LocalTime preferredDeliveryTime) {
+  public CompletableFuture<Void> scheduleNotification(final Account account, final Device device, final LocalTime preferredDeliveryTime) {
     final Instant runAt = SchedulingUtil.getNextRecommendedNotificationTime(account, preferredDeliveryTime, clock);
 
     try {
       return scheduleJob(runAt, SystemMapper.jsonMapper().writeValueAsBytes(
-          new AccountAndDeviceIdentifier(account.getIdentifier(IdentityType.ACI), deviceId)));
+          new JobDescriptor(account.getIdentifier(IdentityType.ACI), device.getId(), device.getLastSeen())));
     } catch (final JsonProcessingException e) {
       // This should never happen when serializing an `AccountAndDeviceIdentifier`
       throw new AssertionError(e);

service/src/main/java/org/whispersystems/textsecuregcm/workers/NotifyIdleDevicesWithoutMessagesCommand.java

@@ -18,6 +18,8 @@
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuples;
 import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
 import java.time.LocalTime;
 
 public class NotifyIdleDevicesWithoutMessagesCommand extends AbstractSinglePassCrawlAccountsCommand {
@@ -33,6 +35,12 @@ public class NotifyIdleDevicesWithoutMessagesCommand extends AbstractSinglePassC
   @VisibleForTesting
   static final LocalTime PREFERRED_NOTIFICATION_TIME = LocalTime.of(14, 0);
 
+  @VisibleForTesting
+  static final Duration MIN_IDLE_DURATION = Duration.ofDays(15);
+
+  @VisibleForTesting
+  static final Duration MAX_IDLE_DURATION = Duration.ofDays(30);
+
   private static final Counter DEVICE_INSPECTED_COUNTER =
       Metrics.counter(MetricsUtil.name(StartPushNotificationExperimentCommand.class, "deviceInspected"));
 
@@ -72,19 +80,20 @@ protected void crawlAccounts(final Flux<Account> accounts) {
 
     final MessagesManager messagesManager = getCommandDependencies().messagesManager();
     final IdleDeviceNotificationScheduler idleDeviceNotificationScheduler = buildIdleDeviceNotificationScheduler();
+    final Clock clock = getClock();
 
     accounts
         .flatMap(account -> Flux.fromIterable(account.getDevices()).map(device -> Tuples.of(account, device)))
         .doOnNext(ignored -> DEVICE_INSPECTED_COUNTER.increment())
-        .flatMap(accountAndDevice -> isDeviceEligible(accountAndDevice.getT1(), accountAndDevice.getT2(), idleDeviceNotificationScheduler, messagesManager)
+        .flatMap(accountAndDevice -> isDeviceEligible(accountAndDevice.getT1(), accountAndDevice.getT2(), messagesManager, clock)
             .mapNotNull(eligible -> eligible ? accountAndDevice : null), maxConcurrency)
         .flatMap(accountAndDevice -> {
           final Account account = accountAndDevice.getT1();
           final Device device = accountAndDevice.getT2();
 
           final Mono<Void> scheduleNotificationMono = dryRun
               ? Mono.empty()
-              : Mono.fromFuture(() -> idleDeviceNotificationScheduler.scheduleNotification(account, device.getId(), PREFERRED_NOTIFICATION_TIME))
+              : Mono.fromFuture(() -> idleDeviceNotificationScheduler.scheduleNotification(account, device, PREFERRED_NOTIFICATION_TIME))
                   .onErrorResume(throwable -> {
                     log.warn("Failed to schedule notification for {}:{}",
                         account.getIdentifier(IdentityType.ACI),
@@ -103,6 +112,11 @@ protected void crawlAccounts(final Flux<Account> accounts) {
         .block();
   }
 
+  @VisibleForTesting
+  protected Clock getClock() {
+    return Clock.systemUTC();
+  }
+
   @VisibleForTesting
   protected IdleDeviceNotificationScheduler buildIdleDeviceNotificationScheduler() {
     final DynamoDbTables.TableWithExpiration tableConfiguration = getConfiguration().getDynamoDbTables().getScheduledJobs();
@@ -119,21 +133,28 @@ protected IdleDeviceNotificationScheduler buildIdleDeviceNotificationScheduler()
   @VisibleForTesting
   static Mono<Boolean> isDeviceEligible(final Account account,
       final Device device,
-      final IdleDeviceNotificationScheduler idleDeviceNotificationScheduler,
-      final MessagesManager messagesManager) {
+      final MessagesManager messagesManager,
+      final Clock clock) {
 
     if (!hasPushToken(device)) {
       return Mono.just(false);
     }
 
-    if (!idleDeviceNotificationScheduler.isIdle(device)) {
+    if (!isIdle(device, clock)) {
       return Mono.just(false);
     }
 
     return Mono.fromFuture(messagesManager.mayHavePersistedMessages(account.getIdentifier(IdentityType.ACI), device))
         .map(mayHavePersistedMessages -> !mayHavePersistedMessages);
   }
 
+  @VisibleForTesting
+  static boolean isIdle(final Device device, final Clock clock) {
+    final Duration idleDuration = Duration.between(Instant.ofEpochMilli(device.getLastSeen()), clock.instant());
+
+    return idleDuration.compareTo(MIN_IDLE_DURATION) >= 0 && idleDuration.compareTo(MAX_IDLE_DURATION) < 0;
+  }
+
   @VisibleForTesting
   static boolean hasPushToken(final Device device) {
     // Exclude VOIP tokens since they have their own, distinct delivery mechanism

service/src/test/java/org/whispersystems/textsecuregcm/push/IdleDeviceNotificationSchedulerTest.java

@@ -1,8 +1,6 @@
 package org.whispersystems.textsecuregcm.push;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyByte;
@@ -19,7 +17,6 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -57,14 +54,14 @@ void setUp() {
   void processJob(final boolean accountPresent,
       final boolean devicePresent,
       final boolean tokenPresent,
-      final Instant deviceLastSeen,
+      final boolean lastSeenChanged,
       final String expectedOutcome) throws JsonProcessingException, NotPushRegisteredException {
 
     final UUID accountIdentifier = UUID.randomUUID();
     final byte deviceId = Device.PRIMARY_ID;
 
     final Device device = mock(Device.class);
-    when(device.getLastSeen()).thenReturn(deviceLastSeen.toEpochMilli());
+    when(device.getLastSeen()).thenReturn(0L);
 
     final Account account = mock(Account.class);
     when(account.getDevice(deviceId)).thenReturn(devicePresent ? Optional.of(device) : Optional.empty());
@@ -82,50 +79,27 @@ void processJob(final boolean accountPresent,
     }
 
     final byte[] jobData = SystemMapper.jsonMapper().writeValueAsBytes(
-        new IdleDeviceNotificationScheduler.AccountAndDeviceIdentifier(accountIdentifier, deviceId));
+        new IdleDeviceNotificationScheduler.JobDescriptor(accountIdentifier, deviceId, lastSeenChanged ? 1 : 0));
 
     assertEquals(expectedOutcome, idleDeviceNotificationScheduler.processJob(jobData).join());
   }
 
   private static List<Arguments> processJob() {
-    final Instant idleDeviceLastSeenTimestamp = CURRENT_TIME
-        .minus(IdleDeviceNotificationScheduler.MIN_IDLE_DURATION)
-        .minus(Duration.ofDays(1));
-
     return List.of(
         // Account present, device present, device has tokens, device is idle
-        Arguments.of(true, true, true, idleDeviceLastSeenTimestamp, "sent"),
+        Arguments.of(true, true, true, false, "sent"),
 
         // Account present, device present, device has tokens, but device is active
-        Arguments.of(true, true, true, CURRENT_TIME, "deviceSeenRecently"),
+        Arguments.of(true, true, true, true, "deviceSeenRecently"),
 
         // Account present, device present, device is idle, but missing tokens
-        Arguments.of(true, true, false, idleDeviceLastSeenTimestamp, "deviceTokenDeleted"),
+        Arguments.of(true, true, false, false, "deviceTokenDeleted"),
 
         // Account present, but device missing
-        Arguments.of(true, false, true, idleDeviceLastSeenTimestamp, "deviceDeleted"),
+        Arguments.of(true, false, true, false, "deviceDeleted"),
 
         // Account missing
-        Arguments.of(false, true, true, idleDeviceLastSeenTimestamp, "accountDeleted")
+        Arguments.of(false, true, true, false, "accountDeleted")
     );
   }
-
-  @Test
-  void isIdle() {
-    {
-      final Device idleDevice = mock(Device.class);
-      when(idleDevice.getLastSeen())
-          .thenReturn(CURRENT_TIME.minus(IdleDeviceNotificationScheduler.MIN_IDLE_DURATION).minus(Duration.ofDays(1))
-              .toEpochMilli());
-
-      assertTrue(idleDeviceNotificationScheduler.isIdle(idleDevice));
-    }
-
-    {
-      final Device activeDevice = mock(Device.class);
-      when(activeDevice.getLastSeen()).thenReturn(CURRENT_TIME.toEpochMilli());
-
-      assertFalse(idleDeviceNotificationScheduler.isIdle(activeDevice));
-    }
-  }
 }