Class: SpreeCmCommissioner::WaitingGuestsCaller

Inherits:
BaseInteractor show all
Defined in:
app/interactors/spree_cm_commissioner/waiting_guests_caller.rb

Constant Summary collapse

FIRESTORE_BATCH_SIZE =

Firestore bounds a batch update by payload size (10 MiB); 500 ops/commit leaves us far under that.

(ENV['WAITING_ROOM_FIRESTORE_BATCH_SIZE'] || 500).to_i
MAX_CALLS_PER_RUN =

Release at most this many guests per run, even when far more slots are free. A capacity spike (cold start / lull → active_sessions low → available_slots ≈ max_sessions, e.g. 900) would otherwise release every queued guest at once, dumping 900 simultaneous session-token requests on the server. Capping paces entry at ~MAX_CALLS_PER_RUN per run (e.g. 30s, per config/schedule.yml); the rest stay queued and drain over the following runs.

(ENV['WAITING_ROOM_MAX_CALLS_PER_RUN'] || 50).to_i

Instance Method Summary collapse

Instance Method Details

#call ⇒ Object 



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 16

def call
  started_at = Time.zone.now
  available_slots = fetch_available_slots
  call_limit = [available_slots, MAX_CALLS_PER_RUN].min

  # Always run through to mark_as/publish_logs — even with no slots — so the lobby keeps a
  # fresh heartbeat during peak/full periods instead of going stale and looking like a dead cron.
  long_waiting_guests = call_limit.positive? ? fetch_long_waiting_guests(call_limit) : []
  calling_all(long_waiting_guests) if long_waiting_guests.any?

  mark_as(
    full: long_waiting_guests.size >= available_slots,
    available_slots: available_slots - long_waiting_guests.size,
    has_queue: waiting_exists?([previous_records_path, records_path]),
    active_sessions: @active_sessions,
    max_sessions: @max_sessions
  )

  publish_logs(started_at: started_at, called_count: long_waiting_guests.size)
end

#calling_all(waiting_guests) ⇒ Object

For alert waiting guests to enter room, we update :allow_to_enter_room_at and (when precreation succeeded) a :session map with a usable token — the app reads it straight off this same snapshot and enters without a separate createSession round trip. Precreating first, then writing both fields in the same batch, is what makes grant + session arrive atomically: a guest can never observe "allowed in" without also having a session, and there's no window for the old race where the app's own createSession beat this job's precreate to the same row.

Commit in Firestore batches (chunks of FIRESTORE_BATCH_SIZE) instead of one update per guest, so e.g. 1000 guests = 2 commits, not 1000 round-trips. update merges, so we only send the changed fields and leave the rest of each doc intact.



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 120

def calling_all(waiting_guests)
  allow_at = Time.zone.now
  sessions_by_guest_id = precreate_sessions(waiting_guests)

  waiting_guests.each_slice(FIRESTORE_BATCH_SIZE) do |slice|
    firestore.batch do |b|
      slice.each do |document|
        session = sessions_by_guest_id[document.ref.document_id]
        data = {
          allow_to_enter_room_at: allow_at,
          session: {
            id: session.id,
            jwt_token: session.jwt_token,
            expired_at: session.expired_at,
            created_at: session.created_at
          }
        }
        b.update(document.ref, data)
      end
    end
  end
end

#current_date ⇒ Object 



188
189
190
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 188

def current_date
  Time.zone.now.strftime('%Y-%m-%d')
end

#default_records_path(date) ⇒ Object 



106
107
108
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 106

def default_records_path(date)
  "waiting_guests/#{date}/records"
end

#eligible_guests_in(records_path, limit) ⇒ Object

This query requires a composite index (allow_to_enter_room_at ==, position !=, queued_at order); create it in Firebase beforehand, same as the plain index #fetch_long_waiting_guests already needed.

The position != nil filter closes the call-before-stamp race: without it, a doc could be called here before StampQueuePositions ever assigns it a position, permanently excluding it from #eligible_guests_in (every future run still sees allow_to_enter_room_at == nil, but the calling_all write below sets it non-nil for good) while still counting toward total_exited_queue's unscoped count. Gating on position != nil makes "already stamped" true by construction before a doc can ever be called, instead of relying on StampQueuePositions reliably beating this job to it (e.g. 10s vs. 30s — see cm-market-server/config/schedule.yml).



73
74
75
76
77
78
79
80
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 73

def eligible_guests_in(records_path, limit)
  firestore.col(records_path)
           .where('allow_to_enter_room_at', '==', nil)
           .where('position', '!=', nil)
           .order('queued_at')
           .limit(limit)
           .get.to_a
end

#fetch_available_slots ⇒ Object 



37
38
39
40
41
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 37

def fetch_available_slots
  @max_sessions = fetch_max_sessions
  @active_sessions = SpreeCmCommissioner::WaitingRoomSession.active.count
  [@max_sessions - @active_sessions, 0].max
end

#fetch_long_waiting_guests(available_slots) ⇒ Object

This query requires an index; create it in Firebase beforehand. Client must create waiting_guests documents with :queued_at and :allow_to_enter_room_at set to nil to allow filter + order queries.

Yesterday's guests are always older than today's, so fill from yesterday first, then use any leftover slots for today. This way no one queued before the midnight rollover gets skipped. e.g. 5 slots, 2 waiting in yesterday -> take both, then take 3 from today.



49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 49

def fetch_long_waiting_guests(available_slots)
  previous_guests = eligible_guests_in(previous_records_path, available_slots)

  # Pre-flip window: the lobby pointer still points at yesterday, so both paths resolve to the
  # same partition — return now to avoid querying (and double-counting) it twice.
  return previous_guests if records_path == previous_records_path

  remaining_slots = available_slots - previous_guests.size
  return previous_guests if remaining_slots <= 0

  previous_guests + eligible_guests_in(records_path, remaining_slots)
end

#fetch_max_sessions ⇒ Object 



244
245
246
247
248
249
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 244

def fetch_max_sessions
  fetcher = SpreeCmCommissioner::WaitingRoomSystemMetadataFetcher.new(firestore: firestore)
  fetcher.load_document_data

  fetcher.max_sessions_count_with_min
end

#firestore ⇒ Object 



251
252
253
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 251

def firestore
  @firestore ||= Google::Cloud::Firestore.new(project_id: [:project_id], credentials: )
end

#lobby_data ⇒ Object 



232
233
234
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 232

def lobby_data
  @lobby_data ||= lobby_document.get.data
end

#lobby_document ⇒ Object 



236
237
238
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 236

def lobby_document
  @lobby_document ||= firestore.col('waiting_rooms').doc('lobby')
end

#logs_document ⇒ Object 



240
241
242
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 240

def logs_document
  @logs_document ||= lobby_document.col('logs').doc('waiting_guests_caller_job')
end

#mark_as(full:, available_slots:, has_queue:, active_sessions: nil, max_sessions: nil) ⇒ Object 



196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 196

def mark_as(
  full:,
  available_slots:,
  has_queue:,
  active_sessions: nil,
  max_sessions: nil
)
  data = {
    full: full,
    available_slots: available_slots,
    has_queue: has_queue,
    updated_at: Time.zone.now
  }

  data[:active_sessions] = active_sessions unless active_sessions.nil?
  data[:max_sessions] = max_sessions unless max_sessions.nil?

  lobby_document.set(data, merge: true)
end

#precreate_sessions(waiting_guests) ⇒ Object

Pre-creates a real, usable WaitingRoomSession row (real signed JWT, not a placeholder) for every called guest, and returns them keyed by guest_identifier so #calling_all can hand each guest their own session in the same Firestore write that grants entry.

A no-show's row simply expires on the normal TTL (no separate reclaim job needed). These rows only contain identity and timing; device attributes and remote_ip are left for the guest's real createSession call to fill in on renewal, keeping compatibility with older clients that never read the Firestore-carried session at all — they just call createSession like before and hit WaitingRoomSessionCreator's existing renew-active-row path.

MAX_CALLS_PER_RUN caps this at ≤50 rows/run, making per-row writes cheap enough to avoid raw SQL batching. The guest_identifier unique index + rescue handles the concurrent-race case (like the guest's app calling createSession first) by looking up whichever row won — unscoped, not .active, so a row that's already inactive by the time we look is still picked up rather than treated as absent. find_by! (not find_by) is deliberate: allow_to_enter_room_at drops a guest out of every future run's eligibility filter for good (see #eligible_guests_in), and the app has no fallback of its own (entry is session-presence-only), so a guest that truly ends up without a row here — the colliding row vanishing between the unique-index hit and this lookup — is a bug worth a loud ActiveRecord::RecordNotFound, not a silently skipped session field.



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 162

def precreate_sessions(waiting_guests)
  return {} if waiting_guests.empty?

  waiting_guests.each_with_object({}) do |doc, sessions|
    guest_id = doc.ref.document_id
    expired_at = precreated_expired_at

    sessions[guest_id] = SpreeCmCommissioner::WaitingRoomSession.create!(
      guest_identifier: guest_id,
      jwt_token: JWT.encode({ exp: expired_at.to_i }, ENV.fetch('WAITING_ROOM_SESSION_SIGNATURE'), 'HS256'),
      expired_at: expired_at,
      max_expired_at: precreated_max_expired_at
    )
  rescue ActiveRecord::RecordNotUnique
    sessions[guest_id] = SpreeCmCommissioner::WaitingRoomSession.find_by!(guest_identifier: guest_id)
  end
end

#precreated_expired_at ⇒ Object 



180
181
182
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 180

def precreated_expired_at
  SpreeCmCommissioner::WaitingRoomSessionCreator::WAITING_ROOM_SESSION_EXPIRE_DURATION_IN_SECOND.seconds.from_now
end

#precreated_max_expired_at ⇒ Object 



184
185
186
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 184

def precreated_max_expired_at
  SpreeCmCommissioner::WaitingRoomSessionCreator::WAITING_ROOM_SESSION_MAX_DURATION_IN_SECOND.seconds.from_now
end

#previous_date ⇒ Object 



192
193
194
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 192

def previous_date
  1.day.ago.strftime('%Y-%m-%d')
end

#previous_records_path ⇒ Object

Drain target is derived from the server date, never the (possibly stale) lobby pointer.



102
103
104
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 102

def previous_records_path
  default_records_path(previous_date)
end

#publish_logs(started_at:, called_count:) ⇒ Object

Point-in-time snapshot (overwritten each run, never cumulative) of this caller run: last-run timing (heartbeat) and how many guests were called. Capacity state (active/max sessions, available_slots) lives on the lobby doc itself, not here — see #mark_as.



219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 219

def publish_logs(started_at:, called_count:)
  finished_at = Time.zone.now

  logs_document.set(
    {
      called_count: called_count,
      last_started_at: started_at,
      last_finished_at: finished_at,
      last_duration_ms: ((finished_at - started_at) * 1000).round
    }
  )
end

#records_path ⇒ Object

Published path is authoritative; fall back to the server's own date if not yet published.



97
98
99
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 97

def records_path
  lobby_data&.dig(:waiting_guests_records_path).presence || default_records_path(current_date)
end

#service_account ⇒ Object 



255
256
257
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 255

def 
  @service_account ||= Rails.application.credentials.
end

#waiting_exists?(records_paths) ⇒ Boolean

Is anyone still waiting across these partitions? A yes/no (limit 1), so we stop at the first hit instead of counting. uniq folds the pre-flip case (both paths == the same partition before the lobby pointer flips) into a single query, matching #fetch_long_waiting_guests's guard. Order matters for read count only (not correctness): yesterday-first exits after one read on the cross-midnight stragglers the caller drains first.

Returns:

  • (Boolean) 


87
88
89
90
91
92
93
94
 
# File 'app/interactors/spree_cm_commissioner/waiting_guests_caller.rb', line 87

def waiting_exists?(records_paths)
  records_paths.uniq.any? do |records_path|
    firestore.col(records_path)
             .where('allow_to_enter_room_at', '==', nil)
             .limit(1)
             .get.any?
  end
end