Design a Notification System

Java — Notification API orchestration: validate → rate-limit → template → enqueue

// Notification API — orchestration entry point
@RestController
@RequestMapping("/notifications")
public class NotificationController {

    @PostMapping
    public ResponseEntity<NotificationResponse> send(@RequestBody NotificationRequest req) {
        // 1. Resolve user contact details and preferences
        UserContact contact = userService.getContact(req.userId());
        Set<Channel> allowedChannels = preferenceService
            .filterChannels(req.userId(), req.channels(), req.priority());

        // 2. Render template once, reuse for all channels
        RenderedTemplate rendered = templateService.render(req.templateId(), req.variables());

        String notificationId = UUID.randomUUID().toString();
        List<ChannelEnqueueResult> results = new ArrayList<>();

        for (Channel channel : allowedChannels) {
            // 3. Per-user, per-channel rate limiting
            if (!rateLimiter.tryAcquire(req.userId(), channel, req.priority())) {
                results.add(new ChannelEnqueueResult(channel, "RATE_LIMITED"));
                continue;
            }

            // 4. Build channel-specific message and publish to Kafka
            NotificationMessage msg = NotificationMessage.builder()
                .notificationId(notificationId)
                .userId(req.userId())
                .channel(channel)
                .priority(req.priority())
                .recipient(contact.recipientFor(channel))
                .subject(rendered.subject())
                .body(rendered.body())
                .build();

            String topic = "notifications-" + channel.name().toLowerCase()
                         + "-" + req.priority().name().toLowerCase();
            kafkaTemplate.send(topic, req.userId(), msg);
            deliveryTracker.record(notificationId, channel, Status.QUEUED);
            results.add(new ChannelEnqueueResult(channel, "QUEUED"));
        }

        return ResponseEntity.accepted()
            .body(new NotificationResponse(notificationId, results));
    }
}

Java — Redis sorted-set sliding window rate limiter with Lua atomicity

@Component
public class SlidingWindowRateLimiter {

    private final RedisTemplate<String, String> redis;

    // Lua script: atomically check + increment sliding window
    private static final String RATE_LIMIT_SCRIPT = """
        local key = KEYS[1]
        local now = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local limit = tonumber(ARGV[3])
        local clearBefore = now - window

        -- Remove entries outside the window
        redis.call('ZREMRANGEBYSCORE', key, '-inf', clearBefore)

        -- Count current entries
        local count = redis.call('ZCARD', key)
        if count >= limit then
            return 0  -- rate limited
        end

        -- Add current request timestamp
        redis.call('ZADD', key, now, now .. '-' .. math.random())
        redis.call('EXPIRE', key, window / 1000 + 1)
        return 1  -- allowed
        """;

    public boolean tryAcquire(String userId, Channel channel, Priority priority) {
        // CRITICAL notifications: 100/day per channel; MARKETING: 5/day
        int limit = priority == Priority.CRITICAL ? 100 : 5;
        long windowMs = Duration.ofDays(1).toMillis();
        String key = "ratelimit:" + userId + ":" + channel.name();

        Long result = redis.execute(
            new DefaultRedisScript<>(RATE_LIMIT_SCRIPT, Long.class),
            List.of(key),
            String.valueOf(System.currentTimeMillis()),
            String.valueOf(windowMs),
            String.valueOf(limit)
        );
        return Long.valueOf(1L).equals(result);
    }
}

Java — Push channel worker with exponential backoff, jitter, circuit breaker, and DLQ

@Service
public class PushChannelWorker {

    private static final int MAX_ATTEMPTS = 5;
    private final FcmClient fcm;
    private final DeliveryTracker tracker;
    private final KafkaTemplate<String, NotificationMessage> kafka;
    private final CircuitBreaker circuitBreaker;

    @KafkaListener(topics = {"notifications-push-critical", "notifications-push-marketing"})
    public void consume(NotificationMessage msg) {
        // Idempotency guard — skip if already delivered
        if (tracker.isDelivered(msg.notificationId(), Channel.PUSH)) return;

        deliver(msg, 1);
    }

    private void deliver(NotificationMessage msg, int attempt) {
        try {
            circuitBreaker.executeRunnable(() -> {
                FcmResponse response = fcm.send(msg.recipient(), msg.subject(), msg.body(),
                    Map.of("notificationId", msg.notificationId())); // idempotency key
                tracker.markSent(msg.notificationId(), Channel.PUSH);
            });
        } catch (CallNotPermittedException e) {
            // Circuit breaker open — send to DLQ immediately
            sendToDlq(msg, "CIRCUIT_OPEN");
        } catch (Exception e) {
            if (attempt >= MAX_ATTEMPTS) {
                tracker.markFailed(msg.notificationId(), Channel.PUSH, e.getMessage());
                sendToDlq(msg, "MAX_RETRIES_EXCEEDED");
                return;
            }
            long delayMs = computeBackoff(attempt);
            log.warn("Push delivery failed (attempt {}), retrying in {}ms: {}", attempt, delayMs, e.getMessage());
            scheduleRetry(msg, attempt + 1, delayMs);
        }
    }

    private long computeBackoff(int attempt) {
        double base = 1000L * Math.pow(2, attempt - 1);  // 1s, 2s, 4s, 8s, 16s
        double jitter = 0.7 + (Math.random() * 0.6);      // ±30% jitter
        return (long) (base * jitter);
    }

    private void scheduleRetry(NotificationMessage msg, int nextAttempt, long delayMs) {
        // Use a delayed queue (Kafka with delayed topic or separate scheduler)
        NotificationMessage retryMsg = msg.withAttempt(nextAttempt);
        kafka.send("notifications-push-retry", retryMsg);
    }

    private void sendToDlq(NotificationMessage msg, String reason) {
        kafka.send("notifications-dlq", msg.withFailureReason(reason));
    }
}

Java — Delivery tracker with upsert, webhook handler, idempotency check, and stats query

// Delivery Tracker — write and query
@Repository
public class DeliveryTrackerRepository {

    private final JdbcTemplate jdbc;

    // Called by Channel Worker on each state transition
    public void updateStatus(String notificationId, Channel channel,
                              DeliveryStatus status, String detail) {
        jdbc.update("""
            INSERT INTO notification_delivery
                (notification_id, channel, status, detail, updated_at)
            VALUES (?, ?, ?, ?, NOW())
            ON CONFLICT (notification_id, channel)
            DO UPDATE SET status = EXCLUDED.status,
                          detail = EXCLUDED.detail,
                          updated_at = EXCLUDED.updated_at
            """,
            notificationId, channel.name(), status.name(), detail
        );
    }

    // Called by Webhook Receiver when provider sends delivery receipt
    public void markDelivered(String providerMessageId, Channel channel, Instant deliveredAt) {
        jdbc.update("""
            UPDATE notification_delivery
            SET status = 'DELIVERED', delivered_at = ?, updated_at = NOW()
            WHERE provider_message_id = ? AND channel = ?
            """,
            Timestamp.from(deliveredAt), providerMessageId, channel.name()
        );
    }

    public boolean isDelivered(String notificationId, Channel channel) {
        Integer count = jdbc.queryForObject("""
            SELECT COUNT(*) FROM notification_delivery
            WHERE notification_id = ? AND channel = ? AND status = 'DELIVERED'
            """,
            Integer.class, notificationId, channel.name()
        );
        return count != null && count > 0;
    }

    // Dashboard: delivery stats for last 24 hours per channel
    public List<DeliveryStats> getDailyStats() {
        return jdbc.query("""
            SELECT channel,
                   COUNT(*) FILTER (WHERE status = 'DELIVERED') AS delivered,
                   COUNT(*) FILTER (WHERE status = 'PERMANENTLY_FAILED') AS failed,
                   AVG(EXTRACT(EPOCH FROM (delivered_at - queued_at))) AS avg_latency_secs
            FROM notification_delivery
            WHERE queued_at > NOW() - INTERVAL '24 hours'
            GROUP BY channel
            """,
            (rs, row) -> new DeliveryStats(
                Channel.valueOf(rs.getString("channel")),
                rs.getLong("delivered"),
                rs.getLong("failed"),
                rs.getDouble("avg_latency_secs")
            )
        );
    }
}