path: root/gajim/common/storage/archive/storage.py

# This file is part of Gajim.
#
# Gajim is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
#
# Gajim is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.

from __future__ import annotations

from typing import Any

import calendar
import datetime
import logging
import sqlite3 as sqlite
import time
from collections import namedtuple
from collections.abc import Iterator
from functools import partial

from nbxmpp import JID

from gajim.common import app
from gajim.common import configpaths
from gajim.common.const import MAX_MESSAGE_CORRECTION_DELAY
from gajim.common.modules.contacts import GroupchatContact
from gajim.common.storage.archive.const import MessageState
from gajim.common.storage.archive.const import MessageType
from gajim.common.storage.archive.migration_v8 import MigrationV8
from gajim.common.storage.archive.statements import ARCHIVE_CREATE_STMT
from gajim.common.storage.archive.statements import FIND_CORRECTED_MSG_STMT
from gajim.common.storage.archive.statements import \
    FIND_CORRECTED_MSG_WITH_RES_STMT
from gajim.common.storage.archive.statements import GET_CONVERATION_JIDS_STMT
from gajim.common.storage.archive.statements import GET_CONVERSATION_STMT
from gajim.common.storage.archive.statements import GET_CORRECTIONS_STMT
from gajim.common.storage.archive.statements import \
    GET_CORRECTIONS_WITH_RES_STMT
from gajim.common.storage.archive.statements import \
    GET_DAYS_CONTAINING_MESSAGES_STMT
from gajim.common.storage.archive.statements import GET_FIRST_MESSAGE_TS_STMT
from gajim.common.storage.archive.statements import GET_LAST_MESSAGE_TS_STMT
from gajim.common.storage.archive.statements import GET_MAM_ARCHIVE_STATE_STMT
from gajim.common.storage.archive.statements import GET_MESSAGE_META_STMT
from gajim.common.storage.archive.statements import JID_INSERT_STMT
from gajim.common.storage.archive.statements import RESET_ARCHIVE_STATE_STMT
from gajim.common.storage.archive.structs import DbChatJoinedData
from gajim.common.storage.archive.structs import DbConversationJoinedData
from gajim.common.storage.archive.structs import DbConversationJoinedRowData
from gajim.common.storage.archive.structs import DbCorrectionRowData
from gajim.common.storage.archive.structs import DbFiletransferRowData
from gajim.common.storage.archive.structs import DbGroupchatJoinedData
from gajim.common.storage.archive.structs import DbInsertCorrectionRowData
from gajim.common.storage.archive.structs import DbInsertRowDataBase
from gajim.common.storage.archive.structs import DbMamArchiveStateRowData
from gajim.common.storage.archive.structs import DbPrivateMessageJoinedData
from gajim.common.storage.archive.structs import DbUpsertRowDataBase
from gajim.common.storage.base import SqliteStorage
from gajim.common.storage.base import timeit
from gajim.common.storage.base import VALUE_MISSING

from gajim.gtk.db_migration import DBMigration

CURRENT_USER_VERSION = 7

log = logging.getLogger('gajim.c.storage.archive')


def _messages_factory(cursor: sqlite.Cursor, row: tuple[Any, ...]) -> Any:
    fields = [col[0] for col in cursor.description]

    match row[6]:
        case MessageType.CHAT:
            return DbChatJoinedData.from_row(fields, row)
        case MessageType.GROUPCHAT:
            return DbGroupchatJoinedData.from_row(fields, row)
        case MessageType.PM:
            return DbPrivateMessageJoinedData.from_row(fields, row)
        case _:
            raise ValueError


def _row_factory(
    row_class: Any,
    cursor: sqlite.Cursor,
    row: tuple[Any, ...]
) -> Any:

    fields = [col[0] for col in cursor.description]
    kv = dict(zip(fields, row, strict=True))
    return row_class.from_query(kv)


def namedtuple_row(cursor: sqlite.Cursor, row: tuple[Any, ...]) -> Any:
    fields = [column[0] for column in cursor.description]
    cls = namedtuple('Row', fields)  # pyright: ignore
    return cls._make(row)


class MessageArchiveStorage(SqliteStorage):
    def __init__(self, in_memory: bool = False):
        path = None if in_memory else configpaths.get('LOG_DB')
        SqliteStorage.__init__(self,
                               log,
                               path,
                               ARCHIVE_CREATE_STMT % CURRENT_USER_VERSION)

        self._jid_eks: dict[JID | str, int] = {}
        self._accounts_eks: dict[str, int] = {}

    def init(self, **kwargs: Any) -> None:
        SqliteStorage.init(self,
                           detect_types=sqlite.PARSE_COLNAMES)

        self._set_journal_mode('WAL')
        self._enable_foreign_keys()
        self._enable_secure_delete()

        self._con.row_factory = namedtuple_row

        self._con.create_function('like', 1, self._like)

        self._load_jids()

    def _migrate(self) -> None:
        user_version = self.user_version
        if user_version == 0:
            # All migrations from 0.16.9 until 1.0.0
            statements = [
                'ALTER TABLE logs ADD COLUMN "account_id" INTEGER',
                'ALTER TABLE logs ADD COLUMN "stanza_id" TEXT',
                'ALTER TABLE logs ADD COLUMN "encryption" TEXT',
                'ALTER TABLE logs ADD COLUMN "encryption_state" TEXT',
                'ALTER TABLE logs ADD COLUMN "marker" INTEGER',
                'ALTER TABLE logs ADD COLUMN "additional_data" TEXT',
                '''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
                    )''',

                '''CREATE INDEX IF NOT EXISTS idx_logs_stanza_id
                   ON logs(stanza_id)''',
                'PRAGMA user_version=1'
            ]

            self._execute_multiple(statements)

        if user_version < 2:
            statements = [
                ('ALTER TABLE last_archive_message '
                 'ADD COLUMN "sync_threshold" INTEGER'),
                'PRAGMA user_version=2'
            ]
            self._execute_multiple(statements)

        if user_version < 3:
            statements = [
                'ALTER TABLE logs ADD COLUMN "message_id" TEXT',
                'PRAGMA user_version=3'
            ]
            self._execute_multiple(statements)

        if user_version < 4:
            statements = [
                'ALTER TABLE logs ADD COLUMN "error" TEXT',
                'PRAGMA user_version=4'
            ]
            self._execute_multiple(statements)

        if user_version < 5:
            statements = [
                '''CREATE INDEX IF NOT EXISTS idx_logs_message_id
                   ON logs (message_id)''',
                'PRAGMA user_version=5'
            ]
            self._execute_multiple(statements)

        if user_version < 7:
            statements = [
                'ALTER TABLE logs ADD COLUMN "real_jid" TEXT',
                'ALTER TABLE logs ADD COLUMN "occupant_id" TEXT',
                'PRAGMA user_version=7'
            ]
            self._execute_multiple(statements)

        if user_version < 8:
            if self._path is not None:
                DBMigration(MigrationV8, self._con, self._path)

    @staticmethod
    def _like(search_str: str) -> str:
        return f'%{search_str}%'

    def _get_cursor(
        self,
        factory: Any = None,
        row_class: Any = None
    ) -> sqlite.Cursor:

        cursor = self._con.cursor()

        if row_class is not None:
            factory = partial(factory, row_class)
        cursor.row_factory = factory
        return cursor

    @timeit
    def _load_jids(self) -> None:
        rows = self._con.execute('SELECT entitykey, jid FROM jid').fetchall()
        for row in rows:
            self._jid_eks[row.jid] = row.entitykey

    def _get_account_ek(self, account: str) -> int:
        account_ek = self._accounts_eks.get(account)
        if account_ek is not None:
            return account_ek

        jid = app.get_jid_from_account(account)

        select_stmt = 'SELECT entitykey FROM account WHERE jid = ?'
        result = self._con.execute(select_stmt, (jid,)).fetchone()
        if result is not None:
            self._accounts_eks[account] = result.entitykey
            return result.entitykey

        insert_stmt = 'INSERT INTO account(jid) VALUES (?)'
        entitykey = self._con.execute(insert_stmt, (jid,)).lastrowid
        assert entitykey is not None
        self._accounts_eks[account] = entitykey
        self._delayed_commit()

        return entitykey

    def _get_jid_ek(self, jid: JID | str) -> int:
        entitykey = self._jid_eks.get(jid)
        if entitykey is not None:
            return entitykey

        entitykey = self._con.execute(JID_INSERT_STMT, {'jid': jid}).lastrowid
        assert entitykey is not None
        self._jid_eks[jid] = entitykey
        return entitykey

    def _get_occupant_ek(self,
                         account_ek: int,
                         jid_ek: int,
                         occupant_id: str) -> int:
        # TODO cache calls?
        stmt = '''
            SELECT entitykey FROM occupant WHERE
                fk_accounts_ek = ? AND
                fk_jids_ek = ? AND
                id = ?
        '''

        row = self._con.execute(
            stmt, (account_ek, jid_ek, occupant_id)).fetchone()
        assert row is not None
        return row.entitykey

    def _get_foreign_keys(
        self,
        db_row_data: Any
    ) -> dict[str, int | None]:

        account = getattr(db_row_data, 'account', None)
        if account is None:
            account_ek = None
        else:
            account_ek = self._get_account_ek(account)

        remote_jid = getattr(db_row_data, 'remote_jid', None)
        if remote_jid is None:
            jid_ek = None
        else:
            jid_ek = self._get_jid_ek(remote_jid)

        real_jid_ek = None
        real_jid = getattr(db_row_data, 'real_jid', None)
        if real_jid not in (None, VALUE_MISSING):
            real_jid_ek = self._get_jid_ek(real_jid)

        return {
            'fk_account_ek': account_ek,
            'fk_jid_ek': jid_ek,
            'fk_real_jid_ek': real_jid_ek,
        }

    def get_conversation_jids(self, account: str) -> list[JID]:
        account_ek = self._get_account_ek(account)

        rows = self._con.execute(
            GET_CONVERATION_JIDS_STMT, (account_ek, )).fetchall()
        return [row.remote_jid for row in rows]

    @timeit
    def get_conversation_before_after(self,
                                      account: str,
                                      jid: JID,
                                      before: bool,
                                      timestamp: float,
                                      n_lines: int
                                      ) -> list[DbConversationJoinedData]:
        '''
        Load n_lines lines of conversation with jid before or after timestamp

        :param account:         The account

        :param jid:             The jid for which we request the conversation

        :param before:          bool for direction (before or after timestamp)

        :param timestamp:       timestamp

        :param nlines:          number of rows to get

        returns a list of namedtuples
        '''

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        if before:
            time_order = ('message.timestamp < ? ORDER BY '
                          'message.timestamp DESC, message.entitykey DESC')
        else:
            time_order = ('message.timestamp > ? ORDER BY '
                          'message.timestamp ASC, message.entitykey ASC')

        stmt = f'''
            {GET_CONVERSATION_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ? AND
            {time_order}
            LIMIT ?
        '''
        cursor = self._get_cursor(_messages_factory)
        return cursor.execute(
            stmt, (jid_ek, account_ek, timestamp, n_lines)).fetchall()

    @timeit
    def get_message_with_entitykey(
        self,
        entitykey: int
    ) -> DbConversationJoinedData:

        stmt = f'''
            {GET_CONVERSATION_STMT}
            message.entitykey = ?
        '''
        cursor = self._get_cursor(_messages_factory)
        return cursor.execute(stmt, (entitykey,)).fetchone()

    def get_row_with_entitykey(self, table: str, entitykey: int) -> Any:
        stmt = f'SELECT * FROM {table} WHERE entitykey=?'
        return self._con.execute(stmt, (entitykey,)).fetchone()

    @timeit
    def get_last_conversation_row(self,
                                  account: str,
                                  jid: JID
                                  ) -> DbConversationJoinedData | None:
        '''
        Load the last line of a conversation with jid for account.
        Loads messages, but no status messages or error messages.

        :param account:         The account

        :param jid:             The jid for which we request the conversation

        returns a namedtuple or None
        '''

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        stmt = f'''
            {GET_CONVERSATION_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ?
            ORDER BY message.timestamp DESC
        '''
        cursor = self._get_cursor(_messages_factory)
        return cursor.execute(stmt, (jid_ek, account_ek)).fetchone()

    @timeit
    def get_last_correctable_message(self,
                                     account: str,
                                     jid: JID,
                                     message_id: str
                                     ) -> DbConversationJoinedData | None:
        '''
        Load the last correctable message of a conversation by message_id.
        Conditions: max 5 min old
        '''
        min_time = time.time() - MAX_MESSAGE_CORRECTION_DELAY

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        # TODO there is no index for message_id

        stmt = f'''
            {GET_CONVERSATION_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ? AND
            message.message_id = ? AND
            message.timestamp > ?
            ORDER BY message.timestamp DESC
        '''
        cursor = self._get_cursor(_messages_factory)
        return cursor.execute(
            stmt, (jid_ek, account_ek, message_id, min_time)).fetchone()

    def get_corrected_message_entitykey(
        self,
        m_type: int,
        corrected_data: DbInsertCorrectionRowData
    ) -> int | None:

        fk_account_ek = self._get_account_ek(corrected_data.account)
        fk_jid_ek = self._get_jid_ek(corrected_data.remote_jid)

        params = corrected_data.asdict()
        params['fk_jid_ek'] = fk_jid_ek
        params['fk_account_ek'] = fk_account_ek

        stmt = FIND_CORRECTED_MSG_STMT
        if (m_type == MessageType.GROUPCHAT and
                corrected_data.fk_occupant_ek is None):
            stmt = FIND_CORRECTED_MSG_WITH_RES_STMT

        cursor = self._get_cursor()
        result = cursor.execute(stmt, params).fetchone()
        if result is None:
            return None
        return result[0]

    def get_corrections(
        self,
        joined_data: DbConversationJoinedRowData
    ) -> list[DbCorrectionRowData]:

        occupant_ek = None
        if joined_data.occupants is not None:
            occupant_ek = joined_data.occupants.entitykey

        resource = None
        if (joined_data.m_type == MessageType.GROUPCHAT and
                joined_data.occupants is None):
            resource = joined_data.resource

        params = {
            'fk_jid_ek': joined_data.remote_ek,
            'fk_account_ek': joined_data.account_ek,
            'message_id': joined_data.message_id,
            'resource': resource,
            'fk_occupant_ek': occupant_ek,
            'direction': joined_data.direction
        }

        stmt = GET_CORRECTIONS_STMT
        if resource is not None:
            stmt = GET_CORRECTIONS_WITH_RES_STMT

        cursor = self._get_cursor(_row_factory, row_class=DbCorrectionRowData)
        return cursor.execute(stmt, params).fetchall()


    def get_filetransfers(self, message_ek: int) -> list[DbFiletransferRowData]:
        cursor = self._get_cursor(_row_factory, row_class=DbFiletransferRowData)
        stmt = DbFiletransferRowData.get_select_stmt()
        return cursor.execute(stmt, (message_ek,)).fetchall()


    @timeit
    def search_archive(self,
                       account: str | None,
                       jid: str | None,
                       query: str,
                       from_users: list[str] | None = None,
                       before: datetime.datetime | None = None,
                       after: datetime.datetime | None = None
                       ) -> Iterator[DbConversationJoinedData]:
        '''
        Search the conversation log for messages containing the `query` string.

        The search can either span the complete log for the given
        `account` and `jid` or be restricted to a single day by
        specifying `date`.

        :param account: The account

        :param jid: The jid for which we request the conversation

        :param query: A search string

        :param from_users: A list of usernames or None

        :param before: A datetime.datetime instance or None

        :param after: A datetime.datetime instance or None

        returns a list of namedtuples
        '''
        if before is None:
            before_ts = datetime.datetime.now().timestamp()
        else:
            before_ts = before.timestamp()

        after_ts = 0
        if after is not None:
            after_ts = after.timestamp()

        contact_stmt = ''
        if account is not None and jid is not None:
            contact_stmt = '''
                message.fk_jid_ek = ? AND
                message.fk_account_ek = ? AND'''

        if from_users is None:
            users_query_stmt = ''
        else:
            users_query_stmt = 'UPPER(message.resource) IN (?) AND'

        stmt = f'''
            {GET_CONVERSATION_STMT}
            {contact_stmt}
            {users_query_stmt}
            IFNULL(correction.corrected_message, message.message)
            LIKE like(?) AND
            message.timestamp BETWEEN ? AND ?
            ORDER BY message.timestamp DESC, message.entitykey
            '''

        cursor = self._get_cursor(_messages_factory)
        if from_users is None:
            if contact_stmt:
                assert account is not None
                assert jid is not None
                account_ek = self._get_account_ek(account)
                jid_ek = self._get_jid_ek(jid)
                cursor.execute(
                    stmt, (jid_ek, account_ek, query, after_ts, before_ts))
            else:
                cursor.execute(
                    stmt, (query, after_ts, before_ts))
            while True:
                results = cursor.fetchmany(25)
                if not results:
                    break
                for result in results:
                    yield result
            return

        users = ','.join([user.upper() for user in from_users])
        if contact_stmt:
            assert account is not None
            assert jid is not None
            account_ek = self._get_account_ek(account)
            jid_ek = self._get_jid_ek(jid)
            cursor.execute(
                stmt, (jid_ek, account_ek, users, query, after_ts, before_ts))
        else:
            cursor.execute(
                stmt, (users, query, after_ts, before_ts))
        while True:
            results = cursor.fetchmany(25)
            if not results:
                break
            for result in results:
                yield result

    @timeit
    def get_days_containing_messages(self,
                                     account: str,
                                     jid: str,
                                     year: int,
                                     month: int
                                     ) -> list[int]:
        '''
        Get days in month of year where messages for account/jid exist
        '''
        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        # Calculate the start and end datetime of the month
        date = datetime.datetime(year, month, 1)
        days = calendar.monthrange(year, month)[1] - 1
        delta = datetime.timedelta(
            days=days, hours=23, minutes=59, seconds=59, microseconds=999999)
        start_ts = date.timestamp()
        end_ts = (date + delta).timestamp()

        stmt = f'''
            {GET_DAYS_CONTAINING_MESSAGES_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ? AND
            message.timestamp BETWEEN ? AND ?
            ORDER BY message.timestamp
            '''
        return [row.day for row in self._con.execute(
            stmt, (jid_ek, account_ek, start_ts, end_ts)).fetchall()]

    @timeit
    def get_last_history_ts(self,
                            account: str,
                            jid: str
                            ) -> float | None:
        '''
        Get the timestamp of the last message we received for the jid
        '''
        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)
        stmt = f'''
            {GET_LAST_MESSAGE_TS_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ?
            '''
        # fetchone() returns always at least one Row with all
        # attributes set to None because of the MAX() function
        return self._con.execute(
            stmt, (jid_ek, account_ek)).fetchone().timestamp

    @timeit
    def get_first_history_ts(self,
                             account: str,
                             jid: str
                             ) -> float | None:
        '''
        Get the timestamp of the first message we received for the jid
        '''
        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)
        stmt = f'''
            {GET_FIRST_MESSAGE_TS_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ?
            '''
        # fetchone() returns always at least one Row with all
        # attributes set to None because of the MIN() function
        return self._con.execute(
            stmt, (jid_ek, account_ek)).fetchone().timestamp

    @timeit
    def get_first_message_meta_for_date(self,
                                        account: str,
                                        jid: str,
                                        date: datetime.datetime
                                        ) -> tuple[int, float] | None:
        '''
        Load meta data (entitykey, timestamp) for the first message of
        a specific date
        '''
        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)
        delta = datetime.timedelta(
            hours=23, minutes=59, seconds=59, microseconds=999999)
        date_ts = date.timestamp()
        delta_ts = (date + delta).timestamp()
        stmt = f'''
            {GET_MESSAGE_META_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ? AND
            message.timestamp BETWEEN ? AND ?
            ORDER BY message.timestamp ASC, message.entitykey ASC
        '''
        return self._con.execute(
            stmt, (jid_ek, account_ek, date_ts, delta_ts)).fetchone()

    @timeit
    def date_has_history(self,
                         account: str,
                         jid: str,
                         date: datetime.datetime
                         ) -> float | None:
        '''
        Get a single meta row of a message for 'jid'
        in time range of one day
        '''
        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)
        delta = datetime.timedelta(
            hours=23, minutes=59, seconds=59, microseconds=999999)
        start_ts = date.timestamp()
        end_ts = (date + delta).timestamp()
        stmt = f'''
            {GET_MESSAGE_META_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ? AND
            message.timestamp BETWEEN ? AND ?
            '''
        return self._con.execute(
            stmt, (jid_ek, account_ek, start_ts, end_ts)).fetchone()

    def _insert_row(
        self,
        row_data: DbInsertRowDataBase,
        with_entitykey: int | None = None,
        raise_on_conflict: bool = True,
        ignore_on_conflict: bool = False,
    ) -> int:

        foreign_keys = self._get_foreign_keys(row_data)

        values = row_data.asdict()
        values.update(foreign_keys)
        if with_entitykey is not None:
            values['entitykey'] = with_entitykey

        log.info('Insert into %s: %s', row_data.get_table_name(), values)
        try:
            cursor = self._con.execute(row_data.get_insert_stmt(), values)
        except sqlite.IntegrityError as error:
            if ignore_on_conflict:
                self._log.debug('Ignore failed insert because of contraint')
                return -1

            if raise_on_conflict:
                raise error

            row = self._con.execute(
                row_data.get_select_stmt(), values).fetchone()
            return row.entitykey

        entitykey = cursor.lastrowid
        assert entitykey is not None
        return entitykey

    def insert_row(
        self,
        row_data: DbInsertRowDataBase,
        dependent_row_data: list[DbInsertRowDataBase] | None = None,
        raise_on_conflict: bool = True,
        ignore_on_conflict: bool = False,
    ) -> int:

        if dependent_row_data is None:
            dependent_row_data = []

        entitykey = self._insert_row(
            row_data,
            raise_on_conflict=raise_on_conflict,
            ignore_on_conflict=ignore_on_conflict)
        if entitykey == -1:
            return -1

        for row_data in dependent_row_data:
            self._insert_row(row_data, with_entitykey=entitykey)

        self._delayed_commit()
        return entitykey

    def upsert_row(self, row_data: DbUpsertRowDataBase) -> int:
        foreign_keys = self._get_foreign_keys(row_data)

        values = row_data.asdict()
        values.update(foreign_keys)

        log.info('Upsert into %s: %s', row_data.get_table_name(), values)
        row = self._con.execute(row_data.get_upsert_stmt(), values).fetchone()
        if row is None:
            row = self._con.execute(
                row_data.get_select_stmt(), values).fetchone()

        self._delayed_commit()
        return row.entitykey

    @timeit
    def get_recent_muc_nicks(self, contact: GroupchatContact) -> set[str]:
        account_ek = self._get_account_ek(contact.account)
        jid_ek = self._get_jid_ek(contact.jid)

        sql = '''
            SELECT resource
            FROM message WHERE
            fk_jid_ek = ? AND
            fk_account_ek = ?
            ORDER BY timestamp DESC'''

        results = self._con.execute(sql, (jid_ek, account_ek)).fetchmany(50)

        nicknames: set[str] = set()
        for row in results:
            if row.resource is None:
                continue
            nicknames.add(row.resource)

        return nicknames

    def get_mam_archive_state(
        self,
        account: str,
        jid: JID
    ) -> DbMamArchiveStateRowData | None:

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        cursor = self._get_cursor(_row_factory, DbMamArchiveStateRowData)
        return cursor.execute(
            GET_MAM_ARCHIVE_STATE_STMT, (jid_ek, account_ek)).fetchone()

    def reset_mam_archive_state(self, account: str, jid: str) -> None:

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        self._con.execute(RESET_ARCHIVE_STATE_STMT, (jid_ek, account_ek))
        self._delayed_commit()

    def _delete_messages(self, find_stmt: str, params: Any) -> None:
        delete_stmts = [
            f'DELETE FROM message WHERE entitykey IN ({find_stmt})',
        ]

        for stmt in delete_stmts:
            self._con.execute(stmt, params)

    def delete_pending_message(
        self,
        account: str,
        remote_jid: JID,
        message_id: str
    ) -> int | None:

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(remote_jid)

        delete_stmt = '''
            DELETE FROM message
            WHERE
                state = ? AND
                message_id = ? AND
                fk_jid_ek = ? AND
                fk_account_ek = ?
            RETURNING entitykey
        '''

        results = self._con.execute(delete_stmt, (
            MessageState.PENDING,
            message_id,
            jid_ek,
            account_ek)).fetchall()

        if results:
            assert len(results) == 1
            self._delayed_commit()
            return results[0].entitykey

    def delete_message(self, entitykey: int) -> None:

        # TODO Delete Corrections
        delete_stmts = [
            'DELETE FROM message WHERE entitykey = ?',
        ]

        for stmt in delete_stmts:
            self._con.execute(stmt, (entitykey,))
        self._commit()

    def remove_history(self, account: str, jid: JID) -> None:
        '''
        Remove history for a specific chat.
        '''

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(jid)

        statements = [
            'DELETE FROM message WHERE fk_jid_ek = ? and fk_account_ek = ?',
            'DELETE FROM error WHERE fk_jid_ek = ? and fk_account_ek = ?',
            'DELETE FROM reaction WHERE fk_jid_ek = ? and fk_account_ek = ?',
            'DELETE FROM retraction WHERE fk_jid_ek = ? and fk_account_ek = ?',
            'DELETE FROM moderation WHERE fk_jid_ek = ? and fk_account_ek = ?',
            'DELETE FROM correction WHERE fk_jid_ek = ? and fk_account_ek = ?',
            'DELETE FROM marker WHERE fk_jid_ek = ? and fk_account_ek = ?',
        ]

        for stmt in statements:
            self._con.execute(stmt, (jid_ek, account_ek))

        self._commit()

        log.info('Removed history for: %s', jid)

    def remove_all_history(self) -> None:
        '''
        Remove all messages for all accounts
        '''
        statements = [
            'DELETE FROM error',
            'DELETE FROM reaction',
            'DELETE FROM retraction',
            'DELETE FROM moderation',
            'DELETE FROM correction',
            'DELETE FROM marker',
            'DELETE FROM message',
        ]
        self._execute_multiple(statements)
        log.info('Removed all chat history')

    def remove_all_from_account(self, account: str) -> None:
        account_ek = self._get_account_ek(account)
        self._con.execute(
            'DELETE FROM account WHERE entitykey = ?', (account_ek,))
        self._commit()

    def cleanup_chat_history(self) -> None:
        '''
        Remove messages from account where messages are older than max_age
        '''
        for account in app.settings.get_accounts():
            max_age = app.settings.get_account_setting(
                account, 'chat_history_max_age')
            if max_age == -1:
                continue

            account_ek = self._get_account_ek(account)
            now = time.time()
            point_in_time = now - int(max_age)

            find_stmt = '''
                SELECT entitykey FROM message
                WHERE fk_account_ek = ? and timestamp < ?
            '''

            results = self._con.execute(
                find_stmt, (account_ek, point_in_time)).fetchall()
            if not results:
                continue

            self._delete_messages(find_stmt, (account_ek, point_in_time))
            self._commit()
            log.info('Removed %s old messages for %s', len(results), account)

    def get_messages_for_export(self,
                                account: str,
                                remote_jid: JID
                                ) -> Iterator[DbConversationJoinedData]:

        account_ek = self._get_account_ek(account)
        jid_ek = self._get_jid_ek(remote_jid)

        stmt = f'''
            {GET_CONVERSATION_STMT}
            message.fk_jid_ek = ? AND
            message.fk_account_ek = ?
            ORDER BY message.timestamp ASC
        '''

        cursor = self._con.execute(stmt, (jid_ek, account_ek))
        while True:
            results = cursor.fetchmany(10)
            if not results:
                break
            yield from results