andras/perfect-postcode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

deploy

Browse source
This commit is contained in:
Andras Schmelczer 2026-05-14 20:42:48 +01:00
parent 273d7a83ee
commit 084117cea8
48 changed files with 2283 additions and 890 deletions
Download patch file Download diff file Expand all files Collapse all files

461
server-rs/src/checkout_sessions.rs
View file

@ -37,6 +37,8 @@ pub enum CheckoutCompletion {
pub struct VerifiedCheckout {
pub reservation_id: String,
pub user_id: String,
pub stripe_session_id: String,
pub payment_intent_id: String,
pub paid_amount_pence: u64,
pub referral_invite_id: String,
}
@ -46,6 +48,7 @@ struct PendingCheckout {
id: String,
user_id: String,
stripe_session_id: String,
stripe_payment_intent_id: String,
checkout_url: String,
amount_pence: u64,
expected_total_pence: u64,
@ -149,6 +152,21 @@ async fn start_license_checkout_locked(
)
.await?;
if let Some(invite_id) = referral_invite_id.filter(|id| !id.is_empty()) {
if let Err(err) =
reserve_referral_invite(state, invite_id, &user.id, &reservation_id, expires_at_unix)
.await
{
if let Err(mark_err) = mark_checkout_status(state, &reservation_id, "failed").await {
warn!(
reservation_id,
"Failed to mark checkout reservation failed: {mark_err}"
);
}
return Err(err);
}
}
let stripe_result = create_stripe_session(
state,
user,
@ -170,6 +188,17 @@ async fn start_license_checkout_locked(
"Failed to mark checkout reservation failed: {mark_err}"
);
}
if let Some(invite_id) = referral_invite_id.filter(|id| !id.is_empty()) {
if let Err(release_err) =
release_referral_invite_reservation(state, invite_id, &reservation_id).await
{
warn!(
reservation_id,
referral_invite_id = invite_id,
"Failed to release referral invite reservation: {release_err}"
);
}
}
return Err(err);
}
};
@ -182,6 +211,17 @@ async fn start_license_checkout_locked(
"Failed to mark checkout reservation failed: {mark_err}"
);
}
if let Some(invite_id) = referral_invite_id.filter(|id| !id.is_empty()) {
if let Err(release_err) =
release_referral_invite_reservation(state, invite_id, &reservation_id).await
{
warn!(
reservation_id,
referral_invite_id = invite_id,
"Failed to release referral invite reservation: {release_err}"
);
}
}
return Err(err);
}
@ -200,6 +240,14 @@ pub async fn verify_checkout_completion(
))
}
};
let payment_intent_id = match session["payment_intent"].as_str() {
Some(id) if is_safe_stripe_session_id(id) => id,
_ => {
return Ok(CheckoutCompletion::Rejected(
"missing or invalid payment intent id".into(),
))
}
};
let checkout = match find_checkout_by_stripe_session(state, session_id).await? {
Some(checkout) => checkout,
@ -287,6 +335,8 @@ pub async fn verify_checkout_completion(
Ok(CheckoutCompletion::Grant(VerifiedCheckout {
reservation_id: checkout.id,
user_id: checkout.user_id,
stripe_session_id: session_id.to_string(),
payment_intent_id: payment_intent_id.to_string(),
paid_amount_pence: amount_total,
referral_invite_id: checkout.referral_invite_id,
}))
@ -296,7 +346,11 @@ pub async fn mark_checkout_completed(
state: &AppState,
reservation_id: &str,
paid_amount_pence: u64,
payment_intent_id: &str,
) -> anyhow::Result<()> {
if !is_safe_stripe_session_id(payment_intent_id) {
return Err(anyhow!("invalid Stripe payment intent id"));
}
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let url = format!("{pb_url}/api/collections/{CHECKOUT_COLLECTION}/records/{reservation_id}");
@ -308,6 +362,7 @@ pub async fn mark_checkout_completed(
"status": "completed",
"paid_amount_pence": paid_amount_pence,
"completed_at_unix": now_unix_secs().to_string(),
"stripe_payment_intent_id": payment_intent_id,
}))
.send()
.await?;
@ -317,7 +372,157 @@ pub async fn mark_checkout_completed(
.context("PocketBase checkout completion update failed")
}
pub async fn complete_verified_checkout(
state: &AppState,
checkout: &VerifiedCheckout,
) -> anyhow::Result<()> {
let _guard = CHECKOUT_RESERVATION_LOCK.lock().await;
let pricing_lock = acquire_pocketbase_lock(
state,
CHECKOUT_PRICING_LOCK_NAME,
CHECKOUT_PRICING_LOCK_TTL_SECS,
)
.await?;
let result = complete_verified_checkout_locked(state, checkout).await;
if let Err(err) = pricing_lock.release().await {
warn!("Failed to release checkout pricing lock: {err}");
}
result
}
async fn complete_verified_checkout_locked(
state: &AppState,
checkout: &VerifiedCheckout,
) -> anyhow::Result<()> {
let live_checkout = find_checkout_by_stripe_session(state, &checkout.stripe_session_id)
.await?
.ok_or_else(|| anyhow!("checkout reservation disappeared before completion"))?;
if live_checkout.status == "completed" {
if !checkout.referral_invite_id.is_empty() {
mark_referral_invite_used(
state,
&checkout.referral_invite_id,
&checkout.user_id,
&checkout.reservation_id,
)
.await?;
}
return Ok(());
}
if live_checkout.id != checkout.reservation_id
|| live_checkout.user_id != checkout.user_id
|| live_checkout.referral_invite_id != checkout.referral_invite_id
{
mark_checkout_status(state, &checkout.reservation_id, "invalid").await?;
return Err(anyhow!("checkout reservation changed before completion"));
}
if live_checkout.status != "pending" && live_checkout.status != "expired" {
return Err(anyhow!("checkout reservation is {}", live_checkout.status));
}
grant_license(state, &checkout.user_id).await?;
mark_checkout_completed(
state,
&checkout.reservation_id,
checkout.paid_amount_pence,
&checkout.payment_intent_id,
)
.await?;
if !checkout.referral_invite_id.is_empty() {
mark_referral_invite_used(
state,
&checkout.referral_invite_id,
&checkout.user_id,
&checkout.reservation_id,
)
.await?;
}
Ok(())
}
pub async fn grant_license_with_pricing_lock(
state: &AppState,
user_id: &str,
) -> anyhow::Result<()> {
let _guard = CHECKOUT_RESERVATION_LOCK.lock().await;
let pricing_lock = acquire_pocketbase_lock(
state,
CHECKOUT_PRICING_LOCK_NAME,
CHECKOUT_PRICING_LOCK_TTL_SECS,
)
.await?;
let result = grant_license(state, user_id).await;
if let Err(err) = pricing_lock.release().await {
warn!("Failed to release checkout pricing lock: {err}");
}
result
}
pub async fn reverse_license_for_payment_intent(
state: &AppState,
payment_intent_id: &str,
reason: &str,
) -> anyhow::Result<Option<String>> {
if !is_safe_stripe_session_id(payment_intent_id) {
return Err(anyhow!("invalid Stripe payment intent id"));
}
let _guard = CHECKOUT_RESERVATION_LOCK.lock().await;
let pricing_lock = acquire_pocketbase_lock(
state,
CHECKOUT_PRICING_LOCK_NAME,
CHECKOUT_PRICING_LOCK_TTL_SECS,
)
.await?;
let result = reverse_license_for_payment_intent_locked(state, payment_intent_id, reason).await;
if let Err(err) = pricing_lock.release().await {
warn!("Failed to release checkout pricing lock: {err}");
}
result
}
async fn reverse_license_for_payment_intent_locked(
state: &AppState,
payment_intent_id: &str,
reason: &str,
) -> anyhow::Result<Option<String>> {
let Some(checkout) = find_checkout_by_payment_intent(state, payment_intent_id).await? else {
return Ok(None);
};
if checkout.stripe_payment_intent_id != payment_intent_id {
return Err(anyhow!("checkout payment intent mismatch"));
}
if checkout.status == "refunded" || checkout.status == "disputed" {
return Ok(Some(checkout.user_id));
}
if checkout.status != "completed" {
return Ok(Some(checkout.user_id));
}
let reversed_status = if reason.contains("dispute") {
"disputed"
} else {
"refunded"
};
revoke_license(state, &checkout.user_id).await?;
mark_checkout_reversed(state, &checkout.id, reversed_status, reason).await?;
Ok(Some(checkout.user_id))
}
pub async fn grant_license(state: &AppState, user_id: &str) -> anyhow::Result<()> {
set_user_subscription(state, user_id, "licensed").await
}
async fn revoke_license(state: &AppState, user_id: &str) -> anyhow::Result<()> {
set_user_subscription(state, user_id, "free").await
}
async fn set_user_subscription(
state: &AppState,
user_id: &str,
subscription: &str,
) -> anyhow::Result<()> {
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
@ -326,7 +531,7 @@ pub async fn grant_license(state: &AppState, user_id: &str) -> anyhow::Result<()
.http_client
.patch(&url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({ "subscription": "licensed" }))
.json(&serde_json::json!({ "subscription": subscription }))
.send()
.await?;
@ -342,23 +547,36 @@ pub async fn mark_referral_invite_used(
state: &AppState,
invite_id: &str,
user_id: &str,
reservation_id: &str,
) -> anyhow::Result<()> {
if invite_id.is_empty() {
return Ok(());
}
if !is_safe_pocketbase_id(invite_id) || !is_safe_pocketbase_id(user_id) {
if !is_safe_pocketbase_id(invite_id)
|| !is_safe_pocketbase_id(user_id)
|| !is_safe_pocketbase_id(reservation_id)
{
return Err(anyhow!("invalid PocketBase id"));
}
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let existing_used_by = fetch_invite_used_by(state, pb_url, &token, invite_id).await?;
let invite = fetch_invite_record(state, pb_url, &token, invite_id).await?;
let existing_used_by = invite["used_by_id"].as_str().unwrap_or_default();
if existing_used_by == user_id {
return Ok(());
}
if !existing_used_by.is_empty() {
return Err(anyhow!("referral invite already used by another account"));
}
let reserved_by_id = invite["reserved_by_id"].as_str().unwrap_or_default();
let reserved_checkout_id = invite["reserved_checkout_id"].as_str().unwrap_or_default();
if !reserved_by_id.is_empty() && reserved_by_id != user_id {
return Err(anyhow!("referral invite reserved by another account"));
}
if !reserved_checkout_id.is_empty() && reserved_checkout_id != reservation_id {
return Err(anyhow!("referral invite reserved by another checkout"));
}
let url = format!("{pb_url}/api/collections/invites/records/{invite_id}");
let resp = state
@ -368,6 +586,9 @@ pub async fn mark_referral_invite_used(
.json(&serde_json::json!({
"used_by_id": user_id,
"used_at": now_unix_secs().to_string(),
"reserved_by_id": "",
"reserved_checkout_id": "",
"reserved_until_unix": 0,
}))
.send()
.await?;
@ -377,12 +598,12 @@ pub async fn mark_referral_invite_used(
.context("PocketBase invite usage update failed")
}
async fn fetch_invite_used_by(
async fn fetch_invite_record(
state: &AppState,
pb_url: &str,
token: &str,
invite_id: &str,
) -> anyhow::Result<String> {
) -> anyhow::Result<Value> {
let url = format!("{pb_url}/api/collections/invites/records/{invite_id}");
let resp = state
.http_client
@ -393,8 +614,98 @@ async fn fetch_invite_used_by(
ensure_success_ref(&resp).await?;
let body: Value = resp.json().await?;
Ok(body["used_by_id"].as_str().unwrap_or_default().to_string())
resp.json().await.map_err(Into::into)
}
async fn reserve_referral_invite(
state: &AppState,
invite_id: &str,
user_id: &str,
reservation_id: &str,
reserved_until_unix: u64,
) -> anyhow::Result<()> {
if !is_safe_pocketbase_id(invite_id)
|| !is_safe_pocketbase_id(user_id)
|| !is_safe_pocketbase_id(reservation_id)
{
return Err(anyhow!("invalid PocketBase id"));
}
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let invite = fetch_invite_record(state, pb_url, &token, invite_id).await?;
let used_by = invite["used_by_id"].as_str().unwrap_or_default();
if !used_by.is_empty() {
return Err(anyhow!("referral invite already used"));
}
let now = now_unix_secs();
let reserved_by_id = invite["reserved_by_id"].as_str().unwrap_or_default();
let reserved_checkout_id = invite["reserved_checkout_id"].as_str().unwrap_or_default();
let existing_reserved_until = number_field(&invite, "reserved_until_unix").unwrap_or(0);
let reservation_is_live = existing_reserved_until >= now;
if reservation_is_live
&& !reserved_checkout_id.is_empty()
&& reserved_checkout_id != reservation_id
{
return Err(anyhow!("referral invite already has an active checkout"));
}
if reservation_is_live && !reserved_by_id.is_empty() && reserved_by_id != user_id {
return Err(anyhow!("referral invite reserved by another account"));
}
let url = format!("{pb_url}/api/collections/invites/records/{invite_id}");
let resp = state
.http_client
.patch(&url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({
"reserved_by_id": user_id,
"reserved_checkout_id": reservation_id,
"reserved_until_unix": reserved_until_unix,
}))
.send()
.await?;
ensure_success(resp)
.await
.context("PocketBase invite reservation update failed")
}
async fn release_referral_invite_reservation(
state: &AppState,
invite_id: &str,
reservation_id: &str,
) -> anyhow::Result<()> {
if !is_safe_pocketbase_id(invite_id) || !is_safe_pocketbase_id(reservation_id) {
return Err(anyhow!("invalid PocketBase id"));
}
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let invite = fetch_invite_record(state, pb_url, &token, invite_id).await?;
let used_by = invite["used_by_id"].as_str().unwrap_or_default();
let reserved_checkout_id = invite["reserved_checkout_id"].as_str().unwrap_or_default();
if !used_by.is_empty() || reserved_checkout_id != reservation_id {
return Ok(());
}
let url = format!("{pb_url}/api/collections/invites/records/{invite_id}");
let resp = state
.http_client
.patch(&url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({
"reserved_by_id": "",
"reserved_checkout_id": "",
"reserved_until_unix": 0,
}))
.send()
.await?;
ensure_success(resp)
.await
.context("PocketBase invite reservation release failed")
}
pub async fn active_referral_checkout_user(
@ -457,8 +768,8 @@ async fn count_active_pending_checkouts(state: &AppState, now: u64) -> anyhow::R
async fn find_active_checkout_for_user(
state: &AppState,
user_id: &str,
discount_coupon_id: &str,
referral_invite_id: &str,
_discount_coupon_id: &str,
_referral_invite_id: &str,
now: u64,
) -> anyhow::Result<Option<PendingCheckout>> {
if !is_safe_pocketbase_id(user_id) {
@ -468,8 +779,8 @@ async fn find_active_checkout_for_user(
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let filter = format!(
"status=\"pending\" && expires_at_unix>={now} && user=\"{}\" && discount_coupon_id=\"{}\" && referral_invite_id=\"{}\"",
user_id, discount_coupon_id, referral_invite_id
"status=\"pending\" && expires_at_unix>={now} && user=\"{}\"",
user_id
);
let url = format!(
"{pb_url}/api/collections/{CHECKOUT_COLLECTION}/records?filter={}&perPage=1",
@ -515,13 +826,28 @@ async fn expire_stale_pending_checkouts(state: &AppState, now: u64) -> anyhow::R
return Ok(());
};
for id in items.iter().filter_map(|item| item["id"].as_str()) {
for item in items {
let Some(id) = item["id"].as_str() else {
continue;
};
if let Err(err) = mark_checkout_status(state, id, "expired").await {
warn!(
reservation_id = id,
"Failed to expire checkout reservation: {err}"
);
}
if let Some(invite_id) = item["referral_invite_id"]
.as_str()
.filter(|invite_id| !invite_id.is_empty())
{
if let Err(err) = release_referral_invite_reservation(state, invite_id, id).await {
warn!(
reservation_id = id,
referral_invite_id = invite_id,
"Failed to release expired referral invite reservation: {err}"
);
}
}
}
Ok(())
@ -551,6 +877,7 @@ async fn create_pending_checkout(
.json(&serde_json::json!({
"user": input.user_id,
"stripe_session_id": "",
"stripe_payment_intent_id": "",
"checkout_url": "",
"amount_pence": input.amount_pence,
"expected_total_pence": input.expected_total_pence,
@ -561,6 +888,7 @@ async fn create_pending_checkout(
"expires_at_unix": input.expires_at_unix,
"paid_amount_pence": 0,
"completed_at_unix": "",
"reversal_reason": "",
}))
.send()
.await?;
@ -574,6 +902,53 @@ async fn create_pending_checkout(
.ok_or_else(|| anyhow!("PocketBase checkout reservation missing id"))
}
/// Fetch a Stripe coupon and ensure its `percent_off` matches the expected
/// referral discount AND that it has no `amount_off` override. This blocks a
/// misconfigured (or maliciously swapped) coupon ID from quietly granting a
/// larger discount than the server's pricing math assumed.
async fn verify_stripe_coupon_discount(state: &AppState, coupon_id: &str) -> anyhow::Result<()> {
if !is_safe_stripe_session_id(coupon_id) {
return Err(anyhow!("unsafe stripe coupon id"));
}
let url = format!(
"https://api.stripe.com/v1/coupons/{}",
urlencoding::encode(coupon_id)
);
let resp = state
.http_client
.get(&url)
.basic_auth(&state.stripe_secret_key, None::<&str>)
.send()
.await
.context("Stripe coupon fetch failed")?;
ensure_success_ref(&resp)
.await
.context("Stripe coupon fetch returned error")?;
let body: Value = resp
.json()
.await
.context("Failed to parse Stripe coupon response")?;
let valid = body["valid"].as_bool().unwrap_or(false);
if !valid {
return Err(anyhow!("stripe coupon is not valid"));
}
if body["amount_off"].is_number() {
return Err(anyhow!(
"stripe coupon uses amount_off; only percent_off is permitted"
));
}
let percent_off = body["percent_off"]
.as_f64()
.ok_or_else(|| anyhow!("stripe coupon missing percent_off"))?;
if percent_off.is_nan() || (percent_off - REFERRAL_DISCOUNT_PERCENT as f64).abs() > 0.001 {
return Err(anyhow!(
"stripe coupon percent_off ({percent_off}) does not match expected {REFERRAL_DISCOUNT_PERCENT}"
));
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn create_stripe_session(
state: &AppState,
@ -585,6 +960,10 @@ async fn create_stripe_session(
expires_at_unix: u64,
discount_coupon_id: Option<&str>,
) -> anyhow::Result<(String, String)> {
if let Some(coupon_id) = discount_coupon_id.filter(|id| !id.is_empty()) {
verify_stripe_coupon_discount(state, coupon_id).await?;
}
let mut form_params = vec![
("mode", "payment".to_string()),
("payment_method_types[0]", "card".to_string()),
@ -697,6 +1076,31 @@ async fn mark_checkout_status(
.with_context(|| format!("PocketBase checkout status update failed for {reservation_id}"))
}
async fn mark_checkout_reversed(
state: &AppState,
reservation_id: &str,
status: &str,
reason: &str,
) -> anyhow::Result<()> {
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let url = format!("{pb_url}/api/collections/{CHECKOUT_COLLECTION}/records/{reservation_id}");
let resp = state
.http_client
.patch(&url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({
"status": status,
"reversal_reason": reason,
}))
.send()
.await?;
ensure_success(resp)
.await
.with_context(|| format!("PocketBase checkout reversal update failed for {reservation_id}"))
}
async fn find_checkout_by_stripe_session(
state: &AppState,
stripe_session_id: &str,
@ -726,6 +1130,35 @@ async fn find_checkout_by_stripe_session(
item.map(parse_pending_checkout).transpose()
}
async fn find_checkout_by_payment_intent(
state: &AppState,
payment_intent_id: &str,
) -> anyhow::Result<Option<PendingCheckout>> {
let token = get_superuser_token(state).await?;
let pb_url = state.pocketbase_url.trim_end_matches('/');
let filter = format!("stripe_payment_intent_id=\"{}\"", payment_intent_id);
let url = format!(
"{pb_url}/api/collections/{CHECKOUT_COLLECTION}/records?filter={}&perPage=1",
urlencoding::encode(&filter)
);
let resp = state
.http_client
.get(&url)
.header("Authorization", format!("Bearer {token}"))
.send()
.await?;
ensure_success_ref(&resp).await?;
let body: Value = resp.json().await?;
let item = body["items"]
.as_array()
.and_then(|items| items.first())
.cloned();
item.map(parse_pending_checkout).transpose()
}
fn parse_pending_checkout(item: Value) -> anyhow::Result<PendingCheckout> {
Ok(PendingCheckout {
id: item["id"]
@ -740,6 +1173,10 @@ fn parse_pending_checkout(item: Value) -> anyhow::Result<PendingCheckout> {
.as_str()
.unwrap_or_default()
.to_string(),
stripe_payment_intent_id: item["stripe_payment_intent_id"]
.as_str()
.unwrap_or_default()
.to_string(),
checkout_url: item["checkout_url"]
.as_str()
.unwrap_or_default()

221
server-rs/src/data/property.rs
View file

@ -15,6 +15,18 @@ const ADDRESS_SEARCH_MAX_POSTINGS_PER_TOKEN: usize = 250_000;
const ADDRESS_SEARCH_PREFIX_MIN_LEN: usize = 4;
const ADDRESS_SEARCH_PREFIX_MAX_LEN: usize = 8;
const NO_POI_METRIC_ROW: u32 = u32::MAX;
const MISSING_COORDINATE_SAMPLE_LIMIT: usize = 10;
const COUNTRY_COLUMN_CANDIDATES: &[&str] = &[
"ctry25cd",
"ctry24cd",
"ctry23cd",
"ctry22cd",
"country_code",
"Country code",
"country",
"Country",
];
const ENGLAND_COUNTRY_VALUES: &[&str] = &["E92000001", "England", "ENGLAND", "england"];
fn is_numeric_dtype(dtype: &DataType) -> bool {
matches!(
@ -38,6 +50,110 @@ fn is_datetime_dtype(dtype: &DataType) -> bool {
matches!(dtype, DataType::Datetime(_, _) | DataType::Date)
}
fn find_country_column(schema: &Schema) -> Option<String> {
COUNTRY_COLUMN_CANDIDATES
.iter()
.find_map(|&name| match schema.get(name) {
Some(dtype) if matches!(dtype, DataType::String) || dtype.is_categorical() => {
Some(name.to_string())
}
_ => None,
})
}
fn england_country_expr(country_column: &str) -> Expr {
ENGLAND_COUNTRY_VALUES.iter().skip(1).fold(
col(country_column)
.cast(DataType::String)
.eq(lit(ENGLAND_COUNTRY_VALUES[0])),
|expr, value| expr.or(col(country_column).cast(DataType::String).eq(lit(*value))),
)
}
#[derive(Debug, PartialEq, Eq)]
struct MissingCoordinateSummary {
row_count: usize,
unique_postcode_count: usize,
sample_postcodes: Vec<String>,
}
fn summarize_missing_coordinate_postcodes<'a>(
postcodes: impl IntoIterator<Item = Option<&'a str>>,
) -> MissingCoordinateSummary {
let mut row_count = 0usize;
let mut unique_postcodes = FxHashSet::default();
for postcode in postcodes {
row_count += 1;
if let Some(postcode) = postcode {
let trimmed = postcode.trim();
if !trimmed.is_empty() {
unique_postcodes.insert(trimmed.to_string());
}
}
}
let mut sample_postcodes: Vec<String> = unique_postcodes.iter().cloned().collect();
sample_postcodes.sort_unstable();
sample_postcodes.truncate(MISSING_COORDINATE_SAMPLE_LIMIT);
MissingCoordinateSummary {
row_count,
unique_postcode_count: unique_postcodes.len(),
sample_postcodes,
}
}
fn missing_england_coordinates_error(
summary: &MissingCoordinateSummary,
country_column: &str,
) -> String {
let samples = if summary.sample_postcodes.is_empty() {
"none".to_string()
} else {
summary.sample_postcodes.join(", ")
};
format!(
"England property rows missing postcode coordinates after joining postcode data: {} rows across {} postcodes (country column '{}'). Sample postcodes: {}",
summary.row_count, summary.unique_postcode_count, country_column, samples
)
}
fn validate_no_england_rows_missing_coordinates(
combined_lf: &LazyFrame,
schema: &Schema,
) -> anyhow::Result<()> {
let Some(country_column) = find_country_column(schema) else {
bail!(
"Postcode feature parquet has no reliable country column; cannot verify that rows with missing coordinates are outside England. Regenerate postcode.parquet with pipeline.transform.merge so it includes ctry25cd."
);
};
let missing_coordinates = col("lat").is_null().or(col("lon").is_null());
let offending_df = combined_lf
.clone()
.filter(missing_coordinates.and(england_country_expr(&country_column)))
.select([col("Postcode")])
.collect()
.context("Failed to validate missing postcode coordinates")?;
let postcode_column = offending_df
.column("Postcode")
.context("Joined frame missing 'Postcode' during coordinate validation")?
.str()
.context("'Postcode' column is not a string during coordinate validation")?;
let summary = summarize_missing_coordinate_postcodes(postcode_column);
if summary.row_count > 0 {
bail!(
"{}",
missing_england_coordinates_error(&summary, &country_column)
);
}
Ok(())
}
#[derive(Clone, Debug)]
struct AddressTermGroup {
alternatives: Vec<String>,
@ -1307,6 +1423,7 @@ impl PropertyData {
.clone()
.collect_schema()
.context("Failed to collect joined schema")?;
validate_no_england_rows_missing_coordinates(&combined_lf, &schema)?;
let numeric_names: Vec<String> = configured_numeric_names
.iter()
.map(|name| (*name).to_string())
@ -1974,6 +2091,110 @@ mod tests {
Bounds::Percentile { low, high }
}
#[test]
fn country_column_detection_prefers_reliable_country_code() {
let df = df!(
"Postcode" => &["SW1A 1AA"],
"ctry25cd" => &["E92000001"],
"lat" => &[Some(51.501_f64)],
"lon" => &[Some(-0.141_f64)],
)
.expect("test dataframe should build");
assert_eq!(
find_country_column(df.schema()).as_deref(),
Some("ctry25cd")
);
}
#[test]
fn missing_coordinate_summary_counts_rows_and_distinct_samples() {
let summary = summarize_missing_coordinate_postcodes([
Some("SW1A 1AA"),
Some("SW1A 1AA"),
Some("E14 2DG"),
Some(""),
None,
]);
assert_eq!(
summary,
MissingCoordinateSummary {
row_count: 5,
unique_postcode_count: 2,
sample_postcodes: vec!["E14 2DG".to_string(), "SW1A 1AA".to_string()],
}
);
}
#[test]
fn missing_england_coordinates_error_includes_counts_and_samples() {
let summary = MissingCoordinateSummary {
row_count: 3,
unique_postcode_count: 2,
sample_postcodes: vec!["E14 2DG".to_string(), "SW1A 1AA".to_string()],
};
let message = missing_england_coordinates_error(&summary, "ctry25cd");
assert!(message.contains("3 rows across 2 postcodes"));
assert!(message.contains("country column 'ctry25cd'"));
assert!(message.contains("E14 2DG, SW1A 1AA"));
}
#[test]
fn coordinate_validation_errors_for_england_rows() {
let lf = df!(
"Postcode" => &["SW1A 1AA", "E14 2DG"],
"ctry25cd" => &["E92000001", "E92000001"],
"lat" => &[Some(51.501_f64), None],
"lon" => &[Some(-0.141_f64), Some(-0.001_f64)],
)
.expect("test dataframe should build")
.lazy();
let schema = lf.clone().collect_schema().expect("schema should collect");
let err = validate_no_england_rows_missing_coordinates(&lf, &schema)
.expect_err("England row with missing coordinate should error");
let message = err.to_string();
assert!(message.contains("1 rows across 1 postcodes"));
assert!(message.contains("E14 2DG"));
}
#[test]
fn coordinate_validation_allows_non_england_rows() {
let lf = df!(
"Postcode" => &["CF10 1AA", "SW1A 1AA"],
"ctry25cd" => &["W92000004", "E92000001"],
"lat" => &[None, Some(51.501_f64)],
"lon" => &[Some(-3.179_f64), Some(-0.141_f64)],
)
.expect("test dataframe should build")
.lazy();
let schema = lf.clone().collect_schema().expect("schema should collect");
validate_no_england_rows_missing_coordinates(&lf, &schema)
.expect("non-England row with missing coordinate should be skipped");
}
#[test]
fn coordinate_validation_requires_country_column() {
let lf = df!(
"Postcode" => &["SW1A 1AA"],
"lat" => &[None::<f64>],
"lon" => &[Some(-0.141_f64)],
)
.expect("test dataframe should build")
.lazy();
let schema = lf.clone().collect_schema().expect("schema should collect");
let err = validate_no_england_rows_missing_coordinates(&lf, &schema)
.expect_err("missing country provenance should error");
assert!(err.to_string().contains("no reliable country column"));
}
#[test]
fn full_postcode_detection_accepts_common_formats() {
assert!(is_full_postcode_compact("SW1A1AA"));

4
server-rs/src/parsing.rs
View file

@ -3,9 +3,7 @@ mod fields;
mod filters;
mod h3;
pub use bounds::{
bounds_intersect, h3_cell_bounds, parse_bounds, require_bounds, require_candidate_count,
};
pub use bounds::{bounds_intersect, h3_cell_bounds, parse_bounds, require_bounds};
pub use fields::{
parse_enum_dist, parse_field_indices, parse_field_indices_with_poi, parse_field_set,
};

155
server-rs/src/pocketbase.rs
View file

@ -529,6 +529,14 @@ async fn ensure_checkout_sessions_fields(
"stripe_session_id",
serde_json::json!({ "name": "stripe_session_id", "type": "text", "required": false }),
);
add_field(
"stripe_payment_intent_id",
serde_json::json!({ "name": "stripe_payment_intent_id", "type": "text", "required": false }),
);
add_field(
"reversal_reason",
serde_json::json!({ "name": "reversal_reason", "type": "text", "required": false }),
);
add_field(
"checkout_url",
serde_json::json!({ "name": "checkout_url", "type": "text", "required": false }),
@ -591,6 +599,66 @@ async fn ensure_checkout_sessions_fields(
Ok(())
}
async fn ensure_invites_fields(client: &Client, base_url: &str, token: &str) -> anyhow::Result<()> {
let url = format!("{base_url}/api/collections/invites");
let resp = client
.get(&url)
.header("Authorization", format!("Bearer {token}"))
.send()
.await?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
anyhow::bail!("Failed to fetch invites collection ({status}): {text}");
}
let body: serde_json::Value = resp.json().await?;
let fields = body["fields"]
.as_array()
.ok_or_else(|| anyhow::anyhow!("invites collection has no fields array"))?;
let mut new_fields = fields.clone();
let mut add_field = |name: &str, field: serde_json::Value| {
if !fields.iter().any(|f| f["name"] == name) {
new_fields.push(field);
}
};
add_field(
"reserved_by_id",
serde_json::json!({ "name": "reserved_by_id", "type": "text", "required": false }),
);
add_field(
"reserved_checkout_id",
serde_json::json!({ "name": "reserved_checkout_id", "type": "text", "required": false }),
);
add_field(
"reserved_until_unix",
serde_json::json!({ "name": "reserved_until_unix", "type": "number" }),
);
if new_fields.len() == fields.len() {
return Ok(());
}
let patch_resp = client
.patch(&url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({ "fields": new_fields }))
.send()
.await?;
if !patch_resp.status().is_success() {
let status = patch_resp.status();
let text = patch_resp.text().await.unwrap_or_default();
anyhow::bail!("Failed to patch invites fields ({status}): {text}");
}
info!("PocketBase invites collection fields updated");
Ok(())
}
async fn ensure_checkout_locks_fields(
client: &Client,
base_url: &str,
@ -655,6 +723,72 @@ async fn ensure_checkout_locks_fields(
Ok(())
}
async fn ensure_short_urls_fields(
client: &Client,
base_url: &str,
token: &str,
) -> anyhow::Result<()> {
let url = format!("{base_url}/api/collections/short_urls");
let resp = client
.get(&url)
.header("Authorization", format!("Bearer {token}"))
.send()
.await?;
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
anyhow::bail!("Failed to fetch short_urls collection ({status}): {text}");
}
let body: serde_json::Value = resp.json().await?;
let fields = body["fields"]
.as_array()
.ok_or_else(|| anyhow::anyhow!("short_urls collection has no fields array"))?;
let mut new_fields = fields.clone();
let mut add_field = |name: &str, field: serde_json::Value| {
if !fields.iter().any(|f| f["name"] == name) {
new_fields.push(field);
}
};
add_field(
"created_by",
serde_json::json!({ "name": "created_by", "type": "text", "required": false }),
);
add_field(
"click_count",
serde_json::json!({ "name": "click_count", "type": "number" }),
);
for field in ["share_south", "share_west", "share_north", "share_east"] {
add_field(
field,
serde_json::json!({ "name": field, "type": "number" }),
);
}
if new_fields.len() == fields.len() {
return Ok(());
}
let patch_resp = client
.patch(&url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({ "fields": new_fields }))
.send()
.await?;
if !patch_resp.status().is_success() {
let status = patch_resp.status();
let text = patch_resp.text().await.unwrap_or_default();
anyhow::bail!("Failed to patch short_urls fields ({status}): {text}");
}
info!("PocketBase short_urls collection fields updated");
Ok(())
}
async fn ensure_collection_indexes(
client: &Client,
base_url: &str,
@ -999,6 +1133,9 @@ pub async fn ensure_collections(
Field::text("invite_type", true),
Field::text("used_by_id", false),
Field::text("used_at", false),
Field::text("reserved_by_id", false),
Field::text("reserved_checkout_id", false),
Field::number("reserved_until_unix"),
Field::autodate("created", true, false),
Field::autodate("updated", true, true),
],
@ -1011,7 +1148,10 @@ pub async fn ensure_collections(
},
)
.await?;
ensure_server_only_rules(client, base_url, &token, "invites").await?;
} else {
ensure_server_only_rules(client, base_url, &token, "invites").await?;
ensure_invites_fields(client, base_url, &token).await?;
ensure_autodate_fields(client, base_url, &token, "invites").await?;
}
@ -1027,6 +1167,7 @@ pub async fn ensure_collections(
fields: vec![
Field::relation("user", &users_id),
Field::text("stripe_session_id", false),
Field::text("stripe_payment_intent_id", false),
Field::text("checkout_url", false),
Field::number("amount_pence"),
Field::number("expected_total_pence"),
@ -1037,6 +1178,7 @@ pub async fn ensure_collections(
Field::number("expires_at_unix"),
Field::number("paid_amount_pence"),
Field::text("completed_at_unix", false),
Field::text("reversal_reason", false),
Field::autodate("created", true, false),
Field::autodate("updated", true, true),
],
@ -1106,6 +1248,12 @@ pub async fn ensure_collections(
fields: vec![
Field::text("code", true),
Field::text("params", true),
Field::text("created_by", false),
Field::number("click_count"),
Field::number("share_south"),
Field::number("share_west"),
Field::number("share_north"),
Field::number("share_east"),
Field::autodate("created", true, false),
Field::autodate("updated", true, true),
],
@ -1118,7 +1266,10 @@ pub async fn ensure_collections(
},
)
.await?;
ensure_server_only_rules(client, base_url, &token, "short_urls").await?;
} else {
ensure_server_only_rules(client, base_url, &token, "short_urls").await?;
ensure_short_urls_fields(client, base_url, &token).await?;
ensure_autodate_fields(client, base_url, &token, "short_urls").await?;
}
@ -1148,7 +1299,9 @@ pub async fn ensure_collections(
},
)
.await?;
ensure_server_only_rules(client, base_url, &token, "location_logs").await?;
} else {
ensure_server_only_rules(client, base_url, &token, "location_logs").await?;
ensure_autodate_fields(client, base_url, &token, "location_logs").await?;
}
@ -1181,7 +1334,9 @@ pub async fn ensure_collections(
},
)
.await?;
ensure_server_only_rules(client, base_url, &token, "ai_query_logs").await?;
} else {
ensure_server_only_rules(client, base_url, &token, "ai_query_logs").await?;
ensure_autodate_fields(client, base_url, &token, "ai_query_logs").await?;
}

2
server-rs/src/routes.rs
View file

@ -50,7 +50,7 @@ pub use pricing::get_pricing;
pub use properties::get_hexagon_properties;
pub use rightmove::get_rightmove_redirect;
pub use screenshot::{fetch_screenshot_bytes, get_screenshot};
pub use shorten::{get_short_url, post_shorten};
pub use shorten::{get_share_links, get_short_url, post_shorten};
pub use streetview::get_streetview;
pub use stripe_webhook::post_stripe_webhook;
pub use telemetry::post_telemetry;

164
server-rs/src/routes/export.rs
View file

@ -4,7 +4,7 @@ use std::sync::Arc;
use std::time::Duration;
use axum::extract::{Query, State};
use axum::http::{header, HeaderMap, StatusCode};
use axum::http::{header, HeaderMap, StatusCode, Uri};
use axum::response::IntoResponse;
use axum::Extension;
use rust_xlsxwriter::{Format, FormatAlign, FormatBorder, Image, Url, Workbook};
@ -16,11 +16,14 @@ use crate::auth::OptionalUser;
use crate::consts::NAN_U16;
use crate::data::{PostcodePoiMetrics, QuantRef};
use crate::features;
use crate::licensing::check_license_bounds;
use crate::licensing::{check_license_bounds, resolve_share_code};
use crate::parsing::{
parse_field_indices_with_poi, parse_filters_with_poi, require_bounds, row_passes_filters,
row_passes_poi_filters,
};
use crate::routes::travel_time::{
load_travel_data, parse_optional_travel, row_passes_travel_filters,
};
use crate::routes::{fetch_screenshot_bytes, FeatureInfo};
use crate::state::SharedState;
@ -29,11 +32,20 @@ const EXPORT_SCREENSHOT_TIMEOUT_SECS: u64 = 12;
/// Height (in pixels) reserved for the screenshot row
const IMAGE_ROW_HEIGHT: f64 = 225.0;
/// Hard cap on the bounding-box area (in degrees²) that may be exported.
/// All of England fits inside ~6° × ~10° ≈ 60 deg². Anything substantially
/// larger is rejected to keep aggregation bounded for non-licensed users
/// who supply share grants outside their expected region, and to avoid
/// minutes-long requests that fan out to millions of rows.
const MAX_EXPORT_BBOX_AREA_DEG2: f64 = 80.0;
#[derive(Deserialize)]
pub struct ExportParams {
bounds: Option<String>,
filters: Option<String>,
travel: Option<String>,
fields: Option<String>,
share: Option<String>,
}
/// Per-postcode accumulator for export aggregation (mean for numeric, mode for enum).
@ -125,6 +137,8 @@ fn build_frontend_params(
center_lon: f64,
zoom: f64,
filters_str: Option<&str>,
travel_params: &[String],
share: Option<&str>,
) -> String {
let mut parts = vec![
format!("lat={:.4}", center_lat),
@ -140,20 +154,53 @@ fn build_frontend_params(
}
}
}
for entry in travel_params {
if !entry.is_empty() {
parts.push(format!("tt={}", urlencoding::encode(entry.trim())));
}
}
if let Some(share) = share.filter(|value| !value.is_empty()) {
parts.push(format!("share={}", urlencoding::encode(share)));
}
parts.join("&")
}
fn collect_travel_state_params(query: Option<&str>) -> Vec<String> {
query
.into_iter()
.flat_map(|qs| url::form_urlencoded::parse(qs.as_bytes()))
.filter_map(|(key, value)| {
if key == "tt" && !value.is_empty() {
Some(value.into_owned())
} else {
None
}
})
.collect()
}
pub async fn get_export(
State(shared): State<Arc<SharedState>>,
headers: HeaderMap,
Extension(user): Extension<OptionalUser>,
uri: Uri,
Query(params): Query<ExportParams>,
) -> Result<impl IntoResponse, axum::response::Response> {
let state = shared.load_state();
let (south, west, north, east) =
require_bounds(params.bounds).map_err(IntoResponse::into_response)?;
check_license_bounds(&user.0, (south, west, north, east), None)?;
let area_deg2 = (north - south).max(0.0) * (east - west).max(0.0);
if area_deg2 > MAX_EXPORT_BBOX_AREA_DEG2 {
return Err((
StatusCode::BAD_REQUEST,
"Export area is too large; zoom in further before exporting",
)
.into_response());
}
let share_bounds = resolve_share_code(&state, params.share.as_deref()).await;
check_license_bounds(&user.0, (south, west, north, east), share_bounds)?;
let quant = state.data.quant_ref();
let poi_quant = state.data.poi_metrics.quant_ref();
@ -168,7 +215,14 @@ pub async fn get_export(
.map_err(|err| (StatusCode::BAD_REQUEST, err).into_response())?;
let has_poi_filters = !parsed_poi_filters.is_empty();
let filters_str = params.filters;
let travel_entries = parse_optional_travel(params.travel.as_deref())
.map_err(|err| (StatusCode::BAD_REQUEST, err).into_response())?;
let has_travel_filters = travel_entries
.iter()
.any(|entry| entry.filter_min.is_some() && entry.filter_max.is_some());
let travel_state_params = collect_travel_state_params(uri.query());
let fields_str = params.fields;
let share_code = params.share;
let public_url = state.public_url.clone();
@ -181,8 +235,14 @@ pub async fn get_export(
} else {
12.0
};
let frontend_params =
build_frontend_params(center_lat, center_lon, zoom, filters_str.as_deref());
let frontend_params = build_frontend_params(
center_lat,
center_lon,
zoom,
filters_str.as_deref(),
&travel_state_params,
share_code.as_deref(),
);
// Fetch screenshot (async, before spawn_blocking)
let auth_header = headers.get(header::AUTHORIZATION);
@ -235,14 +295,17 @@ pub async fn get_export(
let enum_values = &state.data.enum_values;
let postcode_data = &state.postcode_data;
let poi_metrics = &state.data.poi_metrics;
let travel_data = load_travel_data(&state.travel_time_store, &travel_entries)?;
let poi_offset = num_features;
let total_export_features = num_features + poi_metrics.num_features();
let (pc_interner, pc_keys) = state.data.postcode_parts();
// Build set of enum feature indices for quick lookup
let enum_indices: FxHashMap<usize, ()> = enum_values.keys().map(|&idx| (idx, ())).collect();
// Group rows by postcode
let mut postcode_rows: FxHashMap<usize, Vec<usize>> = FxHashMap::default();
// Aggregate directly by postcode so large requests don't retain every
// matching property row before sampling the exported postcodes.
let mut postcode_aggs: FxHashMap<usize, PostcodeExportAgg> = FxHashMap::default();
state
.grid
.for_each_in_bounds(south, west, north, east, |row_idx| {
@ -260,31 +323,31 @@ pub async fn get_export(
{
return;
}
let postcode = state.data.postcode(row);
let postcode = pc_interner.resolve(&pc_keys[row]);
if has_travel_filters
&& !row_passes_travel_filters(postcode, &travel_entries, &travel_data)
{
return;
}
if let Some(&pc_idx) = postcode_data.postcode_to_idx.get(postcode) {
postcode_rows.entry(pc_idx).or_default().push(row);
postcode_aggs
.entry(pc_idx)
.or_insert_with(|| PostcodeExportAgg::new(total_export_features))
.add_row(
feature_data,
row,
num_features,
&enum_indices,
&quant,
poi_metrics,
);
}
});
// Aggregate per postcode
let mut postcode_aggs: Vec<(usize, PostcodeExportAgg)> =
Vec::with_capacity(postcode_rows.len());
for (pc_idx, rows) in postcode_rows {
let mut agg = PostcodeExportAgg::new(total_export_features);
for &row in &rows {
agg.add_row(
feature_data,
row,
num_features,
&enum_indices,
&quant,
poi_metrics,
);
}
if agg.count > 0 {
postcode_aggs.push((pc_idx, agg));
}
}
let mut postcode_aggs: Vec<(usize, PostcodeExportAgg)> = postcode_aggs
.into_iter()
.filter(|(_, agg)| agg.count > 0)
.collect();
// Sort by property count descending
postcode_aggs.sort_unstable_by_key(|agg| std::cmp::Reverse(agg.1.count));
@ -460,7 +523,11 @@ pub async fn get_export(
.set_align(FormatAlign::Left);
// Dashboard URL
let dashboard_url = format!("{}/?{}", public_url, frontend_params);
let dashboard_url = format!(
"{}/dashboard?{}",
public_url.trim_end_matches('/'),
frontend_params
);
// Sheet 1: "Selected" (filter features only) with link + screenshot
// Sheet 2: "All Data" (all features)
@ -680,3 +747,42 @@ pub async fn get_export(
bytes,
))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn collect_travel_state_params_accepts_single_tt_param() {
let entry = "transit:bank-tube-station:Bank%20tube%20station:0:52";
let query = format!("bounds=1,2,3,4&tt={}", urlencoding::encode(entry));
assert_eq!(collect_travel_state_params(Some(&query)), vec![entry]);
}
#[test]
fn collect_travel_state_params_preserves_repeated_tt_params() {
let bank = "transit:bank-tube-station:Bank%20tube%20station:0:52";
let kings_cross = "transit:kings-cross:Kings%20Cross:b:0:30";
let query = format!(
"tt={}&filter=Price%3A0%3A100&tt={}",
urlencoding::encode(bank),
urlencoding::encode(kings_cross)
);
assert_eq!(
collect_travel_state_params(Some(&query)),
vec![bank, kings_cross]
);
}
#[test]
fn export_query_deserializes_when_tt_is_a_single_string() {
let uri: Uri = "/api/export?bounds=1,2,3,4&tt=transit%3Abank%3ABank%2520station%3A0%3A52"
.parse()
.unwrap();
let Query(params) = Query::<ExportParams>::try_from_uri(&uri).unwrap();
assert_eq!(params.bounds.as_deref(), Some("1,2,3,4"));
}
}

142
server-rs/src/routes/invites.rs
View file

@ -10,7 +10,8 @@ use tracing::{info, warn};
use crate::auth::{OptionalUser, PocketBaseUser};
use crate::checkout_sessions::{
active_referral_checkout_user, start_license_checkout, CheckoutStart,
active_referral_checkout_user, grant_license_with_pricing_lock, start_license_checkout,
CheckoutStart,
};
use crate::pocketbase::get_superuser_token;
use crate::pocketbase_locks::acquire_pocketbase_lock;
@ -107,6 +108,25 @@ fn validate_invite_code(code: &str) -> Result<(), &'static str> {
Ok(())
}
/// Sanitize the inviter's display name returned to anonymous clients.
/// The value comes from the inviter's email local-part stored in PocketBase;
/// we don't trust it, so strip control chars and HTML-meaningful characters
/// and cap the length. Returns None if nothing usable remains.
fn sanitize_invited_by(raw: &str) -> Option<String> {
const MAX_LEN: usize = 40;
let cleaned: String = raw
.chars()
.filter(|c| !c.is_control() && !matches!(*c, '<' | '>' | '"' | '\'' | '&' | '\\'))
.take(MAX_LEN)
.collect();
let trimmed = cleaned.trim();
if trimmed.is_empty() {
None
} else {
Some(trimmed.to_string())
}
}
fn generate_invite_code() -> String {
use rand::RngExt;
let mut rng = rand::rng();
@ -131,6 +151,48 @@ fn current_unix_secs_string() -> String {
.to_string()
}
/// Fetch the live `is_admin` flag for a user, bypassing any cached token
/// claims. Returns Err with an HTTP response if PocketBase is unreachable
/// or returns an unexpected payload — the caller should propagate that.
async fn verify_is_admin(
state: &AppState,
pb_url: &str,
token: &str,
user_id: &str,
) -> Result<bool, Response> {
if user_id.is_empty()
|| user_id.len() > 32
|| !user_id.bytes().all(|b| b.is_ascii_alphanumeric())
{
return Err(StatusCode::FORBIDDEN.into_response());
}
let url = format!("{pb_url}/api/collections/users/records/{user_id}");
let resp = match state
.http_client
.get(&url)
.header("Authorization", format!("Bearer {token}"))
.send()
.await
{
Ok(r) => r,
Err(err) => {
warn!("Failed to verify is_admin: {err}");
return Err(StatusCode::BAD_GATEWAY.into_response());
}
};
if !resp.status().is_success() {
return Err(StatusCode::BAD_GATEWAY.into_response());
}
let body: serde_json::Value = match resp.json().await {
Ok(v) => v,
Err(err) => {
warn!("Failed to parse user record for is_admin verify: {err}");
return Err(StatusCode::BAD_GATEWAY.into_response());
}
};
Ok(body["is_admin"].as_bool().unwrap_or(false))
}
async fn lookup_unused_invite(
state: &AppState,
pb_url: &str,
@ -217,35 +279,16 @@ async fn mark_invite_used(
async fn grant_license_for_invite(
state: &AppState,
pb_url: &str,
token: &str,
_pb_url: &str,
_token: &str,
user_id: &str,
) -> Result<(), Response> {
let update_url = format!("{pb_url}/api/collections/users/records/{user_id}");
let resp = match state
.http_client
.patch(&update_url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({ "subscription": "licensed" }))
.send()
grant_license_with_pricing_lock(state, user_id)
.await
{
Ok(resp) => resp,
Err(err) => {
.map_err(|err| {
warn!("Failed to update user subscription for admin invite: {err}");
return Err(StatusCode::BAD_GATEWAY.into_response());
}
};
if !resp.status().is_success() {
let status = resp.status();
let text = resp.text().await.unwrap_or_default();
warn!("PocketBase user subscription update failed ({status}): {text}");
return Err(StatusCode::BAD_GATEWAY.into_response());
}
state.token_cache.invalidate_by_user_id(user_id);
Ok(())
StatusCode::BAD_GATEWAY.into_response()
})
}
async fn create_referral_checkout(
@ -289,12 +332,32 @@ pub async fn post_invites(
None => return StatusCode::UNAUTHORIZED.into_response(),
};
let invite_type = if user.is_admin {
match body.invite_type.as_deref() {
Some("referral") => "referral",
_ => "admin",
// Cached token claims could be stale or, in the worst case, tampered with
// upstream of us. For admin-only actions, re-fetch the live record from
// PocketBase and trust only that.
let wants_admin_invite =
user.is_admin && !matches!(body.invite_type.as_deref(), Some("referral"));
let pb_url = state.pocketbase_url.trim_end_matches('/');
let token = match get_superuser_token(&state).await {
Ok(t) => t,
Err(err) => {
warn!("Failed to auth as PocketBase superuser: {err}");
return StatusCode::BAD_GATEWAY.into_response();
}
} else if user.subscription == "licensed" {
};
let invite_type = if wants_admin_invite {
match verify_is_admin(&state, pb_url, &token, &user.id).await {
Ok(true) => "admin",
Ok(false) => {
warn!(user_id = %user.id, "is_admin claim rejected by live PB lookup");
return (StatusCode::FORBIDDEN, "Not authorised").into_response();
}
Err(response) => return response,
}
} else if user.is_admin || user.subscription == "licensed" {
"referral"
} else {
return (
@ -305,15 +368,6 @@ pub async fn post_invites(
};
let code = generate_invite_code();
let pb_url = state.pocketbase_url.trim_end_matches('/');
let token = match get_superuser_token(&state).await {
Ok(t) => t,
Err(err) => {
warn!("Failed to auth as PocketBase superuser: {err}");
return StatusCode::BAD_GATEWAY.into_response();
}
};
let create_url = format!("{pb_url}/api/collections/invites/records");
let res = state
@ -429,7 +483,7 @@ pub async fn get_invite(
let used = !used_by.is_empty();
let created_by = invite["created_by"].as_str().unwrap_or("");
// Look up inviter's name (email local part)
// Look up inviter's name (email local part) — sanitized before returning.
let invited_by = if !created_by.is_empty() {
let user_url = format!("{pb_url}/api/collections/users/records/{created_by}");
match state
@ -444,7 +498,7 @@ pub async fn get_invite(
user_body["email"]
.as_str()
.and_then(|e| e.split('@').next())
.map(String::from)
.and_then(sanitize_invited_by)
}
_ => None,
}
@ -565,11 +619,11 @@ pub async fn post_redeem_invite(
};
if invite_type == "admin" {
if let Err(response) = mark_invite_used(&state, pb_url, &token, invite_id, &user.id).await {
if let Err(response) = grant_license_for_invite(&state, pb_url, &token, &user.id).await {
return response;
}
if let Err(response) = grant_license_for_invite(&state, pb_url, &token, &user.id).await {
if let Err(response) = mark_invite_used(&state, pb_url, &token, invite_id, &user.id).await {
return response;
}

359
server-rs/src/routes/shorten.rs
View file

@ -3,16 +3,24 @@ use std::sync::Arc;
use axum::extract::{Path, State};
use axum::http::{header, StatusCode};
use axum::response::{Html, IntoResponse, Response};
use axum::Extension;
use axum::Json;
use rand::RngExt;
use serde::{Deserialize, Serialize};
use tracing::warn;
use url::form_urlencoded;
use crate::auth::OptionalUser;
use crate::licensing::{is_valid_share_bounds, share_bounds_from_params, ShareBounds};
use crate::pocketbase::get_superuser_token;
use crate::state::SharedState;
const CODE_LEN: usize = 8;
const CHARSET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789";
const MAX_QUERY_LEN: usize = 4096;
const MAX_QUERY_PAIRS: usize = 80;
const MAX_PARAM_KEY_LEN: usize = 64;
const MAX_PARAM_VALUE_LEN: usize = 512;
fn generate_code() -> String {
let mut rng = rand::rng();
@ -36,15 +44,178 @@ pub struct ShortenResponse {
struct PbRecord {
code: String,
params: String,
#[serde(skip_serializing_if = "Option::is_none")]
created_by: Option<String>,
click_count: u64,
#[serde(skip_serializing_if = "Option::is_none")]
share_south: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
share_west: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
share_north: Option<f64>,
#[serde(skip_serializing_if = "Option::is_none")]
share_east: Option<f64>,
}
#[derive(Serialize)]
struct ShareLinkListItem {
code: String,
url: String,
og_image_url: String,
params: String,
click_count: u64,
created: String,
}
#[derive(Serialize)]
struct ShareLinksResponse {
links: Vec<ShareLinkListItem>,
}
fn json_number_as_u64(value: &serde_json::Value) -> u64 {
value
.as_u64()
.or_else(|| {
value
.as_f64()
.filter(|n| n.is_finite() && *n > 0.0)
.map(|n| n as u64)
})
.unwrap_or(0)
}
fn sanitized_query_params(params: &str, keep_share: bool) -> Result<String, &'static str> {
let params = params.trim_start_matches('?');
if params.len() > MAX_QUERY_LEN {
return Err("query string is too long");
}
let mut pairs = Vec::new();
for (idx, (key, value)) in form_urlencoded::parse(params.as_bytes()).enumerate() {
if idx >= MAX_QUERY_PAIRS {
return Err("query string has too many parameters");
}
if key == "share" && !keep_share {
continue;
}
if !is_allowed_param_key(&key) {
return Err("query string contains an unsupported parameter");
}
if key.len() > MAX_PARAM_KEY_LEN || value.len() > MAX_PARAM_VALUE_LEN {
return Err("query parameter is too long");
}
if key.chars().any(char::is_control) || value.chars().any(char::is_control) {
return Err("query parameter contains control characters");
}
pairs.push((key.into_owned(), value.into_owned()));
}
let mut out = form_urlencoded::Serializer::new(String::new());
for (key, value) in pairs {
out.append_pair(&key, &value);
}
Ok(out.finish())
}
fn is_allowed_param_key(key: &str) -> bool {
matches!(
key,
"lat"
| "lon"
| "zoom"
| "filter"
| "school"
| "crime"
| "voteShare"
| "ethnicity"
| "amenityDistance"
| "transportDistance"
| "amenityCount2km"
| "amenityCount5km"
| "poi"
| "tab"
| "pc"
| "tt"
| "share"
)
}
fn escape_attr(value: &str) -> String {
value
.replace('&', "&amp;")
.replace('"', "&quot;")
.replace('\'', "&#39;")
.replace('<', "&lt;")
.replace('>', "&gt;")
}
fn user_can_create_share_grant(user: &OptionalUser) -> bool {
user.0
.as_ref()
.is_some_and(|u| u.is_admin || u.subscription == "licensed")
}
fn share_fields(
bounds: Option<ShareBounds>,
) -> (Option<f64>, Option<f64>, Option<f64>, Option<f64>) {
match bounds {
Some(bounds) => (
Some(bounds.south),
Some(bounds.west),
Some(bounds.north),
Some(bounds.east),
),
None => (None, None, None, None),
}
}
fn record_share_bounds(item: &serde_json::Value) -> Option<ShareBounds> {
let bounds = ShareBounds {
south: item.get("share_south")?.as_f64()?,
west: item.get("share_west")?.as_f64()?,
north: item.get("share_north")?.as_f64()?,
east: item.get("share_east")?.as_f64()?,
};
is_valid_share_bounds(bounds).then_some(bounds)
}
fn dashboard_redirect_url(params: &str, code: &str, include_share: bool) -> String {
match (params.is_empty(), include_share) {
(true, false) => "/dashboard".to_string(),
(true, true) => format!("/dashboard?share={code}"),
(false, false) => format!("/dashboard?{params}"),
(false, true) => format!("/dashboard?{params}&share={code}"),
}
}
fn og_image_url(public_url: &str, params: &str) -> String {
if params.is_empty() {
format!("{}/api/screenshot?og=1", public_url.trim_end_matches('/'))
} else {
format!(
"{}/api/screenshot?og=1&{params}",
public_url.trim_end_matches('/')
)
}
}
pub async fn post_shorten(
State(shared): State<Arc<SharedState>>,
Extension(user): Extension<OptionalUser>,
Json(req): Json<ShortenRequest>,
) -> Response {
let state = shared.load_state();
let pb_url = state.pocketbase_url.trim_end_matches('/');
let can_create_share_grant = user_can_create_share_grant(&user);
let params = match sanitized_query_params(&req.params, !can_create_share_grant) {
Ok(params) => params,
Err(reason) => {
warn!("Rejected short URL params: {reason}");
return (StatusCode::BAD_REQUEST, reason).into_response();
}
};
let token = match get_superuser_token(&state).await {
Ok(t) => t,
Err(err) => {
@ -54,10 +225,22 @@ pub async fn post_shorten(
};
let code = generate_code();
let share_bounds = if can_create_share_grant {
share_bounds_from_params(&params)
} else {
None
};
let (share_south, share_west, share_north, share_east) = share_fields(share_bounds);
let record = PbRecord {
code: code.clone(),
params: req.params,
params,
created_by: user.0.as_ref().map(|u| u.id.clone()),
click_count: 0,
share_south,
share_west,
share_north,
share_east,
};
let res = state
@ -89,6 +272,85 @@ pub async fn post_shorten(
}
}
pub async fn get_share_links(
State(shared): State<Arc<SharedState>>,
Extension(user): Extension<OptionalUser>,
) -> Response {
let state = shared.load_state();
let user = match user.0 {
Some(u) => u,
None => return StatusCode::UNAUTHORIZED.into_response(),
};
let pb_url = state.pocketbase_url.trim_end_matches('/');
let token = match get_superuser_token(&state).await {
Ok(t) => t,
Err(err) => {
warn!("PocketBase superuser auth failed: {err}");
return StatusCode::BAD_GATEWAY.into_response();
}
};
let filter = format!("created_by=\"{}\"", user.id);
let url = format!(
"{pb_url}/api/collections/short_urls/records?sort=-created&perPage=200&filter={}",
urlencoding::encode(&filter)
);
let res = match state
.http_client
.get(&url)
.header("Authorization", format!("Bearer {token}"))
.send()
.await
{
Ok(r) => r,
Err(err) => {
warn!("Failed to list share links: {err}");
return StatusCode::BAD_GATEWAY.into_response();
}
};
if !res.status().is_success() {
let status = res.status();
let text = res.text().await.unwrap_or_default();
warn!("PocketBase list share links failed ({status}): {text}");
return StatusCode::BAD_GATEWAY.into_response();
}
let body: serde_json::Value = match res.json().await {
Ok(v) => v,
Err(err) => {
warn!("Failed to parse share links response: {err}");
return StatusCode::BAD_GATEWAY.into_response();
}
};
let public_url = state.public_url.trim_end_matches('/');
let links: Vec<ShareLinkListItem> = body["items"]
.as_array()
.map(|arr| {
arr.iter()
.map(|item| {
let code = item["code"].as_str().unwrap_or("").to_string();
let params = item["params"].as_str().unwrap_or("").to_string();
ShareLinkListItem {
url: format!("{public_url}/s/{code}"),
code,
og_image_url: og_image_url(public_url, &params),
params,
click_count: json_number_as_u64(&item["click_count"]),
created: item["created"].as_str().unwrap_or("").to_string(),
}
})
.collect()
})
.unwrap_or_default();
Json(ShareLinksResponse { links }).into_response()
}
pub async fn get_short_url(
State(shared): State<Arc<SharedState>>,
Path(code): Path<String>,
@ -132,22 +394,51 @@ pub async fn get_short_url(
}
};
let params = json["items"]
.as_array()
.and_then(|items| items.first())
.and_then(|item| item["params"].as_str());
let item = json["items"].as_array().and_then(|items| items.first());
match params {
Some(params) => {
let redirect_url = if params.is_empty() {
format!("/dashboard?share={code}")
} else {
format!("/dashboard?{params}&share={code}")
match item.and_then(|item| item["params"].as_str().map(|params| (item, params))) {
Some((item, params)) => {
let record_id = item["id"].as_str().unwrap_or("").to_string();
let next_click_count =
json_number_as_u64(&item["click_count"]).saturating_add(1);
let params = match sanitized_query_params(params, true) {
Ok(params) => params,
Err(reason) => {
warn!("Stored short URL params rejected for {code}: {reason}");
return StatusCode::BAD_REQUEST.into_response();
}
};
let og_image_url = format!("{}/api/screenshot?og=1&{params}", state.public_url);
let og_url = format!("{}/s/{code}", state.public_url);
if !record_id.is_empty() {
let update_url =
format!("{pb_url}/api/collections/short_urls/records/{record_id}");
match state
.http_client
.patch(&update_url)
.header("Authorization", format!("Bearer {token}"))
.json(&serde_json::json!({ "click_count": next_click_count }))
.send()
.await
{
Ok(update_resp) if update_resp.status().is_success() => {}
Ok(update_resp) => {
let status = update_resp.status();
let text = update_resp.text().await.unwrap_or_default();
warn!("PocketBase click count update failed ({status}): {text}");
}
Err(err) => warn!("PocketBase click count update failed: {err}"),
}
}
let redirect_url =
dashboard_redirect_url(&params, &code, record_share_bounds(item).is_some());
let og_image_url = og_image_url(&state.public_url, &params);
let og_url = format!("{}/s/{code}", state.public_url.trim_end_matches('/'));
let og_title = "Perfect Postcode | Every neighbourhood in England";
let og_description = "Explore property prices, energy ratings, crime stats, school ratings, and more across England on one interactive map.";
let redirect_url = escape_attr(&redirect_url);
let og_image_url = escape_attr(&og_image_url);
let og_url = escape_attr(&og_url);
let og_title = escape_attr(og_title);
let og_description = escape_attr(og_description);
let html = format!(
r#"<!DOCTYPE html>
@ -168,7 +459,13 @@ pub async fn get_short_url(
</head><body></body></html>"#
);
(
[(header::CACHE_CONTROL, "public, max-age=86400")],
[
(header::CACHE_CONTROL, "no-store"),
(
header::CONTENT_SECURITY_POLICY,
"default-src 'none'; img-src https: data:; base-uri 'none'; form-action 'none'",
),
],
Html(html),
)
.into_response()
@ -187,3 +484,37 @@ pub async fn get_short_url(
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sanitizes_short_url_params_and_drops_share() {
let params = sanitized_query_params(
"lat=51.5&lon=-0.1&zoom=12&filter=price%3A1%3A2&share=oldcode",
false,
)
.unwrap();
assert_eq!(params, "lat=51.5&lon=-0.1&zoom=12&filter=price%3A1%3A2");
}
#[test]
fn rejects_html_in_unsupported_params() {
assert!(sanitized_query_params("lat=51&x=%22%3E%3Cscript%3E", false).is_err());
}
#[test]
fn can_preserve_existing_share_grant() {
let params =
sanitized_query_params("lat=51.5&lon=-0.1&zoom=12&share=oldcode", true).unwrap();
assert_eq!(params, "lat=51.5&lon=-0.1&zoom=12&share=oldcode");
}
#[test]
fn escapes_html_attributes() {
assert_eq!(escape_attr(r#""'><&"#), "&quot;&#39;&gt;&lt;&amp;");
}
}

127
server-rs/src/routes/stripe_webhook.rs
View file

@ -9,7 +9,7 @@ use sha2::Sha256;
use tracing::{info, warn};
use crate::checkout_sessions::{
grant_license, mark_checkout_completed, mark_referral_invite_used, verify_checkout_completion,
complete_verified_checkout, reverse_license_for_payment_intent, verify_checkout_completion,
CheckoutCompletion,
};
use crate::state::SharedState;
@ -54,16 +54,52 @@ fn verify_signature(payload: &[u8], sig_header: &str, secret: &str) -> bool {
signed_payload.push(b'.');
signed_payload.extend_from_slice(payload);
signatures.into_iter().any(|sig_hex| {
// Verify every candidate signature without short-circuiting, so the total
// time taken doesn't depend on which (if any) signature matched.
let mut matched = false;
for sig_hex in signatures {
let Ok(sig_bytes) = hex::decode(sig_hex) else {
return false;
continue;
};
let Ok(mut mac) = HmacSha256::new_from_slice(secret.as_bytes()) else {
return false;
continue;
};
mac.update(&signed_payload);
mac.verify_slice(&sig_bytes).is_ok()
})
// verify_slice itself is constant-time.
if mac.verify_slice(&sig_bytes).is_ok() {
matched = true;
}
}
matched
}
fn payment_intent_id_from_object(object: &serde_json::Value) -> Option<&str> {
object["payment_intent"]
.as_str()
.filter(|id| is_safe_stripe_id(id))
}
fn is_safe_stripe_id(id: &str) -> bool {
!id.is_empty()
&& id.len() <= 128
&& id
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
}
fn reversal_event_is_actionable(event_type: &str, object: &serde_json::Value) -> bool {
match event_type {
"charge.refunded" => {
object["refunded"].as_bool().unwrap_or(false)
|| object["amount_refunded"].as_u64().unwrap_or(0) > 0
}
"charge.refund.updated" | "refund.created" | "refund.updated" => {
matches!(object["status"].as_str(), Some("succeeded"))
}
"charge.dispute.created" | "charge.dispute.funds_withdrawn" => true,
"charge.dispute.closed" => matches!(object["status"].as_str(), Some("lost")),
_ => false,
}
}
/// Handle Stripe webhook events.
@ -109,40 +145,11 @@ pub async fn post_stripe_webhook(
let session = &event["data"]["object"];
match verify_checkout_completion(&state, session).await {
Ok(CheckoutCompletion::Grant(checkout)) => {
if let Err(err) = mark_referral_invite_used(
&state,
&checkout.referral_invite_id,
&checkout.user_id,
)
.await
{
if let Err(err) = complete_verified_checkout(&state, &checkout).await {
warn!(
user_id = %checkout.user_id,
reservation_id = %checkout.reservation_id,
referral_invite_id = %checkout.referral_invite_id,
"Failed to mark referral invite used after Stripe checkout: {err:?}"
);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
if let Err(err) = grant_license(&state, &checkout.user_id).await {
warn!(
user_id = %checkout.user_id,
reservation_id = %checkout.reservation_id,
"Failed to grant license after Stripe checkout: {err:?}"
);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
if let Err(err) = mark_checkout_completed(
&state,
&checkout.reservation_id,
checkout.paid_amount_pence,
)
.await
{
warn!(
user_id = %checkout.user_id,
reservation_id = %checkout.reservation_id,
"Failed to mark checkout completed after license grant: {err:?}"
"Failed to complete verified Stripe checkout: {err:?}"
);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
@ -163,6 +170,52 @@ pub async fn post_stripe_webhook(
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
}
} else if matches!(
event_type,
"charge.refunded"
| "charge.refund.updated"
| "refund.created"
| "refund.updated"
| "charge.dispute.created"
| "charge.dispute.closed"
| "charge.dispute.funds_withdrawn"
) {
let object = &event["data"]["object"];
let Some(payment_intent_id) = payment_intent_id_from_object(object) else {
warn!(
event_id,
event_type, "Stripe reversal event missing payment intent id"
);
return StatusCode::OK.into_response();
};
if !reversal_event_is_actionable(event_type, object) {
info!(
payment_intent_id,
event_type, "Ignoring non-final Stripe reversal event"
);
return StatusCode::OK.into_response();
}
match reverse_license_for_payment_intent(&state, payment_intent_id, event_type).await {
Ok(Some(user_id)) => {
info!(
user_id,
payment_intent_id, event_type, "Processed Stripe payment reversal event"
);
}
Ok(None) => {
warn!(
payment_intent_id,
event_type, "Stripe reversal event had no matching checkout reservation"
);
}
Err(err) => {
warn!(
payment_intent_id,
event_type, "Failed to process Stripe payment reversal event: {err:?}"
);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
}
}
StatusCode::OK.into_response()