Disable messages proc

src/applications/schedule_sync/activities/__init__.py

@@ -1,6 +1,7 @@
-from .sync import syncronize
+from .sync import syncronize, syncronize_all
 
 
 __all__ = [
     "syncronize",
+    "syncronize_all",
 ]

src/applications/schedule_sync/activities/sync.py

@@ -6,13 +6,24 @@ from applications.schedule_sync.synchronizer import syncronize as syncronize_int
 
 @activity.defn
 async def syncronize(twitch_id: int):
-    streamer = await StreamerConfigRepository.get_by_twitch_id(twitch_id)
-
-    if streamer.integrations.discord is None:
-        return
-
     try:
+        streamer = await StreamerConfigRepository.get_by_twitch_id(twitch_id)
+
+        if streamer.integrations.discord is None:
+            return
+
         await syncronize_internal(streamer.twitch, streamer.integrations.discord.guild_id)
     except Exception as e:
         activity.logger.error(f"Error during synchronization: {e}")
         raise e
+
+
+@activity.defn
+async def syncronize_all():
+    streamers = await StreamerConfigRepository().all()
+
+    for streamer in streamers:
+        if streamer.integrations.discord is None:
+            continue
+
+        await syncronize(streamer.twitch.id)

src/applications/schedule_sync/workflows/sync.py

@@ -3,8 +3,7 @@ from datetime import timedelta
 from temporalio import workflow
 from temporalio.client import Schedule, ScheduleActionStartWorkflow, ScheduleSpec, ScheduleIntervalSpec
 
-from applications.common.repositories.streamers import StreamerConfigRepository
+from applications.schedule_sync.activities import syncronize_all
-from applications.schedule_sync.activities import syncronize
 from applications.temporal_worker.queues import MAIN_QUEUE
 
 
@@ -18,6 +17,7 @@ class ScheduleSyncWorkflow:
                     cls.run,
                     id="ScheduleSyncWorkflow",
                     task_queue=MAIN_QUEUE,
+                    execution_timeout=timedelta(minutes=1),
                 ),
                 spec=ScheduleSpec(
                     intervals=[ScheduleIntervalSpec(every=timedelta(minutes=5))]
@@ -27,15 +27,8 @@ class ScheduleSyncWorkflow:
 
     @workflow.run
     async def run(self):
-        streamers = await StreamerConfigRepository().all()
+        await workflow.execute_activity(
-
+            syncronize_all,
-        for streamer in streamers:
+            task_queue=MAIN_QUEUE,
-            if streamer.integrations.discord is None:
+            schedule_to_close_timeout=timedelta(minutes=5),
-                continue
+        )
-
-            await workflow.start_activity(
-                syncronize,
-                streamer.twitch.id,
-                task_queue=MAIN_QUEUE,
-                schedule_to_close_timeout=timedelta(minutes=5),
-            )

src/applications/temporal_worker/__main__.py

@@ -42,12 +42,14 @@ async def main():
         ],
         activities=[
             schedule_sync_activities.syncronize,
+            schedule_sync_activities.syncronize_all,
             twitch_activities.on_message_activity,
             twitch_activities.on_stream_state_change_activity,
             twitch_activities.check_streams_states,
             twitch_activities.on_redemption_reward_add_activity,
+            twitch_activities.on_channel_update_activity,
         ],
-        workflow_runner=UnsandboxedWorkflowRunner(),
+        workflow_runner=UnsandboxedWorkflowRunner()
     )
 
     await worker.run()

src/applications/twitch_webhook/activities/__init__.py

@@ -1,5 +1,5 @@
 from .message_proc import on_message_activity
-from .on_state_change import on_stream_state_change_activity
+from .on_state_change import on_stream_state_change_activity, on_channel_update_activity
 from .redemption_reward import on_redemption_reward_add_activity
 from .state_checker import check_streams_states
 
@@ -9,4 +9,5 @@ __all__ = [
     "on_stream_state_change_activity",
     "check_streams_states",
     "on_redemption_reward_add_activity",
+    "on_channel_update_activity",
 ]

src/applications/twitch_webhook/activities/on_state_change.py

@@ -1,12 +1,17 @@
+from datetime import datetime, timezone
+
 from temporalio import activity
-from dataclasses import dataclass
 
-from applications.twitch_webhook.state import State, EventType
+from pydantic import BaseModel
+
+from twitchAPI.helper import first
+
+from applications.twitch_webhook.state import State, EventType, UpdateEvent
 from applications.twitch_webhook.watcher import StateWatcher
+from applications.twitch_webhook.twitch.authorize import authorize
 
 
-@dataclass
+class OnStreamStateChangeActivity(BaseModel):
-class OnStreamStateChangeActivity:
     streamer_id: int
     event_type: EventType
     new_state: State | None = None
@@ -21,3 +26,33 @@ async def on_stream_state_change_activity(
         data.event_type,
         data.new_state,
     )
+
+
+class OnChannelUpdateActivity(BaseModel):
+    event: UpdateEvent
+    event_type: EventType
+
+
+@activity.defn
+async def on_channel_update_activity(
+    data: OnChannelUpdateActivity
+):
+    twitch = await authorize(data.event.broadcaster_user_login)
+
+    stream = await first(twitch.get_streams(
+        user_id=[data.event.broadcaster_user_id])
+    )
+    if stream is None:
+        return
+
+    await on_stream_state_change_activity(
+        OnStreamStateChangeActivity(
+            streamer_id=int(data.event.broadcaster_user_id),
+            event_type=data.event_type,
+            new_state=State(
+                title=data.event.title,
+                category=data.event.category_name,
+                last_live_at=datetime.now(timezone.utc)
+            ),
+        )
+    )

src/applications/twitch_webhook/messages_proc.py

@@ -308,6 +308,8 @@ class MessagesProc:
 
     @classmethod
     async def on_message(cls, received_as: str, event: MessageEvent):
+        return
+
         if event.chatter_user_name in cls.FULL_IGNORED_USER_LOGINS:
             return
 

src/applications/twitch_webhook/workflows/on_channel_update.py

@@ -1,13 +1,10 @@
-from datetime import datetime, timezone, timedelta
+from datetime import timedelta
 
 from temporalio import workflow
 
-from twitchAPI.helper import first
-
-from applications.twitch_webhook.state import UpdateEvent, EventType, State
-from applications.twitch_webhook.twitch.authorize import authorize
-from applications.twitch_webhook.activities.on_state_change import on_stream_state_change_activity, OnStreamStateChangeActivity
 from applications.temporal_worker.queues import MAIN_QUEUE
+from applications.twitch_webhook.activities.on_state_change import OnChannelUpdateActivity, on_channel_update_activity
+from applications.twitch_webhook.state import UpdateEvent, EventType
 
 
 @workflow.defn
@@ -18,23 +15,12 @@ class OnChannelUpdateWorkflow:
         event: UpdateEvent,
         event_type: EventType,
     ):
-        twitch = await authorize(event.broadcaster_user_login)
-
-        stream = await first(twitch.get_streams(user_id=[event.broadcaster_user_id]))
-        if stream is None:
-            return
-
         await workflow.start_activity(
-            on_stream_state_change_activity,
+            on_channel_update_activity,
-            OnStreamStateChangeActivity(
+            OnChannelUpdateActivity(
-                int(event.broadcaster_user_id),
+                event=event,
-                event_type,
+                event_type=event_type,
-                State(
-                    title=event.title,
-                    category=event.category_name,
-                    last_live_at=datetime.now(timezone.utc)
-                ),
             ),
             task_queue=MAIN_QUEUE,
-            schedule_to_close_timeout=timedelta(minutes=1)
+            start_to_close_timeout=timedelta(minutes=1),
         )

src/applications/twitch_webhook/workflows/on_message.py

@@ -15,5 +15,5 @@ class OnMessageWorkflow:
             on_message_activity,
             message,
             task_queue=MAIN_QUEUE,
-            schedule_to_close_timeout=timedelta(minutes=1)
+            schedule_to_close_timeout=timedelta(minutes=5)
         )

src/applications/twitch_webhook/workflows/on_stream_online.py

@@ -10,7 +10,7 @@ from applications.temporal_worker.queues import MAIN_QUEUE
 @workflow.defn
 class OnStreamOnlineWorkflow:
     @workflow.run
-    async def run(self, broadcaster_user_id: str, event_type: EventType):
+    async def run(self, broadcaster_user_id: str | int, event_type: EventType):
         await workflow.start_activity(
             on_stream_state_change_activity,
             OnStreamStateChangeActivity(

src/core/temporal.py

@@ -1,5 +1,10 @@
 from temporalio.client import Client
+from temporalio.contrib.pydantic import pydantic_data_converter
 
 
 async def get_client() -> Client:
-    return await Client.connect("temporal:7233", namespace="default")
+    return await Client.connect(
+        "temporal:7233",
+        namespace="default",
+        data_converter=pydantic_data_converter
+    )