""" ChatView — Scrollable chat message display widget for za-tui. Renders a scrollable list of messages with: - Sender grouping (same sender within 90min = show name once) - Auto-scroll on new messages when at bottom - "New messages ↓" indicator when scrolled up - Scroll-to-top triggers older message loading - Message selection via arrow keys - Reply activation via 'r' key on selected message - Reactive updates from MessageSlice pub/sub References: - openspec/changes/plan-03-chat-view/specs/chat-view/spec.md """ from __future__ import annotations from datetime import datetime, timezone, timedelta from typing import Any, Callable, Optional, TYPE_CHECKING from textual.binding import Binding from textual.widget import Widget from za_tui.state.message_models import RenderedMessage, parse_message from za_tui.state.message_slice import MessageSlice if TYPE_CHECKING: from za_tui.state.contact_cache import ContactCache class SenderGroupTracker: """ Tracks last sender + timestamp to determine if sender name should display. Rules: - First message: always show name - Same sender within 90 minutes: hide name - Different sender: always show name - Same sender after >90min gap: show name again """ GROUPING_GAP = timedelta(minutes=90) def __init__(self) -> None: self._last_sender_id: Optional[str] = None self._last_sender_time: Optional[datetime] = None def should_show_sender(self, sender_id: str, timestamp: datetime) -> bool: """Check if sender name should be displayed for this message.""" if self._last_sender_id is None: return True if sender_id != self._last_sender_id: return True if self._last_sender_time is None: return True gap = timestamp - self._last_sender_time if gap > self.GROUPING_GAP: return True return False def record(self, sender_id: str, timestamp: datetime) -> None: """Record that a message was displayed from this sender.""" self._last_sender_id = sender_id self._last_sender_time = timestamp def reset(self) -> None: """Reset tracking state (e.g., when switching chats).""" self._last_sender_id = None self._last_sender_time = None class ChatView(Widget): """ Scrollable chat message display widget. Renders messages from a MessageSlice with sender grouping, auto-scroll, new-messages indicator, and keyboard-driven selection. Reactively updates when MessageSlice changes via on_change pub/sub. """ can_focus = True # Must be focusable to receive key events DEFAULT_CSS = """ ChatView { width: 100%; height: 1fr; overflow-y: auto; padding: 0 1; } #new-messages-indicator { height: 1; color: $accent; background: $surface; text-align: center; } """ BINDINGS = [ Binding("up", "select_up", "Select up", show=False), Binding("down", "select_down", "Select down", show=False), Binding("r", "activate_reply", "Reply", show=False), Binding("escape", "clear_selection", "Clear selection", show=False), Binding("g", "scroll_bottom", "Scroll to bottom", show=False), Binding("end", "scroll_bottom", "Scroll to bottom", show=False), Binding("shift+g", "scroll_top", "Scroll to top", show=False), Binding("home", "scroll_top", "Scroll to top", show=False), ] def __init__( self, message_slice: MessageSlice, chat_id: str, current_user_id: str = "", contact_cache: Optional["ContactCache"] = None, *, name: str | None = None, id: str | None = None, ) -> None: super().__init__(name=name, id=id) self._slice = message_slice self._chat_id = chat_id self._current_user_id = current_user_id self._contact_cache = contact_cache self._group_tracker = SenderGroupTracker() self._selected_index: int = -1 self._at_bottom: bool = True self._on_load_older: Optional[Callable] = None # New messages indicator tracking self._has_new_messages_indicator: bool = False self._new_message_count: int = 0 # Reply activation self._reply_message: Optional[Any] = None self._on_reply_cb: Optional[Callable] = None # Wire up MessageSlice pub/sub self._slice.on_change(self._on_slice_change) @property def is_at_bottom(self) -> bool: """Whether the view is scrolled to the bottom.""" return self._at_bottom @property def reply_message(self) -> Optional[Any]: """The message currently selected for reply, or None.""" return self._reply_message def _compute_at_bottom(self) -> bool: """Compute whether the view is at the bottom based on scroll position. Falls back to True if the widget hasn't been mounted yet or has no virtual height (content fits in viewport). """ try: if not self.is_mounted: return True max_scroll = self.max_scroll_y if max_scroll <= 0: return True return self.scroll_y >= max_scroll - 1 except Exception: return True def _update_scroll_state(self) -> None: """Update _at_bottom from actual scroll position.""" self._at_bottom = self._compute_at_bottom() if self._at_bottom: # Clear new messages indicator when scrolling to bottom self._has_new_messages_indicator = False self._new_message_count = 0 def _on_slice_change(self, chat_id: str, event_type: str) -> None: """Handle MessageSlice change notification.""" if chat_id != self._chat_id: return if event_type == "add_message": if self._at_bottom: # Auto-scroll: schedule scroll after the next render cycle # so the new content is laid out before we measure scroll position. self.call_after_refresh(self._do_auto_scroll) else: # Show indicator self._has_new_messages_indicator = True self._new_message_count += 1 self.refresh() def _do_auto_scroll(self) -> None: """Actually perform the auto-scroll after render.""" self.scroll_end(animate=False) self._has_new_messages_indicator = False self._new_message_count = 0 def on_new_message(self, message: Any) -> None: """Handle a new message arriving (external callback).""" self._slice.add_message(self._chat_id, message) def add_message(self, message: Any) -> None: """Add a message to the current chat (convenience method).""" self.on_new_message(message) # ─── Render ──────────────────────────────────────────────────── def render(self) -> str: """Render all messages for the current chat.""" raw_messages = self._slice.get_messages(self._chat_id) if not raw_messages: return " No messages" self._group_tracker.reset() lines: list[str] = [] for i, raw in enumerate(raw_messages): # Parse if needed if isinstance(raw, dict): msg = parse_message(raw, self._current_user_id, self._contact_cache) else: msg = raw # Check sender grouping show_sender = False if msg.timestamp and not msg.isSelf: show_sender = self._group_tracker.should_show_sender( msg.senderId, msg.timestamp ) self._group_tracker.record(msg.senderId, msg.timestamp) # Render message line prefix = "→" if i == self._selected_index else " " sender_prefix = f"[dim]{msg.senderName}[/dim]\n " if show_sender else "" status_str = "" if isinstance(raw, dict): st = raw.get("status", "") if st == "sending": status_str = " [sending]" elif st == "failed": err = raw.get("error", "") status_str = f" [failed: {err}]" if err else " [failed]" lines.append(f"{prefix} {sender_prefix}{msg.content}{status_str}") # Add "New messages ↓" indicator if needed if self._has_new_messages_indicator and self._new_message_count > 0: lines.append("") lines.append(f"↓ New messages ({self._new_message_count})") return "\n".join(lines) # ─── Scroll ──────────────────────────────────────────────────── def on_scroll(self, event) -> None: """Handle scroll events to update state.""" self._update_scroll_state() def load_older(self) -> None: """Trigger loading of older messages.""" if self._on_load_older: self._on_load_older() def on_load_older(self, callback: Callable) -> None: """Register callback for load-older events.""" self._on_load_older = callback def action_scroll_bottom(self) -> None: """Scroll to the bottom of the chat.""" self._at_bottom = True self._has_new_messages_indicator = False self._new_message_count = 0 try: self.scroll_end(animate=False) except Exception: logger = __import__("logging").getLogger(__name__) logger.debug("scroll_end failed in action_scroll_bottom", exc_info=True) self.refresh() def action_scroll_top(self) -> None: """Scroll to the top of the chat.""" self._at_bottom = False try: self.scroll_home(animate=False) except Exception: logger = __import__("logging").getLogger(__name__) logger.debug("scroll_home failed in action_scroll_top", exc_info=True) self.refresh() # Trigger load-older if applicable if self._slice.has_more(self._chat_id) and not self._slice.is_loading(self._chat_id): if self._on_load_older: self._on_load_older() # ─── Selection ───────────────────────────────────────────────── # ─── Key overrides ───────────────────────────────────────────── # Override default up/down scroll to do selection instead. # Textual's scrollable widgets capture up/down for scrolling. # We want up/down for message selection when ChatView is focused. def key_down(self, event) -> None: """Handle down arrow — select next message.""" self.action_select_down() event.stop() event.prevent_default() def key_up(self, event) -> None: """Handle up arrow — select previous message.""" self.action_select_up() event.stop() event.prevent_default() def action_select_up(self) -> None: """Move selection up.""" self.move_selection(-1) def action_select_down(self) -> None: """Move selection down.""" self.move_selection(1) def move_selection(self, delta: int) -> None: """Move the message selection by delta.""" count = len(self._slice.get_messages(self._chat_id)) if count == 0: return if self._selected_index == -1: self._selected_index = 0 if delta > 0 else count - 1 else: self._selected_index = max(0, min(self._selected_index + delta, count - 1)) self.refresh() def action_clear_selection(self) -> None: """Clear message selection and reply context.""" self._selected_index = -1 self._reply_message = None self.refresh() # ─── Reply ───────────────────────────────────────────────────── def action_activate_reply(self) -> None: """Activate reply mode for selected message.""" if self._selected_index < 0: return messages = self._slice.get_messages(self._chat_id) if self._selected_index >= len(messages): return msg = messages[self._selected_index] self._reply_message = msg # Update any ReplyBanner in the screen from za_tui.widgets.reply_banner import ReplyBanner banners = self.screen.query(ReplyBanner) if len(banners) > 0: banner = banners.first() content = msg.get("content", "") if isinstance(msg, dict) else getattr(msg, "content", "") sender = msg.get("uidFrom", "") if isinstance(msg, dict) else getattr(msg, "senderName", "") banner._sender = sender banner._reply_text = content banner.visible = True banner.refresh() # Notify via callback if registered if self._on_reply_cb: self._on_reply_cb(msg) def on_reply_activated(self, callback: Callable) -> None: """Register callback for reply activation events.""" self._on_reply_cb = callback # ─── Scroll helpers ──────────────────────────────────────────── def scroll_end(self, animate: bool = False) -> None: """Scroll to the very end (bottom) of the content.""" try: max_y = self.max_scroll_y self.scroll_to(0, max_y, animate=animate) except Exception: logger = __import__("logging").getLogger(__name__) logger.debug("scroll_end failed — widget may not be mounted", exc_info=True) def scroll_home(self, animate: bool = False) -> None: """Scroll to the very beginning (top) of the content.""" try: self.scroll_to(0, 0, animate=animate) except Exception: logger = __import__("logging").getLogger(__name__) logger.debug("scroll_home failed — widget may not be mounted", exc_info=True)