Page MenuHomePhabricator
Differential D12938

[Tunnelbroker] add Platform column to `device_token` DDB table
ClosedPublic
Actions

Authored by kamil on Jul 30 2024, 4:04 AM.
Tags
None
Referenced Files
Unknown Object (File)
Sun, Mar 2, 10:47 PM
Unknown Object (File)
Sat, Feb 22, 10:48 AM
Unknown Object (File)
Thu, Feb 20, 1:03 AM
Unknown Object (File)
Tue, Feb 18, 4:25 AM
Unknown Object (File)
Tue, Feb 18, 4:24 AM
Unknown Object (File)
Tue, Feb 18, 4:24 AM
Unknown Object (File)
Feb 1 2025, 11:25 AM
Unknown Object (File)
Feb 1 2025, 10:13 AM
View All 91 Files
Subscribers
ashoat

Details

Reviewers
bartek
tomek
Commits
rCOMM9306dc4214ea: [Tunnelbroker] add Platform column to `device_token` DDB table
Summary

ENG-8903.

Platform is optional to support old clients which uploaded device token without platform.

Depends on D12937

Test Plan

Test inserting and getting device token with platform

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Revision Contents
Changeset List

PathSize
services/
tunnelbroker/
src/
1 line
database/
45 lines
websockets/
6 lines

Diff 43000

View Options

services/tunnelbroker/src/constants.rs

Show All 13 Linesuse lapin::options::{
BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions, BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions,
QueueDeclareOptions, QueueDeleteOptions, QueueDeclareOptions, QueueDeleteOptions,
}; };
use lapin::types::FieldTable; use lapin::types::FieldTable;
use lapin::BasicProperties; use lapin::BasicProperties;
use tokio::io::AsyncRead; use tokio::io::AsyncRead;
use tokio::io::AsyncWrite; use tokio::io::AsyncWrite;
use tracing::{debug, error, info, trace}; use tracing::{debug, error, info, trace};
use tunnelbroker_messages::bad_device_token::BadDeviceToken;
use tunnelbroker_messages::{ use tunnelbroker_messages::{
message_to_device_request_status::Failure, message_to_device_request_status::Failure,
message_to_device_request_status::MessageSentStatus, session::DeviceTypes, message_to_device_request_status::MessageSentStatus, session::DeviceTypes,
DeviceToTunnelbrokerMessage, Heartbeat, MessageToDevice, DeviceToTunnelbrokerMessage, Heartbeat, MessageToDevice,
MessageToDeviceRequest, MessageToTunnelbroker, MessageToDeviceRequest, MessageToTunnelbroker,
}; };
use crate::notifs::apns::response::ErrorReason;
use crate::database::{self, DatabaseClient, MessageToDeviceExt}; use crate::database::{self, DatabaseClient, MessageToDeviceExt};
use crate::identity; use crate::identity;
use crate::notifs::apns::error::Error;
use crate::notifs::apns::headers::NotificationHeaders; use crate::notifs::apns::headers::NotificationHeaders;
use crate::notifs::apns::APNsNotif; use crate::notifs::apns::APNsNotif;
use crate::notifs::fcm::firebase_message::{ use crate::notifs::fcm::firebase_message::{
AndroidConfig, AndroidMessagePriority, FCMMessage, AndroidConfig, AndroidMessagePriority, FCMMessage,
}; };
use crate::notifs::web_push::WebPushNotif; use crate::notifs::web_push::WebPushNotif;
use crate::notifs::NotifClient; use crate::notifs::NotifClient;
▲ Show 20 Lines • Show All 337 Lines • ▼ Show 20 Lines match serialized_message {
debug!("Received APNs notif for {}", notif.device_id); debug!("Received APNs notif for {}", notif.device_id);
let Ok(headers) = let Ok(headers) =
serde_json::from_str::<NotificationHeaders>(&notif.headers) serde_json::from_str::<NotificationHeaders>(&notif.headers)
else { else {
return Some(MessageSentStatus::SerializationError(notif.headers)); return Some(MessageSentStatus::SerializationError(notif.headers));
}; };
let device_token = match self.get_device_token(notif.device_id).await { let device_token =
match self.get_device_token(notif.device_id.clone()).await {
Ok(token) => token, Ok(token) => token,
Err(e) => { Err(e) => {
return Some( return Some(self.get_message_to_device_status(
self &notif.client_message_id,
.get_message_to_device_status(&notif.client_message_id, Err(e)), Err(e),
) ))
} }
}; };
let apns_notif = APNsNotif { let apns_notif = APNsNotif {
device_token, device_token: device_token.clone(),
headers, headers,
payload: notif.payload, payload: notif.payload,
}; };
if let Some(apns) = self.notif_client.apns.clone() { if let Some(apns) = self.notif_client.apns.clone() {
let response = apns.send(apns_notif).await; let response = apns.send(apns_notif).await;
if let Err(Error::ResponseError(body)) = &response {
if matches!(
body.reason,
ErrorReason::BadDeviceToken
| ErrorReason::Unregistered
| ErrorReason::ExpiredToken
) {
if let Err(e) = self
.invalidate_device_token(notif.device_id, device_token)
.await
{
error!("Error invalidating device token: {:?}", e);
};
}
}
return Some( return Some(
self self
.get_message_to_device_status(&notif.client_message_id, response), .get_message_to_device_status(&notif.client_message_id, response),
); );
} }
Some(self.get_message_to_device_status( Some(self.get_message_to_device_status(
&notif.client_message_id, &notif.client_message_id,
▲ Show 20 Lines • Show All 168 Lines • ▼ Show 20 Lines match db_token {
Err(SessionError::InvalidDeviceToken) Err(SessionError::InvalidDeviceToken)
} else { } else {
Ok(token.device_token) Ok(token.device_token)
} }
} }
None => Err(SessionError::MissingDeviceToken), None => Err(SessionError::MissingDeviceToken),
} }
} }
async fn invalidate_device_token(
&self,
device_id: String,
invalidated_token: String,
) -> Result<(), SessionError> {
let bad_device_token_message = BadDeviceToken { invalidated_token };
let payload = serde_json::to_string(&bad_device_token_message)?;
let message_request = MessageToDeviceRequest {
client_message_id: uuid::Uuid::new_v4().to_string(),
device_id: device_id.to_string(),
payload,
};
self.handle_message_to_device(&message_request).await?;
self
.db_client
.mark_device_token_as_invalid(&device_id)
.await
.map_err(SessionError::DatabaseError)?;
Ok(())
}
} }
View Options

services/tunnelbroker/src/database/mod.rs

Show All 13 Linesuse lapin::options::{
BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions, BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions,
QueueDeclareOptions, QueueDeleteOptions, QueueDeclareOptions, QueueDeleteOptions,
}; };
use lapin::types::FieldTable; use lapin::types::FieldTable;
use lapin::BasicProperties; use lapin::BasicProperties;
use tokio::io::AsyncRead; use tokio::io::AsyncRead;
use tokio::io::AsyncWrite; use tokio::io::AsyncWrite;
use tracing::{debug, error, info, trace}; use tracing::{debug, error, info, trace};
use tunnelbroker_messages::bad_device_token::BadDeviceToken;
use tunnelbroker_messages::{ use tunnelbroker_messages::{
message_to_device_request_status::Failure, message_to_device_request_status::Failure,
message_to_device_request_status::MessageSentStatus, session::DeviceTypes, message_to_device_request_status::MessageSentStatus, session::DeviceTypes,
DeviceToTunnelbrokerMessage, Heartbeat, MessageToDevice, DeviceToTunnelbrokerMessage, Heartbeat, MessageToDevice,
MessageToDeviceRequest, MessageToTunnelbroker, MessageToDeviceRequest, MessageToTunnelbroker,
}; };
use crate::notifs::apns::response::ErrorReason;
use crate::database::{self, DatabaseClient, MessageToDeviceExt}; use crate::database::{self, DatabaseClient, MessageToDeviceExt};
use crate::identity; use crate::identity;
use crate::notifs::apns::error::Error;
use crate::notifs::apns::headers::NotificationHeaders; use crate::notifs::apns::headers::NotificationHeaders;
use crate::notifs::apns::APNsNotif; use crate::notifs::apns::APNsNotif;
use crate::notifs::fcm::firebase_message::{ use crate::notifs::fcm::firebase_message::{
AndroidConfig, AndroidMessagePriority, FCMMessage, AndroidConfig, AndroidMessagePriority, FCMMessage,
}; };
use crate::notifs::web_push::WebPushNotif; use crate::notifs::web_push::WebPushNotif;
use crate::notifs::NotifClient; use crate::notifs::NotifClient;
▲ Show 20 Lines • Show All 337 Lines • ▼ Show 20 Lines match serialized_message {
debug!("Received APNs notif for {}", notif.device_id); debug!("Received APNs notif for {}", notif.device_id);
let Ok(headers) = let Ok(headers) =
serde_json::from_str::<NotificationHeaders>(&notif.headers) serde_json::from_str::<NotificationHeaders>(&notif.headers)
else { else {
return Some(MessageSentStatus::SerializationError(notif.headers)); return Some(MessageSentStatus::SerializationError(notif.headers));
}; };
let device_token = match self.get_device_token(notif.device_id).await { let device_token =
match self.get_device_token(notif.device_id.clone()).await {
Ok(token) => token, Ok(token) => token,
Err(e) => { Err(e) => {
return Some( return Some(self.get_message_to_device_status(
self &notif.client_message_id,
.get_message_to_device_status(&notif.client_message_id, Err(e)), Err(e),
) ))
} }
}; };
let apns_notif = APNsNotif { let apns_notif = APNsNotif {
device_token, device_token: device_token.clone(),
headers, headers,
payload: notif.payload, payload: notif.payload,
}; };
if let Some(apns) = self.notif_client.apns.clone() { if let Some(apns) = self.notif_client.apns.clone() {
let response = apns.send(apns_notif).await; let response = apns.send(apns_notif).await;
if let Err(Error::ResponseError(body)) = &response {
if matches!(
body.reason,
ErrorReason::BadDeviceToken
| ErrorReason::Unregistered
| ErrorReason::ExpiredToken
) {
if let Err(e) = self
.invalidate_device_token(notif.device_id, device_token)
.await
{
error!("Error invalidating device token: {:?}", e);
};
}
}
return Some( return Some(
self self
.get_message_to_device_status(&notif.client_message_id, response), .get_message_to_device_status(&notif.client_message_id, response),
); );
} }
Some(self.get_message_to_device_status( Some(self.get_message_to_device_status(
&notif.client_message_id, &notif.client_message_id,
▲ Show 20 Lines • Show All 168 Lines • ▼ Show 20 Lines match db_token {
Err(SessionError::InvalidDeviceToken) Err(SessionError::InvalidDeviceToken)
} else { } else {
Ok(token.device_token) Ok(token.device_token)
} }
} }
None => Err(SessionError::MissingDeviceToken), None => Err(SessionError::MissingDeviceToken),
} }
} }
async fn invalidate_device_token(
&self,
device_id: String,
invalidated_token: String,
) -> Result<(), SessionError> {
let bad_device_token_message = BadDeviceToken { invalidated_token };
let payload = serde_json::to_string(&bad_device_token_message)?;
let message_request = MessageToDeviceRequest {
client_message_id: uuid::Uuid::new_v4().to_string(),
device_id: device_id.to_string(),
payload,
};
self.handle_message_to_device(&message_request).await?;
self
.db_client
.mark_device_token_as_invalid(&device_id)
.await
.map_err(SessionError::DatabaseError)?;
Ok(())
}
} }
View Options

services/tunnelbroker/src/websockets/session.rs

Show All 13 Linesuse lapin::options::{
BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions, BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions,
QueueDeclareOptions, QueueDeleteOptions, QueueDeclareOptions, QueueDeleteOptions,
}; };
use lapin::types::FieldTable; use lapin::types::FieldTable;
use lapin::BasicProperties; use lapin::BasicProperties;
use tokio::io::AsyncRead; use tokio::io::AsyncRead;
use tokio::io::AsyncWrite; use tokio::io::AsyncWrite;
use tracing::{debug, error, info, trace}; use tracing::{debug, error, info, trace};
use tunnelbroker_messages::bad_device_token::BadDeviceToken;
use tunnelbroker_messages::{ use tunnelbroker_messages::{
message_to_device_request_status::Failure, message_to_device_request_status::Failure,
message_to_device_request_status::MessageSentStatus, session::DeviceTypes, message_to_device_request_status::MessageSentStatus, session::DeviceTypes,
DeviceToTunnelbrokerMessage, Heartbeat, MessageToDevice, DeviceToTunnelbrokerMessage, Heartbeat, MessageToDevice,
MessageToDeviceRequest, MessageToTunnelbroker, MessageToDeviceRequest, MessageToTunnelbroker,
}; };
use crate::notifs::apns::response::ErrorReason;
use crate::database::{self, DatabaseClient, MessageToDeviceExt}; use crate::database::{self, DatabaseClient, MessageToDeviceExt};
use crate::identity; use crate::identity;
use crate::notifs::apns::error::Error;
use crate::notifs::apns::headers::NotificationHeaders; use crate::notifs::apns::headers::NotificationHeaders;
use crate::notifs::apns::APNsNotif; use crate::notifs::apns::APNsNotif;
use crate::notifs::fcm::firebase_message::{ use crate::notifs::fcm::firebase_message::{
AndroidConfig, AndroidMessagePriority, FCMMessage, AndroidConfig, AndroidMessagePriority, FCMMessage,
}; };
use crate::notifs::web_push::WebPushNotif; use crate::notifs::web_push::WebPushNotif;
use crate::notifs::NotifClient; use crate::notifs::NotifClient;
▲ Show 20 Lines • Show All 337 Lines • ▼ Show 20 Lines match serialized_message {
debug!("Received APNs notif for {}", notif.device_id); debug!("Received APNs notif for {}", notif.device_id);
let Ok(headers) = let Ok(headers) =
serde_json::from_str::<NotificationHeaders>(&notif.headers) serde_json::from_str::<NotificationHeaders>(&notif.headers)
else { else {
return Some(MessageSentStatus::SerializationError(notif.headers)); return Some(MessageSentStatus::SerializationError(notif.headers));
}; };
let device_token = match self.get_device_token(notif.device_id).await { let device_token =
match self.get_device_token(notif.device_id.clone()).await {
Ok(token) => token, Ok(token) => token,
Err(e) => { Err(e) => {
return Some( return Some(self.get_message_to_device_status(
self &notif.client_message_id,
.get_message_to_device_status(&notif.client_message_id, Err(e)), Err(e),
) ))
} }
}; };
let apns_notif = APNsNotif { let apns_notif = APNsNotif {
device_token, device_token: device_token.clone(),
headers, headers,
payload: notif.payload, payload: notif.payload,
}; };
if let Some(apns) = self.notif_client.apns.clone() { if let Some(apns) = self.notif_client.apns.clone() {
let response = apns.send(apns_notif).await; let response = apns.send(apns_notif).await;
if let Err(Error::ResponseError(body)) = &response {
if matches!(
body.reason,
ErrorReason::BadDeviceToken
| ErrorReason::Unregistered
| ErrorReason::ExpiredToken
) {
if let Err(e) = self
.invalidate_device_token(notif.device_id, device_token)
.await
{
error!("Error invalidating device token: {:?}", e);
};
}
}
return Some( return Some(
self self
.get_message_to_device_status(&notif.client_message_id, response), .get_message_to_device_status(&notif.client_message_id, response),
); );
} }
Some(self.get_message_to_device_status( Some(self.get_message_to_device_status(
&notif.client_message_id, &notif.client_message_id,
▲ Show 20 Lines • Show All 168 Lines • ▼ Show 20 Lines match db_token {
Err(SessionError::InvalidDeviceToken) Err(SessionError::InvalidDeviceToken)
} else { } else {
Ok(token.device_token) Ok(token.device_token)
} }
} }
None => Err(SessionError::MissingDeviceToken), None => Err(SessionError::MissingDeviceToken),
} }
} }
async fn invalidate_device_token(
&self,
device_id: String,
invalidated_token: String,
) -> Result<(), SessionError> {
let bad_device_token_message = BadDeviceToken { invalidated_token };
let payload = serde_json::to_string(&bad_device_token_message)?;
let message_request = MessageToDeviceRequest {
client_message_id: uuid::Uuid::new_v4().to_string(),
device_id: device_id.to_string(),
payload,
};
self.handle_message_to_device(&message_request).await?;
self
.db_client
.mark_device_token_as_invalid(&device_id)
.await
.map_err(SessionError::DatabaseError)?;
Ok(())
}
} }