# 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