"""Pure business logic services for JIGAIDO.""" import time from typing import Optional from core.models import Bounty, RoomData, TrackedBounty, TrackingData from core.ports import RoomStorage, TrackingStorage class BountyService: """Service for bounty operations in a room. A room is identified by room_id: - Negative room_id: Telegram group (e.g., -1001) - Positive room_id: DM/personal context (user's Telegram ID) This service handles both group and personal bounties through room_id. """ def __init__(self, storage: RoomStorage): self._storage = storage def add_bounty( self, room_id: int, user_id: int, text: Optional[str] = None, link: Optional[str] = None, due_date_ts: Optional[int] = None, ) -> Bounty: """Add a new bounty to the room.""" room_data = self._storage.load(room_id) if room_data is None: room_data = RoomData(room_id=room_id, bounties=[], next_id=1) else: room_data.next_id += 1 bounty = Bounty( id=room_data.next_id, created_by_user_id=user_id, text=text, link=link, due_date_ts=due_date_ts, created_at=int(time.time()), ) self._storage.add_bounty(room_id, bounty) return bounty def list_bounties(self, room_id: int) -> list[Bounty]: """List all bounties in a room.""" room_data = self._storage.load(room_id) if room_data is None: return [] return room_data.bounties def get_bounty(self, room_id: int, bounty_id: int) -> Bounty | None: """Get a specific bounty by ID.""" return self._storage.get_bounty(room_id, bounty_id) def update_bounty( self, room_id: int, bounty_id: int, user_id: int, text: Optional[str] = None, link: Optional[str] = None, due_date_ts: Optional[int] = None, clear_link: bool = False, clear_due: bool = False, ) -> bool: """Update a bounty. Only creator can update.""" bounty = self._storage.get_bounty(room_id, bounty_id) if not bounty: return False if bounty.created_by_user_id != user_id: raise PermissionError("Only the creator can edit this bounty.") updated = Bounty( id=bounty.id, created_by_user_id=bounty.created_by_user_id, text=text if text is not None else bounty.text, link=None if clear_link else (link if link is not None else bounty.link), due_date_ts=None if clear_due else (due_date_ts if due_date_ts is not None else bounty.due_date_ts), created_at=bounty.created_at, ) self._storage.update_bounty(room_id, updated) return True def delete_bounty(self, room_id: int, bounty_id: int, user_id: int) -> bool: """Delete a bounty. Only creator can delete.""" bounty = self._storage.get_bounty(room_id, bounty_id) if not bounty: return False if bounty.created_by_user_id != user_id: raise PermissionError("Only the creator can delete this bounty.") self._storage.delete_bounty(room_id, bounty_id) return True class TrackingService: """Service for tracking bounty operations.""" def __init__(self, tracking_storage: TrackingStorage, room_storage: RoomStorage): self._tracking = tracking_storage self._room = room_storage def track_bounty(self, room_id: int, user_id: int, bounty_id: int) -> bool: """Start tracking a bounty. Returns True if newly tracked.""" bounty = self._room.get_bounty(room_id, bounty_id) if not bounty: raise ValueError("Bounty not found.") tracking_data = self._tracking.load(room_id, user_id) if tracking_data is None: tracking_data = TrackingData(room_id=room_id, user_id=user_id, tracked=[]) for tracked in tracking_data.tracked: if tracked.bounty_id == bounty_id: return False tracked = TrackedBounty(bounty_id=bounty_id, created_at=int(time.time())) self._tracking.track_bounty(room_id, user_id, tracked) return True def untrack_bounty(self, room_id: int, user_id: int, bounty_id: int) -> bool: """Stop tracking a bounty. Returns True if was tracking.""" tracking_data = self._tracking.load(room_id, user_id) if tracking_data is None: return False for tracked in tracking_data.tracked: if tracked.bounty_id == bounty_id: self._tracking.untrack_bounty(room_id, user_id, bounty_id) return True return False def get_tracked_bounties(self, room_id: int, user_id: int) -> list[Bounty]: """Get all bounties tracked by a user in a room.""" tracking_data = self._tracking.load(room_id, user_id) if tracking_data is None: return [] room_data = self._room.load(room_id) if room_data is None: return [] bounty_map = {b.id: b for b in room_data.bounties} return [ bounty_map[t.bounty_id] for t in tracking_data.tracked if t.bounty_id in bounty_map ]