ditto

109class Attachment:
110    """A reference to a binary attachment backed by native storage.
111
112    Call :meth:`close` when done with it, or use it as a context manager.
113    The garbage-collector finalizer frees the underlying handle as a
114    backstop only — do not rely on it for prompt cleanup.
115    """
116
117    def __init__(
118        self,
119        ditto: Ditto,
120        handle: Any,
121        id_bytes: bytes,
122        length: int,
123        metadata: Mapping[str, str] | None,
124    ) -> None:
125        self._ditto = weakref.ref(ditto)
126        self._handle = handle
127        self._closed = False
128        self._path: str | None = None
129        self._lock = ProcessLocalRLock()
130        self._finalizer = weakref.finalize(self, _free_attachment, handle)
131        self._token = AttachmentToken(bytes(id_bytes), int(length), dict(metadata or {}))
132
133    @property
134    def id(self) -> str:
135        """The attachment's id, as a URL-safe base64 string."""
136
137        return self._token.id
138
139    @property
140    def length(self) -> int:
141        """The attachment's size in bytes."""
142
143        return self._token.length
144
145    @property
146    def metadata(self) -> dict[str, str]:
147        """A copy of the attachment's string-keyed metadata."""
148
149        return dict(self._token.metadata)
150
151    def _native(self) -> tuple[Ditto, Any]:
152        owner = self._ditto()
153        if self._closed or not self._handle or owner is None or owner.closed:
154            raise DittoClosedError("Attachment is closed")
155        return owner, self._handle
156
157    def _get_path(self) -> str:
158        with self._lock:
159            if self._path is None:
160                owner, handle = self._native()
161                self._path = consume_string(
162                    get_ffi().ditto_get_complete_attachment_path(owner._native_handle, handle)
163                )
164                if self._path is None:
165                    raise DittoAttachmentError("Attachment data is not locally available")
166            return self._path
167
168    async def data(self) -> bytes:
169        """Read the complete attachment into memory, off the event loop thread.
170
171        For large attachments, prefer :meth:`open_stream` to avoid buffering
172        the whole file.
173
174        Raises :class:`~ditto.errors.DittoClosedError` if the attachment or its Ditto instance has
175        been closed.
176        """
177
178        return await run_blocking(Path(self._get_path()).read_bytes)
179
180    async def copy_to_path(self, destination_path: str | Path) -> None:
181        """Copy the attachment's locally available data to a new file.
182
183        Uses exclusive creation, so ``destination_path`` must not already
184        exist — this raises ``FileExistsError`` if it does.
185        ``destination_path`` must not contain ``..`` path segments; a value
186        that does raises DittoValidationError instead of being copied.
187
188        Raises :class:`~ditto.errors.DittoClosedError` if the attachment or its Ditto instance has
189        been closed.
190        """
191
192        source = self._get_path()
193        destination = Path(destination_path)
194        # Destinations often come from document data; refuse traversal
195        # segments so a hostile value cannot climb out of the caller's
196        # intended directory.
197        if ".." in destination.parts:
198            raise DittoValidationError(
199                "copy_to_path destination must not contain '..' path segments"
200            )
201
202        def copy() -> None:
203            with Path(source).open("rb") as source_file, destination.open("xb") as destination_file:
204                shutil.copyfileobj(source_file, destination_file)
205
206        await run_blocking(copy)
207
208    def open_stream(self) -> IO[bytes]:
209        """Open the locally available attachment data for reading.
210
211        The caller owns the returned file object and must close it.
212        """
213
214        return open(self._get_path(), "rb")
215
216    def _cbor_representation(self) -> dict[str, Any]:
217        return {
218            "_ditto_internal_type_jkb12973t4b": 2,
219            "_id": self._token.id_bytes,
220            "_len": self._token.length,
221            "_meta": self._token.metadata,
222        }
223
224    def close(self) -> None:
225        """Release the native attachment handle. Safe to call more than once."""
226
227        with self._lock:
228            if self._closed:
229                return
230            self._closed = True
231            self._finalizer.detach()
232            handle = self._handle
233            self._handle = None
234        if handle:
235            get_ffi().ditto_free_attachment_handle(handle)
236
237    def __enter__(self) -> Attachment:
238        """Context manager: closes the attachment on exit."""
239
240        return self
241
242    def __exit__(self, *_: object) -> None:
243        self.close()

254@dataclass(frozen=True, slots=True)
255class AttachmentFetchCompleted(AttachmentFetchEvent):
256    """The fetch finished; ``attachment`` is now available locally.
257
258    At most one ``Completed`` or ``Deleted`` event is delivered per fetch,
259    after any number of :class:`AttachmentFetchProgress` events.
260    """
261
262    attachment: Attachment

273@dataclass(frozen=True, slots=True)
274class AttachmentFetchDeleted(AttachmentFetchEvent):
275    """The source attachment was deleted upstream before the fetch completed.
276
277    At most one ``Completed`` or ``Deleted`` event is delivered per fetch.
278    """

246class AttachmentFetchEvent:
247    """Base class for events delivered to an :class:`AttachmentFetcher` handler."""
248
249    Completed: type[AttachmentFetchCompleted]
250    Progress: type[AttachmentFetchProgress]
251    Deleted: type[AttachmentFetchDeleted]

265@dataclass(frozen=True, slots=True)
266class AttachmentFetchProgress(AttachmentFetchEvent):
267    """Reports fetch progress. May be delivered any number of times."""
268
269    downloaded_bytes: int
270    total_bytes: int

439class AttachmentFetcher:
440    """Tracks an in-progress attachment fetch and delivers its events.
441
442    The owning :class:`~ditto.store.Store` retains this fetcher internally
443    until a :class:`AttachmentFetchCompleted` or :class:`AttachmentFetchDeleted`
444    event fires, or until :meth:`stop` is called — the caller does not need to
445    hold a reference to keep the fetch alive.
446    """
447
448    def __init__(
449        self,
450        ditto: Ditto,
451        token: Mapping[str, Any],
452        on_event: Callable[[AttachmentFetchEvent], Any],
453    ) -> None:
454        if not callable(on_event):
455            raise TypeError("on_event must be callable")
456        self._ditto = weakref.ref(ditto)
457        self._token = AttachmentToken.from_mapping(token)
458        self._closed = False
459        self._cancel_token = 0
460        self._finish_lock = ProcessLocalRLock()
461        self._loop = asyncio.get_running_loop()
462        store = ditto.store
463        payload = _FetchPayload(self._loop, on_event, self._ditto, self._token)
464        self._registry_key = registry.register(payload)
465        try:
466            payload.fetcher = weakref.ref(self)
467            store._attachment_fetchers.add(self)
468            borrowed_id = borrowed_bytes(self._token.id_bytes)
469            result = get_ffi().ditto_resolve_attachment(
470                ditto._native_handle,
471                borrowed_id.slice,
472                self._registry_key,
473                _RETAIN,
474                _RELEASE,
475                _COMPLETED,
476                _PROGRESS,
477                _DELETED,
478            )
479        except BaseException:
480            self._closed = True
481            store._attachment_fetchers.discard(self)
482            registry.release(self._registry_key)
483            raise
484        if result.status_code != 0:
485            self._closed = True
486            store._attachment_fetchers.discard(self)
487            registry.release(self._registry_key)
488            message = legacy_error_message()
489            if result.status_code == 2:
490                raise DittoAttachmentTokenInvalidError(message)
491            if result.status_code == 3:
492                raise DittoAttachmentNotFoundError(message)
493            raise DittoAttachmentError(message)
494        self._cancel_token = int(result.cancel_token)
495
496    @property
497    def ditto(self) -> Ditto | None:
498        """The owning :class:`~ditto.ditto.Ditto` instance, or ``None`` if garbage collected."""
499
500        return self._ditto()
501
502    @property
503    def closed(self) -> bool:
504        """Whether the fetch has finished, been cancelled, or had its resources released."""
505
506        return self._closed
507
508    def stop(self) -> None:
509        """Cancel an in-flight fetch.
510
511        Safe to call again after the fetch has already completed or the
512        attachment was deleted upstream. ``cancel`` and ``close`` are
513        aliases for this method.
514        """
515
516        self._finish(cancel_native=True)
517
518    def _finish(self, *, cancel_native: bool) -> None:
519        with self._finish_lock:
520            if self._closed:
521                return
522            self._closed = True
523            cancel_token, self._cancel_token = self._cancel_token, 0
524
525        owner = self._ditto()
526        if owner is not None:
527            owner.store._attachment_fetchers.discard(self)
528        try:
529            if owner is not None and cancel_native and not owner.closed and cancel_token:
530                borrowed_id = borrowed_bytes(self._token.id_bytes)
531                status = get_ffi().ditto_cancel_resolve_attachment(
532                    owner._native_handle, borrowed_id.slice, cancel_token
533                )
534                if status not in (0, 3):
535                    _log.warning("Failed to cancel attachment fetch: %s", legacy_error_message())
536        except Exception:
537            # Cancellation is best-effort during concurrent Ditto shutdown. The
538            # Python callback ownership still has to be released exactly once.
539            _log.warning("Failed to cancel attachment fetch during cleanup", exc_info=True)
540        finally:
541            payload = registry.get(self._registry_key)
542            if isinstance(payload, _FetchPayload):
543                with payload.lock:
544                    payload.references -= 1
545                    remove = payload.references <= 0
546                if remove:
547                    registry.release(self._registry_key)
548
549    cancel = stop
550    close = stop
551
552    def __enter__(self) -> AttachmentFetcher:
553        return self
554
555    def __exit__(self, *_: object) -> None:
556        self.stop()

 49@dataclass(frozen=True, slots=True, eq=False)
 50class AttachmentToken:
 51    """The wire representation of an attachment reference.
 52
 53    Mirrors the dict shape produced by native code and returned in a query
 54    result item's document data: ``id`` (URL-safe, unpadded base64), ``len``,
 55    and ``metadata``. :meth:`from_mapping` raises DittoValidationError for
 56    any malformed field.
 57    """
 58
 59    id_bytes: bytes
 60    length: int
 61    metadata: dict[str, str]
 62
 63    @property
 64    def id(self) -> str:
 65        """The token's id, as a URL-safe, unpadded base64 string of ``id_bytes``."""
 66
 67        return base64.urlsafe_b64encode(self.id_bytes).decode("ascii").rstrip("=")
 68
 69    def __eq__(self, other: object) -> bool:
 70        return isinstance(other, AttachmentToken) and self.id_bytes == other.id_bytes
 71
 72    def __hash__(self) -> int:
 73        return hash(self.id_bytes)
 74
 75    @classmethod
 76    def from_mapping(cls, value: Mapping[str, Any]) -> AttachmentToken:
 77        """Build a token from a document's ``id``/``len``/``metadata`` mapping.
 78
 79        Raises :class:`~ditto.errors.DittoValidationError` for any malformed field.
 80        """
 81
 82        try:
 83            encoded_id = value["id"]
 84        except KeyError as error:
 85            raise DittoValidationError("Expected an `id` key as string") from error
 86        if not isinstance(encoded_id, str):
 87            raise DittoValidationError("Expected an `id` key as string")
 88        try:
 89            padding = "=" * (-len(encoded_id) % 4)
 90            id_bytes = base64.b64decode(encoded_id + padding, altchars=b"-_", validate=True)
 91        except (ValueError, TypeError) as error:
 92            raise DittoValidationError("Failed to decode base64 string id") from error
 93        try:
 94            length = int(value["len"])
 95        except (KeyError, TypeError, ValueError, OverflowError) as error:
 96            raise DittoValidationError("Expected a `len` key as a non-negative integer") from error
 97        if length < 0:
 98            raise DittoValidationError("Expected a `len` key as a non-negative integer")
 99        metadata = value.get("metadata")
100        if not isinstance(metadata, Mapping):
101            raise DittoValidationError("Expected a `metadata` key as a dictionary")
102        if not all(
103            isinstance(key, str) and isinstance(item, str) for key, item in metadata.items()
104        ):
105            raise DittoValidationError("Attachment metadata keys and values must be strings")
106        return cls(id_bytes, length, dict(metadata))

29@dataclass(frozen=True, slots=True)
30class AuthenticationProvider:
31    """Identifies which login provider a token should be validated against.
32
33    Passed to :meth:`Authenticator.login`.
34    """
35
36    raw_value: str
37    """The provider identifier.
38
39    Must be non-empty, non-whitespace, and free of control characters — it
40    is passed to native login as a C string.
41    """
42
43    DEVELOPMENT: ClassVar[AuthenticationProvider]
44
45    def __post_init__(self) -> None:
46        if not isinstance(self.raw_value, str) or not self.raw_value.strip():
47            raise ValueError("Authentication provider cannot be empty or whitespace")
48        # The provider flows into native login as a C string; refuse control
49        # characters (including CR/LF) that could smuggle into logs or
50        # protocol frames downstream.
51        if any(ord(char) < 0x20 or ord(char) == 0x7F for char in self.raw_value):
52            raise ValueError("Authentication provider cannot contain control characters")
53
54    @classmethod
55    def custom(cls, raw_value: str) -> AuthenticationProvider:
56        """Create a provider identified by ``raw_value``."""
57
58        return cls(raw_value)
59
60    def __str__(self) -> str:
61        return self.raw_value

67@dataclass(frozen=True, slots=True)
68class AuthenticationStatus:
69    """A point-in-time snapshot of this Ditto instance's authentication state."""
70
71    is_authenticated: bool
72    user_id: str | None
73    """Identifier from the auth webhook when authenticated, else ``None``."""

126class AuthenticationStatusObserver:
127    """A cancellable subscription created by :meth:`Authenticator.observe_status`."""
128
129    def __init__(
130        self, authenticator: Authenticator, handler: Callable[[AuthenticationStatus], Any]
131    ) -> None:
132        self._authenticator = weakref.ref(authenticator)
133        self._handler = handler
134        self._closed = False
135
136    def close(self) -> None:
137        """Cancel the subscription. Safe to call more than once."""
138
139        if self._closed:
140            return
141        self._closed = True
142        authenticator = self._authenticator()
143        if authenticator is not None:
144            authenticator._status_observers.discard(self)
145
146    cancel = close
147
148    def __enter__(self) -> AuthenticationStatusObserver:
149        return self
150
151    def __exit__(self, *_: object) -> None:
152        self.close()

155class Authenticator:
156    """Manages login, logout, and authentication-status/expiration notifications."""
157
158    def __init__(self, ditto: Ditto, loop: asyncio.AbstractEventLoop) -> None:
159        self._ditto = weakref.ref(ditto)
160        self._loop = loop
161        self._lock = ProcessLocalRLock()
162        self._expiration_handler: Callable[[Ditto, timedelta], Any] | None = None
163        self._expiration_observer: Any | None = None
164        self._expiration_registry_key: int | None = None
165        self._status_observers: set[AuthenticationStatusObserver] = set()
166        self._status = self._read_status()
167        self._status_registry_key: int | None = None
168        self._install_status_handler()
169
170    def _owner(self) -> Ditto:
171        owner = self._ditto()
172        if owner is None or owner.closed:
173            raise DittoClosedError("Authenticator's Ditto instance is closed")
174        return owner
175
176    def _read_status(self) -> AuthenticationStatus:
177        owner = self._ditto()
178        if owner is None:
179            return AuthenticationStatus(False, None)
180        ffi = get_ffi()
181        return AuthenticationStatus(
182            bool(ffi.ditto_auth_client_is_web_valid(owner._native_handle)),
183            consume_string(ffi.ditto_auth_client_user_id(owner._native_handle), ffi),
184        )
185
186    def _install_status_handler(self) -> None:
187        from ._ffi._bindings import (
188            BoxDynFnMut1_void_dittoffi_authentication_status_ptr_t,
189        )
190
191        key = registry.register(_NativeCallbackPayload(weakref.ref(self), "status"))
192        call, free = _callbacks(BoxDynFnMut1_void_dittoffi_authentication_status_ptr_t)
193        closure = BoxDynFnMut1_void_dittoffi_authentication_status_ptr_t(key, call, free)
194        self._status_registry_key = key
195        get_ffi().dittoffi_ditto_set_authentication_status_handler(
196            self._owner()._native_handle, closure
197        )
198
199    @property
200    def status(self) -> AuthenticationStatus:
201        """The most recently observed :class:`AuthenticationStatus`."""
202
203        with self._lock:
204            return self._status
205
206    @property
207    def expiration_handler(self) -> Callable[[Ditto, timedelta], Any] | None:
208        """A callback invoked shortly before auth expires, or if not yet authenticated.
209
210        Called with the owning :class:`Ditto` instance and the time
211        remaining until expiration.
212
213        Required in server connect mode: :meth:`Sync.start` raises
214        ``DittoExpirationHandlerMissingError`` if this is unset when sync
215        starts. The handler may be a synchronous or async callable and is
216        dispatched on this :class:`Authenticator`'s event loop (the loop
217        running when its owning :class:`Ditto` instance was opened).
218        """
219
220        with self._lock:
221            return self._expiration_handler
222
223    @expiration_handler.setter
224    def expiration_handler(self, handler: Callable[[Ditto, timedelta], Any] | None) -> None:
225        if handler is not None and not callable(handler):
226            raise TypeError("expiration_handler must be callable or None")
227        with self._lock:
228            self._cancel_expiration_observer()
229            self._expiration_handler = handler
230            if handler is not None:
231                self._register_expiration_observer()
232
233    def _register_expiration_observer(self) -> None:
234        from ._ffi._bindings import BoxDynFnMut1_void_uint32_t
235
236        key = registry.register(_NativeCallbackPayload(weakref.ref(self), "expiration"))
237        call, free = _callbacks(BoxDynFnMut1_void_uint32_t)
238        closure = BoxDynFnMut1_void_uint32_t(key, call, free)
239        try:
240            result = get_ffi().dittoffi_auth_login_provider_register_observer_throws(
241                self._owner()._native_handle, closure
242            )
243            self._expiration_observer = check_result(result)
244            self._expiration_registry_key = key
245        except BaseException:
246            registry.release(key)
247            raise
248
249    def _cancel_expiration_observer(self, ditto_handle: Any | None = None) -> None:
250        observer = self._expiration_observer
251        key = self._expiration_registry_key
252        handle = ditto_handle
253        if observer and handle is None:
254            owner = self._ditto()
255            if owner is not None and not owner.closed:
256                try:
257                    handle = owner._native_handle
258                except DittoClosedError:
259                    handle = None
260            if handle is None:
261                # Ditto.close() marks the peer closing before it acquires this
262                # lock. Leave ownership intact so its coordinator can retry
263                # with the raw handle it already detached from the peer.
264                return
265        self._expiration_observer = None
266        self._expiration_registry_key = None
267        try:
268            if observer:
269                ffi = get_ffi()
270                try:
271                    ffi.dittoffi_auth_login_provider_observer_cancel(handle, observer)
272                finally:
273                    ffi.dittoffi_auth_login_provider_observer_free(observer)
274        finally:
275            if key is not None:
276                registry.release(key)
277
278    def _expiration_requested(self, seconds: int) -> None:
279        handler = self.expiration_handler
280        owner = self._ditto()
281        if handler is not None and owner is not None:
282            dispatch_handler(self._loop, handler, owner, timedelta(seconds=seconds))
283
284    def _authentication_status_updated(self, status: AuthenticationStatus) -> None:
285        with self._lock:
286            if status == self._status:
287                return
288            self._status = status
289            observers = tuple(self._status_observers)
290        for observer in observers:
291            if not observer._closed:
292                dispatch_handler(self._loop, observer._handler, status)
293
294    async def login(self, token: str, provider: AuthenticationProvider | str) -> str | None:
295        """Log in with a token issued by the given authentication provider.
296
297        Returns:
298            The webhook's client-info JSON string, if any was provided.
299
300        Raises:
301            DittoAuthenticationError: login failed. Its ``client_info_json``
302                attribute carries the webhook's ``clientInfo`` payload, if
303                the webhook provided one.
304        """
305
306        owner = self._owner()
307        raw_provider = (
308            provider.raw_value if isinstance(provider, AuthenticationProvider) else provider
309        )
310
311        def invoke() -> str | None:
312            result = get_ffi().ditto_auth_client_login_with_token_and_feedback(
313                owner._native_handle, utf8(token), utf8(raw_provider)
314            )
315            client_info = consume_string(result.c_string)
316            if result.status_code != 0:
317                raise DittoAuthenticationError(
318                    "Failed to authenticate with Login.", client_info_json=client_info
319                )
320            return client_info
321
322        return await run_blocking(invoke)
323
324    def observe_status(
325        self, handler: Callable[[AuthenticationStatus], Any]
326    ) -> AuthenticationStatusObserver:
327        """Call ``handler`` with the new :class:`AuthenticationStatus` on every change.
328
329        Returns an :class:`AuthenticationStatusObserver`; close it to unsubscribe.
330        """
331
332        if not callable(handler):
333            raise TypeError("observe_status requires a callable handler")
334        observer = AuthenticationStatusObserver(self, handler)
335        self._status_observers.add(observer)
336        return observer
337
338    def logout(self, cleanup: Callable[[Ditto], Any] | None = None) -> None:
339        """Log out: purge cached credentials and stop sync.
340
341        Does not remove data already synced into the local store. If given,
342        ``cleanup`` is invoked with the owning :class:`Ditto` instance after
343        credentials are purged and sync is stopped.
344        """
345
346        owner = self._owner()
347        get_ffi().ditto_auth_client_logout(owner._native_handle)
348        owner.sync.stop()
349        if cleanup is not None:
350            cleanup(owner)
351
352    def close(self) -> None:
353        """Release native observers and the expiration handler. Safe to call more than once."""
354
355        self._close_with_native_handle(None)
356
357    def _close_with_native_handle(self, ditto_handle: Any | None) -> None:
358        with self._lock:
359            self._cancel_expiration_observer(ditto_handle)
360            self._expiration_handler = None
361            for observer in tuple(self._status_observers):
362                observer.close()