[Phase 1] Task 4: Create core/services.py — Business logic #8

New Issue

shoko · 2026-04-02T18:29:54+02:00

shoko commented

2026-04-02 18:29:54 +02:00

Task 4: Create core/services.py — Business logic

Labels: phase-1, core
Dependency: Task 1 (#5), Task 2 (#6), Task 3 (#7)

Goal

Implement pure business logic services that use storage ports. No Telegram types, no file I/O.

Files to create

core/services.py

Services to implement

"""Pure business logic services for JIGAIDO."""

import time
from typing import Optional
from core.models import (
    Bounty, RoomData, TrackingData
    TrackedBounty, TrackingData
)
from core.ports import GroupStoragePort, PersonalStoragePort, TrackingStoragePort


class BountyService:
    """Service for group bounty operations."""
    
    def __init__(self, storage: GroupStoragePort):
        self._storage = storage
    
    def add_bounty(
        self,
        group_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 group."""
        data = self._storage.load(group_id)
        bounty = Bounty(
            id=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(group_id, bounty)
        return bounty
    
    def list_bounties(self, group_id: int) -> list[Bounty]:
        """List all bounties in a group."""
        data = self._storage.load(group_id)
        return data.bounties
    
    def update_bounty(
        self,
        group_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(group_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.")
        
        return self._storage.update_bounty(
            group_id, bounty_id,
            text=text, link=link, due_date_ts=due_date_ts,
            clear_link=clear_link, clear_due=clear_due,
        )
    
    def delete_bounty(self, group_id: int, bounty_id: int, user_id: int) -> bool:
        """Delete a bounty. Only creator can delete."""
        bounty = self._storage.get_bounty(group_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.")
        
        return self._storage.delete_bounty(group_id, bounty_id)


class BountyService:
    """Service for personal bounty operations (DM mode)."""
    
    def __init__(self, storage: PersonalStoragePort):
        self._storage = storage
    
    def add_bounty(
        self,
        user_id: int,
        text: Optional[str] = None,
        link: Optional[str] = None,
        due_date_ts: Optional[int] = None,
    ) -> Bounty:
        """Add a new personal bounty."""
        data = self._storage.load(user_id)
        bounty = Bounty(
            id=data.next_id,
            text=text,
            link=link,
            due_date_ts=due_date_ts,
            created_at=int(time.time()),
        )
        self._storage.add_bounty(user_id, bounty)
        return bounty
    
    def list_bounties(self, user_id: int) -> list[Bounty]:
        """List all personal bounties."""
        data = self._storage.load(user_id)
        return data.bounties
    
    def update_bounty(
        self,
        user_id: int,
        bounty_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 personal bounty."""
        return self._storage.update_bounty(
            user_id, bounty_id,
            text=text, link=link, due_date_ts=due_date_ts,
            clear_link=clear_link, clear_due=clear_due,
        )
    
    def delete_bounty(self, user_id: int, bounty_id: int) -> bool:
        """Delete a personal bounty."""
        return self._storage.delete_bounty(user_id, bounty_id)


class TrackingService:
    """Service for tracking bounty operations."""
    
    def __init__(self, tracking_storage: TrackingStoragePort, group_storage: GroupStoragePort):
        self._tracking = tracking_storage
        self._group = group_storage
    
    def track_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool:
        """Start tracking a bounty. Returns True if newly tracked."""
        # Verify bounty exists
        bounty = self._group.get_bounty(group_id, bounty_id)
        if not bounty:
            raise ValueError("Bounty not found.")
        return self._tracking.track_bounty(group_id, user_id, bounty_id)
    
    def untrack_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool:
        """Stop tracking a bounty. Returns True if was tracking."""
        return self._tracking.untrack_bounty(group_id, user_id, bounty_id)
    
    def get_tracked_bounties(self, group_id: int, user_id: int) -> list[Bounty]:
        """Get all bounties tracked by a user in a group."""
        tracking = self._tracking.load(group_id, user_id)
        group_data = self._group.load(group_id)
        bounty_map = {b.id: b for b in group_data.bounties}
        
        result = []
        for tracked in tracking.tracked:
            bounty = bounty_map.get(tracked.bounty_id)
            if bounty:
                result.append(bounty)
        return result

Key Design Decisions

Services do NOT import Telegram types (Update, ContextTypes, etc.)
Errors: ValueError for not found, PermissionError for unauthorized
Timestamps generated at service layer, not storage layer

Acceptance Criteria

All services implement business logic without file I/O
Type hints using domain models from Task 1
PermissionError raised when non-creator tries to edit/delete
ValueError raised when bounty not found

## Task 4: Create core/services.py — Business logic **Labels:** phase-1, core **Dependency:** Task 1 (#5), Task 2 (#6), Task 3 (#7) ### Goal Implement pure business logic services that use storage ports. No Telegram types, no file I/O. ### Files to create - `core/services.py` ### Services to implement ```python """Pure business logic services for JIGAIDO.""" import time from typing import Optional from core.models import ( Bounty, RoomData, TrackingData TrackedBounty, TrackingData ) from core.ports import GroupStoragePort, PersonalStoragePort, TrackingStoragePort class BountyService: """Service for group bounty operations.""" def __init__(self, storage: GroupStoragePort): self._storage = storage def add_bounty( self, group_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 group.""" data = self._storage.load(group_id) bounty = Bounty( id=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(group_id, bounty) return bounty def list_bounties(self, group_id: int) -> list[Bounty]: """List all bounties in a group.""" data = self._storage.load(group_id) return data.bounties def update_bounty( self, group_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(group_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.") return self._storage.update_bounty( group_id, bounty_id, text=text, link=link, due_date_ts=due_date_ts, clear_link=clear_link, clear_due=clear_due, ) def delete_bounty(self, group_id: int, bounty_id: int, user_id: int) -> bool: """Delete a bounty. Only creator can delete.""" bounty = self._storage.get_bounty(group_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.") return self._storage.delete_bounty(group_id, bounty_id) class BountyService: """Service for personal bounty operations (DM mode).""" def __init__(self, storage: PersonalStoragePort): self._storage = storage def add_bounty( self, user_id: int, text: Optional[str] = None, link: Optional[str] = None, due_date_ts: Optional[int] = None, ) -> Bounty: """Add a new personal bounty.""" data = self._storage.load(user_id) bounty = Bounty( id=data.next_id, text=text, link=link, due_date_ts=due_date_ts, created_at=int(time.time()), ) self._storage.add_bounty(user_id, bounty) return bounty def list_bounties(self, user_id: int) -> list[Bounty]: """List all personal bounties.""" data = self._storage.load(user_id) return data.bounties def update_bounty( self, user_id: int, bounty_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 personal bounty.""" return self._storage.update_bounty( user_id, bounty_id, text=text, link=link, due_date_ts=due_date_ts, clear_link=clear_link, clear_due=clear_due, ) def delete_bounty(self, user_id: int, bounty_id: int) -> bool: """Delete a personal bounty.""" return self._storage.delete_bounty(user_id, bounty_id) class TrackingService: """Service for tracking bounty operations.""" def __init__(self, tracking_storage: TrackingStoragePort, group_storage: GroupStoragePort): self._tracking = tracking_storage self._group = group_storage def track_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool: """Start tracking a bounty. Returns True if newly tracked.""" # Verify bounty exists bounty = self._group.get_bounty(group_id, bounty_id) if not bounty: raise ValueError("Bounty not found.") return self._tracking.track_bounty(group_id, user_id, bounty_id) def untrack_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool: """Stop tracking a bounty. Returns True if was tracking.""" return self._tracking.untrack_bounty(group_id, user_id, bounty_id) def get_tracked_bounties(self, group_id: int, user_id: int) -> list[Bounty]: """Get all bounties tracked by a user in a group.""" tracking = self._tracking.load(group_id, user_id) group_data = self._group.load(group_id) bounty_map = {b.id: b for b in group_data.bounties} result = [] for tracked in tracking.tracked: bounty = bounty_map.get(tracked.bounty_id) if bounty: result.append(bounty) return result ``` ### Key Design Decisions - Services do NOT import Telegram types (`Update`, `ContextTypes`, etc.) - Errors: `ValueError` for not found, `PermissionError` for unauthorized - Timestamps generated at service layer, not storage layer ### Acceptance Criteria - [ ] All services implement business logic without file I/O - [ ] Type hints using domain models from Task 1 - [ ] `PermissionError` raised when non-creator tries to edit/delete - [ ] `ValueError` raised when bounty not found

shoko referenced this issue

2026-04-02 18:30:59 +02:00

[Phase 1] Task 6: Create CLI (jigaido-cli) #10

shoko referenced this issue

2026-04-02 18:31:15 +02:00

[Phase 1] Task 6: Create CLI (jigaido-cli) #11

shoko referenced this issue

2026-04-02 18:31:31 +02:00

[Phase 1] Task 7: Refactor telegram-bot/commands.py — Thin Telegram wrappers #12

shoko referenced this issue

2026-04-02 18:31:47 +02:00

[Phase 1] Task 7: Refactor telegram-bot/commands.py — Thin Telegram wrappers #13