public void report(final Optional<String> sourceNumber,
      final Optional<UUID> sourceAci,
      final Optional<UUID> sourcePni,
      final UUID messageGuid,
      final UUID reporterUuid,
      final Optional<byte[]> reportSpamToken,
      final String reporterUserAgent) {

    final boolean found = sourceAci.map(uuid -> reportMessageDynamoDb.remove(hash(messageGuid, uuid.toString())))
        .orElse(false);

    Metrics.counter(REPORT_MESSAGE_COUNTER_NAME,
            Tags.of(FOUND_MESSAGE_TAG, String.valueOf(found),
                    TOKEN_PRESENT_TAG, String.valueOf(reportSpamToken.isPresent()))
                .and(UserAgentTagUtil.getPlatformTag(reporterUserAgent)))
        .increment();

    if (found) {
      rateLimitCluster.useCluster(connection -> {
        sourcePni.ifPresent(pni -> {
          final String reportedSenderKey = getReportedSenderPniKey(pni);
          connection.sync().pfadd(reportedSenderKey, reporterUuid.toString());
          connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
        });

        sourceAci.ifPresent(aci -> {
          final String reportedSenderKey = getReportedSenderAciKey(aci);
          connection.sync().pfadd(reportedSenderKey, reporterUuid.toString());
          connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
        });
      });

      sourceNumber.ifPresent(number ->
          reportedMessageListeners.forEach(listener -> {
            try {
              listener.handleMessageReported(number, messageGuid, reporterUuid, reportSpamToken);
            } catch (final Exception e) {
              logger.error("Failed to notify listener of reported message", e);
            }
          }));
    }
  }
@Test
  void testReport() {
    final ReportedMessageListener listener = mock(ReportedMessageListener.class);
    reportMessageManager.addListener(listener);

    when(reportMessageDynamoDb.remove(any())).thenReturn(false);
    reportMessageManager.report(Optional.of(sourceNumber), Optional.of(sourceAci), Optional.of(sourcePni), messageGuid,
        reporterUuid, Optional.empty(), "user-agent");

    assertEquals(0, reportMessageManager.getRecentReportCount(sourceAccount));

    when(reportMessageDynamoDb.remove(any())).thenReturn(true);
    reportMessageManager.report(Optional.of(sourceNumber), Optional.of(sourceAci), Optional.of(sourcePni), messageGuid,
        reporterUuid, Optional.empty(), "user-agent");

    assertEquals(1, reportMessageManager.getRecentReportCount(sourceAccount));
    verify(listener).handleMessageReported(sourceNumber, messageGuid, reporterUuid, Optional.empty());
  }