[geary/wip/713150-conversations: 1/2] First step: upgrade database, populate conversation table, add new messages to it

[Jim Nelson <jnelson src gnome org>]

commit a1caec679c504b5b5cd7519fcd6941f6e1fe5758
Author: Jim Nelson <jim yorba org>
Date:   Wed Feb 18 16:49:22 2015 -0800

    First step: upgrade database, populate conversation table, add new messages to it

 sql/CMakeLists.txt                           |    1 +
 sql/version-025.sql                          |   13 ++
 src/CMakeLists.txt                           |    1 +
 src/engine/imap-db/imap-db-conversation.vala |  178 ++++++++++++++++++++++++++
 src/engine/imap-db/imap-db-database.vala     |   28 ++++
 src/engine/imap-db/imap-db-folder.vala       |   10 ++
 6 files changed, 231 insertions(+), 0 deletions(-)
---
diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 11de86c..c89b8c9 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -24,3 +24,4 @@ install(FILES version-021.sql DESTINATION ${SQL_DEST})
 install(FILES version-022.sql DESTINATION ${SQL_DEST})
 install(FILES version-023.sql DESTINATION ${SQL_DEST})
 install(FILES version-024.sql DESTINATION ${SQL_DEST})
+install(FILES version-025.sql DESTINATION ${SQL_DEST})
diff --git a/sql/version-025.sql b/sql/version-025.sql
new file mode 100644
index 0000000..5b3fc78
--- /dev/null
+++ b/sql/version-025.sql
@@ -0,0 +1,13 @@
+--
+-- Index tables for fast lookup and association of conversations
+--
+
+CREATE TABLE ConversationTable (
+    id INTEGER PRIMARY KEY,
+    flags TEXT DEFAULT NULL
+);
+
+ALTER TABLE MessageTable ADD COLUMN conversation_id INTEGER REFERENCES ConversationTable DEFAULT NULL;
+
+CREATE INDEX MessageTableConversationIDIndex ON MessageTable(conversation_id);
+
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 78151fc..35984d6 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -166,6 +166,7 @@ engine/imap/transport/imap-serializer.vala
 engine/imap-db/imap-db-account.vala
 engine/imap-db/imap-db-attachment.vala
 engine/imap-db/imap-db-contact.vala
+engine/imap-db/imap-db-conversation.vala
 engine/imap-db/imap-db-database.vala
 engine/imap-db/imap-db-email-identifier.vala
 engine/imap-db/imap-db-folder.vala
diff --git a/src/engine/imap-db/imap-db-conversation.vala b/src/engine/imap-db/imap-db-conversation.vala
new file mode 100644
index 0000000..84d52de
--- /dev/null
+++ b/src/engine/imap-db/imap-db-conversation.vala
@@ -0,0 +1,178 @@
+/* Copyright 2015 Yorba Foundation
+ *
+ * This software is licensed under the GNU Lesser General Public License
+ * (version 2.1 or later).  See the COPYING file in this distribution.
+ */
+
+/**
+ * Associated calls for adding and fetching messages associated as conversations.
+ *
+ * Note that methods prefixed with do_ are intended to be run from within a transaction.
+ */
+ 
+namespace Geary.ImapDB.Conversation {
+
+internal const Geary.Email.Field REQUIRED_FIELDS = Email.Field.REFERENCES;
+
+/**
+ * Should only be called when an email message's { link REQUIRED_FIELDS} are initially fulfilled.
+ */
+internal void do_add_message_to_conversation(Db.Connection cx, int64 message_id, Cancellable? cancellable)
+    throws Error {
+    Db.Statement references_stmt = cx.prepare("""
+        SELECT message_id, in_reply_to, reference_ids
+        FROM MessageTable
+        WHERE id = ?
+    """);
+    references_stmt.bind_rowid(0, message_id);
+    
+    Db.Result references_result = references_stmt.exec(cancellable);
+    
+    // Create a common set of ancestors from In-Reply-To and References
+    Gee.HashSet<RFC822.MessageID> ancestors = new Gee.HashSet<RFC822.MessageID>();
+    add_ancestors(references_result.string_at(1), ancestors);
+    add_ancestors(references_result.string_at(2), ancestors);
+    
+    // Add this message's Message-ID
+    unowned string? rfc822_message_id_text = references_result.string_at(0);
+    if (!String.is_empty(rfc822_message_id_text))
+        ancestors.add(new RFC822.MessageID(rfc822_message_id_text));
+    
+    // in reality, no ancestors indicates that REFERENCES was not complete, so log that and exit
+    if (ancestors.size == 0) {
+        message("Unable to add message %s to conversation table: no ancestors", message_id.to_string());
+        
+        return;
+    }
+    
+    // search for existing conversation(s) for any of these Message-IDs that's not this message
+    StringBuilder sql = new StringBuilder("""
+        SELECT conversation_id
+        FROM MessageTable
+        WHERE message_id IN (
+    """);
+    for (int ctr = 0; ctr < ancestors.size; ctr++)
+        sql.append(ctr == 0 ? "?" : ",?");
+    sql.append(") AND id <> ?");
+
+    Db.Statement search_stmt = cx.prepare(sql.str);
+    int col = 0;
+    foreach (RFC822.MessageID ancestor in ancestors)
+        search_stmt.bind_string(col++, ancestor.value);
+    search_stmt.bind_rowid(col++, message_id);
+    
+    Gee.HashSet<int64?> conversation_ids = new Gee.HashSet<int64?>(Collection.int64_hash_func,
+        Collection.int64_equal_func);
+    
+    Db.Result search_result = search_stmt.exec(cancellable);
+    while (!search_result.finished) {
+        // watch for NULL, which is the default value when a row is added to the MessageTable
+        // without a conversation id (which is almost always for new rows)
+        if (!search_result.is_null_at(0))
+            conversation_ids.add(search_result.rowid_at(0));
+        
+        search_result.next(cancellable);
+    }
+    
+    // Select the message's conversation_id from the following three scenarios:
+    int64 conversation_id;
+    if (conversation_ids.size > 1) {
+        // this indicates that two (or more) conversations were created due to emails arriving
+        // out of order and the complete(r) tree is only being available now; merge the
+        // conversations into one
+        conversation_id = do_merge_conversations(cx, conversation_ids, cancellable);
+        
+        debug("Merged %d conversations to conversation %s", conversation_ids.size - 1,
+            conversation_id.to_string());
+    } else if (conversation_ids.size == 0) {
+        // No conversation for this Message-ID, so generate a new one
+        cx.exec("""
+            INSERT INTO ConversationTable
+            DEFAULT VALUES
+        """);
+        conversation_id = cx.last_insert_rowid;
+        
+        debug("Created new conversation %s for message %s: %s", conversation_id.to_string(),
+            message_id.to_string(), rfc822_message_id_text);
+    } else {
+        // one conversation found, so use that one
+        conversation_id = traverse<int64?>(conversation_ids).first();
+        
+        debug("Expanding existing conversation %s with message %s: %s", conversation_id.to_string(),
+            message_id.to_string(), rfc822_message_id_text);
+    }
+    
+    // Assign the message to this conversation
+    Db.Statement insert = cx.prepare("""
+        UPDATE MessageTable
+        SET conversation_id = ?
+        WHERE id = ?
+    """);
+    insert.bind_rowid(0, conversation_id);
+    insert.bind_rowid(1, message_id);
+    
+    insert.exec(cancellable);
+}
+
+private void add_ancestors(string? text, Gee.Collection<RFC822.MessageID> ancestors) {
+    if (String.is_empty(text))
+        return;
+    
+    RFC822.MessageIDList message_id_list = new RFC822.MessageIDList.from_rfc822_string(text);
+    ancestors.add_all(message_id_list.list);
+}
+
+private int64 do_merge_conversations(Db.Connection cx, Gee.Set<int64?> conversation_ids, Cancellable? 
cancellable)
+    throws Error {
+    // must be at least two in order to merge
+    assert(conversation_ids.size > 1);
+    
+    // doesn't really matter which; use the first one
+    int64 conversation_id = traverse<int64?>(conversation_ids).first();
+    
+    //
+    // TODO: Merge flags together
+    //
+    
+    // reuse this IN block in the following two SQL statements
+    StringBuilder in_sql = new StringBuilder("(");
+    bool first = true;
+    foreach (int64 other_conversation_id in conversation_ids) {
+        if (other_conversation_id == conversation_id)
+            continue;
+        
+        if (!first)
+            in_sql.append(",");
+        
+        in_sql.append(other_conversation_id.to_string());
+        first = false;
+    }
+    in_sql.append(")");
+    
+    // set other messages in the other conversations to the chosen one
+    StringBuilder merge_sql = new StringBuilder("""
+        UPDATE MessageTable
+        SET conversation_id = ?
+        WHERE conversation_id IN
+    """);
+    merge_sql.append(in_sql.str);
+    
+    Db.Statement merge_stmt = cx.prepare(merge_sql.str);
+    merge_stmt.bind_rowid(0, conversation_id);
+    
+    merge_stmt.exec(cancellable);
+    
+    // remove merged conversation(s)
+    StringBuilder delete_sql = new StringBuilder("""
+        DELETE FROM ConversationTable
+        WHERE id IN
+    """);
+    delete_sql.append(in_sql.str);
+    
+    cx.exec(delete_sql.str);
+    
+    return conversation_id;
+}
+
+}
+
diff --git a/src/engine/imap-db/imap-db-database.vala b/src/engine/imap-db/imap-db-database.vala
index 2dfdc33..90c2fc8 100644
--- a/src/engine/imap-db/imap-db-database.vala
+++ b/src/engine/imap-db/imap-db-database.vala
@@ -174,6 +174,10 @@ private class Geary.ImapDB.Database : Geary.Db.VersionedDatabase {
             case 23:
                 post_upgrade_add_tokenizer_table();
             break;
+            
+            case 25:
+                post_upgrade_build_conversation_tables();
+            break;
         }
     }
     
@@ -555,6 +559,30 @@ private class Geary.ImapDB.Database : Geary.Db.VersionedDatabase {
         }
     }
     
+    // Version 25
+    private void post_upgrade_build_conversation_tables() {
+        try {
+            exec_transaction(Db.TransactionType.RW, (cx) => {
+                // Fetch every message that has the required fields for assigning a conversation
+                Db.Result result = query("""
+                    SELECT id, fields
+                    FROM MessageTable
+                """);
+                while (!result.finished) {
+                    Email.Field fields = (Email.Field) result.int_at(1);
+                    if (fields.fulfills(Conversation.REQUIRED_FIELDS))
+                        Conversation.do_add_message_to_conversation(cx, result.rowid_at(0), null);
+                    
+                    result.next();
+                }
+                
+                return Db.TransactionOutcome.COMMIT;
+            }, null);
+        } catch (Error err) {
+            error("Error upgrading conversation tables: %s", err.message);
+        }
+    }
+    
     private void on_prepare_database_connection(Db.Connection cx) throws Error {
         cx.set_busy_timeout_msec(Db.Connection.RECOMMENDED_BUSY_TIMEOUT_MSEC);
         cx.set_foreign_keys(true);
diff --git a/src/engine/imap-db/imap-db-folder.vala b/src/engine/imap-db/imap-db-folder.vala
index 5c8e514..6fef755 100644
--- a/src/engine/imap-db/imap-db-folder.vala
+++ b/src/engine/imap-db/imap-db-folder.vala
@@ -1380,6 +1380,10 @@ private class Geary.ImapDB.Folder : BaseObject, Geary.ReferenceSemantics {
         if (email.fields.fulfills(Attachment.REQUIRED_FIELDS))
             do_save_attachments(cx, message_id, email.get_message().get_attachments(), cancellable);
         
+        // add to conversation, if required fields available
+        if (email.fields.fulfills(Conversation.REQUIRED_FIELDS))
+            Conversation.do_add_message_to_conversation(cx, message_id, cancellable);
+        
         do_add_email_to_search_table(cx, message_id, email, cancellable);
         
         MessageAddresses message_addresses =
@@ -1930,6 +1934,12 @@ private class Geary.ImapDB.Folder : BaseObject, Geary.ReferenceSemantics {
                 do_merge_email_in_search_table(cx, location.message_id, new_fields, combined_email, 
cancellable);
             else
                 do_add_email_to_search_table(cx, location.message_id, combined_email, cancellable);
+            
+            // add to conversation if ready for it
+            if (!fetched_fields.fulfills(Conversation.REQUIRED_FIELDS)
+                && combined_email.fields.fulfills(Conversation.REQUIRED_FIELDS)) {
+                Conversation.do_add_message_to_conversation(cx, location.message_id, cancellable);
+            }
         } else {
             // If the email is ready to go, we still may need to update the unread count.
             Geary.EmailFlags? combined_flags = do_get_email_flags_single(cx, location.message_id,