Actions

async def delete(self, attachment: Attachment) -> None:
    """Delete an attachment from local storage.

    Args:
        attachment: Attachment to delete.
    """
    await self._signal.attachments.delete(attachment)

async def update(
    self,
    update_contact: UpdateContact,
    recipient: str,
) -> None:
    """Update a contact's metadata.

    Args:
        update_contact: Contact update payload.
        recipient: The contact to update.
    """
    recipient = self._recipients.resolve(recipient)
    wire_request = update_contact.to_generated(recipient)
    await self._signal.contacts.update(wire_request)

async def about(self) -> About:
    """Return the signal-cli-rest-api about information."""
    return await self._signal.general.about()

async def update(self, update_group: UpdateGroup, group_id_or_name: str) -> None:
    """Update a group's metadata.

    Args:
        update_group: Group update payload.
        group_id_or_name: The group to update.
    """
    group_id = self._groups.resolve(group_id_or_name)
    if group_id is None:
        raise SignalBotError.cannot_resolve_recipient()

    wire_request = await update_group.to_generated()
    await self._signal.groups.update(wire_request, group_id)

async def edit(
    self, new_message: SendMessage, original_message: SentMessage
) -> SentMessage:
    """Edit a message.

    Args:
        new_message: The message to send.
        original_message: The original message to edit.

    Returns:
        A SentMessage instance.
    """
    new_message.edit_timestamp = original_message.timestamp
    return await self.send(new_message, original_message.recipient)

async def remote_delete(
    self,
    sent_message: SentMessage,
) -> int:
    """Delete a previously sent message.

    Args:
        sent_message: The message to delete.

    Returns:
        The timestamp of the delete action.
    """
    remote_delete_request = RemoteDeleteRequest(
        recipient=self._recipients.resolve(sent_message.recipient),
        timestamp=sent_message.timestamp,
    )

    remote_delete_response = await self._signal.messages.remote_delete(
        remote_delete_request
    )
    ret_timestamp = int(remote_delete_response.timestamp)
    self._logger.info(
        "[Bot] Deleted message with timestamp %s",
        sent_message.timestamp,
    )

    return ret_timestamp

async def send(
    self,
    message: SendMessage,
    recipient: str,
) -> SentMessage:
    """Send or edit a message.

    Args:
        message: The message to send.
        recipient: The contact or group to send the message to.

    Returns:
        A SentMessage instance.
    """
    recipient = self._recipients.resolve(recipient)

    send_message_v2 = await message.to_generated(self._phone_number, [recipient])
    send_message_response = await self._signal.messages.send(send_message_v2)
    timestamp = int(send_message_response.timestamp)
    self._logger.info("[Bot] New message %s sent:\n%s", timestamp, message.text)

    return SentMessage.from_send_message(message, recipient, timestamp)

async def send_multiple(
    self,
    message: SendMessage,
    recipients: list[str],
) -> list[SentMessage]:
    """Send one message to multiple recipients.

    `recipients` must be either one or more 1:1 contacts, or a single
    group. Any other combination (mixing contacts and groups, or more
    than one group) is rejected: no message is sent and a warning is
    logged.

    Args:
        message: The message to send.
        recipients: The contacts or groups to send the message to.

    Returns:
        A list of SentMessage instances, one per recipient. Empty if
        `recipients` was not a valid combination.
    """
    recipients = [self._recipients.resolve(recipient) for recipient in recipients]

    if not self._is_valid_send_multiple_recipients(recipients):
        self._logger.warning(
            "[Bot] send_multiple requires either one or more 1:1 recipients "
            "or a single group, not %s",
            recipients,
        )
        return []

    send_message_v2 = await message.to_generated(self._phone_number, recipients)
    send_message_response = await self._signal.messages.send(send_message_v2)
    timestamp = int(send_message_response.timestamp)

    self._logger.info("[Bot] New message %s sent:\n%s", timestamp, message.text)

    return SentMessage.from_send_message_multiple(message, recipients, timestamp)

async def start_typing(self, recipient: str) -> None:
    """Send a typing indicator to a recipient.

    Args:
        recipient: Message recipient.
    """
    recipient = self._recipients.resolve(recipient)
    await self._signal.messages.start_typing(
        TypingIndicatorRequest(recipient=recipient)
    )

async def stop_typing(self, recipient: str) -> None:
    """Stop a typing indicator for a recipient.

    Args:
        recipient: Message recipient.
    """
    recipient = self._recipients.resolve(recipient)
    await self._signal.messages.stop_typing(
        TypingIndicatorRequest(recipient=recipient)
    )

async def create(
    self,
    create_poll_request: CreatePoll,
    recipient: str,
) -> CreatedPoll:
    """Create a poll.

    Args:
        create_poll_request: The fields to create the poll with.
        recipient: The contact or group to send the poll to.

    Returns:
        A CreatedPoll instance.
    """
    recipient = self._recipients.resolve(recipient)

    wire_request = create_poll_request.to_generated(recipient)
    created_poll = await self._signal.polls.create(wire_request)
    timestamp = int(created_poll.timestamp)
    self._logger.info("[Bot] New poll created:\n%s", create_poll_request.question)

    return CreatedPoll.from_create_poll_request(
        create_poll_request, recipient, timestamp
    )

async def react(self, message: SentMessage | DataMessage, emoji: str) -> None:
    """React to a message with an emoji.

    Args:
        message: The message to react to.
        emoji: Emoji reaction value.
    """
    if isinstance(message, SentMessage):
        recipient = message.recipient
        target_author = self._phone_number
    else:
        recipient = message.source_or_group_id()
        target_author = message.source_uuid or message.source_number

        if message.is_group():
            recipient = self._recipients.resolve(recipient)

            if target_author is None:
                error_msg = "Cannot react to group message without source uuid"
                raise ValueError(error_msg)
        elif target_author is None:
            error_msg = "Message does not contain a source"
            raise ValueError(error_msg)

    reaction_request = SendReactionRequest(
        recipient=recipient,
        reaction=emoji,
        target_author=target_author,
        timestamp=message.timestamp,
    )
    await self._signal.reactions.react(reaction_request)
    self._logger.info("[Bot] New reaction: %s", emoji)

async def send(
    self,
    message: DataMessage | EditMessage,
    receipt_type: ReceiptType,
) -> None:
    """Send a read or viewed receipt for a message if supported.

    Args:
        message: The message to acknowledge.
        receipt_type: The receipt type to send.
    """
    if message.is_group():
        self._logger.warning("[Bot] Receipts are not supported for groups")
        return

    recipient = self._recipients.resolve(message.source_or_group_id())
    receipt_request = Receipt(
        recipient=recipient, receipt_type=receipt_type, timestamp=message.timestamp
    )
    await self._signal.receipts.send(receipt_request)
    self._logger.info("[Bot] Receipt: %s", receipt_type)