1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

# Comprehensive delivery tracking implementation
 
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
import uuid
 
class DeliveryStatus(Enum):
    # Server-side states
    PENDING = "pending"           # Queued, not yet sent
    SENDING = "sending"           # Currently being sent
    SENT = "sent"                 # Accepted by push service
    FAILED_PERMANENT = "failed_permanent"  # Won't retry
    FAILED_TRANSIENT = "failed_transient"  # Will retry
    
    # Delivery states (where trackable)
    DELIVERED = "delivered"       # Confirmed delivered (FCM only)
    DISPLAYED = "displayed"       # Confirmed displayed
    CLICKED = "clicked"           # User clicked notification
    DISMISSED = "dismissed"       # User dismissed notification
    
    # Terminal failure states
    EXPIRED = "expired"           # TTL exceeded before delivery
    INVALID_TOKEN = "invalid_token"  # Token no longer valid
    UNSUBSCRIBED = "unsubscribed" # User unsubscribed
 
@dataclass
class NotificationRecord:
    """
    Complete record of a notification's lifecycle.
    Used for tracking, debugging, and analytics.
    """
    id: str
    user_id: str
    device_token: str
    platform: str  # 'ios', 'android', 'web'
    
    # Content
    title: str
    body: str
    payload_hash: str  # For deduplication
    
    # State tracking
    status: DeliveryStatus
    created_at: datetime
    sent_at: datetime | None = None
    delivered_at: datetime | None = None
    clicked_at: datetime | None = None
    
    # Platform responses
    platform_message_id: str | None = None  # APNs ID, FCM message_id, etc.
    error_code: str | None = None
    error_message: str | None = None
    
    # Retry tracking
    attempt_count: int = 0
    next_retry_at: datetime | None = None
 
 
class DeliveryTracker:
    """
    Tracks notification delivery through the entire pipeline.
    
    Provides:
    - Per-notification lifecycle tracking
    - Aggregated delivery metrics
    - Failure analysis
    """
    
    def __init__(self, db, metrics):
        self.db = db
        self.metrics = metrics
    
    async def record_send_attempt(
        self,
        notification_id: str,
        platform_response: dict
    ) -> None:
        """Record the result of a send attempt."""
        
        if platform_response.get('success'):
            status = DeliveryStatus.SENT
            
            await self.db.update_notification(
                notification_id,
                status=status,
                sent_at=datetime.utcnow(),
                platform_message_id=platform_response.get('message_id'),
                attempt_count=notification_record.attempt_count + 1
            )
            
            self.metrics.increment('notification.sent', tags={
                'platform': notification_record.platform
            })
            
        else:
            error_code = platform_response.get('error_code')
            
            if self._is_permanent_failure(error_code):
                status = DeliveryStatus.FAILED_PERMANENT
                
                # Handle token invalidation
                if error_code in ('UNREGISTERED', 'BadDeviceToken', '410'):
                    status = DeliveryStatus.INVALID_TOKEN
                    await self._handle_invalid_token(
                        notification_record.device_token
                    )
            else:
                status = DeliveryStatus.FAILED_TRANSIENT
            
            await self.db.update_notification(
                notification_id,
                status=status,
                error_code=error_code,
                error_message=platform_response.get('error_message'),
                attempt_count=notification_record.attempt_count + 1,
                next_retry_at=self._compute_retry_time(
                    notification_record.attempt_count
                ) if status == DeliveryStatus.FAILED_TRANSIENT else None
            )
            
            self.metrics.increment('notification.failed', tags={
                'platform': notification_record.platform,
                'error': error_code,
                'permanent': status == DeliveryStatus.FAILED_PERMANENT
            })
    
    async def record_delivery_event(
        self,
        platform_message_id: str,
        event_type: str,  # 'delivered', 'displayed', 'clicked', 'dismissed'
        timestamp: datetime
    ) -> None:
        """
        Record delivery/engagement events from client or platform.
        
        For FCM: Uses Firebase Analytics delivery receipts
        For client confirmation: App reports receipt to your server
        """
        notification = await self.db.find_by_platform_id(platform_message_id)
        
        if not notification:
            return  # Unknown notification, may be older than retention
        
        update_fields = {}
        
        if event_type == 'delivered':
            update_fields['status'] = DeliveryStatus.DELIVERED
            update_fields['delivered_at'] = timestamp
            
        elif event_type == 'clicked':
            update_fields['status'] = DeliveryStatus.CLICKED
            update_fields['clicked_at'] = timestamp
            
        elif event_type == 'dismissed':
            update_fields['status'] = DeliveryStatus.DISMISSED
        
        await self.db.update_notification(notification.id, **update_fields)
        
        self.metrics.increment(f'notification.{event_type}', tags={
            'platform': notification.platform
        })
    
    def _is_permanent_failure(self, error_code: str) -> bool:
        """Determine if error indicates permanent failure."""
        permanent_codes = {
            # APNs
            'BadDeviceToken', 'Unregistered', 'DeviceTokenNotForTopic',
            'InvalidProviderToken', 'Forbidden',
            # FCM
            'UNREGISTERED', 'INVALID_ARGUMENT', 'SENDER_ID_MISMATCH',
            # Web Push
            '404', '410',
        }
        return error_code in permanent_codes
    
    def _compute_retry_time(self, attempt_count: int) -> datetime:
        """Exponential backoff for retry timing."""
        # 1min, 2min, 4min, 8min, max 30min
        delay_seconds = min(60 * (2 ** attempt_count), 1800)
        return datetime.utcnow() + timedelta(seconds=delay_seconds)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

// iOS: Report notification delivery to backend
// Add to your UNUserNotificationCenterDelegate
 
import UserNotifications
 
extension AppDelegate: UNUserNotificationCenterDelegate {
    
    // Called when notification is delivered while app is in foreground
    func userNotificationCenter(
        _ center: UNUserNotificationCenter,
        willPresent notification: UNNotification,
        withCompletionHandler completionHandler: @escaping (UNNotificationPresentationOptions) -> Void
    ) {
        // Report delivery
        reportNotificationDelivered(notification.request.identifier)
        
        // Show notification
        completionHandler([.banner, .sound, .badge])
    }
    
    // Called when user interacts with notification
    func userNotificationCenter(
        _ center: UNUserNotificationCenter,
        didReceive response: UNNotificationResponse,
        withCompletionHandler completionHandler: @escaping () -> Void
    ) {
        let notificationId = response.notification.request.identifier
        
        switch response.actionIdentifier {
        case UNNotificationDefaultActionIdentifier:
            // User tapped notification
            reportNotificationClicked(notificationId)
        case UNNotificationDismissActionIdentifier:
            // User dismissed notification
            reportNotificationDismissed(notificationId)
        default:
            // Custom action
            reportNotificationAction(notificationId, action: response.actionIdentifier)
        }
        
        completionHandler()
    }
    
    private func reportNotificationDelivered(_ notificationId: String) {
        // Fire-and-forget HTTP request to your backend
        let url = URL(string: "https://api.yourapp.com/notifications/delivered")!
        var request = URLRequest(url: url)
        request.httpMethod = "POST"
        request.httpBody = try? JSONEncoder().encode([
            "notification_id": notificationId,
            "timestamp": ISO8601DateFormatter().string(from: Date()),
            "device_id": getDeviceId()
        ])
        request.setValue("application/json", forHTTPHeaderField: "Content-Type")
        
        URLSession.shared.dataTask(with: request).resume()
    }
    
    private func reportNotificationClicked(_ notificationId: String) {
        // Similar to above, with event type "clicked"
    }
}
 
// For background notifications using Notification Service Extension:
class NotificationService: UNNotificationServiceExtension {
    
    override func didReceive(
        _ request: UNNotificationRequest,
        withContentHandler contentHandler: @escaping (UNNotificationContent) -> Void
    ) {
        // Report delivery before modifying content
        reportBackgroundDelivery(request.identifier)
        
        // Continue with content modification...
        contentHandler(request.content)
    }
    
    private func reportBackgroundDelivery(_ notificationId: String) {
        // Note: Limited network access in extension
        // Use background URL session for reliability
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

// Android: Report notification delivery via FirebaseMessagingService
 
class MyFirebaseMessagingService : FirebaseMessagingService() {
    
    override fun onMessageReceived(remoteMessage: RemoteMessage) {
        // Report delivery immediately
        val notificationId = remoteMessage.data["notification_id"]
        if (notificationId != null) {
            reportNotificationDelivered(notificationId)
        }
        
        // Process and display notification
        if (remoteMessage.notification != null) {
            // Display notification (system will handle if app in background)
            showNotification(remoteMessage)
        } else {
            // Data-only message, create notification manually
            showDataNotification(remoteMessage.data)
        }
    }
    
    private fun reportNotificationDelivered(notificationId: String) {
        // Use WorkManager for reliable delivery (survives process death)
        val data = workDataOf(
            "notification_id" to notificationId,
            "event_type" to "delivered",
            "timestamp" to System.currentTimeMillis()
        )
        
        val request = OneTimeWorkRequestBuilder<DeliveryReportWorker>()
            .setInputData(data)
            .setConstraints(Constraints.Builder()
                .setRequiredNetworkType(NetworkType.CONNECTED)
                .build())
            .build()
        
        WorkManager.getInstance(this).enqueue(request)
    }
}
 
// Worker for reliable delivery reporting
class DeliveryReportWorker(
    context: Context,
    params: WorkerParameters
) : CoroutineWorker(context, params) {
    
    override suspend fun doWork(): Result {
        val notificationId = inputData.getString("notification_id") ?: return Result.failure()
        val eventType = inputData.getString("event_type") ?: return Result.failure()
        val timestamp = inputData.getLong("timestamp", System.currentTimeMillis())
        
        return try {
            // Send to your backend
            val api = RetrofitClient.notificationApi
            api.reportDelivery(
                NotificationEvent(
                    notificationId = notificationId,
                    eventType = eventType,
                    timestamp = timestamp,
                    deviceId = getDeviceId()
                )
            )
            Result.success()
        } catch (e: Exception) {
            if (runAttemptCount < 3) {
                Result.retry()
            } else {
                Result.failure()
            }
        }
    }
}
 
// Track notification clicks via BroadcastReceiver or Activity handling
class NotificationClickReceiver : BroadcastReceiver() {
    
    override fun onReceive(context: Context, intent: Intent) {
        val notificationId = intent.getStringExtra("notification_id")
        val action = intent.action
        
        notificationId?.let {
            reportNotificationEvent(context, it, action ?: "clicked")
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

# Diagnostic queries and health checks for push delivery
 
class PushDeliveryDiagnostics:
    """
    Comprehensive diagnostics for push notification delivery issues.
    """
    
    def __init__(self, db, push_client, metrics):
        self.db = db
        self.push = push_client
        self.metrics = metrics
    
    async def diagnose_user_delivery(self, user_id: str) -> DiagnosticReport:
        """
        Complete diagnostic for a specific user's notification delivery.
        """
        report = DiagnosticReport(user_id=user_id)
        
        # 1. Check token status
        tokens = await self.db.get_user_tokens(user_id)
        
        if not tokens:
            report.add_issue(
                severity="critical",
                category="tokens",
                message="No device tokens registered for user"
            )
            return report
        
        for token in tokens:
            # Check token freshness
            if token.last_seen < datetime.utcnow() - timedelta(days=30):
                report.add_issue(
                    severity="warning",
                    category="tokens",
                    message=f"Stale token (last seen: {token.last_seen})",
                    token_id=token.id
                )
            
            if not token.is_valid:
                report.add_issue(
                    severity="critical",
                    category="tokens",
                    message=f"Token marked invalid: {token.invalidation_reason}",
                    token_id=token.id
                )
        
        # 2. Check recent notification history
        recent_notifications = await self.db.get_recent_notifications(
            user_id=user_id,
            limit=20
        )
        
        failure_rate = sum(
            1 for n in recent_notifications 
            if n.status in (DeliveryStatus.FAILED_PERMANENT, DeliveryStatus.EXPIRED)
        ) / len(recent_notifications) if recent_notifications else 0
        
        if failure_rate > 0.3:
            report.add_issue(
                severity="warning",
                category="delivery",
                message=f"High failure rate: {failure_rate:.0%}",
                details=[n.to_dict() for n in recent_notifications[:5]]
            )
        
        # 3. Check common error patterns
        error_counts = await self.db.get_error_counts(
            user_id=user_id,
            since=datetime.utcnow() - timedelta(days=7)
        )
        
        if error_counts.get('UNREGISTERED', 0) > 0:
            report.add_issue(
                severity="critical",
                category="tokens",
                message="Token reported as unregistered by push service",
                recommendation="User may have uninstalled app"
            )
        
        # 4. Verify current token is valid with push service
        for token in tokens:
            if token.is_valid:
                validation = await self._validate_token_with_service(token)
                if not validation.valid:
                    report.add_issue(
                        severity="critical",
                        category="tokens",
                        message=f"Token validation failed: {validation.error}",
                        token_id=token.id
                    )
        
        return report
    
    async def get_delivery_health_metrics(self) -> HealthMetrics:
        """
        System-wide delivery health assessment.
        """
        window = datetime.utcnow() - timedelta(hours=1)
        
        metrics = await self.db.get_aggregated_metrics(since=window)
        
        return HealthMetrics(
            total_sent=metrics['total_sent'],
            success_rate=metrics['success_count'] / metrics['total_sent'],
            
            # Breakdown by platform
            ios_success_rate=metrics['ios_success'] / metrics['ios_total'],
            android_success_rate=metrics['android_success'] / metrics['android_total'],
            web_success_rate=metrics['web_success'] / metrics['web_total'],
            
            # Error analysis
            top_errors=metrics['error_breakdown'],
            invalid_token_rate=metrics['invalid_tokens'] / metrics['total_sent'],
            
            # Latency
            p50_latency_ms=metrics['p50_send_latency'],
            p99_latency_ms=metrics['p99_send_latency'],
        )
    
    def check_common_issues(self, metrics: HealthMetrics) -> list[Issue]:
        """Automated detection of common delivery problems."""
        issues = []
        
        # High invalid token rate
        if metrics.invalid_token_rate > 0.05:
            issues.append(Issue(
                severity="warning",
                title="High invalid token rate",
                message=f"{metrics.invalid_token_rate:.1%} of sends to invalid tokens",
                recommendation="Review token cleanup job; ensure app sends fresh tokens"
            ))
        
        # Platform-specific problems
        if metrics.ios_success_rate < 0.8:
            issues.append(Issue(
                severity="critical",
                title="iOS delivery degradation",
                message=f"iOS success rate: {metrics.ios_success_rate:.1%}",
                recommendation="Check APNs credentials, entitlements, certificate expiry"
            ))
        
        # High latency
        if metrics.p99_latency_ms > 5000:
            issues.append(Issue(
                severity="warning",
                title="High send latency",
                message=f"P99 latency: {metrics.p99_latency_ms}ms",
                recommendation="Check connection pooling, rate limiting, or service issues"
            ))
        
        return issues

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

# Comprehensive metrics collection for push notifications
 
from prometheus_client import Counter, Histogram, Gauge
import time
 
# Define metrics
notifications_sent = Counter(
    'push_notifications_sent_total',
    'Total notifications sent',
    ['platform', 'priority', 'status']
)
 
notification_latency = Histogram(
    'push_notification_send_latency_seconds',
    'Time to send notification to push service',
    ['platform'],
    buckets=[.01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10]
)
 
queue_depth = Gauge(
    'push_notification_queue_depth',
    'Number of notifications waiting to send',
    ['priority']
)
 
token_operations = Counter(
    'push_token_operations_total',
    'Token registry operations',
    ['operation', 'result']  # operation: register, invalidate; result: success, failure
)
 
delivery_funnel = Counter(
    'push_notification_funnel_total',
    'Notification funnel stages',
    ['stage', 'platform']  # stage: sent, delivered, displayed, clicked, dismissed
)
 
 
class MetricsMiddleware:
    """
    Wraps push sending with metrics collection.
    """
    
    def __init__(self, push_service, metrics_backend):
        self.push = push_service
        self.metrics = metrics_backend
    
    async def send(
        self,
        token: str,
        platform: str,
        notification: dict,
        priority: str = 'normal'
    ) -> SendResult:
        """Send with full metrics instrumentation."""
        
        start_time = time.time()
        
        try:
            result = await self.push.send(token, notification)
            
            # Record success
            notifications_sent.labels(
                platform=platform,
                priority=priority,
                status='success'
            ).inc()
            
            delivery_funnel.labels(
                stage='sent',
                platform=platform
            ).inc()
            
        except InvalidTokenError:
            notifications_sent.labels(
                platform=platform,
                priority=priority,
                status='invalid_token'
            ).inc()
            
            token_operations.labels(
                operation='invalidate',
                result='success'  # The invalidation itself succeeded
            ).inc()
            
            raise
            
        except Exception as e:
            notifications_sent.labels(
                platform=platform,
                priority=priority,
                status='error'
            ).inc()
            raise
            
        finally:
            # Always record latency
            duration = time.time() - start_time
            notification_latency.labels(platform=platform).observe(duration)
        
        return result
    
    def record_delivery_event(
        self,
        platform: str,
        event: str  # 'delivered', 'displayed', 'clicked', 'dismissed'
    ):
        """Record downstream delivery events."""
        delivery_funnel.labels(
            stage=event,
            platform=platform
        ).inc()
 
 
# Dashboard queries (Prometheus/Grafana)
"""
# Send success rate (last 5 min)
sum(rate(push_notifications_sent_total{status="success"}[5m])) 
/ 
sum(rate(push_notifications_sent_total[5m])) * 100
 
# Click-through rate
sum(rate(push_notification_funnel_total{stage="clicked"}[1h]))
/
sum(rate(push_notification_funnel_total{stage="delivered"}[1h])) * 100
 
# Invalid token rate
sum(rate(push_notifications_sent_total{status="invalid_token"}[5m]))
/
sum(rate(push_notifications_sent_total[5m])) * 100
 
# P99 latency
histogram_quantile(0.99, 
  sum(rate(push_notification_send_latency_seconds_bucket[5m])) by (le, platform)
)
"""