"""Abstract persistence API for per-conversation share links.""" from __future__ import annotations import secrets from abc import ABC, abstractmethod from typing import Any # Unambiguous alphabet — no 0/O/1/I/L so codes are easy to read and type. _CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789" def generate_share_code(length: int = 8) -> str: """Return a random, unguessable, human-readable share code.""" return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(length)) class ThreadShareStore(ABC): @abstractmethod async def create(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any]: """Create a new share link for one conversation and return its record. ``mode`` is ``"import"`` (recipient copies the conversation) or ``"view"`` (a public read-only static page). """ raise NotImplementedError @abstractmethod async def get(self, share_code: str) -> dict[str, Any] | None: """Look up a share by code regardless of owner (None if missing).""" raise NotImplementedError @abstractmethod async def get_active_for_thread(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any] | None: """Return the owner's non-revoked share of ``mode`` for ``thread_id``, if any. Used so re-sharing the same conversation in the same mode reuses its existing code instead of minting a new one each time. """ raise NotImplementedError @abstractmethod async def list_for_owner(self, owner_user_id: str) -> list[dict[str, Any]]: """List all share links created by ``owner_user_id`` (newest first).""" raise NotImplementedError @abstractmethod async def revoke(self, share_code: str, owner_user_id: str) -> bool: """Revoke a share link. Returns False if missing or not owned.""" raise NotImplementedError