[bug] fix notification for incoming mail with several listeners registered

When setting the listeners in the IMAP Folder, we avoid setting more
than one listener for the same imap mailbox (because in some situations
we were registering way too many listeners).

this was making the pixelated inbox registering the notification and
therefore the imap mailbox not being registered.

this MR also refactors the way pixelated is initialized, so that it
avoid creating a second Account instance. In this way, we make sure that
the pixelated mua and the imap server share the same collections for a
given mailbox, and therefore any of the two is able to get a
notification whenever the other adds a message to the mailbox.

- Resolves: #8846, #8798

7 files changed, 106 insertions, 77 deletions

diff --git a/src/leap/bitmask/core/mail_services.py b/src/leap/bitmask/core/mail_services.py
index abd2f980..4197700b 100644
--- a/src/leap/bitmask/core/mail_services.py
+++ b/src/leap/bitmask/core/mail_services.py
@@ -509,18 +509,19 @@ class StandardMailService(service.MultiService, HookableService):
         sendmail_opts = _get_sendmail_opts(self._basedir, provider, username)
         self._sendmail_opts[userid] = sendmail_opts
 
-        incoming = self.getServiceNamed('incoming_mail')
-        incoming.startInstance(userid)
-
         def registerToken(token):
             self._service_tokens[userid] = token
             self._active_user = userid
             return token
 
-        d = soledad.get_or_create_service_token('mail_auth')
+        incoming = self.getServiceNamed('incoming_mail')
+        d = incoming.startInstance(userid)
+        d.addCallback(
+            lambda _: soledad.get_or_create_service_token('mail_auth'))
         d.addCallback(registerToken)
         d.addCallback(self._write_tokens_file, userid)
-        d.addCallback(self._maybe_start_pixelated, userid, soledad, keymanager)
+        d.addCallback(
+            self._maybe_start_pixelated, userid, soledad, keymanager)
         return d
 
     # hooks
@@ -537,10 +538,9 @@ class StandardMailService(service.MultiService, HookableService):
 
     @defer.inlineCallbacks
     def hook_on_bonafide_auth(self, **kw):
-        # TODO: if it's expired we should renew it
         userid = kw['username']
-
         self._maybe_start_incoming_service(userid)
+        # TODO: if it's expired we should renew it
         yield self._maybe_fetch_smtp_certificate(userid)
 
     def _maybe_start_incoming_service(self, userid):
@@ -642,10 +642,11 @@ class StandardMailService(service.MultiService, HookableService):
             json.dump(token_dict, ftokens)
 
     def _maybe_start_pixelated(self, passthrough, userid, soledad, keymanager):
+        incoming = self.getServiceNamed('incoming_mail')
+        account = incoming.getServiceNamed(userid).account
         if pixelizer.HAS_PIXELATED:
-            reactor.callFromThread(
-                pixelizer.start_pixelated_user_agent,
-                userid, soledad, keymanager)
+            pixelizer.start_pixelated_user_agent(
+                userid, soledad, keymanager, account)
         return passthrough
 
 
@@ -750,12 +751,13 @@ class IncomingMailService(service.MultiService):
     # Individual accounts
 
     def startInstance(self, userid):
+        """returns: a deferred"""
         self._set_status(userid, "starting")
         soledad = self._mail.get_soledad_session(userid)
         keymanager = self._mail.get_keymanager_session(userid)
 
         self.log.info('Setting up Incoming Mail Service for %s' % userid)
-        self._start_incoming_mail_instance(
+        return self._start_incoming_mail_instance(
             keymanager, soledad, userid)
 
     @defer.inlineCallbacks
@@ -777,11 +779,12 @@ class IncomingMailService(service.MultiService):
     def _start_incoming_mail_instance(self, keymanager, soledad,
                                       userid, start_sync=True):
 
-        def setUpIncomingMail(inbox):
+        def setUpIncomingMail(inbox, acc):
             incoming_mail = IncomingMail(
                 keymanager, soledad,
                 inbox, userid,
                 check_period=INCOMING_CHECK_PERIOD)
+            incoming_mail.account = acc
             self.log.debug('Setting Incoming Mail Service for %s' % userid)
             incoming_mail.setName(userid)
             self.addService(incoming_mail)
@@ -790,10 +793,17 @@ class IncomingMailService(service.MultiService):
             self._set_status(userid, "on")
             return res
 
+        # XXX ----------------------------------------------------------------
+        # TODO we probably want to enforce a SINGLE ACCOUNT INSTANCE
+        # earlier in the bootstrap process (ie, upper in the hierarchy of
+        # services) so that the single instance can be shared by the imap and
+        # the pixelated mua.
+        # XXX ----------------------------------------------------------------
+
         acc = Account(soledad, userid)
         d = acc.callWhenReady(
             lambda _: acc.get_collection_by_mailbox(INBOX_NAME))
-        d.addCallback(setUpIncomingMail)
+        d.addCallback(setUpIncomingMail, acc)
         d.addCallback(setStatusOn)
         d.addErrback(self._errback, userid)
         return d
diff --git a/src/leap/bitmask/mail/adaptors/soledad.py b/src/leap/bitmask/mail/adaptors/soledad.py
index a3e011a3..f220aea0 100644
--- a/src/leap/bitmask/mail/adaptors/soledad.py
+++ b/src/leap/bitmask/mail/adaptors/soledad.py
@@ -851,6 +851,7 @@ class SoledadMailAdaptor(SoledadIndexMixin):
     wait_for_indexes = ['get_or_create_mbox', 'update_mbox', 'get_all_mboxes']
 
     mboxwrapper_klass = MailboxWrapper
+    atomic = defer.DeferredLock()
 
     log = Logger()
 
diff --git a/src/leap/bitmask/mail/imap/mailbox.py b/src/leap/bitmask/mail/imap/mailbox.py
index f74910f9..9e74cfc9 100644
--- a/src/leap/bitmask/mail/imap/mailbox.py
+++ b/src/leap/bitmask/mail/imap/mailbox.py
@@ -64,14 +64,14 @@ def make_collection_listener(mailbox):
         def __init__(self, mbox):
             self.mbox = mbox
 
-            # See #8083, pixelated adaptor seems to be misusing this class.
-            self.mailbox_name = self.mbox.mbox_name
+            # See #8083, pixelated adaptor introduces conflicts in the usage
+            self.mailbox_name = self.mbox.mbox_name + 'IMAP'
 
         def __hash__(self):
-            return hash(self.mbox.mbox_name)
+            return hash(self.mailbox_name)
 
         def __eq__(self, other):
-            return self.mbox.mbox_name == other.mbox.mbox_name
+            return self.mailbox_name == other.mbox.mbox_name + 'IMAP'
 
         def notify_new(self):
             self.mbox.notify_new()
@@ -397,8 +397,6 @@ class IMAPMailbox(object):
 
         :param args: ignored.
         """
-        if not NOTIFY_NEW:
-            return
 
         def cbNotifyNew(result):
             exists, recent = result
@@ -887,15 +885,8 @@ class IMAPMailbox(object):
                  uid when the copy succeed.
         :rtype: Deferred
         """
-        # A better place for this would be  the COPY/APPEND dispatcher
-        # in server.py, but qtreactor hangs when I do that, so this seems
-        # to work fine for now.
-        # d.addCallback(lambda r: self.reactor.callLater(0, self.notify_new))
-        # deferLater(self.reactor, 0, self._do_copy, message, d)
-        # return d
-
-        d = self.collection.copy_msg(message.message,
-                                     self.collection.mbox_uuid)
+        d = self.collection.copy_msg(
+            message.message, self.collection.mbox_uuid)
         return d
 
     # convenience fun
diff --git a/src/leap/bitmask/mail/incoming/service.py b/src/leap/bitmask/mail/incoming/service.py
index b1d48cf4..7c7e7f2c 100644
--- a/src/leap/bitmask/mail/incoming/service.py
+++ b/src/leap/bitmask/mail/incoming/service.py
@@ -140,19 +140,6 @@ class IncomingMail(Service):
         # initialize a mail parser only once
         self._parser = Parser()
 
-    def add_listener(self, listener):
-        """
-        Add a listener to inbox insertions.
-
-        This listener function will be called for each message added to the
-        inbox with its uid as parameter. This function should not be blocking
-        or it will block the incoming queue.
-
-        :param listener: the listener function
-        :type listener: callable
-        """
-        self._listeners.append(listener)
-
     #
     # Public API: fetch, start_loop, stop.
     #
@@ -165,17 +152,18 @@ class IncomingMail(Service):
         """
         def _sync_errback(failure):
             self.log.error(
-                'Error while fetching incoming mail {0!r}'.format(failure))
+                'Error while fetching incoming mail: {0!r}'.format(failure))
 
         def syncSoledadCallback(_):
             # XXX this should be moved to adaptors
+            # TODO  we can query the blobs store instead.
             d = self._soledad.get_from_index(
                 fields.JUST_MAIL_IDX, "1", "0")
             d.addCallback(self._process_incoming_mail)
             d.addErrback(_sync_errback)
             return d
 
-        self.log.debug("fetching mail for: %s %s" % (
+        self.log.debug('Fetching mail for: %s %s' % (
             self._soledad.uuid, self._userid))
         d = self._sync_soledad()
         d.addCallbacks(syncSoledadCallback, self._errback)
@@ -231,7 +219,7 @@ class IncomingMail(Service):
         :rtype: iterable or None
         """
         def _log_synced(result):
-            self.log.info('sync finished')
+            self.log.info('Sync finished')
             return result
 
         def _handle_invalid_auth_token_error(failure):
@@ -240,7 +228,7 @@ class IncomingMail(Service):
             self.stopService()
             emit_async(catalog.SOLEDAD_INVALID_AUTH_TOKEN, self._userid)
 
-        self.log.info('starting sync...')
+        self.log.info('Starting sync...')
         d = self._soledad.sync()
         d.addCallbacks(_log_synced, _handle_invalid_auth_token_error)
         return d
@@ -258,7 +246,7 @@ class IncomingMail(Service):
             fetched_ts = time.mktime(time.gmtime())
             num_mails = len(doclist) if doclist is not None else 0
             if num_mails != 0:
-                self.log.info("there are %s mails" % (num_mails,))
+                self.log.info('There are {0!s} mails'.format(num_mails))
             emit_async(catalog.MAIL_FETCHED_INCOMING, self._userid,
                        str(num_mails), str(fetched_ts))
             return doclist
@@ -282,7 +270,7 @@ class IncomingMail(Service):
         deferreds = []
         for index, doc in enumerate(doclist):
             self.log.debug(
-                'processing incoming message: %d of %d'
+                'Processing Incoming Message: %d of %d'
                 % (index + 1, num_mails))
             emit_async(catalog.MAIL_MSG_PROCESSING, self._userid,
                        str(index), str(num_mails))
@@ -292,15 +280,15 @@ class IncomingMail(Service):
             # TODO Compatibility check with the index in pre-0.6 mx
             # that does not write the ERROR_DECRYPTING_KEY
             # This should be removed in 0.7
+            # TODO deprecate this already
 
             has_errors = doc.content.get(fields.ERROR_DECRYPTING_KEY, None)
-
             if has_errors is None:
-                warnings.warn("JUST_MAIL_COMPAT_IDX will be deprecated!",
+                warnings.warn('JUST_MAIL_COMPAT_IDX will be deprecated!',
                               DeprecationWarning)
 
             if has_errors:
-                self.log.debug("Skipping message with decrypting errors...")
+                self.log.debug('Skipping message with decrypting errors...')
             elif self._is_msg(keys):
                 # TODO this pipeline is a bit obscure!
                 d = self._decrypt_doc(doc)
@@ -784,7 +772,7 @@ class IncomingMail(Service):
             else:
                 self.log.debug("No valid url on OpenPGP header %s" % (url,))
         else:
-            self.log.debug("There is no url on the OpenPGP header: %s"
+            self.log.debug('There is no url on the OpenPGP header: %s'
                            % (header,))
         return False
 
@@ -843,12 +831,10 @@ class IncomingMail(Service):
         self.log.info('Adding message %s to local db' % (doc.doc_id,))
 
         def msgSavedCallback(result):
+
             if empty(result):
                 return
 
-            for listener in self._listeners:
-                listener(result)
-
             def signal_deleted(doc_id):
                 emit_async(catalog.MAIL_MSG_DELETED_INCOMING,
                            self._userid)
diff --git a/src/leap/bitmask/mail/mail.py b/src/leap/bitmask/mail/mail.py
index 8cc01e27..92435fdb 100644
--- a/src/leap/bitmask/mail/mail.py
+++ b/src/leap/bitmask/mail/mail.py
@@ -970,6 +970,10 @@ class Account(object):
         return d
 
     def add_mailbox(self, name, creation_ts=None):
+        return self.adaptor.atomic.run(
+            self._add_mailbox, name, creation_ts=creation_ts)
+
+    def _add_mailbox(self, name, creation_ts=None):
 
         if creation_ts is None:
             # by default, we pass an int value
@@ -1004,6 +1008,10 @@ class Account(object):
         return d
 
     def delete_mailbox(self, name):
+        return self.adaptor.atomic.run(
+            self._delete_mailbox, name)
+
+    def _delete_mailbox(self, name):
 
         def delete_uid_table_cb(wrapper):
             d = self.mbox_indexer.delete_table(wrapper.uuid)
@@ -1017,6 +1025,10 @@ class Account(object):
         return d
 
     def rename_mailbox(self, oldname, newname):
+        return self.adaptor.atomic.run(
+            self._rename_mailbox, oldname, newname)
+
+    def _rename_mailbox(self, oldname, newname):
 
         def _rename_mbox(wrapper):
             wrapper.mbox = newname
@@ -1035,6 +1047,10 @@ class Account(object):
         :rtype: deferred
         :return: a deferred that will fire with a MessageCollection
         """
+        return self.adaptor.atomic.run(
+            self._get_collection_by_mailbox, name)
+
+    def _get_collection_by_mailbox(self, name):
         collection = self._collection_mapping[self.user_id].get(
             name, None)
         if collection:
@@ -1056,8 +1072,7 @@ class Account(object):
         :rtype: MessageCollection
         """
         # get a collection of docs by a list of doc_id
-        # get.docs(...) --> it should be a generator. does it behave in the
-        # threadpool?
+        # get.docs(...) --> it should be a generator
         raise NotImplementedError()
 
     def get_collection_by_tag(self, tag):
diff --git a/src/leap/bitmask/mail/smtp/bounces.py b/src/leap/bitmask/mail/smtp/bounces.py
index 8260c1b0..608d87db 100644
--- a/src/leap/bitmask/mail/smtp/bounces.py
+++ b/src/leap/bitmask/mail/smtp/bounces.py
@@ -84,6 +84,8 @@ class Bouncer(object):
 
 def bouncerFactory(soledad):
     user_id = soledad.uuid
+    # TODO should reuse same account that is used in other places, otherwise
+    # new mail in here won't be notified.
     acc = Account(soledad, user_id)
     d = acc.callWhenReady(lambda _: acc.get_collection_by_mailbox(INBOX_NAME))
     d.addCallback(lambda inbox: Bouncer(inbox))
diff --git a/src/leap/bitmask/mua/pixelizer.py b/src/leap/bitmask/mua/pixelizer.py
index ed69afb2..c2dd7144 100644
--- a/src/leap/bitmask/mua/pixelizer.py
+++ b/src/leap/bitmask/mua/pixelizer.py
@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
-# pix.py
-# Copyright (C) 2016 LEAP
+# pixelizer.py
+# Copyright (C) 2016-2017 LEAP
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -14,9 +14,22 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
 """
-Pixelated plugin integration.
+Pixelizer: the pixelated integrator.
+
+Right now this module is kind of hardcoded, but it should be possible for
+bitmask to use this as an optional plugin. For the moment we fail gracefully if
+we cannot find the pixelated modules in the import path.
+
+There's a lot of code duplication right now in this double game of adaptors for
+pixelated adaptors. Refactoring pixelated and bitmask should be possible so
+that we converge the mail apis and stop needing this nasty proliferation of
+adaptors.
+
+However, some care has to be taken to avoid certain types of concurrency bugs.
 """
+
 import json
 import os
 import sys
@@ -25,12 +38,10 @@ from twisted.internet import defer, reactor
 from twisted.logger import Logger
 
 from leap.common.config import get_path_prefix
-from leap.bitmask.mail.mail import Account
 from leap.bitmask.keymanager import KeyNotFound
 
 try:
     from pixelated.adapter.mailstore import LeapMailStore
-    from pixelated.adapter.welcome_mail import add_welcome_mail
     from pixelated.application import SingleUserServicesFactory
     from pixelated.application import UserAgentMode
     from pixelated.application import start_site
@@ -51,10 +62,13 @@ log = Logger()
 # [ ] pre-authenticate
 
 
-def start_pixelated_user_agent(userid, soledad, keymanager):
+def start_pixelated_user_agent(userid, soledad, keymanager, account):
 
-    leap_session = LeapSessionAdapter(
-        userid, soledad, keymanager)
+    try:
+        leap_session = LeapSessionAdapter(
+            userid, soledad, keymanager, account)
+    except Exception as exc:
+        log.error("Got error! %r" % exc)
 
     config = Config()
     leap_home = os.path.join(get_path_prefix(), 'leap')
@@ -134,9 +148,23 @@ class NickNym(object):
         return self.keymanager.send_key()
 
 
+class _LeapMailStore(LeapMailStore):
+
+    def __init__(self, soledad, account):
+        self.account = account
+        super(_LeapMailStore, self).__init__(soledad)
+
+    # We should rewrite the LeapMailStore in the coming pixelated fork so that
+    # we reuse the account instance.
+    @defer.inlineCallbacks
+    def add_mail(self, mailbox_name, raw_msg):
+        inbox = yield self.account.get_collection_by_mailbox(mailbox_name)
+        yield inbox.add_msg(raw_msg, ('\\Recent',), notify_just_mdoc=False)
+
+
 class LeapSessionAdapter(object):
 
-    def __init__(self, userid, soledad, keymanager):
+    def __init__(self, userid, soledad, keymanager, account):
 
         self.userid = userid
         self.soledad = soledad
@@ -144,14 +172,16 @@ class LeapSessionAdapter(object):
         # XXX this needs to be converged with our public apis.
         _n = NickNym(keymanager, userid)
         self.nicknym = self.keymanager = _n
-        self.mail_store = LeapMailStore(soledad)
+
+        self.mail_store = _LeapMailStore(soledad, account)
+
+        self.account = account
 
         self.user_auth = Config()
         self.user_auth.uuid = soledad.uuid
 
         self.fresh_account = False
         self.incoming_mail_fetcher = None
-        self.account = Account(soledad, userid)
 
         username, provider = userid.split('@')
         smtp_client_cert = os.path.join(
@@ -201,13 +231,7 @@ def _start_in_single_user_mode(leap_session, config, resource,
                                services_factory):
     start_site(config, resource)
     reactor.callLater(
-        # workaround for #8798
-        # we need to make pixelated initialization a bit behind
-        # the own leap initialization, because otherwise the inbox is created
-        # without the needed callbacks for IMAP compatibility.
-        # This should be better addressed at pixelated code, by using the mail
-        # api to create the collection.
-        3, start_user_agent_in_single_user_mode,
+        0, start_user_agent_in_single_user_mode,
         resource, services_factory,
         leap_session.config.leap_home, leap_session)
 
@@ -221,10 +245,10 @@ def start_user_agent_in_single_user_mode(root_resource,
     _services = services.Services(leap_session)
     yield _services.setup()
 
-    if leap_session.fresh_account:
-        yield add_welcome_mail(leap_session.mail_store)
+    # TODO we might want to use a Bitmask specific mail
+    # if leap_session.fresh_account:
+    #   yield add_welcome_mail(leap_session.mail_store)
 
     services_factory.add_session(leap_session.user_auth.uuid, _services)
-
     root_resource.initialize(provider=leap_session.provider)
     log.info('Done, the Pixelated User Agent is ready to be used')