package store import ( "context" "encoding/json" "fmt" "strings" "time" "github.com/ponzischeme89/memby/server/internal/notify" ) // The notification log: what Memby sent, to whom, over which channel, and what became of // it. Written by internal/notify — nothing else writes this table, which is the whole // point of it — and read only by the console. // // It is deliberately a separate table from user_notifications rather than a set of extra // columns on it. That table is *state*: one viewer's undismissed list, which they empty. // This is *history*: it keeps a row for a notification that was dismissed, for one that // was never delivered, and for a broadcast that belongs to no viewer at all — none of // which the other table can represent. // NotificationRetention is how far back the log goes. Ninety days is long enough that a // question about "the summary I never got last month" is still answerable, and short // enough that the table cannot outgrow the database on a household gateway. The // housekeeping task prunes to it; the console derives its widest window from it, so the // page can never offer a range the data does not cover. const NotificationRetention = 90 * 24 * time.Hour // notificationTextLimit bounds a stored string. A notification body is a sentence or two // by construction, and this is only here so a bug in a producer cannot write a megabyte // per row into the audit trail. const notificationTextLimit = 2000 // NotificationLogEntry is one delivered — or refused — notification as the console reads // it. type NotificationLogEntry struct { ID int64 `json:"id"` OccurredAt time.Time `json:"occurredAt"` Channel string `json:"channel"` Kind string `json:"kind"` Source string `json:"source"` UserID string `json:"userId,omitempty"` Username string `json:"username,omitempty"` Title string `json:"title"` Body string `json:"body,omitempty"` ItemID string `json:"itemId,omitempty"` Target string `json:"target,omitempty"` SourceKey string `json:"sourceKey,omitempty"` Status string `json:"status"` Detail string `json:"detail,omitempty"` DurationMS int64 `json:"durationMs"` EventAt *time.Time `json:"eventAt,omitempty"` Metadata json.RawMessage `json:"metadata,omitempty"` } // NotificationLogFilter is the console's question. Every field is optional and they // combine with AND, which is what makes the filter bar above the table read the way it // behaves. type NotificationLogFilter struct { UserID string Kinds []string Channels []string Statuses []string Sources []string // Query searches the title, the body, the failure detail and the recipient's name. One // box rather than four, because an operator arriving here is looking for a *thing* they // half remember and does not yet know which column it is in. Query string From time.Time To time.Time Limit int Offset int } // NotificationLogPage is a window on the log plus the counts the page heads itself with. type NotificationLogPage struct { Entries []NotificationLogEntry `json:"entries"` Total int `json:"total"` Limit int `json:"limit"` Offset int `json:"offset"` } // NotificationLogTotals summarises the filtered window. Counted by its own query rather // than tallied from the page, for the reason SearchTotals is: the page is capped, so // adding it up would report the first hundred rows' totals as the window's. type NotificationLogTotals struct { Total int `json:"total"` Sent int `json:"sent"` Delivered int `json:"delivered"` Failed int `json:"failed"` Pending int `json:"pending"` Skipped int `json:"skipped"` Users int `json:"users"` } // NotificationFacet is one value of a filterable column and how many rows carry it. The // console builds its dropdowns from these rather than from a list of constants, so the // filter can neither offer a type that matches nothing nor miss one a feature added after // the page was written — the stance the activity feed's type filter takes. type NotificationFacet struct { Value string `json:"value"` Count int `json:"count"` } // NotificationFacets is every dropdown on the page. type NotificationFacets struct { Kinds []NotificationFacet `json:"kinds"` Channels []NotificationFacet `json:"channels"` Statuses []NotificationFacet `json:"statuses"` Sources []NotificationFacet `json:"sources"` } // RecordNotification writes one row of the audit trail. // // It implements notify.Recorder, which is the only thing that calls it. Text is clamped // here rather than at the caller so one careless producer cannot be the reason the console // takes a second to draw. func (s *Store) RecordNotification(ctx context.Context, record notify.Record) error { if s == nil || s.pool == nil { return nil } occurred := record.OccurredAt if occurred.IsZero() { occurred = time.Now().UTC() } var metadata any if len(record.Metadata) > 0 { metadata = []byte(record.Metadata) } _, err := s.pool.Exec(ctx, ` INSERT INTO notification_log (occurred_at, channel, kind, source, emby_user_id, username, title, body, item_id, target, source_key, status, detail, duration_ms, event_at, metadata) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)`, occurred, string(record.Channel), record.Kind, record.Source, record.UserID, record.Username, clampText(record.Title), clampText(record.Body), record.ItemID, record.Target, record.SourceKey, string(record.Status), clampText(record.Detail), record.DurationMS, record.EventAt, metadata) if err != nil { return fmt.Errorf("store: record notification: %w", err) } return nil } func clampText(value string) string { runes := []rune(value) if len(runes) <= notificationTextLimit { return value } return string(runes[:notificationTextLimit]) + "…" } // notificationWhere builds the shared predicate. The log, the totals and the facets all // answer for the *same* filtered window, so they must be filtered identically — writing // the clause three times is how a page comes to show a total that disagrees with its own // table. func notificationWhere(filter NotificationLogFilter) (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 filter.UserID != "" { add("emby_user_id = $%d", filter.UserID) } if len(filter.Kinds) > 0 { add("kind = ANY($%d)", filter.Kinds) } if len(filter.Channels) > 0 { add("channel = ANY($%d)", filter.Channels) } if len(filter.Statuses) > 0 { add("status = ANY($%d)", filter.Statuses) } if len(filter.Sources) > 0 { add("source = ANY($%d)", filter.Sources) } if !filter.From.IsZero() { add("occurred_at >= $%d", filter.From) } if !filter.To.IsZero() { add("occurred_at < $%d", filter.To) } if query := strings.TrimSpace(filter.Query); query != "" { // ILIKE over four columns rather than a tsvector: this table is a few tens of // thousands of rows on a household gateway, always read with a date bound, and the // operator is looking for a substring of a title or an error message — which is // exactly what full-text search is worst at. add("(title ILIKE $%[1]d OR body ILIKE $%[1]d OR detail ILIKE $%[1]d OR username ILIKE $%[1]d)", "%"+query+"%") } return strings.Join(clauses, " AND "), args } // NotificationLog reads the filtered window, newest first. func (s *Store) NotificationLog( ctx context.Context, filter NotificationLogFilter, ) (NotificationLogPage, error) { limit := filter.Limit if limit <= 0 || limit > 500 { limit = 100 } offset := filter.Offset if offset < 0 { offset = 0 } where, args := notificationWhere(filter) page := NotificationLogPage{Entries: []NotificationLogEntry{}, Limit: limit, Offset: offset} if err := s.pool.QueryRow(ctx, `SELECT count(*) FROM notification_log WHERE `+where, args..., ).Scan(&page.Total); err != nil { return page, fmt.Errorf("store: count notification log: %w", err) } rows, err := s.pool.Query(ctx, ` SELECT id, occurred_at, channel, kind, source, emby_user_id, username, title, body, item_id, target, source_key, status, detail, duration_ms, event_at, metadata FROM notification_log 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: read notification log: %w", err) } defer rows.Close() for rows.Next() { var entry NotificationLogEntry var metadata []byte if err := rows.Scan( &entry.ID, &entry.OccurredAt, &entry.Channel, &entry.Kind, &entry.Source, &entry.UserID, &entry.Username, &entry.Title, &entry.Body, &entry.ItemID, &entry.Target, &entry.SourceKey, &entry.Status, &entry.Detail, &entry.DurationMS, &entry.EventAt, &metadata, ); err != nil { return page, fmt.Errorf("store: scan notification log: %w", err) } if len(metadata) > 0 { entry.Metadata = json.RawMessage(metadata) } page.Entries = append(page.Entries, entry) } if err := rows.Err(); err != nil { return page, fmt.Errorf("store: read notification log: %w", err) } return page, nil } // NotificationLogTotals counts the same window the log is read with. func (s *Store) NotificationLogTotals( ctx context.Context, filter NotificationLogFilter, ) (NotificationLogTotals, error) { where, args := notificationWhere(filter) var totals NotificationLogTotals err := s.pool.QueryRow(ctx, ` SELECT count(*), count(*) FILTER (WHERE status = 'sent'), count(*) FILTER (WHERE status = 'delivered'), count(*) FILTER (WHERE status = 'failed'), count(*) FILTER (WHERE status = 'pending'), count(*) FILTER (WHERE status = 'skipped'), count(DISTINCT emby_user_id) FILTER (WHERE emby_user_id <> '') FROM notification_log WHERE `+where, args...).Scan( &totals.Total, &totals.Sent, &totals.Delivered, &totals.Failed, &totals.Pending, &totals.Skipped, &totals.Users) if err != nil { return totals, fmt.Errorf("store: notification totals: %w", err) } return totals, nil } // NotificationLogFacets lists what the filters may offer. // // Deliberately computed over the retention window rather than over the operator's current // filter: a dropdown whose options disappear as you narrow the table is one you cannot use // to widen the question again. func (s *Store) NotificationLogFacets( ctx context.Context, since time.Time, ) (NotificationFacets, error) { facets := NotificationFacets{ Kinds: []NotificationFacet{}, Channels: []NotificationFacet{}, Statuses: []NotificationFacet{}, Sources: []NotificationFacet{}, } rows, err := s.pool.Query(ctx, ` SELECT 'kind', kind, count(*) FROM notification_log WHERE occurred_at >= $1 AND kind <> '' GROUP BY kind UNION ALL SELECT 'channel', channel, count(*) FROM notification_log WHERE occurred_at >= $1 AND channel <> '' GROUP BY channel UNION ALL SELECT 'status', status, count(*) FROM notification_log WHERE occurred_at >= $1 AND status <> '' GROUP BY status UNION ALL SELECT 'source', source, count(*) FROM notification_log WHERE occurred_at >= $1 AND source <> '' GROUP BY source ORDER BY 3 DESC, 2`, since) if err != nil { return facets, fmt.Errorf("store: notification facets: %w", err) } defer rows.Close() for rows.Next() { var group string var facet NotificationFacet if err := rows.Scan(&group, &facet.Value, &facet.Count); err != nil { return facets, fmt.Errorf("store: scan notification facet: %w", err) } switch group { case "kind": facets.Kinds = append(facets.Kinds, facet) case "channel": facets.Channels = append(facets.Channels, facet) case "status": facets.Statuses = append(facets.Statuses, facet) case "source": facets.Sources = append(facets.Sources, facet) } } return facets, rows.Err() } // NotificationLogDays is the daily shape of the filtered window, for the chart above the // table. Grouped in the database's own timezone, the stance the sign-in history takes, so // an evening notification stays on the day it happened. type NotificationLogDay struct { Day string `json:"day"` Sent int `json:"sent"` Failed int `json:"failed"` Skipped int `json:"skipped"` Delivered int `json:"delivered"` } func (s *Store) NotificationLogDays( ctx context.Context, filter NotificationLogFilter, ) ([]NotificationLogDay, error) { where, args := notificationWhere(filter) rows, err := s.pool.Query(ctx, ` SELECT to_char(date_trunc('day', occurred_at), 'YYYY-MM-DD'), count(*) FILTER (WHERE status = 'sent'), count(*) FILTER (WHERE status = 'failed'), count(*) FILTER (WHERE status = 'skipped'), count(*) FILTER (WHERE status = 'delivered') FROM notification_log WHERE `+where+` GROUP BY 1 ORDER BY 1`, args...) if err != nil { return nil, fmt.Errorf("store: notification days: %w", err) } defer rows.Close() days := []NotificationLogDay{} for rows.Next() { var day NotificationLogDay if err := rows.Scan(&day.Day, &day.Sent, &day.Failed, &day.Skipped, &day.Delivered); err != nil { return nil, fmt.Errorf("store: scan notification day: %w", err) } days = append(days, day) } return days, rows.Err() } // PruneNotificationLog is the retention policy, run by the housekeeping scheduler. func (s *Store) PruneNotificationLog(ctx context.Context, older time.Duration) (int64, error) { if older <= 0 { return 0, nil } tag, err := s.pool.Exec(ctx, `DELETE FROM notification_log WHERE occurred_at < now() - $1::interval`, older.String()) if err != nil { return 0, fmt.Errorf("store: prune notification log: %w", err) } return tag.RowsAffected(), nil }