feat(chat): add transient presence flash and 5s reconnect grace period
CI / test (push) Successful in 1m33s
CI / docker (push) Successful in 2m10s

This commit is contained in:
zwitschi committed 2026-08-13 20:45:52 +02:00
1 parent 4eb88a4594
commit cef002862e
19 files changed
+299 -37

No files matched your search

+2 -2
View File
@@ -21,8 +21,8 @@ const (
// Default HLS values used when neither env nor persisted config is set.
const (
defaultHLSSegmentDuration = 6
defaultHLSWindowSize = 24
defaultHLSSegmentDuration = 2
defaultHLSWindowSize = 12
)
var (
+7 -7
View File
@@ -51,26 +51,26 @@ func TestGettersDefaults(t *testing.T) {
if got := MatomoSiteID(); got != "" {
t.Errorf("MatomoSiteID() = %q, want empty", got)
}
if got := HLSSegmentDuration(); got != 6 {
t.Errorf("HLSSegmentDuration() = %d, want 6", got)
if got := HLSSegmentDuration(); got != 2 {
t.Errorf("HLSSegmentDuration() = %d, want 2", got)
}
if got := HLSWindowSize(); got != 24 {
t.Errorf("HLSWindowSize() = %d, want 24", got)
if got := HLSWindowSize(); got != 12 {
t.Errorf("HLSWindowSize() = %d, want 12", got)
}
}
func TestGettersEnvFallback(t *testing.T) {
resetRuntime(t)
os.Setenv("HLS_SEGMENT_DURATION", "4")
os.Setenv("HLS_WINDOW_SIZE", "12")
os.Setenv("HLS_WINDOW_SIZE", "30")
os.Setenv("MATOMO_URL", "https://analytics.example.com")
os.Setenv("MATOMO_SITE_ID", "7")
if got := HLSSegmentDuration(); got != 4 {
t.Errorf("HLSSegmentDuration() = %d, want 4", got)
}
if got := HLSWindowSize(); got != 12 {
t.Errorf("HLSWindowSize() = %d, want 12", got)
if got := HLSWindowSize(); got != 30 {
t.Errorf("HLSWindowSize() = %d, want 30", got)
}
if got := MatomoURL(); got != "https://analytics.example.com" {
t.Errorf("MatomoURL() = %q", got)
+1 -1
View File
@@ -1829,7 +1829,7 @@ func HandleAdminChatLog(deps *Dependencies) http.HandlerFunc {
limit = parsedLimit
}
dbQuery := deps.DB.Model(&db.Message{})
dbQuery := deps.DB.Model(&db.Message{}).Where("type <> ?", "system")
if room != "" {
dbQuery = dbQuery.Where("room = ?", room)
}
@@ -951,6 +951,32 @@ func TestHandleAdminChatLog_FilterByRoom(t *testing.T) {
}
}
func TestHandleAdminChatLog_ExcludesSystem(t *testing.T) {
deps, gormDB, _ := setupTestDeps(t)
gormDB.Create(&db.Message{Room: "main", Username: "system", Text: "bob joined main", Timestamp: time.Now(), Type: "system"})
gormDB.Create(&db.Message{Room: "main", Username: "u1", Text: "hello", Timestamp: time.Now(), Type: "public"})
handler := HandleAdminChatLog(deps)
req := httptest.NewRequest("GET", "/api/admin/chat-log", nil)
w := httptest.NewRecorder()
handler(w, req)
if w.Code != http.StatusOK {
t.Errorf("status = %d, want 200", w.Code)
}
var resp ChatLogListResponse
json.NewDecoder(w.Body).Decode(&resp)
if resp.Total != 1 {
t.Errorf("Total = %d, want 1 (system messages must be excluded)", resp.Total)
}
if len(resp.Entries) != 1 || resp.Entries[0].Text != "hello" {
t.Errorf("entries = %v, want only the public chat message", resp.Entries)
}
}
func TestHandleAdminChatLogByID_Get(t *testing.T) {
deps, gormDB, _ := setupTestDeps(t)
+61 -14
View File
@@ -41,19 +41,25 @@ type Hub struct {
RateLimitWindow time.Duration
AdminUsername string
DMsSent int64
presenceMu sync.Mutex
presenceTimers map[string]*time.Timer
PresenceGracePeriod time.Duration
}
// NewHub creates a new WebSocket hub.
func NewHub(gormDB *gorm.DB, redis db.RedisClient, instanceID string, rateMax int, rateWindow time.Duration, adminUser string) *Hub {
return &Hub{
clients: make(map[*websocket.Conn]bool),
clientMeta: make(map[*websocket.Conn]ClientInfo),
DB: gormDB,
Redis: redis,
InstanceID: instanceID,
RateLimitMax: rateMax,
RateLimitWindow: rateWindow,
AdminUsername: adminUser,
clients: make(map[*websocket.Conn]bool),
clientMeta: make(map[*websocket.Conn]ClientInfo),
DB: gormDB,
Redis: redis,
InstanceID: instanceID,
RateLimitMax: rateMax,
RateLimitWindow: rateWindow,
AdminUsername: adminUser,
presenceTimers: make(map[string]*time.Timer),
PresenceGracePeriod: 5 * time.Second,
}
}
@@ -65,12 +71,14 @@ func (h *Hub) ConnectedClients() int {
}
// Broadcast sends a message to all connected clients and persists to DB.
// System messages (join/leave announcements, moderation notices) fan out but
// are not persisted, so the chat-log table only holds real chat messages.
func (h *Hub) Broadcast(msg db.Message) {
h.mu.Lock()
defer h.mu.Unlock()
// Persist to configured SQL backend.
if h.DB != nil {
// Persist to configured SQL backend (chat messages only).
if h.DB != nil && msg.Type != "system" {
h.DB.Create(&msg)
}
@@ -93,6 +101,41 @@ func (h *Hub) Broadcast(msg db.Message) {
}
}
// scheduleLeaveAnnouncement broadcasts a "left" message after a short grace
// period so quick reconnects don't produce join/leave spam.
func (h *Hub) scheduleLeaveAnnouncement(room, username string) {
key := room + "\x00" + username
h.presenceMu.Lock()
if h.presenceTimers == nil {
h.presenceTimers = make(map[string]*time.Timer)
}
grace := h.PresenceGracePeriod
if grace <= 0 {
grace = 5 * time.Second
}
h.presenceTimers[key] = time.AfterFunc(grace, func() {
h.presenceMu.Lock()
delete(h.presenceTimers, key)
h.presenceMu.Unlock()
h.Broadcast(db.Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system", Timestamp: time.Now()})
})
h.presenceMu.Unlock()
}
// cancelLeaveAnnouncement cancels a pending "left" announcement for a
// room+user and reports whether one existed (i.e. a reconnect within grace).
func (h *Hub) cancelLeaveAnnouncement(room, username string) bool {
key := room + "\x00" + username
h.presenceMu.Lock()
defer h.presenceMu.Unlock()
if timer, ok := h.presenceTimers[key]; ok {
timer.Stop()
delete(h.presenceTimers, key)
return true
}
return false
}
// BroadcastTyping notifies other users in the room that a user is typing.
func (h *Hub) BroadcastTyping(username, room string) {
h.mu.Lock()
@@ -110,7 +153,7 @@ func (h *Hub) PushChatHistory(conn *websocket.Conn, room string, limit int) {
return
}
var messages []db.Message
h.DB.Where("room = ?", room).Order("timestamp desc").Limit(limit).Find(&messages)
h.DB.Where("room = ? AND type <> ?", room, "system").Order("timestamp desc").Limit(limit).Find(&messages)
// Reverse to chronological order
for i := len(messages)/2 - 1; i >= 0; i-- {
opp := len(messages) - 1 - i
@@ -158,10 +201,14 @@ func (h *Hub) HandleConnection(w http.ResponseWriter, r *http.Request) {
h.clientMeta[conn] = ClientInfo{Username: username, Room: room}
h.mu.Unlock()
// Announce presence (ADR-007)
// Announce presence (ADR-007). A reconnect within the grace window cancels
// the pending "left" announcement and skips a redundant "joined".
rejoined := h.cancelLeaveAnnouncement(room, username)
if h.Redis != nil && room != "" {
_ = h.Redis.Set(context.Background(), fmt.Sprintf("presence:%s:%s", room, username), "online", 60*time.Second)
h.Broadcast(db.Message{Room: room, Username: "system", Text: username + " joined " + room, Type: "system", Timestamp: time.Now()})
if !rejoined {
h.Broadcast(db.Message{Room: room, Username: "system", Text: username + " joined " + room, Type: "system", Timestamp: time.Now()})
}
}
// Push chat history on join (last 50 messages)
@@ -196,7 +243,7 @@ func (h *Hub) HandleConnection(w http.ResponseWriter, r *http.Request) {
h.mu.Unlock()
if h.Redis != nil && room != "" {
_ = h.Redis.Del(context.Background(), fmt.Sprintf("presence:%s:%s", room, username))
h.Broadcast(db.Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system", Timestamp: time.Now()})
h.scheduleLeaveAnnouncement(room, username)
}
conn.Close()
}()
+48
View File
@@ -113,6 +113,11 @@ func setupTestHub(t *testing.T) (*Hub, *gorm.DB, *mockRedis) {
if err != nil {
t.Fatalf("failed to open test DB: %v", err)
}
sqlDB, err := gormDB.DB()
if err != nil {
t.Fatalf("failed to get sql.DB: %v", err)
}
sqlDB.SetMaxOpenConns(1)
gormDB.AutoMigrate(&db.Message{}, &db.ModerationLog{})
mr := newMockRedis()
hub := NewHub(gormDB, mr, "test-instance", 10, 10*time.Second, "admin")
@@ -560,4 +565,47 @@ func TestHub_SubscribeRoomMessages(t *testing.T) {
})
mr.Publish(ctx, "room:main:messages", string(ownPayload))
time.Sleep(50 * time.Millisecond)
}
func TestHub_Broadcast_SystemNotPersisted(t *testing.T) {
hub, _, _ := setupTestHub(t)
hub.Broadcast(db.Message{Room: "main", Username: "system", Text: "bob joined main", Type: "system", Timestamp: time.Now()})
var count int64
hub.DB.Model(&db.Message{}).Where("type = ?", "system").Count(&count)
if count != 0 {
t.Errorf("system message count = %d, want 0 (system messages must not persist)", count)
}
hub.Broadcast(db.Message{Room: "main", Username: "bob", Text: "hi", Type: "public", Timestamp: time.Now()})
hub.DB.Model(&db.Message{}).Where("type = ?", "public").Count(&count)
if count != 1 {
t.Errorf("public message count = %d, want 1", count)
}
}
func TestHub_ReconnectGracePeriod(t *testing.T) {
hub, _, _ := setupTestHub(t)
hub.PresenceGracePeriod = 50 * time.Millisecond
// Disconnect schedules a delayed "left"; reconnect cancels it.
hub.scheduleLeaveAnnouncement("main", "bob")
if !hub.cancelLeaveAnnouncement("main", "bob") {
t.Error("expected a pending leave announcement to be cancelled on reconnect")
}
if hub.cancelLeaveAnnouncement("main", "bob") {
t.Error("expected no pending leave announcement after cancellation")
}
// A fresh disconnect that is never reconnected fires after the grace period
// and cleans up its timer.
hub.scheduleLeaveAnnouncement("main", "alice")
time.Sleep(120 * time.Millisecond)
hub.presenceMu.Lock()
_, exists := hub.presenceTimers["main\x00alice"]
hub.presenceMu.Unlock()
if exists {
t.Error("expected the leave announcement timer to clean itself up after firing")
}
}
+2 -2
View File
@@ -47,8 +47,8 @@ var (
restartWindow = time.Now()
slateDir = "/app/slate"
hlsDir = "/app/hls"
hlsSegmentDuration = envWithDefault("HLS_SEGMENT_DURATION", "6")
hlsWindowSize = envWithDefault("HLS_WINDOW_SIZE", "24")
hlsSegmentDuration = envWithDefault("HLS_SEGMENT_DURATION", "2")
hlsWindowSize = envWithDefault("HLS_WINDOW_SIZE", "12")
// Unique segment naming (STREAMING.md contract: r-{runID}/seg-{seq}.m4s)
runID string