Skip to content

Commit e1a9e3f

Browse files
AchoArnoldCopilot
AchoArnold
and
Copilot
committed
feat(api): count unread thread items
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 332cfd7b-8a60-4e28-9ea8-61854aa712c3
1 parent b38328d commit e1a9e3f

3 files changed

Lines changed: 685 additions & 105 deletions

File tree

‎api/pkg/repositories/gorm_message_thread_repository.go‎

Lines changed: 254 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -45,17 +45,13 @@ func messageThreadActivityUpdates(params MessageThreadActivityUpdate) map[string
4545
if params.Unarchive {
4646
updates["is_archived"] = false
4747
}
48-
if params.MarkAsUnread {
49-
updates["is_read"] = gorm.Expr(
50-
"CASE WHEN last_read_at < ? THEN ? ELSE is_read END",
51-
params.EventTimestamp,
52-
false,
53-
)
54-
}
5548
return updates
5649
}
5750

5851
func messageThreadDeletedUpdates(params MessageThreadDeletedUpdate) map[string]any {
52+
if !params.UpdateLastMessage {
53+
return map[string]any{}
54+
}
5955
return map[string]any{
6056
"last_message_id": params.LastMessageID,
6157
"last_message_content": params.LastMessageContent,
@@ -68,15 +64,65 @@ func messageThreadStatusUpdates(params MessageThreadStatusUpdate) map[string]any
6864
if params.IsArchived != nil {
6965
updates["is_archived"] = *params.IsArchived
7066
}
71-
if params.IsRead != nil {
72-
updates["is_read"] = *params.IsRead
73-
if *params.IsRead {
74-
updates["last_read_at"] = params.ReadAt
75-
}
67+
if params.UnreadCount != nil {
68+
updates["unread_count"] = 0
69+
updates["last_read_at"] = params.ReadAt
7670
}
7771
return updates
7872
}
7973

74+
func lockMessageThread(tx *gorm.DB, userID entities.UserID, threadID uuid.UUID) (*entities.MessageThread, error) {
75+
thread := new(entities.MessageThread)
76+
err := tx.
77+
Clauses(clause.Locking{Strength: "UPDATE"}).
78+
Where("user_id = ?", userID).
79+
Where("id = ?", threadID).
80+
First(thread).
81+
Error
82+
if errors.Is(err, gorm.ErrRecordNotFound) {
83+
return nil, stacktrace.PropagateWithCodef(
84+
err,
85+
ErrCodeNotFound,
86+
"message thread with ID [%s] for user [%s] does not exist",
87+
threadID,
88+
userID,
89+
)
90+
}
91+
if err != nil {
92+
return nil, stacktrace.Propagatef(err, "cannot lock message thread with ID [%s] for user [%s]", threadID, userID)
93+
}
94+
return thread, nil
95+
}
96+
97+
func insertUnreadItem(tx *gorm.DB, item entities.MessageThreadUnreadItem) (bool, error) {
98+
result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&item)
99+
if result.Error != nil {
100+
return false, stacktrace.Propagatef(
101+
result.Error,
102+
"cannot insert unread ledger item for message [%s] in thread [%s]",
103+
item.MessageID,
104+
item.MessageThreadID,
105+
)
106+
}
107+
return result.RowsAffected == 1, nil
108+
}
109+
110+
func deleteUnreadItem(tx *gorm.DB, messageID uuid.UUID, threadID uuid.UUID) (bool, error) {
111+
result := tx.
112+
Where("message_id = ?", messageID).
113+
Where("message_thread_id = ?", threadID).
114+
Delete(&entities.MessageThreadUnreadItem{})
115+
if result.Error != nil {
116+
return false, stacktrace.Propagatef(
117+
result.Error,
118+
"cannot delete unread ledger item for message [%s] in thread [%s]",
119+
messageID,
120+
threadID,
121+
)
122+
}
123+
return result.RowsAffected == 1, nil
124+
}
125+
80126
func (repository *gormMessageThreadRepository) DeleteAllForUser(ctx context.Context, userID entities.UserID) error {
81127
ctx, span := repository.tracer.Start(ctx)
82128
defer span.End()
@@ -106,39 +152,98 @@ func (repository *gormMessageThreadRepository) UpdateAfterDeletedMessage(ctx con
106152
ctx, span := repository.tracer.Start(ctx)
107153
defer span.End()
108154

109-
result := repository.db.WithContext(ctx).
110-
Model(&entities.MessageThread{}).
111-
Where("user_id = ?", params.UserID).
112-
Where("id = ?", params.MessageThreadID).
113-
Updates(messageThreadDeletedUpdates(params))
114-
if result.Error != nil {
115-
return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(result.Error, "cannot update deleted-message metadata for thread [%s]", params.MessageThreadID))
155+
err := repository.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
156+
thread, err := lockMessageThread(tx, params.UserID, params.MessageThreadID)
157+
if err != nil {
158+
return err
159+
}
160+
161+
deleted, err := deleteUnreadItem(tx, params.DeletedMessageID, params.MessageThreadID)
162+
if err != nil {
163+
return err
164+
}
165+
if deleted {
166+
if err := tx.
167+
Model(thread).
168+
Where("user_id = ?", params.UserID).
169+
Where("id = ?", params.MessageThreadID).
170+
UpdateColumn("unread_count", gorm.Expr("GREATEST(unread_count - 1, 0)")).
171+
Error; err != nil {
172+
return stacktrace.Propagatef(
173+
err,
174+
"cannot decrement unread count for thread [%s] and user [%s]",
175+
params.MessageThreadID,
176+
params.UserID,
177+
)
178+
}
179+
if thread.UnreadCount > 0 {
180+
thread.UnreadCount--
181+
}
182+
}
183+
184+
if !params.UpdateLastMessage {
185+
return nil
186+
}
187+
if err := tx.
188+
Model(thread).
189+
Where("user_id = ?", params.UserID).
190+
Where("id = ?", params.MessageThreadID).
191+
Updates(messageThreadDeletedUpdates(params)).
192+
Error; err != nil {
193+
return stacktrace.Propagatef(
194+
err,
195+
"cannot update deleted-message metadata for thread [%s] and user [%s]",
196+
params.MessageThreadID,
197+
params.UserID,
198+
)
199+
}
200+
return nil
201+
})
202+
if err != nil {
203+
return repository.tracer.WrapErrorSpan(
204+
span,
205+
stacktrace.Propagatef(
206+
err,
207+
"cannot apply deleted message [%s] to thread [%s] for user [%s]",
208+
params.DeletedMessageID,
209+
params.MessageThreadID,
210+
params.UserID,
211+
),
212+
)
116213
}
117214

118215
return nil
119216
}
120217

121218
// Store a new entities.MessageThread
122-
func (repository *gormMessageThreadRepository) Store(ctx context.Context, thread *entities.MessageThread) error {
219+
func (repository *gormMessageThreadRepository) Store(ctx context.Context, thread *entities.MessageThread, unreadMessageID *uuid.UUID) error {
123220
ctx, span := repository.tracer.Start(ctx)
124221
defer span.End()
125222

126-
isRead := thread.IsRead
127223
err := repository.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
128224
result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(thread)
129-
thread.IsRead = isRead
130225
if result.Error != nil {
131-
return result.Error
226+
return stacktrace.Propagatef(result.Error, "cannot insert message thread with ID [%s]", thread.ID)
132227
}
133-
if result.RowsAffected == 0 || isRead {
228+
if result.RowsAffected == 0 || unreadMessageID == nil {
134229
return nil
135230
}
136231

137-
return tx.Model(&entities.MessageThread{}).
138-
Where("user_id = ?", thread.UserID).
139-
Where("id = ?", thread.ID).
140-
UpdateColumn("is_read", false).
141-
Error
232+
inserted, err := insertUnreadItem(tx, entities.MessageThreadUnreadItem{
233+
MessageID: *unreadMessageID,
234+
MessageThreadID: thread.ID,
235+
})
236+
if err != nil {
237+
return err
238+
}
239+
if !inserted {
240+
return stacktrace.NewErrorf(
241+
"unread ledger item for message [%s] was not inserted for new thread [%s]",
242+
*unreadMessageID,
243+
thread.ID,
244+
)
245+
}
246+
return nil
142247
})
143248
if err != nil {
144249
return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot save message thread with ID [%s]", thread.ID))
@@ -152,16 +257,63 @@ func (repository *gormMessageThreadRepository) UpdateActivity(ctx context.Contex
152257
ctx, span := repository.tracer.Start(ctx)
153258
defer span.End()
154259

155-
result := repository.db.WithContext(ctx).
156-
Model(&entities.MessageThread{}).
157-
Where("user_id = ?", params.UserID).
158-
Where("id = ?", params.MessageThreadID).
159-
Updates(messageThreadActivityUpdates(params))
160-
if result.Error != nil {
161-
return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(result.Error, "cannot update message activity for thread [%s]", params.MessageThreadID))
162-
}
163-
if result.RowsAffected == 0 {
164-
return repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCodef(gorm.ErrRecordNotFound, ErrCodeNotFound, "thread with id [%s] not found", params.MessageThreadID))
260+
err := repository.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
261+
thread, err := lockMessageThread(tx, params.UserID, params.MessageThreadID)
262+
if err != nil {
263+
return err
264+
}
265+
266+
if err := tx.
267+
Model(thread).
268+
Where("user_id = ?", params.UserID).
269+
Where("id = ?", params.MessageThreadID).
270+
Updates(messageThreadActivityUpdates(params)).
271+
Error; err != nil {
272+
return stacktrace.Propagatef(
273+
err,
274+
"cannot update message activity for thread [%s] and user [%s]",
275+
params.MessageThreadID,
276+
params.UserID,
277+
)
278+
}
279+
if !params.CountAsUnread || !params.EventTimestamp.After(thread.LastReadAt) {
280+
return nil
281+
}
282+
283+
inserted, err := insertUnreadItem(tx, entities.MessageThreadUnreadItem{
284+
MessageID: params.MessageID,
285+
MessageThreadID: params.MessageThreadID,
286+
})
287+
if err != nil || !inserted {
288+
return err
289+
}
290+
291+
if err := tx.
292+
Model(thread).
293+
Where("user_id = ?", params.UserID).
294+
Where("id = ?", params.MessageThreadID).
295+
UpdateColumn("unread_count", gorm.Expr("unread_count + ?", 1)).
296+
Error; err != nil {
297+
return stacktrace.Propagatef(
298+
err,
299+
"cannot increment unread count for thread [%s] and user [%s]",
300+
params.MessageThreadID,
301+
params.UserID,
302+
)
303+
}
304+
thread.UnreadCount++
305+
return nil
306+
})
307+
if err != nil {
308+
return repository.tracer.WrapErrorSpan(
309+
span,
310+
stacktrace.Propagatef(
311+
err,
312+
"cannot update message activity for thread [%s] and user [%s]",
313+
params.MessageThreadID,
314+
params.UserID,
315+
),
316+
)
165317
}
166318

167319
return nil
@@ -177,18 +329,70 @@ func (repository *gormMessageThreadRepository) UpdateStatus(
177329
ctx, span := repository.tracer.Start(ctx)
178330
defer span.End()
179331

180-
thread := new(entities.MessageThread)
181-
result := repository.db.WithContext(ctx).
182-
Model(thread).
183-
Clauses(clause.Returning{}).
184-
Where("user_id = ?", userID).
185-
Where("id = ?", messageThreadID).
186-
Updates(messageThreadStatusUpdates(params))
187-
if result.Error != nil {
188-
return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(result.Error, "cannot update status for thread [%s] and user [%s]", messageThreadID, userID))
332+
if params.UnreadCount != nil && *params.UnreadCount != 0 {
333+
return nil, repository.tracer.WrapErrorSpan(
334+
span,
335+
stacktrace.NewErrorf(
336+
"cannot set unread count to [%d] for thread [%s] and user [%s]: only zero is supported",
337+
*params.UnreadCount,
338+
messageThreadID,
339+
userID,
340+
),
341+
)
189342
}
190-
if result.RowsAffected == 0 {
191-
return nil, repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCodef(gorm.ErrRecordNotFound, ErrCodeNotFound, "thread with id [%s] not found for user with ID [%s]", messageThreadID, userID))
343+
344+
var thread *entities.MessageThread
345+
err := repository.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
346+
var err error
347+
thread, err = lockMessageThread(tx, userID, messageThreadID)
348+
if err != nil {
349+
return err
350+
}
351+
352+
updates := messageThreadStatusUpdates(params)
353+
if len(updates) > 0 {
354+
if err := tx.
355+
Model(thread).
356+
Clauses(clause.Returning{}).
357+
Where("user_id = ?", userID).
358+
Where("id = ?", messageThreadID).
359+
Updates(updates).
360+
Error; err != nil {
361+
return stacktrace.Propagatef(
362+
err,
363+
"cannot update status for thread [%s] and user [%s]",
364+
messageThreadID,
365+
userID,
366+
)
367+
}
368+
}
369+
if params.IsArchived != nil {
370+
thread.IsArchived = *params.IsArchived
371+
}
372+
if params.UnreadCount == nil {
373+
return nil
374+
}
375+
376+
thread.UnreadCount = *params.UnreadCount
377+
thread.LastReadAt = params.ReadAt
378+
if err := tx.
379+
Where("message_thread_id = ?", messageThreadID).
380+
Delete(&entities.MessageThreadUnreadItem{}).
381+
Error; err != nil {
382+
return stacktrace.Propagatef(
383+
err,
384+
"cannot clear unread ledger items for thread [%s] and user [%s]",
385+
messageThreadID,
386+
userID,
387+
)
388+
}
389+
return nil
390+
})
391+
if err != nil {
392+
return nil, repository.tracer.WrapErrorSpan(
393+
span,
394+
stacktrace.Propagatef(err, "cannot update status for thread [%s] and user [%s]", messageThreadID, userID),
395+
)
192396
}
193397

194398
return thread, nil

0 commit comments

Comments
 (0)