motia/bitbylaw/steps/vmh/README_SYNC.md

Beteiligte Sync - Event Handler

📚 Vollständige Dokumentation: ../../docs/SYNC_OVERVIEW.md

Event-driven sync handler für bidirektionale Synchronisation von Beteiligten (Stammdaten).

Implementiert in: steps/vmh/beteiligte_sync_event_step.py

Subscribes

• vmh.beteiligte.create - Neuer Beteiligter in EspoCRM
• vmh.beteiligte.update - Änderung in EspoCRM
• vmh.beteiligte.delete - Löschung in EspoCRM
• vmh.beteiligte.sync_check - Cron-triggered check

Funktionsweise

1. Event empfangen

{
  "entity_id": "68e3e7eab49f09adb",
  "action": "sync_check",
  "source": "cron",
  "timestamp": "2026-02-07T16:00:00"
}

2. Lock acquisition (Redis)

lock_key = f"sync_lock:cbeteiligte:{entity_id}"
acquired = redis.set(lock_key, "locked", nx=True, ex=300)

• Atomar via Redis SET NX
• TTL: 5 Minuten (verhindert Deadlocks)
• Verhindert: Parallele Syncs derselben Entity

3. Routing nach Action

CREATE (kein betnr)

Map EspoCRM → Advoware
  ↓
POST /api/v1/advonet/Beteiligte
  ↓
Response: {betNr: 12345}
  ↓
Update EspoCRM: betnr=12345, syncStatus=clean (combined!)

UPDATE (hat betnr)

GET /api/v1/advonet/Beteiligte/{betnr}
  ↓
Timestamp-Vergleich (modifiedAt vs geaendertAm)
  ↓
├─ espocrm_newer → PUT to Advoware
├─ advoware_newer → PATCH to EspoCRM
├─ conflict → EspoCRM wins + Notification
└─ no_change → Skip

4. Lock release

await sync_utils.release_sync_lock(
    entity_id,
    'clean',
    extra_fields={'betnr': new_betnr}  # Optional: combine operations
)

• Updates syncStatus, advowareLastSync, syncRetryCount
• Optional: Merge zusätzliche Felder (betnr, etc.)
• Löscht Redis lock

Optimierungen

Redis Distributed Lock

# VORHER: Nicht-atomar (Race Condition möglich)
entity = await get_entity(...)
if entity.syncStatus == 'syncing':
    return
await update_entity(..., {'syncStatus': 'syncing'})

# NACHHER: Atomarer Redis lock
acquired = redis.set(lock_key, "locked", nx=True, ex=300)
if not acquired:
    return

Combined API Calls

# VORHER: 2 API calls
await release_sync_lock(entity_id, 'clean')
await update_entity(entity_id, {'betnr': new_betnr})

# NACHHER: 1 API call (33% faster)
await release_sync_lock(
    entity_id,
    'clean',
    extra_fields={'betnr': new_betnr}
)

Merge Utility

# Keine Code-Duplikation mehr (3x → 1x)
merged_data = sync_utils.merge_for_advoware_put(
    advo_entity,
    espo_entity,
    mapper
)

Error Handling

Retriable Errors

• Netzwerk-Timeout → syncStatus=failed, retry beim nächsten Cron
• 500 Server Error → syncStatus=failed, retry
• Redis unavailable → Fallback zu syncStatus-only lock

Non-Retriable Errors

• 400 Bad Request → syncStatus=failed, keine Auto-Retry
• 404 Not Found → Entity gelöscht, markiere als deleted_in_advoware
• 401 Auth Error → syncStatus=failed, keine Auto-Retry

Max Retries

if retry_count >= 5:
    syncStatus = 'permanently_failed'
    send_notification("Max retries exceeded")

Performance

Metric	Value
Latency (CREATE)	~200ms
Latency (UPDATE)	~250ms
API Calls (CREATE)	2
API Calls (UPDATE)	2
Lock Timeout	5 min

Dependencies

• services/espocrm.py - EspoCRM API
• services/advoware.py - Advoware API
• services/espocrm_mapper.py - Entity mapper
• services/beteiligte_sync_utils.py - Sync utilities
• Redis (localhost:6379, DB 1) - Distributed locking

Testing

# Test event
event_data = {
    'entity_id': '68e3e7eab49f09adb',
    'action': 'sync_check',
    'source': 'test'
}

await handler(event_data, context)

# Verify
entity = await espocrm.get_entity('CBeteiligte', entity_id)
assert entity['syncStatus'] == 'clean'
assert entity['betnr'] is not None

Siehe auch

• Beteiligte Sync Docs - Vollständige Dokumentation
• Cron Step - Findet Entities für Sync
• Sync Utils - Helper functions