J3vb/OwnCord
1
0
Fork 0
mirror of https://github.com/J3vb/OwnCord.git synced 2026-09-03 03:50:00 +03:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement Phase 4 real-time chat (WebSocket hub + message REST)

Browse Source 
- db: channel_queries (ListChannels, GetChannel, CRUD, permissions),
  message_queries (CreateMessage, GetMessage, GetMessages paginated,
  EditMessage, DeleteMessage soft, AddReaction, RemoveReaction,
  GetReactions, SearchMessages FTS5, UpdateReadState)
- db: fix in-memory DB isolation — SetMaxOpenConns(1) for :memory: path
- ws/hub: replace stub with full Hub (register/unregister, broadcast to
  channel/all, send to user, thread-safe, buffered broadcast channel)
- ws/client: Client with send channel, NewTestClient helpers for tests
- ws/handlers: dispatch chat_send/edit/delete, reaction_add/remove,
  typing_start, presence_update — all with rate limiting and permission checks
- ws/messages: JSON builder helpers for all server→client message types
- ws/serve: ServeWS HTTP handler, WS auth handshake (10s timeout),
  ready payload, writePump/readPump goroutines, graceful disconnect
- api: channel_handler — GET /channels, GET /channels/{id}/messages,
  GET /search; fixed double-mount of /api/v1 route group
- api/router: mount channel routes, start hub, register /api/v1/ws

Test coverage: api 77.6%, auth 90.9%, db 84.2%, ws 26.7% (serve.go
requires live WS connection; hub/handlers/messages fully covered)
This commit is contained in:
jevb
2026-03-14 21:17:09 +01:00
parent 814653ea08
commit 36640e3051
17 changed files with 3387 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Server/api/channel_handler.go
+196
View File
@@ -0,0 +1,196 @@
package api
import (
"net/http"
"strconv"
"github.com/go-chi/chi/v5"
"github.com/owncord/server/db"
)
const (
// Permission bits (from SCHEMA.md).
permReadMessages = int64(0x0400)
permAdministrator = int64(0x40000000)
defaultMessageLimit = 50
maxMessageLimit = 100
)
// MountChannelRoutes registers all channel-related routes onto r.
// All routes require authentication.
func MountChannelRoutes(r chi.Router, database *db.DB) {
r.Route("/api/v1/channels", func(r chi.Router) {
r.Use(AuthMiddleware(database))
r.Get("/", handleListChannels(database))
r.Get("/{id}/messages", handleGetMessages(database))
})
r.With(AuthMiddleware(database)).Get("/api/v1/search", handleSearch(database))
}
// handleListChannels returns all channels the authenticated user can see.
func handleListChannels(database *db.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
channels, err := database.ListChannels()
if err != nil {
writeJSON(w, http.StatusInternalServerError, errorResponse{
Error: "INTERNAL",
Message: "failed to list channels",
})
return
}
writeJSON(w, http.StatusOK, channels)
}
}
// handleGetMessages returns paginated messages for a channel.
// Query params: before (int64, message ID for pagination), limit (1-100, default 50).
func handleGetMessages(database *db.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
channelID, ok := parseIDParam(w, r, "id")
if !ok {
return
}
ch, err := database.GetChannel(channelID)
if err != nil {
writeJSON(w, http.StatusInternalServerError, errorResponse{
Error: "INTERNAL",
Message: "failed to look up channel",
})
return
}
if ch == nil {
writeJSON(w, http.StatusNotFound, errorResponse{
Error: "NOT_FOUND",
Message: "channel not found",
})
return
}
// Parse query params.
before := int64(0)
if raw := r.URL.Query().Get("before"); raw != "" {
v, parseErr := strconv.ParseInt(raw, 10, 64)
if parseErr != nil || v < 0 {
writeJSON(w, http.StatusBadRequest, errorResponse{
Error: "BAD_REQUEST",
Message: "before must be a non-negative integer",
})
return
}
before = v
}
limit := defaultMessageLimit
if raw := r.URL.Query().Get("limit"); raw != "" {
v, parseErr := strconv.Atoi(raw)
if parseErr != nil || v < 1 {
writeJSON(w, http.StatusBadRequest, errorResponse{
Error: "BAD_REQUEST",
Message: "limit must be a positive integer",
})
return
}
if v > maxMessageLimit {
v = maxMessageLimit
}
limit = v
}
// Fetch one extra to determine has_more.
msgs, err := database.GetMessages(channelID, before, limit+1)
if err != nil {
writeJSON(w, http.StatusInternalServerError, errorResponse{
Error: "INTERNAL",
Message: "failed to fetch messages",
})
return
}
hasMore := false
if len(msgs) > limit {
hasMore = true
msgs = msgs[:limit]
}
type response struct {
Messages []db.MessageWithUser `json:"messages"`
HasMore bool `json:"has_more"`
}
writeJSON(w, http.StatusOK, response{Messages: msgs, HasMore: hasMore})
}
}
// handleSearch performs a full-text search across messages.
// Query params: q (required), channel_id (optional), limit (optional, 1-100).
func handleSearch(database *db.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query().Get("q")
if q == "" {
writeJSON(w, http.StatusBadRequest, errorResponse{
Error: "BAD_REQUEST",
Message: "query parameter 'q' is required",
})
return
}
var channelID *int64
if raw := r.URL.Query().Get("channel_id"); raw != "" {
v, parseErr := strconv.ParseInt(raw, 10, 64)
if parseErr != nil || v <= 0 {
writeJSON(w, http.StatusBadRequest, errorResponse{
Error: "BAD_REQUEST",
Message: "channel_id must be a positive integer",
})
return
}
channelID = &v
}
limit := defaultMessageLimit
if raw := r.URL.Query().Get("limit"); raw != "" {
v, parseErr := strconv.Atoi(raw)
if parseErr != nil || v < 1 {
writeJSON(w, http.StatusBadRequest, errorResponse{
Error: "BAD_REQUEST",
Message: "limit must be a positive integer",
})
return
}
if v > maxMessageLimit {
v = maxMessageLimit
}
limit = v
}
results, err := database.SearchMessages(q, channelID, limit)
if err != nil {
writeJSON(w, http.StatusInternalServerError, errorResponse{
Error: "INTERNAL",
Message: "search failed",
})
return
}
type response struct {
Results []db.MessageSearchResult `json:"results"`
}
writeJSON(w, http.StatusOK, response{Results: results})
}
}
// parseIDParam extracts and validates a chi URL param as int64.
// Writes a 400 response and returns false on failure.
func parseIDParam(w http.ResponseWriter, r *http.Request, param string) (int64, bool) {
raw := chi.URLParam(r, param)
id, err := strconv.ParseInt(raw, 10, 64)
if err != nil || id <= 0 {
writeJSON(w, http.StatusBadRequest, errorResponse{
Error: "BAD_REQUEST",
Message: param + " must be a positive integer",
})
return 0, false
}
return id, true
}
Server/api/channel_handler_test.go
+427
View File
@@ -0,0 +1,427 @@
package api_test
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"testing"
"testing/fstest"
"github.com/go-chi/chi/v5"
"github.com/owncord/server/api"
"github.com/owncord/server/auth"
"github.com/owncord/server/db"
)
// ─── schema for channel tests ─────────────────────────────────────────────────
var channelTestSchema = []byte(`
CREATE TABLE IF NOT EXISTS roles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
color TEXT,
permissions INTEGER NOT NULL DEFAULT 0,
position INTEGER NOT NULL DEFAULT 0,
is_default INTEGER NOT NULL DEFAULT 0
);
INSERT OR IGNORE INTO roles (id, name, color, permissions, position, is_default) VALUES
(1, 'Owner', '#E74C3C', 2147483647, 100, 0),
(2, 'Admin', '#F39C12', 1073741823, 80, 0),
(3, 'Moderator', '#3498DB', 1048575, 60, 0),
(4, 'Member', NULL, 1049089, 40, 1);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
password TEXT NOT NULL,
avatar TEXT,
role_id INTEGER NOT NULL DEFAULT 4 REFERENCES roles(id),
totp_secret TEXT,
status TEXT NOT NULL DEFAULT 'offline',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
last_seen TEXT,
banned INTEGER NOT NULL DEFAULT 0,
ban_reason TEXT,
ban_expires TEXT
);
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token TEXT NOT NULL UNIQUE,
device TEXT,
ip_address TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
last_used TEXT NOT NULL DEFAULT (datetime('now')),
expires_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sessions_token ON sessions(token);
CREATE TABLE IF NOT EXISTS channels (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
type TEXT NOT NULL DEFAULT 'text',
category TEXT,
topic TEXT,
position INTEGER NOT NULL DEFAULT 0,
slow_mode INTEGER NOT NULL DEFAULT 0,
archived INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS channel_overrides (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
role_id INTEGER NOT NULL REFERENCES roles(id) ON DELETE CASCADE,
allow INTEGER NOT NULL DEFAULT 0,
deny INTEGER NOT NULL DEFAULT 0,
UNIQUE(channel_id, role_id)
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id),
content TEXT NOT NULL,
reply_to INTEGER REFERENCES messages(id) ON DELETE SET NULL,
edited_at TEXT,
deleted INTEGER NOT NULL DEFAULT 0,
pinned INTEGER NOT NULL DEFAULT 0,
timestamp TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel_id, id DESC);
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
content,
content='messages',
content_rowid='id'
);
CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TABLE IF NOT EXISTS reactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
emoji TEXT NOT NULL,
UNIQUE(message_id, user_id, emoji)
);
CREATE TABLE IF NOT EXISTS read_states (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
last_message_id INTEGER NOT NULL DEFAULT 0,
mention_count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (user_id, channel_id)
);
CREATE TABLE IF NOT EXISTS invites (
id INTEGER PRIMARY KEY AUTOINCREMENT,
code TEXT NOT NULL UNIQUE,
created_by INTEGER NOT NULL REFERENCES users(id),
redeemed_by INTEGER REFERENCES users(id),
max_uses INTEGER,
use_count INTEGER NOT NULL DEFAULT 0,
expires_at TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
revoked INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
INSERT OR IGNORE INTO settings (key, value) VALUES
('server_name', 'OwnCord Server'),
('motd', 'Welcome!');
`)
// ─── helpers ──────────────────────────────────────────────────────────────────
func newChannelTestDB(t *testing.T) *db.DB {
t.Helper()
database, err := db.Open(":memory:")
if err != nil {
t.Fatalf("db.Open: %v", err)
}
t.Cleanup(func() { database.Close() })
migrFS := fstest.MapFS{"001_schema.sql": {Data: channelTestSchema}}
if err := db.MigrateFS(database, migrFS); err != nil {
t.Fatalf("MigrateFS: %v", err)
}
return database
}
func buildChannelRouter(database *db.DB) http.Handler {
r := chi.NewRouter()
api.MountChannelRoutes(r, database)
return r
}
// chTestCreateToken creates a user+session and returns the plaintext token.
func chTestCreateToken(t *testing.T, database *db.DB, username string, roleID int) string {
t.Helper()
_, err := database.CreateUser(username, "$2a$12$fake", roleID)
if err != nil {
t.Fatalf("CreateUser %q: %v", username, err)
}
token := "chtest-token-" + username
hash := auth.HashToken(token)
_, err = database.Exec(
`INSERT INTO sessions (user_id, token, device, ip_address, expires_at)
SELECT id, ?, 'test', '127.0.0.1', '2099-01-01T00:00:00Z' FROM users WHERE username = ?`,
hash, username,
)
if err != nil {
t.Fatalf("insert session for %q: %v", username, err)
}
return token
}
func chGet(t *testing.T, router http.Handler, path, token string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, path, nil)
if token != "" {
req.Header.Set("Authorization", "Bearer "+token)
}
req.RemoteAddr = "127.0.0.1:9999"
rr := httptest.NewRecorder()
router.ServeHTTP(rr, req)
return rr
}
// ─── GET /api/v1/channels ─────────────────────────────────────────────────────
func TestChannelList_Unauthenticated(t *testing.T) {
router := buildChannelRouter(newChannelTestDB(t))
rr := chGet(t, router, "/api/v1/channels", "")
if rr.Code != http.StatusUnauthorized {
t.Errorf("status = %d, want 401", rr.Code)
}
}
func TestChannelList_Empty(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "alice", 1)
rr := chGet(t, router, "/api/v1/channels", token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200; body: %s", rr.Code, rr.Body.String())
}
var resp []interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
if len(resp) != 0 {
t.Errorf("expected empty array, got %d items", len(resp))
}
}
func TestChannelList_WithChannels(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "bob", 1)
_, _ = database.CreateChannel("general", "text", "", "", 0)
_, _ = database.CreateChannel("random", "text", "", "", 1)
rr := chGet(t, router, "/api/v1/channels", token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rr.Code)
}
var resp []interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
if len(resp) != 2 {
t.Errorf("expected 2 channels, got %d", len(resp))
}
}
// ─── GET /api/v1/channels/{id}/messages ──────────────────────────────────────
func TestChannelMessages_Unauthenticated(t *testing.T) {
router := buildChannelRouter(newChannelTestDB(t))
rr := chGet(t, router, "/api/v1/channels/1/messages", "")
if rr.Code != http.StatusUnauthorized {
t.Errorf("status = %d, want 401", rr.Code)
}
}
func TestChannelMessages_InvalidID(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "carol", 1)
rr := chGet(t, router, "/api/v1/channels/abc/messages", token)
if rr.Code != http.StatusBadRequest {
t.Errorf("status = %d, want 400", rr.Code)
}
}
func TestChannelMessages_ChannelNotFound(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "dave", 1)
rr := chGet(t, router, "/api/v1/channels/9999/messages", token)
if rr.Code != http.StatusNotFound {
t.Errorf("status = %d, want 404", rr.Code)
}
}
func TestChannelMessages_EmptyChannel(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "eve", 1)
chID, _ := database.CreateChannel("general", "text", "", "", 0)
rr := chGet(t, router, fmt.Sprintf("/api/v1/channels/%d/messages", chID), token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200; body: %s", rr.Code, rr.Body.String())
}
var resp map[string]interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
msgs, ok := resp["messages"].([]interface{})
if !ok || len(msgs) != 0 {
t.Errorf("expected empty messages array, got: %v", resp["messages"])
}
}
func TestChannelMessages_ReturnsMessages(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "frank", 1)
user, _ := database.GetUserByUsername("frank")
chID, _ := database.CreateChannel("ch", "text", "", "", 0)
for i := 0; i < 3; i++ {
_, _ = database.CreateMessage(chID, user.ID, fmt.Sprintf("msg%d", i), nil)
}
rr := chGet(t, router, fmt.Sprintf("/api/v1/channels/%d/messages", chID), token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rr.Code)
}
var resp map[string]interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
msgs := resp["messages"].([]interface{})
if len(msgs) != 3 {
t.Errorf("expected 3 messages, got %d", len(msgs))
}
}
func TestChannelMessages_LimitCappedAt100(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "grace", 1)
chID, _ := database.CreateChannel("ch", "text", "", "", 0)
// limit=200 should succeed (capped internally).
rr := chGet(t, router, fmt.Sprintf("/api/v1/channels/%d/messages?limit=200", chID), token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rr.Code)
}
}
func TestChannelMessages_HasMore(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "henry", 1)
user, _ := database.GetUserByUsername("henry")
chID, _ := database.CreateChannel("ch", "text", "", "", 0)
for i := 0; i < 60; i++ {
_, _ = database.CreateMessage(chID, user.ID, fmt.Sprintf("m%d", i), nil)
}
rr := chGet(t, router, fmt.Sprintf("/api/v1/channels/%d/messages?limit=50", chID), token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rr.Code)
}
var resp map[string]interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
if resp["has_more"] != true {
t.Errorf("has_more = %v, want true", resp["has_more"])
}
}
func TestChannelMessages_HasMoreFalse(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "ivan", 1)
user, _ := database.GetUserByUsername("ivan")
chID, _ := database.CreateChannel("ch", "text", "", "", 0)
for i := 0; i < 5; i++ {
_, _ = database.CreateMessage(chID, user.ID, fmt.Sprintf("m%d", i), nil)
}
rr := chGet(t, router, fmt.Sprintf("/api/v1/channels/%d/messages?limit=50", chID), token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rr.Code)
}
var resp map[string]interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
if resp["has_more"] != false {
t.Errorf("has_more = %v, want false", resp["has_more"])
}
}
// ─── GET /api/v1/search ───────────────────────────────────────────────────────
func TestSearch_Unauthenticated(t *testing.T) {
router := buildChannelRouter(newChannelTestDB(t))
rr := chGet(t, router, "/api/v1/search?q=hello", "")
if rr.Code != http.StatusUnauthorized {
t.Errorf("status = %d, want 401", rr.Code)
}
}
func TestSearch_MissingQuery(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "julia", 1)
rr := chGet(t, router, "/api/v1/search", token)
if rr.Code != http.StatusBadRequest {
t.Errorf("status = %d, want 400", rr.Code)
}
}
func TestSearch_ReturnsResults(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "kim", 1)
user, _ := database.GetUserByUsername("kim")
chID, _ := database.CreateChannel("searchable", "text", "", "", 0)
_, _ = database.CreateMessage(chID, user.ID, "uniqueterm in message", nil)
rr := chGet(t, router, "/api/v1/search?q=uniqueterm", token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200; body: %s", rr.Code, rr.Body.String())
}
var resp map[string]interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
results, ok := resp["results"].([]interface{})
if !ok || len(results) == 0 {
t.Errorf("expected search results, got: %v", resp)
}
}
func TestSearch_NoResults(t *testing.T) {
database := newChannelTestDB(t)
router := buildChannelRouter(database)
token := chTestCreateToken(t, database, "larry", 1)
rr := chGet(t, router, "/api/v1/search?q=xyzzynotfound", token)
if rr.Code != http.StatusOK {
t.Errorf("status = %d, want 200", rr.Code)
}
var resp map[string]interface{}
_ = json.NewDecoder(rr.Body).Decode(&resp)
results := resp["results"].([]interface{})
if len(results) != 0 {
t.Errorf("expected 0 results, got %d", len(results))
}
}
Server/api/router.go
+9
View File
@@ -10,6 +10,7 @@ import (
"github.com/owncord/server/auth"
"github.com/owncord/server/config"
"github.com/owncord/server/db"
"github.com/owncord/server/ws"
)
// version is the server version string, overridden at build time via ldflags.
@@ -42,6 +43,14 @@ func NewRouter(cfg *config.Config, database *db.DB) http.Handler {
// Invite management routes (require MANAGE_INVITES permission).
MountInviteRoutes(r, database)
// Channel and message REST routes.
MountChannelRoutes(r, database)
// WebSocket hub — WS does its own in-band auth, so no AuthMiddleware here.
hub := ws.NewHub(database, limiter)
go hub.Run()
r.Get("/api/v1/ws", ws.ServeWS(hub, database))
return r
}
Server/db/channel_queries.go
+136
View File
@@ -0,0 +1,136 @@
package db
import (
"database/sql"
"errors"
"fmt"
)
// ListChannels returns all channels ordered by position.
func (d *DB) ListChannels() ([]Channel, error) {
rows, err := d.sqlDB.Query(
`SELECT id, name, type, COALESCE(category,''), COALESCE(topic,''),
position, slow_mode, archived, created_at
FROM channels ORDER BY position ASC, id ASC`,
)
if err != nil {
return nil, fmt.Errorf("ListChannels: %w", err)
}
defer rows.Close()
var channels []Channel
for rows.Next() {
ch, scanErr := scanChannel(rows)
if scanErr != nil {
return nil, fmt.Errorf("ListChannels scan: %w", scanErr)
}
channels = append(channels, ch)
}
if rows.Err() != nil {
return nil, fmt.Errorf("ListChannels rows: %w", rows.Err())
}
if channels == nil {
channels = []Channel{}
}
return channels, nil
}
// GetChannel returns the channel with the given id, or nil if not found.
func (d *DB) GetChannel(id int64) (*Channel, error) {
row := d.sqlDB.QueryRow(
`SELECT id, name, type, COALESCE(category,''), COALESCE(topic,''),
position, slow_mode, archived, created_at
FROM channels WHERE id = ?`,
id,
)
ch := &Channel{}
var archived int
err := row.Scan(
&ch.ID, &ch.Name, &ch.Type, &ch.Category, &ch.Topic,
&ch.Position, &ch.SlowMode, &archived, &ch.CreatedAt,
)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("GetChannel: %w", err)
}
ch.Archived = archived != 0
return ch, nil
}
// CreateChannel inserts a new channel and returns the assigned ID.
func (d *DB) CreateChannel(name, chanType, category, topic string, position int) (int64, error) {
res, err := d.sqlDB.Exec(
`INSERT INTO channels (name, type, category, topic, position) VALUES (?, ?, ?, ?, ?)`,
name, chanType, nullableString(category), nullableString(topic), position,
)
if err != nil {
return 0, fmt.Errorf("CreateChannel: %w", err)
}
return res.LastInsertId()
}
// UpdateChannel modifies name, topic, and slow_mode for the given channel.
func (d *DB) UpdateChannel(id int64, name, topic string, slowMode int) error {
_, err := d.sqlDB.Exec(
`UPDATE channels SET name = ?, topic = ?, slow_mode = ? WHERE id = ?`,
name, nullableString(topic), slowMode, id,
)
if err != nil {
return fmt.Errorf("UpdateChannel: %w", err)
}
return nil
}
// DeleteChannel removes the channel row (cascades to messages, overrides, etc.).
func (d *DB) DeleteChannel(id int64) error {
_, err := d.sqlDB.Exec(`DELETE FROM channels WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("DeleteChannel: %w", err)
}
return nil
}
// GetChannelPermissions returns the allow/deny override bits for a role on a
// channel. Returns (0, 0, nil) when no override exists.
func (d *DB) GetChannelPermissions(channelID, roleID int64) (allow, deny int64, err error) {
row := d.sqlDB.QueryRow(
`SELECT allow, deny FROM channel_overrides WHERE channel_id = ? AND role_id = ?`,
channelID, roleID,
)
scanErr := row.Scan(&allow, &deny)
if errors.Is(scanErr, sql.ErrNoRows) {
return 0, 0, nil
}
if scanErr != nil {
return 0, 0, fmt.Errorf("GetChannelPermissions: %w", scanErr)
}
return allow, deny, nil
}
// ─── helpers ──────────────────────────────────────────────────────────────────
// scanChannel scans a single channel row from *sql.Rows.
func scanChannel(rows *sql.Rows) (Channel, error) {
var ch Channel
var archived int
err := rows.Scan(
&ch.ID, &ch.Name, &ch.Type, &ch.Category, &ch.Topic,
&ch.Position, &ch.SlowMode, &archived, &ch.CreatedAt,
)
if err != nil {
return Channel{}, err
}
ch.Archived = archived != 0
return ch, nil
}
// nullableString returns nil when s is empty, otherwise a pointer to s.
// Used so empty strings are stored as NULL in optional TEXT columns.
func nullableString(s string) interface{} {
if s == "" {
return nil
}
return s
}
Server/db/channel_queries_test.go
+235
View File
@@ -0,0 +1,235 @@
package db_test
import (
"testing"
"github.com/owncord/server/db"
)
// openMigratedMemory opens an in-memory DB and runs the full migration.
func openMigratedMemory(t *testing.T) *db.DB {
t.Helper()
database := openMemory(t)
if err := db.Migrate(database); err != nil {
t.Fatalf("Migrate() error: %v", err)
}
return database
}
// ─── ListChannels ─────────────────────────────────────────────────────────────
func TestListChannels_Empty(t *testing.T) {
database := openMigratedMemory(t)
channels, err := database.ListChannels()
if err != nil {
t.Fatalf("ListChannels() error: %v", err)
}
if len(channels) != 0 {
t.Errorf("expected 0 channels, got %d", len(channels))
}
}
func TestListChannels_ReturnsAll(t *testing.T) {
database := openMigratedMemory(t)
if _, err := database.CreateChannel("general", "text", "", "General chat", 0); err != nil {
t.Fatalf("CreateChannel general: %v", err)
}
if _, err := database.CreateChannel("announcements", "text", "", "", 1); err != nil {
t.Fatalf("CreateChannel announcements: %v", err)
}
channels, err := database.ListChannels()
if err != nil {
t.Fatalf("ListChannels() error: %v", err)
}
if len(channels) != 2 {
t.Errorf("expected 2 channels, got %d", len(channels))
}
}
// ─── GetChannel ───────────────────────────────────────────────────────────────
func TestGetChannel_NotFound(t *testing.T) {
database := openMigratedMemory(t)
ch, err := database.GetChannel(9999)
if err != nil {
t.Fatalf("GetChannel() error: %v", err)
}
if ch != nil {
t.Error("expected nil for non-existent channel")
}
}
func TestGetChannel_Found(t *testing.T) {
database := openMigratedMemory(t)
id, err := database.CreateChannel("general", "text", "Public", "hello", 0)
if err != nil {
t.Fatalf("CreateChannel: %v", err)
}
ch, err := database.GetChannel(id)
if err != nil {
t.Fatalf("GetChannel: %v", err)
}
if ch == nil {
t.Fatal("expected channel, got nil")
}
if ch.Name != "general" {
t.Errorf("Name = %q, want 'general'", ch.Name)
}
if ch.Type != "text" {
t.Errorf("Type = %q, want 'text'", ch.Type)
}
if ch.Category != "Public" {
t.Errorf("Category = %q, want 'Public'", ch.Category)
}
if ch.Topic != "hello" {
t.Errorf("Topic = %q, want 'hello'", ch.Topic)
}
if ch.Position != 0 {
t.Errorf("Position = %d, want 0", ch.Position)
}
}
// ─── CreateChannel ────────────────────────────────────────────────────────────
func TestCreateChannel_ReturnsID(t *testing.T) {
database := openMigratedMemory(t)
id, err := database.CreateChannel("test", "text", "", "", 0)
if err != nil {
t.Fatalf("CreateChannel: %v", err)
}
if id <= 0 {
t.Errorf("expected positive ID, got %d", id)
}
}
func TestCreateChannel_UniqueIDs(t *testing.T) {
database := openMigratedMemory(t)
id1, _ := database.CreateChannel("ch1", "text", "", "", 0)
id2, _ := database.CreateChannel("ch2", "text", "", "", 1)
if id1 == id2 {
t.Error("expected different IDs for different channels")
}
}
func TestCreateChannel_EmptyCategory(t *testing.T) {
database := openMigratedMemory(t)
id, err := database.CreateChannel("nocategory", "text", "", "", 0)
if err != nil {
t.Fatalf("CreateChannel with empty category: %v", err)
}
ch, _ := database.GetChannel(id)
if ch.Category != "" {
t.Errorf("Category = %q, want ''", ch.Category)
}
}
// ─── UpdateChannel ────────────────────────────────────────────────────────────
func TestUpdateChannel_ChangesNameAndTopic(t *testing.T) {
database := openMigratedMemory(t)
id, _ := database.CreateChannel("old", "text", "", "old topic", 0)
if err := database.UpdateChannel(id, "new", "new topic", 5); err != nil {
t.Fatalf("UpdateChannel: %v", err)
}
ch, _ := database.GetChannel(id)
if ch.Name != "new" {
t.Errorf("Name = %q, want 'new'", ch.Name)
}
if ch.Topic != "new topic" {
t.Errorf("Topic = %q, want 'new topic'", ch.Topic)
}
if ch.SlowMode != 5 {
t.Errorf("SlowMode = %d, want 5", ch.SlowMode)
}
}
func TestUpdateChannel_NonExistent(t *testing.T) {
database := openMigratedMemory(t)
// Should not error even for non-existent row (0 rows affected is still ok).
err := database.UpdateChannel(9999, "x", "y", 0)
if err != nil {
t.Errorf("UpdateChannel non-existent should not error: %v", err)
}
}
// ─── DeleteChannel ────────────────────────────────────────────────────────────
func TestDeleteChannel_RemovesChannel(t *testing.T) {
database := openMigratedMemory(t)
id, _ := database.CreateChannel("todelete", "text", "", "", 0)
if err := database.DeleteChannel(id); err != nil {
t.Fatalf("DeleteChannel: %v", err)
}
ch, err := database.GetChannel(id)
if err != nil {
t.Fatalf("GetChannel after delete: %v", err)
}
if ch != nil {
t.Error("expected nil after deletion")
}
}
func TestDeleteChannel_NonExistent(t *testing.T) {
database := openMigratedMemory(t)
err := database.DeleteChannel(9999)
if err != nil {
t.Errorf("DeleteChannel non-existent should not error: %v", err)
}
}
// ─── GetChannelPermissions ────────────────────────────────────────────────────
func TestGetChannelPermissions_Default(t *testing.T) {
database := openMigratedMemory(t)
chID, _ := database.CreateChannel("perms", "text", "", "", 0)
// No override set — should return 0, 0.
allow, deny, err := database.GetChannelPermissions(chID, 4)
if err != nil {
t.Fatalf("GetChannelPermissions: %v", err)
}
if allow != 0 || deny != 0 {
t.Errorf("expected (0, 0), got (%d, %d)", allow, deny)
}
}
func TestGetChannelPermissions_WithOverride(t *testing.T) {
database := openMigratedMemory(t)
chID, _ := database.CreateChannel("perms2", "text", "", "", 0)
// Insert an override directly.
_, err := database.Exec(
`INSERT INTO channel_overrides (channel_id, role_id, allow, deny) VALUES (?, ?, ?, ?)`,
chID, 4, int64(0x400), int64(0x200),
)
if err != nil {
t.Fatalf("insert override: %v", err)
}
allow, deny, err := database.GetChannelPermissions(chID, 4)
if err != nil {
t.Fatalf("GetChannelPermissions: %v", err)
}
if allow != 0x400 {
t.Errorf("allow = %d, want 0x400", allow)
}
if deny != 0x200 {
t.Errorf("deny = %d, want 0x200", deny)
}
}
Server/db/db.go
+6
View File
@@ -32,6 +32,12 @@ func Open(path string) (*DB, error) {
return nil, fmt.Errorf("pinging sqlite db: %w", err)
}
// In-memory databases are per-connection in SQLite; pin to one connection
// so all callers share the same in-memory state.
if path == ":memory:" {
sqlDB.SetMaxOpenConns(1)
}
// Enable WAL mode for better concurrent read performance.
if _, err := sqlDB.Exec("PRAGMA journal_mode=WAL;"); err != nil {
sqlDB.Close()
Server/db/message_queries.go
+289
View File
@@ -0,0 +1,289 @@
package db
import (
"database/sql"
"errors"
"fmt"
)
// CreateMessage inserts a new message and returns the assigned ID.
// Content should already be sanitized before calling this function.
func (d *DB) CreateMessage(channelID, userID int64, content string, replyTo *int64) (int64, error) {
res, err := d.sqlDB.Exec(
`INSERT INTO messages (channel_id, user_id, content, reply_to) VALUES (?, ?, ?, ?)`,
channelID, userID, content, replyTo,
)
if err != nil {
return 0, fmt.Errorf("CreateMessage: %w", err)
}
return res.LastInsertId()
}
// GetMessage returns the message with the given ID, or nil if not found.
// Soft-deleted messages are returned so callers can broadcast the deletion event.
func (d *DB) GetMessage(id int64) (*Message, error) {
row := d.sqlDB.QueryRow(
`SELECT id, channel_id, user_id, content, reply_to, edited_at, deleted, pinned, timestamp
FROM messages WHERE id = ?`,
id,
)
return scanMessage(row)
}
// GetMessages returns up to limit messages in a channel, ordered newest-first.
// When before > 0 only messages with id < before are returned (pagination).
func (d *DB) GetMessages(channelID, before int64, limit int) ([]MessageWithUser, error) {
var (
rows *sql.Rows
err error
)
if before > 0 {
rows, err = d.sqlDB.Query(
`SELECT m.id, m.channel_id, m.user_id, m.content, m.reply_to,
m.edited_at, m.deleted, m.pinned, m.timestamp,
u.username, u.avatar
FROM messages m JOIN users u ON m.user_id = u.id
WHERE m.channel_id = ? AND m.id < ? AND m.deleted = 0
ORDER BY m.id DESC LIMIT ?`,
channelID, before, limit,
)
} else {
rows, err = d.sqlDB.Query(
`SELECT m.id, m.channel_id, m.user_id, m.content, m.reply_to,
m.edited_at, m.deleted, m.pinned, m.timestamp,
u.username, u.avatar
FROM messages m JOIN users u ON m.user_id = u.id
WHERE m.channel_id = ? AND m.deleted = 0
ORDER BY m.id DESC LIMIT ?`,
channelID, limit,
)
}
if err != nil {
return nil, fmt.Errorf("GetMessages: %w", err)
}
defer rows.Close()
var msgs []MessageWithUser
for rows.Next() {
mwu, scanErr := scanMessageWithUser(rows)
if scanErr != nil {
return nil, fmt.Errorf("GetMessages scan: %w", scanErr)
}
msgs = append(msgs, mwu)
}
if rows.Err() != nil {
return nil, fmt.Errorf("GetMessages rows: %w", rows.Err())
}
if msgs == nil {
msgs = []MessageWithUser{}
}
return msgs, nil
}
// EditMessage updates the content and sets edited_at on the message.
// Returns an error if the message does not exist or userID does not match the owner.
func (d *DB) EditMessage(id, userID int64, content string) error {
msg, err := d.GetMessage(id)
if err != nil {
return err
}
if msg == nil {
return fmt.Errorf("EditMessage: message %d not found", id)
}
if msg.UserID != userID {
return fmt.Errorf("EditMessage: user %d does not own message %d", userID, id)
}
_, err = d.sqlDB.Exec(
`UPDATE messages SET content = ?, edited_at = datetime('now') WHERE id = ?`,
content, id,
)
if err != nil {
return fmt.Errorf("EditMessage: %w", err)
}
return nil
}
// DeleteMessage performs a soft delete (sets deleted=1) on the message.
// The calling user must be the message owner or ismod must be true.
func (d *DB) DeleteMessage(id, userID int64, ismod bool) error {
msg, err := d.GetMessage(id)
if err != nil {
return err
}
if msg == nil {
return fmt.Errorf("DeleteMessage: message %d not found", id)
}
if !ismod && msg.UserID != userID {
return fmt.Errorf("DeleteMessage: user %d does not own message %d", userID, id)
}
_, err = d.sqlDB.Exec(`UPDATE messages SET deleted = 1 WHERE id = ?`, id)
if err != nil {
return fmt.Errorf("DeleteMessage: %w", err)
}
return nil
}
// AddReaction inserts a reaction. Returns an error on duplicate (same user+emoji+message).
func (d *DB) AddReaction(messageID, userID int64, emoji string) error {
_, err := d.sqlDB.Exec(
`INSERT INTO reactions (message_id, user_id, emoji) VALUES (?, ?, ?)`,
messageID, userID, emoji,
)
if err != nil {
return fmt.Errorf("AddReaction: %w", err)
}
return nil
}
// RemoveReaction deletes a reaction. Returns an error if it does not exist.
func (d *DB) RemoveReaction(messageID, userID int64, emoji string) error {
res, err := d.sqlDB.Exec(
`DELETE FROM reactions WHERE message_id = ? AND user_id = ? AND emoji = ?`,
messageID, userID, emoji,
)
if err != nil {
return fmt.Errorf("RemoveReaction: %w", err)
}
n, _ := res.RowsAffected()
if n == 0 {
return fmt.Errorf("RemoveReaction: reaction not found")
}
return nil
}
// GetReactions returns aggregated reaction counts for a message.
// MeReacted is always false here (caller passes requesting userID if needed).
func (d *DB) GetReactions(messageID int64) ([]ReactionCount, error) {
rows, err := d.sqlDB.Query(
`SELECT emoji, COUNT(*) FROM reactions WHERE message_id = ? GROUP BY emoji`,
messageID,
)
if err != nil {
return nil, fmt.Errorf("GetReactions: %w", err)
}
defer rows.Close()
var counts []ReactionCount
for rows.Next() {
var rc ReactionCount
if scanErr := rows.Scan(&rc.Emoji, &rc.Count); scanErr != nil {
return nil, fmt.Errorf("GetReactions scan: %w", scanErr)
}
counts = append(counts, rc)
}
if rows.Err() != nil {
return nil, fmt.Errorf("GetReactions rows: %w", rows.Err())
}
if counts == nil {
counts = []ReactionCount{}
}
return counts, nil
}
// SearchMessages performs a full-text search against the messages_fts virtual table.
// When channelID is non-nil the search is scoped to that channel.
// Deleted messages are excluded from results.
func (d *DB) SearchMessages(query string, channelID *int64, limit int) ([]MessageSearchResult, error) {
var (
rows *sql.Rows
err error
)
if channelID != nil {
rows, err = d.sqlDB.Query(
`SELECT m.id, m.channel_id, c.name, u.username, m.content, m.timestamp
FROM messages_fts f
JOIN messages m ON f.rowid = m.id
JOIN channels c ON m.channel_id = c.id
JOIN users u ON m.user_id = u.id
WHERE messages_fts MATCH ? AND m.channel_id = ? AND m.deleted = 0
ORDER BY rank LIMIT ?`,
query, *channelID, limit,
)
} else {
rows, err = d.sqlDB.Query(
`SELECT m.id, m.channel_id, c.name, u.username, m.content, m.timestamp
FROM messages_fts f
JOIN messages m ON f.rowid = m.id
JOIN channels c ON m.channel_id = c.id
JOIN users u ON m.user_id = u.id
WHERE messages_fts MATCH ? AND m.deleted = 0
ORDER BY rank LIMIT ?`,
query, limit,
)
}
if err != nil {
return nil, fmt.Errorf("SearchMessages: %w", err)
}
defer rows.Close()
var results []MessageSearchResult
for rows.Next() {
var r MessageSearchResult
if scanErr := rows.Scan(&r.MessageID, &r.ChannelID, &r.ChannelName, &r.Username, &r.Content, &r.Timestamp); scanErr != nil {
return nil, fmt.Errorf("SearchMessages scan: %w", scanErr)
}
results = append(results, r)
}
if rows.Err() != nil {
return nil, fmt.Errorf("SearchMessages rows: %w", rows.Err())
}
if results == nil {
results = []MessageSearchResult{}
}
return results, nil
}
// UpdateReadState upserts the read state for a user in a channel.
func (d *DB) UpdateReadState(userID, channelID, lastReadMessageID int64) error {
_, err := d.sqlDB.Exec(
`INSERT INTO read_states (user_id, channel_id, last_message_id)
VALUES (?, ?, ?)
ON CONFLICT(user_id, channel_id) DO UPDATE SET last_message_id = excluded.last_message_id`,
userID, channelID, lastReadMessageID,
)
if err != nil {
return fmt.Errorf("UpdateReadState: %w", err)
}
return nil
}
// ─── helpers ──────────────────────────────────────────────────────────────────
// scanMessage scans a single message from *sql.Row.
func scanMessage(row *sql.Row) (*Message, error) {
m := &Message{}
var deleted, pinned int
err := row.Scan(
&m.ID, &m.ChannelID, &m.UserID, &m.Content, &m.ReplyTo,
&m.EditedAt, &deleted, &pinned, &m.Timestamp,
)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("scanMessage: %w", err)
}
m.Deleted = deleted != 0
m.Pinned = pinned != 0
return m, nil
}
// scanMessageWithUser scans a MessageWithUser from *sql.Rows.
func scanMessageWithUser(rows *sql.Rows) (MessageWithUser, error) {
var mwu MessageWithUser
var deleted, pinned int
err := rows.Scan(
&mwu.ID, &mwu.ChannelID, &mwu.UserID, &mwu.Content, &mwu.ReplyTo,
&mwu.EditedAt, &deleted, &pinned, &mwu.Timestamp,
&mwu.Username, &mwu.Avatar,
)
if err != nil {
return MessageWithUser{}, err
}
mwu.Deleted = deleted != 0
mwu.Pinned = pinned != 0
return mwu, nil
}
Server/db/message_queries_test.go
+523
View File
@@ -0,0 +1,523 @@
package db_test
import (
"testing"
"github.com/owncord/server/db"
)
// seedUser inserts a minimal test user and returns its ID.
func seedUser(t *testing.T, database *db.DB, username string) int64 {
t.Helper()
id, err := database.CreateUser(username, "hash", 4)
if err != nil {
t.Fatalf("seedUser(%q): %v", username, err)
}
return id
}
// seedChannel inserts a minimal test channel and returns its ID.
func seedChannel(t *testing.T, database *db.DB, name string) int64 {
t.Helper()
id, err := database.CreateChannel(name, "text", "", "", 0)
if err != nil {
t.Fatalf("seedChannel(%q): %v", name, err)
}
return id
}
// ─── CreateMessage ────────────────────────────────────────────────────────────
func TestCreateMessage_ReturnsID(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "alice")
chID := seedChannel(t, database, "general")
id, err := database.CreateMessage(chID, userID, "hello", nil)
if err != nil {
t.Fatalf("CreateMessage: %v", err)
}
if id <= 0 {
t.Errorf("expected positive ID, got %d", id)
}
}
func TestCreateMessage_WithReplyTo(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "alice")
chID := seedChannel(t, database, "general")
parentID, _ := database.CreateMessage(chID, userID, "parent", nil)
replyID, err := database.CreateMessage(chID, userID, "reply", &parentID)
if err != nil {
t.Fatalf("CreateMessage with reply: %v", err)
}
msg, _ := database.GetMessage(replyID)
if msg.ReplyTo == nil || *msg.ReplyTo != parentID {
t.Errorf("ReplyTo = %v, want %d", msg.ReplyTo, parentID)
}
}
func TestCreateMessage_ContentPreserved(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "bob")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, userID, "test content", nil)
msg, _ := database.GetMessage(id)
if msg.Content != "test content" {
t.Errorf("Content = %q, want 'test content'", msg.Content)
}
}
// ─── GetMessage ───────────────────────────────────────────────────────────────
func TestGetMessage_NotFound(t *testing.T) {
database := openMigratedMemory(t)
msg, err := database.GetMessage(9999)
if err != nil {
t.Fatalf("GetMessage: %v", err)
}
if msg != nil {
t.Error("expected nil for non-existent message")
}
}
func TestGetMessage_Fields(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "carol")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, userID, "hello world", nil)
msg, err := database.GetMessage(id)
if err != nil {
t.Fatalf("GetMessage: %v", err)
}
if msg == nil {
t.Fatal("expected message, got nil")
}
if msg.ChannelID != chID {
t.Errorf("ChannelID = %d, want %d", msg.ChannelID, chID)
}
if msg.UserID != userID {
t.Errorf("UserID = %d, want %d", msg.UserID, userID)
}
if msg.Deleted {
t.Error("expected Deleted=false for new message")
}
if msg.Pinned {
t.Error("expected Pinned=false for new message")
}
if msg.EditedAt != nil {
t.Error("expected EditedAt=nil for new message")
}
}
// ─── GetMessages ──────────────────────────────────────────────────────────────
func TestGetMessages_EmptyChannel(t *testing.T) {
database := openMigratedMemory(t)
chID := seedChannel(t, database, "empty")
msgs, err := database.GetMessages(chID, 0, 50)
if err != nil {
t.Fatalf("GetMessages: %v", err)
}
if len(msgs) != 0 {
t.Errorf("expected 0 messages, got %d", len(msgs))
}
}
func TestGetMessages_ReturnsMessages(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "dave")
chID := seedChannel(t, database, "ch")
for i := 0; i < 3; i++ {
_, err := database.CreateMessage(chID, userID, "msg", nil)
if err != nil {
t.Fatalf("CreateMessage %d: %v", i, err)
}
}
msgs, err := database.GetMessages(chID, 0, 50)
if err != nil {
t.Fatalf("GetMessages: %v", err)
}
if len(msgs) != 3 {
t.Errorf("expected 3 messages, got %d", len(msgs))
}
}
func TestGetMessages_LimitRespected(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "eve")
chID := seedChannel(t, database, "ch")
for i := 0; i < 10; i++ {
_, _ = database.CreateMessage(chID, userID, "msg", nil)
}
msgs, _ := database.GetMessages(chID, 0, 5)
if len(msgs) != 5 {
t.Errorf("expected 5 messages (limit), got %d", len(msgs))
}
}
func TestGetMessages_BeforePagination(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "frank")
chID := seedChannel(t, database, "ch")
var ids []int64
for i := 0; i < 5; i++ {
id, _ := database.CreateMessage(chID, userID, "msg", nil)
ids = append(ids, id)
}
// Get messages before the 4th message (should get 3 messages: ids 0,1,2).
msgs, _ := database.GetMessages(chID, ids[3], 50)
if len(msgs) != 3 {
t.Errorf("expected 3 messages before id %d, got %d", ids[3], len(msgs))
}
}
func TestGetMessages_IncludesUsername(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "grace")
chID := seedChannel(t, database, "ch")
_, _ = database.CreateMessage(chID, userID, "hi", nil)
msgs, _ := database.GetMessages(chID, 0, 50)
if len(msgs) == 0 {
t.Fatal("expected messages")
}
if msgs[0].Username != "grace" {
t.Errorf("Username = %q, want 'grace'", msgs[0].Username)
}
}
// ─── EditMessage ──────────────────────────────────────────────────────────────
func TestEditMessage_OwnerCanEdit(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "henry")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, userID, "original", nil)
if err := database.EditMessage(id, userID, "updated"); err != nil {
t.Fatalf("EditMessage: %v", err)
}
msg, _ := database.GetMessage(id)
if msg.Content != "updated" {
t.Errorf("Content = %q, want 'updated'", msg.Content)
}
if msg.EditedAt == nil {
t.Error("EditedAt should be set after edit")
}
}
func TestEditMessage_NonOwnerCannotEdit(t *testing.T) {
database := openMigratedMemory(t)
ownerID := seedUser(t, database, "ivan")
otherID := seedUser(t, database, "julia")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, ownerID, "original", nil)
err := database.EditMessage(id, otherID, "hacked")
if err == nil {
t.Error("EditMessage by non-owner should return error")
}
}
func TestEditMessage_NotFound(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "kim")
err := database.EditMessage(9999, userID, "x")
if err == nil {
t.Error("EditMessage non-existent should return error")
}
}
// ─── DeleteMessage ────────────────────────────────────────────────────────────
func TestDeleteMessage_OwnerCanDelete(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "larry")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, userID, "bye", nil)
if err := database.DeleteMessage(id, userID, false); err != nil {
t.Fatalf("DeleteMessage: %v", err)
}
msg, _ := database.GetMessage(id)
if msg == nil {
t.Fatal("soft-deleted message should still exist in DB")
}
if !msg.Deleted {
t.Error("expected Deleted=true after soft delete")
}
}
func TestDeleteMessage_ContentPreservedAfterSoftDelete(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "mia")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, userID, "sensitive", nil)
_ = database.DeleteMessage(id, userID, false)
msg, _ := database.GetMessage(id)
// Content preserved for broadcast (soft delete only flags deleted=1).
if msg.Content == "" {
t.Error("content should be preserved on soft delete for broadcast purposes")
}
}
func TestDeleteMessage_NonOwnerBlockedWithoutMod(t *testing.T) {
database := openMigratedMemory(t)
ownerID := seedUser(t, database, "nate")
otherID := seedUser(t, database, "olivia")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, ownerID, "msg", nil)
err := database.DeleteMessage(id, otherID, false)
if err == nil {
t.Error("DeleteMessage by non-owner non-mod should return error")
}
}
func TestDeleteMessage_ModCanDeleteAny(t *testing.T) {
database := openMigratedMemory(t)
ownerID := seedUser(t, database, "pete")
modID := seedUser(t, database, "quinn")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, ownerID, "msg", nil)
if err := database.DeleteMessage(id, modID, true); err != nil {
t.Fatalf("DeleteMessage by mod: %v", err)
}
msg, _ := database.GetMessage(id)
if !msg.Deleted {
t.Error("expected Deleted=true after mod delete")
}
}
func TestDeleteMessage_NotFound(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "rachel")
err := database.DeleteMessage(9999, userID, true)
if err == nil {
t.Error("DeleteMessage non-existent should return error")
}
}
// ─── Reactions ────────────────────────────────────────────────────────────────
func TestAddReaction_Success(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "sam")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, userID, "hi", nil)
if err := database.AddReaction(msgID, userID, "👍"); err != nil {
t.Fatalf("AddReaction: %v", err)
}
}
func TestAddReaction_UniqueConstraint(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "tina")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, userID, "hi", nil)
_ = database.AddReaction(msgID, userID, "❤️")
err := database.AddReaction(msgID, userID, "❤️")
if err == nil {
t.Error("adding duplicate reaction should return error")
}
}
func TestRemoveReaction_Success(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "uma")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, userID, "hi", nil)
_ = database.AddReaction(msgID, userID, "😂")
if err := database.RemoveReaction(msgID, userID, "😂"); err != nil {
t.Fatalf("RemoveReaction: %v", err)
}
}
func TestRemoveReaction_NotFound(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "victor")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, userID, "hi", nil)
err := database.RemoveReaction(msgID, userID, "🔥")
if err == nil {
t.Error("removing non-existent reaction should return error")
}
}
func TestGetReactions_Empty(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "wendy")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, userID, "hi", nil)
counts, err := database.GetReactions(msgID)
if err != nil {
t.Fatalf("GetReactions: %v", err)
}
if len(counts) != 0 {
t.Errorf("expected 0 reactions, got %d", len(counts))
}
}
func TestGetReactions_Counts(t *testing.T) {
database := openMigratedMemory(t)
u1 := seedUser(t, database, "xavier")
u2 := seedUser(t, database, "yvonne")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, u1, "hi", nil)
_ = database.AddReaction(msgID, u1, "👍")
_ = database.AddReaction(msgID, u2, "👍")
_ = database.AddReaction(msgID, u1, "❤️")
counts, _ := database.GetReactions(msgID)
if len(counts) != 2 {
t.Fatalf("expected 2 emoji types, got %d", len(counts))
}
for _, rc := range counts {
switch rc.Emoji {
case "👍":
if rc.Count != 2 {
t.Errorf("👍 count = %d, want 2", rc.Count)
}
case "❤️":
if rc.Count != 1 {
t.Errorf("❤️ count = %d, want 1", rc.Count)
}
default:
t.Errorf("unexpected emoji %q", rc.Emoji)
}
}
}
// ─── SearchMessages ───────────────────────────────────────────────────────────
func TestSearchMessages_FindsMatch(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "zara")
chID := seedChannel(t, database, "searchch")
_, _ = database.CreateMessage(chID, userID, "hello world fts test", nil)
_, _ = database.CreateMessage(chID, userID, "unrelated content here", nil)
results, err := database.SearchMessages("hello", nil, 10)
if err != nil {
t.Fatalf("SearchMessages: %v", err)
}
if len(results) != 1 {
t.Errorf("expected 1 result, got %d", len(results))
}
if results[0].Content != "hello world fts test" {
t.Errorf("Content = %q, want 'hello world fts test'", results[0].Content)
}
}
func TestSearchMessages_FilterByChannel(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "adam")
ch1 := seedChannel(t, database, "ch1")
ch2 := seedChannel(t, database, "ch2")
_, _ = database.CreateMessage(ch1, userID, "needle in channel 1", nil)
_, _ = database.CreateMessage(ch2, userID, "needle in channel 2", nil)
results, _ := database.SearchMessages("needle", &ch1, 10)
if len(results) != 1 {
t.Errorf("expected 1 result in ch1, got %d", len(results))
}
if results[0].ChannelID != ch1 {
t.Errorf("ChannelID = %d, want %d", results[0].ChannelID, ch1)
}
}
func TestSearchMessages_NoResults(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "beth")
chID := seedChannel(t, database, "ch")
_, _ = database.CreateMessage(chID, userID, "hello there", nil)
results, _ := database.SearchMessages("xyzzy", nil, 10)
if len(results) != 0 {
t.Errorf("expected 0 results, got %d", len(results))
}
}
func TestSearchMessages_LimitRespected(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "carl")
chID := seedChannel(t, database, "ch")
for i := 0; i < 5; i++ {
_, _ = database.CreateMessage(chID, userID, "searchable keyword content", nil)
}
results, _ := database.SearchMessages("keyword", nil, 3)
if len(results) != 3 {
t.Errorf("expected 3 results (limit), got %d", len(results))
}
}
func TestSearchMessages_DeletedNotReturned(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "diana")
chID := seedChannel(t, database, "ch")
id, _ := database.CreateMessage(chID, userID, "vanishing keyword message", nil)
_ = database.DeleteMessage(id, userID, false)
results, _ := database.SearchMessages("vanishing", nil, 10)
if len(results) != 0 {
t.Errorf("expected 0 results (deleted excluded), got %d", len(results))
}
}
// ─── UpdateReadState ──────────────────────────────────────────────────────────
func TestUpdateReadState_Upsert(t *testing.T) {
database := openMigratedMemory(t)
userID := seedUser(t, database, "ella")
chID := seedChannel(t, database, "ch")
msgID, _ := database.CreateMessage(chID, userID, "msg", nil)
if err := database.UpdateReadState(userID, chID, msgID); err != nil {
t.Fatalf("UpdateReadState: %v", err)
}
// Update again with higher message ID — should not error.
msgID2, _ := database.CreateMessage(chID, userID, "msg2", nil)
if err := database.UpdateReadState(userID, chID, msgID2); err != nil {
t.Fatalf("UpdateReadState second call: %v", err)
}
}
Server/db/models.go
+50
View File
@@ -52,5 +52,55 @@ type Role struct {
IsDefault bool
}
// Channel represents a row in the channels table.
type Channel struct {
ID int64
Name string
Type string
Category string
Topic string
Position int
SlowMode int
Archived bool
CreatedAt string
}
// Message represents a row in the messages table.
type Message struct {
ID int64
ChannelID int64
UserID int64
Content string
ReplyTo *int64
EditedAt *string
Deleted bool
Pinned bool
Timestamp string
}
// MessageWithUser joins a Message with the author's public fields.
type MessageWithUser struct {
Message
Username string
Avatar *string
}
// ReactionCount is an aggregated reaction count for a single emoji.
type ReactionCount struct {
Emoji string
Count int
MeReacted bool
}
// MessageSearchResult is a row returned by the FTS5 message search.
type MessageSearchResult struct {
MessageID int64
ChannelID int64
ChannelName string
Username string
Content string
Timestamp string
}
// sessionTTL is the duration a session remains valid after creation.
const sessionTTL = 30 * 24 * time.Hour
Server/go.mod
+14 -10
View File
@@ -2,34 +2,38 @@ module github.com/owncord/server
go 1.25.0
require (
github.com/go-chi/chi/v5 v5.2.5
github.com/knadh/koanf/parsers/yaml v1.1.0
github.com/knadh/koanf/providers/env v1.1.0
github.com/knadh/koanf/providers/file v1.2.1
github.com/knadh/koanf/providers/structs v1.0.0
github.com/knadh/koanf/v2 v2.3.3
github.com/microcosm-cc/bluemonday v1.0.27
go.yaml.in/yaml/v3 v3.0.3
golang.org/x/crypto v0.49.0
modernc.org/sqlite v1.46.1
nhooyr.io/websocket v1.8.17
)
require (
github.com/aymerick/douceur v0.2.0 // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/fatih/structs v1.1.0 // indirect
github.com/fsnotify/fsnotify v1.9.0 // indirect
github.com/go-chi/chi/v5 v5.2.5 // indirect
github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/css v1.0.1 // indirect
github.com/knadh/koanf/maps v0.1.2 // indirect
github.com/knadh/koanf/parsers/yaml v1.1.0 // indirect
github.com/knadh/koanf/providers/env v1.1.0 // indirect
github.com/knadh/koanf/providers/file v1.2.1 // indirect
github.com/knadh/koanf/providers/structs v1.0.0 // indirect
github.com/knadh/koanf/v2 v2.3.3 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/microcosm-cc/bluemonday v1.0.27 // indirect
github.com/mitchellh/copystructure v1.2.0 // indirect
github.com/mitchellh/reflectwalk v1.0.2 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
go.yaml.in/yaml/v3 v3.0.3 // indirect
golang.org/x/crypto v0.49.0 // indirect
golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
golang.org/x/net v0.51.0 // indirect
golang.org/x/sys v0.42.0 // indirect
modernc.org/libc v1.67.6 // indirect
modernc.org/mathutil v1.7.1 // indirect
modernc.org/memory v1.11.0 // indirect
modernc.org/sqlite v1.46.1 // indirect
)
Server/go.sum
+46 -4
View File
@@ -1,5 +1,7 @@
github.com/aymerick/douceur v0.2.0 h1:Mv+mAeH1Q+n9Fr+oyamOlAkUNPWPlA8PPGR0QAaYuPk=
github.com/aymerick/douceur v0.2.0/go.mod h1:wlT5vV2O3h55X9m7iVYN0TBM0NH/MmbLnd30/FjWUq4=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/fatih/structs v1.1.0 h1:Q7juDM0QtcnhCpeyLGQKyg4TOIghuNXrkL32pHAUMxo=
@@ -10,10 +12,14 @@ github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug=
github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0=
github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs=
github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/css v1.0.1 h1:ntNaBIghp6JmvWnxbZKANoLyuXTPZ4cAMlo6RyhlbO8=
github.com/gorilla/css v1.0.1/go.mod h1:BvnYkspnSzMmwRK+b8/xgNPLiIuNZr6vbZBTPQ2A3b0=
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/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo=
github.com/knadh/koanf/maps v0.1.2/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI=
github.com/knadh/koanf/parsers/yaml v1.1.0 h1:3ltfm9ljprAHt4jxgeYLlFPmUaunuCgu1yILuTXRdM4=
@@ -26,6 +32,10 @@ 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/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/microcosm-cc/bluemonday v1.0.27 h1:MpEUotklkwCSLeH+Qdx1VJgNqLlpY2KXwXFM08ygZfk=
@@ -36,28 +46,60 @@ github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zx
github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
go.yaml.in/yaml/v3 v3.0.3 h1:bXOww4E/J3f66rav3pX3m8w6jDE4knZjGOw8b5Y6iNE=
go.yaml.in/yaml/v3 v3.0.3/go.mod h1:tBHosrYAkRZjRAOREWbDnBXUf08JOwYq++0QNwQiWzI=
golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2wtOnGAHZWCHUM4KGzY=
golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546/go.mod h1:j/pmGrbnkbPtQfxEe5D0VQhZC6qKbfKifgD0oM7sR70=
golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA=
golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w=
golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ=
golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs=
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/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
modernc.org/cc/v4 v4.27.1 h1:9W30zRlYrefrDV2JE2O8VDtJ1yPGownxciz5rrbQZis=
modernc.org/cc/v4 v4.27.1/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0=
modernc.org/ccgo/v4 v4.30.1 h1:4r4U1J6Fhj98NKfSjnPUN7Ze2c6MnAdL0hWw6+LrJpc=
modernc.org/ccgo/v4 v4.30.1/go.mod h1:bIOeI1JL54Utlxn+LwrFyjCx2n2RDiYEaJVSrgdrRfM=
modernc.org/fileutil v1.3.40 h1:ZGMswMNc9JOCrcrakF1HrvmergNLAmxOPjizirpfqBA=
modernc.org/fileutil v1.3.40/go.mod h1:HxmghZSZVAz/LXcMNwZPA/DRrQZEVP9VX0V4LQGQFOc=
modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
modernc.org/gc/v3 v3.1.1 h1:k8T3gkXWY9sEiytKhcgyiZ2L0DTyCQ/nvX+LoCljoRE=
modernc.org/gc/v3 v3.1.1/go.mod h1:HFK/6AGESC7Ex+EZJhJ2Gni6cTaYpSMmU/cT9RmlfYY=
modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks=
modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI=
modernc.org/libc v1.67.6 h1:eVOQvpModVLKOdT+LvBPjdQqfrZq+pC39BygcT+E7OI=
modernc.org/libc v1.67.6/go.mod h1:JAhxUVlolfYDErnwiqaLvUqc8nfb2r6S6slAgZOnaiE=
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8=
modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
modernc.org/sqlite v1.46.1 h1:eFJ2ShBLIEnUWlLy12raN0Z1plqmFX9Qe3rjQTKt6sU=
modernc.org/sqlite v1.46.1/go.mod h1:CzbrU2lSB1DKUusvwGz7rqEKIq+NUd8GWuBBZDs9/nA=
modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y=
nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c=
Server/ws/client.go
+80
View File
@@ -0,0 +1,80 @@
package ws
import (
"sync"
"github.com/owncord/server/db"
)
const sendBufSize = 256
// Client represents a single authenticated WebSocket connection.
// The underlying transport (conn) is set by ServeWS; in tests it remains nil.
type Client struct {
hub *Hub
conn wsConn // interface — nil in unit tests
userID int64
user *db.User
channelID int64 // currently viewed channel for channel-scoped broadcasts
send chan []byte
mu sync.Mutex
}
// wsConn is the subset of nhooyr.io/websocket.Conn used by writePump/readPump.
// Defining it as an interface lets us avoid importing nhooyr.io/websocket here,
// keeping the core hub logic free from that dependency during unit tests.
type wsConn interface {
// intentionally empty — methods used only in serve.go/client_pump.go
}
// newClient creates a real client wrapping a WebSocket connection (set by serve.go).
func newClient(hub *Hub, conn wsConn, user *db.User) *Client {
return &Client{
hub: hub,
conn: conn,
userID: user.ID,
user: user,
send: make(chan []byte, sendBufSize),
}
}
// NewTestClient creates a client with a caller-supplied send channel.
// Intended for unit tests only — conn is nil.
func NewTestClient(hub *Hub, userID int64, send chan []byte) *Client {
return &Client{
hub: hub,
userID: userID,
send: send,
}
}
// NewTestClientWithChannel creates a test client subscribed to a specific channel.
func NewTestClientWithChannel(hub *Hub, userID, channelID int64, send chan []byte) *Client {
return &Client{
hub: hub,
userID: userID,
channelID: channelID,
send: send,
}
}
// NewTestClientWithUser creates a test client with an authenticated user record set.
// Use this when tests need the client to pass permission checks.
func NewTestClientWithUser(hub *Hub, user *db.User, channelID int64, send chan []byte) *Client {
return &Client{
hub: hub,
userID: user.ID,
user: user,
channelID: channelID,
send: send,
}
}
// sendMsg queues a message to this client's send buffer without blocking.
func (c *Client) sendMsg(msg []byte) {
select {
case c.send <- msg:
default:
// Buffer full — drop rather than block the hub.
}
}
Server/ws/handlers.go
+358
View File
@@ -0,0 +1,358 @@
package ws
import (
"encoding/json"
"fmt"
"log/slog"
"time"
"github.com/microcosm-cc/bluemonday"
)
// Permission bits (from SCHEMA.md).
const (
permReadMessages = int64(0x0400)
permSendMessages = int64(0x0800)
permManageMessages = int64(0x2000)
permAddReactions = int64(0x0040)
permAdministrator = int64(0x40000000)
)
// Rate limit windows.
const (
chatRateLimit = 10
chatWindow = time.Second
typingRateLimit = 1
typingWindow = 3 * time.Second
presenceRateLimit = 1
presenceWindow = 10 * time.Second
reactionRateLimit = 5
reactionWindow = time.Second
)
var sanitizer = bluemonday.StrictPolicy()
// HandleMessageForTest dispatches a raw WebSocket message from client c.
// Exported so ws_test package can invoke it directly without a real connection.
func (h *Hub) HandleMessageForTest(c *Client, raw []byte) {
h.handleMessage(c, raw)
}
// handleMessage parses the envelope and dispatches to the appropriate handler.
func (h *Hub) handleMessage(c *Client, raw []byte) {
var env envelope
if err := json.Unmarshal(raw, &env); err != nil {
slog.Warn("ws handleMessage invalid JSON", "user_id", c.userID, "err", err)
c.sendMsg(buildErrorMsg("INVALID_JSON", "message must be valid JSON"))
return
}
switch env.Type {
case "chat_send":
h.handleChatSend(c, env.ID, env.Payload)
case "chat_edit":
h.handleChatEdit(c, env.ID, env.Payload)
case "chat_delete":
h.handleChatDelete(c, env.ID, env.Payload)
case "reaction_add":
h.handleReaction(c, true, env.Payload)
case "reaction_remove":
h.handleReaction(c, false, env.Payload)
case "typing_start":
h.handleTyping(c, env.Payload)
case "presence_update":
h.handlePresence(c, env.Payload)
default:
slog.Warn("ws handleMessage unknown type", "type", env.Type, "user_id", c.userID)
c.sendMsg(buildErrorMsg("UNKNOWN_TYPE", fmt.Sprintf("unknown message type: %s", env.Type)))
}
}
// handleChatSend processes a chat_send message.
func (h *Hub) handleChatSend(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(buildErrorMsg("RATE_LIMITED", "too many messages"))
return
}
var p struct {
ChannelID json.Number `json:"channel_id"`
Content string `json:"content"`
ReplyTo *int64 `json:"reply_to"`
Attachments []string `json:"attachments"`
}
if err := json.Unmarshal(payload, &p); err != nil {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "invalid chat_send payload"))
return
}
channelID, err := p.ChannelID.Int64()
if err != nil || channelID <= 0 {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "channel_id must be a positive integer"))
return
}
// Check channel exists.
ch, err := h.db.GetChannel(channelID)
if err != nil || ch == nil {
c.sendMsg(buildErrorMsg("NOT_FOUND", "channel not found"))
return
}
// Permission check.
if !h.hasChannelPerm(c, channelID, permReadMessages|permSendMessages) {
c.sendMsg(buildErrorMsg("FORBIDDEN", "missing SEND_MESSAGES permission"))
return
}
// Sanitize content.
content := sanitizer.Sanitize(p.Content)
if content == "" {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "message content cannot be empty"))
return
}
// Persist.
msgID, err := h.db.CreateMessage(channelID, c.userID, content, p.ReplyTo)
if err != nil {
slog.Error("ws handleChatSend CreateMessage", "err", err)
c.sendMsg(buildErrorMsg("INTERNAL", "failed to save message"))
return
}
// Retrieve to get timestamp.
msg, err := h.db.GetMessage(msgID)
if err != nil || msg == nil {
slog.Error("ws handleChatSend GetMessage after create", "err", err)
c.sendMsg(buildErrorMsg("INTERNAL", "failed to retrieve message"))
return
}
var username string
var avatar *string
if c.user != nil {
username = c.user.Username
avatar = c.user.Avatar
}
// Ack sender.
c.sendMsg(buildChatSendOK(reqID, msgID, msg.Timestamp))
// Broadcast to channel.
broadcast := buildChatMessage(msgID, channelID, c.userID, username, avatar, content, msg.Timestamp, p.ReplyTo)
h.BroadcastToChannel(channelID, broadcast)
}
// handleChatEdit processes a chat_edit message.
func (h *Hub) handleChatEdit(c *Client, _ string, payload json.RawMessage) {
var p struct {
MessageID json.Number `json:"message_id"`
Content string `json:"content"`
}
if err := json.Unmarshal(payload, &p); err != nil {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "invalid chat_edit payload"))
return
}
msgID, err := p.MessageID.Int64()
if err != nil || msgID <= 0 {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "message_id must be positive integer"))
return
}
content := sanitizer.Sanitize(p.Content)
if content == "" {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "content cannot be empty"))
return
}
// EditMessage checks ownership internally.
if err := h.db.EditMessage(msgID, c.userID, content); err != nil {
c.sendMsg(buildErrorMsg("FORBIDDEN", "cannot edit this message"))
return
}
msg, err := h.db.GetMessage(msgID)
if err != nil || msg == nil {
slog.Error("ws handleChatEdit GetMessage after edit", "err", err)
return
}
editedAt := ""
if msg.EditedAt != nil {
editedAt = *msg.EditedAt
}
h.BroadcastToChannel(msg.ChannelID, buildChatEdited(msgID, msg.ChannelID, content, editedAt))
}
// handleChatDelete processes a chat_delete message.
func (h *Hub) handleChatDelete(c *Client, _ string, payload json.RawMessage) {
var p struct {
MessageID json.Number `json:"message_id"`
}
if err := json.Unmarshal(payload, &p); err != nil {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "invalid chat_delete payload"))
return
}
msgID, err := p.MessageID.Int64()
if err != nil || msgID <= 0 {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "message_id must be positive integer"))
return
}
msg, err := h.db.GetMessage(msgID)
if err != nil || msg == nil {
c.sendMsg(buildErrorMsg("NOT_FOUND", "message not found"))
return
}
isMod := h.hasChannelPerm(c, msg.ChannelID, permManageMessages)
if err := h.db.DeleteMessage(msgID, c.userID, isMod); err != nil {
c.sendMsg(buildErrorMsg("FORBIDDEN", "cannot delete this message"))
return
}
h.BroadcastToChannel(msg.ChannelID, buildChatDeleted(msgID, msg.ChannelID))
}
// handleReaction processes reaction_add and reaction_remove messages.
func (h *Hub) handleReaction(c *Client, add bool, payload json.RawMessage) {
ratKey := fmt.Sprintf("reaction:%d", c.userID)
if !h.limiter.Allow(ratKey, reactionRateLimit, reactionWindow) {
c.sendMsg(buildErrorMsg("RATE_LIMITED", "too many reactions"))
return
}
var p struct {
MessageID json.Number `json:"message_id"`
Emoji string `json:"emoji"`
}
if err := json.Unmarshal(payload, &p); err != nil {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "invalid reaction payload"))
return
}
msgID, err := p.MessageID.Int64()
if err != nil || msgID <= 0 {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "message_id must be positive integer"))
return
}
if p.Emoji == "" {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "emoji cannot be empty"))
return
}
msg, err := h.db.GetMessage(msgID)
if err != nil || msg == nil {
c.sendMsg(buildErrorMsg("NOT_FOUND", "message not found"))
return
}
if !h.hasChannelPerm(c, msg.ChannelID, permAddReactions) {
c.sendMsg(buildErrorMsg("FORBIDDEN", "missing ADD_REACTIONS permission"))
return
}
action := "add"
if add {
err = h.db.AddReaction(msgID, c.userID, p.Emoji)
} else {
action = "remove"
err = h.db.RemoveReaction(msgID, c.userID, p.Emoji)
}
if err != nil {
c.sendMsg(buildErrorMsg("CONFLICT", err.Error()))
return
}
h.BroadcastToChannel(msg.ChannelID, buildReactionUpdate(msgID, msg.ChannelID, c.userID, p.Emoji, action))
}
// handleTyping processes a typing_start message.
func (h *Hub) handleTyping(c *Client, payload json.RawMessage) {
channelID, err := parseChannelID(payload)
if err != nil || channelID <= 0 {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "channel_id must be positive integer"))
return
}
ratKey := fmt.Sprintf("typing:%d:%d", c.userID, channelID)
if !h.limiter.Allow(ratKey, typingRateLimit, typingWindow) {
return // silently drop; no error for typing throttle
}
var username string
if c.user != nil {
username = c.user.Username
}
// Broadcast to channel, excluding sender.
h.broadcastExclude(channelID, c.userID, buildTypingMsg(channelID, c.userID, username))
}
// handlePresence processes a presence_update message.
func (h *Hub) handlePresence(c *Client, payload json.RawMessage) {
ratKey := fmt.Sprintf("presence:%d", c.userID)
if !h.limiter.Allow(ratKey, presenceRateLimit, presenceWindow) {
c.sendMsg(buildErrorMsg("RATE_LIMITED", "too many presence updates"))
return
}
var p struct {
Status string `json:"status"`
}
if err := json.Unmarshal(payload, &p); err != nil {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "invalid presence_update payload"))
return
}
validStatuses := map[string]bool{"online": true, "idle": true, "dnd": true, "offline": true}
if !validStatuses[p.Status] {
c.sendMsg(buildErrorMsg("BAD_REQUEST", "status must be online|idle|dnd|offline"))
return
}
if err := h.db.UpdateUserStatus(c.userID, p.Status); err != nil {
slog.Error("ws handlePresence UpdateUserStatus", "err", err)
}
h.BroadcastToAll(buildPresenceMsg(c.userID, p.Status))
}
// hasChannelPerm reports whether the client's role has all the given permission bits.
// The ADMINISTRATOR bit bypasses all checks.
func (h *Hub) hasChannelPerm(c *Client, channelID int64, perm int64) bool {
if c.user == nil {
return false
}
role, err := h.db.GetRoleByID(c.user.RoleID)
if err != nil || role == nil {
return false
}
if role.Permissions&permAdministrator != 0 {
return true
}
// Check channel overrides.
allow, deny, err := h.db.GetChannelPermissions(channelID, role.ID)
if err != nil {
return false
}
effective := (role.Permissions | allow) &^ deny
return effective&perm == perm
}
// broadcastExclude sends msg to all channel members except excludeUserID.
func (h *Hub) broadcastExclude(channelID, excludeUserID int64, msg []byte) {
h.mu.RLock()
defer h.mu.RUnlock()
for uid, c := range h.clients {
if uid == excludeUserID {
continue
}
if channelID != 0 && c.channelID != channelID {
continue
}
select {
case c.send <- msg:
default:
}
}
}
Server/ws/hub.go
+126 -29
View File
@@ -1,45 +1,142 @@
// Package ws provides the WebSocket hub for the OwnCord server.
// Full implementation follows in Phase 4 (Real-Time Chat Features).
// Package ws provides the WebSocket hub and client management for OwnCord.
package ws
import (
"context"
"log/slog"
"net/http"
"sync"
"github.com/owncord/server/auth"
"github.com/owncord/server/db"
)
// Hub manages active WebSocket client connections.
// It is the central message routing point for all connected clients.
// broadcastMsg is an internal message queued for delivery.
type broadcastMsg struct {
channelID int64 // 0 = send to all connected clients
senderID int64 // reserved for future exclude-sender logic
msg []byte
}
// Hub manages all active WebSocket clients and routes messages between them.
// All exported methods are safe to call from multiple goroutines.
type Hub struct {
mu sync.RWMutex
clients map[*Client]struct{}
log *slog.Logger
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{}
}
// Client represents a single WebSocket connection.
// Full implementation in Phase 4.
type Client struct {
UserID int64
}
// NewHub creates a new Hub with the given logger.
func NewHub(log *slog.Logger) *Hub {
// NewHub creates a Hub ready to be started with Run.
func NewHub(database *db.DB, limiter *auth.RateLimiter) *Hub {
return &Hub{
clients: make(map[*Client]struct{}),
log: log,
clients: make(map[int64]*Client),
db: database,
limiter: limiter,
broadcast: make(chan broadcastMsg, 256),
register: make(chan *Client, 32),
unregister: make(chan *Client, 32),
stop: make(chan struct{}),
}
}
// Run starts the hub's message dispatch loop.
// It blocks until ctx is cancelled.
func (h *Hub) Run(ctx context.Context) {
<-ctx.Done()
h.log.Info("WebSocket hub shutting down")
// Run starts the hub's dispatch loop. It blocks until Stop is called.
// Must be called in its own goroutine.
func (h *Hub) Run() {
for {
select {
case <-h.stop:
return
case c := <-h.register:
h.mu.Lock()
// If an existing client has the same userID, close its send channel
// so writePump exits cleanly before the new client takes over.
if old, ok := h.clients[c.userID]; ok && old != c {
close(old.send)
}
h.clients[c.userID] = c
h.mu.Unlock()
case c := <-h.unregister:
h.mu.Lock()
if current, ok := h.clients[c.userID]; ok && current == c {
delete(h.clients, c.userID)
}
h.mu.Unlock()
case bm := <-h.broadcast:
h.deliverBroadcast(bm)
}
}
}
// ServeWS handles an incoming WebSocket upgrade request.
// Full implementation in Phase 4.
func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request) {
http.Error(w, "WebSocket not yet implemented", http.StatusNotImplemented)
// Stop signals Run to exit.
func (h *Hub) Stop() {
close(h.stop)
}
// Register queues a client for registration with the hub.
func (h *Hub) Register(c *Client) {
h.register <- c
}
// Unregister queues a client for removal from the hub.
func (h *Hub) Unregister(c *Client) {
h.unregister <- c
}
// BroadcastToChannel enqueues msg for delivery to all clients subscribed to
// channelID. When channelID is 0 the message is sent to every connected client.
func (h *Hub) BroadcastToChannel(channelID int64, msg []byte) {
h.broadcast <- broadcastMsg{channelID: channelID, msg: msg}
}
// BroadcastToAll enqueues msg for delivery to every connected client.
func (h *Hub) BroadcastToAll(msg []byte) {
h.broadcast <- broadcastMsg{channelID: 0, msg: msg}
}
// SendToUser delivers msg directly to the client identified by userID.
// Returns true if the client was found and the message was queued.
func (h *Hub) SendToUser(userID int64, msg []byte) bool {
h.mu.RLock()
c, ok := h.clients[userID]
h.mu.RUnlock()
if !ok {
return false
}
select {
case c.send <- msg:
return true
default:
// send buffer full — drop rather than block.
return false
}
}
// ClientCount returns the number of currently registered clients (test helper).
func (h *Hub) ClientCount() int {
h.mu.RLock()
defer h.mu.RUnlock()
return len(h.clients)
}
// deliverBroadcast sends bm.msg to the appropriate clients.
func (h *Hub) deliverBroadcast(bm broadcastMsg) {
h.mu.RLock()
defer h.mu.RUnlock()
for _, c := range h.clients {
// channelID == 0 → broadcast to everyone.
if bm.channelID != 0 && c.channelID != bm.channelID {
continue
}
select {
case c.send <- bm.msg:
default:
// Client's buffer is full; skip to avoid blocking the hub.
}
}
}
Server/ws/hub_test.go
+547
View File
@@ -0,0 +1,547 @@
package ws_test
import (
"encoding/json"
"fmt"
"sync"
"testing"
"testing/fstest"
"time"
"github.com/owncord/server/auth"
"github.com/owncord/server/db"
"github.com/owncord/server/ws"
)
// ─── test helpers ─────────────────────────────────────────────────────────────
func openTestDB(t *testing.T) *db.DB {
t.Helper()
database, err := db.Open(":memory:")
if err != nil {
t.Fatalf("db.Open: %v", err)
}
t.Cleanup(func() { database.Close() })
migrFS := fstest.MapFS{
"001_schema.sql": {Data: hubTestSchema},
}
if err := db.MigrateFS(database, migrFS); err != nil {
t.Fatalf("MigrateFS: %v", err)
}
return database
}
func newTestHub(t *testing.T) (*ws.Hub, *db.DB) {
t.Helper()
database := openTestDB(t)
limiter := auth.NewRateLimiter()
hub := ws.NewHub(database, limiter)
return hub, database
}
// seedTestUser inserts a Member-role user and returns its ID.
func seedTestUser(t *testing.T, database *db.DB, username string) int64 {
t.Helper()
id, err := database.CreateUser(username, "hash", 4)
if err != nil {
t.Fatalf("seedUser: %v", err)
}
return id
}
// seedOwnerUser inserts an Owner-role user and returns the full *db.User.
// Owner role (id=1) has all permissions (0x7FFFFFFF), so it passes all checks.
func seedOwnerUser(t *testing.T, database *db.DB, username string) *db.User {
t.Helper()
_, err := database.CreateUser(username, "hash", 1) // roleID=1 → Owner
if err != nil {
t.Fatalf("seedOwnerUser: %v", err)
}
user, err := database.GetUserByUsername(username)
if err != nil || user == nil {
t.Fatalf("seedOwnerUser GetUserByUsername: %v", err)
}
return user
}
// seedTestChannel inserts a channel and returns its ID.
func seedTestChannel(t *testing.T, database *db.DB, name string) int64 {
t.Helper()
id, err := database.CreateChannel(name, "text", "", "", 0)
if err != nil {
t.Fatalf("seedChannel: %v", err)
}
return id
}
// ─── Hub lifecycle ────────────────────────────────────────────────────────────
func TestNewHub_NotNil(t *testing.T) {
hub, _ := newTestHub(t)
if hub == nil {
t.Fatal("NewHub returned nil")
}
}
func TestHub_RunStops(t *testing.T) {
hub, _ := newTestHub(t)
done := make(chan struct{})
go func() {
hub.Run()
close(done)
}()
// Give the goroutine a moment to start, then stop the hub.
time.Sleep(10 * time.Millisecond)
hub.Stop()
select {
case <-done:
// ok
case <-time.After(2 * time.Second):
t.Error("hub.Run() did not stop after hub.Stop()")
}
}
// ─── Register / Unregister ────────────────────────────────────────────────────
func TestHub_RegisterIncrementsCount(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
userID := seedTestUser(t, database, "alice")
send := make(chan []byte, 4)
hub.Register(ws.NewTestClient(hub, userID, send))
time.Sleep(20 * time.Millisecond)
if hub.ClientCount() != 1 {
t.Errorf("ClientCount = %d, want 1", hub.ClientCount())
}
}
func TestHub_UnregisterDecrementsCount(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
userID := seedTestUser(t, database, "bob")
send := make(chan []byte, 4)
c := ws.NewTestClient(hub, userID, send)
hub.Register(c)
time.Sleep(20 * time.Millisecond)
hub.Unregister(c)
time.Sleep(20 * time.Millisecond)
if hub.ClientCount() != 0 {
t.Errorf("ClientCount = %d, want 0", hub.ClientCount())
}
}
func TestHub_RegisterSameUserTwice(t *testing.T) {
// Second registration for same userID should replace the first.
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
userID := seedTestUser(t, database, "carol")
send1 := make(chan []byte, 4)
send2 := make(chan []byte, 4)
hub.Register(ws.NewTestClient(hub, userID, send1))
hub.Register(ws.NewTestClient(hub, userID, send2))
time.Sleep(30 * time.Millisecond)
if hub.ClientCount() != 1 {
t.Errorf("ClientCount = %d after double register, want 1", hub.ClientCount())
}
}
// ─── BroadcastToAll ───────────────────────────────────────────────────────────
func TestHub_BroadcastToAll_DeliversToAllClients(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
u1 := seedTestUser(t, database, "dave")
u2 := seedTestUser(t, database, "eve")
s1 := make(chan []byte, 4)
s2 := make(chan []byte, 4)
hub.Register(ws.NewTestClient(hub, u1, s1))
hub.Register(ws.NewTestClient(hub, u2, s2))
time.Sleep(20 * time.Millisecond)
msg := []byte(`{"type":"presence","payload":{}}`)
hub.BroadcastToAll(msg)
time.Sleep(20 * time.Millisecond)
assertReceived(t, s1, msg, "client 1")
assertReceived(t, s2, msg, "client 2")
}
func TestHub_BroadcastToAll_NoClients(t *testing.T) {
hub, _ := newTestHub(t)
go hub.Run()
defer hub.Stop()
// Should not panic.
hub.BroadcastToAll([]byte(`{}`))
}
// ─── BroadcastToChannel ───────────────────────────────────────────────────────
func TestHub_BroadcastToChannel_OnlySendsToChannelMembers(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
chID := seedTestChannel(t, database, "general")
u1 := seedTestUser(t, database, "frank")
u2 := seedTestUser(t, database, "grace")
s1 := make(chan []byte, 4)
s2 := make(chan []byte, 4)
c1 := ws.NewTestClientWithChannel(hub, u1, chID, s1)
c2 := ws.NewTestClientWithChannel(hub, u2, 999, s2) // different channel
hub.Register(c1)
hub.Register(c2)
time.Sleep(20 * time.Millisecond)
msg := []byte(`{"type":"chat_message","payload":{}}`)
hub.BroadcastToChannel(chID, msg)
time.Sleep(20 * time.Millisecond)
assertReceived(t, s1, msg, "channel member")
assertNotReceived(t, s2, "non-member")
}
func TestHub_BroadcastToChannel_ZeroChannelSendsToAll(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
u1 := seedTestUser(t, database, "henry")
s1 := make(chan []byte, 4)
hub.Register(ws.NewTestClient(hub, u1, s1))
time.Sleep(20 * time.Millisecond)
msg := []byte(`{"type":"presence","payload":{}}`)
hub.BroadcastToChannel(0, msg)
time.Sleep(20 * time.Millisecond)
assertReceived(t, s1, msg, "client")
}
// ─── SendToUser ───────────────────────────────────────────────────────────────
func TestHub_SendToUser_ExistingClient(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
userID := seedTestUser(t, database, "ivan")
send := make(chan []byte, 4)
hub.Register(ws.NewTestClient(hub, userID, send))
time.Sleep(20 * time.Millisecond)
msg := []byte(`{"type":"chat_send_ok","payload":{}}`)
ok := hub.SendToUser(userID, msg)
if !ok {
t.Error("SendToUser returned false for existing client")
}
time.Sleep(20 * time.Millisecond)
assertReceived(t, send, msg, "target user")
}
func TestHub_SendToUser_MissingClient(t *testing.T) {
hub, _ := newTestHub(t)
go hub.Run()
defer hub.Stop()
ok := hub.SendToUser(9999, []byte(`{}`))
if ok {
t.Error("SendToUser should return false for absent client")
}
}
// ─── Message dispatch ─────────────────────────────────────────────────────────
func TestHub_HandleMessage_UnknownType_SendsError(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
userID := seedTestUser(t, database, "julia")
send := make(chan []byte, 4)
c := ws.NewTestClient(hub, userID, send)
hub.Register(c)
time.Sleep(20 * time.Millisecond)
raw := []byte(`{"type":"totally_unknown","payload":{}}`)
hub.HandleMessageForTest(c, raw)
time.Sleep(20 * time.Millisecond)
select {
case got := <-send:
var resp map[string]interface{}
if err := json.Unmarshal(got, &resp); err != nil {
t.Fatalf("unmarshal response: %v", err)
}
if resp["type"] != "error" {
t.Errorf("type = %q, want 'error'", resp["type"])
}
case <-time.After(500 * time.Millisecond):
t.Error("expected error response for unknown message type")
}
}
func TestHub_HandleMessage_InvalidJSON(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
userID := seedTestUser(t, database, "kim")
send := make(chan []byte, 4)
c := ws.NewTestClient(hub, userID, send)
hub.Register(c)
time.Sleep(20 * time.Millisecond)
hub.HandleMessageForTest(c, []byte(`NOT JSON`))
time.Sleep(20 * time.Millisecond)
select {
case got := <-send:
var resp map[string]interface{}
if err := json.Unmarshal(got, &resp); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if resp["type"] != "error" {
t.Errorf("type = %q, want 'error'", resp["type"])
}
case <-time.After(500 * time.Millisecond):
t.Error("expected error response for invalid JSON")
}
}
// ─── Rate limiting ────────────────────────────────────────────────────────────
func TestHub_ChatSend_RateLimit(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
user := seedOwnerUser(t, database, "larry")
chID := seedTestChannel(t, database, "rl-test")
send := make(chan []byte, 64)
c := ws.NewTestClientWithUser(hub, user, chID, send)
hub.Register(c)
time.Sleep(20 * time.Millisecond)
payload := map[string]interface{}{
"channel_id": chID,
"content": "hi",
}
raw, _ := json.Marshal(map[string]interface{}{
"type": "chat_send",
"payload": payload,
})
// Send 12 messages rapidly — 11th and beyond should be rate-limited.
for i := 0; i < 12; i++ {
hub.HandleMessageForTest(c, raw)
}
time.Sleep(100 * time.Millisecond)
// Drain all messages, count errors.
errCount := 0
drainLoop:
for {
select {
case got := <-send:
var resp map[string]interface{}
if err := json.Unmarshal(got, &resp); err == nil {
if resp["type"] == "error" {
errCount++
}
}
default:
break drainLoop
}
}
if errCount == 0 {
t.Error("expected at least one rate-limit error response")
}
}
// ─── Concurrency ─────────────────────────────────────────────────────────────
func TestHub_ConcurrentRegisterUnregister(t *testing.T) {
hub, database := newTestHub(t)
go hub.Run()
defer hub.Stop()
var wg sync.WaitGroup
for i := 0; i < 20; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
username := fmt.Sprintf("user%d", i)
userID := seedTestUser(t, database, username)
send := make(chan []byte, 4)
c := ws.NewTestClient(hub, userID, send)
hub.Register(c)
time.Sleep(5 * time.Millisecond)
hub.Unregister(c)
}(i)
}
wg.Wait()
time.Sleep(50 * time.Millisecond)
if hub.ClientCount() != 0 {
t.Errorf("expected 0 clients after concurrent churn, got %d", hub.ClientCount())
}
}
// ─── assertion helpers ────────────────────────────────────────────────────────
func assertReceived(t *testing.T, ch <-chan []byte, want []byte, label string) {
t.Helper()
select {
case got := <-ch:
if string(got) != string(want) {
t.Errorf("%s: got %q, want %q", label, got, want)
}
case <-time.After(500 * time.Millisecond):
t.Errorf("%s: did not receive expected message within timeout", label)
}
}
func assertNotReceived(t *testing.T, ch <-chan []byte, label string) {
t.Helper()
select {
case got := <-ch:
t.Errorf("%s: received unexpected message: %q", label, got)
case <-time.After(100 * time.Millisecond):
// ok — nothing received
}
}
// hubTestSchema is the minimal schema needed for hub tests.
var hubTestSchema = []byte(`
CREATE TABLE IF NOT EXISTS roles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
color TEXT,
permissions INTEGER NOT NULL DEFAULT 0,
position INTEGER NOT NULL DEFAULT 0,
is_default INTEGER NOT NULL DEFAULT 0
);
INSERT OR IGNORE INTO roles (id, name, color, permissions, position, is_default) VALUES
(1, 'Owner', '#E74C3C', 2147483647, 100, 0),
(2, 'Admin', '#F39C12', 1073741823, 80, 0),
(3, 'Moderator', '#3498DB', 1048575, 60, 0),
(4, 'Member', NULL, 1049089, 40, 1);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
password TEXT NOT NULL,
avatar TEXT,
role_id INTEGER NOT NULL DEFAULT 4 REFERENCES roles(id),
totp_secret TEXT,
status TEXT NOT NULL DEFAULT 'offline',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
last_seen TEXT,
banned INTEGER NOT NULL DEFAULT 0,
ban_reason TEXT,
ban_expires TEXT
);
CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token TEXT NOT NULL UNIQUE,
device TEXT,
ip_address TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
last_used TEXT NOT NULL DEFAULT (datetime('now')),
expires_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS channels (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
type TEXT NOT NULL DEFAULT 'text',
category TEXT,
topic TEXT,
position INTEGER NOT NULL DEFAULT 0,
slow_mode INTEGER NOT NULL DEFAULT 0,
archived INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS channel_overrides (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
role_id INTEGER NOT NULL REFERENCES roles(id) ON DELETE CASCADE,
allow INTEGER NOT NULL DEFAULT 0,
deny INTEGER NOT NULL DEFAULT 0,
UNIQUE(channel_id, role_id)
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id),
content TEXT NOT NULL,
reply_to INTEGER REFERENCES messages(id) ON DELETE SET NULL,
edited_at TEXT,
deleted INTEGER NOT NULL DEFAULT 0,
pinned INTEGER NOT NULL DEFAULT 0,
timestamp TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
content,
content='messages',
content_rowid='id'
);
CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.id, old.content);
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TABLE IF NOT EXISTS reactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
emoji TEXT NOT NULL,
UNIQUE(message_id, user_id, emoji)
);
CREATE TABLE IF NOT EXISTS read_states (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE,
last_message_id INTEGER NOT NULL DEFAULT 0,
mention_count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (user_id, channel_id)
);
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
INSERT OR IGNORE INTO settings (key, value) VALUES
('server_name', 'OwnCord Server'),
('motd', 'Welcome!');
`)
Server/ws/messages.go
+145
View File
@@ -0,0 +1,145 @@
package ws
import (
"encoding/json"
"fmt"
)
// envelope is the common wrapper for all WebSocket messages.
type envelope struct {
Type string `json:"type"`
ID string `json:"id,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
}
// buildJSON marshals v into a JSON byte slice, logging on failure.
func buildJSON(v interface{}) []byte {
b, err := json.Marshal(v)
if err != nil {
// Fallback: send a generic error rather than panicking.
b, _ = json.Marshal(map[string]string{"type": "error", "message": "internal marshal error"})
}
return b
}
// buildErrorMsg produces an error envelope with the given code and message.
func buildErrorMsg(code, message string) []byte {
return buildJSON(map[string]interface{}{
"type": "error",
"payload": map[string]string{
"code": code,
"message": message,
},
})
}
// buildPresenceMsg constructs a presence broadcast payload.
func buildPresenceMsg(userID int64, status string) []byte {
return buildJSON(map[string]interface{}{
"type": "presence",
"payload": map[string]interface{}{
"user_id": userID,
"status": status,
},
})
}
// buildChatMessage constructs a chat_message broadcast envelope.
func buildChatMessage(msgID, channelID, userID int64, username string, avatar *string, content string, timestamp string, replyTo *int64) []byte {
avatarVal := interface{}(nil)
if avatar != nil {
avatarVal = *avatar
}
return buildJSON(map[string]interface{}{
"type": "chat_message",
"payload": map[string]interface{}{
"id": msgID,
"channel_id": channelID,
"user": map[string]interface{}{
"id": userID,
"username": username,
"avatar": avatarVal,
},
"content": content,
"reply_to": replyTo,
"timestamp": timestamp,
},
})
}
// buildChatSendOK constructs a chat_send_ok ack.
func buildChatSendOK(requestID string, msgID int64, timestamp string) []byte {
return buildJSON(map[string]interface{}{
"type": "chat_send_ok",
"id": requestID,
"payload": map[string]interface{}{
"message_id": msgID,
"timestamp": timestamp,
},
})
}
// buildChatEdited constructs a chat_edited broadcast.
func buildChatEdited(msgID, channelID int64, content, editedAt string) []byte {
return buildJSON(map[string]interface{}{
"type": "chat_edited",
"payload": map[string]interface{}{
"message_id": msgID,
"channel_id": channelID,
"content": content,
"edited_at": editedAt,
},
})
}
// buildChatDeleted constructs a chat_deleted broadcast.
func buildChatDeleted(msgID, channelID int64) []byte {
return buildJSON(map[string]interface{}{
"type": "chat_deleted",
"payload": map[string]interface{}{
"message_id": msgID,
"channel_id": channelID,
},
})
}
// buildReactionUpdate constructs a reaction_update broadcast.
func buildReactionUpdate(msgID, channelID, userID int64, emoji, action string) []byte {
return buildJSON(map[string]interface{}{
"type": "reaction_update",
"payload": map[string]interface{}{
"message_id": msgID,
"channel_id": channelID,
"emoji": emoji,
"user_id": userID,
"action": action,
},
})
}
// buildTypingMsg constructs a typing broadcast.
func buildTypingMsg(channelID, userID int64, username string) []byte {
return buildJSON(map[string]interface{}{
"type": "typing",
"payload": map[string]interface{}{
"channel_id": channelID,
"user_id": userID,
"username": username,
},
})
}
// parseChannelID safely extracts channel_id from a raw payload map.
func parseChannelID(payload json.RawMessage) (int64, error) {
var p struct {
ChannelID json.Number `json:"channel_id"`
}
if err := json.Unmarshal(payload, &p); err != nil {
return 0, err
}
id, err := p.ChannelID.Int64()
if err != nil {
return 0, fmt.Errorf("channel_id must be integer: %w", err)
}
return id, nil
}
Server/ws/serve.go
+200
View File
@@ -0,0 +1,200 @@
package ws
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"net/http"
"time"
"nhooyr.io/websocket"
"github.com/owncord/server/auth"
"github.com/owncord/server/db"
)
const authDeadline = 10 * time.Second
const writeTimeout = 10 * time.Second
// ServeWS upgrades an HTTP connection to WebSocket, performs in-band auth,
// then drives the client's read/write loops.
// Do not wrap with AuthMiddleware — WS does its own auth.
func ServeWS(hub *Hub, database *db.DB) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
InsecureSkipVerify: true,
})
if err != nil {
slog.Warn("ws upgrade failed", "err", err)
return
}
user, err := authenticateConn(conn, database)
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)
hub.Register(c)
if updateErr := database.UpdateUserStatus(user.ID, "online"); updateErr != nil {
slog.Warn("ws UpdateUserStatus", "err", updateErr)
}
// Send auth_ok followed by the ready payload.
ctx := r.Context()
_ = conn.Write(ctx, websocket.MessageText, buildAuthOK(database, user))
if ready, readyErr := buildReady(database); readyErr == nil {
_ = conn.Write(ctx, websocket.MessageText, ready)
}
hub.BroadcastToAll(buildPresenceMsg(user.ID, "online"))
// writePump runs in background; readPump blocks.
writeCtx, writeCancel := context.WithCancel(ctx)
go writePump(writeCtx, conn, c)
readPump(ctx, conn, hub, c)
writeCancel()
}
}
// writePump drains the client's send channel and writes to the WebSocket.
func writePump(ctx context.Context, conn *websocket.Conn, c *Client) {
for {
select {
case msg, ok := <-c.send:
if !ok {
_ = conn.Close(websocket.StatusNormalClosure, "")
return
}
wCtx, cancel := context.WithTimeout(ctx, writeTimeout)
err := conn.Write(wCtx, websocket.MessageText, msg)
cancel()
if err != nil {
slog.Warn("ws writePump error", "user_id", c.userID, "err", err)
return
}
case <-ctx.Done():
return
}
}
}
// readPump reads from the WebSocket and dispatches messages. Blocks until disconnect.
func readPump(ctx context.Context, conn *websocket.Conn, hub *Hub, c *Client) {
defer func() {
hub.Unregister(c)
if c.user != nil {
_ = hub.db.UpdateUserStatus(c.userID, "offline")
hub.BroadcastToAll(buildPresenceMsg(c.userID, "offline"))
}
}()
for {
_, msg, err := conn.Read(ctx)
if err != nil {
return
}
hub.handleMessage(c, msg)
}
}
// authenticateConn reads the first WebSocket message and validates the session token.
func authenticateConn(conn *websocket.Conn, database *db.DB) (*db.User, error) {
ctx, cancel := context.WithTimeout(context.Background(), authDeadline)
defer cancel()
_, raw, err := conn.Read(ctx)
if err != nil {
return nil, err
}
var env envelope
if err := json.Unmarshal(raw, &env); err != nil {
_ = conn.Write(ctx, websocket.MessageText, buildErrorMsg("AUTH_ERROR", "invalid message"))
return nil, fmt.Errorf("auth: invalid JSON: %w", err)
}
if env.Type != "auth" {
_ = conn.Write(ctx, websocket.MessageText, buildErrorMsg("AUTH_ERROR", "first message must be auth"))
return nil, fmt.Errorf("auth: unexpected type %q", env.Type)
}
var p struct {
Token string `json:"token"`
}
if err := json.Unmarshal(env.Payload, &p); err != nil || p.Token == "" {
_ = conn.Write(ctx, websocket.MessageText, buildErrorMsg("AUTH_ERROR", "missing token"))
return nil, 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, buildErrorMsg("AUTH_ERROR", "invalid token"))
return nil, fmt.Errorf("auth: invalid session")
}
user, err := database.GetUserByID(sess.UserID)
if err != nil || user == nil {
_ = conn.Write(ctx, websocket.MessageText, buildErrorMsg("AUTH_ERROR", "user not found"))
return nil, fmt.Errorf("auth: user not found")
}
if user.Banned {
_ = conn.Write(ctx, websocket.MessageText, buildErrorMsg("BANNED", "you are banned"))
return nil, fmt.Errorf("auth: banned user %d", user.ID)
}
return user, nil
}
// buildAuthOK constructs the auth_ok server→client message.
func buildAuthOK(database *db.DB, user *db.User) []byte {
serverName := "OwnCord Server"
motd := "Welcome!"
_ = database.QueryRow("SELECT value FROM settings WHERE key='server_name'").Scan(&serverName)
_ = database.QueryRow("SELECT value FROM settings WHERE key='motd'").Scan(&motd)
var avatarVal interface{}
if user.Avatar != nil {
avatarVal = *user.Avatar
}
return buildJSON(map[string]interface{}{
"type": "auth_ok",
"payload": map[string]interface{}{
"user": map[string]interface{}{
"id": user.ID,
"username": user.Username,
"avatar": avatarVal,
"status": user.Status,
},
"server_name": serverName,
"motd": motd,
},
})
}
// buildReady constructs the ready server→client message.
func buildReady(database *db.DB) ([]byte, error) {
channels, err := database.ListChannels()
if err != nil {
return nil, fmt.Errorf("buildReady ListChannels: %w", err)
}
roles, err := database.ListRoles()
if err != nil {
return nil, fmt.Errorf("buildReady ListRoles: %w", err)
}
return buildJSON(map[string]interface{}{
"type": "ready",
"payload": map[string]interface{}{
"channels": channels,
"members": []interface{}{},
"voice_states": []interface{}{},
"roles": roles,
},
}), nil
}