feat: notifications (#109)

database/migrations/015_notifications.up.sql

@@ -0,0 +1,13 @@
+CREATE TYPE getstronger.notification_type AS ENUM ('Follow', 'WorkoutComment');
+
+CREATE TABLE getstronger.notifications
+(
+    id         UUID PRIMARY KEY              NOT NULL DEFAULT uuid_generate_v4(),
+    user_id    UUID                          NOT NULL REFERENCES getstronger.users (id),
+    "type"     getstronger.notification_type NOT NULL,
+    payload    JSONB                         NOT NULL,
+    read_at    TIMESTAMP                     NULL,
+    created_at TIMESTAMP                     NOT NULL DEFAULT (NOW() AT TIME ZONE 'UTC')
+);
+
+CREATE INDEX idx_notifications_user_id_created_at ON getstronger.notifications (user_id, read_at);

protobufs/api/v1/users.proto

@@ -4,6 +4,9 @@ package api.v1;
 
 import "api/v1/options.proto";
 import "api/v1/shared.proto";
+import "api/v1/workouts.proto";
+
+import "google/protobuf/timestamp.proto";
 
 import "buf/validate/validate.proto";
 
@@ -26,6 +29,9 @@ service UserService {
   rpc Search(SearchRequest) returns (SearchResponse) {
     option (auth) = true;
   }
+  rpc ListNotifications(ListNotificationsRequest) returns (ListNotificationsResponse) {
+    option (auth) = true;
+  }
 }
 
 message GetUserRequest {
@@ -67,3 +73,26 @@ message SearchResponse {
   repeated User users = 1;
   PaginationResponse pagination = 2;
 }
+
+message ListNotificationsRequest {
+  bool only_unread = 1;
+  PaginationRequest pagination = 2 [(buf.validate.field).required = true];
+}
+message ListNotificationsResponse {
+  repeated Notification notifications = 1;
+  PaginationResponse pagination = 2;
+}
+
+message Notification {
+  message WorkoutComment {
+    User actor = 1;
+    Workout workout = 2;
+  }
+
+  string id = 1;
+  // DEBT: This should be a timestamp but the client is not able to parse it.
+  int64 notified_at_unix = 2;
+  oneof type {
+    WorkoutComment workout_comment = 3;
+  }
+}

server/bus/events/events.go

@@ -1,3 +1,6 @@
 package events
 
-const RequestTraced = "request:traced"
+const (
+	RequestTraced        = "request:traced"
+	WorkoutCommentPosted = "workout_comment:posted"
+)

server/bus/handlers/handlers.go

@@ -7,14 +7,18 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/crlssn/getstronger/server/bus/payloads"
+	"github.com/crlssn/getstronger/server/pkg/orm"
 	"github.com/crlssn/getstronger/server/pkg/repo"
 )
 
 type Handler interface {
 	HandlePayload(payload any)
 }
 
-var _ Handler = (*RequestTraced)(nil)
+var (
+	_ Handler = (*RequestTraced)(nil)
+	_ Handler = (*WorkoutCommentPosted)(nil)
+)
 
 type RequestTraced struct {
 	log  *zap.Logger
@@ -44,3 +48,42 @@ func (h *RequestTraced) HandlePayload(payload any) {
 		h.log.Error("unexpected event type", zap.Any("event", payload))
 	}
 }
+
+type WorkoutCommentPosted struct {
+	log  *zap.Logger
+	repo *repo.Repo
+}
+
+func NewWorkoutCommentPosted(log *zap.Logger, repo *repo.Repo) *WorkoutCommentPosted {
+	return &WorkoutCommentPosted{log, repo}
+}
+
+func (w *WorkoutCommentPosted) HandlePayload(payload any) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	switch t := payload.(type) {
+	case *payloads.WorkoutCommentPosted:
+		comment, err := w.repo.GetWorkoutComment(ctx,
+			repo.GetWorkoutCommentWithID(t.CommentID),
+			repo.GetWorkoutCommentWithWorkout(),
+		)
+		if err != nil {
+			w.log.Error("get workout comment", zap.Error(err))
+			return
+		}
+
+		if err = w.repo.CreateNotification(ctx, repo.CreateNotificationParams{
+			Type:   orm.NotificationTypeWorkoutComment,
+			UserID: comment.R.Workout.UserID,
+			Payload: repo.NotificationPayload{
+				ActorID:   comment.UserID,
+				WorkoutID: comment.WorkoutID,
+			},
+		}); err != nil {
+			w.log.Error("create notification", zap.Error(err))
+		}
+	default:
+		w.log.Error("unexpected event type", zap.Any("event", payload))
+	}
+}

server/bus/handlers/registry.go

@@ -13,13 +13,15 @@ type Registry struct {
 type RegistryParams struct {
 	fx.In
 
-	RequestTraced *RequestTraced
+	RequestTraced        *RequestTraced
+	WorkoutCommentPosted *WorkoutCommentPosted
 }
 
 func NewRegistry(p RegistryParams) *Registry {
 	return &Registry{
 		handlers: map[string]Handler{
-			events.RequestTraced: p.RequestTraced,
+			events.RequestTraced:        p.RequestTraced,
+			events.WorkoutCommentPosted: p.WorkoutCommentPosted,
 		},
 	}
 }

server/bus/module.go

@@ -15,6 +15,7 @@ func Module() fx.Option {
 			New,
 			handlers.NewRegistry,
 			handlers.NewRequestTraced,
+			handlers.NewWorkoutCommentPosted,
 		),
 		fx.Invoke(
 			func(lc fx.Lifecycle, bus *Bus, registry *handlers.Registry) {

server/bus/payloads/payloads.go

@@ -5,3 +5,7 @@ type RequestTraced struct {
 	DurationMS int
 	StatusCode int
 }
+
+type WorkoutCommentPosted struct {
+	CommentID string
+}