Module: Pgbus::CLI::DLQ

Defined in:
lib/pgbus/cli/dlq.rb

Overview

Dead-letter queue management for headless deployments and incident runbooks. Every operation routes through Web::DataSource so the CLI shares the dashboard's exact semantics — origin-queue re-enqueue, transactional retry, lock release on discard, sensitive-payload filtering — with zero raw SQL or direct PGMQ calls.

Constant Summary collapse

ROW_FORMAT = 
"%-10s %-32s %-28s %-8s %s"

Class Method Summary collapse

Class Method Details

.abort_missing(msg_id) ⇒ Object 



144
145
146
147
 
# File 'lib/pgbus/cli/dlq.rb', line 144

def abort_missing(msg_id)
  warn "No dead-letter message found with msg_id #{msg_id}."
  exit 1
end

.list(args, data_source) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/pgbus/cli/dlq.rb', line 34

def list(args, data_source)
  options = parse_list_options(args)
  messages = data_source.dlq_messages(page: options[:page], per_page: options[:per_page])

  if messages.empty?
    puts "No dead-letter messages."
    return
  end

  print_list_table(messages)
  puts "-" * 96
  puts "Total dead-letter messages: #{data_source.dlq_total_count} " \
       "(page #{options[:page]}, #{options[:per_page]} per page)"
end

.origin_queue(dlq_queue_name) ⇒ Object 



131
132
133
 
# File 'lib/pgbus/cli/dlq.rb', line 131

def origin_queue(dlq_queue_name)
  dlq_queue_name.delete_suffix(Pgbus::DEAD_LETTER_SUFFIX)
end

.parse_list_options(args) ⇒ Object 



112
113
114
115
116
117
118
119
120
 
# File 'lib/pgbus/cli/dlq.rb', line 112

def parse_list_options(args)
  options = { page: 1, per_page: 25 }
  OptionParser.new do |opts|
    opts.banner = "Usage: pgbus dlq list [options]"
    opts.on("--page N", Integer, "Page number (default: 1)") { |v| options[:page] = v }
    opts.on("--per-page N", Integer, "Messages per page (default: 25)") { |v| options[:per_page] = v }
  end.parse!(args.dup)
  options
end

.print_help ⇒ Object 



149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/pgbus/cli/dlq.rb', line 149

def print_help
  puts <<~HELP
    Usage: pgbus dlq <subcommand> [options]

    Subcommands:
      list [--page N] [--per-page N]  List dead-letter messages
      show MSG_ID                     Show one message (payload filtered)
      retry MSG_ID                    Re-enqueue a message to its origin queue
      retry-all                       Re-enqueue every dead-letter message
      purge MSG_ID                    Discard a single message
      purge --all --yes               Discard every dead-letter message
  HELP
end

.print_list_table(messages) ⇒ Object 



122
123
124
125
126
127
128
129
 
# File 'lib/pgbus/cli/dlq.rb', line 122

def print_list_table(messages)
  puts format(ROW_FORMAT, "MSG_ID", "DLQ QUEUE", "ORIGIN QUEUE", "READ_CT", "ENQUEUED_AT")
  puts "-" * 96
  messages.each do |m|
    puts format(ROW_FORMAT, m[:msg_id], m[:queue_name], origin_queue(m[:queue_name]),
                m[:read_ct], m[:enqueued_at])
  end
end

.purge(args, data_source) ⇒ Object 



81
82
83
84
85
86
87
 
# File 'lib/pgbus/cli/dlq.rb', line 81

def purge(args, data_source)
  if args.include?("--all")
    purge_all(args, data_source)
  else
    purge_one(args, data_source)
  end
end

.purge_all(args, data_source) ⇒ Object 



102
103
104
105
106
107
108
109
110
 
# File 'lib/pgbus/cli/dlq.rb', line 102

def purge_all(args, data_source)
  unless args.include?("--yes")
    warn "Refusing to purge all dead-letter messages without --yes."
    exit 1
  end

  count = data_source.discard_all_dlq
  puts "Purged #{count} dead-letter message#{"s" unless count == 1}."
end

.purge_one(args, data_source) ⇒ Object 



89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/pgbus/cli/dlq.rb', line 89

def purge_one(args, data_source)
  msg_id = require_msg_id(args)
  detail = data_source.dlq_message_detail(msg_id)
  abort_missing(msg_id) unless detail

  if data_source.discard_dlq_message(detail[:queue_name], detail[:msg_id])
    puts "Purged msg_id #{detail[:msg_id]} from #{detail[:queue_name]}."
  else
    warn "Failed to purge msg_id #{detail[:msg_id]}."
    exit 1
  end
end

.require_msg_id(args) ⇒ Object 



135
136
137
138
139
140
141
142
 
# File 'lib/pgbus/cli/dlq.rb', line 135

def require_msg_id(args)
  msg_id = args.find { |a| a !~ /\A--/ }
  unless msg_id
    warn "A msg_id is required."
    exit 1
  end
  msg_id
end

.retry_all(data_source) ⇒ Object 



76
77
78
79
 
# File 'lib/pgbus/cli/dlq.rb', line 76

def retry_all(data_source)
  count = data_source.retry_all_dlq
  puts "Re-enqueued #{count} dead-letter message#{"s" unless count == 1}."
end

.retry_one(args, data_source) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/pgbus/cli/dlq.rb', line 63

def retry_one(args, data_source)
  msg_id = require_msg_id(args)
  detail = data_source.dlq_message_detail(msg_id)
  abort_missing(msg_id) unless detail

  if data_source.retry_dlq_message(detail[:queue_name], detail[:msg_id])
    puts "Re-enqueued msg_id #{detail[:msg_id]} to #{origin_queue(detail[:queue_name])}."
  else
    warn "Failed to retry msg_id #{detail[:msg_id]}."
    exit 1
  end
end

.show(args, data_source) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/pgbus/cli/dlq.rb', line 49

def show(args, data_source)
  msg_id = require_msg_id(args)
  detail = data_source.dlq_message_detail(msg_id)
  abort_missing(msg_id) unless detail

  puts "msg_id:      #{detail[:msg_id]}"
  puts "dlq queue:   #{detail[:queue_name]}"
  puts "origin:      #{origin_queue(detail[:queue_name])}"
  puts "read_ct:     #{detail[:read_ct]}"
  puts "enqueued_at: #{detail[:enqueued_at]}"
  puts "payload:"
  puts Pgbus::Web::PayloadFilter.filter_json(detail[:message])
end

.start(args, data_source: Pgbus::Web::DataSource.new) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/pgbus/cli/dlq.rb', line 17

def start(args, data_source: Pgbus::Web::DataSource.new)
  subcommand = args.first
  rest = args[1..] || []

  case subcommand
  when "list"       then list(rest, data_source)
  when "show"       then show(rest, data_source)
  when "retry"      then retry_one(rest, data_source)
  when "retry-all"  then retry_all(data_source)
  when "purge"      then purge(rest, data_source)
  else
    warn "Unknown dlq subcommand: #{subcommand.inspect}" if subcommand
    print_help
    exit 1
  end
end