Changeset 781 in code for trunk/downstream.go

Timestamp:

Feb 11, 2022, 6:41:46 PM (3 years ago)

Author:

delthas

Message:

Add support for the wip soju.im/read capability and READ command

READ lets downstream clients share information between each other about
what messages have been read by other downstreams.

Each target/entity has an optional corresponding read receipt, which is
stored as a timestamp.

• When a downstream sends: READ #chan timestamp=2020-01-01T01:23:45.000Z the read receipt for that target is set to that date
• soju sends READ to downstreams:
  • on JOIN, if the client uses the soju.im/read capability
  • when the read receipt timestamp is set by any downstream

The read receipt date is clamped by the previous receipt date and the
current time.

File:

• trunk/downstream.go (modified) (5 diffs)

trunk/downstream.go

@@ -226 +226 @@
         "soju.im/bouncer-networks":        "",
         "soju.im/bouncer-networks-notify": "",
+        "soju.im/read":                    "",
 }
 
@@ -554 +555 @@
                 return
         }
+        if msg.Command == "READ" && !dc.caps["soju.im/read"] {
+                return
+        }
 
         dc.srv.metrics.downstreamOutMessagesTotal.Inc()
@@ -1470 +1474 @@
                         })
 
-                        forwardChannel(dc, ch)
+                        forwardChannel(ctx, dc, ch)
                 }
         })
@@ -1818 +1822 @@
                                         ch.Key = key
                                 }
-                                uc.network.attach(ch)
+                                uc.network.attach(ctx, ch)
                         } else {
                                 ch = &Channel{
@@ -2751 +2755 @@
                         }
                 })
+        case "READ":
+                var target, criteria string
+                if err := parseMessageParams(msg, &target); err != nil {
+                        return ircError{&irc.Message{
+                                Command: "FAIL",
+                                Params:  []string{"READ", "NEED_MORE_PARAMS", "Missing parameters"},
+                        }}
+                }
+                if len(msg.Params) > 1 {
+                        criteria = msg.Params[1]
+                }
+
+                uc, entity, err := dc.unmarshalEntity(target)
+                if err != nil {
+                        return err
+                }
+                entityCM := uc.network.casemap(entity)
+
+                r, err := dc.srv.db.GetReadReceipt(ctx, uc.network.ID, entityCM)
+                if err != nil {
+                        dc.logger.Printf("failed to get the read receipt for %q: %v", entity, err)
+                        return ircError{&irc.Message{
+                                Command: "FAIL",
+                                Params:  []string{"READ", "INTERNAL_ERROR", target, "Internal error"},
+                        }}
+                } else if r == nil {
+                        r = &ReadReceipt{
+                                Target: entityCM,
+                        }
+                }
+
+                broadcast := false
+                if len(criteria) > 0 {
+                        // TODO: support msgid criteria
+                        criteriaParts := strings.SplitN(criteria, "=", 2)
+                        if len(criteriaParts) != 2 || criteriaParts[0] != "timestamp" {
+                                return ircError{&irc.Message{
+                                        Command: "FAIL",
+                                        Params:  []string{"READ", "INVALID_PARAMS", criteria, "Unknown criteria"},
+                                }}
+                        }
+
+                        timestamp, err := time.Parse(serverTimeLayout, criteriaParts[1])
+                        if err != nil {
+                                return ircError{&irc.Message{
+                                        Command: "FAIL",
+                                        Params:  []string{"READ", "INVALID_PARAMS", criteria, "Invalid criteria"},
+                                }}
+                        }
+                        now := time.Now()
+                        if timestamp.After(now) {
+                                timestamp = now
+                        }
+                        if r.Timestamp.Before(timestamp) {
+                                r.Timestamp = timestamp
+                                if err := dc.srv.db.StoreReadReceipt(ctx, uc.network.ID, r); err != nil {
+                                        dc.logger.Printf("failed to store receipt for %q: %v", entity, err)
+                                        return ircError{&irc.Message{
+                                                Command: "FAIL",
+                                                Params:  []string{"READ", "INTERNAL_ERROR", target, "Internal error"},
+                                        }}
+                                }
+                                broadcast = true
+                        }
+                }
+
+                timestampStr := "*"
+                if !r.Timestamp.IsZero() {
+                        timestampStr = fmt.Sprintf("timestamp=%s", r.Timestamp.UTC().Format(serverTimeLayout))
+                }
+                uc.forEachDownstream(func(d *downstreamConn) {
+                        if broadcast || dc.id == d.id {
+                                d.SendMessage(&irc.Message{
+                                        Prefix:  d.prefix(),
+                                        Command: "READ",
+                                        Params:  []string{d.marshalEntity(uc.network, entity), timestampStr},
+                                })
+                        }
+                })
         case "BOUNCER":
                 var subcommand string