fix: race conditions and rollovers

server/src/_luaScriptsV2/deductFromCustomerEntitlements/contextUtils.lua

@@ -6,7 +6,8 @@
 --[[
   init_context(params)
   
-  Initializes context object with current balances for all customer_entitlements.
+  Initializes context object with current balances for all customer_entitlements
+  and builds a rollover index for fast lookups.
   Reads from Redis once upfront to avoid multiple reads during passes.
   
   params:
@@ -16,6 +17,7 @@
     
   Returns: context table with:
     customer_entitlements: { [cus_ent_id]: { base_path, balance, adjustment, entities } }
+    rollovers: { [rollover_id]: { base_path, balance, usage, entities } }
     pending_writes: {} (empty array to queue writes)
     logs: {} (debug logs)
     logger: { log(fmt, ...): function } (logger that appends to logs)
@@ -25,6 +27,7 @@ local function init_context(params)
   
   local context = {
     customer_entitlements = {},
+    rollovers = {},
     pending_writes = {},
     logs = logs,
     logger = {
@@ -60,6 +63,29 @@ local function init_context(params)
       end
       
       context.customer_entitlements[ent_id] = ent_data
+      
+      -- Build rollover index for this customer_entitlement
+      local cus_ent_rollovers = cus_ent.rollovers
+      if cus_ent_rollovers and type(cus_ent_rollovers) == 'table' then
+        for r_idx, rollover in ipairs(cus_ent_rollovers) do
+          if rollover and rollover.id then
+            local r_idx_0 = r_idx - 1
+            local rollover_path = base_path .. '.rollovers[' .. r_idx_0 .. ']'
+            
+            -- Read fresh rollover data from Redis
+            local rollover_data = read_rollover_data(params.cache_key, rollover_path)
+            
+            if rollover_data then
+              context.rollovers[rollover.id] = {
+                base_path = rollover_path,
+                balance = rollover_data.balance,
+                usage = rollover_data.usage,
+                entities = rollover_data.entities,
+              }
+            end
+          end
+        end
+      end
     end
   end
   
@@ -124,6 +150,56 @@ local function queue_balance_update(params)
   end
 end
 
+--[[
+  queue_rollover_update(params)
+  
+  Queues a rollover balance/usage update to pending_writes.
+  Rollovers track both balance (decrements) and usage (increments).
+  
+  params:
+    context: table (context object)
+    path: string (JSON path to the rollover or entity, WITHOUT .balance/.usage suffix)
+    deduct_amount: number (positive amount to deduct from balance and add to usage)
+]]
+local function queue_rollover_update(params)
+  local context = params.context
+  local path = params.path
+  local deduct_amount = params.deduct_amount
+  
+  -- Queue balance decrement
+  table.insert(context.pending_writes, { path = path .. '.balance', delta = -deduct_amount })
+  
+  -- Queue usage increment
+  table.insert(context.pending_writes, { path = path .. '.usage', delta = deduct_amount })
+end
+
+--[[
+  update_in_memory_rollover(params)
+  
+  Updates balance and usage in-memory on a rollover object.
+  
+  params:
+    target: table (rollover_data or entities table)
+    entity_id: string or nil (if entity-scoped)
+    deduct_amount: number (positive amount to deduct from balance and add to usage)
+]]
+local function update_in_memory_rollover(params)
+  local target = params.target
+  local entity_id = params.entity_id
+  local deduct_amount = params.deduct_amount
+
+  if entity_id then
+    if not target[entity_id] then
+      target[entity_id] = { balance = 0, usage = 0 }
+    end
+    target[entity_id].balance = (target[entity_id].balance or 0) - deduct_amount
+    target[entity_id].usage = (target[entity_id].usage or 0) + deduct_amount
+  else
+    target.balance = (target.balance or 0) - deduct_amount
+    target.usage = (target.usage or 0) + deduct_amount
+  end
+end
+
 --[[
   apply_pending_writes(cache_key, context)