<commit_before># -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ._base import SQLBaseStore

from twisted.internet import defer, reactor

from synapse.events import FrozenEvent, USE_FROZEN_DICTS
from synapse.events.utils import prune_event

from synapse.util.async import ObservableDeferred
from synapse.util.logcontext import (
    preserve_fn, PreserveLoggingContext, preserve_context_over_deferred
)
from synapse.util.logutils import log_function
from synapse.util.metrics import Measure
from synapse.api.constants import EventTypes
from synapse.api.errors import SynapseError
from synapse.state import resolve_events
from synapse.util.caches.descriptors import cached
from synapse.types import get_domain_from_id

from canonicaljson import encode_canonical_json
from collections import deque, namedtuple, OrderedDict
from functools import wraps

import synapse.metrics

import logging
import math
import ujson as json

# these are only included to make the type annotations work
from synapse.events import EventBase    # noqa: F401
from synapse.events.snapshot import EventContext   # noqa: F401

logger = logging.getLogger(__name__)


metrics = synapse.metrics.get_metrics_for(__name__)
persist_event_counter = metrics.register_counter("persisted_events")
event_counter = metrics.register_counter(
    "persisted_events_sep", labels=["type", "origin_type", "origin_entity"]
)


def encode_json(json_object):
    if USE_FROZEN_DICTS:
        # ujson doesn't like frozen_dicts
        return encode_canonical_json(json_object)
    else:
        return json.dumps(json_object, ensure_ascii=False)


# These values are used in the `enqueus_event` and `_do_fetch` methods to
# control how we batch/bulk fetch events from the database.
# The values are plucked out of thing air to make initial sync run faster
# on jki.re
# TODO: Make these configurable.
EVENT_QUEUE_THREADS = 3  # Max number of threads that will fetch events
EVENT_QUEUE_ITERATIONS = 3  # No. times we block waiting for requests for events
EVENT_QUEUE_TIMEOUT_S = 0.1  # Timeout when waiting for requests for events


class _EventPeristenceQueue(object):
    """Queues up events so that they can be persisted in bulk with only one
    concurrent transaction per room.
    """

    _EventPersistQueueItem = namedtuple("_EventPersistQueueItem", (
        "events_and_contexts", "backfilled", "deferred",
    ))

    def __init__(self):
        self._event_persist_queues = {}
        self._currently_persisting_rooms = set()

    def add_to_queue(self, room_id, events_and_contexts, backfilled):
        """Add events to the queue, with the given persist_event options.

        Args:
            room_id (str):
            events_and_contexts (list[(EventBase, EventContext)]):
            backfilled (bool):
        """
        queue = self._event_persist_queues.setdefault(room_id, deque())
        if queue:
            end_item = queue[-1]
            if end_item.backfilled == backfilled:
                end_item.events_and_contexts.extend(events_and_contexts)
                return end_item.deferred.observe()

        deferred = ObservableDeferred(defer.Deferred())

        queue.append(self._EventPersistQueueItem(
            events_and_contexts=events_and_contexts,
            backfilled=backfilled,
            deferred=deferred,
        ))

        return deferred.observe()

    def handle_queue(self, room_id, per_item_callback):
        """Attempts to handle the queue for a room if not already being handled.

        The given callback will be invoked with for each item in the queue,1
        of type _EventPersistQueueItem. The per_item_callback will continuously
        be called with new items, unless the queue becomnes empty. The return
        value of the function will be given to the deferreds waiting on the item,
        exceptions will be passed to the deferres as well.

        This function should therefore be called whenever anything is added
        to the queue.

        If another callback is currently handling the queue then it will not be
        invoked.
        """

        if room_id in self._currently_persisting_rooms:
            return

        self._currently_persisting_rooms.add(room_id)

        @defer.inlineCallbacks
        def handle_queue_loop():
            try:
                queue = self._get_drainining_queue(room_id)
                for item in queue:
                    try:
                        ret = yield per_item_callback(item)
                        item.deferred.callback(ret)
                    except Exception as e:
                        item.deferred.errback(e)
            finally:
                queue = self._event_persist_queues.pop(room_id, None)
                if queue:
                    self._event_persist_queues[room_id] = queue
                self._currently_persisting_rooms.discard(room_id)

        preserve_fn(handle_queue_loop)()

    def _get_drainining_queue(self, room_id):
        queue = self._event_persist_queues.setdefault(room_id, deque())

        try:
            while True:
                yield queue.popleft()
        except IndexError:
            # Queue has been drained.
            pass


_EventCacheEntry = namedtuple("_EventCacheEntry", ("event", "redacted_event"))


def _retry_on_integrity_error(func):
    """Wraps a database function so that it gets retried on IntegrityError,
    with `delete_existing=True` passed in.

    Args:
        func: function that returns a Deferred and accepts a `delete_existing` arg
    """
    @wraps(func)
    @defer.inlineCallbacks
    def f(self, *args, **kwargs):
        try:
            res = yield func(self, *args, **kwargs)
        except self.database_engine.module.IntegrityError:
            logger.exception("IntegrityError, retrying.")
            res = yield func(self, *args, delete_existing=True, **kwargs)
        defer.returnValue(res)

    return f


class EventsStore(SQLBaseStore):
    EVENT_ORIGIN_SERVER_TS_NAME = "event_origin_server_ts"
    EVENT_FIELDS_SENDER_URL_UPDATE_NAME = "event_fields_sender_url"

    def __init__(self, hs):
        super(EventsStore, self).__init__(hs)
        self._clock = hs.get_clock()
        self.register_background_update_handler(
            self.EVENT_ORIGIN_SERVER_TS_NAME, self._background_reindex_origin_server_ts
        )
        self.register_background_update_handler(
            self.EVENT_FIELDS_SENDER_URL_UPDATE_NAME,
            self._background_reindex_fields_sender,
        )

        self.register_background_index_update(
            "event_contains_url_index",
            index_name="event_contains_url_index",
            table="events",
            columns=["room_id", "topological_ordering", "stream_ordering"],
            where_clause="contains_url = true AND outlier = false",
        )

        self._event_persist_queue = _EventPeristenceQueue()

    def persist_events(self, events_and_contexts, backfilled=False):
        """
        Write events to the database
        Args:
            events_and_contexts: list of tuples of (event, context)
            backfilled: ?
        """
        partitioned = {}
        for event, ctx in events_and_contexts:
            partitioned.setdefault(event.room_id, []).append((event, ctx))

        deferreds = []
        for room_id, evs_ctxs in partitioned.iteritems():
            d = preserve_fn(self._event_persist_queue.add_to_queue)(
                room_id, evs_ctxs,
                backfilled=backfilled,
            )
            deferreds.append(d)

        for room_id in partitioned:
            self._maybe_start_persisting(room_id)

        return preserve_context_over_deferred(
            defer.gatherResults(deferreds, consumeErrors=True)
        )

    @defer.inlineCallbacks
    @log_function
    def persist_event(self, event, context, backfilled=False):
        """

        Args:
            event (EventBase):
            context (EventContext):
            backfilled (bool):

        Returns:
            Deferred: resolves to (int, int): the stream ordering of ``event``,
            and the stream ordering of the latest persisted event
        """
        deferred = self._event_persist_queue.add_to_queue(
            event.room_id, [(event, context)],
            backfilled=backfilled,
        )

        self._maybe_start_persisting(event.room_id)

        yield preserve_context_over_deferred(deferred)

        max_persisted_id = yield self._stream_id_gen.get_current_token()
        defer.returnValue((event.internal_metadata.stream_ordering, max_persisted_id))

    def _maybe_start_persisting(self, room_id):
        @defer.inlineCallbacks
        def persisting_queue(item):
            yield self._persist_events(
                item.events_and_contexts,
                backfilled=item.backfilled,
            )

        self._event_persist_queue.handle_queue(room_id, persisting_queue)

    @_retry_on_integrity_error
    @defer.inlineCallbacks
    def _persist_events(self, events_and_contexts, backfilled=False,
                        delete_existing=False):
        """Persist events to db

        Args:
            events_and_contexts (list[(EventBase, EventContext)]):
            backfilled (bool):
            delete_existing (bool):

        Returns:
            Deferred: resolves when the events have been persisted
        """
        if not events_and_contexts:
            return

        if backfilled:
            stream_ordering_manager = self._backfill_id_gen.get_next_mult(
                len(events_and_contexts)
            )
        else:
            stream_ordering_manager = self._stream_id_gen.get_next_mult(
                len(events_and_contexts)
            )

        with stream_ordering_manager as stream_orderings:
            for (event, context), stream, in zip(
                events_and_contexts, stream_orderings
            ):
                event.internal_metadata.stream_ordering = stream

            chunks = [
                events_and_contexts[x:x + 100]
                for x in xrange(0, len(events_and_contexts), 100)
            ]

            for chunk in chunks:
                # We can't easily parallelize these since different chunks
                # might contain the same event. :(

                # NB: Assumes that we are only persisting events for one room
                # at a time.
                new_forward_extremeties = {}
                current_state_for_room = {}
                if not backfilled:
                    with Measure(self._clock, "_calculate_state_and_extrem"):
                        # Work out the new "current state" for each room.
                        # We do this by working out what the new extremities are and then
                        # calculating the state from that.
                        events_by_room = {}
                        for event, context in chunk:
                            events_by_room.setdefault(event.room_id, []).append(
                                (event, context)
                            )

                        for room_id, ev_ctx_rm in events_by_room.iteritems():
                            # Work out new extremities by recursively adding and removing
                            # the new events.
                            latest_event_ids = yield self.get_latest_event_ids_in_room(
                                room_id
                            )
                            new_latest_event_ids = yield self._calculate_new_extremeties(
                                room_id, ev_ctx_rm, latest_event_ids
                            )

                            if new_latest_event_ids == set(latest_event_ids):
                                # No change in extremities, so no change in state
                                continue

                            new_forward_extremeties[room_id] = new_latest_event_ids

                            len_1 = (
                                len(latest_event_ids) == 1
                                and len(new_latest_event_ids) == 1
                            )
                            if len_1:
                                all_single_prev_not_state = all(
                                    len(event.prev_events) == 1
                                    and not event.is_state()
                                    for event, ctx in ev_ctx_rm
                                )
                                # Don't bother calculating state if they're just
                                # a long chain of single ancestor non-state events.
                                if all_single_prev_not_state:
                                    continue

                            state = yield self._calculate_state_delta(
                                room_id, ev_ctx_rm, new_latest_event_ids
                            )
                            if state:
                                current_state_for_room[room_id] = state

                yield self.runInteraction(
                    "persist_events",
                    self._persist_events_txn,
                    events_and_contexts=chunk,
                    backfilled=backfilled,
                    delete_existing=delete_existing,
                    current_state_for_room=current_state_for_room,
                    new_forward_extremeties=new_forward_extremeties,
                )
                persist_event_counter.inc_by(len(chunk))

                for event, context in chunk:
                    if context.app_service:
                        origin_type = "local"
                        origin_entity = context.app_service.id
                    elif self.hs.is_mine_id(event.sender):
                        origin_type = "local"
                        origin_entity = "*client*"
                    else:
                        origin_type = "remote"
                        origin_entity = get_domain_from_id(event.sender)

                    event_counter.inc(event.type, origin_type, origin_entity)

                for room_id, (_, _, new_state) in current_state_for_room.iteritems():
                    self.get_current_state_ids.prefill(
                        (room_id, ), new_state
                    )

    @defer.inlineCallbacks
    def _calculate_new_extremeties(self, room_id, event_contexts, latest_event_ids):
        """Calculates the new forward extremeties for a room given events to
        persist.

        Assumes that we are only persisting events for one room at a time.
        """
        new_latest_event_ids = set(latest_event_ids)
        # First, add all the new events to the list
        new_latest_event_ids.update(
            event.event_id for event, ctx in event_contexts<commit_msg>Add more granular event send metrics<commit_after># -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ._base import SQLBaseStore

from twisted.internet import defer, reactor

from synapse.events import FrozenEvent, USE_FROZEN_DICTS
from synapse.events.utils import prune_event

from synapse.util.async import ObservableDeferred
from synapse.util.logcontext import (
    preserve_fn, PreserveLoggingContext, preserve_context_over_deferred
)
from synapse.util.logutils import log_function
from synapse.util.metrics import Measure
from synapse.api.constants import EventTypes
from synapse.api.errors import SynapseError
from synapse.state import resolve_events
from synapse.util.caches.descriptors import cached
from synapse.types import get_domain_from_id

from canonicaljson import encode_canonical_json
from collections import deque, namedtuple, OrderedDict
from functools import wraps

import synapse.metrics

import logging
import math
import ujson as json

# these are only included to make the type annotations work
from synapse.events import EventBase    # noqa: F401
from synapse.events.snapshot import EventContext   # noqa: F401

logger = logging.getLogger(__name__)


metrics = synapse.metrics.get_metrics_for(__name__)
persist_event_counter = metrics.register_counter("persisted_events")
event_counter = metrics.register_counter(
    "persisted_events_sep", labels=["type", "origin_type", "origin_entity"]
)


def encode_json(json_object):
    if USE_FROZEN_DICTS:
        # ujson doesn't like frozen_dicts
        return encode_canonical_json(json_object)
    else:
        return json.dumps(json_object, ensure_ascii=False)


# These values are used in the `enqueus_event` and `_do_fetch` methods to
# control how we batch/bulk fetch events from the database.
# The values are plucked out of thing air to make initial sync run faster
# on jki.re
# TODO: Make these configurable.
EVENT_QUEUE_THREADS = 3  # Max number of threads that will fetch events
EVENT_QUEUE_ITERATIONS = 3  # No. times we block waiting for requests for events
EVENT_QUEUE_TIMEOUT_S = 0.1  # Timeout when waiting for requests for events


class _EventPeristenceQueue(object):
    """Queues up events so that they can be persisted in bulk with only one
    concurrent transaction per room.
    """

    _EventPersistQueueItem = namedtuple("_EventPersistQueueItem", (
        "events_and_contexts", "backfilled", "deferred",
    ))

    def __init__(self):
        self._event_persist_queues = {}
        self._currently_persisting_rooms = set()

    def add_to_queue(self, room_id, events_and_contexts, backfilled):
        """Add events to the queue, with the given persist_event options.

        Args:
            room_id (str):
            events_and_contexts (list[(EventBase, EventContext)]):
            backfilled (bool):
        """
        queue = self._event_persist_queues.setdefault(room_id, deque())
        if queue:
            end_item = queue[-1]
            if end_item.backfilled == backfilled:
                end_item.events_and_contexts.extend(events_and_contexts)
                return end_item.deferred.observe()

        deferred = ObservableDeferred(defer.Deferred())

        queue.append(self._EventPersistQueueItem(
            events_and_contexts=events_and_contexts,
            backfilled=backfilled,
            deferred=deferred,
        ))

        return deferred.observe()

    def handle_queue(self, room_id, per_item_callback):
        """Attempts to handle the queue for a room if not already being handled.

        The given callback will be invoked with for each item in the queue,1
        of type _EventPersistQueueItem. The per_item_callback will continuously
        be called with new items, unless the queue becomnes empty. The return
        value of the function will be given to the deferreds waiting on the item,
        exceptions will be passed to the deferres as well.

        This function should therefore be called whenever anything is added
        to the queue.

        If another callback is currently handling the queue then it will not be
        invoked.
        """

        if room_id in self._currently_persisting_rooms:
            return

        self._currently_persisting_rooms.add(room_id)

        @defer.inlineCallbacks
        def handle_queue_loop():
            try:
                queue = self._get_drainining_queue(room_id)
                for item in queue:
                    try:
                        ret = yield per_item_callback(item)
                        item.deferred.callback(ret)
                    except Exception as e:
                        item.deferred.errback(e)
            finally:
                queue = self._event_persist_queues.pop(room_id, None)
                if queue:
                    self._event_persist_queues[room_id] = queue
                self._currently_persisting_rooms.discard(room_id)

        preserve_fn(handle_queue_loop)()

    def _get_drainining_queue(self, room_id):
        queue = self._event_persist_queues.setdefault(room_id, deque())

        try:
            while True:
                yield queue.popleft()
        except IndexError:
            # Queue has been drained.
            pass


_EventCacheEntry = namedtuple("_EventCacheEntry", ("event", "redacted_event"))


def _retry_on_integrity_error(func):
    """Wraps a database function so that it gets retried on IntegrityError,
    with `delete_existing=True` passed in.

    Args:
        func: function that returns a Deferred and accepts a `delete_existing` arg
    """
    @wraps(func)
    @defer.inlineCallbacks
    def f(self, *args, **kwargs):
        try:
            res = yield func(self, *args, **kwargs)
        except self.database_engine.module.IntegrityError:
            logger.exception("IntegrityError, retrying.")
            res = yield func(self, *args, delete_existing=True, **kwargs)
        defer.returnValue(res)

    return f


class EventsStore(SQLBaseStore):
    EVENT_ORIGIN_SERVER_TS_NAME = "event_origin_server_ts"
    EVENT_FIELDS_SENDER_URL_UPDATE_NAME = "event_fields_sender_url"

    def __init__(self, hs):
        super(EventsStore, self).__init__(hs)
        self._clock = hs.get_clock()
        self.register_background_update_handler(
            self.EVENT_ORIGIN_SERVER_TS_NAME, self._background_reindex_origin_server_ts
        )
        self.register_background_update_handler(
            self.EVENT_FIELDS_SENDER_URL_UPDATE_NAME,
            self._background_reindex_fields_sender,
        )

        self.register_background_index_update(
            "event_contains_url_index",
            index_name="event_contains_url_index",
            table="events",
            columns=["room_id", "topological_ordering", "stream_ordering"],
            where_clause="contains_url = true AND outlier = false",
        )

        self._event_persist_queue = _EventPeristenceQueue()

    def persist_events(self, events_and_contexts, backfilled=False):
        """
        Write events to the database
        Args:
            events_and_contexts: list of tuples of (event, context)
            backfilled: ?
        """
        partitioned = {}
        for event, ctx in events_and_contexts:
            partitioned.setdefault(event.room_id, []).append((event, ctx))

        deferreds = []
        for room_id, evs_ctxs in partitioned.iteritems():
            d = preserve_fn(self._event_persist_queue.add_to_queue)(
                room_id, evs_ctxs,
                backfilled=backfilled,
            )
            deferreds.append(d)

        for room_id in partitioned:
            self._maybe_start_persisting(room_id)

        return preserve_context_over_deferred(
            defer.gatherResults(deferreds, consumeErrors=True)
        )

    @defer.inlineCallbacks
    @log_function
    def persist_event(self, event, context, backfilled=False):
        """

        Args:
            event (EventBase):
            context (EventContext):
            backfilled (bool):

        Returns:
            Deferred: resolves to (int, int): the stream ordering of ``event``,
            and the stream ordering of the latest persisted event
        """
        deferred = self._event_persist_queue.add_to_queue(
            event.room_id, [(event, context)],
            backfilled=backfilled,
        )

        self._maybe_start_persisting(event.room_id)

        yield preserve_context_over_deferred(deferred)

        max_persisted_id = yield self._stream_id_gen.get_current_token()
        defer.returnValue((event.internal_metadata.stream_ordering, max_persisted_id))

    def _maybe_start_persisting(self, room_id):
        @defer.inlineCallbacks
        def persisting_queue(item):
            yield self._persist_events(
                item.events_and_contexts,
                backfilled=item.backfilled,
            )

        self._event_persist_queue.handle_queue(room_id, persisting_queue)

    @_retry_on_integrity_error
    @defer.inlineCallbacks
    def _persist_events(self, events_and_contexts, backfilled=False,
                        delete_existing=False):
        """Persist events to db

        Args:
            events_and_contexts (list[(EventBase, EventContext)]):
            backfilled (bool):
            delete_existing (bool):

        Returns:
            Deferred: resolves when the events have been persisted
        """
        if not events_and_contexts:
            return

        if backfilled:
            stream_ordering_manager = self._backfill_id_gen.get_next_mult(
                len(events_and_contexts)
            )
        else:
            stream_ordering_manager = self._stream_id_gen.get_next_mult(
                len(events_and_contexts)
            )

        with stream_ordering_manager as stream_orderings:
            for (event, context), stream, in zip(
                events_and_contexts, stream_orderings
            ):
                event.internal_metadata.stream_ordering = stream

            chunks = [
                events_and_contexts[x:x + 100]
                for x in xrange(0, len(events_and_contexts), 100)
            ]

            for chunk in chunks:
                # We can't easily parallelize these since different chunks
                # might contain the same event. :(

                # NB: Assumes that we are only persisting events for one room
                # at a time.
                new_forward_extremeties = {}
                current_state_for_room = {}
                if not backfilled:
                    with Measure(self._clock, "_calculate_state_and_extrem"):
                        # Work out the new "current state" for each room.
                        # We do this by working out what the new extremities are and then
                        # calculating the state from that.
                        events_by_room = {}
                        for event, context in chunk:
                            events_by_room.setdefault(event.room_id, []).append(
                                (event, context)
                            )

                        for room_id, ev_ctx_rm in events_by_room.iteritems():
                            # Work out new extremities by recursively adding and removing
                            # the new events.
                            latest_event_ids = yield self.get_latest_event_ids_in_room(
                                room_id
                            )
                            new_latest_event_ids = yield self._calculate_new_extremeties(
                                room_id, ev_ctx_rm, latest_event_ids
                            )

                            if new_latest_event_ids == set(latest_event_ids):
                                # No change in extremities, so no change in state
                                continue

                            new_forward_extremeties[room_id] = new_latest_event_ids

                            len_1 = (
                                len(latest_event_ids) == 1
                                and len(new_latest_event_ids) == 1
                            )
                            if len_1:
                                all_single_prev_not_state = all(
                                    len(event.prev_events) == 1
                                    and not event.is_state()
                                    for event, ctx in ev_ctx_rm
                                )
                                # Don't bother calculating state if they're just
                                # a long chain of single ancestor non-state events.
                                if all_single_prev_not_state:
                                    continue

                            state = yield self._calculate_state_delta(
                                room_id, ev_ctx_rm, new_latest_event_ids
                            )
                            if state:
                                current_state_for_room[room_id] = state

                yield self.runInteraction(
                    "persist_events",
                    self._persist_events_txn,
                    events_and_contexts=chunk,
                    backfilled=backfilled,
                    delete_existing=delete_existing,
                    current_state_for_room=current_state_for_room,
                    new_forward_extremeties=new_forward_extremeties,
                )
                persist_event_counter.inc_by(len(chunk))
                for event, context in chunk:
                    if context.app_service:
                        origin_type = "local"
                        origin_entity = context.app_service.id
                    elif self.hs.is_mine_id(event.sender):
                        origin_type = "local"
                        origin_entity = "*client*"
                    else:
                        origin_type = "remote"
                        origin_entity = get_domain_from_id(event.sender)

                    event_counter.inc(event.type, origin_type, origin_entity)

                for event, context in chunk:
                    if context.app_service:
                        origin_type = "local"
                        origin_entity = context.app_service.id
                    elif self.hs.is_mine_id(event.sender):
                        origin_type = "local"
                        origin_entity = "*client*"
                    else:
                        origin_type = "remote"
                        origin_entity = get_domain_from_id(event.sender)

                    event_counter.inc(event.type, origin_type, origin_entity)

                for room_id, (_, _, new_state) in current_state_for_room.iteritems():
                    self.get_current_state_ids.prefill(
                        (room_id, ), new_state
                    )

    @defer.inlineCallbacks
    def _calculate_new_extremeties(self, room_id, event_contexts, latest_event_ids):
        """Calculates the new forward extremeties for a room given events to
        persist.

        Assumes that we are only persisting events for one room at a time.
        """
        new_latest_event_ids = set(latest_event_ids)
        # First, add all the new events to the list
        new_latest_event_ids.update(
            event.event_id for event, ctx in event_contexts
{'fraction_non_alphanumeric': 0.06350832266325224, 'fraction_numerical': 0.0023687580025608195, 'mean_word_length': 1.9091163050563367, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3797007', 'n_tokens_mistral': 8463, 'n_tokens_neox': 8017, 'n_words': 2091}