Make stored messages REST accessible.

Add REST endpoints for retrieving and acknowledging pending
messges, beyond the WebSocket.

// FREEBIE

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

@@ -58,6 +58,7 @@ import org.whispersystems.textsecuregcm.providers.TimeProvider;
 import org.whispersystems.textsecuregcm.push.FeedbackHandler;
 import org.whispersystems.textsecuregcm.push.PushSender;
 import org.whispersystems.textsecuregcm.push.PushServiceClient;
+import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.push.WebsocketSender;
 import org.whispersystems.textsecuregcm.sms.NexmoSmsSender;
 import org.whispersystems.textsecuregcm.sms.SmsSender;
@@ -174,6 +175,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     SmsSender                smsSender           = new SmsSender(twilioSmsSender, nexmoSmsSender, config.getTwilioConfiguration().isInternational());
     UrlSigner                urlSigner           = new UrlSigner(config.getS3Configuration());
     PushSender               pushSender          = new PushSender(pushServiceClient, websocketSender);
+    ReceiptSender            receiptSender       = new ReceiptSender(accountsManager, pushSender, federatedClientManager);
     FeedbackHandler          feedbackHandler     = new FeedbackHandler(pushServiceClient, accountsManager);
     Optional<byte[]>         authorizationKey    = config.getRedphoneConfiguration().getAuthorizationKey();
 
@@ -183,7 +185,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     AttachmentController attachmentController = new AttachmentController(rateLimiters, federatedClientManager, urlSigner);
     KeysControllerV1     keysControllerV1     = new KeysControllerV1(rateLimiters, keys, accountsManager, federatedClientManager);
     KeysControllerV2     keysControllerV2     = new KeysControllerV2(rateLimiters, keys, accountsManager, federatedClientManager);
-    MessageController    messageController    = new MessageController(rateLimiters, pushSender, accountsManager, federatedClientManager);
+    MessageController    messageController    = new MessageController(rateLimiters, pushSender, receiptSender, accountsManager, messagesManager, federatedClientManager);
 
     environment.jersey().register(new MultiBasicAuthProvider<>(new FederatedPeerAuthenticator(config.getFederationConfiguration()),
                                                                FederatedPeer.class,
@@ -195,7 +197,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.jersey().register(new DirectoryController(rateLimiters, directory));
     environment.jersey().register(new FederationControllerV1(accountsManager, attachmentController, messageController, keysControllerV1));
     environment.jersey().register(new FederationControllerV2(accountsManager, attachmentController, messageController, keysControllerV2));
-    environment.jersey().register(new ReceiptController(accountsManager, federatedClientManager, pushSender));
+    environment.jersey().register(new ReceiptController(receiptSender));
     environment.jersey().register(new ProvisioningController(rateLimiters, pushSender));
     environment.jersey().register(attachmentController);
     environment.jersey().register(keysControllerV1);
@@ -205,7 +207,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     if (config.getWebsocketConfiguration().isEnabled()) {
       WebSocketEnvironment webSocketEnvironment = new WebSocketEnvironment(environment, config, 90000);
       webSocketEnvironment.setAuthenticator(new WebSocketAccountAuthenticator(deviceAuthenticator));
-      webSocketEnvironment.setConnectListener(new AuthenticatedConnectListener(accountsManager, pushSender, messagesManager, pubSubManager));
+      webSocketEnvironment.setConnectListener(new AuthenticatedConnectListener(accountsManager, pushSender, receiptSender, messagesManager, pubSubManager));
       webSocketEnvironment.jersey().register(new KeepAliveController(pubSubManager));
 
       WebSocketEnvironment provisioningEnvironment = new WebSocketEnvironment(environment, config);

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

@@ -26,6 +26,8 @@ import org.whispersystems.textsecuregcm.entities.IncomingMessageList;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.OutgoingMessageSignal;
 import org.whispersystems.textsecuregcm.entities.MessageResponse;
 import org.whispersystems.textsecuregcm.entities.MismatchedDevices;
+import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity;
+import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntityList;
 import org.whispersystems.textsecuregcm.entities.SendMessageResponse;
 import org.whispersystems.textsecuregcm.entities.StaleDevices;
 import org.whispersystems.textsecuregcm.federation.FederatedClient;
@@ -34,15 +36,19 @@ import org.whispersystems.textsecuregcm.federation.NoSuchPeerException;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.push.NotPushRegisteredException;
 import org.whispersystems.textsecuregcm.push.PushSender;
+import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.push.TransientPushFailureException;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
+import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.util.Base64;
 import org.whispersystems.textsecuregcm.util.Util;
 
 import javax.validation.Valid;
 import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
@@ -66,17 +72,23 @@ public class MessageController {
 
   private final RateLimiters           rateLimiters;
   private final PushSender             pushSender;
+  private final ReceiptSender          receiptSender;
   private final FederatedClientManager federatedClientManager;
   private final AccountsManager        accountsManager;
+  private final MessagesManager        messagesManager;
 
   public MessageController(RateLimiters rateLimiters,
                            PushSender pushSender,
+                           ReceiptSender receiptSender,
                            AccountsManager accountsManager,
+                           MessagesManager messagesManager,
                            FederatedClientManager federatedClientManager)
   {
     this.rateLimiters           = rateLimiters;
     this.pushSender             = pushSender;
+    this.receiptSender          = receiptSender;
     this.accountsManager        = accountsManager;
+    this.messagesManager        = messagesManager;
     this.federatedClientManager = federatedClientManager;
   }
 
@@ -137,6 +149,36 @@ public class MessageController {
     }
   }
 
+  @Timed
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  public OutgoingMessageEntityList getPendingMessages(@Auth Account account) {
+    return new OutgoingMessageEntityList(messagesManager.getMessagesForDevice(account.getNumber(),
+                                                                              account.getAuthenticatedDevice()
+                                                                                     .get().getId()));
+  }
+
+  @Timed
+  @DELETE
+  @Path("/{message_id}")
+  public void removePendingMessage(@Auth Account account, @PathParam("message_id") long id)
+      throws IOException
+  {
+    try {
+      Optional<OutgoingMessageEntity> message = messagesManager.delete(account.getNumber(), id);
+
+      if (message.isPresent() && message.get().getType() != OutgoingMessageSignal.Type.RECEIPT_VALUE) {
+        receiptSender.sendReceipt(account,
+                                  message.get().getSource(),
+                                  message.get().getTimestamp(),
+                                  Optional.fromNullable(message.get().getRelay()));
+      }
+    } catch (NoSuchUserException | NotPushRegisteredException | TransientPushFailureException e) {
+      logger.warn("Sending delivery receipt", e);
+    }
+  }
+
+
   private void sendLocalMessage(Account source,
                                 String destinationName,
                                 IncomingMessageList messages,

src/main/java/org/whispersystems/textsecuregcm/controllers/ReceiptController.java

@@ -2,14 +2,10 @@ package org.whispersystems.textsecuregcm.controllers;
 
 import com.codahale.metrics.annotation.Timed;
 import com.google.common.base.Optional;
-import org.whispersystems.textsecuregcm.federation.FederatedClientManager;
-import org.whispersystems.textsecuregcm.federation.NoSuchPeerException;
 import org.whispersystems.textsecuregcm.push.NotPushRegisteredException;
-import org.whispersystems.textsecuregcm.push.PushSender;
+import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.push.TransientPushFailureException;
 import org.whispersystems.textsecuregcm.storage.Account;
-import org.whispersystems.textsecuregcm.storage.AccountsManager;
-import org.whispersystems.textsecuregcm.storage.Device;
 
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
@@ -18,25 +14,16 @@ import javax.ws.rs.QueryParam;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.Response;
 import java.io.IOException;
-import java.util.Set;
 
 import io.dropwizard.auth.Auth;
-import static org.whispersystems.textsecuregcm.entities.MessageProtos.OutgoingMessageSignal;
 
 @Path("/v1/receipt")
 public class ReceiptController {
 
-  private final AccountsManager        accountManager;
-  private final PushSender             pushSender;
-  private final FederatedClientManager federatedClientManager;
+  private final ReceiptSender receiptSender;
 
-  public ReceiptController(AccountsManager accountManager,
-                           FederatedClientManager federatedClientManager,
-                           PushSender pushSender)
-  {
-    this.accountManager         = accountManager;
-    this.federatedClientManager = federatedClientManager;
-    this.pushSender             = pushSender;
+  public ReceiptController(ReceiptSender receiptSender) {
+    this.receiptSender = receiptSender;
   }
 
   @Timed
@@ -49,11 +36,7 @@ public class ReceiptController {
       throws IOException
   {
     try {
-      if (relay.isPresent() && !relay.get().isEmpty()) {
-        sendRelayedReceipt(source, destination, messageId, relay.get());
-      } else {
-        sendDirectReceipt(source, destination, messageId);
-      }
+      receiptSender.sendReceipt(source, destination, messageId, relay);
     } catch (NoSuchUserException | NotPushRegisteredException e) {
       throw new WebApplicationException(Response.Status.NOT_FOUND);
     } catch (TransientPushFailureException e) {
@@ -61,51 +44,4 @@ public class ReceiptController {
     }
   }
 
-  private void sendRelayedReceipt(Account source, String destination, long messageId, String relay)
-      throws NoSuchUserException, IOException
-  {
-    try {
-      federatedClientManager.getClient(relay)
-                            .sendDeliveryReceipt(source.getNumber(),
-                                                 source.getAuthenticatedDevice().get().getId(),
-                                                 destination, messageId);
-    } catch (NoSuchPeerException e) {
-      throw new NoSuchUserException(e);
-    }
-  }
-
-  private void sendDirectReceipt(Account source, String destination, long messageId)
-      throws NotPushRegisteredException, TransientPushFailureException, NoSuchUserException
-  {
-    Account     destinationAccount = getDestinationAccount(destination);
-    Set<Device> destinationDevices = destinationAccount.getDevices();
-
-    OutgoingMessageSignal.Builder message =
-        OutgoingMessageSignal.newBuilder()
-                             .setSource(source.getNumber())
-                             .setSourceDevice((int) source.getAuthenticatedDevice().get().getId())
-                             .setTimestamp(messageId)
-                             .setType(OutgoingMessageSignal.Type.RECEIPT_VALUE);
-
-    if (source.getRelay().isPresent()) {
-      message.setRelay(source.getRelay().get());
-    }
-
-    for (Device destinationDevice : destinationDevices) {
-      pushSender.sendMessage(destinationAccount, destinationDevice, message.build());
-    }
-  }
-
-  private Account getDestinationAccount(String destination)
-      throws NoSuchUserException
-  {
-    Optional<Account> account = accountManager.get(destination);
-
-    if (!account.isPresent()) {
-      throw new NoSuchUserException(destination);
-    }
-
-    return account.get();
-  }
-
 }

src/main/java/org/whispersystems/textsecuregcm/entities/OutgoingMessageEntity.java

@@ -0,0 +1,70 @@
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class OutgoingMessageEntity {
+
+  @JsonIgnore
+  private long id;
+
+  @JsonProperty
+  private int type;
+
+  @JsonProperty
+  private String relay;
+
+  @JsonProperty
+  private long timestamp;
+
+  @JsonProperty
+  private String source;
+
+  @JsonProperty
+  private int sourceDevice;
+
+  @JsonProperty
+  private byte[] message;
+
+  public OutgoingMessageEntity() {}
+
+  public OutgoingMessageEntity(long id, int type, String relay, long timestamp,
+                               String source, int sourceDevice, byte[] message)
+  {
+    this.id           = id;
+    this.type         = type;
+    this.relay        = relay;
+    this.timestamp    = timestamp;
+    this.source       = source;
+    this.sourceDevice = sourceDevice;
+    this.message      = message;
+  }
+
+  public int getType() {
+    return type;
+  }
+
+  public String getRelay() {
+    return relay;
+  }
+
+  public long getTimestamp() {
+    return timestamp;
+  }
+
+  public String getSource() {
+    return source;
+  }
+
+  public int getSourceDevice() {
+    return sourceDevice;
+  }
+
+  public byte[] getMessage() {
+    return message;
+  }
+
+  public long getId() {
+    return id;
+  }
+}

src/main/java/org/whispersystems/textsecuregcm/entities/OutgoingMessageEntityList.java

@@ -0,0 +1,23 @@
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.annotations.VisibleForTesting;
+
+import java.util.List;
+
+public class OutgoingMessageEntityList {
+
+  @JsonProperty
+  private List<OutgoingMessageEntity> messages;
+
+  public OutgoingMessageEntityList() {}
+
+  public OutgoingMessageEntityList(List<OutgoingMessageEntity> messages) {
+    this.messages = messages;
+  }
+
+  @VisibleForTesting
+  public List<OutgoingMessageEntity> getMessages() {
+    return messages;
+  }
+}

src/main/java/org/whispersystems/textsecuregcm/push/ReceiptSender.java

@@ -0,0 +1,89 @@
+package org.whispersystems.textsecuregcm.push;
+
+import com.google.common.base.Optional;
+import org.whispersystems.textsecuregcm.controllers.NoSuchUserException;
+import org.whispersystems.textsecuregcm.entities.MessageProtos;
+import org.whispersystems.textsecuregcm.federation.FederatedClientManager;
+import org.whispersystems.textsecuregcm.federation.NoSuchPeerException;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.Device;
+
+import java.io.IOException;
+import java.util.Set;
+
+public class ReceiptSender {
+
+  private final PushSender             pushSender;
+  private final FederatedClientManager federatedClientManager;
+  private final AccountsManager        accountManager;
+
+  public ReceiptSender(AccountsManager        accountManager,
+                       PushSender             pushSender,
+                       FederatedClientManager federatedClientManager)
+  {
+    this.federatedClientManager = federatedClientManager;
+    this.accountManager         = accountManager;
+    this.pushSender             = pushSender;
+  }
+
+  public void sendReceipt(Account source, String destination,
+                          long messageId, Optional<String> relay)
+      throws IOException, NoSuchUserException,
+             NotPushRegisteredException, TransientPushFailureException
+  {
+    if (relay.isPresent() && !relay.get().isEmpty()) {
+      sendRelayedReceipt(source, destination, messageId, relay.get());
+    } else {
+      sendDirectReceipt(source, destination, messageId);
+    }
+  }
+
+  private void sendRelayedReceipt(Account source, String destination, long messageId, String relay)
+      throws NoSuchUserException, IOException
+  {
+    try {
+      federatedClientManager.getClient(relay)
+                            .sendDeliveryReceipt(source.getNumber(),
+                                                 source.getAuthenticatedDevice().get().getId(),
+                                                 destination, messageId);
+    } catch (NoSuchPeerException e) {
+      throw new NoSuchUserException(e);
+    }
+  }
+
+  private void sendDirectReceipt(Account source, String destination, long messageId)
+      throws NotPushRegisteredException, TransientPushFailureException, NoSuchUserException
+  {
+    Account     destinationAccount = getDestinationAccount(destination);
+    Set<Device> destinationDevices = destinationAccount.getDevices();
+
+    MessageProtos.OutgoingMessageSignal.Builder message =
+        MessageProtos.OutgoingMessageSignal.newBuilder()
+                                           .setSource(source.getNumber())
+                                           .setSourceDevice((int) source.getAuthenticatedDevice().get().getId())
+                                           .setTimestamp(messageId)
+                                           .setType(MessageProtos.OutgoingMessageSignal.Type.RECEIPT_VALUE);
+
+    if (source.getRelay().isPresent()) {
+      message.setRelay(source.getRelay().get());
+    }
+
+    for (Device destinationDevice : destinationDevices) {
+      pushSender.sendMessage(destinationAccount, destinationDevice, message.build());
+    }
+  }
+
+  private Account getDestinationAccount(String destination)
+      throws NoSuchUserException
+  {
+    Optional<Account> account = accountManager.get(destination);
+
+    if (!account.isPresent()) {
+      throw new NoSuchUserException(destination);
+    }
+
+    return account.get();
+  }
+
+}

src/main/java/org/whispersystems/textsecuregcm/storage/Messages.java

@@ -1,6 +1,5 @@
 package org.whispersystems.textsecuregcm.storage;
 
-import com.google.protobuf.ByteString;
 import org.skife.jdbi.v2.SQLStatement;
 import org.skife.jdbi.v2.StatementContext;
 import org.skife.jdbi.v2.sqlobject.Bind;
@@ -12,7 +11,7 @@ import org.skife.jdbi.v2.sqlobject.SqlUpdate;
 import org.skife.jdbi.v2.sqlobject.customizers.Mapper;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.OutgoingMessageSignal;
-import org.whispersystems.textsecuregcm.util.Pair;
+import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity;
 
 import java.lang.annotation.Annotation;
 import java.lang.annotation.ElementType;
@@ -44,9 +43,14 @@ public abstract class Messages {
 
   @Mapper(MessageMapper.class)
   @SqlQuery("SELECT * FROM messages WHERE " + DESTINATION + " = :destination AND " + DESTINATION_DEVICE + " = :destination_device ORDER BY " + TIMESTAMP + " ASC")
-  abstract List<Pair<Long, OutgoingMessageSignal>> load(@Bind("destination") String destination,
-                                                        @Bind("destination_device") long destinationDevice);
+  abstract List<OutgoingMessageEntity> load(@Bind("destination")        String destination,
+                                            @Bind("destination_device") long destinationDevice);
 
+  @Mapper(MessageMapper.class)
+  @SqlQuery("DELETE FROM messages WHERE " + ID + " IN (SELECT " + ID + " FROM messages WHERE " + DESTINATION + " = :destination AND " + TIMESTAMP + " = :timestamp ORDER BY " + ID + " LIMIT 1) RETURNING *")
+  abstract OutgoingMessageEntity remove(@Bind("destination") String destination, @Bind("timestamp") long timestamp);
+
+  @Mapper(MessageMapper.class)
   @SqlUpdate("DELETE FROM messages WHERE " + ID + " = :id")
   abstract void remove(@Bind("id") long id);
 
@@ -56,20 +60,18 @@ public abstract class Messages {
   @SqlUpdate("VACUUM messages")
   public abstract void vacuum();
 
-  public static class MessageMapper implements ResultSetMapper<Pair<Long, OutgoingMessageSignal>> {
+  public static class MessageMapper implements ResultSetMapper<OutgoingMessageEntity> {
     @Override
-    public Pair<Long, OutgoingMessageSignal> map(int i, ResultSet resultSet, StatementContext statementContext)
+    public OutgoingMessageEntity map(int i, ResultSet resultSet, StatementContext statementContext)
         throws SQLException
     {
-      return new Pair<>(resultSet.getLong(ID),
-                        OutgoingMessageSignal.newBuilder()
-                                             .setType(resultSet.getInt(TYPE))
-                                             .setRelay(resultSet.getString(RELAY))
-                                             .setTimestamp(resultSet.getLong(TIMESTAMP))
-                                             .setSource(resultSet.getString(SOURCE))
-                                             .setSourceDevice(resultSet.getInt(SOURCE_DEVICE))
-                                             .setMessage(ByteString.copyFrom(resultSet.getBytes(MESSAGE)))
-                                             .build());
+      return new OutgoingMessageEntity(resultSet.getLong(ID),
+                                       resultSet.getInt(TYPE),
+                                       resultSet.getString(RELAY),
+                                       resultSet.getLong(TIMESTAMP),
+                                       resultSet.getString(SOURCE),
+                                       resultSet.getInt(SOURCE_DEVICE),
+                                       resultSet.getBytes(MESSAGE));
     }
   }
 

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

@@ -1,7 +1,9 @@
 package org.whispersystems.textsecuregcm.storage;
 
 
+import com.google.common.base.Optional;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.OutgoingMessageSignal;
+import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity;
 import org.whispersystems.textsecuregcm.util.Pair;
 
 import java.util.List;
@@ -18,7 +20,7 @@ public class MessagesManager {
     return this.messages.store(message, destination, destinationDevice) + 1;
   }
 
-  public List<Pair<Long, OutgoingMessageSignal>> getMessagesForDevice(String destination, long destinationDevice) {
+  public List<OutgoingMessageEntity> getMessagesForDevice(String destination, long destinationDevice) {
     return this.messages.load(destination, destinationDevice);
   }
 
@@ -26,6 +28,10 @@ public class MessagesManager {
     this.messages.clear(destination);
   }
 
+  public Optional<OutgoingMessageEntity> delete(String destination, long timestamp) {
+    return Optional.fromNullable(this.messages.remove(destination, timestamp));
+  }
+
   public void delete(long id) {
     this.messages.remove(id);
   }

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

@@ -6,6 +6,7 @@ import com.codahale.metrics.SharedMetricRegistries;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.push.PushSender;
+import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
@@ -27,14 +28,17 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
 
   private final AccountsManager accountsManager;
   private final PushSender      pushSender;
+  private final ReceiptSender   receiptSender;
   private final MessagesManager messagesManager;
   private final PubSubManager   pubSubManager;
 
   public AuthenticatedConnectListener(AccountsManager accountsManager, PushSender pushSender,
-                                      MessagesManager messagesManager, PubSubManager pubSubManager)
+                                      ReceiptSender receiptSender,  MessagesManager messagesManager,
+                                      PubSubManager pubSubManager)
   {
     this.accountsManager = accountsManager;
     this.pushSender      = pushSender;
+    this.receiptSender   = receiptSender;
     this.messagesManager = messagesManager;
     this.pubSubManager   = pubSubManager;
   }
@@ -45,7 +49,7 @@ public class AuthenticatedConnectListener implements WebSocketConnectListener {
     final Device              device      = account.getAuthenticatedDevice().get();
     final long                connectTime = System.currentTimeMillis();
     final WebsocketAddress    address     = new WebsocketAddress(account.getNumber(), device.getId());
-    final WebSocketConnection connection  = new WebSocketConnection(accountsManager, pushSender,
+    final WebSocketConnection connection  = new WebSocketConnection(pushSender, receiptSender,
                                                                     messagesManager, account, device,
                                                                     context.getClient());
 

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

@@ -1,35 +1,33 @@
 package org.whispersystems.textsecuregcm.websocket;
 
-import com.codahale.metrics.Histogram;
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.SharedMetricRegistries;
 import com.google.common.base.Optional;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.dispatch.DispatchChannel;
+import org.whispersystems.textsecuregcm.controllers.NoSuchUserException;
 import org.whispersystems.textsecuregcm.entities.CryptoEncodingException;
 import org.whispersystems.textsecuregcm.entities.EncryptedOutgoingMessage;
+import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity;
 import org.whispersystems.textsecuregcm.push.NotPushRegisteredException;
 import org.whispersystems.textsecuregcm.push.PushSender;
+import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.push.TransientPushFailureException;
 import org.whispersystems.textsecuregcm.storage.Account;
-import org.whispersystems.textsecuregcm.storage.AccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
-import org.whispersystems.textsecuregcm.util.Constants;
-import org.whispersystems.textsecuregcm.util.Pair;
 import org.whispersystems.websocket.WebSocketClient;
 import org.whispersystems.websocket.messages.WebSocketResponseMessage;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.io.IOException;
 import java.util.List;
 
-import static com.codahale.metrics.MetricRegistry.name;
 import static org.whispersystems.textsecuregcm.entities.MessageProtos.OutgoingMessageSignal;
 import static org.whispersystems.textsecuregcm.storage.PubSubProtos.PubSubMessage;
 
@@ -37,7 +35,7 @@ public class WebSocketConnection implements DispatchChannel {
 
   private static final Logger logger = LoggerFactory.getLogger(WebSocketConnection.class);
 
-  private final AccountsManager  accountsManager;
+  private final ReceiptSender    receiptSender;
   private final PushSender       pushSender;
   private final MessagesManager  messagesManager;
 
@@ -45,17 +43,15 @@ public class WebSocketConnection implements DispatchChannel {
   private final Device           device;
   private final WebSocketClient  client;
 
-  private long connectionStartTime;
-
-  public WebSocketConnection(AccountsManager accountsManager,
-                             PushSender pushSender,
+  public WebSocketConnection(PushSender pushSender,
+                             ReceiptSender receiptSender,
                              MessagesManager messagesManager,
                              Account account,
                              Device device,
                              WebSocketClient client)
   {
-    this.accountsManager = accountsManager;
     this.pushSender      = pushSender;
+    this.receiptSender   = receiptSender;
     this.messagesManager = messagesManager;
     this.account         = account;
     this.device          = device;
@@ -107,7 +103,7 @@ public class WebSocketConnection implements DispatchChannel {
           if (isSuccessResponse(response)) {
             if (storedMessageId.isPresent()) messagesManager.delete(storedMessageId.get());
             if (!isReceipt)                  sendDeliveryReceiptFor(message);
-          } else if (!isSuccessResponse(response) & !storedMessageId.isPresent()) {
+          } else if (!isSuccessResponse(response) && !storedMessageId.isPresent()) {
             requeueMessage(message);
           }
         }
@@ -137,34 +133,30 @@ public class WebSocketConnection implements DispatchChannel {
 
   private void sendDeliveryReceiptFor(OutgoingMessageSignal message) {
     try {
-      Optional<Account> source = accountsManager.get(message.getSource());
-
-      if (!source.isPresent()) {
-        logger.warn(String.format("Source account disappeared? (%s)", message.getSource()));
-        return;
-      }
-
-      OutgoingMessageSignal.Builder receipt =
-          OutgoingMessageSignal.newBuilder()
-                               .setSource(account.getNumber())
-                               .setSourceDevice((int) device.getId())
-                               .setTimestamp(message.getTimestamp())
-                               .setType(OutgoingMessageSignal.Type.RECEIPT_VALUE);
-
-      for (Device device : source.get().getDevices()) {
-        pushSender.sendMessage(source.get(), device, receipt.build());
-      }
-    } catch (NotPushRegisteredException | TransientPushFailureException e) {
-      logger.warn("sendDeliveryReceiptFor", "Delivery receipet", e);
+      receiptSender.sendReceipt(account, message.getSource(), message.getTimestamp(),
+                                message.hasRelay() ? Optional.of(message.getRelay()) :
+                                                     Optional.<String>absent());
+    } catch (IOException | NoSuchUserException | TransientPushFailureException | NotPushRegisteredException e) {
+      logger.warn("sendDeliveryReceiptFor", e);
     }
   }
 
   private void processStoredMessages() {
-    List<Pair<Long, OutgoingMessageSignal>> messages = messagesManager.getMessagesForDevice(account.getNumber(),
-                                                                                            device.getId());
+    List<OutgoingMessageEntity> messages = messagesManager.getMessagesForDevice(account.getNumber(), device.getId());
 
-    for (Pair<Long, OutgoingMessageSignal> message : messages) {
-      sendMessage(message.second(), Optional.of(message.first()));
+    for (OutgoingMessageEntity message : messages) {
+      OutgoingMessageSignal.Builder builder = OutgoingMessageSignal.newBuilder()
+                                                                   .setType(message.getType())
+                                                                   .setMessage(ByteString.copyFrom(message.getMessage()))
+                                                                   .setSourceDevice(message.getSourceDevice())
+                                                                   .setSource(message.getSource())
+                                                                   .setTimestamp(message.getTimestamp());
+
+      if (message.getRelay() != null && !message.getRelay().isEmpty()) {
+        builder.setRelay(message.getRelay());
+      }
+
+      sendMessage(builder.build(), Optional.of(message.getId()));
     }
   }