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 } // Rate limit: max 5 attempts per IP per 5 minutes clientIP := r.Header.Get("X-Forwarded-For") if clientIP == "" { clientIP = r.RemoteAddr } if redisClient != nil { loginKey := fmt.Sprintf("login_attempts:%s", clientIP) count, _ := redisClient.Incr(context.Background(), loginKey).Result() if count == 1 { redisClient.Expire(context.Background(), loginKey, 5*time.Minute) } if count > 5 { http.Error(w, "too many login attempts, try again later", http.StatusTooManyRequests) 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) { // IP-based connection rate limiting: max 5 concurrent per IP clientIP := r.Header.Get("X-Forwarded-For") if clientIP == "" { clientIP = r.RemoteAddr } if redisClient != nil { ipKey := fmt.Sprintf("ws_conn:%s", clientIP) count, _ := redisClient.Incr(context.Background(), ipKey).Result() if count == 1 { redisClient.Expire(context.Background(), ipKey, 60*time.Second) } if count > 5 { http.Error(w, "too many connections from this IP", http.StatusTooManyRequests) return } // Decrement on disconnect defer redisClient.Decr(context.Background(), ipKey) } 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 or /ban ", 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)) }