Class: CodexNotify::SlackOutbox

Inherits:
Object
  • Object
show all
Defined in:
lib/codex_notify/slack_outbox.rb

Defined Under Namespace

Classes: CapacityError, Error

Constant Summary collapse

VERSION = 
1
MAX_JOBS = 
10_000
MAX_BYTES = 
64 * 1024 * 1024

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path, clock: -> { Time.now.utc }) ⇒ SlackOutbox

Returns a new instance of SlackOutbox.



18
19
20
21
 
# File 'lib/codex_notify/slack_outbox.rb', line 18

def initialize(path, clock: -> { Time.now.utc })
  @path = Pathname(path)
  @clock = clock
end

Instance Attribute Details

#path ⇒ Object (readonly)

Returns the value of attribute path.



23
24
25
 
# File 'lib/codex_notify/slack_outbox.rb', line 23

def path
  @path
end

Instance Method Details

#complete(job) ⇒ Object 



70
71
72
 
# File 'lib/codex_notify/slack_outbox.rb', line 70

def complete(job)
  with_state_lock { job_path(job, :pending).delete if job_path(job, :pending).exist? }
end

#enqueue(channel:, ordering_key:, generation:, action:, chunks:, recovery_chunks: []) ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/codex_notify/slack_outbox.rb', line 25

def enqueue(channel:, ordering_key:, generation:, action:, chunks:, recovery_chunks: [])
  with_state_lock do
    enforce_capacity!
    sequence = allocate_sequence
    job = {
      'version' => VERSION,
      'id' => SecureRandom.uuid,
      'sequence' => sequence,
      'channel' => channel.to_s,
      'ordering_key' => ordering_key.to_s,
      'generation' => generation.to_i,
      'action' => action.to_s,
      'message_chunks' => redact_chunks(chunks),
      'recovery_root_chunks' => redact_chunks(recovery_chunks),
      'phase' => 'pending',
      'next_chunk' => 0,
      'resolved_thread_ts' => nil,
      'attempt_count' => 0,
      'ambiguous_attempt_count' => 0,
      'next_attempt_at' => nil,
      'last_error' => nil,
      'created_at' => now.iso8601,
      'updated_at' => now.iso8601
    }
    write_job(job)
    job['id']
  end
end

#jobs(status = :pending) ⇒ Object 



61
62
63
 
# File 'lib/codex_notify/slack_outbox.rb', line 61

def jobs(status = :pending)
  directory(status).glob('*.json').map { |file| read_job(file) }.sort_by { |job| job['sequence'] }
end

#move(job, status) ⇒ Object 



74
75
76
77
78
79
80
81
82
 
# File 'lib/codex_notify/slack_outbox.rb', line 74

def move(job, status)
  with_state_lock do
    source = job_path(job, :pending)
    target = job_path(job, status)
    target.dirname.mkpath
    File.chmod(0o700, target.dirname)
    File.rename(source, target) if source.exist?
  end
end

#pending_root?(ordering_key, generation:) ⇒ Boolean

Returns:

  • (Boolean) 


54
55
56
57
58
59
 
# File 'lib/codex_notify/slack_outbox.rb', line 54

def pending_root?(ordering_key, generation:)
  jobs(:pending).any? do |job|
    job['ordering_key'] == ordering_key.to_s && job['generation'] == generation.to_i &&
      %w[ensure_thread root_or_reply].include?(job['action'])
  end
end

#retry(id) ⇒ Object 



107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/codex_notify/slack_outbox.rb', line 107

def retry(id)
  with_state_lock do
    source = %i[needs_review failed].map { |status| directory(status).join("#{id}.json") }.find(&:exist?)
    raise Error, "outbox job not found: #{id}" unless source

    job = read_job(source)
    job['next_attempt_at'] = nil
    job['last_error'] = nil
    job['ambiguous_attempt_count'] = 0
    source.delete
    write_job(job)
  end
end

#status_rows ⇒ Object 



96
97
98
99
100
101
102
103
104
105
 
# File 'lib/codex_notify/slack_outbox.rb', line 96

def status_rows
  %i[pending needs_review failed].flat_map do |status|
    jobs(status).map do |job|
      {
        status: status.to_s.tr('_', '-'), id: job['id'], sequence: job['sequence'],
        created_at: job['created_at'], error: job.dig('last_error', 'code')
      }
    end
  end
end

#try_drain_lock ⇒ Object 



84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/codex_notify/slack_outbox.rb', line 84

def try_drain_lock
  prepare_directories
  File.open(@path.join('locks/drain.lock'), File::RDWR | File::CREAT, 0o600) do |file|
    return false unless file.flock(File::LOCK_EX | File::LOCK_NB)

    yield
    true
  ensure
    file.flock(File::LOCK_UN) rescue nil
  end
end

#update(job) ⇒ Object 



65
66
67
68
 
# File 'lib/codex_notify/slack_outbox.rb', line 65

def update(job)
  job['updated_at'] = now.iso8601
  with_state_lock { write_job(job) }
end