summaryrefslogtreecommitdiff
path: root/poezio/log_loader.py
blob: aa645c30871cee264c16efefa1d07a1848b04b87 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
"""
This modules contains a class that loads messages into a ChatTab, either from
MAM or the local logs, and a class that loads MUC history into the local
logs.


How the log loading works will depend on the poezio configuration:

- if use_log is True, no logs will be fetched dynamically
- if use_log is False, all logs will be fetched from MAM (if available)
- if mam_sync and use_log are True, most chat tabs (all of them except the
  static conversation tab) will try to sync the local
  logs with the MAM history when opening them, or when joining a room.
- all log loading/writing workflows are paused until the MAM sync is complete
  (so that the local log loading can be up-to-date with the MAM history)
- when use_log is False, mam_sync has no effect
"""
from __future__ import annotations
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from typing import List
from poezio import tabs
from poezio.logger import (
    build_log_message,
    iterate_messages_reverse,
    last_message_in_archive,
    Logger,
    LogDict,
)
from poezio.mam import (
    fetch_history,
    NoMAMSupportException,
    MAMQueryException,
    DiscoInfoException,
    make_line,
)
from poezio.common import to_utc
from poezio.ui.types import EndOfArchive, Message, BaseMessage
from poezio.text_buffer import HistoryGap
from slixmpp import JID


# Max number of messages to insert when filling a gap
HARD_LIMIT = 999


log = logging.getLogger(__name__)


def make_line_local(tab: tabs.ChatTab, msg: LogDict) -> Message:
    """Create a UI message from a local log read.

    :param tab: Tab in which that message will be displayed
    :param msg: Log data
    :returns: The UI message
    """
    if isinstance(tab, tabs.MucTab):
        jid = JID(tab.jid)
        jid.resource = msg.get('nickname') or ''
    else:
        jid = JID(tab.jid)
    msg['time'] = msg['time'].astimezone(tz=timezone.utc)
    return make_line(tab, msg['txt'], msg['time'], jid, '', msg['nickname'])


class LogLoader:
    """
    An ephemeral class that loads history in a tab.

    Loading from local logs is blocked until history has been fetched from
    MAM to fill the local archive.
    """
    logger: Logger
    tab: tabs.ChatTab
    mam_only: bool

    def __init__(self, logger: Logger, tab: tabs.ChatTab,
                 local_logs: bool = True):
        self.mam_only = not local_logs
        self.logger = logger
        self.tab = tab

    async def tab_open(self) -> None:
        """Called on a tab opening or a MUC join"""
        amount = 2 * self.tab.text_win.height
        gap = self.tab._text_buffer.find_last_gap_muc()
        messages = []
        if gap is not None:
            if self.mam_only:
                messages = await self.mam_fill_gap(gap)
            else:
                messages = await self.local_fill_gap(gap)
        else:
            if self.mam_only:
                messages = await self.mam_tab_open(amount)
            else:
                messages = await self.local_tab_open(amount)

        log.debug(
            'Fetched %s messages for %s',
            len(messages), self.tab.jid
        )
        if messages:
            self.tab._text_buffer.add_history_messages(messages)
            self.tab.core.refresh_window()

    async def mam_tab_open(self, nb: int) -> List[BaseMessage]:
        """Fetch messages in MAM when opening a new tab.

        :param nb: number of max messages to fetch.
        :returns: list of ui messages to add
        """
        tab = self.tab
        end = datetime.now()
        for message in tab._text_buffer.messages:
            time_ok = to_utc(message.time) < to_utc(end)
            if isinstance(message, Message) and time_ok:
                end = message.time
                break
        end = end - timedelta(microseconds=1)
        try:
            return await fetch_history(tab, end=end, amount=nb)
        except (NoMAMSupportException, MAMQueryException, DiscoInfoException):
            return []
        finally:
            tab.query_status = False

    async def local_tab_open(self, nb: int) -> List[BaseMessage]:
        """Fetch messages locally when opening a new tab.

        :param nb: number of max messages to fetch.
        :returns: list of ui messages to add
        """
        await self.wait_mam()
        results: List[BaseMessage] = []
        filepath = self.logger.get_file_path(self.tab.jid)
        count = 0
        for msg in iterate_messages_reverse(filepath):
            typ_ = msg.pop('type')
            if typ_ == 'message':
                results.append(make_line_local(self.tab, msg))
            if len(results) >= nb:
                break
            count += 1
            if count % 20 == 0:
                await asyncio.sleep(0)
        return results[::-1]

    async def mam_fill_gap(self, gap: HistoryGap) -> List[BaseMessage]:
        """Fill a message gap in an existing tab using MAM.

        :param gap: Object describing the history gap
        :returns: list of ui messages to add
        """
        tab = self.tab

        start = gap.last_timestamp_before_leave
        end = gap.first_timestamp_after_join
        if start:
            start = start + timedelta(seconds=1)
        if end:
            end = end - timedelta(seconds=1)
        try:
            return await fetch_history(
                tab,
                start=start,
                end=end,
                amount=HARD_LIMIT
            )
        except (NoMAMSupportException, MAMQueryException, DiscoInfoException):
            return []
        finally:
            tab.query_status = False

    async def local_fill_gap(self, gap: HistoryGap) -> List[BaseMessage]:
        """Fill a message gap in an existing tab using the local logs.
        Mostly useless when not used with the MAMFiller.

        :param gap: Object describing the history gap
        :returns: list of ui messages to add
        """
        await self.wait_mam()
        start = gap.last_timestamp_before_leave
        end = gap.first_timestamp_after_join
        count = 0

        results: List[BaseMessage] = []
        filepath = self.logger.get_file_path(self.tab.jid)
        for msg in iterate_messages_reverse(filepath):
            typ_ = msg.pop('type')
            if start and msg['time'] < start:
                break
            if typ_ == 'message' and (not end or msg['time'] < end):
                results.append(make_line_local(self.tab, msg))
            if len(results) >= HARD_LIMIT:
                break
            count += 1
            if count % 20 == 0:
                await asyncio.sleep(0)
        return results[::-1]

    async def scroll_requested(self):
        """When a scroll up is requested in a chat tab.

        Try to load more history if there are no more messages in the buffer.
        """
        tab = self.tab
        tw = tab.text_win

        # If position in the tab is < two screen pages, then fetch MAM, so that
        # wa keep some prefetched margin. A first page should also be
        # prefetched on join if not already available.
        total, pos, height = len(tw.built_lines), tw.pos, tw.height
        rest = (total - pos) // height

        if rest > 1:
            return None

        if self.mam_only:
            messages = await self.mam_scroll_requested(height)
        else:
            messages = await self.local_scroll_requested(height)
        if messages:
            tab._text_buffer.add_history_messages(messages)
            tab.core.refresh_window()

    async def local_scroll_requested(self, nb: int) -> List[BaseMessage]:
        """Fetch messages locally on scroll up.

        :param nb: Number of messages to fetch
        :returns: list of ui messages to add
        """
        await self.wait_mam()
        tab = self.tab
        count = 0
        last_message_time = None
        if tab._text_buffer.messages:
            last_message_time = to_utc(tab._text_buffer.messages[0].time)
            last_message_time -= timedelta(microseconds=1)

        results: List[BaseMessage] = []
        filepath = self.logger.get_file_path(self.tab.jid)
        for msg in iterate_messages_reverse(filepath):
            typ_ = msg.pop('type')
            if last_message_time is None or msg['time'] < last_message_time:
                if typ_ == 'message':
                    results.append(make_line_local(self.tab, msg))
            if len(results) >= nb:
                break
            count += 1
            if count % 20 == 0:
                await asyncio.sleep(0)
        return results[::-1]

    async def mam_scroll_requested(self, nb: int) -> List[BaseMessage]:
        """Fetch messages from MAM on scroll up.

        :param nb: Number of messages to fetch
        :returns: list of ui messages to add
        """
        tab = self.tab
        try:
            messages = await fetch_history(tab, amount=nb)
            last_message_exists = False
            if tab._text_buffer.messages:
                last_message = tab._text_buffer.messages[0]
                last_message_exists = True
            if (not messages and
                    last_message_exists
                    and not isinstance(last_message, EndOfArchive)):
                time = tab._text_buffer.messages[0].time
                messages = [EndOfArchive('End of archive reached', time=time)]
            return messages
        except NoMAMSupportException:
            return []
        except (MAMQueryException, DiscoInfoException):
            tab.core.information(
                f'An error occured when fetching MAM for {tab.jid}',
                'Error'
            )
            return []
        finally:
            tab.query_status = False

    async def wait_mam(self) -> None:
        """Wait for the MAM history sync before reading the local logs.

        Does nothing apart from blocking.
        """
        if self.tab.mam_filler is None:
            return
        await self.tab.mam_filler.done.wait()


class MAMFiller:
    """Class that loads messages from MAM history into the local logs.
    """
    tab: tabs.ChatTab
    logger: Logger
    future: asyncio.Future
    done: asyncio.Event
    limit: int

    def __init__(self, logger: Logger, tab: tabs.ChatTab, limit: int = 2000):
        self.tab = tab
        self.logger = logger
        logger.fd_busy(tab.jid)
        self.future = asyncio.ensure_future(self.fetch_routine())
        self.done = asyncio.Event()
        self.limit = limit
        self.result = 0

    def cancel(self) -> None:
        """Cancel the routine and signal the end."""
        self.future.cancel()
        self.end()

    async def fetch_routine(self) -> None:
        """Load logs into the local archive, if possible."""
        filepath = self.logger.get_file_path(self.tab.jid)
        log.debug('Fetching logs for %s', self.tab.jid)
        try:
            last_msg = last_message_in_archive(filepath)
            last_msg_time = None
            if last_msg:
                last_msg_time = last_msg['time'] + timedelta(seconds=1)
            try:
                messages = await fetch_history(
                    self.tab,
                    start=last_msg_time,
                    amount=self.limit,
                )
                log.debug(
                    'Fetched %s messages to fill local logs for %s',
                    len(messages), self.tab.jid,
                )
                self.result = len(messages)
            except NoMAMSupportException:
                log.debug('The entity %s does not support MAM', self.tab.jid)
                return
            except (DiscoInfoException, MAMQueryException):
                log.debug(
                    'Failed fetching logs for %s',
                    self.tab.jid, exc_info=True
                )
                return

            def build_message(msg) -> str:
                return build_log_message(
                    msg.nickname,
                    msg.txt,
                    msg.time,
                    prefix='MR',
                )

            logs = ''.join(map(build_message, messages))
            self.logger.log_raw(self.tab.jid, logs, force=True)
        finally:
            self.end()

    def end(self) -> None:
        """End a MAM fill (error or sucess). Remove references and signal on
        the Event().
        """
        try:
            self.logger.fd_available(self.tab.jid)
        except Exception:
            log.error('Error when restoring log fd:', exc_info=True)
        self.tab.mam_filler = None
        self.done.set()