Design a Notification Service.

# notify.py, part 1: messages, channels, templates and preferences
import heapq
import itertools
import re
import threading
from abc import ABC, abstractmethod
from collections.abc import Callable
from dataclasses import dataclass, field
from enum import Enum


class Priority(Enum):
    NORMAL = 1
    CRITICAL = 2  # a security code: ignores quiet hours


class SendResult(Enum):  # what a channel says about one attempt
    OK = 1
    RETRY = 2      # a temporary failure: a timeout, a busy provider
    PERMANENT = 3  # a dead end: a bad address, a revoked token


class Status(Enum):
    PENDING = 1
    SENT = 2
    RETRYING = 3
    DEFERRED = 4
    DROPPED = 5
    FAILED = 6


class MissingParam(Exception):
    pass


@dataclass(frozen=True)
class Notification:  # what the caller wants said, to whom, and why
    user_id: str
    category: str  # "order", "promo", "security"
    template: str
    params: dict
    dedupe_key: str | None = None
    priority: Priority = Priority.NORMAL


class Channel(ABC):  # an adapter over one provider
    name: str

    @abstractmethod
    def send(self, address: str, text: str) -> SendResult: ...


@dataclass(frozen=True)
class Template:
    id: str
    bodies: dict  # channel name -> text with {placeholders}

    def render(self, channel: str, params: dict) -> str:
        def fill(m: re.Match) -> str:
            if m.group(1) not in params:
                raise MissingParam(f"{self.id}: no value for {m.group(1)}")
            return str(params[m.group(1)])

        return re.sub(r"\{(\w+)\}", fill, self.bodies[channel])


@dataclass
class Preferences:
    muted: set = field(default_factory=set)  # {(category, channel)} the user switched off
    quiet: tuple | None = None               # local hours (start, end): no non-critical sends inside
    tz: int = 0                              # hours from UTC


@dataclass
class User:
    uid: str
    addresses: dict  # channel name -> address
    prefs: Preferences = field(default_factory=Preferences)


@dataclass
class Delivery:  # one message on one channel: the unit that is sent, retried or dropped
    id: int
    channel: str
    notif: Notification
    text: str
    status: Status = Status.PENDING
    attempts: int = 0
    detail: str = ""

// NotifyDemo.java: channels, templates, preferences, a rule chain, retries with backoff, and a dedupe window
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.IntUnaryOperator;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

enum Priority {
    NORMAL,
    CRITICAL // a security code: ignores quiet hours
}

enum SendResult { // what a channel says about one attempt
    OK,
    RETRY,     // a temporary failure: a timeout, a busy provider
    PERMANENT  // a dead end: a bad address, a revoked token
}

enum Status { PENDING, SENT, RETRYING, DEFERRED, DROPPED, FAILED }

class MissingParam extends RuntimeException {
    MissingParam(String message) { super(message); }
}

/** What the caller wants said, to whom, and why. */
record Notification(String userId, String category, String template, Map<String, String> params,
                    String dedupeKey, Priority priority) {}

/** An adapter over one provider. */
interface Channel {
    String name();

    SendResult send(String address, String text);
}

record Template(String id, Map<String, String> bodies) { // channel name -> text with {placeholders}
    String render(String channel, Map<String, String> params) {
        Matcher m = Pattern.compile("\\{(\\w+)\\}").matcher(bodies.get(channel));
        StringBuilder out = new StringBuilder();
        while (m.find()) {
            if (!params.containsKey(m.group(1))) throw new MissingParam(id + ": no value for " + m.group(1));
            m.appendReplacement(out, Matcher.quoteReplacement(params.get(m.group(1))));
        }
        return m.appendTail(out).toString();
    }
}

final class Preferences {
    final Set<String> muted = new HashSet<>(); // "category/channel" the user switched off
    int[] quiet;                               // local hours {start, end}: no non-critical sends inside; null for none
    int tz;                                    // hours from UTC
}

record User(String uid, Map<String, String> addresses, Preferences prefs) {}

/** One message on one channel: the unit that is sent, retried or dropped. */
final class Delivery {
    final int id;
    final String channel;
    final Notification notif;
    final String text;
    Status status = Status.PENDING;
    int attempts = 0;
    String detail = "";

    Delivery(int id, String channel, Notification notif, String text) {
        this.id = id;
        this.channel = channel;
        this.notif = notif;
        this.text = text;
    }
}

// notify.cpp: channels, templates, preferences, a rule chain, retries with backoff, and a dedupe window
#include <algorithm>
#include <cstdio>
#include <functional>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <regex>
#include <set>
#include <stdexcept>
#include <string>
#include <tuple>
#include <utility>
#include <vector>

enum class Priority { Normal, Critical };  // Critical: a security code that ignores quiet hours

enum class SendResult {  // what a channel says about one attempt
    Ok,
    Retry,     // a temporary failure: a timeout, a busy provider
    Permanent  // a dead end: a bad address, a revoked token
};

enum class Status { Pending, Sent, Retrying, Deferred, Dropped, Failed };

std::string name(Status s) {
    static const char* names[] = {"PENDING", "SENT", "RETRYING", "DEFERRED", "DROPPED", "FAILED"};
    return names[static_cast<int>(s)];
}

struct MissingParam : std::runtime_error { using std::runtime_error::runtime_error; };

using Fields = std::vector<std::pair<std::string, std::string>>;  // an ordered map: small and enough here

const std::string* find(const Fields& f, const std::string& key) {
    for (const auto& e : f) if (e.first == key) return &e.second;
    return nullptr;
}

struct Notification {  // what the caller wants said, to whom, and why
    std::string userId, category, templateId;
    Fields params;
    std::string dedupeKey;  // empty: no dedupe
    Priority priority = Priority::Normal;
};

class Channel {  // an adapter over one provider
public:
    virtual ~Channel() = default;
    virtual SendResult send(const std::string& address, const std::string& text) = 0;
};

struct Template {
    std::string id;
    Fields bodies;  // channel name -> text with {placeholders}

    std::string render(const std::string& channel, const Fields& params) const {
        std::string body = *find(bodies, channel), out;
        std::regex re("\\{(\\w+)\\}");
        auto last = body.cbegin();
        for (std::sregex_iterator it(body.begin(), body.end(), re), end; it != end; ++it) {
            const std::string* v = find(params, (*it)[1]);
            if (!v) throw MissingParam(id + ": no value for " + (*it)[1].str());
            out.append(last, (*it)[0].first).append(*v);
            last = (*it)[0].second;
        }
        return out.append(last, body.cend());
    }
};

struct Preferences {
    std::set<std::string> muted;  // "category/channel" the user switched off
    bool hasQuiet = false;
    int quietStart = 0, quietEnd = 0;  // local hours: no non-critical sends inside
    int tz = 0;                        // hours from UTC
};

struct User {
    std::string uid;
    Fields addresses;  // channel name -> address
    Preferences prefs;
};

struct Delivery {  // one message on one channel: the unit that is sent, retried or dropped
    int id;
    std::string channel;
    Notification notif;
    std::string text;
    Status status = Status::Pending;
    int attempts = 0;
    std::string detail;
};

// notify.js: channels, templates, preferences, a rule chain, retries with backoff, and a dedupe window
const Priority = Object.freeze({ NORMAL: "NORMAL", CRITICAL: "CRITICAL" }); // CRITICAL: a security code that ignores quiet hours
const SendResult = Object.freeze({
  OK: "OK",
  RETRY: "RETRY", // a temporary failure: a timeout, a busy provider
  PERMANENT: "PERMANENT", // a dead end: a bad address, a revoked token
});
const Status = Object.freeze({ PENDING: "PENDING", SENT: "SENT", RETRYING: "RETRYING", DEFERRED: "DEFERRED", DROPPED: "DROPPED", FAILED: "FAILED" });

class MissingParam extends Error {}

// what the caller wants said, to whom, and why
const notification = (userId, category, template, params, dedupeKey = null, priority = Priority.NORMAL) =>
  Object.freeze({ userId, category, template, params, dedupeKey, priority });

class Channel { // an adapter over one provider
  send(address, text) { throw new Error("not implemented"); } // returns a SendResult
}

class Template {
  constructor(id, bodies) {
    this.id = id;
    this.bodies = bodies; // channel name -> text with {placeholders}
  }

  render(channel, params) {
    return this.bodies[channel].replace(/\{(\w+)\}/g, (_, key) => {
      if (!(key in params)) throw new MissingParam(`${this.id}: no value for ${key}`);
      return String(params[key]);
    });
  }
}

class Preferences {
  constructor({ muted = [], quiet = null, tz = 0 } = {}) {
    this.muted = new Set(muted); // "category/channel" the user switched off
    this.quiet = quiet; // local hours [start, end]: no non-critical sends inside
    this.tz = tz; // hours from UTC
  }
}

class User {
  constructor(uid, addresses, prefs = new Preferences()) {
    Object.assign(this, { uid, addresses, prefs }); // addresses: channel name -> address
  }
}

// one message on one channel: the unit that is sent, retried or dropped
class Delivery {
  constructor(id, channel, notif, text) {
    Object.assign(this, { id, channel, notif, text });
    this.status = Status.PENDING;
    this.attempts = 0;
    this.detail = "";
  }
}

# part 2: the rule chain (opt-out, quiet hours), retry backoff and the dedupe window
@dataclass(frozen=True)
class Decision:
    kind: str  # "allow", "drop" or "defer"
    reason: str = ""
    until: int = 0


ALLOW = Decision("allow")


class Rule(ABC):  # one link in the chain: allow, drop, or defer to a later time
    @abstractmethod
    def check(self, d: Delivery, user: User, now: int) -> Decision: ...


class OptOutRule(Rule):
    def check(self, d, user, now):
        if (d.notif.category, d.channel) in user.prefs.muted:
            return Decision("drop", "opted out")
        return ALLOW


class QuietHoursRule(Rule):
    def check(self, d, user, now):
        if user.prefs.quiet is None or d.notif.priority is Priority.CRITICAL:
            return ALLOW
        start, end = (h * 3600 for h in user.prefs.quiet)
        local = (now + user.prefs.tz * 3600) % 86400
        inside = start <= local < end if start <= end else local >= start or local < end
        return Decision("defer", "quiet hours", now + (end - local) % 86400) if inside else ALLOW


class BackoffPolicy(ABC):
    @abstractmethod
    def delay(self, failures: int) -> int | None:
        """Seconds to wait before the next try, or None to give up."""


class ExponentialBackoff(BackoffPolicy):
    def __init__(self, base: int = 30, cap: int = 600, max_attempts: int = 4,
                 jitter: Callable[[int], int] = lambda failures: 0) -> None:
        self.base, self.cap, self.max_attempts, self.jitter = base, cap, max_attempts, jitter  # jitter is injected

    def delay(self, failures: int) -> int | None:
        if failures >= self.max_attempts:
            return None
        return min(self.cap, self.base * 2 ** (failures - 1)) + self.jitter(failures)


class DedupeWindow:
    def __init__(self, window: int) -> None:
        self.window, self.seen = window, {}

    def first(self, key: tuple, now: int) -> bool:  # True the first time a key is seen inside the window
        if self.seen.get(key, 0) > now:
            return False
        self.seen[key] = now + self.window
        return True

record Decision(String kind, String reason, int until) { // kind: "allow", "drop" or "defer"
    static final Decision ALLOW = new Decision("allow", "", 0);
}

/** One link in the chain: allow, drop, or defer to a later time. */
interface Rule {
    Decision check(Delivery d, User user, int now);
}

final class OptOutRule implements Rule {
    public Decision check(Delivery d, User user, int now) {
        if (user.prefs().muted.contains(d.notif.category() + "/" + d.channel)) return new Decision("drop", "opted out", 0);
        return Decision.ALLOW;
    }
}

final class QuietHoursRule implements Rule {
    public Decision check(Delivery d, User user, int now) {
        int[] quiet = user.prefs().quiet;
        if (quiet == null || d.notif.priority() == Priority.CRITICAL) return Decision.ALLOW;
        int start = quiet[0] * 3600, end = quiet[1] * 3600;
        int local = Math.floorMod(now + user.prefs().tz * 3600, 86400);
        boolean inside = start <= end ? (start <= local && local < end) : (local >= start || local < end);
        return inside ? new Decision("defer", "quiet hours", now + Math.floorMod(end - local, 86400)) : Decision.ALLOW;
    }
}

interface BackoffPolicy {
    /** Seconds to wait before the next try, or null to give up. */
    Integer delay(int failures);
}

final class ExponentialBackoff implements BackoffPolicy {
    final int base, cap, maxAttempts;
    final IntUnaryOperator jitter; // injected

    ExponentialBackoff(int base, int cap, int maxAttempts, IntUnaryOperator jitter) {
        this.base = base;
        this.cap = cap;
        this.maxAttempts = maxAttempts;
        this.jitter = jitter;
    }

    public Integer delay(int failures) {
        if (failures >= maxAttempts) return null;
        return Math.min(cap, base * (1 << (failures - 1))) + jitter.applyAsInt(failures);
    }
}

final class DedupeWindow {
    final int window;
    final Map<String, Integer> seen = new HashMap<>();

    DedupeWindow(int window) { this.window = window; }

    boolean first(String key, int now) { // true the first time a key is seen inside the window
        if (seen.getOrDefault(key, 0) > now) return false;
        seen.put(key, now + window);
        return true;
    }
}

struct Decision {
    std::string kind;  // "allow", "drop" or "defer"
    std::string reason;
    int until = 0;
};

const Decision ALLOW{"allow", "", 0};

class Rule {  // one link in the chain: allow, drop, or defer to a later time
public:
    virtual ~Rule() = default;
    virtual Decision check(const Delivery& d, const User& user, int now) const = 0;
};

class OptOutRule : public Rule {
public:
    Decision check(const Delivery& d, const User& user, int now) const override {
        if (user.prefs.muted.count(d.notif.category + "/" + d.channel)) return {"drop", "opted out", 0};
        return ALLOW;
    }
};

class QuietHoursRule : public Rule {
public:
    Decision check(const Delivery& d, const User& user, int now) const override {
        const Preferences& p = user.prefs;
        if (!p.hasQuiet || d.notif.priority == Priority::Critical) return ALLOW;
        int start = p.quietStart * 3600, end = p.quietEnd * 3600;
        int local = ((now + p.tz * 3600) % 86400 + 86400) % 86400;
        bool inside = start <= end ? (start <= local && local < end) : (local >= start || local < end);
        return inside ? Decision{"defer", "quiet hours", now + ((end - local) % 86400 + 86400) % 86400} : ALLOW;
    }
};

class BackoffPolicy {
public:
    virtual ~BackoffPolicy() = default;
    // Seconds to wait before the next try, or nullopt to give up.
    virtual std::optional<int> delay(int failures) const = 0;
};

class ExponentialBackoff : public BackoffPolicy {
public:
    ExponentialBackoff(int base, int cap, int maxAttempts, std::function<int(int)> jitter)
        : base_(base), cap_(cap), maxAttempts_(maxAttempts), jitter_(std::move(jitter)) {}  // jitter is injected

    std::optional<int> delay(int failures) const override {
        if (failures >= maxAttempts_) return std::nullopt;
        return std::min(cap_, base_ * (1 << (failures - 1))) + jitter_(failures);
    }

private:
    int base_, cap_, maxAttempts_;
    std::function<int(int)> jitter_;
};

class DedupeWindow {
public:
    explicit DedupeWindow(int window) : window_(window) {}

    bool first(const std::string& key, int now) {  // true the first time a key is seen inside the window
        auto it = seen_.find(key);
        if (it != seen_.end() && it->second > now) return false;
        seen_[key] = now + window_;
        return true;
    }

private:
    int window_;
    std::map<std::string, int> seen_;
};

const ALLOW = Object.freeze({ kind: "allow" });
const mod = (a, n) => ((a % n) + n) % n;

class Rule { // one link in the chain: allow, drop, or defer to a later time
  check(d, user, now) { throw new Error("not implemented"); } // {kind: "allow" | "drop" | "defer", reason, until}
}

class OptOutRule extends Rule {
  check(d, user, now) {
    return user.prefs.muted.has(`${d.notif.category}/${d.channel}`) ? { kind: "drop", reason: "opted out" } : ALLOW;
  }
}

class QuietHoursRule extends Rule {
  check(d, user, now) {
    if (!user.prefs.quiet || d.notif.priority === Priority.CRITICAL) return ALLOW;
    const [start, end] = user.prefs.quiet.map((h) => h * 3600);
    const local = mod(now + user.prefs.tz * 3600, 86400);
    const inside = start <= end ? start <= local && local < end : local >= start || local < end;
    return inside ? { kind: "defer", reason: "quiet hours", until: now + mod(end - local, 86400) } : ALLOW;
  }
}

class BackoffPolicy {
  delay(failures) { throw new Error("not implemented"); } // seconds before the next try, or null to give up
}

class ExponentialBackoff extends BackoffPolicy {
  constructor({ base = 30, cap = 600, maxAttempts = 4, jitter = () => 0 } = {}) {
    super();
    Object.assign(this, { base, cap, maxAttempts, jitter }); // jitter is injected
  }

  delay(failures) {
    if (failures >= this.maxAttempts) return null;
    return Math.min(this.cap, this.base * 2 ** (failures - 1)) + this.jitter(failures);
  }
}

class DedupeWindow {
  constructor(window) {
    this.window = window;
    this.seen = new Map();
  }

  first(key, now) { // true the first time a key is seen inside the window
    if ((this.seen.get(key) ?? 0) > now) return false;
    this.seen.set(key, now + this.window);
    return true;
  }
}

# part 3: the service: fan out to channels, run the rules, send outside the lock, retry or give up
class NotificationService:
    def __init__(self, users: dict, templates: dict, channels: dict, rules: list[Rule],
                 backoff: BackoffPolicy, dedupe_window: int = 3600) -> None:
        self.users, self.templates, self.channels, self.rules, self.backoff = users, templates, channels, rules, backoff
        self.dedupe = DedupeWindow(dedupe_window)
        self.queue: list = []  # heap of (due time, delivery id, delivery)
        self.listeners: list[Callable[[Delivery], None]] = []
        self._ids = itertools.count(1)
        self._lock = threading.RLock()  # guards the queue and the dedupe window, never a provider call

    def notify(self, n: Notification, now: int) -> list[Delivery]:
        with self._lock:
            user, tpl = self.users[n.user_id], self.templates[n.template]
            texts = {ch: tpl.render(ch, n.params) for ch in tpl.bodies if ch in user.addresses}  # a missing value raises here, before anything is queued
            made = []
            for ch, text in texts.items():
                d = Delivery(next(self._ids), ch, n, text)
                made.append(d)
                if n.dedupe_key and not self.dedupe.first((n.user_id, ch, n.dedupe_key), now):
                    d.status, d.detail = Status.DROPPED, "duplicate"
                else:
                    heapq.heappush(self.queue, (now, d.id, d))
        for d in made:
            if d.status is Status.DROPPED:
                self._emit(d)
        self.tick(now)
        return made

    def tick(self, now: int) -> None:  # run everything due; the caller's clock drives retries and deferrals
        with self._lock:
            due = []
            while self.queue and self.queue[0][0] <= now:
                due.append(heapq.heappop(self.queue)[2])
        for d in due:
            user = self.users[d.notif.user_id]
            verdict = next((v for v in (r.check(d, user, now) for r in self.rules) if v.kind != "allow"), ALLOW)
            result = self.channels[d.channel].send(user.addresses[d.channel], d.text) if verdict is ALLOW else None  # slow: no lock held
            with self._lock:
                self._settle(d, verdict, result, now)
            self._emit(d)

    def _settle(self, d: Delivery, verdict: Decision, result: SendResult | None, now: int) -> None:
        if verdict.kind == "drop":
            d.status, d.detail = Status.DROPPED, verdict.reason
        elif verdict.kind == "defer":
            d.status, d.detail = Status.DEFERRED, f"{verdict.reason}, until {clock(verdict.until)}"
            heapq.heappush(self.queue, (verdict.until, d.id, d))
        else:
            d.attempts += 1
            if result is SendResult.OK:
                d.status, d.detail = Status.SENT, f"attempt {d.attempts}"
            elif result is SendResult.PERMANENT:
                d.status, d.detail = Status.FAILED, "rejected by the provider"
            else:
                wait = self.backoff.delay(d.attempts)
                if wait is None:
                    d.status, d.detail = Status.FAILED, f"gave up after {d.attempts} attempts"
                else:
                    d.status, d.detail = Status.RETRYING, f"retry in {wait}s"
                    heapq.heappush(self.queue, (now + wait, d.id, d))

    def _emit(self, d: Delivery) -> None:  # listeners run outside the lock
        for fn in self.listeners:
            fn(d)


def clock(t: int) -> str:  # seconds since the start of day 0, as "d0 07:00:00"
    return f"d{t // 86400} {t % 86400 // 3600:02d}:{t % 3600 // 60:02d}:{t % 60:02d}"

/** The service: fan out to channels, run the rules, send outside the lock, retry or give up. */
final class NotificationService {
    final Map<String, User> users;
    final Map<String, Template> templates;
    final Map<String, Channel> channels;
    final List<Rule> rules;
    final BackoffPolicy backoff;
    final DedupeWindow dedupe;
    final PriorityQueue<int[]> queue = new PriorityQueue<>((a, b) -> a[0] != b[0] ? Integer.compare(a[0], b[0]) : Integer.compare(a[1], b[1])); // {due, delivery id}
    final Map<Integer, Delivery> byId = new HashMap<>();
    final List<Consumer<Delivery>> listeners = new ArrayList<>();
    private int nextId = 1;
    private final ReentrantLock lock = new ReentrantLock(); // guards the queue and the dedupe window, never a provider call

    NotificationService(Map<String, User> users, Map<String, Template> templates, Map<String, Channel> channels,
                        List<Rule> rules, BackoffPolicy backoff, int dedupeWindow) {
        this.users = users;
        this.templates = templates;
        this.channels = channels;
        this.rules = rules;
        this.backoff = backoff;
        this.dedupe = new DedupeWindow(dedupeWindow);
    }

    private void push(int due, Delivery d) {
        queue.add(new int[] {due, d.id});
        byId.put(d.id, d);
    }

    List<Delivery> notify(Notification n, int now) {
        List<Delivery> made = new ArrayList<>();
        lock.lock();
        try {
            User user = users.get(n.userId());
            Template tpl = templates.get(n.template());
            Map<String, String> texts = new LinkedHashMap<>(); // a missing value raises here, before anything is queued
            for (String ch : tpl.bodies().keySet()) if (user.addresses().containsKey(ch)) texts.put(ch, tpl.render(ch, n.params()));
            for (var e : texts.entrySet()) {
                Delivery d = new Delivery(nextId++, e.getKey(), n, e.getValue());
                made.add(d);
                if (n.dedupeKey() != null && !dedupe.first(n.userId() + "/" + e.getKey() + "/" + n.dedupeKey(), now)) {
                    d.status = Status.DROPPED;
                    d.detail = "duplicate";
                } else {
                    push(now, d);
                }
            }
        } finally {
            lock.unlock();
        }
        for (Delivery d : made) if (d.status == Status.DROPPED) emit(d);
        tick(now);
        return made;
    }

    void tick(int now) { // run everything due; the caller's clock drives retries and deferrals
        List<Delivery> due = new ArrayList<>();
        lock.lock();
        try {
            while (!queue.isEmpty() && queue.peek()[0] <= now) due.add(byId.remove(queue.poll()[1]));
        } finally {
            lock.unlock();
        }
        for (Delivery d : due) {
            User user = users.get(d.notif.userId());
            Decision verdict = Decision.ALLOW;
            for (Rule r : rules) {
                Decision v = r.check(d, user, now);
                if (!v.kind().equals("allow")) {
                    verdict = v;
                    break;
                }
            }
            SendResult result = verdict == Decision.ALLOW ? channels.get(d.channel).send(user.addresses().get(d.channel), d.text) : null; // slow: no lock held
            lock.lock();
            try {
                settle(d, verdict, result, now);
            } finally {
                lock.unlock();
            }
            emit(d);
        }
    }

    private void settle(Delivery d, Decision verdict, SendResult result, int now) {
        if (verdict.kind().equals("drop")) {
            d.status = Status.DROPPED;
            d.detail = verdict.reason();
        } else if (verdict.kind().equals("defer")) {
            d.status = Status.DEFERRED;
            d.detail = verdict.reason() + ", until " + clock(verdict.until());
            push(verdict.until(), d);
        } else {
            d.attempts++;
            if (result == SendResult.OK) {
                d.status = Status.SENT;
                d.detail = "attempt " + d.attempts;
            } else if (result == SendResult.PERMANENT) {
                d.status = Status.FAILED;
                d.detail = "rejected by the provider";
            } else {
                Integer wait = backoff.delay(d.attempts);
                if (wait == null) {
                    d.status = Status.FAILED;
                    d.detail = "gave up after " + d.attempts + " attempts";
                } else {
                    d.status = Status.RETRYING;
                    d.detail = "retry in " + wait + "s";
                    push(now + wait, d);
                }
            }
        }
    }

    private void emit(Delivery d) { // listeners run outside the lock
        for (Consumer<Delivery> fn : listeners) fn.accept(d);
    }

    static String clock(int t) { // seconds since the start of day 0, as "d0 07:00:00"
        return String.format("d%d %02d:%02d:%02d", t / 86400, t % 86400 / 3600, t % 3600 / 60, t % 60);
    }
}

std::string clock(int t) {  // seconds since the start of day 0, as "d0 07:00:00"
    char buf[32];
    std::snprintf(buf, sizeof buf, "d%d %02d:%02d:%02d", t / 86400, t % 86400 / 3600, t % 3600 / 60, t % 60);
    return buf;
}

// The service: fan out to channels, run the rules, send outside the lock, retry or give up.
class NotificationService {
public:
    NotificationService(std::map<std::string, User> users, std::map<std::string, Template> templates,
                        std::map<std::string, std::shared_ptr<Channel>> channels, std::vector<std::shared_ptr<Rule>> rules,
                        std::unique_ptr<BackoffPolicy> backoff, int dedupeWindow)
        : users_(std::move(users)), templates_(std::move(templates)), channels_(std::move(channels)),
          rules_(std::move(rules)), backoff_(std::move(backoff)), dedupe_(dedupeWindow) {}

    std::vector<std::shared_ptr<Delivery>> notify(const Notification& n, int now) {
        std::vector<std::shared_ptr<Delivery>> made;
        {
            std::lock_guard<std::recursive_mutex> g(lock_);
            const User& user = users_.at(n.userId);
            const Template& tpl = templates_.at(n.templateId);
            std::vector<std::pair<std::string, std::string>> texts;  // a missing value throws here, before anything is queued
            for (const auto& body : tpl.bodies) if (find(user.addresses, body.first)) texts.push_back({body.first, tpl.render(body.first, n.params)});
            for (const auto& t : texts) {
                auto d = std::make_shared<Delivery>(Delivery{nextId_++, t.first, n, t.second});
                made.push_back(d);
                if (!n.dedupeKey.empty() && !dedupe_.first(n.userId + "/" + t.first + "/" + n.dedupeKey, now)) {
                    d->status = Status::Dropped;
                    d->detail = "duplicate";
                } else {
                    queue_.push({now, d->id, d});
                }
            }
        }
        for (const auto& d : made) if (d->status == Status::Dropped) emit(*d);
        tick(now);
        return made;
    }

    void tick(int now) {  // run everything due; the caller's clock drives retries and deferrals
        std::vector<std::shared_ptr<Delivery>> due;
        {
            std::lock_guard<std::recursive_mutex> g(lock_);
            while (!queue_.empty() && std::get<0>(queue_.top()) <= now) {
                due.push_back(std::get<2>(queue_.top()));
                queue_.pop();
            }
        }
        for (const auto& d : due) {
            const User& user = users_.at(d->notif.userId);
            Decision verdict = ALLOW;
            for (const auto& r : rules_) {
                Decision v = r->check(*d, user, now);
                if (v.kind != "allow") {
                    verdict = v;
                    break;
                }
            }
            std::optional<SendResult> result;
            if (verdict.kind == "allow") result = channels_.at(d->channel)->send(*find(user.addresses, d->channel), d->text);  // slow: no lock held
            {
                std::lock_guard<std::recursive_mutex> g(lock_);
                settle(d, verdict, result, now);
            }
            emit(*d);
        }
    }

    std::vector<std::function<void(const Delivery&)>> listeners;

    std::size_t queued() const { return queue_.size(); }

private:
    using Item = std::tuple<int, int, std::shared_ptr<Delivery>>;  // (due time, delivery id, delivery)
    struct Later {
        bool operator()(const Item& a, const Item& b) const {
            return std::make_pair(std::get<0>(a), std::get<1>(a)) > std::make_pair(std::get<0>(b), std::get<1>(b));
        }
    };

    void settle(const std::shared_ptr<Delivery>& ptr, const Decision& verdict, std::optional<SendResult> result, int now) {
        Delivery& d = *ptr;
        if (verdict.kind == "drop") {
            d.status = Status::Dropped;
            d.detail = verdict.reason;
        } else if (verdict.kind == "defer") {
            d.status = Status::Deferred;
            d.detail = verdict.reason + ", until " + clock(verdict.until);
            queue_.push({verdict.until, d.id, ptr});
        } else {
            d.attempts++;
            if (*result == SendResult::Ok) {
                d.status = Status::Sent;
                d.detail = "attempt " + std::to_string(d.attempts);
            } else if (*result == SendResult::Permanent) {
                d.status = Status::Failed;
                d.detail = "rejected by the provider";
            } else if (auto wait = backoff_->delay(d.attempts)) {
                d.status = Status::Retrying;
                d.detail = "retry in " + std::to_string(*wait) + "s";
                queue_.push({now + *wait, d.id, ptr});
            } else {
                d.status = Status::Failed;
                d.detail = "gave up after " + std::to_string(d.attempts) + " attempts";
            }
        }
    }

    void emit(const Delivery& d) {  // listeners run outside the lock
        for (const auto& fn : listeners) fn(d);
    }

    std::map<std::string, User> users_;
    std::map<std::string, Template> templates_;
    std::map<std::string, std::shared_ptr<Channel>> channels_;
    std::vector<std::shared_ptr<Rule>> rules_;
    std::unique_ptr<BackoffPolicy> backoff_;
    DedupeWindow dedupe_;
    std::priority_queue<Item, std::vector<Item>, Later> queue_;
    int nextId_ = 1;
    std::recursive_mutex lock_;  // guards the queue and the dedupe window, never a provider call
};

const clock = (t) => { // seconds since the start of day 0, as "d0 07:00:00"
  const p = (n) => String(n).padStart(2, "0");
  return `d${Math.floor(t / 86400)} ${p(Math.floor((t % 86400) / 3600))}:${p(Math.floor((t % 3600) / 60))}:${p(t % 60)}`;
};

/** The service: fan out to channels, run the rules, send, retry or give up. (No lock: JS runs one call at a time,
 *  and a real provider call would be awaited, not blocking the queue.) */
class NotificationService {
  constructor(users, templates, channels, rules, backoff, dedupeWindow = 3600) {
    Object.assign(this, { users, templates, channels, rules, backoff });
    this.dedupe = new DedupeWindow(dedupeWindow);
    this.queue = []; // deliveries ordered by [due time, id]
    this.listeners = []; // each: (delivery) => void
    this.nextId = 1;
  }

  #push(due, d) {
    this.queue.push([due, d.id, d]);
    this.queue.sort((a, b) => a[0] - b[0] || a[1] - b[1]);
  }

  notify(n, now) {
    const user = this.users[n.userId];
    const tpl = this.templates[n.template];
    const texts = Object.keys(tpl.bodies).filter((ch) => ch in user.addresses).map((ch) => [ch, tpl.render(ch, n.params)]); // a missing value throws here, before anything is queued
    const made = [];
    for (const [ch, text] of texts) {
      const d = new Delivery(this.nextId++, ch, n, text);
      made.push(d);
      if (n.dedupeKey && !this.dedupe.first(`${n.userId}/${ch}/${n.dedupeKey}`, now)) {
        d.status = Status.DROPPED;
        d.detail = "duplicate";
        this.#emit(d);
      } else {
        this.#push(now, d);
      }
    }
    this.tick(now);
    return made;
  }

  tick(now) { // run everything due; the caller's clock drives retries and deferrals
    const due = [];
    while (this.queue.length && this.queue[0][0] <= now) due.push(this.queue.shift()[2]);
    for (const d of due) {
      const user = this.users[d.notif.userId];
      let verdict = ALLOW;
      for (const r of this.rules) { // the first link that does not allow ends the walk
        const v = r.check(d, user, now);
        if (v.kind !== "allow") {
          verdict = v;
          break;
        }
      }
      const result = verdict === ALLOW ? this.channels[d.channel].send(user.addresses[d.channel], d.text) : null;
      this.#settle(d, verdict, result, now);
      this.#emit(d);
    }
  }

  #settle(d, verdict, result, now) {
    if (verdict.kind === "drop") {
      d.status = Status.DROPPED;
      d.detail = verdict.reason;
    } else if (verdict.kind === "defer") {
      d.status = Status.DEFERRED;
      d.detail = `${verdict.reason}, until ${clock(verdict.until)}`;
      this.#push(verdict.until, d);
    } else {
      d.attempts++;
      if (result === SendResult.OK) {
        d.status = Status.SENT;
        d.detail = `attempt ${d.attempts}`;
      } else if (result === SendResult.PERMANENT) {
        d.status = Status.FAILED;
        d.detail = "rejected by the provider";
      } else {
        const wait = this.backoff.delay(d.attempts);
        if (wait === null) {
          d.status = Status.FAILED;
          d.detail = `gave up after ${d.attempts} attempts`;
        } else {
          d.status = Status.RETRYING;
          d.detail = `retry in ${wait}s`;
          this.#push(now + wait, d);
        }
      }
    }
  }

  #emit(d) {
    for (const fn of this.listeners) fn(d);
  }
}

# part 4: a scripted morning: duplicates, retries, a dead token, quiet hours
class ScriptedChannel(Channel):  # a test double: answers from a script, then OK
    def __init__(self, name: str, script: list[SendResult] | None = None) -> None:
        self.name, self.script, self.sent = name, list(script or []), []

    def send(self, address: str, text: str) -> SendResult:
        result = self.script.pop(0) if self.script else SendResult.OK
        if result is SendResult.OK:
            self.sent.append((address, text))
        return result


def at(day: int, h: int, m: int = 0, s: int = 0) -> int:
    return day * 86400 + h * 3600 + m * 60 + s


if __name__ == "__main__":
    ana = User("ana", {"email": "ana@mail.test", "sms": "+100200300", "push": "tok-ana"},
               Preferences(muted={("promo", "sms"), ("promo", "push")}, quiet=(22, 7)))
    templates = {
        "shipped": Template("shipped", {"email": "Hi {name}, order {order} has shipped.", "sms": "Order {order} shipped.", "push": "Shipped: {order}"}),
        "promo": Template("promo", {"email": "{name}, 20 percent off today.", "sms": "20 off today", "push": "20 off"}),
        "otp": Template("otp", {"sms": "Your code is {code}.", "push": "Code {code}"}),
    }
    email = ScriptedChannel("email")
    sms = ScriptedChannel("sms", [SendResult.RETRY, SendResult.RETRY])
    push = ScriptedChannel("push", [SendResult.PERMANENT])
    svc = NotificationService({"ana": ana}, templates, {"email": email, "sms": sms, "push": push},
                              [OptOutRule(), QuietHoursRule()], ExponentialBackoff(base=30, cap=600, max_attempts=3))
    svc.listeners.append(lambda d: print(f"  #{d.id} {d.channel}: {d.status.name} ({d.detail})"))

    print(clock(at(0, 10)), "shipped, first request")
    svc.notify(Notification("ana", "order", "shipped", {"name": "Ana", "order": "1042"}, "ship-1042"), at(0, 10))
    print(clock(at(0, 10, 0, 5)), "shipped, the same request again")
    svc.notify(Notification("ana", "order", "shipped", {"name": "Ana", "order": "1042"}, "ship-1042"), at(0, 10, 0, 5))
    for t in (at(0, 10, 0, 30), at(0, 10, 1, 30)):
        print(clock(t), "tick")
        svc.tick(t)
    print(clock(at(0, 23, 30)), "a promo, then a security code")
    svc.notify(Notification("ana", "promo", "promo", {"name": "Ana"}), at(0, 23, 30))
    svc.notify(Notification("ana", "security", "otp", {"code": "493021"}, priority=Priority.CRITICAL), at(0, 23, 30))
    print(clock(at(1, 7)), "tick")
    svc.tick(at(1, 7))
    try:
        svc.notify(Notification("ana", "order", "shipped", {"name": "Ana"}), at(1, 8))
    except MissingParam as e:
        print("rejected:", e)
    print("queue left:", len(svc.queue), "| email sent:", len(email.sent), "| sms sent:", len(sms.sent), "| push sent:", len(push.sent))

/** A test double: answers from a script, then OK. */
final class ScriptedChannel implements Channel {
    final String name;
    final List<SendResult> script;
    final List<String[]> sent = new ArrayList<>();

    ScriptedChannel(String name, List<SendResult> script) {
        this.name = name;
        this.script = new ArrayList<>(script);
    }

    public String name() { return name; }

    public SendResult send(String address, String text) {
        SendResult result = script.isEmpty() ? SendResult.OK : script.remove(0);
        if (result == SendResult.OK) sent.add(new String[] {address, text});
        return result;
    }
}

public class NotifyDemo {
    static int at(int day, int h, int m, int s) { return day * 86400 + h * 3600 + m * 60 + s; }

    static int at(int day, int h, int m) { return at(day, h, m, 0); }

    static int at(int day, int h) { return at(day, h, 0, 0); }

    static Map<String, String> map(String... kv) { // a small ordered map
        Map<String, String> out = new LinkedHashMap<>();
        for (int i = 0; i < kv.length; i += 2) out.put(kv[i], kv[i + 1]);
        return out;
    }

    // a scripted morning: duplicates, retries, a dead token, quiet hours
    public static void main(String[] args) {
        Preferences prefs = new Preferences();
        prefs.muted.add("promo/sms");
        prefs.muted.add("promo/push");
        prefs.quiet = new int[] {22, 7};
        User ana = new User("ana", map("email", "ana@mail.test", "sms", "+100200300", "push", "tok-ana"), prefs);
        Map<String, Template> templates = Map.of(
                "shipped", new Template("shipped", map("email", "Hi {name}, order {order} has shipped.", "sms", "Order {order} shipped.", "push", "Shipped: {order}")),
                "promo", new Template("promo", map("email", "{name}, 20 percent off today.", "sms", "20 off today", "push", "20 off")),
                "otp", new Template("otp", map("sms", "Your code is {code}.", "push", "Code {code}")));
        ScriptedChannel email = new ScriptedChannel("email", List.of());
        ScriptedChannel sms = new ScriptedChannel("sms", List.of(SendResult.RETRY, SendResult.RETRY));
        ScriptedChannel push = new ScriptedChannel("push", List.of(SendResult.PERMANENT));
        NotificationService svc = new NotificationService(Map.of("ana", ana), templates,
                Map.of("email", email, "sms", sms, "push", push), List.of(new OptOutRule(), new QuietHoursRule()),
                new ExponentialBackoff(30, 600, 3, failures -> 0), 3600);
        svc.listeners.add(d -> System.out.println("  #" + d.id + " " + d.channel + ": " + d.status + " (" + d.detail + ")"));

        System.out.println(NotificationService.clock(at(0, 10)) + " shipped, first request");
        svc.notify(new Notification("ana", "order", "shipped", map("name", "Ana", "order", "1042"), "ship-1042", Priority.NORMAL), at(0, 10));
        System.out.println(NotificationService.clock(at(0, 10, 0, 5)) + " shipped, the same request again");
        svc.notify(new Notification("ana", "order", "shipped", map("name", "Ana", "order", "1042"), "ship-1042", Priority.NORMAL), at(0, 10, 0, 5));
        for (int t : new int[] {at(0, 10, 0, 30), at(0, 10, 1, 30)}) {
            System.out.println(NotificationService.clock(t) + " tick");
            svc.tick(t);
        }
        System.out.println(NotificationService.clock(at(0, 23, 30)) + " a promo, then a security code");
        svc.notify(new Notification("ana", "promo", "promo", map("name", "Ana"), null, Priority.NORMAL), at(0, 23, 30));
        svc.notify(new Notification("ana", "security", "otp", map("code", "493021"), null, Priority.CRITICAL), at(0, 23, 30));
        System.out.println(NotificationService.clock(at(1, 7)) + " tick");
        svc.tick(at(1, 7));
        try {
            svc.notify(new Notification("ana", "order", "shipped", map("name", "Ana"), null, Priority.NORMAL), at(1, 8));
        } catch (MissingParam e) {
            System.out.println("rejected: " + e.getMessage());
        }
        System.out.println("queue left: " + svc.queue.size() + " | email sent: " + email.sent.size() + " | sms sent: " + sms.sent.size() + " | push sent: " + push.sent.size());
    }
}

// a test double: answers from a script, then OK
class ScriptedChannel : public Channel {
public:
    explicit ScriptedChannel(std::vector<SendResult> script = {}) : script_(std::move(script)) {}

    SendResult send(const std::string& address, const std::string& text) override {
        SendResult result = SendResult::Ok;
        if (next_ < script_.size()) result = script_[next_++];
        if (result == SendResult::Ok) sent++;
        return result;
    }

    int sent = 0;

private:
    std::vector<SendResult> script_;
    std::size_t next_ = 0;
};

int at(int day, int h, int m = 0, int s = 0) { return day * 86400 + h * 3600 + m * 60 + s; }

// a scripted morning: duplicates, retries, a dead token, quiet hours
int main() {
    User ana{"ana", {{"email", "ana@mail.test"}, {"sms", "+100200300"}, {"push", "tok-ana"}}, {}};
    ana.prefs.muted = {"promo/sms", "promo/push"};
    ana.prefs.hasQuiet = true;
    ana.prefs.quietStart = 22;
    ana.prefs.quietEnd = 7;
    std::map<std::string, Template> templates = {
        {"shipped", {"shipped", {{"email", "Hi {name}, order {order} has shipped."}, {"sms", "Order {order} shipped."}, {"push", "Shipped: {order}"}}}},
        {"promo", {"promo", {{"email", "{name}, 20 percent off today."}, {"sms", "20 off today"}, {"push", "20 off"}}}},
        {"otp", {"otp", {{"sms", "Your code is {code}."}, {"push", "Code {code}"}}}},
    };
    auto email = std::make_shared<ScriptedChannel>();
    auto sms = std::make_shared<ScriptedChannel>(std::vector<SendResult>{SendResult::Retry, SendResult::Retry});
    auto push = std::make_shared<ScriptedChannel>(std::vector<SendResult>{SendResult::Permanent});
    NotificationService svc({{"ana", ana}}, templates, {{"email", email}, {"sms", sms}, {"push", push}},
                            {std::make_shared<OptOutRule>(), std::make_shared<QuietHoursRule>()},
                            std::make_unique<ExponentialBackoff>(30, 600, 3, [](int) { return 0; }), 3600);
    svc.listeners.push_back([](const Delivery& d) {
        std::cout << "  #" << d.id << " " << d.channel << ": " << name(d.status) << " (" << d.detail << ")\n";
    });

    std::cout << clock(at(0, 10)) << " shipped, first request\n";
    svc.notify({"ana", "order", "shipped", {{"name", "Ana"}, {"order", "1042"}}, "ship-1042"}, at(0, 10));
    std::cout << clock(at(0, 10, 0, 5)) << " shipped, the same request again\n";
    svc.notify({"ana", "order", "shipped", {{"name", "Ana"}, {"order", "1042"}}, "ship-1042"}, at(0, 10, 0, 5));
    for (int t : {at(0, 10, 0, 30), at(0, 10, 1, 30)}) {
        std::cout << clock(t) << " tick\n";
        svc.tick(t);
    }
    std::cout << clock(at(0, 23, 30)) << " a promo, then a security code\n";
    svc.notify({"ana", "promo", "promo", {{"name", "Ana"}}}, at(0, 23, 30));
    svc.notify({"ana", "security", "otp", {{"code", "493021"}}, "", Priority::Critical}, at(0, 23, 30));
    std::cout << clock(at(1, 7)) << " tick\n";
    svc.tick(at(1, 7));
    try {
        svc.notify({"ana", "order", "shipped", {{"name", "Ana"}}}, at(1, 8));
    } catch (const MissingParam& e) {
        std::cout << "rejected: " << e.what() << "\n";
    }
    std::cout << "queue left: " << svc.queued() << " | email sent: " << email->sent << " | sms sent: " << sms->sent << " | push sent: " << push->sent << "\n";
}

// a test double: answers from a script, then OK
class ScriptedChannel extends Channel {
  constructor(script = []) {
    super();
    this.script = [...script];
    this.sent = [];
  }

  send(address, text) {
    const result = this.script.length ? this.script.shift() : SendResult.OK;
    if (result === SendResult.OK) this.sent.push([address, text]);
    return result;
  }
}

const at = (day, h, m = 0, s = 0) => day * 86400 + h * 3600 + m * 60 + s;

// a scripted morning: duplicates, retries, a dead token, quiet hours
const ana = new User("ana", { email: "ana@mail.test", sms: "+100200300", push: "tok-ana" },
  new Preferences({ muted: ["promo/sms", "promo/push"], quiet: [22, 7] }));
const templates = {
  shipped: new Template("shipped", { email: "Hi {name}, order {order} has shipped.", sms: "Order {order} shipped.", push: "Shipped: {order}" }),
  promo: new Template("promo", { email: "{name}, 20 percent off today.", sms: "20 off today", push: "20 off" }),
  otp: new Template("otp", { sms: "Your code is {code}.", push: "Code {code}" }),
};
const email = new ScriptedChannel();
const sms = new ScriptedChannel([SendResult.RETRY, SendResult.RETRY]);
const push = new ScriptedChannel([SendResult.PERMANENT]);
const svc = new NotificationService({ ana }, templates, { email, sms, push },
  [new OptOutRule(), new QuietHoursRule()], new ExponentialBackoff({ base: 30, cap: 600, maxAttempts: 3 }));
svc.listeners.push((d) => console.log(`  #${d.id} ${d.channel}: ${d.status} (${d.detail})`));

console.log(clock(at(0, 10)), "shipped, first request");
svc.notify(notification("ana", "order", "shipped", { name: "Ana", order: "1042" }, "ship-1042"), at(0, 10));
console.log(clock(at(0, 10, 0, 5)), "shipped, the same request again");
svc.notify(notification("ana", "order", "shipped", { name: "Ana", order: "1042" }, "ship-1042"), at(0, 10, 0, 5));
for (const t of [at(0, 10, 0, 30), at(0, 10, 1, 30)]) {
  console.log(clock(t), "tick");
  svc.tick(t);
}
console.log(clock(at(0, 23, 30)), "a promo, then a security code");
svc.notify(notification("ana", "promo", "promo", { name: "Ana" }), at(0, 23, 30));
svc.notify(notification("ana", "security", "otp", { code: "493021" }, null, Priority.CRITICAL), at(0, 23, 30));
console.log(clock(at(1, 7)), "tick");
svc.tick(at(1, 7));
try {
  svc.notify(notification("ana", "order", "shipped", { name: "Ana" }), at(1, 8));
} catch (e) {
  if (!(e instanceof MissingParam)) throw e;
  console.log("rejected:", e.message);
}
console.log("queue left:", svc.queue.length, "| email sent:", email.sent.length, "| sms sent:", sms.sent.length, "| push sent:", push.sent.length);

d0 10:00:00 shipped, first request
  #1 email: SENT (attempt 1)
  #2 sms: RETRYING (retry in 30s)
  #3 push: FAILED (rejected by the provider)
d0 10:00:05 shipped, the same request again
  #4 email: DROPPED (duplicate)
  #5 sms: DROPPED (duplicate)
  #6 push: DROPPED (duplicate)
d0 10:00:30 tick
  #2 sms: RETRYING (retry in 60s)
d0 10:01:30 tick
  #2 sms: SENT (attempt 3)
d0 23:30:00 a promo, then a security code
  #7 email: DEFERRED (quiet hours, until d1 07:00:00)
  #8 sms: DROPPED (opted out)
  #9 push: DROPPED (opted out)
  #10 sms: SENT (attempt 1)
  #11 push: SENT (attempt 1)
d1 07:00:00 tick
  #7 email: SENT (attempt 1)
rejected: shipped: no value for order
queue left: 0 | email sent: 2 | sms sent: 2 | push sent: 1