136 lines
4.5 KiB
Python
136 lines
4.5 KiB
Python
"""Business logic services for JIGAIDO bounty tracker.
|
|
|
|
These services orchestrate operations through storage ports.
|
|
No implementation details - pure orchestration layer.
|
|
"""
|
|
|
|
from core.models import Bounty, RoomData, TrackingData, TrackedBounty
|
|
from core.ports import RoomStorage, TrackingStorage
|
|
|
|
|
|
class RoomBountyService:
|
|
"""Service for managing bounties within a room.
|
|
|
|
Orchestrates bounty operations through a RoomStorage port.
|
|
"""
|
|
|
|
def __init__(self, storage: RoomStorage):
|
|
self._storage = storage
|
|
|
|
def create_bounty(
|
|
self,
|
|
room_id: int,
|
|
text: str | None,
|
|
link: str | None,
|
|
due_date_ts: int | None,
|
|
created_by_user_id: int,
|
|
) -> Bounty:
|
|
"""Create a new bounty in a room. Returns the created bounty."""
|
|
room_data = self._storage.load(room_id)
|
|
if room_data is None:
|
|
room_data = RoomData(room_id=room_id, bounties=[], next_id=1)
|
|
|
|
bounty = Bounty(
|
|
id=room_data.next_id,
|
|
text=text,
|
|
link=link,
|
|
due_date_ts=due_date_ts,
|
|
created_at=0,
|
|
created_by_user_id=created_by_user_id,
|
|
)
|
|
|
|
room_data.bounties.append(bounty)
|
|
room_data.next_id += 1
|
|
self._storage.save(room_data)
|
|
|
|
return bounty
|
|
|
|
def get_bounty(self, room_id: int, bounty_id: int) -> Bounty | None:
|
|
"""Get a bounty by ID. Returns None if not found."""
|
|
return self._storage.get_bounty(room_id, bounty_id)
|
|
|
|
def list_bounties(self, room_id: int) -> list[Bounty]:
|
|
"""List all bounties in a room. Returns empty list if room not found."""
|
|
room_data = self._storage.load(room_id)
|
|
if room_data is None:
|
|
return []
|
|
return room_data.bounties.copy()
|
|
|
|
def update_bounty(
|
|
self,
|
|
room_id: int,
|
|
bounty_id: int,
|
|
text: str | None = None,
|
|
link: str | None = None,
|
|
due_date_ts: int | None = None,
|
|
) -> bool:
|
|
"""Update a bounty. Returns False if not found."""
|
|
bounty = self._storage.get_bounty(room_id, bounty_id)
|
|
if bounty is None:
|
|
return False
|
|
|
|
updated = Bounty(
|
|
id=bounty.id,
|
|
text=text if text is not None else bounty.text,
|
|
link=link if link is not None else bounty.link,
|
|
due_date_ts=due_date_ts if due_date_ts is not None else bounty.due_date_ts,
|
|
created_at=bounty.created_at,
|
|
created_by_user_id=bounty.created_by_user_id,
|
|
)
|
|
|
|
self._storage.update_bounty(room_id, updated)
|
|
return True
|
|
|
|
def delete_bounty(self, room_id: int, bounty_id: int) -> bool:
|
|
"""Delete a bounty. Returns False if not found."""
|
|
bounty = self._storage.get_bounty(room_id, bounty_id)
|
|
if bounty is None:
|
|
return False
|
|
|
|
self._storage.delete_bounty(room_id, bounty_id)
|
|
return True
|
|
|
|
|
|
class TrackingService:
|
|
"""Service for managing bounty tracking.
|
|
|
|
Orchestrates tracking operations through a TrackingStorage port.
|
|
"""
|
|
|
|
def __init__(self, storage: TrackingStorage):
|
|
self._storage = storage
|
|
|
|
def track_bounty(self, room_id: int, user_id: int, bounty_id: int) -> bool:
|
|
"""Start tracking a bounty. Returns False if already tracking."""
|
|
tracking_data = self._storage.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=0)
|
|
self._storage.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 False if not tracking."""
|
|
tracking_data = self._storage.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._storage.untrack_bounty(room_id, user_id, bounty_id)
|
|
return True
|
|
|
|
return False
|
|
|
|
def list_tracked(self, room_id: int, user_id: int) -> list[TrackedBounty]:
|
|
"""List all tracked bounties for a user in a room. Returns empty list if not tracking."""
|
|
tracking_data = self._storage.load(room_id, user_id)
|
|
if tracking_data is None:
|
|
return []
|
|
return tracking_data.tracked.copy()
|