Design for Notification aggregate

Architectural Concepts
1

Hello,
I have a SendNotification command. The actual sending of the mail occurs in the constructor, but I’m thinking that it might be better to separate the command handler from the email sending logic. Seems like we had some sort of network “blip” and the same command was sent multiple times which resulted in multiple notifications being sent and lots of errors stating: [AXONIQ-2000] Invalid sequence number 0 for aggregate , expected 1. I understand what the message means, but I’m curious what might have lead to that? There were SMTP related errors in the log as well where an exception might have been thrown from the handler.

What I think might have happened is that the command was published, but there was no response because the handler was maybe hung up. Then it was retried. Both commands eventually sent the email and tried to publish the event. One went thru, but the other resulted in AXONIQ-2000.

Aside from that, in order to separate things, I’m thinking that perhaps from the constructor, I just set a really short deadline so that the sending occurs in a different thread. But, is there still some sort of retry mechanism if there’s an exception?

2

Your approach here is wrong. Your command handler should only be validating the notification creation which in result should only create the notification in ready to send state, no external processing. On all ready to send notifications you execute actual SendNotificationCommand which in result emits appropriate event. Another event handler component now actually listens to these events and actually sends the notification through external service. When sending is successful or failed you can execute appropriate command to reflect your notification state.

3

Yep, I agree with this statement. Assuming the command handler for the SendNotification is in an Aggregate, I would not perform any other tasks aside state validation and event publication.

Although there are outliers when this is necessary, 9 out of 10 it is best to separate that concern.

So, I agree with this statement from @vaelyr:

4

Thanks for the responses. Sorry for the delay, this is a work-related question and priorities can change daily so I haven’t been checking back. Anyhow…

Here’s how I envision it…
SendNotification → NotificationCreated → Send messages in different thread → NotificationSent.

SendNotification contains the plain data to be used in the message along with a list of recipients, etc…
The command handler performs the validation and publishes NotificationCreated?
Can you elaborate on the proper way to get to the next step? I know it’s not a saga because it’s the same aggregate. This is where I was thinking a deadline or something, but not completely sure.

In NotificationSent, I include details that pertain to the actual message(s) as well as some timing info. I actually use it to construct a PDF in the primary system for people who didn’t receive the message.

5

FTR, If I publish an event from a @DeadlineHandler, I see it in the event store. However, my @EventSourcingHandler doesn’t pick it up if I re-run the @DeadlineHandler. If it allows me to publish, I don’t quite understand why it won’t read.

6

No worries, Brian! I think we are all familiar with shifting priorities because of work.

I think that I would have the steps “Send messages in different thread” and “NotificationSent” performed by a regular Event Handling Component.
So, to provide some pseudo-code:

@ProcessingGroup("notification-sender")
class NotificationSender {

    private final EventGateway eventGateway = ...;
    private final NotificationService notificationService = ...;

    @EventHandler
    public void on(NotificationCreated event) {
        try {
            notificationService.send(event.notification());
            eventGateway.publish(new NotificationSent(event.identifier(), ...);
        } catch (Exception e) {
            // React as desired to failures from the NotificationService or EventGateway.
        }
    }
}

I think the difference between my setup and your suggestion, @Brian_Sanders, is clear. I suggest using an Event Handler as part of an Event Processor that handles the event. Upon handling, I invoked the notification service to send the event. Once that completes successfully, I publish the NotificationSent event.

Taking my flow does mean that the NotificationSent event is not part of the aggregate’s event stream. In other words, it will not have the aggregateIdentifier and sequenceNumber fields set as other events originating from the aggregate.

However, I don’t think that’s a big deal. A NotificationSent, to me, sounds like a technical event. A pure trigger for others to know that something occurred. So, it is not necessarily something that is important for the aggregate to know until the end of time.

Whether that is okay for your domain is, of course, an assumption on my part! So, if this isn’t feasible, just let me know.

1 Like
7

Hello. Just wanted to follow up on this. While I did end up trying out the way I had envisioned, I will keep your suggestion in mind for the future just to keep my options open. It’s been in production for about 8 days now and it is actually pretty solid now. I’ve seen exceptions in the log and it just kept chugging along. This is the trimmed down version of the code…

@Aggregate
@Slf4j
@Data
@NoArgsConstructor
public class Notification {

    @AggregateIdentifier
    private String id;
    
    @CommandHandler
    public Notification(SendNotification cmd, DeadlineManager deadlineManager) {
        log.info("Queueing : " + cmd.toString().replaceAll("\n", "\\n"));
        apply(new NotificationQueued(cmd.getId()));
        deadlineManager.schedule(Duration.ofSeconds(10), "sendNotifications", cmd);
    }
    
    @EventSourcingHandler
    public void queued(NotificationQueued event) {
        setId(event.getId());
    }
    
    @DeadlineHandler(deadlineName = "sendNotifications")
    public void sendNotifications(
        SendNotification cmd,
        ...
        EmailService emailService
    ) {
        log.info(cmd.toString().replaceAll("\n", "\\n"));

        ...
        
        StopWatch sendAllMessagesTimer = new StopWatch();

        sendAllMessagesTimer.start();

        Map<Integer, Message> messageByUserId = toUsers.parallelStream()
            .map(user -> {
                try {
                    StopWatch sendMessageTimer = new StopWatch();
                    
                    sendMessageTimer.start();
                    
                    MimeMessage mimeMessage = emailService.sendEmail(Email.builder()
                        .from(new InternetAddress(fromUser.getEmail(), fromUser.getName()))
                        .to(new InternetAddress(user.getEmail(), user.getName()))
                        .subject(subject)
                        .attachments(attachments)
                        .body(body)
                        .retry(Retry.builder()
                            .limit(3)
                            .min(Duration.ofMinutes(1))
                            .max(Duration.ofMinutes(5))
                            .build())
                        .build());
                    
                    sendMessageTimer.stop();
                    
                    return Map.entry(user.getId(), Message.builder()
                        .id(mimeMessage.getMessageID())
                        .sentDate(mimeMessage.getSentDate().toInstant())
                        .duration(sendMessageTimer.getTotalTimeMillis())
                        .build());
                }
                catch (Exception e) {
                    log.error("Couldn't send notification: " + new MapBuilder<>()
                        .put("inspectionId", cmdCopy.getInspection().getId())
                        .put("userId", user.getId())
                        .put("email", user.getEmail()).toMap(), e);
                }
                
                return null;
            })
        .filter(Objects::nonNull)
        .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));

        sendAllMessagesTimer.stop();
        
        log.info("NotificationSent : " + cmd.getId());
        
        AggregateLifecycle.apply(NotificationSent.builder()
            .id(cmd.getId())
            .inspectionId(cmd.getInspection().getId().toString())
            .fromUserId(fromUser.getId())
            .messageByUserId(messageByUserId)
            .attachments(new ArrayList<>(attachments.keySet()))
            .subject(subject)
            .body(body)
            .duration(sendAllMessagesTimer.getTotalTimeMillis())
            .build());
    }
    ...
}
1 Like