Implement atomic locking in API and cron steps for per-employee parallel sync

bitbylaw/steps/advoware_cal_sync/calendar_sync_api_step.py

@@ -12,8 +12,7 @@ config = {
     'flows': ['advoware']
 }
 
-async def handler(req):
-    context = req.context
+async def handler(req, context):
     try:
         # Konfiguration aus Request-Body
         body = req.get('body', {})
@@ -37,7 +36,7 @@ async def handler(req):
             socket_timeout=Config.REDIS_TIMEOUT_SECONDS
         )
         
-        if redis_client.get(employee_lock_key):
+        if redis_client.set(employee_lock_key, 'api', ex=1800, nx=True) is None:
             context.logger.info(f"Calendar Sync API: Sync bereits aktiv für {kuerzel}, überspringe")
             return {
                 'status': 409,
@@ -51,9 +50,7 @@ async def handler(req):
 
         context.logger.info(f"Calendar Sync API aufgerufen für {kuerzel}")
         
-        # Setze Lock für 30 Minuten
-        redis_client.set(employee_lock_key, 'api', ex=1800)
-        context.logger.info(f"Calendar Sync API: Lock gesetzt für {kuerzel}")
+        # Lock erfolgreich gesetzt, jetzt emittieren
 
         # Emit Event für den Sync
         await context.emit({

bitbylaw/steps/advoware_cal_sync/calendar_sync_cron_step.py

@@ -50,6 +50,19 @@ async def handler(context):
                 context.logger.info(f"DEBUG: Überspringe {kuerzel}, nur SB wird gesynct")
                 continue
 
+            employee_lock_key = f'calendar_sync_lock_{kuerzel}'
+
+            redis_client = redis.Redis(
+                host=Config.REDIS_HOST,
+                port=int(Config.REDIS_PORT),
+                db=int(Config.REDIS_DB_CALENDAR_SYNC),
+                socket_timeout=Config.REDIS_TIMEOUT_SECONDS
+            )
+
+            if redis_client.set(employee_lock_key, 'cron', ex=1800, nx=True) is None:
+                context.logger.info(f"Calendar Sync Cron: Sync bereits aktiv für {kuerzel}, überspringe")
+                continue
+
             # Emit event for this employee
             await context.emit({
                 "topic": "calendar_sync_employee",

bitbylaw/steps/advoware_cal_sync/calendar_sync_event_step.py

@@ -848,9 +848,9 @@ async def process_updates(state, conn, service, calendar_id, kuerzel, advoware,
                     await conn.execute("UPDATE calendar_sync SET sync_status = 'failed' WHERE sync_id = $1;", row['sync_id'])
 CALENDAR_SYNC_LOCK_KEY = 'calendar_sync_lock'
 
-async def handler(context):
+async def handler(event_data, context):
     """Main event handler for calendar sync."""
-    kuerzel = context.input.data.get('kuerzel')
+    kuerzel = event_data.get('kuerzel')
     if not kuerzel:
         log_operation('error', "No kuerzel provided in event", context=context)
         return {'status': 400, 'body': {'error': 'No kuerzel provided'}}
@@ -867,13 +867,6 @@ async def handler(context):
     )
 
     try:
-        if redis_client.get(employee_lock_key):
-            log_operation('info', f"Sync already running for {kuerzel}, skipping", context=context)
-            return {'status': 200, 'body': {'status': 'skipped', 'reason': 'sync_already_running', 'kuerzel': kuerzel}}
-
-        # Set lock for 30 minutes
-        redis_client.set(employee_lock_key, 'event', ex=1800)
-        log_operation('info', f"Lock set for {kuerzel}, starting sync", context=context)
 
         log_operation('debug', "Initializing Advoware API", context=context)
         advoware = AdvowareAPI(context)

bitbylaw/types.d.ts

@@ -25,7 +25,7 @@ declare module 'motia' {
     'Advoware Proxy GET': ApiRouteHandler<Record<string, unknown>, unknown, never>
     'Advoware Proxy DELETE': ApiRouteHandler<Record<string, unknown>, unknown, never>
     'Calendar Sync Event Step': EventHandler<never, never>
-    'Calendar Sync Cron Job': CronHandler<{ topic: 'calendar.sync.triggered'; data: never }>
-    'Calendar Sync API Trigger': ApiRouteHandler<Record<string, unknown>, unknown, { topic: 'calendar.sync.triggered'; data: never }>
+    'Calendar Sync Cron Job': CronHandler<{ topic: 'calendar_sync_employee'; data: never }>
+    'Calendar Sync API Trigger': ApiRouteHandler<Record<string, unknown>, unknown, { topic: 'calendar_sync_employee'; data: never }>
   }
 }