| 1 | from __future__ import annotations |
| 2 | |
| 3 | import threading |
| 4 | import time |
| 5 | from typing import Literal |
| 6 | |
| 7 | from helpers import kvp |
| 8 | |
| 9 | PinKind = Literal["chat", "task"] |
| 10 | |
| 11 | STORE_KEY = "plugin_pin_to_top" |
| 12 | KINDS: tuple[PinKind, ...] = ("chat", "task") |
| 13 | _lock = threading.RLock() |
| 14 | |
| 15 | |
| 16 | def get_pins() -> dict[PinKind, dict[str, float]]: |
| 17 | """Return normalized persisted pins grouped by sidebar list.""" |
| 18 | with _lock: |
| 19 | return _normalize(kvp.get_persistent(STORE_KEY, {})) |
| 20 | |
| 21 | |
| 22 | def toggle_pin(kind: str, item_id: str) -> tuple[bool, float]: |
| 23 | """Toggle a pin and return its new state and timestamp.""" |
| 24 | normalized_kind = _require_kind(kind) |
| 25 | normalized_id = _require_item_id(item_id) |
| 26 | |
| 27 | with _lock: |
| 28 | pins = _normalize(kvp.get_persistent(STORE_KEY, {})) |
| 29 | kind_pins = pins[normalized_kind] |
| 30 | if normalized_id in kind_pins: |
| 31 | del kind_pins[normalized_id] |
| 32 | timestamp = 0.0 |
| 33 | pinned = False |
| 34 | else: |
| 35 | timestamp = time.time() |
| 36 | kind_pins[normalized_id] = timestamp |
| 37 | pinned = True |
| 38 | |
| 39 | kvp.set_persistent(STORE_KEY, pins) |
| 40 | return pinned, timestamp |
| 41 | |
| 42 | |
| 43 | def _normalize(value: object) -> dict[PinKind, dict[str, float]]: |
| 44 | normalized: dict[PinKind, dict[str, float]] = {"chat": {}, "task": {}} |
| 45 | if not isinstance(value, dict): |
| 46 | return normalized |
| 47 | |
| 48 | for kind in KINDS: |
| 49 | entries = value.get(kind) |
| 50 | if not isinstance(entries, dict): |
| 51 | continue |
| 52 | for item_id, timestamp in entries.items(): |
| 53 | try: |
| 54 | clean_timestamp = float(timestamp) |
| 55 | except (TypeError, ValueError): |
| 56 | continue |
| 57 | clean_id = str(item_id).strip() |
| 58 | if clean_id and clean_timestamp > 0: |
| 59 | normalized[kind][clean_id] = clean_timestamp |
| 60 | |
| 61 | return normalized |
| 62 | |
| 63 | |
| 64 | def _require_kind(kind: str) -> PinKind: |
| 65 | if kind not in KINDS: |
| 66 | raise ValueError("kind must be 'chat' or 'task'") |
| 67 | return kind |
| 68 | |
| 69 | |
| 70 | def _require_item_id(item_id: str) -> str: |
| 71 | normalized = item_id.strip() |
| 72 | if not normalized: |
| 73 | raise ValueError("item_id is required") |
| 74 | if len(normalized) > 512: |
| 75 | raise ValueError("item_id is too long") |
| 76 | return normalized |