ashwin/nxtgauge-backend-rust
0
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(companies): contact-unlock double-spend and job-quota race

Browse source 
view_contact previously decremented free_contact_views/
purchased_contact_views on every call with no record of which
applications had already had their contact unlocked -- refreshing the
same application's contact panel silently re-spent the allowance every
time, not just under concurrency, and the check-then-act shape also let
concurrent requests both pass the exhausted-quota check before either
UPDATE committed. Now the whole claim (dedup check + conditional
decrement + job_applications.contact_unlocked_at write) runs in one
transaction with company_profiles locked FOR UPDATE, gated on a
WHERE ... > 0 guard on the decrement itself.

create_job had the same shape for the 1-free-job/month rule and
purchased_job_slots: two concurrent requests could both observe
"quota available" before either committed, letting a company publish
2+ free jobs in a month or drive purchased_job_slots negative. Same
fix: lock company_profiles FOR UPDATE for the whole check+mutate+
insert sequence.

Also adds CHECK (>= 0) constraints on all four company_profiles
counters as defense-in-depth, matching the pattern tracecoin_wallets
already uses for balance/reserved.
This commit is contained in:
Ashwin Kumar Sivakumar 2026-08-18 00:51:02 +05:30
parent cdf78635f6
commit 617a75971f
5 changed files with 275 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

296
apps/companies/src/handlers/mod.rs
View file

@ -199,36 +199,80 @@ async fn create_job(
).into_response();
}
// --- New Quota Logic ---
let jobs_this_month = match JobRepository::count_by_company_id_this_month(&state.pool, company.id).await {
Ok(count) => count,
// --- Quota logic ---
// The free-job-per-month count and the purchased-slot decrement both
// ran as independent, unlocked statements against a company_profiles
// row read before either of them -- two concurrent create_job calls for
// the same company could each observe "0 jobs this month" (or "slots
// remaining > 0") before either commits, letting a company publish 2+
// free jobs in a month or drive purchased_job_slots negative. Same
// class of bug as the job_applications duplicate-application race.
//
// Locking company_profiles FOR UPDATE for the whole check+mutate+insert
// sequence serializes concurrent create_job calls for one company: the
// second call blocks until the first commits, then sees the real,
// already-updated counts.
let mut tx = match state.pool.begin().await {
Ok(tx) => tx,
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
};
if let Err(e) = sqlx::query("SELECT id FROM company_profiles WHERE id = $1 FOR UPDATE")
.bind(company.id)
.execute(&mut *tx)
.await
{
let _ = tx.rollback().await;
return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response();
}
let jobs_this_month: i64 = match sqlx::query_scalar(
r#"
SELECT COUNT(*)
FROM jobs
WHERE company_id = $1
AND created_at >= date_trunc('month', now())
AND status != 'REJECTED'
"#,
)
.bind(company.id)
.fetch_one(&mut *tx)
.await
{
Ok(count) => count,
Err(e) => {
let _ = tx.rollback().await;
return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response();
}
};
if jobs_this_month >= 1 {
// Must use a purchased slot if they've already used their monthly freebie
if company.purchased_job_slots <= 0 {
return (
StatusCode::PAYMENT_REQUIRED,
Json(serde_json::json!({
"error": "Monthly free job quota exhausted. Please purchase job slots.",
"code": "QUOTA_EXHAUSTED",
"requires_tracecoins": true
}))
).into_response();
}
// Deduct ONE purchased slot
let deduct_result = sqlx::query(
"UPDATE company_profiles SET purchased_job_slots = purchased_job_slots - 1 WHERE id = $1",
"UPDATE company_profiles SET purchased_job_slots = purchased_job_slots - 1 WHERE id = $1 AND purchased_job_slots > 0",
)
.bind(company.id)
.execute(&state.pool)
.execute(&mut *tx)
.await;
if let Err(e) = deduct_result {
tracing::error!("Failed to deduct job slot: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to deduct quota" }))).into_response();
match deduct_result {
Ok(r) if r.rows_affected() == 1 => {}
Ok(_) => {
let _ = tx.rollback().await;
return (
StatusCode::PAYMENT_REQUIRED,
Json(serde_json::json!({
"error": "Monthly free job quota exhausted. Please purchase job slots.",
"code": "QUOTA_EXHAUSTED",
"requires_tracecoins": true
}))
).into_response();
}
Err(e) => {
let _ = tx.rollback().await;
tracing::error!("Failed to deduct job slot: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to deduct quota" }))).into_response();
}
}
}
// -----------------------
@ -246,20 +290,50 @@ async fn create_job(
skills: payload.skills,
};
match JobRepository::create(&state.pool, db_payload).await {
Ok(job) => {
// Invalidate company's job list cache
let mut redis = state.redis.clone();
let pattern = format!("jobs:company:{}:*", company.id);
if let Ok(keys) = redis.keys::<_, Vec<String>>(pattern).await {
if !keys.is_empty() {
let _ = redis.del::<_, ()>(keys).await;
}
}
(StatusCode::CREATED, Json(job)).into_response()
let job = match sqlx::query_as::<_, db::models::job::Job>(
r#"
INSERT INTO jobs (
company_id, title, category, description, location,
job_type, salary_min, salary_max, experience_years, skills
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
RETURNING *
"#,
)
.bind(db_payload.company_id)
.bind(db_payload.title)
.bind(db_payload.category)
.bind(db_payload.description)
.bind(db_payload.location)
.bind(db_payload.job_type.unwrap_or_else(|| "FULL_TIME".to_string()))
.bind(db_payload.salary_min)
.bind(db_payload.salary_max)
.bind(db_payload.experience_years)
.bind(db_payload.skills.unwrap_or_default())
.fetch_one(&mut *tx)
.await
{
Ok(job) => job,
Err(e) => {
let _ = tx.rollback().await;
return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response();
}
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
};
if let Err(e) = tx.commit().await {
return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response();
}
// Invalidate company's job list cache
let mut redis = state.redis.clone();
let pattern = format!("jobs:company:{}:*", company.id);
if let Ok(keys) = redis.keys::<_, Vec<String>>(pattern).await {
if !keys.is_empty() {
let _ = redis.del::<_, ()>(keys).await;
}
}
(StatusCode::CREATED, Json(job)).into_response()
}
async fn get_job(
@ -672,36 +746,113 @@ async fn view_contact(
return (StatusCode::FORBIDDEN, "Access denied").into_response();
}
let free_views = company.free_contact_views;
let purchased_views = company.purchased_contact_views;
// Claiming the quota (or confirming it's already been spent on this
// application) all happens in one transaction with company_profiles'
// row locked FOR UPDATE, so two concurrent view_contact calls for the
// same company can't both read the same pre-decrement counters and
// both pass the quota check -- the second call blocks until the first
// commits, then sees the real, already-decremented balance.
//
// job_applications.contact_unlocked_at is what makes this idempotent:
// without it, re-opening (or simply refreshing) the same application's
// contact panel would re-spend the allowance every time, not just under
// concurrency -- there was previously no record of "already unlocked
// this one" at all.
let mut tx = match state.pool.begin().await {
Ok(tx) => tx,
Err(e) => {
tracing::error!("Failed to start contact-unlock transaction: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to fetch contact info").into_response();
}
};
if free_views <= 0 && purchased_views <= 0 {
return (StatusCode::PAYMENT_REQUIRED, Json(serde_json::json!({
"error": "Contact view quota exhausted",
"code": "QUOTA_EXHAUSTED",
"requires_purchase": true,
"message": "You have used all your free contact views. Please purchase a contact view package to continue."
}))).into_response();
}
let already_unlocked: Option<Option<chrono::DateTime<chrono::Utc>>> = sqlx::query_scalar(
"SELECT contact_unlocked_at FROM job_applications WHERE id = $1 FOR UPDATE",
)
.bind(id)
.fetch_optional(&mut *tx)
.await
.unwrap_or(None);
let used_free = free_views > 0;
let already_unlocked = matches!(already_unlocked, Some(Some(_)));
if used_free {
sqlx::query(
"UPDATE company_profiles SET free_contact_views = free_contact_views - 1, updated_at = NOW() WHERE id = $1"
)
.bind(company.id)
.execute(&state.pool)
.await
.ok();
let (used_free, new_free, new_purchased) = if already_unlocked {
(false, company.free_contact_views, company.purchased_contact_views)
} else {
sqlx::query(
"UPDATE company_profiles SET purchased_contact_views = purchased_contact_views - 1, updated_at = NOW() WHERE id = $1"
let (free_views, purchased_views): (i32, i32) = match sqlx::query_as(
"SELECT free_contact_views, purchased_contact_views FROM company_profiles WHERE id = $1 FOR UPDATE",
)
.bind(company.id)
.execute(&state.pool)
.fetch_one(&mut *tx)
.await
.ok();
{
Ok(row) => row,
Err(e) => {
let _ = tx.rollback().await;
tracing::error!("Failed to lock company_profiles for contact unlock: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to fetch contact info").into_response();
}
};
if free_views <= 0 && purchased_views <= 0 {
let _ = tx.rollback().await;
return (StatusCode::PAYMENT_REQUIRED, Json(serde_json::json!({
"error": "Contact view quota exhausted",
"code": "QUOTA_EXHAUSTED",
"requires_purchase": true,
"message": "You have used all your free contact views. Please purchase a contact view package to continue."
}))).into_response();
}
let used_free = free_views > 0;
let decrement_result = if used_free {
sqlx::query(
"UPDATE company_profiles SET free_contact_views = free_contact_views - 1, updated_at = NOW() WHERE id = $1 AND free_contact_views > 0"
)
.bind(company.id)
.execute(&mut *tx)
.await
} else {
sqlx::query(
"UPDATE company_profiles SET purchased_contact_views = purchased_contact_views - 1, updated_at = NOW() WHERE id = $1 AND purchased_contact_views > 0"
)
.bind(company.id)
.execute(&mut *tx)
.await
};
match decrement_result {
Ok(r) if r.rows_affected() == 1 => {}
_ => {
let _ = tx.rollback().await;
return (StatusCode::PAYMENT_REQUIRED, Json(serde_json::json!({
"error": "Contact view quota exhausted",
"code": "QUOTA_EXHAUSTED",
"requires_purchase": true,
"message": "You have used all your free contact views. Please purchase a contact view package to continue."
}))).into_response();
}
}
if let Err(e) = sqlx::query("UPDATE job_applications SET contact_unlocked_at = NOW() WHERE id = $1")
.bind(id)
.execute(&mut *tx)
.await
{
let _ = tx.rollback().await;
tracing::error!("Failed to record contact unlock for application {}: {}", id, e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to fetch contact info").into_response();
}
let new_free = if used_free { free_views - 1 } else { free_views };
let new_purchased = if used_free { purchased_views } else { purchased_views - 1 };
(used_free, new_free, new_purchased)
};
if let Err(e) = tx.commit().await {
tracing::error!("Failed to commit contact-unlock transaction: {}", e);
return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to fetch contact info").into_response();
}
let contact = sqlx::query_as::<_, (Option<String>, String, Option<String>)>(
@ -717,23 +868,24 @@ async fn view_contact(
match contact {
Ok(Some((name, email, phone))) => {
let new_free = if used_free { free_views - 1 } else { free_views };
let new_purchased = if used_free { purchased_views } else { purchased_views - 1 };
let _ = sqlx::query(
r#"
INSERT INTO notifications (user_id, title, body, type, reference_id)
VALUES ($1, $2, $3, $4, $5)
"#
)
.bind(app.applicant_user_id)
.bind("Your contact was viewed")
.bind(format!("{} viewed your application for {}", company.company_name, job.title))
.bind("APPLICATION")
.bind(id)
.execute(&state.pool)
.await
.ok();
// Only notify the applicant the first time this contact is
// actually unlocked -- not on every idempotent re-fetch.
if !already_unlocked {
let _ = sqlx::query(
r#"
INSERT INTO notifications (user_id, title, body, type, reference_id)
VALUES ($1, $2, $3, $4, $5)
"#
)
.bind(app.applicant_user_id)
.bind("Your contact was viewed")
.bind(format!("{} viewed your application for {}", company.company_name, job.title))
.bind("APPLICATION")
.bind(id)
.execute(&state.pool)
.await
.ok();
}
(StatusCode::OK, Json(serde_json::json!({
"application_id": id,

3
crates/db/migrations/20260818000000_job_applications_contact_unlock_dedup.down.sql Normal file
View file

@ -0,0 +1,3 @@
BEGIN;
ALTER TABLE job_applications DROP COLUMN IF EXISTS contact_unlocked_at;
COMMIT;

18
crates/db/migrations/20260818000000_job_applications_contact_unlock_dedup.up.sql Normal file
View file

@ -0,0 +1,18 @@
-- apps/companies/src/handlers/mod.rs::view_contact spends a company's
-- free_contact_views/purchased_contact_views allowance on every call, with
-- no record of which applications' contacts a company has already unlocked.
-- That means simply reloading the page re-spends the allowance for the same
-- applicant every time (not just a concurrency race -- guaranteed
-- double-consumption even one request at a time), and concurrent requests
-- can additionally race the check-then-decrement on company_profiles's
-- counters past zero.
--
-- This column lets view_contact skip the charge (and the counter race)
-- entirely once an application's contact has already been unlocked by its
-- company -- see the accompanying handler fix.
BEGIN;
ALTER TABLE job_applications
ADD COLUMN IF NOT EXISTS contact_unlocked_at TIMESTAMPTZ;
COMMIT;

9
crates/db/migrations/20260818000001_company_profiles_counters_non_negative.down.sql Normal file
View file

@ -0,0 +1,9 @@
BEGIN;
ALTER TABLE company_profiles
DROP CONSTRAINT IF EXISTS company_profiles_free_job_slots_non_negative,
DROP CONSTRAINT IF EXISTS company_profiles_purchased_job_slots_non_negative,
DROP CONSTRAINT IF EXISTS company_profiles_free_contact_views_non_negative,
DROP CONSTRAINT IF EXISTS company_profiles_purchased_contact_views_non_negative;
COMMIT;

21
crates/db/migrations/20260818000001_company_profiles_counters_non_negative.up.sql Normal file
View file

@ -0,0 +1,21 @@
-- Defense-in-depth alongside the app-layer fixes to view_contact (contact
-- unlock dedup) and create_job (locked quota check): these counters were
-- previously decremented with plain `x = x - 1` and no floor, so a bug in
-- either handler (or any future caller) could still drive them negative
-- with nothing at the DB level to stop it. tracecoin_wallets already has
-- this kind of CHECK for balance/reserved -- these four counters are the
-- same class of "must never go negative" value and deserve the same
-- guarantee, not just app-layer discipline.
--
-- Safe to run live: fails loudly rather than silently if any row has
-- already gone negative, same reasoning as the job_applications unique
-- constraint migration.
BEGIN;
ALTER TABLE company_profiles
ADD CONSTRAINT company_profiles_free_job_slots_non_negative CHECK (free_job_slots >= 0),
ADD CONSTRAINT company_profiles_purchased_job_slots_non_negative CHECK (purchased_job_slots >= 0),
ADD CONSTRAINT company_profiles_free_contact_views_non_negative CHECK (free_contact_views >= 0),
ADD CONSTRAINT company_profiles_purchased_contact_views_non_negative CHECK (purchased_contact_views >= 0);
COMMIT;