Files
linearvoid/backend/chat/src/main.go
T
zwitschi b5710548b1
CI / test (push) Successful in 2m21s
CI / docker (push) Successful in 3m2s
CI / release (push) Skipped
Release / release (push) Successful in 26s
fix: botched edit
2026-08-06 18:08:23 +02:00

2652 lines
80 KiB
Go

package main
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"os/exec"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/gorilla/websocket"
"github.com/redis/go-redis/v9"
"gorm.io/driver/postgres"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
// Config from environment (ADR-003 and ADR-007)
var (
redisClient *redis.Client
db *gorm.DB
upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool { return true },
}
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"`
Room string `json:"room"`
Username string `json:"username"`
Text string `json:"text"`
Timestamp time.Time `json:"timestamp"`
Type string `json:"type"` // "public", "private", "system", "moderation"
}
// ModerationLog for audit trail (immutable per ADR-005)
type ModerationLog struct {
ID uint `gorm:"primarykey" json:"id"`
Action string `json:"action"` // "mute", "ban", "delete", "timeout"
Target string `json:"target"`
Moderator string `json:"moderator"`
Reason string `json:"reason"`
Timestamp time.Time `json:"timestamp"`
}
// Media stores admin media library metadata (filesystem + DB model per ADR-006)
type Media struct {
ID uint `gorm:"primarykey" json:"id"`
Filename string `gorm:"index" json:"filename"`
OriginalName string `json:"originalName"`
Path string `json:"path"`
SizeBytes int64 `json:"sizeBytes"`
DurationSeconds int `json:"durationSeconds"`
Status string `gorm:"index" json:"status"`
Title string `json:"title"`
Tags string `json:"tags"`
UploadedAt time.Time `gorm:"autoCreateTime" json:"uploadedAt"`
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"`
Room string `json:"room"`
Online bool `json:"online"`
}
type AdminOverview struct {
Viewers int `json:"viewers"`
StoredMessages int64 `json:"storedMessages"`
ModerationActions int64 `json:"moderationActions"`
MediaCount int64 `json:"mediaCount"`
MediaTotalDurationSeconds int64 `json:"mediaTotalDurationSeconds"`
MediaTotalRuntime string `json:"mediaTotalRuntime"`
RedisEnabled bool `json:"redisEnabled"`
RedisClients int `json:"redisClients"`
RedisMemoryUsed string `json:"redisMemoryUsed"`
DatabaseDriver string `json:"databaseDriver"`
GeneratedAt string `json:"generatedAt"`
}
type PlaylistPayload struct {
Content string `json:"content"`
}
type ModerationActionPayload struct {
Action string `json:"action"`
Target string `json:"target"`
Moderator string `json:"moderator"`
Reason string `json:"reason"`
Room string `json:"room"`
}
type ModerationLogsResponse struct {
Logs []ModerationLog `json:"logs"`
}
type RoomAnalytics struct {
Room string `json:"room"`
MessageCount int64 `json:"messageCount"`
}
type AdminAnalytics struct {
TotalMessages int64 `json:"totalMessages"`
MessagesLast24Hours int64 `json:"messagesLast24Hours"`
DistinctRooms int64 `json:"distinctRooms"`
ModerationActions int64 `json:"moderationActions"`
MediaCount int64 `json:"mediaCount"`
MediaTotalDurationSeconds int64 `json:"mediaTotalDurationSeconds"`
MediaTotalRuntime string `json:"mediaTotalRuntime"`
TopRooms []RoomAnalytics `json:"topRooms"`
GeneratedAt string `json:"generatedAt"`
}
type ServiceHealth struct {
Name string `json:"name"`
Status string `json:"status"`
Detail string `json:"detail"`
Checked string `json:"checked"`
}
type AdminSystemSummary struct {
DatabaseDriver string `json:"databaseDriver"`
PlaylistPath string `json:"playlistPath"`
PlaylistExists bool `json:"playlistExists"`
RedisEnabled bool `json:"redisEnabled"`
RedisMemoryUsed string `json:"redisMemoryUsed"`
Services []ServiceHealth `json:"services"`
GeneratedAt string `json:"generatedAt"`
}
type ActiveUser struct {
Username string `json:"username"`
Room string `json:"room"`
Online bool `json:"online"`
}
type AdminUsersResponse struct {
Users []ActiveUser `json:"users"`
RedisEnabled bool `json:"redisEnabled"`
GeneratedAt string `json:"generatedAt"`
}
type ChatLogEntry struct {
ID uint `json:"id"`
Room string `json:"room"`
Username string `json:"username"`
Text string `json:"text"`
Timestamp time.Time `json:"timestamp"`
Type string `json:"type"`
}
type ChatLogListResponse struct {
Entries []ChatLogEntry `json:"entries"`
Total int64 `json:"total"`
}
type ChatLogBulkDeleteRequest struct {
IDs []uint `json:"ids"`
}
type ScheduleEntry struct {
Index int `json:"index"`
Path string `json:"path"`
Label string `json:"label"`
DurationSeconds int `json:"durationSeconds"`
Runtime string `json:"runtime"`
Status string `json:"status"`
Title string `json:"title"`
}
type AdminScheduleResponse struct {
Entries []ScheduleEntry `json:"entries"`
TotalEntries int `json:"totalEntries"`
MatchedEntries int `json:"matchedEntries"`
TotalDurationSeconds int `json:"totalDurationSeconds"`
TotalRuntime string `json:"totalRuntime"`
GeneratedAt string `json:"generatedAt"`
}
type MediaFileEntry struct {
ID uint `json:"id"`
Name string `json:"name"`
OriginalName string `json:"originalName"`
Path string `json:"path"`
Size int64 `json:"size"`
DurationSeconds int `json:"durationSeconds"`
Runtime string `json:"runtime"`
Status string `json:"status"`
Title string `json:"title"`
Tags string `json:"tags"`
UploadedAt string `json:"uploadedAt"`
ModifiedAt string `json:"modifiedAt"`
}
type AdminMediaResponse struct {
Files []MediaFileEntry `json:"files"`
GeneratedAt string `json:"generatedAt"`
}
type AdminMediaUploadInitRequest struct {
UploadID string `json:"uploadId"`
FileName string `json:"fileName"`
FileSize int64 `json:"fileSize"`
MimeType string `json:"mimeType"`
ChunkSize int64 `json:"chunkSize"`
TotalChunks int `json:"totalChunks"`
}
type AdminMediaUploadSession struct {
UploadID string `json:"uploadId"`
FileName string `json:"fileName"`
FileSize int64 `json:"fileSize"`
MimeType string `json:"mimeType"`
ChunkSize int64 `json:"chunkSize"`
TotalChunks int `json:"totalChunks"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
}
type AdminMediaUploadStatusResponse struct {
UploadID string `json:"uploadId"`
FileName string `json:"fileName"`
FileSize int64 `json:"fileSize"`
ChunkSize int64 `json:"chunkSize"`
TotalChunks int `json:"totalChunks"`
UploadedChunks int `json:"uploadedChunks"`
Progress float64 `json:"progress"`
}
type AdminLoginPayload struct {
Username string `json:"username"`
Password string `json:"password"`
}
type AdminSessionResponse struct {
Authenticated bool `json:"authenticated"`
Username string `json:"username,omitempty"`
}
var databaseDriverName = "sqlite"
const adminSessionCookieName = "linearvoid_admin_session"
const (
maxAdminUploadBytes int64 = 2 * 1024 * 1024 * 1024
defaultAdminChunkSize int64 = 8 * 1024 * 1024
maxAdminChunkSize int64 = 32 * 1024 * 1024
)
// initRedis initializes Redis for presence, rate limiting, and pub/sub (ADR-007)
func initRedis() {
redisAddress := os.Getenv("REDIS_URL")
if redisAddress == "" {
redisAddress = "redis:6379"
}
var options *redis.Options
var err error
if strings.Contains(redisAddress, "://") {
options, err = redis.ParseURL(redisAddress)
if err != nil {
log.Printf("Redis configuration warning: %v (rate limiting and presence will be disabled)", err)
redisClient = nil
return
}
} else {
options = &redis.Options{Addr: redisAddress}
}
redisClient = redis.NewClient(options)
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
databaseURL := strings.TrimSpace(os.Getenv("DATABASE_URL"))
driverName := "sqlite"
switch {
case databaseURL == "":
db, err = gorm.Open(sqlite.Open("chat.db"), &gorm.Config{})
case strings.HasPrefix(databaseURL, "sqlite://"):
sqlitePath := strings.TrimPrefix(databaseURL, "sqlite://")
if sqlitePath == "" {
sqlitePath = "chat.db"
}
db, err = gorm.Open(sqlite.Open(sqlitePath), &gorm.Config{})
case strings.HasPrefix(databaseURL, "postgres://") || strings.HasPrefix(databaseURL, "postgresql://"):
driverName = "postgres"
db, err = gorm.Open(postgres.Open(databaseURL), &gorm.Config{})
default:
log.Fatalf("Unsupported DATABASE_URL scheme: %s", databaseURL)
}
if err != nil {
log.Fatalf("Failed to connect to %s database: %v", driverName, err)
}
databaseDriverName = driverName
db.AutoMigrate(&Message{}, &ModerationLog{}, &Media{}, &ScheduleOverride{})
log.Printf("%s database initialized (chat history, moderation log, media metadata)", strings.Title(driverName))
}
func parseRedisInfoValue(info, key string) string {
prefix := key + ":"
for _, line := range strings.Split(info, "\n") {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, prefix) {
return strings.TrimSpace(strings.TrimPrefix(line, prefix))
}
}
return ""
}
func getAdminOverview() AdminOverview {
overview := AdminOverview{
RedisEnabled: redisClient != nil,
RedisMemoryUsed: "unavailable",
DatabaseDriver: databaseDriverName,
MediaTotalRuntime: "00:00:00",
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
}
mutex.Lock()
overview.Viewers = len(clients)
mutex.Unlock()
if db != nil {
db.Model(&Message{}).Count(&overview.StoredMessages)
db.Model(&ModerationLog{}).Count(&overview.ModerationActions)
db.Model(&Media{}).Count(&overview.MediaCount)
db.Model(&Media{}).Select("COALESCE(SUM(duration_seconds), 0)").Scan(&overview.MediaTotalDurationSeconds)
overview.MediaTotalRuntime = formatDuration(int(overview.MediaTotalDurationSeconds))
}
if redisClient != nil {
ctx := context.Background()
if info, err := redisClient.Info(ctx, "clients", "memory").Result(); err == nil {
if value := parseRedisInfoValue(info, "connected_clients"); value != "" {
if count, convErr := strconv.Atoi(value); convErr == nil {
overview.RedisClients = count
}
}
if value := parseRedisInfoValue(info, "used_memory_human"); value != "" {
overview.RedisMemoryUsed = value
}
}
}
return overview
}
func handleAdminLogin(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var payload AdminLoginPayload
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
if payload.Username != getAdminUsername() || payload.Password != getAdminPassword() {
http.Error(w, "invalid credentials", http.StatusUnauthorized)
return
}
token, err := generateSessionToken()
if err != nil {
http.Error(w, fmt.Sprintf("failed to create session: %v", err), http.StatusInternalServerError)
return
}
expiresAt := time.Now().Add(getAdminSessionTTL())
authSessionsMu.Lock()
authSessions[token] = expiresAt
authSessionsMu.Unlock()
setAdminSessionCookie(w, r, token, expiresAt)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminSessionResponse{Authenticated: true, Username: getAdminUsername()})
}
func handleAdminLogout(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if cookie, err := r.Cookie(adminSessionCookieName); err == nil {
authSessionsMu.Lock()
delete(authSessions, cookie.Value)
authSessionsMu.Unlock()
}
clearAdminSessionCookie(w, r)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminSessionResponse{Authenticated: false})
}
func handleAdminSession(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if !validateAdminSession(r) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminSessionResponse{Authenticated: true, Username: getAdminUsername()})
}
func handleAdminOverview(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(getAdminOverview())
}
func getPlaylistPath() string {
playlistPath := strings.TrimSpace(os.Getenv("PLAYLIST_PATH"))
if playlistPath == "" {
playlistPath = "/app/playlist/playlist.txt"
}
return playlistPath
}
func getMediaDir() string {
mediaDir := strings.TrimSpace(os.Getenv("MEDIA_DIR"))
if mediaDir == "" {
mediaDir = "/app/media"
}
return mediaDir
}
func getAdminUsername() string {
username := strings.TrimSpace(os.Getenv("ADMIN_USERNAME"))
if username == "" {
username = "admin"
}
return username
}
func getAdminPassword() string {
password := strings.TrimSpace(os.Getenv("ADMIN_PASSWORD"))
if password == "" {
password = "admin"
}
return password
}
func getAdminSessionTTL() time.Duration {
if rawHours := strings.TrimSpace(os.Getenv("ADMIN_SESSION_TTL_HOURS")); rawHours != "" {
if hours, err := strconv.Atoi(rawHours); err == nil && hours > 0 {
return time.Duration(hours) * time.Hour
}
}
return 12 * time.Hour
}
func generateSessionToken() (string, error) {
buf := make([]byte, 32)
if _, err := rand.Read(buf); err != nil {
return "", err
}
return hex.EncodeToString(buf), nil
}
func sessionCookieSecure(r *http.Request) bool {
if forwardedProto := strings.TrimSpace(r.Header.Get("X-Forwarded-Proto")); forwardedProto != "" {
return strings.EqualFold(forwardedProto, "https")
}
return r.TLS != nil
}
func setAdminSessionCookie(w http.ResponseWriter, r *http.Request, token string, expiresAt time.Time) {
http.SetCookie(w, &http.Cookie{
Name: adminSessionCookieName,
Value: token,
Path: "/",
HttpOnly: true,
SameSite: http.SameSiteStrictMode,
Secure: sessionCookieSecure(r),
Expires: expiresAt,
MaxAge: int(time.Until(expiresAt).Seconds()),
})
}
func clearAdminSessionCookie(w http.ResponseWriter, r *http.Request) {
http.SetCookie(w, &http.Cookie{
Name: adminSessionCookieName,
Value: "",
Path: "/",
HttpOnly: true,
SameSite: http.SameSiteStrictMode,
Secure: sessionCookieSecure(r),
Expires: time.Unix(0, 0),
MaxAge: -1,
})
}
func validateAdminSession(r *http.Request) bool {
cookie, err := r.Cookie(adminSessionCookieName)
if err != nil || cookie.Value == "" {
return false
}
authSessionsMu.Lock()
defer authSessionsMu.Unlock()
expiresAt, ok := authSessions[cookie.Value]
if !ok {
return false
}
if time.Now().After(expiresAt) {
delete(authSessions, cookie.Value)
return false
}
return true
}
func requireAdminAuth(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if !validateAdminSession(r) {
http.Error(w, "unauthorized", http.StatusUnauthorized)
return
}
next(w, r)
}
}
func parsePlaylistEntries(content string) []ScheduleEntry {
entries := []ScheduleEntry{}
lines := strings.Split(content, "\n")
for _, rawLine := range lines {
line := strings.TrimSpace(rawLine)
if line == "" || strings.HasPrefix(line, "#") {
continue
}
if strings.HasPrefix(line, "file ") {
line = strings.TrimSpace(strings.TrimPrefix(line, "file "))
line = strings.Trim(line, "'\"")
}
entries = append(entries, ScheduleEntry{
Index: len(entries) + 1,
Path: line,
Label: filepath.Base(line),
})
}
return entries
}
func handleAdminPlaylist(w http.ResponseWriter, r *http.Request) {
playlistPath := getPlaylistPath()
switch r.Method {
case http.MethodGet:
content, err := os.ReadFile(playlistPath)
if err != nil {
http.Error(w, fmt.Sprintf("failed to read playlist: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(PlaylistPayload{Content: string(content)})
case http.MethodPut:
var payload PlaylistPayload
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
if err := os.MkdirAll(filepath.Dir(playlistPath), 0o755); err != nil {
http.Error(w, fmt.Sprintf("failed to ensure playlist directory: %v", err), http.StatusInternalServerError)
return
}
content := payload.Content
if content != "" && !strings.HasSuffix(content, "\n") {
content += "\n"
}
if err := os.WriteFile(playlistPath, []byte(content), 0o644); err != nil {
http.Error(w, fmt.Sprintf("failed to save playlist: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(PlaylistPayload{Content: content})
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
// 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)
}
}
func handleAdminSchedule(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, fmt.Sprintf("failed to read schedule playlist: %v", err), http.StatusInternalServerError)
return
}
entries := parsePlaylistEntries(string(content))
var mediaRecords []Media
if err := db.Find(&mediaRecords).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to load media metadata: %v", err), http.StatusInternalServerError)
return
}
mediaByFilename := make(map[string]Media, len(mediaRecords))
mediaByPath := make(map[string]Media, len(mediaRecords))
for _, record := range mediaRecords {
mediaByFilename[strings.ToLower(record.Filename)] = record
mediaByPath[strings.ToLower(filepath.Clean(record.Path))] = record
}
totalDurationSeconds := 0
matchedEntries := 0
for i := range entries {
entry := &entries[i]
entry.Status = "missing"
entry.Runtime = "00:00:00"
cleanPath := strings.ToLower(filepath.Clean(entry.Path))
baseName := strings.ToLower(filepath.Base(entry.Path))
record, ok := mediaByPath[cleanPath]
if !ok {
record, ok = mediaByFilename[baseName]
}
if !ok {
continue
}
entry.DurationSeconds = record.DurationSeconds
entry.Runtime = formatDuration(record.DurationSeconds)
entry.Status = record.Status
entry.Title = record.Title
totalDurationSeconds += record.DurationSeconds
matchedEntries++
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminScheduleResponse{
Entries: entries,
TotalEntries: len(entries),
MatchedEntries: matchedEntries,
TotalDurationSeconds: totalDurationSeconds,
TotalRuntime: formatDuration(totalDurationSeconds),
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
})
}
func handleAdminModerationLogs(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
limit := 50
if rawLimit := strings.TrimSpace(r.URL.Query().Get("limit")); rawLimit != "" {
parsedLimit, err := strconv.Atoi(rawLimit)
if err != nil || parsedLimit <= 0 {
http.Error(w, "invalid limit", http.StatusBadRequest)
return
}
if parsedLimit > 200 {
parsedLimit = 200
}
limit = parsedLimit
}
var logs []ModerationLog
if err := db.Order("timestamp desc").Limit(limit).Find(&logs).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to load moderation logs: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(ModerationLogsResponse{Logs: logs})
}
func handleAdminModerationAction(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var payload ModerationActionPayload
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
payload.Action = strings.TrimSpace(payload.Action)
payload.Target = strings.TrimSpace(payload.Target)
payload.Moderator = strings.TrimSpace(payload.Moderator)
payload.Reason = strings.TrimSpace(payload.Reason)
payload.Room = strings.TrimSpace(payload.Room)
if payload.Action == "" || payload.Target == "" || payload.Reason == "" {
http.Error(w, "action, target, and reason are required", http.StatusBadRequest)
return
}
if payload.Moderator == "" {
payload.Moderator = "admin"
}
if payload.Room == "" {
payload.Room = "main"
}
logModerationAction(payload.Action, payload.Target, payload.Moderator, payload.Reason)
broadcast(Message{
Room: payload.Room,
Username: "system",
Type: "system",
Text: fmt.Sprintf("Moderator %s %s %s for: %s", payload.Moderator, payload.Action, payload.Target, payload.Reason),
Timestamp: time.Now(),
})
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(payload)
}
func getAdminAnalytics() AdminAnalytics {
analytics := AdminAnalytics{
TopRooms: []RoomAnalytics{},
MediaTotalRuntime: "00:00:00",
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
}
if db == nil {
return analytics
}
db.Model(&Message{}).Count(&analytics.TotalMessages)
db.Model(&Message{}).
Where("timestamp >= ?", time.Now().Add(-24*time.Hour)).
Count(&analytics.MessagesLast24Hours)
db.Model(&Message{}).Distinct("room").Count(&analytics.DistinctRooms)
db.Model(&ModerationLog{}).Count(&analytics.ModerationActions)
db.Model(&Media{}).Count(&analytics.MediaCount)
db.Model(&Media{}).Select("COALESCE(SUM(duration_seconds), 0)").Scan(&analytics.MediaTotalDurationSeconds)
analytics.MediaTotalRuntime = formatDuration(int(analytics.MediaTotalDurationSeconds))
var topRooms []RoomAnalytics
db.Model(&Message{}).
Select("room, COUNT(*) as message_count").
Group("room").
Order("message_count DESC").
Limit(5).
Scan(&topRooms)
analytics.TopRooms = topRooms
return analytics
}
func handleAdminAnalytics(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(getAdminAnalytics())
}
func checkServiceHealth(name, url string) ServiceHealth {
checked := time.Now().UTC().Format(time.RFC3339)
resp, err := http.Get(url)
if err != nil {
return ServiceHealth{Name: name, Status: "down", Detail: err.Error(), Checked: checked}
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
status := "healthy"
if resp.StatusCode >= 400 {
status = "degraded"
}
detail := strings.TrimSpace(string(body))
if detail == "" {
detail = resp.Status
}
return ServiceHealth{Name: name, Status: status, Detail: detail, Checked: checked}
}
func getAdminSystemSummary() AdminSystemSummary {
summary := AdminSystemSummary{
DatabaseDriver: databaseDriverName,
PlaylistPath: getPlaylistPath(),
RedisEnabled: redisClient != nil,
RedisMemoryUsed: "unavailable",
GeneratedAt: time.Now().UTC().Format(time.RFC3339),
}
if _, err := os.Stat(summary.PlaylistPath); err == nil {
summary.PlaylistExists = true
}
if redisClient != nil {
if info, err := redisClient.Info(context.Background(), "memory").Result(); err == nil {
if value := parseRedisInfoValue(info, "used_memory_human"); value != "" {
summary.RedisMemoryUsed = value
}
}
}
summary.Services = []ServiceHealth{
{Name: "chat", Status: "healthy", Detail: "local process", Checked: summary.GeneratedAt},
checkServiceHealth("playout", "http://playout:8081/health"),
}
if redisClient != nil {
if err := redisClient.Ping(context.Background()).Err(); err != nil {
summary.Services = append(summary.Services, ServiceHealth{Name: "redis", Status: "down", Detail: err.Error(), Checked: summary.GeneratedAt})
} else {
summary.Services = append(summary.Services, ServiceHealth{Name: "redis", Status: "healthy", Detail: "ping ok", Checked: summary.GeneratedAt})
}
} else {
summary.Services = append(summary.Services, ServiceHealth{Name: "redis", Status: "disabled", Detail: "client unavailable", Checked: summary.GeneratedAt})
}
if sqlDB, err := db.DB(); err == nil {
if err := sqlDB.Ping(); err != nil {
summary.Services = append(summary.Services, ServiceHealth{Name: "database", Status: "down", Detail: err.Error(), Checked: summary.GeneratedAt})
} else {
summary.Services = append(summary.Services, ServiceHealth{Name: "database", Status: "healthy", Detail: "ping ok", Checked: summary.GeneratedAt})
}
}
return summary
}
// 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),
})
}
func handleAdminSystemSummary(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
summary := getAdminSystemSummary()
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(summary)
}
func handleAdminUsers(w http.ResponseWriter, r *http.Request) {
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)
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
}
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
}
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)
}
}
func listMediaFiles() ([]MediaFileEntry, error) {
var records []Media
if err := db.Order("uploaded_at desc").Find(&records).Error; err != nil {
return nil, err
}
files := make([]MediaFileEntry, 0, len(records))
for _, record := range records {
files = append(files, MediaFileEntry{
ID: record.ID,
Name: record.Filename,
OriginalName: record.OriginalName,
Path: record.Path,
Size: record.SizeBytes,
DurationSeconds: record.DurationSeconds,
Runtime: formatDuration(record.DurationSeconds),
Status: record.Status,
Title: record.Title,
Tags: record.Tags,
UploadedAt: record.UploadedAt.UTC().Format(time.RFC3339),
ModifiedAt: record.UpdatedAt.UTC().Format(time.RFC3339),
})
}
sort.Slice(files, func(i, j int) bool {
if files[i].UploadedAt == files[j].UploadedAt {
return files[i].Name < files[j].Name
}
return files[i].UploadedAt > files[j].UploadedAt
})
return files, nil
}
func formatDuration(seconds int) string {
if seconds < 0 {
seconds = 0
}
hours := seconds / 3600
minutes := (seconds % 3600) / 60
remainingSeconds := seconds % 60
return fmt.Sprintf("%02d:%02d:%02d", hours, minutes, remainingSeconds)
}
func getMediaIDFromPath(path string) (uint, error) {
prefix := "/api/admin/media/"
if !strings.HasPrefix(path, prefix) {
return 0, fmt.Errorf("invalid media path")
}
idPart := strings.TrimPrefix(path, prefix)
idPart = strings.TrimSpace(idPart)
if idPart == "" || strings.Contains(idPart, "/") {
return 0, fmt.Errorf("invalid media id")
}
id, err := strconv.ParseUint(idPart, 10, 64)
if err != nil || id == 0 {
return 0, fmt.Errorf("invalid media id")
}
return uint(id), nil
}
func isAllowedMediaExtension(fileName string) bool {
ext := strings.ToLower(filepath.Ext(fileName))
switch ext {
case ".mp4", ".mov", ".mkv", ".webm", ".m4v", ".ts":
return true
default:
return false
}
}
func isAllowedMediaMime(mime string) bool {
mime = strings.TrimSpace(strings.ToLower(mime))
if mime == "" {
return true
}
if strings.HasPrefix(mime, "video/") {
return true
}
return mime == "application/octet-stream"
}
func getUploadWorkspaceDir() string {
return filepath.Join(getMediaDir(), ".uploads")
}
func getUploadSessionDir(uploadID string) string {
return filepath.Join(getUploadWorkspaceDir(), uploadID)
}
func getUploadSessionMetaPath(uploadID string) string {
return filepath.Join(getUploadSessionDir(uploadID), "session.json")
}
func getUploadChunkPath(uploadID string, chunkIndex int) string {
return filepath.Join(getUploadSessionDir(uploadID), fmt.Sprintf("chunk-%06d.part", chunkIndex))
}
func loadUploadSession(uploadID string) (AdminMediaUploadSession, error) {
var session AdminMediaUploadSession
data, err := os.ReadFile(getUploadSessionMetaPath(uploadID))
if err != nil {
return session, err
}
if err := json.Unmarshal(data, &session); err != nil {
return session, err
}
return session, nil
}
func saveUploadSession(session AdminMediaUploadSession) error {
if err := os.MkdirAll(getUploadSessionDir(session.UploadID), 0o755); err != nil {
return err
}
data, err := json.MarshalIndent(session, "", " ")
if err != nil {
return err
}
return os.WriteFile(getUploadSessionMetaPath(session.UploadID), data, 0o644)
}
func countUploadedChunks(uploadID string, totalChunks int) int {
uploaded := 0
for i := 0; i < totalChunks; i++ {
if _, err := os.Stat(getUploadChunkPath(uploadID, i)); err == nil {
uploaded++
}
}
return uploaded
}
func validateUploadID(uploadID string) bool {
if len(uploadID) < 16 || len(uploadID) > 128 {
return false
}
for _, r := range uploadID {
if (r < 'a' || r > 'f') && (r < '0' || r > '9') {
return false
}
}
return true
}
func buildUploadStatus(session AdminMediaUploadSession) AdminMediaUploadStatusResponse {
uploadedChunks := countUploadedChunks(session.UploadID, session.TotalChunks)
progress := 0.0
if session.TotalChunks > 0 {
progress = (float64(uploadedChunks) / float64(session.TotalChunks)) * 100
}
return AdminMediaUploadStatusResponse{
UploadID: session.UploadID,
FileName: session.FileName,
FileSize: session.FileSize,
ChunkSize: session.ChunkSize,
TotalChunks: session.TotalChunks,
UploadedChunks: uploadedChunks,
Progress: progress,
}
}
func uniqueMediaTargetPath(mediaDir, fileName string) string {
baseName := filepath.Base(fileName)
ext := filepath.Ext(baseName)
nameOnly := strings.TrimSuffix(baseName, ext)
candidate := filepath.Join(mediaDir, baseName)
for i := 1; ; i++ {
if _, err := os.Stat(candidate); os.IsNotExist(err) {
return candidate
}
candidate = filepath.Join(mediaDir, fmt.Sprintf("%s-%d%s", nameOnly, i, ext))
}
}
func detectMediaMime(filePath string) (string, error) {
f, err := os.Open(filePath)
if err != nil {
return "", err
}
defer f.Close()
head := make([]byte, 512)
n, err := f.Read(head)
if err != nil && err != io.EOF {
return "", err
}
return http.DetectContentType(head[:n]), nil
}
func mediaEntryFromRecord(record Media) MediaFileEntry {
return MediaFileEntry{
ID: record.ID,
Name: record.Filename,
OriginalName: record.OriginalName,
Path: record.Path,
Size: record.SizeBytes,
DurationSeconds: record.DurationSeconds,
Runtime: formatDuration(record.DurationSeconds),
Status: record.Status,
Title: record.Title,
Tags: record.Tags,
UploadedAt: record.UploadedAt.UTC().Format(time.RFC3339),
ModifiedAt: record.UpdatedAt.UTC().Format(time.RFC3339),
}
}
// 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
}
func probeMediaDurationSeconds(mediaPath string) (int, error) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
out, err := exec.CommandContext(
ctx,
"ffprobe",
"-v", "error",
"-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1",
mediaPath,
).CombinedOutput()
if ctx.Err() == context.DeadlineExceeded {
return 0, fmt.Errorf("ffprobe timed out")
}
if err != nil {
return 0, fmt.Errorf("ffprobe failed: %w (%s)", err, strings.TrimSpace(string(out)))
}
rawDuration := strings.TrimSpace(string(out))
if rawDuration == "" {
return 0, fmt.Errorf("ffprobe returned empty duration")
}
secondsFloat, err := strconv.ParseFloat(rawDuration, 64)
if err != nil {
return 0, fmt.Errorf("failed to parse duration %q: %w", rawDuration, err)
}
if secondsFloat < 0 {
secondsFloat = 0
}
seconds := int(secondsFloat + 0.5)
if secondsFloat > 0 && seconds == 0 {
seconds = 1
}
return seconds, nil
}
func handleAdminMedia(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
files, err := listMediaFiles()
if err != nil {
http.Error(w, fmt.Sprintf("failed to list media files: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminMediaResponse{Files: files, GeneratedAt: time.Now().UTC().Format(time.RFC3339)})
case http.MethodPost:
if err := r.ParseMultipartForm(256 << 20); err != nil {
http.Error(w, fmt.Sprintf("failed to parse upload: %v", err), http.StatusBadRequest)
return
}
file, header, err := r.FormFile("file")
if err != nil {
http.Error(w, fmt.Sprintf("missing file upload: %v", err), http.StatusBadRequest)
return
}
defer file.Close()
fileName := filepath.Base(strings.TrimSpace(header.Filename))
if fileName == "" {
http.Error(w, "missing file name", http.StatusBadRequest)
return
}
if !isAllowedMediaExtension(fileName) {
http.Error(w, "unsupported file type", http.StatusBadRequest)
return
}
if header.Size > maxAdminUploadBytes {
http.Error(w, "file exceeds 2GB limit", http.StatusRequestEntityTooLarge)
return
}
mediaDir := getMediaDir()
if err := os.MkdirAll(mediaDir, 0o755); err != nil {
http.Error(w, fmt.Sprintf("failed to ensure media directory: %v", err), http.StatusInternalServerError)
return
}
targetPath := uniqueMediaTargetPath(mediaDir, fileName)
targetFile, err := os.Create(targetPath)
if err != nil {
http.Error(w, fmt.Sprintf("failed to create media file: %v", err), http.StatusInternalServerError)
return
}
if _, err := io.Copy(targetFile, file); err != nil {
targetFile.Close()
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to store media file: %v", err), http.StatusInternalServerError)
return
}
if err := targetFile.Close(); err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to finalize media file: %v", err), http.StatusInternalServerError)
return
}
fileInfo, err := os.Stat(targetPath)
if err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to read media file info: %v", err), http.StatusInternalServerError)
return
}
if fileInfo.Size() > maxAdminUploadBytes {
_ = os.Remove(targetPath)
http.Error(w, "file exceeds 2GB limit", http.StatusRequestEntityTooLarge)
return
}
detectedMime, err := detectMediaMime(targetPath)
if err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to inspect media file: %v", err), http.StatusBadRequest)
return
}
if !strings.HasPrefix(strings.ToLower(detectedMime), "video/") {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("unsupported media content type: %s", detectedMime), http.StatusBadRequest)
return
}
durationSeconds, err := probeMediaDurationSeconds(targetPath)
if err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to probe media duration: %v", err), http.StatusBadRequest)
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,
OriginalName: fileName,
Path: targetPath,
SizeBytes: fileInfo.Size(),
DurationSeconds: durationSeconds,
Status: status,
Title: strings.TrimSuffix(filename, filepath.Ext(filename)),
Tags: fmt.Sprintf(`"normalization":%v`, normResult.Valid),
}
if err := db.Create(&mediaRecord).Error; err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to persist media metadata: %v", err), http.StatusInternalServerError)
return
}
files, err := listMediaFiles()
if err != nil {
http.Error(w, fmt.Sprintf("failed to list media files: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminMediaResponse{Files: files, GeneratedAt: time.Now().UTC().Format(time.RFC3339)})
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
func handleAdminMediaUploads(w http.ResponseWriter, r *http.Request) {
const basePath = "/api/admin/media/uploads"
if r.URL.Path == basePath {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var payload AdminMediaUploadInitRequest
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
fileName := filepath.Base(strings.TrimSpace(payload.FileName))
if fileName == "" {
http.Error(w, "file name is required", http.StatusBadRequest)
return
}
if !isAllowedMediaExtension(fileName) {
http.Error(w, "unsupported file type", http.StatusBadRequest)
return
}
if payload.FileSize <= 0 {
http.Error(w, "file size must be positive", http.StatusBadRequest)
return
}
if payload.FileSize > maxAdminUploadBytes {
http.Error(w, "file exceeds 2GB limit", http.StatusRequestEntityTooLarge)
return
}
if !isAllowedMediaMime(payload.MimeType) {
http.Error(w, "unsupported mime type", http.StatusBadRequest)
return
}
chunkSize := payload.ChunkSize
if chunkSize <= 0 {
chunkSize = defaultAdminChunkSize
}
if chunkSize > maxAdminChunkSize {
http.Error(w, "chunk size exceeds server limit", http.StatusBadRequest)
return
}
expectedChunks := int((payload.FileSize + chunkSize - 1) / chunkSize)
if payload.TotalChunks <= 0 {
payload.TotalChunks = expectedChunks
}
if payload.TotalChunks != expectedChunks {
http.Error(w, "total chunk count mismatch", http.StatusBadRequest)
return
}
uploadID := strings.TrimSpace(payload.UploadID)
if uploadID != "" && !validateUploadID(uploadID) {
http.Error(w, "invalid upload id", http.StatusBadRequest)
return
}
now := time.Now().UTC().Format(time.RFC3339)
session := AdminMediaUploadSession{}
if uploadID != "" {
loaded, err := loadUploadSession(uploadID)
if err == nil {
if loaded.FileName != fileName || loaded.FileSize != payload.FileSize {
http.Error(w, "upload id does not match selected file", http.StatusConflict)
return
}
session = loaded
} else if !os.IsNotExist(err) {
http.Error(w, fmt.Sprintf("failed to resume upload session: %v", err), http.StatusInternalServerError)
return
}
}
if session.UploadID == "" {
randomID, err := generateSessionToken()
if err != nil {
http.Error(w, fmt.Sprintf("failed to create upload session: %v", err), http.StatusInternalServerError)
return
}
session = AdminMediaUploadSession{
UploadID: randomID,
FileName: fileName,
FileSize: payload.FileSize,
MimeType: payload.MimeType,
ChunkSize: chunkSize,
TotalChunks: payload.TotalChunks,
CreatedAt: now,
UpdatedAt: now,
}
} else {
session.UpdatedAt = now
}
if err := saveUploadSession(session); err != nil {
http.Error(w, fmt.Sprintf("failed to persist upload session: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(buildUploadStatus(session))
return
}
if !strings.HasPrefix(r.URL.Path, basePath+"/") {
http.NotFound(w, r)
return
}
remainder := strings.TrimPrefix(r.URL.Path, basePath+"/")
parts := strings.Split(strings.Trim(remainder, "/"), "/")
if len(parts) == 0 || strings.TrimSpace(parts[0]) == "" {
http.Error(w, "missing upload id", http.StatusBadRequest)
return
}
uploadID := strings.TrimSpace(parts[0])
if !validateUploadID(uploadID) {
http.Error(w, "invalid upload id", http.StatusBadRequest)
return
}
session, err := loadUploadSession(uploadID)
if err != nil {
if os.IsNotExist(err) {
http.Error(w, "upload session not found", http.StatusNotFound)
return
}
http.Error(w, fmt.Sprintf("failed to read upload session: %v", err), http.StatusInternalServerError)
return
}
if len(parts) == 1 {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(buildUploadStatus(session))
return
}
if len(parts) != 2 {
http.NotFound(w, r)
return
}
subresource := parts[1]
switch subresource {
case "chunk":
if r.Method != http.MethodPut {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
chunkIndexRaw := strings.TrimSpace(r.URL.Query().Get("index"))
chunkIndex, err := strconv.Atoi(chunkIndexRaw)
if err != nil || chunkIndex < 0 || chunkIndex >= session.TotalChunks {
http.Error(w, "invalid chunk index", http.StatusBadRequest)
return
}
chunkPath := getUploadChunkPath(uploadID, chunkIndex)
tmpPath := chunkPath + ".tmp"
chunkFile, err := os.Create(tmpPath)
if err != nil {
http.Error(w, fmt.Sprintf("failed to open chunk target: %v", err), http.StatusInternalServerError)
return
}
maxRead := session.ChunkSize + 1
written, copyErr := io.Copy(chunkFile, io.LimitReader(r.Body, maxRead))
closeErr := chunkFile.Close()
if copyErr != nil || closeErr != nil {
_ = os.Remove(tmpPath)
http.Error(w, "failed to persist upload chunk", http.StatusInternalServerError)
return
}
expectedMax := session.ChunkSize
if chunkIndex == session.TotalChunks-1 {
remaining := session.FileSize - (int64(chunkIndex) * session.ChunkSize)
if remaining > 0 {
expectedMax = remaining
}
}
if written <= 0 || written > expectedMax {
_ = os.Remove(tmpPath)
http.Error(w, "invalid chunk size", http.StatusBadRequest)
return
}
if err := os.Rename(tmpPath, chunkPath); err != nil {
_ = os.Remove(tmpPath)
http.Error(w, fmt.Sprintf("failed to finalize chunk: %v", err), http.StatusInternalServerError)
return
}
session.UpdatedAt = time.Now().UTC().Format(time.RFC3339)
if err := saveUploadSession(session); err != nil {
http.Error(w, fmt.Sprintf("failed to update upload session: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(buildUploadStatus(session))
case "complete":
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if countUploadedChunks(uploadID, session.TotalChunks) != session.TotalChunks {
http.Error(w, "upload is incomplete", http.StatusBadRequest)
return
}
mediaDir := getMediaDir()
if err := os.MkdirAll(mediaDir, 0o755); err != nil {
http.Error(w, fmt.Sprintf("failed to ensure media directory: %v", err), http.StatusInternalServerError)
return
}
targetPath := uniqueMediaTargetPath(mediaDir, session.FileName)
targetFile, err := os.Create(targetPath)
if err != nil {
http.Error(w, fmt.Sprintf("failed to create media file: %v", err), http.StatusInternalServerError)
return
}
var totalWritten int64
for i := 0; i < session.TotalChunks; i++ {
chunkPath := getUploadChunkPath(uploadID, i)
chunkReader, err := os.Open(chunkPath)
if err != nil {
targetFile.Close()
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("missing chunk %d", i), http.StatusBadRequest)
return
}
written, err := io.Copy(targetFile, chunkReader)
_ = chunkReader.Close()
if err != nil {
targetFile.Close()
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to assemble media file: %v", err), http.StatusInternalServerError)
return
}
totalWritten += written
}
if err := targetFile.Close(); err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to finalize media file: %v", err), http.StatusInternalServerError)
return
}
if totalWritten != session.FileSize {
_ = os.Remove(targetPath)
http.Error(w, "assembled file size mismatch", http.StatusBadRequest)
return
}
detectedMime, err := detectMediaMime(targetPath)
if err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to inspect media file: %v", err), http.StatusBadRequest)
return
}
if !strings.HasPrefix(strings.ToLower(detectedMime), "video/") {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("unsupported media content type: %s", detectedMime), http.StatusBadRequest)
return
}
durationSeconds, err := probeMediaDurationSeconds(targetPath)
if err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to probe media duration: %v", err), http.StatusBadRequest)
return
}
fileInfo, err := os.Stat(targetPath)
if err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to read media file info: %v", err), http.StatusInternalServerError)
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,
OriginalName: session.FileName,
Path: targetPath,
SizeBytes: fileInfo.Size(),
DurationSeconds: durationSeconds,
Status: status,
Title: strings.TrimSuffix(filename, filepath.Ext(filename)),
Tags: fmt.Sprintf(`"normalization":%v`, normResult.Valid),
}
if err := db.Create(&mediaRecord).Error; err != nil {
_ = os.Remove(targetPath)
http.Error(w, fmt.Sprintf("failed to persist media metadata: %v", err), http.StatusInternalServerError)
return
}
_ = os.RemoveAll(getUploadSessionDir(uploadID))
files, err := listMediaFiles()
if err != nil {
http.Error(w, fmt.Sprintf("failed to list media files: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(AdminMediaResponse{Files: files, GeneratedAt: time.Now().UTC().Format(time.RFC3339)})
default:
http.NotFound(w, r)
}
}
func handleAdminMediaByID(w http.ResponseWriter, r *http.Request) {
mediaID, err := getMediaIDFromPath(r.URL.Path)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
var record Media
if err := db.First(&record, mediaID).Error; err != nil {
http.Error(w, "media item not found", http.StatusNotFound)
return
}
switch r.Method {
case http.MethodGet:
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(mediaEntryFromRecord(record))
case http.MethodPut:
var payload struct {
Status string `json:"status"`
Title string `json:"title"`
Tags string `json:"tags"`
}
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
updates := map[string]any{}
if payload.Status != "" {
updates["status"] = strings.TrimSpace(payload.Status)
}
if payload.Title != "" {
updates["title"] = strings.TrimSpace(payload.Title)
}
updates["tags"] = strings.TrimSpace(payload.Tags)
if err := db.Model(&record).Updates(updates).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to update media metadata: %v", err), http.StatusInternalServerError)
return
}
if err := db.First(&record, mediaID).Error; err != nil {
http.Error(w, "media item not found after update", http.StatusNotFound)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(mediaEntryFromRecord(record))
case http.MethodDelete:
absMediaDir, err := filepath.Abs(getMediaDir())
if err != nil {
http.Error(w, fmt.Sprintf("failed to resolve media directory: %v", err), http.StatusInternalServerError)
return
}
absTargetPath, err := filepath.Abs(record.Path)
if err != nil {
http.Error(w, fmt.Sprintf("failed to resolve media path: %v", err), http.StatusInternalServerError)
return
}
prefix := absMediaDir + string(os.PathSeparator)
if absTargetPath != absMediaDir && !strings.HasPrefix(absTargetPath, prefix) {
http.Error(w, "media path consistency check failed", http.StatusConflict)
return
}
if err := os.Remove(absTargetPath); err != nil && !os.IsNotExist(err) {
http.Error(w, fmt.Sprintf("failed to delete media file: %v", err), http.StatusInternalServerError)
return
}
if err := db.Delete(&record).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to delete media record: %v", err), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{"deleted": true, "id": mediaID})
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
func countConnectedClients() int {
mutex.Lock()
defer mutex.Unlock()
return len(clients)
}
func countActiveAdminSessions() int {
now := time.Now()
authSessionsMu.Lock()
defer authSessionsMu.Unlock()
active := 0
for token, expiresAt := range authSessions {
if now.After(expiresAt) {
delete(authSessions, token)
continue
}
active++
}
return active
}
func writePrometheusMetricLine(w io.Writer, name string, value any) {
fmt.Fprintf(w, "%s %v\n", name, value)
}
func handleMetrics(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var storedMessages int64
if err := db.Model(&Message{}).Count(&storedMessages).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to count messages: %v", err), http.StatusInternalServerError)
return
}
var moderationActions int64
if err := db.Model(&ModerationLog{}).Count(&moderationActions).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to count moderation actions: %v", err), http.StatusInternalServerError)
return
}
redisEnabled := 0
redisUp := 0
redisClients := 0
if redisClient != nil {
redisEnabled = 1
if err := redisClient.Ping(context.Background()).Err(); err == nil {
redisUp = 1
if info, err := redisClient.Info(context.Background(), "clients").Result(); err == nil {
for _, line := range strings.Split(info, "\n") {
if strings.HasPrefix(line, "connected_clients:") {
if value, err := strconv.Atoi(strings.TrimSpace(strings.TrimPrefix(line, "connected_clients:"))); err == nil {
redisClients = value
}
break
}
}
}
}
}
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
fmt.Fprintln(w, "# HELP linearvoid_chat_connected_clients Current connected WebSocket clients.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_connected_clients gauge")
writePrometheusMetricLine(w, "linearvoid_chat_connected_clients", countConnectedClients())
fmt.Fprintln(w, "# HELP linearvoid_chat_messages_total Total stored chat messages.")
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)
fmt.Fprintln(w, "# HELP linearvoid_chat_redis_enabled Redis configured for chat service.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_enabled gauge")
writePrometheusMetricLine(w, "linearvoid_chat_redis_enabled", redisEnabled)
fmt.Fprintln(w, "# HELP linearvoid_chat_redis_up Redis reachable from chat service.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_up gauge")
writePrometheusMetricLine(w, "linearvoid_chat_redis_up", redisUp)
fmt.Fprintln(w, "# HELP linearvoid_chat_redis_connected_clients Redis connected client count seen by chat service.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_connected_clients gauge")
writePrometheusMetricLine(w, "linearvoid_chat_redis_connected_clients", redisClients)
fmt.Fprintln(w, "# HELP linearvoid_chat_admin_sessions_active Active unexpired admin sessions.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_admin_sessions_active gauge")
writePrometheusMetricLine(w, "linearvoid_chat_admin_sessions_active", countActiveAdminSessions())
fmt.Fprintln(w, "# HELP linearvoid_chat_build_info Static build info for chat service.")
fmt.Fprintln(w, "# TYPE linearvoid_chat_build_info gauge")
fmt.Fprintln(w, "linearvoid_chat_build_info{service=\"chat\",database_driver=\""+databaseDriverName+"\"} 1")
}
// broadcast sends a message to all connected clients in a room and saves to DB
func broadcast(msg Message) {
mutex.Lock()
defer mutex.Unlock()
// Persist to configured SQL backend.
db.Create(&msg)
for client := range clients {
if err := client.WriteJSON(msg); err != nil {
client.Close()
delete(clients, client)
}
}
}
// handleConnection handles WebSocket connections for chat (public rooms, private DMs, moderation)
func handleConnection(w http.ResponseWriter, r *http.Request) {
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
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)
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", 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 && room != "" {
redisClient.Del(context.Background(), fmt.Sprintf("presence:%s:%s", room, username))
broadcast(Message{Room: room, Username: "system", Text: username + " left " + room, Type: "system", Timestamp: time.Now()})
}
conn.Close()
}()
for {
var msg Message
if err := conn.ReadJSON(&msg); err != nil {
break
}
msg.Timestamp = time.Now()
msg.Username = username // enforce from query param for security
// 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, rateLimitWindow)
}
if count > int64(rateLimitMax) {
conn.WriteJSON(Message{Type: "system", Text: "Rate limit exceeded. Please slow down.", Timestamp: time.Now()})
continue
}
}
// Handle moderation commands
if strings.HasPrefix(msg.Text, "/mod ") || strings.HasPrefix(msg.Text, "/mute ") || strings.HasPrefix(msg.Text, "/ban ") {
handleModerationCommand(msg, conn)
continue
}
// Handle private DMs
if strings.HasPrefix(msg.Room, "dm:") {
handlePrivateDM(msg)
continue
}
msg.Type = "public"
log.Printf("Chat message in room %s from %s: %s", msg.Room, msg.Username, msg.Text)
broadcast(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) {
msg.Type = "private"
dmsSent++
// Persist for history
db.Create(&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)
}
// logModerationAction logs an action to the immutable ModerationLog table (ADR-005/ADR-007)
func logModerationAction(action, target, moderator, reason string) {
logEntry := ModerationLog{
Action: action,
Target: target,
Moderator: moderator,
Reason: reason,
Timestamp: time.Now(),
}
db.Create(&logEntry)
log.Printf("Moderation action logged: %s %s by %s - %s", action, target, moderator, reason)
}
// handleModerationCommand processes /mod, /mute, /ban commands and logs to ModerationLog
func handleModerationCommand(msg Message, conn *websocket.Conn) {
parts := strings.Fields(msg.Text)
if len(parts) < 3 {
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
}
action := strings.TrimPrefix(parts[0], "/")
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,
Username: "system",
Type: "system",
Text: fmt.Sprintf("%s was %s by %s: %s", target, action, msg.Username, reason),
Timestamp: time.Now(),
})
}
// handleAdminChatLog lists chat log entries with optional filtering and pagination
func handleAdminChatLog(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"))
limit := 50
if rawLimit := strings.TrimSpace(query.Get("limit")); rawLimit != "" {
parsedLimit, err := strconv.Atoi(rawLimit)
if err != nil || parsedLimit <= 0 {
http.Error(w, "invalid limit", http.StatusBadRequest)
return
}
if parsedLimit > 200 {
parsedLimit = 200
}
limit = parsedLimit
}
dbQuery := db.Model(&Message{})
if room != "" {
dbQuery = dbQuery.Where("room = ?", room)
}
if beforeRaw := strings.TrimSpace(query.Get("before")); beforeRaw != "" {
beforeTime, err := time.Parse(time.RFC3339, beforeRaw)
if err != nil {
http.Error(w, "invalid before timestamp, use RFC3339 format", http.StatusBadRequest)
return
}
dbQuery = dbQuery.Where("timestamp < ?", beforeTime)
}
if username := strings.TrimSpace(query.Get("username")); username != "" {
dbQuery = dbQuery.Where("username LIKE ?", "%"+username+"%")
}
if text := strings.TrimSpace(query.Get("text")); text != "" {
dbQuery = dbQuery.Where("text LIKE ?", "%"+text+"%")
}
var total int64
if err := dbQuery.Count(&total).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to count chat log entries: %v", err), http.StatusInternalServerError)
return
}
var messages []Message
if err := dbQuery.Order("timestamp desc").Limit(limit).Find(&messages).Error; err != nil {
http.Error(w, fmt.Sprintf("failed to load chat log entries: %v", err), http.StatusInternalServerError)
return
}
entries := make([]ChatLogEntry, len(messages))
for i, msg := range messages {
entries[i] = ChatLogEntry{
ID: msg.ID,
Room: msg.Room,
Username: msg.Username,
Text: msg.Text,
Timestamp: msg.Timestamp,
Type: msg.Type,
}
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(ChatLogListResponse{Entries: entries, Total: total})
}
// handleAdminChatLogByID returns or deletes a single chat log entry
func handleAdminChatLogByID(w http.ResponseWriter, r *http.Request) {
idPart := strings.TrimPrefix(r.URL.Path, "/api/admin/chat-log/")
idPart = strings.TrimSpace(idPart)
if idPart == "" || strings.Contains(idPart, "/") {
http.Error(w, "invalid chat log id", http.StatusBadRequest)
return
}
id, err := strconv.ParseUint(idPart, 10, 64)
if err != nil || id == 0 {
http.Error(w, "invalid chat log id", http.StatusBadRequest)
return
}
switch r.Method {
case http.MethodGet:
var msg Message
if err := db.First(&msg, id).Error; err != nil {
http.Error(w, "chat log entry not found", http.StatusNotFound)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(ChatLogEntry{
ID: msg.ID,
Room: msg.Room,
Username: msg.Username,
Text: msg.Text,
Timestamp: msg.Timestamp,
Type: msg.Type,
})
case http.MethodDelete:
result := db.Delete(&Message{}, id)
if result.Error != nil {
http.Error(w, fmt.Sprintf("failed to delete chat log entry: %v", result.Error), http.StatusInternalServerError)
return
}
if result.RowsAffected == 0 {
http.Error(w, "chat log entry not found", http.StatusNotFound)
return
}
w.WriteHeader(http.StatusNoContent)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
}
// 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 {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
var payload ChatLogBulkDeleteRequest
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, fmt.Sprintf("invalid request body: %v", err), http.StatusBadRequest)
return
}
if len(payload.IDs) == 0 {
http.Error(w, "ids array is required", http.StatusBadRequest)
return
}
if len(payload.IDs) > 500 {
http.Error(w, "maximum 500 ids per bulk request", http.StatusBadRequest)
return
}
result := db.Delete(&Message{}, payload.IDs)
if result.Error != nil {
http.Error(w, fmt.Sprintf("failed to bulk delete chat log entries: %v", result.Error), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{"deleted": result.RowsAffected})
}
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()
http.HandleFunc("/ws", handleConnection)
http.HandleFunc("/api/admin/login", handleAdminLogin)
http.HandleFunc("/api/admin/logout", handleAdminLogout)
http.HandleFunc("/api/admin/session", handleAdminSession)
http.HandleFunc("/api/admin/overview", requireAdminAuth(handleAdminOverview))
http.HandleFunc("/api/admin/playlist", requireAdminAuth(handleAdminPlaylist))
http.HandleFunc("/api/admin/moderation/logs", requireAdminAuth(handleAdminModerationLogs))
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)
http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
fmt.Fprint(w, `{"status":"healthy","service":"chat","timestamp":"`+time.Now().UTC().Format(time.RFC3339)+`"}`)
})
// Catch-all: return 404 JSON for all unhandled paths
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusNotFound)
fmt.Fprint(w, `{"error":"not found"}`)
})
port := os.Getenv("CHAT_PORT")
if port == "" {
port = "3000"
}
log.Printf("Chat Backend starting on :%s (WebSocket at /ws, health at /health, database persistence enabled)", port)
log.Fatal(http.ListenAndServe(":"+port, nil))
}