@@ 2,6 2,27 @@
require "glimmer-dsl-libui"
require "blather/client"
+require "securerandom"
+require "sqlite3"
+require "xdg"
+
+DATA_DIR = XDG::Data.new.home + "jabber-client-demo"
+DATA_DIR.mkpath
+DB = SQLite3::Database.new(DATA_DIR + "db.sqlite3")
+
+if DB.user_version < 1
+ DB.execute(<<~SQL)
+ CREATE TABLE messages (
+ mam_id TEXT PRIMARY KEY,
+ stanza_id TEXT NOT NULL,
+ conversation TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ stanza TEXT NOT NULL
+ )
+ SQL
+ DB.execute("CREATE TABLE data (key TEXT PRIMARY KEY, value TEXT)")
+ DB.user_version = 1
+end
BLATHER = self
include Glimmer
@@ 12,16 33,34 @@ $roster = [["", ""]]
class Conversation
include Glimmer
- def self.open(jid, m=nil)
+ def self.open(jid)
return if $conversations[jid]
- ($conversations[jid] = new(jid, m)).launch
+ ($conversations[jid] = new(jid)).launch
end
- def initialize(jid, m=nil)
+ def initialize(jid)
@jid = jid
@messages = [["", ""]]
- new_message(m) if m
+ EM.defer do
+ mam_messages = []
+ DB.execute(<<~SQL, [@jid]) do |row|
+ SELECT stanza
+ FROM messages
+ WHERE conversation=?
+ ORDER BY created_at
+ SQL
+ m = Blather::XMPPNode.import(
+ Nokogiri::XML.parse(row[0]).root
+ )
+ mam_messages << m
+ end
+
+ LibUI.queue_main do
+ mam_messages.map! { |m| message_row(m) }
+ @messages.replace(mam_messages + @messages)
+ end
+ end
end
def launch
@@ 32,20 71,28 @@ class Conversation
text_column("Message")
editable false
cell_rows @messages
- @messages.clear
+ @messages.clear if @messages.length == 1 && @messages.first.last == ""
}
horizontal_box {
stretchy false
- message_entry = entry
+ @message_entry = entry
button("Send") {
stretchy false
on_clicked do
- BLATHER.say(@jid, message_entry.text)
- @messages << [ARGV[0], message_entry.text]
- message_entry.text = ""
+ m = message
+ EM.defer do
+ BLATHER << m
+ DB.execute(<<~SQL, [nil, m.id, @jid, m.to_s])
+ INSERT INTO messages
+ (mam_id, stanza_id, conversation, created_at, stanza)
+ VALUES (?,?,?,unixepoch(),?)
+ SQL
+ end
+ @messages << message_row(m)
+ @message_entry.text = ""
end
}
}
@@ 57,6 104,12 @@ class Conversation
}.show
end
+ def message
+ Blather::Stanza::Message.new(@jid, @message_entry.text, :chat).tap { |m|
+ m.id = SecureRandom.uuid
+ }
+ end
+
def format_sender(jid)
BLATHER.my_roster[jid]&.name || jid
end
@@ 123,7 176,69 @@ def xml_child(parent, name, namespace)
child
end
+def sync_mam(last_id)
+ start_mam = Blather::Stanza::Iq.new(:set).tap { |iq|
+ xml_child(iq, :query, "urn:xmpp:mam:2").tap do |query|
+ xml_child(query, :set, "http://jabber.org/protocol/rsm").tap do |rsm|
+ xml_child(rsm, :max, "http://jabber.org/protocol/rsm").tap do |max|
+ max.content = (EM.threadpool_size * 5).to_s
+ end
+ next unless last_id
+
+ xml_child(rsm, :after, "http://jabber.org/protocol/rsm").tap do |after|
+ after.content = last_id
+ end
+ end
+ end
+ }
+
+ client.write_with_handler(start_mam) do |reply|
+ next if reply.error?
+
+ fin = reply.find_first("./ns:fin", ns: "urn:xmpp:mam:2")
+ next unless fin
+
+ handle_rsm_reply_when_idle(fin)
+ end
+end
+
+def handle_rsm_reply_when_idle(fin)
+ unless EM.defers_finished?
+ EM.add_timer(0.1) { handle_rsm_reply_when_idle(fin) }
+ return
+ end
+
+ last = fin.find_first(
+ "./ns:set/ns:last",
+ ns: "http://jabber.org/protocol/rsm"
+ )&.content
+
+ if last
+ DB.execute(<<~SQL, [last, last])
+ INSERT INTO data VALUES ('last_mam_id', ?)
+ ON CONFLICT(key) DO UPDATE SET value=? WHERE key='last_mam_id'
+ SQL
+ end
+ return if fin["complete"].to_s == "true"
+
+ sync_mam(last)
+end
+
def handle_live_message(m, counterpart: m.from.stripped.to_s)
+ mam_id = m.xpath("./ns:stanza-id", ns: "urn:xmpp:sid:0").find { |el|
+ el["by"] == jid.stripped.to_s
+ }&.[]("id")
+ delay = m.delay&.stamp&.to_i || Time.now.to_i
+ DB.execute(<<~SQL, [mam_id, m.id, counterpart, delay, m.to_s])
+ INSERT INTO messages (mam_id, stanza_id, conversation, created_at, stanza) VALUES (?,?,?,?,?)
+ SQL
+
+ if mam_id
+ DB.execute(<<~SQL, [mam_id])
+ UPDATE data SET value=? WHERE key='last_mam_id'
+ SQL
+ end
+
LibUI.queue_main do
conversation = $conversations[counterpart]
if conversation
@@ 148,6 263,11 @@ when_ready do
self << Blather::Stanza::Iq.new(:set).tap { |iq|
xml_child(iq, :enable, "urn:xmpp:carbons:2")
}
+
+ last_mam_id = DB.execute(<<~SQL)[0]&.first
+ SELECT value FROM data WHERE key='last_mam_id' LIMIT 1
+ SQL
+ sync_mam(last_mam_id)
end
message(
@@ 166,6 286,40 @@ message(
handle_carbons(fwd, counterpart: ->(m) { m.to.stripped.to_s })
end
+message "./ns:result", ns: "urn:xmpp:mam:2" do |_, result|
+ fwd = result.xpath("./ns:forwarded", ns: "urn:xmpp:forward:0").first
+ fwd = fwd.find_first("./ns:message", ns: "jabber:client")
+ m = Blather::XMPPNode.import(fwd)
+ next unless m.is_a?(Blather::Stanza::Message) && m.body.present?
+
+ mam_id = result.first["id"]&.to_s
+ # Can't really race because we're checking for something from the past
+ # Any new message inserted isn't the one we're looking for here anyway
+ sent = DB.execute(<<~SQL, [m.id])[0][0]
+ SELECT count(*) FROM messages WHERE stanza_id=? AND mam_id IS NULL
+ SQL
+ if sent < 1
+ counterpart = if m.from.stripped.to_s == jid.stripped.to_s
+ m.to.stripped.to_s
+ else
+ m.from.stripped.to_s
+ end
+ delay =
+ fwd.find_first("./ns:delay", ns: "urn:xmpp:delay")
+ &.[]("stamp")&.then(Time.method(:parse))
+ delay = delay&.to_i || m.delay&.stamp&.to_i || Time.now.to_i
+ DB.execute(<<~SQL, [mam_id, m.id, counterpart, delay, m.to_s])
+ INSERT OR IGNORE INTO messages
+ (mam_id, stanza_id, conversation, created_at, stanza)
+ VALUES (?,?,?,?,?)
+ SQL
+ else
+ DB.execute(<<~SQL, [mam_id, m.id])
+ UPDATE messages SET mam_id=? WHERE stanza_id=?
+ SQL
+ end
+end
+
message :body do |m|
handle_live_message(m)
end