Class: Lucerna::Identity::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/lucerna/identity/client.rb

Overview

Fire-and-forget delivery of identify calls to People (POST /sdk/v1/people/identify). Mirrors the wire behavior of @lucerna/identity's sync layer, adapted to a server process: a bounded queue drained by one lazy background worker replaces the JS microtask, and flush/close exist because Ruby processes don't drain a microtask queue on exit.

identify validates strictly (bad input is a programming error and fails loud); delivery never raises into the caller — failures report through on_error, and the server's idempotent upsert makes the next identify the retry.

Constant Summary collapse

DEFAULT_BASE_URL = 
"https://api.uselucerna.app"
DEFAULT_REQUEST_TIMEOUT = 
5.0
DEFAULT_MAX_QUEUE = 
1000

Instance Method Summary collapse

Constructor Details

#initialize(api_key:, base_url: nil, request_timeout: nil, on_error: nil, logger: nil, transport: nil, max_queue: DEFAULT_MAX_QUEUE, background: true) ⇒ Client

Returns a new instance of Client.

Raises:

  • (ArgumentError) 


28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/lucerna/identity/client.rb', line 28

def initialize(api_key:, base_url: nil, request_timeout: nil, on_error: nil, logger: nil,
  transport: nil, max_queue: DEFAULT_MAX_QUEUE, background: true)
  raise ArgumentError, "api_key is required" if api_key.nil? || api_key.to_s.empty?

  @url = "#{(base_url || DEFAULT_BASE_URL).sub(%r{/+\z}, "")}/sdk/v1/people/identify"
  @request_timeout = (request_timeout || DEFAULT_REQUEST_TIMEOUT).to_f
  @reporter = Reporting.reporter(on_error, logger)
  @transport = transport || HTTP::NetHttpTransport.new
  @max_queue = max_queue
  @background = background
  @headers = {
    "Authorization" => "Bearer #{api_key}",
    "Content-Type" => "application/json",
    "x-lucerna-sdk" => SDK_IDENTITY,
  }.freeze

  @queue = Thread::Queue.new
  @worker = nil
  @pid = Process.pid
  @lifecycle_mutex = Mutex.new
  @last_sent = nil # worker-thread-private after spawn
end

Instance Method Details

#close(timeout: 2) ⇒ Object

Flushes, then stops the worker. The client stays usable — the next identify re-arms it.



88
89
90
91
92
93
94
95
96
 
# File 'lib/lucerna/identity/client.rb', line 88

def close(timeout: 2)
  drained = flush(timeout: timeout)
  worker = @lifecycle_mutex.synchronize { @worker }
  if worker&.alive?
    @queue << :stop
    worker.join(1)
  end
  drained
end

#flush(timeout: 2) ⇒ Object

Blocks until everything queued before the call has been attempted, or the timeout elapses. Returns true when drained.



78
79
80
81
82
83
84
 
# File 'lib/lucerna/identity/client.rb', line 78

def flush(timeout: 2)
  return true unless @background && @worker&.alive?

  ack = Thread::Queue.new
  @queue << [:flush, ack]
  !ack.pop(timeout: timeout).nil?
end

#identify(identity = nil, **kwargs) ⇒ Object

Accepts a Lucerna::Identity, a hash form, or the same keywords as Identity.new. Never blocks the caller: past max_queue the event is dropped and reported instead of applying backpressure.

Raises:

  • (ArgumentError) 


54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/lucerna/identity/client.rb', line 54

def identify(identity = nil, **kwargs)
  identity = identity.nil? && !kwargs.empty? ? Identity.new(**kwargs) : Identity.wrap(identity)
  raise ArgumentError, "identify requires an identity" if identity.nil?

  payload = JSON.generate(identity.to_payload)
  if @queue.size >= @max_queue
    @reporter.call(Error.new("identity queue full — dropped identify for #{identity.user_id}"))
    return nil
  end

  if @background
    # Fork guard first: in a freshly forked child it clears the
    # inherited backlog — the enqueue below must survive that.
    ensure_worker
    @queue << payload
  else
    @queue << payload
    drain_now # test seam: deliveries run synchronously on the caller
  end
  nil
end