1// Package admin implements the HTTP adapter for the admin API.
2package admin
3
4import (
5 "encoding/json"
6 "errors"
7 "fmt"
8 "net/http"
9 "strconv"
10 "strings"
11
12 "github.com/bnema/zerowrap"
13
14 "gordon/internal/boundaries/in"
15 "gordon/internal/boundaries/out"
16 "gordon/internal/domain"
17)
18
19// maxAdminRequestSize is the maximum allowed size for admin API request bodies.
20const maxAdminRequestSize = 1 << 20 // 1MB
21
22// maxLogLines is the maximum allowed number of log lines that can be requested.
23const maxLogLines = 10000
24
25// Handler implements the HTTP handler for the admin API.
26type Handler struct {
27 configSvc in.ConfigService
28 authSvc in.AuthService
29 containerSvc in.ContainerService
30 healthSvc in.HealthService
31 secretSvc in.SecretService
32 logSvc in.LogService
33 eventBus out.EventPublisher
34 log zerowrap.Logger
35}
36
37// NewHandler creates a new admin HTTP handler.
38func NewHandler(
39 configSvc in.ConfigService,
40 authSvc in.AuthService,
41 containerSvc in.ContainerService,
42 healthSvc in.HealthService,
43 secretSvc in.SecretService,
44 logSvc in.LogService,
45 eventBus out.EventPublisher,
46 log zerowrap.Logger,
47) *Handler {
48 return &Handler{
49 configSvc: configSvc,
50 authSvc: authSvc,
51 containerSvc: containerSvc,
52 healthSvc: healthSvc,
53 secretSvc: secretSvc,
54 logSvc: logSvc,
55 eventBus: eventBus,
56 log: log,
57 }
58}
59
60// RegisterRoutes registers the admin routes on the given mux.
61func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
62 mux.HandleFunc("/admin/", h.handleAdminRoutes)
63}
64
65// ServeHTTP implements http.Handler.
66func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
67 h.handleAdminRoutes(w, r)
68}
69
70func (h *Handler) handleAdminRoutes(w http.ResponseWriter, r *http.Request) {
71 ctx := zerowrap.CtxWithFields(r.Context(), map[string]any{
72 zerowrap.FieldLayer: "adapter",
73 zerowrap.FieldAdapter: "http",
74 zerowrap.FieldHandler: "admin",
75 zerowrap.FieldMethod: r.Method,
76 zerowrap.FieldPath: r.URL.Path,
77 })
78 r = r.WithContext(ctx)
79
80 path := strings.TrimPrefix(r.URL.Path, "/admin")
81
82 // Route to appropriate handler
83 switch {
84 case path == "/routes" || strings.HasPrefix(path, "/routes/"):
85 h.handleRoutes(w, r, path)
86 case path == "/secrets" || strings.HasPrefix(path, "/secrets/"):
87 h.handleSecrets(w, r, path)
88 case path == "/deploy" || strings.HasPrefix(path, "/deploy/"):
89 h.handleDeploy(w, r, path)
90 case path == "/logs" || strings.HasPrefix(path, "/logs/"):
91 h.handleLogs(w, r, path)
92 case path == "/status":
93 h.handleStatus(w, r)
94 case path == "/health":
95 h.handleHealth(w, r)
96 case path == "/reload":
97 h.handleReload(w, r)
98 case path == "/config":
99 h.handleConfig(w, r)
100 default:
101 http.NotFound(w, r)
102 }
103}
104
105// sendJSON sends a JSON response.
106func (h *Handler) sendJSON(w http.ResponseWriter, status int, data any) {
107 w.Header().Set("Content-Type", "application/json")
108 w.WriteHeader(status)
109 if data != nil {
110 _ = json.NewEncoder(w).Encode(data)
111 }
112}
113
114// sendError sends an error response.
115func (h *Handler) sendError(w http.ResponseWriter, status int, message string) {
116 h.sendJSON(w, status, map[string]string{"error": message})
117}
118
119// handleRoutes handles /admin/routes endpoints.
120func (h *Handler) handleRoutes(w http.ResponseWriter, r *http.Request, path string) {
121 // Parse domain from path if present
122 routeDomain := strings.TrimPrefix(path, "/routes/")
123 if routeDomain == "/routes" {
124 routeDomain = ""
125 }
126
127 switch r.Method {
128 case http.MethodGet:
129 h.handleRoutesGet(w, r, routeDomain)
130 case http.MethodPost:
131 h.handleRoutesPost(w, r)
132 case http.MethodPut:
133 h.handleRoutesPut(w, r, routeDomain)
134 case http.MethodDelete:
135 h.handleRoutesDelete(w, r, routeDomain)
136 default:
137 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
138 }
139}
140
141func (h *Handler) handleRoutesGet(w http.ResponseWriter, r *http.Request, routeDomain string) {
142 ctx := r.Context()
143
144 // Check read permission
145 if !HasAccess(ctx, domain.AdminResourceRoutes, domain.AdminActionRead) {
146 h.sendError(w, http.StatusForbidden, "insufficient permissions for routes:read")
147 return
148 }
149
150 if routeDomain == "" {
151 routes := h.configSvc.GetRoutes(ctx)
152 h.sendJSON(w, http.StatusOK, map[string]any{"routes": routes})
153 return
154 }
155
156 route, err := h.configSvc.GetRoute(ctx, routeDomain)
157 if err != nil {
158 h.sendError(w, http.StatusNotFound, "route not found")
159 return
160 }
161 h.sendJSON(w, http.StatusOK, route)
162}
163
164func (h *Handler) handleRoutesPost(w http.ResponseWriter, r *http.Request) {
165 ctx := r.Context()
166 log := zerowrap.FromCtx(ctx)
167
168 // Check write permission
169 if !HasAccess(ctx, domain.AdminResourceRoutes, domain.AdminActionWrite) {
170 h.sendError(w, http.StatusForbidden, "insufficient permissions for routes:write")
171 return
172 }
173
174 // Limit request body size
175 r.Body = http.MaxBytesReader(w, r.Body, maxAdminRequestSize)
176
177 var route domain.Route
178 if err := json.NewDecoder(r.Body).Decode(&route); err != nil {
179 log.Warn().Err(err).Msg("invalid route JSON")
180 h.sendError(w, http.StatusBadRequest, "invalid JSON")
181 return
182 }
183
184 if err := h.configSvc.AddRoute(ctx, route); err != nil {
185 log.Error().Err(err).Str("domain", route.Domain).Msg("failed to add route")
186 switch {
187 case errors.Is(err, domain.ErrRouteDomainEmpty), errors.Is(err, domain.ErrRouteImageEmpty):
188 h.sendError(w, http.StatusBadRequest, err.Error())
189 default:
190 h.sendError(w, http.StatusInternalServerError, "failed to add route")
191 }
192 return
193 }
194
195 log.Info().Str("domain", route.Domain).Str("image", route.Image).Msg("route added")
196 h.sendJSON(w, http.StatusCreated, route)
197}
198
199func (h *Handler) handleRoutesPut(w http.ResponseWriter, r *http.Request, routeDomain string) {
200 ctx := r.Context()
201 log := zerowrap.FromCtx(ctx)
202
203 // Check write permission
204 if !HasAccess(ctx, domain.AdminResourceRoutes, domain.AdminActionWrite) {
205 h.sendError(w, http.StatusForbidden, "insufficient permissions for routes:write")
206 return
207 }
208
209 if routeDomain == "" {
210 h.sendError(w, http.StatusBadRequest, "domain required in path")
211 return
212 }
213
214 // Limit request body size
215 r.Body = http.MaxBytesReader(w, r.Body, maxAdminRequestSize)
216
217 var route domain.Route
218 if err := json.NewDecoder(r.Body).Decode(&route); err != nil {
219 log.Warn().Err(err).Msg("invalid route JSON")
220 h.sendError(w, http.StatusBadRequest, "invalid JSON")
221 return
222 }
223
224 route.Domain = routeDomain
225
226 if err := h.configSvc.UpdateRoute(ctx, route); err != nil {
227 log.Error().Err(err).Str("domain", routeDomain).Msg("failed to update route")
228 switch {
229 case errors.Is(err, domain.ErrRouteNotFound):
230 h.sendError(w, http.StatusNotFound, "route not found")
231 case errors.Is(err, domain.ErrRouteDomainEmpty), errors.Is(err, domain.ErrRouteImageEmpty):
232 h.sendError(w, http.StatusBadRequest, err.Error())
233 default:
234 h.sendError(w, http.StatusInternalServerError, "failed to update route")
235 }
236 return
237 }
238
239 log.Info().Str("domain", route.Domain).Str("image", route.Image).Msg("route updated")
240 h.sendJSON(w, http.StatusOK, route)
241}
242
243func (h *Handler) handleRoutesDelete(w http.ResponseWriter, r *http.Request, routeDomain string) {
244 ctx := r.Context()
245 log := zerowrap.FromCtx(ctx)
246
247 // Check write permission
248 if !HasAccess(ctx, domain.AdminResourceRoutes, domain.AdminActionWrite) {
249 h.sendError(w, http.StatusForbidden, "insufficient permissions for routes:write")
250 return
251 }
252
253 if routeDomain == "" {
254 h.sendError(w, http.StatusBadRequest, "domain required in path")
255 return
256 }
257
258 if err := h.configSvc.RemoveRoute(ctx, routeDomain); err != nil {
259 log.Error().Err(err).Str("domain", routeDomain).Msg("failed to remove route")
260 h.sendError(w, http.StatusInternalServerError, "failed to remove route")
261 return
262 }
263
264 log.Info().Str("domain", routeDomain).Msg("route removed")
265 h.sendJSON(w, http.StatusOK, map[string]string{"status": "removed"})
266}
267
268// handleSecrets handles /admin/secrets endpoints.
269func (h *Handler) handleSecrets(w http.ResponseWriter, r *http.Request, path string) {
270 ctx := r.Context()
271 log := zerowrap.FromCtx(ctx)
272
273 // Parse path: /secrets/{domain} or /secrets/{domain}/{key}
274 parts := strings.Split(strings.TrimPrefix(path, "/secrets/"), "/")
275 if len(parts) == 0 || parts[0] == "" {
276 h.sendError(w, http.StatusBadRequest, "domain required")
277 return
278 }
279
280 secretDomain := parts[0]
281 secretKey := ""
282 if len(parts) > 1 {
283 secretKey = parts[1]
284 }
285
286 switch r.Method {
287 case http.MethodGet:
288 // Check read permission
289 if !HasAccess(ctx, domain.AdminResourceSecrets, domain.AdminActionRead) {
290 h.sendError(w, http.StatusForbidden, "insufficient permissions for secrets:read")
291 return
292 }
293 // List secrets for domain (names only, not values)
294 keys, err := h.secretSvc.ListKeys(ctx, secretDomain)
295 if err != nil {
296 log.Error().Err(err).Str("domain", secretDomain).Msg("failed to list secrets")
297 h.sendError(w, http.StatusBadRequest, "invalid domain")
298 return
299 }
300 h.sendJSON(w, http.StatusOK, map[string]any{"domain": secretDomain, "keys": keys})
301
302 case http.MethodPost:
303 // Check write permission
304 if !HasAccess(ctx, domain.AdminResourceSecrets, domain.AdminActionWrite) {
305 h.sendError(w, http.StatusForbidden, "insufficient permissions for secrets:write")
306 return
307 }
308
309 // Limit request body size
310 r.Body = http.MaxBytesReader(w, r.Body, maxAdminRequestSize)
311
312 // Set secret(s)
313 var data map[string]string
314 if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
315 log.Warn().Err(err).Msg("invalid secrets JSON")
316 h.sendError(w, http.StatusBadRequest, "invalid JSON")
317 return
318 }
319
320 if err := h.secretSvc.Set(ctx, secretDomain, data); err != nil {
321 log.Error().Err(err).Str("domain", secretDomain).Msg("failed to set secrets")
322 h.sendError(w, http.StatusBadRequest, "invalid domain")
323 return
324 }
325
326 log.Info().Str("domain", secretDomain).Int("count", len(data)).Msg("secrets set")
327 h.sendJSON(w, http.StatusOK, map[string]string{"status": "updated"})
328
329 case http.MethodDelete:
330 // Check write permission
331 if !HasAccess(ctx, domain.AdminResourceSecrets, domain.AdminActionWrite) {
332 h.sendError(w, http.StatusForbidden, "insufficient permissions for secrets:write")
333 return
334 }
335 if secretKey == "" {
336 h.sendError(w, http.StatusBadRequest, "key required in path")
337 return
338 }
339
340 if err := h.secretSvc.Delete(ctx, secretDomain, secretKey); err != nil {
341 log.Error().Err(err).Str("domain", secretDomain).Str("key", secretKey).Msg("failed to delete secret")
342 h.sendError(w, http.StatusBadRequest, "invalid domain")
343 return
344 }
345
346 log.Info().Str("domain", secretDomain).Str("key", secretKey).Msg("secret deleted")
347 h.sendJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
348
349 default:
350 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
351 }
352}
353
354// handleHealth handles /admin/health endpoint.
355// Returns detailed health status for all routes including HTTP probe results.
356func (h *Handler) handleHealth(w http.ResponseWriter, r *http.Request) {
357 if r.Method != http.MethodGet {
358 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
359 return
360 }
361
362 ctx := r.Context()
363
364 // Check read permission
365 if !HasAccess(ctx, domain.AdminResourceStatus, domain.AdminActionRead) {
366 h.sendError(w, http.StatusForbidden, "insufficient permissions for status:read")
367 return
368 }
369
370 health := h.healthSvc.CheckAllRoutes(ctx)
371
372 // Convert to JSON-friendly format
373 result := make(map[string]any, len(health))
374 for domain, h := range health {
375 result[domain] = map[string]any{
376 "container_status": h.ContainerStatus,
377 "http_status": h.HTTPStatus,
378 "response_time_ms": h.ResponseTimeMs,
379 "healthy": h.Healthy,
380 "error": h.Error,
381 }
382 }
383
384 h.sendJSON(w, http.StatusOK, map[string]any{"health": result})
385}
386
387// handleStatus handles /admin/status endpoint.
388func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) {
389 if r.Method != http.MethodGet {
390 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
391 return
392 }
393
394 ctx := r.Context()
395
396 // Check read permission
397 if !HasAccess(ctx, domain.AdminResourceStatus, domain.AdminActionRead) {
398 h.sendError(w, http.StatusForbidden, "insufficient permissions for status:read")
399 return
400 }
401
402 routes := h.configSvc.GetRoutes(ctx)
403
404 // Get container statuses
405 statuses := make(map[string]string)
406 for _, route := range routes {
407 status := "unknown"
408 container, ok := h.containerSvc.Get(ctx, route.Domain)
409 if ok && container != nil {
410 status = container.Status
411 }
412 statuses[route.Domain] = status
413 }
414
415 status := map[string]any{
416 "routes": len(routes),
417 "registry_domain": h.configSvc.GetRegistryDomain(),
418 "registry_port": h.configSvc.GetRegistryPort(),
419 "server_port": h.configSvc.GetServerPort(),
420 "auto_route": h.configSvc.IsAutoRouteEnabled(),
421 "network_isolation": h.configSvc.IsNetworkIsolationEnabled(),
422 "container_status": statuses,
423 }
424
425 h.sendJSON(w, http.StatusOK, status)
426}
427
428// handleReload handles /admin/reload endpoint.
429// This reloads configuration from file into memory and triggers container sync.
430func (h *Handler) handleReload(w http.ResponseWriter, r *http.Request) {
431 if r.Method != http.MethodPost {
432 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
433 return
434 }
435
436 ctx := r.Context()
437 log := zerowrap.FromCtx(ctx)
438
439 // Check write permission (reload modifies state)
440 if !HasAccess(ctx, domain.AdminResourceConfig, domain.AdminActionWrite) {
441 h.sendError(w, http.StatusForbidden, "insufficient permissions for config:write")
442 return
443 }
444
445 // Reload configuration from file into memory
446 if err := h.configSvc.Load(ctx); err != nil {
447 log.Error().Err(err).Msg("failed to reload config")
448 h.sendError(w, http.StatusInternalServerError, "failed to reload config")
449 return
450 }
451
452 // Publish manual reload event to sync containers
453 // ManualReloadHandler starts missing containers without restarting running ones
454 if h.eventBus != nil {
455 if err := h.eventBus.Publish(domain.EventManualReload, nil); err != nil {
456 log.Warn().Err(err).Msg("failed to publish manual reload event")
457 }
458 }
459
460 log.Info().Msg("config reloaded via admin API")
461 h.sendJSON(w, http.StatusOK, map[string]string{"status": "reloaded"})
462}
463
464// handleConfig handles /admin/config endpoint.
465func (h *Handler) handleConfig(w http.ResponseWriter, r *http.Request) {
466 if r.Method != http.MethodGet {
467 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
468 return
469 }
470
471 ctx := r.Context()
472
473 // Check read permission
474 if !HasAccess(ctx, domain.AdminResourceConfig, domain.AdminActionRead) {
475 h.sendError(w, http.StatusForbidden, "insufficient permissions for config:read")
476 return
477 }
478
479 config := map[string]any{
480 "server": map[string]any{
481 "port": h.configSvc.GetServerPort(),
482 "registry_port": h.configSvc.GetRegistryPort(),
483 "registry_domain": h.configSvc.GetRegistryDomain(),
484 "data_dir": h.configSvc.GetDataDir(),
485 },
486 "auto_route": map[string]any{
487 "enabled": h.configSvc.IsAutoRouteEnabled(),
488 },
489 "network_isolation": map[string]any{
490 "enabled": h.configSvc.IsNetworkIsolationEnabled(),
491 "prefix": h.configSvc.GetNetworkPrefix(),
492 },
493 "routes": h.configSvc.GetRoutes(ctx),
494 "external_routes": h.configSvc.GetExternalRoutes(),
495 }
496
497 h.sendJSON(w, http.StatusOK, config)
498}
499
500// handleDeploy handles /admin/deploy/:domain endpoint.
501// POST triggers a deployment for the specified domain.
502func (h *Handler) handleDeploy(w http.ResponseWriter, r *http.Request, path string) {
503 if r.Method != http.MethodPost {
504 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
505 return
506 }
507
508 ctx := r.Context()
509 log := zerowrap.FromCtx(ctx)
510
511 // Check write permission
512 if !HasAccess(ctx, domain.AdminResourceConfig, domain.AdminActionWrite) {
513 h.sendError(w, http.StatusForbidden, "insufficient permissions for config:write")
514 return
515 }
516
517 // Parse domain from path
518 deployDomain := strings.TrimPrefix(path, "/deploy/")
519 if deployDomain == "" || deployDomain == "/deploy" {
520 h.sendError(w, http.StatusBadRequest, "domain required in path")
521 return
522 }
523
524 // Get the route for this domain
525 route, err := h.configSvc.GetRoute(ctx, deployDomain)
526 if err != nil {
527 h.sendError(w, http.StatusNotFound, "route not found")
528 return
529 }
530
531 // Deploy the container
532 container, err := h.containerSvc.Deploy(ctx, *route)
533 if err != nil {
534 log.Error().Err(err).Str("domain", deployDomain).Msg("failed to deploy container")
535 h.sendError(w, http.StatusInternalServerError, "failed to deploy container")
536 return
537 }
538
539 log.Info().Str("domain", deployDomain).Str("container_id", container.ID).Msg("container deployed via admin API")
540 h.sendJSON(w, http.StatusOK, map[string]any{
541 "status": "deployed",
542 "container_id": container.ID,
543 "domain": deployDomain,
544 })
545}
546
547// handleLogs handles /admin/logs endpoints.
548// GET /admin/logs - Gordon process logs
549// GET /admin/logs/:domain - Container logs for a specific domain
550func (h *Handler) handleLogs(w http.ResponseWriter, r *http.Request, path string) {
551 if r.Method != http.MethodGet {
552 http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
553 return
554 }
555
556 ctx := r.Context()
557 log := zerowrap.FromCtx(ctx)
558
559 // Check read permission
560 if !HasAccess(ctx, domain.AdminResourceStatus, domain.AdminActionRead) {
561 h.sendError(w, http.StatusForbidden, "insufficient permissions for status:read")
562 return
563 }
564
565 // Check if LogService is available
566 if h.logSvc == nil {
567 h.sendError(w, http.StatusServiceUnavailable, "log service not available")
568 return
569 }
570
571 // Parse query parameters
572 lines := 50 // default
573 if linesStr := r.URL.Query().Get("lines"); linesStr != "" {
574 if n, err := strconv.Atoi(linesStr); err == nil && n > 0 {
575 lines = n
576 }
577 }
578 if lines > maxLogLines {
579 lines = maxLogLines
580 }
581 follow := r.URL.Query().Get("follow") == "true"
582
583 // Parse domain from path
584 logDomain := strings.TrimPrefix(path, "/logs/")
585 if logDomain == "/logs" {
586 logDomain = ""
587 }
588
589 if logDomain == "" {
590 // Gordon process logs
591 h.handleProcessLogs(w, r, lines, follow)
592 } else {
593 // Container logs
594 h.handleContainerLogs(w, r, logDomain, lines, follow)
595 }
596
597 // Prevent unused variable warning when follow is implemented
598 _ = log
599}
600
601// handleProcessLogs handles Gordon process logs.
602func (h *Handler) handleProcessLogs(w http.ResponseWriter, r *http.Request, lines int, follow bool) {
603 ctx := r.Context()
604 log := zerowrap.FromCtx(ctx)
605
606 if follow {
607 // SSE streaming
608 h.streamProcessLogs(w, r, lines)
609 return
610 }
611
612 // Return last N lines as JSON
613 logLines, err := h.logSvc.GetProcessLogs(ctx, lines)
614 if err != nil {
615 log.Warn().Err(err).Msg("failed to get process logs")
616 h.sendError(w, http.StatusInternalServerError, "failed to get logs")
617 return
618 }
619
620 h.sendJSON(w, http.StatusOK, map[string]any{
621 "lines": logLines,
622 })
623}
624
625// handleContainerLogs handles container logs for a specific domain.
626func (h *Handler) handleContainerLogs(w http.ResponseWriter, r *http.Request, logDomain string, lines int, follow bool) {
627 ctx := r.Context()
628 log := zerowrap.FromCtx(ctx)
629
630 if follow {
631 // SSE streaming
632 h.streamContainerLogs(w, r, logDomain, lines)
633 return
634 }
635
636 // Return last N lines as JSON
637 logLines, err := h.logSvc.GetContainerLogs(ctx, logDomain, lines)
638 if err != nil {
639 log.Warn().Err(err).Str("domain", logDomain).Msg("failed to get container logs")
640 h.sendError(w, http.StatusInternalServerError, fmt.Sprintf("failed to get logs: %v", err))
641 return
642 }
643
644 h.sendJSON(w, http.StatusOK, map[string]any{
645 "domain": logDomain,
646 "lines": logLines,
647 })
648}
649
650// streamProcessLogs streams Gordon process logs via SSE.
651func (h *Handler) streamProcessLogs(w http.ResponseWriter, r *http.Request, lines int) {
652 ctx := r.Context()
653 log := zerowrap.FromCtx(ctx)
654
655 // Check for flusher support before setting up SSE
656 flusher, ok := w.(http.Flusher)
657 if !ok {
658 h.sendError(w, http.StatusInternalServerError, "streaming not supported")
659 return
660 }
661
662 // Set up SSE
663 w.Header().Set("Content-Type", "text/event-stream")
664 w.Header().Set("Cache-Control", "no-cache")
665 w.Header().Set("Connection", "keep-alive")
666 w.WriteHeader(http.StatusOK)
667
668 ch, err := h.logSvc.FollowProcessLogs(ctx, lines)
669 if err != nil {
670 log.Warn().Err(err).Msg("failed to follow process logs")
671 _, _ = fmt.Fprintf(w, "event: error\ndata: failed to stream logs\n\n")
672 flusher.Flush()
673 return
674 }
675
676 for {
677 select {
678 case line, ok := <-ch:
679 if !ok {
680 return
681 }
682 _, _ = fmt.Fprintf(w, "data: %s\n\n", line)
683 flusher.Flush()
684 case <-ctx.Done():
685 return
686 }
687 }
688}
689
690// streamContainerLogs streams container logs via SSE.
691func (h *Handler) streamContainerLogs(w http.ResponseWriter, r *http.Request, logDomain string, lines int) {
692 ctx := r.Context()
693 log := zerowrap.FromCtx(ctx)
694
695 // Check for flusher support before setting up SSE
696 flusher, ok := w.(http.Flusher)
697 if !ok {
698 h.sendError(w, http.StatusInternalServerError, "streaming not supported")
699 return
700 }
701
702 // Set up SSE
703 w.Header().Set("Content-Type", "text/event-stream")
704 w.Header().Set("Cache-Control", "no-cache")
705 w.Header().Set("Connection", "keep-alive")
706 w.WriteHeader(http.StatusOK)
707
708 ch, err := h.logSvc.FollowContainerLogs(ctx, logDomain, lines)
709 if err != nil {
710 log.Warn().Err(err).Str("domain", logDomain).Msg("failed to follow container logs")
711 _, _ = fmt.Fprintf(w, "event: error\ndata: failed to stream container logs\n\n")
712 flusher.Flush()
713 return
714 }
715
716 for {
717 select {
718 case line, ok := <-ch:
719 if !ok {
720 return
721 }
722 _, _ = fmt.Fprintf(w, "data: %s\n\n", line)
723 flusher.Flush()
724 case <-ctx.Done():
725 return
726 }
727 }
728}