"""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" ) # Check if user exists target_user = db.query(User).filter(User.id == share_data.user_id).first() if not target_user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Check if already shared existing_share = db.query(MapShare).filter( MapShare.map_id == map_id, MapShare.user_id == share_data.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=share_data.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 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" ) db.delete(share) db.commit() 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