Here we add the ability for the application service ephemeral message
processor to handle new events on the "to_device" stream.

We keep track of a stream id (token) per application service, and every
time a new to-device message comes in, for each appservice we pull the
messages between the last-recorded and current stream id and check
whether any of the messages are for a user in that appservice's user
namespace.

get_new_messages is implemented in the next commit.

since we rebased off latest develop.

This method is quite similar to the one below, except that it doesn't
support device ids, and supports querying with more than one user id,
both of which are relevant to application services.

The results are also formatted in a different data structure, so I'm
not sure how much we could really share here between the two methods.

This is for tracking the stream id that each application service has
been sent up to. In other words, there shouldn't be any need to process
stream ids below the recorded one here as the AS should have already
received them.

Note that there is no reliability built-in here. Reliability of delivery
if intended for a separate PR.

I decided to spin up another test class for this as the existing one is
1. quite old and 2. was mocking away too much of the infrastructure to
my liking. I've named the new class alluding to ephemeral messages, and
while we already have some ephemeral tests in AppServiceHandlerTestCase,
ideally I'd like to migrate those over.

There's two new tests here. One for testing that to-device messages for
a local user are received by any application services that have
registered interest in that user - and that those that haven't won't
receive those messages.

The next test is similar, but tests with a whole bunch of to-device
messages. Rather than actually registering tons of devices - which would
make for a very slow unit test - we just directly insert them into the
database.

…heck

The second check for self._msc2409_to_device_messages_enabled was not necessary. It's
already checked in notify_interested_services_ephemeral earlier.

Test cases needed to be updated, as we now always call
submit_ephemeral_events_for_as, it may just be with an
empty events list.

There is so much mocking going on here. I look forward to replacing these one day.

…e_pos

…actions

Things were starting to get a little inflexible as we kept adding new
types of data to send to application services. It's better to just
have one method for adding data to AS transactions, than one for
each type of data.

Note that subsequent PRs will need to add device lists, one-time keys
and fallback keys to these transactions. Adding those are additional
arguments to a method is much nicer than a new method for each one.

Plus with this setup we can add multiple different types of data at
once without kicking off an AS transaction for each type. This will
be useful for OTK/fallback keys, as we plan to lazily attach those
when handling other event types.

…heduler.enqueue*

With enqueue_for_appservice being called per-event per-appservice, it's
not a great target for mocking. So we use _TransactionController.send
instead, to track things that are actually sent out to AS's.

This also has the benefit of testing a wider bit of the AS txn pipeline

…_as_to_device

…appservice_stream_type.sql

Co-authored-by: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>

…_as_to_device

…_messages

This commit refactors the previously rather duplicated 'get_new_messages_for_device' and
'get_new_messages' methods into one new private method with combined logic, and two small
public methods. The public methods expose the correct interface for querying to-device
messages for either a single device (where a limit can be used) and multiple devices
(where using a limit is infeasible).

…_as_to_device

…-deps

Co-authored-by: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>

…sers

Co-authored-by: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>

…_as_to_device