mirror of
https://github.com/J3vb/OwnCord.git
synced 2026-09-03 03:50:00 +03:00
test: add WS coverage tests + refactor handlers, update docs
Add hub, livekit, export, and coverage boost tests for Server/ws. Refactor handlers_chat.go and serve.go for testability. Sync docs: fix backup endpoint path, add DELETE /auth/account, add audit logging + account deletion to security.md.
This commit is contained in:
@@ -8,6 +8,9 @@ CLAUDE.md
|
||||
.github/copilot-instructions.md
|
||||
.github/instructions/
|
||||
|
||||
# Agent worktrees (Mission Control fleet isolation)
|
||||
.worktrees/
|
||||
|
||||
# AI-specific / internal planning docs
|
||||
docs/brain/
|
||||
docs/CODEMAPS/
|
||||
|
||||
@@ -2220,3 +2220,477 @@ func TestHandleVoiceJoin_InvalidQualityFallsBackToMedium(t *testing.T) {
|
||||
}
|
||||
t.Error("expected voice_config with medium quality fallback")
|
||||
}
|
||||
|
||||
// ─── getLastActivity (client.go:153) ─────────────────────────────────────────
|
||||
|
||||
func TestGetLastActivity_ReturnsZeroForNewTestClient(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
got := ws.GetLastActivityForTest(c)
|
||||
if !got.IsZero() {
|
||||
t.Fatalf("expected zero time for new test client, got %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetLastActivity_UpdatedByTouch(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
before := time.Now()
|
||||
ws.TouchForTest(c)
|
||||
after := time.Now()
|
||||
|
||||
got := ws.GetLastActivityForTest(c)
|
||||
if got.Before(before) || got.After(after) {
|
||||
t.Fatalf("lastActivity = %v, expected between %v and %v", got, before, after)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetLastActivity_MultipleTouch(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
ws.TouchForTest(c)
|
||||
first := ws.GetLastActivityForTest(c)
|
||||
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
ws.TouchForTest(c)
|
||||
second := ws.GetLastActivityForTest(c)
|
||||
|
||||
if !second.After(first) {
|
||||
t.Fatalf("second touch (%v) should be after first (%v)", second, first)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── setVoiceChID (client.go:186) ───────────────────────────────────────────
|
||||
|
||||
func TestSetVoiceChID_SetsAndGetsValue(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 77)
|
||||
if got := ws.GetClientVoiceChIDForTest(c); got != 77 {
|
||||
t.Fatalf("voiceChID = %d, want 77", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetVoiceChID_OverwritesPreviousValue(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 10)
|
||||
ws.SetVoiceChIDForTest(c, 20)
|
||||
if got := ws.GetClientVoiceChIDForTest(c); got != 20 {
|
||||
t.Fatalf("voiceChID = %d, want 20", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetVoiceChID_ZeroMeansNotInVoice(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 50)
|
||||
ws.SetVoiceChIDForTest(c, 0)
|
||||
if got := ws.GetClientVoiceChIDForTest(c); got != 0 {
|
||||
t.Fatalf("voiceChID = %d, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── clearVoiceChID (client.go:203) ─────────────────────────────────────────
|
||||
|
||||
func TestClearVoiceChID_ReturnsOldValueAndClearsToZero(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 42)
|
||||
old := ws.ClearVoiceChIDForTest(c)
|
||||
if old != 42 {
|
||||
t.Fatalf("clearVoiceChID returned %d, want 42", old)
|
||||
}
|
||||
if got := ws.GetClientVoiceChIDForTest(c); got != 0 {
|
||||
t.Fatalf("voiceChID after clear = %d, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClearVoiceChID_ReturnsZeroWhenNotInVoice(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
old := ws.ClearVoiceChIDForTest(c)
|
||||
if old != 0 {
|
||||
t.Fatalf("clearVoiceChID returned %d, want 0", old)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClearVoiceChID_DoubleClearReturnsZero(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 99)
|
||||
first := ws.ClearVoiceChIDForTest(c)
|
||||
second := ws.ClearVoiceChIDForTest(c)
|
||||
if first != 99 {
|
||||
t.Fatalf("first clear = %d, want 99", first)
|
||||
}
|
||||
if second != 0 {
|
||||
t.Fatalf("second clear = %d, want 0", second)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── handleVoiceTokenRefresh (voice_join.go:188) ────────────────────────────
|
||||
|
||||
func voiceTokenRefreshMsg() []byte {
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_token_refresh",
|
||||
"payload": map[string]any{},
|
||||
})
|
||||
return raw
|
||||
}
|
||||
|
||||
func TestHandleVoiceTokenRefresh_NotInVoice(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "vtr-notinvoice")
|
||||
send := make(chan []byte, 16)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
hub.HandleMessageForTest(c, voiceTokenRefreshMsg())
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
code := drainForErrorCode(send, 200*time.Millisecond)
|
||||
if code != "BAD_REQUEST" {
|
||||
t.Errorf("error code = %q, want BAD_REQUEST", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleVoiceTokenRefresh_InVoice_ReturnsToken(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "vtr-invc")
|
||||
vcID := seedVoiceChannel(t, database, "vtr-vc")
|
||||
send := make(chan []byte, 64)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
raw, _ := json.Marshal(map[string]any{
|
||||
"type": "voice_join",
|
||||
"payload": map[string]any{"channel_id": vcID},
|
||||
})
|
||||
hub.HandleMessageForTest(c, raw)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
drainChanBuf(send)
|
||||
|
||||
hub.HandleMessageForTest(c, voiceTokenRefreshMsg())
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
msgs := drainChanTimeout(send, 300*time.Millisecond)
|
||||
foundToken := false
|
||||
for _, msg := range msgs {
|
||||
var env map[string]any
|
||||
if json.Unmarshal(msg, &env) == nil && env["type"] == "voice_token" {
|
||||
foundToken = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !foundToken {
|
||||
t.Error("expected voice_token message after token refresh")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleVoiceTokenRefresh_NilUser(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
send := make(chan []byte, 16)
|
||||
c := ws.NewTestClient(hub, 1, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 42)
|
||||
|
||||
hub.HandleMessageForTest(c, voiceTokenRefreshMsg())
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
code := drainForErrorCode(send, 200*time.Millisecond)
|
||||
if code != "INTERNAL" {
|
||||
t.Errorf("error code = %q, want INTERNAL", code)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── rollbackVoiceJoin (voice_join.go:239) ──────────────────────────────────
|
||||
|
||||
func TestRollbackVoiceJoin_ClearsVoiceStateAndBroadcasts(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "rb-user")
|
||||
vcID := seedVoiceChannel(t, database, "rb-vc")
|
||||
send := make(chan []byte, 64)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
if err := database.JoinVoiceChannel(user.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel: %v", err)
|
||||
}
|
||||
ws.SetVoiceChIDForTest(c, vcID)
|
||||
|
||||
hub.RollbackVoiceJoinForTest(c, vcID)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
if got := ws.GetClientVoiceChIDForTest(c); got != 0 {
|
||||
t.Fatalf("voiceChID after rollback = %d, want 0", got)
|
||||
}
|
||||
|
||||
state, _ := database.GetVoiceState(user.ID)
|
||||
if state != nil {
|
||||
t.Fatal("voice state should be nil after rollback")
|
||||
}
|
||||
|
||||
msgs := drainChanTimeout(send, 300*time.Millisecond)
|
||||
foundLeave := false
|
||||
for _, msg := range msgs {
|
||||
var env map[string]any
|
||||
if json.Unmarshal(msg, &env) == nil && env["type"] == "voice_leave" {
|
||||
foundLeave = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !foundLeave {
|
||||
t.Error("expected voice_leave broadcast after rollback")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRollbackVoiceJoin_NoDBState_DoesNotPanic(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "rb-nostate")
|
||||
send := make(chan []byte, 16)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
ws.SetVoiceChIDForTest(c, 999)
|
||||
hub.RollbackVoiceJoinForTest(c, 999)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
if got := ws.GetClientVoiceChIDForTest(c); got != 0 {
|
||||
t.Fatalf("voiceChID after rollback = %d, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── leaveVoiceChannelWithRetry (voice_leave.go:57) ─────────────────────────
|
||||
|
||||
func TestLeaveVoiceChannelWithRetry_SuccessOnFirstAttempt(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "lvcr-ok")
|
||||
vcID := seedVoiceChannel(t, database, "lvcr-ok-vc")
|
||||
|
||||
if err := database.JoinVoiceChannel(user.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel: %v", err)
|
||||
}
|
||||
|
||||
state, _ := database.GetVoiceState(user.ID)
|
||||
if state == nil {
|
||||
t.Fatal("voice state should exist before leave")
|
||||
}
|
||||
|
||||
err := ws.LeaveVoiceChannelWithRetryForTest(hub, user.ID, vcID, state.JoinedAt)
|
||||
if err != nil {
|
||||
t.Fatalf("leaveVoiceChannelWithRetry returned error: %v", err)
|
||||
}
|
||||
|
||||
state, _ = database.GetVoiceState(user.ID)
|
||||
if state != nil {
|
||||
t.Fatal("voice state should be nil after successful leave")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLeaveVoiceChannelWithRetry_NoVoiceState_NilReturn(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
_ = seedCoverageOwner(t, database, "lvcr-nostate")
|
||||
|
||||
err := ws.LeaveVoiceChannelWithRetryForTest(hub, 9999, 1, "")
|
||||
if err != nil {
|
||||
t.Fatalf("expected nil error for non-existent voice state, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ─── CleanupVoiceForChannel (hub.go:237) — additional paths ─────────────────
|
||||
|
||||
func TestCleanupVoiceForChannel_WithClientsInChannel(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user1 := seedCoverageOwner(t, database, "cvfc-u1")
|
||||
user2 := seedCoverageOwner(t, database, "cvfc-u2")
|
||||
vcID := seedVoiceChannel(t, database, "cvfc-vc")
|
||||
|
||||
send1 := make(chan []byte, 64)
|
||||
send2 := make(chan []byte, 64)
|
||||
c1 := ws.NewTestClientWithUser(hub, user1, 0, send1)
|
||||
c2 := ws.NewTestClientWithUser(hub, user2, 0, send2)
|
||||
hub.Register(c1)
|
||||
hub.Register(c2)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
if err := database.JoinVoiceChannel(user1.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel u1: %v", err)
|
||||
}
|
||||
if err := database.JoinVoiceChannel(user2.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel u2: %v", err)
|
||||
}
|
||||
ws.SetVoiceChIDForTest(c1, vcID)
|
||||
ws.SetVoiceChIDForTest(c2, vcID)
|
||||
|
||||
hub.CleanupVoiceForChannel(vcID)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
if got := ws.GetClientVoiceChIDForTest(c1); got != 0 {
|
||||
t.Errorf("c1 voiceChID = %d, want 0", got)
|
||||
}
|
||||
if got := ws.GetClientVoiceChIDForTest(c2); got != 0 {
|
||||
t.Errorf("c2 voiceChID = %d, want 0", got)
|
||||
}
|
||||
|
||||
states, _ := database.GetChannelVoiceStates(vcID)
|
||||
if len(states) != 0 {
|
||||
t.Errorf("expected 0 voice states after cleanup, got %d", len(states))
|
||||
}
|
||||
}
|
||||
|
||||
func TestCleanupVoiceForChannel_EmptyChannel(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
vcID := seedVoiceChannel(t, database, "cvfc-empty-vc")
|
||||
hub.CleanupVoiceForChannel(vcID)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
|
||||
func TestCleanupVoiceForChannel_DBStateButNoClient(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "cvfc-noclient")
|
||||
vcID := seedVoiceChannel(t, database, "cvfc-noclient-vc")
|
||||
|
||||
if err := database.JoinVoiceChannel(user.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel: %v", err)
|
||||
}
|
||||
|
||||
hub.CleanupVoiceForChannel(vcID)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
state, _ := database.GetVoiceState(user.ID)
|
||||
if state != nil {
|
||||
t.Error("voice state should be nil after cleanup")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── sweepStaleVoiceStates (hub.go:489) ─────────────────────────────────────
|
||||
|
||||
func TestSweepStaleVoiceStates_RemovesGhostState(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "sweep-ghost")
|
||||
vcID := seedVoiceChannel(t, database, "sweep-ghost-vc")
|
||||
|
||||
// Put user in voice in DB but don't register a client — ghost state.
|
||||
if err := database.JoinVoiceChannel(user.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel: %v", err)
|
||||
}
|
||||
|
||||
// Verify it exists.
|
||||
state, _ := database.GetVoiceState(user.ID)
|
||||
if state == nil {
|
||||
t.Fatal("voice state should exist before sweep")
|
||||
}
|
||||
|
||||
hub.SweepStaleVoiceStatesForTest()
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Ghost state should be removed.
|
||||
state, _ = database.GetVoiceState(user.ID)
|
||||
if state != nil {
|
||||
t.Error("ghost voice state should be nil after sweep")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSweepStaleVoiceStates_PreservesActiveClientState(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "sweep-active")
|
||||
vcID := seedVoiceChannel(t, database, "sweep-active-vc")
|
||||
|
||||
// Register client and set voice channel.
|
||||
send := make(chan []byte, 64)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
if err := database.JoinVoiceChannel(user.ID, vcID); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel: %v", err)
|
||||
}
|
||||
ws.SetVoiceChIDForTest(c, vcID)
|
||||
|
||||
hub.SweepStaleVoiceStatesForTest()
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Active client's state should be preserved.
|
||||
state, _ := database.GetVoiceState(user.ID)
|
||||
if state == nil {
|
||||
t.Error("active client's voice state should be preserved after sweep")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSweepStaleVoiceStates_NoStatesNoPanic(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
hub.SweepStaleVoiceStatesForTest()
|
||||
}
|
||||
|
||||
func TestSweepStaleVoiceStates_MismatchedChannelIsGhost(t *testing.T) {
|
||||
hub, database := newCoverageHub(t)
|
||||
user := seedCoverageOwner(t, database, "sweep-mismatch")
|
||||
vc1 := seedVoiceChannel(t, database, "sweep-mismatch-vc1")
|
||||
vc2 := seedVoiceChannel(t, database, "sweep-mismatch-vc2")
|
||||
|
||||
// Register client in vc1 but DB says vc2.
|
||||
send := make(chan []byte, 64)
|
||||
c := ws.NewTestClientWithUser(hub, user, 0, send)
|
||||
hub.Register(c)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
if err := database.JoinVoiceChannel(user.ID, vc2); err != nil {
|
||||
t.Fatalf("JoinVoiceChannel: %v", err)
|
||||
}
|
||||
ws.SetVoiceChIDForTest(c, vc1) // Client thinks vc1, DB says vc2 — mismatch.
|
||||
|
||||
hub.SweepStaleVoiceStatesForTest()
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Mismatched state should be removed from DB.
|
||||
state, _ := database.GetVoiceState(user.ID)
|
||||
if state != nil {
|
||||
t.Error("mismatched voice state should be removed after sweep")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── BroadcastToChannel / BroadcastToAll full-channel path ──────────────────
|
||||
|
||||
func TestBroadcastToChannel_DropsWhenFull(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
// Don't start Run() — broadcast channel will fill up.
|
||||
// The broadcast channel capacity is 256.
|
||||
for i := 0; i < 260; i++ {
|
||||
hub.BroadcastToChannel(1, []byte(`{"type":"test"}`))
|
||||
}
|
||||
// No panic and no block = pass. Some messages will be dropped.
|
||||
}
|
||||
|
||||
func TestBroadcastToAll_DropsWhenFull(t *testing.T) {
|
||||
hub, _ := newCoverageHub(t)
|
||||
for i := 0; i < 260; i++ {
|
||||
hub.BroadcastToAll([]byte(`{"type":"test"}`))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -5,12 +5,92 @@ package ws
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"time"
|
||||
|
||||
"github.com/livekit/protocol/livekit"
|
||||
"github.com/owncord/server/db"
|
||||
)
|
||||
|
||||
// ─── hub sweep helpers ─────────────────────────────────────────────────────
|
||||
|
||||
// SweepStaleClientsForTest exposes sweepStaleClients for external tests.
|
||||
func (h *Hub) SweepStaleClientsForTest() {
|
||||
h.sweepStaleClients()
|
||||
}
|
||||
|
||||
// SweepStaleVoiceStatesForTest exposes sweepStaleVoiceStates for external tests.
|
||||
func (h *Hub) SweepStaleVoiceStatesForTest() {
|
||||
h.sweepStaleVoiceStates()
|
||||
}
|
||||
|
||||
// SetClientLastActivityForTest overwrites a client's lastActivity timestamp.
|
||||
func SetClientLastActivityForTest(c *Client, t time.Time) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.lastActivity = t
|
||||
}
|
||||
|
||||
// ─── client getter/setter helpers ──────────────────────────────────────────
|
||||
|
||||
// GetLastActivityForTest exposes Client.getLastActivity for external tests.
|
||||
func GetLastActivityForTest(c *Client) time.Time {
|
||||
return c.getLastActivity()
|
||||
}
|
||||
|
||||
// ClearVoiceChIDForTest exposes Client.clearVoiceChID for external tests.
|
||||
func ClearVoiceChIDForTest(c *Client) int64 {
|
||||
return c.clearVoiceChID()
|
||||
}
|
||||
|
||||
// SetVoiceChIDForTest exposes Client.setVoiceChID for external tests.
|
||||
func SetVoiceChIDForTest(c *Client, chID int64) {
|
||||
c.setVoiceChID(chID)
|
||||
}
|
||||
|
||||
// TouchForTest exposes Client.touch for external tests.
|
||||
func TouchForTest(c *Client) {
|
||||
c.touch()
|
||||
}
|
||||
|
||||
// RollbackVoiceJoinForTest exposes Hub.rollbackVoiceJoin for external tests.
|
||||
func (h *Hub) RollbackVoiceJoinForTest(c *Client, channelID int64) {
|
||||
h.rollbackVoiceJoin(c, channelID)
|
||||
}
|
||||
|
||||
// LeaveVoiceChannelWithRetryForTest exposes leaveVoiceChannelWithRetry for external tests.
|
||||
func LeaveVoiceChannelWithRetryForTest(h *Hub, userID int64, channelID int64, joinToken string) error {
|
||||
return leaveVoiceChannelWithRetry(h, userID, channelID, joinToken)
|
||||
}
|
||||
|
||||
// ─── livekit process/webhook helpers ───────────────────────────────────────
|
||||
|
||||
// GenerateConfigForTest exposes LiveKitProcess.generateConfig for external tests.
|
||||
func (p *LiveKitProcess) GenerateConfigForTest() (string, error) {
|
||||
return p.generateConfig()
|
||||
}
|
||||
|
||||
// SetProcessCmdForTest sets cmd to a non-nil value to simulate "already running".
|
||||
func (p *LiveKitProcess) SetProcessCmdForTest() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.cmd = &exec.Cmd{}
|
||||
}
|
||||
|
||||
// SetProcessStoppedForTest sets stopped=true to simulate a stopped process.
|
||||
func (p *LiveKitProcess) SetProcessStoppedForTest() {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.stopped = true
|
||||
}
|
||||
|
||||
// NewHubForTest creates a minimal Hub with no DB or limiter for webhook testing.
|
||||
func NewHubForTest() *Hub {
|
||||
return &Hub{
|
||||
clients: make(map[int64]*Client),
|
||||
}
|
||||
}
|
||||
|
||||
// BuildAuthOKForTest exposes Hub.buildAuthOK for external tests.
|
||||
func (h *Hub) BuildAuthOKForTest(user *db.User, roleName string) []byte {
|
||||
return h.buildAuthOK(user, roleName)
|
||||
|
||||
+138
-113
@@ -7,6 +7,7 @@ import (
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/owncord/server/db"
|
||||
"github.com/owncord/server/permissions"
|
||||
)
|
||||
|
||||
@@ -23,21 +24,22 @@ func registerChatHandlers(r *HandlerRegistry) {
|
||||
})
|
||||
}
|
||||
|
||||
type chatSendPayload struct {
|
||||
ChannelID json.Number `json:"channel_id"`
|
||||
Content string `json:"content"`
|
||||
ReplyTo *int64 `json:"reply_to"`
|
||||
Attachments []string `json:"attachments"`
|
||||
}
|
||||
|
||||
// handleChatSend processes a chat_send message.
|
||||
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) {
|
||||
c.sendMsg(buildRateLimitError("too many messages", chatWindow.Seconds()))
|
||||
return
|
||||
}
|
||||
|
||||
var p struct {
|
||||
ChannelID json.Number `json:"channel_id"`
|
||||
Content string `json:"content"`
|
||||
ReplyTo *int64 `json:"reply_to"`
|
||||
Attachments []string `json:"attachments"`
|
||||
}
|
||||
var p chatSendPayload
|
||||
if err := json.Unmarshal(payload, &p); err != nil {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "invalid chat_send payload"))
|
||||
return
|
||||
@@ -48,102 +50,36 @@ func (h *Hub) handleChatSend(ctx context.Context, c *Client, reqID string, paylo
|
||||
return
|
||||
}
|
||||
|
||||
// Check channel exists.
|
||||
ch, err := h.db.GetChannel(channelID)
|
||||
if err != nil || ch == nil {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeNotFound, "channel not found"))
|
||||
return
|
||||
}
|
||||
|
||||
// DM channels use participant-based auth instead of role permissions.
|
||||
isDM := ch.Type == "dm"
|
||||
if isDM {
|
||||
ok, dmErr := h.db.IsDMParticipant(c.userID, channelID)
|
||||
if dmErr != nil {
|
||||
slog.Error("ws handleChatSend IsDMParticipant", "err", dmErr)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to check DM participation"))
|
||||
return
|
||||
}
|
||||
if !ok {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeForbidden, "you are not a participant in this DM"))
|
||||
return
|
||||
}
|
||||
} else {
|
||||
// Permission check for non-DM channels.
|
||||
if !h.requireChannelPerm(c, channelID, permissions.ReadMessages|permissions.SendMessages, "SEND_MESSAGES") {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Slow mode enforcement: moderators with MANAGE_MESSAGES bypass it.
|
||||
// DM channels do not have slow mode.
|
||||
if !isDM && ch.SlowMode > 0 && !h.hasChannelPerm(c, channelID, permissions.ManageMessages) {
|
||||
slowKey := fmt.Sprintf("slow:%d:%d", c.userID, channelID)
|
||||
if !h.limiter.Allow(slowKey, 1, time.Duration(ch.SlowMode)*time.Second) {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeSlowMode, fmt.Sprintf("channel has %ds slow mode", ch.SlowMode)))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Sanitize and validate content length.
|
||||
content := sanitizer.Sanitize(p.Content)
|
||||
if content == "" && len(p.Attachments) == 0 {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content cannot be empty"))
|
||||
if !h.checkChatSendPermission(c, channelID, isDM) {
|
||||
return
|
||||
}
|
||||
if len([]rune(content)) > maxMessageLen {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content exceeds maximum length of 4000 characters"))
|
||||
if !h.checkSlowMode(c, ch, channelID, isDM) {
|
||||
return
|
||||
}
|
||||
|
||||
content, ok := h.validateChatContent(c, p.Content, p.Attachments)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
// Check attachment permission before persisting anything.
|
||||
// DM channels use participant-based auth (already checked above), not role permissions.
|
||||
if !isDM && len(p.Attachments) > 0 {
|
||||
if !h.requireChannelPerm(c, channelID, permissions.AttachFiles, "ATTACH_FILES") {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Persist message.
|
||||
msgID, err := h.db.CreateMessage(channelID, c.userID, content, p.ReplyTo)
|
||||
if err != nil {
|
||||
slog.Error("ws handleChatSend CreateMessage", "err", err)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to save message"))
|
||||
msgID, attachments, ok := h.persistChatMessage(c, channelID, content, p.ReplyTo, p.Attachments)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
// Link attachments if provided.
|
||||
var attachments []map[string]any
|
||||
if len(p.Attachments) > 0 {
|
||||
linked, linkErr := h.db.LinkAttachmentsToMessage(msgID, p.Attachments)
|
||||
if linkErr != nil {
|
||||
slog.Error("ws handleChatSend LinkAttachments", "err", linkErr, "msg_id", msgID)
|
||||
// Delete the orphaned message so it doesn't persist without its attachments.
|
||||
if delErr := h.db.DeleteMessage(msgID, c.userID, true); delErr != nil {
|
||||
slog.Error("ws handleChatSend DeleteMessage (cleanup)", "err", delErr, "msg_id", msgID)
|
||||
}
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to send message with attachments"))
|
||||
return
|
||||
}
|
||||
if linked > 0 {
|
||||
attMap, attErr := h.db.GetAttachmentsByMessageIDs([]int64{msgID})
|
||||
if attErr != nil {
|
||||
slog.Error("ws handleChatSend GetAttachments", "err", attErr)
|
||||
} else {
|
||||
for _, ai := range attMap[msgID] {
|
||||
attachments = append(attachments, map[string]any{
|
||||
"id": ai.ID,
|
||||
"filename": ai.Filename,
|
||||
"size": ai.Size,
|
||||
"mime": ai.Mime,
|
||||
"url": ai.URL,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Retrieve to get timestamp.
|
||||
msg, err := h.db.GetMessage(msgID)
|
||||
if err != nil || msg == nil {
|
||||
slog.Error("ws handleChatSend GetMessage after create", "err", err)
|
||||
@@ -159,47 +95,136 @@ func (h *Hub) handleChatSend(ctx context.Context, c *Client, reqID string, paylo
|
||||
}
|
||||
|
||||
slog.Debug("message sent", "user", username, "channel_id", channelID, "msg_id", msgID)
|
||||
|
||||
// Ack sender.
|
||||
c.sendMsg(buildChatSendOK(reqID, msgID, msg.Timestamp))
|
||||
|
||||
// Broadcast message.
|
||||
broadcast := buildChatMessage(msgID, channelID, c.userID, username, avatar, c.roleName, content, msg.Timestamp, p.ReplyTo, attachments)
|
||||
h.broadcastChatMessage(c, channelID, isDM, broadcast)
|
||||
}
|
||||
|
||||
func (h *Hub) checkChatSendPermission(c *Client, channelID int64, isDM bool) bool {
|
||||
if isDM {
|
||||
// DM: send directly to both participants instead of channel broadcast.
|
||||
participantIDs, pErr := h.db.GetDMParticipantIDs(channelID)
|
||||
if pErr != nil {
|
||||
slog.Error("ws handleChatSend GetDMParticipantIDs", "err", pErr, "channel_id", channelID)
|
||||
// Message is already persisted but we cannot deliver it. Inform the
|
||||
// sender so the failure is not silent.
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "message saved but delivery failed — please retry"))
|
||||
return
|
||||
ok, dmErr := h.db.IsDMParticipant(c.userID, channelID)
|
||||
if dmErr != nil {
|
||||
slog.Error("ws handleChatSend IsDMParticipant", "err", dmErr)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to check DM participation"))
|
||||
return false
|
||||
}
|
||||
if !ok {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeForbidden, "you are not a participant in this DM"))
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
return h.requireChannelPerm(c, channelID, permissions.ReadMessages|permissions.SendMessages, "SEND_MESSAGES")
|
||||
}
|
||||
|
||||
// Deliver chat_message to all DM participants.
|
||||
for _, pid := range participantIDs {
|
||||
h.SendToUser(pid, broadcast)
|
||||
}
|
||||
func (h *Hub) checkSlowMode(c *Client, ch *db.Channel, channelID int64, isDM bool) bool {
|
||||
if isDM || ch.SlowMode <= 0 || h.hasChannelPerm(c, channelID, permissions.ManageMessages) {
|
||||
return true
|
||||
}
|
||||
slowKey := fmt.Sprintf("slow:%d:%d", c.userID, channelID)
|
||||
if !h.limiter.Allow(slowKey, 1, time.Duration(ch.SlowMode)*time.Second) {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeSlowMode, fmt.Sprintf("channel has %ds slow mode", ch.SlowMode)))
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Auto-reopen the DM for the recipient if it was closed.
|
||||
for _, pid := range participantIDs {
|
||||
if pid == c.userID {
|
||||
continue
|
||||
}
|
||||
if openErr := h.db.OpenDM(pid, channelID); openErr != nil {
|
||||
slog.Error("ws handleChatSend OpenDM", "err", openErr,
|
||||
"recipient_id", pid, "channel_id", channelID)
|
||||
continue
|
||||
}
|
||||
// Notify the recipient that the DM was (re)opened.
|
||||
// Build the event with the sender as the recipient's "other user".
|
||||
if c.user != nil {
|
||||
h.SendToUser(pid, buildDMChannelOpen(channelID, c.user))
|
||||
}
|
||||
func (h *Hub) validateChatContent(c *Client, raw string, attachments []string) (string, bool) {
|
||||
content := sanitizer.Sanitize(raw)
|
||||
if content == "" && len(attachments) == 0 {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content cannot be empty"))
|
||||
return "", false
|
||||
}
|
||||
if len([]rune(content)) > maxMessageLen {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeBadRequest, "message content exceeds maximum length of 4000 characters"))
|
||||
return "", false
|
||||
}
|
||||
return content, true
|
||||
}
|
||||
|
||||
func (h *Hub) persistChatMessage(c *Client, channelID int64, content string, replyTo *int64, attIDs []string) (int64, []map[string]any, bool) {
|
||||
msgID, err := h.db.CreateMessage(channelID, c.userID, content, replyTo)
|
||||
if err != nil {
|
||||
slog.Error("ws handleChatSend CreateMessage", "err", err)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to save message"))
|
||||
return 0, nil, false
|
||||
}
|
||||
|
||||
attachments, ok := h.linkAttachments(c, msgID, attIDs)
|
||||
if !ok {
|
||||
return 0, nil, false
|
||||
}
|
||||
return msgID, attachments, true
|
||||
}
|
||||
|
||||
func (h *Hub) linkAttachments(c *Client, msgID int64, attIDs []string) ([]map[string]any, bool) {
|
||||
if len(attIDs) == 0 {
|
||||
return nil, true
|
||||
}
|
||||
|
||||
linked, linkErr := h.db.LinkAttachmentsToMessage(msgID, attIDs)
|
||||
if linkErr != nil {
|
||||
slog.Error("ws handleChatSend LinkAttachments", "err", linkErr, "msg_id", msgID)
|
||||
if delErr := h.db.DeleteMessage(msgID, c.userID, true); delErr != nil {
|
||||
slog.Error("ws handleChatSend DeleteMessage (cleanup)", "err", delErr, "msg_id", msgID)
|
||||
}
|
||||
} else {
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "failed to send message with attachments"))
|
||||
return nil, false
|
||||
}
|
||||
|
||||
if linked == 0 {
|
||||
return nil, true
|
||||
}
|
||||
|
||||
attMap, attErr := h.db.GetAttachmentsByMessageIDs([]int64{msgID})
|
||||
if attErr != nil {
|
||||
slog.Error("ws handleChatSend GetAttachments", "err", attErr)
|
||||
return nil, true
|
||||
}
|
||||
|
||||
var attachments []map[string]any
|
||||
for _, ai := range attMap[msgID] {
|
||||
attachments = append(attachments, map[string]any{
|
||||
"id": ai.ID,
|
||||
"filename": ai.Filename,
|
||||
"size": ai.Size,
|
||||
"mime": ai.Mime,
|
||||
"url": ai.URL,
|
||||
})
|
||||
}
|
||||
return attachments, true
|
||||
}
|
||||
|
||||
func (h *Hub) broadcastChatMessage(c *Client, channelID int64, isDM bool, broadcast []byte) {
|
||||
if !isDM {
|
||||
h.BroadcastToChannel(channelID, broadcast)
|
||||
return
|
||||
}
|
||||
|
||||
participantIDs, pErr := h.db.GetDMParticipantIDs(channelID)
|
||||
if pErr != nil {
|
||||
slog.Error("ws handleChatSend GetDMParticipantIDs", "err", pErr, "channel_id", channelID)
|
||||
c.sendMsg(buildErrorMsg(ErrCodeInternal, "message saved but delivery failed — please retry"))
|
||||
return
|
||||
}
|
||||
|
||||
for _, pid := range participantIDs {
|
||||
h.SendToUser(pid, broadcast)
|
||||
}
|
||||
|
||||
for _, pid := range participantIDs {
|
||||
if pid == c.userID {
|
||||
continue
|
||||
}
|
||||
if openErr := h.db.OpenDM(pid, channelID); openErr != nil {
|
||||
slog.Error("ws handleChatSend OpenDM", "err", openErr,
|
||||
"recipient_id", pid, "channel_id", channelID)
|
||||
continue
|
||||
}
|
||||
if c.user != nil {
|
||||
h.SendToUser(pid, buildDMChannelOpen(channelID, c.user))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -518,6 +518,129 @@ func TestHub_CleanupVoiceForChannel_NoVoiceState_NoPanic(t *testing.T) {
|
||||
// the client map entry; voice cleanup for disconnects is handled by
|
||||
// handleVoiceLeave called from readPump/ICE monitor.
|
||||
|
||||
// ─── sweepStaleClients ──────────────────────────────────────────────────────
|
||||
|
||||
func TestHub_SweepStaleClients_RemovesInactiveClients(t *testing.T) {
|
||||
hub, database := newTestHub(t)
|
||||
go hub.Run()
|
||||
defer hub.Stop()
|
||||
|
||||
u1 := seedTestUser(t, database, "stale-alice")
|
||||
u2 := seedTestUser(t, database, "fresh-bob")
|
||||
|
||||
s1 := make(chan []byte, 4)
|
||||
s2 := make(chan []byte, 4)
|
||||
c1 := ws.NewTestClient(hub, u1, s1)
|
||||
c2 := ws.NewTestClient(hub, u2, s2)
|
||||
|
||||
hub.Register(c1)
|
||||
hub.Register(c2)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
ws.SetClientLastActivityForTest(c1, time.Now().Add(-2*time.Minute))
|
||||
ws.SetClientLastActivityForTest(c2, time.Now())
|
||||
|
||||
hub.SweepStaleClientsForTest()
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
if hub.ClientCount() != 1 {
|
||||
t.Errorf("ClientCount = %d after sweep, want 1", hub.ClientCount())
|
||||
}
|
||||
if hub.GetClient(u1) != nil {
|
||||
t.Error("stale client should have been removed")
|
||||
}
|
||||
if hub.GetClient(u2) == nil {
|
||||
t.Error("fresh client should still be present")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHub_SweepStaleClients_NoClientsNoPanic(t *testing.T) {
|
||||
hub, _ := newTestHub(t)
|
||||
hub.SweepStaleClientsForTest()
|
||||
}
|
||||
|
||||
func TestHub_SweepStaleClients_AllFresh(t *testing.T) {
|
||||
hub, database := newTestHub(t)
|
||||
go hub.Run()
|
||||
defer hub.Stop()
|
||||
|
||||
u1 := seedTestUser(t, database, "fresh-carol")
|
||||
s1 := make(chan []byte, 4)
|
||||
c1 := ws.NewTestClient(hub, u1, s1)
|
||||
hub.Register(c1)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
ws.SetClientLastActivityForTest(c1, time.Now())
|
||||
hub.SweepStaleClientsForTest()
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
|
||||
if hub.ClientCount() != 1 {
|
||||
t.Errorf("ClientCount = %d after sweep of fresh clients, want 1", hub.ClientCount())
|
||||
}
|
||||
}
|
||||
|
||||
// ─── LiveKitHealthCheck ─────────────────────────────────────────────────────
|
||||
|
||||
func TestHub_LiveKitHealthCheck_NilReturnsError(t *testing.T) {
|
||||
hub, _ := newTestHub(t)
|
||||
ok, err := hub.LiveKitHealthCheck()
|
||||
if ok {
|
||||
t.Error("expected ok=false when LiveKit is nil")
|
||||
}
|
||||
if err == nil {
|
||||
t.Error("expected non-nil error when LiveKit is nil")
|
||||
}
|
||||
}
|
||||
|
||||
// ─── SetLiveKitProcess ──────────────────────────────────────────────────────
|
||||
|
||||
func TestHub_SetLiveKitProcess(t *testing.T) {
|
||||
hub, _ := newTestHub(t)
|
||||
hub.SetLiveKitProcess(nil)
|
||||
go hub.Run()
|
||||
hub.GracefulStop()
|
||||
}
|
||||
|
||||
// ─── VoiceSessionCount ─────────────────────────────────────────────────────
|
||||
|
||||
func TestHub_VoiceSessionCount(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
voiceChs []int64
|
||||
wantCount int
|
||||
}{
|
||||
{"no clients", nil, 0},
|
||||
{"all in voice", []int64{100, 200, 300}, 3},
|
||||
{"none in voice", []int64{0, 0}, 0},
|
||||
{"mixed", []int64{100, 0, 200, 0}, 2},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
hub, database := newTestHub(t)
|
||||
go hub.Run()
|
||||
defer hub.Stop()
|
||||
|
||||
for i, vch := range tc.voiceChs {
|
||||
username := fmt.Sprintf("voice-%s-%d", tc.name, i)
|
||||
uid := seedTestUser(t, database, username)
|
||||
send := make(chan []byte, 4)
|
||||
c := ws.NewTestClient(hub, uid, send)
|
||||
if vch != 0 {
|
||||
ws.SetClientVoiceChID(c, vch)
|
||||
}
|
||||
hub.Register(c)
|
||||
}
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
|
||||
got := hub.VoiceSessionCount()
|
||||
if got != tc.wantCount {
|
||||
t.Errorf("VoiceSessionCount() = %d, want %d", got, tc.wantCount)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// hubTestSchema is the minimal schema needed for hub tests.
|
||||
var hubTestSchema = []byte(`
|
||||
CREATE TABLE IF NOT EXISTS roles (
|
||||
|
||||
@@ -2,13 +2,19 @@ package ws_test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/owncord/server/config"
|
||||
"github.com/owncord/server/ws"
|
||||
)
|
||||
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// livekit.go tests
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -508,3 +514,351 @@ func TestWebhook_ParticipantLeft_OldToken_DoesNotTeardownReplacement(t *testing.
|
||||
t.Fatalf("replacement join token = %q, want %q", vs.JoinedAt, newState.JoinedAt)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// livekit_process.go – generateConfig tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestGenerateConfig_WritesYAML(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dataDir := t.TempDir()
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "testkey",
|
||||
LiveKitAPISecret: "testsecret",
|
||||
LiveKitURL: "ws://localhost:7880",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, dataDir)
|
||||
|
||||
cfgPath, err := proc.GenerateConfigForTest()
|
||||
if err != nil {
|
||||
t.Fatalf("generateConfig: %v", err)
|
||||
}
|
||||
|
||||
content, err := os.ReadFile(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatalf("reading config file: %v", err)
|
||||
}
|
||||
|
||||
got := string(content)
|
||||
|
||||
for _, want := range []string{
|
||||
"port: 7880",
|
||||
`"testkey": "testsecret"`,
|
||||
"port_range_start: 50000",
|
||||
"port_range_end: 60000",
|
||||
"use_external_ip: true",
|
||||
"level: info",
|
||||
} {
|
||||
if !strings.Contains(got, want) {
|
||||
t.Errorf("config missing %q.\nGot:\n%s", want, got)
|
||||
}
|
||||
}
|
||||
|
||||
if strings.Contains(got, "node_ip") {
|
||||
t.Error("config should not contain node_ip when NodeIP is empty")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateConfig_WithNodeIP(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dataDir := t.TempDir()
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "key1",
|
||||
LiveKitAPISecret: "secret1",
|
||||
LiveKitURL: "ws://localhost:7880",
|
||||
NodeIP: "203.0.113.10",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, dataDir)
|
||||
|
||||
cfgPath, err := proc.GenerateConfigForTest()
|
||||
if err != nil {
|
||||
t.Fatalf("generateConfig: %v", err)
|
||||
}
|
||||
|
||||
content, err := os.ReadFile(cfgPath)
|
||||
if err != nil {
|
||||
t.Fatalf("reading config file: %v", err)
|
||||
}
|
||||
|
||||
got := string(content)
|
||||
if !strings.Contains(got, `node_ip: "203.0.113.10"`) {
|
||||
t.Errorf("expected node_ip in config.\nGot:\n%s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateConfig_UnsafeCredentialChars(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
key string
|
||||
secret string
|
||||
}{
|
||||
{"colon in key", "bad:key", "secret"},
|
||||
{"newline in secret", "key", "bad\nsecret"},
|
||||
{"hash in key", "bad#key", "secret"},
|
||||
{"brace in secret", "key", "bad{secret"},
|
||||
{"backslash in key", `bad\key`, "secret"},
|
||||
{"quote in secret", "key", `bad"secret`},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: tt.key,
|
||||
LiveKitAPISecret: tt.secret,
|
||||
LiveKitURL: "ws://localhost:7880",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
_, err := proc.GenerateConfigForTest()
|
||||
if err == nil {
|
||||
t.Error("expected error for unsafe YAML character, got nil")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateConfig_UnsafeNodeIPChars(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "safekey",
|
||||
LiveKitAPISecret: "safesecret",
|
||||
LiveKitURL: "ws://localhost:7880",
|
||||
NodeIP: "192.168.1.1\n evil: true",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
_, err := proc.GenerateConfigForTest()
|
||||
if err == nil {
|
||||
t.Error("expected error for unsafe node_ip character, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// livekit_process.go – Start guard tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestStart_AlreadyRunningGuard(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "key",
|
||||
LiveKitAPISecret: "secret",
|
||||
LiveKitURL: "ws://localhost:7880",
|
||||
LiveKitBinaryPath: "/nonexistent/livekit-server",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
proc.SetProcessCmdForTest()
|
||||
|
||||
err := proc.Start()
|
||||
if err == nil {
|
||||
t.Fatal("expected error when process already running, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "already running") {
|
||||
t.Errorf("expected 'already running' error, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStart_StoppedGuard(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "key",
|
||||
LiveKitAPISecret: "secret",
|
||||
LiveKitURL: "ws://localhost:7880",
|
||||
LiveKitBinaryPath: "/nonexistent/livekit-server",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
proc.SetProcessStoppedForTest()
|
||||
|
||||
err := proc.Start()
|
||||
if err != nil {
|
||||
t.Fatalf("Start() on stopped process returned error: %v", err)
|
||||
}
|
||||
|
||||
proc.Stop()
|
||||
if proc.IsRunning() {
|
||||
t.Error("expected IsRunning() = false after Stop()")
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// livekit_process.go – HealthCheck tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestHealthCheck_Success(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws://" + srv.Listener.Addr().String()
|
||||
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "key",
|
||||
LiveKitAPISecret: "secret",
|
||||
LiveKitURL: wsURL,
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
ok, err := proc.HealthCheck()
|
||||
if err != nil {
|
||||
t.Fatalf("HealthCheck: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Error("expected HealthCheck to return true")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthCheck_ServerDown(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "key",
|
||||
LiveKitAPISecret: "secret",
|
||||
LiveKitURL: "ws://127.0.0.1:1",
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
ok, err := proc.HealthCheck()
|
||||
if err == nil {
|
||||
t.Fatal("expected error for unreachable server, got nil")
|
||||
}
|
||||
if ok {
|
||||
t.Error("expected HealthCheck to return false on error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthCheck_NonOKStatus(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
wsURL := "ws://" + srv.Listener.Addr().String()
|
||||
cfg := &config.VoiceConfig{
|
||||
LiveKitAPIKey: "key",
|
||||
LiveKitAPISecret: "secret",
|
||||
LiveKitURL: wsURL,
|
||||
}
|
||||
tlsCfg := &config.TLSConfig{}
|
||||
proc := ws.NewLiveKitProcess(cfg, tlsCfg, t.TempDir())
|
||||
|
||||
ok, err := proc.HealthCheck()
|
||||
if err != nil {
|
||||
t.Fatalf("HealthCheck: %v", err)
|
||||
}
|
||||
if !ok {
|
||||
t.Error("expected HealthCheck to return true even for 500 status")
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// livekit_webhook.go – NewLiveKitWebhookHandler tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestWebhookHandler_MissingAuthHeader(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
hub := ws.NewHubForTest()
|
||||
handler := hub.NewLiveKitWebhookHandler("api-key", "api-secret")
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/livekit/webhook", strings.NewReader(`{}`))
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
handler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected 401, got %d", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookHandler_InvalidToken(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
hub := ws.NewHubForTest()
|
||||
handler := hub.NewLiveKitWebhookHandler("api-key", "api-secret")
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/livekit/webhook",
|
||||
strings.NewReader(`{}`))
|
||||
req.Header.Set("Authorization", "Bearer not-a-valid-jwt-token")
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
handler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected 401 for invalid token, got %d", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebhookHandler_EmptyBody(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
hub := ws.NewHubForTest()
|
||||
handler := hub.NewLiveKitWebhookHandler("api-key", "api-secret")
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/livekit/webhook",
|
||||
strings.NewReader(""))
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
handler(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected 401 for missing auth, got %d", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// livekit_webhook.go – MountWebhookRoute tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestMountWebhookRoute_RegistersRoute(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
hub := ws.NewHubForTest()
|
||||
handler := ws.MountWebhookRoute(hub, "key", "secret")
|
||||
|
||||
if handler == nil {
|
||||
t.Fatal("MountWebhookRoute returned nil handler")
|
||||
}
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Post("/livekit/webhook", handler)
|
||||
|
||||
req := httptest.NewRequest(http.MethodPost, "/livekit/webhook",
|
||||
strings.NewReader(`{}`))
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
r.ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code == http.StatusNotFound {
|
||||
t.Error("expected route to be registered, got 404")
|
||||
}
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected 401 from mounted webhook handler, got %d", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
+100
-144
@@ -36,39 +36,12 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun
|
||||
}
|
||||
conn.SetReadLimit(1 << 20) // 1 MB — match client-side limit
|
||||
|
||||
user, tokenHash, lastSeq, err := authenticateConn(conn, database)
|
||||
c, lastSeq, err := hub.upgradeAndAuth(conn, database, r)
|
||||
if err != nil {
|
||||
slog.Warn("ws auth failed", "err", err, "remote", r.RemoteAddr)
|
||||
_ = conn.Close(websocket.StatusPolicyViolation, "authentication failed")
|
||||
return
|
||||
}
|
||||
|
||||
c := newClient(hub, conn, user, tokenHash, lastSeq, r.Context())
|
||||
c.remoteAddr = r.RemoteAddr
|
||||
|
||||
// Look up role name for protocol-compliant payloads and cache on client.
|
||||
roleName := "member"
|
||||
if role, roleErr := database.GetRoleByID(user.RoleID); roleErr == nil && role != nil {
|
||||
roleName = strings.ToLower(role.Name)
|
||||
}
|
||||
c.roleName = roleName
|
||||
|
||||
slog.Info("websocket connected", "username", user.Username, "user_id", user.ID, "remote", r.RemoteAddr)
|
||||
_ = database.LogAudit(user.ID, "ws_connect", "user", user.ID,
|
||||
"WebSocket connected from "+r.RemoteAddr)
|
||||
|
||||
ctx := r.Context()
|
||||
hydrateVoiceJoinToken := func() {
|
||||
voiceChID := c.getVoiceChID()
|
||||
if voiceChID == 0 || c.getVoiceJoinToken() != "" {
|
||||
return
|
||||
}
|
||||
vs, vsErr := database.GetVoiceState(user.ID)
|
||||
if vsErr != nil || vs == nil || vs.ChannelID != voiceChID || vs.JoinedAt == "" {
|
||||
return
|
||||
}
|
||||
c.setVoiceState(voiceChID, vs.JoinedAt)
|
||||
}
|
||||
startPumps := func() {
|
||||
writeCtx, writeCancel := context.WithCancel(ctx)
|
||||
go writePump(writeCtx, conn, c)
|
||||
@@ -81,130 +54,17 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun
|
||||
// try to replay missed events from the ring buffer instead of
|
||||
// sending a full ready payload.
|
||||
if lastSeq > 0 {
|
||||
events := hub.ReplayBuffer().EventsSince(lastSeq)
|
||||
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)
|
||||
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 {
|
||||
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)
|
||||
hydrateVoiceJoinToken()
|
||||
|
||||
// Update presence but skip member_join — user was already known.
|
||||
if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil {
|
||||
slog.Warn("ws UpdateUserStatus", "err", updateErr)
|
||||
}
|
||||
hub.BroadcastToAll(buildPresenceMsg(user.ID, "online"))
|
||||
|
||||
// Start pumps.
|
||||
if hub.handleReconnect(ctx, conn, c, database, lastSeq) {
|
||||
startPumps()
|
||||
return
|
||||
}
|
||||
// Replay failed (seq too old) — fall through to full ready payload.
|
||||
slog.Info("ws replay failed (seq too old), sending full ready", "user_id", user.ID, "last_seq", lastSeq)
|
||||
slog.Info("ws replay failed (seq too old), sending full ready", "user_id", c.userID, "last_seq", lastSeq)
|
||||
}
|
||||
|
||||
// Fresh connection or replay fallback: full auth_ok + ready flow.
|
||||
|
||||
// Clean any stale voice state BEFORE building the ready payload so
|
||||
// the user doesn't appear as a ghost in a voice channel they left
|
||||
// abruptly (e.g. F5 reload). Only for truly fresh connections
|
||||
// (lastSeq == 0); when lastSeq > 0 the client still has its
|
||||
// JS context with a LiveKit room — it just needs a new ready payload
|
||||
// because the replay buffer was too old.
|
||||
//
|
||||
// This is the SINGLE authoritative cleanup path for fresh connections.
|
||||
// registerNow does NOT duplicate this — it only handles in-memory
|
||||
// client replacement and voice state transfer for lastSeq > 0.
|
||||
if lastSeq == 0 {
|
||||
vs, vsErr := database.GetVoiceState(user.ID)
|
||||
if vsErr != nil {
|
||||
// DB read failure — fail closed. A transient read failure
|
||||
// could leak stale voice state into the ready payload.
|
||||
slog.Error("ws: GetVoiceState failed — aborting connection",
|
||||
"user_id", user.ID, "err", vsErr)
|
||||
_ = conn.Write(ctx, websocket.MessageText,
|
||||
buildErrorMsg(ErrCodeInternal, "voice state check failed"))
|
||||
_ = conn.Close(websocket.StatusInternalError, "voice state check failed")
|
||||
return
|
||||
}
|
||||
if vs != nil {
|
||||
staleChID := vs.ChannelID
|
||||
slog.Info("ws cleaning stale voice state before ready",
|
||||
"user_id", user.ID, "stale_channel_id", staleChID)
|
||||
// Channel-conditional delete: only removes the row if it still
|
||||
// points at staleChID. If the old connection moved the user to
|
||||
// a different channel between GetVoiceState and now, the delete
|
||||
// is a safe no-op and we skip the broadcast.
|
||||
deleted, dbErr := database.LeaveVoiceChannelIfMatch(user.ID, staleChID, vs.JoinedAt)
|
||||
if dbErr != nil {
|
||||
slog.Error("ws: stale voice cleanup failed — aborting connection",
|
||||
"user_id", user.ID, "channel_id", staleChID, "err", dbErr)
|
||||
_ = conn.Write(ctx, websocket.MessageText,
|
||||
buildErrorMsg(ErrCodeInternal, "voice state cleanup failed"))
|
||||
_ = conn.Close(websocket.StatusInternalError, "voice cleanup failed")
|
||||
return
|
||||
}
|
||||
if deleted {
|
||||
// Clear the old client's in-memory voice state BEFORE
|
||||
// calling RemoveParticipant. RemoveParticipant triggers a
|
||||
// LiveKit participant_left webhook; if the old client
|
||||
// still carries the matching join token, the webhook
|
||||
// handler's token-match branch would broadcast a second
|
||||
// voice_leave. Clearing first makes that branch a no-op.
|
||||
hub.mu.RLock()
|
||||
if oldClient, ok := hub.clients[user.ID]; ok {
|
||||
oldClient.clearVoiceState()
|
||||
}
|
||||
hub.mu.RUnlock()
|
||||
|
||||
hub.BroadcastToAll(buildVoiceLeave(staleChID, user.ID))
|
||||
if hub.livekit != nil {
|
||||
_ = hub.livekit.RemoveParticipant(staleChID, user.ID, vs.JoinedAt)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
slog.Info("ws sending auth_ok", "user_id", user.ID, "username", user.Username, "role", 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")
|
||||
if err := hub.handleFreshConnect(ctx, conn, c, database); err != nil {
|
||||
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))
|
||||
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)
|
||||
hydrateVoiceJoinToken()
|
||||
|
||||
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))
|
||||
hub.BroadcastToAll(buildPresenceMsg(user.ID, "online"))
|
||||
|
||||
// writePump runs in background; readPump blocks.
|
||||
// When readPump returns (disconnect), close the send channel first
|
||||
@@ -213,6 +73,102 @@ func ServeWS(hub *Hub, database *db.DB, allowedOrigins []string) http.HandlerFun
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) upgradeAndAuth(
|
||||
conn *websocket.Conn, database *db.DB, r *http.Request,
|
||||
) (*Client, uint64, error) {
|
||||
user, tokenHash, lastSeq, err := authenticateConn(conn, database)
|
||||
if err != nil {
|
||||
slog.Warn("ws auth failed", "err", err, "remote", r.RemoteAddr)
|
||||
_ = conn.Close(websocket.StatusPolicyViolation, "authentication failed")
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
c := newClient(h, conn, user, tokenHash, lastSeq, r.Context())
|
||||
c.remoteAddr = r.RemoteAddr
|
||||
|
||||
// Look up role name for protocol-compliant payloads and cache on client.
|
||||
roleName := "member"
|
||||
if role, roleErr := database.GetRoleByID(user.RoleID); roleErr == nil && role != nil {
|
||||
roleName = strings.ToLower(role.Name)
|
||||
}
|
||||
c.roleName = roleName
|
||||
|
||||
slog.Info("websocket connected", "username", user.Username, "user_id", user.ID, "remote", r.RemoteAddr)
|
||||
_ = database.LogAudit(user.ID, "ws_connect", "user", user.ID,
|
||||
"WebSocket connected from "+r.RemoteAddr)
|
||||
|
||||
return c, lastSeq, nil
|
||||
}
|
||||
|
||||
func (h *Hub) handleReconnect(
|
||||
ctx context.Context, conn *websocket.Conn, c *Client, database *db.DB, lastSeq uint64,
|
||||
) bool {
|
||||
events := h.ReplayBuffer().EventsSince(lastSeq)
|
||||
if events == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Replay succeeded — send auth_ok then missed events.
|
||||
slog.Info("ws sending auth_ok (reconnect)", "user_id", c.userID, "username", c.user.Username, "role", c.roleName)
|
||||
if err := conn.Write(ctx, websocket.MessageText, h.buildAuthOK(c.user, c.roleName)); err != nil {
|
||||
slog.Warn("ws: failed to send auth_ok (reconnect)", "user_id", c.userID, "err", err)
|
||||
_ = conn.Close(websocket.StatusInternalError, "handshake failed")
|
||||
return true
|
||||
}
|
||||
for _, evt := range events {
|
||||
if err := conn.Write(ctx, websocket.MessageText, evt); err != nil {
|
||||
slog.Warn("ws: failed to send replay event", "user_id", c.userID, "err", err)
|
||||
_ = conn.Close(websocket.StatusInternalError, "handshake failed")
|
||||
return true
|
||||
}
|
||||
}
|
||||
slog.Info("ws replay completed", "user_id", c.userID, "events_replayed", len(events), "from_seq", lastSeq)
|
||||
h.registerNow(c)
|
||||
|
||||
// Update presence but skip member_join — user was already known.
|
||||
if updateErr := database.UpdateUserStatus(c.userID, "online"); updateErr != nil {
|
||||
slog.Warn("ws UpdateUserStatus", "err", updateErr)
|
||||
}
|
||||
h.BroadcastToAll(buildPresenceMsg(c.userID, "online"))
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (h *Hub) handleFreshConnect(
|
||||
ctx context.Context, conn *websocket.Conn, c *Client, database *db.DB,
|
||||
) error {
|
||||
// Fresh connection or replay fallback: full auth_ok + ready flow.
|
||||
slog.Info("ws sending auth_ok", "user_id", c.userID, "username", c.user.Username, "role", c.roleName)
|
||||
if err := conn.Write(ctx, websocket.MessageText, h.buildAuthOK(c.user, c.roleName)); err != nil {
|
||||
slog.Warn("ws: failed to send auth_ok", "user_id", c.userID, "err", err)
|
||||
_ = conn.Close(websocket.StatusInternalError, "handshake failed")
|
||||
return err
|
||||
}
|
||||
if ready, readyErr := h.buildReady(database, c.userID); readyErr == nil {
|
||||
slog.Info("ws sending ready payload", "user_id", c.userID, "payload_bytes", len(ready))
|
||||
if err := conn.Write(ctx, websocket.MessageText, ready); err != nil {
|
||||
slog.Warn("ws: failed to send ready payload", "user_id", c.userID, "err", err)
|
||||
_ = conn.Close(websocket.StatusInternalError, "handshake failed")
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
slog.Error("buildReady failed", "user_id", c.userID, "err", readyErr)
|
||||
_ = conn.Write(ctx, websocket.MessageText,
|
||||
buildErrorMsg(ErrCodeInternal, "failed to build ready payload"))
|
||||
}
|
||||
h.registerNow(c)
|
||||
|
||||
if updateErr := database.UpdateUserStatus(c.userID, "online"); updateErr != nil {
|
||||
slog.Warn("ws UpdateUserStatus", "err", updateErr)
|
||||
}
|
||||
|
||||
slog.Info("ws broadcasting member_join and presence", "user_id", c.userID, "username", c.user.Username)
|
||||
h.BroadcastToAll(buildMemberJoin(c.user, c.roleName))
|
||||
h.BroadcastToAll(buildPresenceMsg(c.userID, "online"))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// writePump drains the client's send channel and writes to the WebSocket.
|
||||
func writePump(ctx context.Context, conn *websocket.Conn, c *Client) {
|
||||
for {
|
||||
|
||||
+30
@@ -248,6 +248,36 @@ Invalidate the current session token.
|
||||
|
||||
---
|
||||
|
||||
### DELETE /api/v1/auth/account
|
||||
|
||||
Permanently delete the authenticated user's account. Requires password confirmation.
|
||||
|
||||
**Auth:** Required (Bearer token)
|
||||
**Rate limit:** 5 requests/minute per IP. After 3 failed password attempts, the endpoint locks out for 15 minutes per user.
|
||||
|
||||
#### Request
|
||||
|
||||
```json
|
||||
{
|
||||
"password": "MyStr0ng!Pass"
|
||||
}
|
||||
```
|
||||
|
||||
#### Response 204 No Content
|
||||
|
||||
Account deleted successfully. All sessions, messages (soft-deleted), and associated data are cleaned up.
|
||||
|
||||
#### Errors
|
||||
|
||||
| Status | Code | Cause |
|
||||
| ------ | ---- | ----- |
|
||||
| 400 | `INVALID_INPUT` | Missing or incorrect password |
|
||||
| 403 | `FORBIDDEN` | Cannot delete the last admin account |
|
||||
| 429 | `RATE_LIMITED` | Locked out after 3 failed password attempts (15 min cooldown) |
|
||||
| 500 | `SERVER_ERROR` | Database error during deletion |
|
||||
|
||||
---
|
||||
|
||||
### POST /api/v1/users/me/totp/enable
|
||||
|
||||
Start TOTP enrollment for the authenticated user. The secret is not persisted until `/api/v1/users/me/totp/confirm` succeeds.
|
||||
|
||||
@@ -7,7 +7,7 @@ How to set up the development environment and contribute to OwnCord.
|
||||
### Prerequisites
|
||||
|
||||
- **Windows 10+** (x64)
|
||||
- **Go 1.22+** (server)
|
||||
- **Go 1.25+** (server)
|
||||
- **Node.js 20+** (client)
|
||||
- **Rust / Cargo** (Tauri client)
|
||||
|
||||
@@ -40,6 +40,7 @@ How to set up the development environment and contribute to OwnCord.
|
||||
| `npm run test:watch` | Vitest watch mode |
|
||||
| `npm run test:coverage` | Coverage report |
|
||||
| `npm run typecheck` | Full typecheck (all sources) |
|
||||
| `npm run typecheck:build` | Typecheck build config only |
|
||||
| `npm run lint` | ESLint check (src/) |
|
||||
| `npm run lint:fix` | ESLint auto-fix |
|
||||
|
||||
|
||||
+5
-5
@@ -5,7 +5,7 @@ Production deployment guide for OwnCord server on Windows.
|
||||
## Prerequisites
|
||||
|
||||
- **Windows 10+** (x64)
|
||||
- **Go 1.22+** (only if building from source)
|
||||
- **Go 1.25+** (only if building from source)
|
||||
- **LiveKit Server** binary (for voice/video) -- see [LiveKit Setup](livekit-setup.md)
|
||||
- Ports available: `8443` (default), `7880` (LiveKit), `80` (if using ACME/Let's Encrypt)
|
||||
|
||||
@@ -115,10 +115,10 @@ The database uses SQLite WAL mode. Do NOT copy the `.db` file directly while the
|
||||
|
||||
| Endpoint | Method | Description |
|
||||
|----------|--------|-------------|
|
||||
| `/admin/api/backups` | POST | Create a new backup |
|
||||
| `/admin/api/backup` | POST | Create a new backup (owner-only) |
|
||||
| `/admin/api/backups` | GET | List all backups (newest first) |
|
||||
| `/admin/api/backups/{name}` | DELETE | Delete a backup |
|
||||
| `/admin/api/backups/{name}/restore` | POST | Restore from backup (creates pre-restore safety backup first) |
|
||||
| `/admin/api/backups/{name}` | DELETE | Delete a backup (owner-only) |
|
||||
| `/admin/api/backups/{name}/restore` | POST | Restore from backup (owner-only; creates pre-restore safety backup first) |
|
||||
|
||||
Backups are stored in `data/backups/` with timestamps.
|
||||
|
||||
@@ -128,7 +128,7 @@ Use Windows Task Scheduler with PowerShell:
|
||||
|
||||
```powershell
|
||||
$headers = @{ "Cookie" = "session=<admin-session-token>" }
|
||||
Invoke-RestMethod -Uri "https://localhost:8443/admin/api/backups" -Method POST -Headers $headers -SkipCertificateCheck
|
||||
Invoke-RestMethod -Uri "https://localhost:8443/admin/api/backup" -Method POST -Headers $headers -SkipCertificateCheck
|
||||
```
|
||||
|
||||
### Restore
|
||||
|
||||
@@ -23,6 +23,21 @@ OwnCord supports TOTP-based 2FA:
|
||||
- `require_2fa` requires all users to have 2FA enabled and registration to be closed
|
||||
- Login flow returns `requires_2fa: true` with a `partial_token` (10-min TTL, 5-attempt limit)
|
||||
- Auth challenges are rate-limited to 10 req/min per IP
|
||||
- TOTP code verification uses constant-time comparison (`subtle.ConstantTimeCompare`) to prevent timing side-channel attacks
|
||||
|
||||
## Account Deletion
|
||||
|
||||
Users can delete their own account via `DELETE /api/v1/auth/account` with password confirmation. The last admin account cannot be deleted. After 3 failed password attempts, the endpoint locks out for 15 minutes.
|
||||
|
||||
## Audit Logging
|
||||
|
||||
Security-relevant actions are recorded in the `audit_log` table with actor, action, target, and detail:
|
||||
|
||||
- **Auth:** `user_register`, `user_login`, `user_logout`, `login_blocked_banned`, `account_deleted`
|
||||
- **2FA:** `totp_enabled`, `totp_verified`, `totp_disabled`
|
||||
- **Admin:** `role_change`, `user_ban`, `user_unban`, `force_logout`, `setting_change`, `server_setup`
|
||||
- **Content:** `channel_create`, `channel_update`, `channel_delete`, `message_delete`
|
||||
- **Ops:** `backup_create`, `backup_delete`, `backup_restore`, `ws_connect`
|
||||
|
||||
## Known Limitations
|
||||
|
||||
|
||||
Reference in New Issue
Block a user