Some tweaks for performance and logging.

spec/more_speech/db/xtdb_spec.clj

@@ -3,7 +3,8 @@
             [more-speech.db.gateway :as gateway]
             [more-speech.db.xtdb :as db]
             [more-speech.config :as config]
-            [more-speech.util.files :refer :all]))
+            [more-speech.util.files :refer :all]
+            [more-speech.db.xtdb :as xtdb]))
 
 (declare db)
 (describe "xtdb gateway implementations"
@@ -16,12 +17,14 @@
     (it "adds and fetches profiles"
       (let [profile {:name "name"}]
         (gateway/add-profile @db 1 profile)
+        (xtdb/sync-db @db)
         (should= profile (gateway/get-profile @db 1)))
       (db/delete-profile @db 1)
       (should-be-nil (gateway/get-profile @db 1)))
 
-    (it "gets and id from a user name"
+    (it "gets an id from a user name"
       (gateway/add-profile @db 1 {:name "name"})
+      (xtdb/sync-db @db)
       (should= 1 (gateway/get-id-from-username @db "name"))))
 
   (it "adds a map of profiles"
@@ -34,12 +37,14 @@
     (it "adds and fetches events"
       (let [event {:id 1 :content "content"}]
         (gateway/add-event @db event)
+        (xtdb/sync-db @db)
         (should= event (gateway/get-event @db 1)))
       (db/delete-event @db 1)
       (should-be-nil (gateway/get-event @db 1)))
 
     (it "checks whether events exist"
       (gateway/add-event @db {:id 1 :content "blah"})
+      (xtdb/sync-db @db)
       (should (gateway/event-exists? @db 1))
       (should-not (gateway/event-exists? @db 2))
       (db/delete-event @db 1)
@@ -83,6 +88,7 @@
   (context "contacts"
     (it "adds and fetches contacts"
       (gateway/add-contacts @db 1 {:name "contact"})
+      (xtdb/sync-db @db)
       (should= {:name "contact"} (gateway/get-contacts @db 1))
       (db/delete-contacts @db 1)
       (should-be-nil (gateway/get-contacts @db 1)))

src/more_speech/config.clj

@@ -9,8 +9,8 @@
 
 (def article-width 120)
 
-(def days-to-read 3) ;how many daily message files to read in.
-(def read-contact-lists-days-ago 2)
+(def days-to-read 0.5) ;how far back in time to load old messages from the database.
+(def read-contact-lists-days-ago 0.5)
 
 (def migration-level 10)
 (def version "2023-02-02T10:17")
@@ -60,3 +60,7 @@
     (xtdb/get-db "prod-db")
 
     (throw (Exception. "No Database Specified"))))
+
+;------Websocket backlog
+
+(def websocket-backlog (atom 0))

src/more_speech/data_storage.clj

@@ -67,14 +67,15 @@
     (relays/load-relays-from-file @config/relays-filename)))
 
 (defn load-events [old-events handler]
-  (loop [events (reverse old-events)
+  (Thread/sleep 60000)
+  (loop [events old-events
          event-count 0]
     (if (empty? events)
       (prn 'done-loading-events)
       (let [event (first events)]
         (when (zero? (rem event-count 100))
-          (prn event-count 'events-loaded (fu/format-time (:created-at event)))
-          (Thread/sleep 100))
+          (prn event-count 'events-loaded (fu/format-time (:created-at event)) 'backlog @config/websocket-backlog)
+          (Thread/sleep 5000))
         (try
           (handlers/handle-text-event handler event)
           (catch Exception e
@@ -146,7 +147,7 @@
   (prn 'read-in-last-n-days 'starting)
   (let [db (get-db)
         now (quot (System/currentTimeMillis) 1000)
-        start-time (- now (* n 86400))
+        start-time (int (- now (* n 86400)))
         event-ids (gateway/get-event-ids-since db start-time)
         events (map #(gateway/get-event db %) event-ids)
         times (map :created-at events)

src/more_speech/db/xtdb.clj

@@ -3,13 +3,16 @@
             [clojure.java.io :as io]
             [xtdb.api :as xt]))
 
+(defn sync-db [db]
+  (xt/sync (:node db)))
+
 (defn add-entity [db type id entity]
-  (let [node (:node db)
-        tx (xt/submit-tx
-             node
-             [[::xt/put
-               (assoc entity :xt/id {:type type :id (bigint id)})]])]
-    (xt/await-tx node tx)))
+  (let [node (:node db)]
+    (xt/submit-tx
+      node
+      [[::xt/put
+        (assoc entity :xt/id {:type type :id (bigint id)})]])
+    ))
 
 (defn make-profile-transaction [id profile]
   (let [id (bigint id)]
@@ -104,7 +107,7 @@
                        :where [[event :created-at event-time]
                                [event :id id]
                                [(>= event-time start-time)]]
-                       :order-by [[event-time :asc]]}
+                       :order-by [[event-time :desc]]}
                      start-time)]
     (map first result))
   )

src/more_speech/nostr/event_handlers.clj

@@ -163,18 +163,23 @@
 (def event-counter (atom {:total 0}))
 
 (defn count-event [envelope url]
-  (let [source (second envelope)
+  (let [type (first envelope)
+        source (second envelope)
         key (str url "|" source)]
-    (swap! event-counter update :total inc)
-    (swap! event-counter update key #(inc (if (nil? %) 0 %)))
-    (when (zero? (mod (:total @event-counter) 1000))
-      (clojure.pprint/pprint @event-counter))))
+    (when (= type "EVENT")
+      (swap! event-counter update :total inc)
+      (swap! event-counter update key #(inc (if (nil? %) 0 %)))
+      (when (zero? (mod (:total @event-counter) 1000))
+        (prn 'websocket-backlog @config/websocket-backlog)
+        (clojure.pprint/pprint @event-counter)))))
 
 (defn handle-notification [envelope url]
   (prn 'NOTICE url envelope))
 
 (defn handle-event [_agent envelope url]
-  (if (not (.startsWith (second envelope) "more-speech"))
+  (swap! config/websocket-backlog dec)
+  (if (and (not= "OK" (first envelope))
+           (not (.startsWith (second envelope) "more-speech")))
     (prn 'strange-message-source url envelope)
     (count-event envelope url))
   (if (not= "EVENT" (first envelope))

src/more_speech/nostr/main.clj

@@ -33,9 +33,9 @@
     (protocol/connect-to-relays)
     (protocol/request-contact-lists-from-relays contact-lists-request-id)
     (when (user-configuration/should-import-metadata? now-in-seconds)
-      (protocol/request-metadata-from-relays metadata-request-id)
+      (protocol/request-metadata-from-relays metadata-request-id (- now-in-seconds 86400))
       (user-configuration/set-last-time-metadata-imported now-in-seconds))
-    (protocol/subscribe-to-relays subscription-id subscription-time)
+    (protocol/subscribe-to-relays subscription-id subscription-time now-in-seconds)
     (handlers/update-relay-panel event-handler)
     (if (user-configuration/should-export-profile? now-in-seconds)
       (do

src/more_speech/nostr/protocol.clj

@@ -18,18 +18,20 @@
 (defn request-contact-lists [relay id]
   (let [now (quot (System/currentTimeMillis) 1000)
         days-ago config/read-contact-lists-days-ago
-        seconds-ago (* days-ago 86400)
+        seconds-ago (int (* days-ago 86400))
         since (int (- now seconds-ago))]
     (relay/send relay ["REQ" id {"kinds" [3] "since" since}])))
 
-(defn request-metadata [relay id]
-  (relay/send relay ["REQ" id {"kinds" [0] "since" 0}]))
+(defn request-metadata [relay id since]
+  (relay/send relay ["REQ" id {"kinds" [0] "since" since}]))
 
 (defn subscribe
   ([relay id]
-   (subscribe relay id (int (- (quot (System/currentTimeMillis) 1000) 86400))))
-  ([relay id since]
-   (relay/send relay ["REQ" id {"since" since}])))
+   (let [now (int (quot (System/currentTimeMillis) 1000))]
+     (subscribe relay id (- now 86400) now)))
+  ([relay id since now]
+   (relay/send relay ["REQ" id {"since" since "until" now}])
+   (relay/send relay ["REQ" (str id "-now") {"since" now}])))
 
 (defn unsubscribe [relay id]
   (relay/send relay ["CLOSE" id]))
@@ -40,7 +42,8 @@
 
 (defn handle-relay-message [relay message]
   (let [url (::ws-relay/url relay)]
-    (send events/event-agent handlers/handle-event message url)))
+    (swap! config/websocket-backlog inc)
+    (send-off events/event-agent handlers/handle-event message url)))
 
 (defn connect-to-relays []
   (let [urls (if (config/is-test-run?)
@@ -67,16 +70,16 @@
         (unsubscribe relay id)
         (request-contact-lists relay id)))))
 
-(defn request-metadata-from-relays [id]
+(defn request-metadata-from-relays [id since]
   (prn 'requesting-metadata)
   (doseq [url (keys @relays)]
     (let [relay (get-in @relays [url :connection])
           read? (get-in @relays [url :read])]
       (when (and read? (some? relay))
         (unsubscribe relay id)
-        (request-metadata relay id)))))
+        (request-metadata relay id since)))))
 
-(defn subscribe-to-relays [id subscription-time]
+(defn subscribe-to-relays [id subscription-time now]
   (let [date (- subscription-time 100)]
     (prn 'subscription-date date (format-time date))
     (doseq [url (keys @relays)]
@@ -84,7 +87,7 @@
             read? (get-in @relays [url :read])]
         (when (and read? (some? relay))
           (unsubscribe relay id)
-          (subscribe relay id date)
+          (subscribe relay id date now)
           (swap! relays assoc-in [url :subscribed] true))))))
 
 (defn unsubscribe-from-relays [id]

src/more_speech/ui/swing/main_window.clj

@@ -18,7 +18,7 @@
 (defrecord seesawHandler []
   handlers/event-handler
   (handle-text-event [_handler event]
-    (invoke-now (article-tree/add-event event)))
+    (invoke-later (article-tree/add-event event)))
   (update-relay-panel [_handler]
     (invoke-later (relay-panel/update-relay-panel))))