memby/server/internal/store/admin_events.go

package store

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"
)

// AdminEventRetention is how long an administrative event is kept. The feed is a rolling
// operational record, not an archive: anything worth keeping longer is already a log line
// or a row in the table the event was about.
const AdminEventRetention = 30 * 24 * time.Hour

// Severities. Three, not six — this decides whether a row is worth an operator's
// attention, and a scale finer than that is one nobody applies consistently.
const (
	SeverityInfo    = "info"
	SeverityWarning = "warning"
	SeverityError   = "error"
)

// AdminEvent is one thing that happened, in the shape the console and every integration
// read it in.
//
// The vocabulary is deliberately about *shape* rather than about any particular subject:
// Type is a routing key, Title is the headline, Summary is the sentence, Actor is who or
// what caused it and Target is what it happened to. A publisher that cannot fill one
// leaves it blank. Metadata is for the fields only that publisher's readers understand,
// and nothing in the console depends on any key in it, so adding one is never a
// migration.
//
// Link is a console path (not a URL), so an event can carry the operator to the page that
// explains it. It is a path because the console does not reliably know its own external
// address, exactly as the gateway does not for a subtitle it serves.
type AdminEvent struct {
	ID         int64           `json:"id"`
	OccurredAt time.Time       `json:"occurredAt"`
	Type       string          `json:"type"`
	Severity   string          `json:"severity"`
	Title      string          `json:"title"`
	Summary    string          `json:"summary"`
	Actor      string          `json:"actor,omitempty"`
	Target     string          `json:"target,omitempty"`
	Link       string          `json:"link,omitempty"`
	Metadata   json.RawMessage `json:"metadata,omitempty"`
	ReadAt     *time.Time      `json:"readAt,omitempty"`
}

// Read reports whether the operator has already seen this one.
func (e AdminEvent) Read() bool { return e.ReadAt != nil }

// AdminEventFilter narrows the feed. Types is a set rather than a single value because
// the natural question is "show me the authentication ones", which is several types.
type AdminEventFilter struct {
	Types      []string
	Severities []string
	UnreadOnly bool
	Since      time.Time
	Limit      int
	Offset     int
}

// AdminEventPage carries the unread count alongside the rows, because the bell needs it
// and asking for it separately would mean the badge and the list could disagree.
type AdminEventPage struct {
	Events []AdminEvent `json:"events"`
	Total  int          `json:"total"`
	Unread int          `json:"unread"`
	Limit  int          `json:"limit"`
	Offset int          `json:"offset"`
}

// RecordAdminEvent persists one event and fills in its assigned id and timestamp, which
// the caller needs in order to broadcast the same row it stored — a subscriber shown an
// event with no id could neither mark it read nor recognise it on a later poll.
func (s *Store) RecordAdminEvent(ctx context.Context, event AdminEvent) (AdminEvent, error) {
	if event.Severity == "" {
		event.Severity = SeverityInfo
	}
	if len(event.Metadata) == 0 {
		event.Metadata = json.RawMessage(`{}`)
	}
	err := s.pool.QueryRow(ctx, `
		INSERT INTO admin_events (type, severity, title, summary, actor, target, link, metadata)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
		RETURNING id, occurred_at`,
		event.Type, event.Severity, event.Title, event.Summary,
		event.Actor, event.Target, event.Link, []byte(event.Metadata),
	).Scan(&event.ID, &event.OccurredAt)
	if err != nil {
		return event, fmt.Errorf("store: record admin event: %w", err)
	}
	return event, nil
}

func adminEventWhere(filter AdminEventFilter) (string, []any) {
	clauses := []string{"TRUE"}
	args := []any{}
	add := func(clause string, value any) {
		args = append(args, value)
		clauses = append(clauses, fmt.Sprintf(clause, len(args)))
	}
	if len(filter.Types) > 0 {
		add("type = ANY($%d::text[])", filter.Types)
	}
	if len(filter.Severities) > 0 {
		add("severity = ANY($%d::text[])", filter.Severities)
	}
	if filter.UnreadOnly {
		clauses = append(clauses, "read_at IS NULL")
	}
	if !filter.Since.IsZero() {
		add("occurred_at >= $%d", filter.Since)
	}
	return strings.Join(clauses, " AND "), args
}

// AdminEvents answers the filtered feed, newest first.
func (s *Store) AdminEvents(ctx context.Context, filter AdminEventFilter) (AdminEventPage, error) {
	limit := filter.Limit
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	offset := filter.Offset
	if offset < 0 {
		offset = 0
	}
	where, args := adminEventWhere(filter)

	page := AdminEventPage{Events: []AdminEvent{}, Limit: limit, Offset: offset}
	// The unread count is over the *whole* feed rather than the filter: the badge says how
	// much news there is, and a filtered view that also shrank the badge would report the
	// household as caught up because the operator had narrowed the list.
	if err := s.pool.QueryRow(ctx, `
		SELECT count(*) FILTER (WHERE `+where+`), count(*) FILTER (WHERE read_at IS NULL)
		FROM admin_events`, args...,
	).Scan(&page.Total, &page.Unread); err != nil {
		return page, fmt.Errorf("store: count admin events: %w", err)
	}

	rows, err := s.pool.Query(ctx, `
		SELECT id, occurred_at, type, severity, title, summary, actor, target, link,
			metadata, read_at
		FROM admin_events
		WHERE `+where+`
		ORDER BY occurred_at DESC, id DESC
		LIMIT $`+fmt.Sprint(len(args)+1)+` OFFSET $`+fmt.Sprint(len(args)+2),
		append(args, limit, offset)...)
	if err != nil {
		return page, fmt.Errorf("store: list admin events: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var event AdminEvent
		var metadata []byte
		if err := rows.Scan(&event.ID, &event.OccurredAt, &event.Type, &event.Severity,
			&event.Title, &event.Summary, &event.Actor, &event.Target, &event.Link,
			&metadata, &event.ReadAt); err != nil {
			return page, fmt.Errorf("store: scan admin event: %w", err)
		}
		event.Metadata = json.RawMessage(metadata)
		page.Events = append(page.Events, event)
	}
	return page, rows.Err()
}

// MarkAdminEventsRead marks the given ids, or every unread event when none are given.
func (s *Store) MarkAdminEventsRead(ctx context.Context, ids []int64) (int64, error) {
	if len(ids) == 0 {
		tag, err := s.pool.Exec(ctx,
			`UPDATE admin_events SET read_at = now() WHERE read_at IS NULL`)
		if err != nil {
			return 0, fmt.Errorf("store: mark admin events read: %w", err)
		}
		return tag.RowsAffected(), nil
	}
	tag, err := s.pool.Exec(ctx,
		`UPDATE admin_events SET read_at = now() WHERE read_at IS NULL AND id = ANY($1::bigint[])`,
		ids)
	if err != nil {
		return 0, fmt.Errorf("store: mark admin events read: %w", err)
	}
	return tag.RowsAffected(), nil
}

// UnreadAdminEvents is the badge on its own, for callers that want the count without the
// rows behind it.
func (s *Store) UnreadAdminEvents(ctx context.Context) (int, error) {
	var unread int
	err := s.pool.QueryRow(ctx,
		`SELECT count(*) FROM admin_events WHERE read_at IS NULL`).Scan(&unread)
	if err != nil {
		return 0, fmt.Errorf("store: count unread admin events: %w", err)
	}
	return unread, nil
}

// AdminEventTypeCount is one type's share of the feed, for the filter control — a list of
// types built from what has actually been published cannot offer a filter that matches
// nothing, and cannot miss a type added after this was written.
type AdminEventTypeCount struct {
	Type  string `json:"type"`
	Count int    `json:"count"`
}

func (s *Store) AdminEventTypes(ctx context.Context, since time.Time) ([]AdminEventTypeCount, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT type, count(*) FROM admin_events
		WHERE ($1::timestamptz IS NULL OR occurred_at >= $1)
		GROUP BY type ORDER BY count(*) DESC, type`, nullableTime(since))
	if err != nil {
		return nil, fmt.Errorf("store: admin event types: %w", err)
	}
	defer rows.Close()
	counts := []AdminEventTypeCount{}
	for rows.Next() {
		var count AdminEventTypeCount
		if err := rows.Scan(&count.Type, &count.Count); err != nil {
			return nil, fmt.Errorf("store: scan admin event type: %w", err)
		}
		counts = append(counts, count)
	}
	return counts, rows.Err()
}

// PruneAdminEvents drops events past the retention period.
func (s *Store) PruneAdminEvents(ctx context.Context, retention time.Duration) (int64, error) {
	if retention <= 0 {
		retention = AdminEventRetention
	}
	tag, err := s.pool.Exec(ctx,
		`DELETE FROM admin_events WHERE occurred_at < now() - $1::interval`,
		fmt.Sprintf("%d seconds", int64(retention.Seconds())))
	if err != nil {
		return 0, fmt.Errorf("store: prune admin events: %w", err)
	}
	return tag.RowsAffected(), nil
}

func nullableTime(value time.Time) any {
	if value.IsZero() {
		return nil
	}
	return value
}