feat(notifications): persistent read cursor + delivery history

notification_cursors table stores one read-cursor per user per tenant;
GET /v1/activity returns unreadCount + readAt, POST /v1/activity/read
persists the cursor. Reminder dispatches now append reminder.sent /
reminder.failed system events so delivery history reaches the feed.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomas DvorakandDevin committed 2026-09-18 16:55:21 +02:00
1 parent 76e45d4178
commit 2464aa45fa
14 files changed
+332 -8

No files matched your search

+24 -1
View File
@@ -3,6 +3,7 @@ package activity
import (
"context"
"errors"
"time"
"bookra/apps/backend/internal/db"
"bookra/apps/backend/internal/domain"
@@ -35,6 +36,11 @@ func (s *Service) List(ctx context.Context, principal domain.Principal, limit in
if err != nil {
return domain.ActivityFeedResponse{}, err
}
readAt, err := s.repo.GetNotificationCursor(ctx, membership.Tenant.ID, principal.Subject)
if err != nil {
return domain.ActivityFeedResponse{}, err
}
unread := 0
events := make([]domain.ActivityEvent, 0, len(records))
for _, rec := range records {
event := domain.ActivityEvent{
@@ -49,7 +55,24 @@ func (s *Service) List(ctx context.Context, principal domain.Principal, limit in
if rec.EntityID != nil {
event.EntityID = *rec.EntityID
}
if readAt == nil || rec.CreatedAt.After(*readAt) {
unread++
}
events = append(events, event)
}
return domain.ActivityFeedResponse{Events: events}, nil
return domain.ActivityFeedResponse{Events: events, UnreadCount: unread, ReadAt: readAt}, nil
}
// MarkAllRead persists the user's cursor — events at or before now count as
// read for this subject on every subsequent load.
func (s *Service) MarkAllRead(ctx context.Context, principal domain.Principal) (time.Time, error) {
membership, err := s.repo.GetTenantMembershipBySubject(ctx, principal.Subject)
if err != nil {
return time.Time{}, ErrTenantMembership
}
readAt := time.Now().UTC()
if err := s.repo.SetNotificationCursor(ctx, membership.Tenant.ID, principal.Subject, readAt); err != nil {
return time.Time{}, err
}
return readAt, nil
}
@@ -0,0 +1,82 @@
package activity
import (
"context"
"testing"
"bookra/apps/backend/internal/db"
"bookra/apps/backend/internal/domain"
)
var demoPrincipal = domain.Principal{Subject: "demo-owner", Role: "owner"}
func TestListCountsUnreadAgainstCursor(t *testing.T) {
repo := db.NewMemoryRepository()
svc := NewService(repo)
ctx := context.Background()
feed, err := svc.List(ctx, demoPrincipal, 10)
if err != nil {
t.Fatalf("list: %v", err)
}
if len(feed.Events) == 0 {
t.Fatal("expected seeded demo events")
}
if feed.UnreadCount != len(feed.Events) {
t.Fatalf("expected all %d events unread without cursor, got %d", len(feed.Events), feed.UnreadCount)
}
if feed.ReadAt != nil {
t.Fatalf("expected nil readAt before first mark, got %v", feed.ReadAt)
}
}
func TestMarkAllReadPersistsCursor(t *testing.T) {
repo := db.NewMemoryRepository()
svc := NewService(repo)
ctx := context.Background()
readAt, err := svc.MarkAllRead(ctx, demoPrincipal)
if err != nil {
t.Fatalf("mark all read: %v", err)
}
if readAt.IsZero() {
t.Fatal("expected non-zero readAt")
}
feed, err := svc.List(ctx, demoPrincipal, 10)
if err != nil {
t.Fatalf("list: %v", err)
}
if feed.UnreadCount != 0 {
t.Fatalf("expected 0 unread after mark, got %d", feed.UnreadCount)
}
if feed.ReadAt == nil || !feed.ReadAt.Equal(readAt) {
t.Fatalf("expected readAt %v, got %v", readAt, feed.ReadAt)
}
// A new event after the cursor is unread again.
if err := repo.AppendActivityEvent(ctx, db.ActivityEventParams{
TenantID: "5d6b3551-0a3e-4b86-bdf0-e9df20a47148",
ActorKind: "staff",
Action: "booking.created",
EntityType: "booking",
Summary: "New — Booking",
}); err != nil {
t.Fatalf("append event: %v", err)
}
feed, err = svc.List(ctx, demoPrincipal, 10)
if err != nil {
t.Fatalf("list: %v", err)
}
if feed.UnreadCount != 1 {
t.Fatalf("expected 1 unread after new event, got %d", feed.UnreadCount)
}
}
func TestMarkAllReadRequiresMembership(t *testing.T) {
repo := db.NewMemoryRepository()
svc := NewService(repo)
if _, err := svc.MarkAllRead(context.Background(), domain.Principal{Subject: "nobody"}); err == nil {
t.Fatal("expected membership error for unknown subject")
}
}
+13
View File
@@ -1296,6 +1296,19 @@ func NewServer(cfg config.Config, pools *db.Pools) (*Server, error) {
c.JSON(http.StatusOK, response)
})
protected.POST("/activity/read", func(c *gin.Context) {
readAt, err := activityService.MarkAllRead(c.Request.Context(), auth.PrincipalFromContext(c))
if err != nil {
status := http.StatusInternalServerError
if errors.Is(err, activity.ErrTenantMembership) {
status = http.StatusNotFound
}
c.JSON(status, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"readAt": readAt})
})
protected.GET("/metrics/staff", func(c *gin.Context) {
days, _ := strconv.Atoi(c.DefaultQuery("days", "30"))
response, err := bookingService.StaffMetrics(c.Request.Context(), auth.PrincipalFromContext(c), days)
+29
View File
@@ -2,6 +2,10 @@ package db
import (
"context"
"errors"
"time"
"github.com/jackc/pgx/v5"
)
func (r *PGRepository) AppendActivityEvent(ctx context.Context, params ActivityEventParams) error {
@@ -40,3 +44,28 @@ func (r *PGRepository) ListActivityEvents(ctx context.Context, tenantID string,
}
return events, rows.Err()
}
func (r *PGRepository) GetNotificationCursor(ctx context.Context, tenantID string, userSubject string) (*time.Time, error) {
var readAt time.Time
err := r.pool.QueryRow(ctx, `
SELECT read_at FROM notification_cursors
WHERE tenant_id = $1 AND user_subject = $2
`, tenantID, userSubject).Scan(&readAt)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, err
}
return &readAt, nil
}
func (r *PGRepository) SetNotificationCursor(ctx context.Context, tenantID string, userSubject string, readAt time.Time) error {
_, err := r.pool.Exec(ctx, `
INSERT INTO notification_cursors (tenant_id, user_subject, read_at)
VALUES ($1, $2, $3)
ON CONFLICT (tenant_id, user_subject)
DO UPDATE SET read_at = EXCLUDED.read_at, updated_at = now()
`, tenantID, userSubject, readAt)
return err
}
+18
View File
@@ -146,6 +146,9 @@ type Repository interface {
// Activity feed
AppendActivityEvent(ctx context.Context, params ActivityEventParams) error
ListActivityEvents(ctx context.Context, tenantID string, limit int) ([]ActivityEventRecord, error)
// GetNotificationCursor returns nil when the user has no cursor yet.
GetNotificationCursor(ctx context.Context, tenantID string, userSubject string) (*time.Time, error)
SetNotificationCursor(ctx context.Context, tenantID string, userSubject string, readAt time.Time) error
// Staff metrics
GetStaffMetrics(ctx context.Context, tenantID string, from time.Time, to time.Time) ([]StaffMetricsRecord, error)
@@ -864,6 +867,7 @@ type MemoryRepository struct {
tenantFeedToken string
staffFeedTokens map[string]string
activityEvents []ActivityEventRecord
notificationCursors map[string]time.Time
}
func NewMemoryRepository() *MemoryRepository {
@@ -990,6 +994,7 @@ func NewMemoryRepository() *MemoryRepository {
staffFeedTokens: map[string]string{
staffID: "demo-staff-feed-token",
},
notificationCursors: map[string]time.Time{},
activityEvents: []ActivityEventRecord{
{
ID: uuid.NewString(), TenantID: tenantID, ActorKind: "staff", ActorName: "Demo Stylist",
@@ -1409,6 +1414,19 @@ func (r *MemoryRepository) ListActivityEvents(_ context.Context, tenantID string
return events, nil
}
func (r *MemoryRepository) GetNotificationCursor(_ context.Context, tenantID string, userSubject string) (*time.Time, error) {
readAt, ok := r.notificationCursors[tenantID+"/"+userSubject]
if !ok {
return nil, nil
}
return &readAt, nil
}
func (r *MemoryRepository) SetNotificationCursor(_ context.Context, tenantID string, userSubject string, readAt time.Time) error {
r.notificationCursors[tenantID+"/"+userSubject] = readAt
return nil
}
func (r *MemoryRepository) ListAvailabilityRulesByTenant(_ context.Context, tenantID string) ([]AvailabilityRuleRecord, error) {
if tenantID != r.tenant.ID {
return nil, nil
+3 -1
View File
@@ -553,7 +553,9 @@ type ActivityEvent struct {
}
type ActivityFeedResponse struct {
Events []ActivityEvent `json:"events"`
Events []ActivityEvent `json:"events"`
UnreadCount int `json:"unreadCount"`
ReadAt *time.Time `json:"readAt,omitempty"`
}
// StaffMetric aggregates appointment workload for one staff member over a
@@ -141,6 +141,19 @@ func (s *Service) DispatchDue(ctx context.Context, limit int) (domain.DispatchRe
return domain.DispatchReminderJobsResponse{}, err
}
action := "reminder.sent"
if status != "sent" {
action = "reminder.failed"
}
_ = s.repo.AppendActivityEvent(ctx, db.ActivityEventParams{
TenantID: job.TenantID,
ActorKind: "system",
Action: action,
EntityType: "booking",
EntityID: &job.BookingID,
Summary: job.CustomerName + " — " + job.Channel,
})
if status == "sent" {
response.SentCount++
} else {
@@ -61,6 +61,20 @@ func TestDispatchDueProcessesPendingEmailReminders(t *testing.T) {
if len(pending) != 0 {
t.Fatalf("expected no pending jobs after dispatch, got %d", len(pending))
}
events, err := repo.ListActivityEvents(context.Background(), "5d6b3551-0a3e-4b86-bdf0-e9df20a47148", 10)
if err != nil {
t.Fatalf("list activity: %v", err)
}
for _, event := range events {
if event.Action == "reminder.sent" && event.ActorKind == "system" {
if event.EntityID == nil || *event.EntityID != created.ID {
t.Fatalf("expected entity id %s, got %v", created.ID, event.EntityID)
}
return
}
}
t.Fatal("expected system reminder.sent activity event")
}
func TestDispatchDueFailsUnknownChannel(t *testing.T) {