Notification System

public enum NotificationChannel {
  EMAIL,
  SMS,
  PUSH
}

public enum DeliveryStatus {
  QUEUED,
  SENDING,
  SENT,
  FAILED
}

public record NotificationRequest(
      String recipientId,
      String destination,
      NotificationChannel channel,
      String subject,
      String body) {
  public NotificationRequest {
      if (recipientId == null || recipientId.isBlank()
              || destination == null || destination.isBlank()
              || channel == null || body == null || body.isBlank()) {
          throw new IllegalArgumentException("Complete notification details are required");
      }
      subject = subject == null ? "" : subject;
  }
}

public record SubmissionReceipt(String jobId) {
  public SubmissionReceipt {
      if (jobId == null || jobId.isBlank()) {
          throw new IllegalArgumentException("Job id is required");
      }
  }
}

public record DeliverySnapshot(
      String jobId,
      DeliveryStatus status,
      int attempts,
      String lastError) {
}

public record RetryPolicy(int maxAttempts) {
  public RetryPolicy {
      if (maxAttempts < 1) {
          throw new IllegalArgumentException("At least one attempt is required");
      }
  }

  public boolean canRetry(int completedAttempts) {
      return completedAttempts < maxAttempts;
  }
}

public final class DeliveryException extends Exception {
  public DeliveryException(String message) {
      super(message);
  }
}

public interface NotificationSender {
  NotificationChannel channel();

  void send(NotificationRequest request) throws DeliveryException;
}

public final class EmailSender implements NotificationSender {
  @Override
  public NotificationChannel channel() {
      return NotificationChannel.EMAIL;
  }

  @Override
  public void send(NotificationRequest request) throws DeliveryException {
      System.out.println("EMAIL -> " + request.destination() + ": " + request.body());
  }
}

public final class SmsSender implements NotificationSender {
  @Override
  public NotificationChannel channel() {
      return NotificationChannel.SMS;
  }

  @Override
  public void send(NotificationRequest request) throws DeliveryException {
      System.out.println("SMS -> " + request.destination() + ": " + request.body());
  }
}

public final class PushSender implements NotificationSender {
  @Override
  public NotificationChannel channel() {
      return NotificationChannel.PUSH;
  }

  @Override
  public void send(NotificationRequest request) throws DeliveryException {
      System.out.println("PUSH -> " + request.destination() + ": " + request.body());
  }
}

import java.util.Objects;

final class DeliveryJob {
  private final String jobId;
  private final NotificationRequest request;
  private DeliveryStatus status = DeliveryStatus.QUEUED;
  private int attempts;
  private String lastError;

  DeliveryJob(String jobId, NotificationRequest request) {
      this.jobId = Objects.requireNonNull(jobId);
      this.request = Objects.requireNonNull(request);
  }

  NotificationRequest request() {
      return request;
  }

  synchronized void startAttempt() {
      if (status == DeliveryStatus.SENT || status == DeliveryStatus.FAILED) {
          throw new IllegalStateException("Terminal job cannot restart");
      }
      attempts++;
      status = DeliveryStatus.SENDING;
  }

  synchronized void recordAttemptFailure(String message) {
      if (status != DeliveryStatus.SENDING) {
          throw new IllegalStateException("No active attempt");
      }
      lastError = Objects.requireNonNull(message);
  }

  synchronized void markSent() {
      if (status != DeliveryStatus.SENDING) {
          throw new IllegalStateException("No active attempt");
      }
      status = DeliveryStatus.SENT;
      lastError = null;
  }

  synchronized void markFailed() {
      if (status != DeliveryStatus.SENDING) {
          throw new IllegalStateException("No active attempt");
      }
      status = DeliveryStatus.FAILED;
  }

  synchronized int attempts() {
      return attempts;
  }

  synchronized DeliverySnapshot snapshot() {
      return new DeliverySnapshot(jobId, status, attempts, lastError);
  }
}

import java.util.EnumMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

public final class NotificationService implements AutoCloseable {
  private final Map<NotificationChannel, NotificationSender> senders;
  private final ConcurrentMap<String, DeliveryJob> jobs = new ConcurrentHashMap<>();
  private final ExecutorService executor;
  private final RetryPolicy retryPolicy;
  private final AtomicLong sequence = new AtomicLong(1);
  private final AtomicBoolean accepting = new AtomicBoolean(true);

  public NotificationService(List<NotificationSender> senders,
          ExecutorService executor, RetryPolicy retryPolicy) {
      this.executor = Objects.requireNonNull(executor);
      this.retryPolicy = Objects.requireNonNull(retryPolicy);
      EnumMap<NotificationChannel, NotificationSender> registry =
              new EnumMap<>(NotificationChannel.class);
      for (NotificationSender sender : List.copyOf(senders)) {
          if (registry.putIfAbsent(sender.channel(), sender) != null) {
              throw new IllegalArgumentException("Duplicate sender: " + sender.channel());
          }
      }
      this.senders = Map.copyOf(registry);
  }

  public SubmissionReceipt submit(NotificationRequest request) {
      Objects.requireNonNull(request);
      if (!accepting.get()) {
          throw new IllegalStateException("Service is closed");
      }
      if (!senders.containsKey(request.channel())) {
          throw new IllegalArgumentException("No sender for " + request.channel());
      }
      String jobId = "N" + String.format("%03d", sequence.getAndIncrement());
      DeliveryJob job = new DeliveryJob(jobId, request);
      jobs.put(jobId, job);
      try {
          executor.submit(() -> deliver(job));
      } catch (RejectedExecutionException failure) {
          jobs.remove(jobId);
          throw failure;
      }
      return new SubmissionReceipt(jobId);
  }

  private void deliver(DeliveryJob job) {
      NotificationSender sender = senders.get(job.request().channel());
      while (true) {
          job.startAttempt();
          try {
              sender.send(job.request());
              job.markSent();
              return;
          } catch (DeliveryException failure) {
              String message = failure.getMessage() == null
                      ? "Delivery failed"
                      : failure.getMessage();
              job.recordAttemptFailure(message);
              if (!retryPolicy.canRetry(job.attempts())) {
                  job.markFailed();
                  return;
              }
          }
      }
  }

  public Optional<DeliverySnapshot> getStatus(String jobId) {
      if (jobId == null || jobId.isBlank()) {
          return Optional.empty();
      }
      DeliveryJob job = jobs.get(jobId);
      return job == null ? Optional.empty() : Optional.of(job.snapshot());
  }

  @Override
  public void close() {
      if (accepting.compareAndSet(true, false)) {
          executor.shutdown();
      }
  }
}

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public final class NotificationDemo {
  public static void main(String[] args) throws InterruptedException {
      ExecutorService workers = Executors.newFixedThreadPool(2);
      try (NotificationService service = new NotificationService(
              List.of(new EmailSender(), new SmsSender(), new PushSender()),
              workers,
              new RetryPolicy(3))) {
          SubmissionReceipt receipt = service.submit(new NotificationRequest(
                  "U1", "learner@example.com", NotificationChannel.EMAIL,
                  "Interview update", "Your interview starts at 3 PM."));
          System.out.println("Accepted " + receipt.jobId());
          workers.shutdown();
          workers.awaitTermination(2, TimeUnit.SECONDS);
          System.out.println(service.getStatus(receipt.jobId()).orElseThrow());
      }
  }
}