Add sync strategy documentation and templates for bidirectional sync between EspoCRM and Advoware
- Introduced SYNC_STRATEGY_ARCHIVE.md detailing the sync process, status values, and flow for updating entities from EspoCRM to Advoware and vice versa. - Created SYNC_TEMPLATE.md as a guide for implementing new syncs, including field definitions, mapper examples, sync utilities, event handlers, and cron jobs. - Added README_SYNC.md for the Beteiligte sync event handler, outlining its functionality, event subscriptions, optimizations, error handling, and performance metrics.
This commit is contained in:
442
bitbylaw/docs/SYNC_TEMPLATE.md
Normal file
442
bitbylaw/docs/SYNC_TEMPLATE.md
Normal file
@@ -0,0 +1,442 @@
|
||||
# Advoware Sync Template
|
||||
|
||||
Template für neue bidirektionale Syncs zwischen EspoCRM und Advoware.
|
||||
|
||||
## Quick Start
|
||||
|
||||
Für neuen Sync von Entity `XYZ`:
|
||||
|
||||
### 1. EspoCRM Custom Fields
|
||||
```sql
|
||||
-- In EspoCRM Admin → Entity Manager → XYZ
|
||||
advowareId (int, unique) -- Foreign Key
|
||||
syncStatus (enum: clean|dirty|...) -- Status
|
||||
advowareLastSync (datetime) -- Letzter Sync
|
||||
syncErrorMessage (text, 2000) -- Fehler
|
||||
syncRetryCount (int) -- Retries
|
||||
```
|
||||
|
||||
### 2. Mapper erstellen
|
||||
```python
|
||||
# services/xyz_mapper.py
|
||||
class XYZMapper:
|
||||
@staticmethod
|
||||
def map_espo_to_advoware(espo_entity: Dict) -> Dict:
|
||||
"""EspoCRM → Advoware transformation"""
|
||||
return {
|
||||
'field1': espo_entity.get('espoField1'),
|
||||
'field2': espo_entity.get('espoField2'),
|
||||
# Nur relevante Felder mappen!
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def map_advoware_to_espo(advo_entity: Dict) -> Dict:
|
||||
"""Advoware → EspoCRM transformation"""
|
||||
return {
|
||||
'espoField1': advo_entity.get('field1'),
|
||||
'espoField2': advo_entity.get('field2'),
|
||||
}
|
||||
```
|
||||
|
||||
### 3. Sync Utils erstellen
|
||||
```python
|
||||
# services/xyz_sync_utils.py
|
||||
import redis
|
||||
from typing import Dict, Any, Optional
|
||||
from datetime import datetime
|
||||
import pytz
|
||||
|
||||
MAX_SYNC_RETRIES = 5
|
||||
LOCK_TTL_SECONDS = 300
|
||||
|
||||
class XYZSync:
|
||||
def __init__(self, espocrm_api, redis_client: redis.Redis, context=None):
|
||||
self.espocrm = espocrm_api
|
||||
self.redis = redis_client
|
||||
self.context = context
|
||||
|
||||
async def acquire_sync_lock(self, entity_id: str) -> bool:
|
||||
"""Atomic distributed lock via Redis"""
|
||||
if self.redis:
|
||||
lock_key = f"sync_lock:xyz:{entity_id}"
|
||||
acquired = self.redis.set(lock_key, "locked", nx=True, ex=LOCK_TTL_SECONDS)
|
||||
if not acquired:
|
||||
return False
|
||||
|
||||
await self.espocrm.update_entity('XYZ', entity_id, {'syncStatus': 'syncing'})
|
||||
return True
|
||||
|
||||
async def release_sync_lock(
|
||||
self,
|
||||
entity_id: str,
|
||||
new_status: str = 'clean',
|
||||
error_message: Optional[str] = None,
|
||||
increment_retry: bool = False,
|
||||
extra_fields: Optional[Dict[str, Any]] = None
|
||||
):
|
||||
"""Release lock and update status (combined operation)"""
|
||||
update_data = {
|
||||
'syncStatus': new_status,
|
||||
'advowareLastSync': datetime.now(pytz.UTC).isoformat()
|
||||
}
|
||||
|
||||
if error_message:
|
||||
update_data['syncErrorMessage'] = error_message[:2000]
|
||||
else:
|
||||
update_data['syncErrorMessage'] = None
|
||||
|
||||
if increment_retry:
|
||||
entity = await self.espocrm.get_entity('XYZ', entity_id)
|
||||
retry_count = (entity.get('syncRetryCount') or 0) + 1
|
||||
update_data['syncRetryCount'] = retry_count
|
||||
|
||||
if retry_count >= MAX_SYNC_RETRIES:
|
||||
update_data['syncStatus'] = 'permanently_failed'
|
||||
await self.send_notification(
|
||||
entity_id,
|
||||
f"Sync failed after {MAX_SYNC_RETRIES} attempts"
|
||||
)
|
||||
else:
|
||||
update_data['syncRetryCount'] = 0
|
||||
|
||||
if extra_fields:
|
||||
update_data.update(extra_fields)
|
||||
|
||||
await self.espocrm.update_entity('XYZ', entity_id, update_data)
|
||||
|
||||
if self.redis:
|
||||
self.redis.delete(f"sync_lock:xyz:{entity_id}")
|
||||
|
||||
def compare_timestamps(self, espo_ts, advo_ts, last_sync_ts):
|
||||
"""Compare timestamps and determine sync direction"""
|
||||
# Parse timestamps
|
||||
espo = self._parse_ts(espo_ts)
|
||||
advo = self._parse_ts(advo_ts)
|
||||
sync = self._parse_ts(last_sync_ts)
|
||||
|
||||
if not sync:
|
||||
if not espo or not advo:
|
||||
return "no_change"
|
||||
return "espocrm_newer" if espo > advo else "advoware_newer"
|
||||
|
||||
espo_changed = espo and espo > sync
|
||||
advo_changed = advo and advo > sync
|
||||
|
||||
if espo_changed and advo_changed:
|
||||
return "conflict"
|
||||
elif espo_changed:
|
||||
return "espocrm_newer"
|
||||
elif advo_changed:
|
||||
return "advoware_newer"
|
||||
else:
|
||||
return "no_change"
|
||||
|
||||
def merge_for_advoware_put(self, advo_entity, espo_entity, mapper):
|
||||
"""Merge EspoCRM updates into Advoware entity (Read-Modify-Write)"""
|
||||
advo_updates = mapper.map_espo_to_advoware(espo_entity)
|
||||
merged = {**advo_entity, **advo_updates}
|
||||
|
||||
self._log(f"📝 Merge: {len(advo_updates)} updates → {len(merged)} total")
|
||||
return merged
|
||||
|
||||
async def send_notification(self, entity_id, message):
|
||||
"""Send in-app notification to EspoCRM"""
|
||||
# Implementation...
|
||||
pass
|
||||
|
||||
def _parse_ts(self, ts):
|
||||
"""Parse timestamp string to datetime"""
|
||||
# Implementation...
|
||||
pass
|
||||
|
||||
def _log(self, msg, level='info'):
|
||||
"""Log with context support"""
|
||||
if self.context:
|
||||
getattr(self.context.logger, level)(msg)
|
||||
```
|
||||
|
||||
### 4. Event Handler erstellen
|
||||
```python
|
||||
# steps/vmh/xyz_sync_event_step.py
|
||||
from services.advoware import AdvowareAPI
|
||||
from services.espocrm import EspoCRMAPI
|
||||
from services.xyz_mapper import XYZMapper
|
||||
from services.xyz_sync_utils import XYZSync
|
||||
import redis
|
||||
from config import Config
|
||||
|
||||
config = {
|
||||
'type': 'event',
|
||||
'name': 'VMH XYZ Sync Handler',
|
||||
'description': 'Bidirectional sync for XYZ entities',
|
||||
'subscribes': [
|
||||
'vmh.xyz.create',
|
||||
'vmh.xyz.update',
|
||||
'vmh.xyz.delete',
|
||||
'vmh.xyz.sync_check'
|
||||
],
|
||||
'flows': ['vmh']
|
||||
}
|
||||
|
||||
async def handler(event_data, context):
|
||||
entity_id = event_data.get('entity_id')
|
||||
action = event_data.get('action', 'sync_check')
|
||||
|
||||
if not entity_id:
|
||||
context.logger.error("No entity_id in event")
|
||||
return
|
||||
|
||||
# Initialize
|
||||
redis_client = redis.Redis(
|
||||
host=Config.REDIS_HOST,
|
||||
port=int(Config.REDIS_PORT),
|
||||
db=int(Config.REDIS_DB_ADVOWARE_CACHE),
|
||||
decode_responses=True
|
||||
)
|
||||
|
||||
espocrm = EspoCRMAPI()
|
||||
advoware = AdvowareAPI(context)
|
||||
sync_utils = XYZSync(espocrm, redis_client, context)
|
||||
mapper = XYZMapper()
|
||||
|
||||
try:
|
||||
# Acquire lock
|
||||
if not await sync_utils.acquire_sync_lock(entity_id):
|
||||
context.logger.warning(f"Already syncing: {entity_id}")
|
||||
return
|
||||
|
||||
# Load entity
|
||||
espo_entity = await espocrm.get_entity('XYZ', entity_id)
|
||||
advoware_id = espo_entity.get('advowareId')
|
||||
|
||||
# Route to handler
|
||||
if not advoware_id and action in ['create', 'sync_check']:
|
||||
await handle_create(entity_id, espo_entity, espocrm, advoware, sync_utils, mapper, context)
|
||||
elif advoware_id:
|
||||
await handle_update(entity_id, advoware_id, espo_entity, espocrm, advoware, sync_utils, mapper, context)
|
||||
|
||||
except Exception as e:
|
||||
context.logger.error(f"Sync failed: {e}")
|
||||
await sync_utils.release_sync_lock(entity_id, 'failed', str(e), increment_retry=True)
|
||||
|
||||
|
||||
async def handle_create(entity_id, espo_entity, espocrm, advoware, sync_utils, mapper, context):
|
||||
"""Create new entity in Advoware"""
|
||||
try:
|
||||
advo_data = mapper.map_espo_to_advoware(espo_entity)
|
||||
|
||||
result = await advoware.api_call(
|
||||
'api/v1/advonet/XYZ',
|
||||
method='POST',
|
||||
data=advo_data
|
||||
)
|
||||
|
||||
new_id = result.get('id')
|
||||
if not new_id:
|
||||
raise Exception(f"No ID in response: {result}")
|
||||
|
||||
# Combined API call: release lock + save foreign key
|
||||
await sync_utils.release_sync_lock(
|
||||
entity_id,
|
||||
'clean',
|
||||
extra_fields={'advowareId': new_id}
|
||||
)
|
||||
|
||||
context.logger.info(f"✅ Created in Advoware: {new_id}")
|
||||
|
||||
except Exception as e:
|
||||
context.logger.error(f"❌ Create failed: {e}")
|
||||
await sync_utils.release_sync_lock(entity_id, 'failed', str(e), increment_retry=True)
|
||||
|
||||
|
||||
async def handle_update(entity_id, advoware_id, espo_entity, espocrm, advoware, sync_utils, mapper, context):
|
||||
"""Sync existing entity"""
|
||||
try:
|
||||
# Fetch from Advoware
|
||||
advo_result = await advoware.api_call(f'api/v1/advonet/XYZ/{advoware_id}', method='GET')
|
||||
advo_entity = advo_result[0] if isinstance(advo_result, list) else advo_result
|
||||
|
||||
if not advo_entity:
|
||||
context.logger.error(f"Entity not found in Advoware: {advoware_id}")
|
||||
await sync_utils.release_sync_lock(entity_id, 'failed', "Not found in Advoware")
|
||||
return
|
||||
|
||||
# Compare timestamps
|
||||
comparison = sync_utils.compare_timestamps(
|
||||
espo_entity.get('modifiedAt'),
|
||||
advo_entity.get('modifiedAt'), # Advoware timestamp field
|
||||
espo_entity.get('advowareLastSync')
|
||||
)
|
||||
|
||||
# Initial sync (no last_sync)
|
||||
if not espo_entity.get('advowareLastSync'):
|
||||
merged_data = sync_utils.merge_for_advoware_put(advo_entity, espo_entity, mapper)
|
||||
await advoware.api_call(f'api/v1/advonet/XYZ/{advoware_id}', method='PUT', data=merged_data)
|
||||
await sync_utils.release_sync_lock(entity_id, 'clean')
|
||||
return
|
||||
|
||||
# No change
|
||||
if comparison == 'no_change':
|
||||
await sync_utils.release_sync_lock(entity_id, 'clean')
|
||||
return
|
||||
|
||||
# EspoCRM newer
|
||||
if comparison == 'espocrm_newer':
|
||||
merged_data = sync_utils.merge_for_advoware_put(advo_entity, espo_entity, mapper)
|
||||
await advoware.api_call(f'api/v1/advonet/XYZ/{advoware_id}', method='PUT', data=merged_data)
|
||||
await sync_utils.release_sync_lock(entity_id, 'clean')
|
||||
|
||||
# Advoware newer
|
||||
elif comparison == 'advoware_newer':
|
||||
espo_data = mapper.map_advoware_to_espo(advo_entity)
|
||||
await espocrm.update_entity('XYZ', entity_id, espo_data)
|
||||
await sync_utils.release_sync_lock(entity_id, 'clean')
|
||||
|
||||
# Conflict → EspoCRM wins
|
||||
elif comparison == 'conflict':
|
||||
merged_data = sync_utils.merge_for_advoware_put(advo_entity, espo_entity, mapper)
|
||||
await advoware.api_call(f'api/v1/advonet/XYZ/{advoware_id}', method='PUT', data=merged_data)
|
||||
await sync_utils.send_notification(entity_id, "Conflict resolved: EspoCRM won")
|
||||
await sync_utils.release_sync_lock(entity_id, 'clean')
|
||||
|
||||
except Exception as e:
|
||||
context.logger.error(f"❌ Update failed: {e}")
|
||||
await sync_utils.release_sync_lock(entity_id, 'failed', str(e), increment_retry=True)
|
||||
```
|
||||
|
||||
### 5. Cron erstellen
|
||||
```python
|
||||
# steps/vmh/xyz_sync_cron_step.py
|
||||
import asyncio
|
||||
from services.espocrm import EspoCRMAPI
|
||||
import datetime
|
||||
|
||||
config = {
|
||||
'type': 'cron',
|
||||
'name': 'VMH XYZ Sync Cron',
|
||||
'description': 'Check for XYZ entities needing sync',
|
||||
'schedule': '*/15 * * * *', # Every 15 minutes
|
||||
'flows': ['vmh'],
|
||||
'emits': ['vmh.xyz.sync_check']
|
||||
}
|
||||
|
||||
async def handler(context):
|
||||
context.logger.info("🕐 XYZ Sync Cron started")
|
||||
|
||||
espocrm = EspoCRMAPI()
|
||||
threshold = datetime.datetime.now() - datetime.timedelta(hours=24)
|
||||
|
||||
# Find entities needing sync
|
||||
unclean_filter = {
|
||||
'where': [{
|
||||
'type': 'or',
|
||||
'value': [
|
||||
{'type': 'equals', 'attribute': 'syncStatus', 'value': 'pending_sync'},
|
||||
{'type': 'equals', 'attribute': 'syncStatus', 'value': 'dirty'},
|
||||
{'type': 'equals', 'attribute': 'syncStatus', 'value': 'failed'},
|
||||
]
|
||||
}]
|
||||
}
|
||||
|
||||
result = await espocrm.search_entities('XYZ', unclean_filter, max_size=100)
|
||||
entities = result.get('list', [])
|
||||
entity_ids = [e['id'] for e in entities]
|
||||
|
||||
context.logger.info(f"Found {len(entity_ids)} entities to sync")
|
||||
|
||||
if not entity_ids:
|
||||
return
|
||||
|
||||
# Batch emit (parallel)
|
||||
tasks = [
|
||||
context.emit({
|
||||
'topic': 'vmh.xyz.sync_check',
|
||||
'data': {
|
||||
'entity_id': eid,
|
||||
'action': 'sync_check',
|
||||
'source': 'cron',
|
||||
'timestamp': datetime.datetime.now().isoformat()
|
||||
}
|
||||
})
|
||||
for eid in entity_ids
|
||||
]
|
||||
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
success_count = sum(1 for r in results if not isinstance(r, Exception))
|
||||
|
||||
context.logger.info(f"✅ Emitted {success_count}/{len(entity_ids)} events")
|
||||
```
|
||||
|
||||
## Best Practices
|
||||
|
||||
### ✅ DO
|
||||
- Use Redis distributed lock (atomicity)
|
||||
- Combine API calls with `extra_fields`
|
||||
- Use `merge_for_advoware_put()` utility
|
||||
- Implement max retries (5x)
|
||||
- Batch emit in cron with `asyncio.gather()`
|
||||
- Map only relevant fields (avoid overhead)
|
||||
- Add proper error logging
|
||||
|
||||
### ❌ DON'T
|
||||
- Don't use GET-then-PUT for locks (race condition)
|
||||
- Don't make unnecessary API calls
|
||||
- Don't duplicate merge logic
|
||||
- Don't retry infinitely
|
||||
- Don't emit events sequentially in cron
|
||||
- Don't map every field (performance)
|
||||
- Don't swallow exceptions silently
|
||||
|
||||
## Architecture Principles
|
||||
|
||||
1. **Atomicity**: Redis lock + TTL
|
||||
2. **Efficiency**: Combined operations
|
||||
3. **Reusability**: Utility functions
|
||||
4. **Robustness**: Max retries + notifications
|
||||
5. **Scalability**: Batch processing
|
||||
6. **Maintainability**: Clear separation of concerns
|
||||
|
||||
## Performance Targets
|
||||
|
||||
| Metric | Target |
|
||||
|--------|--------|
|
||||
| Single sync latency | < 500ms |
|
||||
| API calls per operation | ≤ 3 |
|
||||
| Cron execution (100 entities) | < 2s |
|
||||
| Lock timeout | 5 min |
|
||||
| Max retries | 5 |
|
||||
|
||||
## Testing
|
||||
|
||||
```python
|
||||
# Test script template
|
||||
async def main():
|
||||
entity_id = 'test-id'
|
||||
espo = EspoCRMAPI()
|
||||
|
||||
# Reset entity
|
||||
await espo.update_entity('XYZ', entity_id, {
|
||||
'advowareLastSync': None,
|
||||
'syncStatus': 'clean',
|
||||
'syncRetryCount': 0
|
||||
})
|
||||
|
||||
# Trigger sync
|
||||
event_data = {
|
||||
'entity_id': entity_id,
|
||||
'action': 'sync_check',
|
||||
'source': 'test'
|
||||
}
|
||||
|
||||
await xyz_sync_event_step.handler(event_data, MockContext())
|
||||
|
||||
# Verify
|
||||
entity_after = await espo.get_entity('XYZ', entity_id)
|
||||
assert entity_after['syncStatus'] == 'clean'
|
||||
```
|
||||
|
||||
## Siehe auch
|
||||
|
||||
- [Beteiligte Sync](BETEILIGTE_SYNC.md) - Reference implementation
|
||||
- [Advoware API Docs](advoware/)
|
||||
- [EspoCRM API Docs](API.md)
|
||||
Reference in New Issue
Block a user