Publish "messages persisted" events

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

@@ -58,6 +58,11 @@ public class PubSubClientEventManager extends RedisClusterPubSubAdapter<byte[],
       .build()
       .toByteArray();
 
+  private final byte[] MESSAGES_PERSISTED_EVENT_BYTES = ClientEvent.newBuilder()
+      .setMessagesPersisted(MessagesPersistedEvent.getDefaultInstance())
+      .build()
+      .toByteArray();
+
   @Nullable
   private FaultTolerantPubSubClusterConnection<byte[], byte[]> pubSubConnection;
 
@@ -240,10 +245,29 @@ public class PubSubClientEventManager extends RedisClusterPubSubAdapter<byte[],
     }
 
     return pubSubConnection.withPubSubConnection(connection ->
-        connection.async().spublish(getClientPresenceKey(accountIdentifier, deviceId), NEW_MESSAGE_EVENT_BYTES))
+            connection.async().spublish(getClientPresenceKey(accountIdentifier, deviceId), NEW_MESSAGE_EVENT_BYTES))
         .thenApply(listeners -> listeners > 0);
   }
 
+  /**
+   * Publishes an event notifying a specific device that messages have been persisted from short-term to long-term
+   * storage.
+   *
+   * @param accountIdentifier the account identifier for which messages have been persisted
+   * @param deviceId the ID of the device within the target account
+   *
+   * @return a future that completes when the event has been published
+   */
+  public CompletionStage<Void> handleMessagesPersisted(final UUID accountIdentifier, final byte deviceId) {
+    if (pubSubConnection == null) {
+      throw new IllegalStateException("Presence manager not started");
+    }
+
+    return pubSubConnection.withPubSubConnection(connection ->
+            connection.async().spublish(getClientPresenceKey(accountIdentifier, deviceId), MESSAGES_PERSISTED_EVENT_BYTES))
+        .thenRun(Util.NOOP);
+  }
+
   /**
    * Tests whether a client with the given account/device is connected to this presence manager instance.
    *

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

@@ -22,6 +22,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
+import org.whispersystems.textsecuregcm.push.PubSubClientEventManager;
 import org.whispersystems.textsecuregcm.util.Util;
 import software.amazon.awssdk.services.dynamodb.model.ItemCollectionSizeLimitExceededException;
 
@@ -30,6 +31,7 @@ public class MessagePersister implements Managed {
   private final MessagesCache messagesCache;
   private final MessagesManager messagesManager;
   private final AccountsManager accountsManager;
+  private final PubSubClientEventManager pubSubClientEventManager;
 
   private final Duration persistDelay;
 
@@ -63,13 +65,16 @@ public class MessagePersister implements Managed {
 
   public MessagePersister(final MessagesCache messagesCache, final MessagesManager messagesManager,
       final AccountsManager accountsManager,
-      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager, final Duration persistDelay,
+      final PubSubClientEventManager pubSubClientEventManager,
+      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
+      final Duration persistDelay,
       final int dedicatedProcessWorkerThreadCount
   ) {
 
     this.messagesCache = messagesCache;
     this.messagesManager = messagesManager;
     this.accountsManager = accountsManager;
+    this.pubSubClientEventManager = pubSubClientEventManager;
     this.persistDelay = persistDelay;
     this.workerThreads = new Thread[dedicatedProcessWorkerThreadCount];
 
@@ -206,6 +211,7 @@ public class MessagePersister implements Managed {
       maybeUnlink(account, deviceId); // may throw, in which case we'll retry later by the usual mechanism
     } finally {
       messagesCache.unlockQueueForPersistence(accountUuid, deviceId);
+      pubSubClientEventManager.handleMessagesPersisted(accountUuid, deviceId);
       sample.stop(persistQueueTimer);
     }
 

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

@@ -76,6 +76,7 @@ record CommandDependencies(
     ReportMessageManager reportMessageManager,
     MessagesCache messagesCache,
     MessagesManager messagesManager,
+    PubSubClientEventManager pubSubClientEventManager,
     KeysManager keysManager,
     APNSender apnSender,
     FcmSender fcmSender,
@@ -271,6 +272,7 @@ record CommandDependencies(
         reportMessageManager,
         messagesCache,
         messagesManager,
+        pubSubClientEventManager,
         keys,
         apnSender,
         fcmSender,

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

@@ -60,8 +60,11 @@ public class MessagePersisterServiceCommand extends ServerCommand<WhisperServerC
           });
     }
 
-    final MessagePersister messagePersister = new MessagePersister(deps.messagesCache(), deps.messagesManager(),
-        deps.accountsManager(), deps.dynamicConfigurationManager(),
+    final MessagePersister messagePersister = new MessagePersister(deps.messagesCache(),
+        deps.messagesManager(),
+        deps.accountsManager(),
+        deps.pubSubClientEventManager(),
+        deps.dynamicConfigurationManager(),
         Duration.ofMinutes(configuration.getMessageCacheConfiguration().getPersistDelayMinutes()),
         namespace.getInt(WORKER_COUNT));