Some code to work with conversations

README.md

@@ -9,7 +9,7 @@ A Java implementation for the Bitmessage protocol. To build, use command `./grad
 
 Please note that it still has its limitations, but the API should now be stable. Jabit uses Semantic Versioning, meaning as long as the major version doesn't change, nothing should break if you update.
 
-Be aware though that this doesn't necessarily applies for SNAPSHOT builds and the development branch, notably when it comes to database updates. _In other words, they may break your installation!_ 
+Be aware though that this doesn't necessarily applies for SNAPSHOT builds and the development branch, notably when it comes to database updates. In other words, they may break your installation!_ 
 
 #### Master
 [![Build Status](https://travis-ci.org/Dissem/Jabit.svg?branch=master)](https://travis-ci.org/Dissem/Jabit) 

core/build.gradle

@@ -25,7 +25,7 @@ artifacts {
 
 dependencies {
     compile 'org.slf4j:slf4j-api:1.7.12'
-    compile 'ch.dissem.msgpack:msgpack:development-SNAPSHOT'
+    compile 'ch.dissem.msgpack:msgpack:1.0.0'
     testCompile 'junit:junit:4.12'
     testCompile 'org.hamcrest:hamcrest-library:1.3'
     testCompile 'org.mockito:mockito-core:1.10.19'

core/src/main/java/ch/dissem/bitmessage/DefaultMessageListener.java

@@ -87,7 +87,7 @@ class DefaultMessageListener implements NetworkHandler.MessageListener, Internal
         BitmessageAddress identity = ctx.getAddressRepository().findIdentity(getPubkey.getRipeTag());
         if (identity != null && identity.getPrivateKey() != null && !identity.isChan()) {
             LOG.info("Got pubkey request for identity " + identity);
-            // FIXME: only send pubkey if it wasn't sent in the last 28 days
+            // FIXME: only send pubkey if it wasn't sent in the last TTL.pubkey() days
             ctx.sendPubkey(identity, object.getStream());
         }
     }

core/src/main/java/ch/dissem/bitmessage/entity/Plaintext.java

@@ -18,22 +18,20 @@ package ch.dissem.bitmessage.entity;
 
 import ch.dissem.bitmessage.entity.payload.Msg;
 import ch.dissem.bitmessage.entity.payload.Pubkey.Feature;
-import ch.dissem.bitmessage.entity.valueobject.extended.Attachment;
 import ch.dissem.bitmessage.entity.valueobject.ExtendedEncoding;
 import ch.dissem.bitmessage.entity.valueobject.InventoryVector;
 import ch.dissem.bitmessage.entity.valueobject.Label;
+import ch.dissem.bitmessage.entity.valueobject.extended.Attachment;
 import ch.dissem.bitmessage.entity.valueobject.extended.Message;
 import ch.dissem.bitmessage.exception.ApplicationException;
 import ch.dissem.bitmessage.factory.ExtendedEncodingFactory;
 import ch.dissem.bitmessage.factory.Factory;
-import ch.dissem.bitmessage.utils.Decode;
+import ch.dissem.bitmessage.utils.*;
-import ch.dissem.bitmessage.utils.Encode;
-import ch.dissem.bitmessage.utils.TTL;
-import ch.dissem.bitmessage.utils.UnixTime;
 
 import java.io.*;
 import java.nio.ByteBuffer;
 import java.util.*;
+import java.util.Collections;
 
 import static ch.dissem.bitmessage.entity.Plaintext.Encoding.EXTENDED;
 import static ch.dissem.bitmessage.entity.Plaintext.Encoding.SIMPLE;
@@ -50,6 +48,7 @@ public class Plaintext implements Streamable {
     private final long encoding;
     private final byte[] message;
     private final byte[] ackData;
+    private final UUID conversationId;
     private ExtendedEncoding extendedData;
     private ObjectMessage ackMessage;
     private Object id;
@@ -90,6 +89,7 @@ public class Plaintext implements Streamable {
         ttl = builder.ttl;
         retries = builder.retries;
         nextTry = builder.nextTry;
+        conversationId = builder.conversation;
     }
 
     public static Plaintext read(Type type, InputStream in) throws IOException {
@@ -390,7 +390,7 @@ public class Plaintext implements Streamable {
     }
 
     public List<InventoryVector> getParents() {
-        if (Message.TYPE.equals(getExtendedData().getType())) {
+        if (getExtendedData() != null && Message.TYPE.equals(getExtendedData().getType())) {
             return ((Message) extendedData.getContent()).getParents();
         } else {
             return Collections.emptyList();
@@ -405,6 +405,10 @@ public class Plaintext implements Streamable {
         }
     }
 
+    public UUID getConversationId() {
+        return conversationId;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -470,6 +474,16 @@ public class Plaintext implements Streamable {
         return initialHash;
     }
 
+    @Override
+    public String toString() {
+        String subject = getSubject();
+        if (subject == null || subject.length() == 0) {
+            return Strings.hex(initialHash).toString();
+        } else {
+            return subject;
+        }
+    }
+
     public enum Encoding {
         IGNORE(0), TRIVIAL(1), SIMPLE(2), EXTENDED(3);
 
@@ -527,12 +541,13 @@ public class Plaintext implements Streamable {
         private byte[] ackMessage;
         private byte[] signature;
         private long sent;
-        private long received;
+        private Long received;
         private Status status;
         private Set<Label> labels = new HashSet<>();
         private long ttl;
         private int retries;
         private Long nextTry;
+        private UUID conversation;
 
         public Builder(Type type) {
             this.type = type;
@@ -685,6 +700,11 @@ public class Plaintext implements Streamable {
             return this;
         }
 
+        public Builder conversation(UUID id) {
+            this.conversation = id;
+            return this;
+        }
+
         public Plaintext build() {
             if (from == null) {
                 from = new BitmessageAddress(Factory.createPubkey(
@@ -706,6 +726,9 @@ public class Plaintext implements Streamable {
             if (ttl <= 0) {
                 ttl = TTL.msg();
             }
+            if (conversation == null) {
+                conversation = UUID.randomUUID();
+            }
             return new Plaintext(this);
         }
     }

core/src/main/java/ch/dissem/bitmessage/ports/AbstractMessageRepository.java

@@ -19,13 +19,16 @@ package ch.dissem.bitmessage.ports;
 import ch.dissem.bitmessage.InternalContext;
 import ch.dissem.bitmessage.entity.BitmessageAddress;
 import ch.dissem.bitmessage.entity.Plaintext;
+import ch.dissem.bitmessage.entity.valueobject.InventoryVector;
 import ch.dissem.bitmessage.entity.valueobject.Label;
 import ch.dissem.bitmessage.exception.ApplicationException;
 import ch.dissem.bitmessage.utils.Strings;
 import ch.dissem.bitmessage.utils.UnixTime;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.UUID;
 
 import static ch.dissem.bitmessage.utils.SqlStrings.join;
 
@@ -71,6 +74,11 @@ public abstract class AbstractMessageRepository implements MessageRepository, In
         }
     }
 
+    @Override
+    public Plaintext getMessage(InventoryVector iv) {
+        return single(find("iv=X'" + Strings.hex(iv.getHash()) + "'"));
+    }
+
     @Override
     public Plaintext getMessage(byte[] initialHash) {
         return single(find("initial_hash=X'" + Strings.hex(initialHash) + "'"));
@@ -111,6 +119,20 @@ public abstract class AbstractMessageRepository implements MessageRepository, In
             " AND next_try < " + UnixTime.now());
     }
 
+    @Override
+    public List<Plaintext> findResponses(Plaintext parent) {
+        if (parent.getInventoryVector() == null) {
+            return Collections.emptyList();
+        }
+        return find("iv IN (SELECT child FROM Message_Parent"
+            + " WHERE parent=X'" + Strings.hex(parent.getInventoryVector().getHash()) + "')");
+    }
+
+    @Override
+    public List<Plaintext> getConversation(UUID conversationId) {
+        return find("conversation=X'" + conversationId.toString().replace("-", "") + "'");
+    }
+
     @Override
     public List<Label> getLabels() {
         return findLabels("1=1");

core/src/main/java/ch/dissem/bitmessage/ports/MessageRepository.java

@@ -19,9 +19,12 @@ package ch.dissem.bitmessage.ports;
 import ch.dissem.bitmessage.entity.BitmessageAddress;
 import ch.dissem.bitmessage.entity.Plaintext;
 import ch.dissem.bitmessage.entity.Plaintext.Status;
+import ch.dissem.bitmessage.entity.valueobject.InventoryVector;
 import ch.dissem.bitmessage.entity.valueobject.Label;
 
+import java.util.Collection;
 import java.util.List;
+import java.util.UUID;
 
 public interface MessageRepository {
     List<Label> getLabels();
@@ -32,10 +35,18 @@ public interface MessageRepository {
 
     Plaintext getMessage(Object id);
 
+    Plaintext getMessage(InventoryVector iv);
+
     Plaintext getMessage(byte[] initialHash);
 
     Plaintext getMessageForAck(byte[] ackData);
 
+    /**
+     * @param label to search for
+     * @return a distinct list of all conversations that have at least one message with the given label.
+     */
+    List<UUID> findConversations(Label label);
+
     List<Plaintext> findMessages(Label label);
 
     List<Plaintext> findMessages(Status status);
@@ -44,9 +55,21 @@ public interface MessageRepository {
 
     List<Plaintext> findMessages(BitmessageAddress sender);
 
+    List<Plaintext> findResponses(Plaintext parent);
+
     List<Plaintext> findMessagesToResend();
 
     void save(Plaintext message);
 
     void remove(Plaintext message);
+
+    /**
+     * Returns all messages with this conversation ID. The returned messages aren't sorted in any way,
+     * so you may prefer to use {@link ch.dissem.bitmessage.utils.ConversationService#getConversation(UUID)}
+     * instead.
+     *
+     * @param conversationId ID of the requested conversation
+     * @return all messages with the given conversation ID
+     */
+    Collection<Plaintext> getConversation(UUID conversationId);
 }

core/src/main/java/ch/dissem/bitmessage/ports/NodeRegistry.java

@@ -24,6 +24,12 @@ import java.util.List;
  * Stores and provides known peers.
  */
 public interface NodeRegistry {
+    /**
+     * Removes all known nodes from registry. This should work around connection issues
+     * when there are many invalid nodes in the registry.
+     */
+    void clear();
+
     List<NetworkAddress> getKnownAddresses(int limit, long... streams);
 
     void offerAddresses(List<NetworkAddress> addresses);

core/src/main/java/ch/dissem/bitmessage/utils/ConversationService.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2017 Christian Basler
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ch.dissem.bitmessage.utils;
+
+import ch.dissem.bitmessage.entity.Plaintext;
+import ch.dissem.bitmessage.entity.valueobject.InventoryVector;
+import ch.dissem.bitmessage.ports.MessageRepository;
+
+import java.util.*;
+import java.util.Collections;
+
+/**
+ * Helper service to work with conversations
+ */
+public class ConversationService {
+    private final MessageRepository messageRepository;
+
+    public ConversationService(MessageRepository messageRepository) {
+        this.messageRepository = messageRepository;
+    }
+
+    public List<Plaintext> getConversation(Plaintext message) {
+        return getConversation(message.getConversationId());
+    }
+
+    private LinkedList<Plaintext> sorted(Collection<Plaintext> collection) {
+        LinkedList<Plaintext> result = new LinkedList<>(collection);
+        Collections.sort(result, new Comparator<Plaintext>() {
+            @Override
+            public int compare(Plaintext o1, Plaintext o2) {
+                //noinspection NumberEquality - if both are null (if both are the same, it's a bonus)
+                if (o1.getReceived() == o2.getReceived()) {
+                    return 0;
+                }
+                if (o1.getReceived() == null) {
+                    return -1;
+                }
+                if (o2.getReceived() == null) {
+                    return 1;
+                }
+                return -o1.getReceived().compareTo(o2.getReceived());
+            }
+        });
+        return result;
+    }
+
+    public List<Plaintext> getConversation(UUID conversationId) {
+        LinkedList<Plaintext> messages = sorted(messageRepository.getConversation(conversationId));
+        Map<InventoryVector, Plaintext> map = new HashMap<>(messages.size());
+        for (Plaintext message : messages) {
+            if (message.getInventoryVector() != null) {
+                map.put(message.getInventoryVector(), message);
+            }
+        }
+
+        LinkedList<Plaintext> result = new LinkedList<>();
+        while (!messages.isEmpty()) {
+            Plaintext last = messages.poll();
+            int pos = lastParentPosition(last, result);
+            result.add(pos, last);
+            addAncestors(last, result, messages, map);
+        }
+        return result;
+    }
+
+    private int lastParentPosition(Plaintext child, LinkedList<Plaintext> messages) {
+        Iterator<Plaintext> plaintextIterator = messages.descendingIterator();
+        int i = 0;
+        while (plaintextIterator.hasNext()) {
+            Plaintext next = plaintextIterator.next();
+            if (isParent(next, child)) {
+                break;
+            }
+            i++;
+        }
+        return messages.size() - i;
+    }
+
+    private boolean isParent(Plaintext item, Plaintext child) {
+        for (InventoryVector parentId : child.getParents()) {
+            if (parentId.equals(item.getInventoryVector())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private void addAncestors(Plaintext message, LinkedList<Plaintext> result, LinkedList<Plaintext> messages, Map<InventoryVector, Plaintext> map) {
+        for (InventoryVector parentKey : message.getParents()) {
+            Plaintext parent = map.remove(parentKey);
+            if (parent != null) {
+                messages.remove(parent);
+                result.addFirst(parent);
+                addAncestors(parent, result, messages, map);
+            }
+        }
+    }
+}

core/src/main/java/ch/dissem/bitmessage/utils/Decode.java

@@ -138,7 +138,7 @@ public class Decode {
 
     public static String varString(InputStream in, AccessCounter counter) throws IOException {
         int length = (int) varInt(in, counter);
-        // FIXME: technically, it says the length in characters, but I think this one might be correct
+        // technically, it says the length in characters, but I think this one might be correct
         // otherwise it will get complicated, as we'll need to read UTF-8 char by char...
         return new String(bytes(in, length, counter), "utf-8");
     }

core/src/test/java/ch/dissem/bitmessage/utils/ConversationServiceTest.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2017 Christian Basler
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ch.dissem.bitmessage.utils;
+
+import ch.dissem.bitmessage.cryptography.bc.BouncyCryptography;
+import ch.dissem.bitmessage.entity.BitmessageAddress;
+import ch.dissem.bitmessage.entity.Plaintext;
+import ch.dissem.bitmessage.entity.valueobject.ExtendedEncoding;
+import ch.dissem.bitmessage.entity.valueobject.extended.Message;
+import ch.dissem.bitmessage.ports.MessageRepository;
+import org.junit.Test;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.UUID;
+
+import static ch.dissem.bitmessage.entity.Plaintext.Type.MSG;
+import static ch.dissem.bitmessage.utils.TestUtils.RANDOM;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConversationServiceTest {
+    private BitmessageAddress alice = new BitmessageAddress("BM-2cSqjfJ8xK6UUn5Rw3RpdGQ9RsDkBhWnS8");
+    private BitmessageAddress bob = new BitmessageAddress("BM-2cTtkBnb4BUYDndTKun6D9PjtueP2h1bQj");
+
+    private MessageRepository messageRepository = mock(MessageRepository.class);
+    private ConversationService conversationService = new ConversationService(messageRepository);
+
+    static {
+        Singleton.initialize(new BouncyCryptography());
+    }
+
+    @Test
+    public void ensureConversationIsSortedProperly() {
+        List<Plaintext> expected = getConversation();
+
+        when(conversationService.getConversation(any(UUID.class))).thenReturn(expected);
+        List<Plaintext> actual = conversationService.getConversation(UUID.randomUUID());
+        assertThat(actual, is(expected));
+    }
+
+    private List<Plaintext> getConversation() {
+        List<Plaintext> result = new LinkedList<>();
+
+        Plaintext older = plaintext(alice, bob,
+            new Message.Builder()
+                .subject("hey there")
+                .body("does it work?")
+                .build(),
+            Plaintext.Status.SENT);
+        result.add(older);
+
+        Plaintext root = plaintext(alice, bob,
+            new Message.Builder()
+                .subject("new test")
+                .body("There's a new test in town!")
+                .build(),
+            Plaintext.Status.SENT);
+        result.add(root);
+
+        result.add(
+            plaintext(bob, alice,
+                new Message.Builder()
+                    .subject("Re: new test (1a)")
+                    .body("Nice!")
+                    .addParent(root)
+                    .build(),
+                Plaintext.Status.RECEIVED)
+        );
+
+        Plaintext latest = plaintext(bob, alice,
+            new Message.Builder()
+                .subject("Re: new test (2b)")
+                .body("PS: it did work!")
+                .addParent(root)
+                .addParent(older)
+                .build(),
+            Plaintext.Status.RECEIVED);
+        result.add(latest);
+
+        result.add(
+            plaintext(alice, bob,
+                new Message.Builder()
+                    .subject("Re: new test (2)")
+                    .body("")
+                    .addParent(latest)
+                    .build(),
+                Plaintext.Status.DRAFT)
+        );
+
+        return result;
+    }
+
+    private int timer = 2;
+
+    private Plaintext plaintext(BitmessageAddress from, BitmessageAddress to,
+                                ExtendedEncoding content, Plaintext.Status status) {
+        Plaintext.Builder builder = new Plaintext.Builder(MSG)
+            .IV(TestUtils.randomInventoryVector())
+            .from(from)
+            .to(to)
+            .message(content)
+            .status(status);
+        if (status != Plaintext.Status.DRAFT && status != Plaintext.Status.DOING_PROOF_OF_WORK) {
+            builder.received(5 * ++timer - RANDOM.nextInt(10));
+        }
+        return builder.build();
+    }
+}

demo/src/test/java/ch/dissem/bitmessage/TestNodeRegistry.java

@@ -39,6 +39,11 @@ class TestNodeRegistry implements NodeRegistry {
         }
     }
 
+    @Override
+    public void clear() {
+        // NO OP
+    }
+
     @Override
     public List<NetworkAddress> getKnownAddresses(int limit, long... streams) {
         return nodes;

repositories/src/main/java/ch/dissem/bitmessage/repository/JdbcMessageRepository.java

@@ -31,6 +31,7 @@ import java.sql.*;
 import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.UUID;
 
 import static ch.dissem.bitmessage.repository.JdbcHelper.writeBlob;
 
@@ -99,7 +100,7 @@ public class JdbcMessageRepository extends AbstractMessageRepository implements
             Connection connection = config.getConnection();
             Statement stmt = connection.createStatement();
             ResultSet rs = stmt.executeQuery(
-                "SELECT id, iv, type, sender, recipient, data, ack_data, sent, received, initial_hash, status, ttl, retries, next_try " +
+                "SELECT id, iv, type, sender, recipient, data, ack_data, sent, received, initial_hash, status, ttl, retries, next_try, conversation " +
                     "FROM Message WHERE " + where)
         ) {
             while (rs.next()) {
@@ -119,6 +120,7 @@ public class JdbcMessageRepository extends AbstractMessageRepository implements
                 builder.ttl(rs.getLong("ttl"));
                 builder.retries(rs.getInt("retries"));
                 builder.nextTry(rs.getLong("next_try"));
+                builder.conversation((UUID) rs.getObject("conversation"));
                 builder.labels(findLabels(connection,
                     "id IN (SELECT label_id FROM Message_Label WHERE message_id=" + id + ") ORDER BY ord"));
                 Plaintext message = builder.build();
@@ -155,6 +157,7 @@ public class JdbcMessageRepository extends AbstractMessageRepository implements
             try {
                 connection.setAutoCommit(false);
                 save(connection, message);
+                updateParents(connection, message);
                 updateLabels(connection, message);
                 connection.commit();
             } catch (IOException | SQLException e) {
@@ -189,11 +192,38 @@ public class JdbcMessageRepository extends AbstractMessageRepository implements
         }
     }
 
+    private void updateParents(Connection connection, Plaintext message) throws SQLException {
+        if (message.getInventoryVector() == null || message.getParents().isEmpty()) {
+            // There are no parents to save yet (they are saved in the extended data, that's enough for now)
+            return;
+        }
+        // remove existing parents
+        try (PreparedStatement ps = connection.prepareStatement("DELETE FROM Message_Parent WHERE child=?")) {
+            ps.setBytes(1, message.getInitialHash());
+            ps.executeUpdate();
+        }
+        byte[] childIV = message.getInventoryVector().getHash();
+        // save new parents
+        int order = 0;
+        for (InventoryVector parentIV : message.getParents()) {
+            Plaintext parent = getMessage(parentIV);
+            mergeConversations(connection, parent.getConversationId(), message.getConversationId());
+            order++;
+            try (PreparedStatement ps = connection.prepareStatement("INSERT INTO Message_Parent VALUES (?, ?, ?, ?)")) {
+                ps.setBytes(1, parentIV.getHash());
+                ps.setBytes(2, childIV);
+                ps.setInt(3, order); // FIXME: this might not be necessary
+                ps.setObject(4, message.getConversationId());
+                ps.executeUpdate();
+            }
+        }
+    }
+
     private void insert(Connection connection, Plaintext message) throws SQLException, IOException {
         try (PreparedStatement ps = connection.prepareStatement(
             "INSERT INTO Message (iv, type, sender, recipient, data, ack_data, sent, received, " +
-                "status, initial_hash, ttl, retries, next_try) " +
+                "status, initial_hash, ttl, retries, next_try, conversation) " +
-                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
             Statement.RETURN_GENERATED_KEYS)
         ) {
             ps.setBytes(1, message.getInventoryVector() == null ? null : message.getInventoryVector().getHash());
@@ -209,6 +239,7 @@ public class JdbcMessageRepository extends AbstractMessageRepository implements
             ps.setLong(11, message.getTTL());
             ps.setInt(12, message.getRetries());
             ps.setObject(13, message.getNextTry());
+            ps.setObject(14, message.getConversationId());
 
             ps.executeUpdate();
             // get generated id
@@ -262,4 +293,52 @@ public class JdbcMessageRepository extends AbstractMessageRepository implements
             LOG.error(e.getMessage(), e);
         }
     }
+
+    @Override
+    public List<UUID> findConversations(Label label) {
+        String where;
+        if (label == null) {
+            where = "id NOT IN (SELECT message_id FROM Message_Label)";
+        } else {
+            where = "id IN (SELECT message_id FROM Message_Label WHERE label_id=" + label.getId() + ")";
+        }
+        List<UUID> result = new LinkedList<>();
+        try (
+            Connection connection = config.getConnection();
+            Statement stmt = connection.createStatement();
+            ResultSet rs = stmt.executeQuery(
+                "SELECT DISTINCT conversation FROM Message WHERE " + where)
+        ) {
+            while (rs.next()) {
+                result.add((UUID) rs.getObject(1));
+            }
+        } catch (SQLException e) {
+            LOG.error(e.getMessage(), e);
+        }
+        return result;
+    }
+
+    /**
+     * Replaces every occurrence of the source conversation ID with the target ID
+     *
+     * @param source ID of the conversation to be merged
+     * @param target ID of the merge target
+     */
+    private void mergeConversations(Connection connection, UUID source, UUID target) {
+        try (
+            PreparedStatement ps1 = connection.prepareStatement(
+                "UPDATE Message SET conversation=? WHERE conversation=?");
+            PreparedStatement ps2 = connection.prepareStatement(
+                "UPDATE Message_Parent SET conversation=? WHERE conversation=?")
+        ) {
+            ps1.setObject(1, target);
+            ps1.setObject(2, source);
+            ps1.executeUpdate();
+            ps2.setObject(1, target);
+            ps2.setObject(2, source);
+            ps2.executeUpdate();
+        } catch (SQLException e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
 }

repositories/src/main/java/ch/dissem/bitmessage/repository/JdbcNodeRegistry.java

@@ -69,6 +69,19 @@ public class JdbcNodeRegistry extends JdbcHelper implements NodeRegistry {
         }
     }
 
+    @Override
+    public void clear() {
+        try (
+            Connection connection = config.getConnection();
+            PreparedStatement ps = connection.prepareStatement(
+                "DELETE FROM Node")
+        ) {
+            ps.executeUpdate();
+        } catch (SQLException e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
+
     @Override
     public List<NetworkAddress> getKnownAddresses(int limit, long... streams) {
         List<NetworkAddress> result = new LinkedList<>();

repositories/src/main/resources/db/migration/V4.0__Create_table_message_parent.sql

@@ -0,0 +1,11 @@
+ALTER TABLE Message ADD COLUMN conversation UUID NOT NULL DEFAULT RANDOM_UUID();
+
+CREATE TABLE Message_Parent (
+    parent       BINARY(64) NOT NULL,
+    child        BINARY(64) NOT NULL,
+    pos          INT NOT NULL,
+    conversation UUID,
+
+    PRIMARY KEY (parent, child),
+    FOREIGN KEY (child) REFERENCES Message (iv)
+);

repositories/src/test/java/ch/dissem/bitmessage/repository/JdbcMessageRepositoryTest.java

@@ -21,18 +21,23 @@ import ch.dissem.bitmessage.InternalContext;
 import ch.dissem.bitmessage.entity.BitmessageAddress;
 import ch.dissem.bitmessage.entity.ObjectMessage;
 import ch.dissem.bitmessage.entity.Plaintext;
-import ch.dissem.bitmessage.entity.valueobject.InventoryVector;
+import ch.dissem.bitmessage.entity.valueobject.ExtendedEncoding;
 import ch.dissem.bitmessage.entity.valueobject.Label;
 import ch.dissem.bitmessage.entity.valueobject.PrivateKey;
+import ch.dissem.bitmessage.entity.valueobject.extended.Message;
 import ch.dissem.bitmessage.ports.AddressRepository;
 import ch.dissem.bitmessage.ports.MessageRepository;
 import ch.dissem.bitmessage.utils.TestUtils;
 import ch.dissem.bitmessage.utils.UnixTime;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.UUID;
 
 import static ch.dissem.bitmessage.entity.Plaintext.Type.MSG;
 import static ch.dissem.bitmessage.entity.payload.Pubkey.Feature.DOES_ACK;
@@ -49,6 +54,7 @@ public class JdbcMessageRepositoryTest extends TestBase {
     private MessageRepository repo;
 
     private Label inbox;
+    private Label sent;
     private Label drafts;
     private Label unread;
 
@@ -59,9 +65,9 @@ public class JdbcMessageRepositoryTest extends TestBase {
         AddressRepository addressRepo = new JdbcAddressRepository(config);
         repo = new JdbcMessageRepository(config);
         new InternalContext(new BitmessageContext.Builder()
-                .cryptography(cryptography())
+            .cryptography(cryptography())
-                .addressRepo(addressRepo)
+            .addressRepo(addressRepo)
-                .messageRepo(repo)
+            .messageRepo(repo)
         );
 
         BitmessageAddress tmp = new BitmessageAddress(new PrivateKey(false, 1, 1000, 1000, DOES_ACK));
@@ -75,6 +81,7 @@ public class JdbcMessageRepositoryTest extends TestBase {
         addressRepo.save(identity);
 
         inbox = repo.getLabels(Label.Type.INBOX).get(0);
+        sent = repo.getLabels(Label.Type.SENT).get(0);
         drafts = repo.getLabels(Label.Type.DRAFT).get(0);
         unread = repo.getLabels(Label.Type.UNREAD).get(0);
 
@@ -163,12 +170,12 @@ public class JdbcMessageRepositoryTest extends TestBase {
     @Test
     public void testSave() throws Exception {
         Plaintext message = new Plaintext.Builder(MSG)
-                .IV(TestUtils.randomInventoryVector())
+            .IV(TestUtils.randomInventoryVector())
-                .from(identity)
+            .from(identity)
-                .to(contactA)
+            .to(contactA)
-                .message("Subject", "Message")
+            .message("Subject", "Message")
-                .status(Plaintext.Status.DOING_PROOF_OF_WORK)
+            .status(Plaintext.Status.DOING_PROOF_OF_WORK)
-                .build();
+            .build();
         repo.save(message);
 
         assertNotNull(message.getId());
@@ -207,19 +214,19 @@ public class JdbcMessageRepositoryTest extends TestBase {
     @Test
     public void ensureUnacknowledgedMessagesAreFoundForResend() throws Exception {
         Plaintext message = new Plaintext.Builder(MSG)
-                .IV(TestUtils.randomInventoryVector())
+            .IV(TestUtils.randomInventoryVector())
-                .from(identity)
+            .from(identity)
-                .to(contactA)
+            .to(contactA)
-                .message("Subject", "Message")
+            .message("Subject", "Message")
-                .status(Plaintext.Status.SENT)
+            .status(Plaintext.Status.SENT)
-                .ttl(2)
+            .ttl(2)
-                .build();
+            .build();
         message.updateNextTry();
         assertThat(message.getRetries(), is(1));
         assertThat(message.getNextTry(), greaterThan(UnixTime.now()));
         assertThat(message.getNextTry(), lessThanOrEqualTo(UnixTime.now(+2)));
         repo.save(message);
-        Thread.sleep(4100);
+        Thread.sleep(4100); // somewhat longer than 2*TTL
         List<Plaintext> messagesToResend = repo.findMessagesToResend();
         assertThat(messagesToResend, hasSize(1));
 
@@ -231,14 +238,105 @@ public class JdbcMessageRepositoryTest extends TestBase {
         assertThat(messagesToResend, empty());
     }
 
-    private void addMessage(BitmessageAddress from, BitmessageAddress to, Plaintext.Status status, Label... labels) {
+    @Test
+    public void ensureParentsAreSaved() {
+        Plaintext parent = storeConversation();
+
+        List<Plaintext> responses = repo.findResponses(parent);
+        assertThat(responses, hasSize(2));
+        assertThat(responses, hasItem(hasMessage("Re: new test", "Nice!")));
+        assertThat(responses, hasItem(hasMessage("Re: new test", "PS: it did work!")));
+    }
+
+    @Test
+    public void ensureConversationCanBeRetrieved() {
+        Plaintext root = storeConversation();
+        List<UUID> conversations = repo.findConversations(inbox);
+        assertThat(conversations, hasSize(2));
+        assertThat(conversations, hasItem(root.getConversationId()));
+    }
+
+    private Plaintext addMessage(BitmessageAddress from, BitmessageAddress to, Plaintext.Status status, Label... labels) {
+        ExtendedEncoding content = new Message.Builder()
+            .subject("Subject")
+            .body("Message")
+            .build();
+        return addMessage(from, to, content, status, labels);
+    }
+
+    private Plaintext addMessage(BitmessageAddress from, BitmessageAddress to,
+                                 ExtendedEncoding content, Plaintext.Status status, Label... labels) {
         Plaintext message = new Plaintext.Builder(MSG)
-                .from(from)
+            .IV(TestUtils.randomInventoryVector())
-                .to(to)
+            .from(from)
-                .message("Subject", "Message")
+            .to(to)
-                .status(status)
+            .message(content)
-                .labels(Arrays.asList(labels))
+            .status(status)
-                .build();
+            .labels(Arrays.asList(labels))
+            .build();
         repo.save(message);
+        return message;
+    }
+
+    private Plaintext storeConversation() {
+        Plaintext older = addMessage(identity, contactA,
+            new Message.Builder()
+                .subject("hey there")
+                .body("does it work?")
+                .build(),
+            Plaintext.Status.SENT, sent);
+
+        Plaintext root = addMessage(identity, contactA,
+            new Message.Builder()
+                .subject("new test")
+                .body("There's a new test in town!")
+                .build(),
+            Plaintext.Status.SENT, sent);
+
+        addMessage(contactA, identity,
+            new Message.Builder()
+                .subject("Re: new test")
+                .body("Nice!")
+                .addParent(root)
+                .build(),
+            Plaintext.Status.RECEIVED, inbox);
+
+        addMessage(contactA, identity,
+            new Message.Builder()
+                .subject("Re: new test")
+                .body("PS: it did work!")
+                .addParent(root)
+                .addParent(older)
+                .build(),
+            Plaintext.Status.RECEIVED, inbox);
+
+        return repo.getMessage(root.getId());
+    }
+
+    private Matcher<Plaintext> hasMessage(String subject, String body) {
+        return new BaseMatcher<Plaintext>() {
+            @Override
+            public void describeTo(Description description) {
+                description.appendText("Subject: ").appendText(subject);
+                description.appendText(", ");
+                description.appendText("Body: ").appendText(body);
+            }
+
+            @Override
+            public boolean matches(Object item) {
+                if (item instanceof Plaintext) {
+                    Plaintext message = (Plaintext) item;
+                    if (subject != null && !subject.equals(message.getSubject())) {
+                        return false;
+                    }
+                    if (body != null && !body.equals(message.getText())) {
+                        return false;
+                    }
+                    return true;
+                } else {
+                    return false;
+                }
+            }
+        };
     }
 }

repositories/src/test/java/ch/dissem/bitmessage/repository/TestJdbcConfig.java

@@ -16,11 +16,27 @@
 
 package ch.dissem.bitmessage.repository;
 
+import org.h2.tools.Server;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.SQLException;
+
 /**
  * JdbcConfig to be used for tests. Uses an in-memory database and adds a useful {@link #reset()} method resetting
  * the database.
  */
 public class TestJdbcConfig extends JdbcConfig {
+    private static final Logger LOG = LoggerFactory.getLogger(TestJdbcConfig.class);
+
+    static {
+        try {
+            Server.createTcpServer().start();
+        } catch (SQLException e) {
+            LOG.error(e.getMessage(), e);
+        }
+    }
+
     public TestJdbcConfig() {
         super("jdbc:h2:mem:test;DB_CLOSE_DELAY=-1", "sa", null);
     }