2525 action_result_to_dict ,
2626 channel_cmd_discord ,
2727 channel_events_discord ,
28+ channel_events_discord_guild ,
2829 channel_reply_discord ,
2930 gateway_instance_key_from_token ,
3031 gateway_result_envelope_from_dict ,
3132 normalized_event_from_dict ,
3233 parse_event_envelope ,
34+ registry_key_persona_guilds ,
3335)
3436
3537logger = logging .getLogger (__name__ )
@@ -78,6 +80,7 @@ def __init__(
7880 ) -> None :
7981 self ._token = token
8082 self ._key = gateway_instance_key_from_token (token )
83+ # Legacy broadcast channel kept for direct-socket / dev-fallback reference.
8184 self ._events_ch = channel_events_discord (self ._key )
8285 self ._cmd_ch = channel_cmd_discord (self ._key )
8386 self ._redis_cmd = redis_cmd
@@ -87,25 +90,83 @@ def __init__(
8790 self ._stop = asyncio .Event ()
8891 self ._reader_task : asyncio .Task [None ] | None = None
8992 self ._cb : Callable [[NormalizedEvent ], Awaitable [None ]] | None = None
93+ # Live pubsub handle shared between reader loop and update_guild_channels.
94+ self ._ps : Any | None = None
95+ # Currently active per-guild subscription channels.
96+ self ._subscribed_guild_channels : set [str ] = set ()
9097
9198 @property
9299 def gateway_instance_key (self ) -> str :
93100 return self ._key
94101
95102 @property
96103 def events_channel (self ) -> str :
104+ # Returns legacy broadcast channel for reference; production path uses per-guild channels.
97105 return self ._events_ch
98106
99107 @property
100108 def cmd_channel (self ) -> str :
101109 return self ._cmd_ch
102110
103- async def start_event_consumer (self , on_event : Callable [[NormalizedEvent ], Awaitable [None ]]) -> None :
111+ async def start_event_consumer (
112+ self ,
113+ on_event : Callable [[NormalizedEvent ], Awaitable [None ]],
114+ guild_ids : set [int ] | None = None ,
115+ ) -> None :
104116 self ._cb = on_event
105117 self ._stop .clear ()
106118 if self ._redis_pubsub is None :
107119 self ._redis_pubsub = redis_pubsub_client_from_env ()
108- self ._reader_task = asyncio .create_task (self ._event_reader_loop ())
120+ initial_channels = {
121+ channel_events_discord_guild (self ._key , str (g ))
122+ for g in (guild_ids or set ())
123+ }
124+ self ._subscribed_guild_channels = initial_channels
125+ self ._reader_task = asyncio .create_task (self ._event_reader_loop (initial_channels ))
126+
127+ async def update_guild_channels (self , guild_ids : set [int ]) -> None :
128+ """Subscribe to new per-guild channels and drop channels no longer in the allowed set."""
129+ new_channels = {channel_events_discord_guild (self ._key , str (g )) for g in guild_ids }
130+ to_add = new_channels - self ._subscribed_guild_channels
131+ to_remove = self ._subscribed_guild_channels - new_channels
132+ ps = self ._ps
133+ if ps is None :
134+ # Reader loop not yet running; track for when it starts.
135+ self ._subscribed_guild_channels = new_channels
136+ return
137+ try :
138+ if to_add :
139+ await ps .subscribe (* to_add )
140+ if to_remove :
141+ await ps .unsubscribe (* to_remove )
142+ except (RedisConnectionError , OSError , RuntimeError ) as e :
143+ logger .warning ("update_guild_channels: %s %s" , type (e ).__name__ , e )
144+ self ._subscribed_guild_channels = new_channels
145+
146+ async def register_persona_guilds (self , persona_id : str , guild_ids : set [int ]) -> None :
147+ """Write persona->guild Set in Redis with a 300s TTL; called on connect and refresh."""
148+ if self ._redis_cmd is None :
149+ self ._redis_cmd = redis_client_from_env ()
150+ reg_key = registry_key_persona_guilds (self ._key , persona_id )
151+ try :
152+ pipe = self ._redis_cmd .pipeline ()
153+ pipe .delete (reg_key )
154+ if guild_ids :
155+ pipe .sadd (reg_key , * [str (g ) for g in guild_ids ])
156+ pipe .expire (reg_key , 300 )
157+ await pipe .execute ()
158+ except (RedisConnectionError , RedisTimeoutError , OSError , RuntimeError ) as e :
159+ logger .warning ("register_persona_guilds: %s %s" , type (e ).__name__ , e )
160+
161+ async def unregister_persona_guilds (self , persona_id : str ) -> None :
162+ """Remove persona->guild Set from Redis on clean disconnect."""
163+ if self ._redis_cmd is None :
164+ return
165+ reg_key = registry_key_persona_guilds (self ._key , persona_id )
166+ try :
167+ await self ._redis_cmd .delete (reg_key )
168+ except (RedisConnectionError , RedisTimeoutError , OSError , RuntimeError ) as e :
169+ logger .warning ("unregister_persona_guilds: %s %s" , type (e ).__name__ , e )
109170
110171 async def stop_event_consumer (self ) -> None :
111172 self ._stop .set ()
@@ -127,12 +188,14 @@ async def close(self) -> None:
127188 await self ._redis_cmd .close ()
128189 self ._redis_cmd = None
129190
130- async def _event_reader_loop (self ) -> None :
191+ async def _event_reader_loop (self , initial_channels : set [ str ] ) -> None :
131192 r = self ._redis_pubsub
132193 if r is None :
133194 return
134195 ps = r .pubsub ()
135- await ps .subscribe (self ._events_ch )
196+ self ._ps = ps
197+ if initial_channels :
198+ await ps .subscribe (* initial_channels )
136199 try :
137200 while not self ._stop .is_set () and not ckit_shutdown .shutdown_event .is_set ():
138201 try :
@@ -161,8 +224,8 @@ async def _event_reader_loop(self) -> None:
161224 except asyncio .CancelledError :
162225 raise
163226 finally :
227+ self ._ps = None
164228 try :
165- await ps .unsubscribe (self ._events_ch )
166229 await ps .close ()
167230 except (RedisConnectionError , OSError , RuntimeError ):
168231 pass
0 commit comments