bsiggel/motia-iii
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Refactor AI Knowledge sync processes to remove full sync parameter and ensure Blake3 verification is always performed

Browse Source
This commit is contained in:
bsiggel
2026-03-12 22:41:19 +00:00
parent 8f1533337c
commit 46c0bbf381
4 changed files with 20 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
steps/vmh/aiknowledge_full_sync_cron_step.py
View File

@@ -1,11 +1,11 @@
"""AI Knowledge Full Sync - Daily Cron Job"""
"""AI Knowledge Daily Sync - Cron Job"""
from typing import Any
from motia import FlowContext, cron
config = {
"name": "AI Knowledge Full Sync",
"description": "Daily full sync of all CAIKnowledge entities (catches missed webhooks)",
"name": "AI Knowledge Daily Sync",
"description": "Daily sync of all CAIKnowledge entities (catches missed webhooks, Blake3 verification included)",
"flows": ["aiknowledge-full-sync"],
"triggers": [
cron("0 0 2 * * *"), # Daily at 2:00 AM
@@ -16,16 +16,17 @@ config = {
async def handler(input_data: None, ctx: FlowContext[Any]) -> None:
"""
Daily full sync handler.
Daily sync handler - ensures all active knowledge bases are synchronized.
Loads all CAIKnowledge entities that need sync and emits events.
Blake3 hash verification is always performed (hash available from JunctionData API).
Runs every day at 02:00:00.
"""
from services.espocrm import EspoCRMAPI
from services.models import AIKnowledgeActivationStatus, AIKnowledgeSyncStatus
ctx.logger.info("=" * 80)
ctx.logger.info("🌙 DAILY FULL SYNC STARTED")
ctx.logger.info("🌙 DAILY AI KNOWLEDGE SYNC STARTED")
ctx.logger.info("=" * 80)
espocrm = EspoCRMAPI(ctx)
@@ -63,20 +64,22 @@ async def handler(input_data: None, ctx: FlowContext[Any]) -> None:
ctx.logger.info("=" * 80)
return
# Enqueue sync events for all
# Enqueue sync events for all (Blake3 verification always enabled)
for i, entity in enumerate(entities, 1):
await ctx.enqueue({
'topic': 'aiknowledge.sync',
'data': {
'knowledge_id': entity['id'],
'source': 'daily_full_sync',
'full_sync': True # Enable Blake3 verification
'source': 'daily_cron'
}
})
ctx.logger.info(
f"📤 [{i}/{total}] Enqueued: {entity['name']} "
f"(syncStatus={entity.get('syncStatus')})"
)
ctx.logger.info("=" * 80)
ctx.logger.info(f"✅ Full sync complete: {total} events enqueued")
ctx.logger.info(f"✅ Daily sync complete: {total} events enqueued")
ctx.logger.info("=" * 80)
except Exception as e:

6
steps/vmh/aiknowledge_sync_event_step.py
View File

@@ -36,7 +36,6 @@ async def handler(event_data: Dict[str, Any], ctx: FlowContext[Any]) -> None:
# Extract data
knowledge_id = event_data.get('knowledge_id')
source = event_data.get('source', 'unknown')
full_sync = event_data.get('full_sync', False) # Blake3 verification mode
if not knowledge_id:
ctx.logger.error("❌ Missing knowledge_id in event data")
@@ -44,7 +43,6 @@ async def handler(event_data: Dict[str, Any], ctx: FlowContext[Any]) -> None:
ctx.logger.info(f"📋 Knowledge ID: {knowledge_id}")
ctx.logger.info(f"📋 Source: {source}")
ctx.logger.info(f"📋 Full Sync Mode: {full_sync}")
ctx.logger.info("=" * 80)
# Get Redis for locking
@@ -62,8 +60,8 @@ async def handler(event_data: Dict[str, Any], ctx: FlowContext[Any]) -> None:
raise RuntimeError(f"Lock busy for {knowledge_id}") # Motia will retry
try:
# Perform sync
await sync_utils.sync_knowledge_to_xai(knowledge_id, ctx, full_sync=full_sync)
# Perform sync (Blake3 hash verification always enabled)
await sync_utils.sync_knowledge_to_xai(knowledge_id, ctx)
ctx.logger.info("=" * 80)
ctx.logger.info("✅ AI KNOWLEDGE SYNC COMPLETED")