hermes-chatto-plugin/vendor/chattolib/types.py

"""Dataclasses and enums mirroring Chatto's Connect API messages.

Enum ``value`` strings match the proto JSON encoding used on the wire
(e.g. ``PRESENCE_STATUS_ONLINE``). Parsers below accept both the fully-qualified
form and the short/legacy tail (``ONLINE``) for robustness against older
snapshots or clients that construct short enum strings by hand.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum, StrEnum
from typing import Any

# --- Enums ---------------------------------------------------------------


def _parse_enum(cls: type[Enum], raw: Any, default: Enum) -> Enum:
    """Accept either the full ``FOO_BAR_BAZ`` form or its tail (``BAZ``)."""
    if raw is None or raw == "":
        return default
    if isinstance(raw, cls):
        return raw
    try:
        return cls(raw)
    except ValueError:
        pass
    # Tolerate integer values from proto JSON alternate encoding.
    if isinstance(raw, int):
        for member in cls:
            if getattr(member, "_proto_index", None) == raw:
                return member
    text = str(raw)
    for member in cls:
        value = member.value if isinstance(member.value, str) else ""
        if value.endswith("_" + text) or value == text:
            return member
    return default


class PresenceStatus(StrEnum):
    UNSPECIFIED = "PRESENCE_STATUS_UNSPECIFIED"
    ONLINE = "PRESENCE_STATUS_ONLINE"
    AWAY = "PRESENCE_STATUS_AWAY"
    DO_NOT_DISTURB = "PRESENCE_STATUS_DO_NOT_DISTURB"
    OFFLINE = "PRESENCE_STATUS_OFFLINE"


class TimeFormat(StrEnum):
    UNSPECIFIED = "TIME_FORMAT_UNSPECIFIED"
    AUTO = "TIME_FORMAT_AUTO"
    HOUR_12 = "TIME_FORMAT_12_HOUR"
    HOUR_24 = "TIME_FORMAT_24_HOUR"


class RoomKind(StrEnum):
    UNSPECIFIED = "ROOM_KIND_UNSPECIFIED"
    CHANNEL = "ROOM_KIND_CHANNEL"
    DM = "ROOM_KIND_DM"


class RoomDirectoryScope(StrEnum):
    UNSPECIFIED = "ROOM_DIRECTORY_SCOPE_UNSPECIFIED"
    ALL = "ROOM_DIRECTORY_SCOPE_ALL"
    CHANNELS = "ROOM_DIRECTORY_SCOPE_CHANNELS"
    DMS = "ROOM_DIRECTORY_SCOPE_DMS"


class NotificationLevel(StrEnum):
    UNSPECIFIED = "NOTIFICATION_LEVEL_UNSPECIFIED"
    DEFAULT = "NOTIFICATION_LEVEL_DEFAULT"
    MUTED = "NOTIFICATION_LEVEL_MUTED"
    NORMAL = "NOTIFICATION_LEVEL_NORMAL"
    ALL_MESSAGES = "NOTIFICATION_LEVEL_ALL_MESSAGES"


class CallEventSource(StrEnum):
    UNSPECIFIED = "REALTIME_CALL_EVENT_SOURCE_UNSPECIFIED"
    USER = "REALTIME_CALL_EVENT_SOURCE_USER"
    LIVEKIT = "REALTIME_CALL_EVENT_SOURCE_LIVEKIT"
    RECONCILIATION = "REALTIME_CALL_EVENT_SOURCE_RECONCILIATION"


class ImageFitMode(StrEnum):
    UNSPECIFIED = "IMAGE_FIT_MODE_UNSPECIFIED"
    CONTAIN = "IMAGE_FIT_MODE_CONTAIN"
    COVER = "IMAGE_FIT_MODE_COVER"


class VideoProcessingStatus(StrEnum):
    UNSPECIFIED = "MESSAGE_VIDEO_PROCESSING_STATUS_UNSPECIFIED"
    PROCESSING = "MESSAGE_VIDEO_PROCESSING_STATUS_PROCESSING"
    COMPLETED = "MESSAGE_VIDEO_PROCESSING_STATUS_COMPLETED"
    FAILED = "MESSAGE_VIDEO_PROCESSING_STATUS_FAILED"


class AssetUploadStatus(StrEnum):
    UNSPECIFIED = "ASSET_UPLOAD_STATUS_UNSPECIFIED"
    OPEN = "ASSET_UPLOAD_STATUS_OPEN"
    COMPLETED = "ASSET_UPLOAD_STATUS_COMPLETED"
    CANCELLED = "ASSET_UPLOAD_STATUS_CANCELLED"


class AdminRoomLayoutItemKind(StrEnum):
    UNSPECIFIED = "ADMIN_ROOM_LAYOUT_ITEM_KIND_UNSPECIFIED"
    ROOM = "ADMIN_ROOM_LAYOUT_ITEM_KIND_ROOM"
    SIDEBAR_LINK = "ADMIN_ROOM_LAYOUT_ITEM_KIND_SIDEBAR_LINK"


class ExternalIdentityFlowKind(StrEnum):
    UNSPECIFIED = "EXTERNAL_IDENTITY_FLOW_KIND_UNSPECIFIED"
    CREATE_ACCOUNT = "EXTERNAL_IDENTITY_FLOW_KIND_CREATE_ACCOUNT"
    LINK_ACCOUNT = "EXTERNAL_IDENTITY_FLOW_KIND_LINK_ACCOUNT"


class PermissionDecision(StrEnum):
    UNSPECIFIED = "PERMISSION_DECISION_UNSPECIFIED"
    ALLOW = "PERMISSION_DECISION_ALLOW"
    DENY = "PERMISSION_DECISION_DENY"
    NONE = "PERMISSION_DECISION_NONE"


class PermissionScopeKind(StrEnum):
    UNSPECIFIED = "PERMISSION_SCOPE_KIND_UNSPECIFIED"
    SERVER = "PERMISSION_SCOPE_KIND_SERVER"
    GROUP = "PERMISSION_SCOPE_KIND_GROUP"
    ROOM = "PERMISSION_SCOPE_KIND_ROOM"


# --- Time helpers --------------------------------------------------------


def parse_datetime(value: Any) -> datetime | None:
    """Parse an RFC 3339 timestamp string (proto JSON encoding for Timestamp)."""
    if not value:
        return None
    if isinstance(value, datetime):
        return value
    text = str(value)
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        return datetime.fromisoformat(text)
    except ValueError:
        return None


def format_datetime(value: datetime) -> str:
    """Format a ``datetime`` as an RFC 3339 UTC timestamp (``...Z``)."""
    if value.tzinfo is None:
        value = value.replace(tzinfo=UTC)
    return value.astimezone(UTC).isoformat().replace("+00:00", "Z")


# --- Core resource dataclasses ------------------------------------------


@dataclass
class CustomUserStatus:
    emoji: str
    text: str
    expires_at: datetime | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> CustomUserStatus | None:
        if not data:
            return None
        return cls(
            emoji=data.get("emoji", ""),
            text=data.get("text", ""),
            expires_at=parse_datetime(data.get("expiresAt")),
        )


@dataclass
class User:
    id: str
    login: str
    display_name: str
    presence_status: PresenceStatus = PresenceStatus.UNSPECIFIED
    avatar_url: str | None = None
    deleted: bool = False
    custom_status: CustomUserStatus | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> User | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            login=data.get("login", ""),
            display_name=data.get("displayName", ""),
            presence_status=_parse_enum(
                PresenceStatus, data.get("presenceStatus"), PresenceStatus.UNSPECIFIED
            ),  # type: ignore[arg-type]
            avatar_url=data.get("avatarUrl"),
            deleted=bool(data.get("deleted", False)),
            custom_status=CustomUserStatus.parse(data.get("customStatus")),
        )


@dataclass
class UserSettings:
    timezone: str | None = None
    time_format: TimeFormat = TimeFormat.UNSPECIFIED

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> UserSettings:
        data = data or {}
        return cls(
            timezone=data.get("timezone"),
            time_format=_parse_enum(
                TimeFormat, data.get("timeFormat"), TimeFormat.UNSPECIFIED
            ),  # type: ignore[arg-type]
        )


@dataclass
class ViewerUser:
    profile: User | None
    settings: UserSettings
    has_verified_email: bool = False
    viewer_can_delete_account: bool = False
    last_login_change: datetime | None = None
    has_password: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ViewerUser | None:
        if not data:
            return None
        return cls(
            profile=User.parse(data.get("profile")),
            settings=UserSettings.parse(data.get("settings")),
            has_verified_email=bool(data.get("hasVerifiedEmail", False)),
            viewer_can_delete_account=bool(data.get("viewerCanDeleteAccount", False)),
            last_login_change=parse_datetime(data.get("lastLoginChange")),
            has_password=bool(data.get("hasPassword", False)),
        )


@dataclass
class DirectoryMember:
    user: User | None
    roles: list[str] = field(default_factory=list)
    created_at: datetime | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> DirectoryMember | None:
        if not data:
            return None
        return cls(
            user=User.parse(data.get("user")),
            roles=list(data.get("roles") or []),
            created_at=parse_datetime(data.get("createdAt")),
        )


@dataclass
class Page:
    total_count: int = 0
    has_more: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> Page:
        data = data or {}
        return cls(
            total_count=int(data.get("totalCount", 0)),
            has_more=bool(data.get("hasMore", False)),
        )


@dataclass
class ServerProfile:
    name: str
    version: str = ""
    logo_url: str | None = None
    banner_url: str | None = None
    welcome_message: str | None = None
    description: str | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ServerProfile:
        data = data or {}
        return cls(
            name=data.get("name", ""),
            version=data.get("version", ""),
            logo_url=data.get("logoUrl"),
            banner_url=data.get("bannerUrl"),
            welcome_message=data.get("welcomeMessage"),
            description=data.get("description"),
        )


@dataclass
class ProviderMetadata:
    id: str
    type: str
    label: str
    login_url: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any]) -> ProviderMetadata:
        return cls(
            id=data.get("id", ""),
            type=data.get("type", ""),
            label=data.get("label", ""),
            login_url=data.get("loginUrl", ""),
        )


@dataclass
class ServerLogin:
    direct_registration_enabled: bool = False
    providers: list[ProviderMetadata] = field(default_factory=list)
    authorize_url: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ServerLogin:
        data = data or {}
        return cls(
            direct_registration_enabled=bool(data.get("directRegistrationEnabled", False)),
            providers=[ProviderMetadata.parse(p) for p in data.get("providers") or []],
            authorize_url=data.get("authorizeUrl", ""),
        )


@dataclass
class ServerRuntimeConfig:
    push_notifications_enabled: bool = False
    vapid_public_key: str | None = None
    livekit_url: str | None = None
    video_processing_enabled: bool = False
    max_upload_size: int = 0
    max_video_upload_size: int = 0
    message_edit_window_seconds: int = 0

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ServerRuntimeConfig:
        data = data or {}
        return cls(
            push_notifications_enabled=bool(data.get("pushNotificationsEnabled", False)),
            vapid_public_key=data.get("vapidPublicKey"),
            livekit_url=data.get("livekitUrl"),
            video_processing_enabled=bool(data.get("videoProcessingEnabled", False)),
            max_upload_size=int(data.get("maxUploadSize", 0)),
            max_video_upload_size=int(data.get("maxVideoUploadSize", 0)),
            message_edit_window_seconds=int(data.get("messageEditWindowSeconds", 0)),
        )


@dataclass
class Room:
    id: str
    kind: RoomKind = RoomKind.UNSPECIFIED
    name: str = ""
    description: str = ""
    archived: bool = False
    group_id: str = ""
    universal: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> Room | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            kind=_parse_enum(RoomKind, data.get("kind"), RoomKind.UNSPECIFIED),  # type: ignore[arg-type]
            name=data.get("name", ""),
            description=data.get("description", ""),
            archived=bool(data.get("archived", False)),
            group_id=data.get("groupId", ""),
            universal=bool(data.get("universal", False)),
        )


@dataclass
class RoomSummary:
    id: str
    kind: RoomKind = RoomKind.UNSPECIFIED
    name: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RoomSummary | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            kind=_parse_enum(RoomKind, data.get("kind"), RoomKind.UNSPECIFIED),  # type: ignore[arg-type]
            name=data.get("name", ""),
        )


@dataclass
class RoomViewerState:
    is_member: bool = False
    has_unread: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RoomViewerState:
        data = data or {}
        return cls(
            is_member=bool(data.get("isMember", False)),
            has_unread=bool(data.get("hasUnread", False)),
        )


@dataclass
class RoomWithViewerState:
    room: Room | None
    viewer_state: RoomViewerState

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RoomWithViewerState | None:
        if not data:
            return None
        return cls(
            room=Room.parse(data.get("room")),
            viewer_state=RoomViewerState.parse(data.get("viewerState")),
        )


@dataclass
class SidebarLink:
    id: str
    label: str
    url: str

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> SidebarLink | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            label=data.get("label", ""),
            url=data.get("url", ""),
        )


@dataclass
class RoomGroup:
    id: str
    name: str
    description: str = ""
    rooms: list[RoomWithViewerState] = field(default_factory=list)
    sidebar_links: list[SidebarLink] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RoomGroup | None:
        if not data:
            return None
        rooms: list[RoomWithViewerState] = []
        links: list[SidebarLink] = []
        for item in data.get("items") or []:
            room = RoomWithViewerState.parse(item.get("room"))
            if room:
                rooms.append(room)
            sl = item.get("sidebarLink")
            if sl:
                links.append(
                    SidebarLink(
                        id=sl.get("id", ""),
                        label=sl.get("label", ""),
                        url=sl.get("url", ""),
                    )
                )
        return cls(
            id=data.get("id", ""),
            name=data.get("name", ""),
            description=data.get("description", ""),
            rooms=rooms,
            sidebar_links=links,
        )


@dataclass
class RoomBan:
    id: str
    room_id: str
    user_id: str
    moderator_id: str = ""
    reason: str = ""
    created_at: datetime | None = None
    expires_at: datetime | None = None
    room: Room | None = None
    user: DirectoryMember | None = None
    moderator: DirectoryMember | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RoomBan | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            room_id=data.get("roomId", ""),
            user_id=data.get("userId", ""),
            moderator_id=data.get("moderatorId", ""),
            reason=data.get("reason", ""),
            created_at=parse_datetime(data.get("createdAt")),
            expires_at=parse_datetime(data.get("expiresAt")),
            room=Room.parse(data.get("room")),
            user=DirectoryMember.parse(data.get("user")),
            moderator=DirectoryMember.parse(data.get("moderator")),
        )


@dataclass
class AssetUrl:
    url: str
    expires_at: datetime | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AssetUrl | None:
        if not data:
            return None
        return cls(
            url=data.get("url", ""),
            expires_at=parse_datetime(data.get("expiresAt")),
        )


@dataclass
class VideoVariant:
    quality: str
    width: int
    height: int
    size: int
    asset_url: AssetUrl | None

    @classmethod
    def parse(cls, data: dict[str, Any]) -> VideoVariant:
        return cls(
            quality=data.get("quality", ""),
            width=int(data.get("width", 0)),
            height=int(data.get("height", 0)),
            size=int(data.get("size", 0)),
            asset_url=AssetUrl.parse(data.get("assetUrl")),
        )


@dataclass
class VideoProcessing:
    status: VideoProcessingStatus = VideoProcessingStatus.UNSPECIFIED
    duration_ms: int = 0
    width: int = 0
    height: int = 0
    source_available: bool = False
    reason_code: str = ""
    thumbnail_asset_url: AssetUrl | None = None
    variants: list[VideoVariant] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> VideoProcessing | None:
        if not data:
            return None
        return cls(
            status=_parse_enum(
                VideoProcessingStatus,
                data.get("status"),
                VideoProcessingStatus.UNSPECIFIED,
            ),  # type: ignore[arg-type]
            duration_ms=int(data.get("durationMs", 0)),
            width=int(data.get("width", 0)),
            height=int(data.get("height", 0)),
            source_available=bool(data.get("sourceAvailable", False)),
            reason_code=data.get("reasonCode", ""),
            thumbnail_asset_url=AssetUrl.parse(data.get("thumbnailAssetUrl")),
            variants=[VideoVariant.parse(v) for v in data.get("variants") or []],
        )


@dataclass
class MessageAttachment:
    id: str
    filename: str
    content_type: str
    width: int = 0
    height: int = 0
    asset_url: AssetUrl | None = None
    thumbnail_asset_url: AssetUrl | None = None
    video_processing: VideoProcessing | None = None

    @classmethod
    def parse(cls, data: dict[str, Any]) -> MessageAttachment:
        return cls(
            id=data.get("id", ""),
            filename=data.get("filename", ""),
            content_type=data.get("contentType", ""),
            width=int(data.get("width", 0)),
            height=int(data.get("height", 0)),
            asset_url=AssetUrl.parse(data.get("assetUrl")),
            thumbnail_asset_url=AssetUrl.parse(data.get("thumbnailAssetUrl")),
            video_processing=VideoProcessing.parse(data.get("videoProcessing")),
        )


@dataclass
class LinkPreview:
    url: str
    title: str | None = None
    description: str | None = None
    image_url: str | None = None
    image_asset_id: str | None = None
    site_name: str | None = None
    embed_type: str | None = None
    embed_id: str | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> LinkPreview | None:
        if not data:
            return None
        return cls(
            url=data.get("url", ""),
            title=data.get("title"),
            description=data.get("description"),
            image_url=data.get("imageUrl"),
            image_asset_id=data.get("imageAssetId"),
            site_name=data.get("siteName"),
            embed_type=data.get("embedType"),
            embed_id=data.get("embedId"),
        )


@dataclass
class MessageReaction:
    emoji: str
    count: int = 0
    has_reacted: bool = False
    preview_user_ids: list[str] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any]) -> MessageReaction:
        return cls(
            emoji=data.get("emoji", ""),
            count=int(data.get("count", 0)),
            has_reacted=bool(data.get("hasReacted", False)),
            preview_user_ids=list(data.get("previewUserIds") or []),
        )


@dataclass
class ThreadSummary:
    thread_root_event_id: str
    reply_count: int = 0
    last_reply_at: datetime | None = None
    participant_preview_user_ids: list[str] = field(default_factory=list)
    participant_count: int = 0
    is_following: bool | None = None
    has_unread: bool | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ThreadSummary | None:
        if not data:
            return None
        viewer = data.get("viewerState") or {}
        return cls(
            thread_root_event_id=data.get("threadRootEventId", ""),
            reply_count=int(data.get("replyCount", 0)),
            last_reply_at=parse_datetime(data.get("lastReplyAt")),
            participant_preview_user_ids=list(
                data.get("participantPreviewUserIds") or []
            ),
            participant_count=int(data.get("participantCount", 0)),
            is_following=viewer.get("isFollowing") if "isFollowing" in viewer else None,
            has_unread=viewer.get("hasUnread") if "hasUnread" in viewer else None,
        )


@dataclass
class Message:
    id: str
    room_id: str
    created_at: datetime | None
    actor_id: str
    body: str | None = None
    attachments: list[MessageAttachment] = field(default_factory=list)
    link_preview: LinkPreview | None = None
    updated_at: datetime | None = None
    in_reply_to: str = ""
    thread_root_event_id: str = ""
    echo_of_event_id: str = ""
    echo_from_thread_root_event_id: str = ""
    channel_echo_event_id: str = ""
    reactions: list[MessageReaction] = field(default_factory=list)
    thread: ThreadSummary | None = None
    deleted_at: datetime | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> Message | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            room_id=data.get("roomId", ""),
            created_at=parse_datetime(data.get("createdAt")),
            actor_id=data.get("actorId", ""),
            body=data.get("body"),
            attachments=[MessageAttachment.parse(a) for a in data.get("attachments") or []],
            link_preview=LinkPreview.parse(data.get("linkPreview")),
            updated_at=parse_datetime(data.get("updatedAt")),
            in_reply_to=data.get("inReplyTo", ""),
            thread_root_event_id=data.get("threadRootEventId", ""),
            echo_of_event_id=data.get("echoOfEventId", ""),
            echo_from_thread_root_event_id=data.get("echoFromThreadRootEventId", ""),
            channel_echo_event_id=data.get("channelEchoEventId", ""),
            reactions=[MessageReaction.parse(r) for r in data.get("reactions") or []],
            thread=ThreadSummary.parse(data.get("thread")),
            deleted_at=parse_datetime(data.get("deletedAt")),
        )


@dataclass
class TimelineEvent:
    id: str
    created_at: datetime | None
    actor_id: str = ""
    kind: str = ""  # "message_posted", "room_created", "user_joined_room", ...
    message: Message | None = None
    room_id: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any]) -> TimelineEvent:
        kind = ""
        message: Message | None = None
        room_id = ""
        for candidate in (
            "messagePosted",
            "roomCreated",
            "roomUpdated",
            "roomDeleted",
            "roomArchived",
            "roomUnarchived",
            "userJoinedRoom",
            "userLeftRoom",
        ):
            payload = data.get(candidate)
            if payload is None:
                continue
            kind = _snake_case(candidate)
            if candidate == "messagePosted":
                message = Message.parse(payload.get("message"))
            else:
                room_id = payload.get("roomId", "")
            break
        return cls(
            id=data.get("id", ""),
            created_at=parse_datetime(data.get("createdAt")),
            actor_id=data.get("actorId", ""),
            kind=kind,
            message=message,
            room_id=room_id,
        )


def _snake_case(name: str) -> str:
    out: list[str] = []
    for ch in name:
        if ch.isupper() and out:
            out.append("_")
        out.append(ch.lower())
    return "".join(out)


@dataclass
class TimelinePage:
    events: list[TimelineEvent] = field(default_factory=list)
    start_cursor: str = ""
    end_cursor: str = ""
    has_older: bool = False
    has_newer: bool = False
    users_by_id: dict[str, User] = field(default_factory=dict)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> TimelinePage:
        data = data or {}
        includes = data.get("includes") or {}
        users_raw = includes.get("users") or {}
        users: dict[str, User] = {}
        for uid, user_data in users_raw.items():
            parsed_user = User.parse(user_data)
            if parsed_user is not None:
                users[uid] = parsed_user
        return cls(
            events=[TimelineEvent.parse(e) for e in data.get("events") or []],
            start_cursor=data.get("startCursor", ""),
            end_cursor=data.get("endCursor", ""),
            has_older=bool(data.get("hasOlder", False)),
            has_newer=bool(data.get("hasNewer", False)),
            users_by_id=users,
        )


@dataclass
class FollowedThread:
    room: RoomSummary | None
    root_message: Message | None
    thread: ThreadSummary | None

    @classmethod
    def parse(cls, data: dict[str, Any]) -> FollowedThread:
        return cls(
            room=RoomSummary.parse(data.get("room")),
            root_message=Message.parse(data.get("rootMessage")),
            thread=ThreadSummary.parse(data.get("thread")),
        )


@dataclass
class FollowedThreadsPage:
    threads: list[FollowedThread] = field(default_factory=list)
    page: Page = field(default_factory=Page)
    users_by_id: dict[str, User] = field(default_factory=dict)


@dataclass
class Notification:
    id: str
    created_at: datetime | None
    actor: User | None
    kind: str  # "direct_message" | "mention" | "reply" | "room_message"
    room: RoomSummary | None = None
    event_id: str = ""
    in_reply_to_id: str = ""
    thread_root_event_id: str | None = None

    @classmethod
    def parse(cls, data: dict[str, Any]) -> Notification:
        kind = ""
        room: RoomSummary | None = None
        event_id = ""
        in_reply_to_id = ""
        thread_root: str | None = None
        for candidate in ("directMessage", "mention", "reply", "roomMessage"):
            payload = data.get(candidate)
            if payload is None:
                continue
            kind = _snake_case(candidate)
            room = RoomSummary.parse(payload.get("room"))
            event_id = payload.get("eventId", "")
            in_reply_to_id = payload.get("inReplyToId", "") if candidate == "reply" else ""
            thread_root = payload.get("threadRootEventId")
            break
        return cls(
            id=data.get("id", ""),
            created_at=parse_datetime(data.get("createdAt")),
            actor=User.parse(data.get("actor")),
            kind=kind,
            room=room,
            event_id=event_id,
            in_reply_to_id=in_reply_to_id,
            thread_root_event_id=thread_root,
        )


@dataclass
class NotificationsPage:
    notifications: list[Notification] = field(default_factory=list)
    page: Page = field(default_factory=Page)


@dataclass
class NotificationPreference:
    level: NotificationLevel = NotificationLevel.UNSPECIFIED
    effective_level: NotificationLevel = NotificationLevel.UNSPECIFIED

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> NotificationPreference:
        data = data or {}
        return cls(
            level=_parse_enum(
                NotificationLevel, data.get("level"), NotificationLevel.UNSPECIFIED
            ),  # type: ignore[arg-type]
            effective_level=_parse_enum(
                NotificationLevel,
                data.get("effectiveLevel"),
                NotificationLevel.UNSPECIFIED,
            ),  # type: ignore[arg-type]
        )


@dataclass
class Asset:
    id: str
    filename: str
    content_type: str
    size: int
    width: int = 0
    height: int = 0
    asset_url: AssetUrl | None = None
    thumbnail_asset_url: AssetUrl | None = None
    video_processing: VideoProcessing | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> Asset | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            filename=data.get("filename", ""),
            content_type=data.get("contentType", ""),
            size=int(data.get("size", 0)),
            width=int(data.get("width", 0)),
            height=int(data.get("height", 0)),
            asset_url=AssetUrl.parse(data.get("assetUrl")),
            thumbnail_asset_url=AssetUrl.parse(data.get("thumbnailAssetUrl")),
            video_processing=VideoProcessing.parse(data.get("videoProcessing")),
        )


@dataclass
class ImageTransformOptions:
    width: int
    height: int
    fit: ImageFitMode = ImageFitMode.UNSPECIFIED

    def to_wire(self) -> dict[str, Any]:
        return {
            "width": self.width,
            "height": self.height,
            "fit": self.fit.value,
        }


@dataclass
class ActiveCallParticipant:
    user: User | None = None
    joined_at: datetime | None = None
    call_id: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any]) -> ActiveCallParticipant:
        return cls(
            user=User.parse(data.get("user")),
            joined_at=parse_datetime(data.get("joinedAt")),
            call_id=data.get("callId", ""),
        )


@dataclass
class ActiveCall:
    call_id: str
    room: RoomSummary | None
    participants: list[ActiveCallParticipant] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any]) -> ActiveCall:
        return cls(
            call_id=data.get("callId", ""),
            room=RoomSummary.parse(data.get("room")),
            participants=[
                ActiveCallParticipant.parse(p) for p in data.get("participants") or []
            ],
        )


# --- Roles ----------------------------------------------------------------


@dataclass
class Role:
    name: str
    display_name: str = ""
    description: str = ""
    is_system: bool = False
    position: int = 0
    pingable: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> Role | None:
        if not data:
            return None
        return cls(
            name=data.get("name", ""),
            display_name=data.get("displayName", ""),
            description=data.get("description", ""),
            is_system=bool(data.get("isSystem", False)),
            position=int(data.get("position", 0)),
            pingable=bool(data.get("pingable", False)),
        )


@dataclass
class AdminRole:
    role: Role | None
    permissions: list[str] = field(default_factory=list)
    permission_denials: list[str] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminRole | None:
        if not data:
            return None
        return cls(
            role=Role.parse(data.get("role")),
            permissions=list(data.get("permissions") or []),
            permission_denials=list(data.get("permissionDenials") or []),
        )


# --- Asset uploads --------------------------------------------------------


@dataclass
class AssetUpload:
    upload_id: str
    room_id: str
    status: AssetUploadStatus = AssetUploadStatus.UNSPECIFIED
    committed_offset: int = 0
    size: int = 0
    max_chunk_size: int = 0
    sha256: str = ""
    expires_at: datetime | None = None
    asset_id: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AssetUpload | None:
        if not data:
            return None
        return cls(
            upload_id=data.get("uploadId", ""),
            room_id=data.get("roomId", ""),
            status=_parse_enum(
                AssetUploadStatus,
                data.get("status"),
                AssetUploadStatus.UNSPECIFIED,
            ),  # type: ignore[arg-type]
            committed_offset=int(data.get("committedOffset", 0)),
            size=int(data.get("size", 0)),
            max_chunk_size=int(data.get("maxChunkSize", 0)),
            sha256=data.get("sha256", ""),
            expires_at=parse_datetime(data.get("expiresAt")),
            asset_id=data.get("assetId", ""),
        )


# --- External identity ----------------------------------------------------


@dataclass
class ExternalIdentityProvider:
    provider: ProviderMetadata | None
    link_url: str = ""
    linked: bool = False
    linked_identity_subject_hash: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any]) -> ExternalIdentityProvider:
        provider_data = data.get("provider")
        provider = ProviderMetadata.parse(provider_data) if provider_data else None
        return cls(
            provider=provider,
            link_url=data.get("linkUrl", ""),
            linked=bool(data.get("linked", False)),
            linked_identity_subject_hash=data.get("linkedIdentitySubjectHash", ""),
        )


@dataclass
class LinkedExternalIdentity:
    provider_id: str
    provider_type: str = ""
    provider_label: str = ""
    subject_hash: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any]) -> LinkedExternalIdentity:
        return cls(
            provider_id=data.get("providerId", ""),
            provider_type=data.get("providerType", ""),
            provider_label=data.get("providerLabel", ""),
            subject_hash=data.get("subjectHash", ""),
        )


# --- External identity flow types ----------------------------------------


@dataclass
class PendingExternalIdentity:
    kind: ExternalIdentityFlowKind = ExternalIdentityFlowKind.UNSPECIFIED
    provider_id: str = ""
    provider_type: str = ""
    provider_label: str = ""
    verified_email: str = ""
    login_hint: str = ""
    display_name_hint: str = ""
    bound_user_id: str = ""
    redirect_path: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> PendingExternalIdentity | None:
        if not data:
            return None
        return cls(
            kind=_parse_enum(  # type: ignore[arg-type]
                ExternalIdentityFlowKind,
                data.get("kind"),
                ExternalIdentityFlowKind.UNSPECIFIED,
            ),
            provider_id=data.get("providerId", ""),
            provider_type=data.get("providerType", ""),
            provider_label=data.get("providerLabel", ""),
            verified_email=data.get("verifiedEmail", ""),
            login_hint=data.get("loginHint", ""),
            display_name_hint=data.get("displayNameHint", ""),
            bound_user_id=data.get("boundUserId", ""),
            redirect_path=data.get("redirectPath", ""),
        )


@dataclass
class ExternalIdentityAccount:
    user_id: str = ""
    login: str = ""
    token: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ExternalIdentityAccount | None:
        if not data:
            return None
        return cls(
            user_id=data.get("userId", ""),
            login=data.get("login", ""),
            token=data.get("token", ""),
        )


# --- Admin: server / members ---------------------------------------------


@dataclass
class ServerConfig:
    server_name: str = ""
    description: str = ""
    motd: str = ""
    welcome_message: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ServerConfig:
        data = data or {}
        return cls(
            server_name=data.get("serverName", ""),
            description=data.get("description", ""),
            motd=data.get("motd", ""),
            welcome_message=data.get("welcomeMessage", ""),
        )


@dataclass
class AdminMember:
    user: User | None
    roles: list[str] = field(default_factory=list)
    created_at: datetime | None = None
    has_verified_email: bool = False
    verified_emails: list[str] = field(default_factory=list)
    viewer_can_delete_account: bool = False
    last_login_change: datetime | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminMember | None:
        if not data:
            return None
        return cls(
            user=User.parse(data.get("user")),
            roles=list(data.get("roles") or []),
            created_at=parse_datetime(data.get("createdAt")),
            has_verified_email=bool(data.get("hasVerifiedEmail", False)),
            verified_emails=list(data.get("verifiedEmails") or []),
            viewer_can_delete_account=bool(data.get("viewerCanDeleteAccount", False)),
            last_login_change=parse_datetime(data.get("lastLoginChange")),
        )


# --- Admin: member/role detail envelopes ---------------------------------


@dataclass
class AdminMemberDetail:
    member: AdminMember | None
    roles: list[AdminRole] = field(default_factory=list)
    available_permissions: list[str] = field(default_factory=list)
    viewer_can_assign_roles: bool = False
    viewer_can_manage_roles: bool = False
    viewer_can_manage_user_permissions: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminMemberDetail:
        data = data or {}
        roles = [
            r
            for raw in data.get("roles") or []
            if (r := AdminRole.parse(raw)) is not None
        ]
        return cls(
            member=AdminMember.parse(data.get("member")),
            roles=roles,
            available_permissions=list(data.get("availablePermissions") or []),
            viewer_can_assign_roles=bool(data.get("viewerCanAssignRoles", False)),
            viewer_can_manage_roles=bool(data.get("viewerCanManageRoles", False)),
            viewer_can_manage_user_permissions=bool(
                data.get("viewerCanManageUserPermissions", False)
            ),
        )


@dataclass
class AdminRoleDetail:
    role: AdminRole | None
    users: list[User] = field(default_factory=list)
    viewer_can_manage_roles: bool = False
    viewer_can_assign_roles: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminRoleDetail:
        data = data or {}
        users = [
            u
            for raw in data.get("users") or []
            if (u := User.parse(raw)) is not None
        ]
        return cls(
            role=AdminRole.parse(data.get("role")),
            users=users,
            viewer_can_manage_roles=bool(data.get("viewerCanManageRoles", False)),
            viewer_can_assign_roles=bool(data.get("viewerCanAssignRoles", False)),
        )


# --- Admin: room layout --------------------------------------------------


@dataclass
class AdminRoomLayoutGroup:
    id: str
    name: str
    description: str = ""
    rooms: list[Room] = field(default_factory=list)
    sidebar_links: list[SidebarLink] = field(default_factory=list)
    can_create_room: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminRoomLayoutGroup | None:
        if not data:
            return None
        rooms: list[Room] = []
        links: list[SidebarLink] = []
        for item in data.get("items") or []:
            room = Room.parse(item.get("room"))
            if room is not None:
                rooms.append(room)
            sl = item.get("sidebarLink")
            if sl:
                links.append(
                    SidebarLink(
                        id=sl.get("id", ""),
                        label=sl.get("label", ""),
                        url=sl.get("url", ""),
                    )
                )
        return cls(
            id=data.get("id", ""),
            name=data.get("name", ""),
            description=data.get("description", ""),
            rooms=rooms,
            sidebar_links=links,
            can_create_room=bool(data.get("canCreateRoom", False)),
        )


# --- Viewer snapshot (GetViewerResponse) ----------------------------------


@dataclass
class CapabilityGrant:
    """Effective decision for one capability key (e.g. ``admin.view``)."""

    capability: str
    granted: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> CapabilityGrant | None:
        if data is None:
            return None
        return cls(
            capability=data.get("capability", ""),
            granted=bool(data.get("granted", False)),
        )


@dataclass
class ViewerCapabilities:
    """Permission-derived capabilities for the authenticated user."""

    grants: list[CapabilityGrant] = field(default_factory=list)
    has_unread_followed_threads: bool = False
    has_pending_followed_thread_notifications: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ViewerCapabilities | None:
        if data is None:
            return None
        grants = [
            g
            for raw in data.get("grants") or []
            if (g := CapabilityGrant.parse(raw)) is not None
        ]
        return cls(
            grants=grants,
            has_unread_followed_threads=bool(
                data.get("hasUnreadFollowedThreads", False)
            ),
            has_pending_followed_thread_notifications=bool(
                data.get("hasPendingFollowedThreadNotifications", False)
            ),
        )


@dataclass
class PermissionGrant:
    """Effective decision for one permission key (e.g. ``server.manage``)."""

    permission: str
    granted: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> PermissionGrant | None:
        if data is None:
            return None
        return cls(
            permission=data.get("permission", ""),
            granted=bool(data.get("granted", False)),
        )


@dataclass
class ViewerState:
    """Non-permission server state for the authenticated user."""

    has_unread_rooms: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ViewerState | None:
        if data is None:
            return None
        return cls(
            has_unread_rooms=bool(data.get("hasUnreadRooms", False)),
        )


@dataclass
class RoomNotificationPreference:
    """Notification preference for one joined room."""

    room_id: str
    preference: NotificationPreference | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RoomNotificationPreference | None:
        if data is None:
            return None
        raw_pref = data.get("preference")
        return cls(
            room_id=data.get("roomId", ""),
            preference=NotificationPreference.parse(raw_pref) if raw_pref is not None else None,
        )


@dataclass
class AdminEventLogEntry:
    """One durable EVT entry decoded for operator inspection (``AdminEventLogEntry``)."""

    sequence: str = ""
    subject: str = ""
    aggregate_type: str = ""
    aggregate_id: str = ""
    event_type: str = ""
    event_id: str = ""
    actor_id: str = ""
    created_at: datetime | None = None
    payload_json: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminEventLogEntry | None:
        if data is None:
            return None
        return cls(
            sequence=data.get("sequence", ""),
            subject=data.get("subject", ""),
            aggregate_type=data.get("aggregateType", ""),
            aggregate_id=data.get("aggregateId", ""),
            event_type=data.get("eventType", ""),
            event_id=data.get("eventId", ""),
            actor_id=data.get("actorId", ""),
            created_at=parse_datetime(data.get("createdAt")),
            payload_json=data.get("payloadJson", ""),
        )


@dataclass
class AdminEventLogPage:
    """Newest-first page of durable EVT entries (``ListEventsResponse``)."""

    entries: list[AdminEventLogEntry] = field(default_factory=list)
    has_older: bool = False
    end_cursor: str | None = None
    total_count: int = 0
    scanned_count: int = 0
    scan_limit: int = 0
    scan_limited: bool = False

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminEventLogPage:
        data = data or {}
        entries = [
            e
            for raw in data.get("entries") or []
            if (e := AdminEventLogEntry.parse(raw)) is not None
        ]
        return cls(
            entries=entries,
            has_older=bool(data.get("hasOlder", False)),
            end_cursor=data.get("endCursor"),
            total_count=int(data.get("totalCount", 0)),
            scanned_count=int(data.get("scannedCount", 0)),
            scan_limit=int(data.get("scanLimit", 0)),
            scan_limited=bool(data.get("scanLimited", False)),
        )


@dataclass
class ViewerSnapshot:
    """Full authenticated viewer snapshot (``GetViewerResponse``)."""

    user: ViewerUser | None = None
    capabilities: ViewerCapabilities | None = None
    server_notification_preference: NotificationPreference | None = None
    room_notification_preferences: list[RoomNotificationPreference] = field(
        default_factory=list
    )
    viewer_permissions: list[PermissionGrant] = field(default_factory=list)
    viewer_state: ViewerState | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ViewerSnapshot:
        data = data or {}
        # ServerViewerPermissions wraps a "permissions" list; flatten it.
        vp_raw = data.get("viewerPermissions") or {}
        viewer_permissions = [
            p
            for raw in vp_raw.get("permissions") or []
            if (p := PermissionGrant.parse(raw)) is not None
        ]
        room_prefs = [
            rp
            for raw in data.get("roomNotificationPreferences") or []
            if (rp := RoomNotificationPreference.parse(raw)) is not None
        ]
        raw_snp = data.get("serverNotificationPreference")
        return cls(
            user=ViewerUser.parse(data.get("user")),
            capabilities=ViewerCapabilities.parse(data.get("capabilities")),
            server_notification_preference=(
                NotificationPreference.parse(raw_snp) if raw_snp is not None else None
            ),
            room_notification_preferences=room_prefs,
            viewer_permissions=viewer_permissions,
            viewer_state=ViewerState.parse(data.get("viewerState")),
        )


# --- Admin permissions ----------------------------------------------------


@dataclass
class PermissionScope:
    kind: PermissionScopeKind = PermissionScopeKind.UNSPECIFIED
    id: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> PermissionScope | None:
        if not data:
            return None
        return cls(
            kind=_parse_enum(  # type: ignore[arg-type]
                PermissionScopeKind, data.get("kind"), PermissionScopeKind.UNSPECIFIED
            ),
            id=data.get("id", ""),
        )


@dataclass
class ScopedPermissionDecision:
    permission: str = ""
    scope: PermissionScope | None = None
    override: PermissionDecision = PermissionDecision.UNSPECIFIED
    effective: PermissionDecision = PermissionDecision.UNSPECIFIED

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> ScopedPermissionDecision | None:
        if not data:
            return None
        return cls(
            permission=data.get("permission", ""),
            scope=PermissionScope.parse(data.get("scope")),
            override=_parse_enum(  # type: ignore[arg-type]
                PermissionDecision, data.get("override"), PermissionDecision.UNSPECIFIED
            ),
            effective=_parse_enum(  # type: ignore[arg-type]
                PermissionDecision, data.get("effective"), PermissionDecision.UNSPECIFIED
            ),
        )


@dataclass
class PermissionMatrixScope:
    id: str = ""
    label: str = ""
    kind: PermissionScopeKind = PermissionScopeKind.UNSPECIFIED
    parent_group_id: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> PermissionMatrixScope | None:
        if not data:
            return None
        return cls(
            id=data.get("id", ""),
            label=data.get("label", ""),
            kind=_parse_enum(  # type: ignore[arg-type]
                PermissionScopeKind, data.get("kind"), PermissionScopeKind.UNSPECIFIED
            ),
            parent_group_id=data.get("parentGroupId", ""),
        )


@dataclass
class PermissionMatrixCell:
    permission: str = ""
    scope_id: str = ""
    override: PermissionDecision = PermissionDecision.UNSPECIFIED
    effective: PermissionDecision = PermissionDecision.UNSPECIFIED

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> PermissionMatrixCell | None:
        if not data:
            return None
        return cls(
            permission=data.get("permission", ""),
            scope_id=data.get("scopeId", ""),
            override=_parse_enum(  # type: ignore[arg-type]
                PermissionDecision, data.get("override"), PermissionDecision.UNSPECIFIED
            ),
            effective=_parse_enum(  # type: ignore[arg-type]
                PermissionDecision, data.get("effective"), PermissionDecision.UNSPECIFIED
            ),
        )


@dataclass
class RolePermissionDecisions:
    """Resource-oriented permission decisions for one role.

    Maps to ``ListRolePermissionDecisionsResponse``.
    """

    role_name: str = ""
    decisions: list[ScopedPermissionDecision] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RolePermissionDecisions:
        data = data or {}
        decisions = [
            d
            for raw in data.get("decisions") or []
            if (d := ScopedPermissionDecision.parse(raw)) is not None
        ]
        return cls(
            role_name=data.get("roleName", ""),
            decisions=decisions,
        )


@dataclass
class UserPermissionDecisions:
    """Resource-oriented permission decisions for one user.

    Maps to ``ListUserPermissionDecisionsResponse``.
    """

    user_id: str = ""
    decisions: list[ScopedPermissionDecision] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> UserPermissionDecisions:
        data = data or {}
        decisions = [
            d
            for raw in data.get("decisions") or []
            if (d := ScopedPermissionDecision.parse(raw)) is not None
        ]
        return cls(
            user_id=data.get("userId", ""),
            decisions=decisions,
        )


@dataclass
class RolePermissionMatrix:
    """Permission matrix for one role across all scopes (``RolePermissionMatrix``)."""

    role_name: str = ""
    applicable_permissions: list[str] = field(default_factory=list)
    scopes: list[PermissionMatrixScope] = field(default_factory=list)
    cells: list[PermissionMatrixCell] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> RolePermissionMatrix | None:
        if not data:
            return None
        scopes = [
            s
            for raw in data.get("scopes") or []
            if (s := PermissionMatrixScope.parse(raw)) is not None
        ]
        cells = [
            c
            for raw in data.get("cells") or []
            if (c := PermissionMatrixCell.parse(raw)) is not None
        ]
        return cls(
            role_name=data.get("roleName", ""),
            applicable_permissions=list(data.get("applicablePermissions") or []),
            scopes=scopes,
            cells=cells,
        )


@dataclass
class UserPermissionMatrix:
    """Permission matrix for one user across all scopes (``UserPermissionMatrix``)."""

    user_id: str = ""
    applicable_permissions: list[str] = field(default_factory=list)
    scopes: list[PermissionMatrixScope] = field(default_factory=list)
    cells: list[PermissionMatrixCell] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> UserPermissionMatrix | None:
        if not data:
            return None
        scopes = [
            s
            for raw in data.get("scopes") or []
            if (s := PermissionMatrixScope.parse(raw)) is not None
        ]
        cells = [
            c
            for raw in data.get("cells") or []
            if (c := PermissionMatrixCell.parse(raw)) is not None
        ]
        return cls(
            user_id=data.get("userId", ""),
            applicable_permissions=list(data.get("applicablePermissions") or []),
            scopes=scopes,
            cells=cells,
        )


# --- Group G: admin diagnostics tree (diagnostics.proto) -----------------


@dataclass
class AdminProjectionMetric:
    """Projection-specific diagnostic metric (``AdminProjectionMetric``)."""

    name: str = ""
    value: int = 0
    bytes: int = 0

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminProjectionMetric | None:
        if data is None:
            return None
        return cls(
            name=data.get("name", ""),
            value=int(data.get("value", 0)),
            bytes=int(data.get("bytes", 0)),
        )


@dataclass
class AdminProjectionState:
    """Event-sourced projection health and memory estimates (``AdminProjectionState``)."""

    key: str = ""
    name: str = ""
    subjects: list[str] = field(default_factory=list)
    started: bool = False
    startup_duration_seconds: float | None = None
    last_applied_sequence: str = ""
    matching_stream_sequence: str = ""
    stream_last_sequence: str = ""
    lag: int = 0
    failed: bool = False
    failed_sequence: str = ""
    failure: str = ""
    entry_count: int = 0
    estimated_bytes: int = 0
    average_entry_bytes: int = 0
    metrics: list[AdminProjectionMetric] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminProjectionState | None:
        if data is None:
            return None
        raw_sds = data.get("startupDurationSeconds")
        metrics = [
            m
            for raw in data.get("metrics") or []
            if (m := AdminProjectionMetric.parse(raw)) is not None
        ]
        return cls(
            key=data.get("key", ""),
            name=data.get("name", ""),
            subjects=list(data.get("subjects") or []),
            started=bool(data.get("started", False)),
            startup_duration_seconds=(
                float(raw_sds) if raw_sds is not None else None
            ),
            last_applied_sequence=data.get("lastAppliedSequence", ""),
            matching_stream_sequence=data.get("matchingStreamSequence", ""),
            stream_last_sequence=data.get("streamLastSequence", ""),
            lag=int(data.get("lag", 0)),
            failed=bool(data.get("failed", False)),
            failed_sequence=data.get("failedSequence", ""),
            failure=data.get("failure", ""),
            entry_count=int(data.get("entryCount", 0)),
            estimated_bytes=int(data.get("estimatedBytes", 0)),
            average_entry_bytes=int(data.get("averageEntryBytes", 0)),
            metrics=metrics,
        )


@dataclass
class AdminConnectionInfo:
    """Current NATS server connection metadata (``AdminConnectionInfo``)."""

    connected: bool = False
    server_id: str = ""
    server_name: str = ""
    version: str = ""
    max_payload: int = 0
    rtt: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminConnectionInfo | None:
        if data is None:
            return None
        return cls(
            connected=bool(data.get("connected", False)),
            server_id=data.get("serverId", ""),
            server_name=data.get("serverName", ""),
            version=data.get("version", ""),
            max_payload=int(data.get("maxPayload", 0)),
            rtt=data.get("rtt", ""),
        )


@dataclass
class AdminAccountInfo:
    """JetStream account limits and usage (``AdminAccountInfo``)."""

    memory: int = 0
    memory_used: int = 0
    storage: int = 0
    storage_used: int = 0
    streams: int = 0
    streams_used: int = 0
    consumers: int = 0
    consumers_used: int = 0

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminAccountInfo | None:
        if data is None:
            return None
        return cls(
            memory=int(data.get("memory", 0)),
            memory_used=int(data.get("memoryUsed", 0)),
            storage=int(data.get("storage", 0)),
            storage_used=int(data.get("storageUsed", 0)),
            streams=int(data.get("streams", 0)),
            streams_used=int(data.get("streamsUsed", 0)),
            consumers=int(data.get("consumers", 0)),
            consumers_used=int(data.get("consumersUsed", 0)),
        )


@dataclass
class AdminServerStats:
    """Aggregate Chatto object counts (``AdminServerStats``)."""

    user_count: int = 0
    channel_room_count: int = 0
    dm_room_count: int = 0

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminServerStats | None:
        if data is None:
            return None
        return cls(
            user_count=int(data.get("userCount", 0)),
            channel_room_count=int(data.get("channelRoomCount", 0)),
            dm_room_count=int(data.get("dmRoomCount", 0)),
        )


@dataclass
class AdminNatsStreamInfo:
    """JetStream stream state (``AdminNatsStreamInfo``)."""

    name: str = ""
    description: str = ""
    subjects: list[str] = field(default_factory=list)
    storage: str = ""
    messages: int = 0
    bytes: int = 0
    first_sequence: str = ""
    last_sequence: str = ""
    consumer_count: int = 0
    replicas: int = 0
    cluster_leader: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminNatsStreamInfo | None:
        if data is None:
            return None
        return cls(
            name=data.get("name", ""),
            description=data.get("description", ""),
            subjects=list(data.get("subjects") or []),
            storage=data.get("storage", ""),
            messages=int(data.get("messages", 0)),
            bytes=int(data.get("bytes", 0)),
            first_sequence=data.get("firstSequence", ""),
            last_sequence=data.get("lastSequence", ""),
            consumer_count=int(data.get("consumerCount", 0)),
            replicas=int(data.get("replicas", 0)),
            cluster_leader=data.get("clusterLeader", ""),
        )


@dataclass
class AdminNatsConsumerInfo:
    """JetStream consumer state (``AdminNatsConsumerInfo``)."""

    stream: str = ""
    name: str = ""
    durable: str = ""
    filter_subject: str = ""
    filter_subjects: list[str] = field(default_factory=list)
    ack_policy: str = ""
    pull_based: bool = False
    push_bound: bool = False
    pending: int = 0
    ack_pending: int = 0
    redelivered: int = 0
    waiting: int = 0
    delivered_consumer_sequence: str = ""
    delivered_stream_sequence: str = ""
    ack_floor_consumer_sequence: str = ""
    ack_floor_stream_sequence: str = ""

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminNatsConsumerInfo | None:
        if data is None:
            return None
        return cls(
            stream=data.get("stream", ""),
            name=data.get("name", ""),
            durable=data.get("durable", ""),
            filter_subject=data.get("filterSubject", ""),
            filter_subjects=list(data.get("filterSubjects") or []),
            ack_policy=data.get("ackPolicy", ""),
            pull_based=bool(data.get("pullBased", False)),
            push_bound=bool(data.get("pushBound", False)),
            pending=int(data.get("pending", 0)),
            ack_pending=int(data.get("ackPending", 0)),
            redelivered=int(data.get("redelivered", 0)),
            waiting=int(data.get("waiting", 0)),
            delivered_consumer_sequence=data.get("deliveredConsumerSequence", ""),
            delivered_stream_sequence=data.get("deliveredStreamSequence", ""),
            ack_floor_consumer_sequence=data.get("ackFloorConsumerSequence", ""),
            ack_floor_stream_sequence=data.get("ackFloorStreamSequence", ""),
        )


@dataclass
class AdminNatsStats:
    """JetStream stream and consumer diagnostics (``AdminNatsStats``)."""

    total_messages: int = 0
    total_bytes: int = 0
    total_consumer_pending: int = 0
    total_ack_pending: int = 0
    streams: list[AdminNatsStreamInfo] = field(default_factory=list)
    consumers: list[AdminNatsConsumerInfo] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminNatsStats | None:
        if data is None:
            return None
        streams = [
            s
            for raw in data.get("streams") or []
            if (s := AdminNatsStreamInfo.parse(raw)) is not None
        ]
        consumers = [
            c
            for raw in data.get("consumers") or []
            if (c := AdminNatsConsumerInfo.parse(raw)) is not None
        ]
        return cls(
            total_messages=int(data.get("totalMessages", 0)),
            total_bytes=int(data.get("totalBytes", 0)),
            total_consumer_pending=int(data.get("totalConsumerPending", 0)),
            total_ack_pending=int(data.get("totalAckPending", 0)),
            streams=streams,
            consumers=consumers,
        )


@dataclass
class AdminSystemInfo:
    """Broker and application diagnostics (``AdminSystemInfo``)."""

    connection: AdminConnectionInfo | None = None
    account: AdminAccountInfo | None = None
    nats: AdminNatsStats | None = None
    stats: AdminServerStats | None = None

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminSystemInfo | None:
        if data is None:
            return None
        return cls(
            connection=AdminConnectionInfo.parse(data.get("connection")),
            account=AdminAccountInfo.parse(data.get("account")),
            nats=AdminNatsStats.parse(data.get("nats")),
            stats=AdminServerStats.parse(data.get("stats")),
        )


@dataclass
class AdminSystemInfoSnapshot:
    """Envelope for ``GetSystemInfoResponse``.

    Contains broker/JetStream diagnostics and projection health snapshots.
    """

    system_info: AdminSystemInfo | None = None
    projections: list[AdminProjectionState] = field(default_factory=list)

    @classmethod
    def parse(cls, data: dict[str, Any] | None) -> AdminSystemInfoSnapshot:
        data = data or {}
        projections = [
            p
            for raw in data.get("projections") or []
            if (p := AdminProjectionState.parse(raw)) is not None
        ]
        return cls(
            system_info=AdminSystemInfo.parse(data.get("systemInfo")),
            projections=projections,
        )