248 lines
8.6 KiB
Go
248 lines
8.6 KiB
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
|
|
}
|