Merge pull request #1926 from slingamn/readmarker.6

implement draft/read-marker capability

gencapdefs.py

@@ -183,6 +183,12 @@ CAPDEFS = [
         url="https://github.com/ircv3/ircv3-specifications/pull/466",
         standard="draft IRCv3",
     ),
+    CapDef(
+        identifier="ReadMarker",
+        name="draft/read-marker",
+        url="https://github.com/ircv3/ircv3-specifications/pull/489",
+        standard="draft IRCv3",
+    ),
 ]
 
 def validate_defs():

irc/accounts.go

@@ -41,6 +41,7 @@ const (
 	keyCertToAccount           = "account.creds.certfp %s"
 	keyAccountChannels         = "account.channels %s" // channels registered to the account
 	keyAccountLastSeen         = "account.lastseen %s"
+	keyAccountReadMarkers      = "account.readmarkers %s"
 	keyAccountModes            = "account.modes %s"     // user modes for the always-on client as a string
 	keyAccountRealname         = "account.realname %s"  // client realname stored as string
 	keyAccountSuspended        = "account.suspended %s" // client realname stored as string
@@ -647,9 +648,18 @@ func (am *AccountManager) loadModes(account string) (uModes modes.Modes) {
 
 func (am *AccountManager) saveLastSeen(account string, lastSeen map[string]time.Time) {
 	key := fmt.Sprintf(keyAccountLastSeen, account)
+	am.saveTimeMap(account, key, lastSeen)
+}
+
+func (am *AccountManager) saveReadMarkers(account string, readMarkers map[string]time.Time) {
+	key := fmt.Sprintf(keyAccountReadMarkers, account)
+	am.saveTimeMap(account, key, readMarkers)
+}
+
+func (am *AccountManager) saveTimeMap(account, key string, timeMap map[string]time.Time) {
 	var val string
-	if len(lastSeen) != 0 {
+	if len(timeMap) != 0 {
-		text, _ := json.Marshal(lastSeen)
+		text, _ := json.Marshal(timeMap)
 		val = string(text)
 	}
 	err := am.server.store.Update(func(tx *buntdb.Tx) error {
@@ -661,7 +671,7 @@ func (am *AccountManager) saveLastSeen(account string, lastSeen map[string]time.
 		return nil
 	})
 	if err != nil {
-		am.server.logger.Error("internal", "error persisting lastSeen", account, err.Error())
+		am.server.logger.Error("internal", "error persisting timeMap", key, err.Error())
 	}
 }
 
@@ -1739,6 +1749,7 @@ func (am *AccountManager) Unregister(account string, erase bool) error {
 	channelsKey := fmt.Sprintf(keyAccountChannels, casefoldedAccount)
 	joinedChannelsKey := fmt.Sprintf(keyAccountChannelToModes, casefoldedAccount)
 	lastSeenKey := fmt.Sprintf(keyAccountLastSeen, casefoldedAccount)
+	readMarkersKey := fmt.Sprintf(keyAccountReadMarkers, casefoldedAccount)
 	unregisteredKey := fmt.Sprintf(keyAccountUnregistered, casefoldedAccount)
 	modesKey := fmt.Sprintf(keyAccountModes, casefoldedAccount)
 	realnameKey := fmt.Sprintf(keyAccountRealname, casefoldedAccount)
@@ -1801,6 +1812,7 @@ func (am *AccountManager) Unregister(account string, erase bool) error {
 		tx.Delete(channelsKey)
 		tx.Delete(joinedChannelsKey)
 		tx.Delete(lastSeenKey)
+		tx.Delete(readMarkersKey)
 		tx.Delete(modesKey)
 		tx.Delete(realnameKey)
 		tx.Delete(suspendedKey)

irc/caps/defs.go

@@ -7,7 +7,7 @@ package caps
 
 const (
 	// number of recognized capabilities:
-	numCapabs = 28
+	numCapabs = 29
 	// length of the uint64 array that represents the bitset:
 	bitsetLen = 1
 )
@@ -65,6 +65,10 @@ const (
 	// https://github.com/ircv3/ircv3-specifications/pull/398
 	Multiline Capability = iota
 
+	// ReadMarker is the draft IRCv3 capability named "draft/read-marker":
+	// https://github.com/ircv3/ircv3-specifications/pull/489
+	ReadMarker Capability = iota
+
 	// Relaymsg is the proposed IRCv3 capability named "draft/relaymsg":
 	// https://github.com/ircv3/ircv3-specifications/pull/417
 	Relaymsg Capability = iota
@@ -142,6 +146,7 @@ var (
 		"draft/extended-monitor",
 		"draft/languages",
 		"draft/multiline",
+		"draft/read-marker",
 		"draft/relaymsg",
 		"echo-message",
 		"ergo.chat/nope",

irc/channel.go

@@ -881,6 +881,10 @@ func (channel *Channel) Join(client *Client, key string, isSajoin bool, rb *Resp
 		rb.AddFromClient(message.Time, message.Msgid, details.nickMask, details.accountName, isBot, nil, "JOIN", chname)
 	}
 
+	if rb.session.capabilities.Has(caps.ReadMarker) {
+		rb.Add(nil, client.server.name, "MARKREAD", chname, client.GetReadMarker(chcfname))
+	}
+
 	if rb.session.client == client {
 		// don't send topic and names for a SAJOIN of a different client
 		channel.SendTopic(client, rb, false)
@@ -964,10 +968,15 @@ func (channel *Channel) playJoinForSession(session *Session) {
 	client := session.client
 	sessionRb := NewResponseBuffer(session)
 	details := client.Details()
+	chname := channel.Name()
 	if session.capabilities.Has(caps.ExtendedJoin) {
-		sessionRb.Add(nil, details.nickMask, "JOIN", channel.Name(), details.accountName, details.realname)
+		sessionRb.Add(nil, details.nickMask, "JOIN", chname, details.accountName, details.realname)
 	} else {
-		sessionRb.Add(nil, details.nickMask, "JOIN", channel.Name())
+		sessionRb.Add(nil, details.nickMask, "JOIN", chname)
+	}
+	if session.capabilities.Has(caps.ReadMarker) {
+		chcfname := channel.NameCasefolded()
+		sessionRb.Add(nil, client.server.name, "MARKREAD", chname, client.GetReadMarker(chcfname))
 	}
 	channel.SendTopic(client, sessionRb, false)
 	channel.Names(client, sessionRb)

irc/client.go

@@ -40,9 +40,9 @@ const (
 	IRCv3TimestampFormat = utils.IRCv3TimestampFormat
 	// limit the number of device IDs a client can use, as a DoS mitigation
 	maxDeviceIDsPerClient = 64
-	// controls how often often we write an autoreplay-missed client's
+	// maximum total read markers that can be stored
-	// deviceid->lastseentime mapping to the database
+	// (writeback of read markers is controlled by lastSeen logic)
-	lastSeenWriteInterval = time.Hour
+	maxReadMarkers = 256
 )
 
 const (
@@ -83,7 +83,7 @@ type Client struct {
 	languages          []string
 	lastActive         time.Time            // last time they sent a command that wasn't PONG or similar
 	lastSeen           map[string]time.Time // maps device ID (including "") to time of last received command
-	lastSeenLastWrite  time.Time            // last time `lastSeen` was written to the datastore
+	readMarkers        map[string]time.Time // maps casefolded target to time of last read marker
 	loginThrottle      connection_limits.GenericThrottle
 	nextSessionID      int64 // Incremented when a new session is established
 	nick               string
@@ -101,6 +101,7 @@ type Client struct {
 	requireSASL        bool
 	registered         bool
 	registerCmdSent    bool // already sent the draft/register command, can't send it again
+	dirtyTimestamps    bool // lastSeen or readMarkers is dirty
 	registrationTimer  *time.Timer
 	server             *Server
 	skeleton           string
@@ -745,41 +746,23 @@ func (client *Client) playReattachMessages(session *Session) {
 // Touch indicates that we received a line from the client (so the connection is healthy
 // at this time, modulo network latency and fakelag).
 func (client *Client) Touch(session *Session) {
-	var markDirty bool
 	now := time.Now().UTC()
 	client.stateMutex.Lock()
 	if client.registered {
 		client.updateIdleTimer(session, now)
 		if client.alwaysOn {
 			client.setLastSeen(now, session.deviceID)
-			if now.Sub(client.lastSeenLastWrite) > lastSeenWriteInterval {
+			client.dirtyTimestamps = true
-				markDirty = true
-				client.lastSeenLastWrite = now
-			}
 		}
 	}
 	client.stateMutex.Unlock()
-	if markDirty {
-		client.markDirty(IncludeLastSeen)
-	}
 }
 
 func (client *Client) setLastSeen(now time.Time, deviceID string) {
 	if client.lastSeen == nil {
 		client.lastSeen = make(map[string]time.Time)
 	}
-	client.lastSeen[deviceID] = now
+	updateLRUMap(client.lastSeen, deviceID, now, maxDeviceIDsPerClient)
-	// evict the least-recently-used entry if necessary
-	if maxDeviceIDsPerClient < len(client.lastSeen) {
-		var minLastSeen time.Time
-		var minClientId string
-		for deviceID, lastSeen := range client.lastSeen {
-			if minLastSeen.IsZero() || lastSeen.Before(minLastSeen) {
-				minClientId, minLastSeen = deviceID, lastSeen
-			}
-		}
-		delete(client.lastSeen, minClientId)
-	}
 }
 
 func (client *Client) updateIdleTimer(session *Session, now time.Time) {
@@ -1191,7 +1174,6 @@ func (client *Client) Quit(message string, session *Session) {
 func (client *Client) destroy(session *Session) {
 	config := client.server.Config()
 	var sessionsToDestroy []*Session
-	var saveLastSeen bool
 	var quitMessage string
 
 	client.stateMutex.Lock()
@@ -1223,20 +1205,6 @@ func (client *Client) destroy(session *Session) {
 		}
 	}
 
-	// save last seen if applicable:
-	if alwaysOn {
-		if client.accountSettings.AutoreplayMissed {
-			saveLastSeen = true
-		} else {
-			for _, session := range sessionsToDestroy {
-				if session.deviceID != "" {
-					saveLastSeen = true
-					break
-				}
-			}
-		}
-	}
-
 	// should we destroy the whole client this time?
 	shouldDestroy := !client.destroyed && remainingSessions == 0 && !alwaysOn
 	// decrement stats on a true destroy, or for the removal of the last connected session
@@ -1246,9 +1214,6 @@ func (client *Client) destroy(session *Session) {
 		// if it's our job to destroy it, don't let anyone else try
 		client.destroyed = true
 	}
-	if saveLastSeen {
-		client.dirtyBits |= IncludeLastSeen
-	}
 
 	becameAutoAway := false
 	var awayMessage string
@@ -1266,14 +1231,6 @@ func (client *Client) destroy(session *Session) {
 
 	client.stateMutex.Unlock()
 
-	// XXX there is no particular reason to persist this state here rather than
-	// any other place: it would be correct to persist it after every `Touch`. However,
-	// I'm not comfortable introducing that many database writes, and I don't want to
-	// design a throttle.
-	if saveLastSeen {
-		client.wakeWriter()
-	}
-
 	// destroy all applicable sessions:
 	for _, session := range sessionsToDestroy {
 		if session.client != client {
@@ -1784,18 +1741,13 @@ func (client *Client) handleRegisterTimeout() {
 func (client *Client) copyLastSeen() (result map[string]time.Time) {
 	client.stateMutex.RLock()
 	defer client.stateMutex.RUnlock()
-	result = make(map[string]time.Time, len(client.lastSeen))
+	return utils.CopyMap(client.lastSeen)
-	for id, lastSeen := range client.lastSeen {
-		result[id] = lastSeen
-	}
-	return
 }
 
 // these are bit flags indicating what part of the client status is "dirty"
 // and needs to be read from memory and written to the db
 const (
 	IncludeChannels uint = 1 << iota
-	IncludeLastSeen
 	IncludeUserModes
 	IncludeRealname
 )
@@ -1853,9 +1805,6 @@ func (client *Client) performWrite(additionalDirtyBits uint) {
 		}
 		client.server.accounts.saveChannels(account, channelToModes)
 	}
-	if (dirtyBits & IncludeLastSeen) != 0 {
-		client.server.accounts.saveLastSeen(account, client.copyLastSeen())
-	}
 	if (dirtyBits & IncludeUserModes) != 0 {
 		uModes := make(modes.Modes, 0, len(modes.SupportedUserModes))
 		for _, m := range modes.SupportedUserModes {

irc/commands.go

@@ -53,7 +53,7 @@ func (cmd *Command) Run(server *Server, client *Client, session *Session, msg ir
 	}
 
 	if client.registered {
-		client.Touch(session)
+		client.Touch(session) // even if `exiting`, we bump the lastSeen timestamp
 	}
 
 	return exiting
@@ -178,6 +178,10 @@ func init() {
 			handler:   lusersHandler,
 			minParams: 0,
 		},
+		"MARKREAD": {
+			handler:   markReadHandler,
+			minParams: 0, // send FAIL instead of ERR_NEEDMOREPARAMS
+		},
 		"MODE": {
 			handler:   modeHandler,
 			minParams: 1,

irc/getters.go

@@ -493,6 +493,63 @@ func (client *Client) checkAlwaysOnExpirationNoMutex(config *Config, ignoreRegis
 	return true
 }
 
+func (client *Client) GetReadMarker(cfname string) (result string) {
+	client.stateMutex.RLock()
+	t, ok := client.readMarkers[cfname]
+	client.stateMutex.RUnlock()
+	if ok {
+		return t.Format(IRCv3TimestampFormat)
+	}
+	return "*"
+}
+
+func (client *Client) copyReadMarkers() (result map[string]time.Time) {
+	client.stateMutex.RLock()
+	defer client.stateMutex.RUnlock()
+	return utils.CopyMap(client.readMarkers)
+}
+
+func (client *Client) SetReadMarker(cfname string, now time.Time) (result time.Time) {
+	client.stateMutex.Lock()
+	defer client.stateMutex.Unlock()
+
+	if client.readMarkers == nil {
+		client.readMarkers = make(map[string]time.Time)
+	}
+	result = updateLRUMap(client.readMarkers, cfname, now, maxReadMarkers)
+	client.dirtyTimestamps = true
+	return
+}
+
+func updateLRUMap(lru map[string]time.Time, key string, val time.Time, maxItems int) (result time.Time) {
+	if currentVal := lru[key]; currentVal.After(val) {
+		return currentVal
+	}
+
+	lru[key] = val
+	// evict the least-recently-used entry if necessary
+	if maxItems < len(lru) {
+		var minKey string
+		var minVal time.Time
+		for key, val := range lru {
+			if minVal.IsZero() || val.Before(minVal) {
+				minKey, minVal = key, val
+			}
+		}
+		delete(lru, minKey)
+	}
+	return val
+}
+
+func (client *Client) shouldFlushTimestamps() (result bool) {
+	client.stateMutex.Lock()
+	defer client.stateMutex.Unlock()
+
+	result = client.dirtyTimestamps && client.registered && client.alwaysOn
+	client.dirtyTimestamps = false
+	return
+}
+
 func (channel *Channel) Name() string {
 	channel.stateMutex.RLock()
 	defer channel.stateMutex.RUnlock()

irc/handlers.go

@@ -2697,6 +2697,50 @@ func verifyHandler(server *Server, client *Client, msg ircmsg.Message, rb *Respo
 	return
 }
 
+// MARKREAD <target> [timestamp]
+func markReadHandler(server *Server, client *Client, msg ircmsg.Message, rb *ResponseBuffer) (exiting bool) {
+	if len(msg.Params) == 0 {
+		rb.Add(nil, server.name, "FAIL", "MARKREAD", "NEED_MORE_PARAMS", client.t("Missing parameters"))
+		return
+	}
+
+	target := msg.Params[0]
+	cftarget, err := CasefoldTarget(target)
+	if err != nil {
+		rb.Add(nil, server.name, "FAIL", "MARKREAD", "INVALID_PARAMS", utils.SafeErrorParam(target), client.t("Invalid target"))
+		return
+	}
+	unfoldedTarget := server.UnfoldName(cftarget)
+
+	// "MARKREAD client get command": MARKREAD <target>
+	if len(msg.Params) == 1 {
+		rb.Add(nil, client.server.name, "MARKREAD", unfoldedTarget, client.GetReadMarker(cftarget))
+		return
+	}
+
+	// "MARKREAD client set command": MARKREAD <target> <timestamp>
+	readTimestamp := msg.Params[1]
+	readTime, err := time.Parse(IRCv3TimestampFormat, readTimestamp)
+	if err != nil {
+		rb.Add(nil, server.name, "FAIL", "MARKREAD", "INVALID_PARAMS", utils.SafeErrorParam(readTimestamp), client.t("Invalid timestamp"))
+		return
+	}
+	result := client.SetReadMarker(cftarget, readTime)
+	readTimestamp = result.Format(IRCv3TimestampFormat)
+	// inform the originating session whether it was a success or a no-op:
+	rb.Add(nil, server.name, "MARKREAD", unfoldedTarget, readTimestamp)
+	if result.Equal(readTime) {
+		// successful update (i.e. it moved the stored timestamp forward):
+		// inform other sessions
+		for _, session := range client.Sessions() {
+			if session != rb.session {
+				session.Send(nil, server.name, "MARKREAD", unfoldedTarget, readTimestamp)
+			}
+		}
+	}
+	return
+}
+
 // REHASH
 func rehashHandler(server *Server, client *Client, msg ircmsg.Message, rb *ResponseBuffer) bool {
 	nick := client.Nick()

irc/help.go

@@ -320,6 +320,13 @@ channels). <elistcond>s modify how the channels are selected.`,
 Shows statistics about the size of the network. If <mask> is given, only
 returns stats for servers matching the given mask.  If <server> is given, the
 command is processed by that server.`,
+	},
+	"markread": {
+		text: `MARKREAD <target> [timestamp]
+
+MARKREAD updates an IRCv3 read message marker. It is not intended for use by
+end users. For more details, see the latest draft of the read-marker
+specification.`,
 	},
 	"mode": {
 		text: `MODE <target> [<modestring> [<mode arguments>...]]

irc/server.go

@@ -36,7 +36,7 @@ import (
 )
 
 const (
-	alwaysOnExpirationPollPeriod = time.Hour
+	alwaysOnMaintenanceInterval = 30 * time.Minute
 )
 
 var (
@@ -119,7 +119,7 @@ func NewServer(config *Config, logger *logger.Manager) (*Server, error) {
 	signal.Notify(server.exitSignals, utils.ServerExitSignals...)
 	signal.Notify(server.rehashSignal, syscall.SIGHUP)
 
-	time.AfterFunc(alwaysOnExpirationPollPeriod, server.handleAlwaysOnExpirations)
+	time.AfterFunc(alwaysOnMaintenanceInterval, server.periodicAlwaysOnMaintenance)
 
 	return server, nil
 }
@@ -132,11 +132,11 @@ func (server *Server) Shutdown() {
 	//TODO(dan): Make sure we disallow new nicks
 	for _, client := range server.clients.AllClients() {
 		client.Notice("Server is shutting down")
-		if client.AlwaysOn() {
-			client.Store(IncludeLastSeen)
-		}
 	}
 
+	// flush data associated with always-on clients:
+	server.performAlwaysOnMaintenance(false, true)
+
 	if err := server.store.Close(); err != nil {
 		server.logger.Error("shutdown", fmt.Sprintln("Could not close datastore:", err))
 	}
@@ -244,25 +244,32 @@ func (server *Server) checkTorLimits() (banned bool, message string) {
 	}
 }
 
-func (server *Server) handleAlwaysOnExpirations() {
+func (server *Server) periodicAlwaysOnMaintenance() {
 	defer func() {
 		// reschedule whether or not there was a panic
-		time.AfterFunc(alwaysOnExpirationPollPeriod, server.handleAlwaysOnExpirations)
+		time.AfterFunc(alwaysOnMaintenanceInterval, server.periodicAlwaysOnMaintenance)
 	}()
 
 	defer server.HandlePanic()
 
+	server.logger.Info("accounts", "Performing periodic always-on client checks")
+	server.performAlwaysOnMaintenance(true, true)
+}
+
+func (server *Server) performAlwaysOnMaintenance(checkExpiration, flushTimestamps bool) {
 	config := server.Config()
-	deadline := time.Duration(config.Accounts.Multiclient.AlwaysOnExpiration)
-	if deadline == 0 {
-		return
-	}
-	server.logger.Info("accounts", "Checking always-on clients for expiration")
 	for _, client := range server.clients.AllClients() {
-		if client.IsExpiredAlwaysOn(config) {
+		if checkExpiration && client.IsExpiredAlwaysOn(config) {
 			// TODO save the channels list, use it for autojoin if/when they return?
 			server.logger.Info("accounts", "Expiring always-on client", client.AccountName())
 			client.destroy(nil)
+			continue
+		}
+
+		if flushTimestamps && client.shouldFlushTimestamps() {
+			account := client.Account()
+			server.accounts.saveLastSeen(account, client.copyLastSeen())
+			server.accounts.saveReadMarkers(account, client.copyReadMarkers())
 		}
 	}
 }

irc/strings.go

@@ -165,6 +165,17 @@ func CasefoldName(name string) (string, error) {
 	return lowered, err
 }
 
+// CasefoldTarget returns a casefolded version of an IRC target, i.e.
+// it determines whether the target is a channel name or nickname and
+// applies the appropriate casefolding rules.
+func CasefoldTarget(name string) (string, error) {
+	if strings.HasPrefix(name, "#") {
+		return CasefoldChannel(name)
+	} else {
+		return CasefoldName(name)
+	}
+}
+
 // returns true if the given name is a valid ident, using a mix of Insp and
 // Chary's ident restrictions.
 func isIdent(name string) bool {

irctest

@@ -1 +1 @@
-Subproject commit 8356ace0144297885f6d51e334763df5223b933f
+Subproject commit f52f21897bc7f37075fca4e9c2a0121b9edf9392