Система уведомлений

<?php

declare(strict_types=1);

final class NotificationService
{
    public function __construct(
        private readonly NotificationRepository $repository,
        private readonly TemplateRenderer $templateRenderer,
        private readonly PreferenceChecker $preferenceChecker,
        private readonly QueuePublisher $queue,
        private readonly DeduplicationService $dedup,
        private readonly RateLimiter $rateLimiter,
    ) {}

    public function send(NotificationRequest $request): void
    {
        // 1. Deduplicate
        if (!$this->dedup->isUnique($request->idempotencyKey)) {
            return; // Already sent
        }

        // 2. Check user preferences
        if (!$this->preferenceChecker->canSend(
            $request->userId,
            $request->category,
            $request->channel,
        )) {
            return; // User opted out
        }

        // 3. Check quiet hours
        if ($this->preferenceChecker->isQuietHour($request->userId, $request->channel)) {
            // Schedule for after quiet hours
            $this->scheduleAfterQuietHours($request);
            return;
        }

        // 4. Rate limit per user
        $rateLimitResult = $this->rateLimiter->allow(
            key: "notif:{$request->userId}:{$request->channel}",
            capacity: $this->getChannelLimit($request->channel),
            rate: 1.0,
        );

        if (!$rateLimitResult['allowed']) {
            // Queue for later
            $this->scheduleWithDelay($request, 60);
            return;
        }

        // 5. Render template
        $rendered = $this->templateRenderer->render(
            $request->templateCode,
            $request->variables,
        );

        // 6. Save to database
        $notification = $this->repository->create(
            userId: $request->userId,
            templateId: $rendered->templateId,
            channel: $request->channel,
            priority: $request->priority,
            payload: $rendered->toArray(),
            idempotencyKey: $request->idempotencyKey,
        );

        // 7. Publish to priority queue
        $this->queue->publish(
            queue: $this->getQueueName($request->channel, $request->priority),
            message: new NotificationMessage(
                notificationId: $notification->id,
                channel: $request->channel,
                payload: $rendered->toArray(),
            ),
        );
    }

    public function sendBatch(BatchNotificationRequest $request): string
    {
        $batchId = bin2hex(random_bytes(16));

        // Publish batch job to dedicated queue
        $this->queue->publish('notification.batch', [
            'batch_id' => $batchId,
            'template_code' => $request->templateCode,
            'user_segment' => $request->userSegment,
            'variables' => $request->variables,
            'channel' => $request->channel,
            'scheduled_at' => $request->scheduledAt?->format('c'),
        ]);

        return $batchId;
    }

    private function getQueueName(string $channel, int $priority): string
    {
        $priorityName = match ($priority) {
            0 => 'critical',
            1 => 'high',
            2 => 'normal',
            3 => 'low',
            default => 'normal',
        };

        return "notification.{$channel}.{$priorityName}";
    }

    private function getChannelLimit(string $channel): int
    {
        return match ($channel) {
            'push' => 30,   // 30 per hour
            'email' => 10,  // 10 per hour
            'sms' => 5,     // 5 per hour
            'in_app' => 100,
            default => 20,
        };
    }
}

package notification

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
)

// NotificationService dispatches notifications to the appropriate channels.
type NotificationService struct {
	repo        NotificationRepository
	renderer    *TemplateRenderer
	prefs       *PreferenceChecker
	queue       QueuePublisher
	dedup       *DeduplicationService
	rateLimiter *RateLimiter
}

func NewNotificationService(
	repo NotificationRepository,
	renderer *TemplateRenderer,
	prefs *PreferenceChecker,
	queue QueuePublisher,
	dedup *DeduplicationService,
	rateLimiter *RateLimiter,
) *NotificationService {
	return &NotificationService{
		repo: repo, renderer: renderer, prefs: prefs,
		queue: queue, dedup: dedup, rateLimiter: rateLimiter,
	}
}

// Send dispatches a single notification after checking dedup, preferences, and rate limits.
func (s *NotificationService) Send(ctx context.Context, req NotificationRequest) error {
	// 1. Deduplicate
	unique, err := s.dedup.IsUnique(ctx, req.IdempotencyKey)
	if err != nil {
		return fmt.Errorf("dedup check: %w", err)
	}
	if !unique {
		return nil // Already sent
	}

	// 2. Check user preferences
	if !s.prefs.CanSend(ctx, req.UserID, req.Category, req.Channel) {
		return nil
	}

	// 3. Check quiet hours
	if s.prefs.IsQuietHour(ctx, req.UserID, req.Channel) {
		return s.scheduleAfterQuietHours(ctx, req)
	}

	// 4. Rate limit per user
	key := fmt.Sprintf("notif:%s:%s", req.UserID, req.Channel)
	allowed, err := s.rateLimiter.Allow(ctx, key, s.channelLimit(req.Channel), 1.0)
	if err != nil {
		return fmt.Errorf("rate limit check: %w", err)
	}
	if !allowed {
		return s.scheduleWithDelay(ctx, req, 60)
	}

	// 5. Render template
	rendered, err := s.renderer.Render(ctx, req.TemplateCode, req.Variables)
	if err != nil {
		return fmt.Errorf("render template: %w", err)
	}

	// 6. Save to database
	notif, err := s.repo.Create(ctx, CreateNotificationParams{
		UserID:         req.UserID,
		TemplateID:     rendered.TemplateID,
		Channel:        req.Channel,
		Priority:       req.Priority,
		Payload:        rendered.ToMap(),
		IdempotencyKey: req.IdempotencyKey,
	})
	if err != nil {
		return fmt.Errorf("create notification: %w", err)
	}

	// 7. Publish to priority queue
	queueName := s.queueName(req.Channel, req.Priority)
	return s.queue.Publish(ctx, queueName, NotificationMessage{
		NotificationID: notif.ID,
		Channel:        req.Channel,
		Payload:        rendered.ToMap(),
	})
}

// SendBatch publishes a batch notification job.
func (s *NotificationService) SendBatch(ctx context.Context, req BatchNotificationRequest) (string, error) {
	b := make([]byte, 16)
	rand.Read(b)
	batchID := hex.EncodeToString(b)

	err := s.queue.Publish(ctx, "notification.batch", map[string]any{
		"batch_id":      batchID,
		"template_code": req.TemplateCode,
		"user_segment":  req.UserSegment,
		"variables":     req.Variables,
		"channel":       req.Channel,
		"scheduled_at":  req.ScheduledAt,
	})
	return batchID, err
}

func (s *NotificationService) queueName(channel string, priority int) string {
	names := map[int]string{0: "critical", 1: "high", 2: "normal", 3: "low"}
	p, ok := names[priority]
	if !ok {
		p = "normal"
	}
	return fmt.Sprintf("notification.%s.%s", channel, p)
}

func (s *NotificationService) channelLimit(channel string) int {
	switch channel {
	case "push":
		return 30
	case "email":
		return 10
	case "sms":
		return 5
	case "in_app":
		return 100
	default:
		return 20
	}
}

func (s *NotificationService) scheduleAfterQuietHours(ctx context.Context, req NotificationRequest) error {
	return nil // Implementation: re-queue with delay until quiet hours end
}

func (s *NotificationService) scheduleWithDelay(ctx context.Context, req NotificationRequest, delaySec int) error {
	return nil // Implementation: publish to delayed queue
}

using System.Security.Cryptography;

namespace Notifications;

// NotificationService dispatches notifications to the appropriate channels.
public sealed class NotificationService(
    INotificationRepository repository,
    TemplateRenderer renderer,
    PreferenceChecker preferences,
    IQueuePublisher queue,
    DeduplicationService dedup,
    RateLimiter rateLimiter)
{
    public async Task SendAsync(NotificationRequest request, CancellationToken ct = default)
    {
        // 1. Deduplicate
        if (!await dedup.IsUniqueAsync(request.IdempotencyKey))
        {
            return; // Already sent
        }

        // 2. Check user preferences
        if (!await preferences.CanSendAsync(request.UserId, request.Category, request.Channel, ct))
        {
            return; // User opted out
        }

        // 3. Check quiet hours
        if (await preferences.IsQuietHourAsync(request.UserId, request.Channel, ct))
        {
            // Schedule for after quiet hours
            await ScheduleAfterQuietHoursAsync(request, ct);
            return;
        }

        // 4. Rate limit per user
        var key = $"notif:{request.UserId}:{request.Channel}";
        var allowed = await rateLimiter.AllowAsync(key, ChannelLimit(request.Channel), rate: 1.0, ct);
        if (!allowed)
        {
            // Queue for later
            await ScheduleWithDelayAsync(request, TimeSpan.FromSeconds(60), ct);
            return;
        }

        // 5. Render template
        var rendered = await renderer.RenderAsync(request.TemplateCode, request.Variables, ct);

        // 6. Save to database
        var notification = await repository.CreateAsync(new CreateNotificationParams(
            UserId: request.UserId,
            TemplateId: rendered.TemplateId,
            Channel: request.Channel,
            Priority: request.Priority,
            Payload: rendered.ToDictionary(),
            IdempotencyKey: request.IdempotencyKey), ct);

        // 7. Publish to priority queue
        await queue.PublishAsync(
            QueueName(request.Channel, request.Priority),
            new NotificationMessage(notification.Id, request.Channel, rendered.ToDictionary()),
            ct);
    }

    // SendBatchAsync publishes a batch notification job.
    public async Task<string> SendBatchAsync(BatchNotificationRequest request, CancellationToken ct = default)
    {
        var batchId = Convert.ToHexString(RandomNumberGenerator.GetBytes(16)).ToLowerInvariant();

        await queue.PublishAsync("notification.batch", new Dictionary<string, object?>
        {
            ["batch_id"] = batchId,
            ["template_code"] = request.TemplateCode,
            ["user_segment"] = request.UserSegment,
            ["variables"] = request.Variables,
            ["channel"] = request.Channel,
            ["scheduled_at"] = request.ScheduledAt?.ToString("O"),
        }, ct);

        return batchId;
    }

    private static string QueueName(string channel, int priority)
    {
        var priorityName = priority switch
        {
            0 => "critical",
            1 => "high",
            2 => "normal",
            3 => "low",
            _ => "normal",
        };

        return $"notification.{channel}.{priorityName}";
    }

    private static int ChannelLimit(string channel) => channel switch
    {
        "push" => 30,   // 30 per hour
        "email" => 10,  // 10 per hour
        "sms" => 5,     // 5 per hour
        "in_app" => 100,
        _ => 20,
    };

    private async Task ScheduleAfterQuietHoursAsync(NotificationRequest request, CancellationToken ct)
    {
        var delay = await preferences.TimeUntilQuietHoursEndAsync(request.UserId, request.Channel, ct);
        await queue.PublishWithDelayAsync($"notification.{request.Channel}.scheduled", request, delay, ct);
    }

    private Task ScheduleWithDelayAsync(NotificationRequest request, TimeSpan delay, CancellationToken ct)
        => queue.PublishWithDelayAsync($"notification.{request.Channel}.delayed", request, delay, ct);
}

from __future__ import annotations

import secrets
from dataclasses import dataclass
from datetime import timedelta

CHANNEL_LIMITS = {"push": 30, "email": 10, "sms": 5, "in_app": 100}  # per hour
DEFAULT_CHANNEL_LIMIT = 20
PRIORITY_NAMES = {0: "critical", 1: "high", 2: "normal", 3: "low"}
RATE_LIMIT_RETRY = timedelta(seconds=60)

@dataclass(slots=True)
class NotificationService:
    """Dispatches notifications to the appropriate channels."""

    repository: NotificationRepository
    renderer: TemplateRenderer
    preferences: PreferenceChecker
    queue: QueuePublisher
    dedup: DeduplicationService
    rate_limiter: RateLimiter

    async def send(self, request: NotificationRequest) -> None:
        # 1. Deduplicate
        if not await self.dedup.is_unique(request.idempotency_key):
            return  # Already sent

        # 2. Check user preferences
        if not await self.preferences.can_send(request.user_id, request.category, request.channel):
            return  # User opted out

        # 3. Check quiet hours
        if await self.preferences.is_quiet_hour(request.user_id, request.channel):
            # Schedule for after quiet hours
            await self._schedule_after_quiet_hours(request)
            return

        # 4. Rate limit per user
        key = f"notif:{request.user_id}:{request.channel}"
        allowed = await self.rate_limiter.allow(key, capacity=_channel_limit(request.channel), rate=1.0)
        if not allowed:
            # Queue for later
            await self._schedule_with_delay(request, RATE_LIMIT_RETRY)
            return

        # 5. Render template
        rendered = await self.renderer.render(request.template_code, request.variables)

        # 6. Save to database
        notification = await self.repository.create(
            user_id=request.user_id,
            template_id=rendered.template_id,
            channel=request.channel,
            priority=request.priority,
            payload=rendered.to_dict(),
            idempotency_key=request.idempotency_key,
        )

        # 7. Publish to priority queue
        await self.queue.publish(
            _queue_name(request.channel, request.priority),
            NotificationMessage(
                notification_id=notification.id,
                channel=request.channel,
                payload=rendered.to_dict(),
            ),
        )

    async def send_batch(self, request: BatchNotificationRequest) -> str:
        """Publish a batch notification job."""
        batch_id = secrets.token_hex(16)

        await self.queue.publish(
            "notification.batch",
            {
                "batch_id": batch_id,
                "template_code": request.template_code,
                "user_segment": request.user_segment,
                "variables": request.variables,
                "channel": request.channel,
                "scheduled_at": request.scheduled_at.isoformat() if request.scheduled_at else None,
            },
        )

        return batch_id

    async def _schedule_after_quiet_hours(self, request: NotificationRequest) -> None:
        delay = await self.preferences.time_until_quiet_hours_end(request.user_id, request.channel)
        await self.queue.publish_with_delay(f"notification.{request.channel}.scheduled", request, delay)

    async def _schedule_with_delay(self, request: NotificationRequest, delay: timedelta) -> None:
        await self.queue.publish_with_delay(f"notification.{request.channel}.delayed", request, delay)

def _queue_name(channel: str, priority: int) -> str:
    return f"notification.{channel}.{PRIORITY_NAMES.get(priority, 'normal')}"

def _channel_limit(channel: str) -> int:
    return CHANNEL_LIMITS.get(channel, DEFAULT_CHANNEL_LIMIT)

<?php

declare(strict_types=1);

final class TemplateRenderer
{
    public function __construct(
        private readonly TemplateRepository $templates,
    ) {}

    public function render(string $templateCode, array $variables): RenderedNotification
    {
        $template = $this->templates->findByCode($templateCode);

        if ($template === null) {
            throw new TemplateNotFoundException("Template not found: {$templateCode}");
        }

        // Validate required variables
        $requiredVars = json_decode($template->variables, true);
        foreach ($requiredVars as $var) {
            if (!isset($variables[$var])) {
                throw new MissingVariableException("Missing variable: {$var}");
            }
        }

        // Render subject and body
        $subject = $this->interpolate($template->subject ?? '', $variables);
        $body = $this->interpolate($template->body, $variables);

        return new RenderedNotification(
            templateId: $template->id,
            subject: $subject,
            body: $body,
            channel: $template->channel,
        );
    }

    private function interpolate(string $template, array $variables): string
    {
        return preg_replace_callback(
            '/\{\{(\w+)\}\}/',
            fn (array $matches) => $variables[$matches[1]] ?? $matches[0],
            $template,
        );
    }
}

final readonly class RenderedNotification
{
    public function __construct(
        public string $templateId,
        public string $subject,
        public string $body,
        public string $channel,
    ) {}

    public function toArray(): array
    {
        return [
            'template_id' => $this->templateId,
            'subject' => $this->subject,
            'body' => $this->body,
            'channel' => $this->channel,
        ];
    }
}

package notification

import (
	"context"
	"fmt"
	"regexp"
)

// RenderedNotification holds a rendered template result.
type RenderedNotification struct {
	TemplateID string
	Subject    string
	Body       string
	Channel    string
}

func (r *RenderedNotification) ToMap() map[string]string {
	return map[string]string{
		"template_id": r.TemplateID,
		"subject":     r.Subject,
		"body":        r.Body,
		"channel":     r.Channel,
	}
}

// TemplateRenderer renders notification templates with variable interpolation.
type TemplateRenderer struct {
	templates TemplateRepository
}

func NewTemplateRenderer(templates TemplateRepository) *TemplateRenderer {
	return &TemplateRenderer{templates: templates}
}

var varPattern = regexp.MustCompile(`\{\{(\w+)\}\}`)

// Render finds the template and interpolates variables.
func (tr *TemplateRenderer) Render(ctx context.Context, templateCode string, variables map[string]string) (*RenderedNotification, error) {
	tmpl, err := tr.templates.FindByCode(ctx, templateCode)
	if err != nil {
		return nil, fmt.Errorf("find template %q: %w", templateCode, err)
	}

	// Validate required variables
	for _, v := range tmpl.RequiredVars {
		if _, ok := variables[v]; !ok {
			return nil, fmt.Errorf("missing variable: %s", v)
		}
	}

	subject := interpolate(tmpl.Subject, variables)
	body := interpolate(tmpl.Body, variables)

	return &RenderedNotification{
		TemplateID: tmpl.ID,
		Subject:    subject,
		Body:       body,
		Channel:    tmpl.Channel,
	}, nil
}

func interpolate(text string, vars map[string]string) string {
	return varPattern.ReplaceAllStringFunc(text, func(match string) string {
		key := match[2 : len(match)-2] // strip {{ and }}
		if val, ok := vars[key]; ok {
			return val
		}
		return match
	})
}

using System.Text.RegularExpressions;

namespace Notifications;

// RenderedNotification holds a rendered template result.
public sealed record RenderedNotification(string TemplateId, string Subject, string Body, string Channel)
{
    public Dictionary<string, string> ToDictionary() => new()
    {
        ["template_id"] = TemplateId,
        ["subject"] = Subject,
        ["body"] = Body,
        ["channel"] = Channel,
    };
}

// TemplateRenderer renders notification templates with variable interpolation.
public sealed partial class TemplateRenderer(ITemplateRepository templates)
{
    // Source-generated regex is compiled at build time, so there is no runtime cost per call.
    [GeneratedRegex(@"\{\{(\w+)\}\}")]
    private static partial Regex VarPattern();

    public async Task<RenderedNotification> RenderAsync(
        string templateCode,
        IReadOnlyDictionary<string, string> variables,
        CancellationToken ct = default)
    {
        var template = await templates.FindByCodeAsync(templateCode, ct)
            ?? throw new TemplateNotFoundException($"Template not found: {templateCode}");

        // Validate required variables
        foreach (var name in template.RequiredVars)
        {
            if (!variables.ContainsKey(name))
            {
                throw new MissingVariableException($"Missing variable: {name}");
            }
        }

        // Render subject and body
        return new RenderedNotification(
            template.Id,
            Interpolate(template.Subject ?? string.Empty, variables),
            Interpolate(template.Body, variables),
            template.Channel);
    }

    private static string Interpolate(string text, IReadOnlyDictionary<string, string> variables) =>
        VarPattern().Replace(text, m => variables.TryGetValue(m.Groups[1].Value, out var value) ? value : m.Value);
}

from __future__ import annotations

import re
from dataclasses import dataclass

VAR_PATTERN = re.compile(r"\{\{(\w+)\}\}")

class TemplateNotFoundError(Exception):
    """Raised when the template code is unknown."""

class MissingVariableError(Exception):
    """Raised when a required template variable is absent."""

@dataclass(frozen=True, slots=True)
class RenderedNotification:
    """Holds a rendered template result."""

    template_id: str
    subject: str
    body: str
    channel: str

    def to_dict(self) -> dict[str, str]:
        return {
            "template_id": self.template_id,
            "subject": self.subject,
            "body": self.body,
            "channel": self.channel,
        }

@dataclass(slots=True)
class TemplateRenderer:
    """Renders notification templates with variable interpolation."""

    templates: TemplateRepository

    async def render(self, template_code: str, variables: dict[str, str]) -> RenderedNotification:
        template = await self.templates.find_by_code(template_code)
        if template is None:
            raise TemplateNotFoundError(f"Template not found: {template_code}")

        # Validate required variables
        for name in template.required_vars:
            if name not in variables:
                raise MissingVariableError(f"Missing variable: {name}")

        # Render subject and body
        return RenderedNotification(
            template_id=template.id,
            subject=_interpolate(template.subject or "", variables),
            body=_interpolate(template.body, variables),
            channel=template.channel,
        )

def _interpolate(text: str, variables: dict[str, str]) -> str:
    # Unknown placeholders are left untouched
    return VAR_PATTERN.sub(lambda m: variables.get(m.group(1), m.group(0)), text)

<?php

declare(strict_types=1);

interface NotificationSender
{
    public function send(string $notificationId, array $payload): DeliveryResult;
}

final class PushNotificationSender implements NotificationSender
{
    public function __construct(
        private readonly DeviceTokenRepository $deviceTokens,
        private readonly ApnsClient $apns,
        private readonly FcmClient $fcm,
    ) {}

    public function send(string $notificationId, array $payload): DeliveryResult
    {
        $userId = $payload['user_id'];
        $tokens = $this->deviceTokens->getActiveTokens($userId);

        $results = [];
        foreach ($tokens as $token) {
            try {
                $result = match ($token->platform) {
                    'ios' => $this->apns->send($token->token, [
                        'title' => $payload['subject'],
                        'body' => $payload['body'],
                        'data' => $payload['data'] ?? [],
                    ]),
                    'android' => $this->fcm->send($token->token, [
                        'title' => $payload['subject'],
                        'body' => $payload['body'],
                        'data' => $payload['data'] ?? [],
                    ]),
                    default => throw new \InvalidArgumentException("Unknown platform: {$token->platform}"),
                };

                $results[] = $result;
            } catch (InvalidTokenException $e) {
                // Deactivate invalid token
                $this->deviceTokens->deactivate($token->id);
            }
        }

        $success = count(array_filter($results, fn ($r) => $r->isSuccess()));

        return new DeliveryResult(
            success: $success > 0,
            deliveredCount: $success,
            totalCount: count($tokens),
        );
    }
}

final class EmailNotificationSender implements NotificationSender
{
    public function __construct(
        private readonly EmailClient $emailClient,
        private readonly UserRepository $users,
    ) {}

    public function send(string $notificationId, array $payload): DeliveryResult
    {
        $user = $this->users->find($payload['user_id']);

        $result = $this->emailClient->send(
            to: $user->email,
            subject: $payload['subject'],
            htmlBody: $payload['body'],
            headers: [
                'X-Notification-Id' => $notificationId,
            ],
        );

        return new DeliveryResult(
            success: $result->isSuccess(),
            deliveredCount: $result->isSuccess() ? 1 : 0,
            totalCount: 1,
        );
    }
}

package notification

import (
	"context"
	"fmt"
)

// DeliveryResult holds the outcome of a notification delivery attempt.
type DeliveryResult struct {
	Success        bool
	DeliveredCount int
	TotalCount     int
}

// Sender is the interface for channel-specific notification delivery.
type Sender interface {
	Send(ctx context.Context, notificationID string, payload map[string]string) (DeliveryResult, error)
}

// PushSender delivers push notifications via APNs and FCM.
type PushSender struct {
	tokens DeviceTokenRepository
	apns   APNsClient
	fcm    FCMClient
}

func NewPushSender(tokens DeviceTokenRepository, apns APNsClient, fcm FCMClient) *PushSender {
	return &PushSender{tokens: tokens, apns: apns, fcm: fcm}
}

func (s *PushSender) Send(ctx context.Context, notificationID string, payload map[string]string) (DeliveryResult, error) {
	userID := payload["user_id"]
	tokens, err := s.tokens.GetActiveTokens(ctx, userID)
	if err != nil {
		return DeliveryResult{}, fmt.Errorf("get tokens: %w", err)
	}

	delivered := 0
	for _, tok := range tokens {
		msg := PushMessage{
			Title: payload["subject"],
			Body:  payload["body"],
		}

		var sendErr error
		switch tok.Platform {
		case "ios":
			sendErr = s.apns.Send(ctx, tok.Token, msg)
		case "android":
			sendErr = s.fcm.Send(ctx, tok.Token, msg)
		default:
			continue
		}

		if sendErr != nil {
			// Deactivate invalid tokens
			s.tokens.Deactivate(ctx, tok.ID)
			continue
		}
		delivered++
	}

	return DeliveryResult{
		Success:        delivered > 0,
		DeliveredCount: delivered,
		TotalCount:     len(tokens),
	}, nil
}

// EmailSender delivers email notifications.
type EmailSender struct {
	emailClient EmailClient
	users       UserRepository
}

func NewEmailSender(emailClient EmailClient, users UserRepository) *EmailSender {
	return &EmailSender{emailClient: emailClient, users: users}
}

func (s *EmailSender) Send(ctx context.Context, notificationID string, payload map[string]string) (DeliveryResult, error) {
	user, err := s.users.Find(ctx, payload["user_id"])
	if err != nil {
		return DeliveryResult{}, fmt.Errorf("find user: %w", err)
	}

	err = s.emailClient.Send(ctx, EmailMessage{
		To:      user.Email,
		Subject: payload["subject"],
		Body:    payload["body"],
		Headers: map[string]string{"X-Notification-Id": notificationID},
	})

	success := err == nil
	count := 0
	if success {
		count = 1
	}

	return DeliveryResult{
		Success:        success,
		DeliveredCount: count,
		TotalCount:     1,
	}, err
}

namespace Notifications;

// DeliveryResult holds the outcome of a notification delivery attempt.
public readonly record struct DeliveryResult(bool Success, int DeliveredCount, int TotalCount);

// INotificationSender is the contract for channel-specific delivery.
public interface INotificationSender
{
    Task<DeliveryResult> SendAsync(
        string notificationId,
        IReadOnlyDictionary<string, string> payload,
        CancellationToken ct = default);
}

// PushNotificationSender delivers push notifications via APNs and FCM.
public sealed class PushNotificationSender(
    IDeviceTokenRepository deviceTokens,
    IApnsClient apns,
    IFcmClient fcm) : INotificationSender
{
    public async Task<DeliveryResult> SendAsync(
        string notificationId,
        IReadOnlyDictionary<string, string> payload,
        CancellationToken ct = default)
    {
        var tokens = await deviceTokens.GetActiveTokensAsync(payload["user_id"], ct);
        var message = new PushMessage(payload["subject"], payload["body"]);

        var delivered = 0;
        foreach (var token in tokens)
        {
            try
            {
                switch (token.Platform)
                {
                    case "ios":
                        await apns.SendAsync(token.Value, message, ct);
                        break;
                    case "android":
                        await fcm.SendAsync(token.Value, message, ct);
                        break;
                    default:
                        throw new ArgumentException($"Unknown platform: {token.Platform}");
                }

                delivered++;
            }
            catch (InvalidTokenException)
            {
                // Deactivate invalid token
                await deviceTokens.DeactivateAsync(token.Id, ct);
            }
        }

        return new DeliveryResult(delivered > 0, delivered, tokens.Count);
    }
}

// EmailNotificationSender delivers email notifications.
public sealed class EmailNotificationSender(IEmailClient emailClient, IUserRepository users) : INotificationSender
{
    public async Task<DeliveryResult> SendAsync(
        string notificationId,
        IReadOnlyDictionary<string, string> payload,
        CancellationToken ct = default)
    {
        var user = await users.FindAsync(payload["user_id"], ct);

        var sent = await emailClient.SendAsync(new EmailMessage(
            To: user.Email,
            Subject: payload["subject"],
            HtmlBody: payload["body"],
            Headers: new Dictionary<string, string> { ["X-Notification-Id"] = notificationId }), ct);

        return new DeliveryResult(sent, sent ? 1 : 0, 1);
    }
}

from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol

@dataclass(frozen=True, slots=True)
class DeliveryResult:
    """Outcome of a notification delivery attempt."""

    success: bool
    delivered_count: int
    total_count: int

class NotificationSender(Protocol):
    """Channel-specific delivery contract; Python uses structural typing instead of `implements`."""

    async def send(self, notification_id: str, payload: dict[str, str]) -> DeliveryResult: ...

@dataclass(slots=True)
class PushNotificationSender:
    """Delivers push notifications via APNs and FCM."""

    device_tokens: DeviceTokenRepository
    apns: ApnsClient
    fcm: FcmClient

    async def send(self, notification_id: str, payload: dict[str, str]) -> DeliveryResult:
        tokens = await self.device_tokens.get_active(payload["user_id"])
        message = {"title": payload["subject"], "body": payload["body"], "data": payload.get("data", {})}

        delivered = 0
        for token in tokens:
            client = {"ios": self.apns, "android": self.fcm}.get(token.platform)
            if client is None:
                raise ValueError(f"Unknown platform: {token.platform}")

            try:
                await client.send(token.value, message)
                delivered += 1
            except InvalidTokenError:
                # Deactivate invalid token
                await self.device_tokens.deactivate(token.id)

        return DeliveryResult(delivered > 0, delivered, len(tokens))

@dataclass(slots=True)
class EmailNotificationSender:
    """Delivers email notifications."""

    email_client: EmailClient
    users: UserRepository

    async def send(self, notification_id: str, payload: dict[str, str]) -> DeliveryResult:
        user = await self.users.find(payload["user_id"])

        sent = await self.email_client.send(
            to=user.email,
            subject=payload["subject"],
            html_body=payload["body"],
            headers={"X-Notification-Id": notification_id},
        )

        return DeliveryResult(sent, 1 if sent else 0, 1)

<?php

declare(strict_types=1);

final class NotificationWorker
{
    private const MAX_RETRIES = 3;
    private const BASE_DELAY_SECONDS = 60;

    public function __construct(
        private readonly NotificationSender $sender,
        private readonly NotificationRepository $repository,
        private readonly QueuePublisher $queue,
    ) {}

    public function process(NotificationMessage $message): void
    {
        try {
            $result = $this->sender->send(
                $message->notificationId,
                $message->payload,
            );

            if ($result->success) {
                $this->repository->markDelivered($message->notificationId);
            } else {
                $this->handleFailure($message, 'Delivery failed');
            }
        } catch (\Throwable $e) {
            $this->handleFailure($message, $e->getMessage());
        }
    }

    private function handleFailure(NotificationMessage $message, string $error): void
    {
        $notification = $this->repository->find($message->notificationId);

        if ($notification->retryCount >= self::MAX_RETRIES) {
            $this->repository->markFailed(
                $message->notificationId,
                $error,
            );
            return;
        }

        // Exponential backoff: 60s, 120s, 240s
        $delay = self::BASE_DELAY_SECONDS * (2 ** $notification->retryCount);

        $this->repository->incrementRetry($message->notificationId, $error);

        // Re-queue with delay
        $this->queue->publishWithDelay(
            queue: "notification.{$message->channel}.retry",
            message: $message,
            delaySeconds: $delay,
        );
    }
}

package notification

import (
	"context"
	"fmt"
	"math"
	"time"
)

const (
	maxRetries       = 3
	baseDelaySeconds = 60
)

// Worker processes notification messages with retry and exponential backoff.
type Worker struct {
	sender Sender
	repo   NotificationRepository
	queue  QueuePublisher
}

func NewWorker(sender Sender, repo NotificationRepository, queue QueuePublisher) *Worker {
	return &Worker{sender: sender, repo: repo, queue: queue}
}

// Process attempts to deliver a notification and handles failures.
func (w *Worker) Process(ctx context.Context, msg NotificationMessage) error {
	result, err := w.sender.Send(ctx, msg.NotificationID, msg.Payload)

	if err == nil && result.Success {
		return w.repo.MarkDelivered(ctx, msg.NotificationID)
	}

	errMsg := "delivery failed"
	if err != nil {
		errMsg = err.Error()
	}

	return w.handleFailure(ctx, msg, errMsg)
}

func (w *Worker) handleFailure(ctx context.Context, msg NotificationMessage, errMsg string) error {
	notif, err := w.repo.Find(ctx, msg.NotificationID)
	if err != nil {
		return fmt.Errorf("find notification: %w", err)
	}

	if notif.RetryCount >= maxRetries {
		return w.repo.MarkFailed(ctx, msg.NotificationID, errMsg)
	}

	// Exponential backoff: 60s, 120s, 240s
	delay := time.Duration(baseDelaySeconds*math.Pow(2, float64(notif.RetryCount))) * time.Second

	if err := w.repo.IncrementRetry(ctx, msg.NotificationID, errMsg); err != nil {
		return fmt.Errorf("increment retry: %w", err)
	}

	queue := fmt.Sprintf("notification.%s.retry", msg.Channel)
	return w.queue.PublishWithDelay(ctx, queue, msg, delay)
}

using Microsoft.Extensions.Logging;

namespace Notifications;

// NotificationWorker delivers a message and retries it with exponential backoff.
public sealed class NotificationWorker(
    INotificationSender sender,
    INotificationRepository repository,
    IQueuePublisher queue,
    ILogger<NotificationWorker> logger)
{
    private const int MaxRetries = 3;
    private static readonly TimeSpan BaseDelay = TimeSpan.FromSeconds(60);

    public async Task ProcessAsync(NotificationMessage message, CancellationToken ct = default)
    {
        try
        {
            var result = await sender.SendAsync(message.NotificationId, message.Payload, ct);

            if (result.Success)
            {
                await repository.MarkDeliveredAsync(message.NotificationId, ct);
            }
            else
            {
                await HandleFailureAsync(message, "Delivery failed", ct);
            }
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            // Never swallow shutdown cancellation, retry everything else
            logger.LogWarning(ex, "Delivery failed for {NotificationId}", message.NotificationId);
            await HandleFailureAsync(message, ex.Message, ct);
        }
    }

    private async Task HandleFailureAsync(NotificationMessage message, string error, CancellationToken ct)
    {
        var notification = await repository.FindAsync(message.NotificationId, ct);

        if (notification.RetryCount >= MaxRetries)
        {
            await repository.MarkFailedAsync(message.NotificationId, error, ct);
            return;
        }

        // Exponential backoff: 60s, 120s, 240s
        var delay = BaseDelay * Math.Pow(2, notification.RetryCount);

        await repository.IncrementRetryAsync(message.NotificationId, error, ct);

        // Re-queue with delay
        await queue.PublishWithDelayAsync($"notification.{message.Channel}.retry", message, delay, ct);
    }
}

from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass
from datetime import timedelta

MAX_RETRIES = 3
BASE_DELAY = timedelta(seconds=60)

logger = logging.getLogger(__name__)

@dataclass(slots=True)
class NotificationWorker:
    """Delivers a message and retries it with exponential backoff."""

    sender: NotificationSender
    repository: NotificationRepository
    queue: QueuePublisher

    async def process(self, message: NotificationMessage) -> None:
        try:
            result = await self.sender.send(message.notification_id, message.payload)
        except asyncio.CancelledError:
            raise  # Never swallow shutdown cancellation
        except Exception as exc:
            logger.warning("delivery failed for %s: %s", message.notification_id, exc)
            await self._handle_failure(message, str(exc))
            return

        if result.success:
            await self.repository.mark_delivered(message.notification_id)
        else:
            await self._handle_failure(message, "Delivery failed")

    async def _handle_failure(self, message: NotificationMessage, error: str) -> None:
        notification = await self.repository.find(message.notification_id)

        if notification.retry_count >= MAX_RETRIES:
            await self.repository.mark_failed(message.notification_id, error)
            return

        # Exponential backoff: 60s, 120s, 240s
        delay = BASE_DELAY * (2**notification.retry_count)

        await self.repository.increment_retry(message.notification_id, error)

        # Re-queue with delay
        await self.queue.publish_with_delay(f"notification.{message.channel}.retry", message, delay)

<?php

declare(strict_types=1);

final class DeduplicationService
{
    private const TTL = 86400; // 24 hours

    public function __construct(
        private readonly \Redis $redis,
    ) {}

    public function isUnique(string $idempotencyKey): bool
    {
        $key = "dedup:{$idempotencyKey}";

        // SET NX returns true only if key didn't exist
        $result = $this->redis->set($key, '1', ['NX', 'EX' => self::TTL]);

        return $result !== false;
    }
}

package notification

import (
	"context"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

const dedupTTL = 24 * time.Hour

// DeduplicationService prevents duplicate notification delivery.
type DeduplicationService struct {
	rdb *redis.Client
}

func NewDeduplicationService(rdb *redis.Client) *DeduplicationService {
	return &DeduplicationService{rdb: rdb}
}

// IsUnique returns true if this idempotency key has not been seen before.
func (d *DeduplicationService) IsUnique(ctx context.Context, idempotencyKey string) (bool, error) {
	key := "dedup:" + idempotencyKey

	ok, err := d.rdb.SetNX(ctx, key, "1", dedupTTL).Result()
	if err != nil {
		return false, fmt.Errorf("dedup set nx: %w", err)
	}

	return ok, nil
}

using StackExchange.Redis;

namespace Notifications;

// DeduplicationService prevents duplicate notification delivery.
public sealed class DeduplicationService(IConnectionMultiplexer redis)
{
    private static readonly TimeSpan Ttl = TimeSpan.FromHours(24);

    // IsUniqueAsync returns true only if this idempotency key has not been seen before.
    public async Task<bool> IsUniqueAsync(string idempotencyKey)
    {
        var db = redis.GetDatabase();

        // SET NX succeeds only if the key did not exist
        return await db.StringSetAsync($"dedup:{idempotencyKey}", "1", Ttl, When.NotExists);
    }
}

from __future__ import annotations

from dataclasses import dataclass

import redis.asyncio as redis

DEDUP_TTL_SECONDS = 24 * 60 * 60

@dataclass(slots=True)
class DeduplicationService:
    """Prevents duplicate notification delivery."""

    client: redis.Redis

    async def is_unique(self, idempotency_key: str) -> bool:
        # SET NX returns None when the key already exists, True when it was stored
        stored = await self.client.set(
            f"dedup:{idempotency_key}", "1", nx=True, ex=DEDUP_TTL_SECONDS
        )

        return stored is True