NK-610 Improve channel message db ts precision (#1294)

Write persisted channel messages create time timestamp with ns precision. This should ensure that the messages are ordered correctly when listed.
heroiclabs · Dec 3, 2024 · 8a72e2c · 8a72e2c
1 parent 0457b78
commit 8a72e2c
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 17 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,8 @@ All notable changes to this project are documented below.
 The format is based on [keep a changelog](http://keepachangelog.com) and this project uses [semantic versioning](http://semver.org).
 
 ## [Unreleased]
+### Fixed
+- Ensure persisted chat messages listing returns correct order.
 
 ## [3.25.0] - 2024-11-25
 ### Added

diff --git a/server/core_channel.go b/server/core_channel.go
@@ -291,16 +291,16 @@ WHERE stream_mode = $1 AND stream_subject = $2::UUID AND stream_descriptor = $3:
 }
 
 func ChannelMessageSend(ctx context.Context, logger *zap.Logger, db *sql.DB, router MessageRouter, channelStream PresenceStream, channelId, content, senderId, senderUsername string, persist bool) (*rtapi.ChannelMessageAck, error) {
-	ts := time.Now().Unix()
+	ts := timestamppb.New(time.Now().UTC())
 	message := &api.ChannelMessage{
 		ChannelId:  channelId,
 		MessageId:  uuid.Must(uuid.NewV4()).String(),
 		Code:       &wrapperspb.Int32Value{Value: ChannelMessageTypeChat},
 		SenderId:   senderId,
 		Username:   senderUsername,
 		Content:    content,
-		CreateTime: &timestamppb.Timestamp{Seconds: ts},
-		UpdateTime: &timestamppb.Timestamp{Seconds: ts},
+		CreateTime: ts,
+		UpdateTime: ts,
 		Persistent: &wrapperspb.BoolValue{Value: persist},
 	}
 
@@ -326,10 +326,9 @@ func ChannelMessageSend(ctx context.Context, logger *zap.Logger, db *sql.DB, rou
 	if persist {
 		query := `INSERT INTO message (id, code, sender_id, username, stream_mode, stream_subject, stream_descriptor, stream_label, content, create_time, update_time)
 VALUES ($1, $2, $3, $4, $5, $6::UUID, $7::UUID, $8, $9, $10, $10)`
-		_, err := db.ExecContext(ctx, query, message.MessageId, message.Code.Value, message.SenderId, message.Username, channelStream.Mode, channelStream.Subject, channelStream.Subcontext, channelStream.Label, message.Content, time.Unix(message.CreateTime.Seconds, 0).UTC())
+		_, err := db.ExecContext(ctx, query, message.MessageId, message.Code.Value, message.SenderId, message.Username, channelStream.Mode, channelStream.Subject, channelStream.Subcontext, channelStream.Label, message.Content, message.CreateTime.AsTime())
 		if err != nil {
 			logger.Error("Error persisting channel message", zap.Error(err))
-
 			return nil, errChannelMessagePersist
 		}
 	}
@@ -340,16 +339,16 @@ VALUES ($1, $2, $3, $4, $5, $6::UUID, $7::UUID, $8, $9, $10, $10)`
 }
 
 func ChannelMessageUpdate(ctx context.Context, logger *zap.Logger, db *sql.DB, router MessageRouter, channelStream PresenceStream, channelId, messageId, content, senderId, senderUsername string, persist bool) (*rtapi.ChannelMessageAck, error) {
-	ts := time.Now().Unix()
+	ts := time.Now().UTC()
 	message := &api.ChannelMessage{
 		ChannelId:  channelId,
 		MessageId:  messageId,
 		Code:       &wrapperspb.Int32Value{Value: ChannelMessageTypeChatUpdate},
 		SenderId:   senderId,
 		Username:   senderUsername,
 		Content:    content,
-		CreateTime: &timestamppb.Timestamp{Seconds: ts},
-		UpdateTime: &timestamppb.Timestamp{Seconds: ts},
+		CreateTime: timestamppb.New(ts),
+		UpdateTime: timestamppb.New(ts),
 		Persistent: &wrapperspb.BoolValue{Value: persist},
 	}
 
@@ -377,7 +376,7 @@ func ChannelMessageUpdate(ctx context.Context, logger *zap.Logger, db *sql.DB, r
 		// First find and update the referenced message.
 		var dbCreateTime pgtype.Timestamptz
 		query := "UPDATE message SET update_time = $5, username = $4, content = $3 WHERE id = $1 AND sender_id = $2 RETURNING create_time"
-		err := db.QueryRowContext(ctx, query, messageId, message.SenderId, message.Content, message.Username, time.Unix(message.UpdateTime.Seconds, 0).UTC()).Scan(&dbCreateTime)
+		err := db.QueryRowContext(ctx, query, messageId, message.SenderId, message.Content, message.Username, message.UpdateTime.AsTime()).Scan(&dbCreateTime)
 		if err != nil {
 			if err == sql.ErrNoRows {
 				return nil, errChannelMessageNotFound
@@ -386,7 +385,7 @@ func ChannelMessageUpdate(ctx context.Context, logger *zap.Logger, db *sql.DB, r
 			return nil, errChannelMessagePersist
 		}
 		// Replace the message create time with the real one from DB.
-		message.CreateTime = &timestamppb.Timestamp{Seconds: dbCreateTime.Time.Unix()}
+		*message.CreateTime = *timestamppb.New(dbCreateTime.Time)
 	}
 
 	router.SendToStream(logger, channelStream, &rtapi.Envelope{Message: &rtapi.Envelope_ChannelMessage{ChannelMessage: message}}, true)
@@ -395,16 +394,16 @@ func ChannelMessageUpdate(ctx context.Context, logger *zap.Logger, db *sql.DB, r
 }
 
 func ChannelMessageRemove(ctx context.Context, logger *zap.Logger, db *sql.DB, router MessageRouter, channelStream PresenceStream, channelId, messageId, senderId, senderUsername string, persist bool) (*rtapi.ChannelMessageAck, error) {
-	ts := time.Now().Unix()
+	ts := time.Now().UTC()
 	message := &api.ChannelMessage{
 		ChannelId:  channelId,
 		MessageId:  messageId,
 		Code:       &wrapperspb.Int32Value{Value: ChannelMessageTypeChatRemove},
 		SenderId:   senderId,
 		Username:   senderUsername,
 		Content:    "{}",
-		CreateTime: &timestamppb.Timestamp{Seconds: ts},
-		UpdateTime: &timestamppb.Timestamp{Seconds: ts},
+		CreateTime: timestamppb.New(ts),
+		UpdateTime: timestamppb.New(ts),
 		Persistent: &wrapperspb.BoolValue{Value: persist},
 	}
 
@@ -430,9 +429,9 @@ func ChannelMessageRemove(ctx context.Context, logger *zap.Logger, db *sql.DB, r
 
 	if persist {
 		// First find and remove the referenced message.
-		var dbCreateTime pgtype.Timestamptz
-		query := "DELETE FROM message WHERE id = $1 AND sender_id = $2 RETURNING create_time"
-		err := db.QueryRowContext(ctx, query, messageId, message.SenderId).Scan(&dbCreateTime)
+		var dbCreateTime, dbUpdateTime pgtype.Timestamptz
+		query := "DELETE FROM message WHERE id = $1 AND sender_id = $2 RETURNING create_time, update_time"
+		err := db.QueryRowContext(ctx, query, messageId, message.SenderId).Scan(&dbCreateTime, &dbUpdateTime)
 		if err != nil {
 			if err == sql.ErrNoRows {
 				return nil, errChannelMessageNotFound
@@ -441,7 +440,8 @@ func ChannelMessageRemove(ctx context.Context, logger *zap.Logger, db *sql.DB, r
 			return nil, errChannelMessagePersist
 		}
 		// Replace the message create time with the real one from DB.
-		message.CreateTime = &timestamppb.Timestamp{Seconds: dbCreateTime.Time.Unix()}
+		*message.CreateTime = *timestamppb.New(dbCreateTime.Time)
+		*message.UpdateTime = *timestamppb.New(dbUpdateTime.Time)
 	}
 
 	router.SendToStream(logger, channelStream, &rtapi.Envelope{Message: &rtapi.Envelope_ChannelMessage{ChannelMessage: message}}, true)