Class: Pinmark::Mcp::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/pinmark/mcp/queue.rb

Overview

Persists the pinmark queue file with atomic writes so concurrent readers (the Rails endpoint and the MCP tools) never see a half-written JSON document.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path = nil) ⇒ Queue

Returns a new instance of Queue.



19
20
21
22
 
# File 'lib/pinmark/mcp/queue.rb', line 19

def initialize(path = nil)
  @path = Pathname.new(path || self.class.default_path).expand_path
  ensure_file
end

Instance Attribute Details

#path ⇒ Object (readonly)

Returns the value of attribute path.



13
14
15
 
# File 'lib/pinmark/mcp/queue.rb', line 13

def path
  @path
end

Class Method Details

.default_path ⇒ Object 



15
16
17
 
# File 'lib/pinmark/mcp/queue.rb', line 15

def self.default_path
  Rails.root.join("tmp/pinmark/queue.json")
end

Instance Method Details

#addressed ⇒ Object 



50
51
52
 
# File 'lib/pinmark/mcp/queue.rb', line 50

def addressed
  read.fetch("annotations", []).select { |entry| entry["status"] == "addressed" }
end

#append(entries) ⇒ Object 



54
55
56
57
58
59
 
# File 'lib/pinmark/mcp/queue.rb', line 54

def append(entries)
  data = read
  data["annotations"] = data.fetch("annotations", []) + Array(entries)
  write(data)
  data["annotations"]
end

#clear_addressed ⇒ Object 



80
81
82
83
84
85
86
87
 
# File 'lib/pinmark/mcp/queue.rb', line 80

def clear_addressed
  data = read
  before = data.fetch("annotations", []).size
  data["annotations"] = data.fetch("annotations", []).reject { |entry| entry["status"] == "addressed" }
  removed = before - data["annotations"].size
  write(data)
  { removed:, remaining: data["annotations"].size }
end

#ensure_file ⇒ Object 



24
25
26
27
28
29
 
# File 'lib/pinmark/mcp/queue.rb', line 24

def ensure_file
  FileUtils.mkdir_p(@path.dirname)
  return if @path.exist?

  write({ "annotations" => [] })
end

#mark_addressed(id) ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/pinmark/mcp/queue.rb', line 61

def mark_addressed(id)
  data = read
  found = false
  already_addressed = false

  data["annotations"] = data.fetch("annotations", []).map do |entry|
    next entry unless entry["id"] == id

    found = true
    already_addressed = true if entry["status"] == "addressed"
    entry.merge("status" => "addressed", "addressed_at" => Time.now.utc.iso8601)
  end

  return { found: false } unless found

  write(data)
  { found: true, already_addressed: }
end

#pending ⇒ Object 



46
47
48
 
# File 'lib/pinmark/mcp/queue.rb', line 46

def pending
  read.fetch("annotations", []).select { |entry| entry["status"] == "pending" }
end

#read ⇒ Object 



31
32
33
34
35
36
37
38
 
# File 'lib/pinmark/mcp/queue.rb', line 31

def read
  ensure_file
  parsed = JSON.parse(@path.read)
  parsed["annotations"] = [] unless parsed["annotations"].is_a?(Array)
  parsed
rescue JSON::ParserError
  { "annotations" => [] }
end

#write(data) ⇒ Object 



40
41
42
43
44
 
# File 'lib/pinmark/mcp/queue.rb', line 40

def write(data)
  tmp = Pathname.new("#{@path}.tmp")
  tmp.write(JSON.pretty_generate(data))
  File.rename(tmp, @path)
end