models.py

"""Datastore model classes."""
import copy
from datetime import timedelta, timezone
from functools import lru_cache
import itertools
import json
import logging
import random
import re
from threading import Lock
from urllib.parse import quote, urlparse

from arroba.util import parse_at_uri
import cachetools
from Crypto.PublicKey import RSA
from flask import request
from google.cloud import ndb
from granary import as1, as2, atom, bluesky, microformats2
from granary.bluesky import AT_URI_PATTERN, BSKY_APP_URL_RE
from granary.source import html_to_text
import humanize
from oauth_dropins.webutil import util
from oauth_dropins.webutil.appengine_info import DEBUG
from oauth_dropins.webutil.flask_util import error
from oauth_dropins.webutil.models import JsonProperty, StringIdModel
from oauth_dropins.webutil.util import ellipsize, json_dumps, json_loads
from requests import RequestException

import common
from common import (
    base64_to_long,
    DOMAIN_RE,
    long_to_base64,
    OLD_ACCOUNT_AGE,
    PROTOCOL_DOMAINS,
    report_error,
    unwrap,
)
import ids
import memcache

# maps string label to Protocol subclass. values are populated by ProtocolUserMeta.
# (we used to wait for ProtocolUserMeta to populate the keys as well, but that was
# awkward to use in datastore model properties with choices, below; it required
# overriding them in reset_model_properties, which was always flaky.)
PROTOCOLS = {label: None for label in (
    'activitypub',
    'ap',
    'atproto',
    'bsky',
    'ostatus',
    'web',
    'webmention',
    'ui',
)}
if DEBUG:
    PROTOCOLS.update({label: None for label in (
        'fa',
        'fake',
        'efake',
        'other',
    )})

# maps string kind (eg 'MagicKey') to Protocol subclass.
# populated in ProtocolUserMeta
PROTOCOLS_BY_KIND = {}

# 2048 bits makes tests slow, so use 1024 for them
KEY_BITS = 1024 if DEBUG else 2048
PAGE_SIZE = 20

# auto delete most old objects via the Object.expire property
# https://cloud.google.com/datastore/docs/ttl
#
# need to keep follows because we attach them to Followers and use them for
# unfollows
DONT_EXPIRE_OBJECT_TYPES = (as1.ACTOR_TYPES | as1.POST_TYPES
                            | set(['block', 'flag', 'follow', 'like', 'share']))
OBJECT_EXPIRE_AGE = timedelta(days=90)

GET_ORIGINALS_CACHE_EXPIRATION = timedelta(days=1)
FOLLOWERS_CACHE_EXPIRATION = timedelta(hours=2)

# See https://www.cloudimage.io/
IMAGE_PROXY_URL_BASE = 'https://aujtzahimq.cloudimg.io/v7/'
IMAGE_PROXY_DOMAINS = ('threads.net',)

USER_STATUS_DESCRIPTIONS = {  # keep in sync with DM.type!
    'no-feed-or-webmention': "your web site doesn't have an RSS or Atom feed or webmention endpoint",
    'nobot': "your profile has 'nobot' in it",
    'nobridge': "your profile has 'nobridge' in it",
    'opt-out': 'your account or instance has requested to be opted out',
    'owns-webfinger': 'your web site looks like a fediverse instance because it already serves Webfinger',
    'private': 'your account is set as private or protected',
    'requires-avatar': "you haven't set a profile picture",
    'requires-name': "you haven't set a profile name that's different from your username",
    'requires-old-account': f"your account is less than {humanize.naturaldelta(OLD_ACCOUNT_AGE)} old",
    'unsupported-handle-ap': f"<a href='https://fed.brid.gy/docs#fediverse-get-started'>your username has characters that Bridgy Fed doesn't currently support</a>",
}

logger = logging.getLogger(__name__)


class Target(ndb.Model):
    r""":class:`protocol.Protocol` + URI pairs for identifying objects.

    These are currently used for:

    * delivery destinations, eg ActivityPub inboxes, webmention targets, etc.
    * copies of :class:`Object`\s and :class:`User`\s elsewhere,
      eg ``at://`` URIs for ATProto records, nevent etc bech32-encoded Nostr ids,
      ATProto user DIDs, etc.

    Used in :class:`google.cloud.ndb.model.StructuredProperty`\s inside
    :class:`Object` and :class:`User`; not stored as top-level entities in the
    datastore.

    ndb implements this by hoisting each property here into a corresponding
    property on the parent entity, prefixed by the StructuredProperty name
    below, eg ``delivered.uri``, ``delivered.protocol``, etc.

    For repeated StructuredPropertys, the hoisted properties are all repeated on
    the parent entity, and reconstructed into StructuredPropertys based on their
    order.

    https://googleapis.dev/python/python-ndb/latest/model.html#google.cloud.ndb.model.StructuredProperty
    """
    uri = ndb.StringProperty(required=True)
    ''
    protocol = ndb.StringProperty(choices=list(PROTOCOLS.keys()), required=True)
    ''

    def __eq__(self, other):
        """Equality excludes Targets' :class:`Key`."""
        return self.uri == other.uri and self.protocol == other.protocol

    def __hash__(self):
        """Allow hashing so these can be dict keys."""
        return hash((self.protocol, self.uri))


class DM(ndb.Model):
    """:class:`protocol.Protocol` + type pairs for identifying sent DMs.

    Used in :attr:`User.sent_dms`.

    https://googleapis.dev/python/python-ndb/latest/model.html#google.cloud.ndb.model.StructuredProperty
    """
    type = ndb.StringProperty(required=True)
    """Known values (keep in sync with USER_STATUS_DESCRIPTIONS!):
      * no-feed-or-webmention
      * opt-out
      * owns-webfinger
      * private
      * replied_to_bridged_user
      * request_bridging
      * requires-avatar
      * requires-name
      * requires-old-account
      * unsupported-handle-ap
      * welcome
    """
    protocol = ndb.StringProperty(choices=list(PROTOCOLS.keys()), required=True)
    ''

    def __eq__(self, other):
        """Equality excludes Targets' :class:`Key`."""
        return self.type == other.type and self.protocol == other.protocol


class ProtocolUserMeta(type(ndb.Model)):
    """:class:`User` metaclass. Registers all subclasses in the ``PROTOCOLS`` global."""
    def __new__(meta, name, bases, class_dict):
        cls = super().__new__(meta, name, bases, class_dict)

        if hasattr(cls, 'LABEL') and cls.LABEL not in ('protocol', 'user'):
            for label in (cls.LABEL, cls.ABBREV) + cls.OTHER_LABELS:
                if label:
                    PROTOCOLS[label] = cls

        PROTOCOLS_BY_KIND[cls._get_kind()] = cls

        return cls


def reset_protocol_properties():
    """Recreates various protocol properties to include choices from ``PROTOCOLS``."""
    abbrevs = f'({"|".join(PROTOCOLS.keys())}|fed)'
    common.SUBDOMAIN_BASE_URL_RE = re.compile(
        rf'^https?://({abbrevs}\.brid\.gy|localhost(:8080)?)/(convert/|r/)?({abbrevs}/)?(?P<path>.+)')
    ids.COPIES_PROTOCOLS = tuple(label for label, proto in PROTOCOLS.items()
                                 if proto and proto.HAS_COPIES)


class User(StringIdModel, metaclass=ProtocolUserMeta):
    """Abstract base class for a Bridgy Fed user.

    Stores some protocols' keypairs. Currently:

    * RSA keypair for ActivityPub HTTP Signatures
      properties: ``mod``, ``public_exponent``, ``private_exponent``, all
      encoded as base64url (ie URL-safe base64) strings as described in RFC
      4648 and section 5.1 of the Magic Signatures spec:
      https://tools.ietf.org/html/draft-cavage-http-signatures-12
    * *Not* K-256 signing or rotation keys for AT Protocol, those are stored in
      :class:`arroba.datastore_storage.AtpRepo` entities
    """
    obj_key = ndb.KeyProperty(kind='Object')  # user profile
    ''
    mod = ndb.StringProperty()
    ''
    use_instead = ndb.KeyProperty()
    ''

    copies = ndb.StructuredProperty(Target, repeated=True)
    """Proxy copies of this user elsewhere, eg DIDs for ATProto records, bech32
    npub Nostr ids, etc. Similar to ``rel-me`` links in microformats2,
    ``alsoKnownAs`` in DID docs (and now AS2), etc.
    """

    public_exponent = ndb.StringProperty()
    """Part of this user's bridged ActivityPub actor's private key."""
    private_exponent = ndb.StringProperty()
    """Part of this user's bridged ActivityPub actor's private key."""

    manual_opt_out = ndb.BooleanProperty()
    """Set to True for users who asked to be opted out."""

    enabled_protocols = ndb.StringProperty(repeated=True,
                                           choices=list(PROTOCOLS.keys()))
    """Protocols that this user has explicitly opted into.

    Protocols that don't require explicit opt in are omitted here. ``choices``
    is populated in :func:`reset_protocol_properties`.
    """

    sent_dms = ndb.StructuredProperty(DM, repeated=True)
    """DMs that we've attempted to send to this user."""

    created = ndb.DateTimeProperty(auto_now_add=True)
    ''
    updated = ndb.DateTimeProperty(auto_now=True)
    ''

    # `existing` attr is set by get_or_create

    # OLD. some stored entities still have these; do not reuse.
    # direct = ndb.BooleanProperty(default=False)
    # actor_as2 = JsonProperty()
    # protocol-specific state
    # atproto_notifs_indexed_at = ndb.TextProperty()
    # atproto_feed_indexed_at = ndb.TextProperty()

    def __init__(self, **kwargs):
        """Constructor.

        Sets :attr:`obj` explicitly because however
        :class:`google.cloud.ndb.model.Model` sets it doesn't work with
        ``@property`` and ``@obj.setter`` below.
        """
        obj = kwargs.pop('obj', None)
        super().__init__(**kwargs)

        if obj:
            self.obj = obj

        self.lock = Lock()

    @classmethod
    def new(cls, **kwargs):
        """Try to prevent instantiation. Use subclasses instead."""
        raise NotImplementedError()

    def _post_put_hook(self, future):
        logger.debug(f'Wrote {self.key}')

    def add(self, prop, val):
        """Adds a value to a multiply-valued property. Uses ``self.lock``.

        Args:
          prop (str)
          val
        """
        with self.lock:
            added = util.add(getattr(self, prop), val)

        if prop == 'copies' and added:
            memcache.pickle_memcache.set(memcache.memoize_key(
                get_original_user_key, val.uri), self.key)

    @classmethod
    def get_by_id(cls, id, allow_opt_out=False, **kwargs):
        """Override to follow ``use_instead`` property and ``status``.

        Returns None if the user is opted out.
        """
        user = cls._get_by_id(id, **kwargs)
        if user and user.use_instead:
            logger.info(f'{user.key} use_instead => {user.use_instead}')
            user = user.use_instead.get()

        if not user:
            return None

        if user.status and not allow_opt_out:
            logger.info(f'{user.key} is {user.status}')
            return None

        return user

    @classmethod
    def get_or_create(cls, id, propagate=False, allow_opt_out=False,
                      reload=False, **kwargs):
        """Loads and returns a :class:`User`. Creates it if necessary.

        Not transactional because transactions don't read or write memcache. :/
        Fortunately we don't really depend on atomicity for anything, last
        writer wins is pretty much always fine.

        Args:
          propagate (bool): whether to create copies of this user in push-based
            protocols, eg ATProto and Nostr.
          allow_opt_out (bool): whether to allow and create the user if they're
            currently opted out
          reload (bool): whether to reload profile always, vs only if necessary
          kwargs: passed through to ``cls`` constructor

        Returns:
          User: existing or new user, or None if the user is opted out
        """
        assert cls != User

        user = cls.get_by_id(id, allow_opt_out=True)
        if user:
            if reload:
                user.reload_profile(gateway=True, raise_=False)

            if user.status and not allow_opt_out:
                return None
            user.existing = True

            # TODO: propagate more fields?
            changed = False
            for field in ['obj', 'obj_key']:
                old_val = getattr(user, field, None)
                new_val = kwargs.get(field)
                if old_val is None and new_val is not None:
                    setattr(user, field, new_val)
                    changed = True

            if enabled_protocols := kwargs.get('enabled_protocols'):
                user.enabled_protocols = (set(user.enabled_protocols)
                                          | set(enabled_protocols))
                changed = True

            if not propagate:
                if changed:
                    user.put()
                return user

        else:
            if orig_key := get_original_user_key(id):
                orig = orig_key.get()
                if orig.status and not allow_opt_out:
                    return None
                orig.existing = False
                return orig

            user = cls(id=id, **kwargs)
            user.existing = False
            user.reload_profile(gateway=True, raise_=False)
            if user.status and not allow_opt_out:
                return None

        if propagate and not user.status:
            for label in user.enabled_protocols + list(user.DEFAULT_ENABLED_PROTOCOLS):
                proto = PROTOCOLS[label]
                if proto == cls:
                    continue
                elif proto.HAS_COPIES:
                    if not user.get_copy(proto) and user.is_enabled(proto):
                        try:
                            proto.create_for(user)
                        except (ValueError, AssertionError):
                            logger.info(f'failed creating {proto.LABEL} copy',
                                        exc_info=True)
                            util.remove(user.enabled_protocols, proto.LABEL)
                    else:
                        logger.debug(f'{proto.LABEL} not enabled or user copy already exists, skipping propagate')

        # generate keys for all protocols _except_ our own
        #
        # these can use urandom() and do nontrivial math, so they can take time
        # depending on the amount of randomness available and compute needed.
        if cls.LABEL != 'activitypub':
            if (not user.public_exponent or not user.private_exponent or not user.mod):
                assert (not user.public_exponent and not user.private_exponent
                        and not user.mod), id
                key = RSA.generate(KEY_BITS,
                                   randfunc=random.randbytes if DEBUG else None)
                user.mod = long_to_base64(key.n)
                user.public_exponent = long_to_base64(key.e)
                user.private_exponent = long_to_base64(key.d)

        try:
            user.put()
        except AssertionError as e:
            error(f'Bad {cls.__name__} id {id} : {e}')

        logger.debug(('Updated ' if user.existing else 'Created new ') + str(user))
        return user

    @property
    def obj(self):
        """Convenience accessor that loads :attr:`obj_key` from the datastore."""
        if self.obj_key:
            if not hasattr(self, '_obj'):
                self._obj = self.obj_key.get()
            return self._obj

    @obj.setter
    def obj(self, obj):
        if obj:
            assert isinstance(obj, Object)
            assert obj.key
            self._obj = obj
            self.obj_key = obj.key
        else:
            self._obj = self.obj_key = None

    def delete(self, proto=None):
        """Deletes a user's bridged actors in all protocols or a specific one.

        Args:
          proto (Protocol): optional
        """
        now = util.now().isoformat()
        proto_label = proto.LABEL if proto else 'all'
        delete_id = f'{self.profile_id()}#bridgy-fed-delete-user-{proto_label}-{now}'
        delete = Object(id=delete_id, source_protocol=self.LABEL, our_as1={
            'id': delete_id,
            'objectType': 'activity',
            'verb': 'delete',
            'actor': self.key.id(),
            'object': self.key.id(),
        })
        self.deliver(delete, from_user=self, to_proto=proto)

    @classmethod
    def load_multi(cls, users):
        """Loads :attr:`obj` for multiple users in parallel.

        Args:
          users (sequence of User)
        """
        objs = ndb.get_multi(u.obj_key for u in users if u.obj_key)
        keys_to_objs = {o.key: o for o in objs if o}

        for u in users:
            u._obj = keys_to_objs.get(u.obj_key)

    @ndb.ComputedProperty
    def handle(self):
        """This user's unique, human-chosen handle, eg ``@me@snarfed.org``.

        To be implemented by subclasses.
        """
        raise NotImplementedError()

    @ndb.ComputedProperty
    def status(self):
        """Whether this user is blocked or opted out.

        Optional. Current possible values:
          * ``opt-out``: the user or domain has manually opted out
          * ``owns-webfinger``: a :class:`web.Web` user that looks like a
            fediverse server
          * ``nobridge``: the user's profile has ``#nobridge`` in it
          * ``nobot``: the user's profile has ``#nobot`` in it
          * ``no-feed-or-webmention``: a :class:`web.Web` user that doesn't have
            an RSS or Atom feed or webmention endpoint and has never sent us a
            webmention
          * ``private``: the account is set to be protected or private in its
            native protocol
          * ``requires-avatar``
          * ``requires-name``
          * ``requires-old-account``

        Duplicates ``util.is_opt_out`` in Bridgy!

        https://github.com/snarfed/bridgy-fed/issues/666
        """
        if self.manual_opt_out:
            return 'opt-out'

        if not self.obj or not self.obj.as1:
            return None

        if self.obj.as1.get('bridgeable') is False:  # FEP-0036
            return 'opt-out'

        if self.REQUIRES_AVATAR and not self.obj.as1.get('image'):
            return 'requires-avatar'

        name = self.obj.as1.get('displayName')
        if self.REQUIRES_NAME and (not name or name in (self.handle, self.key.id())):
            return 'requires-name'

        if self.REQUIRES_OLD_ACCOUNT:
            if published := self.obj.as1.get('published'):
                if util.now() - util.parse_iso8601(published) < OLD_ACCOUNT_AGE:
                    return 'requires-old-account'

        summary = html_to_text(self.obj.as1.get('summary', ''), ignore_links=True)
        name = html_to_text(self.obj.as1.get('displayName', ''), ignore_links=True)

        # #nobridge overrides enabled_protocols
        if '#nobridge' in summary or '#nobridge' in name:
            return 'nobridge'

        # user has explicitly opted in. should go after spam filter (REQUIRES_*)
        # checks, but before is_public and #nobot
        #
        # !!! WARNING: keep in sync with User.enable_protocol!
        if self.enabled_protocols:
            return None

        if not as1.is_public(self.obj.as1, unlisted=False):
            return 'private'

        # enabled_protocols overrides #nobot
        if '#nobot' in summary or '#nobot' in name:
            return 'nobot'

    def is_enabled(self, to_proto, explicit=False):
        """Returns True if this user can be bridged to a given protocol.

        Reasons this might return False:
        * We haven't turned on bridging these two protocols yet.
        * The user is opted out or blocked.
        * The user is on a domain that's opted out or blocked.
        * The from protocol requires opt in, and the user hasn't opted in.
        * ``explicit`` is True, and this protocol supports ``to_proto`` by, but the user hasn't explicitly opted into it.

        Args:
          to_proto (Protocol subclass)
          explicit (bool)

        Returns:
          bool:
        """
        from protocol import Protocol
        assert issubclass(to_proto, Protocol)

        if self.__class__ == to_proto:
            return True

        from_label = self.LABEL
        to_label = to_proto.LABEL

        if bot_protocol := Protocol.for_bridgy_subdomain(self.key.id()):
            return to_proto != bot_protocol

        elif self.manual_opt_out:
            return False

        elif to_label in self.enabled_protocols:
            return True

        elif self.status:
            return False

        elif to_label in self.DEFAULT_ENABLED_PROTOCOLS and not explicit:
            return True

        return False

    def enable_protocol(self, to_proto):
        """Adds ``to_proto`` to :attr:`enabled_protocols`.

        Also sends a welcome DM to the user (via a send task) if their protocol
        supports DMs.

        Args:
          to_proto (:class:`protocol.Protocol` subclass)
        """
        import dms

        # explicit opt-in overrides some status
        # !!! WARNING: keep in sync with User.status!
        ineligible = """Hi! Your account isn't eligible for bridging yet because {desc}. <a href="https://fed.brid.gy/docs#troubleshooting">More details here.</a> You can try again once that's fixed by unfollowing and re-following this account."""
        if self.status and self.status not in ('nobot', 'private'):
            if desc := USER_STATUS_DESCRIPTIONS.get(self.status):
                dms.maybe_send(from_proto=to_proto, to_user=self, type=self.status,
                               text=ineligible.format(desc=desc))
            common.error(f'Nope, user {self.key.id()} is {self.status}', status=299)

        try:
            self.handle_as(to_proto)
        except ValueError as e:
            dms.maybe_send(from_proto=to_proto, to_user=self,
                           type=f'unsupported-handle-{to_proto.ABBREV}',
                           text=ineligible.format(desc=e))
            common.error(str(e), status=299)

        added = False

        if to_proto.LABEL in ids.COPIES_PROTOCOLS:
            # do this even if there's an existing copy since we might need to
            # reactivate it, which create_for should do
            to_proto.create_for(self)

        @ndb.transactional()
        def enable():
            user = self.key.get()
            if to_proto.LABEL not in user.enabled_protocols:
                user.enabled_protocols.append(to_proto.LABEL)
                util.add(user.sent_dms, DM(protocol=to_proto.LABEL, type='welcome'))
                user.put()
                nonlocal added
                added = True

            return user

        new_self = enable()
        # populate newly enabled protocol in this instance
        self.enabled_protocols = new_self.enabled_protocols
        self.copies = new_self.copies
        if self.obj:
            self.obj.copies = new_self.obj.copies

        if added:
            dms.maybe_send(from_proto=to_proto, to_user=self, type='welcome', text=f"""\
Welcome to Bridgy Fed! Your account will soon be bridged to {to_proto.PHRASE} at {self.user_link(proto=to_proto, name=False)}. <a href="https://fed.brid.gy/docs">See the docs</a> and <a href="https://{common.PRIMARY_DOMAIN}{self.user_page_path()}">your user page</a> for more information. To disable this and delete your bridged profile, block this account.""")

        msg = f'Enabled {to_proto.LABEL} for {self.key.id()} : {self.user_page_path()}'
        logger.info(msg)

    def disable_protocol(self, to_proto):
        """Removes ``to_proto` from :attr:`enabled_protocols``.

        Args:
          to_proto (:class:`protocol.Protocol` subclass)
        """
        @ndb.transactional()
        def disable():
            user = self.key.get()
            util.remove(user.enabled_protocols, to_proto.LABEL)
            user.put()

        disable()
        util.remove(self.enabled_protocols, to_proto.LABEL)

        msg = f'Disabled {to_proto.LABEL} for {self.key.id()} : {self.user_page_path()}'
        logger.info(msg)

    def handle_as(self, to_proto):
        """Returns this user's handle in a different protocol.

        Args:
          to_proto (str or Protocol)

        Returns:
          str
        """
        if isinstance(to_proto, str):
            to_proto = PROTOCOLS[to_proto]

        # override to-ATProto to use custom domain handle in DID doc
        from atproto import ATProto, did_to_handle
        if to_proto == ATProto:
            if did := self.get_copy(ATProto):
                if handle := did_to_handle(did, remote=False):
                    return handle

        # override web users to always use domain instead of custom username
        # TODO: fall back to id if handle is unset?
        handle = self.key.id() if self.LABEL == 'web' else self.handle
        if not handle:
            return None

        return ids.translate_handle(handle=handle, from_=self.__class__,
                                    to=to_proto, enhanced=False)

    def id_as(self, to_proto):
        """Returns this user's id in a different protocol.

        Args:
          to_proto (str or Protocol)

        Returns:
          str
        """
        if isinstance(to_proto, str):
            to_proto = PROTOCOLS[to_proto]

        return ids.translate_user_id(id=self.key.id(), from_=self.__class__,
                                     to=to_proto)

    def handle_or_id(self):
        """Returns handle if we know it, otherwise id."""
        return self.handle or self.key.id()

    def public_pem(self):
        """
        Returns:
          bytes:
        """
        rsa = RSA.construct((base64_to_long(str(self.mod)),
                             base64_to_long(str(self.public_exponent))))
        return rsa.exportKey(format='PEM')

    def private_pem(self):
        """
        Returns:
          bytes:
        """
        assert self.mod and self.public_exponent and self.private_exponent, str(self)
        rsa = RSA.construct((base64_to_long(str(self.mod)),
                             base64_to_long(str(self.public_exponent)),
                             base64_to_long(str(self.private_exponent))))
        return rsa.exportKey(format='PEM')

    def name(self):
        """Returns this user's human-readable name, eg ``Ryan Barrett``."""
        if self.obj and self.obj.as1:
            name = self.obj.as1.get('displayName')
            if name:
                return name

        return self.handle_or_id()

    def web_url(self):
        """Returns this user's web URL (homepage), eg ``https://foo.com/``.

        To be implemented by subclasses.

        Returns:
          str
        """
        raise NotImplementedError()

    def is_web_url(self, url, ignore_www=False):
        """Returns True if the given URL is this user's web URL (homepage).

        Args:
          url (str)
          ignore_www (bool): if True, ignores ``www.`` subdomains

        Returns:
          bool:
        """
        if not url:
            return False

        url = url.strip().rstrip('/')
        url = re.sub(r'^(https?://)www\.', r'\1', url)
        parsed_url = urlparse(url)
        if parsed_url.scheme not in ('http', 'https', ''):
            return False

        this = self.web_url().rstrip('/')
        this = re.sub(r'^(https?://)www\.', r'\1', this)
        parsed_this = urlparse(this)

        return (url == this or url == parsed_this.netloc or
                parsed_url[1:] == parsed_this[1:])  # ignore http vs https

    def id_uri(self):
        """Returns the user id as a URI.

        Sometimes this is the user id itself, eg ActivityPub actor ids.
        Sometimes it's a bit different, eg at://did:plc:... for ATProto user,
        https://site.com for Web users.

        Returns:
          str
        """
        return self.key.id()

    def profile_id(self):
        """Returns the id of this user's profile object in its native protocol.

        Examples:

        * Web: home page URL, eg ``https://me.com/``
        * ActivityPub: actor URL, eg ``https://instance.com/users/me``
        * ATProto: profile AT URI, eg ``at://did:plc:123/app.bsky.actor.profile/self``

        Defaults to this user's key id.

        Returns:
          str or None:
        """
        return ids.profile_id(id=self.key.id(), proto=self)

    def reload_profile(self, **kwargs):
        """Reloads this user's identity and profile from their native protocol.

        Populates the reloaded profile :class:`Object` in ``self.obj``.

        Args:
          kwargs: passed through to :meth:`Protocol.load`
        """
        obj = self.load(self.profile_id(), remote=True, **kwargs)
        if obj:
            self.obj = obj

        # write the user so that we re-populate any computed properties
        self.put()

    def user_page_path(self, rest=None, prefer_id=False):
        """Returns the user's Bridgy Fed user page path.

        Args:
          rest (str): additional path and/or query to add to the end
          prefer_id (bool): whether to prefer to use the account's id in the path
            instead of handle. Defaults to ``False``.
        """
        path = f'/{self.ABBREV}/{self.key.id() if prefer_id else self.handle_or_id()}'

        if rest:
            if not (rest.startswith('?') or rest.startswith('/')):
                path += '/'
            path += rest

        return path

    def get_copy(self, proto):
        """Returns the id for the copy of this user in a given protocol.

        ...or None if no such copy exists. If ``proto`` is this user, returns
        this user's key id.

        Args:
          proto: :class:`Protocol` subclass

        Returns:
          str:
        """
        # don't use isinstance because the testutil Fake protocol has subclasses
        if self.LABEL == proto.LABEL:
            return self.key.id()

        for copy in self.copies:
            if copy.protocol in (proto.LABEL, proto.ABBREV):
                return copy.uri

    def user_link(self, name=True, handle=True, pictures=False, proto=None,
                  proto_fallback=False):
        """Returns a pretty HTML link to the user's profile.

        Can optionally include display name, handle, profile
        picture, and/or link to a different protocol that they've enabled.

        TODO: unify with :meth:`Object.actor_link`?

        Args:
          name (bool): include display name
          handle (bool): include handle
          pictures (bool): include profile picture and protocol logo
          proto (protocol.Protocol): link to this protocol instead of the user's
            native protocol
          proto_fallback (bool): if True, and ``proto`` is provided and has no
            no canonical profile URL for bridged users, uses the user's profile
            URL in their native protocol
        """
        img = name_str = handle_str = dot = logo = a_open = a_close = ''

        if proto:
            assert self.is_enabled(proto), f"{proto.LABEL} isn't enabled"
            url = proto.bridged_web_url_for(self, fallback=proto_fallback)
        else:
            proto = self.__class__
            url = self.web_url()

        if pictures:
            logo = f'<span class="logo" title="{proto.__name__}">{proto.LOGO_HTML}</span> '
            if pic := self.profile_picture():
                img = f'<img src="{pic}" class="profile"> '

        if handle:
            handle_str = self.handle_as(proto) or ''

        if name and self.name() != handle_str:
            name_str = self.name() or ''

        if handle_str and name_str:
            dot = ' &middot; '

        if url:
            a_open = f'<a class="h-card u-author" rel="me" href="{url}" title="{name_str}{dot}{handle_str}">'
            a_close = '</a>'

        name_html = f'<span style="unicode-bidi: isolate">{ellipsize(name_str, chars=40)}</span>' if name_str else ''
        return f'{logo}{a_open}{img}{name_html}{dot}{ellipsize(handle_str, chars=40)}{a_close}'

    def profile_picture(self):
        """Returns the user's profile picture image URL, if available, or None."""
        if self.obj and self.obj.as1:
            return util.get_url(self.obj.as1, 'image')

    # can't use functools.lru_cache here because we want the cache key to be
    # just the user id, not the whole entity
    @cachetools.cached(
        cachetools.TTLCache(50000, FOLLOWERS_CACHE_EXPIRATION.total_seconds()),
        key=lambda user: user.key.id(), lock=Lock())
    @memcache.memoize(key=lambda self: self.key.id(),
                      expire=FOLLOWERS_CACHE_EXPIRATION)
    def count_followers(self):
        """Counts this user's followers and followings.

        Returns:
          (int, int) tuple: (number of followers, number following)
        """
        if self.key.id() in PROTOCOL_DOMAINS:
            # we don't store Followers for protocol bot users any more, so
            # follower counts are inaccurate, so don't return them
            return (0, 0)

        num_followers = Follower.query(Follower.to == self.key,
                                       Follower.status == 'active')\
                                .count_async()
        num_following = Follower.query(Follower.from_ == self.key,
                                       Follower.status == 'active')\
                                .count_async()
        return num_followers.get_result(), num_following.get_result()


class Object(StringIdModel):
    """An activity or other object, eg actor.

    Key name is the id. We synthesize ids if necessary.
    """
    users = ndb.KeyProperty(repeated=True)
    'User(s) who created or otherwise own this object.'

    notify = ndb.KeyProperty(repeated=True)
    """User who should see this in their user page, eg in reply to, reaction to,
    share of, etc.
    """
    feed = ndb.KeyProperty(repeated=True)
    'User who should see this in their feeds, eg followers of its creator'

    source_protocol = ndb.StringProperty(choices=list(PROTOCOLS.keys()))
    """The protocol this object originally came from.

    ``choices`` is populated in :func:`reset_protocol_properties`, after all
    :class:`User` subclasses are created, so that :attr:`PROTOCOLS` is fully
    populated.

    TODO: nail down whether this is :attr:`ABBREV`` or :attr:`LABEL`
    """

    # TODO: switch back to ndb.JsonProperty if/when they fix it for the web console
    # https://github.com/googleapis/python-ndb/issues/874
    as2 = JsonProperty()
    'ActivityStreams 2, for ActivityPub'
    bsky = JsonProperty()
    'AT Protocol lexicon, for Bluesky'
    mf2 = JsonProperty()
    'HTML microformats2 item, (ie _not_ top level parse object with ``items`` field'
    our_as1 = JsonProperty()
    'ActivityStreams 1, for activities that we generate or modify ourselves'
    raw = JsonProperty()
    'Other standalone data format, eg DID document'

    # TODO: remove and actually delete Objects instead!
    deleted = ndb.BooleanProperty()
    ''

    copies = ndb.StructuredProperty(Target, repeated=True)
    """Copies of this object elsewhere, eg at:// URIs for ATProto records and
    nevent etc bech32-encoded Nostr ids, where this object is the original.
    Similar to u-syndication links in microformats2 and
    upstream/downstreamDuplicates in AS1.
    """

    created = ndb.DateTimeProperty(auto_now_add=True)
    ''
    updated = ndb.DateTimeProperty(auto_now=True)
    ''

    new = None
    """True if this object is new, ie this is the first time we've seen it,
    False otherwise, None if we don't know.
    """
    changed = None
    """True if this object's contents have changed from our existing copy in the
    datastore, False otherwise, None if we don't know. :class:`Object` is
    new/changed. See :meth:`activity_changed()` for more details.
    """

    lock = None
    """Synchronizes :meth:`add` and :meth:`remove`."""

    # DEPRECATED
    # These were for full feeds with multiple items, not just this one, so they were
    # stored as audit records only, not used in to_as1. for Atom/RSS
    # based Objects, our_as1 was populated with an feed_index top-level