Implementacion de sistema distribuido de boton de ayuda

This commit is contained in:
2026-02-18 14:25:24 -06:00
parent bbe74906ec
commit 62315e5b24
20 changed files with 1305 additions and 35 deletions

View File

@@ -0,0 +1,22 @@
import uuid
from datetime import datetime, timezone
from sqlalchemy import Column, String, Text, DateTime
from sqlalchemy.dialects.postgresql import UUID
from core.database import Base
class HelpArticle(Base):
"""
Modelo para los artículos de ayuda (Base de Conocimientos).
Sincronizado entre Servidor Central y Clientes.
"""
__tablename__ = "help_articles"
uuid = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
slug = Column(String(255), unique=True, index=True, nullable=False)
title = Column(String(255), nullable=False)
content = Column(Text, nullable=False)
updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc), nullable=False)
last_editor = Column(String(255), nullable=False)
def __repr__(self):
return f"<HelpArticle(title='{self.title}', slug='{self.slug}')>"

View File

@@ -0,0 +1,116 @@
from datetime import datetime
from typing import List, Optional
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Header, status
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.config import settings
from .schemas import HelpArticleInDB, HelpArticleCreate, HelpArticleUpdate, HelpSyncRequest, HelpSyncResponse
from .services import HelpCenterService
from .tasks import sync_single_article_task
router = APIRouter(prefix="/help-center", tags=["Help Center"])
def verify_sync_token(x_sync_token: str = Header(...)):
if x_sync_token != settings.SYNC_SECRET_TOKEN:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid Sync Token"
)
def trigger_sync_or_broadcast(article_uuid: UUID):
"""
Helper function to handle synchronization logic.
- If we are a Client (CENTRAL_SERVER_URL is set): Trigger upstream sync.
- If we are the Hub (No CENTRAL_SERVER, but SPOKE_URLS set): Trigger broadcast.
"""
# 1. Upstream Sync (Client -> Hub)
if settings.CENTRAL_SERVER_URL:
sync_single_article_task.delay(str(article_uuid))
# 2. Downstream Broadcast (Hub -> Spokes)
# Only if we are the Hub (no upstream) and have spokes configured.
elif not settings.CENTRAL_SERVER_URL and settings.SPOKE_URLS:
from .tasks import broadcast_help_update
# origin_client_uuid is None because this change originated on the Hub itself
broadcast_help_update.delay(str(article_uuid), None)
@router.post("/sync/", response_model=HelpSyncResponse, dependencies=[Depends(verify_sync_token)])
def sync_help_article(sync_data: HelpSyncRequest, db: Session = Depends(get_core_db)):
"""
Endpoint de sincronización inteligente para artículos de ayuda.
Requiere X-Sync-Token en los headers.
"""
result = HelpCenterService.sync_article(db, sync_data)
# Broadcast to other spokes (Hub logic)
import logging
logger = logging.getLogger(__name__)
logger.info(f"DEBUG: Hub Sync Check. CENTRAL_SERVER_URL='{settings.CENTRAL_SERVER_URL}' SPOKE_URLS='{settings.SPOKE_URLS}'")
if not settings.CENTRAL_SERVER_URL and settings.SPOKE_URLS:
# We are the Hub (no central server to push to) and have Spokes configured
from .tasks import broadcast_help_update
logger.info(f"DEBUG: Triggering broadcast for article {sync_data.article_uuid}")
broadcast_help_update.delay(
str(sync_data.article_uuid),
str(sync_data.origin_client_uuid) if sync_data.origin_client_uuid else None
)
else:
logger.info("DEBUG: Broadcast skipped (Condition failed)")
return result
@router.get("/articles/", response_model=List[HelpArticleInDB])
def list_articles(db: Session = Depends(get_core_db)):
"""Lista todos los artículos de ayuda."""
return HelpCenterService.get_all(db)
@router.get("/modifications/", response_model=List[HelpArticleInDB])
def get_modifications(since: datetime, db: Session = Depends(get_core_db)):
"""Obtiene artículos modificados desde la fecha indicada (Polling)."""
return HelpCenterService.get_modifications(db, since)
@router.get("/articles/{article_uuid}/", response_model=HelpArticleInDB)
def get_article(article_uuid: UUID, db: Session = Depends(get_core_db)):
"""Obtiene un artículo por UUID."""
article = HelpCenterService.get_by_uuid(db, article_uuid)
if not article:
raise HTTPException(status_code=404, detail="Article not found")
return article
@router.post("/articles/", response_model=HelpArticleInDB, status_code=status.HTTP_201_CREATED)
def create_article(article: HelpArticleCreate, db: Session = Depends(get_core_db)):
"""Crea un nuevo artículo."""
new_article = HelpCenterService.create(db, article)
trigger_sync_or_broadcast(new_article.uuid)
return new_article
@router.patch("/articles/{article_uuid}/", response_model=HelpArticleInDB)
def update_article(article_uuid: UUID, article_data: HelpArticleUpdate, db: Session = Depends(get_core_db)):
"""Actualiza un artículo."""
article = HelpCenterService.update(db, article_uuid, article_data)
if not article:
raise HTTPException(status_code=404, detail="Article not found")
trigger_sync_or_broadcast(article.uuid)
return article
@router.delete("/articles/{article_uuid}/", status_code=status.HTTP_204_NO_CONTENT)
def delete_article(article_uuid: UUID, db: Session = Depends(get_core_db)):
"""Elimina un artículo."""
if not HelpCenterService.delete(db, article_uuid):
raise HTTPException(status_code=404, detail="Article not found")
# Broadcast or Sync the deletion?
# Current sync logic relies on sending the *content*. Deletion sync is harder because the article is gone.
# For now, let's at least trigger the logic.
# WARNING: sync_single_article_task expects the article to exist to send it.
# If we deleted it locally, sync_single_article_task will fail or send nothing.
# We need a dedicated 'sync_deletion' task or similar.
# Since the user didn't explicitly ask for deletion sync, I will SKIP adding complex deletion sync
# logic right now to avoid breaking things, but I'll add the hook for completeness.
# Actually, better to NOT trigger sync on delete if we don't handle it, to avoid errors in logs.
return None

View File

@@ -0,0 +1,43 @@
from datetime import datetime
from typing import Optional
from uuid import UUID
from pydantic import BaseModel, Field
class HelpArticleBase(BaseModel):
slug: str
title: str
content: str
last_editor: str
class HelpArticleCreate(HelpArticleBase):
pass
class HelpArticleUpdate(BaseModel):
slug: Optional[str] = None
title: Optional[str] = None
content: Optional[str] = None
last_editor: Optional[str] = None
class HelpArticleInDB(HelpArticleBase):
uuid: UUID
updated_at: datetime
class Config:
from_attributes = True
class HelpSyncRequest(BaseModel):
article_uuid: UUID
client_updated_at: datetime
client_content: str
client_title: str
client_slug: str
last_editor: str
origin_client_uuid: Optional[UUID] = None
class HelpSyncResponse(BaseModel):
status: str
server_updated_at: Optional[datetime] = None
server_content: Optional[str] = None
server_title: Optional[str] = None
server_slug: Optional[str] = None
message: str

View File

@@ -0,0 +1,116 @@
from datetime import datetime, timezone
from typing import List, Optional
from uuid import UUID
from sqlalchemy.orm import Session
from .models import HelpArticle
from .schemas import HelpArticleCreate, HelpArticleUpdate, HelpSyncRequest, HelpSyncResponse
class HelpCenterService:
@staticmethod
def get_all(db: Session) -> List[HelpArticle]:
return db.query(HelpArticle).all()
@staticmethod
def get_by_uuid(db: Session, article_uuid: UUID) -> Optional[HelpArticle]:
return db.query(HelpArticle).filter(HelpArticle.uuid == article_uuid).first()
@staticmethod
def get_by_slug(db: Session, slug: str) -> Optional[HelpArticle]:
return db.query(HelpArticle).filter(HelpArticle.slug == slug).first()
@staticmethod
def get_modifications(db: Session, since: datetime) -> List[HelpArticle]:
# Ensure timezone awareness
if since.tzinfo is None:
since = since.replace(tzinfo=timezone.utc)
return db.query(HelpArticle).filter(HelpArticle.updated_at > since).all()
@staticmethod
def create(db: Session, article: HelpArticleCreate) -> HelpArticle:
db_article = HelpArticle(**article.model_dump())
db.add(db_article)
db.commit()
db.refresh(db_article)
return db_article
@staticmethod
def update(db: Session, article_uuid: UUID, article_data: HelpArticleUpdate) -> Optional[HelpArticle]:
db_article = db.query(HelpArticle).filter(HelpArticle.uuid == article_uuid).first()
if not db_article:
return None
update_data = article_data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(db_article, key, value)
db.commit()
db.refresh(db_article)
return db_article
@staticmethod
def delete(db: Session, article_uuid: UUID) -> bool:
db_article = db.query(HelpArticle).filter(HelpArticle.uuid == article_uuid).first()
if not db_article:
return False
db.delete(db_article)
db.commit()
return True
@staticmethod
def sync_article(db: Session, sync_data: HelpSyncRequest) -> HelpSyncResponse:
"""
Lógica de sincronización "Smart Sync" (Last Write Wins).
"""
# Loop Prevention: If the update originated from this client, ignore it.
from core.config import settings
if settings.CLIENT_UUID and sync_data.origin_client_uuid and str(sync_data.origin_client_uuid) == settings.CLIENT_UUID:
return HelpSyncResponse(status="OK", message="Skipped: I am the origin.")
db_article = db.query(HelpArticle).filter(HelpArticle.uuid == sync_data.article_uuid).first()
client_updated_at = sync_data.client_updated_at
if client_updated_at.tzinfo is None:
client_updated_at = client_updated_at.replace(tzinfo=timezone.utc)
if not db_article:
# Caso A: Artículo nuevo desde el cliente
new_article = HelpArticle(
uuid=sync_data.article_uuid,
slug=sync_data.client_slug,
title=sync_data.client_title,
content=sync_data.client_content,
updated_at=client_updated_at,
last_editor=sync_data.last_editor
)
db.add(new_article)
db.commit()
return HelpSyncResponse(status="OK", message="Article created on server.")
server_updated_at = db_article.updated_at
if server_updated_at.tzinfo is None:
server_updated_at = server_updated_at.replace(tzinfo=timezone.utc)
# Caso A: Cliente es más nuevo
if client_updated_at > server_updated_at:
db_article.content = sync_data.client_content
db_article.title = sync_data.client_title
db_article.slug = sync_data.client_slug
db_article.updated_at = client_updated_at
db_article.last_editor = sync_data.last_editor
db.commit()
return HelpSyncResponse(status="OK", message="Server updated with client data.")
# Caso B: Servidor es más nuevo
elif server_updated_at > client_updated_at:
return HelpSyncResponse(
status="UPDATE_REQUIRED",
server_updated_at=server_updated_at,
server_content=db_article.content,
server_title=db_article.title,
server_slug=db_article.slug,
message="Client is outdated. Update required."
)
# Caso C: Iguales
else:
return HelpSyncResponse(status="OK", message="Already in sync.")

View File

@@ -0,0 +1,236 @@
import logging
import httpx
from celery import shared_task
from datetime import datetime, timezone
from core.database import CoreSessionLocal
from core.config import settings
from .models import HelpArticle
from .schemas import HelpSyncRequest, HelpSyncResponse
logger = logging.getLogger(__name__)
@shared_task(name="sync_all_articles_task")
def sync_all_articles_task():
"""
Tarea periódica que recorre todos los artículos locales y los sincroniza con el Central.
Solo se ejecuta si hay un CENTRAL_SERVER_URL configurado (Rol: Cliente/Spoke).
"""
if not settings.CENTRAL_SERVER_URL:
logger.info("Skipping sync: No CENTRAL_SERVER_URL configured (Hub mode).")
return
db = CoreSessionLocal()
try:
articles = db.query(HelpArticle).all()
for article in articles:
sync_single_article(article.uuid)
except Exception as e:
logger.error(f"Error in sync_all_articles_task: {e}")
finally:
db.close()
@shared_task(name="sync_single_article_task")
def sync_single_article_task(article_uuid_str: str):
"""
Sincroniza un único artículo inmediatamente después de una edición local.
"""
sync_single_article(article_uuid_str)
@shared_task(name="broadcast_help_update")
def broadcast_help_update(article_uuid_str: str, origin_client_uuid_str: str = None):
"""
Difunde una actualización de artículo a todos los spokes configurados,
excepto al que originó el cambio (si existe).
"""
if not settings.SPOKE_URLS:
logger.info("No SPOKE_URLS configured. Skipping broadcast.")
return
spokes = [s.strip() for s in settings.SPOKE_URLS.split(",") if s.strip()]
headers = {"X-Sync-Token": settings.SYNC_SECRET_TOKEN}
db = CoreSessionLocal()
try:
article = db.query(HelpArticle).filter(HelpArticle.uuid == article_uuid_str).first()
if not article:
logger.error(f"Article {article_uuid_str} not found for broadcast.")
return
sync_payload = HelpSyncRequest(
article_uuid=article.uuid,
client_updated_at=article.updated_at,
client_content=article.content,
client_title=article.title,
client_slug=article.slug,
last_editor=article.last_editor,
origin_client_uuid=UUID(origin_client_uuid_str) if origin_client_uuid_str else None
).model_dump(mode='json')
with httpx.Client() as client:
for spoke_url in spokes:
# Loop Prevention: Skip if the spoke is the origin
# NOTE: This assumes SPOKE_URLS contains the unique Sync URL of the client.
# A more robust check would be comparing CLIENT_UUID if we knew the URL->UUID mapping ahead of time.
# For this implementation, we rely on the `origin_client_uuid` being passed in the payload
# and the receiving client checking it against its own settings.CLIENT_UUID.
try:
logger.info(f"Broadcasting update to {spoke_url}")
response = client.post(
spoke_url,
json=sync_payload,
headers=headers,
timeout=5.0
)
if response.status_code != 200:
logger.warning(f"Broadcast to {spoke_url} failed: {response.status_code}")
except Exception as e:
logger.error(f"Error broadcasting to {spoke_url}: {e}")
except Exception as e:
logger.error(f"Broadcast error: {e}")
finally:
db.close()
def sync_single_article(article_uuid):
"""
Lógica compartida para sincronizar un artículo con el servidor central.
"""
logger.info(f"DEBUG: Syncing article {article_uuid}. CENTRAL_SERVER_URL='{settings.CENTRAL_SERVER_URL}' (Type: {type(settings.CENTRAL_SERVER_URL)})")
if not settings.CENTRAL_SERVER_URL or settings.CENTRAL_SERVER_URL == '""':
# Enhanced check to catch literal empty quotes if they slip through
return
db = CoreSessionLocal()
try:
article = db.query(HelpArticle).filter(HelpArticle.uuid == article_uuid).first()
if not article:
return
sync_data = HelpSyncRequest(
article_uuid=article.uuid,
client_updated_at=article.updated_at,
client_content=article.content,
client_title=article.title,
client_slug=article.slug,
last_editor=article.last_editor,
origin_client_uuid=UUID(settings.CLIENT_UUID) if settings.CLIENT_UUID else None
)
headers = {"X-Sync-Token": settings.SYNC_SECRET_TOKEN}
with httpx.Client() as client:
response = client.post(
settings.CENTRAL_SERVER_URL,
json=sync_data.model_dump(mode='json'),
headers=headers,
timeout=10.0
)
if response.status_code == 200:
result = HelpSyncResponse(**response.json())
if result.status == "UPDATE_REQUIRED":
# El servidor tiene una versión más nueva, actualizamos localmente
article.content = result.server_content
article.title = result.server_title
article.slug = result.server_slug
article.updated_at = result.server_updated_at
db.commit()
logger.info(f"Article {article.uuid} updated from server.")
else:
logger.info(f"Article {article.uuid} sync OK: {result.message}")
else:
logger.error(f"Sync failed for article {article.uuid}: {response.status_code} - {response.text}")
except Exception as e:
logger.error(f"Error syncing article {article.uuid}: {e}")
finally:
db.close()
from sqlalchemy import func
@shared_task(name="sync_from_hub_task")
def sync_from_hub_task():
"""
Tarea de POLLING que el Cliente ejecuta periódicamente.
Consulta al Hub (CENTRAL_SERVER_URL) por artículos modificados desde
la última actualización local.
"""
if not settings.CENTRAL_SERVER_URL:
return
db = CoreSessionLocal()
try:
# 1. Obtener la fecha de la última actualización local
last_local_update = db.query(func.max(HelpArticle.updated_at)).scalar()
if not last_local_update:
# Si no hay datos, traer todo desde el principio de los tiempos
last_local_update = datetime(2000, 1, 1, tzinfo=timezone.utc)
# Asegurar timezone awareness
if last_local_update.tzinfo is None:
last_local_update = last_local_update.replace(tzinfo=timezone.utc)
logger.info(f"Polling Hub for updates since {last_local_update}")
# 2. Consultar al Hub
headers = {"X-Sync-Token": settings.SYNC_SECRET_TOKEN}
# CENTRAL_SERVER_URL es ".../help-center/sync/"
# Queremos ".../help-center/modifications/"
hub_url = settings.CENTRAL_SERVER_URL.replace("/sync/", "/modifications/")
with httpx.Client() as client:
response = client.get(
hub_url,
params={"since": last_local_update.isoformat()},
headers=headers,
timeout=10.0
)
if response.status_code == 200:
articles_data = response.json()
if not articles_data:
logger.info("No updates found.")
return
logger.info(f"Found {len(articles_data)} updates from Hub. Applying...")
# 3. Aplicar actualizaciones
for art_data in articles_data:
# Convertir string a datetime
server_updated_at = datetime.fromisoformat(art_data['updated_at'])
# Logic similar to sync_article but simpler (Force Update from Hub)
# We assume Hub is Truth in this Polling flow
local_article = db.query(HelpArticle).filter(HelpArticle.uuid == art_data['uuid']).first()
if not local_article:
new_article = HelpArticle(
uuid=art_data['uuid'],
slug=art_data['slug'],
title=art_data['title'],
content=art_data['content'],
updated_at=server_updated_at,
last_editor=art_data['last_editor']
)
db.add(new_article)
else:
if server_updated_at > local_article.updated_at.replace(tzinfo=timezone.utc):
local_article.slug = art_data['slug']
local_article.title = art_data['title']
local_article.content = art_data['content']
local_article.updated_at = server_updated_at
local_article.last_editor = art_data['last_editor']
db.commit()
logger.info("Polling sync completed successfully.")
else:
logger.error(f"Polling failed: {response.status_code} - {response.text}")
except Exception as e:
logger.error(f"Error in sync_from_hub_task: {e}")
finally:
db.close()

View File

@@ -5,6 +5,7 @@ from .tenants.routes import router as tenants_router
from .user_tenant.routes import router as user_tenant_router
from .users.routes import router as users_router
from .dashboard.routes import router as dashboard_router
from .help_center.routes import router as help_center_router
from fastapi import APIRouter
router = APIRouter()
@@ -16,3 +17,4 @@ router.include_router(users_router, prefix="/core", tags=["core / users"])
router.include_router(licenses_router, prefix="/core", tags=["core / licenses"])
router.include_router(permissions_router, prefix="/core", tags=["core / permissions"])
router.include_router(dashboard_router, prefix="/core", tags=["core / dashboard"])
router.include_router(help_center_router, prefix="/core", tags=["core / help-center"])