memby/server/internal/api/admin_notifications.go

package api

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/ponzischeme89/memby/server/internal/adminevents"
	"github.com/ponzischeme89/memby/server/internal/store"
)

// streamHeartbeat keeps the notification stream's connection open through anything that
// drops an idle one — a reverse proxy's read timeout being the usual culprit. It is a
// comment line rather than an event, so a client that has been told nothing has still
// been told the connection is alive.
const streamHeartbeat = 25 * time.Second

type adminNotificationsResponse struct {
	Events      []store.AdminEvent          `json:"events"`
	Total       int                         `json:"total"`
	Unread      int                         `json:"unread"`
	Limit       int                         `json:"limit"`
	Offset      int                         `json:"offset"`
	Types       []store.AdminEventTypeCount `json:"types"`
	Subscribers int                         `json:"subscribers"`
}

// handleAdminNotifications is the bell's dropdown and the full activity page alike — the
// same feed, read with a bigger limit.
func (s *Server) handleAdminNotifications(w http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	filter := store.AdminEventFilter{
		Types:      splitCSV(query.Get("type")),
		Severities: splitCSV(query.Get("severity")),
		UnreadOnly: query.Get("unread") == "true",
		Limit:      queryInt(r, "limit", 50, 200),
		Offset:     queryInt(r, "offset", 0, 10000),
	}
	if days := queryInt(r, "days", 0, 90); days > 0 {
		filter.Since = time.Now().Add(-time.Duration(days) * 24 * time.Hour)
	}
	page, err := s.store.AdminEvents(r.Context(), filter)
	if err != nil {
		s.log.Error("admin notifications failed", "error", err)
		writeError(w, http.StatusInternalServerError, "could not read the activity feed")
		return
	}
	response := adminNotificationsResponse{
		Events: page.Events, Total: page.Total, Unread: page.Unread,
		Limit: page.Limit, Offset: page.Offset,
		Types: []store.AdminEventTypeCount{}, Subscribers: s.adminEvents.Subscribers(),
	}
	// The type list is built from what has been published rather than from the constants
	// in adminevents, so the filter can neither offer a type that matches nothing nor miss
	// one a service added after this handler was written.
	if types, err := s.store.AdminEventTypes(r.Context(), time.Time{}); err == nil {
		response.Types = types
	}
	w.Header().Set("Cache-Control", "no-store")
	writeJSON(w, http.StatusOK, response)
}

type markReadRequest struct {
	IDs []int64 `json:"ids"`
	All bool    `json:"all"`
}

// handleAdminNotificationsRead marks events read. An empty id list with `all` marks the
// whole feed, which is the "mark all read" button; anything else marks exactly what was
// named, which is what opening the dropdown does for the rows it showed.
func (s *Server) handleAdminNotificationsRead(w http.ResponseWriter, r *http.Request) {
	var req markReadRequest
	if r.Body != nil {
		_ = json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10)).Decode(&req)
	}
	if len(req.IDs) == 0 && !req.All {
		req.IDs = queryInt64s(r, "id")
	}
	if len(req.IDs) == 0 && !req.All {
		writeError(w, http.StatusBadRequest, "name some events, or ask for all of them")
		return
	}
	marked, err := s.store.MarkAdminEventsRead(r.Context(), req.IDs)
	if err != nil {
		s.log.Error("marking notifications read failed", "error", err)
		writeError(w, http.StatusInternalServerError, "could not update the activity feed")
		return
	}
	unread, _ := s.store.UnreadAdminEvents(r.Context())
	writeJSON(w, http.StatusOK, map[string]any{"marked": marked, "unread": unread})
}

// handleAdminNotificationStream is the live feed: server-sent events, not a WebSocket.
//
// The traffic is one-way and low-volume, which is exactly what SSE is for — and it rides
// ordinary HTTP, so it needs nothing from the reverse proxy in front of the gateway, uses
// the same admin cookie every other route does, and reconnects by itself when a laptop
// wakes up. A WebSocket would have bought a second protocol to authenticate and proxy for
// no capability this feature wants.
//
// It opens by replaying everything after the id the client last saw, which is what makes
// the dropped-event handling in adminevents.Bus recoverable: a reader that fell behind, or
// was asleep, catches up on reconnect rather than having a hole in its feed.
func (s *Server) handleAdminNotificationStream(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		writeError(w, http.StatusInternalServerError, "streaming is not available")
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("Connection", "keep-alive")
	// Nginx buffers a proxied response by default, which for a stream means the browser
	// receives nothing until the connection closes. This is the one header that stops it.
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	// Subscribe *before* the replay, or an event published between the two is delivered to
	// nobody: it would be too late for the replay and too early for the subscription.
	live, unsubscribe := s.adminEvents.Subscribe()
	defer unsubscribe()

	after := lastEventID(r)
	if page, err := s.store.AdminEvents(r.Context(), store.AdminEventFilter{Limit: 50}); err == nil {
		// Oldest first, so the client applies them in the order they happened.
		for i := len(page.Events) - 1; i >= 0; i-- {
			if page.Events[i].ID > after {
				writeSSE(w, flusher, page.Events[i])
			}
			after = max(after, page.Events[i].ID)
		}
	}

	heartbeat := time.NewTicker(streamHeartbeat)
	defer heartbeat.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case <-heartbeat.C:
			fmt.Fprint(w, ": keep-alive\n\n")
			flusher.Flush()
		case event, open := <-live:
			if !open {
				return
			}
			// The replay may overlap the live feed by an event or two; the id filter is
			// what stops the client seeing one twice.
			if event.ID != 0 && event.ID <= after {
				continue
			}
			after = max(after, event.ID)
			writeSSE(w, flusher, event)
		}
	}
}

// lastEventID reads where a reconnecting client got to. The browser sends it back in
// Last-Event-ID automatically after a dropped connection; the query parameter is for a
// first connection that already has a feed on screen.
func lastEventID(r *http.Request) int64 {
	if header := strings.TrimSpace(r.Header.Get("Last-Event-ID")); header != "" {
		if ids := parseInt64(header); ids > 0 {
			return ids
		}
	}
	return int64(queryInt(r, "after", 0, 1<<62))
}

func parseInt64(raw string) int64 {
	var value int64
	if _, err := fmt.Sscanf(raw, "%d", &value); err != nil {
		return 0
	}
	return value
}

func writeSSE(w http.ResponseWriter, flusher http.Flusher, event store.AdminEvent) {
	payload, err := json.Marshal(event)
	if err != nil {
		return
	}
	fmt.Fprintf(w, "id: %d\nevent: admin\ndata: %s\n\n", event.ID, payload)
	flusher.Flush()
}

func splitCSV(raw string) []string {
	values := []string{}
	for _, part := range strings.Split(raw, ",") {
		if trimmed := strings.TrimSpace(part); trimmed != "" {
			values = append(values, trimmed)
		}
	}
	return values
}

// AnnounceServerStart puts the gateway coming up into the feed.
//
// It is a method on Server rather than a call in main so it carries the same build
// information the rest of the console reports, and it is deliberately published *after*
// the listener is up: an event announcing a start that then fails to bind would be the
// one misleading row in the feed.
func (s *Server) AnnounceServerStart(version string) {
	s.publishAdmin(context.Background(), adminevents.Event{
		Type:     adminevents.TypeServerStarted,
		Severity: adminevents.SeverityInfo,
		Title:    "Memby server started",
		Summary:  "The gateway is running version " + version + ".",
		Actor:    "memby-server",
		Link:     "/admin",
		Metadata: adminevents.Meta(map[string]any{"version": version}),
	})
}