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

# Simplified example of device token registration flow
# This runs on your application server
 
class DeviceTokenService:
    """
    Manages the relationship between users and their device tokens.
    A single user may have multiple devices (tokens), and each device
    may have multiple apps (requiring separate token-per-app storage).
    """
    
    def register_token(
        self,
        user_id: str,
        device_token: str,
        platform: str,  # 'ios' or 'android'
        app_version: str,
        device_model: str
    ) -> None:
        """
        Registers or updates a device token for a user.
        
        Key considerations:
        1. Upsert pattern: Token may already exist (update metadata)
           or be new (insert new record)
        2. Associate token with user for targeted messaging
        3. Track platform to route through correct push service
        4. Store metadata for analytics and debugging
        """
        existing = self.db.find_token(device_token)
        
        if existing:
            # Token exists - update metadata and user association
            # This handles case where user logs into different account
            self.db.update_token(
                token=device_token,
                user_id=user_id,
                platform=platform,
                app_version=app_version,
                device_model=device_model,
                last_seen=datetime.utcnow()
            )
        else:
            # New token - insert new record
            self.db.insert_token(
                token=device_token,
                user_id=user_id,
                platform=platform,
                app_version=app_version,
                device_model=device_model,
                registered_at=datetime.utcnow(),
                last_seen=datetime.utcnow()
            )
    
    def invalidate_token(self, device_token: str) -> None:
        """
        Marks a token as invalid. Called when push service reports
        the token is no longer valid (app uninstalled, etc.)
        
        Important: Don't delete immediately - mark as invalid for
        analytics and debugging purposes. Purge old invalid tokens
        via batch job.
        """
        self.db.update_token(
            token=device_token,
            is_valid=False,
            invalidated_at=datetime.utcnow()
        )

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

# High-level architecture of a push notification system
 
class NotificationSystem:
    """
    Orchestrates the complete push notification pipeline.
    
    Architecture follows a layered approach:
    1. Event ingestion (trigger detection)
    2. Notification composition (payload creation)
    3. Dispatch orchestration (delivery coordination)
    4. Platform delivery (APNs/FCM communication)
    5. Tracking and feedback (analytics and token maintenance)
    """
    
    def __init__(self):
        # Core services
        self.token_registry = TokenRegistryService()
        self.composer = NotificationComposer()
        self.dispatcher = DispatchOrchestrator()
        
        # Platform-specific clients
        self.apns_client = APNsClient(
            key_id=config.APNS_KEY_ID,
            team_id=config.APNS_TEAM_ID,
            private_key=config.APNS_PRIVATE_KEY,
            use_sandbox=config.IS_DEVELOPMENT
        )
        self.fcm_client = FCMClient(
            credentials=config.FCM_SERVICE_ACCOUNT
        )
        
        # Tracking and analytics
        self.tracker = DeliveryTracker()
    
    async def send_notification(
        self,
        event: NotificationEvent,
        target: NotificationTarget  # Can be user_id, segment, or broadcast
    ) -> NotificationResult:
        """
        Main entry point for sending notifications.
        
        Flow:
        1. Resolve target to device tokens
        2. Compose platform-specific payloads
        3. Dispatch to appropriate platform services
        4. Track results and handle failures
        """
        # Step 1: Target Resolution
        tokens = await self.token_registry.resolve_target(target)
        
        if not tokens:
            return NotificationResult(
                status="no_tokens",
                message="No valid device tokens for target"
            )
        
        # Step 2: Group tokens by platform
        ios_tokens = [t for t in tokens if t.platform == "ios"]
        android_tokens = [t for t in tokens if t.platform == "android"]
        
        # Step 3: Compose payloads
        apns_payload = self.composer.compose_apns(event)
        fcm_payload = self.composer.compose_fcm(event)
        
        # Step 4: Dispatch in parallel
        results = await asyncio.gather(
            self._send_ios(ios_tokens, apns_payload),
            self._send_android(android_tokens, fcm_payload),
            return_exceptions=True
        )
        
        # Step 5: Process results
        return self._aggregate_results(results, tokens)
    
    async def _send_ios(
        self,
        tokens: List[DeviceToken],
        payload: APNsPayload
    ) -> Dict[str, SendResult]:
        """
        Send notifications to iOS devices via APNs.
        
        Important considerations:
        - APNs uses HTTP/2, allowing multiplexed requests
        - Each token requires a separate request
        - Rate limit is ~100k notifications per second
        - Connection should be kept alive and reused
        """
        results = {}
        
        for token in tokens:
            try:
                response = await self.apns_client.send(
                    device_token=token.token,
                    payload=payload,
                    topic=config.APNS_BUNDLE_ID,  # App's bundle ID
                    priority=10 if payload.is_urgent else 5,
                    expiration=self._compute_expiration(payload)
                )
                
                results[token.token] = SendResult(
                    status="success",
                    apns_id=response.apns_id
                )
                
            except TokenInvalidError:
                # Token no longer valid - mark for removal
                results[token.token] = SendResult(status="invalid_token")
                await self.token_registry.invalidate_token(token.token)
                
            except APNsError as e:
                results[token.token] = SendResult(
                    status="error",
                    error=str(e)
                )
        
        return results

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

# Token lifecycle management implementation
 
class TokenLifecycleManager:
    """
    Manages the complete lifecycle of device tokens,
    including cleanup of stale tokens and metric tracking.
    """
    
    # Thresholds for token health assessment
    STALE_THRESHOLD_DAYS = 90  # Mark tokens as stale after 90 days
    CLEANUP_THRESHOLD_DAYS = 180  # Remove invalid tokens after 180 days
    
    def __init__(self, db, metrics):
        self.db = db
        self.metrics = metrics
    
    async def handle_push_failure(
        self,
        token: str,
        error_code: str,
        platform: str
    ) -> None:
        """
        Process push failure feedback from platform services.
        
        Critical to handle this promptly - continuing to send to
        invalid tokens wastes resources and can affect your
        reputation with the push service.
        """
        # Definitive invalid token errors
        INVALID_TOKEN_CODES = {
            'apns': ['BadDeviceToken', 'Unregistered', 'DeviceTokenNotForTopic'],
            'fcm': ['UNREGISTERED', 'INVALID_ARGUMENT']
        }
        
        if error_code in INVALID_TOKEN_CODES.get(platform, []):
            # Immediate invalidation
            await self.db.execute(
                """
                UPDATE device_tokens 
                SET is_valid = false,
                    invalidated_at = NOW(),
                    invalidation_reason = :reason
                WHERE token = :token
                """,
                {'token': token, 'reason': error_code}
            )
            self.metrics.increment('token.invalidated', tags={'reason': error_code})
        
        elif self._is_retriable_error(error_code):
            # Transient error - track but don't invalidate
            await self.db.execute(
                """
                UPDATE device_tokens 
                SET consecutive_failures = consecutive_failures + 1,
                    last_failure = NOW(),
                    last_failure_reason = :reason
                WHERE token = :token
                """,
                {'token': token, 'reason': error_code}
            )
            self.metrics.increment('token.transient_failure', tags={'reason': error_code})
    
    async def run_cleanup_job(self) -> Dict[str, int]:
        """
        Periodic job to clean up stale and invalid tokens.
        
        Run this as a scheduled job (e.g., daily at low-traffic hours).
        Uses batch processing to avoid long-running transactions.
        """
        results = {'stale_marked': 0, 'invalid_removed': 0}
        
        # Step 1: Mark stale tokens (not seen recently, but not yet invalid)
        stale_cutoff = datetime.utcnow() - timedelta(days=self.STALE_THRESHOLD_DAYS)
        
        result = await self.db.execute(
            """
            UPDATE device_tokens
            SET is_stale = true, stale_marked_at = NOW()
            WHERE last_seen < :cutoff
              AND is_valid = true
              AND is_stale = false
            """,
            {'cutoff': stale_cutoff}
        )
        results['stale_marked'] = result.rowcount
        
        # Step 2: Archive and remove old invalid tokens
        cleanup_cutoff = datetime.utcnow() - timedelta(days=self.CLEANUP_THRESHOLD_DAYS)
        
        # Archive to cold storage first
        await self.db.execute(
            """
            INSERT INTO device_tokens_archive
            SELECT * FROM device_tokens
            WHERE is_valid = false
              AND invalidated_at < :cutoff
            """,
            {'cutoff': cleanup_cutoff}
        )
        
        # Then delete from hot storage
        result = await self.db.execute(
            """
            DELETE FROM device_tokens
            WHERE is_valid = false
              AND invalidated_at < :cutoff
            """,
            {'cutoff': cleanup_cutoff}
        )
        results['invalid_removed'] = result.rowcount
        
        # Emit metrics
        self.metrics.gauge('token.stale_marked', results['stale_marked'])
        self.metrics.gauge('token.cleanup_removed', results['invalid_removed'])
        
        return results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

{
    "aps": {
        "alert": {
            "title": "New Message",
            "subtitle": "From John Doe",
            "body": "Hey, are you available for a quick call?"
        },
        "badge": 5,
        "sound": "default",
        "category": "MESSAGE_CATEGORY",
        "mutable-content": 1,
        "content-available": 1,
        "thread-id": "conversation-456"
    },
    "custom_data": {
        "conversation_id": "abc123",
        "sender_id": "user_789",
        "message_type": "text"
    }
}

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

{
    "message": {
        "token": "device_token_here",
        "notification": {
            "title": "New Message",
            "body": "Hey, are you available for a quick call?"
        },
        "android": {
            "priority": "high",
            "notification": {
                "icon": "message_icon",
                "color": "#4A90D9",
                "channel_id": "messages",
                "click_action": "OPEN_CONVERSATION",
                "tag": "conversation-456"
            },
            "data": {
                "conversation_id": "abc123",
                "sender_id": "user_789",
                "message_type": "text"
            },
            "ttl": "86400s"
        }
    }
}

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

import jwt
import time
 
class APNsTokenManager:
    """
    Manages JWT token generation and refresh for APNs authentication.
    
    Key points:
    - Tokens are valid for 1 hour
    - Refresh proactively before expiration
    - Sign with ES256 algorithm using your APNs auth key
    """
    
    TOKEN_LIFETIME = 3600  # 1 hour
    REFRESH_BUFFER = 300   # Refresh 5 minutes before expiration
    
    def __init__(self, key_id: str, team_id: str, private_key: str):
        self.key_id = key_id
        self.team_id = team_id
        self.private_key = private_key
        self._cached_token = None
        self._token_expiry = 0
    
    def get_token(self) -> str:
        """
        Returns a valid JWT token, generating a new one if necessary.
        Thread-safe implementation for concurrent access.
        """
        current_time = time.time()
        
        # Check if cached token is still valid (with buffer)
        if (self._cached_token and 
            current_time < self._token_expiry - self.REFRESH_BUFFER):
            return self._cached_token
        
        # Generate new token
        self._cached_token = self._generate_token(current_time)
        self._token_expiry = current_time + self.TOKEN_LIFETIME
        
        return self._cached_token
    
    def _generate_token(self, issued_at: float) -> str:
        """
        Generates a new JWT for APNs authentication.
        
        Token structure:
        - Header: algorithm (ES256) and key ID
        - Payload: issuer (team ID) and issued-at timestamp
        """
        headers = {
            'alg': 'ES256',
            'kid': self.key_id,
        }
        
        payload = {
            'iss': self.team_id,
            'iat': int(issued_at),
        }
        
        return jwt.encode(
            payload,
            self.private_key,
            algorithm='ES256',
            headers=headers
        )

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

# Queue-based notification architecture
 
class NotificationRequestHandler:
    """
    Handles notification requests by enqueueing to message queue.
    Actual sending is done by separate worker processes.
    """
    
    def __init__(self, queue_client):
        self.queue = queue_client
    
    async def request_notification(
        self,
        event: NotificationEvent,
        target: NotificationTarget,
        priority: str = "normal"
    ) -> str:
        """
        Enqueues a notification request for asynchronous processing.
        
        Returns a request_id that can be used to track the notification.
        """
        request_id = str(uuid.uuid4())
        
        message = NotificationMessage(
            request_id=request_id,
            event=event,
            target=target,
            priority=priority,
            created_at=datetime.utcnow(),
            attempt=0
        )
        
        # Route to appropriate queue based on priority
        queue_name = f"notifications-{priority}"
        
        await self.queue.send(
            queue=queue_name,
            message=message.to_json(),
            message_id=request_id
        )
        
        return request_id
 
 
class NotificationWorker:
    """
    Worker process that consumes from notification queue
    and sends to platform push services.
    """
    
    def __init__(self, queue_client, push_service):
        self.queue = queue_client
        self.push = push_service
    
    async def run(self):
        """Main worker loop - consume and process messages."""
        while True:
            messages = await self.queue.receive(
                queue="notifications-high",  # Process high priority first
                max_messages=10,
                wait_time=20
            )
            
            if not messages:
                messages = await self.queue.receive(
                    queue="notifications-normal",
                    max_messages=10,
                    wait_time=20
                )
            
            for msg in messages:
                await self._process_message(msg)
    
    async def _process_message(self, message):
        try:
            notification = NotificationMessage.from_json(message.body)
            result = await self.push.send_notification(
                notification.event,
                notification.target
            )
            
            if result.should_retry:
                # Re-enqueue with incremented attempt count
                await self._retry_message(notification)
            else:
                await self.queue.delete(message)
                
        except Exception as e:
            # Log error and potentially DLQ the message
            logger.error(f"Failed to process: {e}")
            await self._handle_failure(message, notification)