wip

Author: luke-beamcloud

examples/slackdemo.py

@@ -4,11 +4,12 @@
 from fastloop import FastLoop, LoopContext
 from fastloop.integrations.slack import (
     SlackAppMentionEvent,
+    SlackConfig,
     SlackFileSharedEvent,
     SlackIntegration,
     SlackMessageEvent,
+    SlackSetupInput,
 )
-from fastloop.models import LoopEvent
 
 app = FastLoop(name="slackdemo")
 
@@ -17,18 +18,26 @@ class AppContext(LoopContext):
     client: Any
 
 
-async def resolve_bot_token(context: LoopContext, event: LoopEvent) -> str:
+async def resolve_slack_config(setup_input: SlackSetupInput) -> SlackConfig:
     """
-    Resolve the bot token for this workspace.
+    Resolve the Slack config for this workspace.
 
-    In a real app, you would look this up from a database based on the team_id
-    that was stored when the workspace installed your Slack app via OAuth.
+    In a real app, you would look up the bot token and signing secret from a database
+    based on the team_id that was stored when the workspace installed your Slack app via OAuth.
 
     Example:
-        team_id = getattr(event, 'team', None)
-        return await db.get_bot_token(team_id)
+        workspace = await db.get_workspace(setup_input.team_id)
+        return SlackConfig(
+            bot_token=workspace.bot_token,
+            signing_secret=workspace.signing_secret,
+            team_id=setup_input.team_id,
+        )
     """
-    return os.getenv("SLACK_BOT_TOKEN") or ""
+    return SlackConfig(
+        bot_token=os.getenv("SLACK_BOT_TOKEN") or "",
+        signing_secret=os.getenv("SLACK_SIGNING_SECRET") or "",
+        team_id=setup_input.team_id,
+    )
 
 
 async def analyze_file(context: AppContext):
@@ -45,12 +54,7 @@ async def analyze_file(context: AppContext):
 
 @app.loop(
     "filebot",
-    integrations=[
-        SlackIntegration(
-            signing_secret=os.getenv("SLACK_SIGNING_SECRET"),
-            setup=resolve_bot_token,
-        )
-    ],
+    integrations=[SlackIntegration(setup=resolve_slack_config)],
 )
 async def test_slack_bot(context: AppContext):
     mention: SlackAppMentionEvent | None = await context.wait_for(

fastloop/context.py

@@ -115,10 +115,13 @@ async def sleep_until(self, timestamp: float) -> None:
 
     async def wait_for(
         self,
-        event: type[E],
+        *event_types: type[E],
         timeout: float | int = 10.0,
         raise_on_timeout: bool = True,
     ) -> E | None:
+        if not event_types:
+            raise ValueError("At least one event type must be provided")
+
         wait_for_start = time.monotonic()
         start = asyncio.get_event_loop().time()
         pubsub = await self.state_manager.subscribe_to_events(self.loop_id)
@@ -139,14 +142,15 @@ async def wait_for(
                 if self.should_stop:
                     raise LoopStoppedError()
 
-                event_result = await self.state_manager.pop_event(
-                    self.loop_id,
-                    event,  # type: ignore
-                    sender=LoopEventSender.CLIENT,
-                )
-                if event_result is not None:
-                    self.event_this_cycle = True
-                    return cast(E, event_result)  # noqa
+                for event_type in event_types:
+                    event_result = await self.state_manager.pop_event(
+                        self.loop_id,
+                        event_type,  # type: ignore
+                        sender=LoopEventSender.CLIENT,
+                    )
+                    if event_result is not None:
+                        self.event_this_cycle = True
+                        return cast(E, event_result)  # noqa
 
                 remaining_timeout = timeout - (asyncio.get_event_loop().time() - start)
                 if remaining_timeout <= 0:
@@ -166,7 +170,8 @@ async def wait_for(
                 await pubsub.close()  # type: ignore
 
         if raise_on_timeout:
-            raise EventTimeoutError(f"Timeout waiting for event {event.type}")
+            type_names = ", ".join(et.__name__ for et in event_types)
+            raise EventTimeoutError(f"Timeout waiting for event(s): {type_names}")
         else:
             return None
 

fastloop/integrations/slack.py

@@ -1,5 +1,6 @@
 import re
 from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, cast
 
@@ -16,7 +17,26 @@
     from ..context import LoopContext
     from ..fastloop import FastLoop
 
-SlackSetupCallback = Callable[["LoopContext", "LoopEvent"], Awaitable[str]]
+
+@dataclass
+class SlackConfig:
+    bot_token: str
+    signing_secret: str
+    team_id: str | None = None
+
+
+@dataclass
+class SlackSetupInput:
+    loop_name: str
+    team_id: str
+    channel: str
+    root_ts: str
+    event_type: str
+    payload: dict[str, Any] = field(default_factory=dict)
+    event: dict[str, Any] = field(default_factory=dict)
+
+
+SlackSetupCallback = Callable[["SlackSetupInput"], Awaitable["SlackConfig"]]
 
 IGNORED_MESSAGE_SUBTYPES = frozenset(
     [
@@ -224,21 +244,44 @@ class SlackFileUploadEvent(LoopEvent):
 
 
 class SlackIntegration(Integration):
-    def __init__(self, *, signing_secret: str, setup: SlackSetupCallback):
+    def __init__(self, *, setup: SlackSetupCallback):
         super().__init__()
         self._setup_callback = setup
-        self._signing_secret = signing_secret
-        self.verifier = SignatureVerifier(signing_secret)
+        self._config_cache: dict[str, SlackConfig] = {}
 
     def type(self) -> IntegrationType:
         return IntegrationType.SLACK
 
+    async def _resolve_config(self, setup_input: SlackSetupInput) -> SlackConfig:
+        team_id = setup_input.team_id
+        if team_id in self._config_cache:
+            return self._config_cache[team_id]
+        config = await self._setup_callback(setup_input)
+        self._config_cache[team_id] = config
+        return config
+
     async def setup_for_context(
         self, context: "LoopContext", event: "LoopEvent"
     ) -> AsyncWebClient:
-        bot_token = await self._setup_callback(context, event)
-        client = AsyncWebClient(token=bot_token)
+        team_id = getattr(event, "team", "")
+        channel = getattr(event, "channel", "")
+        root_ts = getattr(event, "root_ts", None) or getattr(event, "ts", "")
+        event_type = getattr(event, "type", "")
+        event_dict = getattr(event, "raw_event", None) or {}
+
+        setup_input = SlackSetupInput(
+            loop_name=self.loop_name,
+            team_id=team_id,
+            channel=channel,
+            root_ts=root_ts,
+            event_type=event_type,
+            payload={},
+            event=event_dict,
+        )
+        config = await self._resolve_config(setup_input)
+        client = AsyncWebClient(token=config.bot_token)
         context.set_integration_client(self.type(), client)
+        context.set_integration_client(f"{self.type()}_config", config)
         return client
 
     def get_client_for_context(self, context: "LoopContext") -> AsyncWebClient:
@@ -250,6 +293,12 @@ def get_client_for_context(self, context: "LoopContext") -> AsyncWebClient:
             )
         return cast("AsyncWebClient", client)
 
+    def get_config_for_context(self, context: "LoopContext") -> SlackConfig:
+        config = context.get_integration_client(f"{self.type()}_config")
+        if config is None:
+            raise ValueError("Slack config not initialized for this context.")
+        return cast("SlackConfig", config)
+
     def register(self, fastloop: "FastLoop", loop_name: str) -> None:
         fastloop.register_events(SLACK_EVENT_TYPES)
         self._fastloop: FastLoop = fastloop
@@ -265,18 +314,69 @@ def events(self) -> list[Any]:
         return list(SLACK_EVENT_TYPES)
 
     async def _handle_slack_event(self, request: Request):
+        from ..logging import setup_logger
+
+        logger = setup_logger(__name__)
+
         body = await request.body()
-        if not self.verifier.is_valid_request(body, dict(request.headers)):
-            raise HTTPException(
-                status_code=HTTPStatus.FORBIDDEN, detail="Invalid signature"
-            )
 
-        payload = await request.json()
+        logger.debug(
+            "Slack webhook received",
+            extra={"body_length": len(body), "path": str(request.url)},
+        )
+
+        try:
+            payload = await request.json()
+        except Exception as e:
+            logger.error("Failed to parse Slack payload", extra={"error": str(e)})
+            raise
+
+        logger.debug(
+            "Parsed Slack payload",
+            extra={
+                "type": payload.get("type"),
+                "event_type": payload.get("event", {}).get("type"),
+            },
+        )
+
         if payload.get("type") == "url_verification":
             return {"challenge": payload["challenge"]}
 
+        team_id = payload.get("team_id", "")
         event: dict[str, Any] = payload.get("event", {})
-        event_type = event.get("type")
+        event_type = event.get("type", "")
+        channel = event.get("channel", "")
+        root_ts = event.get("thread_ts") or event.get("ts", "")
+
+        logger.info(
+            "Received Slack event",
+            extra={
+                "event_type": event_type,
+                "channel": channel,
+                "root_ts": root_ts,
+                "thread_ts": event.get("thread_ts"),
+                "ts": event.get("ts"),
+                "subtype": event.get("subtype"),
+                "has_bot_id": bool(event.get("bot_id")),
+            },
+        )
+
+        setup_input = SlackSetupInput(
+            loop_name=self.loop_name,
+            team_id=team_id,
+            channel=channel,
+            root_ts=root_ts,
+            event_type=event_type,
+            payload=payload,
+            event=event,
+        )
+        config = await self._resolve_config(setup_input)
+        verifier = SignatureVerifier(config.signing_secret)
+
+        if not verifier.is_valid_request(body, dict(request.headers)):
+            raise HTTPException(
+                status_code=HTTPStatus.FORBIDDEN, detail="Invalid signature"
+            )
 
         if event_type not in SUPPORTED_SLACK_EVENTS:
             return {"ok": True}
@@ -288,18 +388,29 @@ async def _handle_slack_event(self, request: Request):
 
         handler = self._fastloop.loop_event_handlers.get(self.loop_name)
         if not handler:
+            logger.warning("No handler found", extra={"loop_name": self.loop_name})
             return {"ok": True}
 
-        channel = event.get("channel", "")
-        root_ts = event.get("thread_ts") or event.get("ts", "")
-        loop_id = await self._fastloop.state_manager.get_loop_mapping(
-            f"slack_thread:{channel}:{root_ts}"
+        mapping_key = f"slack_thread:{channel}:{root_ts}"
+        loop_id = await self._fastloop.state_manager.get_loop_mapping(mapping_key)
+
+        logger.info(
+            "Loop mapping lookup",
+            extra={"mapping_key": mapping_key, "loop_id": loop_id},
         )
 
         loop_event = self._map_event(event, event_type, payload, loop_id)
         if loop_event is None:
+            logger.warning(
+                "Event could not be mapped", extra={"event_type": event_type}
+            )
             return {"ok": True}
 
+        logger.info(
+            "Dispatching event to handler",
+            extra={"event_type": loop_event.type, "loop_id": loop_id},
+        )
+
         loop: LoopState = await handler(loop_event.to_dict())
         if loop.loop_id:
             await self._fastloop.state_manager.set_loop_mapping(

fastloop/state/state_redis.py

@@ -604,12 +604,13 @@ async def push_event(self, loop_id: str, event: "LoopEvent"):
             pipe.lpush(queue_key, event_str)
             pipe.lpush(history_key, event_str)
             pipe.ltrim(history_key, 0, MAX_EVENT_HISTORY - 1)
-
-            if event.sender == LoopEventSender.SERVER:
-                pipe.publish(channel_key, "new_event")  # type: ignore
+            pipe.publish(channel_key, "new_event")  # type: ignore
 
             await pipe.execute()
 
+        if event.sender == LoopEventSender.CLIENT:
+            self.wake_queue.put_nowait(loop_id)
+
     async def get_context_value(self, loop_id: str, key: str) -> Any:
         value_str = await self.rdb.get(
             RedisKeys.LOOP_CONTEXT.format(

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastloop"
-version = "0.1.107"
+version = "0.1.113"
 description = "A Python package for deploying stateful loops"
 readme = "README.md"
 requires-python = ">=3.12"