Include readable user identity in activity and stream logs

internal/app/activity_logging_integration_test.go

@@ -36,52 +36,80 @@ func captureActivityLogs(t *testing.T) func() string {
 }
 
 func TestSignInAndSettingsActivityLogs(t *testing.T) {
-	logs := captureActivityLogs(t)
-	p := newAuthTestProvider(t)
-	p.role = "admin"
-	a := newAuthTestApp(t, p, nil)
-	var identity struct{ User AuthUser }
-	if err := json.Unmarshal([]byte(a.login(t)), &identity); err != nil || identity.User.ID == "" {
-		t.Fatal("missing signed-in identity", err)
-	}
-	for _, expected := range []string{"Sign-in started", "User signed in", "user_id=" + identity.User.ID, "role=admin"} {
-		if !strings.Contains(logs(), expected) {
-			t.Fatalf("missing %q: %s", expected, logs())
-		}
-	}
-	before := logs()
-	for range 5 {
-		a.get(t, "/api/auth").Body.Close()
-		a.get(t, "/api/status").Body.Close()
-	}
-	if logs() != before {
-		t.Fatal("routine auth/status polling produced activity logs", logs())
-	}
-	resp := authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`)
-	resp.Body.Close()
-	if resp.StatusCode != http.StatusOK || !strings.Contains(logs(), "Settings updated") || !strings.Contains(logs(), "sections=[discovery]") {
-		t.Fatal("settings change was not logged", resp.StatusCode, logs())
-	}
-	before = logs()
-	resp = authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`)
-	resp.Body.Close()
-	if logs() != before {
-		t.Fatal("unchanged settings produced activity logs", logs())
-	}
-	resp = authPlaybackRequest(t, a, "POST", "/auth/logout", "")
-	resp.Body.Close()
-	if resp.StatusCode != http.StatusNoContent || !strings.Contains(logs(), "User signed out") {
-		t.Fatal("sign-out was not logged", resp.StatusCode, logs())
-	}
-	resp = a.get(t, "/auth/callback?code=private-code&state=private-state&error=private-provider-error")
-	resp.Body.Close()
-	if !strings.Contains(logs(), "Sign-in failed") || !strings.Contains(logs(), "stage=callback_validation") {
-		t.Fatal("failed sign-in stage was not logged", logs())
-	}
-	for _, secret := range []string{"private-", "opaque-access-token", "opaque-refresh-token", "test-secret", p.name, p.server.URL} {
-		if strings.Contains(logs(), secret) {
-			t.Fatalf("activity logs exposed %q: %s", secret, logs())
-		}
+	for _, tc := range []struct{ label, name, username, wantName string }{
+		{"display name", "Alex", "alex-user", "Alex"},
+		{"username", "  ", "alex-user", "alex-user"},
+		{"user ID", "", "", ""},
+	} {
+		t.Run(tc.label, func(t *testing.T) {
+			logs := captureActivityLogs(t)
+			p := newAuthTestProvider(t)
+			p.role = "admin"
+			p.name = tc.name
+			p.claims = func(claims map[string]any) { claims["preferred_username"] = tc.username }
+			a := newAuthTestApp(t, p, nil)
+			var identity struct{ User AuthUser }
+			if err := json.Unmarshal([]byte(a.login(t)), &identity); err != nil || identity.User.ID == "" || identity.User.Name != tc.wantName {
+				t.Fatal("missing signed-in identity", err)
+			}
+			for _, expected := range []string{"Sign-in started", "User signed in", "user_id=" + identity.User.ID, "role=admin"} {
+				if !strings.Contains(logs(), expected) {
+					t.Fatalf("missing %q: %s", expected, logs())
+				}
+			}
+			before := logs()
+			for range 5 {
+				a.get(t, "/api/auth").Body.Close()
+				a.get(t, "/api/status").Body.Close()
+			}
+			if logs() != before {
+				t.Fatal("routine auth/status polling produced activity logs", logs())
+			}
+			resp := authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`)
+			resp.Body.Close()
+			if resp.StatusCode != http.StatusOK || !strings.Contains(logs(), "Settings updated") || !strings.Contains(logs(), "sections=[discovery]") {
+				t.Fatal("settings change was not logged", resp.StatusCode, logs())
+			}
+			before = logs()
+			resp = authPlaybackRequest(t, a, "PATCH", "/api/settings", `{"discovery":{"tmdbReadAccessToken":"private-metadata-token"}}`)
+			resp.Body.Close()
+			if logs() != before {
+				t.Fatal("unchanged settings produced activity logs", logs())
+			}
+			resp = authPlaybackRequest(t, a, "POST", "/auth/logout", "")
+			resp.Body.Close()
+			if resp.StatusCode != http.StatusNoContent || !strings.Contains(logs(), "User signed out") {
+				t.Fatal("sign-out was not logged", resp.StatusCode, logs())
+			}
+			resp = a.get(t, "/auth/callback?code=private-code&state=private-state&error=private-provider-error")
+			resp.Body.Close()
+			if !strings.Contains(logs(), "Sign-in failed") || !strings.Contains(logs(), "stage=callback_validation") {
+				t.Fatal("failed sign-in stage was not logged", logs())
+			}
+			wantUser := tc.wantName
+			if wantUser == "" {
+				wantUser = identity.User.ID
+			}
+			for _, message := range []string{"User signed in", "Settings updated", "User signed out"} {
+				found := false
+				for _, line := range strings.Split(logs(), "\n") {
+					if strings.Contains(line, `msg="`+message+`"`) {
+						found = true
+						if !strings.Contains(line, "user="+wantUser) || !strings.Contains(line, "user_id="+identity.User.ID) {
+							t.Fatalf("missing user context for %q: %s", message, line)
+						}
+					}
+				}
+				if !found {
+					t.Fatalf("missing activity %q", message)
+				}
+			}
+			for _, secret := range []string{"private-", "opaque-access-token", "opaque-refresh-token", "test-secret", p.server.URL} {
+				if strings.Contains(logs(), secret) {
+					t.Fatalf("activity logs exposed %q: %s", secret, logs())
+				}
+			}
+		})
 	}
 }
 

internal/app/auth.go

@@ -88,10 +88,10 @@ type authClaims struct {
 
 func (c authClaims) valid() bool { return c.Role == "viewer" || c.Role == "admin" }
 func (c authClaims) name() string {
-	if c.Name != "" {
-		return c.Name
+	if name := strings.TrimSpace(c.Name); name != "" {
+		return name
 	}
-	return c.Username
+	return strings.TrimSpace(c.Username)
 }
 
 func newAuth(cfg AuthConfig, db *sql.DB) (*authManager, error) {
@@ -350,18 +350,24 @@ func (a *authManager) callback(w http.ResponseWriter, r *http.Request) {
 	a.sessions.Clear(ctx)
 	a.sessions.Put(ctx, "user_id", userID)
 	a.sessions.Put(ctx, "user_version", version)
-	slog.Info("User signed in", "user_id", userID, "role", claims.Role)
+	slog.Info("User signed in", append(userLogAttrs(&AuthUser{ID: userID, Name: claims.name()}), "role", claims.Role)...)
 	http.Redirect(w, r, "/", http.StatusSeeOther)
 }
 func (a *authManager) logout(w http.ResponseWriter, r *http.Request) {
 	userID := a.sessions.GetString(r.Context(), "user_id")
+	var user *AuthUser
+	if userID != "" {
+		user = &AuthUser{ID: userID}
+		// Read the saved display name without refreshing provider credentials.
+		_ = a.db.QueryRowContext(r.Context(), `SELECT name FROM auth_users WHERE id=?`, userID).Scan(&user.Name)
+	}
 	if err := a.sessions.Destroy(r.Context()); err != nil {
-		slog.Error("Sign-out failed", "stage", "session_destroy", "user_id", userID)
+		slog.Error("Sign-out failed", append(userLogAttrs(user), "stage", "session_destroy")...)
 		fail(w, 500, "auth", "Unable to sign out.")
 		return
 	}
 	if userID != "" {
-		slog.Info("User signed out", "user_id", userID)
+		slog.Info("User signed out", userLogAttrs(user)...)
 	}
 	w.WriteHeader(http.StatusNoContent)
 }
@@ -405,7 +411,7 @@ func (a *authManager) user(r *http.Request) *AuthUser {
 		ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), 3*time.Second)
 		defer cancel()
 		_, _ = a.db.ExecContext(ctx, `UPDATE auth_users SET role='',checked_at=0,session_version=session_version+1 WHERE id=?`, id)
-		slog.Warn("User access revoked", "user_id", id, "stage", "session_revalidation", "reason", reason)
+		slog.Warn("User access revoked", append(userLogAttrs(&u), "stage", "session_revalidation", "reason", reason)...)
 		return nil
 	}
 	var token oauth2.Token
@@ -445,9 +451,10 @@ func (a *authManager) user(r *http.Request) *AuthUser {
 	if err != nil {
 		return revoke("user_save_failed")
 	}
-	if u.Role != claims.Role {
-		slog.Info("User role changed", "user_id", id, "previous_role", u.Role, "role", claims.Role)
-	}
+	previousRole := u.Role
 	u.Name, u.Role = claims.name(), claims.Role
+	if previousRole != u.Role {
+		slog.Info("User role changed", append(userLogAttrs(&u), "previous_role", previousRole, "role", u.Role)...)
+	}
 	return &u
 }

internal/app/auth_playback_integration_test.go

@@ -100,7 +100,7 @@ func TestOIDCPlaybackBelongsToItsUser(t *testing.T) {
 		t.Fatal("owner could not stop playback", resp.StatusCode)
 	}
 	output := logs()
-	if strings.Count(output, "Stream starting") != 1 || !strings.Contains(output, "session_id=test-1") || !strings.Contains(output, "user_id=") {
+	if strings.Count(output, "Stream starting") != 1 || !strings.Contains(output, "session_id=test-1") || !strings.Contains(output, "user_id=") || !strings.Contains(output, "user=Alex") {
 		t.Fatal("stream start is missing its user/session context", output)
 	}
 	if strings.Contains(output, "secret-command") || strings.Contains(output, "opaque-access-token") || strings.Contains(output, "opaque-refresh-token") {

internal/app/logging.go

@@ -3,17 +3,26 @@ package app
 import (
 	"errors"
 	"net/http"
+	"strings"
 
 	"github.com/pushpinderbal/restream/internal/stalker"
 )
 
 // Log classifications instead of upstream error text: URLs, tokens and response
-// bodies can contain credentials. Internal user IDs link activity without names.
+// bodies can contain credentials. Keep user IDs alongside readable identities.
 func activityAttrs(r *http.Request) []any {
-	if user := authenticatedUser(r); user != nil {
-		return []any{"user_id", user.ID}
+	return userLogAttrs(authenticatedUser(r))
+}
+
+func userLogAttrs(user *AuthUser) []any {
+	if user == nil {
+		return []any{"user", "anonymous"}
+	}
+	name := strings.TrimSpace(user.Name)
+	if name == "" {
+		name = user.ID
 	}
-	return nil
+	return []any{"user_id", user.ID, "user", name}
 }
 
 func refreshFailureAttrs(err error) []any {

internal/app/server.go

@@ -757,7 +757,8 @@ func (s *Server) start(w http.ResponseWriter, r *http.Request) {
 		fail(w, 400, "invalid", "Invalid start position.")
 		return
 	}
-	session, err := s.streams.Start(r.Context(), item, body.Start)
+	ctx := stream.WithLogger(r.Context(), slog.With(activityAttrs(r)...))
+	session, err := s.streams.Start(ctx, item, body.Start)
 	if err != nil {
 		if r.Context().Err() == nil {
 			reason := "stream_unavailable"

internal/stream/stream.go

@@ -69,6 +69,15 @@ type entry struct {
 	gen           *generation
 	heartbeat     time.Time
 	probe         *probeResult
+	logger        *slog.Logger
+}
+
+type loggerContextKey struct{}
+
+// WithLogger attaches request identity to a stream's lifecycle logs. Start
+// captures the logger so background work and later seeks retain the owner.
+func WithLogger(ctx context.Context, logger *slog.Logger) context.Context {
+	return context.WithValue(ctx, loggerContextKey{}, logger)
 }
 
 type Manager struct {
@@ -184,7 +193,11 @@ func (m *Manager) Start(ctx context.Context, item model.Item, start float64) (Se
 	}
 	m.next++
 	id := strconv.FormatUint(uint64(time.Now().UnixNano()), 36) + "-" + strconv.FormatUint(m.next, 36)
-	e := &entry{item: item, transcodeMode: m.cfg.TranscodeMode, heartbeat: time.Now()}
+	logger, _ := ctx.Value(loggerContextKey{}).(*slog.Logger)
+	if logger == nil {
+		logger = slog.Default()
+	}
+	e := &entry{item: item, transcodeMode: m.cfg.TranscodeMode, heartbeat: time.Now(), logger: logger.With("session_id", id)}
 	e.session = Session{ID: id, Duration: validDuration(item.Duration)}
 	m.sessions[id] = e
 	m.launchLocked(e, start, nil)
@@ -261,7 +274,7 @@ func (m *Manager) Seek(ctx context.Context, id string, position float64) (Sessio
 		return Session{}, ErrInvalid
 	}
 	old := e.gen
-	slog.Debug("Stream seeking", "session_id", e.session.ID, "offset", position)
+	e.logger.Debug("Stream seeking", "offset", position)
 	old.cancel()
 	go m.discard(old)
 	m.launchLocked(e, position, old.done)
@@ -275,7 +288,7 @@ func (m *Manager) Stop(id string) error {
 		delete(m.sessions, id)
 		e.gen.cancel()
 		m.stopping++
-		slog.Info("Stream stopped", "session_id", id, "reason", "requested", "state", e.session.State)
+		e.logger.Info("Stream stopped", "reason", "requested", "state", e.session.State)
 	}
 	m.mu.Unlock()
 	if e == nil {
@@ -307,7 +320,7 @@ func (m *Manager) Close() error {
 		for id, e := range m.sessions {
 			e.gen.cancel()
 			delete(m.sessions, id)
-			slog.Info("Stream stopped", "session_id", id, "reason", "server_shutdown", "state", e.session.State)
+			e.logger.Info("Stream stopped", "reason", "server_shutdown", "state", e.session.State)
 		}
 	}
 	m.mu.Unlock()
@@ -339,7 +352,7 @@ func (m *Manager) sweep() {
 					delete(m.sessions, id)
 					e.gen.cancel()
 					m.stopping++
-					slog.Info("Stream stopped", "session_id", id, "reason", "idle_timeout", "state", e.session.State)
+					e.logger.Info("Stream stopped", "reason", "idle_timeout", "state", e.session.State)
 					go m.retire(e.gen)
 				}
 			}
@@ -532,7 +545,7 @@ func (m *Manager) run(ctx context.Context, e *entry, g *generation, pos float64,
 			if ready && err == nil && !live {
 				m.update(e, g, func(s *Session) {
 					s.State = "ended"
-					slog.Info("Stream processing completed", "session_id", s.ID, "elapsed", time.Since(started).Round(time.Millisecond))
+					e.logger.Info("Stream processing completed", "elapsed", time.Since(started).Round(time.Millisecond))
 				})
 			} else if live && ctx.Err() == nil {
 				reason := exitClass(err, stderr.String())
@@ -567,7 +580,7 @@ func (m *Manager) run(ctx context.Context, e *entry, g *generation, pos float64,
 func (m *Manager) logReady(e *entry, g *generation, mode string, started time.Time) {
 	m.update(e, g, func(s *Session) {
 		s.State = "ready"
-		slog.Info("Stream ready", "session_id", s.ID, "processing", mode, "elapsed", time.Since(started).Round(time.Millisecond))
+		e.logger.Info("Stream ready", "processing", mode, "elapsed", time.Since(started).Round(time.Millisecond))
 	})
 }
 
@@ -609,7 +622,7 @@ func (m *Manager) logFailure(e *entry, stage, reason string, started time.Time)
 	} else if stage == "ffmpeg_exit" {
 		message = "Stream processing failed"
 	}
-	slog.Warn(message, "session_id", e.session.ID, "item_id", safeLogID(e.item.ID), "stage", stage, "reason", reason, "elapsed", time.Since(started).Round(time.Millisecond))
+	e.logger.Warn(message, "item_id", safeLogID(e.item.ID), "stage", stage, "reason", reason, "elapsed", time.Since(started).Round(time.Millisecond))
 }
 
 func safeLogID(id string) string {

internal/stream/stream_test.go

@@ -78,7 +78,8 @@ func TestFFmpegSessionLifecycle(t *testing.T) {
 	}
 	defer m.Close()
 	item := model.Item{ID: "v", Kind: "movie", Duration: 60}
-	s, err := m.Start(context.Background(), item, 0)
+	owner := WithLogger(context.Background(), slog.With("user_id", "stream-owner", "user", "Alex"))
+	s, err := m.Start(owner, item, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -163,6 +164,11 @@ func TestFFmpegSessionLifecycle(t *testing.T) {
 			t.Fatalf("missing %q in stream lifecycle logs: %s", expected, output)
 		}
 	}
+	for _, line := range strings.Split(string(output), "\n") {
+		if strings.Contains(line, "session_id="+s.ID) && (!strings.Contains(line, "user=Alex") || !strings.Contains(line, "user_id=stream-owner")) {
+			t.Fatalf("stream lost its owner across lifecycle/seek: %s", line)
+		}
+	}
 	if _, err = m.Start(context.Background(), item, 0); err != nil {
 		t.Fatalf("slot not reusable: %v", err)
 	}
@@ -318,7 +324,8 @@ func TestResolverDiagnosticDoesNotLogErrorText(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer m.Close()
-	s, err := m.Start(context.Background(), model.Item{ID: "episode:1:2:3", Kind: "episode"}, 0)
+	ctx := WithLogger(context.Background(), slog.With("user_id", "stream-owner", "user", "stream-owner"))
+	s, err := m.Start(ctx, model.Item{ID: "episode:1:2:3", Kind: "episode"}, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -334,7 +341,7 @@ func TestResolverDiagnosticDoesNotLogErrorText(t *testing.T) {
 		t.Fatalf("state: %s", got.State)
 	}
 	logText := logs.String()
-	if !strings.Contains(logText, "stage=resolve") || !strings.Contains(logText, "reason=error") || strings.Contains(logText, "secret-token") || strings.Contains(logText, "private.invalid") {
+	if !strings.Contains(logText, "user=stream-owner") || !strings.Contains(logText, "user_id=stream-owner") || !strings.Contains(logText, "stage=resolve") || !strings.Contains(logText, "reason=error") || strings.Contains(logText, "secret-token") || strings.Contains(logText, "private.invalid") {
 		t.Fatalf("unsafe or incomplete resolver diagnostic: %s", logText)
 	}
 }