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

# APNs HTTP/2 Request Structure
 
:method = POST
:path = /3/device/{device_token}
:scheme = https
authorization = bearer {jwt_token}          # Token-based auth
apns-topic = com.yourcompany.app            # Bundle ID
apns-push-type = alert                      # alert, background, voip, complication, fileprovider, mdm
apns-priority = 10                          # 10 (immediate) or 5 (throttleable)
apns-expiration = 0                         # 0 = immediate delivery only, or Unix timestamp
apns-collapse-id = collapse-123             # Optional: replaces previous notification with same ID
 
# Request Body (JSON payload)
{
    "aps": {
        "alert": {
            "title": "Message Title",
            "body": "Message body text"
        },
        "sound": "default",
        "badge": 1
    },
    "customData": "your-custom-payload"
}

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

import jwt
import time
from pathlib import Path
 
class APNsAuthenticator:
    """
    Manages JWT generation for APNs token-based authentication.
    
    Key components:
    - Key ID: 10-character identifier from Apple Developer Portal
    - Team ID: 10-character Apple Developer Team identifier
    - Private Key: The .p8 file contents (ES256 private key)
    """
    
    ALGORITHM = "ES256"
    TOKEN_LIFETIME_SECONDS = 3600  # 1 hour maximum
    
    def __init__(
        self,
        key_id: str,
        team_id: str,
        private_key_path: str
    ):
        self.key_id = key_id
        self.team_id = team_id
        self.private_key = Path(private_key_path).read_text()
        
        # Token caching
        self._cached_token: str | None = None
        self._token_generation_time: float = 0
    
    def get_authorization_header(self) -> str:
        """
        Returns the Bearer token for APNs authentication.
        Caches token and refreshes before expiration.
        """
        current_time = time.time()
        
        # Refresh if token doesn't exist or is within 5 minutes of expiry
        if (
            not self._cached_token or 
            current_time - self._token_generation_time > self.TOKEN_LIFETIME_SECONDS - 300
        ):
            self._cached_token = self._generate_token()
            self._token_generation_time = current_time
        
        return f"bearer {self._cached_token}"
    
    def _generate_token(self) -> str:
        """
        Generates a new JWT for APNs authentication.
        
        Token structure (RFC 7519):
        - Header: algorithm and key ID
        - Payload: issuer (team ID) and issued-at timestamp
        - Signature: ES256 signature using private key
        """
        headers = {
            "alg": self.ALGORITHM,
            "kid": self.key_id,
        }
        
        payload = {
            "iss": self.team_id,
            "iat": int(time.time()),
        }
        
        token = jwt.encode(
            payload,
            self.private_key,
            algorithm=self.ALGORITHM,
            headers=headers
        )
        
        return token
    
    def force_refresh(self) -> None:
        """
        Forces token regeneration on next request.
        Useful after receiving 403 InvalidProviderToken errors.
        """
        self._cached_token = None
        self._token_generation_time = 0
 
 
# Production usage
authenticator = APNsAuthenticator(
    key_id="ABC123DEFG",
    team_id="TEAM123456",
    private_key_path="/secure/keys/AuthKey_ABC123DEFG.p8"
)
 
# Get header for each request
auth_header = authenticator.get_authorization_header()

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

# Comprehensive APNs error handling implementation
 
class APNsErrorHandler:
    """
    Handles APNs error responses with appropriate recovery actions.
    
    Error categories:
    1. Token-related: Invalidate and stop sending
    2. Payload-related: Fix and retry
    3. Transient: Retry with backoff
    4. Authentication: Refresh credentials
    """
    
    # Errors that indicate permanent token invalidity
    TOKEN_INVALID_REASONS = {
        'BadDeviceToken',        # Token format is invalid
        'Unregistered',          # App was uninstalled
        'DeviceTokenNotForTopic', # Token doesn't match the request topic
        'ExpiredToken',          # Token has expired (rare)
    }
    
    # Errors that require payload fixes
    PAYLOAD_ERRORS = {
        'PayloadEmpty',          # Payload is empty
        'PayloadTooLarge',       # Exceeds 4KB limit
        'BadPayload',            # Invalid JSON
        'BadMessageId',          # Invalid apns-id header
        'BadCollapseId',         # Invalid collapse ID
    }
    
    # Errors that should trigger retry
    TRANSIENT_ERRORS = {
        'TooManyRequests',       # Rate limited
        'InternalServerError',   # APNs internal error
        'ServiceUnavailable',    # APNs temporarily unavailable
        'Shutdown',              # Server is shutting down
    }
    
    # Authentication errors
    AUTH_ERRORS = {
        'InvalidProviderToken',  # JWT is invalid or expired
        'ExpiredProviderToken',  # JWT expired
        'MissingProviderToken',  # No JWT provided
        'BadCertificate',        # Certificate issue
        'BadCertificateEnvironment', # Cert doesn't match environment
        'Forbidden',             # Not authorized for this topic
    }
    
    def __init__(self, token_service, metrics, auth_manager):
        self.token_service = token_service
        self.metrics = metrics
        self.auth = auth_manager
    
    async def handle_error(
        self,
        device_token: str,
        status_code: int,
        reason: str | None,
        response_body: dict | None
    ) -> ErrorHandleResult:
        """
        Processes APNs error and returns appropriate action.
        
        Returns:
            ErrorHandleResult with:
                - should_retry: Whether to retry this notification
                - retry_after: Seconds to wait before retry (if applicable)
                - should_invalidate_token: Whether to remove this token
                - action: Human-readable description of action taken
        """
        self.metrics.increment(
            'apns.error',
            tags={'status': status_code, 'reason': reason}
        )
        
        # Handle by category
        if reason in self.TOKEN_INVALID_REASONS:
            return await self._handle_invalid_token(device_token, reason)
        
        if reason in self.PAYLOAD_ERRORS:
            return self._handle_payload_error(reason)
        
        if reason in self.AUTH_ERRORS:
            return self._handle_auth_error(reason)
        
        if reason in self.TRANSIENT_ERRORS or status_code >= 500:
            return self._handle_transient_error(reason, response_body)
        
        # Unknown error - log and don't retry
        logger.error(f"Unknown APNs error: {status_code} - {reason}")
        return ErrorHandleResult(
            should_retry=False,
            should_invalidate_token=False,
            action=f"Unknown error: {reason}"
        )
    
    async def _handle_invalid_token(
        self,
        device_token: str,
        reason: str
    ) -> ErrorHandleResult:
        """
        Handle permanently invalid tokens.
        These should be removed from the database immediately.
        """
        await self.token_service.invalidate_token(
            token=device_token,
            reason=reason
        )
        
        return ErrorHandleResult(
            should_retry=False,
            should_invalidate_token=True,
            action=f"Token invalidated: {reason}"
        )
    
    def _handle_transient_error(
        self,
        reason: str,
        response_body: dict | None
    ) -> ErrorHandleResult:
        """
        Handle transient errors with retry.
        
        Uses exponential backoff with jitter.
        Respects Retry-After header if present.
        """
        if reason == 'TooManyRequests':
            # Rate limited - check for Retry-After hint
            retry_after = 60  # Default 1 minute
            
            return ErrorHandleResult(
                should_retry=True,
                retry_after=retry_after,
                should_invalidate_token=False,
                action="Rate limited, will retry"
            )
        
        # Server errors - retry with backoff
        return ErrorHandleResult(
            should_retry=True,
            retry_after=5,  # Start with 5 seconds
            should_invalidate_token=False,
            action=f"Transient error: {reason}, will retry"
        )
    
    def _handle_auth_error(self, reason: str) -> ErrorHandleResult:
        """
        Handle authentication errors.
        
        Typically requires refreshing the JWT token.
        """
        if reason in ('InvalidProviderToken', 'ExpiredProviderToken'):
            self.auth.force_refresh()
            
            return ErrorHandleResult(
                should_retry=True,
                retry_after=1,
                should_invalidate_token=False,
                action="Token refreshed, will retry"
            )
        
        # Other auth errors may require configuration fixes
        return ErrorHandleResult(
            should_retry=False,
            should_invalidate_token=False,
            action=f"Auth error requires manual intervention: {reason}"
        )

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

// FCM HTTP v1 API Request Structure
POST https://fcm.googleapis.com/v1/projects/{project_id}/messages:send
Authorization: Bearer {oauth2_access_token}
Content-Type: application/json
 
{
    "message": {
        "token": "device_registration_token",
        
        // Notification payload (displayed by system)
        "notification": {
            "title": "Message Title",
            "body": "Message body text",
            "image": "https://example.com/notification-image.png"
        },
        
        // Data payload (passed to app)
        "data": {
            "conversation_id": "abc123",
            "message_type": "chat"
        },
        
        // Android-specific configuration
        "android": {
            "priority": "high",
            "ttl": "86400s",
            "collapse_key": "new_messages",
            "notification": {
                "icon": "notification_icon",
                "color": "#4285F4",
                "channel_id": "messages",
                "click_action": "OPEN_CHAT"
            }
        },
        
        // Web push configuration
        "webpush": {
            "headers": {
                "Urgency": "high"
            },
            "notification": {
                "icon": "https://example.com/icon.png"
            }
        }
    }
}

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

from google.oauth2 import service_account
from google.auth.transport.requests import Request
import google.auth
 
class FCMAuthenticator:
    """
    Manages OAuth 2.0 access token generation for FCM.
    
    Uses Google's official auth libraries which handle:
    - Token generation from service account
    - Automatic token refresh before expiration
    - Caching to minimize token generation calls
    """
    
    # FCM requires this specific scope
    FCM_SCOPE = 'https://www.googleapis.com/auth/firebase.messaging'
    
    def __init__(self, service_account_path: str):
        """
        Initialize with path to service account JSON key file.
        
        The key file contains:
        - project_id: Your Firebase project ID
        - private_key: RSA private key for signing
        - client_email: Service account email
        """
        self.credentials = service_account.Credentials.from_service_account_file(
            service_account_path,
            scopes=[self.FCM_SCOPE]
        )
        self._project_id = self._extract_project_id(service_account_path)
    
    def _extract_project_id(self, path: str) -> str:
        """Extract project ID from service account file."""
        import json
        with open(path) as f:
            return json.load(f)['project_id']
    
    @property
    def project_id(self) -> str:
        """Firebase project ID for constructing FCM endpoint."""
        return self._project_id
    
    def get_access_token(self) -> str:
        """
        Returns a valid access token, refreshing if necessary.
        
        Google's credentials object handles token lifecycle:
        - First call generates token
        - Subsequent calls return cached token
        - Automatically refreshes when expired
        """
        if not self.credentials.valid:
            self.credentials.refresh(Request())
        
        return self.credentials.token
    
    def get_authorization_header(self) -> str:
        """Returns the Bearer token header value."""
        return f"Bearer {self.get_access_token()}"
    
    def get_endpoint(self) -> str:
        """Returns the FCM HTTP v1 API endpoint."""
        return f"https://fcm.googleapis.com/v1/projects/{self.project_id}/messages:send"
 
 
# Production usage
auth = FCMAuthenticator('/secure/credentials/firebase-service-account.json')
 
# Make FCM request
import httpx
 
async def send_fcm_message(message: dict) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            auth.get_endpoint(),
            headers={
                'Authorization': auth.get_authorization_header(),
                'Content-Type': 'application/json'
            },
            json={'message': message}
        )
        return response.json()

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

# FCM error handling implementation
 
class FCMErrorHandler:
    """
    Processes FCM error responses with appropriate recovery actions.
    
    FCM error codes follow Google's error model with:
    - HTTP status code (4xx, 5xx)
    - Error code in response body
    - Detailed error message
    """
    
    # Errors indicating invalid registration token
    TOKEN_INVALID_CODES = {
        'UNREGISTERED',          # App uninstalled or token invalid
        'INVALID_ARGUMENT',      # Token format is invalid (when targeting token)
    }
    
    # Transient errors that should be retried
    TRANSIENT_CODES = {
        'UNAVAILABLE',           # FCM service temporarily unavailable
        'INTERNAL',              # FCM internal error
        'QUOTA_EXCEEDED',        # Rate limit or quota exceeded
    }
    
    # Errors requiring configuration changes
    CONFIG_ERRORS = {
        'SENDER_ID_MISMATCH',    # Token was generated with different sender
        'THIRD_PARTY_AUTH_ERROR', # APNs auth error (when sending to iOS via FCM)
    }
    
    def __init__(self, token_service, metrics):
        self.token_service = token_service
        self.metrics = metrics
    
    async def handle_error(
        self,
        token: str,
        status_code: int,
        error_response: dict
    ) -> ErrorHandleResult:
        """
        Process FCM error and determine appropriate action.
        
        Args:
            token: The registration token that failed
            status_code: HTTP status code
            error_response: Parsed JSON error response
        
        Returns:
            ErrorHandleResult with retry guidance
        """
        # Extract error code from response
        error_code = self._extract_error_code(error_response)
        
        self.metrics.increment(
            'fcm.error',
            tags={'status': status_code, 'code': error_code}
        )
        
        if error_code in self.TOKEN_INVALID_CODES:
            return await self._handle_invalid_token(token, error_code)
        
        if error_code in self.TRANSIENT_CODES:
            return self._handle_transient_error(error_code, error_response)
        
        if error_code in self.CONFIG_ERRORS:
            return self._handle_config_error(error_code)
        
        # Handle by HTTP status code
        if status_code == 400:
            return ErrorHandleResult(
                should_retry=False,
                action=f"Bad request: {error_response}"
            )
        
        if status_code == 401:
            return ErrorHandleResult(
                should_retry=True,
                retry_after=0,  # Retry immediately after refreshing auth
                action="Auth error - refreshing credentials"
            )
        
        if status_code >= 500:
            return self._handle_transient_error('SERVER_ERROR', error_response)
        
        return ErrorHandleResult(
            should_retry=False,
            action=f"Unhandled error: {status_code} - {error_code}"
        )
    
    def _extract_error_code(self, error_response: dict) -> str:
        """
        Extracts error code from FCM error response.
        
        FCM error format:
        {
            "error": {
                "code": 400,
                "message": "...",
                "status": "INVALID_ARGUMENT",
                "details": [
                    {
                        "@type": "type.googleapis.com/google.firebase.fcm.v1.FcmError",
                        "errorCode": "UNREGISTERED"
                    }
                ]
            }
        }
        """
        try:
            error = error_response.get('error', {})
            
            # Check details for FCM-specific error code
            for detail in error.get('details', []):
                if 'errorCode' in detail:
                    return detail['errorCode']
            
            # Fall back to status
            return error.get('status', 'UNKNOWN')
        except (KeyError, TypeError):
            return 'UNKNOWN'
    
    async def _handle_invalid_token(
        self,
        token: str,
        error_code: str
    ) -> ErrorHandleResult:
        """
        Handle permanently invalid tokens.
        """
        await self.token_service.invalidate_token(
            token=token,
            reason=error_code
        )
        
        return ErrorHandleResult(
            should_retry=False,
            should_invalidate_token=True,
            action=f"Token invalidated: {error_code}"
        )
    
    def _handle_transient_error(
        self,
        error_code: str,
        error_response: dict
    ) -> ErrorHandleResult:
        """
        Handle transient errors with retry.
        
        FCM may include Retry-After header for rate limiting.
        """
        retry_after = 60  # Default
        
        if error_code == 'QUOTA_EXCEEDED':
            # Back off more aggressively for quota errors
            retry_after = 300
        
        return ErrorHandleResult(
            should_retry=True,
            retry_after=retry_after,
            action=f"Transient error: {error_code}, retry in {retry_after}s"
        )

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

# Unified notification layer abstraction
 
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
 
class Platform(Enum):
    IOS = "ios"
    ANDROID = "android"
 
@dataclass
class UnifiedNotification:
    """Platform-agnostic notification structure."""
    title: str
    body: str
    badge: int | None = None
    sound: str | None = "default"
    data: dict | None = None
    image_url: str | None = None
    priority: str = "normal"  # "normal" or "high"
    ttl_seconds: int = 86400  # Default 24 hours
    collapse_key: str | None = None
 
 
class PushProvider(ABC):
    """Abstract base for platform-specific providers."""
    
    @abstractmethod
    async def send(
        self,
        token: str,
        notification: UnifiedNotification
    ) -> SendResult:
        pass
 
 
class UnifiedPushService:
    """
    Unified push notification service that abstracts platform differences.
    """
    
    def __init__(
        self,
        apns_provider: APNsProvider,
        fcm_provider: FCMProvider,
        token_registry: TokenRegistry
    ):
        self.providers = {
            Platform.IOS: apns_provider,
            Platform.ANDROID: fcm_provider,
        }
        self.tokens = token_registry
    
    async def send_to_user(
        self,
        user_id: str,
        notification: UnifiedNotification
    ) -> list[SendResult]:
        """
        Send notification to all devices registered to a user.
        
        Returns results for each device attempt.
        """
        tokens = await self.tokens.get_user_tokens(user_id)
        
        results = []
        for token_info in tokens:
            provider = self.providers[token_info.platform]
            result = await provider.send(token_info.token, notification)
            results.append(result)
        
        return results
    
    async def send_to_segment(
        self,
        segment_query: dict,
        notification: UnifiedNotification
    ) -> BatchSendResult:
        """
        Send notification to all users matching segment criteria.
        
        Uses streaming/pagination to handle large segments efficiently.
        """
        total_sent = 0
        total_failed = 0
        
        async for token_batch in self.tokens.stream_by_segment(segment_query):
            # Group by platform for batched sending
            ios_tokens = [t for t in token_batch if t.platform == Platform.IOS]
            android_tokens = [t for t in token_batch if t.platform == Platform.ANDROID]
            
            # Send to each platform
            for token in ios_tokens:
                result = await self.providers[Platform.IOS].send(
                    token.token, notification
                )
                if result.success:
                    total_sent += 1
                else:
                    total_failed += 1
            
            for token in android_tokens:
                result = await self.providers[Platform.ANDROID].send(
                    token.token, notification
                )
                if result.success:
                    total_sent += 1
                else:
                    total_failed += 1
        
        return BatchSendResult(
            total_sent=total_sent,
            total_failed=total_failed
        )