diff --git a/.env.example b/.env.example index 913807a..7b5584f 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,12 @@ MEDIA_DOMAIN=media.linearvoid.tv VITE_PUBLIC_APP_URL=https://linearvoid.tv VITE_PUBLIC_WS_URL=wss://linearvoid.tv/ws VITE_PUBLIC_HLS_URL=https://media.linearvoid.tv/hls/playlist.m3u8 +VITE_PUBLIC_ADMIN_DEFAULT_MODERATOR=admin +VITE_PUBLIC_ADMIN_DEFAULT_ROOM=main +VITE_PUBLIC_ADMIN_MODERATION_LOG_LIMIT=50 +ADMIN_USERNAME=admin +ADMIN_PASSWORD=change-me +ADMIN_SESSION_TTL_HOURS=12 # Optional direct-local fallback when bypassing the public proxy: # VITE_PUBLIC_APP_URL=http://localhost # VITE_PUBLIC_WS_URL=ws://localhost/ws diff --git a/.gitignore b/.gitignore index 0b91d78..e69b273 100644 --- a/.gitignore +++ b/.gitignore @@ -82,7 +82,7 @@ grafana/ # Agent / Tool files .github/instructions/TODO.md .github/instructions/DONE.md -# (keep AGENTS.md and other instruction files) +# (keep AGENTS.md, MEMORY.md, and other instruction files) # Temporary architecture artifacts **/*.tmp diff --git a/CHANGELOG.md b/CHANGELOG.md index 7900f26..be4751d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ +## [1.0.3] - 2026-08-03 + +### Added + +- Added Prometheus-format `/metrics` endpoint for chat runtime and admin-session monitoring. + +### Changed + +- Updated Prometheus scrape targets and Grafana dashboard queries to use live chat metrics. + +### Fixed + +- Fixed nginx routing so `/metrics` is reachable through localhost and `linearvoid.tv`. + ## [1.0.2] - 2026-08-03 ### Changed diff --git a/SETUP.md b/SETUP.md index 435573b..d943bfa 100644 --- a/SETUP.md +++ b/SETUP.md @@ -58,6 +58,12 @@ This guide covers prerequisites, environment setup, running the application loca VITE_PUBLIC_APP_URL=https://linearvoid.tv VITE_PUBLIC_WS_URL=wss://linearvoid.tv/ws VITE_PUBLIC_HLS_URL=https://media.linearvoid.tv/hls/playlist.m3u8 + VITE_PUBLIC_ADMIN_DEFAULT_MODERATOR=admin + VITE_PUBLIC_ADMIN_DEFAULT_ROOM=main + VITE_PUBLIC_ADMIN_MODERATION_LOG_LIMIT=50 + ADMIN_USERNAME=admin + ADMIN_PASSWORD=change-me + ADMIN_SESSION_TTL_HOURS=12 # Optional direct-local fallback when bypassing the public proxy: # VITE_PUBLIC_APP_URL=http://localhost # VITE_PUBLIC_WS_URL=ws://localhost/ws @@ -188,7 +194,7 @@ This guide covers prerequisites, environment setup, running the application loca 2. **Access the Application**: - Preferred local access with the documented proxy setup: - - Admin Interface: + - Admin Interface: (redirects to `/admin/login/` until authenticated) - Chat WebSocket: `wss://linearvoid.tv/ws` - Direct-local fallback when bypassing Cloudflare/Nginx Proxy Manager: , , and `ws://localhost/ws` - Health Checks: @@ -299,7 +305,7 @@ See `docs/architecture/README.md` for the full arc42 documentation and the ADRs - **Chat Not Connecting**: Check Redis connection and WebSocket URL. The health check at `/health` should return 200. - **Cache Not Working**: Verify Cloudflare cache rules (`.ts` = long TTL immutable, `.m3u8` = short TTL). Check cache hit ratio in Cloudflare dashboard. - **Docker Issues**: Run `docker compose logs` for service-specific errors. Ensure ports are not in use. -- **Admin Interface**: Ensure the protected route is configured in nginx and authentication is set up (default is open for local development). +- **Admin Interface**: Ensure `ADMIN_USERNAME`, `ADMIN_PASSWORD`, and `ADMIN_SESSION_TTL_HOURS` are set. `/admin/` and `/api/admin/*` are protected by a session-based login flow. - **Production Pull Fails**: Confirm `docker login` to Gitea succeeded on the server and the image tags referenced by `docker-compose.prod.yml` exist. ## Further Reading diff --git a/backend/chat/src/main.go b/backend/chat/src/main.go index c75de02..a422c56 100644 --- a/backend/chat/src/main.go +++ b/backend/chat/src/main.go @@ -2,10 +2,17 @@ package main import ( "context" + "crypto/rand" + "encoding/hex" + "encoding/json" "fmt" + "io" "log" "net/http" "os" + "path/filepath" + "sort" + "strconv" "strings" "sync" "time" @@ -26,6 +33,8 @@ var ( } clients = make(map[*websocket.Conn]bool) mutex = sync.Mutex{} + authSessions = map[string]time.Time{} + authSessionsMu = sync.Mutex{} ) // Message represents a chat message (persisted in SQLite per ADR-006) @@ -55,6 +64,112 @@ type Presence struct { Online bool `json:"online"` } +type AdminOverview struct { + Viewers int `json:"viewers"` + StoredMessages int64 `json:"storedMessages"` + ModerationActions int64 `json:"moderationActions"` + 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"` + 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 ScheduleEntry struct { + Index int `json:"index"` + Path string `json:"path"` + Label string `json:"label"` +} + +type AdminScheduleResponse struct { + Entries []ScheduleEntry `json:"entries"` + GeneratedAt string `json:"generatedAt"` +} + +type MediaFileEntry struct { + Name string `json:"name"` + Size int64 `json:"size"` + ModifiedAt string `json:"modifiedAt"` +} + +type AdminMediaResponse struct { + Files []MediaFileEntry `json:"files"` + GeneratedAt string `json:"generatedAt"` +} + +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" + // initRedis initializes Redis for presence, rate limiting, and pub/sub (ADR-007) func initRedis() { redisAddress := os.Getenv("REDIS_URL") @@ -110,10 +225,752 @@ func initDB() { log.Fatalf("Failed to connect to %s database: %v", driverName, err) } + databaseDriverName = driverName db.AutoMigrate(&Message{}, &ModerationLog{}) log.Printf("%s database initialized (chat history and moderation log)", 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, + 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) + } + + 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) + } +} + +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 + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(AdminScheduleResponse{ + Entries: parsePlaylistEntries(string(content)), + 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{}, + 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) + + 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 +} + +func handleAdminSystemSummary(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(getAdminSystemSummary()) +} + +func handleAdminUsers(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + 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) +} + +func listMediaFiles() ([]MediaFileEntry, error) { + mediaDir := getMediaDir() + entries, err := os.ReadDir(mediaDir) + if err != nil { + return nil, err + } + + files := make([]MediaFileEntry, 0, len(entries)) + for _, entry := range entries { + if entry.IsDir() { + continue + } + + info, err := entry.Info() + if err != nil { + continue + } + + files = append(files, MediaFileEntry{ + Name: entry.Name(), + Size: info.Size(), + ModifiedAt: info.ModTime().UTC().Format(time.RFC3339), + }) + } + + sort.Slice(files, func(i, j int) bool { + return files[i].Name < files[j].Name + }) + + return files, 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() + + 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 := filepath.Join(mediaDir, filepath.Base(header.Filename)) + targetFile, err := os.Create(targetPath) + if err != nil { + http.Error(w, fmt.Sprintf("failed to create media file: %v", err), http.StatusInternalServerError) + return + } + defer targetFile.Close() + + if _, err := io.Copy(targetFile, file); err != nil { + http.Error(w, fmt.Sprintf("failed to store media file: %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 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_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() @@ -253,6 +1110,19 @@ func main() { 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/users", requireAdminAuth(handleAdminUsers)) + http.HandleFunc("/api/admin/schedule", requireAdminAuth(handleAdminSchedule)) + http.HandleFunc("/api/admin/media", requireAdminAuth(handleAdminMedia)) + 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) diff --git a/backend/playout/src/playlist.go b/backend/playout/src/playlist.go index 65e4061..479924f 100644 --- a/backend/playout/src/playlist.go +++ b/backend/playout/src/playlist.go @@ -143,6 +143,46 @@ func HealthHandler(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{"status":"healthy","service":"playout","timestamp":"`+time.Now().UTC().Format(time.RFC3339)+`"}`) } +func resetHLSOutput(hlsDir string) error { + if err := os.MkdirAll(hlsDir, 0o755); err != nil { + return err + } + + matches, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.ts")) + if err != nil { + return err + } + + for _, match := range matches { + if err := os.Remove(match); err != nil && !os.IsNotExist(err) { + return err + } + } + + fmp4Matches, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.m4s")) + if err != nil { + return err + } + + for _, match := range fmp4Matches { + if err := os.Remove(match); err != nil && !os.IsNotExist(err) { + return err + } + } + + manifestPath := filepath.Join(hlsDir, "playlist.m3u8") + if err := os.Remove(manifestPath); err != nil && !os.IsNotExist(err) { + return err + } + + initPath := filepath.Join(hlsDir, "init.mp4") + if err := os.Remove(initPath); err != nil && !os.IsNotExist(err) { + return err + } + + return nil +} + func main() { playlistPath := "/app/playlist/playlist.txt" if len(os.Args) > 1 { @@ -176,16 +216,29 @@ func main() { // Run FFmpeg in a loop for continuous playback with R2 sync hook for { - fmt.Printf("[%s] Starting FFmpeg playout with -c copy (immutable segments for Cloudflare caching)...\n", time.Now().Format(time.RFC3339)) + if err := resetHLSOutput("/app/hls"); err != nil { + fmt.Println("Failed to reset HLS output:", err) + time.Sleep(5 * time.Second) + continue + } + + fmt.Printf("[%s] Starting FFmpeg playout with browser-safe H.264/AAC HLS encoding...\n", time.Now().Format(time.RFC3339)) cmd := exec.Command("ffmpeg", "-re", "-f", "concat", "-safe", "0", "-i", playlistPath, - "-c:v", "copy", "-c:a", "copy", + "-map", "0:v:0", "-map", "0:a:0?", + "-vf", "scale=1280:720:force_original_aspect_ratio=decrease,pad=1280:720:(ow-iw)/2:(oh-ih)/2,fps=30,format=yuv420p", + "-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p", + "-profile:v", "high", "-level", "4.0", + "-r", "30", "-g", "60", "-keyint_min", "60", "-sc_threshold", "0", + "-c:a", "aac", "-ar", "48000", "-ac", "2", "-b:a", "128k", "-f", "hls", "-hls_time", "6", "-hls_list_size", "24", + "-hls_segment_type", "fmp4", + "-hls_fmp4_init_filename", "init.mp4", "-strftime", "1", - "-hls_flags", "delete_segments+append_list+independent_segments", - "-hls_segment_filename", "/app/hls/segment_%Y%m%d_%H%M%S.ts", + "-hls_flags", "delete_segments+independent_segments", + "-hls_segment_filename", "/app/hls/segment_%Y%m%d_%H%M%S.m4s", "/app/hls/playlist.m3u8") if err := cmd.Run(); err != nil { @@ -203,7 +256,7 @@ func main() { // syncLatestToR2 uploads new segments and the playlist manifest to R2 func syncLatestToR2(hlsDir string) error { - files, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.ts")) + files, err := filepath.Glob(filepath.Join(hlsDir, "segment_*.m4s")) if err != nil { return err } @@ -222,6 +275,10 @@ func syncLatestToR2(hlsDir string) error { if err := SyncToR2(manifestPath, "live/playlist.m3u8"); err != nil { return err } + initPath := filepath.Join(hlsDir, "init.mp4") + if err := SyncToR2(initPath, "live/init.mp4"); err != nil { + return err + } fmt.Println("Synced live playlist.m3u8 to R2") return nil diff --git a/backend/playout/start-playout.sh b/backend/playout/start-playout.sh index 9ec1f09..b9977a1 100644 --- a/backend/playout/start-playout.sh +++ b/backend/playout/start-playout.sh @@ -26,17 +26,27 @@ fi echo "Playlist validated successfully. Starting continuous HLS playout..." +mkdir -p /app/hls +rm -f /app/hls/playlist.m3u8 /app/hls/init.mp4 /app/hls/segment_*.ts /app/hls/segment_*.m4s + # Run FFmpeg in a loop for continuous playback while true; do - echo "[$(date)] Starting FFmpeg playout with -c copy (stream copy mode)..." + echo "[$(date)] Starting FFmpeg playout with browser-safe HLS encoding..." ffmpeg -re -f concat -safe 0 -i "$PLAYLIST_PATH" \ - -c:v copy -c:a copy \ + -map 0:v:0 -map 0:a:0? \ + -vf "scale=1280:720:force_original_aspect_ratio=decrease,pad=1280:720:(ow-iw)/2:(oh-ih)/2,fps=30,format=yuv420p" \ + -c:v libx264 -preset veryfast -pix_fmt yuv420p \ + -profile:v high -level 4.0 \ + -r 30 -g 60 -keyint_min 60 -sc_threshold 0 \ + -c:a aac -ar 48000 -ac 2 -b:a 128k \ -f hls \ -hls_time "${HLS_SEGMENT_DURATION}" \ -hls_list_size "${HLS_WINDOW_SIZE}" \ + -hls_segment_type fmp4 \ + -hls_fmp4_init_filename init.mp4 \ -strftime 1 \ - -hls_flags delete_segments+append_list \ - -hls_segment_filename "/app/hls/segment_%Y%m%d_%H%M%S.ts" \ + -hls_flags delete_segments+independent_segments \ + -hls_segment_filename "/app/hls/segment_%Y%m%d_%H%M%S.m4s" \ "/app/hls/playlist.m3u8" || echo "FFmpeg exited with error (will restart in 5s per ADR-001 resilience requirements)..." echo "Playout loop crashed or ended. Restarting in 5 seconds..." diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 618518f..462f17a 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -30,6 +30,8 @@ services: - "${CHAT_PORT:-3000}:3000" volumes: - ./data/chat:/app:rw + - ./data/playlist:/app/playlist:rw + - ./data/media:/app/media:rw networks: - linearvoid-net depends_on: diff --git a/docker-compose.yml b/docker-compose.yml index 34fd56b..f4f404e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -36,6 +36,8 @@ services: - "${CHAT_PORT:-3000}:3000" volumes: - ./data/chat:/app:rw + - ./data/playlist:/app/playlist:rw + - ./data/media:/app/media:rw networks: - linearvoid-net depends_on: diff --git a/frontend/admin/index.html b/frontend/admin/index.html index 57e0811..c605532 100644 --- a/frontend/admin/index.html +++ b/frontend/admin/index.html @@ -10,7 +10,7 @@ - +
@@ -18,15 +18,18 @@
LV
Admin
+
+ +
@@ -41,26 +44,272 @@
-
Cache Hit Ratio
-
95%
+
Stored Messages
+
-
R2 Usage
-
12.4 GB
+
Moderation Actions
+
+
+ + + + + + + + @@ -75,15 +324,15 @@
Memory Used - 248 MB +
- Connected Clients - 12 + Connected Redis Clients +
- Rate Limit Threshold - + Redis Enabled +
@@ -91,41 +340,462 @@

Storage

- SQLite Size - 142 MB + Database Driver + +
+
+ Stored Messages + +
+
+ Last Refresh + +
+
+ Playlist Path + +
+
+ Playlist Present +
+
+
+

Service Health

+ +
+
+ +
+
+
+ + + + + + + + +
+
+
Linear Void
+

Admin Login

+

Authenticate to access protected admin controls.

+
+ +
+ + + +
+
+
+ + + + + \ No newline at end of file diff --git a/frontend/index.html b/frontend/index.html index df7b454..0dbcbf0 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -11,7 +11,7 @@ -
+
@@ -29,7 +29,7 @@
- +