Changeset 489 in code for trunk

Timestamp:

Mar 31, 2021, 4:04:13 PM (4 years ago)

Author:

contact

Message:

Save delivery receipts in DB

This avoids loosing history on restart for clients that don't
support chathistory.

Closes: ​https://todo.sr.ht/~emersion/soju/80

Location:

trunk

Files:

• db.go (modified) (4 diffs)
• upstream.go (modified) (1 diff)
• user.go (modified) (9 diffs)

trunk/db.go

@@ -119 +119 @@
         DetachAfter   time.Duration
         DetachOn      MessageFilter
+}
+
+type DeliveryReceipt struct {
+        ID            int64
+        Target        string // channel or nick
+        Client        string
+        InternalMsgID string
 }
 
@@ -161 +168 @@
         FOREIGN KEY(network) REFERENCES Network(id),
         UNIQUE(network, name)
+);
+
+CREATE TABLE DeliveryReceipt (
+        id INTEGER PRIMARY KEY,
+        network INTEGER NOT NULL,
+        target VARCHAR(255) NOT NULL,
+        client VARCHAR(255),
+        internal_msgid VARCHAR(255) NOT NULL,
+        FOREIGN KEY(network) REFERENCES Network(id),
+        UNIQUE(network, target, client)
 );
 `
@@ -218 +235 @@
                 ALTER TABLE Channel ADD COLUMN detach_on INTEGER NOT NULL DEFAULT 0;
         `,
+        `
+                CREATE TABLE DeliveryReceipt (
+                        id INTEGER PRIMARY KEY,
+                        network INTEGER NOT NULL,
+                        target VARCHAR(255) NOT NULL,
+                        client VARCHAR(255),
+                        internal_msgid VARCHAR(255) NOT NULL,
+                        FOREIGN KEY(network) REFERENCES Network(id),
+                        UNIQUE(network, target, client)
+                );
+        `,
 }
 
@@ -579 +607 @@
         return err
 }
+
+func (db *DB) ListDeliveryReceipts(networkID int64) ([]DeliveryReceipt, error) {
+        db.lock.RLock()
+        defer db.lock.RUnlock()
+
+        rows, err := db.db.Query(`SELECT id, target, client, internal_msgid
+                FROM DeliveryReceipt
+                WHERE network = ?`, networkID)
+        if err != nil {
+                return nil, err
+        }
+        defer rows.Close()
+
+        var receipts []DeliveryReceipt
+        for rows.Next() {
+                var rcpt DeliveryReceipt
+                var client sql.NullString
+                if err := rows.Scan(&rcpt.ID, &rcpt.Target, &client, &rcpt.InternalMsgID); err != nil {
+                        return nil, err
+                }
+                rcpt.Client = client.String
+                receipts = append(receipts, rcpt)
+        }
+        if err := rows.Err(); err != nil {
+                return nil, err
+        }
+
+        return receipts, nil
+}
+
+func (db *DB) StoreClientDeliveryReceipts(networkID int64, client string, receipts []DeliveryReceipt) error {
+        db.lock.Lock()
+        defer db.lock.Unlock()
+
+        tx, err := db.db.Begin()
+        if err != nil {
+                return err
+        }
+        defer tx.Rollback()
+
+        _, err = tx.Exec("DELETE FROM DeliveryReceipt WHERE network = ? AND client = ?",
+                networkID, toNullString(client))
+        if err != nil {
+                return err
+        }
+
+        for i := range receipts {
+                rcpt := &receipts[i]
+
+                res, err := tx.Exec("INSERT INTO DeliveryReceipt(network, target, client, internal_msgid) VALUES (?, ?, ?, ?)",
+                        networkID, rcpt.Target, toNullString(client), rcpt.InternalMsgID)
+                if err != nil {
+                        return err
+                }
+                rcpt.ID, err = res.LastInsertId()
+                if err != nil {
+                        return err
+                }
+        }
+
+        return tx.Commit()
+}

trunk/upstream.go

@@ -1753 +1753 @@
                 }
 
-                for clientName, _ := range uc.user.clientNames {
+                uc.network.delivered.ForEachClient(func(clientName string) {
                         uc.network.delivered.StoreID(entity, clientName, lastID)
-                }
+                })
         }
 

trunk/user.go

@@ -90 +90 @@
         for _, entry := range ds.m.innerMap {
                 f(entry.originalKey)
+        }
+}
+
+func (ds deliveredStore) ForEachClient(f func(clientName string)) {
+        clients := make(map[string]struct{})
+        for _, entry := range ds.m.innerMap {
+                delivered := entry.value.(deliveredClientMap)
+                for clientName := range delivered {
+                        clients[clientName] = struct{}{}
+                }
+        }
+
+        for clientName := range clients {
+                f(clientName)
         }
 }
@@ -299 +313 @@
 }
 
+func (net *network) storeClientDeliveryReceipts(clientName string) {
+        if !net.user.hasPersistentMsgStore() {
+                return
+        }
+
+        var receipts []DeliveryReceipt
+        net.delivered.ForEachTarget(func(target string) {
+                msgID := net.delivered.LoadID(target, clientName)
+                if msgID == "" {
+                        return
+                }
+                receipts = append(receipts, DeliveryReceipt{
+                        Target:        target,
+                        InternalMsgID: msgID,
+                })
+        })
+
+        if err := net.user.srv.db.StoreClientDeliveryReceipts(net.ID, clientName, receipts); err != nil {
+                net.user.srv.Logger.Printf("failed to store delivery receipts for user %q, client %q, network %q: %v", net.user.Username, clientName, net.GetName(), err)
+        }
+}
+
 type user struct {
         User
@@ -309 +345 @@
         downstreamConns []*downstreamConn
         msgStore        messageStore
-        clientNames     map[string]struct{}
 
         // LIST commands in progress
@@ -330 +365 @@
 
         return &user{
-                User:        *record,
-                srv:         srv,
-                events:      make(chan event, 64),
-                done:        make(chan struct{}),
-                msgStore:    msgStore,
-                clientNames: make(map[string]struct{}),
+                User:     *record,
+                srv:      srv,
+                events:   make(chan event, 64),
+                done:     make(chan struct{}),
+                msgStore: msgStore,
         }
 }
@@ -407 +441 @@
                 network := newNetwork(u, &record, channels)
                 u.networks = append(u.networks, network)
+
+                if u.hasPersistentMsgStore() {
+                        receipts, err := u.srv.db.ListDeliveryReceipts(record.ID)
+                        if err != nil {
+                                u.srv.Logger.Printf("failed to load delivery receipts for user %q, network %q: %v", u.Username, network.GetName(), err)
+                                return
+                        }
+
+                        for _, rcpt := range receipts {
+                                network.delivered.StoreID(rcpt.Target, rcpt.Client, rcpt.InternalMsgID)
+                        }
+                }
 
                 go network.run()
@@ -490 +536 @@
                                 uc.updateAway()
                         })
-
-                        u.clientNames[dc.clientName] = struct{}{}
                 case eventDownstreamDisconnected:
                         dc := e.dc
@@ -501 +545 @@
                                 }
                         }
+
+                        dc.forEachNetwork(func(net *network) {
+                                net.storeClientDeliveryReceipts(dc.clientName)
+                        })
 
                         u.forEachUpstream(func(uc *upstreamConn) {
@@ -525 +573 @@
                         for _, n := range u.networks {
                                 n.stop()
+
+                                n.delivered.ForEachClient(func(clientName string) {
+                                        n.storeClientDeliveryReceipts(clientName)
+                                })
                         }
                         return
@@ -666 +718 @@
         <-u.done
 }
+
+func (u *user) hasPersistentMsgStore() bool {
+        if u.msgStore == nil {
+                return false
+        }
+        _, isMem := u.msgStore.(*memoryMessageStore)
+        return !isMem
+}