70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.model.memory import MemorySyncOutbox
|
|
from app.service.relationship_service import RelationshipService
|
|
|
|
|
|
class GraphProjectionWorker:
|
|
"""Projects approved domain events to Neo4j; callers provide durable dedup storage."""
|
|
|
|
def __init__(
|
|
self, relationships: RelationshipService, session: AsyncSession | None = None
|
|
) -> None:
|
|
self.relationships = relationships
|
|
self.session = session
|
|
self.processed_event_ids: set[str] = set()
|
|
|
|
async def project(self, event_id: str, payload: dict[str, Any]) -> bool:
|
|
if event_id in self.processed_event_ids:
|
|
return False
|
|
if self.session is not None:
|
|
existing = await self.session.scalar(
|
|
select(MemorySyncOutbox).where(MemorySyncOutbox.event_uuid == event_id)
|
|
)
|
|
if existing is not None and existing.status == "processed":
|
|
return False
|
|
relation = payload.get("relationship")
|
|
if relation not in RelationshipService.ALLOWED_RELATIONSHIPS:
|
|
raise ValueError("relationship is not allowed")
|
|
query = (
|
|
"MERGE (a:Entity {entity_id: $source_id}) "
|
|
"MERGE (b:Entity {entity_id: $target_id}) "
|
|
f"MERGE (a)-[r:{relation}]->(b) "
|
|
"SET r.trace_id = $trace_id, r.confidence = $confidence"
|
|
)
|
|
try:
|
|
await self.relationships.driver.execute_query(
|
|
query,
|
|
source_id=str(payload["source_id"]),
|
|
target_id=str(payload["target_id"]),
|
|
trace_id=str(payload.get("trace_id", "")),
|
|
confidence=float(payload.get("confidence", 0.0)),
|
|
)
|
|
except Exception as exc:
|
|
if self.session is not None:
|
|
record = await self.session.scalar(
|
|
select(MemorySyncOutbox).where(MemorySyncOutbox.event_uuid == event_id)
|
|
)
|
|
if record is not None:
|
|
record.retry_count += 1
|
|
record.last_error = str(exc)[:500]
|
|
record.status = "dead" if record.retry_count >= 5 else "pending"
|
|
record.next_retry_at = datetime.now(UTC).replace(tzinfo=None) + timedelta(
|
|
seconds=min(300, 2 ** record.retry_count)
|
|
)
|
|
await self.session.commit()
|
|
raise
|
|
self.processed_event_ids.add(event_id)
|
|
if self.session is not None:
|
|
record = await self.session.scalar(
|
|
select(MemorySyncOutbox).where(MemorySyncOutbox.event_uuid == event_id)
|
|
)
|
|
if record is not None:
|
|
record.status = "processed"
|
|
await self.session.commit()
|
|
return True
|