From 8a060eaea7d0aa5e70a756b17d392d9d3e00e1b9 Mon Sep 17 00:00:00 2001 From: PianoNic <79938743+Pianonic@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:48:05 +0100 Subject: [PATCH] Added a live chat stream for the open chat page --- .claude/CLAUDE.md | 5 +- requirements.txt | 2 +- src/api/controllers/chats_controller.py | 6 + src/api/extensions/chats_extensions.py | 5 + src/application/abstractions/i_chat_live.py | 12 ++ .../abstractions/i_graph_client.py | 3 + .../queries/open_chat_live_query.py | 31 +++ src/infrastructure/teams/chat_live.py | 104 ++++++++++ tests/test_chat_live.py | 193 ++++++++++++++++++ 9 files changed, 358 insertions(+), 3 deletions(-) create mode 100644 src/application/abstractions/i_chat_live.py create mode 100644 src/application/queries/open_chat_live_query.py create mode 100644 src/infrastructure/teams/chat_live.py create mode 100644 tests/test_chat_live.py diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index a2272d7..a4e3823 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -6,7 +6,7 @@ The Schuly **Microsoft plugin** is the only caller. ## What it is / isn't -- **Stateless** apart from the short-lived in-memory password sign-in jobs. No database, no stored tokens. The caller owns the FOCI refresh token and +- **Stateless** apart from the short-lived in-memory password sign-in jobs and the live chat streams, which last only as long as their request. No database, no stored tokens. The caller owns the FOCI refresh token and sends it back in the `X-MS-Refresh-Token` header; reads return the rotated token in the same response header. - Sign-in is the **browser sign-in** (authorization code + PKCE in a WebView, used by the @@ -44,6 +44,7 @@ The Schuly **Microsoft plugin** is the only caller. - `AccountRunner` strips `_native` from every result. No route calls OutNote's native reads today: chat messages stay on Graph's list because `ChatReader` and the callers need its attachments, mentions and reactions. `emails()` gets an explicit `$select` because the web's list shows `toRecipients`. +- `GET /api/chats/live` (`ChatLive`) runs `note.listen(reconnect=False)` in a daemon thread for as long as the caller stays, so one thread and one Teams connection per open chat page, capped per process and per account. `listen()` stops only on `close()`, so the stream's `finally` must reach it. It forwards only `{kind, chatId, fromMe}` for messages, edits and deletions: no content, typing or presence. The route is undocumented and has no Graph fallback; the web app polls while it is down. - Never add `SQLiteCache` or any other on-disk cache. `stream()` cursors would have to live in the plugin. ## Cold start @@ -109,7 +110,7 @@ src/ logging_config.py # logger names microsoft/ # AccountRunner, OutNoteClientFactory, BrokerCache, SharedBroker, AccountCache, MicrosoftHttpClients, DeviceLoginClient, BrowserSignInClient, ProfileReader graph/ # TeamsFiles, GraphSearch, OneNoteClient, OneNoteRunner, PdfConverter, FileThumbnails, ChannelMessages, PeoplePhotos, MailFolderRoles - teams/ # TeamsChatClient (skype token, AMS), ChatReader, ChatSender, ChatPresence, EmojiCatalogue + teams/ # TeamsChatClient (skype token, AMS), ChatReader, ChatSender, ChatLive, ChatPresence, EmojiCatalogue giphy/ # GifSearch (Teams' own GIF command with the account's sign-in; 503 when Teams refuses) files/ # FileTextExtractor (lazy imports, zip-bomb guard, caps, paging) password_signin/ # JobStore, SignInJob (worker thread per job, MfaPrompt bridge), RateLimiter diff --git a/requirements.txt b/requirements.txt index 7622dae..b40c92b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,7 +2,7 @@ fastapi==0.122.0 uvicorn==0.38.0 httpx==0.28.1 python-multipart==0.0.20 -outnote==2.2.0 +outnote==2.3.0 ms-entrance==2.0.0 wreq==0.13.0 python-docx==1.2.0 diff --git a/src/api/controllers/chats_controller.py b/src/api/controllers/chats_controller.py index 026d891..fd3dd76 100644 --- a/src/api/controllers/chats_controller.py +++ b/src/api/controllers/chats_controller.py @@ -19,6 +19,7 @@ from src.application.queries.get_chat_presence_query import GetChatPresenceQuery from src.application.queries.list_chat_messages_query import ListChatMessagesQuery from src.application.queries.list_chats_query import ListChatsQuery +from src.application.queries.open_chat_live_query import OpenChatLiveQuery router = APIRouter(prefix="/api") @@ -73,6 +74,11 @@ async def chat_presence(self, chat_id: str, token: str = Header(..., alias=HttpH """The Teams status of the other person in a 1:1 chat: `{availability, activity}`, both null when there is none to show.""" return await self.mediator.send(GetChatPresenceQuery(token, chat_id)) + @router.get("/chats/live", tags=["Microsoft 365"], response_class=StreamingResponse, responses={200: {"content": {"application/x-ndjson": {}}}}) + async def chats_live(self, token: str = Header(..., alias=HttpHeaders.REFRESH_TOKEN_HEADER)): + """Live chat events, one JSON object per line, for as long as the caller stays: `connected`, `message`, `message_updated` and `message_deleted` with `chatId` and `fromMe`, and `ping` to keep the connection up. It says what changed where and nothing else. When it ends, reconnect.""" + return StreamingResponse(await self.mediator.send(OpenChatLiveQuery(token)), media_type="application/x-ndjson", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}) + @router.get("/chats/{chat_id}/photo", tags=["Microsoft 365"], response_class=Response, responses={200: {"content": {"image/jpeg": {}}}}) async def chat_photo(self, chat_id: str, token: str = Header(..., alias=HttpHeaders.REFRESH_TOKEN_HEADER)): return FileResponses.image(await self.mediator.send(GetChatPhotoQuery(token, chat_id))) diff --git a/src/api/extensions/chats_extensions.py b/src/api/extensions/chats_extensions.py index e659e22..694bd8a 100644 --- a/src/api/extensions/chats_extensions.py +++ b/src/api/extensions/chats_extensions.py @@ -1,5 +1,6 @@ from src.api.di.service_collection import ServiceCollection from src.application.abstractions.i_chat_presence import IChatPresence +from src.application.abstractions.i_chat_live import IChatLive from src.application.abstractions.i_chat_reader import IChatReader from src.application.abstractions.i_chat_sender import IChatSender from src.application.abstractions.i_teams_chat import ITeamsChat @@ -14,8 +15,10 @@ from src.application.queries.list_chat_messages_query import ListChatMessagesQuery, ListChatMessagesHandler from src.application.queries.get_chat_presence_query import GetChatPresenceQuery, GetChatPresenceHandler from src.application.queries.list_chats_query import ListChatsQuery, ListChatsHandler +from src.application.queries.open_chat_live_query import OpenChatLiveQuery, OpenChatLiveHandler from src.application.services.chat_html_sanitizer import ChatHtmlSanitizer from src.infrastructure.teams.chat_presence import ChatPresence +from src.infrastructure.teams.chat_live import ChatLive from src.infrastructure.teams.chat_reader import ChatReader from src.infrastructure.teams.chat_sender import ChatSender from src.infrastructure.teams.teams_chat_client import TeamsChatClient @@ -27,6 +30,7 @@ def add_chats(services: ServiceCollection) -> ServiceCollection: services.add_singleton(ChatHtmlSanitizer) services.add_singleton(ITeamsChat, TeamsChatClient) services.add_singleton(IChatReader, ChatReader) + services.add_singleton(IChatLive, ChatLive) services.add_singleton(IChatPresence, ChatPresence) services.add_singleton(IChatSender, ChatSender) @@ -40,5 +44,6 @@ def add_chats(services: ServiceCollection) -> ServiceCollection: services.add_handler(ScheduleChatMeetingCommand, ScheduleChatMeetingHandler) services.add_handler(MarkChatReadCommand, MarkChatReadHandler) services.add_handler(GetChatPhotoQuery, GetChatPhotoHandler) + services.add_handler(OpenChatLiveQuery, OpenChatLiveHandler) services.add_handler(GetChatPresenceQuery, GetChatPresenceHandler) return services diff --git a/src/application/abstractions/i_chat_live.py b/src/application/abstractions/i_chat_live.py new file mode 100644 index 0000000..c240d37 --- /dev/null +++ b/src/application/abstractions/i_chat_live.py @@ -0,0 +1,12 @@ +from abc import ABC, abstractmethod +from collections.abc import AsyncIterator + +from src.application.abstractions.i_account_broker import IAccountBroker +from src.application.abstractions.i_graph_client import IGraphClient + + +class IChatLive(ABC): + """Live chat events as NDJSON lines, for as long as the caller reads them.""" + + @abstractmethod + def open(self, note: IGraphClient, broker: IAccountBroker) -> AsyncIterator[bytes]: ... diff --git a/src/application/abstractions/i_graph_client.py b/src/application/abstractions/i_graph_client.py index e1c4a68..b7b5752 100644 --- a/src/application/abstractions/i_graph_client.py +++ b/src/application/abstractions/i_graph_client.py @@ -13,6 +13,9 @@ def list(self, path, params=None, limit=None, client=None): ... @abstractmethod def post(self, path, body=None, client=None): ... + @abstractmethod + def listen(self, reconnect=True, presence=()): ... + @abstractmethod def presences(self, user_ids): ... diff --git a/src/application/queries/open_chat_live_query.py b/src/application/queries/open_chat_live_query.py new file mode 100644 index 0000000..963cab6 --- /dev/null +++ b/src/application/queries/open_chat_live_query.py @@ -0,0 +1,31 @@ +from collections.abc import AsyncIterator +from dataclasses import dataclass, field + +from mediatorx import IQuery, IQueryHandler + +from src.application.abstractions.i_account_runner import IAccountRunner +from src.application.abstractions.i_chat_live import IChatLive +from src.application.options.microsoft_options import MicrosoftOptions +from src.domain.validation.refresh_token_format import RefreshTokenFormat + + +@dataclass +class OpenChatLiveQuery(IQuery[AsyncIterator[bytes]]): + refresh_token: str = field(repr=False) + + +class OpenChatLiveHandler(IQueryHandler[OpenChatLiveQuery, AsyncIterator[bytes]]): + def __init__(self, accounts: IAccountRunner, live: IChatLive, options: MicrosoftOptions): + self._accounts = accounts + self._live = live + self._options = options + + async def handle(self, query: OpenChatLiveQuery) -> AsyncIterator[bytes]: + RefreshTokenFormat.check(query.refresh_token) + + def run(note, broker): + # A dead sign-in is refused here, as a real status, before the stream starts. + broker(self._options.device_client_id) + return self._live.open(note, broker) + + return await self._accounts.run(query.refresh_token, run) diff --git a/src/infrastructure/teams/chat_live.py b/src/infrastructure/teams/chat_live.py new file mode 100644 index 0000000..fadc185 --- /dev/null +++ b/src/infrastructure/teams/chat_live.py @@ -0,0 +1,104 @@ +"""Live chat events for one open Schuly chat page, over OutNote's `listen()` (Teams' own push service, undocumented). + +The feed lives exactly as long as the request: no listener outlives its caller, and nothing is stored. It says only that something +changed in a chat, never what: the caller reads the messages the normal way. `reconnect=False` on purpose, so a broken feed ends the +stream and the caller sees it, falls back to polling and reconnects, rather than a stream that looks up and delivers nothing. +""" + +import asyncio +import json +import logging +import threading +from collections import Counter +from collections.abc import AsyncIterator + +from src.application.abstractions.i_chat_live import IChatLive +from src.domain.errors.service_error import ServiceError +from src.domain.validation.graph_ids import GraphIds +from src.infrastructure.logging_config import LoggerNames + + +class ChatLive(IChatLive): + """One thread and one Teams connection per open stream, capped because this service is one process.""" + + log = logging.getLogger(LoggerNames.SERVICE) + HEARTBEAT = 20.0 # seconds; keeps proxies from closing a quiet stream + MAX_STREAMS = 200 + MAX_PER_ACCOUNT = 3 # a few tabs; each one is its own connection to Teams + PING = b'{"kind":"ping"}\n' + CHANGES = frozenset({"message", "message_updated", "message_deleted"}) + + def __init__(self): + self._guard = threading.Lock() + self._streams = Counter() + + def _has_room(self, broker) -> bool: + return self._streams.total() < self.MAX_STREAMS and self._streams[broker] < self.MAX_PER_ACCOUNT + + def _enter(self, broker) -> bool: + with self._guard: + if not self._has_room(broker): + return False + self._streams[broker] += 1 + return True + + def _leave(self, broker) -> None: + with self._guard: + self._streams[broker] -= 1 + if self._streams[broker] <= 0: + del self._streams[broker] + + def line(self, event) -> bytes | None: + """What the caller may see of an event, or None: everything else (typing, presence, read receipts) stays here.""" + if event.kind == "connected": + return b'{"kind":"connected"}\n' + if event.kind not in self.CHANGES or not isinstance(event.chat_id, str): + return None + if not GraphIds.CHAT_ID_PATTERN.fullmatch(event.chat_id) or GraphIds.has_dot_segment(event.chat_id): + return None + return (json.dumps({"kind": event.kind, "chatId": event.chat_id, "fromMe": bool(event.from_me)}) + "\n").encode() + + def open(self, note, broker) -> AsyncIterator[bytes]: + with self._guard: + if not self._has_room(broker): + raise ServiceError(503, "Too many live chat connections.", headers={"Retry-After": "60"}) + return self._stream(note.listen(reconnect=False), broker) + + def _pump(self, live, loop, queue) -> None: + def put(item) -> None: + try: + loop.call_soon_threadsafe(queue.put_nowait, item) + except RuntimeError: + live.close() # the loop is gone, and so is the caller + + try: + for event in live: + line = self.line(event) + if line: + put(line) + except Exception as ex: # network, auth or protocol: the caller reconnects + self.log.warning("Live chat feed ended: %s", type(ex).__name__) + finally: + put(None) + + async def _stream(self, live, broker) -> AsyncIterator[bytes]: + # The slot is taken here and not in open(): a generator that never starts never reaches a finally. + if not self._enter(broker): + live.close() + return + try: + queue = asyncio.Queue() + threading.Thread(target=self._pump, args=(live, asyncio.get_running_loop(), queue), daemon=True, name="chat-live").start() + yield self.PING + while True: + try: + line = await asyncio.wait_for(queue.get(), self.HEARTBEAT) + except TimeoutError: + yield self.PING + continue + if line is None: + return + yield line + finally: + live.close() + self._leave(broker) diff --git a/tests/test_chat_live.py b/tests/test_chat_live.py new file mode 100644 index 0000000..29a85af --- /dev/null +++ b/tests/test_chat_live.py @@ -0,0 +1,193 @@ +import asyncio +import json +import socket +import threading +import time +from types import SimpleNamespace + +import httpx +import pytest +import uvicorn +from fastapi.testclient import TestClient + +from src.api.app import app +from src.application.abstractions.i_chat_live import IChatLive +from src.domain.errors.service_error import ServiceError +from tests.services import config, outnote_client, services + +live_chats = services.get(IChatLive) +H = {config.REFRESH_TOKEN_HEADER: "frt-1"} +CHAT = "19:g@thread.v2" + + +def event(kind, chat_id=CHAT, from_me=False, **rest): + return SimpleNamespace(kind=kind, chat_id=chat_id, from_me=from_me, sender_name="Bo", message={"body": "secret"}, raw={"x": 1}, **rest) + + +class Feed: + """Like OutNote's LiveConnection: iterating blocks until close(), and only close() stops it.""" + + def __init__(self, events=(), end=False, fails=None): + self.events, self.end, self.fails = list(events), end, fails + self.closed = threading.Event() + + def close(self): + self.closed.set() + + def __iter__(self): + yield from self.events + if self.fails: + raise self.fails + if not self.end: + self.closed.wait(30) + + +class Broker: + refresh_token = "frt-1" + + def __call__(self, *args): + return "token" + + +class Note: + def __init__(self, feed): + self.feed = feed + self.reconnect = None + + def listen(self, reconnect=True, presence=()): + self.reconnect = reconnect + return self.feed + + +@pytest.fixture +def client(monkeypatch): + def build(feed): + note = Note(feed) + broker = Broker() + monkeypatch.setattr(outnote_client, "build_client", lambda frt: (note, broker)) + return note, broker + + return build + + +async def _read(stream): + return [json.loads(line) async for line in stream] + + +def test_a_stream_says_only_what_changed_where(client): + feed = Feed([event("connected", None), event("message", from_me=True), event("message_updated"), event("message_deleted", "48:notes"), event("typing"), event("stop_typing"), event("presence"), event("other"), event("message", "../../me"), event("message", None), event("message", "bad id")], end=True) + note, _ = client(feed) + r = TestClient(app).get("/api/chats/live", headers=H) + assert r.status_code == 200 and r.headers["content-type"].startswith("application/x-ndjson") and r.headers["cache-control"] == "no-store" + assert [json.loads(line) for line in r.text.splitlines()] == [ + {"kind": "ping"}, + {"kind": "connected"}, + {"kind": "message", "chatId": CHAT, "fromMe": True}, + {"kind": "message_updated", "chatId": CHAT, "fromMe": False}, + {"kind": "message_deleted", "chatId": "48:notes", "fromMe": False}, + ] + assert note.reconnect is False # a broken feed must end the stream, so the caller knows + assert feed.closed.is_set() + + +def test_a_broken_feed_ends_the_stream_without_its_error(client): + feed = Feed([event("connected", None)], fails=OSError("token abc in this message")) + client(feed) + r = TestClient(app).get("/api/chats/live", headers=H) + assert r.status_code == 200 and "abc" not in r.text and feed.closed.is_set() + + +def test_a_bad_token_is_refused_before_the_stream(client): + client(Feed()) + assert TestClient(app).get("/api/chats/live", headers={config.REFRESH_TOKEN_HEADER: "has spaces in it"}).status_code == 401 + assert TestClient(app).get("/api/chats/live").status_code == 422 + + +def test_quiet_streams_send_heartbeats(monkeypatch): + monkeypatch.setattr(live_chats, "HEARTBEAT", 0.01) + + async def run(): + feed = Feed() + stream = live_chats.open(Note(feed), Broker()) + lines = [await stream.__anext__() for _ in range(3)] + await stream.aclose() + return lines, feed + + lines, feed = asyncio.run(run()) + assert lines == [live_chats.PING] * 3 and feed.closed.is_set() + + +def test_streams_are_capped_per_account_and_a_closed_one_frees_its_place(): + async def run(): + broker = Broker() + streams = [live_chats.open(Note(Feed()), broker) for _ in range(live_chats.MAX_PER_ACCOUNT)] + for stream in streams: + await stream.__anext__() + with pytest.raises(ServiceError) as refused: + live_chats.open(Note(Feed()), broker) + other = live_chats.open(Note(Feed()), Broker()) + await other.__anext__() + await other.aclose() + await streams[0].aclose() + again = live_chats.open(Note(Feed()), broker) + await again.__anext__() + for stream in [*streams[1:], again]: + await stream.aclose() + return refused.value + + refused = asyncio.run(run()) + assert refused.status == 503 and refused.headers["Retry-After"] == "60" + assert not live_chats._streams + + +def test_streams_are_capped_per_process(monkeypatch): + monkeypatch.setattr(live_chats, "MAX_STREAMS", 1) + + async def run(): + first = live_chats.open(Note(Feed()), Broker()) + await first.__anext__() + with pytest.raises(ServiceError): + live_chats.open(Note(Feed()), Broker()) + await first.aclose() + + asyncio.run(run()) + + +def test_a_stream_opened_but_never_read_holds_no_place(): + feed = Feed() + + async def run(): + live_chats.open(Note(feed), Broker()) + + asyncio.run(run()) + assert not live_chats._streams + + +def test_the_caller_leaving_closes_the_feed(client, monkeypatch): + # A real server and a real disconnect: `listen()` stops only on close(), so this is what stops a Teams connection leaking. + monkeypatch.setattr(live_chats, "HEARTBEAT", 0.2) + feed = Feed([event("connected", None)]) + client(feed) + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning", lifespan="off")) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + try: + deadline = time.monotonic() + 10 + while not server.started and time.monotonic() < deadline: + time.sleep(0.02) + with httpx.stream("GET", f"http://127.0.0.1:{port}/api/chats/live", headers=H, timeout=10) as response: + lines = response.iter_lines() + assert json.loads(next(lines)) == {"kind": "ping"} + assert json.loads(next(lines)) == {"kind": "connected"} + assert not feed.closed.is_set() + assert feed.closed.wait(10), "the feed stayed open after the caller left" + finally: + server.should_exit = True + thread.join(10) + deadline = time.monotonic() + 5 + while live_chats._streams and time.monotonic() < deadline: + time.sleep(0.02) + assert not live_chats._streams