@Timed
  @Path("/{destination}")
  @PUT
  @Consumes(MediaType.APPLICATION_JSON)
  @Produces(MediaType.APPLICATION_JSON)
  @ManagedAsync
    @Operation(
      summary = "Send a message",
      description = """
          Deliver a message to a single recipient. May be authenticated or unauthenticated; if unauthenticated,
          an unidentifed-access key or group-send endorsement token must be provided, unless the message is a story.
          """)
  @ApiResponse(responseCode="200", description="Message was successfully sent", useReturnTypeSchema=true)
  @ApiResponse(
      responseCode="401",
      description="The message is not a story and the authorization, unauthorized access key, or group send endorsement token is missing or incorrect")
  @ApiResponse(
      responseCode="404",
      description="The message is not a story and some the recipient service ID does not correspond to a registered Signal user")
  @ApiResponse(
      responseCode = "409", description = "Incorrect set of devices supplied for recipient",
      content = @Content(schema = @Schema(implementation = AccountMismatchedDevices[].class)))
  @ApiResponse(
      responseCode = "410", description = "Mismatched registration ids supplied for some recipient devices",
      content = @Content(schema = @Schema(implementation = AccountStaleDevices[].class)))
  public Response sendMessage(@ReadOnly @Auth Optional<AuthenticatedDevice> source,
      @Parameter(description="The recipient's unidentified access key")
      @HeaderParam(HeaderUtils.UNIDENTIFIED_ACCESS_KEY) Optional<Anonymous> accessKey,

      @Parameter(description="A group send endorsement token covering the recipient. Must not be combined with `Unidentified-Access-Key` or set on a story message.")
      @HeaderParam(HeaderUtils.GROUP_SEND_TOKEN)
      @Nullable GroupSendTokenHeader groupSendToken,

      @HeaderParam(HttpHeaders.USER_AGENT) String userAgent,

      @Parameter(description="If true, deliver the message only to recipients that are online when it is sent")
      @PathParam("destination") ServiceIdentifier destinationIdentifier,

      @Parameter(description="If true, the message is a story; access tokens are not checked and sending to nonexistent recipients is permitted")
      @QueryParam("story") boolean isStory,

      @Parameter(description="The encrypted message payloads for each recipient device")
      @NotNull @Valid IncomingMessageList messages,

      @Context ContainerRequestContext context) throws RateLimitExceededException {

    final Sample sample = Timer.start();
    try {
      if (source.isEmpty() && accessKey.isEmpty() && groupSendToken == null && !isStory) {
        throw new WebApplicationException(Response.Status.UNAUTHORIZED);
      }

      if (groupSendToken != null) {
        if (!source.isEmpty() || !accessKey.isEmpty()) {
            throw new BadRequestException("Group send endorsement tokens should not be combined with other authentication");
        } else if (isStory) {
            throw new BadRequestException("Group send endorsement tokens should not be sent for story messages");
        }
      }

      final String senderType;
      if (source.isPresent()) {
        if (source.get().getAccount().isIdentifiedBy(destinationIdentifier)) {
          senderType = SENDER_TYPE_SELF;
        } else {
          senderType = SENDER_TYPE_IDENTIFIED;
        }
      } else {
        senderType = SENDER_TYPE_UNIDENTIFIED;
      }

      boolean isSyncMessage = source.isPresent() && source.get().getAccount().isIdentifiedBy(destinationIdentifier);

      if (isSyncMessage && destinationIdentifier.identityType() == IdentityType.PNI) {
        throw new WebApplicationException(Status.FORBIDDEN);
      }

      Optional<Account> destination;
      if (!isSyncMessage) {
        destination = accountsManager.getByServiceIdentifier(destinationIdentifier);
      } else {
        destination = source.map(AuthenticatedDevice::getAccount);
      }

      final Optional<Response> spamCheck = spamChecker.checkForSpam(
          context, source.map(AuthenticatedDevice::getAccount), destination);
      if (spamCheck.isPresent()) {
        return spamCheck.get();
      }

      final Optional<byte[]> spamReportToken = switch (senderType) {
        case SENDER_TYPE_IDENTIFIED ->
        reportSpamTokenProvider.makeReportSpamToken(context, source.get(), destination);
        default -> Optional.empty();
      };

      int totalContentLength = 0;

      for (final IncomingMessage message : messages.messages()) {
        int contentLength = 0;

        if (StringUtils.isNotEmpty(message.content())) {
          contentLength += message.content().length();
        }

        validateContentLength(contentLength, false, userAgent);
        validateEnvelopeType(message.type(), userAgent);

        totalContentLength += contentLength;
      }

      try {
        rateLimiters.getInboundMessageBytes().validate(destinationIdentifier.uuid(), totalContentLength);
      } catch (final RateLimitExceededException e) {
        if (dynamicConfigurationManager.getConfiguration().getInboundMessageByteLimitConfiguration().enforceInboundLimit()) {
          messageByteLimitEstimator.add(destinationIdentifier.uuid().toString());
          throw e;
        }
      }

      try {
        if (isStory) {
          // Stories will be checked by the client; we bypass access checks here for stories.
        } else if (groupSendToken != null) {
          checkGroupSendToken(List.of(destinationIdentifier.toLibsignal()), groupSendToken);
          if (destination.isEmpty()) {
            throw new NotFoundException();
          }
        } else {
          OptionalAccess.verify(source.map(AuthenticatedDevice::getAccount), accessKey, destination,
              destinationIdentifier);
        }

        boolean needsSync = !isSyncMessage && source.isPresent() && source.get().getAccount().getDevices().size() > 1;

        // We return 200 when stories are sent to a non-existent account. Since story sends bypass OptionalAccess.verify
        // we leak information about whether a destination UUID exists if we return any other code (e.g. 404) from
        // these requests.
        if (isStory && destination.isEmpty()) {
          return Response.ok(new SendMessageResponse(needsSync)).build();
        }

        // if destination is empty we would either throw an exception in OptionalAccess.verify when isStory is false
        // or else return a 200 response when isStory is true.
        assert destination.isPresent();

        if (source.isPresent() && !isSyncMessage) {
          checkMessageRateLimit(source.get(), destination.get(), userAgent);
        }

        if (isStory) {
          rateLimiters.getStoriesLimiter().validate(destination.get().getUuid());
        }

        final Set<Byte> excludedDeviceIds;

        if (isSyncMessage) {
          excludedDeviceIds = Set.of(source.get().getAuthenticatedDevice().getId());
        } else {
          excludedDeviceIds = Collections.emptySet();
        }

        DestinationDeviceValidator.validateCompleteDeviceList(destination.get(),
            messages.messages().stream().map(IncomingMessage::destinationDeviceId).collect(Collectors.toSet()),
            excludedDeviceIds);

        DestinationDeviceValidator.validateRegistrationIds(destination.get(),
            messages.messages(),
            IncomingMessage::destinationDeviceId,
            IncomingMessage::destinationRegistrationId,
            destination.get().getPhoneNumberIdentifier().equals(destinationIdentifier.uuid()));

        final String authType;
        if (SENDER_TYPE_IDENTIFIED.equals(senderType)) {
          authType = AUTH_TYPE_IDENTIFIED;
        } else if (isStory) {
          authType = AUTH_TYPE_STORY;
        } else if (groupSendToken != null) {
          authType = AUTH_TYPE_GROUP_SEND_TOKEN;
        } else {
          authType = AUTH_TYPE_ACCESS_KEY;
        }

        final List<Tag> tags = List.of(UserAgentTagUtil.getPlatformTag(userAgent),
            Tag.of(ENDPOINT_TYPE_TAG_NAME, ENDPOINT_TYPE_SINGLE),
            Tag.of(EPHEMERAL_TAG_NAME, String.valueOf(messages.online())),
            Tag.of(SENDER_TYPE_TAG_NAME, senderType),
            Tag.of(AUTH_TYPE_TAG_NAME, authType),
            Tag.of(IDENTITY_TYPE_TAG_NAME, destinationIdentifier.identityType().name()));

        for (IncomingMessage incomingMessage : messages.messages()) {
          Optional<Device> destinationDevice = destination.get().getDevice(incomingMessage.destinationDeviceId());

          if (destinationDevice.isPresent()) {
            Metrics.counter(SENT_MESSAGE_COUNTER_NAME, tags).increment();
            sendIndividualMessage(
                source,
                destination.get(),
                destinationDevice.get(),
                destinationIdentifier,
                messages.timestamp(),
                messages.online(),
                isStory,
                messages.urgent(),
                incomingMessage,
                userAgent,
                spamReportToken);
          }
        }

        return Response.ok(new SendMessageResponse(needsSync)).build();
      } catch (MismatchedDevicesException e) {
        throw new WebApplicationException(Response.status(409)
            .type(MediaType.APPLICATION_JSON_TYPE)
            .entity(new MismatchedDevices(e.getMissingDevices(),
                    e.getExtraDevices()))
            .build());
      } catch (StaleDevicesException e) {
        throw new WebApplicationException(Response.status(410)
            .type(MediaType.APPLICATION_JSON)
            .entity(new StaleDevices(e.getStaleDevices()))
            .build());
      }
    } finally {
      sample.stop(SEND_MESSAGE_LATENCY_TIMER);
    }
  }
@Test
  void testMultiDeviceByPni() throws Exception {
    try (final Response response =
        resources.getJerseyTest()
            .target(String.format("/v1/messages/PNI:%s", MULTI_DEVICE_PNI))
            .request()
            .header("Authorization", AuthHelper.getAuthHeader(AuthHelper.VALID_UUID, AuthHelper.VALID_PASSWORD))
            .put(Entity.entity(SystemMapper.jsonMapper().readValue(jsonFixture("fixtures/current_message_multi_device_pni.json"),
                    IncomingMessageList.class),
                MediaType.APPLICATION_JSON_TYPE))) {

      assertThat("Good Response Code", response.getStatus(), is(equalTo(200)));

      verify(messageSender, times(3))
          .sendMessage(any(Account.class), any(Device.class), any(Envelope.class), eq(false));
    }
  }