Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: improve create client performance #337

Merged
merged 6 commits into from
May 21, 2024
Merged

fix: improve create client performance #337

Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a4d9a9c
fix: improve create client performance
chris13524 May 20, 2024
f8c7dc0
fix: failing tests
chris13524 May 21, 2024
5ffce16
fix: tests
chris13524 May 21, 2024
2cfcca4
fix: metrics
chris13524 May 21, 2024
f7f260f
fix: missing transaction
chris13524 May 21, 2024
776802a
chore: use functional mapping
chris13524 May 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 88 additions & 48 deletions src/stores/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,63 +47,103 @@ impl ClientStore for sqlx::PgPool {
client.token
);

let mut transaction = self.begin().await?;

// Statement for locking based on the client id to prevent an issue #230
// and locking based on the token to prevent an issue #292
let start = Instant::now();
sqlx::query(
"SELECT
pg_advisory_xact_lock(abs(hashtext($1::text))),
pg_advisory_xact_lock(abs(hashtext($2::text)))",
)
.bind(id)
.bind(client.token.clone())
.execute(&mut transaction)
.await?;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_pg_advisory_xact_lock", start);
#[derive(Debug, Clone, PartialEq, Eq, sqlx::FromRow)]
pub struct ClientSelect {
pub id: String,
pub device_token: String,
}

let query = "
SELECT *
FROM public.clients
WHERE id = $1
OR device_token = $2
FOR UPDATE
";
let start = Instant::now();
sqlx::query("DELETE FROM public.clients WHERE id = $1 OR device_token = $2")
let res = sqlx::query_as::<sqlx::postgres::Postgres, ClientSelect>(query)
.bind(id)
.bind(client.token.clone())
.execute(&mut transaction)
.await?;
.fetch_one(self)
.await;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_delete", start);
}

let start = Instant::now();
let mut insert_query = sqlx::QueryBuilder::new(
"INSERT INTO public.clients (id, tenant_id, push_type, device_token, always_raw)",
);
insert_query.push_values(
vec![(
id,
tenant_id,
client.push_type,
client.token,
client.always_raw,
)],
|mut b, client| {
b.push_bind(client.0)
.push_bind(client.1)
.push_bind(client.2)
.push_bind(client.3)
.push_bind(client.4);
},
);
insert_query.build().execute(&mut transaction).await?;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_insert", start);
}

let start = Instant::now();
transaction.commit().await?;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_commit", start);
let existing_client = match res {
Err(sqlx::Error::RowNotFound) => {
let start = Instant::now();
let mut insert_query = sqlx::QueryBuilder::new(
"INSERT INTO public.clients (id, tenant_id, push_type, device_token, always_raw)",
);
insert_query.push_values(
vec![(
id,
tenant_id,
client.push_type,
client.token,
client.always_raw,
)],
|mut b, client| {
b.push_bind(client.0)
.push_bind(client.1)
.push_bind(client.2)
.push_bind(client.3)
.push_bind(client.4);
},
);
insert_query.build().execute(self).await?;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_insert", start);
}
return Ok(());
}
Err(e) => return Err(e.into()),
Ok(row) => row,
};

if existing_client.id == id && existing_client.device_token != client.token {
let query = "
UPDATE public.clients
SET device_token = $2,
push_type = $3,
always_raw = $4,
tenant_id = $5
WHERE id = $1
";
let start = Instant::now();
sqlx::query(query)
.bind(id)
.bind(client.token)
.bind(client.push_type)
.bind(client.always_raw)
.bind(tenant_id)
.execute(self)
.await?;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_update_device_token", start);
}
} else if existing_client.device_token == client.token && existing_client.id != id {
let query = "
UPDATE public.clients
SET id = $2,
push_type = $3,
always_raw = $4,
tenant_id = $5
WHERE device_token = $1
";
let start = Instant::now();
sqlx::query(query)
.bind(client.token)
.bind(id)
.bind(client.push_type)
.bind(client.always_raw)
.bind(tenant_id)
.execute(self)
.await?;
if let Some(metrics) = metrics {
metrics.postgres_query("create_client_update_id", start);
}
}

Ok(())
Expand Down
Loading