Design a Notification System

Problem এর পরিচয়

যেকোনো product শেষমেশ notification পাঠায়, আর প্রায় সব team প্রথমে সহজ পথে করে: order service সরাসরি SMS gateway কে call করে। এটা চলে যতক্ষণ না gateway ধীর হয়, একটা campaign এসে ওটাকে ডুবিয়ে দেয়, একটা retry একই OTP তিনবার পাঠায়, বা কেউ রাত 3 টায় sale এর SMS পায়। এই problem হলো সেই shared service নিয়ে যেটা এসব ঠিক করে, আর এর মূল শিক্ষা যেকোনো background কাজে খাটে: request তাড়াতাড়ি নিয়ে নাও, ধীর আর অনির্ভরযোগ্য অংশটা পরে করো, আর প্রতিটা ধাপ এমন বানাও যাতে বারবার করলেও ক্ষতি না হয়।

Problem statement

নুসরাত ঢাকার একটা দোকান থেকে একটা phone order করেছে। পরের দুই দিনে কয়েকটা system তাকে কিছু জানাবে: order service ("আপনার order পাঠানো হয়েছে"), courier ("আজকে পৌঁছাবে"), payment service ("refund হয়ে গেছে"), আর marketing team ("Eid sale, 30% ছাড়")। এর কিছু তাকে পেতেই হবে (login এর OTP), কিছু সে বন্ধ করে রেখেছে (marketing email), আর জরুরি না হলে কোনোটাই যেন রাতে তার ঘুম না ভাঙায়।

কী বানাচ্ছি: একটা notification service, যেটাকে ভেতরের যেকোনো service বলতে পারে "user Y কে template X পাঠাও", আর সেটা user এর পছন্দমতো push, SMS বা email এ পৌঁছে দেয়, নির্ভরযোগ্যভাবে, duplicate ছাড়া, আর যে call করেছে তাকে আটকে না রেখে।

কে ব্যবহার করে: এক দিকে ভেতরের service গুলো (order, payment, auth, chat, courier) আর marketing team এর campaign tool; অন্য দিকে কয়েক কোটি user; আর support team, যাদের জানতে হয় "customer কি SMS টা পেয়েছিল?"।

কী সমস্যা মেটায়: প্রতিটা team কে আর নিজের retry, template, opt-out এর নিয়ম আর provider integration লিখতে হয় না; user রা একই রকম অভিজ্ঞতা পায়; আর কী পাঠানো হলো সেটা দেখার একটাই জায়গা থাকে।

Technically কোথায় মজা:

• অনির্ভরযোগ্য, ধীর provider: SMS gateway timeout দেয়, push service throttle করে, email provider address ফিরিয়ে দেয়। Call এ কয়েকশো millisecond লাগে, আর আলাদা আলাদা ভাবে fail করে।
• Duplicate: caller retry করে, queue আবার দেয়, worker মাঝপথে পড়ে যায়। দুইটা OTP এলে মানুষ বিভ্রান্ত হয়; দুইটা "টাকা পেয়েছি" এলে ভয় পায়।
• হঠাৎ চাপ: আড়াই কোটি user এর campaign যেন login এর OTP দেরি না করায়।
• প্রতি user এর নিয়ম: opt-out, quiet hours, দিনের সীমা, ভাষা, কার কোন channel আছেই।
• খরচ: প্রতিটা SMS এ টাকা লাগে; push আর email প্রায় ফ্রি। কোন কাজে কোন channel, সেটা design ই ঠিক করে।

Requirement

Functional requirement

1. Send: ভেতরের একটা service পাঠায় {user, template, params, idempotency key}; service টা template এর channel গুলোতে (push, SMS, email) পৌঁছে দেয়।
2. পছন্দ: user প্রতিটা channel এ marketing বন্ধ করতে পারে আর quiet hours ঠিক করতে পারে; transactional message (OTP, order আর payment update) সবসময় যাবে।
3. Template: message আসে version করা template থেকে, parameter আর অনুবাদ সহ (বাংলা আর English)।
4. নির্ভরযোগ্যতা: সাময়িক failure হলে retry; যেগুলো পৌঁছানো যায় না সেগুলো dead-letter queue তে যায়, দেখা আর আবার পাঠানোর জন্য।
5. Retry থেকে duplicate না: একই idempotency key কখনো দ্বিতীয় notification বানাবে না।
6. Campaign: marketing লাখো মানুষের একটা segment এ পাঠাতে পারবে, provider যতটা নিতে পারে সেই গতিতে।
7. Tracking: প্রতিটা message এর অবস্থা (queued, sent, delivered, failed) দেখা যাবে।
8. Scope এর বাইরে: app এর ভেতরের inbox UI, message এর লেখা নিয়ে A/B test, আর নিজের SMS বা push infrastructure বানানো (আমরা provider ব্যবহার করি)।

Non-functional requirement

Accept latency, durability, duplicate না হওয়া আর আলাদা রাখা, এগুলোই design ঠিক করে। Delivery latency ঢিলা, শুধু OTP বাদে, আর সেজন্যই OTP এর নিজের আলাদা পথ।

Capacity আন্দাজ

ধরে নেওয়া (মুখে বলা, গোল সংখ্যা):

• 5 কোটি user; marketing এর জন্য পৌঁছানো যায় মোটামুটি আড়াই কোটি (opt in করেছে, channel আছে)।
• দিনে: 1 কোটি order আর delivery update, 50 লাখ OTP, 3.5 কোটি app push (chat, ride এর অবস্থা), আর আড়াই কোটি করে দুইটা marketing campaign: মোট প্রায় দিনে 10 কোটি message।
• Channel এর ভাগ: 6.5 কোটি push, 2.5 কোটি email, 1 কোটি SMS (SMS এ টাকা লাগে, তাই শুধু OTP আর জরুরি update এ)।
• একটা campaign মোটামুটি এক ঘণ্টায় যায়; স্বাভাবিক traffic এর peak গড়ের 3 গুণ।

Traffic

Storage

Worker (Little's law): একসাথে চলতে থাকা call = rate × প্রতি call এর সময়। সেকেন্ডে 10,000 পাঠানো আর প্রতি provider call এ ~200 ms হলে যেকোনো মুহূর্তে মোটামুটি 2,000 টা call চলছে। Asynchronous I/O দিয়ে 20 টা worker instance, প্রতিটা একসাথে 100 টা call, peak সামলে নেয়। আসল সীমা সাধারণত আমাদের CPU না, provider এর rate limit: একটা SMS account হয়তো সেকেন্ডে কয়েকশোর বেশি নেয় না, তাই worker কে সেই গতিতে throttle করতে হয়।

সংখ্যাগুলো কী বলছে: একটা queue এর জন্য পরিমাণটা মাঝারি, কিন্তু API call এর ভেতরে করার জন্য অনেক ধীর আর অনির্ভরযোগ্য। Storage এর বড় অংশ status log। কঠিন অংশ size না, আচরণ: duplicate, হঠাৎ চাপ, ধীর provider আর প্রতি user এর নিয়ম।

API design

• 200 OK না, 202 Accepted: API কথা দেয় যে notification টা durable ভাবে queue তে আছে, পৌঁছে গেছে সেটা না। এজন্যই সে millisecond এ উত্তর দিতে পারে।
• Idempotency key: caller বেছে নেয়, আর এটা business event এর সাথে বাঁধা (order-BD-7790-shipped), প্রতি চেষ্টায় random না। একই key দিয়ে retry হলে প্রথম ফলটাই ফেরত যায়, নতুন কিছু যায় না।
• Authentication: service থেকে service (mTLS বা signed token); প্রতিটা caller এর quota আছে, আর শুধু marketing tool campaign বানাতে বা marketing template পাঠাতে পারে।
• Caller template এর নাম পাঠায়, লেখা না: কথা, অনুবাদ আর আইনি footer এক জায়গায় থাকে, আর একটা service hack হলেও সে সব user কে যা ইচ্ছা লিখে পাঠাতে পারে না।
• Webhook provider sign করে পাঠায় আর আমরা যাচাই করি; এগুলো status কে "sent" থেকে "delivered" বা "failed" করে।

Database design

Access pattern: প্রতিটা request এ পছন্দ আর contact পড়া হয় (বেশি পড়া, ছোট, user ধরে)। Template সারাক্ষণ পড়া হয়, কদাচিৎ বদলায়। প্রতিটা message এ status record লেখা হয় আর কয়েকবার update হয় (বেশি লেখা, notification বা user ধরে খোঁজা)। Dedup key প্রতি request এ একবার দেখা হয় আর একদিন পর মুছে যায়।

-- Preferences and contacts (PostgreSQL, cached in memory with a short TTL).
CREATE TABLE user_prefs (
  user_id       BIGINT PRIMARY KEY,
  locale        TEXT NOT NULL DEFAULT 'bn',
  tz            TEXT NOT NULL DEFAULT 'Asia/Dhaka',
  quiet_from    TIME,                       -- marketing only
  quiet_to      TIME,
  optout_push   BOOLEAN NOT NULL DEFAULT false,   -- marketing opt-outs per channel
  optout_sms    BOOLEAN NOT NULL DEFAULT false,
  optout_email  BOOLEAN NOT NULL DEFAULT false
);
CREATE TABLE contacts (
  user_id   BIGINT NOT NULL,
  channel   TEXT   NOT NULL,                -- push, sms, email
  address   TEXT   NOT NULL,                -- device token, phone, email
  verified  BOOLEAN NOT NULL,
  active    BOOLEAN NOT NULL DEFAULT true,  -- false after the provider says "invalid"
  PRIMARY KEY (user_id, channel, address)
);

-- Templates: versioned, one row per language and channel.
CREATE TABLE templates (
  name     TEXT, version INT, locale TEXT, channel TEXT,
  kind     TEXT NOT NULL,                   -- transactional | marketing
  body     TEXT NOT NULL,
  PRIMARY KEY (name, version, locale, channel)
);

-- Status of every message (Cassandra / DynamoDB style: partition by user, newest first).
--   notifications_by_user: PRIMARY KEY ((user_id), created_at DESC, message_key)
--     columns: channel, template, status, attempts, provider, provider_msg_id, updated_at
--   by_provider_msg: PRIMARY KEY (provider, provider_msg_id) -> user_id, message_key  (for webhooks)

-- Dedup: Redis  SET dedup:<idempotency key> <result> NX EX 86400

• পছন্দের জন্য PostgreSQL কেন: ছোট (10 GB), relational, user রা settings page থেকে বদলায়, আর constraint দরকার। Read আসে cache থেকে, তাই database এর উপর চাপ কম।
• Status এর জন্য wide-column store কেন: দিনে প্রায় 10 কোটি insert আর তার update, সহজ key lookup, পুরনো record এর জন্য স্বাভাবিক TTL, আর সহজে horizontal scaling। ছোট scale এ partition করা PostgreSQL table পুরোপুরি ঠিক আছে; লেখার পরিমাণ বা কতদিন রাখতে হবে সেটা যখন যন্ত্রণা দেয় তখন বদলাও।
• Dedup এর জন্য Redis কেন: millisecond এর কমে atomic একটা "না থাকলে set করো, expiry সহ", ঠিক এই কাজটাই দরকার, আর key গুলোর একদিন বাঁচলেই চলে। Redis স্থায়ী কোনো কিছুর source of truth না, তাই এটা হারালে অসুবিধা, data হারানো না (Failure দেখো)।
• Partitioning: status কে user_id ধরে (তাতে "এই user এর notification" একটাই partition); প্রতিটা topic এর ভেতরে queue কে user ID ধরে ভাগ, যাতে একজন user এর message গুলো ক্রমে থাকে।

Architecture এর ছবি

যেগুলো ইচ্ছা করে রাখিনি: API পথে প্রতি message এ database write (durability র জন্য শুধু queue তে লেখাই দরকার), প্রথম দিন থেকে প্রতিটা channel এর আলাদা microservice (এক codebase, আলাদা queue আর worker pool), বা নিজের push বা SMS infrastructure।

একটা request এর জীবন

একটা request নেওয়া

1. Order service → LB → Notify API: Idempotency-Key: evt-2 সহ POST /v1/notifications।
2. API caller এর quota, template আর parameter দেখে (parameter না থাকলে 400, কখনো ফাঁকা জায়গা সহ message না)।
3. Redis এ SET dedup:evt-2 NX EX 86400। Key আগে থেকে থাকলে রাখা ফলটা 200 সহ ফেরত দাও: এটা retry।
4. পছন্দ আর contact পড়ো (cache থেকে)। Channel ঠিক করো: template বলছে push + SMS; নুসরাতের device token নেই, তাই push বাদ।
5. শুধু marketing: opt-out, quiet hours (window শেষ হওয়া পর্যন্ত পিছিয়ে দাও) আর দিনের সীমা দেখো।
6. User এর ভাষায় template render করো আর প্রতি channel এ একটা message রাখো (evt-2:sms), user ধরে key করা, যাতে একজনের message গুলো ক্রমে থাকে।
7. কী queue তে গেল আর কী বাদ গেল সেটা সহ 202 দাও। মোট: কয়েক millisecond, কোনো provider call নেই।

একটা message পাঠানো (worker এর দিক)

1. একটা SMS worker queue থেকে evt-2:sms নেয় (lease এ নেয়, worker acknowledge না করা পর্যন্ত মোছা হয় না)।
2. Status store দেখে: এই message key আগেই sent হলে acknowledge করে থেমে যায়। Crash এর পর আবার আসা message এভাবে ধরা পড়ে।
3. Message টাকে sending দাগায় আর timeout (যেমন 5 s) সহ gateway কে call করে; provider idempotency key নিলে message key টা পাঠায়।
4. সফল: provider এর message ID সহ sent লেখো, queue এর message acknowledge করো।
5. সাময়িক failure (timeout, 5xx, throttled): base × 2^(attempt-1) আর একটু random jitter পরে retry ঠিক করো, আসলটা acknowledge করো। সর্বোচ্চ চেষ্টা শেষ হলে dead-letter queue তে সরাও।
6. স্থায়ী failure (ভুল number, অচেনা address): retry না। Dead-letter করো আর contact টা inactive দাগাও, যাতে ওই address এ আর চেষ্টা না হয়।
7. পরে provider এর webhook জানায় delivered (বা failed); by_provider_msg দিয়ে status update হয়।

Component গুলো গভীরে

Duplicate: কোথা থেকে আসে আর প্রতিটা কে কী থামায়

ফোন পর্যন্ত exactly-once delivery কোনো broker setting দিয়ে পাওয়া যায় না: শেষ ধাপটা বাইরের একটা call, যেটা সফল হতে পারে অথচ আমাদের acknowledgement fail করতে পারে। আমরা যা পারি: প্রতিটা ধাপ idempotent বানানো, আর ঝুঁকির ফাঁকটা ছোট করে আনা "provider এর উত্তর আর আমাদের status লেখার মাঝের কয়েক millisecond এ crash" পর্যন্ত। এই সৎ উত্তরটা interviewer পছন্দ করে।

প্রতি channel এ একটা queue, আর জরুরি message এর আলাদা lane

আলাদা queue মানে প্রতিটা channel এর নিজের গতি, provider এর rate limit মেপে নিজের worker সংখ্যা, আর নিজের failure এর গণ্ডি। Priority তেও একই কথা: OTP আর security alert যায় একটা high-priority topic এ, নিজের worker সহ, তাই normal queue তে বসে থাকা আড়াই কোটি message এর campaign কখনো login code দেরি করায় না। Kafka র ভাষায়: প্রতি channel (আর প্রতি priority) এ একটা topic, user ID ধরে partition।

Retry, backoff আর dead-letter queue

সাথে সাথে retry করলে চাপে থাকা provider আরো খারাপ হয়: সব worker একসাথে ওকে পেটাতে থাকে। Exponential backoff চেষ্টাগুলোর মাঝে ফাঁক বাড়ায় (1, 2, 4, 8..., আমাদের demo তে সেকেন্ড; production এ মিনিট), আর jitter (অপেক্ষার একটা random অংশ) একসাথে fail করা হাজারো message কে একসাথে retry করতে দেয় না। সর্বোচ্চ সংখ্যক চেষ্টার পর, বা স্থায়ী error হলে, message যায় dead-letter queue তে: হারায় না, DLQ বাড়লে alert বাজে, আর কারণ ঠিক হলে কেউ আবার পাঠাতে পারে। উপরে থাকে একটা circuit breaker: কিছুক্ষণ একটা provider বেশিরভাগ call এ fail করলে অল্প সময়ের জন্য ওকে call করা বন্ধ (বা দ্বিতীয় provider এ চলে যাও), retry জমাতে থাকার বদলে।

পিছিয়ে দেওয়া retry কে কোথাও অপেক্ষা করতে হয়। উপায়: প্রতিটা delay এর জন্য একটা "retry" topic (sms-retry-1m, sms-retry-10m) যেটা worker শুধু সময় হলে পড়ে, বা due time ধরে সাজানো একটা scheduled-job table/sorted set। আমাদের core প্রতিটা message এ একটা Due সময় রাখে, এটাই sorted-set এর ছোট রূপ।

পছন্দ, quiet hours আর সীমা

দুই ধরনের message, আলাদা নিয়ম:

নিয়মগুলো নেওয়ার সময় চলে (opt-out, সীমা), আর যে message গুলো অপেক্ষা করেছে সেগুলোর জন্য পাঠানোর সময় আবার দেখা হয় (campaign queue তে থাকার সময় user opt out করতে পারে)। পছন্দ অল্প সময় cache থাকে (কয়েক সেকেন্ড থেকে এক মিনিট); আবার দেখার নিয়মটা সেই পুরনো থাকাকে নিরীহ করে দেয়।

Campaign: বন্যা না বানিয়ে fan-out

আড়াই কোটি user এর campaign মানে আড়াই কোটি প্রাপক সহ একটা API call না। একটা scheduler segment টা page ধরে পড়ে, ধরো 10,000 করে batch এ request বানায়, camp-7:<user_id> এর মতো idempotency key সহ (যাতে scheduler restart হলে দুইবার না যায়), আর নিয়ন্ত্রিত গতিতে normal-priority queue তে দেয়। মোট গতি provider এর সীমার নিচে, আর এমন যাতে বাকি traffic দেরি না হয়। Campaign থামানো আর আবার চালু করা যায়, কারণ অগ্রগতি মানে শুধু "segment এর কতদূর এলাম"।

Provider failover

প্রতিটা channel এ দুইটা provider থাকতে পারে। Worker প্রতিটা provider এর error rate আর latency দেখে; primary র circuit breaker খুলে গেলে traffic secondary তে যায়। SMS এ এতে খরচ আর নির্দিষ্ট operator এর delivery সমস্যাতেও সুবিধা হয়। Message key message এর সাথেই যায়, তাই যে provider ই পাঠাক, status আর webhook মিলে যায়।

চালিয়ে দেখার মতো core

এই program টা design এর মূল অংশ, এক বসায় পড়ে ফেলার মতো ছোট: parameter সহ template, contact, opt-out আর quiet hours সহ user, API এর দিকের Send যেটা dedup আর নিয়ম একসাথে atomic ভাবে মানে, প্রতি channel এ একটা queue, আর worker যারা provider কে call করে, সাময়িক failure এ exponential backoff সহ retry করে আর বাকিগুলো dead-letter করে। একটা simulated clock সময়টা একদম ঠিক রাখে, আর নকল provider গুলো ঠিক সময়মতো fail করে।

notify/
├── go.mod
├── model.go           Channel, Kind, Template, User, Request, Message, Render, Clock
├── provider.go        Outcome, Provider, FakeProvider (scripted failures)
├── service.go         Service: Send (dedup, prefs, quiet hours, limits), Work/Step (retry, DLQ)
├── notify_test.go     duplicates under -race, backoff timing, DLQ, quiet hours, opt-outs, 4 workers
└── cmd/demo/main.go   order updates, a duplicate, a flaky gateway, a campaign at night

// model.go
package notify

import (
	"errors"
	"fmt"
	"strings"
)

// Channel is one way to reach a person. Each channel has its own queue and its own provider,
// so a slow SMS gateway never holds up push or email.
type Channel string

const (
	Push  Channel = "push"
	SMS   Channel = "sms"
	Email Channel = "email"
)

// Kind decides which rules apply. Users can't opt out of transactional messages (an order
// update, a one-time code), but marketing respects opt-outs, quiet hours and a daily limit.
type Kind int

const (
	Transactional Kind = iota
	Marketing
)

// Template is a message shape: which channels it goes to and its text with {placeholders}.
type Template struct {
	Kind     Kind
	Channels []Channel
	Text     string
}

// User is what the preference service knows about a person.
type User struct {
	Contacts  map[Channel]string // device token, phone number, email address
	OptOut    map[Channel]bool   // marketing opt-outs per channel
	QuietFrom int                // quiet hours for marketing, e.g. 22 to 8 (0, 0 = none)
	QuietTo   int
}

// Request is what another service sends us: "tell this user about this event".
// ID is the idempotency key: the same ID twice means the same notification.
type Request struct {
	ID, User, Template string
	Params             map[string]string
}

// Message is one notification on one channel, waiting in that channel's queue.
type Message struct {
	Key     string // request ID + channel, e.g. "evt-1:sms"
	Channel Channel
	To      string
	Body    string
	Kind    Kind
	Attempt int   // attempts made so far
	Due     int64 // simulated seconds; not sent before this
	seq     int   // enqueue order, breaks ties between equal Due times
}

var (
	ErrDuplicate       = errors.New("duplicate request, already accepted")
	ErrUnknownUser     = errors.New("unknown user")
	ErrUnknownTemplate = errors.New("unknown template")
	ErrRateLimited     = errors.New("marketing limit reached for today")
)

// Render fills {name} placeholders; a missing value is an error, never a blank in a message.
func Render(text string, params map[string]string) (string, error) {
	var b strings.Builder
	for {
		i := strings.IndexByte(text, '{')
		if i < 0 {
			b.WriteString(text)
			return b.String(), nil
		}
		j := strings.IndexByte(text[i:], '}')
		if j < 0 {
			return "", fmt.Errorf("unclosed placeholder in %q", text)
		}
		name := text[i+1 : i+j]
		v, ok := params[name]
		if !ok {
			return "", fmt.Errorf("missing parameter %q", name)
		}
		b.WriteString(text[:i])
		b.WriteString(v)
		text = text[i+j+1:]
	}
}

// Clock formats simulated seconds as "D1 21:59:50".
func Clock(t int64) string {
	s := t % 86400
	return fmt.Sprintf("D%d %02d:%02d:%02d", t/86400+1, s/3600, s/60%60, s%60)
}

// provider.go
package notify

import "sync"

// Outcome is what a provider (APNs/FCM, an SMS gateway, an email service) said.
type Outcome int

const (
	OK        Outcome = iota
	Transient         // timeout, 5xx, throttled: worth retrying
	Permanent         // invalid number, unknown address: retrying won't help
)

func (o Outcome) String() string {
	switch o {
	case OK:
		return "delivered"
	case Transient:
		return "failed (timeout)"
	default:
		return "rejected (bad address)"
	}
}

// Provider sends one message on one channel.
type Provider interface {
	Send(m Message) Outcome
}

// FakeProvider plays back scripted outcomes per address (attempt 1, 2, ...); after the
// script runs out, sends succeed. It also counts sends per message key.
type FakeProvider struct {
	mu     sync.Mutex
	Script map[string][]Outcome
	Sent   map[string]int
}

func NewFakeProvider(script map[string][]Outcome) *FakeProvider {
	if script == nil {
		script = map[string][]Outcome{}
	}
	return &FakeProvider{Script: script, Sent: map[string]int{}}
}

func (p *FakeProvider) Send(m Message) Outcome {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.Sent[m.Key]++
	if s := p.Script[m.To]; m.Attempt < len(s) {
		return s[m.Attempt]
	}
	return OK
}

// service.go
// Package notify is the core of a notification system: requests are deduplicated, checked
// against preferences, rendered from templates and put on one queue per channel; workers send
// them through providers, retry transient failures with exponential backoff, and move
// messages that can't be delivered to a dead-letter queue.
package notify

import (
	"strconv"
	"sync"
)

// MaxAttempts is how many times a message is tried before it goes to the dead-letter queue.
const MaxAttempts = 4

// Backoff is the wait before retry number n (1, 2, 4, ... seconds). Production adds jitter
// and uses minutes, not seconds; the demo keeps it exact so every language prints the same.
func Backoff(attempt int) int64 { return int64(1) << (attempt - 1) }

// Event is one delivery attempt, for logs and the demo.
type Event struct {
	Time    int64
	Msg     Message
	Outcome Outcome
	RetryAt int64 // set when the message was put back on the queue
	Dead    bool  // set when it went to the dead-letter queue
}

// Service accepts requests and runs the per-channel queues.
type Service struct {
	mu                 sync.Mutex
	users              map[string]User
	templates          map[string]Template
	providers          map[Channel]Provider
	seen               map[string]bool // accepted request IDs (production: a store with a TTL)
	marketing          map[string]int  // "user/day" -> marketing messages accepted
	MaxMarketingPerDay int
	queues             map[Channel][]Message
	seq                int
	Dead               []Message
	Delivered          int
}

func NewService(users map[string]User, templates map[string]Template, providers map[Channel]Provider) *Service {
	return &Service{
		users: users, templates: templates, providers: providers,
		seen: map[string]bool{}, marketing: map[string]int{}, MaxMarketingPerDay: 1,
		queues: map[Channel][]Message{},
	}
}

func inQuiet(u User, now int64) bool {
	if u.QuietFrom == u.QuietTo {
		return false
	}
	h := int(now % 86400 / 3600)
	if u.QuietFrom < u.QuietTo {
		return h >= u.QuietFrom && h < u.QuietTo
	}
	return h >= u.QuietFrom || h < u.QuietTo
}

// quietEnd is the next time the clock reads QuietTo:00.
func quietEnd(u User, now int64) int64 {
	day := now / 86400 * 86400
	end := day + int64(u.QuietTo)*3600
	if end <= now {
		end += 86400
	}
	return end
}

// Accepted says what happened to a request: channels queued, channels skipped (with the
// reason), and when sending starts (later than now during quiet hours).
type Accepted struct {
	Queued  []Channel
	Skipped []string
	Due     int64
}

// Send accepts one request. The whole check-and-accept runs under one lock, so two copies of
// a request racing each other can't both get through.
func (s *Service) Send(r Request, now int64) (Accepted, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.seen[r.ID] {
		return Accepted{}, ErrDuplicate
	}
	u, ok := s.users[r.User]
	if !ok {
		return Accepted{}, ErrUnknownUser
	}
	t, ok := s.templates[r.Template]
	if !ok {
		return Accepted{}, ErrUnknownTemplate
	}
	body, err := Render(t.Text, r.Params)
	if err != nil {
		return Accepted{}, err
	}
	var a Accepted
	dayKey := r.User + "/" + strconv.FormatInt(now/86400, 10)
	if t.Kind == Marketing && s.marketing[dayKey] >= s.MaxMarketingPerDay {
		return Accepted{}, ErrRateLimited
	}
	a.Due = now
	if t.Kind == Marketing && inQuiet(u, now) {
		a.Due = quietEnd(u, now)
	}
	for _, ch := range t.Channels {
		to, ok := u.Contacts[ch]
		switch {
		case !ok:
			a.Skipped = append(a.Skipped, string(ch)+": no address")
		case t.Kind == Marketing && u.OptOut[ch]:
			a.Skipped = append(a.Skipped, string(ch)+": opted out")
		default:
			s.seq++
			s.queues[ch] = append(s.queues[ch], Message{Key: r.ID + ":" + string(ch), Channel: ch, To: to, Body: body, Kind: t.Kind, Due: a.Due, seq: s.seq})
			a.Queued = append(a.Queued, ch)
		}
	}
	s.seen[r.ID] = true
	if t.Kind == Marketing && len(a.Queued) > 0 {
		s.marketing[dayKey]++
	}
	return a, nil
}

// take removes the earliest due message (on one channel, or on any if ch is ""), if one is due.
func (s *Service) take(ch Channel, now int64) (Message, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	bestCh, best := Channel(""), -1
	for _, c := range []Channel{Push, SMS, Email} {
		if ch != "" && c != ch {
			continue
		}
		for i, m := range s.queues[c] {
			if m.Due > now {
				continue
			}
			if best < 0 || m.Due < s.queues[bestCh][best].Due ||
				(m.Due == s.queues[bestCh][best].Due && m.seq < s.queues[bestCh][best].seq) {
				bestCh, best = c, i
			}
		}
	}
	if best < 0 {
		return Message{}, false
	}
	q := s.queues[bestCh]
	m := q[best]
	s.queues[bestCh] = append(q[:best:best], q[best+1:]...)
	return m, true
}

// deliver calls the provider outside the lock (it's network I/O), then records the result.
func (s *Service) deliver(m Message, now int64) Event {
	out := s.providers[m.Channel].Send(m)
	m.Attempt++
	ev := Event{Time: now, Msg: m, Outcome: out}
	s.mu.Lock()
	defer s.mu.Unlock()
	switch {
	case out == OK:
		s.Delivered++
	case out == Transient && m.Attempt < MaxAttempts:
		m.Due = now + Backoff(m.Attempt)
		s.queues[m.Channel] = append(s.queues[m.Channel], m)
		ev.RetryAt = m.Due
	default:
		s.Dead = append(s.Dead, m)
		ev.Dead = true
	}
	return ev
}

// Work is one worker step on one channel's queue: send the earliest due message, if any.
func (s *Service) Work(ch Channel, now int64) (Event, bool) {
	m, ok := s.take(ch, now)
	if !ok {
		return Event{}, false
	}
	return s.deliver(m, now), true
}

// Step sends the earliest due message on any channel (the demo's single worker).
func (s *Service) Step(now int64) (Event, bool) { return s.Work("", now) }

// NextDue is the earliest time any queued message becomes due.
func (s *Service) NextDue() (int64, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var next int64
	found := false
	for _, q := range s.queues {
		for _, m := range q {
			if !found || m.Due < next {
				next, found = m.Due, true
			}
		}
	}
	return next, found
}

// Pending is how many messages are still queued.
func (s *Service) Pending() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	n := 0
	for _, q := range s.queues {
		n += len(q)
	}
	return n
}

// cmd/demo/main.go
// Command demo sends a few order updates and a marketing campaign through the notification
// core, with a flaky SMS gateway, a bad email address, quiet hours and duplicate requests.
package main

import (
	"fmt"
	"strings"

	"notify"
)

func main() {
	users := map[string]notify.User{
		"rafi": {Contacts: map[notify.Channel]string{notify.Push: "tok-rafi", notify.SMS: "+8801711000001", notify.Email: "rafi@example.com"},
			QuietFrom: 22, QuietTo: 8},
		"nusrat": {Contacts: map[notify.Channel]string{notify.SMS: "+8801811000002", notify.Email: "nusrat@example.com"},
			OptOut: map[notify.Channel]bool{notify.Email: true}},
		"tanvir": {Contacts: map[notify.Channel]string{notify.Email: "tanvir@bad-domain"}},
	}
	templates := map[string]notify.Template{
		"order_shipped": {Kind: notify.Transactional, Channels: []notify.Channel{notify.Push, notify.SMS}, Text: "Order {order} is on the way, arriving {day}"},
		"eid_sale":      {Kind: notify.Marketing, Channels: []notify.Channel{notify.Push, notify.Email}, Text: "Eid sale: {pct}% off until Friday"},
	}
	providers := map[notify.Channel]notify.Provider{
		notify.Push:  notify.NewFakeProvider(nil),
		notify.SMS:   notify.NewFakeProvider(map[string][]notify.Outcome{"+8801811000002": {notify.Transient, notify.Transient}}),
		notify.Email: notify.NewFakeProvider(map[string][]notify.Outcome{"tanvir@bad-domain": {notify.Permanent}}),
	}
	s := notify.NewService(users, templates, providers)

	send := func(now int64, r notify.Request) {
		a, err := s.Send(r, now)
		head := fmt.Sprintf("  %-16s %-13s %-6s ->", r.ID, r.Template, r.User)
		if err != nil {
			fmt.Printf("%s error: %v\n", head, err)
			return
		}
		parts := []string{}
		for _, ch := range a.Queued {
			parts = append(parts, string(ch))
		}
		line := "queued " + strings.Join(parts, ", ")
		if a.Due > now {
			line += " for " + notify.Clock(a.Due) + " (quiet hours)"
		}
		if len(a.Queued) == 0 {
			line = "nothing to send"
		}
		if len(a.Skipped) > 0 {
			line += " (skipped " + strings.Join(a.Skipped, "; ") + ")"
		}
		fmt.Printf("%s %s\n", head, line)
	}
	run := func(until int64) {
		for {
			next, ok := s.NextDue()
			if !ok || next > until {
				return
			}
			ev, _ := s.Step(next)
			line := ev.Outcome.String()
			switch {
			case ev.RetryAt > 0:
				line += ", retry at " + notify.Clock(ev.RetryAt)
			case ev.Dead:
				line += ", moved to dead letters"
			default:
				line += ": " + ev.Msg.Body
			}
			fmt.Printf("  %s %-5s %-19s try %d: %s\n", notify.Clock(ev.Time), ev.Msg.Channel, ev.Msg.Key, ev.Msg.Attempt, line)
		}
	}

	t1 := int64(21*3600 + 59*60 + 50)
	fmt.Println("D1 21:59:50, the order service sends three requests:")
	send(t1, notify.Request{ID: "evt-1", User: "rafi", Template: "order_shipped", Params: map[string]string{"order": "BD-7781", "day": "Sunday"}})
	send(t1, notify.Request{ID: "evt-1", User: "rafi", Template: "order_shipped", Params: map[string]string{"order": "BD-7781", "day": "Sunday"}})
	send(t1, notify.Request{ID: "evt-2", User: "nusrat", Template: "order_shipped", Params: map[string]string{"order": "BD-7790", "day": "Monday"}})
	fmt.Println("workers:")
	run(t1 + 10)

	t2 := int64(22*3600 + 5)
	fmt.Println("D1 22:00:05, the marketing team starts a campaign:")
	send(t2, notify.Request{ID: "camp-7:rafi", User: "rafi", Template: "eid_sale", Params: map[string]string{"pct": "30"}})
	send(t2, notify.Request{ID: "camp-7:nusrat", User: "nusrat", Template: "eid_sale", Params: map[string]string{"pct": "30"}})
	send(t2, notify.Request{ID: "camp-7:tanvir", User: "tanvir", Template: "eid_sale", Params: map[string]string{"pct": "30"}})
	send(t2, notify.Request{ID: "camp-8:rafi", User: "rafi", Template: "eid_sale", Params: map[string]string{"pct": "40"}})
	send(t2, notify.Request{ID: "camp-9:tanvir", User: "tanvir", Template: "eid_sale", Params: map[string]string{}})
	fmt.Println("workers:")
	run(2 * 86400)
	fmt.Printf("delivered %d, dead letters %d, still queued %d\n", s.Delivered, len(s.Dead), s.Pending())
}

Demo চালালে:

$ go run ./cmd/demo
D1 21:59:50, the order service sends three requests:
  evt-1            order_shipped rafi   -> queued push, sms
  evt-1            order_shipped rafi   -> error: duplicate request, already accepted
  evt-2            order_shipped nusrat -> queued sms (skipped push: no address)
workers:
  D1 21:59:50 push  evt-1:push          try 1: delivered: Order BD-7781 is on the way, arriving Sunday
  D1 21:59:50 sms   evt-1:sms           try 1: delivered: Order BD-7781 is on the way, arriving Sunday
  D1 21:59:50 sms   evt-2:sms           try 1: failed (timeout), retry at D1 21:59:51
  D1 21:59:51 sms   evt-2:sms           try 2: failed (timeout), retry at D1 21:59:53
  D1 21:59:53 sms   evt-2:sms           try 3: delivered: Order BD-7790 is on the way, arriving Monday
D1 22:00:05, the marketing team starts a campaign:
  camp-7:rafi      eid_sale      rafi   -> queued push, email for D2 08:00:00 (quiet hours)
  camp-7:nusrat    eid_sale      nusrat -> nothing to send (skipped push: no address; email: opted out)
  camp-7:tanvir    eid_sale      tanvir -> queued email (skipped push: no address)
  camp-8:rafi      eid_sale      rafi   -> error: marketing limit reached for today
  camp-9:tanvir    eid_sale      tanvir -> error: missing parameter "pct"
workers:
  D1 22:00:05 email camp-7:tanvir:email try 1: rejected (bad address), moved to dead letters
  D2 08:00:00 push  camp-7:rafi:push    try 1: delivered: Eid sale: 30% off until Friday
  D2 08:00:00 email camp-7:rafi:email   try 1: delivered: Eid sale: 30% off until Friday
delivered 5, dead letters 1, still queued 0

Output টা পড়া:

• Duplicate: দ্বিতীয় evt-1 আগেই নেওয়া হয়েছে বলে ফেরত যায়। রাফি একটা push আর একটা SMS পায়, প্রতিটার দুইটা না।
• বাদ পড়া channel: নুসরাতের device token নেই, তাই evt-2 শুধু SMS এ যায়, আর response বলে push কেন বাদ গেল। কী হলো caller সেটা ঠিকঠাক জানে।
• মাঝে মাঝে fail করা gateway: নুসরাতের SMS 21:59:50 এ timeout, 1 s পর আবার চেষ্টা, আবার timeout, 2 s অপেক্ষা, আর 21:59:53 এ try 3 এ পৌঁছায়। একই মুহূর্তে queue তে যাওয়া রাফির message গুলো আগে চলে গেছে, আটকে থাকেনি।
• Quiet hours: 22:00:05 এ রাফি quiet hours এ (22 থেকে 8), তাই campaign টা D2 08:00:00 এর জন্য queue তে যায় আর তখনই পৌঁছায়। এক ঘণ্টা আগের order update দেরি হয়নি, কারণ quiet hours শুধু marketing এ খাটে।
• Opt-out: নুসরাত marketing email বন্ধ করেছে আর তার push token নেই, তাই campaign এর তাকে পাঠানোর কিছু নেই। এটা শূন্য message সহ একটা সফল request, error না।
• স্থায়ী failure: তানভীরের email address ভুল। Provider try 1 এই ফিরিয়ে দেয়, সোজা dead letter এ যায়; retry করলে শুধু call নষ্ট হতো।
• সীমা আর যাচাই: একই দিনে রাফিকে দ্বিতীয় campaign দিনের marketing সীমায় আটকায়; pct ছাড়া একটা request ফিরিয়ে দেওয়া হয়, "Eid sale: % off" পাঠানোর বদলে।
• মোট: 5 টা পৌঁছেছে, 1 টা dead letter, queue তে কিছু বাকি নেই।

Test গুলো সবচেয়ে জরুরি জিনিসগুলো যাচাই করে: একসাথে 50 টা একই request থেকে ঠিক একটা message queue তে যায় (-race), retry হয় 0, 1, 3 আর 7 s এ, তারপর DLQ, স্থায়ী failure এ retry হয় না, quiet hours শুধু marketing পিছায় আর মাঝরাত পেরিয়েও ঠিক কাজ করে, opt-out transactional message আটকায় না, fail করা request তার idempotency key নষ্ট করে না, আর একটা queue থেকে চারটা worker একসাথে নিলেও প্রতিটা message ঠিক একবার যায়। Duplicate যাচাই আর নেওয়ার মাঝে lock ছেড়ে দিলে duplicate test fail করে (এক run এ 4 টা copy ঢুকে গেছে), তাই test টা সত্যিই ওই ধাপটা পাহারা দেয়।

$ go test -race -v ./...
=== RUN   TestDuplicateRequestsQueueOnce
--- PASS: TestDuplicateRequestsQueueOnce (0.00s)
=== RUN   TestRetriesBackOffThenDeadLetter
--- PASS: TestRetriesBackOffThenDeadLetter (0.00s)
=== RUN   TestPermanentFailureIsNotRetried
--- PASS: TestPermanentFailureIsNotRetried (0.00s)
=== RUN   TestQuietHoursDelayMarketingOnly
--- PASS: TestQuietHoursDelayMarketingOnly (0.00s)
=== RUN   TestOptOutAppliesToMarketingOnly
--- PASS: TestOptOutAppliesToMarketingOnly (0.00s)
=== RUN   TestMarketingLimitAndValidation
--- PASS: TestMarketingLimitAndValidation (0.00s)
=== RUN   TestConcurrentWorkersSendEachMessageOnce
--- PASS: TestConcurrentWorkersSendEachMessageOnce (0.02s)
PASS
ok  	notify	1.034s
?   	notify/cmd/demo	[no test files]

একই core আরো পাঁচটা language এ

Single-threaded version, একই class, একই output। এখানে worker loop একটার পর একটা চলে; worker একসাথে চলে Go version এ। যা আলাদা:

• Python: message গুলো frozen dataclass, আর retry replace() দিয়ে বদলানো একটা copy বানায়, তাই queue তে থাকা message কখনো জায়গায় বদলায় না।
• TypeScript: Outcome একটা enum না, const object, কারণ Node .ts file চালায় type মুছে দিয়ে, আর enum শুধু type না। SendRequest আর Attempt নাম যাতে DOM এর global Request আর Event এর সাথে না লাগে।
• Rust: Channel, Kind আর Outcome হলো enum, আর match এ সব case ধরতে হয়, তাই নতুন channel যোগ করলে প্রতিটা match কে সেটা সামলাতে হয়। render আগেভাগে ফেরার জন্য let ... else ব্যবহার করে।
• Java: অপরিবর্তনীয় message এর জন্য record; send আর take হলো synchronized, আর provider call lock এর বাইরে, Go এর মতোই।
• C++: একই critical section এর চারপাশে std::lock_guard সহ একটা std::mutex; "কোনো message due না" বোঝাতে std::optional।

"""A notification system's core: requests are deduplicated, checked against preferences,
rendered from templates and put on one queue per channel; workers send them through
providers, retry transient failures with exponential backoff, and move messages that can't be
delivered to a dead-letter queue."""
from dataclasses import dataclass, field, replace
from enum import Enum

PUSH, SMS, EMAIL = "push", "sms", "email"
CHANNELS = [PUSH, SMS, EMAIL]
TRANSACTIONAL, MARKETING = 0, 1
MAX_ATTEMPTS = 4


class Outcome(Enum):
    OK = "delivered"
    TRANSIENT = "failed (timeout)"
    PERMANENT = "rejected (bad address)"


@dataclass
class Template:
    kind: int
    channels: list[str]
    text: str


@dataclass
class User:
    contacts: dict[str, str]
    opt_out: set[str] = field(default_factory=set)  # marketing opt-outs per channel
    quiet_from: int = 0  # quiet hours for marketing, e.g. 22 to 8 (0, 0 = none)
    quiet_to: int = 0


@dataclass
class Request:
    id: str  # the idempotency key
    user: str
    template: str
    params: dict[str, str]


@dataclass(frozen=True)
class Message:
    key: str  # request ID + channel, e.g. "evt-1:sms"
    channel: str
    to: str
    body: str
    kind: int
    attempt: int  # attempts made so far
    due: int  # simulated seconds; not sent before this
    seq: int  # enqueue order, breaks ties


@dataclass
class Accepted:
    queued: list[str] = field(default_factory=list)
    skipped: list[str] = field(default_factory=list)
    due: int = 0


@dataclass
class Event:
    time: int
    msg: Message
    outcome: Outcome
    retry_at: int = 0
    dead: bool = False


class Rejected(Exception):
    pass


def render(text: str, params: dict[str, str]) -> str:
    """Fills {name} placeholders; a missing value is an error, never a blank in a message."""
    out = []
    while True:
        i = text.find("{")
        if i < 0:
            return "".join(out) + text
        j = text.find("}", i)
        if j < 0:
            raise Rejected(f"unclosed placeholder in {text!r}")
        name = text[i + 1 : j]
        if name not in params:
            raise Rejected(f'missing parameter "{name}"')
        out += [text[:i], params[name]]
        text = text[j + 1 :]


def clock(t: int) -> str:
    s = t % 86400
    return f"D{t // 86400 + 1} {s // 3600:02d}:{s // 60 % 60:02d}:{s % 60:02d}"


def backoff(attempt: int) -> int:
    """Wait before retry n: 1, 2, 4... seconds (production: minutes, with jitter)."""
    return 1 << (attempt - 1)


class FakeProvider:
    """Plays back scripted outcomes per address; after the script runs out, sends succeed."""

    def __init__(self, script: dict[str, list[Outcome]] | None = None):
        self.script = script or {}
        self.sent: dict[str, int] = {}

    def send(self, m: Message) -> Outcome:
        self.sent[m.key] = self.sent.get(m.key, 0) + 1
        s = self.script.get(m.to, [])
        return s[m.attempt] if m.attempt < len(s) else Outcome.OK


def in_quiet(u: User, now: int) -> bool:
    if u.quiet_from == u.quiet_to:
        return False
    h = now % 86400 // 3600
    if u.quiet_from < u.quiet_to:
        return u.quiet_from <= h < u.quiet_to
    return h >= u.quiet_from or h < u.quiet_to


def quiet_end(u: User, now: int) -> int:
    end = now // 86400 * 86400 + u.quiet_to * 3600
    return end + 86400 if end <= now else end


class Service:
    def __init__(self, users: dict[str, User], templates: dict[str, Template], providers: dict[str, FakeProvider]):
        self.users, self.templates, self.providers = users, templates, providers
        self.seen: set[str] = set()  # accepted request IDs (production: a store with a TTL)
        self.marketing: dict[str, int] = {}  # "user/day" -> marketing messages accepted
        self.max_marketing_per_day = 1
        self.queues: dict[str, list[Message]] = {c: [] for c in CHANNELS}
        self.seq = 0
        self.dead: list[Message] = []
        self.delivered = 0

    def send(self, r: Request, now: int) -> Accepted:
        if r.id in self.seen:
            raise Rejected("duplicate request, already accepted")
        u = self.users.get(r.user)
        if u is None:
            raise Rejected("unknown user")
        t = self.templates.get(r.template)
        if t is None:
            raise Rejected("unknown template")
        body = render(t.text, r.params)
        day_key = f"{r.user}/{now // 86400}"
        if t.kind == MARKETING and self.marketing.get(day_key, 0) >= self.max_marketing_per_day:
            raise Rejected("marketing limit reached for today")
        a = Accepted(due=now)
        if t.kind == MARKETING and in_quiet(u, now):
            a.due = quiet_end(u, now)
        for ch in t.channels:
            if ch not in u.contacts:
                a.skipped.append(f"{ch}: no address")
            elif t.kind == MARKETING and ch in u.opt_out:
                a.skipped.append(f"{ch}: opted out")
            else:
                self.seq += 1
                self.queues[ch].append(Message(f"{r.id}:{ch}", ch, u.contacts[ch], body, t.kind, 0, a.due, self.seq))
                a.queued.append(ch)
        self.seen.add(r.id)
        if t.kind == MARKETING and a.queued:
            self.marketing[day_key] = self.marketing.get(day_key, 0) + 1
        return a

    def _take(self, now: int) -> Message | None:
        due = [m for c in CHANNELS for m in self.queues[c] if m.due <= now]
        if not due:
            return None
        m = min(due, key=lambda m: (m.due, m.seq))
        self.queues[m.channel].remove(m)
        return m

    def step(self, now: int) -> Event | None:
        m = self._take(now)
        if m is None:
            return None
        out = self.providers[m.channel].send(m)
        m = replace(m, attempt=m.attempt + 1)
        ev = Event(now, m, out)
        if out == Outcome.OK:
            self.delivered += 1
        elif out == Outcome.TRANSIENT and m.attempt < MAX_ATTEMPTS:
            m = replace(m, due=now + backoff(m.attempt))
            self.queues[m.channel].append(m)
            ev.retry_at = m.due
        else:
            self.dead.append(m)
            ev.dead = True
        return ev

    def next_due(self) -> int | None:
        dues = [m.due for q in self.queues.values() for m in q]
        return min(dues) if dues else None

    def pending(self) -> int:
        return sum(len(q) for q in self.queues.values())


def main() -> None:
    users = {
        "rafi": User({PUSH: "tok-rafi", SMS: "+8801711000001", EMAIL: "rafi@example.com"}, quiet_from=22, quiet_to=8),
        "nusrat": User({SMS: "+8801811000002", EMAIL: "nusrat@example.com"}, opt_out={EMAIL}),
        "tanvir": User({EMAIL: "tanvir@bad-domain"}),
    }
    templates = {
        "order_shipped": Template(TRANSACTIONAL, [PUSH, SMS], "Order {order} is on the way, arriving {day}"),
        "eid_sale": Template(MARKETING, [PUSH, EMAIL], "Eid sale: {pct}% off until Friday"),
    }
    providers = {
        PUSH: FakeProvider(),
        SMS: FakeProvider({"+8801811000002": [Outcome.TRANSIENT, Outcome.TRANSIENT]}),
        EMAIL: FakeProvider({"tanvir@bad-domain": [Outcome.PERMANENT]}),
    }
    s = Service(users, templates, providers)

    def send(now: int, r: Request) -> None:
        head = f"  {r.id:<16} {r.template:<13} {r.user:<6} ->"
        try:
            a = s.send(r, now)
        except Rejected as e:
            print(f"{head} error: {e}")
            return
        line = "queued " + ", ".join(a.queued)
        if a.due > now:
            line += f" for {clock(a.due)} (quiet hours)"
        if not a.queued:
            line = "nothing to send"
        if a.skipped:
            line += " (skipped " + "; ".join(a.skipped) + ")"
        print(f"{head} {line}")

    def run(until: int) -> None:
        while (nxt := s.next_due()) is not None and nxt <= until:
            ev = s.step(nxt)
            assert ev is not None
            line = ev.outcome.value
            if ev.retry_at:
                line += f", retry at {clock(ev.retry_at)}"
            elif ev.dead:
                line += ", moved to dead letters"
            else:
                line += f": {ev.msg.body}"
            print(f"  {clock(ev.time)} {ev.msg.channel:<5} {ev.msg.key:<19} try {ev.msg.attempt}: {line}")

    t1 = 21 * 3600 + 59 * 60 + 50
    print("D1 21:59:50, the order service sends three requests:")
    send(t1, Request("evt-1", "rafi", "order_shipped", {"order": "BD-7781", "day": "Sunday"}))
    send(t1, Request("evt-1", "rafi", "order_shipped", {"order": "BD-7781", "day": "Sunday"}))
    send(t1, Request("evt-2", "nusrat", "order_shipped", {"order": "BD-7790", "day": "Monday"}))
    print("workers:")
    run(t1 + 10)

    t2 = 22 * 3600 + 5
    print("D1 22:00:05, the marketing team starts a campaign:")
    send(t2, Request("camp-7:rafi", "rafi", "eid_sale", {"pct": "30"}))
    send(t2, Request("camp-7:nusrat", "nusrat", "eid_sale", {"pct": "30"}))
    send(t2, Request("camp-7:tanvir", "tanvir", "eid_sale", {"pct": "30"}))
    send(t2, Request("camp-8:rafi", "rafi", "eid_sale", {"pct": "40"}))
    send(t2, Request("camp-9:tanvir", "tanvir", "eid_sale", {}))
    print("workers:")
    run(2 * 86400)
    print(f"delivered {s.delivered}, dead letters {len(s.dead)}, still queued {s.pending()}")


if __name__ == "__main__":
    main()

// A notification system's core: requests are deduplicated, checked against preferences,
// rendered from templates and put on one queue per channel; workers send them through
// providers, retry transient failures with exponential backoff, and move messages that can't
// be delivered to a dead-letter queue.

type Channel = "push" | "sms" | "email";
const CHANNELS: Channel[] = ["push", "sms", "email"];
type Kind = "transactional" | "marketing";
const MAX_ATTEMPTS = 4;

// A const object instead of an enum: Node runs this file by stripping types, and enums aren't types.
const Outcome = { OK: "delivered", Transient: "failed (timeout)", Permanent: "rejected (bad address)" } as const;
type Outcome = (typeof Outcome)[keyof typeof Outcome];

type Template = { kind: Kind; channels: Channel[]; text: string };
type User = {
  contacts: Partial<Record<Channel, string>>;
  optOut?: Set<Channel>; // marketing opt-outs per channel
  quietFrom?: number; // quiet hours for marketing, e.g. 22 to 8
  quietTo?: number;
};
// SendRequest and Attempt, not Request and Event: those names are DOM globals.
type SendRequest = { id: string; user: string; template: string; params: Record<string, string> };
type Message = {
  readonly key: string; // request ID + channel, e.g. "evt-1:sms"
  readonly channel: Channel;
  readonly to: string;
  readonly body: string;
  readonly kind: Kind;
  readonly attempt: number; // attempts made so far
  readonly due: number; // simulated seconds; not sent before this
  readonly seq: number; // enqueue order, breaks ties
};
type Accepted = { queued: Channel[]; skipped: string[]; due: number };
type Attempt = { time: number; msg: Message; outcome: Outcome; retryAt: number; dead: boolean };

class Rejected extends Error {}

/** Fills {name} placeholders; a missing value is an error, never a blank in a message. */
function render(text: string, params: Record<string, string>): string {
  let out = "";
  for (;;) {
    const i = text.indexOf("{");
    if (i < 0) return out + text;
    const j = text.indexOf("}", i);
    if (j < 0) throw new Rejected(`unclosed placeholder in ${JSON.stringify(text)}`);
    const name = text.slice(i + 1, j);
    if (!(name in params)) throw new Rejected(`missing parameter "${name}"`);
    out += text.slice(0, i) + params[name];
    text = text.slice(j + 1);
  }
}

const pad2 = (n: number) => String(n).padStart(2, "0");
function clock(t: number): string {
  const s = t % 86400;
  return `D${Math.floor(t / 86400) + 1} ${pad2(Math.floor(s / 3600))}:${pad2(Math.floor(s / 60) % 60)}:${pad2(s % 60)}`;
}

/** Wait before retry n: 1, 2, 4... seconds (production: minutes, with jitter). */
const backoff = (attempt: number) => 2 ** (attempt - 1);

/** Plays back scripted outcomes per address; after the script runs out, sends succeed. */
class FakeProvider {
  sent = new Map<string, number>();
  readonly script: Map<string, Outcome[]>;
  constructor(script: Record<string, Outcome[]> = {}) {
    this.script = new Map(Object.entries(script));
  }
  send(m: Message): Outcome {
    this.sent.set(m.key, (this.sent.get(m.key) ?? 0) + 1);
    const s = this.script.get(m.to) ?? [];
    return m.attempt < s.length ? s[m.attempt]! : Outcome.OK;
  }
}

function inQuiet(u: User, now: number): boolean {
  const from = u.quietFrom ?? 0;
  const to = u.quietTo ?? 0;
  if (from === to) return false;
  const h = Math.floor((now % 86400) / 3600);
  return from < to ? h >= from && h < to : h >= from || h < to;
}

function quietEnd(u: User, now: number): number {
  const end = Math.floor(now / 86400) * 86400 + (u.quietTo ?? 0) * 3600;
  return end <= now ? end + 86400 : end;
}

class Service {
  seen = new Set<string>(); // accepted request IDs (production: a store with a TTL)
  marketing = new Map<string, number>(); // "user/day" -> marketing messages accepted
  maxMarketingPerDay = 1;
  queues = new Map<Channel, Message[]>(CHANNELS.map((c) => [c, []]));
  seq = 0;
  dead: Message[] = [];
  delivered = 0;
  readonly users: Map<string, User>;
  readonly templates: Map<string, Template>;
  readonly providers: Map<Channel, FakeProvider>;

  constructor(users: Record<string, User>, templates: Record<string, Template>, providers: Record<Channel, FakeProvider>) {
    this.users = new Map(Object.entries(users));
    this.templates = new Map(Object.entries(templates));
    this.providers = new Map(Object.entries(providers) as [Channel, FakeProvider][]);
  }

  send(r: SendRequest, now: number): Accepted {
    if (this.seen.has(r.id)) throw new Rejected("duplicate request, already accepted");
    const u = this.users.get(r.user);
    if (!u) throw new Rejected("unknown user");
    const t = this.templates.get(r.template);
    if (!t) throw new Rejected("unknown template");
    const body = render(t.text, r.params);
    const dayKey = `${r.user}/${Math.floor(now / 86400)}`;
    if (t.kind === "marketing" && (this.marketing.get(dayKey) ?? 0) >= this.maxMarketingPerDay)
      throw new Rejected("marketing limit reached for today");
    const a: Accepted = { queued: [], skipped: [], due: now };
    if (t.kind === "marketing" && inQuiet(u, now)) a.due = quietEnd(u, now);
    for (const ch of t.channels) {
      const to = u.contacts[ch];
      if (to === undefined) a.skipped.push(`${ch}: no address`);
      else if (t.kind === "marketing" && u.optOut?.has(ch)) a.skipped.push(`${ch}: opted out`);
      else {
        this.queues.get(ch)!.push({ key: `${r.id}:${ch}`, channel: ch, to, body, kind: t.kind, attempt: 0, due: a.due, seq: ++this.seq });
        a.queued.push(ch);
      }
    }
    this.seen.add(r.id);
    if (t.kind === "marketing" && a.queued.length > 0) this.marketing.set(dayKey, (this.marketing.get(dayKey) ?? 0) + 1);
    return a;
  }

  private take(now: number): Message | undefined {
    let best: Message | undefined;
    for (const c of CHANNELS)
      for (const m of this.queues.get(c)!)
        if (m.due <= now && (!best || m.due < best.due || (m.due === best.due && m.seq < best.seq))) best = m;
    if (best) {
      const q = this.queues.get(best.channel)!;
      q.splice(q.indexOf(best), 1);
    }
    return best;
  }

  step(now: number): Attempt | undefined {
    const taken = this.take(now);
    if (!taken) return undefined;
    const outcome = this.providers.get(taken.channel)!.send(taken);
    const m: Message = { ...taken, attempt: taken.attempt + 1 };
    const ev: Attempt = { time: now, msg: m, outcome, retryAt: 0, dead: false };
    if (outcome === Outcome.OK) this.delivered++;
    else if (outcome === Outcome.Transient && m.attempt < MAX_ATTEMPTS) {
      ev.retryAt = now + backoff(m.attempt);
      this.queues.get(m.channel)!.push({ ...m, due: ev.retryAt });
    } else {
      this.dead.push(m);
      ev.dead = true;
    }
    return ev;
  }

  nextDue(): number | undefined {
    let next: number | undefined;
    for (const q of this.queues.values()) for (const m of q) if (next === undefined || m.due < next) next = m.due;
    return next;
  }

  pending(): number {
    let n = 0;
    for (const q of this.queues.values()) n += q.length;
    return n;
  }
}

const s = new Service(
  {
    rafi: { contacts: { push: "tok-rafi", sms: "+8801711000001", email: "rafi@example.com" }, quietFrom: 22, quietTo: 8 },
    nusrat: { contacts: { sms: "+8801811000002", email: "nusrat@example.com" }, optOut: new Set<Channel>(["email"]) },
    tanvir: { contacts: { email: "tanvir@bad-domain" } },
  },
  {
    order_shipped: { kind: "transactional", channels: ["push", "sms"], text: "Order {order} is on the way, arriving {day}" },
    eid_sale: { kind: "marketing", channels: ["push", "email"], text: "Eid sale: {pct}% off until Friday" },
  },
  {
    push: new FakeProvider(),
    sms: new FakeProvider({ "+8801811000002": [Outcome.Transient, Outcome.Transient] }),
    email: new FakeProvider({ "tanvir@bad-domain": [Outcome.Permanent] }),
  },
);

function send(now: number, r: SendRequest): void {
  const head = `  ${r.id.padEnd(16)} ${r.template.padEnd(13)} ${r.user.padEnd(6)} ->`;
  let a: Accepted;
  try {
    a = s.send(r, now);
  } catch (e) {
    if (!(e instanceof Rejected)) throw e;
    console.log(`${head} error: ${e.message}`);
    return;
  }
  let line = "queued " + a.queued.join(", ");
  if (a.due > now) line += ` for ${clock(a.due)} (quiet hours)`;
  if (a.queued.length === 0) line = "nothing to send";
  if (a.skipped.length > 0) line += ` (skipped ${a.skipped.join("; ")})`;
  console.log(`${head} ${line}`);
}

function run(until: number): void {
  for (let next = s.nextDue(); next !== undefined && next <= until; next = s.nextDue()) {
    const ev = s.step(next)!;
    let line: string = ev.outcome;
    if (ev.retryAt) line += `, retry at ${clock(ev.retryAt)}`;
    else if (ev.dead) line += ", moved to dead letters";
    else line += `: ${ev.msg.body}`;
    console.log(`  ${clock(ev.time)} ${ev.msg.channel.padEnd(5)} ${ev.msg.key.padEnd(19)} try ${ev.msg.attempt}: ${line}`);
  }
}

const t1 = 21 * 3600 + 59 * 60 + 50;
console.log("D1 21:59:50, the order service sends three requests:");
send(t1, { id: "evt-1", user: "rafi", template: "order_shipped", params: { order: "BD-7781", day: "Sunday" } });
send(t1, { id: "evt-1", user: "rafi", template: "order_shipped", params: { order: "BD-7781", day: "Sunday" } });
send(t1, { id: "evt-2", user: "nusrat", template: "order_shipped", params: { order: "BD-7790", day: "Monday" } });
console.log("workers:");
run(t1 + 10);

const t2 = 22 * 3600 + 5;
console.log("D1 22:00:05, the marketing team starts a campaign:");
send(t2, { id: "camp-7:rafi", user: "rafi", template: "eid_sale", params: { pct: "30" } });
send(t2, { id: "camp-7:nusrat", user: "nusrat", template: "eid_sale", params: { pct: "30" } });
send(t2, { id: "camp-7:tanvir", user: "tanvir", template: "eid_sale", params: { pct: "30" } });
send(t2, { id: "camp-8:rafi", user: "rafi", template: "eid_sale", params: { pct: "40" } });
send(t2, { id: "camp-9:tanvir", user: "tanvir", template: "eid_sale", params: {} });
console.log("workers:");
run(2 * 86400);
console.log(`delivered ${s.delivered}, dead letters ${s.dead.length}, still queued ${s.pending()}`);