326 lines
8.8 KiB
Python
326 lines
8.8 KiB
Python
"""Map sharing service for business logic."""
|
|
from typing import List, Optional
|
|
from uuid import UUID
|
|
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException, status
|
|
from datetime import datetime
|
|
import secrets
|
|
|
|
from app.models.map_share import MapShare, MapShareLink, SharePermission
|
|
from app.models.user import User
|
|
from app.models.map import Map
|
|
from app.schemas.map_share import MapShareCreate, MapShareUpdate, MapShareLinkCreate
|
|
from app.services.map_service import get_map_by_id
|
|
|
|
|
|
def generate_share_token() -> str:
|
|
"""Generate a random share token."""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
def create_map_share(
|
|
db: Session,
|
|
map_id: UUID,
|
|
share_data: MapShareCreate,
|
|
current_user: User
|
|
) -> MapShare:
|
|
"""Share a map with a specific user."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can share it"
|
|
)
|
|
|
|
# Look up user by username, email, or UUID
|
|
target_user = None
|
|
user_identifier = share_data.user_identifier.strip()
|
|
|
|
# Try UUID first
|
|
try:
|
|
user_uuid = UUID(user_identifier)
|
|
target_user = db.query(User).filter(User.id == user_uuid).first()
|
|
except ValueError:
|
|
# Not a valid UUID, try username or email
|
|
target_user = db.query(User).filter(
|
|
(User.username == user_identifier) | (User.email == user_identifier)
|
|
).first()
|
|
|
|
if not target_user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"User not found with identifier: {user_identifier}"
|
|
)
|
|
|
|
# Prevent sharing with self
|
|
if target_user.id == current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot share map with yourself"
|
|
)
|
|
|
|
# Check if already shared
|
|
existing_share = db.query(MapShare).filter(
|
|
MapShare.map_id == map_id,
|
|
MapShare.user_id == target_user.id
|
|
).first()
|
|
|
|
if existing_share:
|
|
# Update existing share
|
|
existing_share.permission = share_data.permission
|
|
existing_share.updated_at = datetime.utcnow()
|
|
db.commit()
|
|
db.refresh(existing_share)
|
|
return existing_share
|
|
|
|
# Create new share
|
|
share = MapShare(
|
|
map_id=map_id,
|
|
user_id=target_user.id,
|
|
permission=share_data.permission,
|
|
shared_by=current_user.id
|
|
)
|
|
|
|
db.add(share)
|
|
db.commit()
|
|
db.refresh(share)
|
|
|
|
return share
|
|
|
|
|
|
def get_map_shares(db: Session, map_id: UUID, current_user: User) -> List[MapShare]:
|
|
"""Get all shares for a map."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can view shares"
|
|
)
|
|
|
|
shares = db.query(MapShare).filter(MapShare.map_id == map_id).all()
|
|
return shares
|
|
|
|
|
|
def update_map_share(
|
|
db: Session,
|
|
map_id: UUID,
|
|
share_id: UUID,
|
|
update_data: MapShareUpdate,
|
|
current_user: User
|
|
) -> MapShare:
|
|
"""Update map share permissions."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can update shares"
|
|
)
|
|
|
|
share = db.query(MapShare).filter(
|
|
MapShare.id == share_id,
|
|
MapShare.map_id == map_id
|
|
).first()
|
|
|
|
if not share:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Share not found"
|
|
)
|
|
|
|
share.permission = update_data.permission
|
|
share.updated_at = datetime.utcnow()
|
|
|
|
db.commit()
|
|
db.refresh(share)
|
|
|
|
return share
|
|
|
|
|
|
async def delete_map_share(
|
|
db: Session,
|
|
map_id: UUID,
|
|
share_id: UUID,
|
|
current_user: User
|
|
) -> None:
|
|
"""Revoke map share."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can revoke shares"
|
|
)
|
|
|
|
share = db.query(MapShare).filter(
|
|
MapShare.id == share_id,
|
|
MapShare.map_id == map_id
|
|
).first()
|
|
|
|
if not share:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Share not found"
|
|
)
|
|
|
|
# Get user_id before deleting the share
|
|
revoked_user_id = share.user_id
|
|
|
|
db.delete(share)
|
|
db.commit()
|
|
|
|
# Disconnect the user's WebSocket connections
|
|
from app.websocket.connection_manager import manager
|
|
await manager.disconnect_user(map_id, revoked_user_id)
|
|
|
|
|
|
def create_share_link(
|
|
db: Session,
|
|
map_id: UUID,
|
|
link_data: MapShareLinkCreate,
|
|
current_user: User
|
|
) -> MapShareLink:
|
|
"""Create a public/guest share link."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can create share links"
|
|
)
|
|
|
|
token = generate_share_token()
|
|
|
|
link = MapShareLink(
|
|
map_id=map_id,
|
|
token=token,
|
|
permission=link_data.permission,
|
|
is_active=True,
|
|
created_by=current_user.id,
|
|
expires_at=link_data.expires_at
|
|
)
|
|
|
|
db.add(link)
|
|
db.commit()
|
|
db.refresh(link)
|
|
|
|
return link
|
|
|
|
|
|
def get_share_links(db: Session, map_id: UUID, current_user: User) -> List[MapShareLink]:
|
|
"""Get all share links for a map."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can view share links"
|
|
)
|
|
|
|
links = db.query(MapShareLink).filter(MapShareLink.map_id == map_id).all()
|
|
return links
|
|
|
|
|
|
def delete_share_link(
|
|
db: Session,
|
|
map_id: UUID,
|
|
link_id: UUID,
|
|
current_user: User
|
|
) -> None:
|
|
"""Delete a share link."""
|
|
# Verify user owns the map
|
|
map_obj = get_map_by_id(db, map_id, current_user)
|
|
if map_obj.owner_id != current_user.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only the map owner can delete share links"
|
|
)
|
|
|
|
link = db.query(MapShareLink).filter(
|
|
MapShareLink.id == link_id,
|
|
MapShareLink.map_id == map_id
|
|
).first()
|
|
|
|
if not link:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Share link not found"
|
|
)
|
|
|
|
db.delete(link)
|
|
db.commit()
|
|
|
|
|
|
def get_map_by_share_token(db: Session, token: str) -> tuple[Map, SharePermission]:
|
|
"""Get map by share token (for guest access)."""
|
|
link = db.query(MapShareLink).filter(
|
|
MapShareLink.token == token,
|
|
MapShareLink.is_active == True
|
|
).first()
|
|
|
|
if not link:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invalid or expired share link"
|
|
)
|
|
|
|
# Check if link is expired
|
|
if link.expires_at and link.expires_at < datetime.utcnow():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_410_GONE,
|
|
detail="Share link has expired"
|
|
)
|
|
|
|
map_obj = db.query(Map).filter(Map.id == link.map_id).first()
|
|
if not map_obj:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Map not found"
|
|
)
|
|
|
|
return map_obj, link.permission
|
|
|
|
|
|
def check_map_access(
|
|
db: Session,
|
|
map_id: UUID,
|
|
user: Optional[User] = None,
|
|
token: Optional[str] = None
|
|
) -> tuple[bool, SharePermission]:
|
|
"""
|
|
Check if user has access to a map and return their permission level.
|
|
Returns (has_access, permission_level)
|
|
"""
|
|
map_obj = db.query(Map).filter(Map.id == map_id).first()
|
|
if not map_obj:
|
|
return False, SharePermission.READ
|
|
|
|
# Owner has full edit access
|
|
if user and map_obj.owner_id == user.id:
|
|
return True, SharePermission.EDIT
|
|
|
|
# Check user share
|
|
if user:
|
|
share = db.query(MapShare).filter(
|
|
MapShare.map_id == map_id,
|
|
MapShare.user_id == user.id
|
|
).first()
|
|
if share:
|
|
return True, share.permission
|
|
|
|
# Check share token
|
|
if token:
|
|
try:
|
|
_, permission = get_map_by_share_token(db, token)
|
|
return True, permission
|
|
except HTTPException:
|
|
pass
|
|
|
|
# Check if map is public
|
|
if map_obj.is_default_public:
|
|
return True, SharePermission.READ
|
|
|
|
return False, SharePermission.READ
|