be9f14ce34
A small helpdesk system: an office Pixel running a patched GrapheneOS Dialer answers technician calls, records both call legs as separate channels, and a Ruby backend transcribes them through Whisper and files an AI summary against the caller. Squashed to a single commit for sharing. No credentials are included; secrets live outside the repo in /etc/helpdesk/env on the server or a gitignored .claude/env.local locally. See .claude/env.local.example for the shape. Start at README.md, then docs/architecture.md.
186 lines
9.9 KiB
Ruby
186 lines
9.9 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
# PgStore — a PostgreSQL-backed drop-in for Helpdesk::Store (the JSON Store lives in domain.rb).
|
|
# Same public interface as the JSON Store: the 9 collection attr_readers, next_id, the three indexes,
|
|
# and the explicit write interface (put_*/delete_*/cas_person_ai_context). Only plain Ruby hashes cross
|
|
# the seam — no Sequel objects leak into Service.
|
|
#
|
|
# Model: a WRITE-THROUGH in-memory CACHE. All durable rows are loaded into symbol-keyed record hashes at
|
|
# boot (jsonb columns → plain string-keyed Ruby, timestamptz → "...Z" ISO8601 strings) so Service READS
|
|
# (`events.values.find`, `people[id]`, `each_value`) are unchanged. Every write goes through an explicit
|
|
# put_*/delete_* that updates the cache AND UPSERTs/DELETEs the row (parameterized via Sequel). Transient
|
|
# broadcasts/commands stay in-memory only (never persisted — same as the JSON Store).
|
|
#
|
|
# Single-writer-process assumption: the cache is authoritative within the one WEBrick process; the
|
|
# reentrant Monitor in Service serializes writes. cas_person_ai_context still takes a real
|
|
# SELECT..FOR UPDATE row lock so the AI-memory read-modify-write is correct even multi-process.
|
|
|
|
# Pin the two runtime gems at the require site: bundler is not installed on the server and the app is
|
|
# deliberately launched with plain `ruby` (dev/prod parity, no bundler dependency).
|
|
# These activate the intended versions from the user-install gem path and fail loud if a wrong major/minor
|
|
# is present. See components/backend/Gemfile for the manifest used by `bundle install` on a server.
|
|
gem "pg", "~> 1.5"
|
|
gem "sequel", "~> 5.106"
|
|
require "sequel"
|
|
require "json"
|
|
require "time"
|
|
|
|
Sequel.extension :pg_json
|
|
|
|
module Helpdesk
|
|
class PgStore
|
|
attr_reader :people, :phone_numbers, :orgs, :campaigns, :dids, :events, :quarantine, :broadcasts, :commands
|
|
|
|
# column spec per table (mirrors db/migrate/001_init.sql). jsonb: plain-Ruby<->pg_jsonb; ts: string<->timestamptz.
|
|
SPECS = {
|
|
orgs: { table: :orgs, pk: :id, cols: %i[id name operator], jsonb: [], ts: [] },
|
|
campaigns: { table: :campaigns, pk: :id, cols: %i[id name operator], jsonb: [], ts: [] },
|
|
dids: { table: :dids, pk: :e164, cols: %i[e164 operator campaign_id], jsonb: [], ts: [] },
|
|
people: { table: :people, pk: :id, cols: %i[id name operator email context position org_id
|
|
source notes_rev ai_context ai_context_rev
|
|
ai_context_pinned_sections ai_context_edited
|
|
ai_context_built_from resolutions_ledger
|
|
ledger_applied_events],
|
|
jsonb: %i[ai_context ai_context_pinned_sections ai_context_built_from resolutions_ledger
|
|
ledger_applied_events], ts: [] },
|
|
phone_numbers: { table: :phone_numbers, pk: :e164, cols: %i[e164 person_id raw_input label], jsonb: [], ts: [] },
|
|
events: { table: :events, pk: :id, cols: %i[id call_id person_id campaign_id did_id
|
|
recording_uuid recording_url started_at answered_at
|
|
ended_at duration_s direction operator_context
|
|
presentation state notes transcript ai_summary
|
|
action_items resolution prior_event_id
|
|
pomoc_ticket_id device_id flags muted on_hold
|
|
audio_route dtmf_log disconnect_cause number
|
|
dialed_did audio_path context_digest caller_facts
|
|
ledger_deltas failure_reason failure_retryable
|
|
suggested_position],
|
|
jsonb: %i[action_items flags dtmf_log caller_facts ledger_deltas],
|
|
ts: %i[started_at answered_at ended_at] }
|
|
}.freeze
|
|
SEQ = { org: "org_id_seq", person: "person_id_seq", campaign: "campaign_id_seq",
|
|
event: "event_id_seq", ledger: "ledger_seq" }.freeze
|
|
|
|
def initialize(url)
|
|
@db = Sequel.connect(url, max_connections: 10)
|
|
@db.extension :pg_json
|
|
@broadcasts = Hash.new { |h, k| h[k] = [] }
|
|
@commands = Hash.new { |h, k| h[k] = [] }
|
|
reload!
|
|
end
|
|
|
|
attr_reader :db
|
|
|
|
# (re)build every cache from the DB. Called at boot; also usable after the importer.
|
|
def reload!
|
|
@orgs = index(load_all(:orgs), :id)
|
|
@campaigns = index(load_all(:campaigns), :id)
|
|
@people = index(load_all(:people), :id)
|
|
@phone_numbers = index(load_all(:phone_numbers), :e164)
|
|
@dids = index(load_all(:dids), :e164)
|
|
@events = index(load_all(:events), :id)
|
|
@quarantine = @db[:quarantine].all.each_with_object({}) do |row, acc|
|
|
acc[row[:recording_uuid]] = { call_id: row[:call_id], url: row[:url],
|
|
audio_path: row[:audio_path], at: iso(row[:at]) }
|
|
end
|
|
self
|
|
end
|
|
|
|
# ---- indexes (over the cache — identical semantics to the JSON Store) ----
|
|
def person_by_e164(e164) = @phone_numbers[e164]&.then { |pn| @people[pn[:person_id]] }
|
|
def did_by_e164(e164) = @dids[e164]
|
|
def event_by_call_id(cid) = @events.values.find { |e| e[:call_id] == cid }
|
|
|
|
def next_id(kind) = @db.get(Sequel.function(:nextval, SEQ.fetch(kind))).to_i
|
|
|
|
# ---- write interface (DB write first, then cache; a DB error propagates and leaves the cache clean) ----
|
|
def put_org(o) = (upsert(:orgs, o); @orgs[o[:id]] = o)
|
|
def put_campaign(c) = (upsert(:campaigns, c); @campaigns[c[:id]] = c)
|
|
def put_person(p) = (upsert(:people, p); @people[p[:id]] = p)
|
|
def put_event(ev) = (upsert(:events, ev); @events[ev[:id]] = ev)
|
|
def put_phone_number(pn) = (upsert(:phone_numbers, pn); @phone_numbers[pn[:e164]] = pn)
|
|
def put_did(d) = (upsert(:dids, d); @dids[d[:e164]] = d)
|
|
|
|
def put_quarantine(uuid, h)
|
|
@db[:quarantine].insert_conflict(
|
|
target: :recording_uuid,
|
|
update: { call_id: Sequel[:excluded][:call_id], url: Sequel[:excluded][:url],
|
|
audio_path: Sequel[:excluded][:audio_path], at: Sequel[:excluded][:at] }
|
|
).insert(recording_uuid: uuid, call_id: h[:call_id], url: h[:url],
|
|
audio_path: h[:audio_path], at: parse_ts(h[:at]))
|
|
@quarantine[uuid] = h
|
|
end
|
|
|
|
def delete_person(id) = (@db[:people].where(id: id).delete; @people.delete(id))
|
|
def delete_phone_number(e164) = (@db[:phone_numbers].where(e164: e164).delete; @phone_numbers.delete(e164))
|
|
def delete_quarantine(uuid) = (@db[:quarantine].where(recording_uuid: uuid).delete; @quarantine.delete(uuid))
|
|
|
|
# AI-memory compare-and-set under a real row lock (correct multi-process). LLM call is done by the
|
|
# caller OUTSIDE this method / txn. Returns the updated person hash, or nil if stale (caller retries).
|
|
def cas_person_ai_context(id, expected_rev, ai_context, built_from)
|
|
ok = false
|
|
@db.transaction do
|
|
row = @db[:people].where(id: id).for_update.first
|
|
break if row.nil? || (row[:ai_context_rev] || 0) != expected_rev
|
|
@db[:people].where(id: id).update(
|
|
ai_context: Sequel.pg_jsonb(ai_context),
|
|
ai_context_rev: expected_rev + 1,
|
|
ai_context_built_from: built_from ? Sequel.pg_jsonb(built_from) : Sequel[:ai_context_built_from]
|
|
)
|
|
ok = true
|
|
end
|
|
return nil unless ok
|
|
p = @people[id]
|
|
p[:ai_context] = ai_context
|
|
p[:ai_context_rev] = expected_rev + 1
|
|
p[:ai_context_built_from] = built_from if built_from
|
|
p
|
|
end
|
|
|
|
# transaction seam for multi-row Service ops (update_person/delete_person/upsert_contacts). The block
|
|
# runs inside a DB transaction; Service still calls put_*/delete_* which participate in it.
|
|
def transaction(&blk) = @db.transaction(&blk)
|
|
|
|
def save(_path) = nil # no-op: writes are immediate
|
|
|
|
private
|
|
|
|
def index(list, key) = list.each_with_object({}) { |r, acc| acc[r[key]] = r }
|
|
def load_all(kind) = @db[SPECS[kind][:table]].all.map { |row| row_to_record(kind, row) }
|
|
|
|
def upsert(kind, rec)
|
|
spec = SPECS[kind]
|
|
row = record_to_row(kind, rec)
|
|
update = spec[:cols].reject { |c| c == spec[:pk] }
|
|
.each_with_object({}) { |c, u| u[c] = Sequel[:excluded][c] }
|
|
@db[spec[:table]].insert_conflict(target: spec[:pk], update: update).insert(row)
|
|
end
|
|
|
|
def record_to_row(kind, rec)
|
|
spec = SPECS[kind]
|
|
spec[:cols].each_with_object({}) do |c, row|
|
|
v = rec[c]
|
|
row[c] = if spec[:jsonb].include?(c) then (v.nil? ? nil : Sequel.pg_jsonb(v))
|
|
elsif spec[:ts].include?(c) then parse_ts(v)
|
|
else v
|
|
end
|
|
end
|
|
end
|
|
|
|
def row_to_record(kind, row)
|
|
spec = SPECS[kind]
|
|
spec[:cols].each_with_object({}) do |c, rec|
|
|
v = row[c]
|
|
rec[c] = if spec[:jsonb].include?(c) then (v.nil? ? nil : plainify(v))
|
|
elsif spec[:ts].include?(c) then iso(v)
|
|
else v
|
|
end
|
|
end
|
|
end
|
|
|
|
# jsonb <-> plain string-keyed Ruby (matches the "nested = string keys" contract). Sequel's pg_json
|
|
# returns JSONBHash/JSONBArray wrappers on read; deep-convert to plain Hash/Array.
|
|
def plainify(o) = JSON.parse(JSON.generate(o))
|
|
def parse_ts(v) = v.nil? ? nil : (v.is_a?(Time) ? v : Time.parse(v).utc)
|
|
def iso(v) = v.nil? ? nil : (v.is_a?(String) ? v : v.utc.iso8601)
|
|
end
|
|
end
|