1// Package logs implements the log access use case.
2package logs
3
4import (
5 "bufio"
6 "bytes"
7 "context"
8 "encoding/binary"
9 "fmt"
10 "io"
11 "os"
12 "os/exec"
13 "strconv"
14 "strings"
15 "time"
16
17 "github.com/bnema/zerowrap"
18
19 "github.com/bnema/gordon/internal/boundaries/out"
20 "github.com/bnema/gordon/internal/domain"
21)
22
23// appStateProvider resolves app services from the durable ACTIVE record.
24type appStateProvider interface {
25 LoadActive(ctx context.Context, app string) (domain.AppActive, bool, error)
26}
27
28// Service implements the LogService interface.
29type Service struct {
30 logFilePath string
31 fileLoggingEnabled bool
32 runtime out.ContainerRuntime
33 log zerowrap.Logger
34 // appState resolves log references to containers recorded in ACTIVE.
35 appState appStateProvider
36}
37
38var execCommandContext = exec.CommandContext
39
40// NewService creates a new log service. App state is wired after the store opens.
41func NewService(
42 logFilePath string,
43 fileLoggingEnabled bool,
44 runtime out.ContainerRuntime,
45 log zerowrap.Logger,
46) *Service {
47 return &Service{
48 logFilePath: logFilePath,
49 fileLoggingEnabled: fileLoggingEnabled,
50 runtime: runtime,
51 log: log,
52 }
53}
54
55// WithAppState wires the durable ACTIVE app-state reader used for log resolution.
56func (s *Service) WithAppState(provider appStateProvider) *Service {
57 s.appState = provider
58 return s
59}
60
61// containerForService resolves an app/service reference exclusively through ACTIVE.
62func (s *Service) containerForService(ctx context.Context, ref string) (string, error) {
63 app, service, ok := strings.Cut(ref, "/")
64 if !ok || app == "" || service == "" || strings.Contains(service, "/") || s.appState == nil {
65 return "", fmt.Errorf("invalid app/service log reference: %s", ref)
66 }
67 active, found, err := s.appState.LoadActive(ctx, app)
68 if err != nil {
69 return "", fmt.Errorf("failed to load active app %s: %w", app, err)
70 }
71 if !found {
72 return "", fmt.Errorf("active app/service not found: %s", ref)
73 }
74 target, found := active.Services[service]
75 if !found || target.Container == "" {
76 return "", fmt.Errorf("active app/service not found: %s", ref)
77 }
78 return target.Container, nil
79}
80
81// GetProcessLogs returns the last N lines of Gordon process logs.
82func (s *Service) GetProcessLogs(ctx context.Context, lines int) ([]string, error) {
83 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
84 zerowrap.FieldLayer: "usecase",
85 zerowrap.FieldUseCase: "GetProcessLogs",
86 "lines": lines,
87 })
88 log := zerowrap.FromCtx(ctx)
89
90 if s.useFileProcessLogs() {
91 if s.logFilePath == "" {
92 return nil, fmt.Errorf("log file path not configured")
93 }
94
95 file, err := os.Open(s.logFilePath)
96 if err != nil {
97 if os.IsNotExist(err) {
98 log.Warn().Str("log_path", s.logFilePath).
99 Msg("process log file missing, falling back to journalctl")
100 return s.getProcessLogsFromJournal(ctx, lines)
101 }
102 return nil, log.WrapErr(err, "failed to open log file")
103 }
104 defer file.Close()
105
106 return tailLines(file, lines)
107 }
108
109 return s.getProcessLogsFromJournal(ctx, lines)
110}
111
112// FollowProcessLogs returns a channel that streams Gordon process log lines.
113func (s *Service) FollowProcessLogs(ctx context.Context, initialLines int) (<-chan string, error) {
114 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
115 zerowrap.FieldLayer: "usecase",
116 zerowrap.FieldUseCase: "FollowProcessLogs",
117 "initial_lines": initialLines,
118 })
119 if !s.useFileProcessLogs() {
120 return s.followProcessLogsFromJournal(ctx, initialLines)
121 }
122
123 return s.followProcessLogsFromFileOrJournal(ctx, initialLines)
124}
125
126func (s *Service) useFileProcessLogs() bool {
127 return s.fileLoggingEnabled
128}
129
130func (s *Service) followProcessLogsFromFileOrJournal(
131 ctx context.Context,
132 initialLines int,
133) (<-chan string, error) {
134 log := zerowrap.FromCtx(ctx)
135
136 file, err := s.openProcessLogFile()
137 if err != nil {
138 if os.IsNotExist(err) {
139 log.Warn().Str("log_path", s.logFilePath).
140 Msg("process log file missing, falling back to journalctl follow")
141 return s.followProcessLogsFromJournal(ctx, initialLines)
142 }
143 return nil, log.WrapErr(err, "failed to open log file")
144 }
145
146 return s.followProcessLogsFromFile(ctx, file, initialLines), nil
147}
148
149func (s *Service) openProcessLogFile() (*os.File, error) {
150 if s.logFilePath == "" {
151 return nil, fmt.Errorf("log file path not configured")
152 }
153
154 return os.Open(s.logFilePath)
155}
156
157func (s *Service) followProcessLogsFromFile(
158 ctx context.Context,
159 file *os.File,
160 initialLines int,
161) <-chan string {
162 log := zerowrap.FromCtx(ctx)
163 ch := make(chan string, 100)
164
165 go func() {
166 defer close(ch)
167 defer file.Close()
168
169 if !s.emitInitialFileLogs(ctx, file, initialLines, ch) {
170 return
171 }
172 if err := s.seekToEndForFileFollow(file); err != nil {
173 log.Warn().Err(err).Msg("failed to seek to end")
174 return
175 }
176
177 s.streamFileFollowLogs(ctx, file, ch)
178 }()
179
180 return ch
181}
182
183func (s *Service) emitInitialFileLogs(
184 ctx context.Context,
185 file *os.File,
186 initialLines int,
187 ch chan<- string,
188) bool {
189 if initialLines <= 0 {
190 return true
191 }
192
193 log := zerowrap.FromCtx(ctx)
194 lines, err := tailLines(file, initialLines)
195 if err != nil {
196 log.Warn().Err(err).Msg("failed to read initial lines")
197 return true
198 }
199
200 for _, line := range lines {
201 if !sendLogLineWithChannel(ctx, ch, line) {
202 return false
203 }
204 }
205
206 return true
207}
208
209func (s *Service) seekToEndForFileFollow(file *os.File) error {
210 _, err := file.Seek(0, io.SeekEnd)
211 return err
212}
213
214func (s *Service) streamFileFollowLogs(ctx context.Context, file *os.File, ch chan<- string) {
215 log := zerowrap.FromCtx(ctx)
216 reader := bufio.NewReader(file)
217
218 for {
219 if ctx.Err() != nil {
220 return
221 }
222
223 line, err := reader.ReadString('\n')
224 if err != nil {
225 if err == io.EOF {
226 time.Sleep(100 * time.Millisecond)
227 continue
228 }
229 log.Warn().Err(err).Msg("error reading log file")
230 return
231 }
232
233 line = strings.TrimRight(line, "\n\r")
234 if !sendLogLineWithChannel(ctx, ch, line) {
235 return
236 }
237 }
238}
239
240func sendLogLineWithChannel(ctx context.Context, ch chan<- string, line string) bool {
241 select {
242 case ch <- line:
243 return true
244 case <-ctx.Done():
245 return false
246 }
247}
248
249func (s *Service) getProcessLogsFromJournal(ctx context.Context, lines int) ([]string, error) {
250 userScope, err := resolveJournalctlScope(ctx)
251 if err != nil {
252 return nil, err
253 }
254
255 args := []string{"-u", "gordon", "-n", strconv.Itoa(lines), "--no-pager", "-o", "cat"}
256 if userScope {
257 args = append([]string{"--user"}, args...)
258 }
259
260 out, err := execCommandContext(ctx, "journalctl", args...).Output()
261 if err != nil {
262 return nil, fmt.Errorf("failed to read journalctl logs: %w", err)
263 }
264
265 return outputLines(out), nil
266}
267
268func (s *Service) followProcessLogsFromJournal(ctx context.Context, initialLines int) (<-chan string, error) {
269 userScope, err := resolveJournalctlScope(ctx)
270 if err != nil {
271 return nil, err
272 }
273
274 args := []string{
275 "-u", "gordon",
276 "-n", strconv.Itoa(initialLines),
277 "--no-pager",
278 "-o", "cat",
279 "-f",
280 }
281 if userScope {
282 args = append([]string{"--user"}, args...)
283 }
284
285 cmd := execCommandContext(ctx, "journalctl", args...) // #nosec G204 -- internal args only
286 stdout, err := cmd.StdoutPipe()
287 if err != nil {
288 return nil, fmt.Errorf("failed to open journalctl stdout: %w", err)
289 }
290 cmd.Stderr = io.Discard
291
292 if err := cmd.Start(); err != nil {
293 return nil, fmt.Errorf("failed to start journalctl follow: %w", err)
294 }
295
296 ch := make(chan string, 100)
297
298 go func() {
299 defer close(ch)
300 defer stdout.Close()
301 defer cmd.Wait() //nolint:errcheck // best-effort cleanup
302
303 scanner := bufio.NewScanner(stdout)
304 scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
305
306 for scanner.Scan() {
307 select {
308 case ch <- scanner.Text():
309 case <-ctx.Done():
310 return
311 }
312 }
313 }()
314
315 return ch, nil
316}
317
318func resolveJournalctlScope(ctx context.Context) (bool, error) {
319 userLogs, userErr := hasJournalctlLogs(ctx, true)
320 if userErr == nil && userLogs {
321 return true, nil
322 }
323
324 systemLogs, systemErr := hasJournalctlLogs(ctx, false)
325 if systemErr == nil && systemLogs {
326 return false, nil
327 }
328
329 if userErr != nil && systemErr != nil {
330 return false, fmt.Errorf("journalctl unavailable for user and system services")
331 }
332
333 // Default to user scope when logs are currently empty; this matches rootless setups.
334 return true, nil
335}
336
337func hasJournalctlLogs(ctx context.Context, userService bool) (bool, error) {
338 args := []string{"-u", "gordon", "-n", "1", "--no-pager", "-q", "-o", "cat"}
339 if userService {
340 args = append([]string{"--user"}, args...)
341 }
342
343 out, err := execCommandContext(ctx, "journalctl", args...).Output()
344 if err != nil {
345 return false, err
346 }
347
348 return len(bytes.TrimSpace(out)) > 0, nil
349}
350
351func outputLines(out []byte) []string {
352 trimmed := strings.TrimRight(string(out), "\n\r")
353 if trimmed == "" {
354 return []string{}
355 }
356 return strings.Split(trimmed, "\n")
357}
358
359// validateDomain checks that domain is safe for use in logs and error messages.
360func validateDomain(domain string) error {
361 if domain == "" {
362 return fmt.Errorf("domain cannot be empty")
363 }
364 if len(domain) > 253 {
365 return fmt.Errorf("domain too long")
366 }
367 if strings.Contains(domain, "..") {
368 return fmt.Errorf("invalid domain")
369 }
370 if strings.ContainsAny(domain, "\x00\n\r") {
371 return fmt.Errorf("invalid domain")
372 }
373 return nil
374}
375
376// GetContainerLogs returns the last N lines of container logs for a domain.
377func (s *Service) GetContainerLogs(ctx context.Context, domain string, lines int) ([]string, error) {
378 if err := validateDomain(domain); err != nil {
379 return nil, err
380 }
381
382 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
383 zerowrap.FieldLayer: "usecase",
384 zerowrap.FieldUseCase: "GetContainerLogs",
385 "domain": domain,
386 "lines": lines,
387 })
388 log := zerowrap.FromCtx(ctx)
389
390 // Resolve the app/service to its ACTIVE container.
391 containerID, err := s.containerForService(ctx, domain)
392 if err != nil {
393 return nil, err
394 }
395
396 // Get logs from container runtime (non-follow mode)
397 reader, err := s.runtime.GetContainerLogs(ctx, containerID, false)
398 if err != nil {
399 return nil, log.WrapErr(err, "failed to get container logs")
400 }
401 defer reader.Close()
402
403 // Parse Docker log stream and get lines
404 logLines, err := parseDockerLogs(reader)
405 if err != nil {
406 return nil, log.WrapErr(err, "failed to parse container logs")
407 }
408
409 // Return last N lines
410 if len(logLines) > lines {
411 logLines = logLines[len(logLines)-lines:]
412 }
413
414 return logLines, nil
415}
416
417// FollowContainerLogs returns a channel that streams container log lines.
418func (s *Service) FollowContainerLogs(ctx context.Context, domain string, initialLines int) (<-chan string, error) {
419 if err := validateDomain(domain); err != nil {
420 return nil, err
421 }
422
423 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
424 zerowrap.FieldLayer: "usecase",
425 zerowrap.FieldUseCase: "FollowContainerLogs",
426 "domain": domain,
427 "initial_lines": initialLines,
428 })
429 log := zerowrap.FromCtx(ctx)
430
431 // Resolve the app/service to its ACTIVE container.
432 containerID, err := s.containerForService(ctx, domain)
433 if err != nil {
434 return nil, err
435 }
436
437 // Get logs from container runtime (follow mode)
438 reader, err := s.runtime.GetContainerLogs(ctx, containerID, true)
439 if err != nil {
440 return nil, log.WrapErr(err, "failed to get container logs")
441 }
442
443 ch := make(chan string, 100)
444
445 go func() {
446 defer close(ch)
447 defer reader.Close()
448
449 // Stream Docker logs
450 err := streamDockerLogs(ctx, reader, ch)
451 if err != nil && ctx.Err() == nil {
452 log.Warn().Err(err).Msg("error streaming container logs")
453 }
454 }()
455
456 return ch, nil
457}
458
459// tailLines reads the last N lines from a file using a ring buffer.
460func tailLines(file *os.File, n int) ([]string, error) {
461 if n <= 0 {
462 return []string{}, nil
463 }
464
465 // Seek to beginning
466 if _, err := file.Seek(0, io.SeekStart); err != nil {
467 return nil, err
468 }
469
470 scanner := bufio.NewScanner(file)
471 buffer := make([]string, n)
472 index := 0
473 total := 0
474
475 for scanner.Scan() {
476 buffer[index] = scanner.Text()
477 index = (index + 1) % n
478 total++
479 }
480
481 if err := scanner.Err(); err != nil {
482 return nil, err
483 }
484
485 if total == 0 {
486 return []string{}, nil
487 }
488
489 // Build result in correct order
490 var result []string
491 if total <= n {
492 result = make([]string, total)
493 for i := 0; i < total; i++ {
494 result[i] = buffer[i]
495 }
496 } else {
497 result = make([]string, n)
498 for i := 0; i < n; i++ {
499 result[i] = buffer[(index+i)%n]
500 }
501 }
502
503 return result, nil
504}
505
506// parseDockerLogs parses Docker multiplexed log stream and returns lines.
507// Docker logs have an 8-byte header: [stream type (1), padding (3), size (4)]
508func parseDockerLogs(reader io.Reader) ([]string, error) {
509 var lines []string
510 header := make([]byte, 8)
511
512 for {
513 // Read header
514 _, err := io.ReadFull(reader, header)
515 if err != nil {
516 if err == io.EOF {
517 break
518 }
519 return lines, nil // Return what we have
520 }
521
522 // Get payload size from header (big endian)
523 size := binary.BigEndian.Uint32(header[4:8])
524 if size == 0 {
525 continue
526 }
527
528 // Read payload
529 payload := make([]byte, size)
530 _, err = io.ReadFull(reader, payload)
531 if err != nil {
532 if err == io.EOF {
533 break
534 }
535 return lines, nil
536 }
537
538 // Split by newlines and add to result
539 content := strings.TrimRight(string(payload), "\n\r")
540 for _, line := range strings.Split(content, "\n") {
541 if line != "" {
542 lines = append(lines, line)
543 }
544 }
545 }
546
547 return lines, nil
548}
549
550// streamDockerLogs streams Docker multiplexed log stream to a channel.
551func streamDockerLogs(ctx context.Context, reader io.Reader, ch chan<- string) error {
552 header := make([]byte, 8)
553
554 for {
555 select {
556 case <-ctx.Done():
557 return ctx.Err()
558 default:
559 }
560
561 // Read header
562 _, err := io.ReadFull(reader, header)
563 if err != nil {
564 if err == io.EOF {
565 return nil
566 }
567 return err
568 }
569
570 // Get payload size
571 size := binary.BigEndian.Uint32(header[4:8])
572 if size == 0 {
573 continue
574 }
575
576 // Read payload
577 payload := make([]byte, size)
578 _, err = io.ReadFull(reader, payload)
579 if err != nil {
580 if err == io.EOF {
581 return nil
582 }
583 return err
584 }
585
586 // Split by newlines and send to channel
587 content := strings.TrimRight(string(payload), "\n\r")
588 for _, line := range strings.Split(content, "\n") {
589 if line != "" {
590 select {
591 case ch <- line:
592 case <-ctx.Done():
593 return ctx.Err()
594 }
595 }
596 }
597 }
598}