395 lines
12 KiB
Python
395 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.core.crypto import hash_opaque_token
|
|
from app.core.database import session_scope
|
|
from app.modules.sync.models import (
|
|
Enterprise,
|
|
EnterpriseMember,
|
|
FarmHub,
|
|
HubCredential,
|
|
HubPairingSession,
|
|
SyncAppliedEvent,
|
|
SyncConflict,
|
|
SyncCursor,
|
|
SyncEventLog,
|
|
SyncOutbox,
|
|
SyncRecordState,
|
|
UserFarmAccess,
|
|
)
|
|
|
|
|
|
def _detach(db, instance):
|
|
db.refresh(instance)
|
|
db.expunge(instance)
|
|
return instance
|
|
|
|
|
|
def get_enterprise_by_id(enterprise_id: str) -> Enterprise | None:
|
|
with session_scope() as db:
|
|
row = db.get(Enterprise, enterprise_id)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def get_enterprise_by_slug(slug: str) -> Enterprise | None:
|
|
with session_scope() as db:
|
|
row = db.scalar(select(Enterprise).where(Enterprise.slug == slug))
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def create_enterprise(name: str, slug: str) -> Enterprise:
|
|
with session_scope() as db:
|
|
row = Enterprise(name=name, slug=slug, status="active")
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def get_member(user_id: str, enterprise_id: str) -> EnterpriseMember | None:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(EnterpriseMember).where(
|
|
EnterpriseMember.user_id == user_id,
|
|
EnterpriseMember.enterprise_id == enterprise_id,
|
|
)
|
|
)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def add_member(user_id: str, enterprise_id: str, role: str) -> EnterpriseMember:
|
|
with session_scope() as db:
|
|
row = EnterpriseMember(user_id=user_id, enterprise_id=enterprise_id, role=role)
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def list_members(enterprise_id: str) -> list[EnterpriseMember]:
|
|
with session_scope() as db:
|
|
rows = list(
|
|
db.scalars(
|
|
select(EnterpriseMember).where(EnterpriseMember.enterprise_id == enterprise_id)
|
|
)
|
|
)
|
|
return [_detach(db, row) for row in rows]
|
|
|
|
|
|
def set_member_role(user_id: str, enterprise_id: str, role: str) -> EnterpriseMember | None:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(EnterpriseMember).where(
|
|
EnterpriseMember.user_id == user_id,
|
|
EnterpriseMember.enterprise_id == enterprise_id,
|
|
)
|
|
)
|
|
if not row:
|
|
return None
|
|
row.role = role
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def list_farm_access(user_id: str, enterprise_id: str) -> list[str]:
|
|
with session_scope() as db:
|
|
rows = db.scalars(
|
|
select(UserFarmAccess.farm_hub_id)
|
|
.join(FarmHub, FarmHub.id == UserFarmAccess.farm_hub_id)
|
|
.where(UserFarmAccess.user_id == user_id, FarmHub.enterprise_id == enterprise_id)
|
|
).all()
|
|
return list(rows)
|
|
|
|
|
|
def grant_farm_access(user_id: str, farm_hub_id: str) -> UserFarmAccess:
|
|
with session_scope() as db:
|
|
existing = db.scalar(
|
|
select(UserFarmAccess).where(
|
|
UserFarmAccess.user_id == user_id,
|
|
UserFarmAccess.farm_hub_id == farm_hub_id,
|
|
)
|
|
)
|
|
if existing:
|
|
return _detach(db, existing)
|
|
row = UserFarmAccess(user_id=user_id, farm_hub_id=farm_hub_id)
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def get_farm_hub_by_site_id(hub_site_id: str) -> FarmHub | None:
|
|
with session_scope() as db:
|
|
row = db.scalar(select(FarmHub).where(FarmHub.hub_site_id == hub_site_id))
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def get_farm_hub_by_id(farm_hub_id: str) -> FarmHub | None:
|
|
with session_scope() as db:
|
|
row = db.get(FarmHub, farm_hub_id)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def list_farm_hubs(enterprise_id: str) -> list[FarmHub]:
|
|
with session_scope() as db:
|
|
rows = list(db.scalars(select(FarmHub).where(FarmHub.enterprise_id == enterprise_id).order_by(FarmHub.name)))
|
|
return [_detach(db, row) for row in rows]
|
|
|
|
|
|
def create_pairing_session(enterprise_id: str, code: str, code_hash: str, ttl_minutes: int = 15) -> HubPairingSession:
|
|
with session_scope() as db:
|
|
row = HubPairingSession(
|
|
enterprise_id=enterprise_id,
|
|
code=code,
|
|
code_hash=code_hash,
|
|
expires_at=datetime.now(UTC) + timedelta(minutes=ttl_minutes),
|
|
)
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def get_pairing_session_by_code_hash(code_hash: str) -> HubPairingSession | None:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(HubPairingSession).where(
|
|
HubPairingSession.code_hash == code_hash,
|
|
HubPairingSession.confirmed_at.is_(None),
|
|
)
|
|
)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def confirm_pairing_session(session_id: str, farm_hub_id: str) -> None:
|
|
with session_scope() as db:
|
|
row = db.get(HubPairingSession, session_id)
|
|
if row:
|
|
row.confirmed_at = datetime.now(UTC)
|
|
row.farm_hub_id = farm_hub_id
|
|
|
|
|
|
def create_farm_hub(
|
|
enterprise_id: str,
|
|
name: str,
|
|
hub_site_id: str,
|
|
url: str | None = None,
|
|
) -> FarmHub:
|
|
with session_scope() as db:
|
|
row = FarmHub(
|
|
enterprise_id=enterprise_id,
|
|
name=name,
|
|
hub_site_id=hub_site_id,
|
|
url=url,
|
|
status="active",
|
|
)
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def create_hub_credential(farm_hub_id: str, api_key: str) -> HubCredential:
|
|
from app.core.crypto import hash_opaque_token as _hash
|
|
|
|
with session_scope() as db:
|
|
row = HubCredential(farm_hub_id=farm_hub_id, secret_hash=_hash(api_key))
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def verify_hub_credential(hub_site_id: str, api_key: str) -> FarmHub | None:
|
|
from app.core.crypto import hash_opaque_token as _hash
|
|
|
|
key_hash = _hash(api_key)
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(FarmHub)
|
|
.join(HubCredential, HubCredential.farm_hub_id == FarmHub.id)
|
|
.where(
|
|
FarmHub.hub_site_id == hub_site_id,
|
|
HubCredential.secret_hash == key_hash,
|
|
HubCredential.revoked_at.is_(None),
|
|
)
|
|
)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def update_hub_heartbeat(farm_hub_id: str, wesp_version: str | None) -> None:
|
|
with session_scope() as db:
|
|
row = db.get(FarmHub, farm_hub_id)
|
|
if row:
|
|
row.last_seen = datetime.now(UTC)
|
|
if wesp_version:
|
|
row.wesp_version = wesp_version
|
|
|
|
|
|
def next_seq(enterprise_id: str) -> int:
|
|
with session_scope() as db:
|
|
current = db.scalar(
|
|
select(func.max(SyncEventLog.seq)).where(SyncEventLog.enterprise_id == enterprise_id)
|
|
)
|
|
return int(current or 0) + 1
|
|
|
|
|
|
def event_exists(event_id: str) -> bool:
|
|
with session_scope() as db:
|
|
row = db.scalar(select(SyncEventLog.id).where(SyncEventLog.event_id == event_id))
|
|
return row is not None
|
|
|
|
|
|
def append_event_log(**kwargs) -> SyncEventLog:
|
|
with session_scope() as db:
|
|
row = SyncEventLog(**kwargs)
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def applied_event_exists(site_id: str, event_id: str) -> bool:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(SyncAppliedEvent.id).where(
|
|
SyncAppliedEvent.site_id == site_id,
|
|
SyncAppliedEvent.event_id == event_id,
|
|
)
|
|
)
|
|
return row is not None
|
|
|
|
|
|
def mark_applied(enterprise_id: str, site_id: str, event_id: str) -> None:
|
|
with session_scope() as db:
|
|
db.add(
|
|
SyncAppliedEvent(
|
|
enterprise_id=enterprise_id,
|
|
site_id=site_id,
|
|
event_id=event_id,
|
|
)
|
|
)
|
|
|
|
|
|
def get_record_state(enterprise_id: str, table_name: str, record_id: str) -> SyncRecordState | None:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(SyncRecordState).where(
|
|
SyncRecordState.enterprise_id == enterprise_id,
|
|
SyncRecordState.table_name == table_name,
|
|
SyncRecordState.record_id == record_id,
|
|
)
|
|
)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def upsert_record_state(
|
|
enterprise_id: str,
|
|
table_name: str,
|
|
record_id: str,
|
|
agreed_version: int,
|
|
agreed_hash: str,
|
|
last_event_id: str,
|
|
) -> None:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(SyncRecordState).where(
|
|
SyncRecordState.enterprise_id == enterprise_id,
|
|
SyncRecordState.table_name == table_name,
|
|
SyncRecordState.record_id == record_id,
|
|
)
|
|
)
|
|
if row:
|
|
row.agreed_version = agreed_version
|
|
row.agreed_hash = agreed_hash
|
|
row.last_event_id = last_event_id
|
|
row.updated_at = datetime.now(UTC)
|
|
else:
|
|
db.add(
|
|
SyncRecordState(
|
|
enterprise_id=enterprise_id,
|
|
table_name=table_name,
|
|
record_id=record_id,
|
|
agreed_version=agreed_version,
|
|
agreed_hash=agreed_hash,
|
|
last_event_id=last_event_id,
|
|
)
|
|
)
|
|
|
|
|
|
def list_conflicts(enterprise_id: str, status: str = "pending") -> list[SyncConflict]:
|
|
with session_scope() as db:
|
|
rows = list(
|
|
db.scalars(
|
|
select(SyncConflict)
|
|
.where(SyncConflict.enterprise_id == enterprise_id, SyncConflict.status == status)
|
|
.order_by(SyncConflict.created_at.desc())
|
|
)
|
|
)
|
|
return [_detach(db, row) for row in rows]
|
|
|
|
|
|
def get_conflict(conflict_id: str) -> SyncConflict | None:
|
|
with session_scope() as db:
|
|
row = db.get(SyncConflict, conflict_id)
|
|
return _detach(db, row) if row else None
|
|
|
|
|
|
def get_or_create_cursor(enterprise_id: str, farm_hub_id: str, direction: str) -> SyncCursor:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(SyncCursor).where(
|
|
SyncCursor.farm_hub_id == farm_hub_id,
|
|
SyncCursor.direction == direction,
|
|
)
|
|
)
|
|
if row:
|
|
return _detach(db, row)
|
|
row = SyncCursor(
|
|
enterprise_id=enterprise_id,
|
|
farm_hub_id=farm_hub_id,
|
|
direction=direction,
|
|
last_acked_seq=0,
|
|
last_pulled_seq=0,
|
|
)
|
|
db.add(row)
|
|
db.flush()
|
|
return _detach(db, row)
|
|
|
|
|
|
def update_cursor_ack(farm_hub_id: str, direction: str, last_acked_seq: int) -> None:
|
|
with session_scope() as db:
|
|
row = db.scalar(
|
|
select(SyncCursor).where(
|
|
SyncCursor.farm_hub_id == farm_hub_id,
|
|
SyncCursor.direction == direction,
|
|
)
|
|
)
|
|
if row:
|
|
row.last_acked_seq = max(row.last_acked_seq, last_acked_seq)
|
|
row.updated_at = datetime.now(UTC)
|
|
|
|
|
|
def pull_events_since(
|
|
enterprise_id: str,
|
|
cursor: int,
|
|
limit: int,
|
|
*,
|
|
exclude_origin_site_id: str | None = None,
|
|
) -> list[SyncEventLog]:
|
|
with session_scope() as db:
|
|
query = select(SyncEventLog).where(
|
|
SyncEventLog.enterprise_id == enterprise_id,
|
|
SyncEventLog.seq > cursor,
|
|
)
|
|
if exclude_origin_site_id:
|
|
query = query.where(
|
|
(SyncEventLog.origin_site_id.is_(None))
|
|
| (SyncEventLog.origin_site_id != exclude_origin_site_id)
|
|
)
|
|
rows = list(db.scalars(query.order_by(SyncEventLog.seq.asc()).limit(limit)))
|
|
return [_detach(db, row) for row in rows]
|
|
|
|
|
|
def get_farm_hub_site_id(farm_hub_id: str) -> str | None:
|
|
with session_scope() as db:
|
|
row = db.get(FarmHub, farm_hub_id)
|
|
return row.hub_site_id if row else None
|