<commit_before># Documented in https://zulip.readthedocs.io/en/latest/subsystems/queuing.html
from typing import Any, Callable, Dict, List, Mapping, Optional, cast, TypeVar, Type

import copy
import signal
import tempfile
from functools import wraps
from threading import Timer

import smtplib
import socket

from django.conf import settings
from django.db import connection
from django.core.handlers.wsgi import WSGIRequest
from django.core.handlers.base import BaseHandler
from zerver.models import \
    get_client, get_system_bot, PreregistrationUser, \
    get_user_profile_by_id, Message, Realm, UserMessage, UserProfile, \
    Client
from zerver.lib.context_managers import lockfile
from zerver.lib.error_notify import do_report_error
from zerver.lib.feedback import handle_feedback
from zerver.lib.queue import SimpleQueueClient, queue_json_publish, retry_event
from zerver.lib.timestamp import timestamp_to_datetime
from zerver.lib.email_notifications import handle_missedmessage_emails
from zerver.lib.push_notifications import handle_push_notification, handle_remove_push_notification, \
    initialize_push_notifications
from zerver.lib.actions import do_send_confirmation_email, \
    do_update_user_activity, do_update_user_activity_interval, do_update_user_presence, \
    internal_send_message, internal_send_private_message, notify_export_completed, \
    render_incoming_message, do_update_embedded_data, do_mark_stream_messages_as_read
from zerver.lib.url_preview import preview as url_preview
from zerver.lib.digest import handle_digest_email
from zerver.lib.send_email import send_future_email, send_email_from_dict, \
    FromAddress, EmailNotDeliveredException, handle_send_email_format_changes
from zerver.lib.email_mirror import process_message as mirror_email, rate_limit_mirror_by_realm, \
    is_missed_message_address, extract_and_validate
from zerver.lib.streams import access_stream_by_id
from zerver.tornado.socket import req_redis_key, respond_send_message
from confirmation.models import Confirmation, create_confirmation_link
from zerver.lib.db import reset_queries
from zerver.lib.redis_utils import get_redis_client
from zerver.context_processors import common_context
from zerver.lib.outgoing_webhook import do_rest_call, get_outgoing_webhook_service_handler
from zerver.models import get_bot_services, RealmAuditLog
from zulip_bots.lib import extract_query_without_mention
from zerver.lib.bot_lib import EmbeddedBotHandler, get_bot_handler, EmbeddedBotQuitException
from zerver.lib.exceptions import RateLimited
from zerver.lib.export import export_realm_wrapper

import os
import sys
import ujson
from collections import defaultdict
import email
import time
import datetime
import logging
import requests
from io import StringIO
import urllib

logger = logging.getLogger(__name__)

class WorkerDeclarationException(Exception):
    pass

ConcreteQueueWorker = TypeVar('ConcreteQueueWorker', bound='QueueProcessingWorker')

def assign_queue(
        queue_name: str, enabled: bool=True, queue_type: str="consumer"
) -> Callable[[Type[ConcreteQueueWorker]], Type[ConcreteQueueWorker]]:
    def decorate(clazz: Type[ConcreteQueueWorker]) -> Type[ConcreteQueueWorker]:
        clazz.queue_name = queue_name
        if enabled:
            register_worker(queue_name, clazz, queue_type)
        return clazz
    return decorate

worker_classes = {}  # type: Dict[str, Type[QueueProcessingWorker]]
queues = {}  # type: Dict[str, Dict[str, Type[QueueProcessingWorker]]]
def register_worker(queue_name: str, clazz: Type['QueueProcessingWorker'], queue_type: str) -> None:
    if queue_type not in queues:
        queues[queue_type] = {}
    queues[queue_type][queue_name] = clazz
    worker_classes[queue_name] = clazz

def get_worker(queue_name: str) -> 'QueueProcessingWorker':
    return worker_classes[queue_name]()

def get_active_worker_queues(queue_type: Optional[str]=None) -> List[str]:
    """Returns all the non-test worker queues."""
    if queue_type is None:
        return list(worker_classes.keys())
    return list(queues[queue_type].keys())

def check_and_send_restart_signal() -> None:
    try:
        if not connection.is_usable():
            logging.warning("*** Sending self SIGUSR1 to trigger a restart.")
            os.kill(os.getpid(), signal.SIGUSR1)
    except Exception:
        pass

def retry_send_email_failures(
        func: Callable[[ConcreteQueueWorker, Dict[str, Any]], None]
) -> Callable[['QueueProcessingWorker', Dict[str, Any]], None]:

    @wraps(func)
    def wrapper(worker: ConcreteQueueWorker, data: Dict[str, Any]) -> None:
        try:
            func(worker, data)
        except (smtplib.SMTPServerDisconnected, socket.gaierror, EmailNotDeliveredException):
            def on_failure(event: Dict[str, Any]) -> None:
                logging.exception("Event {} failed".format(event))

            retry_event(worker.queue_name, data, on_failure)

    return wrapper

class QueueProcessingWorker:
    queue_name = None  # type: str

    def __init__(self) -> None:
        self.q = None  # type: SimpleQueueClient
        if self.queue_name is None:
            raise WorkerDeclarationException("Queue worker declared without queue_name")

    def consume(self, data: Dict[str, Any]) -> None:
        raise WorkerDeclarationException("No consumer defined!")

    def consume_wrapper(self, data: Dict[str, Any]) -> None:
        try:
            self.consume(data)
        except Exception:
            self._log_problem()
            if not os.path.exists(settings.QUEUE_ERROR_DIR):
                os.mkdir(settings.QUEUE_ERROR_DIR)  # nocoverage
            fname = '%s.errors' % (self.queue_name,)
            fn = os.path.join(settings.QUEUE_ERROR_DIR, fname)
            line = '%s\t%s\n' % (time.asctime(), ujson.dumps(data))
            lock_fn = fn + '.lock'
            with lockfile(lock_fn):
                with open(fn, 'ab') as f:
                    f.write(line.encode('utf-8'))
            check_and_send_restart_signal()
        finally:
            reset_queries()

    def _log_problem(self) -> None:
        logging.exception("Problem handling data on queue %s" % (self.queue_name,))

    def setup(self) -> None:
        self.q = SimpleQueueClient()

    def start(self) -> None:
        self.q.register_json_consumer(self.queue_name, self.consume_wrapper)
        self.q.start_consuming()

    def stop(self) -> None:  # nocoverage
        self.q.stop_consuming()

class LoopQueueProcessingWorker(QueueProcessingWorker):
    sleep_delay = 0

    def start(self) -> None:  # nocoverage
        while True:
            # TODO: Probably it'd be better to share code with consume_wrapper()
            events = self.q.drain_queue(self.queue_name, json=True)
            try:
                self.consume_batch(events)
            finally:
                reset_queries()
            time.sleep(self.sleep_delay)

    def consume_batch(self, event: List[Dict[str, Any]]) -> None:
        raise NotImplementedError

    def consume(self, event: Dict[str, Any]) -> None:
        """In LoopQueueProcessingWorker, consume is used just for automated tests"""
        self.consume_batch([event])

@assign_queue('signups')
class SignupWorker(QueueProcessingWorker):
    def consume(self, data: Dict[str, Any]) -> None:
        # TODO: This is the only implementation with Dict cf Mapping; should we simplify?
        user_profile = get_user_profile_by_id(data['user_id'])
        logging.info("Processing signup for user %s in realm %s" % (
            user_profile.email, user_profile.realm.string_id))
        if settings.MAILCHIMP_API_KEY and settings.PRODUCTION:
            endpoint = "https://%s.api.mailchimp.com/3.0/lists/%s/members" % \
                       (settings.MAILCHIMP_API_KEY.split('-')[1], settings.ZULIP_FRIENDS_LIST_ID)
            params = dict(data)
            del params['user_id']
            params['list_id'] = settings.ZULIP_FRIENDS_LIST_ID
            params['status'] = 'subscribed'
            r = requests.post(endpoint, auth=('apikey', settings.MAILCHIMP_API_KEY), json=params, timeout=10)
            if r.status_code == 400 and ujson.loads(r.text)['title'] == 'Member Exists':
                logging.warning("Attempted to sign up already existing email to list: %s" %
                                (data['email_address'],))
            elif r.status_code == 400:
                retry_event('signups', data, lambda e: r.raise_for_status())
            else:
                r.raise_for_status()

@assign_queue('invites')
class ConfirmationEmailWorker(QueueProcessingWorker):
    def consume(self, data: Mapping[str, Any]) -> None:
        if "email" in data:
            # When upgrading from a version up through 1.7.1, there may be
            # existing items in the queue with `email` instead of `prereg_id`.
            invitee = PreregistrationUser.objects.filter(
                email__iexact=data["email"].strip()).latest("invited_at")
        else:
            invitee = PreregistrationUser.objects.filter(id=data["prereg_id"]).first()
            if invitee is None:
                # The invitation could have been revoked
                return

        referrer = get_user_profile_by_id(data["referrer_id"])
        logger.info("Sending invitation for realm %s to %s" % (referrer.realm.string_id, invitee.email))
        do_send_confirmation_email(invitee, referrer)

        # queue invitation reminder for two days from now.
        link = create_confirmation_link(invitee, referrer.realm.host, Confirmation.INVITATION)
        context = common_context(referrer)
        context.update({
            'activate_url': link,
            'referrer_name': referrer.full_name,
            'referrer_email': referrer.email,
            'referrer_realm_name': referrer.realm.name,
        })
        send_future_email(
            "zerver/emails/invitation_reminder",
            referrer.realm,
            to_emails=[invitee.email],
            from_address=FromAddress.tokenized_no_reply_address(),
            language=referrer.realm.default_language,
            context=context,
            delay=datetime.timedelta(days=2))

@assign_queue('user_activity')
class UserActivityWorker(QueueProcessingWorker):
    def consume(self, event: Mapping[str, Any]) -> None:
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        client = get_client(event["client"])
        log_time = timestamp_to_datetime(event["time"])
        query = event["query"]
        do_update_user_activity(user_profile, client, query, log_time)

@assign_queue('user_activity_interval')
class UserActivityIntervalWorker(QueueProcessingWorker):
    def consume(self, event: Mapping[str, Any]) -> None:
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        log_time = timestamp_to_datetime(event["time"])
        do_update_user_activity_interval(user_profile, log_time)

@assign_queue('user_presence')
class UserPresenceWorker(QueueProcessingWorker):
    def consume(self, event: Mapping[str, Any]) -> None:
        logging.debug("Received presence event: %s" % (event,),)
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        client = get_client(event["client"])
        log_time = timestamp_to_datetime(event["time"])
        status = event["status"]
        do_update_user_presence(user_profile, client, log_time, status)

@assign_queue('missedmessage_emails', queue_type="loop")
class MissedMessageWorker(QueueProcessingWorker):
    # Aggregate all messages received over the last BATCH_DURATION
    # seconds to let someone finish sending a batch of messages and/or
    # editing them before they are sent out as emails to recipients.
    #
    # The timer is running whenever; we poll at most every TIMER_FREQUENCY
    # seconds, to avoid excessive activity.
    #
    # TODO: Since this process keeps events in memory for up to 2
    # minutes, it now will lose approximately BATCH_DURATION worth of
    # missed_message emails whenever it is restarted as part of a
    # server restart.  We should probably add some sort of save/reload
    # mechanism for that case.
    TIMER_FREQUENCY = 5
    BATCH_DURATION = 120
    timer_event = None  # type: Optional[Timer]
    events_by_recipient = defaultdict(list)  # type: Dict[int, List[Dict[str, Any]]]
    batch_start_by_recipient = {}  # type: Dict[int, float]

    def consume(self, event: Dict[str, Any]) -> None:
        logging.debug("Received missedmessage_emails event: %s" % (event,))

        # When we process an event, just put it into the queue and ensure we have a timer going.
        user_profile_id = event['user_profile_id']
        if user_profile_id not in self.batch_start_by_recipient:
            self.batch_start_by_recipient[user_profile_id] = time.time()
        self.events_by_recipient[user_profile_id].append(event)

        self.ensure_timer()

    def ensure_timer(self) -> None:
        if self.timer_event is not None:
            return
        self.timer_event = Timer(self.TIMER_FREQUENCY, MissedMessageWorker.maybe_send_batched_emails, [self])
        self.timer_event.start()

    def stop_timer(self) -> None:
        if self.timer_event and self.timer_event.is_alive():  # type: ignore # Report mypy bug.
            self.timer_event.cancel()
            self.timer_event = None

    def maybe_send_batched_emails(self) -> None:
        self.stop_timer()

        current_time = time.time()
        for user_profile_id, timestamp in list(self.batch_start_by_recipient.items()):
            if current_time - timestamp < self.BATCH_DURATION:
                continue
            events = self.events_by_recipient[user_profile_id]
            logging.info("Batch-processing %s missedmessage_emails events for user %s" %
                         (len(events), user_profile_id))
            handle_missedmessage_emails(user_profile_id, events)
            del self.events_by_recipient[user_profile_id]
            del self.batch_start_by_recipient[user_profile_id]

        # By only restarting the timer if there are actually events in
        # the queue, we ensure this queue processor is idle when there
        # are no missed-message emails to process.
        if len(self.batch_start_by_recipient) > 0:
            self.ensure_timer()<commit_msg>mypy: Remove type ignore on MissedMessageWorker.stop_timer.<commit_after># Documented in https://zulip.readthedocs.io/en/latest/subsystems/queuing.html
from typing import Any, Callable, Dict, List, Mapping, Optional, cast, TypeVar, Type

import copy
import signal
import tempfile
from functools import wraps
from threading import Timer

import smtplib
import socket

from django.conf import settings
from django.db import connection
from django.core.handlers.wsgi import WSGIRequest
from django.core.handlers.base import BaseHandler
from zerver.models import \
    get_client, get_system_bot, PreregistrationUser, \
    get_user_profile_by_id, Message, Realm, UserMessage, UserProfile, \
    Client
from zerver.lib.context_managers import lockfile
from zerver.lib.error_notify import do_report_error
from zerver.lib.feedback import handle_feedback
from zerver.lib.queue import SimpleQueueClient, queue_json_publish, retry_event
from zerver.lib.timestamp import timestamp_to_datetime
from zerver.lib.email_notifications import handle_missedmessage_emails
from zerver.lib.push_notifications import handle_push_notification, handle_remove_push_notification, \
    initialize_push_notifications
from zerver.lib.actions import do_send_confirmation_email, \
    do_update_user_activity, do_update_user_activity_interval, do_update_user_presence, \
    internal_send_message, internal_send_private_message, notify_export_completed, \
    render_incoming_message, do_update_embedded_data, do_mark_stream_messages_as_read
from zerver.lib.url_preview import preview as url_preview
from zerver.lib.digest import handle_digest_email
from zerver.lib.send_email import send_future_email, send_email_from_dict, \
    FromAddress, EmailNotDeliveredException, handle_send_email_format_changes
from zerver.lib.email_mirror import process_message as mirror_email, rate_limit_mirror_by_realm, \
    is_missed_message_address, extract_and_validate
from zerver.lib.streams import access_stream_by_id
from zerver.tornado.socket import req_redis_key, respond_send_message
from confirmation.models import Confirmation, create_confirmation_link
from zerver.lib.db import reset_queries
from zerver.lib.redis_utils import get_redis_client
from zerver.context_processors import common_context
from zerver.lib.outgoing_webhook import do_rest_call, get_outgoing_webhook_service_handler
from zerver.models import get_bot_services, RealmAuditLog
from zulip_bots.lib import extract_query_without_mention
from zerver.lib.bot_lib import EmbeddedBotHandler, get_bot_handler, EmbeddedBotQuitException
from zerver.lib.exceptions import RateLimited
from zerver.lib.export import export_realm_wrapper

import os
import sys
import ujson
from collections import defaultdict
import email
import time
import datetime
import logging
import requests
from io import StringIO
import urllib

logger = logging.getLogger(__name__)

class WorkerDeclarationException(Exception):
    pass

ConcreteQueueWorker = TypeVar('ConcreteQueueWorker', bound='QueueProcessingWorker')

def assign_queue(
        queue_name: str, enabled: bool=True, queue_type: str="consumer"
) -> Callable[[Type[ConcreteQueueWorker]], Type[ConcreteQueueWorker]]:
    def decorate(clazz: Type[ConcreteQueueWorker]) -> Type[ConcreteQueueWorker]:
        clazz.queue_name = queue_name
        if enabled:
            register_worker(queue_name, clazz, queue_type)
        return clazz
    return decorate

worker_classes = {}  # type: Dict[str, Type[QueueProcessingWorker]]
queues = {}  # type: Dict[str, Dict[str, Type[QueueProcessingWorker]]]
def register_worker(queue_name: str, clazz: Type['QueueProcessingWorker'], queue_type: str) -> None:
    if queue_type not in queues:
        queues[queue_type] = {}
    queues[queue_type][queue_name] = clazz
    worker_classes[queue_name] = clazz

def get_worker(queue_name: str) -> 'QueueProcessingWorker':
    return worker_classes[queue_name]()

def get_active_worker_queues(queue_type: Optional[str]=None) -> List[str]:
    """Returns all the non-test worker queues."""
    if queue_type is None:
        return list(worker_classes.keys())
    return list(queues[queue_type].keys())

def check_and_send_restart_signal() -> None:
    try:
        if not connection.is_usable():
            logging.warning("*** Sending self SIGUSR1 to trigger a restart.")
            os.kill(os.getpid(), signal.SIGUSR1)
    except Exception:
        pass

def retry_send_email_failures(
        func: Callable[[ConcreteQueueWorker, Dict[str, Any]], None]
) -> Callable[['QueueProcessingWorker', Dict[str, Any]], None]:

    @wraps(func)
    def wrapper(worker: ConcreteQueueWorker, data: Dict[str, Any]) -> None:
        try:
            func(worker, data)
        except (smtplib.SMTPServerDisconnected, socket.gaierror, EmailNotDeliveredException):
            def on_failure(event: Dict[str, Any]) -> None:
                logging.exception("Event {} failed".format(event))

            retry_event(worker.queue_name, data, on_failure)

    return wrapper

class QueueProcessingWorker:
    queue_name = None  # type: str

    def __init__(self) -> None:
        self.q = None  # type: SimpleQueueClient
        if self.queue_name is None:
            raise WorkerDeclarationException("Queue worker declared without queue_name")

    def consume(self, data: Dict[str, Any]) -> None:
        raise WorkerDeclarationException("No consumer defined!")

    def consume_wrapper(self, data: Dict[str, Any]) -> None:
        try:
            self.consume(data)
        except Exception:
            self._log_problem()
            if not os.path.exists(settings.QUEUE_ERROR_DIR):
                os.mkdir(settings.QUEUE_ERROR_DIR)  # nocoverage
            fname = '%s.errors' % (self.queue_name,)
            fn = os.path.join(settings.QUEUE_ERROR_DIR, fname)
            line = '%s\t%s\n' % (time.asctime(), ujson.dumps(data))
            lock_fn = fn + '.lock'
            with lockfile(lock_fn):
                with open(fn, 'ab') as f:
                    f.write(line.encode('utf-8'))
            check_and_send_restart_signal()
        finally:
            reset_queries()

    def _log_problem(self) -> None:
        logging.exception("Problem handling data on queue %s" % (self.queue_name,))

    def setup(self) -> None:
        self.q = SimpleQueueClient()

    def start(self) -> None:
        self.q.register_json_consumer(self.queue_name, self.consume_wrapper)
        self.q.start_consuming()

    def stop(self) -> None:  # nocoverage
        self.q.stop_consuming()

class LoopQueueProcessingWorker(QueueProcessingWorker):
    sleep_delay = 0

    def start(self) -> None:  # nocoverage
        while True:
            # TODO: Probably it'd be better to share code with consume_wrapper()
            events = self.q.drain_queue(self.queue_name, json=True)
            try:
                self.consume_batch(events)
            finally:
                reset_queries()
            time.sleep(self.sleep_delay)

    def consume_batch(self, event: List[Dict[str, Any]]) -> None:
        raise NotImplementedError

    def consume(self, event: Dict[str, Any]) -> None:
        """In LoopQueueProcessingWorker, consume is used just for automated tests"""
        self.consume_batch([event])

@assign_queue('signups')
class SignupWorker(QueueProcessingWorker):
    def consume(self, data: Dict[str, Any]) -> None:
        # TODO: This is the only implementation with Dict cf Mapping; should we simplify?
        user_profile = get_user_profile_by_id(data['user_id'])
        logging.info("Processing signup for user %s in realm %s" % (
            user_profile.email, user_profile.realm.string_id))
        if settings.MAILCHIMP_API_KEY and settings.PRODUCTION:
            endpoint = "https://%s.api.mailchimp.com/3.0/lists/%s/members" % \
                       (settings.MAILCHIMP_API_KEY.split('-')[1], settings.ZULIP_FRIENDS_LIST_ID)
            params = dict(data)
            del params['user_id']
            params['list_id'] = settings.ZULIP_FRIENDS_LIST_ID
            params['status'] = 'subscribed'
            r = requests.post(endpoint, auth=('apikey', settings.MAILCHIMP_API_KEY), json=params, timeout=10)
            if r.status_code == 400 and ujson.loads(r.text)['title'] == 'Member Exists':
                logging.warning("Attempted to sign up already existing email to list: %s" %
                                (data['email_address'],))
            elif r.status_code == 400:
                retry_event('signups', data, lambda e: r.raise_for_status())
            else:
                r.raise_for_status()

@assign_queue('invites')
class ConfirmationEmailWorker(QueueProcessingWorker):
    def consume(self, data: Mapping[str, Any]) -> None:
        if "email" in data:
            # When upgrading from a version up through 1.7.1, there may be
            # existing items in the queue with `email` instead of `prereg_id`.
            invitee = PreregistrationUser.objects.filter(
                email__iexact=data["email"].strip()).latest("invited_at")
        else:
            invitee = PreregistrationUser.objects.filter(id=data["prereg_id"]).first()
            if invitee is None:
                # The invitation could have been revoked
                return

        referrer = get_user_profile_by_id(data["referrer_id"])
        logger.info("Sending invitation for realm %s to %s" % (referrer.realm.string_id, invitee.email))
        do_send_confirmation_email(invitee, referrer)

        # queue invitation reminder for two days from now.
        link = create_confirmation_link(invitee, referrer.realm.host, Confirmation.INVITATION)
        context = common_context(referrer)
        context.update({
            'activate_url': link,
            'referrer_name': referrer.full_name,
            'referrer_email': referrer.email,
            'referrer_realm_name': referrer.realm.name,
        })
        send_future_email(
            "zerver/emails/invitation_reminder",
            referrer.realm,
            to_emails=[invitee.email],
            from_address=FromAddress.tokenized_no_reply_address(),
            language=referrer.realm.default_language,
            context=context,
            delay=datetime.timedelta(days=2))

@assign_queue('user_activity')
class UserActivityWorker(QueueProcessingWorker):
    def consume(self, event: Mapping[str, Any]) -> None:
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        client = get_client(event["client"])
        log_time = timestamp_to_datetime(event["time"])
        query = event["query"]
        do_update_user_activity(user_profile, client, query, log_time)

@assign_queue('user_activity_interval')
class UserActivityIntervalWorker(QueueProcessingWorker):
    def consume(self, event: Mapping[str, Any]) -> None:
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        log_time = timestamp_to_datetime(event["time"])
        do_update_user_activity_interval(user_profile, log_time)

@assign_queue('user_presence')
class UserPresenceWorker(QueueProcessingWorker):
    def consume(self, event: Mapping[str, Any]) -> None:
        logging.debug("Received presence event: %s" % (event,),)
        user_profile = get_user_profile_by_id(event["user_profile_id"])
        client = get_client(event["client"])
        log_time = timestamp_to_datetime(event["time"])
        status = event["status"]
        do_update_user_presence(user_profile, client, log_time, status)

@assign_queue('missedmessage_emails', queue_type="loop")
class MissedMessageWorker(QueueProcessingWorker):
    # Aggregate all messages received over the last BATCH_DURATION
    # seconds to let someone finish sending a batch of messages and/or
    # editing them before they are sent out as emails to recipients.
    #
    # The timer is running whenever; we poll at most every TIMER_FREQUENCY
    # seconds, to avoid excessive activity.
    #
    # TODO: Since this process keeps events in memory for up to 2
    # minutes, it now will lose approximately BATCH_DURATION worth of
    # missed_message emails whenever it is restarted as part of a
    # server restart.  We should probably add some sort of save/reload
    # mechanism for that case.
    TIMER_FREQUENCY = 5
    BATCH_DURATION = 120
    timer_event = None  # type: Optional[Timer]
    events_by_recipient = defaultdict(list)  # type: Dict[int, List[Dict[str, Any]]]
    batch_start_by_recipient = {}  # type: Dict[int, float]

    def consume(self, event: Dict[str, Any]) -> None:
        logging.debug("Received missedmessage_emails event: %s" % (event,))

        # When we process an event, just put it into the queue and ensure we have a timer going.
        user_profile_id = event['user_profile_id']
        if user_profile_id not in self.batch_start_by_recipient:
            self.batch_start_by_recipient[user_profile_id] = time.time()
        self.events_by_recipient[user_profile_id].append(event)

        self.ensure_timer()

    def ensure_timer(self) -> None:
        if self.timer_event is not None:
            return
        self.timer_event = Timer(self.TIMER_FREQUENCY, MissedMessageWorker.maybe_send_batched_emails, [self])
        self.timer_event.start()

    def stop_timer(self) -> None:
        if self.timer_event and self.timer_event.is_alive():
            self.timer_event.cancel()
            self.timer_event = None

    def maybe_send_batched_emails(self) -> None:
        self.stop_timer()

        current_time = time.time()
        for user_profile_id, timestamp in list(self.batch_start_by_recipient.items()):
            if current_time - timestamp < self.BATCH_DURATION:
                continue
            events = self.events_by_recipient[user_profile_id]
            logging.info("Batch-processing %s missedmessage_emails events for user %s" %
                         (len(events), user_profile_id))
            handle_missedmessage_emails(user_profile_id, events)
            del self.events_by_recipient[user_profile_id]
            del self.batch_start_by_recipient[user_profile_id]

        # By only restarting the timer if there are actually events in
        # the queue, we ensure this queue processor is idle when there
        # are no missed-message emails to process.
        if len(self.batch_start_by_recipient) > 0:
            self.ensure_timer()
{'fraction_non_alphanumeric': 0.09040828422997803, 'fraction_numerical': 0.0017433143893169694, 'mean_word_length': 3.370257504190157, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 61, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3810842', 'n_tokens_mistral': 8641, 'n_tokens_neox': 8421, 'n_words': 1989}