chore: release v1.0.7
CI / test (push) Successful in 9m11s
CI / docker (push) Failing after 2m39s
CI / release (push) Skipped
Release / release (push) Canceled after 2m11s

Playout: slate fallback, strict ffprobe validation, Prometheus metrics
Chat: history on join, DM offline delivery, env-config rate limits,
  Redis auto-reconnect, heartbeat, ban/mute enforcement, DM metrics
Viewer: presence indicators, schedule overlay, public /api/schedule
Admin: playlist drag-reorder, schedule overrides CRUD, user actions,
  inline mod buttons, CSV export, config editor, restart buttons,
  upload normalization gate, analytics time range
nginx: CORS preflight, JSON health, stub_status
Docs: test-lab vs production distinction across all configs/docs
Infra: Redis maxmemory, CD health IP-agnostic, slate volume mount
This commit is contained in:
zwitschi committed 2026-08-06 17:41:52 +02:00
1 parent 17ac04856b
commit 5b5cdc4cd8
26 files changed
+1723 -110

No files matched your search

+560 -69
View File
@@ -32,12 +32,22 @@ var (
upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}
clients = make(map[*websocket.Conn]bool)
mutex = sync.Mutex{}
clients = make(map[*websocket.Conn]bool)
clientMeta = make(map[*websocket.Conn]clientInfo)
mutex = sync.Mutex{}
authSessions = map[string]time.Time{}
authSessionsMu = sync.Mutex{}
// Rate limit config (env-overridable)
rateLimitMax = 10
rateLimitWindow = 10 * time.Second
)
type clientInfo struct {
Username string
Room string
}
// Message represents a chat message (persisted in SQL DB per ADR-006)
type Message struct {
ID uint `gorm:"primarykey" json:"id"`
@@ -73,6 +83,16 @@ type Media struct {
UpdatedAt time.Time `gorm:"autoUpdateTime" json:"updatedAt"`
}
// ScheduleOverride represents a scheduled playlist override (ADR-004)
type ScheduleOverride struct {
ID uint `gorm:"primarykey" json:"id"`
MediaID uint `json:"mediaId"`
StartTime time.Time `json:"startTime"`
EndTime time.Time `json:"endTime"`
Reason string `json:"reason"`
CreatedAt time.Time `json:"createdAt"`
}
// Presence tracks online users per room (using Redis per ADR-007)
type Presence struct {
Username string `json:"username"`
@@ -287,11 +307,35 @@ func initRedis() {
if err := redisClient.Ping(context.Background()).Err(); err != nil {
log.Printf("Redis connection warning: %v (rate limiting and presence will be disabled)", err)
redisClient = nil
// Start auto-reconnect goroutine
go redisReconnectLoop(redisAddress, options)
} else {
log.Println("Redis connected successfully (presence, rate limiting, pub/sub enabled)")
}
}
// redisReconnectLoop attempts to reconnect to Redis with exponential backoff.
func redisReconnectLoop(addr string, baseOptions *redis.Options) {
backoff := 2 * time.Second
maxBackoff := 60 * time.Second
for {
time.Sleep(backoff)
testClient := redis.NewClient(baseOptions)
if err := testClient.Ping(context.Background()).Err(); err != nil {
log.Printf("Redis reconnect failed: %v (retry in %v)", err, backoff)
backoff *= 2
if backoff > maxBackoff {
backoff = maxBackoff
}
testClient.Close()
continue
}
redisClient = testClient
log.Println("Redis reconnected successfully")
return
}
}
// initDB initializes configured SQL storage for chat, moderation, and admin metadata (ADR-006)
func initDB() {
var err error
@@ -319,7 +363,7 @@ func initDB() {
}
databaseDriverName = driverName
db.AutoMigrate(&Message{}, &ModerationLog{}, &Media{})
db.AutoMigrate(&Message{}, &ModerationLog{}, &Media{}, &ScheduleOverride{})
log.Printf("%s database initialized (chat history, moderation log, media metadata)", strings.Title(driverName))
}
@@ -626,7 +670,103 @@ func handleAdminPlaylist(w http.ResponseWriter, r *http.Request) {
}
}
func handleAdminSchedule(w http.ResponseWriter, r *http.Request) {
// handlePublicSchedule returns playlist schedule without auth (viewer-facing).
func handlePublicSchedule(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
content, err := os.ReadFile(getPlaylistPath())
if err != nil {
http.Error(w, "schedule unavailable", http.StatusServiceUnavailable)
return
}
entries := parsePlaylistEntries(string(content))
var mediaRecords []Media
db.Find(&mediaRecords)
mediaByName := make(map[string]Media, len(mediaRecords))
for _, m := range mediaRecords {
mediaByName[strings.ToLower(m.Filename)] = m
}
type PublicEntry struct {
Index int `json:"index"`
Label string `json:"label"`
Title string `json:"title"`
Status string `json:"status"`
}
output := make([]PublicEntry, 0, len(entries))
for _, e := range entries {
pe := PublicEntry{Index: e.Index, Label: e.Label, Status: "pending"}
if m, ok := mediaByName[strings.ToLower(filepath.Base(e.Path))]; ok {
pe.Title = m.Title
pe.Status = m.Status
} else {
pe.Title = filepath.Base(e.Path)
}
output = append(output, pe)
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{
"entries": output,
"total": len(output),
"generatedAt": time.Now().UTC().Format(time.RFC3339),
})
}
// handleAdminScheduleOverrides manages CRUD for schedule overrides.
func handleAdminScheduleOverrides(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
var overrides []ScheduleOverride
db.Order("start_time asc").Find(&overrides)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(overrides)
case http.MethodPost:
var ov ScheduleOverride
if err := json.NewDecoder(r.Body).Decode(&ov); err != nil {
http.Error(w, "invalid request body", http.StatusBadRequest)
return
}
if ov.MediaID == 0 || ov.StartTime.IsZero() {
http.Error(w, "mediaId and startTime are required", http.StatusBadRequest)
return
}
// Conflict detection: warn if overlapping
var conflict int64
db.Model(&ScheduleOverride{}).
Where("start_time < ? AND end_time > ?", ov.EndTime, ov.StartTime).
Count(&conflict)
if conflict > 0 {
http.Error(w, fmt.Sprintf("conflict: %d overlapping override(s)", conflict), http.StatusConflict)
return
}
db.Create(&ov)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusCreated)
json.NewEncoder(w).Encode(ov)
case http.MethodDelete:
idStr := r.URL.Query().Get("id")
if idStr == "" {
http.Error(w, "id query param required", http.StatusBadRequest)
return
}
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
http.Error(w, "invalid id", http.StatusBadRequest)
return
}
if err := db.Delete(&ScheduleOverride{}, id).Error; err != nil {
http.Error(w, "failed to delete override", http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
@@ -875,7 +1015,53 @@ func getAdminSystemSummary() AdminSystemSummary {
return summary
}
func handleAdminSystemSummary(w http.ResponseWriter, r *http.Request) {
// handleAdminConfig returns runtime configuration values for the admin config editor.
func handleAdminConfig(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{
"rateLimitMax": rateLimitMax,
"rateLimitWindowSec": int(rateLimitWindow.Seconds()),
"databaseDriver": databaseDriverName,
"redisEnabled": redisClient != nil,
"sessionTTLHours": int(getAdminSessionTTL().Hours()),
})
case http.MethodPut:
var payload map[string]any
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, "invalid body", http.StatusBadRequest)
return
}
if v, ok := payload["rateLimitMax"].(float64); ok && v > 0 {
rateLimitMax = int(v)
log.Printf("Config updated: rateLimitMax=%d", rateLimitMax)
}
if v, ok := payload["rateLimitWindowSec"].(float64); ok && v > 0 {
rateLimitWindow = time.Duration(v) * time.Second
log.Printf("Config updated: rateLimitWindow=%v", rateLimitWindow)
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
// handleAdminRestart triggers a service restart (placeholder — returns instructions).
func handleAdminRestart(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
service := r.URL.Query().Get("service")
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{
"status": "restart_not_automated",
"service": service,
"message": fmt.Sprintf("Restart %s manually: docker compose -f docker-compose.prod.yml restart %s", service, service),
})
}
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
@@ -886,47 +1072,83 @@ func handleAdminSystemSummary(w http.ResponseWriter, r *http.Request) {
}
func handleAdminUsers(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
switch r.Method {
case http.MethodGet:
response := AdminUsersResponse{
Users: []ActiveUser{},
RedisEnabled: redisClient != nil,
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
}
if redisClient != nil {
ctx := context.Background()
keys, _, err := redisClient.Scan(ctx, 0, "presence:*", 200).Result()
if err != nil {
http.Error(w, fmt.Sprintf("failed to scan presence keys: %v", err), http.StatusInternalServerError)
return
}
for _, key := range keys {
parts := strings.SplitN(key, ":", 3)
if len(parts) != 3 {
continue
}
response.Users = append(response.Users, ActiveUser{
Room: parts[1],
Username: parts[2],
Online: true,
})
}
}
sort.Slice(response.Users, func(i, j int) bool {
if response.Users[i].Room == response.Users[j].Room {
return response.Users[i].Username < response.Users[j].Username
}
return response.Users[i].Room < response.Users[j].Room
})
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response)
response := AdminUsersResponse{
Users: []ActiveUser{},
RedisEnabled: redisClient != nil,
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
}
if redisClient != nil {
ctx := context.Background()
keys, _, err := redisClient.Scan(ctx, 0, "presence:*", 200).Result()
if err != nil {
http.Error(w, fmt.Sprintf("failed to scan presence keys: %v", err), http.StatusInternalServerError)
case http.MethodPost:
var payload struct {
Action string `json:"action"` // mute, unmute, ban, unban
Username string `json:"username"`
Duration int `json:"duration"` // minutes, for mute
}
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, "invalid body", http.StatusBadRequest)
return
}
for _, key := range keys {
parts := strings.SplitN(key, ":", 3)
if len(parts) != 3 {
continue
if payload.Action == "" || payload.Username == "" {
http.Error(w, "action and username required", http.StatusBadRequest)
return
}
if redisClient == nil {
http.Error(w, "Redis unavailable", http.StatusServiceUnavailable)
return
}
ctx := context.Background()
switch payload.Action {
case "mute":
dur := time.Duration(payload.Duration) * time.Minute
if dur <= 0 {
dur = 10 * time.Minute
}
response.Users = append(response.Users, ActiveUser{
Room: parts[1],
Username: parts[2],
Online: true,
})
redisClient.Set(ctx, fmt.Sprintf("muted:%s", payload.Username), "1", dur)
case "unmute":
redisClient.Del(ctx, fmt.Sprintf("muted:%s", payload.Username))
case "ban":
redisClient.Set(ctx, fmt.Sprintf("banned:%s", payload.Username), "1", 0)
case "unban":
redisClient.Del(ctx, fmt.Sprintf("banned:%s", payload.Username))
default:
http.Error(w, "unknown action", http.StatusBadRequest)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{"status": "ok", "action": payload.Action, "username": payload.Username})
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
sort.Slice(response.Users, func(i, j int) bool {
if response.Users[i].Room == response.Users[j].Room {
return response.Users[i].Username < response.Users[j].Username
}
return response.Users[i].Room < response.Users[j].Room
})
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response)
}
func listMediaFiles() ([]MediaFileEntry, error) {
@@ -1139,7 +1361,90 @@ func mediaEntryFromRecord(record Media) MediaFileEntry {
}
}
func probeMediaDurationSeconds(mediaPath string) (int, error) {
// MediaNormalizationResult holds ffprobe validation output for admin uploads.
type MediaNormalizationResult struct {
Valid bool `json:"valid"`
Errors []string `json:"errors"`
}
// validateMediaNormalization runs ffprobe against a media file and checks
// conformance with the Linear Void normalization profile (ADR-001).
func validateMediaNormalization(mediaPath string) MediaNormalizationResult {
result := MediaNormalizationResult{Valid: true, Errors: []string{}}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
out, err := exec.CommandContext(ctx,
"ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", mediaPath,
).CombinedOutput()
if ctx.Err() == context.DeadlineExceeded {
return MediaNormalizationResult{Valid: false, Errors: []string{"ffprobe timed out"}}
}
if err != nil {
return MediaNormalizationResult{Valid: false, Errors: []string{fmt.Sprintf("ffprobe failed: %v (%s)", err, strings.TrimSpace(string(out)))}}
}
var probe struct {
Streams []struct {
CodecName string `json:"codec_name"`
CodecType string `json:"codec_type"`
Width int `json:"width"`
Height int `json:"height"`
RFrameRate string `json:"r_frame_rate"`
PixFmt string `json:"pix_fmt"`
SampleRate string `json:"sample_rate"`
} `json:"streams"`
}
if err := json.Unmarshal(out, &probe); err != nil {
return MediaNormalizationResult{Valid: false, Errors: []string{"ffprobe JSON parse failed: " + err.Error()}}
}
var hasVideo, hasAudio bool
for _, s := range probe.Streams {
switch s.CodecType {
case "video":
hasVideo = true
if s.CodecName != "h264" {
result.Errors = append(result.Errors, fmt.Sprintf("video codec must be h264, got %s", s.CodecName))
}
if s.RFrameRate != "30/1" {
result.Errors = append(result.Errors, fmt.Sprintf("framerate must be 30 fps constant, got %s", s.RFrameRate))
}
if s.Width != 1280 && s.Width != 1920 {
result.Errors = append(result.Errors, fmt.Sprintf("width must be 1280 or 1920, got %d", s.Width))
}
if s.Height != 720 && s.Height != 1080 {
result.Errors = append(result.Errors, fmt.Sprintf("height must be 720 or 1080, got %d", s.Height))
}
if s.PixFmt != "yuv420p" {
result.Errors = append(result.Errors, fmt.Sprintf("pixel format must be yuv420p, got %s", s.PixFmt))
}
case "audio":
hasAudio = true
if s.CodecName != "aac" {
result.Errors = append(result.Errors, fmt.Sprintf("audio codec must be aac, got %s", s.CodecName))
}
if s.SampleRate != "48000" {
result.Errors = append(result.Errors, fmt.Sprintf("sample rate must be 48000 Hz, got %s", s.SampleRate))
}
}
}
if !hasVideo {
result.Errors = append(result.Errors, "no video stream found")
}
if !hasAudio {
result.Errors = append(result.Errors, "no audio stream found")
}
if len(result.Errors) > 0 {
result.Valid = false
}
return result
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
@@ -1274,6 +1579,13 @@ func handleAdminMedia(w http.ResponseWriter, r *http.Request) {
return
}
// Validate normalization profile (ADR-001)
normResult := validateMediaNormalization(targetPath)
status := "ready"
if !normResult.Valid {
status = "needs_normalization"
}
filename := filepath.Base(targetPath)
mediaRecord := Media{
Filename: filename,
@@ -1281,9 +1593,9 @@ func handleAdminMedia(w http.ResponseWriter, r *http.Request) {
Path: targetPath,
SizeBytes: fileInfo.Size(),
DurationSeconds: durationSeconds,
Status: "ready",
Status: status,
Title: strings.TrimSuffix(filename, filepath.Ext(filename)),
Tags: "",
Tags: fmt.Sprintf(`"normalization":%v`, normResult.Valid),
}
if err := db.Create(&mediaRecord).Error; err != nil {
_ = os.Remove(targetPath)
@@ -1602,6 +1914,13 @@ func handleAdminMediaUploads(w http.ResponseWriter, r *http.Request) {
return
}
// Validate normalization profile (ADR-001)
normResult := validateMediaNormalization(targetPath)
status := "ready"
if !normResult.Valid {
status = "needs_normalization"
}
filename := filepath.Base(targetPath)
mediaRecord := Media{
Filename: filename,
@@ -1609,9 +1928,9 @@ func handleAdminMediaUploads(w http.ResponseWriter, r *http.Request) {
Path: targetPath,
SizeBytes: fileInfo.Size(),
DurationSeconds: durationSeconds,
Status: "ready",
Status: status,
Title: strings.TrimSuffix(filename, filepath.Ext(filename)),
Tags: "",
Tags: fmt.Sprintf(`"normalization":%v`, normResult.Valid),
}
if err := db.Create(&mediaRecord).Error; err != nil {
_ = os.Remove(targetPath)
@@ -1793,6 +2112,10 @@ func handleMetrics(w http.ResponseWriter, r *http.Request) {
fmt.Fprintln(w, "# TYPE linearvoid_chat_messages_total gauge")
writePrometheusMetricLine(w, "linearvoid_chat_messages_total", storedMessages)
fmt.Fprintln(w, "# HELP linearvoid_chat_dms_total Total private messages sent.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_dms_total counter")
writePrometheusMetricLine(w, "linearvoid_chat_dms_total", dmsSent)
fmt.Fprintln(w, "# HELP linearvoid_chat_moderation_actions_total Total moderation log entries.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_moderation_actions_total gauge")
writePrometheusMetricLine(w, "linearvoid_chat_moderation_actions_total", moderationActions)
@@ -1841,25 +2164,57 @@ func handleConnection(w http.ResponseWriter, r *http.Request) {
log.Println("Upgrade error:", err)
return
}
username := r.URL.Query().Get("username")
room := r.URL.Query().Get("room")
if username == "" {
username = fmt.Sprintf("anon_%d", time.Now().UnixNano()%10000)
}
mutex.Lock()
clients[conn] = true
clientMeta[conn] = clientInfo{Username: username, Room: room}
mutex.Unlock()
// Announce presence (ADR-007)
username := r.URL.Query().Get("username")
room := r.URL.Query().Get("room")
if redisClient != nil && username != "" && room != "" {
if redisClient != nil && room != "" {
redisClient.Set(context.Background(), fmt.Sprintf("presence:%s:%s", room, username), "online", 60*time.Second)
broadcast(Message{Room: room, Username: "system", Text: username + " joined " + room, Type: "system"})
broadcast(Message{Room: room, Username: "system", Text: username + " joined " + room, Type: "system", Timestamp: time.Now()})
}
// Push chat history on join (last 50 messages)
pushChatHistory(conn, room, 50)
// Heartbeat goroutine: refresh presence TTL every 30s
heartbeatStop := make(chan struct{})
go func() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if redisClient != nil && room != "" {
redisClient.Set(context.Background(), fmt.Sprintf("presence:%s:%s", room, username), "online", 60*time.Second)
}
// Ping to detect dead connections
if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
case <-heartbeatStop:
return
}
}
}()
defer func() {
close(heartbeatStop)
mutex.Lock()
delete(clients, conn)
delete(clientMeta, conn)
mutex.Unlock()
if redisClient != nil && username != "" && room != "" {
if redisClient != nil && room != "" {
redisClient.Del(context.Background(), fmt.Sprintf("presence:%s:%s", room, username))
broadcast(Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system"})
broadcast(Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system", Timestamp: time.Now()})
}
conn.Close()
}()
@@ -1873,15 +2228,27 @@ func handleConnection(w http.ResponseWriter, r *http.Request) {
msg.Timestamp = time.Now()
msg.Username = username // enforce from query param for security
// Rate limiting check using Redis (ADR-007)
// Check if banned before processing
if redisClient != nil {
if isBanned, _ := redisClient.Exists(context.Background(), fmt.Sprintf("banned:%s", username)).Result(); isBanned > 0 {
conn.WriteJSON(Message{Type: "system", Text: "You are banned from chat.", Timestamp: time.Now()})
continue
}
if isMuted, _ := redisClient.Exists(context.Background(), fmt.Sprintf("muted:%s", username)).Result(); isMuted > 0 {
conn.WriteJSON(Message{Type: "system", Text: "You are muted.", Timestamp: time.Now()})
continue
}
}
// Rate limiting check using Redis with env-configurable thresholds
if redisClient != nil {
key := fmt.Sprintf("rate:%s", msg.Username)
count, _ := redisClient.Incr(context.Background(), key).Result()
if count == 1 {
redisClient.Expire(context.Background(), key, 10*time.Second)
redisClient.Expire(context.Background(), key, rateLimitWindow)
}
if count > 10 {
conn.WriteJSON(Message{Type: "system", Text: "Rate limit exceeded. Please slow down."})
if count > int64(rateLimitMax) {
conn.WriteJSON(Message{Type: "system", Text: "Rate limit exceeded. Please slow down.", Timestamp: time.Now()})
continue
}
}
@@ -1892,29 +2259,73 @@ func handleConnection(w http.ResponseWriter, r *http.Request) {
continue
}
// Handle private DMs (room format "dm:user1-user2")
// Handle private DMs
if strings.HasPrefix(msg.Room, "dm:") {
handlePrivateDM(msg)
return
continue
}
msg.Type = "public"
log.Printf("Chat message in room %s from %s: %s", msg.Room, msg.Username, msg.Text)
broadcast(msg)
// Save private messages explicitly as well.
if msg.Type == "private" {
db.Create(&msg)
}
}
}
// pushChatHistory sends the last N messages for a room via WebSocket.
func pushChatHistory(conn *websocket.Conn, room string, limit int) {
if db == nil || room == "" {
return
}
var messages []Message
db.Where("room = ?", room).Order("timestamp desc").Limit(limit).Find(&messages)
// Reverse to chronological order
for i := len(messages)/2 - 1; i >= 0; i-- {
opp := len(messages) - 1 - i
messages[i], messages[opp] = messages[opp], messages[i]
}
for _, msg := range messages {
conn.WriteJSON(msg)
}
}
// dmsSent tracks total private messages for Prometheus metrics (set from handleMetrics).
var dmsSent int64
// handlePrivateDM routes private messages to the specific user pair
func handlePrivateDM(msg Message) {
// Persist private messages for history.
msg.Type = "private"
dmsSent++
// Persist for history
db.Create(&msg)
// Broadcast only to participants (simplified - in production use a map of user connections)
broadcast(msg)
// Deliver to target if online
mutex.Lock()
targetOnline := false
for conn, info := range clientMeta {
if info.Username == msg.Room || strings.Contains(msg.Room, info.Username) {
if err := conn.WriteJSON(msg); err == nil {
targetOnline = true
}
}
}
mutex.Unlock()
if !targetOnline {
// Target offline — message persisted, will appear in history on next join
log.Printf("DM stored for offline user: %s -> room %s", msg.Username, msg.Room)
}
// Also deliver to sender for confirmation
mutex.Lock()
for conn, info := range clientMeta {
if info.Username == msg.Username {
conn.WriteJSON(msg)
break
}
}
mutex.Unlock()
log.Printf("Private DM from %s to room %s: %s", msg.Username, msg.Room, msg.Text)
}
@@ -1935,7 +2346,14 @@ func logModerationAction(action, target, moderator, reason string) {
func handleModerationCommand(msg Message, conn *websocket.Conn) {
parts := strings.Fields(msg.Text)
if len(parts) < 3 {
conn.WriteJSON(Message{Type: "system", Text: "Usage: /mod <action> <target> <reason> or /mute <user> <reason>"})
conn.WriteJSON(Message{Type: "system", Text: "Usage: /mute <user> <duration_m> <reason> or /ban <user> <reason>", Timestamp: time.Now()})
return
}
// Basic moderator check: admin username or configured moderators
adminUser := getAdminUsername()
if msg.Username != adminUser {
conn.WriteJSON(Message{Type: "system", Text: "Only moderators can use moderation commands.", Timestamp: time.Now()})
return
}
@@ -1943,12 +2361,34 @@ func handleModerationCommand(msg Message, conn *websocket.Conn) {
target := parts[1]
reason := strings.Join(parts[2:], " ")
if redisClient != nil {
switch action {
case "mute", "mod":
durationMinutes := 10 // default
if len(parts) >= 3 {
if d, err := strconv.Atoi(parts[2]); err == nil && d > 0 {
durationMinutes = d
if len(parts) > 3 {
reason = strings.Join(parts[3:], " ")
} else {
reason = "no reason given"
}
}
}
redisClient.Set(context.Background(), fmt.Sprintf("muted:%s", target), "1", time.Duration(durationMinutes)*time.Minute)
case "ban":
redisClient.Set(context.Background(), fmt.Sprintf("banned:%s", target), "1", 0) // no expiry
}
}
logModerationAction(action, target, msg.Username, reason)
broadcast(Message{
Room: msg.Room,
Type: "system",
Text: fmt.Sprintf("Moderator %s %s %s for: %s", msg.Username, action, target, reason),
Room: msg.Room,
Username: "system",
Type: "system",
Text: fmt.Sprintf("%s was %s by %s: %s", target, action, msg.Username, reason),
Timestamp: time.Now(),
})
}
@@ -2075,6 +2515,39 @@ func handleAdminChatLogByID(w http.ResponseWriter, r *http.Request) {
}
}
// handleAdminChatLogExport returns chat log as CSV file.
func handleAdminChatLogExport(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
query := r.URL.Query()
room := strings.TrimSpace(query.Get("room"))
dbQuery := db.Model(&Message{})
if room != "" {
dbQuery = dbQuery.Where("room = ?", room)
}
if username := strings.TrimSpace(query.Get("username")); username != "" {
dbQuery = dbQuery.Where("username LIKE ?", "%"+username+"%")
}
var messages []Message
if err := dbQuery.Order("timestamp desc").Limit(10000).Find(&messages).Error; err != nil {
http.Error(w, "failed to load messages", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/csv; charset=utf-8")
w.Header().Set("Content-Disposition", "attachment; filename=chat-log.csv")
fmt.Fprintln(w, "id,room,username,type,text,timestamp")
for _, msg := range messages {
text := strings.ReplaceAll(msg.Text, `"`, `""`)
fmt.Fprintf(w, "%d,%s,%s,%s,\"%s\",%s\n",
msg.ID, msg.Room, msg.Username, msg.Type, text, msg.Timestamp.Format(time.RFC3339))
}
}
// handleAdminChatLogBulk deletes multiple chat log entries by ID
func handleAdminChatLogBulk(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodDelete {
@@ -2109,6 +2582,19 @@ func handleAdminChatLogBulk(w http.ResponseWriter, r *http.Request) {
}
func main() {
// Load env-configurable rate limits
if raw := strings.TrimSpace(os.Getenv("CHAT_RATE_LIMIT_MAX")); raw != "" {
if v, err := strconv.Atoi(raw); err == nil && v > 0 {
rateLimitMax = v
}
}
if raw := strings.TrimSpace(os.Getenv("CHAT_RATE_LIMIT_WINDOW_S")); raw != "" {
if v, err := strconv.Atoi(raw); err == nil && v > 0 {
rateLimitWindow = time.Duration(v) * time.Second
}
}
log.Printf("Rate limit: %d messages per %v", rateLimitMax, rateLimitWindow)
initRedis()
initDB()
@@ -2122,13 +2608,18 @@ func main() {
http.HandleFunc("/api/admin/moderation/actions", requireAdminAuth(handleAdminModerationAction))
http.HandleFunc("/api/admin/analytics", requireAdminAuth(handleAdminAnalytics))
http.HandleFunc("/api/admin/system", requireAdminAuth(handleAdminSystemSummary))
http.HandleFunc("/api/admin/config", requireAdminAuth(handleAdminConfig))
http.HandleFunc("/api/admin/restart", requireAdminAuth(handleAdminRestart))
http.HandleFunc("/api/admin/users", requireAdminAuth(handleAdminUsers))
http.HandleFunc("/api/admin/schedule", requireAdminAuth(handleAdminSchedule))
http.HandleFunc("/api/admin/schedule/overrides", requireAdminAuth(handleAdminScheduleOverrides))
http.HandleFunc("/api/schedule", handlePublicSchedule)
http.HandleFunc("/api/admin/media", requireAdminAuth(handleAdminMedia))
http.HandleFunc("/api/admin/media/uploads", requireAdminAuth(handleAdminMediaUploads))
http.HandleFunc("/api/admin/media/uploads/", requireAdminAuth(handleAdminMediaUploads))
http.HandleFunc("/api/admin/media/", requireAdminAuth(handleAdminMediaByID))
http.HandleFunc("/api/admin/chat-log/bulk", requireAdminAuth(handleAdminChatLogBulk))
http.HandleFunc("/api/admin/chat-log/export", requireAdminAuth(handleAdminChatLogExport))
http.HandleFunc("/api/admin/chat-log/", requireAdminAuth(handleAdminChatLogByID))
http.HandleFunc("/api/admin/chat-log", requireAdminAuth(handleAdminChatLog))
http.HandleFunc("/metrics", handleMetrics)