Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

NK-610 Improve channel message db ts precision #1294

Merged
merged 5 commits into from
Dec 3, 2024
Merged

NK-610 Improve channel message db ts precision #1294

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5f61977
NK-610 Ensure message ts ns precision in db.
sesposito Nov 29, 2024
41e7324
Update changelog
sesposito Nov 29, 2024
184be9d
Fix message ack timestamps
sesposito Nov 29, 2024
63262e7
Ensure ts are in UTC
sesposito Nov 29, 2024
63604e3
Minor fix
sesposito Dec 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
34 changes: 17 additions & 17 deletions server/core_channel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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},
}

Expand All @@ -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
}
}
Expand All @@ -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},
}

Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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},
}

Expand All @@ -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
Expand All @@ -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)
Expand Down
Loading