🔥 Feature: Add broadcast to bot and space

[grivera64]

Description

This PR adds a broadcast method to matrix/bot.py and matrix/space.py.

matrix/bot.py:

    async def broadcast(
        self,
        rooms: list[Room],
        content: str | None = None,
        *,
        raw: bool = False,
        notice: bool = False,
        file: File | None = None,
    ) -> list[Message]

Behavior: Calls send() on the provided list of rooms (skips over spaces).

matrix/space.py:

    async def broadcast(
        self,
        content: str | None = None,
        *,
        raw: bool = False,
        notice: bool = False,
        file: File | None = None,
        depth: int = 1
    ) -> list[Message]

Behavior: Calls send() on all child rooms of the given space up to the specified depth (default is 1, which means only broadcast to immediate sub rooms).

---

Fixes: #66

[PenguinBoi12]

In general the PR looks pretty good.

There's one big thing we need to address regarding rate limite and retry on failing send. I've left a comment that talks about it but we can discuss on matrix if needed.

There's also a couple of missing tests, I haven't listed all of them so you should make sure every cases are well tested, similarily, some cases were tested in one broadcast and notntested in the other, it's important to test the same cases for both broadcast.

Make sure the docstring also reflect those tests or special edge case and to remove what's not part of the things you've done.

[PenguinBoi12]

There's a good chance we get rate limited doing this, so we need to think about how to avoid it. Here's what I have in mind. To avoid rate limiting, we first need to implement this (the BroadcastResult return type), so a rate-limited room doesn't blow up the whole broadcast and we have a place to report per-room failures.

Then, broadcast would look something like:

async def broadcast(self, rooms, content=None, *, max_concurrent=8, **kwargs) -> list[BroadcastResult]:
    semaphore = asyncio.Semaphore(max_concurrent)
    rooms = filter(lambda r: not isinstance(r, Space), rooms)

    async def _send(room):
        async with semaphore:
            try:
                msg = await with_retry(lambda: room.send(content, **kwargs))
                return BroadcastResult(room=room, message=msg, error=None)
            except Exception as e:
                return BroadcastResult(room=room, message=None, error=e)

    return await asyncio.gather(*[_send(room) for room in rooms])

It's also a good opportunity to add a new helper in api.py called with_retry, which would let us retry any coroutine on failure:

async def with_retry(coro, *, retries=3):
    ...

I am also debating if it would be worth it to extract the semaphore logic into another bounded_gather helper or something like that:

async def bounded_gather(coros, *, max_concurrent=8):
    semaphore = asyncio.Semaphore(max_concurrent)
    async def _run(coro):
        async with semaphore:
            return await coro
    return await asyncio.gather(*[_run(c) for c in coros])

This is mainly for consistency with the with_retry and to avoid too much repetitions but it's not blocking either if it's directly in the broadcast so maybe we should

Of course, all this would require some tests as well.

---

As a FYI, the with_retry and the semaphore solve 2 different problems. The semaphore keeps us from causing rate limit and with_retr handles it if # it happens anyway (or other error happen)

[PenguinBoi12]

@grivera64 to help you and unblock this PR, I've made the with_retry (#116) and bounded_gather (#117) . All you have to change is the broadcast code to use those two mechanism:

async def broadcast(self, ...) -> list[BroadcastResult]:
    """Broadcasts a message to the specified rooms.

    ...

    """
    rooms = filter(lambda child: not isinstance(child, Space), rooms)

    async def _send(room: Room) -> BroadcastResult:
        try:
            message = await with_retry(lambda: room.send(content, **kwargs))
            return BroadcastResult(room=room, message=message, error=None)
        except Exception as e:
            return BroadcastResult(room=room, message=None, error=e)
    
    return await bounded_gather(
        [with_retry(lambda room=room: room.send(content)) for room in rooms],
        max_concurrent=8,
    )

[PenguinBoi12]

Using asyncio.gather(*async_send) without return_exceptions=True means one failed room.send() (e.g. bot not joined, rate limit, permission error) aborts the whole broadcast and the caller gets an exception with no indication of which rooms succeeded.

So, I think in the case of broadcasting, it would be more relevant to set return_exceptions=Trueand return something that captures both successes and failures per-room rather than letting one failed send blow up everything:

@dataclass
class BroadcastResult:
    room: Room
    message: Message | None
    error: Exception | None

    @property
    def ok(self) -> bool:
        return self.error is None

broadcast() would return a list[BroadcastResult] instead.

[grivera64]

Since broadcast is the entire operation, should we consider a different naming for the individual room.send() results? Like for example, SendResult or SendStatus? e.g. broadcast(...) -> list[SendResult]. We could also stick with the BroadcastResult naming for simplicity.

If we do the SendResult naming, should we also extend it to just a single room.send() method too or would it be better to make this exclusive to space.broadcast()?

[PenguinBoi12]

@grivera64 No I'd keep this scoped to BroadcastResult. send() still just returns a Message or raises, and that's fine for one call. The wrapper only makes sense once you're sending off a bunch of sends and need to keep going even if one fails, which is really just the broadcast case.

[PenguinBoi12]

The doc says that space are silently omitted and the code seems to do just that but there is not test for it.

[PenguinBoi12]

I am not sure if this is possible; we should validate that children of space can return children that the bot isn't in. I also don't see where it does that in the code.

If we don't want that we can remove that part of the comment. If we do, we need a test to validate that and something that actually check it.

But as in the earlier comment, with the changes I propose we might be able to let if fail without too much cost.

We probably still want a test because it might be a recurring edge case.

[PenguinBoi12]

In both broadcast we probably want to do filter duplicate room. I'm not sure if we can just use a set in that situation but it's worth looking into it and afterwards make a test for this.

[PenguinBoi12]

main's history has to be rewritten to adjust a commit's message, no file content changed but your PR's diff may look off until you either click "Update branch" or rebase locally:

git fetch origin
git rebase origin/main
git push --force-with-lease

[PenguinBoi12]

This should be list[Room | Space]

[PenguinBoi12]

Although room.send has a validation for content = None and file = None, we should add another validation here to avoid raising in every room