382 lines
15 KiB
Python
382 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
from app.core.database import session_scope
|
|
from app.modules.sync import repository as repo
|
|
from app.modules.sync.models import SyncConflict, SyncOutbox
|
|
from app.modules.sync.schemas import (
|
|
AckChangesRequest,
|
|
AckChangesResponse,
|
|
ChangeEventIn,
|
|
ChangeEventOut,
|
|
PullChangesResponse,
|
|
PushChangesResponse,
|
|
)
|
|
from app.modules.zootech.catalog_apply import apply_catalog_change
|
|
|
|
|
|
class SyncEngine:
|
|
ORCHESTRATOR_SITE_ID = "orchestrator"
|
|
|
|
def __init__(self, enterprise_id: str, site_id: str) -> None:
|
|
self.enterprise_id = enterprise_id
|
|
self.site_id = site_id
|
|
|
|
def process_push(self, events: list[ChangeEventIn], origin: str) -> PushChangesResponse:
|
|
applied: list[str] = []
|
|
held: list[str] = []
|
|
conflict_ids: list[str] = []
|
|
|
|
for event in events:
|
|
if repo.event_exists(event.event_id):
|
|
applied.append(event.event_id)
|
|
continue
|
|
if repo.applied_event_exists(self.site_id, event.event_id):
|
|
applied.append(event.event_id)
|
|
continue
|
|
|
|
seq = event.seq or repo.next_seq(self.enterprise_id)
|
|
payload_json = json.dumps(event.payload, ensure_ascii=False)
|
|
repo.append_event_log(
|
|
enterprise_id=self.enterprise_id,
|
|
event_id=event.event_id,
|
|
origin=origin,
|
|
origin_site_id=event.origin_site_id or self.site_id,
|
|
seq=seq,
|
|
domain=event.domain,
|
|
table_name=event.table,
|
|
record_id=event.record_id,
|
|
action=event.action,
|
|
version=event.version,
|
|
content_hash=event.content_hash,
|
|
payload_json=payload_json,
|
|
)
|
|
|
|
outcome = self._apply_or_hold(event, origin)
|
|
if outcome == "applied":
|
|
repo.mark_applied(self.enterprise_id, self.ORCHESTRATOR_SITE_ID, event.event_id)
|
|
applied.append(event.event_id)
|
|
elif outcome == "held":
|
|
held.append(event.event_id)
|
|
elif isinstance(outcome, str) and outcome.startswith("conflict:"):
|
|
conflict_ids.append(outcome.split(":", 1)[1])
|
|
|
|
return PushChangesResponse(
|
|
applied_event_ids=applied,
|
|
held_event_ids=held,
|
|
conflicts=conflict_ids,
|
|
)
|
|
|
|
def _apply_or_hold(self, event: ChangeEventIn, origin: str) -> str:
|
|
state = repo.get_record_state(self.enterprise_id, event.table, event.record_id)
|
|
if state and state.agreed_hash and state.agreed_hash != event.content_hash:
|
|
agreed_version = int(state.agreed_version or 0)
|
|
if event.version <= agreed_version:
|
|
pending = self._find_pending_conflict(event.table, event.record_id)
|
|
if pending:
|
|
self._append_held_event(pending.id, event.event_id)
|
|
return "held"
|
|
conflict_id = self._create_conflict(event, origin, state)
|
|
self._hold_outbox_for_record(event.table, event.record_id)
|
|
return f"conflict:{conflict_id}"
|
|
|
|
if event.domain in ("report", "reports"):
|
|
from app.modules.zootech.report_apply import apply_report_change
|
|
|
|
apply_report_change(
|
|
self.enterprise_id,
|
|
event.table,
|
|
event.record_id,
|
|
event.action,
|
|
event.payload,
|
|
event.version,
|
|
event.content_hash,
|
|
farm_hub_id=event.origin_site_id if origin == "hub" else None,
|
|
)
|
|
else:
|
|
apply_catalog_change(
|
|
self.enterprise_id,
|
|
event.table,
|
|
event.record_id,
|
|
event.action,
|
|
event.payload,
|
|
event.version,
|
|
event.content_hash,
|
|
)
|
|
repo.upsert_record_state(
|
|
self.enterprise_id,
|
|
event.table,
|
|
event.record_id,
|
|
event.version,
|
|
event.content_hash,
|
|
event.event_id,
|
|
)
|
|
if origin == "orchestrator" or self.site_id == self.ORCHESTRATOR_SITE_ID:
|
|
self._enqueue_fanout(event)
|
|
return "applied"
|
|
|
|
def _append_held_event(self, conflict_id: str, event_id: str) -> None:
|
|
with session_scope() as db:
|
|
row = db.get(SyncConflict, conflict_id)
|
|
if not row:
|
|
return
|
|
held = json.loads(row.held_event_ids_json or "[]")
|
|
if event_id not in held:
|
|
held.append(event_id)
|
|
row.held_event_ids_json = json.dumps(held, ensure_ascii=False)
|
|
|
|
def _hold_outbox_for_record(self, table_name: str, record_id: str) -> None:
|
|
with session_scope() as db:
|
|
from sqlalchemy import select
|
|
|
|
rows = list(
|
|
db.scalars(
|
|
select(SyncOutbox).where(
|
|
SyncOutbox.enterprise_id == self.enterprise_id,
|
|
SyncOutbox.table_name == table_name,
|
|
SyncOutbox.record_id == record_id,
|
|
SyncOutbox.status.in_(("pending", "sent")),
|
|
)
|
|
)
|
|
)
|
|
for row in rows:
|
|
row.status = "held"
|
|
|
|
def _has_pending_orchestrator_change(self, table_name: str, record_id: str, origin: str) -> bool:
|
|
if origin == "hub":
|
|
with session_scope() as db:
|
|
from sqlalchemy import select
|
|
|
|
row = db.scalar(
|
|
select(SyncOutbox).where(
|
|
SyncOutbox.enterprise_id == self.enterprise_id,
|
|
SyncOutbox.table_name == table_name,
|
|
SyncOutbox.record_id == record_id,
|
|
SyncOutbox.status.in_(("pending", "sent", "held")),
|
|
SyncOutbox.origin == "orchestrator",
|
|
)
|
|
)
|
|
return row is not None
|
|
return origin == "orchestrator"
|
|
|
|
def _find_pending_conflict(self, table_name: str, record_id: str) -> SyncConflict | None:
|
|
with session_scope() as db:
|
|
from sqlalchemy import select
|
|
|
|
row = db.scalar(
|
|
select(SyncConflict).where(
|
|
SyncConflict.enterprise_id == self.enterprise_id,
|
|
SyncConflict.table_name == table_name,
|
|
SyncConflict.record_id == record_id,
|
|
SyncConflict.status == "pending",
|
|
)
|
|
)
|
|
if row:
|
|
db.refresh(row)
|
|
db.expunge(row)
|
|
return row
|
|
return None
|
|
|
|
def _create_conflict(self, event: ChangeEventIn, origin: str, state) -> str:
|
|
from app.modules.zootech.catalog_apply import load_catalog_row
|
|
|
|
orchestrator_row = load_catalog_row(self.enterprise_id, event.table, event.record_id) or {}
|
|
if origin == "hub":
|
|
hub_snapshot = event.payload
|
|
orch_snapshot = orchestrator_row
|
|
else:
|
|
hub_snapshot = {}
|
|
orch_snapshot = event.payload or orchestrator_row
|
|
|
|
conflict_id = str(uuid4())
|
|
with session_scope() as db:
|
|
db.add(
|
|
SyncConflict(
|
|
id=conflict_id,
|
|
enterprise_id=self.enterprise_id,
|
|
table_name=event.table,
|
|
record_id=event.record_id,
|
|
farm_hub_id=event.origin_site_id if origin == "hub" else None,
|
|
orchestrator_snapshot_json=json.dumps(orch_snapshot, ensure_ascii=False),
|
|
hub_snapshot_json=json.dumps(hub_snapshot, ensure_ascii=False),
|
|
held_event_ids_json=json.dumps([event.event_id]),
|
|
status="pending",
|
|
)
|
|
)
|
|
return conflict_id
|
|
|
|
def _enqueue_fanout(self, event: ChangeEventIn) -> None:
|
|
seq = repo.next_seq(self.enterprise_id)
|
|
with session_scope() as db:
|
|
existing = db.scalar(
|
|
__import__("sqlalchemy").select(SyncOutbox).where(
|
|
SyncOutbox.enterprise_id == self.enterprise_id,
|
|
SyncOutbox.table_name == event.table,
|
|
SyncOutbox.record_id == event.record_id,
|
|
SyncOutbox.status.in_(("pending", "sent")),
|
|
)
|
|
)
|
|
payload_json = json.dumps(event.payload, ensure_ascii=False)
|
|
if existing:
|
|
existing.event_id = event.event_id
|
|
existing.action = event.action
|
|
existing.version = event.version
|
|
existing.content_hash = event.content_hash
|
|
existing.payload_json = payload_json
|
|
existing.seq = seq
|
|
existing.status = "pending"
|
|
existing.emitted_at = event.emitted_at
|
|
else:
|
|
db.add(
|
|
SyncOutbox(
|
|
enterprise_id=self.enterprise_id,
|
|
event_id=event.event_id,
|
|
origin="orchestrator",
|
|
origin_site_id=self.ORCHESTRATOR_SITE_ID,
|
|
seq=seq,
|
|
domain=event.domain,
|
|
table_name=event.table,
|
|
record_id=event.record_id,
|
|
action=event.action,
|
|
version=event.version,
|
|
content_hash=event.content_hash,
|
|
payload_json=payload_json,
|
|
status="pending",
|
|
emitted_at=event.emitted_at,
|
|
)
|
|
)
|
|
|
|
def process_pull(self, farm_hub_id: str, cursor: int, limit: int) -> PullChangesResponse:
|
|
rows = repo.pull_events_since(
|
|
self.enterprise_id,
|
|
cursor,
|
|
limit,
|
|
exclude_origin_site_id=self.site_id,
|
|
)
|
|
events = [
|
|
ChangeEventOut(
|
|
event_id=row.event_id,
|
|
seq=row.seq,
|
|
domain=row.domain, # type: ignore[arg-type]
|
|
table=row.table_name,
|
|
record_id=row.record_id,
|
|
action=row.action, # type: ignore[arg-type]
|
|
version=row.version,
|
|
content_hash=row.content_hash,
|
|
payload=json.loads(row.payload_json or "{}"),
|
|
emitted_at=row.received_at,
|
|
origin_site_id=row.origin_site_id,
|
|
)
|
|
for row in rows
|
|
]
|
|
next_cursor = events[-1].seq if events else cursor
|
|
repo.get_or_create_cursor(self.enterprise_id, farm_hub_id, "outbound")
|
|
return PullChangesResponse(events=events, next_cursor=next_cursor)
|
|
|
|
def process_ack(self, farm_hub_id: str, body: AckChangesRequest) -> AckChangesResponse:
|
|
cursor = repo.get_or_create_cursor(self.enterprise_id, farm_hub_id, body.direction)
|
|
max_seq = cursor.last_acked_seq
|
|
for event_id in body.event_ids:
|
|
row = self._event_log_by_id(event_id)
|
|
if row:
|
|
max_seq = max(max_seq, row.seq)
|
|
repo.mark_applied(self.enterprise_id, farm_hub_id, event_id)
|
|
repo.update_cursor_ack(farm_hub_id, body.direction, max_seq)
|
|
return AckChangesResponse(last_acked_seq=max_seq)
|
|
|
|
def _event_log_by_id(self, event_id: str):
|
|
with session_scope() as db:
|
|
from sqlalchemy import select
|
|
|
|
from app.modules.sync.models import SyncEventLog
|
|
|
|
row = db.scalar(select(SyncEventLog).where(SyncEventLog.event_id == event_id))
|
|
if row:
|
|
db.refresh(row)
|
|
db.expunge(row)
|
|
return row
|
|
|
|
def resolve_conflict(self, conflict_id: str, user_id: str, resolution: str) -> None:
|
|
conflict = repo.get_conflict(conflict_id)
|
|
if not conflict or conflict.enterprise_id != self.enterprise_id:
|
|
raise ValueError("CONFLICT_NOT_FOUND")
|
|
snapshot = (
|
|
json.loads(conflict.orchestrator_snapshot_json)
|
|
if resolution == "keep_orchestrator"
|
|
else json.loads(conflict.hub_snapshot_json)
|
|
)
|
|
event = ChangeEventIn(
|
|
event_id=str(uuid4()),
|
|
seq=repo.next_seq(self.enterprise_id),
|
|
domain="global",
|
|
table=conflict.table_name,
|
|
record_id=conflict.record_id,
|
|
action="upsert",
|
|
version=int(snapshot.get("version", 1)),
|
|
content_hash=str(snapshot.get("content_hash", "")),
|
|
payload=snapshot,
|
|
emitted_at=datetime.now(UTC),
|
|
origin_site_id=self.ORCHESTRATOR_SITE_ID,
|
|
)
|
|
apply_catalog_change(
|
|
self.enterprise_id,
|
|
event.table,
|
|
event.record_id,
|
|
event.action,
|
|
event.payload,
|
|
event.version,
|
|
event.content_hash,
|
|
)
|
|
repo.upsert_record_state(
|
|
self.enterprise_id,
|
|
event.table,
|
|
event.record_id,
|
|
event.version,
|
|
event.content_hash,
|
|
event.event_id,
|
|
)
|
|
repo.append_event_log(
|
|
enterprise_id=self.enterprise_id,
|
|
event_id=event.event_id,
|
|
origin="orchestrator",
|
|
origin_site_id=self.ORCHESTRATOR_SITE_ID,
|
|
seq=event.seq or repo.next_seq(self.enterprise_id),
|
|
domain=event.domain,
|
|
table_name=event.table,
|
|
record_id=event.record_id,
|
|
action=event.action,
|
|
version=event.version,
|
|
content_hash=event.content_hash,
|
|
payload_json=json.dumps(event.payload, ensure_ascii=False),
|
|
)
|
|
repo.mark_applied(self.enterprise_id, self.ORCHESTRATOR_SITE_ID, event.event_id)
|
|
self._enqueue_fanout(event)
|
|
self._release_held_outbox(conflict.table_name, conflict.record_id)
|
|
with session_scope() as db:
|
|
row = db.get(SyncConflict, conflict_id)
|
|
if row:
|
|
row.status = "resolved"
|
|
row.resolution = resolution
|
|
row.resolved_by = user_id
|
|
row.resolved_at = datetime.now(UTC)
|
|
|
|
def _release_held_outbox(self, table_name: str, record_id: str) -> None:
|
|
with session_scope() as db:
|
|
from sqlalchemy import select
|
|
|
|
rows = list(
|
|
db.scalars(
|
|
select(SyncOutbox).where(
|
|
SyncOutbox.enterprise_id == self.enterprise_id,
|
|
SyncOutbox.table_name == table_name,
|
|
SyncOutbox.record_id == record_id,
|
|
SyncOutbox.status == "held",
|
|
)
|
|
)
|
|
)
|
|
for row in rows:
|
|
row.status = "pending"
|