GitLab

at 2017-11-19T19:36:20+01:00

Add MAM for MUCs

Remove Support for mam:0

Add new DB table last_archive_message

Add account_id column to logs table

When we search for duplicated stanza-ids we have to do it
in the context of an account

Add new DB methods

Save account_id in logs table

Save last archive timestamps in DB

- A Migration from last_mam_id config value is included

Move common code out of HistoryWindow

- move it into message_archiving.py

Parse from jid correctly

Request one day history when first joining MUCs

Remove log message

Improve deduplication

A stanza-id is only unique within a specific archive
This adds the archive jid to the query

Merge branch 'master' into 'master'

Add MAM for MUCs

See merge request gajim/gajim!152
 import subprocess

 

-__version__ = "0.16.11.1"

+__version__ = "0.16.11.2"

 

 try:

     node = subprocess.Popen('git rev-parse --short=12 HEAD', shell=True,

 

 import base64

 import hashlib

+from collections import namedtuple

 

 import logging

 log = logging.getLogger('gajim.c.caps_cache')

 

+import nbxmpp

 from nbxmpp import (NS_XHTML_IM, NS_ESESSION, NS_CHATSTATES,

     NS_JINGLE_ICE_UDP, NS_JINGLE_RTP_AUDIO, NS_JINGLE_RTP_VIDEO,

     NS_JINGLE_FILE_TRANSFER_5)

 FEATURE_BLACKLIST = [NS_CHATSTATES, NS_XHTML_IM, NS_ESESSION,

     NS_JINGLE_ICE_UDP, NS_JINGLE_RTP_AUDIO, NS_JINGLE_RTP_VIDEO,

     NS_JINGLE_FILE_TRANSFER_5]

-

+from gajim.common import app

 # Query entry status codes

 NEW = 0

 QUERIED = 1

 ################################################################################

 

 capscache = None

+muc_caps_cache = None

 def initialize(logger):

     """

     Initialize this module

     """

     global capscache

+    global muc_caps_cache

     capscache = CapsCache(logger)

+    muc_caps_cache = MucCapsCache()

 

 def client_supports(client_caps, requested_feature):

     lookup_item = client_caps.get_cache_lookup_strategy()

         key = (hash_method, hash)

         if key in self.__cache:

             del self.__cache[key]

+

+

+class MucCapsCache:

+

+    DiscoInfo = namedtuple('DiscoInfo', ['identities', 'features', 'data'])

+

+    def __init__(self):

+        self.cache = {}

+

+    def append(self, stanza):

+        jid = stanza.getFrom()

+        identities, features, data = [], [], []

+        query_childs = stanza.getQueryChildren()

+        if not query_childs:

+            app.log('gajim.muc').warning('%s returned empty disco info', jid)

+            return

+

+        for child in query_childs:

+            if child.getName() == 'identity':

+                attr = {}

+                for key in child.getAttrs().keys():

+                    attr[key] = child.getAttr(key)

+                identities.append(attr)

+            elif child.getName() == 'feature':

+                features.append(child.getAttr('var'))

+            elif child.getName() == 'x':

+                if child.getNamespace() == nbxmpp.NS_DATA:

+                    data.append(nbxmpp.DataForm(node=child))

+

+        self.cache[jid] = self.DiscoInfo(identities, features, data)

+

+    def is_cached(self, jid):

+        return jid in self.cache

+

+    def supports(self, jid, feature):

+        if jid in self.cache:

+            if feature in self.cache[jid].features:

+                return True

+        return False

+

+    def has_mam(self, jid):

+        try:

+            if nbxmpp.NS_MAM_2 in self.cache[jid].features:

+                return True

+            if nbxmpp.NS_MAM_1 in self.cache[jid].features:

+                return True

+        except (KeyError, AttributeError):

+            return False
 

             CREATE TABLE logs(

                     log_line_id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE,

+                    account_id INTEGER,

                     jid_id INTEGER,

                     contact_name TEXT,

                     time INTEGER,

                     marker INTEGER

             );

 

+            CREATE TABLE last_archive_message(

+                    jid_id INTEGER PRIMARY KEY UNIQUE,

+                    last_mam_id TEXT,

+                    oldest_mam_timestamp TEXT,

+                    last_muc_timestamp TEXT

+            );

+

             CREATE INDEX idx_logs_jid_id_time ON logs (jid_id, time DESC);

             '''

             )

                     'oauth2_client_id': [ opt_str, '0000000044077801', _('client_id for OAuth 2.0 authentication.')],

                     'oauth2_redirect_url': [ opt_str, 'https%3A%2F%2Fgajim.org%2Fmsnauth%2Findex.cgi', _('redirect_url for OAuth 2.0 authentication.')],

                     'opened_chat_controls': [opt_str, '', _('Space separated list of JIDs for which we want to re-open a chat window on next startup.')],

-                    'last_mam_id': [opt_str, '', _('Last MAM id we are syncronized with')],

-                    'mam_start_date': [opt_int, 0, _('The earliest date we requested MAM history for')],

             }, {}),

             'statusmsg': ({

                     'message': [ opt_str, '' ],

 import hmac

 import hashlib

 import json

+from functools import partial

 

 try:

     randomsource = random.SystemRandom()

         if obj.message is None:

             return

 

-        app.logger.insert_into_logs(jid, obj.timestamp, obj.kind,

+        app.logger.insert_into_logs(self.name, jid, obj.timestamp, obj.kind,

                                     message=obj.message,

                                     subject=obj.subject,

                                     additional_data=obj.additional_data,

                     self.archiving_namespace = nbxmpp.NS_MAM_2

                 elif nbxmpp.NS_MAM_1 in obj.features:

                     self.archiving_namespace = nbxmpp.NS_MAM_1

-                elif nbxmpp.NS_MAM in obj.features:

-                    self.archiving_namespace = nbxmpp.NS_MAM

                 if self.archiving_namespace:

                     self.archiving_supported = True

                     self.archiving_313_supported = True

             # Never join a room when invisible

             return

 

+        self.discoverMUC(

+            room_jid, partial(self._join_gc, nick, show, room_jid,

+                              password, change_nick, rejoin))

+

+    def _join_gc(self, nick, show, room_jid, password, change_nick, rejoin):

         # Check time first in the FAST table

         last_date = app.logger.get_room_last_message_time(

             self.name, room_jid)

         if app.config.get('send_sha_in_gc_presence'):

             p = self.add_sha(p)

         self.add_lang(p)

-        if not change_nick:

-            t = p.setTag(nbxmpp.NS_MUC + ' x')

+        if change_nick:

+            self.connection.send(p)

+            return

+

+        t = p.setTag(nbxmpp.NS_MUC + ' x')

+        if muc_caps_cache.has_mam(room_jid):

+            # The room is MAM capable dont get MUC History

+            t.setTag('history', {'maxchars': '0'})

+        else:

+            # Request MUC History (not MAM)

             tags = {}

             timeout = app.config.get_per('rooms', room_jid,

-                'muc_restore_timeout')

+                                         'muc_restore_timeout')

             if timeout is None or timeout == -2:

                 timeout = app.config.get('muc_restore_timeout')

             if last_date == 0 and timeout >= 0:

                 tags['maxstanzas'] = nb

             if tags:

                 t.setTag('history', tags)

-            if password:

-                t.setTagData('password', password)

+

+        if password:

+            t.setTagData('password', password)

         self.connection.send(p)

 

     def _nec_gc_message_outgoing(self, obj):

 from gajim.common import app

 from gajim.common import dataforms

 from gajim.common import jingle_xtls

+from gajim.common.caps_cache import muc_caps_cache

 from gajim.common.commands import ConnectionCommands

 from gajim.common.pubsub import ConnectionPubSub

 from gajim.common.protocol.caps import ConnectionCaps

         id_ = self._discover(nbxmpp.NS_DISCO_INFO, jid, node, id_prefix)

         self.disco_info_ids.append(id_)

 

+    def discoverMUC(self, jid, callback):

+        disco_info = nbxmpp.Iq(typ='get', to=jid, queryNS=nbxmpp.NS_DISCO_INFO)

+        self.connection.SendAndCallForResponse(

+            disco_info, self.received_muc_info, {'callback': callback})

+

+    def received_muc_info(self, conn, stanza, callback):

+        if nbxmpp.isResultNode(stanza):

+            app.log('gajim.muc').info(

+                'Received MUC DiscoInfo for %s', stanza.getFrom())

+            muc_caps_cache.append(stanza)

+            callback()

+        else:

+            error = stanza.getError()

+            if error == 'item-not-found':

+                # Groupchat does not exist

+                callback()

+                return

+            app.nec.push_incoming_event(

+                InformationEvent(None, conn=self,

+                                 level='error',

+                                 pri_txt=_('Unable to join Groupchat'),

+                                 sec_txt=error))

+

     def request_register_agent_info(self, agent):

         if not self.connection or self.connected < 2:

             return None

             self._nec_message_received)

         app.ged.register_event_handler('mam-message-received', ged.CORE,

             self._nec_message_received)

+        app.ged.register_event_handler('mam-gc-message-received', ged.CORE,

+            self._nec_message_received)

         app.ged.register_event_handler('decrypted-message-received', ged.CORE,

             self._nec_decrypted_message_received)

         app.ged.register_event_handler('gc-message-received', ged.CORE,

             self._nec_message_received)

         app.ged.remove_event_handler('mam-message-received', ged.CORE,

             self._nec_message_received)

+        app.ged.remove_event_handler('mam-gc-message-received', ged.CORE,

+            self._nec_message_received)

         app.ged.remove_event_handler('decrypted-message-received', ged.CORE,

             self._nec_decrypted_message_received)

         app.ged.remove_event_handler('gc-message-received', ged.CORE,

         app.config.should_log(self.name, obj.jid):

             show = app.logger.convert_show_values_to_db_api_values(obj.show)

             if show is not None:

-                app.logger.insert_into_logs(nbxmpp.JID(obj.jid).getStripped(),

+                app.logger.insert_into_logs(self.name,

+                                            nbxmpp.JID(obj.jid).getStripped(),

                                             time_time(),

                                             KindConstant.STATUS,

                                             message=obj.status,

             # if not obj.nick, it means message comes from room itself

             # usually it hold description and can be send at each connection

             # so don't store it in logs

-            app.logger.insert_into_logs(obj.jid,

+            app.logger.insert_into_logs(self.name,

+                                        obj.jid,

                                         obj.timestamp,

                                         KindConstant.GC_MSG,

                                         message=obj.msgtxt,

         subject = msg.getSubject()

 

         if session.is_loggable():

-            app.logger.insert_into_logs(nbxmpp.JID(frm).getStripped(),

+            app.logger.insert_into_logs(self.name,

+                                        nbxmpp.JID(frm).getStripped(),

                                         tim,

                                         KindConstant.ERROR,

                                         message=error_msg,

         app.nec.register_incoming_event(ArchivingErrorReceivedEvent)

         app.nec.register_incoming_event(

             Archiving313PreferencesChangedReceivedEvent)

-        app.nec.register_incoming_event(

-            ArchivingFinishedLegacyReceivedEvent)

-        app.nec.register_incoming_event(

-            ArchivingFinishedReceivedEvent)

         app.nec.register_incoming_event(NotificationEvent)

 

         app.ged.register_event_handler('http-auth-received', ged.CORE,

         con.RegisterHandler('iq', self._IqPingCB, 'get', nbxmpp.NS_PING)

         con.RegisterHandler('iq', self._SearchCB, 'result', nbxmpp.NS_SEARCH)

         con.RegisterHandler('iq', self._PrivacySetCB, 'set', nbxmpp.NS_PRIVACY)

-        con.RegisterHandler('iq', self._ArchiveCB, ns=nbxmpp.NS_MAM)

         con.RegisterHandler('iq', self._ArchiveCB, ns=nbxmpp.NS_MAM_1)

         con.RegisterHandler('iq', self._ArchiveCB, ns=nbxmpp.NS_MAM_2)

         con.RegisterHandler('iq', self._PubSubCB, 'result')

             show = app.logger.convert_show_values_to_db_api_values(self.show)

             if show is not None:

                 fjid = nbxmpp.JID(self.fjid)

-                app.logger.insert_into_logs(fjid.getStripped(),

+                app.logger.insert_into_logs(self.conn.name,

+                                            fjid.getStripped(),

                                             time_time(),

                                             KindConstant.GCSTATUS,

                                             contact_name=fjid.getResource(),

         :stanza:        Complete stanza Node

         :forwarded:     Forwarded Node

         :result:        Result Node

-        :unique_id:     The unique stable id

         '''

         self._set_base_event_vars_as_attributes(base_event)

         self.additional_data = {}

         self.encrypted = False

         self.groupchat = False

+        self.nick = None

 

     def generate(self):

         archive_jid = self.stanza.getFrom()

-        own_jid = self.conn.get_own_jid()

+        own_jid = self.conn.get_own_jid().getStripped()

         if archive_jid and not archive_jid.bareMatch(own_jid):

             # MAM Message not from our Archive

-            log.info('MAM message not from our user archive')

             return False

 

         self.msg_ = self.forwarded.getTag('message', protocol=True)

 

         if self.msg_.getType() == 'groupchat':

-            log.info('Received groupchat message from user archive')

             return False

 

         # use stanza-id as unique-id

         self.unique_id, origin_id = self.get_unique_id()

 

         # Check for duplicates

-        if app.logger.find_stanza_id(self.unique_id, origin_id):

+        if app.logger.find_stanza_id(own_jid, self.unique_id, origin_id):

             return

 

         self.msgtxt = self.msg_.getTagData('body')

-        self.query_id = self.result.getAttr('queryid')

 

         frm = self.msg_.getFrom()

         # Some servers dont set the 'to' attribute when

         return True

 

     def get_unique_id(self):

+        stanza_id = self.get_stanza_id(self.result, query=True)

         if self.conn.get_own_jid().bareMatch(self.msg_.getFrom()):

-            # On our own Messages we have to check for both

-            # stanza-id and origin-id, because other resources

-            # maybe not support origin-id

-            stanza_id = None

-            if self.result.getNamespace() == nbxmpp.NS_MAM_2:

-                # Only mam:2 ensures valid stanza-id

-                stanza_id = self.get_stanza_id(self.result, query=True)

-

-            # try always to get origin-id because its a message

-            # we sent.

+            # message we sent

             origin_id = self.msg_.getOriginID()

-

             return stanza_id, origin_id

 

         # A message we received

-        elif self.result.getNamespace() == nbxmpp.NS_MAM_2:

-            # Only mam:2 ensures valid stanza-id

-            return self.get_stanza_id(self.result, query=True), None

-        return None, None

+        return stanza_id, None

+

+class MamGcMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):

+    name = 'mam-gc-message-received'

+    base_network_events = ['raw-mam-message-received']

+

+    def __init__(self, name, base_event):

+        '''

+        Pre-Generated attributes on self:

+

+        :conn:          Connection instance

+        :stanza:        Complete stanza Node

+        :forwarded:     Forwarded Node

+        :result:        Result Node

+        '''

+        self._set_base_event_vars_as_attributes(base_event)

+        self.additional_data = {}

+        self.encrypted = False

+        self.groupchat = True

+        self.kind = KindConstant.GC_MSG

+

+    def generate(self):

+        self.msg_ = self.forwarded.getTag('message', protocol=True)

+

+        if self.msg_.getType() != 'groupchat':

+            return False

+

+        self.room_jid = self.stanza.getFrom().getStripped()

+

+        self.unique_id = self.get_stanza_id(self.result, query=True)

+

+        # Check for duplicates

+        if app.logger.find_stanza_id(self.room_jid, self.unique_id,

+                                     groupchat=True):

+            return

+

+        self.msgtxt = self.msg_.getTagData('body')

+        self.with_ = self.msg_.getFrom().getStripped()

+        self.nick = self.msg_.getFrom().getResource()

+

+        # Get the real jid if we have it

+        self.real_jid = None

+        muc_user = self.msg_.getTag('x', namespace=nbxmpp.NS_MUC_USER)

+        if muc_user is not None:

+            self.real_jid = muc_user.getTagAttr('item', 'jid')

+

+        delay = self.forwarded.getTagAttr(

+            'delay', 'stamp', namespace=nbxmpp.NS_DELAY2)

+        if delay is None:

+            log.error('Received MAM message without timestamp')

+            return

+

+        self.timestamp = helpers.parse_datetime(

+            delay, check_utc=True, epoch=True)

+        if self.timestamp is None:

+            log.error('Received MAM message with invalid timestamp: %s', delay)

+            return

+

+        # Save timestamp added by the user

+        user_delay = self.msg_.getTagAttr(

+            'delay', 'stamp', namespace=nbxmpp.NS_DELAY2)

+        if user_delay is not None:

+            self.user_timestamp = helpers.parse_datetime(

+                user_delay, check_utc=True, epoch=True)

+            if self.user_timestamp is None:

+                log.warning('Received MAM message with '

+                            'invalid user timestamp: %s', user_delay)

+

+        log.debug('Received mam-gc-message: unique id: %s', self.unique_id)

+        return True

 

 class MamDecryptedMessageReceivedEvent(nec.NetworkIncomingEvent, HelperEvent):

     name = 'mam-decrypted-message-received'

 

         self.get_oob_data(self.msg_)

 

+        if self.groupchat:

+            return True

+

         self.is_pm = app.logger.jid_is_room_jid(self.with_.getStripped())

         if self.is_pm is None:

             # Check if this event is triggered after a disco, so we dont

         # Check groupchat messages for duplicates,

         # We do this because of MUC History messages

         if self.stanza.getType() == 'groupchat':

-            if app.logger.find_stanza_id(self.unique_id):

+            if app.logger.find_stanza_id(self.stanza.getFrom().getStripped(),

+                                         self.unique_id,

+                                         groupchat=True):

                 return

 

         address_tag = self.stanza.getTag('addresses',

                 self.forwarded = True

 

         result = self.stanza.getTag('result', protocol=True)

-        if result and result.getNamespace() in (nbxmpp.NS_MAM,

-                                                nbxmpp.NS_MAM_1,

+        if result and result.getNamespace() in (nbxmpp.NS_MAM_1,

                                                 nbxmpp.NS_MAM_2):

+

+            if result.getAttr('queryid') not in self.conn.mam_query_ids:

+                log.warning('Invalid MAM Message: unknown query id')

+                log.debug(self.stanza)

+                return

+

             forwarded = result.getTag('forwarded',

                                       namespace=nbxmpp.NS_FORWARD,

                                       protocol=True)

                              conn=self.conn,

                              stanza=self.stanza,

                              forwarded=forwarded,

-                             result=result,

-                             stanza_id=self.unique_id))

+                             result=result))

             return

 

         # Mediated invitation?

             self.error_msg = self.stanza.getErrorMsg()

             return True

 

+class ArchivingCountReceived(nec.NetworkIncomingEvent):

+    name = 'archiving-count-received'

+    base_network_events = []

+

+    def generate(self):

+        return True

+

+class ArchivingIntervalFinished(nec.NetworkIncomingEvent):

+    name = 'archiving-interval-finished'

+    base_network_events = []

+

+    def generate(self):

+        return True

+

+class ArchivingQueryID(nec.NetworkIncomingEvent):

+    name = 'archiving-query-id'

+    base_network_events = []

+

+    def generate(self):

+        return True

+

 class Archiving313PreferencesChangedReceivedEvent(nec.NetworkIncomingEvent):

     name = 'archiving-313-preferences-changed-received'

     base_network_events = ['archiving-received']

 

         return True

 

-class ArchivingFinishedReceivedEvent(nec.NetworkIncomingEvent):

-    name = 'archiving-finished'

-    base_network_events = ['archiving-received']

-

-    def generate(self):

-        self.conn = self.base_event.conn

-        self.stanza = self.base_event.stanza

-        self.type_ = self.base_event.type_

-        self.fin = self.stanza.getTag('fin')

-

-        if self.type_ != 'result' or not self.fin:

-            return

-

-        self.query_id = self.fin.getAttr('queryid')

-        if not self.query_id:

-            return

-

-        return True

-

-class ArchivingFinishedLegacyReceivedEvent(nec.NetworkIncomingEvent):

-    name = 'archiving-finished-legacy'

-    base_network_events = ['raw-message-received']

-

-    def generate(self):

-        self.conn = self.base_event.conn

-        self.stanza = self.base_event.stanza

-        self.fin = self.stanza.getTag('fin', namespace=nbxmpp.NS_MAM)

-

-        if not self.fin:

-            return

-

-        self.query_id = self.fin.getAttr('queryid')

-        if not self.query_id:

-            return

-

-        return True

-

 class AccountCreatedEvent(nec.NetworkIncomingEvent):

     name = 'account-created'

     base_network_events = []

     TOOLTIP = 125

     VCARD = 200

 

+class ArchiveState(IntEnum):

+    NEVER = 0

+    ALL = 1

+

 

 THANKS = u"""\

 Alexander Futász

             return [user['jid'] for user in family]

         return [jid]

 

+    def get_account_id(self, account):

+        jid = app.get_jid_from_account(account)

+        return self.get_jid_id(jid, type_=JIDConstant.NORMAL_TYPE)

+

     def get_jid_id(self, jid, kind=None, type_=None):

         """

         Get the jid id from a jid.

             return True

         return False

 

-    def find_stanza_id(self, stanza_id, origin_id=None):

+    def find_stanza_id(self, archive_jid, stanza_id, origin_id=None,

+                       groupchat=False):

         """

         Checks if a stanza-id is already in the `logs` table

 

+        :param archive_jid: The jid of the archive the stanza-id belongs to

+

         :param stanza_id:   The stanza-id

 

+        :param origin_id:   The origin-id

+

+        :param groupchat:   stanza-id is from a groupchat

+

         return True if the stanza-id was found

         """

         ids = []

         if not ids:

             return False

 

+        archive_id = self.get_jid_id(archive_jid)

+        if groupchat:

+            column = 'jid_id'

+        else:

+            column = 'account_id'

+

         sql = '''

               SELECT stanza_id FROM logs

-              WHERE stanza_id IN ({values}) LIMIT 1

-              '''.format(values=', '.join('?' * len(ids)))

+              WHERE stanza_id IN ({values}) AND {archive} = ? LIMIT 1

+              '''.format(values=', '.join('?' * len(ids)),

+                         archive=column)

 

-        result = self.con.execute(sql, tuple(ids)).fetchone()

+        result = self.con.execute(sql, tuple(ids) + (archive_id,)).fetchone()

 

         if result is not None:

             log.info('Found duplicated message, stanza-id: %s, origin-id: %s',

         """

         return self.get_jid_id(jid, kind, type_)

 

-    def insert_into_logs(self, jid, time_, kind, unread=True, **kwargs):

+    def insert_into_logs(self, account, jid, time_, kind,

+                         unread=True, **kwargs):

         """

         Insert a new message into the `logs` table

 

                         a field in the `logs` table

         """

         jid_id = self.get_jid_id(jid, kind=kind)

- 

+        account_id = self.get_account_id(account)

+

         if 'additional_data' in kwargs:

             if not kwargs['additional_data']:

                 del kwargs['additional_data']

             else:

                 kwargs['additional_data'] = json.dumps(kwargs["additional_data"])

- 

+

         sql = '''

-              INSERT INTO logs (jid_id, time, kind, {columns})

-              VALUES (?, ?, ?, {values})

+              INSERT INTO logs (account_id, jid_id, time, kind, {columns})

+              VALUES (?, ?, ?, ?, {values})

               '''.format(columns=', '.join(kwargs.keys()),

                          values=', '.join('?' * len(kwargs)))

 

-        lastrowid = self.con.execute(sql, (jid_id, time_, kind) + tuple(kwargs.values())).lastrowid

+        lastrowid = self.con.execute(

+            sql, (account_id, jid_id, time_, kind) + tuple(kwargs.values())).lastrowid

 

         log.info('Insert into DB: jid: %s, time: %s, kind: %s, stanza_id: %s',

                  jid, time_, kind, kwargs.get('stanza_id', None))

             '''

         self.con.execute(sql, (sha, account_jid_id, jid_id))

         self._timeout_commit()

+

+    def get_archive_timestamp(self, jid, type_=None):

+        """

+        Get the last archive id/timestamp for a jid

+

+        :param jid:     The jid that belongs to the avatar

+

+        """

+        jid_id = self.get_jid_id(jid, type_=type_)

+        sql = '''SELECT * FROM last_archive_message WHERE jid_id = ?'''

+        return self.con.execute(sql, (jid_id,)).fetchone()

+

+    def set_archive_timestamp(self, jid, **kwargs):

+        """

+        Set the last archive id/timestamp

+

+        :param jid:                     The jid that belongs to the avatar

+

+        :param last_mam_id:             The last MAM result id

+

+        :param oldest_mam_timestamp:    The oldest date we requested MAM

+                                        history for

+

+        :param last_muc_timestamp:      The timestamp of the last message we

+                                        received in a MUC

+

+        """

+        jid_id = self.get_jid_id(jid)

+        exists = self.get_archive_timestamp(jid)

+        if not exists:

+            sql = '''INSERT INTO last_archive_message VALUES (?, ?, ?, ?)'''

+            self.con.execute(sql, (jid_id,

+                                   kwargs.get('last_mam_id', None),

+                                   kwargs.get('oldest_mam_timestamp', None),

+                                   kwargs.get('last_muc_timestamp', None)))

+        else:

+            args = ' = ?, '.join(kwargs.keys()) + ' = ?'

+            sql = '''UPDATE last_archive_message SET {}

+                     WHERE jid_id = ?'''.format(args)

+            self.con.execute(sql, tuple(kwargs.values()) + (jid_id,))

+        log.info('Save archive timestamps: %s', kwargs)

+        self._timeout_commit()
 from gajim.common import app

 from gajim.common import ged

 from gajim.common.logger import KindConstant, JIDConstant

+from gajim.common.const import ArchiveState

 import gajim.common.connection_handlers_events as ev

 

 log = logging.getLogger('gajim.c.message_archiving')

         self.archiving_313_supported = False

         self.mam_awaiting_disco_result = {}

         self.iq_answer = []

-        self.mam_query_date = None

-        self.mam_query_id = None

+        self.mam_query_ids = []

         app.nec.register_incoming_event(ev.MamMessageReceivedEvent)

-        app.ged.register_event_handler('archiving-finished-legacy', ged.CORE,

-            self._nec_result_finished)

-        app.ged.register_event_handler('archiving-finished', ged.CORE,

-            self._nec_result_finished)

+        app.nec.register_incoming_event(ev.MamGcMessageReceivedEvent)

         app.ged.register_event_handler('agent-info-error-received', ged.CORE,

             self._nec_agent_info_error)

         app.ged.register_event_handler('agent-info-received', ged.CORE,

             self._nec_archiving_313_preferences_changed_received)

 

     def cleanup(self):

-        app.ged.remove_event_handler('archiving-finished-legacy', ged.CORE,

-            self._nec_result_finished)

-        app.ged.remove_event_handler('archiving-finished', ged.CORE,

-            self._nec_result_finished)

         app.ged.remove_event_handler('agent-info-error-received', ged.CORE,

             self._nec_agent_info_error)

         app.ged.remove_event_handler('agent-info-received', ged.CORE,

                     None, disco=True, **vars(msg_obj)))

         del self.mam_awaiting_disco_result[obj.jid]

 

-    def _nec_result_finished(self, obj):

-        if obj.conn.name != self.name:

+    @staticmethod

+    def parse_iq(stanza, query_id):

+        if not nbxmpp.isResultNode(stanza):

+            log.error('Error on MAM query: %s', stanza.getError())

+            raise InvalidMamIQ

+

+        fin = stanza.getTag('fin')

+        if fin is None:

+            log.error('Malformed MAM query result received: %s', stanza)

+            raise InvalidMamIQ

+

+        if fin.getAttr('queryid') != query_id:

+            log.error('Result with unknown query id received')

+            raise InvalidMamIQ

+

+        set_ = fin.getTag('set', namespace=nbxmpp.NS_RSM)

+        if set_ is None:

+            log.error(

+                'Malformed MAM query result received (no "set" Node): %s',

+                stanza)

+            raise InvalidMamIQ

+        return fin, set_

+

+    def parse_from_jid(self, stanza):

+        jid = stanza.getFrom()

+        if jid is None:

+            # No from means, iq from our own archive

+            jid = self.get_own_jid().getStripped()

+        else:

+            jid = jid.getStripped()

+        return jid

+

+    def _result_finished(self, conn, stanza, query_id, start_date, groupchat):

+        try:

+            fin, set_ = self.parse_iq(stanza, query_id)

+        except InvalidMamIQ:

             return

 

-        if obj.query_id != self.mam_query_id:

+        last = set_.getTagData('last')

+        if last is None:

+            log.info('End of MAM query, no items retrieved')

+            return

+

+        jid = self.parse_from_jid(stanza)

+        complete = fin.getAttr('complete')

+        app.logger.set_archive_timestamp(jid, last_mam_id=last)

+        if complete != 'true':

+            self.mam_query_ids.remove(query_id)

+            query_id = self.get_query_id()

+            query = self.get_archive_query(query_id, jid=jid, after=last)

+            self._send_archive_query(query, query_id, groupchat=groupchat)

+        else:

+            self.mam_query_ids.remove(query_id)

+            if start_date is not None:

+                app.logger.set_archive_timestamp(

+                    jid,

+                    last_mam_id=last,

+                    oldest_mam_timestamp=start_date.timestamp())

+            log.info('End of MAM query, last mam id: %s', last)

+

+    def _intervall_result_finished(self, conn, stanza, query_id,

+                                   start_date, end_date, event_id):

+        try:

+            fin, set_ = self.parse_iq(stanza, query_id)

+        except InvalidMamIQ:

+            return

+

+        self.mam_query_ids.remove(query_id)

+        jid = self.parse_from_jid(stanza)

+        if start_date:

+            timestamp = start_date.timestamp()

+        else:

+            timestamp = ArchiveState.ALL

+

+        last = set_.getTagData('last')

+        if last is None:

+            app.nec.push_incoming_event(ev.ArchivingIntervalFinished(

+                None, event_id=event_id))

+            app.logger.set_archive_timestamp(

+                jid, oldest_mam_timestamp=timestamp)

+            log.info('End of MAM query, no items retrieved')

+            return

+

+        complete = fin.getAttr('complete')

+        if complete != 'true':

+            self.request_archive_interval(event_id, start_date, end_date, last)

+        else:

+            log.info('query finished')

+            app.logger.set_archive_timestamp(

+                jid, oldest_mam_timestamp=timestamp)

+            app.nec.push_incoming_event(ev.ArchivingIntervalFinished(

+                None, event_id=event_id, stanza=stanza))

+

+    def _received_count(self, conn, stanza, query_id, event_id):

+        try:

+            _, set_ = self.parse_iq(stanza, query_id)

+        except InvalidMamIQ:

             return

 

-        set_ = obj.fin.getTag('set', namespace=nbxmpp.NS_RSM)

-        if set_:

-            last = set_.getTagData('last')

-            complete = obj.fin.getAttr('complete')

-            if last:

-                app.config.set_per('accounts', self.name, 'last_mam_id', last)

-                if complete != 'true':

-                    self.request_archive(self.get_query_id(), after=last)

-            if complete == 'true':

-                self.mam_query_id = None

-                if self.mam_query_date:

-                    app.config.set_per(

-                        'accounts', self.name,

-                        'mam_start_date', self.mam_query_date.timestamp())

-                    self.mam_query_date = None

+        self.mam_query_ids.remove(query_id)

+

+        count = set_.getTagData('count')

+        log.info('message count received: %s', count)

+        app.nec.push_incoming_event(ev.ArchivingCountReceived(

+            None, event_id=event_id, count=count))

 

     def _nec_mam_decrypted_message_received(self, obj):

         if obj.conn.name != self.name:

         duplicate = app.logger.search_for_duplicate(

             obj.with_, obj.timestamp, obj.msgtxt)

         if duplicate:

-            return

-        app.logger.insert_into_logs(

-            obj.with_, obj.timestamp, obj.kind,

-            unread=False,

-            message=obj.msgtxt,

-            additional_data=obj.additional_data,

-            stanza_id=obj.unique_id)

+            # dont propagate the event further

+            return True

+        app.logger.insert_into_logs(self.name,

+                                    obj.with_,

+                                    obj.timestamp,

+                                    obj.kind,

+                                    unread=False,

+                                    message=obj.msgtxt,

+                                    contact_name=obj.nick,

+                                    additional_data=obj.additional_data,

+                                    stanza_id=obj.unique_id)

 

     def get_query_id(self):

-        self.mam_query_id = self.connection.getAnID()

-        return self.mam_query_id

+        query_id = self.connection.getAnID()

+        self.mam_query_ids.append(query_id)

+        return query_id

 

     def request_archive_on_signin(self):

-        mam_id = app.config.get_per('accounts', self.name, 'last_mam_id')

+        own_jid = self.get_own_jid().getStripped()

+        archive = app.logger.get_archive_timestamp(own_jid)

+

+        # Migration of last_mam_id from config to DB

+        if archive is not None:

+            mam_id = archive.last_mam_id

+        else:

+            mam_id = app.config.get_per('accounts', self.name, 'last_mam_id')

+

+        start_date = None

         query_id = self.get_query_id()

         if mam_id:

-            self.request_archive(query_id, after=mam_id)

+            log.info('MAM query after: %s', mam_id)

+            query = self.get_archive_query(query_id, after=mam_id)

         else:

             # First Start, we request the last week

-            self.mam_query_date = datetime.utcnow() - timedelta(days=7)

-            log.info('First start: query archive start: %s', self.mam_query_date)

-            self.request_archive(query_id, start=self.mam_query_date)

+            start_date = datetime.utcnow() - timedelta(days=7)

+            log.info('First start: query archive start: %s', start_date)

+            query = self.get_archive_query(query_id, start=start_date)

+        self._send_archive_query(query, query_id, start_date)

+

+    def request_archive_on_muc_join(self, jid):

+        archive = app.logger.get_archive_timestamp(

+            jid, type_=JIDConstant.ROOM_TYPE)

+        query_id = self.get_query_id()

+        start_date = None

+        if archive is not None:

+            log.info('Query Groupchat MAM Archive %s after %s:',

+                     jid, archive.last_mam_id)

+            query = self.get_archive_query(

+                query_id, jid=jid, after=archive.last_mam_id)

+        else:

+            # First Start, we dont request history

+            # Depending on what a MUC saves, there could be thousands

+            # of Messages even in just one day.

+            start_date = datetime.utcnow() - timedelta(days=1)

+            log.info('First join: query archive %s from: %s', jid, start_date)

+            query = self.get_archive_query(query_id, jid=jid, start=start_date)

+        self._send_archive_query(query, query_id, start_date, groupchat=True)

+

+    def request_archive_count(self, event_id, start_date, end_date):

+        query_id = self.get_query_id()

+        query = self.get_archive_query(

+            query_id, start=start_date, end=end_date, max_=0)

+        self.connection.SendAndCallForResponse(

+            query, self._received_count, {'query_id': query_id,

+                                          'event_id': event_id})

+

+    def request_archive_interval(self, event_id, start_date,

+                                 end_date, after=None):

+        query_id = self.get_query_id()

+        query = self.get_archive_query(query_id, start=start_date,

+                                       end=end_date, after=after, max_=30)

+        app.nec.push_incoming_event(ev.ArchivingQueryID(

+            None, event_id=event_id, query_id=query_id))

+        self.connection.SendAndCallForResponse(

+            query, self._intervall_result_finished, {'query_id': query_id,

+                                                     'start_date': start_date,

+                                                     'end_date': end_date,

+                                                     'event_id': event_id})

 

-    def request_archive(self, query_id, start=None, end=None, with_=None,

-                        after=None, max_=30):

+    def _send_archive_query(self, query, query_id, start_date=None,

+                            groupchat=False):

+        self.connection.SendAndCallForResponse(

+            query, self._result_finished, {'query_id': query_id,

+                                           'start_date': start_date,

+                                           'groupchat': groupchat})

+

+    def get_archive_query(self, query_id, jid=None, start=None, end=None, with_=None,

+                          after=None, max_=30):

         namespace = self.archiving_namespace

-        iq = nbxmpp.Iq('set')

+        iq = nbxmpp.Iq('set', to=jid)

         query = iq.addChild('query', namespace=namespace)

         form = query.addChild(node=nbxmpp.DataForm(typ='submit'))

         field = nbxmpp.DataField(typ='hidden',

         if after:

             set_.setTagData('after', after)

         query.setAttr('queryid', query_id)

-        id_ = self.connection.getAnID()

-        iq.setID(id_)

-        self.connection.send(iq)

+        return iq

 

     def request_archive_preferences(self):

         if not app.account_is_connected(self.name):

         app.nec.push_incoming_event(ev.ArchivingReceivedEvent(None, conn=self,

             stanza=iq_obj))

         raise nbxmpp.NodeProcessed

+

+

+class InvalidMamIQ(Exception):

+    pass
             self.update_config_to_016105()

         if old < [0, 16, 11, 1] and new >= [0, 16, 11, 1]:

             self.update_config_to_016111()

+        if old < [0, 16, 11, 2] and new >= [0, 16, 11, 2]:

+            self.update_config_to_016112()

 

         app.logger.init_vars()

         app.logger.attach_cache_database()

             log.exception('Error')

         con.close()

         app.config.set('version', '0.16.11.1')

+

+    def update_config_to_016112(self):

+        con = sqlite.connect(logger.LOG_DB_PATH)

+        cur = con.cursor()

+        try:

+            cur.executescript(

+                    '''

+                    CREATE TABLE IF NOT EXISTS last_archive_message(

+                        jid_id INTEGER PRIMARY KEY UNIQUE,

+                        last_mam_id TEXT,

+                        oldest_mam_timestamp TEXT,

+                        last_muc_timestamp TEXT

+                        );

+                    ALTER TABLE logs ADD COLUMN 'account_id' INTEGER;

+                    '''

+            )

+            con.commit()

+        except sqlite.OperationalError:

+            log.exception('Error')

+        con.close()

+        app.config.set('version', '0.16.11.2')
 from gajim import cell_renderer_image

 from gajim import dataforms_widget

 from gajim.common.const import AvatarSize

+from gajim.common.caps_cache import muc_caps_cache

 import nbxmpp

 

 from enum import IntEnum, unique

             self._nec_gc_presence_received)

         app.ged.register_event_handler('gc-message-received', ged.GUI1,

             self._nec_gc_message_received)

+        app.ged.register_event_handler('mam-decrypted-message-received',

+            ged.GUI1, self._nec_mam_decrypted_message_received)

         app.ged.register_event_handler('vcard-published', ged.GUI1,

             self._nec_vcard_published)

         app.ged.register_event_handler('update-gc-avatar', ged.GUI1,

                                 obj.contact.name, obj.contact.avatar_sha)

         self.draw_avatar(obj.contact)

 

+    def _nec_mam_decrypted_message_received(self, obj):

+        if not obj.groupchat:

+            return

+        if obj.room_jid != self.room_jid:

+            return

+        self.print_conversation(

+            obj.msgtxt, contact=obj.nick,

+            tim=obj.timestamp, encrypted=obj.encrypted,

+            msg_stanza_id=obj.unique_id,

+            additional_data=obj.additional_data)

+

     def _nec_gc_message_received(self, obj):

         if obj.room_jid != self.room_jid or obj.conn.name != self.account:

             return

             GLib.source_remove(self.autorejoin)

         self.autorejoin = None

 

+        if muc_caps_cache.has_mam(self.room_jid):

+            # Request MAM

+            app.connections[self.account].request_archive_on_muc_join(

+                self.room_jid)

+

         app.gc_connected[self.account][self.room_jid] = True

         ChatControlBase.got_connected(self)

         self.list_treeview.set_model(self.model)

 

 import logging

 from enum import IntEnum

-from datetime import datetime, timedelta, timezone

+from datetime import datetime, timedelta

 

 import nbxmpp

 from gi.repository import Gtk, GLib

 from gajim.common import app

 from gajim.common import ged

 from gajim.gtkgui_helpers import get_icon_pixmap

+from gajim.common.const import ArchiveState

 

 log = logging.getLogger('gajim.c.message_archiving')

 

+

 class Pages(IntEnum):

     TIME = 0

     SYNC = 1

     SUMMARY = 2

 

-class ArchiveState(IntEnum):

-    NEVER = 0

-    ALL = 1

 

 class HistorySyncAssistant(Gtk.Assistant):

     def __init__(self, account, parent):

         self.timedelta = None

         self.now = datetime.utcnow()

         self.query_id = None

-        self.count_query_id = None

         self.start = None

         self.end = None

         self.next = None

         self.hide_buttons()

+        self.event_id = id(self)

+

+        own_jid = self.con.get_own_jid().getStripped()

+        archive = app.logger.get_archive_timestamp(own_jid)

+

+        if archive is not None:

+            mam_start = float(archive.oldest_mam_timestamp)

+        else:

+            # Migration from old config value

+            mam_start = app.config.get_per(

+                'accounts', account, 'mam_start_date')

 

-        mam_start = app.config.get_per('accounts', account, 'mam_start_date')

         if not mam_start or mam_start == ArchiveState.NEVER:

             self.current_start = self.now

         elif mam_start == ArchiveState.ALL:

 

         self.download_history = DownloadHistoryPage(self)

         self.append_page(self.download_history)

-        self.set_page_type(self.download_history, Gtk.AssistantPageType.PROGRESS)

+        self.set_page_type(self.download_history,

+                           Gtk.AssistantPageType.PROGRESS)

         self.set_page_complete(self.download_history, True)

 

         self.summary = SummaryPage(self)

         self.set_page_type(self.summary, Gtk.AssistantPageType.SUMMARY)

         self.set_page_complete(self.summary, True)

 

-        app.ged.register_event_handler('archiving-finished',

-                                         ged.PRECORE,

-                                         self._nec_archiving_finished)

+        app.ged.register_event_handler('archiving-count-received',

+                                       ged.GUI1,

+                                       self._received_count)

+        app.ged.register_event_handler('archiving-query-id',

+                                       ged.GUI1,

+                                       self._new_query_id)

+        app.ged.register_event_handler('archiving-interval-finished',

+                                       ged.GUI1,

+                                       self._received_finished)

         app.ged.register_event_handler('raw-mam-message-received',

-                                         ged.PRECORE,

-                                         self._nec_mam_message_received)

+                                       ged.PRECORE,

+                                       self._nec_mam_message_received)

 

         self.connect('prepare', self.on_page_change)

         self.connect('destroy', self.on_destroy)

             self.set_current_page(Pages.SUMMARY)

             self.summary.nothing_to_do()

 

-        if self.con.mam_query_id:

-            self.set_current_page(Pages.SUMMARY)

-            self.summary.query_already_running()

+        # if self.con.mam_query_ids:

+        #     self.set_current_page(Pages.SUMMARY)

+        #     self.summary.query_already_running()

 

         self.show_all()

 

             self.start = self.now - self.timedelta

         self.end = self.current_start

 

-        log.info('config: get mam_start_date: %s', self.current_start)

+        log.info('get mam_start_date: %s', self.current_start)

         log.info('now: %s', self.now)

         log.info('start: %s', self.start)

         log.info('end: %s', self.end)

 

-        self.query_count()

+        self.con.request_archive_count(self.event_id, self.start, self.end)

+

+    def _received_count(self, event):

+        if event.event_id != self.event_id:

+            return

+        if event.count is not None:

+            self.download_history.count = int(event.count)

+        self.con.request_archive_interval(self.event_id, self.start, self.end)

+

+    def _received_finished(self, event):

+        if event.event_id != self.event_id:

+            return

+        log.info('query finished')

+        GLib.idle_add(self.download_history.finished)

+        self.set_current_page(Pages.SUMMARY)

+        self.summary.finished()

+

+    def _new_query_id(self, event):

+        if event.event_id != self.event_id:

+            return

+        self.query_id = event.query_id

+

+    def _nec_mam_message_received(self, obj):

+        if obj.conn.name != self.account:

+            return

 

-    def query_count(self):

-        self.count_query_id = self.con.connection.getAnID()

-        self.con.request_archive(self.count_query_id,

-                                 start=self.start,

-                                 end=self.end,

-                                 max_=0)

+        if obj.result.getAttr('queryid') != self.query_id:

+            return

 

-    def query_messages(self, last=None):

-        self.query_id = self.con.connection.getAnID()

-        self.con.request_archive(self.query_id,

-                                 start=self.start,

-                                 end=self.end,

-                                 after=last,

-                                 max_=30)

+        log.debug('received message')

+        GLib.idle_add(self.download_history.set_fraction)

 

     def on_row_selected(self, listbox, row):

         self.timedelta = row.get_child().get_delta()

             self.prepare_query()

 

     def on_destroy(self, *args):

-        app.ged.remove_event_handler('archiving-finished',

-                                       ged.PRECORE,

-                                       self._nec_archiving_finished)

+        app.ged.remove_event_handler('archiving-count-received',

+                                     ged.GUI1,

+                                     self._received_count)

+        app.ged.remove_event_handler('archiving-query-id',

+                                     ged.GUI1,

+                                     self._new_query_id)

+        app.ged.remove_event_handler('archiving-interval-finished',

+                                     ged.GUI1,

+                                     self._received_finished)

         app.ged.remove_event_handler('raw-mam-message-received',

-                                       ged.PRECORE,

-                                       self._nec_mam_message_received)

+                                     ged.PRECORE,

+                                     self._nec_mam_message_received)

         del app.interface.instances[self.account]['history_sync']

 

     def on_close_clicked(self, *args):

         self.destroy()

 

-    def _nec_mam_message_received(self, obj):

-        if obj.conn.name != self.account:

-            return

-

-        if obj.result.getAttr('queryid') != self.query_id:

-            return

-

-        log.debug('received message')

-        GLib.idle_add(self.download_history.set_fraction)

-

-    def _nec_archiving_finished(self, obj):

-        if obj.conn.name != self.account:

-            return

-

-        if obj.query_id not in (self.query_id, self.count_query_id):

-            return

-

-        set_ = obj.fin.getTag('set', namespace=nbxmpp.NS_RSM)

-        if not set_:

-            log.error('invalid result')

-            log.error(obj.fin)

-            return

-

-        if obj.query_id == self.count_query_id:

-            count = set_.getTagData('count')

-            log.info('message count received: %s', count)

-            if count:

-                self.download_history.count = int(count)

-            self.query_messages()

-            return

-

-        if obj.query_id == self.query_id:

-            last = set_.getTagData('last')

-            complete = obj.fin.getAttr('complete')

-            if not last and complete != 'true':

-                log.error('invalid result')

-                log.error(obj.fin)

-                return

-

-            if complete != 'true':

-                self.query_messages(last)

-            else:

-                log.info('query finished')

-                GLib.idle_add(self.download_history.finished)

-                if self.start:

-                    timestamp = self.start.timestamp()

-                else:

-                    timestamp = ArchiveState.ALL

-                app.config.set_per('accounts', self.account,

-                                     'mam_start_date', timestamp)

-                log.debug('config: set mam_start_date: %s', timestamp)

-                self.set_current_page(Pages.SUMMARY)

-                self.summary.finished()

-

 

 class SelectTimePage(Gtk.Box):

     def __init__(self, assistant):

         self.pack_start(label, True, True, 0)

         self.pack_start(listbox, False, False, 0)

 

+

 class DownloadHistoryPage(Gtk.Box):

     def __init__(self, assistant):

         super().__init__(orientation=Gtk.Orientation.VERTICAL)

     def finished(self):

         self.progress.set_fraction(1)

 

+

 class SummaryPage(Gtk.Box):

     def __init__(self, assistant):

         super().__init__(orientation=Gtk.Orientation.VERTICAL)

     def finished(self):

         received = self.assistant.download_history.received

         finished = _('''

-        Finshed synchronising your History. 

+        Finshed synchronising your History.

         {received} Messages downloaded.

         '''.format(received=received))

         self.label.set_text(finished)

 

     def nothing_to_do(self):

         nothing_to_do = _('''

-        Gajim is fully synchronised 

+        Gajim is fully synchronised

         with the Archive.

         ''')

         self.label.set_text(nothing_to_do)

 

     def query_already_running(self):

         already_running = _('''

-        There is already a synchronisation in 

+        There is already a synchronisation in

         progress. Please try later.

         ''')

         self.label.set_text(already_running)

 

+

 class TimeOption(Gtk.Label):

     def __init__(self, label, months=None):

         super().__init__(label=label)

                 jid = obj.jid

 

             obj.msg_log_id = app.logger.insert_into_logs(

-                jid, obj.timestamp, log_type,

+                self.conn.name, jid, obj.timestamp, log_type,

                 message=msg_to_log,

                 subject=obj.subject,

                 additional_data=obj.additional_data,

Philipp Hörist pushed to branch master at gajim / gajim

Commits:

14 changed files:

Changes: