Refactor separate message stores

[snichme]

WHAT is this pull request doing?

Separate Message stores

Queue message stores and Stream message stores has different semantics, one shift out messages while the other just read message at a certain offset.

HOW can this pull request be tested?

....

[claude]

Claude finished @snichme's task —— View job

---

Code Review Complete

After analyzing the message store refactoring, I found several issues that need fixing:

Todo List:

• Analyze the PR context and setup todo list
• Read key files to understand the message store refactoring
• Review new store architecture for bugs and issues
• Check for performance issues and error handling problems
• Examine Crystal-specific anti-patterns
• Post review findings

Issues Found:

1. Potential nil pointer crash in src/lavinmq/store/offset.cr:86

headers.not_nil!("Message lacks headers")["x-stream-offset"].as(Int64)

• The .not_nil! call will crash the program if headers is nil
• This is used in multiple places: offset_from_headers method and duplicated in src/lavinmq/amqp/stream/message_store.cr:209
• Should use proper nil handling instead of crashing

2. Commented out error handling in src/lavinmq/store/offset.cr:80-82

# rescue ex : IndexError # first segment can be empty if message size >= segment size
#   return offset_at(seg + 1, 4_u32, true) unless retried
#   raise ex

• Critical error handling code is commented out in offset_at method
• Could lead to unhandled IndexError exceptions
• Should either remove the commented code or implement proper error handling

3. Inconsistent error handling change in get_last_offset

• The diff shows the rescue block was removed: rescue e; 0i64 → explicit empty check
• While the explicit empty check is better, this changes error handling behavior
• May cause crashes that were previously handled silently

4. Potential resource leak in src/lavinmq/store/offset.cr:194-204

• replace_offsets_file method creates temporary files and uses async cleanup
• If an exception occurs between file creation and the spawn block, the old file handle may leak
• Should use proper exception handling around file operations

---