Unconditionally write messages to the messages table as envelopes

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

@@ -342,7 +342,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     Keys keys = new Keys(dynamoDbClient, config.getDynamoDbTables().getKeys().getTableName());
     MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(dynamoDbClient,
         config.getDynamoDbTables().getMessages().getTableName(),
-        config.getDynamoDbTables().getMessages().getExpiration(), dynamicConfigurationManager);
+        config.getDynamoDbTables().getMessages().getExpiration());
     RemoteConfigs remoteConfigs = new RemoteConfigs(dynamoDbClient,
         config.getDynamoDbTables().getRemoteConfig().getTableName());
     PushChallengeDynamoDb pushChallengeDynamoDb = new PushChallengeDynamoDb(dynamoDbClient,

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicConfiguration.java

@@ -64,10 +64,6 @@ public class DynamicConfiguration {
   @Valid
   DynamicMessagePersisterConfiguration messagePersister = new DynamicMessagePersisterConfiguration();
 
-  @JsonProperty
-  @Valid
-  DynamicMessageTableConfiguration messageTable = new DynamicMessageTableConfiguration();
-
   public Optional<DynamicExperimentEnrollmentConfiguration> getExperimentEnrollmentConfiguration(
       final String experimentName) {
     return Optional.ofNullable(experiments.get(experimentName));
@@ -130,8 +126,4 @@ public class DynamicConfiguration {
   public DynamicMessagePersisterConfiguration getMessagePersisterConfiguration() {
     return messagePersister;
   }
-
-  public DynamicMessageTableConfiguration getMessageTableConfiguration() {
-    return messageTable;
-  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicMessageTableConfiguration.java

@@ -1,18 +0,0 @@
-/*
- * Copyright 2013-2022 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-package org.whispersystems.textsecuregcm.configuration.dynamic;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-public class DynamicMessageTableConfiguration {
-
-  @JsonProperty
-  private boolean writeEnvelopes = false;
-
-  public boolean isWriteEnvelopes() {
-    return writeEnvelopes;
-  }
-}

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

@@ -8,6 +8,7 @@ package org.whispersystems.textsecuregcm.storage;
 import static com.codahale.metrics.MetricRegistry.name;
 import static io.micrometer.core.instrument.Metrics.timer;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import com.google.protobuf.InvalidProtocolBufferException;
 import io.micrometer.core.instrument.Counter;
@@ -24,7 +25,6 @@ import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 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.entities.OutgoingMessageEntity;
 import org.whispersystems.textsecuregcm.util.AttributeValues;
@@ -48,17 +48,34 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore {
   private static final String LOCAL_INDEX_MESSAGE_UUID_NAME = "Message_UUID_Index";
   private static final String LOCAL_INDEX_MESSAGE_UUID_KEY_SORT = "U";
 
-  private static final String KEY_TYPE = "T";
-  private static final String KEY_TIMESTAMP = "TS";
-  private static final String KEY_SOURCE = "SN";
-  private static final String KEY_SOURCE_UUID = "SU";
-  private static final String KEY_SOURCE_DEVICE = "SD";
-  private static final String KEY_DESTINATION_UUID = "DU";
-  private static final String KEY_UPDATED_PNI = "UP";
-  private static final String KEY_CONTENT = "C";
   private static final String KEY_TTL = "E";
   private static final String KEY_ENVELOPE_BYTES = "EB";
 
+  // TODO Stop reading messages by attribute value after DATE
+  @Deprecated
+  private static final String KEY_TYPE = "T";
+
+  @Deprecated
+  private static final String KEY_TIMESTAMP = "TS";
+
+  @Deprecated
+  private static final String KEY_SOURCE = "SN";
+
+  @Deprecated
+  private static final String KEY_SOURCE_UUID = "SU";
+
+  @Deprecated
+  private static final String KEY_SOURCE_DEVICE = "SD";
+
+  @Deprecated
+  private static final String KEY_DESTINATION_UUID = "DU";
+
+  @Deprecated
+  private static final String KEY_UPDATED_PNI = "UP";
+
+  @Deprecated
+  private static final String KEY_CONTENT = "C";
+
   private final Timer storeTimer = timer(name(getClass(), "store"));
   private final Timer loadTimer = timer(name(getClass(), "load"));
   private final Timer deleteByGuid = timer(name(getClass(), "delete", "guid"));
@@ -69,20 +86,16 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore {
   private final String tableName;
   private final Duration timeToLive;
 
-  private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
-
   private static final Counter GET_MESSAGE_WITH_ATTRIBUTES_COUNTER = Metrics.counter(name(MessagesDynamoDb.class, "loadMessage"), "format", "attributes");
   private static final Counter GET_MESSAGE_WITH_ENVELOPE_COUNTER = Metrics.counter(name(MessagesDynamoDb.class, "loadMessage"), "format", "envelope");
 
   private static final Logger logger = LoggerFactory.getLogger(MessagesDynamoDb.class);
 
-  public MessagesDynamoDb(DynamoDbClient dynamoDb, String tableName, Duration timeToLive,
+  public MessagesDynamoDb(DynamoDbClient dynamoDb, String tableName, Duration timeToLive) {
-      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager) {
     super(dynamoDb);
 
     this.tableName = tableName;
     this.timeToLive = timeToLive;
-    this.dynamicConfigurationManager = dynamicConfigurationManager;
   }
 
   public void store(final List<MessageProtos.Envelope> messages, final UUID destinationAccountUuid, final long destinationDeviceId) {
@@ -103,31 +116,8 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore {
           .put(KEY_PARTITION, partitionKey)
           .put(KEY_SORT, convertSortKey(destinationDeviceId, message.getServerTimestamp(), messageUuid))
           .put(LOCAL_INDEX_MESSAGE_UUID_KEY_SORT, convertLocalIndexMessageUuidSortKey(messageUuid))
-          .put(KEY_TTL, AttributeValues.fromLong(getTtlForMessage(message)));
+          .put(KEY_TTL, AttributeValues.fromLong(getTtlForMessage(message)))
-
+          .put(KEY_ENVELOPE_BYTES, AttributeValue.builder().b(SdkBytes.fromByteArray(message.toByteArray())).build());
-      if (dynamicConfigurationManager.getConfiguration().getMessageTableConfiguration().isWriteEnvelopes()) {
-        item.put(KEY_ENVELOPE_BYTES, AttributeValue.builder().b(SdkBytes.fromByteArray(message.toByteArray())).build());
-      } else {
-        item.put(KEY_TYPE, AttributeValues.fromInt(message.getType().getNumber()))
-            .put(KEY_TIMESTAMP, AttributeValues.fromLong(message.getTimestamp()))
-            .put(KEY_DESTINATION_UUID, AttributeValues.fromUUID(UUID.fromString(message.getDestinationUuid())));
-
-        if (message.hasUpdatedPni()) {
-          item.put(KEY_UPDATED_PNI, AttributeValues.fromUUID(UUID.fromString(message.getUpdatedPni())));
-        }
-        if (message.hasSource()) {
-          item.put(KEY_SOURCE, AttributeValues.fromString(message.getSource()));
-        }
-        if (message.hasSourceUuid()) {
-          item.put(KEY_SOURCE_UUID, AttributeValues.fromUUID(UUID.fromString(message.getSourceUuid())));
-        }
-        if (message.hasSourceDevice()) {
-          item.put(KEY_SOURCE_DEVICE, AttributeValues.fromInt(message.getSourceDevice()));
-        }
-        if (message.hasContent()) {
-          item.put(KEY_CONTENT, AttributeValues.fromByteArray(message.getContent().toByteArray()));
-        }
-      }
 
       writeItems.add(WriteRequest.builder().putRequest(PutRequest.builder()
           .item(item.build())
@@ -272,7 +262,8 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore {
     });
   }
 
-  private MessageProtos.Envelope convertItemToEnvelope(final Map<String, AttributeValue> item)
+  @VisibleForTesting
+  static MessageProtos.Envelope convertItemToEnvelope(final Map<String, AttributeValue> item)
       throws InvalidProtocolBufferException {
     final MessageProtos.Envelope envelope;
 

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

@@ -152,7 +152,7 @@ public class AssignUsernameCommand extends EnvironmentCommand<WhisperServerConfi
         configuration.getDynamoDbTables().getKeys().getTableName());
     MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(dynamoDbClient,
         configuration.getDynamoDbTables().getMessages().getTableName(),
-        configuration.getDynamoDbTables().getMessages().getExpiration(), dynamicConfigurationManager);
+        configuration.getDynamoDbTables().getMessages().getExpiration());
     FaultTolerantRedisCluster messageInsertCacheCluster = new FaultTolerantRedisCluster("message_insert_cluster",
         configuration.getMessageCacheConfiguration().getRedisClusterConfiguration(), redisClusterClientResources);
     FaultTolerantRedisCluster messageReadDeleteCluster = new FaultTolerantRedisCluster("message_read_delete_cluster",

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

@@ -155,7 +155,7 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
           configuration.getDynamoDbTables().getKeys().getTableName());
       MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(dynamoDbClient,
           configuration.getDynamoDbTables().getMessages().getTableName(),
-          configuration.getDynamoDbTables().getMessages().getExpiration(), dynamicConfigurationManager);
+          configuration.getDynamoDbTables().getMessages().getExpiration());
       FaultTolerantRedisCluster messageInsertCacheCluster = new FaultTolerantRedisCluster("message_insert_cluster",
           configuration.getMessageCacheConfiguration().getRedisClusterConfiguration(), redisClusterClientResources);
       FaultTolerantRedisCluster messageReadDeleteCluster = new FaultTolerantRedisCluster("message_read_delete_cluster",

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

@@ -158,7 +158,7 @@ public class SetUserDiscoverabilityCommand extends EnvironmentCommand<WhisperSer
           configuration.getDynamoDbTables().getKeys().getTableName());
       MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(dynamoDbClient,
           configuration.getDynamoDbTables().getMessages().getTableName(),
-          configuration.getDynamoDbTables().getMessages().getExpiration(), dynamicConfigurationManager);
+          configuration.getDynamoDbTables().getMessages().getExpiration());
       FaultTolerantRedisCluster messageInsertCacheCluster = new FaultTolerantRedisCluster("message_insert_cluster",
           configuration.getMessageCacheConfiguration().getRedisClusterConfiguration(), redisClusterClientResources);
       FaultTolerantRedisCluster messageReadDeleteCluster = new FaultTolerantRedisCluster("message_read_delete_cluster",

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

@@ -7,17 +7,18 @@ package org.whispersystems.textsecuregcm.storage;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
 import io.lettuce.core.cluster.SlotHash;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
@@ -31,13 +32,10 @@ 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.entities.MessageProtos.Envelope.Type;
 import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import org.whispersystems.textsecuregcm.tests.util.MessagesDynamoDbExtension;
-import org.whispersystems.textsecuregcm.util.AttributeValues;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
-import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
 import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
 
 class MessagePersisterIntegrationTest {
@@ -66,8 +64,10 @@ class MessagePersisterIntegrationTest {
     @SuppressWarnings("unchecked") final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager =
         mock(DynamicConfigurationManager.class);
 
+    when(dynamicConfigurationManager.getConfiguration()).thenReturn(new DynamicConfiguration());
+
     final MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(dynamoDbExtension.getDynamoDbClient(),
-        MessagesDynamoDbExtension.TABLE_NAME, Duration.ofDays(14), dynamicConfigurationManager);
+        MessagesDynamoDbExtension.TABLE_NAME, Duration.ofDays(14));
     final AccountsManager accountsManager = mock(AccountsManager.class);
 
     notificationExecutorService = Executors.newSingleThreadExecutor();
@@ -146,20 +146,19 @@ class MessagePersisterIntegrationTest {
 
       messagePersister.stop();
 
-      final List<MessageProtos.Envelope> persistedMessages = new ArrayList<>(messageCount);
-
       DynamoDbClient dynamoDB = dynamoDbExtension.getDynamoDbClient();
-      for (Map<String, AttributeValue> item : dynamoDB
+
-          .scan(ScanRequest.builder().tableName(MessagesDynamoDbExtension.TABLE_NAME).build()).items()) {
+      final List<MessageProtos.Envelope> persistedMessages =
-        persistedMessages.add(MessageProtos.Envelope.newBuilder()
+          dynamoDB.scan(ScanRequest.builder().tableName(MessagesDynamoDbExtension.TABLE_NAME).build()).items().stream()
-            .setServerGuid(AttributeValues.getUUID(item, "U", null).toString())
+              .map(item -> {
-            .setType(Type.forNumber(AttributeValues.getInt(item, "T", -1)))
+                try {
-            .setTimestamp(AttributeValues.getLong(item, "TS", -1))
+                  return MessagesDynamoDb.convertItemToEnvelope(item);
-            .setServerTimestamp(extractServerTimestamp(AttributeValues.getByteArray(item, "S", null)))
+                } catch (InvalidProtocolBufferException e) {
-            .setContent(ByteString.copyFrom(AttributeValues.getByteArray(item, "C", null)))
+                  fail("Could not parse stored message", e);
-            .setDestinationUuid(AttributeValues.getUUID(item, "DU", null).toString())
+                  return null;
-            .build());
+                }
-      }
+              })
+              .toList();
 
       assertEquals(expectedMessages, persistedMessages);
     });

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

@@ -6,8 +6,6 @@
 package org.whispersystems.textsecuregcm.storage;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 import com.google.protobuf.ByteString;
 import java.time.Duration;
@@ -17,10 +15,6 @@ import java.util.UUID;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
-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.DynamicMessageTableConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.tests.util.MessagesDynamoDbExtension;
 
@@ -67,7 +61,6 @@ class MessagesDynamoDbTest {
     MESSAGE3 = builder.build();
   }
 
-  private DynamicMessageTableConfiguration dynamicMessageTableConfiguration;
   private MessagesDynamoDb messagesDynamoDb;
 
 
@@ -76,24 +69,12 @@ class MessagesDynamoDbTest {
 
   @BeforeEach
   void setup() {
-    @SuppressWarnings("unchecked") final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager =
-        mock(DynamicConfigurationManager.class);
-
-    final DynamicConfiguration dynamicConfiguration = mock(DynamicConfiguration.class);
-    dynamicMessageTableConfiguration = mock(DynamicMessageTableConfiguration.class);
-
-    when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
-    when(dynamicConfiguration.getMessageTableConfiguration()).thenReturn(dynamicMessageTableConfiguration);
-
     messagesDynamoDb = new MessagesDynamoDb(dynamoDbExtension.getDynamoDbClient(), MessagesDynamoDbExtension.TABLE_NAME,
-        Duration.ofDays(14), dynamicConfigurationManager);
+        Duration.ofDays(14));
   }
 
-  @ParameterizedTest
+  @Test
-  @ValueSource(booleans = {true, false})
+  void testSimpleFetchAfterInsert() {
-  void testSimpleFetchAfterInsert(final boolean writeEnvelopes) {
-    when(dynamicMessageTableConfiguration.isWriteEnvelopes()).thenReturn(writeEnvelopes);
-
     final UUID destinationUuid = UUID.randomUUID();
     final int destinationDeviceId = random.nextInt(255) + 1;
     messagesDynamoDb.store(List.of(MESSAGE1, MESSAGE2, MESSAGE3), destinationUuid, destinationDeviceId);

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

@@ -41,8 +41,6 @@ import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.ArgumentCaptor;
 import org.mockito.stubbing.Answer;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
-import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
-import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicMessageTableConfiguration;
 import org.whispersystems.textsecuregcm.entities.MessageProtos;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
 import org.whispersystems.textsecuregcm.metrics.PushLatencyManager;
@@ -50,7 +48,6 @@ import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.redis.RedisClusterExtension;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.Device;
-import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
 import org.whispersystems.textsecuregcm.storage.DynamoDbExtension;
 import org.whispersystems.textsecuregcm.storage.MessagesCache;
 import org.whispersystems.textsecuregcm.storage.MessagesDynamoDb;
@@ -86,21 +83,11 @@ class WebSocketConnectionIntegrationTest {
   @BeforeEach
   void setUp() throws Exception {
 
-    @SuppressWarnings("unchecked") final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager =
-        mock(DynamicConfigurationManager.class);
-
-    final DynamicConfiguration dynamicConfiguration = mock(DynamicConfiguration.class);
-    final DynamicMessageTableConfiguration dynamicMessageTableConfiguration = mock(DynamicMessageTableConfiguration.class);
-
-    when(dynamicConfigurationManager.getConfiguration()).thenReturn(dynamicConfiguration);
-    when(dynamicConfiguration.getMessageTableConfiguration()).thenReturn(dynamicMessageTableConfiguration);
-
-
     executorService = Executors.newSingleThreadExecutor();
     messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(),
         REDIS_CLUSTER_EXTENSION.getRedisCluster(), executorService);
     messagesDynamoDb = new MessagesDynamoDb(dynamoDbExtension.getDynamoDbClient(), MessagesDynamoDbExtension.TABLE_NAME,
-        Duration.ofDays(7), dynamicConfigurationManager);
+        Duration.ofDays(7));
     reportMessageManager = mock(ReportMessageManager.class);
     account = mock(Account.class);
     device = mock(Device.class);