From 6b6a6fbea8682f89d73ae192b73edd56b170aaf0 Mon Sep 17 00:00:00 2001 From: jevb Date: Sun, 29 Mar 2026 19:39:46 +0200 Subject: [PATCH] refactor: context propagation, LogAudit deadlock fix, ESLint v9, code quality - Propagate context.Context from WS upgrade through all 17 handlers - Add ExecContext/QueryRowContext/QueryContext/BeginTx to DB wrapper - Fix LogAudit deadlock: move audit writes after tx.Commit to avoid SQLite write-lock contention (TestAdminAPI_PatchUser_UnbanUser) - Add ESLint v9 with no-floating-promises, no-unused-vars - Refactor livekitSession.ts: remove duplicate audio pipeline (267 lines) - Add delete account UI tests (7 tests) - Expand WS integration tests --- Server/admin/admin_handler_test.go | 12 ++ Server/admin/api.go | 11 +- Server/admin/api_edge_cases_test.go | 99 +++++++++++ Server/admin/handlers_backup.go | 18 ++ Server/admin/handlers_users.go | 72 ++++++-- Server/admin/logstream.go | 134 ++++++++++++++- Server/admin/setup_handler.go | 18 +- Server/admin/static/index.html | 26 ++- Server/db/db.go | 21 +++ Server/go.mod | 13 +- Server/go.sum | 34 ++-- Server/main.go | 3 +- Server/ws/client.go | 9 +- Server/ws/export_test.go | 5 + Server/ws/handlers.go | 5 +- Server/ws/handlers_chat.go | 19 ++- Server/ws/handlers_ping.go | 7 +- Server/ws/handlers_presence.go | 19 ++- Server/ws/handlers_reaction.go | 11 +- Server/ws/handlers_voice.go | 33 ++-- Server/ws/hub.go | 76 +++++---- Server/ws/registry.go | 15 +- Server/ws/registry_test.go | 7 +- Server/ws/serve.go | 86 ++++++---- Server/ws/voice_controls.go | 9 +- Server/ws/voice_join.go | 7 +- Server/ws/voice_leave.go | 59 ++++--- Server/ws/ws_integration_test.go | 251 +++++++++++++++++++++++++++- 28 files changed, 865 insertions(+), 214 deletions(-) diff --git a/Server/admin/admin_handler_test.go b/Server/admin/admin_handler_test.go index 6e0edaeb..e7d7bce0 100644 --- a/Server/admin/admin_handler_test.go +++ b/Server/admin/admin_handler_test.go @@ -4,6 +4,7 @@ import ( "net/http" "net/http/httptest" "os" + "strings" "testing" "github.com/owncord/server/admin" @@ -41,6 +42,17 @@ func TestNewHandler_ServesStaticRoot(t *testing.T) { if ct == "" { t.Error("Content-Type header missing on / response") } + + body := w.Body.String() + if !strings.Contains(body, "api('POST','/logs/ticket')") { + t.Error("admin root should request log stream tickets before opening EventSource") + } + if !strings.Contains(body, "/admin/api/logs/stream?ticket=") { + t.Error("admin root should connect to log stream with a ticket query parameter") + } + if strings.Contains(body, "/admin/api/logs/stream?token=") { + t.Error("admin root should not use the deprecated token-based log stream URL") + } } // TestNewHandler_SetsCSPOnRoot verifies that the root path response includes a diff --git a/Server/admin/api.go b/Server/admin/api.go index d29fe724..45be6303 100644 --- a/Server/admin/api.go +++ b/Server/admin/api.go @@ -20,16 +20,21 @@ func NewAdminAPI(database *db.DB, version string, hub HubBroadcaster, u *updater r.Get("/setup/status", handleSetupStatus(database)) r.Post("/setup", handleSetup(database)) - // SSE log stream — does its own auth via query param token because - // EventSource cannot send Authorization headers. + // SSE log stream — auth is via a single-use ticket from POST /logs/ticket. + // EventSource cannot send Authorization headers, so the client first + // obtains a short-lived ticket via the authenticated ticket endpoint, + // then passes it as ?ticket= to the SSE stream. if logBuf != nil { - r.Get("/logs/stream", handleLogStream(logBuf, database)) + r.Get("/logs/stream", handleLogStream(database, logBuf)) } // All remaining routes require authentication and ADMINISTRATOR permission. r.Group(func(r chi.Router) { r.Use(adminAuthMiddleware(database)) + // Log stream ticket — issues a single-use, 30s TTL ticket for SSE auth. + r.Post("/logs/ticket", handleLogTicket(database)) + r.Get("/stats", handleGetStats(database, hub)) r.Get("/users", handleListUsers(database)) r.Patch("/users/{id}", handlePatchUser(database, hub)) diff --git a/Server/admin/api_edge_cases_test.go b/Server/admin/api_edge_cases_test.go index 5879eec3..9828eff6 100644 --- a/Server/admin/api_edge_cases_test.go +++ b/Server/admin/api_edge_cases_test.go @@ -4,12 +4,15 @@ package admin_test import ( "bytes" + "context" "encoding/json" + "io" "net/http" "net/http/httptest" "testing" "github.com/owncord/server/admin" + "github.com/owncord/server/auth" ) // ─── handlePatchUser — self-modification guard ───────────────────────────── @@ -337,6 +340,102 @@ func TestAdminAPI_PatchUser_BanNilHubDoesNotPanic(t *testing.T) { } } +// TestAdminAPI_LogStreamTicketFlow verifies that the admin API issues +// single-use log stream tickets and rejects both ticket reuse and the old +// token-in-query flow. +func TestAdminAPI_LogStreamTicketFlow(t *testing.T) { + database := openAdminTestDB(t) + logBuf := admin.NewRingBuffer(8) + handler := admin.NewAdminAPI(database, "1.0.0", &mockHub{}, nil, logBuf) + token := createAdminUser(t, database) + + ticketResp := doRequest(t, handler, http.MethodPost, "/logs/ticket", token, nil) + if ticketResp.Code != http.StatusOK { + t.Fatalf("POST /logs/ticket status = %d, want 200; body: %s", ticketResp.Code, ticketResp.Body.String()) + } + + var payload struct { + Ticket string `json:"ticket"` + } + if err := json.Unmarshal(ticketResp.Body.Bytes(), &payload); err != nil { + t.Fatalf("unmarshal ticket response: %v", err) + } + if payload.Ticket == "" { + t.Fatal("expected non-empty log stream ticket") + } + if err := database.DeleteSession(auth.HashToken(token)); err != nil { + t.Fatalf("DeleteSession: %v", err) + } + + srv := httptest.NewServer(handler) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/logs/stream?ticket="+payload.Ticket, nil) + if err != nil { + t.Fatalf("NewRequestWithContext: %v", err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("stream request failed: %v", err) + } + if resp.StatusCode != http.StatusUnauthorized { + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + t.Fatalf("GET /logs/stream?ticket=... after session revocation status = %d, want 401; body: %s", resp.StatusCode, string(body)) + } + body, _ := io.ReadAll(resp.Body) + if !bytes.Contains(body, []byte("invalid or expired session")) { + _ = resp.Body.Close() + t.Fatalf("expected revoked-session error body, got: %s", string(body)) + } + cancel() + _ = resp.Body.Close() + + reuseResp, err := http.Get(srv.URL + "/logs/stream?ticket=" + payload.Ticket) + if err != nil { + t.Fatalf("reuse request failed: %v", err) + } + defer reuseResp.Body.Close() + if reuseResp.StatusCode != http.StatusUnauthorized { + body, _ := io.ReadAll(reuseResp.Body) + t.Fatalf("reused ticket status = %d, want 401; body: %s", reuseResp.StatusCode, string(body)) + } + + legacyResp, err := http.Get(srv.URL + "/logs/stream?token=" + token) + if err != nil { + t.Fatalf("legacy request failed: %v", err) + } + defer legacyResp.Body.Close() + if legacyResp.StatusCode != http.StatusUnauthorized { + body, _ := io.ReadAll(legacyResp.Body) + t.Fatalf("legacy token stream status = %d, want 401; body: %s", legacyResp.StatusCode, string(body)) + } + + if _, err := database.CreateSession(1, auth.HashToken(token), "test", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + ticketResp = doRequest(t, handler, http.MethodPost, "/logs/ticket", token, nil) + if ticketResp.Code != http.StatusOK { + t.Fatalf("POST /logs/ticket after restoring session status = %d, want 200; body: %s", ticketResp.Code, ticketResp.Body.String()) + } + if err := json.Unmarshal(ticketResp.Body.Bytes(), &payload); err != nil { + t.Fatalf("unmarshal restored ticket response: %v", err) + } + if err := database.UpdateUserRole(1, 3); err != nil { + t.Fatalf("UpdateUserRole: %v", err) + } + demotedResp, err := http.Get(srv.URL + "/logs/stream?ticket=" + payload.Ticket) + if err != nil { + t.Fatalf("demoted-role request failed: %v", err) + } + defer demotedResp.Body.Close() + if demotedResp.StatusCode != http.StatusForbidden { + body, _ := io.ReadAll(demotedResp.Body) + t.Fatalf("demoted-role ticket status = %d, want 403; body: %s", demotedResp.StatusCode, string(body)) + } +} + // TestAdminAPI_PatchUser_RoleChangeNilHubDoesNotPanic verifies that changing a // user's role when hub is nil does not panic (exercises the hub != nil guard // around BroadcastMemberUpdate). diff --git a/Server/admin/handlers_backup.go b/Server/admin/handlers_backup.go index a9807204..40303755 100644 --- a/Server/admin/handlers_backup.go +++ b/Server/admin/handlers_backup.go @@ -102,6 +102,15 @@ func handleDeleteBackup(database *db.DB) http.Handler { return } + // Resolve to absolute path and verify it stays within the backups directory + // to prevent path traversal via Windows drive-letter prefixes (e.g. "C:evil.db"). + absDir, _ := filepath.Abs(filepath.Join("data", "backups")) + target := filepath.Join(absDir, name) + if !strings.HasPrefix(target, absDir+string(filepath.Separator)) { + writeErr(w, http.StatusBadRequest, "BAD_REQUEST", "invalid backup name") + return + } + backupPath := filepath.Join("data", "backups", name) if _, err := os.Stat(backupPath); os.IsNotExist(err) { writeErr(w, http.StatusNotFound, "NOT_FOUND", "backup not found") @@ -129,6 +138,15 @@ func handleRestoreBackup(database *db.DB) http.Handler { return } + // Resolve to absolute path and verify it stays within the backups directory + // to prevent path traversal via Windows drive-letter prefixes (e.g. "C:evil.db"). + absDir, _ := filepath.Abs(filepath.Join("data", "backups")) + target := filepath.Join(absDir, name) + if !strings.HasPrefix(target, absDir+string(filepath.Separator)) { + writeErr(w, http.StatusBadRequest, "BAD_REQUEST", "invalid backup name") + return + } + backupPath := filepath.Join("data", "backups", name) if _, err := os.Stat(backupPath); os.IsNotExist(err) { writeErr(w, http.StatusNotFound, "NOT_FOUND", "backup not found") diff --git a/Server/admin/handlers_users.go b/Server/admin/handlers_users.go index 299f6afe..a86ddc16 100644 --- a/Server/admin/handlers_users.go +++ b/Server/admin/handlers_users.go @@ -84,44 +84,80 @@ func handlePatchUser(database *db.DB, hub HubBroadcaster) http.HandlerFunc { return } + // Wrap role + ban updates in a transaction so both succeed or fail atomically. + tx, txErr := database.Begin() + if txErr != nil { + writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to begin transaction") + return + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + if req.RoleID != nil { - if err := database.UpdateUserRole(id, *req.RoleID); err != nil { + if _, err := tx.Exec(`UPDATE users SET role_id = ? WHERE id = ?`, *req.RoleID, id); err != nil { writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to update role") return } slog.Info("role changed", "actor_id", actor, "target_user", user.Username, "new_role_id", *req.RoleID) + } + + banReason := "" + if req.Banned != nil { + if req.BanReason != nil { + banReason = *req.BanReason + } + if *req.Banned { + var expiresStr *string + if _, err := tx.Exec( + `UPDATE users SET banned = 1, ban_reason = ?, ban_expires = ? WHERE id = ?`, + banReason, expiresStr, id, + ); err != nil { + writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to ban user") + return + } + slog.Warn("user banned", "actor_id", actor, "target_user", user.Username, "reason", banReason) + } else { + if _, err := tx.Exec( + `UPDATE users SET banned = 0, ban_reason = NULL, ban_expires = NULL WHERE id = ?`, + id, + ); err != nil { + writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to unban user") + return + } + slog.Info("user unbanned", "actor_id", actor, "target_user", user.Username) + } + } + + if err := tx.Commit(); err != nil { + writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to commit user update") + return + } + committed = true + + // Post-commit side effects: audit logging and broadcasts. + // These run outside the transaction to avoid SQLite write-lock + // contention (LogAudit uses the main *sql.DB, not the tx). + if req.RoleID != nil { _ = database.LogAudit(actor, "role_change", "user", id, fmt.Sprintf("changed %s role to %d", user.Username, *req.RoleID)) - // Broadcast member_update with the new role name. if role, err := database.GetRoleByID(*req.RoleID); err == nil && role != nil { if hub != nil { hub.BroadcastMemberUpdate(id, role.Name) } } } - if req.Banned != nil { - reason := "" - if req.BanReason != nil { - reason = *req.BanReason - } if *req.Banned { - if err := database.BanUser(id, reason, nil); err != nil { - writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to ban user") - return - } - slog.Warn("user banned", "actor_id", actor, "target_user", user.Username, "reason", reason) _ = database.LogAudit(actor, "user_ban", "user", id, - fmt.Sprintf("banned %s: %s", user.Username, reason)) + fmt.Sprintf("banned %s: %s", user.Username, banReason)) if hub != nil { hub.BroadcastMemberBan(id) } } else { - if err := database.UnbanUser(id); err != nil { - writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to unban user") - return - } - slog.Info("user unbanned", "actor_id", actor, "target_user", user.Username) _ = database.LogAudit(actor, "user_unban", "user", id, fmt.Sprintf("unbanned %s", user.Username)) } diff --git a/Server/admin/logstream.go b/Server/admin/logstream.go index 2aacede4..82808049 100644 --- a/Server/admin/logstream.go +++ b/Server/admin/logstream.go @@ -2,6 +2,8 @@ package admin import ( "context" + "crypto/rand" + "encoding/hex" "encoding/json" "fmt" "log/slog" @@ -16,6 +18,86 @@ import ( "github.com/owncord/server/permissions" ) +// ─── Ticket Store for SSE Log Stream ──────────────────────────────────────── + +// ticketEntry holds a single-use ticket with a creation timestamp for TTL. +type ticketEntry struct { + createdAt time.Time + tokenHash string +} + +// ticketStore manages short-lived, single-use tickets for SSE authentication. +type ticketStore struct { + mu sync.Mutex + tickets map[string]ticketEntry +} + +var logTickets = &ticketStore{ + tickets: make(map[string]ticketEntry), +} + +const ticketTTL = 30 * time.Second + +// issue creates a new single-use ticket and returns its hex string. +func (ts *ticketStore) issue(tokenHash string) (string, error) { + b := make([]byte, 32) + if _, err := rand.Read(b); err != nil { + return "", fmt.Errorf("generating ticket: %w", err) + } + ticket := hex.EncodeToString(b) + + ts.mu.Lock() + defer ts.mu.Unlock() + + // Opportunistic cleanup of expired tickets. + now := time.Now() + for k, v := range ts.tickets { + if now.Sub(v.createdAt) > ticketTTL { + delete(ts.tickets, k) + } + } + + ts.tickets[ticket] = ticketEntry{createdAt: now, tokenHash: tokenHash} + return ticket, nil +} + +// redeem validates and consumes a ticket. +func (ts *ticketStore) redeem(ticket string) (ticketEntry, bool) { + ts.mu.Lock() + defer ts.mu.Unlock() + + entry, ok := ts.tickets[ticket] + if !ok { + return ticketEntry{}, false + } + delete(ts.tickets, ticket) // single-use: delete immediately + + if time.Since(entry.createdAt) > ticketTTL { + return ticketEntry{}, false + } + return entry, true +} + +// handleLogTicket issues a short-lived, single-use ticket for the SSE log stream. +// POST /admin/api/logs/ticket — requires normal admin auth (cookie/header). +func handleLogTicket(database *db.DB) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + sess, ok := r.Context().Value(adminSessionKey).(*db.Session) + if !ok || sess == nil || sess.TokenHash == "" { + writeErr(w, http.StatusUnauthorized, "UNAUTHORIZED", "invalid or expired session") + return + } + + ticket, err := logTickets.issue(sess.TokenHash) + if err != nil { + slog.Error("failed to issue log stream ticket", "err", err) + writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to generate ticket") + return + } + writeJSON(w, http.StatusOK, map[string]string{"ticket": ticket}) + } +} + // LogEntry holds a single structured log record for the ring buffer. type LogEntry struct { Timestamp string `json:"ts"` @@ -269,19 +351,53 @@ func authenticateAdmin(database *db.DB, rawToken string) (*db.User, error) { } // handleLogStream serves an SSE endpoint that streams log entries in real-time. -// Auth is via query param ?token= since EventSource cannot send headers. -func handleLogStream(ringBuf *RingBuffer, database *db.DB) http.HandlerFunc { +// Auth is via query param ?ticket= — a short-lived single-use ticket obtained +// from POST /admin/api/logs/ticket (which requires normal admin auth). +func handleLogStream(database *db.DB, ringBuf *RingBuffer) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { - // Authenticate via query param. - rawToken := r.URL.Query().Get("token") - if _, err := authenticateAdmin(database, rawToken); err != nil { + // Authenticate via single-use ticket. + ticket := r.URL.Query().Get("ticket") + entry, ok := logTickets.redeem(ticket) + if ticket == "" || !ok { errResp, _ := json.Marshal(map[string]string{ "error": "UNAUTHORIZED", - "message": err.Error(), + "message": "invalid or expired ticket", }) http.Error(w, string(errResp), http.StatusUnauthorized) return } + sess, err := database.GetSessionByTokenHash(entry.tokenHash) + if err != nil || sess == nil || auth.IsSessionExpired(sess.ExpiresAt) { + errResp, _ := json.Marshal(map[string]string{ + "error": "UNAUTHORIZED", + "message": "invalid or expired session", + }) + http.Error(w, string(errResp), http.StatusUnauthorized) + return + } + sessionStillAuthorized := func() bool { + current, currentErr := database.GetSessionByTokenHash(entry.tokenHash) + if currentErr != nil || current == nil || auth.IsSessionExpired(current.ExpiresAt) { + return false + } + user, userErr := database.GetUserByID(current.UserID) + if userErr != nil || user == nil { + return false + } + role, roleErr := database.GetRoleByID(user.RoleID) + if roleErr != nil || role == nil { + return false + } + return permissions.HasAdmin(role.Permissions) + } + if !sessionStillAuthorized() { + errResp, _ := json.Marshal(map[string]string{ + "error": "FORBIDDEN", + "message": "administrator permission required", + }) + http.Error(w, string(errResp), http.StatusForbidden) + return + } // Check that we can flush (required for SSE). flusher, ok := w.(http.Flusher) @@ -318,11 +434,17 @@ func handleLogStream(ringBuf *RingBuffer, database *db.DB) http.HandlerFunc { for { select { case entry := <-ch: + if !sessionStillAuthorized() { + return + } if data, err := json.Marshal(entry); err == nil { _, _ = fmt.Fprintf(w, "data: %s\n\n", data) flusher.Flush() } case <-keepalive.C: + if !sessionStillAuthorized() { + return + } _, _ = fmt.Fprint(w, ": keepalive\n\n") flusher.Flush() case <-ctx.Done(): diff --git a/Server/admin/setup_handler.go b/Server/admin/setup_handler.go index 996f2b3f..8809888f 100644 --- a/Server/admin/setup_handler.go +++ b/Server/admin/setup_handler.go @@ -3,6 +3,7 @@ package admin import ( "encoding/json" "log/slog" + "net" "net/http" "strings" "time" @@ -63,8 +64,13 @@ func handleSetupStatus(database *db.DB) http.HandlerFunc { func handleSetup(database *db.DB) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { // Rate limit: 5 attempts per minute per IP. - ip := r.RemoteAddr - setupKey := "setup:" + ip + // Strip the port so that different source ports from the same IP + // are correctly grouped under a single rate-limit bucket. + host, _, err := net.SplitHostPort(r.RemoteAddr) + if err != nil { + host = r.RemoteAddr + } + setupKey := "setup:" + host if !setupLimiter.Allow(setupKey, 5, time.Minute) { writeErr(w, http.StatusTooManyRequests, "RATE_LIMITED", "too many setup attempts, try again later") return @@ -93,6 +99,12 @@ func handleSetup(database *db.DB) http.HandlerFunc { return } + // Validate username format (length, no control/invisible chars). + if err := auth.ValidateUsername(req.Username); err != nil { + writeErr(w, http.StatusBadRequest, "BAD_REQUEST", err.Error()) + return + } + if err := auth.ValidatePasswordStrength(req.Password); err != nil { writeErr(w, http.StatusBadRequest, "BAD_REQUEST", err.Error()) return @@ -120,7 +132,7 @@ func handleSetup(database *db.DB) http.HandlerFunc { } device := r.Header.Get("User-Agent") - if _, err := database.CreateSession(uid, auth.HashToken(token), device, ip); err != nil { + if _, err := database.CreateSession(uid, auth.HashToken(token), device, host); err != nil { writeErr(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to create session") return } diff --git a/Server/admin/static/index.html b/Server/admin/static/index.html index dc7c7c54..1c44f155 100644 --- a/Server/admin/static/index.html +++ b/Server/admin/static/index.html @@ -285,7 +285,7 @@ const state={section:'dashboard',token:localStorage.getItem('admin_token')||'', usersPage:1,auditPage:1,auditSearch:'',auditActionFilter:'all',auditCache:[],settingsChanged:false,backupRunning:false,updateApplying:false, cachedStats:null,cachedUpdate:null, logEntries:[],logLevels:{DEBUG:true,INFO:true,WARN:true,ERROR:true}, - logSearch:'',logAutoScroll:true,logPaused:false,logEventSource:null,logMaxLines:2000}; + logSearch:'',logAutoScroll:true,logPaused:false,logEventSource:null,logReconnectTimer:null,logConnectSeq:0,logMaxLines:2000}; /* ═══ API ═══ */ async function api(method,path,body){ @@ -388,11 +388,11 @@ function renderNav(){ } function navigateTo(id){ - if(state.logEventSource&&state.section==='logs'&&id!=='logs'){state.logEventSource.close();state.logEventSource=null} + if(state.section==='logs'&&id!=='logs'){state.logConnectSeq++;if(state.logEventSource){state.logEventSource.close();state.logEventSource=null}if(state.logReconnectTimer){clearTimeout(state.logReconnectTimer);state.logReconnectTimer=null}} state.section=id;renderNav();renderContent(); } -function doLogout(){state.token='';localStorage.removeItem('admin_token');showOverlay('loginOverlay')} +function doLogout(){state.logConnectSeq++;if(state.logEventSource){state.logEventSource.close();state.logEventSource=null}if(state.logReconnectTimer){clearTimeout(state.logReconnectTimer);state.logReconnectTimer=null}state.token='';localStorage.removeItem('admin_token');showOverlay('loginOverlay')} /* ═══ Content Router ═══ */ function renderContent(){ @@ -662,14 +662,24 @@ function toggleLogPause(){ const btn=document.getElementById('pauseBtn');if(btn)btn.textContent=state.logPaused?'▶ Resume':'⏸ Pause'; const dot=document.getElementById('logDot');if(dot)dot.className=state.logPaused?'dot-off':'dot-live'; const txt=document.getElementById('logStatusText'); - if(state.logPaused){if(state.logEventSource){state.logEventSource.close();state.logEventSource=null}if(txt)txt.textContent='Paused'} + if(state.logPaused){state.logConnectSeq++;if(state.logEventSource){state.logEventSource.close();state.logEventSource=null}if(state.logReconnectTimer){clearTimeout(state.logReconnectTimer);state.logReconnectTimer=null}if(txt)txt.textContent='Paused'} else{connectLogStream()} } -function connectLogStream(){ - if(state.logEventSource){state.logEventSource.close()} +function scheduleLogReconnect(){ + if(state.logPaused||state.section!=='logs'||state.logReconnectTimer)return; + state.logReconnectTimer=setTimeout(function(){state.logReconnectTimer=null;connectLogStream()},1500); +} + +async function connectLogStream(){ + if(state.logReconnectTimer){clearTimeout(state.logReconnectTimer);state.logReconnectTimer=null} + if(state.logEventSource){state.logEventSource.close();state.logEventSource=null} if(state.logPaused||state.section!=='logs')return; - const es=new EventSource('/admin/api/logs/stream?token='+encodeURIComponent(state.token)); + const connectSeq=++state.logConnectSeq; + let ticket; + try{const res=await api('POST','/logs/ticket');ticket=res.ticket}catch(err){const t=document.getElementById('logStatusText');const d=document.getElementById('logDot');const msg=(err&&err.message)||'';if(/authorization|invalid or expired session|session has expired|missing or invalid|administrator permission required/i.test(msg)){state.logPaused=true;state.logConnectSeq++;if(state.logReconnectTimer){clearTimeout(state.logReconnectTimer);state.logReconnectTimer=null}if(state.logEventSource){state.logEventSource.close();state.logEventSource=null}state.token='';localStorage.removeItem('admin_token');if(t)t.textContent='Session expired';if(d)d.className='dot-off';showOverlay('loginOverlay');return}if(t)t.textContent='Reconnect failed';if(d)d.className='dot-off';scheduleLogReconnect();return} + if(connectSeq!==state.logConnectSeq||state.logPaused||state.section!=='logs')return; + const es=new EventSource('/admin/api/logs/stream?ticket='+encodeURIComponent(ticket)); state.logEventSource=es; es.onopen=function(){const t=document.getElementById('logStatusText');if(t)t.textContent='Connected'}; es.onmessage=function(e){ @@ -679,7 +689,7 @@ function connectLogStream(){ const c=document.getElementById('logCount');if(c)c.textContent=state.logEntries.length+' entries'; }catch(err){} }; - es.onerror=function(){const t=document.getElementById('logStatusText');if(t)t.textContent='Reconnecting...';const d=document.getElementById('logDot');if(d)d.className='dot-off'}; + es.onerror=function(){if(state.logEventSource===es){state.logEventSource.close();state.logEventSource=null}const t=document.getElementById('logStatusText');if(t)t.textContent='Reconnecting...';const d=document.getElementById('logDot');if(d)d.className='dot-off';scheduleLogReconnect()}; } function matchesLogFilter(entry){ diff --git a/Server/db/db.go b/Server/db/db.go index 922e7457..da2903cd 100644 --- a/Server/db/db.go +++ b/Server/db/db.go @@ -3,6 +3,7 @@ package db import ( + "context" "database/sql" "fmt" @@ -94,21 +95,41 @@ func (d *DB) QueryRow(query string, args ...any) *sql.Row { return d.sqlDB.QueryRow(query, args...) } +// QueryRowContext executes a query that returns at most one row, with context. +func (d *DB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row { + return d.sqlDB.QueryRowContext(ctx, query, args...) +} + // Exec executes a query that doesn't return rows. func (d *DB) Exec(query string, args ...any) (sql.Result, error) { return d.sqlDB.Exec(query, args...) } +// ExecContext executes a query that doesn't return rows, with context. +func (d *DB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) { + return d.sqlDB.ExecContext(ctx, query, args...) +} + // Query executes a query that returns multiple rows. func (d *DB) Query(query string, args ...any) (*sql.Rows, error) { return d.sqlDB.Query(query, args...) } +// QueryContext executes a query that returns multiple rows, with context. +func (d *DB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) { + return d.sqlDB.QueryContext(ctx, query, args...) +} + // Begin starts a database transaction. func (d *DB) Begin() (*sql.Tx, error) { return d.sqlDB.Begin() } +// BeginTx starts a database transaction with context and options. +func (d *DB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) { + return d.sqlDB.BeginTx(ctx, opts) +} + // SQLDb returns the underlying *sql.DB for cases requiring direct access. func (d *DB) SQLDb() *sql.DB { return d.sqlDB diff --git a/Server/go.mod b/Server/go.mod index 4867e0fa..93fa7a19 100644 --- a/Server/go.mod +++ b/Server/go.mod @@ -28,7 +28,6 @@ require ( github.com/antlr4-go/antlr/v4 v4.13.1 // indirect github.com/aymerick/douceur v0.2.0 // indirect github.com/benbjohnson/clock v1.3.5 // indirect - github.com/beorn7/perks v1.0.1 // indirect github.com/bep/debounce v1.2.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dennwc/iters v1.2.2 // indirect @@ -45,12 +44,11 @@ require ( github.com/google/cel-go v0.27.0 // indirect github.com/gorilla/css v1.0.1 // indirect github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect - github.com/hashicorp/go-cleanhttp v0.5.2 // indirect - github.com/hashicorp/go-retryablehttp v0.7.7 // indirect github.com/jxskiss/base62 v1.1.0 // indirect github.com/klauspost/compress v1.18.4 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/knadh/koanf/maps v0.1.2 // indirect + github.com/kr/pretty v0.3.1 // indirect github.com/lithammer/shortuuid/v4 v4.2.0 // indirect github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 // indirect github.com/livekit/mediatransportutil v0.0.0-20251128105421-19c7a7b81c22 // indirect @@ -60,7 +58,6 @@ require ( github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/moby/sys/user v0.4.0 // indirect - github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/nats-io/nats.go v1.48.0 // indirect github.com/nats-io/nkeys v0.4.15 // indirect github.com/nats-io/nuid v1.0.1 // indirect @@ -81,13 +78,10 @@ require ( github.com/pion/transport/v4 v4.0.1 // indirect github.com/pion/turn/v4 v4.1.4 // indirect github.com/pion/webrtc/v4 v4.2.9 // indirect - github.com/prometheus/client_golang v1.22.0 // indirect - github.com/prometheus/client_model v0.6.2 // indirect - github.com/prometheus/common v0.64.0 // indirect - github.com/prometheus/procfs v0.19.2 // indirect github.com/puzpuzpuz/xsync/v3 v3.5.1 // indirect github.com/redis/go-redis/v9 v9.17.2 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/rogpeppe/go-internal v1.10.0 // indirect github.com/twitchtv/twirp v8.1.3+incompatible // indirect github.com/wlynxg/anet v0.0.5 // indirect github.com/zeebo/xxh3 v1.1.0 // indirect @@ -104,8 +98,9 @@ require ( golang.org/x/time v0.14.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect - google.golang.org/grpc v1.79.1 // indirect + google.golang.org/grpc v1.79.3 // indirect google.golang.org/protobuf v1.36.11 // indirect + gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect gopkg.in/yaml.v3 v3.0.1 // indirect modernc.org/libc v1.67.6 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/Server/go.sum b/Server/go.sum index 2a68833b..a1e880ec 100644 --- a/Server/go.sum +++ b/Server/go.sum @@ -20,8 +20,6 @@ github.com/aymerick/douceur v0.2.0 h1:Mv+mAeH1Q+n9Fr+oyamOlAkUNPWPlA8PPGR0QAaYuP github.com/aymerick/douceur v0.2.0/go.mod h1:wlT5vV2O3h55X9m7iVYN0TBM0NH/MmbLnd30/FjWUq4= github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o= github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= -github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= -github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bep/debounce v1.2.1 h1:v67fRdBA9UQu2NhLFXrSg0Brw7CexQekrBwDMM8bzeY= github.com/bep/debounce v1.2.1/go.mod h1:H8yggRPQKLUhUoqrJC1bO2xNya7vanpDl7xR3ISbCJ0= github.com/brianvoe/gofakeit/v6 v6.28.0 h1:Xib46XXuQfmlLS2EXRuJpqcw8St6qSZz75OUo0tgAW4= @@ -40,6 +38,7 @@ github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -95,10 +94,6 @@ github.com/gorilla/css v1.0.1 h1:ntNaBIghp6JmvWnxbZKANoLyuXTPZ4cAMlo6RyhlbO8= github.com/gorilla/css v1.0.1/go.mod h1:BvnYkspnSzMmwRK+b8/xgNPLiIuNZr6vbZBTPQ2A3b0= github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo= github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA= -github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= -github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48= -github.com/hashicorp/go-retryablehttp v0.7.7 h1:C8hUCYzor8PIfXHa4UrZkU4VvK8o9ISHxT2Q8+VepXU= -github.com/hashicorp/go-retryablehttp v0.7.7/go.mod h1:pkQpWZeYWskR+D1tR2O5OcBFOxfA7DoAO6xtkuQnHTk= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/jxskiss/base62 v1.1.0 h1:A5zbF8v8WXx2xixnAKD2w+abC+sIzYJX+nxmhA6HWFw= @@ -119,8 +114,11 @@ github.com/knadh/koanf/providers/structs v1.0.0 h1:DznjB7NQykhqCar2LvNug3MuxEQsZ github.com/knadh/koanf/providers/structs v1.0.0/go.mod h1:kjo5TFtgpaZORlpoJqcbeLowM2cINodv8kX+oFAeQ1w= github.com/knadh/koanf/v2 v2.3.3 h1:jLJC8XCRfLC7n4F+ZKKdBsbq1bfXTpuFhf4L7t94D94= github.com/knadh/koanf/v2 v2.3.3/go.mod h1:gRb40VRAbd4iJMYYD5IxZ6hfuopFcXBpc9bbQpZwo28= -github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lithammer/shortuuid/v4 v4.2.0 h1:LMFOzVB3996a7b8aBuEXxqOBflbfPQAiVzkIcHO0h8c= @@ -155,8 +153,6 @@ github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= -github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= -github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nats-io/nats.go v1.48.0 h1:pSFyXApG+yWU/TgbKCjmm5K4wrHu86231/w84qRVR+U= github.com/nats-io/nats.go v1.48.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g= github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= @@ -207,19 +203,12 @@ github.com/pion/turn/v4 v4.1.4 h1:EU11yMXKIsK43FhcUnjLlrhE4nboHZq+TXBIi3QpcxQ= github.com/pion/turn/v4 v4.1.4/go.mod h1:ES1DXVFKnOhuDkqn9hn5VJlSWmZPaRJLyBXoOeO/BmQ= github.com/pion/webrtc/v4 v4.2.9 h1:DZIh1HAhPIL3RvwEDFsmL5hfPSLEpxsQk9/Jir2vkJE= github.com/pion/webrtc/v4 v4.2.9/go.mod h1:9EmLZve0H76eTzf8v2FmchZ6tcBXtDgpfTEu+drW6SY= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/prometheus/client_golang v1.22.0 h1:rb93p9lokFEsctTys46VnV1kLCDpVZ0a/Y92Vm0Zc6Q= -github.com/prometheus/client_golang v1.22.0/go.mod h1:R7ljNsLXhuQXYZYtw6GAE9AZg8Y7vEW5scdCXrWRXC0= -github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= -github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= -github.com/prometheus/common v0.64.0 h1:pdZeA+g617P7oGv1CzdTzyeShxAGrTBsolKNOLQPGO4= -github.com/prometheus/common v0.64.0/go.mod h1:0gZns+BLRQ3V6NdaerOhMbwwRbNh9hkGINtQAsP5GS8= -github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws= -github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw= github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg= github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA= github.com/redis/go-redis/v9 v9.17.2 h1:P2EGsA4qVIM3Pp+aPocCJ7DguDHhqrXNhVcEp4ViluI= @@ -228,6 +217,9 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rodaine/protogofakeit v0.1.1 h1:ZKouljuRM3A+TArppfBqnH8tGZHOwM/pjvtXe9DaXH8= github.com/rodaine/protogofakeit v0.1.1/go.mod h1:pXn/AstBYMaSfc1/RqH3N82pBuxtWgejz1AlYpY1mI0= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/shoenig/test v1.7.0 h1:eWcHtTXa6QLnBvm0jgEabMRN/uJ4DMV3M8xUGgRkZmk= github.com/shoenig/test v1.7.0/go.mod h1:UxJ6u/x2v/TNs/LoLxBNJRV9DiwBBKYxXSyczsBHFoI= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= @@ -333,13 +325,13 @@ google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 h1: google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:kSJwQxqmFXeo79zOmbrALdflXQeAYcUbgS7PbpMknCY= google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 h1:mWPCjDEyshlQYzBpMNHaEof6UX1PmHcaUODUywQ0uac= google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= -google.golang.org/grpc v1.79.1 h1:zGhSi45ODB9/p3VAawt9a+O/MULLl9dpizzNNpq7flY= -google.golang.org/grpc v1.79.1/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= +google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= +google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/Server/main.go b/Server/main.go index bdcfb67a..94cf9b8f 100644 --- a/Server/main.go +++ b/Server/main.go @@ -107,7 +107,8 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error { } // ── 5. Build HTTP router ─────────────────────────────────────────────── - router, hub := api.NewRouter(cfg, database, version, logBuf) + router, hub, routerCleanup := api.NewRouter(cfg, database, version, logBuf) + defer routerCleanup() // ── 6. Start server ──────────────────────────────────────────────────── addr := fmt.Sprintf(":%d", cfg.Server.Port) diff --git a/Server/ws/client.go b/Server/ws/client.go index 75b01a35..dc19e202 100644 --- a/Server/ws/client.go +++ b/Server/ws/client.go @@ -1,6 +1,7 @@ package ws import ( + "context" "sync" "time" @@ -19,6 +20,7 @@ const SessionCheckInterval = 10 type Client struct { hub *Hub conn wsConn // interface — nil in unit tests + ctx context.Context // derived from WS upgrade request; cancelled on disconnect userID int64 user *db.User channelID int64 // currently viewed channel for channel-scoped broadcasts @@ -47,11 +49,12 @@ type wsConn interface { } // newClient creates a real client wrapping a WebSocket connection (set by serve.go). -func newClient(hub *Hub, conn wsConn, user *db.User, tokenHash string) *Client { +func newClient(hub *Hub, conn wsConn, user *db.User, tokenHash string, ctx context.Context) *Client { now := time.Now() return &Client{ hub: hub, conn: conn, + ctx: ctx, userID: user.ID, user: user, tokenHash: tokenHash, @@ -72,6 +75,7 @@ func (c *Client) GetTokenHash() string { func NewTestClient(hub *Hub, userID int64, send chan []byte) *Client { return &Client{ hub: hub, + ctx: context.Background(), userID: userID, send: send, } @@ -81,6 +85,7 @@ func NewTestClient(hub *Hub, userID int64, send chan []byte) *Client { func NewTestClientWithChannel(hub *Hub, userID, channelID int64, send chan []byte) *Client { return &Client{ hub: hub, + ctx: context.Background(), userID: userID, channelID: channelID, send: send, @@ -92,6 +97,7 @@ func NewTestClientWithChannel(hub *Hub, userID, channelID int64, send chan []byt func NewTestClientWithUser(hub *Hub, user *db.User, channelID int64, send chan []byte) *Client { return &Client{ hub: hub, + ctx: context.Background(), userID: user.ID, user: user, channelID: channelID, @@ -111,6 +117,7 @@ func SetClientVoiceChID(c *Client, channelID int64) { func NewTestClientWithTokenHash(hub *Hub, user *db.User, tokenHash string, channelID int64, send chan []byte) *Client { return &Client{ hub: hub, + ctx: context.Background(), userID: user.ID, user: user, tokenHash: tokenHash, diff --git a/Server/ws/export_test.go b/Server/ws/export_test.go index aac37233..9bb8baf1 100644 --- a/Server/ws/export_test.go +++ b/Server/ws/export_test.go @@ -24,6 +24,11 @@ func (h *Hub) GetCachedSettingsForTest() (string, string) { return h.getCachedSettings() } +// GetClientVoiceChIDForTest exposes Client.getVoiceChID for external tests. +func GetClientVoiceChIDForTest(c *Client) int64 { + return c.getVoiceChID() +} + // ExpireSettingsCacheForTest forces the settings cache to appear stale so that // the next call to getCachedSettings triggers a DB refresh. func (h *Hub) ExpireSettingsCacheForTest() { diff --git a/Server/ws/handlers.go b/Server/ws/handlers.go index 654a7b74..fe5ba6ea 100644 --- a/Server/ws/handlers.go +++ b/Server/ws/handlers.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -38,7 +39,7 @@ func (h *Hub) HandleMessageForTest(c *Client, raw []byte) { // disconnect-triggered cleanup without an explicit voice_leave message. // Exported for ws_test package use only. func (h *Hub) HandleVoiceLeaveForTest(c *Client) { - h.handleVoiceLeave(c) + h.handleVoiceLeave(context.Background(), c) } // handleMessage parses the envelope and dispatches to the appropriate handler. @@ -101,7 +102,7 @@ func (h *Hub) handleMessage(c *Client, raw []byte) { reqLog.Debug("ws ← client message") - if !h.registry.Dispatch(env.Type, h, c, env.ID, env.Payload) { + if !h.registry.Dispatch(c.ctx, env.Type, h, c, env.ID, env.Payload) { reqLog.Warn("ws handleMessage unknown type") c.sendMsg(buildErrorMsg(ErrCodeUnknownType, fmt.Sprintf("unknown message type: %s", env.Type))) } diff --git a/Server/ws/handlers_chat.go b/Server/ws/handlers_chat.go index 14ef54d0..271039e8 100644 --- a/Server/ws/handlers_chat.go +++ b/Server/ws/handlers_chat.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -11,19 +12,19 @@ import ( // registerChatHandlers registers all chat-related message handlers. func registerChatHandlers(r *HandlerRegistry) { - r.Register(MsgTypeChatSend, func(h *Hub, c *Client, reqID string, payload json.RawMessage) { - h.handleChatSend(c, reqID, payload) + r.Register(MsgTypeChatSend, func(ctx context.Context, h *Hub, c *Client, reqID string, payload json.RawMessage) { + h.handleChatSend(ctx, c, reqID, payload) }) - r.Register(MsgTypeChatEdit, func(h *Hub, c *Client, reqID string, payload json.RawMessage) { - h.handleChatEdit(c, reqID, payload) + r.Register(MsgTypeChatEdit, func(ctx context.Context, h *Hub, c *Client, reqID string, payload json.RawMessage) { + h.handleChatEdit(ctx, c, reqID, payload) }) - r.Register(MsgTypeChatDelete, func(h *Hub, c *Client, reqID string, payload json.RawMessage) { - h.handleChatDelete(c, reqID, payload) + r.Register(MsgTypeChatDelete, func(ctx context.Context, h *Hub, c *Client, reqID string, payload json.RawMessage) { + h.handleChatDelete(ctx, c, reqID, payload) }) } // handleChatSend processes a chat_send message. -func (h *Hub) handleChatSend(c *Client, reqID string, payload json.RawMessage) { +func (h *Hub) handleChatSend(ctx context.Context, c *Client, reqID string, payload json.RawMessage) { // Rate limit. ratKey := fmt.Sprintf("chat:%d", c.userID) if !h.limiter.Allow(ratKey, chatRateLimit, chatWindow) { @@ -203,7 +204,7 @@ func (h *Hub) handleChatSend(c *Client, reqID string, payload json.RawMessage) { } // handleChatEdit processes a chat_edit message. -func (h *Hub) handleChatEdit(c *Client, _ string, payload json.RawMessage) { +func (h *Hub) handleChatEdit(ctx context.Context, c *Client, _ string, payload json.RawMessage) { ratKey := fmt.Sprintf("chat_edit:%d", c.userID) if !h.limiter.Allow(ratKey, chatRateLimit, chatWindow) { c.sendMsg(buildRateLimitError("too many edits", chatWindow.Seconds())) @@ -289,7 +290,7 @@ func (h *Hub) handleChatEdit(c *Client, _ string, payload json.RawMessage) { } // handleChatDelete processes a chat_delete message. -func (h *Hub) handleChatDelete(c *Client, _ string, payload json.RawMessage) { +func (h *Hub) handleChatDelete(ctx context.Context, c *Client, _ string, payload json.RawMessage) { ratKey := fmt.Sprintf("chat_delete:%d", c.userID) if !h.limiter.Allow(ratKey, chatRateLimit, chatWindow) { c.sendMsg(buildRateLimitError("too many deletes", chatWindow.Seconds())) diff --git a/Server/ws/handlers_ping.go b/Server/ws/handlers_ping.go index b206dd2e..3aada77a 100644 --- a/Server/ws/handlers_ping.go +++ b/Server/ws/handlers_ping.go @@ -1,10 +1,13 @@ package ws -import "encoding/json" +import ( + "context" + "encoding/json" +) // registerPingHandler registers the ping/pong handler. func registerPingHandler(r *HandlerRegistry) { - r.Register(MsgTypePing, func(_ *Hub, c *Client, _ string, _ json.RawMessage) { + r.Register(MsgTypePing, func(_ context.Context, _ *Hub, c *Client, _ string, _ json.RawMessage) { c.sendMsg(buildJSON(map[string]any{"type": MsgTypePong})) }) } diff --git a/Server/ws/handlers_presence.go b/Server/ws/handlers_presence.go index b52095d0..7e27451d 100644 --- a/Server/ws/handlers_presence.go +++ b/Server/ws/handlers_presence.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -10,19 +11,19 @@ import ( // registerPresenceHandlers registers presence, typing, and channel focus handlers. func registerPresenceHandlers(r *HandlerRegistry) { - r.Register(MsgTypeTypingStart, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleTyping(c, payload) + r.Register(MsgTypeTypingStart, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleTyping(ctx, c, payload) }) - r.Register(MsgTypePresenceUpdate, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handlePresence(c, payload) + r.Register(MsgTypePresenceUpdate, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handlePresence(ctx, c, payload) }) - r.Register(MsgTypeChannelFocus, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleChannelFocus(c, payload) + r.Register(MsgTypeChannelFocus, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleChannelFocus(ctx, c, payload) }) } // handleTyping processes a typing_start message. -func (h *Hub) handleTyping(c *Client, payload json.RawMessage) { +func (h *Hub) handleTyping(ctx context.Context, c *Client, payload json.RawMessage) { channelID, err := parseChannelID(payload) if err != nil || channelID <= 0 { c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "channel_id must be positive integer")) @@ -64,7 +65,7 @@ func (h *Hub) handleTyping(c *Client, payload json.RawMessage) { } // handlePresence processes a presence_update message. -func (h *Hub) handlePresence(c *Client, payload json.RawMessage) { +func (h *Hub) handlePresence(ctx context.Context, c *Client, payload json.RawMessage) { ratKey := fmt.Sprintf("presence:%d", c.userID) if !h.limiter.Allow(ratKey, presenceRateLimit, presenceWindow) { c.sendMsg(buildRateLimitError("too many presence updates", presenceWindow.Seconds())) @@ -96,7 +97,7 @@ func (h *Hub) handlePresence(c *Client, payload json.RawMessage) { // handleChannelFocus sets which channel the client is currently viewing, // so channel-scoped broadcasts (chat messages, typing) reach them. // Also updates read_states so unread counts decrease when the user views a channel. -func (h *Hub) handleChannelFocus(c *Client, payload json.RawMessage) { +func (h *Hub) handleChannelFocus(ctx context.Context, c *Client, payload json.RawMessage) { chID, err := parseChannelID(payload) if err != nil || chID <= 0 { slog.Debug("handleChannelFocus: invalid channel_id", "user_id", c.userID, "err", err) diff --git a/Server/ws/handlers_reaction.go b/Server/ws/handlers_reaction.go index 3d1904fa..76e5ab9e 100644 --- a/Server/ws/handlers_reaction.go +++ b/Server/ws/handlers_reaction.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -10,16 +11,16 @@ import ( // registerReactionHandlers registers reaction_add and reaction_remove handlers. func registerReactionHandlers(r *HandlerRegistry) { - r.Register(MsgTypeReactionAdd, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleReaction(c, true, payload) + r.Register(MsgTypeReactionAdd, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleReaction(ctx, c, true, payload) }) - r.Register(MsgTypeReactionRemove, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleReaction(c, false, payload) + r.Register(MsgTypeReactionRemove, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleReaction(ctx, c, false, payload) }) } // handleReaction processes reaction_add and reaction_remove messages. -func (h *Hub) handleReaction(c *Client, add bool, payload json.RawMessage) { +func (h *Hub) handleReaction(ctx context.Context, c *Client, add bool, payload json.RawMessage) { ratKey := fmt.Sprintf("reaction:%d", c.userID) if !h.limiter.Allow(ratKey, reactionRateLimit, reactionWindow) { c.sendMsg(buildRateLimitError("too many reactions", reactionWindow.Seconds())) diff --git a/Server/ws/handlers_voice.go b/Server/ws/handlers_voice.go index f2f887d4..79d7ae34 100644 --- a/Server/ws/handlers_voice.go +++ b/Server/ws/handlers_voice.go @@ -1,31 +1,34 @@ package ws -import "encoding/json" +import ( + "context" + "encoding/json" +) // registerVoiceHandlers registers all voice-related message handlers. // The handler methods themselves live in voice_join.go, voice_leave.go, // voice_controls.go, and voice_broadcast.go — this function only wires // them into the registry. func registerVoiceHandlers(r *HandlerRegistry) { - r.Register(MsgTypeVoiceJoin, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleVoiceJoin(c, payload) + r.Register(MsgTypeVoiceJoin, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleVoiceJoin(ctx, c, payload) }) - r.Register(MsgTypeVoiceLeave, func(h *Hub, c *Client, _ string, _ json.RawMessage) { - h.handleVoiceLeave(c) + r.Register(MsgTypeVoiceLeave, func(ctx context.Context, h *Hub, c *Client, _ string, _ json.RawMessage) { + h.handleVoiceLeave(ctx, c) }) - r.Register(MsgTypeVoiceTokenRefresh, func(h *Hub, c *Client, _ string, _ json.RawMessage) { - h.handleVoiceTokenRefresh(c) + r.Register(MsgTypeVoiceTokenRefresh, func(ctx context.Context, h *Hub, c *Client, _ string, _ json.RawMessage) { + h.handleVoiceTokenRefresh(ctx, c) }) - r.Register(MsgTypeVoiceMute, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleVoiceMute(c, payload) + r.Register(MsgTypeVoiceMute, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleVoiceMute(ctx, c, payload) }) - r.Register(MsgTypeVoiceDeafen, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleVoiceDeafen(c, payload) + r.Register(MsgTypeVoiceDeafen, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleVoiceDeafen(ctx, c, payload) }) - r.Register(MsgTypeVoiceCamera, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleVoiceCamera(c, payload) + r.Register(MsgTypeVoiceCamera, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleVoiceCamera(ctx, c, payload) }) - r.Register(MsgTypeVoiceScreenshare, func(h *Hub, c *Client, _ string, payload json.RawMessage) { - h.handleVoiceScreenshare(c, payload) + r.Register(MsgTypeVoiceScreenshare, func(ctx context.Context, h *Hub, c *Client, _ string, payload json.RawMessage) { + h.handleVoiceScreenshare(ctx, c, payload) }) } diff --git a/Server/ws/hub.go b/Server/ws/hub.go index 5802320e..5feeac1c 100644 --- a/Server/ws/hub.go +++ b/Server/ws/hub.go @@ -23,19 +23,19 @@ type broadcastMsg struct { // Hub manages all active WebSocket clients and routes messages between them. // All exported methods are safe to call from multiple goroutines. type Hub struct { - clients map[int64]*Client - mu sync.RWMutex - db *db.DB - limiter *auth.RateLimiter - broadcast chan broadcastMsg - register chan *Client - unregister chan *Client - stop chan struct{} - stopOnce sync.Once - livekit *LiveKitClient - lkProcess *LiveKitProcess - registry *HandlerRegistry - permChecker *permissions.Checker + clients map[int64]*Client + mu sync.RWMutex + db *db.DB + limiter *auth.RateLimiter + broadcast chan broadcastMsg + register chan *Client + unregister chan *Client + stop chan struct{} + stopOnce sync.Once + livekit *LiveKitClient + lkProcess *LiveKitProcess + registry *HandlerRegistry + permChecker *permissions.Checker seq uint64 // atomic monotonic sequence counter replayBuf *EventRingBuffer // recent broadcast events for reconnection replay @@ -175,24 +175,9 @@ func (h *Hub) Run() { case <-h.stop: return case c := <-h.register: - h.mu.Lock() - if old, exists := h.clients[c.userID]; exists { - // Kick the stale connection atomically before registering - // the new one — prevents TOCTOU races on duplicate login. - slog.Warn("hub: kicking stale connection for re-registering user", - "user_id", c.userID) - old.closeSend() - } - h.clients[c.userID] = c - slog.Info("hub: client registered", "user_id", c.userID, "total_clients", len(h.clients)) - h.mu.Unlock() + h.registerNow(c) case c := <-h.unregister: - h.mu.Lock() - if current, ok := h.clients[c.userID]; ok && current == c { - delete(h.clients, c.userID) - slog.Info("hub: client unregistered", "user_id", c.userID, "total_clients", len(h.clients)) - } - h.mu.Unlock() + h.unregisterNow(c) case bm := <-h.broadcast: h.deliverBroadcast(bm) case <-staleTicker.C: @@ -258,7 +243,9 @@ func (h *Hub) CleanupVoiceForChannel(channelID int64) { // Clean up DB state and LiveKit for each participant. for _, vs := range states { - _ = h.db.LeaveVoiceChannel(vs.UserID) + if err := h.db.LeaveVoiceChannel(vs.UserID); err != nil { + slog.Error("CleanupVoiceForChannel LeaveVoiceChannel", "err", err, "user_id", vs.UserID, "channel_id", channelID) + } // Clear client voice state. h.mu.RLock() @@ -306,6 +293,33 @@ func (h *Hub) Unregister(c *Client) { h.unregister <- c } +func (h *Hub) registerNow(c *Client) { + h.mu.Lock() + if old, exists := h.clients[c.userID]; exists { + oldVoiceChID := old.clearVoiceChID() + if c.getVoiceChID() == 0 { + c.setVoiceChID(oldVoiceChID) + } + // Kick the stale connection atomically before registering + // the new one — prevents TOCTOU races on duplicate login. + slog.Warn("hub: kicking stale connection for re-registering user", + "user_id", c.userID) + old.closeSend() + } + h.clients[c.userID] = c + slog.Info("hub: client registered", "user_id", c.userID, "total_clients", len(h.clients)) + h.mu.Unlock() +} + +func (h *Hub) unregisterNow(c *Client) { + h.mu.Lock() + if current, ok := h.clients[c.userID]; ok && current == c { + delete(h.clients, c.userID) + slog.Info("hub: client unregistered", "user_id", c.userID, "total_clients", len(h.clients)) + } + h.mu.Unlock() +} + // BroadcastToChannel enqueues msg for delivery to all clients subscribed to // channelID. When channelID is 0 the message is sent to every connected client. // Non-blocking: if the broadcast channel is full the message is dropped with a warning. diff --git a/Server/ws/registry.go b/Server/ws/registry.go index 8bce15e0..b196d095 100644 --- a/Server/ws/registry.go +++ b/Server/ws/registry.go @@ -1,11 +1,14 @@ package ws -import "encoding/json" +import ( + "context" + "encoding/json" +) // MessageHandler is the function signature for all WebSocket message handlers. -// It receives the hub, the sending client, the request ID from the envelope, -// and the raw JSON payload. -type MessageHandler func(h *Hub, c *Client, reqID string, payload json.RawMessage) +// It receives a context (derived from the client's WS connection), the hub, +// the sending client, the request ID from the envelope, and the raw JSON payload. +type MessageHandler func(ctx context.Context, h *Hub, c *Client, reqID string, payload json.RawMessage) // HandlerRegistry maps message type strings to their handler functions. // It is not safe for concurrent use after initialization; all Register @@ -28,12 +31,12 @@ func (r *HandlerRegistry) Register(msgType string, handler MessageHandler) { // Dispatch looks up the handler for msgType and invokes it. Returns true if a // handler was found and called, false if no handler is registered for the type. -func (r *HandlerRegistry) Dispatch(msgType string, h *Hub, c *Client, reqID string, payload json.RawMessage) bool { +func (r *HandlerRegistry) Dispatch(ctx context.Context, msgType string, h *Hub, c *Client, reqID string, payload json.RawMessage) bool { handler, ok := r.handlers[msgType] if !ok { return false } - handler(h, c, reqID, payload) + handler(ctx, h, c, reqID, payload) return true } diff --git a/Server/ws/registry_test.go b/Server/ws/registry_test.go index 3960b595..5885b9f4 100644 --- a/Server/ws/registry_test.go +++ b/Server/ws/registry_test.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "sort" "testing" @@ -10,14 +11,14 @@ func TestHandlerRegistry_RegisterAndDispatch(t *testing.T) { r := NewHandlerRegistry() called := false - r.Register("test_type", func(h *Hub, c *Client, reqID string, payload json.RawMessage) { + r.Register("test_type", func(ctx context.Context, h *Hub, c *Client, reqID string, payload json.RawMessage) { called = true if reqID != "req-1" { t.Errorf("expected reqID %q, got %q", "req-1", reqID) } }) - ok := r.Dispatch("test_type", nil, nil, "req-1", nil) + ok := r.Dispatch(context.Background(), "test_type", nil, nil, "req-1", nil) if !ok { t.Fatal("Dispatch returned false for registered type") } @@ -29,7 +30,7 @@ func TestHandlerRegistry_RegisterAndDispatch(t *testing.T) { func TestHandlerRegistry_DispatchUnknownType(t *testing.T) { r := NewHandlerRegistry() - ok := r.Dispatch("nonexistent", nil, nil, "", nil) + ok := r.Dispatch(context.Background(), "nonexistent", nil, nil, "", nil) if ok { t.Fatal("Dispatch returned true for unregistered type") } diff --git a/Server/ws/serve.go b/Server/ws/serve.go index d6766688..0bfbbe36 100644 --- a/Server/ws/serve.go +++ b/Server/ws/serve.go @@ -43,9 +43,8 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun return } - c := newClient(hub, conn, user, tokenHash) + c := newClient(hub, conn, user, tokenHash, r.Context()) c.remoteAddr = r.RemoteAddr - hub.Register(c) // Look up role name for protocol-compliant payloads and cache on client. roleName := "member" @@ -59,6 +58,13 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun "WebSocket connected from "+r.RemoteAddr) ctx := r.Context() + startPumps := func() { + writeCtx, writeCancel := context.WithCancel(ctx) + go writePump(writeCtx, conn, c) + readPump(ctx, conn, hub, c) + c.closeSend() + writeCancel() + } // Reconnection with state recovery: if the client sent a last_seq, // try to replay missed events from the ring buffer instead of @@ -68,11 +74,20 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun if events != nil { // Replay succeeded — send auth_ok then missed events. slog.Info("ws sending auth_ok (reconnect)", "user_id", user.ID, "username", user.Username, "role", roleName) - _ = conn.Write(ctx, websocket.MessageText, hub.buildAuthOK(user, roleName)) + if err := conn.Write(ctx, websocket.MessageText, hub.buildAuthOK(user, roleName)); err != nil { + slog.Warn("ws: failed to send auth_ok (reconnect)", "user_id", user.ID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return + } for _, evt := range events { - _ = conn.Write(ctx, websocket.MessageText, evt) + if err := conn.Write(ctx, websocket.MessageText, evt); err != nil { + slog.Warn("ws: failed to send replay event", "user_id", user.ID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return + } } slog.Info("ws replay completed", "user_id", user.ID, "events_replayed", len(events), "from_seq", lastSeq) + hub.registerNow(c) // Update presence but skip member_join — user was already known. if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil { @@ -81,11 +96,7 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun hub.BroadcastToAll(buildPresenceMsg(user.ID, "online")) // Start pumps. - writeCtx, writeCancel := context.WithCancel(ctx) - go writePump(writeCtx, conn, c) - readPump(ctx, conn, hub, c) - c.closeSend() - writeCancel() + startPumps() return } // Replay failed (seq too old) — fall through to full ready payload. @@ -93,20 +104,29 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun } // Fresh connection or replay fallback: full auth_ok + ready flow. - if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil { - slog.Warn("ws UpdateUserStatus", "err", updateErr) - } - slog.Info("ws sending auth_ok", "user_id", user.ID, "username", user.Username, "role", roleName) - _ = conn.Write(ctx, websocket.MessageText, hub.buildAuthOK(user, roleName)) + if err := conn.Write(ctx, websocket.MessageText, hub.buildAuthOK(user, roleName)); err != nil { + slog.Warn("ws: failed to send auth_ok", "user_id", user.ID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return + } if ready, readyErr := hub.buildReady(database, user.ID); readyErr == nil { slog.Info("ws sending ready payload", "user_id", user.ID, "payload_bytes", len(ready)) - _ = conn.Write(ctx, websocket.MessageText, ready) + if err := conn.Write(ctx, websocket.MessageText, ready); err != nil { + slog.Warn("ws: failed to send ready payload", "user_id", user.ID, "err", err) + _ = conn.Close(websocket.StatusInternalError, "handshake failed") + return + } } else { slog.Error("buildReady failed", "user_id", user.ID, "err", readyErr) _ = conn.Write(ctx, websocket.MessageText, buildErrorMsg(ErrCodeInternal, "failed to build ready payload")) } + hub.registerNow(c) + + if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil { + slog.Warn("ws UpdateUserStatus", "err", updateErr) + } slog.Info("ws broadcasting member_join and presence", "user_id", user.ID, "username", user.Username) hub.BroadcastToAll(buildMemberJoin(user, roleName)) @@ -115,11 +135,7 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun // writePump runs in background; readPump blocks. // When readPump returns (disconnect), close the send channel first // so writePump drains any remaining messages, then cancel its context. - writeCtx, writeCancel := context.WithCancel(ctx) - go writePump(writeCtx, conn, c) - readPump(ctx, conn, hub, c) - c.closeSend() - writeCancel() + startPumps() } } @@ -149,10 +165,13 @@ func writePump(ctx context.Context, conn *websocket.Conn, c *Client) { func readPump(ctx context.Context, conn *websocket.Conn, hub *Hub, c *Client) { var lastReadErr error defer func() { - voiceChID := c.getVoiceChID() // capture BEFORE handleVoiceLeave clears it - hub.Unregister(c) - hub.handleVoiceLeave(c) + hub.unregisterNow(c) if c.user != nil { + replaced := hub.IsUserConnected(c.userID) + voiceChID := c.getVoiceChID() + if !replaced { + hub.handleVoiceLeave(ctx, c) + } c.mu.Lock() received := c.msgsReceived sent := c.msgsSent @@ -172,13 +191,18 @@ func readPump(ctx context.Context, conn *websocket.Conn, hub *Hub, c *Client) { if voiceChID > 0 { attrs = append(attrs, "voice_channel_id", voiceChID) } + if replaced { + attrs = append(attrs, "replaced", true) + } if lastReadErr != nil { attrs = append(attrs, "last_error", lastReadErr.Error()) } slog.Info("websocket disconnected", attrs...) - _ = hub.db.UpdateUserStatus(c.userID, "offline") - hub.BroadcastToAll(buildPresenceMsg(c.userID, "offline")) + if !replaced { + _ = hub.db.UpdateUserStatus(c.userID, "offline") + hub.BroadcastToAll(buildPresenceMsg(c.userID, "offline")) + } } }() @@ -207,11 +231,11 @@ func authenticateConn(conn *websocket.Conn, database *db.DB) (*db.User, string, var env envelope if err := json.Unmarshal(raw, &env); err != nil { - _ = conn.Write(ctx, websocket.MessageText, buildAuthError( "invalid message")) + _ = conn.Write(ctx, websocket.MessageText, buildAuthError("invalid message")) return nil, "", 0, fmt.Errorf("auth: invalid JSON: %w", err) } if env.Type != "auth" { - _ = conn.Write(ctx, websocket.MessageText, buildAuthError( "first message must be auth")) + _ = conn.Write(ctx, websocket.MessageText, buildAuthError("first message must be auth")) return nil, "", 0, fmt.Errorf("auth: unexpected type %q", env.Type) } @@ -220,25 +244,25 @@ func authenticateConn(conn *websocket.Conn, database *db.DB) (*db.User, string, LastSeq uint64 `json:"last_seq"` } if err := json.Unmarshal(env.Payload, &p); err != nil || p.Token == "" { - _ = conn.Write(ctx, websocket.MessageText, buildAuthError( "missing token")) + _ = conn.Write(ctx, websocket.MessageText, buildAuthError("missing token")) return nil, "", 0, fmt.Errorf("auth: missing token") } hash := auth.HashToken(p.Token) sess, err := database.GetSessionByTokenHash(hash) if err != nil || sess == nil { - _ = conn.Write(ctx, websocket.MessageText, buildAuthError( "invalid token")) + _ = conn.Write(ctx, websocket.MessageText, buildAuthError("invalid token")) return nil, "", 0, fmt.Errorf("auth: invalid session") } if auth.IsSessionExpired(sess.ExpiresAt) { - _ = conn.Write(ctx, websocket.MessageText, buildAuthError( "session expired")) + _ = conn.Write(ctx, websocket.MessageText, buildAuthError("session expired")) return nil, "", 0, fmt.Errorf("auth: session expired") } user, err := database.GetUserByID(sess.UserID) if err != nil || user == nil { - _ = conn.Write(ctx, websocket.MessageText, buildAuthError( "user not found")) + _ = conn.Write(ctx, websocket.MessageText, buildAuthError("user not found")) return nil, "", 0, fmt.Errorf("auth: user not found") } diff --git a/Server/ws/voice_controls.go b/Server/ws/voice_controls.go index 3fc5d217..08d19436 100644 --- a/Server/ws/voice_controls.go +++ b/Server/ws/voice_controls.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -12,7 +13,7 @@ import ( // 1. Parses muted bool. // 2. Updates DB. // 3. Broadcasts voice_state update to channel. -func (h *Hub) handleVoiceMute(c *Client, payload json.RawMessage) { +func (h *Hub) handleVoiceMute(ctx context.Context, c *Client, payload json.RawMessage) { ratKey := fmt.Sprintf("voice_mute:%d", c.userID) if !h.limiter.Allow(ratKey, voiceMuteRateLimit, voiceMuteWindow) { c.sendMsg(buildRateLimitError("too many mute toggles", voiceMuteWindow.Seconds())) @@ -46,7 +47,7 @@ func (h *Hub) handleVoiceMute(c *Client, payload json.RawMessage) { // 1. Parses deafened bool. // 2. Updates DB. // 3. Broadcasts voice_state update to channel. -func (h *Hub) handleVoiceDeafen(c *Client, payload json.RawMessage) { +func (h *Hub) handleVoiceDeafen(ctx context.Context, c *Client, payload json.RawMessage) { ratKey := fmt.Sprintf("voice_deafen:%d", c.userID) if !h.limiter.Allow(ratKey, voiceDeafenRateLimit, voiceDeafenWindow) { c.sendMsg(buildRateLimitError("too many deafen toggles", voiceDeafenWindow.Seconds())) @@ -83,7 +84,7 @@ func (h *Hub) handleVoiceDeafen(c *Client, payload json.RawMessage) { // 4. Enforces MaxVideo limit via DB count (race-free). // 5. Updates DB. // 6. Broadcasts voice_state update to channel. -func (h *Hub) handleVoiceCamera(c *Client, payload json.RawMessage) { +func (h *Hub) handleVoiceCamera(ctx context.Context, c *Client, payload json.RawMessage) { ratKey := fmt.Sprintf("voice_camera:%d", c.userID) if !h.limiter.Allow(ratKey, voiceCameraRateLimit, voiceCameraWindow) { c.sendMsg(buildRateLimitError("too many camera toggles", voiceCameraWindow.Seconds())) @@ -142,7 +143,7 @@ func (h *Hub) handleVoiceCamera(c *Client, payload json.RawMessage) { // 3. Parses enabled bool. // 4. Updates DB. // 5. Broadcasts voice_state update to channel. -func (h *Hub) handleVoiceScreenshare(c *Client, payload json.RawMessage) { +func (h *Hub) handleVoiceScreenshare(ctx context.Context, c *Client, payload json.RawMessage) { ratKey := fmt.Sprintf("voice_screenshare:%d", c.userID) if !h.limiter.Allow(ratKey, voiceScreenshareRateLimit, voiceScreenshareWindow) { c.sendMsg(buildRateLimitError("too many screenshare toggles", voiceScreenshareWindow.Seconds())) diff --git a/Server/ws/voice_join.go b/Server/ws/voice_join.go index 4f763e5a..31f67ca9 100644 --- a/Server/ws/voice_join.go +++ b/Server/ws/voice_join.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "log/slog" @@ -26,7 +27,7 @@ func validVoiceQuality(q string) bool { // 7. Sends existing voice states to the joiner. // 8. Broadcasts voice_state to all clients. // 9. Sends voice_config to the joiner. -func (h *Hub) handleVoiceJoin(c *Client, payload json.RawMessage) { +func (h *Hub) handleVoiceJoin(ctx context.Context, c *Client, payload json.RawMessage) { channelID, err := parseChannelID(payload) if err != nil || channelID <= 0 { c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "channel_id must be a positive integer")) @@ -70,7 +71,7 @@ func (h *Hub) handleVoiceJoin(c *Client, payload json.RawMessage) { // If user is already in a different voice channel, leave it first. if currentChID > 0 { - h.handleVoiceLeave(c) + h.handleVoiceLeave(ctx, c) } // Check channel capacity. @@ -184,7 +185,7 @@ func (h *Hub) handleVoiceJoin(c *Client, payload json.RawMessage) { // handleVoiceTokenRefresh generates a fresh LiveKit token for a client // that is already in a voice channel. This lets clients request a new token // (e.g. before a manual reconnect) without leaving and rejoining voice. -func (h *Hub) handleVoiceTokenRefresh(c *Client) { +func (h *Hub) handleVoiceTokenRefresh(ctx context.Context, c *Client) { ratKey := fmt.Sprintf("voice_token_refresh:%d", c.userID) if !h.limiter.Allow(ratKey, 1, 60*time.Second) { c.sendMsg(buildRateLimitError("token refresh rate limited", 60)) diff --git a/Server/ws/voice_leave.go b/Server/ws/voice_leave.go index 4f043fe4..ef7dbdcd 100644 --- a/Server/ws/voice_leave.go +++ b/Server/ws/voice_leave.go @@ -1,6 +1,7 @@ package ws import ( + "context" "log/slog" "time" ) @@ -9,7 +10,7 @@ import ( // 1. Gets old voiceChID from clearVoiceChID(). // 2. If was in voice: remove from DB (with retry), broadcast voice_leave. // 3. Call livekit.RemoveParticipant (ignore errors — participant may already be gone). -func (h *Hub) handleVoiceLeave(c *Client) { +func (h *Hub) handleVoiceLeave(ctx context.Context, c *Client) { oldChID := c.clearVoiceChID() if oldChID == 0 { slog.Debug("handleVoiceLeave no-op (already cleared)", "user_id", c.userID) @@ -42,33 +43,45 @@ func (h *Hub) handleVoiceLeave(c *Client) { } } -// leaveVoiceChannelWithRetry attempts to remove the voice state from the DB -// with up to 3 retries and exponential backoff (100ms, 200ms, 400ms). -// Returns nil on success, the last error on exhaustion. +// leaveVoiceChannelWithRetry attempts to remove the voice state from the DB. +// The first attempt is synchronous. If it fails, subsequent retries run in a +// background goroutine with exponential backoff so the caller (readPump) is +// not blocked by time.Sleep. +// Returns nil on first-attempt success, the first error otherwise (retries +// continue in the background). func leaveVoiceChannelWithRetry(h *Hub, userID int64, channelID int64) error { - const maxRetries = 3 - delay := 100 * time.Millisecond + // Synchronous first attempt. + if err := h.db.LeaveVoiceChannel(userID); err != nil { + slog.Warn("LeaveVoiceChannel failed, retrying in background", + "err", err, "user_id", userID, "channel_id", channelID, + "attempt", 1, "max_retries", 3) - for attempt := 1; attempt <= maxRetries; attempt++ { - if err := h.db.LeaveVoiceChannel(userID); err != nil { - slog.Warn("LeaveVoiceChannel failed, retrying", - "err", err, "user_id", userID, "channel_id", channelID, - "attempt", attempt, "max_retries", maxRetries) - if attempt < maxRetries { + // Background retries so the readPump goroutine is not blocked. + go func() { + const maxRetries = 3 + delay := 200 * time.Millisecond + + for attempt := 2; attempt <= maxRetries; attempt++ { time.Sleep(delay) delay *= 2 - } else { - slog.Error("LeaveVoiceChannel exhausted retries — ghost state may persist", - "err", err, "user_id", userID, "channel_id", channelID) - return err + + if retryErr := h.db.LeaveVoiceChannel(userID); retryErr != nil { + slog.Warn("LeaveVoiceChannel retry failed", + "err", retryErr, "user_id", userID, "channel_id", channelID, + "attempt", attempt, "max_retries", maxRetries) + if attempt == maxRetries { + slog.Error("LeaveVoiceChannel exhausted retries — ghost state may persist", + "err", retryErr, "user_id", userID, "channel_id", channelID) + } + } else { + slog.Info("LeaveVoiceChannel succeeded on retry", + "user_id", userID, "attempt", attempt) + return + } } - } else { - if attempt > 1 { - slog.Info("LeaveVoiceChannel succeeded on retry", - "user_id", userID, "attempt", attempt) - } - return nil - } + }() + + return err } return nil } diff --git a/Server/ws/ws_integration_test.go b/Server/ws/ws_integration_test.go index 6a44d502..9c1eb8b6 100644 --- a/Server/ws/ws_integration_test.go +++ b/Server/ws/ws_integration_test.go @@ -343,6 +343,256 @@ func TestServeWS_ValidAuth_FullHandshake(t *testing.T) { } } +// TestServeWS_ImmediateDisconnect_DoesNotLeaveGhostClient verifies that a +// client that drops immediately after sending auth does not remain registered +// or stuck online. +func TestServeWS_ImmediateDisconnect_DoesNotLeaveGhostClient(t *testing.T) { + database := openServeTestDB(t) + limiter := auth.NewRateLimiter() + hub := ws.NewHub(database, limiter) + go hub.Run() + defer hub.Stop() + + userID, err := database.CreateUser("abruptclose", "hash", 4) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + token := "abrupt-close-token" + tokenHash := auth.HashToken(token) + if _, err := database.CreateSession(userID, tokenHash, "test", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + handler := ws.ServeWS(hub, database, []string{"*"}) + srv := httptest.NewServer(http.HandlerFunc(handler)) + defer srv.Close() + + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + conn, _, err := websocket.Dial(ctx, wsURL, nil) + if err != nil { + t.Fatalf("websocket.Dial: %v", err) + } + defer conn.CloseNow() + + authMsg := map[string]any{ + "type": "auth", + "payload": map[string]string{"token": token}, + } + raw, err := json.Marshal(authMsg) + if err != nil { + t.Fatalf("marshal auth: %v", err) + } + if err := conn.Write(ctx, websocket.MessageText, raw); err != nil { + t.Fatalf("write auth: %v", err) + } + conn.CloseNow() + + deadline := time.Now().Add(2 * time.Second) + cleanedUp := false + for time.Now().Before(deadline) { + user, getErr := database.GetUserByID(userID) + if getErr != nil { + t.Fatalf("GetUserByID: %v", getErr) + } + if hub.ClientCount() == 0 && user.Status == "offline" { + cleanedUp = true + break + } + time.Sleep(20 * time.Millisecond) + } + + if !cleanedUp { + user, getErr := database.GetUserByID(userID) + if getErr != nil { + t.Fatalf("GetUserByID final: %v", getErr) + } + t.Fatalf("immediate disconnect left stale state: client_count=%d user_status=%q", hub.ClientCount(), user.Status) + } +} + +// TestServeWS_DuplicateLogin_KeepsUserOnline verifies that replacing an +// existing connection does not broadcast or persist a false offline state for +// the still-connected replacement session. +func TestServeWS_DuplicateLogin_KeepsUserOnline(t *testing.T) { + database := openServeTestDB(t) + limiter := auth.NewRateLimiter() + hub := ws.NewHub(database, limiter) + go hub.Run() + defer hub.Stop() + + userID, err := database.CreateUser("ws-reconnect-user", "hash", 1) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + token, err := auth.GenerateToken() + if err != nil { + t.Fatalf("GenerateToken: %v", err) + } + tokenHash := auth.HashToken(token) + if _, err := database.CreateSession(userID, tokenHash, "test", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + handler := ws.ServeWS(hub, database, []string{"*"}) + srv := httptest.NewServer(http.HandlerFunc(handler)) + defer srv.Close() + + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + dialAndAuth := func() *websocket.Conn { + conn, _, dialErr := websocket.Dial(ctx, wsURL, nil) + if dialErr != nil { + t.Fatalf("websocket.Dial: %v", dialErr) + } + authMsg := map[string]any{ + "type": "auth", + "payload": map[string]string{"token": token}, + } + raw, marshalErr := json.Marshal(authMsg) + if marshalErr != nil { + t.Fatalf("marshal auth: %v", marshalErr) + } + if writeErr := conn.Write(ctx, websocket.MessageText, raw); writeErr != nil { + t.Fatalf("write auth: %v", writeErr) + } + for i := 0; i < 2; i++ { + if _, _, readErr := conn.Read(ctx); readErr != nil { + t.Fatalf("read handshake message %d: %v", i, readErr) + } + } + return conn + } + + conn1 := dialAndAuth() + defer func() { _ = conn1.Close(websocket.StatusNormalClosure, "") }() + conn2 := dialAndAuth() + defer func() { _ = conn2.Close(websocket.StatusNormalClosure, "") }() + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + user, getErr := database.GetUserByID(userID) + if getErr != nil { + t.Fatalf("GetUserByID: %v", getErr) + } + if hub.ClientCount() == 1 && user.Status == "online" { + return + } + time.Sleep(20 * time.Millisecond) + } + + user, getErr := database.GetUserByID(userID) + if getErr != nil { + t.Fatalf("GetUserByID final: %v", getErr) + } + t.Fatalf("duplicate login left wrong state: client_count=%d user_status=%q", hub.ClientCount(), user.Status) +} + +// TestServeWS_DuplicateLogin_DoesNotBroadcastVoiceLeave verifies that +// replacing a connection does not emit a spurious voice_leave for the same +// still-connected user. +func TestServeWS_DuplicateLogin_DoesNotBroadcastVoiceLeave(t *testing.T) { + database := openServeTestDB(t) + limiter := auth.NewRateLimiter() + hub := ws.NewHub(database, limiter) + go hub.Run() + defer hub.Stop() + + userID, err := database.CreateUser("ws-voice-reconnect", "hash", 1) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + token, err := auth.GenerateToken() + if err != nil { + t.Fatalf("GenerateToken: %v", err) + } + tokenHash := auth.HashToken(token) + if _, err := database.CreateSession(userID, tokenHash, "test", "127.0.0.1"); err != nil { + t.Fatalf("CreateSession: %v", err) + } + + handler := ws.ServeWS(hub, database, []string{"*"}) + srv := httptest.NewServer(http.HandlerFunc(handler)) + defer srv.Close() + + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + dialAndAuth := func() *websocket.Conn { + conn, _, dialErr := websocket.Dial(ctx, wsURL, nil) + if dialErr != nil { + t.Fatalf("websocket.Dial: %v", dialErr) + } + authMsg := map[string]any{ + "type": "auth", + "payload": map[string]string{"token": token}, + } + raw, marshalErr := json.Marshal(authMsg) + if marshalErr != nil { + t.Fatalf("marshal auth: %v", marshalErr) + } + if writeErr := conn.Write(ctx, websocket.MessageText, raw); writeErr != nil { + t.Fatalf("write auth: %v", writeErr) + } + for i := 0; i < 2; i++ { + if _, _, readErr := conn.Read(ctx); readErr != nil { + t.Fatalf("read handshake message %d: %v", i, readErr) + } + } + return conn + } + + conn1 := dialAndAuth() + defer func() { _ = conn1.Close(websocket.StatusNormalClosure, "") }() + + var originalClient *ws.Client + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + originalClient = hub.GetClient(userID) + if originalClient != nil { + break + } + time.Sleep(20 * time.Millisecond) + } + if originalClient == nil { + t.Fatal("expected first client to be registered") + } + ws.SetClientVoiceChID(originalClient, 99) + + conn2 := dialAndAuth() + defer func() { _ = conn2.Close(websocket.StatusNormalClosure, "") }() + + replacementClient := hub.GetClient(userID) + if replacementClient == nil { + t.Fatal("expected replacement client to be registered") + } + if got := ws.GetClientVoiceChIDForTest(replacementClient); got != 99 { + t.Fatalf("replacement client voiceChID = %d, want 99", got) + } + + readDeadline := time.Now().Add(400 * time.Millisecond) + for time.Now().Before(readDeadline) { + readCtx, readCancel := context.WithTimeout(ctx, 100*time.Millisecond) + _, raw, readErr := conn2.Read(readCtx) + readCancel() + if readErr != nil { + break + } + var msg map[string]any + if err := json.Unmarshal(raw, &msg); err != nil { + continue + } + if msg["type"] == "voice_leave" { + t.Fatalf("duplicate login should not broadcast voice_leave for replacement connection: %s", string(raw)) + } + } +} + // TestServeWS_writePump_MessageDelivered verifies that messages queued on the // hub are written through writePump to the connected client. func TestServeWS_writePump_MessageDelivered(t *testing.T) { @@ -728,4 +978,3 @@ func TestServeWS_BannedUser_ReceivesError(t *testing.T) { } } } -