"""JSON file storage adapter for JIGAIDO. Implements RoomStorage and TrackingStorage ports using JSON file persistence. Data stored at: - Rooms: ~/.jigaido/data/room/.json - Tracking: ~/.jigaido/data/tracking/_.json """ import json import os import tempfile from pathlib import Path from core.models import Bounty, Category, RoomData, TrackingData, TrackedBounty class JsonFileRoomStorage: """RoomStorage implementation using JSON files. Stores room data at ~/.jigaido/data/room/.json """ def __init__(self, data_dir: Path | None = None): if data_dir is None: data_dir = Path.home() / ".jigaido" / "data" / "room" self._data_dir = data_dir self._data_dir.mkdir(parents=True, exist_ok=True) def _get_file_path(self, room_id: int) -> Path: return self._data_dir / f"{room_id}.json" def _atomic_write(self, path: Path, data: dict) -> None: """Write data atomically using tempfile + rename.""" with tempfile.NamedTemporaryFile( mode="w", dir=self._data_dir, delete=False ) as tmp: json.dump(data, tmp, indent=2) tmp_path = tmp.name os.rename(tmp_path, path) def load(self, room_id: int) -> RoomData | None: """Load room data from JSON file. Returns None if not found.""" file_path = self._get_file_path(room_id) if not file_path.exists(): return None with open(file_path, "r") as f: data = json.load(f) bounties = [ Bounty( id=b["id"], text=b.get("text"), link=b.get("link"), due_date_ts=b.get("due_date_ts"), created_at=b["created_at"], created_by_user_id=b["created_by_user_id"], deleted_at=b.get("deleted_at"), created_by_username=b.get("created_by_username"), category_ids=b.get("category_ids", []), ) for b in data.get("bounties", []) ] categories = [ Category( id=c["id"], name=c["name"], created_at=c["created_at"], deleted_at=c.get("deleted_at"), ) for c in data.get("categories", []) ] return RoomData( room_id=data["room_id"], bounties=bounties, next_id=data["next_id"], timezone=data.get("timezone"), admin_usernames=data.get("admin_usernames", []), categories=categories, ) def save(self, room_data: RoomData) -> None: """Save room data to JSON file.""" data = { "room_id": room_data.room_id, "next_id": room_data.next_id, "timezone": room_data.timezone, "admin_usernames": room_data.admin_usernames or [], "categories": [ { "id": c.id, "name": c.name, "created_at": c.created_at, "deleted_at": c.deleted_at, } for c in room_data.categories ], "bounties": [ { "id": b.id, "text": b.text, "link": b.link, "due_date_ts": b.due_date_ts, "created_at": b.created_at, "created_by_user_id": b.created_by_user_id, "deleted_at": b.deleted_at, "created_by_username": b.created_by_username, "category_ids": b.category_ids, } for b in room_data.bounties ], } self._atomic_write(self._get_file_path(room_data.room_id), data) def add_bounty(self, room_id: int, bounty: Bounty) -> None: """Add a bounty to a room, creating the room if necessary.""" room_data = self.load(room_id) if room_data is None: room_data = RoomData(room_id=room_id, bounties=[], next_id=1) room_data.bounties.append(bounty) if bounty.id >= room_data.next_id: room_data.next_id = bounty.id + 1 self.save(room_data) def update_bounty(self, room_id: int, bounty: Bounty) -> None: """Update an existing bounty in a room.""" room_data = self.load(room_id) if room_data is None: return for i, b in enumerate(room_data.bounties): if b.id == bounty.id: room_data.bounties[i] = bounty break self.save(room_data) def get_bounty(self, room_id: int, bounty_id: int) -> Bounty | None: """Get a specific bounty from a room by ID.""" room_data = self.load(room_id) if room_data is None: return None for b in room_data.bounties: if b.id == bounty_id: return b return None def list_bounties(self, room_id: int) -> list[Bounty]: """List all non-deleted bounties in a room. This is the default method for normal queries - soft-deleted bounties are excluded from results. """ room_data = self.load(room_id) if room_data is None: return [] return [b for b in room_data.bounties if b.deleted_at is None] def list_all_bounties( self, room_id: int, include_deleted: bool = True ) -> list[Bounty]: """List all bounties including or excluding soft-deleted. Args: room_id: The room ID include_deleted: If True, return all bounties including soft-deleted. If False, return only non-deleted bounties. Defaults to True for /recover functionality. """ room_data = self.load(room_id) if room_data is None: return [] if include_deleted: return room_data.bounties return [b for b in room_data.bounties if b.deleted_at is None] class JsonFileTrackingStorage: """TrackingStorage implementation using JSON files. Stores tracking data at ~/.jigaido/data/tracking/_.json """ def __init__(self, tracking_dir: Path | None = None): if tracking_dir is None: tracking_dir = Path.home() / ".jigaido" / "data" / "tracking" self._tracking_dir = tracking_dir self._tracking_dir.mkdir(parents=True, exist_ok=True) def _get_file_path(self, room_id: int, user_id: int) -> Path: return self._tracking_dir / f"{room_id}_{user_id}.json" def _atomic_write(self, path: Path, data: dict) -> None: """Write data atomically using tempfile + rename.""" with tempfile.NamedTemporaryFile( mode="w", dir=self._tracking_dir, delete=False ) as tmp: json.dump(data, tmp, indent=2) tmp_path = tmp.name os.rename(tmp_path, path) def load(self, room_id: int, user_id: int) -> TrackingData | None: """Load tracking data from JSON file. Returns None if not found.""" file_path = self._get_file_path(room_id, user_id) if not file_path.exists(): return None with open(file_path, "r") as f: data = json.load(f) tracked = [ TrackedBounty( bounty_id=t["bounty_id"], created_at=t["created_at"], ) for t in data.get("tracked", []) ] return TrackingData( room_id=data["room_id"], user_id=data["user_id"], tracked=tracked, ) def save(self, tracking_data: TrackingData) -> None: """Save tracking data.""" data = { "room_id": tracking_data.room_id, "user_id": tracking_data.user_id, "tracked": [ { "bounty_id": t.bounty_id, "created_at": t.created_at, } for t in tracking_data.tracked ], } self._atomic_write( self._get_file_path(tracking_data.room_id, tracking_data.user_id), data ) def track_bounty(self, room_id: int, user_id: int, tracked: TrackedBounty) -> None: """Add a bounty to a user's tracking list, creating the tracking entry if needed.""" tracking_data = self.load(room_id, user_id) if tracking_data is None: tracking_data = TrackingData(room_id=room_id, user_id=user_id, tracked=[]) tracking_data.tracked.append(tracked) self.save(tracking_data) def untrack_bounty(self, room_id: int, user_id: int, bounty_id: int) -> None: """Remove a bounty from a user's tracking list.""" tracking_data = self.load(room_id, user_id) if tracking_data is None: return tracking_data.tracked = [ t for t in tracking_data.tracked if t.bounty_id != bounty_id ] self.save(tracking_data)