Support for synchronizing read state to/from desktop

// FREEBIE

build.gradle

@@ -72,7 +72,7 @@ dependencies {
     compile 'org.whispersystems:jobmanager:1.0.2'
     compile 'org.whispersystems:libpastelog:1.0.7'
     compile 'com.amulyakhare:com.amulyakhare.textdrawable:1.0.1'
-    compile 'org.whispersystems:textsecure-android:1.8.6'
+    compile 'org.whispersystems:textsecure-android:1.8.7'
     compile 'com.h6ah4i.android.compat:mulsellistprefcompat:1.0.0'
     compile 'com.google.zxing:core:3.2.1'
 
@@ -126,7 +126,7 @@ dependencyVerification {
             'org.whispersystems:jobmanager:506f679fc2fcf7bb6d10f00f41d6f6ea0abf75c70dc95b913398661ad538a181',
             'org.whispersystems:libpastelog:bb331d9a98240fc139101128ba836c1edec3c40e000597cdbb29ebf4cbf34d88',
             'com.amulyakhare:com.amulyakhare.textdrawable:54c92b5fba38cfd316a07e5a30528068f45ce8515a6890f1297df4c401af5dcb',
-            'org.whispersystems:textsecure-android:0405821f479985677d6d5f4032eaaa732e04562c1975969cbaea37939030ec96',
+            'org.whispersystems:textsecure-android:5aa9fe94799570da35c8ff2faf517924ca602284971c60a5a7208818d6d00df4',
             'com.h6ah4i.android.compat:mulsellistprefcompat:47167c5cb796de1a854788e9ff318358e36c8fb88123baaa6e38fb78511dfabe',
             'com.google.zxing:core:b4d82452e7a6bf6ec2698904b332431717ed8f9a850224f295aec89de80f2259',
             'com.google.android.gms:play-services-base:ef36e50fa5c0415ed41f74dd399a889efd2fa327c449036e140c7c3786aa0e1f',
@@ -134,7 +134,7 @@ dependencyVerification {
             'com.nineoldandroids:library:68025a14e3e7673d6ad2f95e4b46d78d7d068343aa99256b686fe59de1b3163a',
             'javax.inject:javax.inject:91c77044a50c481636c32d916fd89c9118a72195390452c81065080f957de7ff',
             'com.madgag.spongycastle:core:8d6240b974b0aca4d3da9c7dd44d42339d8a374358aca5fc98e50a995764511f',
-            'org.whispersystems:textsecure-java:d4ee0d0537693f82b7a9f76fe32cf6b61f79e27d8657a486ee4959b54b738c5a',
+            'org.whispersystems:textsecure-java:d9e366c2ff9ed208d0fd2dd76e9097604425b2824739e59057b6afef0fd34e3d',
             'org.whispersystems:axolotl-android:40d3db5004a84749a73f68d2f0d01b2ae35a73c54df96d8c6c6723b96efb6fc0',
             'com.google.android.gms:play-services-basement:e1d29b21e02fd2a63e5a31807415cbb17a59568e27e3254181c01ffae10659bf',
             'com.googlecode.libphonenumber:libphonenumber:9625de9d2270e9a280ff4e6d9ef3106573fb4828773fd32c9b7614f4e17d2811',

src/org/thoughtcrime/securesms/ConversationActivity.java

@@ -90,9 +90,12 @@ import org.thoughtcrime.securesms.database.DraftDatabase;
 import org.thoughtcrime.securesms.database.DraftDatabase.Draft;
 import org.thoughtcrime.securesms.database.DraftDatabase.Drafts;
 import org.thoughtcrime.securesms.database.GroupDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.MmsSmsColumns.Types;
 import org.thoughtcrime.securesms.database.RecipientPreferenceDatabase.RecipientsPreferences;
 import org.thoughtcrime.securesms.database.ThreadDatabase;
+import org.thoughtcrime.securesms.jobs.MultiDeviceReadUpdateJob;
 import org.thoughtcrime.securesms.mms.AttachmentManager;
 import org.thoughtcrime.securesms.mms.AttachmentManager.MediaType;
 import org.thoughtcrime.securesms.mms.AttachmentTypeSelectorAdapter;
@@ -1232,8 +1235,15 @@ public class ConversationActivity extends PassphraseRequiredActionBarActivity
     new AsyncTask<Long, Void, Void>() {
       @Override
       protected Void doInBackground(Long... params) {
-        DatabaseFactory.getThreadDatabase(ConversationActivity.this).setRead(params[0]);
+        Context             context    = ConversationActivity.this;
-        MessageNotifier.updateNotification(ConversationActivity.this, masterSecret);
+        List<SyncMessageId> messageIds = DatabaseFactory.getThreadDatabase(context).setRead(params[0]);
+
+        MessageNotifier.updateNotification(context, masterSecret);
+
+        if (!messageIds.isEmpty()) {
+          ApplicationContext.getInstance(context).getJobManager().add(new MultiDeviceReadUpdateJob(context, messageIds));
+        }
+
         return null;
       }
     }.execute(threadId);

src/org/thoughtcrime/securesms/ConversationListFragment.java

@@ -63,7 +63,10 @@ import org.thoughtcrime.securesms.components.reminder.ShareReminder;
 import org.thoughtcrime.securesms.components.reminder.SystemSmsImportReminder;
 import org.thoughtcrime.securesms.crypto.MasterSecret;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.MessagingDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.loaders.ConversationListLoader;
+import org.thoughtcrime.securesms.jobs.MultiDeviceReadUpdateJob;
 import org.thoughtcrime.securesms.notifications.MessageNotifier;
 import org.thoughtcrime.securesms.recipients.Recipients;
 import org.thoughtcrime.securesms.util.Util;
@@ -72,6 +75,7 @@ import org.thoughtcrime.securesms.util.task.SnackbarAsyncTask;
 import org.whispersystems.libaxolotl.util.guava.Optional;
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Locale;
 import java.util.Set;
 
@@ -468,8 +472,14 @@ public class ConversationListFragment extends Fragment
             DatabaseFactory.getThreadDatabase(getActivity()).archiveConversation(threadId);
 
             if (!read) {
-              DatabaseFactory.getThreadDatabase(getActivity()).setRead(threadId);
+              List<SyncMessageId> messageIds = DatabaseFactory.getThreadDatabase(getActivity()).setRead(threadId);
               MessageNotifier.updateNotification(getActivity(), masterSecret);
+
+              if (!messageIds.isEmpty()) {
+                ApplicationContext.getInstance(getActivity())
+                                  .getJobManager()
+                                  .add(new MultiDeviceReadUpdateJob(getActivity(), messageIds));
+              }
             }
           }
 

src/org/thoughtcrime/securesms/database/MessagingDatabase.java

@@ -165,4 +165,23 @@ public abstract class MessagingDatabase extends Database implements MmsSmsColumn
         cursor.close();
     }
   }
+
+  public static class SyncMessageId {
+
+    private final String address;
+    private final long   timetamp;
+
+    public SyncMessageId(String address, long timetamp) {
+      this.address = address;
+      this.timetamp = timetamp;
+    }
+
+    public String getAddress() {
+      return address;
+    }
+
+    public long getTimetamp() {
+      return timetamp;
+    }
+  }
 }

src/org/thoughtcrime/securesms/database/MmsDatabase.java

@@ -190,14 +190,14 @@ public class MmsDatabase extends MessagingDatabase {
     }
   }
 
-  public void incrementDeliveryReceiptCount(String address, long timestamp) {
+  public void incrementDeliveryReceiptCount(SyncMessageId messageId) {
     MmsAddressDatabase addressDatabase = DatabaseFactory.getMmsAddressDatabase(context);
     SQLiteDatabase     database        = databaseHelper.getWritableDatabase();
     Cursor             cursor          = null;
     boolean            found           = false;
 
     try {
-      cursor = database.query(TABLE_NAME, new String[] {ID, THREAD_ID, MESSAGE_BOX}, DATE_SENT + " = ?", new String[] {String.valueOf(timestamp)}, null, null, null, null);
+      cursor = database.query(TABLE_NAME, new String[] {ID, THREAD_ID, MESSAGE_BOX}, DATE_SENT + " = ?", new String[] {String.valueOf(messageId.getTimetamp())}, null, null, null, null);
 
       while (cursor.moveToNext()) {
         if (Types.isOutgoingMessageType(cursor.getLong(cursor.getColumnIndexOrThrow(MESSAGE_BOX)))) {
@@ -205,7 +205,7 @@ public class MmsDatabase extends MessagingDatabase {
 
           for (String storedAddress : addresses) {
             try {
-              String ourAddress   = canonicalizeNumber(context, address);
+              String ourAddress   = canonicalizeNumber(context, messageId.getAddress());
               String theirAddress = canonicalizeNumberOrGroup(context, storedAddress);
 
               if (ourAddress.equals(theirAddress) || GroupUtil.isEncodedGroup(theirAddress)) {
@@ -230,7 +230,7 @@ public class MmsDatabase extends MessagingDatabase {
 
       if (!found) {
         try {
-          earlyReceiptCache.increment(timestamp, canonicalizeNumber(context, address));
+          earlyReceiptCache.increment(messageId.getTimetamp(), canonicalizeNumber(context, messageId.getAddress()));
         } catch (InvalidNumberException e) {
           Log.w(TAG, e);
         }
@@ -432,12 +432,72 @@ public class MmsDatabase extends MessagingDatabase {
     notifyConversationListeners(threadId);
   }
 
-  public void setMessagesRead(long threadId) {
+  public List<SyncMessageId> setMessagesRead(long threadId) {
     SQLiteDatabase      database  = databaseHelper.getWritableDatabase();
+    String              where     = THREAD_ID + " = ? AND " + READ + " = 0";
+    String[]            selection = new String[]{String.valueOf(threadId)};
+    List<SyncMessageId> result    = new LinkedList<>();
+    Cursor              cursor    = null;
+
+    database.beginTransaction();
+
+    try {
+      cursor = database.query(TABLE_NAME, new String[] {ADDRESS, DATE_SENT, MESSAGE_BOX}, where, selection, null, null, null);
+
+      while(cursor != null && cursor.moveToNext()) {
+        if (Types.isSecureType(cursor.getLong(2))) {
+          result.add(new SyncMessageId(cursor.getString(0), cursor.getLong(1)));
+        }
+      }
+
       ContentValues contentValues = new ContentValues();
       contentValues.put(READ, 1);
 
-    database.update(TABLE_NAME, contentValues, THREAD_ID + " = ?", new String[] {threadId + ""});
+      database.update(TABLE_NAME, contentValues, where, selection);
+      database.setTransactionSuccessful();
+    } finally {
+      if (cursor != null) cursor.close();
+      database.endTransaction();
+    }
+
+    return result;
+  }
+
+  public void setTimestampRead(SyncMessageId messageId) {
+    MmsAddressDatabase addressDatabase = DatabaseFactory.getMmsAddressDatabase(context);
+    SQLiteDatabase     database        = databaseHelper.getWritableDatabase();
+    Cursor             cursor          = null;
+
+    try {
+      cursor = database.query(TABLE_NAME, new String[] {ID, THREAD_ID, MESSAGE_BOX}, DATE_SENT + " = ?", new String[] {String.valueOf(messageId.getTimetamp())}, null, null, null, null);
+
+      while (cursor.moveToNext()) {
+        List<String> addresses = addressDatabase.getAddressesListForId(cursor.getLong(cursor.getColumnIndexOrThrow(ID)));
+
+        for (String storedAddress : addresses) {
+          try {
+            String ourAddress   = canonicalizeNumber(context, messageId.getAddress());
+            String theirAddress = canonicalizeNumberOrGroup(context, storedAddress);
+
+            if (ourAddress.equals(theirAddress) || GroupUtil.isEncodedGroup(theirAddress)) {
+              long id       = cursor.getLong(cursor.getColumnIndexOrThrow(ID));
+              long threadId = cursor.getLong(cursor.getColumnIndexOrThrow(THREAD_ID));
+
+              database.execSQL("UPDATE " + TABLE_NAME + " SET " + READ + " = 1 WHERE " + ID + " = ?",
+                               new String[] {String.valueOf(id)});
+
+              DatabaseFactory.getThreadDatabase(context).updateReadState(threadId);
+              notifyConversationListeners(threadId);
+            }
+          } catch (InvalidNumberException e) {
+            Log.w("MmsDatabase", e);
+          }
+        }
+      }
+    } finally {
+      if (cursor != null)
+        cursor.close();
+    }
   }
 
   public void setAllMessagesRead() {

src/org/thoughtcrime/securesms/database/MmsSmsDatabase.java

@@ -26,12 +26,15 @@ import android.support.annotation.Nullable;
 import android.util.Log;
 
 import org.thoughtcrime.securesms.crypto.MasterSecret;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
 import org.whispersystems.libaxolotl.util.guava.Optional;
 
 import java.util.HashSet;
 import java.util.Set;
 
+import ws.com.google.android.mms.pdu.PduHeaders;
+
 public class MmsSmsDatabase extends Database {
 
   private static final String TAG = MmsSmsDatabase.class.getSimpleName();
@@ -107,6 +110,17 @@ public class MmsSmsDatabase extends Database {
     return queryTables(PROJECTION, selection, order, null);
   }
 
+  public int getUnreadCount(long threadId) {
+    String selection = MmsSmsColumns.READ + " = 0 AND " + MmsSmsColumns.THREAD_ID + " = " + threadId;
+    Cursor cursor    = queryTables(PROJECTION, selection, null, null);
+
+    try {
+      return cursor != null ? cursor.getCount() : 0;
+    } finally {
+      if (cursor != null) cursor.close();;
+    }
+  }
+
   public int getConversationCount(long threadId) {
     int count = DatabaseFactory.getSmsDatabase(context).getMessageCountForThread(threadId);
     count    += DatabaseFactory.getMmsDatabase(context).getMessageCountForThread(threadId);
@@ -114,9 +128,9 @@ public class MmsSmsDatabase extends Database {
     return count;
   }
 
-  public void incrementDeliveryReceiptCount(String address, long timestamp) {
+  public void incrementDeliveryReceiptCount(SyncMessageId syncMessageId) {
-    DatabaseFactory.getSmsDatabase(context).incrementDeliveryReceiptCount(address, timestamp);
+    DatabaseFactory.getSmsDatabase(context).incrementDeliveryReceiptCount(syncMessageId);
-    DatabaseFactory.getMmsDatabase(context).incrementDeliveryReceiptCount(address, timestamp);
+    DatabaseFactory.getMmsDatabase(context).incrementDeliveryReceiptCount(syncMessageId);
   }
 
   private Cursor queryTables(String[] projection, String selection, String order, String limit) {

src/org/thoughtcrime/securesms/database/SmsDatabase.java

@@ -252,20 +252,20 @@ public class SmsDatabase extends MessagingDatabase {
     updateTypeBitmask(id, Types.BASE_TYPE_MASK, Types.BASE_SENT_FAILED_TYPE);
   }
 
-  public void incrementDeliveryReceiptCount(String address, long timestamp) {
+  public void incrementDeliveryReceiptCount(SyncMessageId messageId) {
     SQLiteDatabase database     = databaseHelper.getWritableDatabase();
     Cursor         cursor       = null;
     boolean        foundMessage = false;
 
     try {
       cursor = database.query(TABLE_NAME, new String[] {ID, THREAD_ID, ADDRESS, TYPE},
-                              DATE_SENT + " = ?", new String[] {String.valueOf(timestamp)},
+                              DATE_SENT + " = ?", new String[] {String.valueOf(messageId.getTimetamp())},
                               null, null, null, null);
 
       while (cursor.moveToNext()) {
         if (Types.isOutgoingMessageType(cursor.getLong(cursor.getColumnIndexOrThrow(TYPE)))) {
           try {
-            String theirAddress = canonicalizeNumber(context, address);
+            String theirAddress = canonicalizeNumber(context, messageId.getAddress());
             String ourAddress   = canonicalizeNumber(context, cursor.getString(cursor.getColumnIndexOrThrow(ADDRESS)));
 
             if (ourAddress.equals(theirAddress)) {
@@ -288,7 +288,7 @@ public class SmsDatabase extends MessagingDatabase {
 
       if (!foundMessage) {
         try {
-          earlyReceiptCache.increment(timestamp, canonicalizeNumber(context, address));
+          earlyReceiptCache.increment(messageId.getTimetamp(), canonicalizeNumber(context, messageId.getAddress()));
         } catch (InvalidNumberException e) {
           Log.w(TAG, e);
         }
@@ -300,14 +300,68 @@ public class SmsDatabase extends MessagingDatabase {
     }
   }
 
-  public void setMessagesRead(long threadId) {
+  public List<SyncMessageId> setMessagesRead(long threadId) {
     SQLiteDatabase      database  = databaseHelper.getWritableDatabase();
+    String              where     = THREAD_ID + " = ? AND " + READ + " = 0";
+    String[]            selection = new String[]{String.valueOf(threadId)};
+    List<SyncMessageId> results   = new LinkedList<>();
+    Cursor              cursor    = null;
+
+    database.beginTransaction();
+    try {
+      cursor = database.query(TABLE_NAME, new String[] {ADDRESS, DATE_SENT, TYPE}, where, selection, null, null, null);
+
+      while (cursor != null && cursor.moveToNext()) {
+        if (Types.isSecureType(cursor.getLong(2))) {
+          results.add(new SyncMessageId(cursor.getString(0), cursor.getLong(1)));
+        }
+      }
+
       ContentValues contentValues = new ContentValues();
       contentValues.put(READ, 1);
 
-    database.update(TABLE_NAME, contentValues,
+      database.update(TABLE_NAME, contentValues, where, selection);
-                    THREAD_ID + " = ? AND " + READ + " = 0",
+      database.setTransactionSuccessful();
-                    new String[] {threadId+""});
+    } finally {
+      if (cursor != null) cursor.close();
+      database.endTransaction();
+    }
+
+    return results;
+  }
+
+  public void setTimestampRead(SyncMessageId messageId) {
+    SQLiteDatabase database     = databaseHelper.getWritableDatabase();
+    Cursor         cursor       = null;
+
+    try {
+      cursor = database.query(TABLE_NAME, new String[] {ID, THREAD_ID, ADDRESS, TYPE},
+                              DATE_SENT + " = ?", new String[] {String.valueOf(messageId.getTimetamp())},
+                              null, null, null, null);
+
+      while (cursor.moveToNext()) {
+        try {
+          String theirAddress = canonicalizeNumber(context, messageId.getAddress());
+          String ourAddress   = canonicalizeNumber(context, cursor.getString(cursor.getColumnIndexOrThrow(ADDRESS)));
+
+          if (ourAddress.equals(theirAddress)) {
+            long threadId = cursor.getLong(cursor.getColumnIndexOrThrow(THREAD_ID));
+
+            ContentValues contentValues = new ContentValues();
+            contentValues.put(READ, 1);
+
+            database.update(TABLE_NAME, contentValues, ID_WHERE, new String[] {cursor.getLong(cursor.getColumnIndexOrThrow(ID)) + ""});
+
+            DatabaseFactory.getThreadDatabase(context).updateReadState(threadId);
+            notifyConversationListeners(threadId);
+          }
+        } catch (InvalidNumberException e) {
+          Log.w(TAG, e);
+        }
+      }
+    } finally {
+      if (cursor != null) cursor.close();
+    }
   }
 
   public void setAllMessagesRead() {

src/org/thoughtcrime/securesms/database/ThreadDatabase.java

@@ -29,6 +29,7 @@ import android.util.Log;
 
 import org.thoughtcrime.securesms.R;
 import org.thoughtcrime.securesms.crypto.MasterCipher;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.model.DisplayRecord;
 import org.thoughtcrime.securesms.database.model.MediaMmsMessageRecord;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
@@ -257,16 +258,22 @@ public class ThreadDatabase extends Database {
     notifyConversationListListeners();
   }
 
-  public void setRead(long threadId) {
+  public List<SyncMessageId> setRead(long threadId) {
     ContentValues contentValues = new ContentValues(1);
     contentValues.put(READ, 1);
 
     SQLiteDatabase db = databaseHelper.getWritableDatabase();
     db.update(TABLE_NAME, contentValues, ID_WHERE, new String[] {threadId+""});
 
-    DatabaseFactory.getSmsDatabase(context).setMessagesRead(threadId);
+    final List<SyncMessageId> smsRecords = DatabaseFactory.getSmsDatabase(context).setMessagesRead(threadId);
-    DatabaseFactory.getMmsDatabase(context).setMessagesRead(threadId);
+    final List<SyncMessageId> mmsRecords = DatabaseFactory.getMmsDatabase(context).setMessagesRead(threadId);
+
     notifyConversationListListeners();
+
+    return new LinkedList<SyncMessageId>() {{
+      addAll(smsRecords);
+      addAll(mmsRecords);
+    }};
   }
 
   public void setUnread(long threadId) {
@@ -465,6 +472,18 @@ public class ThreadDatabase extends Database {
     return null;
   }
 
+  public void updateReadState(long threadId) {
+    int unreadCount = DatabaseFactory.getMmsSmsDatabase(context).getUnreadCount(threadId);
+
+    ContentValues contentValues = new ContentValues();
+    contentValues.put(READ, unreadCount == 0);
+
+    databaseHelper.getWritableDatabase().update(TABLE_NAME, contentValues,ID_WHERE,
+                                                new String[] {String.valueOf(threadId)});
+
+    notifyConversationListListeners();
+  }
+
   public boolean update(long threadId, boolean unarchive) {
     MmsSmsDatabase mmsSmsDatabase = DatabaseFactory.getMmsSmsDatabase(context);
     long count                    = mmsSmsDatabase.getConversationCount(threadId);

src/org/thoughtcrime/securesms/dependencies/TextSecureCommunicationModule.java

@@ -12,6 +12,7 @@ import org.thoughtcrime.securesms.jobs.DeliveryReceiptJob;
 import org.thoughtcrime.securesms.jobs.GcmRefreshJob;
 import org.thoughtcrime.securesms.jobs.MultiDeviceContactUpdateJob;
 import org.thoughtcrime.securesms.jobs.MultiDeviceGroupUpdateJob;
+import org.thoughtcrime.securesms.jobs.MultiDeviceReadUpdateJob;
 import org.thoughtcrime.securesms.jobs.PushGroupSendJob;
 import org.thoughtcrime.securesms.jobs.PushMediaSendJob;
 import org.thoughtcrime.securesms.jobs.PushNotificationReceiveJob;
@@ -43,6 +44,7 @@ import dagger.Provides;
                                      PushNotificationReceiveJob.class,
                                      MultiDeviceContactUpdateJob.class,
                                      MultiDeviceGroupUpdateJob.class,
+                                     MultiDeviceReadUpdateJob.class,
                                      DeviceListFragment.class,
                                      RefreshAttributesJob.class,
                                      GcmRefreshJob.class})

src/org/thoughtcrime/securesms/jobs/MultiDeviceReadUpdateJob.java

@@ -0,0 +1,89 @@
+package org.thoughtcrime.securesms.jobs;
+
+import android.content.Context;
+
+import org.thoughtcrime.securesms.crypto.MasterSecret;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
+import org.thoughtcrime.securesms.dependencies.InjectableType;
+import org.thoughtcrime.securesms.dependencies.TextSecureCommunicationModule;
+import org.thoughtcrime.securesms.jobs.requirements.MasterSecretRequirement;
+import org.whispersystems.jobqueue.JobParameters;
+import org.whispersystems.jobqueue.requirements.NetworkRequirement;
+import org.whispersystems.textsecure.api.TextSecureMessageSender;
+import org.whispersystems.textsecure.api.crypto.UntrustedIdentityException;
+import org.whispersystems.textsecure.api.messages.multidevice.ReadMessage;
+import org.whispersystems.textsecure.api.messages.multidevice.TextSecureSyncMessage;
+import org.whispersystems.textsecure.api.push.exceptions.PushNetworkException;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.inject.Inject;
+
+public class MultiDeviceReadUpdateJob extends MasterSecretJob implements InjectableType {
+
+  private static final long serialVersionUID = 1L;
+  private static final String TAG = MultiDeviceReadUpdateJob.class.getSimpleName();
+
+  private final List<SerializableSyncMessageId> messageIds;
+
+  @Inject
+  transient TextSecureCommunicationModule.TextSecureMessageSenderFactory messageSenderFactory;
+
+  public MultiDeviceReadUpdateJob(Context context, List<SyncMessageId> messageIds) {
+    super(context, JobParameters.newBuilder()
+                                .withRequirement(new NetworkRequirement(context))
+                                .withRequirement(new MasterSecretRequirement(context))
+                                .withPersistence()
+                                .create());
+
+    this.messageIds = new LinkedList<>();
+
+    for (SyncMessageId messageId : messageIds) {
+      this.messageIds.add(new SerializableSyncMessageId(messageId.getAddress(), messageId.getTimetamp()));
+    }
+  }
+
+
+  @Override
+  public void onRun(MasterSecret masterSecret) throws IOException, UntrustedIdentityException {
+    List<ReadMessage> readMessages = new LinkedList<>();
+
+    for (SerializableSyncMessageId messageId : messageIds) {
+      readMessages.add(new ReadMessage(messageId.sender, messageId.timestamp));
+    }
+
+    TextSecureMessageSender messageSender = messageSenderFactory.create();
+    messageSender.sendMessage(TextSecureSyncMessage.forRead(readMessages));
+  }
+
+  @Override
+  public boolean onShouldRetryThrowable(Exception exception) {
+    return exception instanceof PushNetworkException;
+  }
+
+  @Override
+  public void onAdded() {
+
+  }
+
+  @Override
+  public void onCanceled() {
+
+  }
+
+  private static class SerializableSyncMessageId implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final String sender;
+    private final long   timestamp;
+
+    private SerializableSyncMessageId(String sender, long timestamp) {
+      this.sender = sender;
+      this.timestamp = timestamp;
+    }
+  }
+}

src/org/thoughtcrime/securesms/jobs/PushDecryptJob.java

@@ -17,6 +17,8 @@ import org.thoughtcrime.securesms.crypto.storage.TextSecureAxolotlStore;
 import org.thoughtcrime.securesms.crypto.storage.TextSecureSessionStore;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
 import org.thoughtcrime.securesms.database.EncryptingSmsDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.MmsDatabase;
 import org.thoughtcrime.securesms.database.NoSuchMessageException;
 import org.thoughtcrime.securesms.database.PushDatabase;
@@ -56,6 +58,7 @@ import org.whispersystems.textsecure.api.messages.TextSecureContent;
 import org.whispersystems.textsecure.api.messages.TextSecureDataMessage;
 import org.whispersystems.textsecure.api.messages.TextSecureEnvelope;
 import org.whispersystems.textsecure.api.messages.TextSecureGroup;
+import org.whispersystems.textsecure.api.messages.multidevice.ReadMessage;
 import org.whispersystems.textsecure.api.messages.multidevice.RequestMessage;
 import org.whispersystems.textsecure.api.messages.multidevice.SentTranscriptMessage;
 import org.whispersystems.textsecure.api.messages.multidevice.TextSecureSyncMessage;
@@ -146,6 +149,8 @@ public class PushDecryptJob extends ContextJob {
 
         if      (syncMessage.getSent().isPresent())    handleSynchronizeSentMessage(masterSecret, envelope, syncMessage.getSent().get(), smsMessageId);
         else if (syncMessage.getRequest().isPresent()) handleSynchronizeRequestMessage(masterSecret, syncMessage.getRequest().get());
+        else if (syncMessage.getRead().isPresent())    handleSynchronizeReadMessage(masterSecret, syncMessage.getRead().get());
+        else                                           Log.w(TAG, "Contains no known sync types...");
       }
 
       if (envelope.isPreKeyWhisperMessage()) {
@@ -252,6 +257,17 @@ public class PushDecryptJob extends ContextJob {
     }
   }
 
+  private void handleSynchronizeReadMessage(@NonNull MasterSecretUnion masterSecret,
+                                            @NonNull List<ReadMessage> readMessages)
+  {
+    for (ReadMessage readMessage : readMessages) {
+      DatabaseFactory.getSmsDatabase(context).setTimestampRead(new SyncMessageId(readMessage.getSender(), readMessage.getTimestamp()));
+      DatabaseFactory.getMmsDatabase(context).setTimestampRead(new SyncMessageId(readMessage.getSender(), readMessage.getTimestamp()));
+    }
+
+    MessageNotifier.updateNotification(context, masterSecret.getMasterSecret().orNull());
+  }
+
   private void handleMediaMessage(@NonNull MasterSecretUnion masterSecret,
                                   @NonNull TextSecureEnvelope envelope,
                                   @NonNull TextSecureDataMessage message,

src/org/thoughtcrime/securesms/jobs/PushReceivedJob.java

@@ -5,6 +5,8 @@ import android.util.Log;
 
 import org.thoughtcrime.securesms.ApplicationContext;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.MessagingDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.NotInDirectoryException;
 import org.thoughtcrime.securesms.database.TextSecureDirectory;
 import org.thoughtcrime.securesms.recipients.RecipientFactory;
@@ -64,8 +66,8 @@ public abstract class PushReceivedJob extends ContextJob {
 
   private void handleReceipt(TextSecureEnvelope envelope) {
     Log.w(TAG, String.format("Received receipt: (XXXXX, %d)", envelope.getTimestamp()));
-    DatabaseFactory.getMmsSmsDatabase(context).incrementDeliveryReceiptCount(envelope.getSource(),
+    DatabaseFactory.getMmsSmsDatabase(context).incrementDeliveryReceiptCount(new SyncMessageId(envelope.getSource(),
-                                                                             envelope.getTimestamp());
+                                                                                               envelope.getTimestamp()));
   }
 
   private boolean isActiveNumber(Context context, String e164number) {

src/org/thoughtcrime/securesms/notifications/MarkReadReceiver.java

@@ -1,15 +1,20 @@
 package org.thoughtcrime.securesms.notifications;
 
 import android.app.NotificationManager;
-import android.content.BroadcastReceiver;
 import android.content.Context;
 import android.content.Intent;
 import android.os.AsyncTask;
 import android.support.annotation.Nullable;
 import android.util.Log;
 
+import org.thoughtcrime.securesms.ApplicationContext;
 import org.thoughtcrime.securesms.crypto.MasterSecret;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
+import org.thoughtcrime.securesms.jobs.MultiDeviceReadUpdateJob;
+
+import java.util.LinkedList;
+import java.util.List;
 
 public class MarkReadReceiver extends MasterSecretBroadcastReceiver {
 
@@ -35,12 +40,22 @@ public class MarkReadReceiver extends MasterSecretBroadcastReceiver {
       new AsyncTask<Void, Void, Void>() {
         @Override
         protected Void doInBackground(Void... params) {
+          List<SyncMessageId> messageIdsCollection = new LinkedList<>();
+
           for (long threadId : threadIds) {
             Log.w(TAG, "Marking as read: " + threadId);
-            DatabaseFactory.getThreadDatabase(context).setRead(threadId);
+            List<SyncMessageId> messageIds = DatabaseFactory.getThreadDatabase(context).setRead(threadId);
+            messageIdsCollection.addAll(messageIds);
           }
 
           MessageNotifier.updateNotification(context, masterSecret);
+
+          if (!messageIdsCollection.isEmpty()) {
+            ApplicationContext.getInstance(context)
+                              .getJobManager()
+                              .add(new MultiDeviceReadUpdateJob(context, messageIdsCollection));
+          }
+
           return null;
         }
       }.execute();

src/org/thoughtcrime/securesms/notifications/MessageNotifier.java

@@ -38,16 +38,20 @@ import android.text.TextUtils;
 import android.text.style.StyleSpan;
 import android.util.Log;
 
+import org.thoughtcrime.securesms.ApplicationContext;
 import org.thoughtcrime.securesms.ConversationActivity;
 import org.thoughtcrime.securesms.R;
 import org.thoughtcrime.securesms.crypto.MasterSecret;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.MessagingDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.MmsSmsDatabase;
 import org.thoughtcrime.securesms.database.PushDatabase;
 import org.thoughtcrime.securesms.database.SmsDatabase;
 import org.thoughtcrime.securesms.database.ThreadDatabase;
 import org.thoughtcrime.securesms.database.model.MediaMmsMessageRecord;
 import org.thoughtcrime.securesms.database.model.MessageRecord;
+import org.thoughtcrime.securesms.jobs.MultiDeviceReadUpdateJob;
 import org.thoughtcrime.securesms.mms.SlideDeck;
 import org.thoughtcrime.securesms.recipients.Recipient;
 import org.thoughtcrime.securesms.recipients.RecipientFactory;
@@ -136,7 +140,13 @@ public class MessageNotifier {
                                                .getRecipientsForThreadId(threadId);
 
     if (isVisible) {
-      threads.setRead(threadId);
+      List<SyncMessageId> messageIds = threads.setRead(threadId);
+
+      if (!messageIds.isEmpty()) {
+        ApplicationContext.getInstance(context)
+                          .getJobManager()
+                          .add(new MultiDeviceReadUpdateJob(context, messageIds));
+      }
     }
 
     if (!TextSecurePreferences.isNotificationsEnabled(context) ||

src/org/thoughtcrime/securesms/notifications/WearReplyReceiver.java

@@ -24,10 +24,14 @@ import android.os.Bundle;
 import android.support.annotation.Nullable;
 import android.support.v4.app.RemoteInput;
 
+import org.thoughtcrime.securesms.ApplicationContext;
 import org.thoughtcrime.securesms.attachments.Attachment;
 import org.thoughtcrime.securesms.crypto.MasterSecret;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.MessagingDatabase;
+import org.thoughtcrime.securesms.database.MessagingDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.RecipientPreferenceDatabase.RecipientsPreferences;
+import org.thoughtcrime.securesms.jobs.MultiDeviceReadUpdateJob;
 import org.thoughtcrime.securesms.mms.OutgoingMediaMessage;
 import org.thoughtcrime.securesms.recipients.RecipientFactory;
 import org.thoughtcrime.securesms.recipients.Recipients;
@@ -36,6 +40,7 @@ import org.thoughtcrime.securesms.sms.OutgoingTextMessage;
 import org.whispersystems.libaxolotl.util.guava.Optional;
 
 import java.util.LinkedList;
+import java.util.List;
 
 /**
  * Get the response text from the Wearable Device and sends an message as a reply
@@ -77,9 +82,15 @@ public class WearReplyReceiver extends MasterSecretBroadcastReceiver {
             threadId = MessageSender.send(context, masterSecret, reply, -1, false);
           }
 
-          DatabaseFactory.getThreadDatabase(context).setRead(threadId);
+          List<SyncMessageId> messageIds = DatabaseFactory.getThreadDatabase(context).setRead(threadId);
           MessageNotifier.updateNotification(context, masterSecret);
 
+          if (!messageIds.isEmpty()) {
+            ApplicationContext.getInstance(context)
+                              .getJobManager()
+                              .add(new MultiDeviceReadUpdateJob(context, messageIds));
+          }
+
           return null;
         }
       }.execute();