LobotomyLabs/dumpsterChat
2
0
Fork 0
Code Issues Pull Requests Actions Packages Releases 1 Wiki Activity

feat: realtime status, mentions, push notifications

Browse Source
This commit is contained in:
Dustin Newkirk
2026-06-29 15:52:34 -04:00
parent 282a436995
commit 6c8defb6fd
10 changed files with 387 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/auth/handlers.go
+27 -3
View File
@@ -11,6 +11,7 @@ import (
"strings"
"git.dustin.coffee/hobokenchicken/dumpsterChat/internal/config"
"git.dustin.coffee/hobokenchicken/dumpsterChat/internal/gateway"
"git.dustin.coffee/hobokenchicken/dumpsterChat/internal/middleware"
"github.com/go-chi/chi/v5"
)
@@ -19,13 +20,15 @@ type Handler struct {
db *sql.DB
cfg *config.Config
sessions *SessionStore
hub *gateway.Hub
}
func NewHandler(db *sql.DB, cfg *config.Config) *Handler {
func NewHandler(db *sql.DB, cfg *config.Config, hub *gateway.Hub) *Handler {
return &Handler{
db: db,
cfg: cfg,
sessions: NewSessionStore(db, cfg),
hub: hub,
}
}
@@ -59,6 +62,7 @@ type updateProfileRequest struct {
Bio string `json:"bio"`
AccentColor string `json:"accent_color"`
StatusText string `json:"status_text"`
Status string `json:"status"`
}
type changePasswordRequest struct {
@@ -279,17 +283,29 @@ func (h *Handler) UpdateProfile(w http.ResponseWriter, r *http.Request) {
http.Error(w, `{"error":"accent color must be 7 char hex"}`, http.StatusBadRequest)
return
}
if req.Status != "" && !isValidStatus(req.Status) {
http.Error(w, `{"error":"invalid status"}`, http.StatusBadRequest)
return
}
_, err := h.db.ExecContext(r.Context(), `
UPDATE users
SET display_name = $2, bio = $3, accent_color = $4, status_text = $5
SET display_name = $2, bio = $3, accent_color = $4, status_text = $5, status = COALESCE(NULLIF($6, ''), status)
WHERE id = $1
`, userID, req.DisplayName, req.Bio, req.AccentColor, req.StatusText)
`, userID, req.DisplayName, req.Bio, req.AccentColor, req.StatusText, req.Status)
if err != nil {
http.Error(w, `{"error":"update failed"}`, http.StatusInternalServerError)
return
}
if req.Status != "" && h.hub != nil {
user, err := h.getProfile(r.Context(), userID)
if err == nil {
h.hub.SetUserStatus(userID, req.Status)
h.hub.BroadcastPresence(userID, user.Username, req.Status)
}
}
user, err := h.getProfile(r.Context(), userID)
if err != nil {
http.Error(w, `{"error":"not found"}`, http.StatusNotFound)
@@ -300,6 +316,14 @@ func (h *Handler) UpdateProfile(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(user)
}
func isValidStatus(status string) bool {
switch status {
case "online", "idle", "dnd", "offline":
return true
}
return false
}
func (h *Handler) getProfile(ctx context.Context, userID string) (UserProfile, error) {
var user UserProfile
return user, h.db.QueryRowContext(ctx, `
internal/gateway/hub.go
+42 -14
View File
@@ -18,16 +18,17 @@ const (
// PresenceData is the data payload for PRESENCE_UPDATE events.
type PresenceData struct {
UserID string `json:"user_id"`
Status string `json:"status"`
UserID string `json:"user_id"`
Username string `json:"username"`
Status string `json:"status"`
}
// Hub maintains the set of active clients and broadcasts messages to them.
type Hub struct {
mu sync.RWMutex
clients map[*Client]struct{}
presence map[string]string // userID -> status
lastActivity map[string]time.Time // userID -> last activity time
presence map[string]string // userID -> status
lastActivity map[string]time.Time // userID -> last activity time
userServers map[string]map[string]struct{} // userID -> set of serverIDs
register chan *Client
unregister chan *Client
@@ -162,16 +163,7 @@ func (h *Hub) checkIdleUsers() {
// setUserStatus persists the user's status to the database.
func (h *Hub) setUserStatus(userID, status string) {
if h.db == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := h.db.ExecContext(ctx, `UPDATE users SET status = $1 WHERE id = $2`, status, userID)
if err != nil {
h.logger.Error("failed to update user status", "user_id", userID, "status", status, "error", err)
}
setUserStatus(h.db, h.logger, userID, status)
}
// broadcastPresence sends a PRESENCE_UPDATE event to all connected clients.
@@ -185,6 +177,29 @@ func (h *Hub) broadcastPresence(userID, status string) {
})
}
// SetUserStatus is the exported form of setUserStatus, used by non-gateway
// handlers that need to persist a status change.
func (h *Hub) SetUserStatus(userID, status string) {
h.setUserStatus(userID, status)
}
// BroadcastPresence sends a PRESENCE_UPDATE event including the username.
func (h *Hub) BroadcastPresence(userID, username, status string) {
h.mu.Lock()
h.presence[userID] = status
h.lastActivity[userID] = time.Now()
h.mu.Unlock()
h.BroadcastEvent(Event{
Type: EventPresenceUpdate,
Data: PresenceData{
UserID: userID,
Username: username,
Status: status,
},
})
}
// Register queues a client for registration with the hub.
func (h *Hub) Register(client *Client) {
h.register <- client
@@ -273,3 +288,16 @@ func (h *Hub) ServerIDForChannel(ctx context.Context, channelID string) (string,
err := h.db.QueryRowContext(ctx, `SELECT server_id FROM channels WHERE id = $1`, channelID).Scan(&serverID)
return serverID, err
}
func setUserStatus(db *sql.DB, logger *slog.Logger, userID, status string) {
if db == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := db.ExecContext(ctx, `UPDATE users SET status = $1 WHERE id = $2`, status, userID)
if err != nil {
logger.Error("failed to update user status", "user_id", userID, "status", status, "error", err)
}
}
internal/message/handlers.go
+18
View File
@@ -16,6 +16,14 @@ import (
)
type Handler struct {
db *sql.DB
hub *gateway.Hub
mentions *MentionHandler
pushHandler *push.Handler
logger *slog.Logger
}
type Handler_old struct {
db *sql.DB
hub *gateway.Hub
mentions *MentionHandler
@@ -137,6 +145,16 @@ func (h *Handler) Create(w http.ResponseWriter, r *http.Request) {
// Dispatch push notifications for @mentions
go h.mentions.ParseAndNotify(r.Context(), channelID, userID, req.Content)
// Dispatch push notifications to channel members
go func() {
var channelName string
_ = h.db.QueryRowContext(context.Background(), `SELECT name FROM channels WHERE id = $1`, channelID).Scan(&channelName)
if channelName == "" {
channelName = channelID
}
h.pushHandler.SendChannelNotification(context.Background(), channelID, userID, channelName, req.Content)
}()
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(msg)
internal/push/handlers.go
+71
View File
@@ -183,6 +183,77 @@ func (h *Handler) SendPush(ctx context.Context, userID string, payload map[strin
}
}
// SendChannelNotification sends a push notification to all server members (except the author)
// when a new message is posted in a channel.
func (h *Handler) SendChannelNotification(ctx context.Context, channelID, authorID, channelName, content string) {
if h.vapidPub == "" || h.vapidPriv == "" {
return
}
var serverID string
err := h.db.QueryRowContext(ctx, `SELECT server_id FROM channels WHERE id = $1`, channelID).Scan(&serverID)
if err != nil {
h.logger.Error("failed to find channel server", "error", err)
return
}
body := content
if len(body) > 200 {
body = body[:200] + "..."
}
payload, _ := json.Marshal(map[string]interface{}{
"title": "New message in #" + channelName,
"body": body,
"url": "/channels/" + channelID,
})
rows, err := h.db.QueryContext(ctx, `
SELECT DISTINCT ps.user_id, ps.endpoint, ps.p256dh, ps.auth
FROM push_subscriptions ps
JOIN members m ON m.user_id = ps.user_id
WHERE m.server_id = $1 AND ps.user_id != $2
`, serverID, authorID)
if err != nil {
h.logger.Error("failed to query push subscriptions", "error", err)
return
}
defer rows.Close()
for rows.Next() {
var userID, endpoint, p256dh, auth string
if err := rows.Scan(&userID, &endpoint, &p256dh, &auth); err != nil {
continue
}
sub := &webpush.Subscription{
Endpoint: endpoint,
Keys: webpush.Keys{
P256dh: p256dh,
Auth: auth,
},
}
resp, err := webpush.SendNotification(payload, sub, &webpush.Options{
Subscriber: h.vapidSubj,
VAPIDPublicKey: h.vapidPub,
VAPIDPrivateKey: h.vapidPriv,
TTL: 30,
})
if err != nil {
h.logger.Error("failed to send push", "error", err, "user_id", userID)
if resp != nil && resp.StatusCode == 410 {
h.db.ExecContext(ctx, `DELETE FROM push_subscriptions WHERE endpoint = $1`, endpoint)
}
continue
}
if resp != nil {
resp.Body.Close()
}
}
}
// GenerateVAPIDKeys generates a new VAPID key pair.
func GenerateVAPIDKeys() (publicKey, privateKey string, err error) {
privateKeyBytes := make([]byte, 32)