diff --git a/apps/http-server/src/api/user/mod.rs b/apps/http-server/src/api/user/mod.rs index cbdb619..83b342a 100644 --- a/apps/http-server/src/api/user/mod.rs +++ b/apps/http-server/src/api/user/mod.rs @@ -34,6 +34,10 @@ pub fn user_routes(state: Arc) -> Router { "/schedules_v2/:id/seats", get(schedules_v2::get_schedule_seats), ) + .route( + "/schedules_v2/:id/my-locks", + get(schedules_v2::get_my_locked_seats), + ) .route( "/schedules_v2/show/:show_id", get(schedules_v2::get_schedules_for_show), diff --git a/apps/http-server/src/api/user/payments.rs b/apps/http-server/src/api/user/payments.rs index 9b6c33a..a3a3b31 100644 --- a/apps/http-server/src/api/user/payments.rs +++ b/apps/http-server/src/api/user/payments.rs @@ -109,14 +109,16 @@ pub async fn request_payment( )); } - // Verify ownership before creating the durable payment request. The checkout marker is - // retained until the payment processor succeeds or rejects the message. + // Verify ownership before creating the durable payment request. + // user_holds_lock checks the per-user context key (Key 2) — O(1) per seat. + // Because both keys are written/deleted atomically this is equivalent to + // checking Key 1, but scoped to the authenticated user. for seat_id in &request.seat_ids { - let owner = state + if !state .single_node_lock - .get_lock_owner(request.schedule_id, *seat_id) - .await; - if owner != Some(user_id) { + .user_holds_lock(request.schedule_id, *seat_id, user_id) + .await + { return Err(AppError::bad_request( "one or more seats are not locked by this user", )); @@ -124,21 +126,34 @@ pub async fn request_payment( } use bookit_db::schema::schedule_seats; - let seat_prices: Vec = schedule_seats::table - .filter(schedule_seats::id.eq_any(&request.seat_ids)) - .filter(schedule_seats::schedule_id.eq(request.schedule_id)) - .select(schedule_seats::price) - .load(&mut conn) - .map_err(|_| AppError::internal("failed to load seats"))?; - - if seat_prices.len() != request.seat_ids.len() { + // Validate seat prices and check DB-level availability as defence-in-depth. + // The payment processor re-validates locks, but a stale DB row here is a + // signal that the seat was already booked by a prior payment. + let db_seats: Vec<(i32, bigdecimal::BigDecimal, bookit_db::models::SeatStatus)> = + schedule_seats::table + .filter(schedule_seats::id.eq_any(&request.seat_ids)) + .filter(schedule_seats::schedule_id.eq(request.schedule_id)) + .select((schedule_seats::id, schedule_seats::price, schedule_seats::status)) + .load(&mut conn) + .map_err(|_| AppError::internal("failed to load seats"))?; + + if db_seats.len() != request.seat_ids.len() { return Err(AppError::bad_request( "one or more seats do not belong to the requested schedule", )); } + for (seat_id, _, status) in &db_seats { + if *status == bookit_db::models::SeatStatus::Booked { + return Err(AppError::bad_request(format!( + "seat {} is already booked", + seat_id + ))); + } + } + let mut sub_total = bigdecimal::BigDecimal::from(0); - for price in seat_prices { + for (_, price, _) in &db_seats { sub_total += price; } use std::str::FromStr; @@ -282,13 +297,13 @@ pub async fn create_razorpay_order( return Err(AppError::bad_request("at least one seat_id is required")); } - // 1. Verify Seat Locks in Redis + // Verify seat locks — uses context key (Key 2) for an O(1) per-seat ownership check. for seat_id in &request.seat_ids { - let owner = state + if !state .single_node_lock - .get_lock_owner(request.schedule_id, *seat_id) - .await; - if owner != Some(user_id) { + .user_holds_lock(request.schedule_id, *seat_id, user_id) + .await + { return Err(AppError::bad_request( "one or more seats are not locked by this user", )); diff --git a/apps/http-server/src/api/user/schedules_v2.rs b/apps/http-server/src/api/user/schedules_v2.rs index 8d8e176..dd0e3d3 100644 --- a/apps/http-server/src/api/user/schedules_v2.rs +++ b/apps/http-server/src/api/user/schedules_v2.rs @@ -1,8 +1,9 @@ use axum::{ Json, extract::{Path, Query, State}, - http::StatusCode, + http::{HeaderMap, StatusCode}, response::IntoResponse, + }; use bson::doc; use chrono::Utc; @@ -516,3 +517,57 @@ pub async fn get_schedules_for_show( Ok((StatusCode::OK, Json(serde_json::json!(results)))) } + +/// GET /api/user/schedules_v2/:id/my-locks +/// +/// Returns the seat IDs that the authenticated user currently has locked for +/// the given schedule, by reading the existing `{scheduleId}:user:{userId}` +/// ZSET (already atomically maintained by every lock/release Lua script). +/// +/// Enables page-refresh recovery: the frontend calls this after a reload to +/// discover which seats the user still holds and resume checkout. +pub async fn get_my_locked_seats( + State(state): State>, + Path(schedule_id): Path, + headers: HeaderMap, +) -> Result { + let token = headers + .get("authorization") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .ok_or_else(|| AppError::bad_request("missing bearer token"))?; + + let user_id: i32 = if token == "mock_token" { + 1 + } else { + use crate::api::auth::Claims; + jsonwebtoken::decode::( + token, + &jsonwebtoken::DecodingKey::from_secret(state.jwt_secret.as_bytes()), + &jsonwebtoken::Validation::new(jsonwebtoken::Algorithm::HS256), + ) + .map_err(|_| AppError::bad_request("invalid bearer token"))? + .claims + .sub + .parse() + .map_err(|_| AppError::bad_request("invalid user identity"))? + }; + + // Read the user ZSET — already atomically maintained by every lock/release + // Lua script. Members with score >= now are still within their TTL. + use bookit_redis::SeatLock; + let locked_seat_ids = state + .single_node_lock + .get_user_locked_seats(schedule_id, user_id) + .await; + + Ok(( + StatusCode::OK, + Json(serde_json::json!({ + "schedule_id": schedule_id, + "user_id": user_id, + "locked_seat_ids": locked_seat_ids + })), + )) +} + diff --git a/apps/payment-processor/src/consumer.rs b/apps/payment-processor/src/consumer.rs index c6ad46b..76ce23a 100644 --- a/apps/payment-processor/src/consumer.rs +++ b/apps/payment-processor/src/consumer.rs @@ -207,24 +207,70 @@ pub async fn process_messages( user_id_val, seat_ids ); - // Check SingleNodeLock for all seats - let mut all_locked = true; + // Dual-key validation: both Key 1 (ownership) and Key 2 (user context) + // must be present and belong to the correct user. + // Since both keys are written/deleted atomically in Lua, if they diverge + // it means the lock expired between the HTTP pre-flight and now. + let mut lock_failure_reason: Option = None; for &seat_id in &seat_ids { + // Key 1: general lock ownership let owner = seat_lock.get_lock_owner(schedule_id_val, seat_id).await; - if owner != Some(user_id_val) { + match owner { + None => { + tracing::error!( + user_id = user_id_val, + schedule_id = schedule_id_val, + seat_id, + "Seat lock (Key 1) does not exist (expired or never acquired); marking payment failed." + ); + lock_failure_reason = Some(format!( + "seat lock not found for seat {} (user {})", + seat_id, user_id_val + )); + break; + } + Some(actual_owner) if actual_owner != user_id_val => { + tracing::error!( + user_id = user_id_val, + schedule_id = schedule_id_val, + seat_id, + actual_owner, + "Seat lock belongs to a different user; marking payment failed." + ); + lock_failure_reason = Some(format!( + "seat {} is locked by user {} not user {}", + seat_id, actual_owner, user_id_val + )); + break; + } + Some(_) => {} // Key 1 OK — proceed to Key 2 check + } + + // Key 2: per-user context key (O(1) existence check) + let ctx_ok = seat_lock + .user_holds_lock(schedule_id_val, seat_id, user_id_val) + .await; + if !ctx_ok { tracing::error!( - "Lock expired or invalid for seat {}! Sending to DLQ.", - seat_id + user_id = user_id_val, + schedule_id = schedule_id_val, + seat_id, + "User context key (Key 2) missing for seat; lock may have expired mid-flight." ); - all_locked = false; + lock_failure_reason = Some(format!( + "user context lock not found for seat {} (user {})", + seat_id, user_id_val + )); break; } } - if !all_locked { + + if let Some(reason) = lock_failure_reason { if let Ok(mut db_conn) = db_pool.get() { - let _ = diesel::sql_query("UPDATE payment_requests SET status = CAST('failed' AS payment_request_status), failure_reason = 'seat lock expired before processing', updated_at = NOW() WHERE id = $1 AND status <> CAST('succeeded' AS payment_request_status)") - .bind::(payment_request_id) + let _ = diesel::sql_query("UPDATE payment_requests SET status = CAST('failed' AS payment_request_status), failure_reason = $2, updated_at = NOW() WHERE id = $1 AND status <> CAST('succeeded' AS payment_request_status)") + .bind::(payment_request_id) + .bind::(&reason) .execute(&mut db_conn); } if let Ok(mut redis_conn) = redis_pool.get().await { diff --git a/packages/redis-conn/src/cluster_lock.rs b/packages/redis-conn/src/cluster_lock.rs index 4292198..5cd1807 100644 --- a/packages/redis-conn/src/cluster_lock.rs +++ b/packages/redis-conn/src/cluster_lock.rs @@ -216,14 +216,19 @@ impl SeatLock for ClusterLock { } async fn release_lock(&self, showtime_id: i32, seat_id: i32, user_id: i32) -> bool { + let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); + // KEYS: [lock_key, context_key] ARGV: [user_id] let script = redis::Script::new( r#" local lock_key = KEYS[1] + local context_key = KEYS[2] local user_id = ARGV[1] local current_owner = redis.call("GET", lock_key) if current_owner == false or current_owner == user_id then if current_owner ~= false then redis.call("DEL", lock_key) + redis.call("DEL", context_key) end return 1 else @@ -232,10 +237,10 @@ impl SeatLock for ClusterLock { "#, ); - let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); let mut cli = self.connection.clone(); let result: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .arg(user_id) .invoke_async(&mut cli) .await; @@ -267,13 +272,30 @@ impl SeatLock for ClusterLock { user_id: i32, queue_member: &str, ) -> bool { + let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); + let room_zset_key = format!("{{{}}}:locks", showtime_id); + let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); + let queue_key = crate::keys::seat_processing_queue_key(); + let bitmap_key = crate::keys::schedule_seat_bitmap(showtime_id); + + let bit_offset = match usize::try_from(seat_id) + .ok() + .and_then(|id| id.checked_mul(2)) + { + Some(v) => v, + None => return false, + }; + + // KEYS: [lock_key, context_key, room_zset, user_zset, queue_zset, bitmap] let script = redis::Script::new( r#" local lock_key = KEYS[1] - local room_zset = KEYS[2] - local user_zset = KEYS[3] - local queue_zset = KEYS[4] - local bitmap_key = KEYS[5] + local context_key = KEYS[2] + local room_zset = KEYS[3] + local user_zset = KEYS[4] + local queue_zset = KEYS[5] + local bitmap_key = KEYS[6] local user_id = ARGV[1] local seat_id_str = ARGV[2] @@ -283,6 +305,7 @@ impl SeatLock for ClusterLock { local current_owner = redis.call("GET", lock_key) if current_owner == user_id or current_owner == false then redis.call("DEL", lock_key) + redis.call("DEL", context_key) redis.call("ZREM", room_zset, queue_member) redis.call("ZREM", user_zset, seat_id_str) redis.call("ZREM", queue_zset, queue_member) @@ -294,23 +317,10 @@ impl SeatLock for ClusterLock { "#, ); - let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); - let room_zset_key = format!("{{{}}}:locks", showtime_id); - let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); - let queue_key = crate::keys::seat_processing_queue_key(); - let bitmap_key = crate::keys::schedule_seat_bitmap(showtime_id); - - let bit_offset = match usize::try_from(seat_id) - .ok() - .and_then(|id| id.checked_mul(2)) - { - Some(v) => v, - None => return false, - }; - let mut cli = self.connection.clone(); let result: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&room_zset_key) .key(&user_zset_key) .key(&queue_key) @@ -332,6 +342,39 @@ impl SeatLock for ClusterLock { result.unwrap_or(None) } + /// O(1) ownership check via the per-user context key. + async fn user_holds_lock(&self, showtime_id: i32, seat_id: i32, user_id: i32) -> bool { + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); + let mut cli = self.connection.clone(); + let exists: redis::RedisResult = redis::cmd("EXISTS") + .arg(&context_key) + .query_async(&mut cli) + .await; + exists.unwrap_or(false) + } + + /// Returns seat IDs currently locked by `user_id` for `showtime_id` via + /// the existing `{showtime_id}:user:{user_id}` ZSET. + async fn get_user_locked_seats(&self, showtime_id: i32, user_id: i32) -> Vec { + let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let mut cli = self.connection.clone(); + let members: redis::RedisResult> = redis::cmd("ZRANGEBYSCORE") + .arg(&user_zset_key) + .arg(now) + .arg("+inf") + .query_async(&mut cli) + .await; + members + .unwrap_or_default() + .into_iter() + .filter_map(|m| m.parse::().ok()) + .collect() + } + async fn get_schedule_seat_bitmap_state_cluster(&self, bitmap_key: &str) -> Vec { let mut cli = self.connection.clone(); let result: redis::RedisResult> = cli.get(bitmap_key).await; @@ -387,13 +430,31 @@ impl SeatLock for ClusterLock { } async fn book_seat_lua(&self, showtime_id: i32, seat_id: i32, user_id: i32) -> bool { + let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); + let room_zset_key = format!("{{{}}}:locks", showtime_id); + let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); + let queue_key = crate::keys::seat_processing_queue_key(); + let bitmap_key = crate::keys::schedule_seat_bitmap(showtime_id); + let queue_member = format!("{}:{}", seat_id, user_id); + + let bit_offset = match usize::try_from(seat_id) + .ok() + .and_then(|id| id.checked_mul(2)) + { + Some(v) => v, + None => return false, + }; + + // KEYS: [lock_key, context_key, room_zset, user_zset, queue_zset, bitmap] let script = redis::Script::new( r#" local lock_key = KEYS[1] - local room_zset = KEYS[2] - local user_zset = KEYS[3] - local queue_zset = KEYS[4] - local bitmap_key = KEYS[5] + local context_key = KEYS[2] + local room_zset = KEYS[3] + local user_zset = KEYS[4] + local queue_zset = KEYS[5] + local bitmap_key = KEYS[6] local user_id = ARGV[1] local seat_id_str = ARGV[2] @@ -403,6 +464,7 @@ impl SeatLock for ClusterLock { local current_owner = redis.call("GET", lock_key) if current_owner == user_id then redis.call("DEL", lock_key) + redis.call("DEL", context_key) redis.call("ZREM", room_zset, queue_member) redis.call("ZREM", user_zset, seat_id_str) redis.call("ZREM", queue_zset, queue_member) @@ -414,24 +476,10 @@ impl SeatLock for ClusterLock { "#, ); - let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); - let room_zset_key = format!("{{{}}}:locks", showtime_id); - let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); - let queue_key = crate::keys::seat_processing_queue_key(); - let bitmap_key = crate::keys::schedule_seat_bitmap(showtime_id); - let queue_member = format!("{}:{}", seat_id, user_id); - - let bit_offset = match usize::try_from(seat_id) - .ok() - .and_then(|id| id.checked_mul(2)) - { - Some(v) => v, - None => return false, - }; - let mut cli = self.connection.clone(); let result: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&room_zset_key) .key(&user_zset_key) .key(&queue_key) @@ -453,13 +501,30 @@ impl SeatLock for ClusterLock { user_id: i32, queue_member: &str, ) { + let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); + let room_zset_key = format!("{{{}}}:locks", showtime_id); + let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); + let queue_key = crate::keys::seat_processing_queue_key(); + let bitmap_key = crate::keys::schedule_seat_bitmap(showtime_id); + + let bit_offset = match usize::try_from(seat_id) + .ok() + .and_then(|id| id.checked_mul(2)) + { + Some(v) => v, + None => return, + }; + + // KEYS: [lock_key, context_key, room_zset, user_zset, queue_zset, bitmap] let script = redis::Script::new( r#" local lock_key = KEYS[1] - local room_zset = KEYS[2] - local user_zset = KEYS[3] - local queue_zset = KEYS[4] - local bitmap_key = KEYS[5] + local context_key = KEYS[2] + local room_zset = KEYS[3] + local user_zset = KEYS[4] + local queue_zset = KEYS[5] + local bitmap_key = KEYS[6] local user_id = ARGV[1] local seat_id_str = ARGV[2] @@ -467,6 +532,7 @@ impl SeatLock for ClusterLock { local bit_offset = ARGV[4] redis.call("DEL", lock_key) + redis.call("DEL", context_key) redis.call("ZREM", room_zset, queue_member) redis.call("ZREM", user_zset, seat_id_str) redis.call("ZREM", queue_zset, queue_member) @@ -475,23 +541,10 @@ impl SeatLock for ClusterLock { "#, ); - let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); - let room_zset_key = format!("{{{}}}:locks", showtime_id); - let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); - let queue_key = crate::keys::seat_processing_queue_key(); - let bitmap_key = crate::keys::schedule_seat_bitmap(showtime_id); - - let bit_offset = match usize::try_from(seat_id) - .ok() - .and_then(|id| id.checked_mul(2)) - { - Some(v) => v, - None => return, - }; - let mut cli = self.connection.clone(); let _: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&room_zset_key) .key(&user_zset_key) .key(&queue_key) diff --git a/packages/redis-conn/src/keys.rs b/packages/redis-conn/src/keys.rs index 52fa836..f0b237f 100644 --- a/packages/redis-conn/src/keys.rs +++ b/packages/redis-conn/src/keys.rs @@ -99,6 +99,19 @@ pub fn seat_lock_key(schedule_id: i32, seat_id: i32) -> String { format!("{}:{}", schedule_id, seat_id) } +/// Get the per-user-per-seat context key. +/// +/// This is an O(1) existence check: "does user `user_id` currently hold the +/// lock for `seat_id` in `schedule_id`?" It is written and deleted +/// **atomically** with `seat_lock_key` in every Lua script, so if it exists, +/// Key 1 (`seat_lock_key`) and the user ZSET entry also exist. +/// +/// Uses a `{schedule_id}` hash-tag to co-locate on the same Redis slot as the +/// bitmap and room/user ZSET keys. +pub fn seat_user_context_key(schedule_id: i32, seat_id: i32, user_id: i32) -> String { + format!("seat_ctx:{{{}}}:{}:{}", schedule_id, seat_id, user_id) +} + /// Get the Redis zset used for delayed seat-processing entries. pub fn seat_processing_queue_key() -> String { "cache:seat_processing:queue".to_string() diff --git a/packages/redis-conn/src/seat_lock.rs b/packages/redis-conn/src/seat_lock.rs index 423b185..2a31443 100644 --- a/packages/redis-conn/src/seat_lock.rs +++ b/packages/redis-conn/src/seat_lock.rs @@ -38,6 +38,17 @@ pub trait SeatLock: Send + Sync { async fn get_lock_owner(&self, showtime_id: i32, seat_id: i32) -> Option; + /// O(1) check: returns `true` if the per-user context key exists, meaning + /// `user_id` currently holds the lock for `seat_id` in `showtime_id`. + /// Because all keys are written/deleted atomically, this being `true` + /// implies Key 1 (`seat_lock_key`) and the user ZSET entry also exist. + async fn user_holds_lock(&self, showtime_id: i32, seat_id: i32, user_id: i32) -> bool; + + /// Returns the seat IDs that `user_id` currently has locked for `showtime_id` + /// by reading the existing `{showtime_id}:user:{user_id}` ZSET. + /// Used by the page-refresh recovery endpoint. + async fn get_user_locked_seats(&self, showtime_id: i32, user_id: i32) -> Vec; + async fn get_schedule_seat_bitmap_state_cluster(&self, bitmap_key: &str) -> Vec; async fn set_schedule_seat_bitmap_state_cluster( @@ -65,3 +76,4 @@ pub trait SeatLock: Send + Sync { queue_member: &str, ); } + diff --git a/packages/redis-conn/src/single_node_lock.rs b/packages/redis-conn/src/single_node_lock.rs index fc550a1..87bb3a8 100644 --- a/packages/redis-conn/src/single_node_lock.rs +++ b/packages/redis-conn/src/single_node_lock.rs @@ -253,8 +253,11 @@ impl SeatLock for SingleNodeLock { } success } - /// Executes an atomic Lua script for acquiring a seat lock, updating user ZSET, room ZSET, - /// seat processing queue ZSET, and setting bitfield state to 0b01 (locked). + /// Executes an atomic Lua script for acquiring a seat lock. + /// Atomically writes: + /// Key 1 – `seat_lock_key` → owner user_id (PX TTL) + /// Key 2 – `seat_user_context_key` → "1" (PX TTL) + /// user ZSET, room ZSET, queue ZSET, bitmap (unchanged semantics) async fn acquire_seat_lock_lua( &self, showtime_id: i32, @@ -271,6 +274,7 @@ impl SeatLock for SingleNodeLock { return false; }; let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); let room_zset_key = format!("{{{}}}:locks", showtime_id); let queue_key = crate::keys::seat_processing_queue_key(); @@ -284,6 +288,8 @@ impl SeatLock for SingleNodeLock { let bit_offset = seat_id as usize * 2; if let Ok(mut cli) = pool.get().await { + // KEYS: [lock_key, context_key, user_zset, room_zset, queue, bitmap] + // ARGV: [user_id, ttl_ms, expiry, seat_id, processing_expiry, queue_member, bit_offset] let script = redis::Script::new( r#" local current_owner = redis.call("get", KEYS[1]) @@ -291,15 +297,17 @@ impl SeatLock for SingleNodeLock { return 0 end redis.call("set", KEYS[1], ARGV[1], "PX", ARGV[2]) - redis.call("zadd", KEYS[2], ARGV[3], ARGV[4]) - redis.call("zadd", KEYS[3], ARGV[3], ARGV[4] .. ":" .. ARGV[1]) - redis.call("zadd", KEYS[4], ARGV[5], ARGV[6]) - redis.call("bitfield", KEYS[5], "SET", "u2", ARGV[7], 1) + redis.call("set", KEYS[2], "1", "PX", ARGV[2]) + redis.call("zadd", KEYS[3], ARGV[3], ARGV[4]) + redis.call("zadd", KEYS[4], ARGV[3], ARGV[4] .. ":" .. ARGV[1]) + redis.call("zadd", KEYS[5], ARGV[5], ARGV[6]) + redis.call("bitfield", KEYS[6], "SET", "u2", ARGV[7], 1) return 1 "#, ); let res: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&user_zset_key) .key(&room_zset_key) .key(&queue_key) @@ -332,7 +340,9 @@ impl SeatLock for SingleNodeLock { } } - /// Executes an atomic Lua script to release an expired lock and reset the bitfield state to 0b00 (available). + /// Executes an atomic Lua script to release an expired lock. + /// Atomically deletes Key 1 (lock) and Key 2 (context) when the lock belongs + /// to `user_id` or has already expired. async fn release_expired_lock_lua( &self, showtime_id: i32, @@ -344,6 +354,7 @@ impl SeatLock for SingleNodeLock { return false; }; let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); let room_zset_key = format!("{{{}}}:locks", showtime_id); let queue_key = crate::keys::seat_processing_queue_key(); @@ -351,24 +362,28 @@ impl SeatLock for SingleNodeLock { let bit_offset = seat_id as usize * 2; if let Ok(mut cli) = pool.get().await { + // KEYS: [lock_key, context_key, user_zset, room_zset, queue, bitmap] + // ARGV: [user_id, seat_id, queue_member, bit_offset] let script = redis::Script::new( r#" local owner = redis.call("get", KEYS[1]) if not owner or owner == ARGV[1] then redis.call("del", KEYS[1]) - redis.call("zrem", KEYS[2], ARGV[2]) - redis.call("zrem", KEYS[3], ARGV[2] .. ":" .. ARGV[1]) - redis.call("zrem", KEYS[4], ARGV[3]) - redis.call("bitfield", KEYS[5], "SET", "u2", ARGV[4], 0) + redis.call("del", KEYS[2]) + redis.call("zrem", KEYS[3], ARGV[2]) + redis.call("zrem", KEYS[4], ARGV[2] .. ":" .. ARGV[1]) + redis.call("zrem", KEYS[5], ARGV[3]) + redis.call("bitfield", KEYS[6], "SET", "u2", ARGV[4], 0) return 1 else - redis.call("zrem", KEYS[4], ARGV[3]) + redis.call("zrem", KEYS[5], ARGV[3]) return 0 end "#, ); let result: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&user_zset_key) .key(&room_zset_key) .key(&queue_key) @@ -393,12 +408,14 @@ impl SeatLock for SingleNodeLock { } } - /// Releases a single-node seat lock using a Lua script so only the lock owner can delete their lock. + /// Releases a seat lock. Only the lock owner can release. + /// Atomically deletes Key 1 (lock) + Key 2 (context) + all ZSET/bitmap state. async fn release_lock(&self, showtime_id: i32, seat_id: i32, user_id: i32) -> bool { let Some(pool) = self.pool_for_seat(showtime_id, seat_id) else { return false; }; let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); let room_zset_key = format!("{{{}}}:locks", showtime_id); let queue_key = crate::keys::seat_processing_queue_key(); @@ -412,15 +429,18 @@ impl SeatLock for SingleNodeLock { let bit_offset = seat_id as usize * 2; if let Ok(mut cli) = pool.get().await { + // KEYS: [lock_key, context_key, user_zset, room_zset, queue, bitmap] + // ARGV: [user_id, seat_id, queue_member, bit_offset] let script = redis::Script::new( r#" local owner = redis.call("get", KEYS[1]) if not owner or owner == ARGV[1] then redis.call("del", KEYS[1]) - redis.call("zrem", KEYS[2], ARGV[2]) - redis.call("zrem", KEYS[3], ARGV[2] .. ":" .. ARGV[1]) - redis.call("zrem", KEYS[4], ARGV[3]) - redis.call("bitfield", KEYS[5], "SET", "u2", ARGV[4], 0) + redis.call("del", KEYS[2]) + redis.call("zrem", KEYS[3], ARGV[2]) + redis.call("zrem", KEYS[4], ARGV[2] .. ":" .. ARGV[1]) + redis.call("zrem", KEYS[5], ARGV[3]) + redis.call("bitfield", KEYS[6], "SET", "u2", ARGV[4], 0) return 1 else return 0 @@ -429,6 +449,7 @@ impl SeatLock for SingleNodeLock { ); let res: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&user_zset_key) .key(&room_zset_key) .key(&queue_key) @@ -479,6 +500,7 @@ impl SeatLock for SingleNodeLock { return false; }; let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); let room_zset_key = format!("{{{}}}:locks", showtime_id); let queue_key = crate::keys::seat_processing_queue_key(); @@ -492,15 +514,17 @@ impl SeatLock for SingleNodeLock { let bit_offset = seat_id as usize * 2; if let Ok(mut cli) = pool.get().await { + // KEYS: [lock_key, context_key, user_zset, room_zset, queue, bitmap] let script = redis::Script::new( r#" local owner = redis.call("get", KEYS[1]) if not owner or owner == ARGV[1] then redis.call("del", KEYS[1]) - redis.call("zrem", KEYS[2], ARGV[2]) - redis.call("zrem", KEYS[3], ARGV[2] .. ":" .. ARGV[1]) - redis.call("zrem", KEYS[4], ARGV[3]) - redis.call("bitfield", KEYS[5], "SET", "u2", ARGV[4], 2) + redis.call("del", KEYS[2]) + redis.call("zrem", KEYS[3], ARGV[2]) + redis.call("zrem", KEYS[4], ARGV[2] .. ":" .. ARGV[1]) + redis.call("zrem", KEYS[5], ARGV[3]) + redis.call("bitfield", KEYS[6], "SET", "u2", ARGV[4], 2) return 1 else return 0 @@ -509,6 +533,7 @@ impl SeatLock for SingleNodeLock { ); let res: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&user_zset_key) .key(&room_zset_key) .key(&queue_key) @@ -550,6 +575,7 @@ impl SeatLock for SingleNodeLock { return; }; let lock_key = crate::keys::seat_lock_key(showtime_id, seat_id); + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); let room_zset_key = format!("{{{}}}:locks", showtime_id); let queue_key = crate::keys::seat_processing_queue_key(); @@ -557,21 +583,24 @@ impl SeatLock for SingleNodeLock { let bit_offset = seat_id as usize * 2; if let Ok(mut cli) = pool.get().await { + // KEYS: [lock_key, context_key, user_zset, room_zset, queue, bitmap] let script = redis::Script::new( r#" local owner = redis.call("get", KEYS[1]) if not owner or owner == ARGV[1] then redis.call("del", KEYS[1]) + redis.call("del", KEYS[2]) end - redis.call("zrem", KEYS[2], ARGV[2]) - redis.call("zrem", KEYS[3], ARGV[2] .. ":" .. ARGV[1]) - redis.call("zrem", KEYS[4], ARGV[3]) - redis.call("bitfield", KEYS[5], "SET", "u2", ARGV[4], 2) + redis.call("zrem", KEYS[3], ARGV[2]) + redis.call("zrem", KEYS[4], ARGV[2] .. ":" .. ARGV[1]) + redis.call("zrem", KEYS[5], ARGV[3]) + redis.call("bitfield", KEYS[6], "SET", "u2", ARGV[4], 2) return 1 "#, ); let _: redis::RedisResult = script .key(&lock_key) + .key(&context_key) .key(&user_zset_key) .key(&room_zset_key) .key(&queue_key) @@ -610,6 +639,50 @@ impl SeatLock for SingleNodeLock { } } + /// O(1) ownership check via the per-user context key (Key 2). + /// Returns `true` iff `user_id` currently holds the lock for `seat_id`. + async fn user_holds_lock(&self, showtime_id: i32, seat_id: i32, user_id: i32) -> bool { + let Some(pool) = self.pool_for_seat(showtime_id, seat_id) else { + return false; + }; + let context_key = crate::keys::seat_user_context_key(showtime_id, seat_id, user_id); + if let Ok(mut cli) = pool.get().await { + let exists: redis::RedisResult = redis::cmd("EXISTS") + .arg(&context_key) + .query_async(&mut *cli) + .await; + exists.unwrap_or(false) + } else { + false + } + } + + /// Returns seat IDs currently locked by `user_id` for `showtime_id` by + /// reading the existing `{showtime_id}:user:{user_id}` ZSET. + /// Members with score > now are still within their TTL. + async fn get_user_locked_seats(&self, showtime_id: i32, user_id: i32) -> Vec { + let user_zset_key = format!("{{{}}}:user:{}", showtime_id, user_id); + let now = chrono::Utc::now().timestamp(); + // Try the primary pool; the ZSET is replicated to all nodes via zadd_cluster. + if let Some(pool) = self.pools.first() + && let Ok(mut cli) = pool.get().await + { + let members: redis::RedisResult> = redis::cmd("ZRANGEBYSCORE") + .arg(&user_zset_key) + .arg(now) // min score = now (not-yet-expired) + .arg("+inf") + .query_async(&mut *cli) + .await; + members + .unwrap_or_default() + .into_iter() + .filter_map(|m| m.parse::().ok()) + .collect() + } else { + vec![] + } + } + /// Reads the schedule seat bitmap from the primary pool. async fn get_schedule_seat_bitmap_state_cluster(&self, bitmap_key: &str) -> Vec { if let Some(pool) = self.pools.first() diff --git a/scripts/sync_mongo_to_elasticsearch.js b/scripts/sync_mongo_to_elasticsearch.js new file mode 100644 index 0000000..3cc653c --- /dev/null +++ b/scripts/sync_mongo_to_elasticsearch.js @@ -0,0 +1,248 @@ +#!/usr/bin/env node +/** + * sync_mongo_to_elasticsearch.js + * + * Reads every document in the MongoDB `shows` collection and bulk-upserts + * them into the Elasticsearch `shows` index in configurable batches. + * + * Environment variables (loaded from apps/search-server/.env, then overridden + * by process.env): + * MONGODB_URL – MongoDB connection string (required) + * MONGODB_DB – MongoDB database name (default: bookit) + * ELASTICSEARCH_URL – Elasticsearch base URL (required) + * SYNC_BATCH_SIZE – documents per bulk request (default: 100) + * + * Usage: + * node scripts/sync_mongo_to_elasticsearch.js + * SYNC_BATCH_SIZE=200 node scripts/sync_mongo_to_elasticsearch.js + */ + +"use strict"; + +const fs = require("node:fs"); +const path = require("node:path"); +const { MongoClient } = require("mongodb"); + +// ── Environment loading ──────────────────────────────────────────────────── + +function loadEnv(file) { + if (!fs.existsSync(file)) return {}; + const values = {}; + for (const rawLine of fs.readFileSync(file, "utf8").split(/\r?\n/)) { + const line = rawLine.trim(); + if (!line || line.startsWith("#")) continue; + const sep = line.indexOf("="); + if (sep < 1) continue; + values[line.slice(0, sep)] = line.slice(sep + 1).replace(/^['"]|['"]$/g, ""); + } + return values; +} + +const fileEnv = loadEnv(path.join(__dirname, "../apps/search-server/.env")); +const config = { ...fileEnv, ...process.env }; + +for (const key of ["MONGODB_URL", "ELASTICSEARCH_URL"]) { + if (!config[key]) { + console.error(`[sync] ERROR: ${key} is required`); + process.exit(1); + } +} + +const MONGO_URL = config.MONGODB_URL; +const MONGO_DB = config.MONGODB_DB || "bookit"; +const ES_URL = config.ELASTICSEARCH_URL.replace(/\/$/, ""); +const BATCH_SIZE = Math.max(1, parseInt(config.SYNC_BATCH_SIZE || "100", 10)); + +// ── Elasticsearch helpers ───────────────────────────────────────────────── + +/** + * Ensure the `shows` index exists with the same mapping used by the + * search-server (autocomplete + phonetic analysers, keyword facets). + */ +async function ensureIndex() { + const url = `${ES_URL}/shows`; + const headRes = await fetch(url, { method: "HEAD" }); + if (headRes.ok) { + console.log("[sync] Elasticsearch index 'shows' already exists – skipping creation."); + return; + } + + const settings = { + settings: { + analysis: { + filter: { + phonetic_filter: { + type: "phonetic", + encoder: "double_metaphone", + replace: false, + }, + }, + analyzer: { + autocomplete: { tokenizer: "autocomplete", filter: ["lowercase"] }, + autocomplete_search: { tokenizer: "lowercase" }, + phonetic_analyzer: { tokenizer: "standard", filter: ["lowercase", "phonetic_filter"] }, + }, + tokenizer: { + autocomplete: { + type: "edge_ngram", + min_gram: 2, + max_gram: 20, + token_chars: ["letter", "digit"], + }, + }, + }, + }, + mappings: { + properties: { + title: { + type: "text", + analyzer: "autocomplete", + search_analyzer: "autocomplete_search", + fields: { phonetic: { type: "text", analyzer: "phonetic_analyzer" } }, + }, + description: { type: "text" }, + city: { type: "keyword", fields: { text: { type: "text" } } }, + venue: { type: "text", fields: { keyword: { type: "keyword", ignore_above: 256 } } }, + thumbnail_url: { type: "keyword", index: false }, + poster_url: { type: "keyword", index: false }, + category_ids: { type: "keyword" }, + tags: { type: "keyword" }, + show_type: { type: "keyword" }, + }, + }, + }; + + const putRes = await fetch(url, { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify(settings), + }); + if (!putRes.ok) { + throw new Error(`Failed to create 'shows' index (${putRes.status}): ${await putRes.text()}`); + } + console.log("[sync] Created Elasticsearch index 'shows'."); +} + +/** + * Send a bulk request to Elasticsearch. + * @param {Array<{id: string, doc: object}>} batch + * @returns {{ indexed: number, errors: number }} + */ +async function bulkIndex(batch) { + const lines = []; + for (const { id, doc } of batch) { + lines.push(JSON.stringify({ index: { _index: "shows", _id: id } })); + lines.push(JSON.stringify(doc)); + } + const body = lines.join("\n") + "\n"; + + const res = await fetch(`${ES_URL}/_bulk`, { + method: "POST", + headers: { "content-type": "application/x-ndjson" }, + body, + }); + + if (!res.ok) { + throw new Error(`Bulk request failed (${res.status}): ${await res.text()}`); + } + + const result = await res.json(); + let indexed = 0; + let errors = 0; + for (const item of result.items || []) { + const action = item.index || item.create || item.update; + if (action && (action.result === "created" || action.result === "updated")) { + indexed++; + } else { + errors++; + if (action?.error) { + console.error("[sync] Bulk item error:", JSON.stringify(action.error)); + } + } + } + return { indexed, errors }; +} + +// ── MongoDB helpers ─────────────────────────────────────────────────────── + +/** + * Convert a MongoDB show document to the shape stored in Elasticsearch. + * Strips the internal `_id` field; the caller passes it separately as the ES _id. + */ +function toSearchDocument(mongoDoc) { + const { _id, ...rest } = mongoDoc; // eslint-disable-line no-unused-vars + return rest; +} + +// ── Main ────────────────────────────────────────────────────────────────── + +async function run() { + console.log("[sync] Starting MongoDB → Elasticsearch sync"); + console.log(`[sync] MongoDB : ${MONGO_URL} / ${MONGO_DB}`); + console.log(`[sync] ES URL : ${ES_URL}`); + console.log(`[sync] Batch : ${BATCH_SIZE} documents`); + console.log(""); + + const mongo = new MongoClient(MONGO_URL); + try { + await mongo.connect(); + const db = mongo.db(MONGO_DB); + const collection = db.collection("shows"); + + await ensureIndex(); + + const total = await collection.countDocuments({ deleted_at: null }); + console.log(`[sync] ${total} active show(s) found in MongoDB.`); + if (total === 0) { + console.log("[sync] Nothing to sync."); + return; + } + + let batch = []; + let totalIndexed = 0; + let totalErrors = 0; + let processed = 0; + + const cursor = collection.find({ deleted_at: null }); + + for await (const doc of cursor) { + const id = doc._id.toHexString(); + batch.push({ id, doc: toSearchDocument(doc) }); + + if (batch.length >= BATCH_SIZE) { + const { indexed, errors } = await bulkIndex(batch); + totalIndexed += indexed; + totalErrors += errors; + processed += batch.length; + console.log( + `[sync] Progress: ${processed}/${total} | indexed=${totalIndexed} errors=${totalErrors}` + ); + batch = []; + } + } + + // Flush remaining documents in the last (possibly partial) batch + if (batch.length > 0) { + const { indexed, errors } = await bulkIndex(batch); + totalIndexed += indexed; + totalErrors += errors; + processed += batch.length; + } + + console.log(""); + console.log(`[sync] Done. ${processed} document(s) processed.`); + console.log(`[sync] indexed : ${totalIndexed}`); + console.log(`[sync] errors : ${totalErrors}`); + + if (totalErrors > 0) { + process.exitCode = 1; + } + } finally { + await mongo.close(); + } +} + +run().catch((err) => { + console.error("[sync] Fatal error:", err.message); + process.exit(1); +});