feat(core): implement services for issue #8
- Add BountyService for room bounty operations (group and personal) - Add TrackingService for tracking bounty operations - Uses RoomStorage and TrackingStorage ports - PermissionError raised when non-creator edits/deletes - ValueError raised when bounty not found in tracking
This commit is contained in:
155
core/services.py
Normal file
155
core/services.py
Normal file
@@ -0,0 +1,155 @@
|
||||
"""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
|
||||
]
|
||||
Reference in New Issue
Block a user