Refactor scheduled APNs notifications in preparation for future development

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -136,14 +136,14 @@ import org.whispersystems.textsecuregcm.metrics.MicrometerRegistryManager;
 import org.whispersystems.textsecuregcm.metrics.NetworkReceivedGauge;
 import org.whispersystems.textsecuregcm.metrics.NetworkSentGauge;
 import org.whispersystems.textsecuregcm.metrics.OperatingSystemMemoryGauge;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
 import org.whispersystems.textsecuregcm.metrics.ReportedMessageMetricsListener;
 import org.whispersystems.textsecuregcm.metrics.TrafficSource;
 import org.whispersystems.textsecuregcm.providers.MultiRecipientMessageProvider;
 import org.whispersystems.textsecuregcm.providers.RedisClientFactory;
 import org.whispersystems.textsecuregcm.providers.RedisClusterHealthCheck;
 import org.whispersystems.textsecuregcm.push.APNSender;
-import org.whispersystems.textsecuregcm.push.ApnFallbackManager;
+import org.whispersystems.textsecuregcm.push.ApnPushNotificationScheduler;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.push.FcmSender;
 import org.whispersystems.textsecuregcm.push.MessageSender;
@@ -443,7 +443,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     MessagesCache              messagesCache              = new MessagesCache(messagesCluster, messagesCluster, keyspaceNotificationDispatchExecutor);
     PushLatencyManager         pushLatencyManager         = new PushLatencyManager(metricsCluster, dynamicConfigurationManager);
     ReportMessageManager       reportMessageManager       = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster, config.getReportMessageConfiguration().getCounterTtl());
-    MessagesManager            messagesManager            = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager, reportMessageManager);
+    MessagesManager            messagesManager            = new MessagesManager(messagesDynamoDb, messagesCache, reportMessageManager);
     DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
         deletedAccountsLockDynamoDbClient, config.getDynamoDbTables().getDeletedAccountsLock().getTableName());
     AccountsManager accountsManager = new AccountsManager(accounts, phoneNumberIdentifiers, cacheCluster,
@@ -454,8 +454,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     PubSubManager              pubSubManager              = new PubSubManager(pubsubClient, dispatchManager);
     APNSender                  apnSender                  = new APNSender(apnSenderExecutor, config.getApnConfiguration());
     FcmSender                  fcmSender                  = new FcmSender(fcmSenderExecutor, config.getFcmConfiguration().credentials());
-    ApnFallbackManager         apnFallbackManager         = new ApnFallbackManager(pushSchedulerCluster, apnSender, accountsManager);
+    ApnPushNotificationScheduler apnPushNotificationScheduler = new ApnPushNotificationScheduler(pushSchedulerCluster, apnSender, accountsManager);
-    PushNotificationManager    pushNotificationManager    = new PushNotificationManager(accountsManager, apnSender, fcmSender, apnFallbackManager);
+    PushNotificationManager    pushNotificationManager    = new PushNotificationManager(accountsManager, apnSender, fcmSender, apnPushNotificationScheduler, pushLatencyManager);
     RateLimiters               rateLimiters               = new RateLimiters(config.getLimitsConfiguration(), rateLimitersCluster);
     DynamicRateLimiters        dynamicRateLimiters        = new DynamicRateLimiters(rateLimitersCluster, dynamicConfigurationManager);
     ProvisioningManager        provisioningManager        = new ProvisioningManager(pubSubManager);
@@ -567,7 +567,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     CurrencyConversionManager currencyManager = new CurrencyConversionManager(fixerClient, ftxClient, config.getPaymentsServiceConfiguration().getPaymentCurrencies());
 
     environment.lifecycle().manage(apnSender);
-    environment.lifecycle().manage(apnFallbackManager);
+    environment.lifecycle().manage(apnPushNotificationScheduler);
     environment.lifecycle().manage(pubSubManager);
     environment.lifecycle().manage(accountDatabaseCrawler);
     environment.lifecycle().manage(directoryReconciliationAccountDatabaseCrawler);
@@ -625,7 +625,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         config.getWebSocketConfiguration(), 90000);
     webSocketEnvironment.setAuthenticator(new WebSocketAccountAuthenticator(accountAuthenticator));
     webSocketEnvironment.setConnectListener(
-        new AuthenticatedConnectListener(receiptSender, messagesManager, pushNotificationManager, apnFallbackManager,
+        new AuthenticatedConnectListener(receiptSender, messagesManager, pushNotificationManager,
             clientPresenceManager, websocketScheduledExecutor));
     webSocketEnvironment.jersey().register(new WebsocketRefreshApplicationEventListener(accountsManager, clientPresenceManager));
     webSocketEnvironment.jersey().register(new ContentLengthFilter(TrafficSource.WEBSOCKET));
@@ -652,8 +652,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new DirectoryV2Controller(directoryV2CredentialsGenerator),
         new DonationController(clock, zkReceiptOperations, redeemedReceiptsManager, accountsManager, config.getBadges(),
             ReceiptCredentialPresentation::new, stripeExecutor, config.getDonationConfiguration(), config.getStripe()),
-        new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, deletedAccountsManager,
+        new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, deletedAccountsManager, messagesManager, pushNotificationManager, reportMessageManager, multiRecipientMessageExecutor),
-            messagesManager, apnFallbackManager, reportMessageManager, multiRecipientMessageExecutor),
         new PaymentsController(currencyManager, paymentsCredentialsGenerator),
         new ProfileController(clock, rateLimiters, accountsManager, profilesManager, dynamicConfigurationManager, profileBadgeConverter, config.getBadges(), cdnS3Client, profileCdnPolicyGenerator, profileCdnPolicySigner, config.getCdnConfiguration().getBucket(), zkProfileOperations, batchIdentityCheckExecutor),
         new ProvisioningController(rateLimiters, provisioningManager),

service/src/main/java/org/whispersystems/textsecuregcm/controllers/MessageController.java

@@ -77,11 +77,10 @@ import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.metrics.MessageMetrics;
 import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
 import org.whispersystems.textsecuregcm.providers.MultiRecipientMessageProvider;
-import org.whispersystems.textsecuregcm.push.ApnFallbackManager;
 import org.whispersystems.textsecuregcm.push.MessageSender;
 import org.whispersystems.textsecuregcm.push.NotPushRegisteredException;
+import org.whispersystems.textsecuregcm.push.PushNotificationManager;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
-import org.whispersystems.textsecuregcm.redis.RedisOperation;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.DeletedAccountsManager;
@@ -107,7 +106,7 @@ public class MessageController {
   private final AccountsManager accountsManager;
   private final DeletedAccountsManager deletedAccountsManager;
   private final MessagesManager messagesManager;
-  private final ApnFallbackManager apnFallbackManager;
+  private final PushNotificationManager pushNotificationManager;
   private final ReportMessageManager reportMessageManager;
   private final ExecutorService multiRecipientMessageExecutor;
 
@@ -138,7 +137,7 @@ public class MessageController {
       AccountsManager accountsManager,
       DeletedAccountsManager deletedAccountsManager,
       MessagesManager messagesManager,
-      ApnFallbackManager apnFallbackManager,
+      PushNotificationManager pushNotificationManager,
       ReportMessageManager reportMessageManager,
       @Nonnull ExecutorService multiRecipientMessageExecutor) {
     this.rateLimiters = rateLimiters;
@@ -147,7 +146,7 @@ public class MessageController {
     this.accountsManager = accountsManager;
     this.deletedAccountsManager = deletedAccountsManager;
     this.messagesManager = messagesManager;
-    this.apnFallbackManager = apnFallbackManager;
+    this.pushNotificationManager = pushNotificationManager;
     this.reportMessageManager = reportMessageManager;
     this.multiRecipientMessageExecutor = Objects.requireNonNull(multiRecipientMessageExecutor);
   }
@@ -408,18 +407,14 @@ public class MessageController {
   @Produces(MediaType.APPLICATION_JSON)
   public OutgoingMessageEntityList getPendingMessages(@Auth AuthenticatedAccount auth,
       @HeaderParam("User-Agent") String userAgent) {
-    assert auth.getAuthenticatedDevice() != null;
 
-    if (!Util.isEmpty(auth.getAuthenticatedDevice().getApnId())) {
+    pushNotificationManager.handleMessagesRetrieved(auth.getAccount(), auth.getAuthenticatedDevice(), userAgent);
-      RedisOperation.unchecked(() -> apnFallbackManager.cancel(auth.getAccount(), auth.getAuthenticatedDevice()));
-    }
 
     final OutgoingMessageEntityList outgoingMessages;
     {
       final Pair<List<Envelope>, Boolean> messagesAndHasMore = messagesManager.getMessagesForDevice(
           auth.getAccount().getUuid(),
           auth.getAuthenticatedDevice().getId(),
-          userAgent,
           false);
 
       outgoingMessages = new OutgoingMessageEntityList(messagesAndHasMore.first().stream()

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

@@ -5,26 +5,12 @@
 
 package org.whispersystems.textsecuregcm.push;
 
-import com.codahale.metrics.Meter;
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.RatioGauge;
-import com.codahale.metrics.SharedMetricRegistries;
 import com.google.common.annotations.VisibleForTesting;
 import io.dropwizard.lifecycle.Managed;
 import io.lettuce.core.ScriptOutputType;
 import io.lettuce.core.cluster.SlotHash;
-import org.slf4j.Logger;
+import io.micrometer.core.instrument.Counter;
-import org.slf4j.LoggerFactory;
+import io.micrometer.core.instrument.Metrics;
-import org.whispersystems.textsecuregcm.redis.ClusterLuaScript;
-import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
-import org.whispersystems.textsecuregcm.storage.Account;
-import org.whispersystems.textsecuregcm.storage.AccountsManager;
-import org.whispersystems.textsecuregcm.storage.Device;
-import org.whispersystems.textsecuregcm.util.Constants;
-import org.whispersystems.textsecuregcm.util.Pair;
-import org.whispersystems.textsecuregcm.util.RedisClusterUtil;
-import org.whispersystems.textsecuregcm.util.Util;
-
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
@@ -32,33 +18,40 @@ import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.redis.ClusterLuaScript;
+import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.Device;
+import org.whispersystems.textsecuregcm.util.Pair;
+import org.whispersystems.textsecuregcm.util.RedisClusterUtil;
+import org.whispersystems.textsecuregcm.util.Util;
 
-import static com.codahale.metrics.MetricRegistry.name;
+import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
 
-public class ApnFallbackManager implements Managed {
+public class ApnPushNotificationScheduler implements Managed {
 
-  private static final Logger logger = LoggerFactory.getLogger(ApnFallbackManager.class);
+  private static final Logger logger = LoggerFactory.getLogger(ApnPushNotificationScheduler.class);
 
   private static final String PENDING_NOTIFICATIONS_KEY = "PENDING_APN";
-          static final String NEXT_SLOT_TO_PERSIST_KEY  = "pending_notification_next_slot";
 
-  private static final MetricRegistry metricRegistry        = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  @VisibleForTesting
-  private static final Meter          delivered             = metricRegistry.meter(name(ApnFallbackManager.class, "voip_delivered"));
+  static final String NEXT_SLOT_TO_PERSIST_KEY  = "pending_notification_next_slot";
-  private static final Meter          sent                  = metricRegistry.meter(name(ApnFallbackManager.class, "voip_sent"     ));
-  private static final Meter          retry                 = metricRegistry.meter(name(ApnFallbackManager.class, "voip_retry"));
-  private static final Meter          evicted               = metricRegistry.meter(name(ApnFallbackManager.class, "voip_evicted"));
 
-  static {
+  private static final Counter delivered = Metrics.counter(name(ApnPushNotificationScheduler.class, "voip_delivered"));
-    metricRegistry.register(name(ApnFallbackManager.class, "voip_ratio"), new VoipRatioGauge(delivered, sent));
+  private static final Counter sent = Metrics.counter(name(ApnPushNotificationScheduler.class, "voip_sent"));
-  }
+  private static final Counter retry = Metrics.counter(name(ApnPushNotificationScheduler.class, "voip_retry"));
+  private static final Counter evicted = Metrics.counter(name(ApnPushNotificationScheduler.class, "voip_evicted"));
 
-  private final APNSender                 apnSender;
+  private final APNSender apnSender;
-  private final AccountsManager           accountsManager;
+  private final AccountsManager accountsManager;
-  private final FaultTolerantRedisCluster cluster;
+  private final FaultTolerantRedisCluster pushSchedulingCluster;
 
-  private final ClusterLuaScript getScript;
+  private final ClusterLuaScript getPendingVoipDestinationsScript;
-  private final ClusterLuaScript insertScript;
+  private final ClusterLuaScript insertPendingVoipDestinationScript;
-  private final ClusterLuaScript removeScript;
+  private final ClusterLuaScript removePendingVoipDestinationScript;
 
   private final Thread[] workerThreads = new Thread[WORKER_THREAD_COUNT];
 
@@ -90,7 +83,7 @@ public class ApnFallbackManager implements Managed {
       long entriesProcessed = 0;
 
       do {
-        pendingDestinations = getPendingDestinations(slot, 100);
+        pendingDestinations = getPendingDestinationsForRecurringVoipNotifications(slot, 100);
         entriesProcessed += pendingDestinations.size();
 
         for (final String uuidAndDevice : pendingDestinations) {
@@ -104,9 +97,9 @@ public class ApnFallbackManager implements Managed {
                                                         .flatMap(deviceId -> maybeAccount.flatMap(account -> account.getDevice(deviceId)));
 
           if (maybeAccount.isPresent() && maybeDevice.isPresent()) {
-            sendNotification(maybeAccount.get(), maybeDevice.get());
+            sendRecurringVoipNotification(maybeAccount.get(), maybeDevice.get());
           } else {
-            remove(uuidAndDevice);
+            removeRecurringVoipNotificationEntry(uuidAndDevice);
           }
         }
       } while (!pendingDestinations.isEmpty());
@@ -115,37 +108,37 @@ public class ApnFallbackManager implements Managed {
     }
   }
 
-  public ApnFallbackManager(FaultTolerantRedisCluster cluster,
+  public ApnPushNotificationScheduler(FaultTolerantRedisCluster pushSchedulingCluster,
                             APNSender apnSender,
                             AccountsManager accountsManager)
       throws IOException
   {
-    this.apnSender       = apnSender;
+    this.apnSender = apnSender;
     this.accountsManager = accountsManager;
-    this.cluster         = cluster;
+    this.pushSchedulingCluster = pushSchedulingCluster;
 
-    this.getScript = ClusterLuaScript.fromResource(cluster, "lua/apn/get.lua", ScriptOutputType.MULTI);
+    this.getPendingVoipDestinationsScript = ClusterLuaScript.fromResource(pushSchedulingCluster, "lua/apn/get.lua", ScriptOutputType.MULTI);
-    this.insertScript = ClusterLuaScript.fromResource(cluster, "lua/apn/insert.lua", ScriptOutputType.VALUE);
+    this.insertPendingVoipDestinationScript = ClusterLuaScript.fromResource(pushSchedulingCluster, "lua/apn/insert.lua", ScriptOutputType.VALUE);
-    this.removeScript = ClusterLuaScript.fromResource(cluster, "lua/apn/remove.lua", ScriptOutputType.INTEGER);
+    this.removePendingVoipDestinationScript = ClusterLuaScript.fromResource(pushSchedulingCluster, "lua/apn/remove.lua", ScriptOutputType.INTEGER);
 
     for (int i = 0; i < this.workerThreads.length; i++) {
       this.workerThreads[i] = new Thread(new NotificationWorker(), "ApnFallbackManagerWorker-" + i);
     }
   }
 
-  public void schedule(Account account, Device device) {
+  public void scheduleRecurringVoipNotification(Account account, Device device) {
-    schedule(account, device, System.currentTimeMillis());
+    scheduleRecurringVoipNotification(account, device, System.currentTimeMillis());
   }
 
   @VisibleForTesting
-  void schedule(Account account, Device device, long timestamp) {
+  void scheduleRecurringVoipNotification(Account account, Device device, long timestamp) {
-    sent.mark();
+    sent.increment();
-    insert(account, device, timestamp + (15 * 1000), (15 * 1000));
+    insertRecurringVoipNotificationEntry(account, device, timestamp + (15 * 1000), (15 * 1000));
   }
 
-  public void cancel(Account account, Device device) {
+  public void cancelRecurringVoipNotification(Account account, Device device) {
-    if (remove(account, device)) {
+    if (removeRecurringVoipNotificationEntry(account, device)) {
-      delivered.mark();
+      delivered.increment();
     }
   }
 
@@ -167,24 +160,24 @@ public class ApnFallbackManager implements Managed {
     }
   }
 
-  private void sendNotification(final Account account, final Device device) {
+  private void sendRecurringVoipNotification(final Account account, final Device device) {
     String apnId = device.getVoipApnId();
 
     if (apnId == null) {
-      remove(account, device);
+      removeRecurringVoipNotificationEntry(account, device);
       return;
     }
 
     long deviceLastSeen = device.getLastSeen();
 
     if (deviceLastSeen < System.currentTimeMillis() - TimeUnit.DAYS.toMillis(7)) {
-      evicted.mark();
+      evicted.increment();
-      remove(account, device);
+      removeRecurringVoipNotificationEntry(account, device);
       return;
     }
 
     apnSender.sendNotification(new PushNotification(apnId, PushNotification.TokenType.APN_VOIP, PushNotification.NotificationType.NOTIFICATION, null, account, device));
-    retry.mark();
+    retry.increment();
   }
 
   @VisibleForTesting
@@ -206,30 +199,33 @@ public class ApnFallbackManager implements Managed {
     }
   }
 
-  private boolean remove(Account account, Device device) {
+  private boolean removeRecurringVoipNotificationEntry(Account account, Device device) {
-    return remove(getEndpointKey(account, device));
+    return removeRecurringVoipNotificationEntry(getEndpointKey(account, device));
   }
 
-  private boolean remove(final String endpoint) {
+  private boolean removeRecurringVoipNotificationEntry(final String endpoint) {
-    return (long)removeScript.execute(List.of(getPendingNotificationQueueKey(endpoint), endpoint),
+    return (long) removePendingVoipDestinationScript.execute(
-                                      Collections.emptyList()) > 0;
+        List.of(getPendingRecurringVoipNotificationQueueKey(endpoint), endpoint),
+        Collections.emptyList()) > 0;
   }
 
   @SuppressWarnings("unchecked")
   @VisibleForTesting
-  List<String> getPendingDestinations(final int slot, final int limit) {
+  List<String> getPendingDestinationsForRecurringVoipNotifications(final int slot, final int limit) {
-    return (List<String>)getScript.execute(List.of(getPendingNotificationQueueKey(slot)),
+    return (List<String>) getPendingVoipDestinationsScript.execute(
-                                           List.of(String.valueOf(System.currentTimeMillis()), String.valueOf(limit)));
+        List.of(getPendingRecurringVoipNotificationQueueKey(slot)),
+        List.of(String.valueOf(System.currentTimeMillis()), String.valueOf(limit)));
   }
 
-  private void insert(final Account account, final Device device, final long timestamp, final long interval) {
+  private void insertRecurringVoipNotificationEntry(final Account account, final Device device, final long timestamp, final long interval) {
     final String endpoint = getEndpointKey(account, device);
 
-    insertScript.execute(List.of(getPendingNotificationQueueKey(endpoint), endpoint),
+    insertPendingVoipDestinationScript.execute(
-                         List.of(String.valueOf(timestamp),
+        List.of(getPendingRecurringVoipNotificationQueueKey(endpoint), endpoint),
-                                 String.valueOf(interval),
+        List.of(String.valueOf(timestamp),
-                                 account.getUuid().toString(),
+            String.valueOf(interval),
-                                 String.valueOf(device.getId())));
+            account.getUuid().toString(),
+            String.valueOf(device.getId())));
   }
 
   @VisibleForTesting
@@ -237,32 +233,15 @@ public class ApnFallbackManager implements Managed {
     return "apn_device::{" + account.getUuid() + "::" + device.getId() + "}";
   }
 
-  private String getPendingNotificationQueueKey(final String endpoint) {
+  private String getPendingRecurringVoipNotificationQueueKey(final String endpoint) {
-    return getPendingNotificationQueueKey(SlotHash.getSlot(endpoint));
+    return getPendingRecurringVoipNotificationQueueKey(SlotHash.getSlot(endpoint));
   }
 
-  private String getPendingNotificationQueueKey(final int slot) {
+  private String getPendingRecurringVoipNotificationQueueKey(final int slot) {
     return PENDING_NOTIFICATIONS_KEY + "::{" + RedisClusterUtil.getMinimalHashTag(slot) + "}";
   }
 
   private int getNextSlot() {
-    return (int)(cluster.withCluster(connection -> connection.sync().incr(NEXT_SLOT_TO_PERSIST_KEY)) % SlotHash.SLOT_COUNT);
+    return (int)(pushSchedulingCluster.withCluster(connection -> connection.sync().incr(NEXT_SLOT_TO_PERSIST_KEY)) % SlotHash.SLOT_COUNT);
   }
-
-  private static class VoipRatioGauge extends RatioGauge {
-
-    private final Meter success;
-    private final Meter attempts;
-
-    private VoipRatioGauge(Meter success, Meter attempts) {
-      this.success  = success;
-      this.attempts = attempts;
-    }
-
-    @Override
-    protected Ratio getRatio() {
-      return RatioGauge.Ratio.of(success.getFiveMinuteRate(), attempts.getFiveMinuteRate());
-    }
-  }
-
 }

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

@@ -9,7 +9,6 @@ import static org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
 
 import io.micrometer.core.instrument.Metrics;
 import org.apache.commons.lang3.StringUtils;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
 import org.whispersystems.textsecuregcm.redis.RedisOperation;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Device;

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

@@ -1,9 +1,9 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2022 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
-package org.whispersystems.textsecuregcm.metrics;
+package org.whispersystems.textsecuregcm.push;
 
 import com.codahale.metrics.MetricRegistry;
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -28,6 +28,7 @@ import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 import org.whispersystems.textsecuregcm.util.SystemMapper;
@@ -103,7 +104,7 @@ public class PushLatencyManager {
     this.clock = clock;
   }
 
-  public void recordPushSent(final UUID accountUuid, final long deviceId, final boolean isVoip) {
+  void recordPushSent(final UUID accountUuid, final long deviceId, final boolean isVoip) {
     try {
       final String recordJson = SystemMapper.getMapper().writeValueAsString(
           new PushRecord(Instant.now(clock), isVoip ? PushType.VOIP : PushType.STANDARD));
@@ -118,7 +119,7 @@ public class PushLatencyManager {
     }
   }
 
-  public void recordQueueRead(final UUID accountUuid, final long deviceId, final String userAgentString) {
+  void recordQueueRead(final UUID accountUuid, final long deviceId, final String userAgentString) {
     takePushRecord(accountUuid, deviceId).thenAccept(pushRecord -> {
       if (pushRecord != null) {
         final Duration latency = Duration.between(pushRecord.getTimestamp(), Instant.now());

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

@@ -5,6 +5,8 @@
 
 package org.whispersystems.textsecuregcm.push;
 
+import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
+
 import com.google.common.annotations.VisibleForTesting;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.Tags;
@@ -18,14 +20,13 @@ import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.util.Pair;
 import org.whispersystems.textsecuregcm.util.Util;
 
-import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
-
 public class PushNotificationManager {
 
   private final AccountsManager accountsManager;
   private final APNSender apnSender;
   private final FcmSender fcmSender;
-  private final ApnFallbackManager fallbackManager;
+  private final ApnPushNotificationScheduler apnPushNotificationScheduler;
+  private final PushLatencyManager pushLatencyManager;
 
   private static final String SENT_NOTIFICATION_COUNTER_NAME = name(PushNotificationManager.class, "sentPushNotification");
   private static final String FAILED_NOTIFICATION_COUNTER_NAME = name(PushNotificationManager.class, "failedPushNotification");
@@ -35,12 +36,14 @@ public class PushNotificationManager {
   public PushNotificationManager(final AccountsManager accountsManager,
       final APNSender apnSender,
       final FcmSender fcmSender,
-      final ApnFallbackManager fallbackManager) {
+      final ApnPushNotificationScheduler apnPushNotificationScheduler,
+      final PushLatencyManager pushLatencyManager) {
 
     this.accountsManager = accountsManager;
     this.apnSender = apnSender;
     this.fcmSender = fcmSender;
-    this.fallbackManager = fallbackManager;
+    this.apnPushNotificationScheduler = apnPushNotificationScheduler;
+    this.pushLatencyManager = pushLatencyManager;
   }
 
   public void sendNewMessageNotification(final Account destination, final long destinationDeviceId) throws NotPushRegisteredException {
@@ -65,6 +68,11 @@ public class PushNotificationManager {
         PushNotification.NotificationType.RATE_LIMIT_CHALLENGE, challengeToken, destination, device));
   }
 
+  public void handleMessagesRetrieved(final Account account, final Device device, final String userAgent) {
+    RedisOperation.unchecked(() -> pushLatencyManager.recordQueueRead(account.getUuid(), device.getId(), userAgent));
+    RedisOperation.unchecked(() -> apnPushNotificationScheduler.cancelRecurringVoipNotification(account, device));
+  }
+
   @VisibleForTesting
   Pair<String, PushNotification.TokenType> getToken(final Device device) throws NotPushRegisteredException {
     final Pair<String, PushNotification.TokenType> tokenAndType;
@@ -112,7 +120,7 @@ public class PushNotificationManager {
             pushNotification.destination() != null &&
             pushNotification.destinationDevice() != null) {
 
-          RedisOperation.unchecked(() -> fallbackManager.schedule(pushNotification.destination(),
+          RedisOperation.unchecked(() -> apnPushNotificationScheduler.scheduleRecurringVoipNotification(pushNotification.destination(),
               pushNotification.destinationDevice()));
         }
       } else {
@@ -131,7 +139,7 @@ public class PushNotificationManager {
             d.setUninstalledFeedbackTimestamp(Util.todayInMillis()));
       }
     } else {
-      RedisOperation.unchecked(() -> fallbackManager.cancel(account, device));
+      RedisOperation.unchecked(() -> apnPushNotificationScheduler.cancelRecurringVoipNotification(account, device));
     }
   }
 }

service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java

@@ -15,8 +15,6 @@ import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Collectors;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
-import org.whispersystems.textsecuregcm.redis.RedisOperation;
 import org.whispersystems.textsecuregcm.util.Constants;
 import org.whispersystems.textsecuregcm.util.Pair;
 
@@ -32,17 +30,14 @@ public class MessagesManager {
 
   private final MessagesDynamoDb messagesDynamoDb;
   private final MessagesCache messagesCache;
-  private final PushLatencyManager pushLatencyManager;
   private final ReportMessageManager reportMessageManager;
 
   public MessagesManager(
       final MessagesDynamoDb messagesDynamoDb,
       final MessagesCache messagesCache,
-      final PushLatencyManager pushLatencyManager,
       final ReportMessageManager reportMessageManager) {
     this.messagesDynamoDb = messagesDynamoDb;
     this.messagesCache = messagesCache;
-    this.pushLatencyManager = pushLatencyManager;
     this.reportMessageManager = reportMessageManager;
   }
 
@@ -60,9 +55,7 @@ public class MessagesManager {
     return messagesCache.hasMessages(destinationUuid, destinationDevice);
   }
 
-  public Pair<List<Envelope>, Boolean> getMessagesForDevice(UUID destinationUuid, long destinationDevice, final String userAgent, final boolean cachedMessagesOnly) {
+  public Pair<List<Envelope>, Boolean> getMessagesForDevice(UUID destinationUuid, long destinationDevice, final boolean cachedMessagesOnly) {
-    RedisOperation.unchecked(() -> pushLatencyManager.recordQueueRead(destinationUuid, destinationDevice, userAgent));
-
     List<Envelope> messageList = new ArrayList<>();
 
     if (!cachedMessagesOnly) {

service/src/main/java/org/whispersystems/textsecuregcm/websocket/AuthenticatedConnectListener.java

@@ -18,7 +18,6 @@ import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
-import org.whispersystems.textsecuregcm.push.ApnFallbackManager;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.push.NotPushRegisteredException;
 import org.whispersystems.textsecuregcm.push.PushNotificationManager;
@@ -44,21 +43,18 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
   private final ReceiptSender         receiptSender;
   private final MessagesManager       messagesManager;
   private final PushNotificationManager pushNotificationManager;
-  private final ApnFallbackManager    apnFallbackManager;
   private final ClientPresenceManager clientPresenceManager;
   private final ScheduledExecutorService scheduledExecutorService;
 
   public AuthenticatedConnectListener(ReceiptSender receiptSender,
       MessagesManager messagesManager,
       PushNotificationManager pushNotificationManager,
-      ApnFallbackManager apnFallbackManager,
       ClientPresenceManager clientPresenceManager,
       ScheduledExecutorService scheduledExecutorService)
   {
     this.receiptSender         = receiptSender;
     this.messagesManager       = messagesManager;
     this.pushNotificationManager = pushNotificationManager;
-    this.apnFallbackManager    = apnFallbackManager;
     this.clientPresenceManager = clientPresenceManager;
     this.scheduledExecutorService = scheduledExecutorService;
   }
@@ -75,7 +71,7 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
           scheduledExecutorService);
 
       openWebsocketCounter.inc();
-      RedisOperation.unchecked(() -> apnFallbackManager.cancel(auth.getAccount(), device));
+      pushNotificationManager.handleMessagesRetrieved(auth.getAccount(), device, context.getClient().getUserAgent());
 
       final AtomicReference<ScheduledFuture<?>> renewPresenceFutureReference = new AtomicReference<>();
 

service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java

@@ -320,8 +320,8 @@ public class WebSocketConnection implements MessageAvailabilityListener, Displac
 
   private void sendNextMessagePage(final boolean cachedMessagesOnly, final CompletableFuture<Void> queueClearedFuture) {
     try {
-      final Pair<List<Envelope>, Boolean> messagesAndHasMore = messagesManager
+      final Pair<List<Envelope>, Boolean> messagesAndHasMore = messagesManager.getMessagesForDevice(
-          .getMessagesForDevice(auth.getAccount().getUuid(), device.getId(), client.getUserAgent(), cachedMessagesOnly);
+          auth.getAccount().getUuid(), device.getId(), cachedMessagesOnly);
 
       final List<Envelope> messages = messagesAndHasMore.first();
       final boolean hasMore = messagesAndHasMore.second();

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

@@ -25,7 +25,7 @@ import net.sourceforge.argparse4j.inf.Subparser;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialGenerator;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
@@ -180,7 +180,7 @@ public class AssignUsernameCommand extends EnvironmentCommand<WhisperServerConfi
         configuration.getReportMessageConfiguration().getReportTtl());
     ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster,
         configuration.getReportMessageConfiguration().getCounterTtl());
-    MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager,
+    MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache,
         reportMessageManager);
     DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
         deletedAccountsLockDynamoDbClient,

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

@@ -27,7 +27,7 @@ import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialGenerator;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
@@ -183,8 +183,8 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
           configuration.getReportMessageConfiguration().getReportTtl());
       ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster,
               configuration.getReportMessageConfiguration().getCounterTtl());
-      MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager,
+      MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache,
-          reportMessageManager);
+              reportMessageManager);
       DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
           deletedAccountsLockDynamoDbClient,
           configuration.getDynamoDbTables().getDeletedAccountsLock().getTableName());

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

@@ -26,7 +26,7 @@ import net.sourceforge.argparse4j.inf.Subparser;
 import org.whispersystems.textsecuregcm.WhisperServerConfiguration;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialGenerator;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.securebackup.SecureBackupClient;
@@ -184,8 +184,8 @@ public class SetUserDiscoverabilityCommand extends EnvironmentCommand<WhisperSer
           configuration.getReportMessageConfiguration().getReportTtl());
       ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster,
               configuration.getReportMessageConfiguration().getCounterTtl());
-      MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager,
+      MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache,
-          reportMessageManager);
+              reportMessageManager);
       DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
           deletedAccountsLockDynamoDbClient,
           configuration.getDynamoDbTables().getDeletedAccountsLock().getTableName());

service/src/test/java/org/whispersystems/textsecuregcm/controllers/MessageControllerTest.java

@@ -67,8 +67,8 @@ import org.whispersystems.textsecuregcm.entities.StaleDevices;
 import org.whispersystems.textsecuregcm.limits.RateLimiter;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.mappers.RateLimitExceededExceptionMapper;
-import org.whispersystems.textsecuregcm.push.ApnFallbackManager;
 import org.whispersystems.textsecuregcm.push.MessageSender;
+import org.whispersystems.textsecuregcm.push.PushNotificationManager;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
@@ -107,7 +107,7 @@ class MessageControllerTest {
   private static final MessagesManager messagesManager = mock(MessagesManager.class);
   private static final RateLimiters rateLimiters = mock(RateLimiters.class);
   private static final RateLimiter rateLimiter = mock(RateLimiter.class);
-  private static final ApnFallbackManager apnFallbackManager = mock(ApnFallbackManager.class);
+  private static final PushNotificationManager pushNotificationManager = mock(PushNotificationManager.class);
   private static final ReportMessageManager reportMessageManager = mock(ReportMessageManager.class);
   private static final ExecutorService multiRecipientMessageExecutor = mock(ExecutorService.class);
 
@@ -119,7 +119,7 @@ class MessageControllerTest {
       .setTestContainerFactory(new GrizzlyWebTestContainerFactory())
       .addResource(
           new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, deletedAccountsManager,
-              messagesManager, apnFallbackManager, reportMessageManager, multiRecipientMessageExecutor))
+              messagesManager, pushNotificationManager, reportMessageManager, multiRecipientMessageExecutor))
       .build();
 
   @BeforeEach
@@ -170,7 +170,7 @@ class MessageControllerTest {
         messagesManager,
         rateLimiters,
         rateLimiter,
-        apnFallbackManager,
+        pushNotificationManager,
         reportMessageManager
     );
   }
@@ -435,13 +435,16 @@ class MessageControllerTest {
         .map(OutgoingMessageEntity::fromEnvelope)
         .toList(), false);
 
-    when(messagesManager.getMessagesForDevice(eq(AuthHelper.VALID_UUID), eq(1L), anyString(), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(AuthHelper.VALID_UUID), eq(1L), anyBoolean()))
         .thenReturn(new Pair<>(messages, false));
 
+    final String userAgent = "Test-UA";
+
     OutgoingMessageEntityList response =
         resources.getJerseyTest().target("/v1/messages/")
             .request()
             .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_UUID, AuthHelper.VALID_PASSWORD))
+            .header("USer-Agent", userAgent)
             .accept(MediaType.APPLICATION_JSON_TYPE)
             .get(OutgoingMessageEntityList.class);
 
@@ -458,6 +461,8 @@ class MessageControllerTest {
 
     assertEquals(updatedPniOne, response.messages().get(0).updatedPni());
     assertNull(response.messages().get(1).updatedPni());
+
+    verify(pushNotificationManager).handleMessagesRetrieved(AuthHelper.VALID_ACCOUNT, AuthHelper.VALID_DEVICE, userAgent);
   }
 
   @Test
@@ -472,7 +477,7 @@ class MessageControllerTest {
             UUID.randomUUID(), 2, AuthHelper.VALID_UUID, null, null, 0)
     );
 
-    when(messagesManager.getMessagesForDevice(eq(AuthHelper.VALID_UUID), eq(1L), anyString(), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(AuthHelper.VALID_UUID), eq(1L), anyBoolean()))
         .thenReturn(new Pair<>(messages, false));
 
     Response response =

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

@@ -26,7 +26,7 @@ import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.util.Pair;
 
-class ApnFallbackManagerTest {
+class ApnPushNotificationSchedulerTest {
 
   @RegisterExtension
   static final RedisClusterExtension REDIS_CLUSTER_EXTENSION = RedisClusterExtension.builder().build();
@@ -36,7 +36,7 @@ class ApnFallbackManagerTest {
 
   private APNSender apnSender;
 
-  private ApnFallbackManager apnFallbackManager;
+  private ApnPushNotificationScheduler apnPushNotificationScheduler;
 
   private static final UUID ACCOUNT_UUID = UUID.randomUUID();
   private static final String ACCOUNT_NUMBER = "+18005551234";
@@ -62,41 +62,43 @@ class ApnFallbackManagerTest {
 
     apnSender = mock(APNSender.class);
 
-    apnFallbackManager = new ApnFallbackManager(REDIS_CLUSTER_EXTENSION.getRedisCluster(), apnSender, accountsManager);
+    apnPushNotificationScheduler = new ApnPushNotificationScheduler(REDIS_CLUSTER_EXTENSION.getRedisCluster(), apnSender, accountsManager);
   }
 
   @Test
   void testClusterInsert() {
-    final String endpoint = apnFallbackManager.getEndpointKey(account, device);
+    final String endpoint = apnPushNotificationScheduler.getEndpointKey(account, device);
 
-    assertTrue(apnFallbackManager.getPendingDestinations(SlotHash.getSlot(endpoint), 1).isEmpty());
+    assertTrue(
+        apnPushNotificationScheduler.getPendingDestinationsForRecurringVoipNotifications(SlotHash.getSlot(endpoint), 1).isEmpty());
 
-    apnFallbackManager.schedule(account, device, System.currentTimeMillis() - 30_000);
+    apnPushNotificationScheduler.scheduleRecurringVoipNotification(account, device, System.currentTimeMillis() - 30_000);
 
-    final List<String> pendingDestinations = apnFallbackManager.getPendingDestinations(SlotHash.getSlot(endpoint), 2);
+    final List<String> pendingDestinations = apnPushNotificationScheduler.getPendingDestinationsForRecurringVoipNotifications(SlotHash.getSlot(endpoint), 2);
     assertEquals(1, pendingDestinations.size());
 
-    final Optional<Pair<String, Long>> maybeUuidAndDeviceId = ApnFallbackManager.getSeparated(
+    final Optional<Pair<String, Long>> maybeUuidAndDeviceId = ApnPushNotificationScheduler.getSeparated(
         pendingDestinations.get(0));
 
     assertTrue(maybeUuidAndDeviceId.isPresent());
     assertEquals(ACCOUNT_UUID.toString(), maybeUuidAndDeviceId.get().first());
     assertEquals(DEVICE_ID, (long) maybeUuidAndDeviceId.get().second());
 
-    assertTrue(apnFallbackManager.getPendingDestinations(SlotHash.getSlot(endpoint), 1).isEmpty());
+    assertTrue(
+        apnPushNotificationScheduler.getPendingDestinationsForRecurringVoipNotifications(SlotHash.getSlot(endpoint), 1).isEmpty());
   }
 
   @Test
   void testProcessNextSlot() {
-    final ApnFallbackManager.NotificationWorker worker = apnFallbackManager.new NotificationWorker();
+    final ApnPushNotificationScheduler.NotificationWorker worker = apnPushNotificationScheduler.new NotificationWorker();
 
-    apnFallbackManager.schedule(account, device, System.currentTimeMillis() - 30_000);
+    apnPushNotificationScheduler.scheduleRecurringVoipNotification(account, device, System.currentTimeMillis() - 30_000);
 
-    final int slot = SlotHash.getSlot(apnFallbackManager.getEndpointKey(account, device));
+    final int slot = SlotHash.getSlot(apnPushNotificationScheduler.getEndpointKey(account, device));
     final int previousSlot = (slot + SlotHash.SLOT_COUNT - 1) % SlotHash.SLOT_COUNT;
 
     REDIS_CLUSTER_EXTENSION.getRedisCluster().withCluster(connection -> connection.sync()
-        .set(ApnFallbackManager.NEXT_SLOT_TO_PERSIST_KEY, String.valueOf(previousSlot)));
+        .set(ApnPushNotificationScheduler.NEXT_SLOT_TO_PERSIST_KEY, String.valueOf(previousSlot)));
 
     assertEquals(1, worker.processNextSlot());
 

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

@@ -26,7 +26,6 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;

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

@@ -1,9 +1,9 @@
 /*
- * Copyright 2013-2020 Signal Messenger, LLC
+ * Copyright 2013-2022 Signal Messenger, LLC
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
-package org.whispersystems.textsecuregcm.metrics;
+package org.whispersystems.textsecuregcm.push;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -23,8 +23,9 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicPushLatencyConfiguration;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager.PushRecord;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager.PushType;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager.PushRecord;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager.PushType;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 

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

@@ -14,6 +14,7 @@ import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 import java.util.Optional;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -28,7 +29,8 @@ class PushNotificationManagerTest {
   private AccountsManager accountsManager;
   private APNSender apnSender;
   private FcmSender fcmSender;
-  private ApnFallbackManager apnFallbackManager;
+  private ApnPushNotificationScheduler apnPushNotificationScheduler;
+  private PushLatencyManager pushLatencyManager;
 
   private PushNotificationManager pushNotificationManager;
 
@@ -37,11 +39,13 @@ class PushNotificationManagerTest {
     accountsManager = mock(AccountsManager.class);
     apnSender = mock(APNSender.class);
     fcmSender = mock(FcmSender.class);
-    apnFallbackManager = mock(ApnFallbackManager.class);
+    apnPushNotificationScheduler = mock(ApnPushNotificationScheduler.class);
+    pushLatencyManager = mock(PushLatencyManager.class);
 
     AccountsHelper.setupMockUpdate(accountsManager);
 
-    pushNotificationManager = new PushNotificationManager(accountsManager, apnSender, fcmSender, apnFallbackManager);
+    pushNotificationManager = new PushNotificationManager(accountsManager, apnSender, fcmSender,
+        apnPushNotificationScheduler, pushLatencyManager);
   }
 
   @Test
@@ -113,7 +117,7 @@ class PushNotificationManagerTest {
     verifyNoInteractions(apnSender);
     verify(accountsManager, never()).updateDevice(eq(account), eq(Device.MASTER_ID), any());
     verify(device, never()).setUninstalledFeedbackTimestamp(Util.todayInMillis());
-    verifyNoInteractions(apnFallbackManager);
+    verifyNoInteractions(apnPushNotificationScheduler);
   }
 
   @Test
@@ -136,7 +140,7 @@ class PushNotificationManagerTest {
     verifyNoInteractions(fcmSender);
     verify(accountsManager, never()).updateDevice(eq(account), eq(Device.MASTER_ID), any());
     verify(device, never()).setUninstalledFeedbackTimestamp(Util.todayInMillis());
-    verify(apnFallbackManager).schedule(account, device);
+    verify(apnPushNotificationScheduler).scheduleRecurringVoipNotification(account, device);
   }
 
   @Test
@@ -159,7 +163,7 @@ class PushNotificationManagerTest {
     verify(accountsManager).updateDevice(eq(account), eq(Device.MASTER_ID), any());
     verify(device).setUninstalledFeedbackTimestamp(Util.todayInMillis());
     verifyNoInteractions(apnSender);
-    verifyNoInteractions(apnFallbackManager);
+    verifyNoInteractions(apnPushNotificationScheduler);
   }
 
   @Test
@@ -181,6 +185,22 @@ class PushNotificationManagerTest {
     verifyNoInteractions(fcmSender);
     verify(accountsManager, never()).updateDevice(eq(account), eq(Device.MASTER_ID), any());
     verify(device, never()).setUninstalledFeedbackTimestamp(Util.todayInMillis());
-    verify(apnFallbackManager).cancel(account, device);
+    verify(apnPushNotificationScheduler).cancelRecurringVoipNotification(account, device);
+  }
+
+  @Test
+  void testHandleMessagesRetrieved() {
+    final UUID accountIdentifier = UUID.randomUUID();
+    final Account account = mock(Account.class);
+    final Device device = mock(Device.class);
+    final String userAgent = "User-Agent";
+
+    when(account.getUuid()).thenReturn(accountIdentifier);
+    when(device.getId()).thenReturn(Device.MASTER_ID);
+
+    pushNotificationManager.handleMessagesRetrieved(account, device, userAgent);
+
+    verify(pushLatencyManager).recordQueueRead(accountIdentifier, Device.MASTER_ID, userAgent);
+    verify(apnPushNotificationScheduler).cancelRecurringVoipNotification(account, device);
   }
 }

service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterIntegrationTest.java

@@ -32,7 +32,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import org.whispersystems.textsecuregcm.tests.util.MessagesDynamoDbExtension;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
@@ -73,8 +73,7 @@ class MessagePersisterIntegrationTest {
     notificationExecutorService = Executors.newSingleThreadExecutor();
     messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(),
         REDIS_CLUSTER_EXTENSION.getRedisCluster(), notificationExecutorService);
-    messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, mock(PushLatencyManager.class),
+    messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, mock(ReportMessageManager.class));
-        mock(ReportMessageManager.class));
     messagePersister = new MessagePersister(messagesCache, messagesManager, accountsManager,
         dynamicConfigurationManager, PERSIST_DELAY);
 

service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesManagerTest.java

@@ -14,7 +14,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
 import java.util.UUID;
 import org.junit.jupiter.api.Test;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
+import org.whispersystems.textsecuregcm.push.PushLatencyManager;
 
 class MessagesManagerTest {
 
@@ -24,7 +24,7 @@ class MessagesManagerTest {
   private final ReportMessageManager reportMessageManager = mock(ReportMessageManager.class);
 
   private final MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache,
-      pushLatencyManager, reportMessageManager);
+      reportMessageManager);
 
   @Test
   void insert() {

service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java

@@ -43,7 +43,6 @@ import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
-import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import org.whispersystems.textsecuregcm.storage.Account;
@@ -100,7 +99,7 @@ class WebSocketConnectionIntegrationTest {
 
     webSocketConnection = new WebSocketConnection(
         mock(ReceiptSender.class),
-        new MessagesManager(messagesDynamoDb, messagesCache, mock(PushLatencyManager.class), reportMessageManager),
+        new MessagesManager(messagesDynamoDb, messagesCache, reportMessageManager),
         new AuthenticatedAccount(() -> new Pair<>(account, device)),
         device,
         webSocketClient,

service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java

@@ -52,7 +52,7 @@ import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.auth.AccountAuthenticator;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
 import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntityList;
-import org.whispersystems.textsecuregcm.push.ApnFallbackManager;
+import org.whispersystems.textsecuregcm.push.ApnPushNotificationScheduler;
 import org.whispersystems.textsecuregcm.push.ClientPresenceManager;
 import org.whispersystems.textsecuregcm.push.PushNotificationManager;
 import org.whispersystems.textsecuregcm.push.ReceiptSender;
@@ -83,7 +83,7 @@ class WebSocketConnectionTest {
   private AuthenticatedAccount auth;
   private UpgradeRequest upgradeRequest;
   private ReceiptSender receiptSender;
-  private ApnFallbackManager apnFallbackManager;
+  private PushNotificationManager pushNotificationManager;
   private ScheduledExecutorService retrySchedulingExecutor;
 
   @BeforeEach
@@ -95,7 +95,7 @@ class WebSocketConnectionTest {
     auth = new AuthenticatedAccount(() -> new Pair<>(account, device));
     upgradeRequest = mock(UpgradeRequest.class);
     receiptSender = mock(ReceiptSender.class);
-    apnFallbackManager = mock(ApnFallbackManager.class);
+    pushNotificationManager = mock(PushNotificationManager.class);
     retrySchedulingExecutor = mock(ScheduledExecutorService.class);
   }
 
@@ -104,7 +104,7 @@ class WebSocketConnectionTest {
     MessagesManager storedMessages = mock(MessagesManager.class);
     WebSocketAccountAuthenticator webSocketAuthenticator = new WebSocketAccountAuthenticator(accountAuthenticator);
     AuthenticatedConnectListener connectListener = new AuthenticatedConnectListener(receiptSender, storedMessages,
-        mock(PushNotificationManager.class), apnFallbackManager, mock(ClientPresenceManager.class),
+        mock(PushNotificationManager.class), mock(ClientPresenceManager.class),
         retrySchedulingExecutor);
     WebSocketSessionContext sessionContext = mock(WebSocketSessionContext.class);
 
@@ -166,7 +166,7 @@ class WebSocketConnectionTest {
 
     String userAgent = "user-agent";
 
-    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), userAgent, false))
+    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), false))
         .thenReturn(new Pair<>(outgoingMessages, false));
 
     final List<CompletableFuture<WebSocketResponseMessage>> futures = new LinkedList<>();
@@ -221,9 +221,8 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(accountUuid);
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
-    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), eq("Test-UA"), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), anyBoolean()))
         .thenReturn(new Pair<>(Collections.emptyList(), false))
         .thenReturn(new Pair<>(List.of(createMessage(UUID.randomUUID(), UUID.randomUUID(), 1111, "first")), false))
         .thenReturn(new Pair<>(List.of(createMessage(UUID.randomUUID(), UUID.randomUUID(), 2222, "second")), false));
@@ -316,7 +315,7 @@ class WebSocketConnectionTest {
 
     String userAgent = "user-agent";
 
-    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), userAgent, false))
+    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), false))
         .thenReturn(new Pair<>(pendingMessages, false));
 
     final List<CompletableFuture<WebSocketResponseMessage>> futures = new LinkedList<>();
@@ -362,12 +361,11 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(UUID.randomUUID());
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
     final AtomicBoolean threadWaiting = new AtomicBoolean(false);
     final AtomicBoolean returnMessageList = new AtomicBoolean(false);
 
-    when(messagesManager.getMessagesForDevice(account.getUuid(), 1L, client.getUserAgent(), false)).thenAnswer(
+    when(messagesManager.getMessagesForDevice(account.getUuid(), 1L, false)).thenAnswer(
         (Answer<OutgoingMessageEntityList>) invocation -> {
       synchronized (threadWaiting) {
         threadWaiting.set(true);
@@ -415,7 +413,7 @@ class WebSocketConnectionTest {
       }
     });
 
-    verify(messagesManager).getMessagesForDevice(any(UUID.class), anyLong(), anyString(), eq(false));
+    verify(messagesManager).getMessagesForDevice(any(UUID.class), anyLong(), eq(false));
   }
 
   @Test
@@ -429,7 +427,6 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(UUID.randomUUID());
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
     final List<Envelope> firstPageMessages =
         List.of(createMessage(UUID.randomUUID(), UUID.randomUUID(), 1111, "first"),
@@ -438,7 +435,7 @@ class WebSocketConnectionTest {
     final List<Envelope> secondPageMessages =
         List.of(createMessage(UUID.randomUUID(), UUID.randomUUID(), 3333, "third"));
 
-    when(messagesManager.getMessagesForDevice(account.getUuid(), 1L, client.getUserAgent(), false))
+    when(messagesManager.getMessagesForDevice(account.getUuid(), 1L, false))
             .thenReturn(new Pair<>(firstPageMessages, true))
             .thenReturn(new Pair<>(secondPageMessages, false));
 
@@ -473,13 +470,12 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(UUID.randomUUID());
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
     final UUID senderUuid = UUID.randomUUID();
     final List<Envelope> messages = List.of(
         createMessage(senderUuid, UUID.randomUUID(), 1111L, "message the first"));
 
-    when(messagesManager.getMessagesForDevice(account.getUuid(), 1L, client.getUserAgent(), false))
+    when(messagesManager.getMessagesForDevice(account.getUuid(), 1L, false))
         .thenReturn(new Pair<>(messages, false));
 
     final WebSocketResponseMessage successResponse = mock(WebSocketResponseMessage.class);
@@ -530,9 +526,8 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(accountUuid);
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
-    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), eq("Test-UA"), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), anyBoolean()))
         .thenReturn(new Pair<>(Collections.emptyList(), false));
 
     final WebSocketResponseMessage successResponse = mock(WebSocketResponseMessage.class);
@@ -560,7 +555,6 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(accountUuid);
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
     final List<Envelope> firstPageMessages =
         List.of(createMessage(UUID.randomUUID(), UUID.randomUUID(), 1111, "first"),
@@ -569,7 +563,7 @@ class WebSocketConnectionTest {
     final List<Envelope> secondPageMessages =
         List.of(createMessage(UUID.randomUUID(), UUID.randomUUID(), 3333, "third"));
 
-    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), eq("Test-UA"), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), anyBoolean()))
             .thenReturn(new Pair<>(firstPageMessages, false))
             .thenReturn(new Pair<>(secondPageMessages, false))
             .thenReturn(new Pair<>(Collections.emptyList(), false));
@@ -609,9 +603,8 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(accountUuid);
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
-    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), eq("Test-UA"), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), anyBoolean()))
         .thenReturn(new Pair<>(Collections.emptyList(), false));
 
     final WebSocketResponseMessage successResponse = mock(WebSocketResponseMessage.class);
@@ -623,11 +616,11 @@ class WebSocketConnectionTest {
     // anything.
     connection.processStoredMessages();
 
-    verify(messagesManager).getMessagesForDevice(account.getUuid(), device.getId(), client.getUserAgent(), false);
+    verify(messagesManager).getMessagesForDevice(account.getUuid(), device.getId(), false);
 
     connection.handleNewMessagesAvailable();
 
-    verify(messagesManager).getMessagesForDevice(account.getUuid(), device.getId(), client.getUserAgent(), true);
+    verify(messagesManager).getMessagesForDevice(account.getUuid(), device.getId(), true);
   }
 
   @Test
@@ -643,9 +636,8 @@ class WebSocketConnectionTest {
     when(account.getUuid()).thenReturn(accountUuid);
     when(device.getId()).thenReturn(1L);
     when(client.isOpen()).thenReturn(true);
-    when(client.getUserAgent()).thenReturn("Test-UA");
 
-    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), eq("Test-UA"), anyBoolean()))
+    when(messagesManager.getMessagesForDevice(eq(accountUuid), eq(1L), anyBoolean()))
         .thenReturn(new Pair<>(Collections.emptyList(), false));
 
     final WebSocketResponseMessage successResponse = mock(WebSocketResponseMessage.class);
@@ -658,7 +650,7 @@ class WebSocketConnectionTest {
     connection.processStoredMessages();
     connection.handleMessagesPersisted();
 
-    verify(messagesManager, times(2)).getMessagesForDevice(account.getUuid(), device.getId(), client.getUserAgent(), false);
+    verify(messagesManager, times(2)).getMessagesForDevice(account.getUuid(), device.getId(), false);
   }
 
   @Test
@@ -692,7 +684,7 @@ class WebSocketConnectionTest {
 
     String userAgent = "Signal-Desktop/1.2.3";
 
-    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), userAgent, false))
+    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), false))
             .thenReturn(new Pair<>(outgoingMessages, false));
 
     final List<CompletableFuture<WebSocketResponseMessage>> futures = new LinkedList<>();
@@ -762,7 +754,7 @@ class WebSocketConnectionTest {
 
     String userAgent = "Signal-Android/4.68.3";
 
-    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), userAgent, false))
+    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), false))
             .thenReturn(new Pair<>(outgoingMessages, false));
 
     final List<CompletableFuture<WebSocketResponseMessage>> futures = new LinkedList<>();
@@ -814,7 +806,7 @@ class WebSocketConnectionTest {
 
     String userAgent = "Signal-Android/4.68.3";
 
-    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), userAgent, false))
+    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), false))
         .thenThrow(new RedisException("OH NO"));
 
     when(retrySchedulingExecutor.schedule(any(Runnable.class), anyLong(), any())).thenAnswer(
@@ -848,7 +840,7 @@ class WebSocketConnectionTest {
 
     String userAgent = "Signal-Android/4.68.3";
 
-    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), userAgent, false))
+    when(storedMessages.getMessagesForDevice(account.getUuid(), device.getId(), false))
         .thenThrow(new RedisException("OH NO"));
 
     final WebSocketClient client = mock(WebSocketClient.class);