From 301be79f0a752b24550621fecbc367a0bf8f2749 Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Sun, 31 May 2026 14:11:55 -0400 Subject: [PATCH] Per-user reload --- user/access_cache.go | 87 +++++++++++++++++++++++++++++---- user/manager.go | 102 ++++++++++++++++++++++++--------------- user/manager_postgres.go | 7 +++ user/manager_sqlite.go | 7 +++ user/types.go | 1 + 5 files changed, 156 insertions(+), 48 deletions(-) diff --git a/user/access_cache.go b/user/access_cache.go index cfd983b3..a1518058 100644 --- a/user/access_cache.go +++ b/user/access_cache.go @@ -64,17 +64,15 @@ func (c *accessCache) reload(d *db.DB, query string) error { patterns := make(map[string][]aclEntry) for rows.Next() { var username, topic string - var entry aclEntry - if err := rows.Scan(&username, &topic, &entry.read, &entry.write); err != nil { + var read, write bool + if err := rows.Scan(&username, &topic, &read, &write); err != nil { return err } - entry.length = len(topic) - if strings.Contains(topic, "%") { - re, err := compileLikeToRegex(topic) - if err != nil { - return err - } - entry.pattern = re + entry, isWildcard, err := newACLEntry(topic, read, write) + if err != nil { + return err + } + if isWildcard { patterns[username] = append(patterns[username], entry) } else { if exacts[username] == nil { @@ -93,6 +91,77 @@ func (c *accessCache) reload(d *db.DB, query string) error { return nil } +// reloadUser refreshes just one user's rules from the database and swaps them +// into the cache. Used as a cheaper, owner-cascade-safe alternative to the +// full bulk reload after a mutation that affects a known set of users. The +// caller is expected to invoke reloadUser for every user whose row set may +// have changed -- typically the targeted user plus Everyone, since most +// reservation flows touch both. +// +// The query must return (topic, read, write) for every row whose user_id +// matches the given username. An empty result set is treated as "this user +// has no rules": the user's entries are removed from both maps so the inner +// maps don't grow unbounded under churn. +func (c *accessCache) reloadUser(d *db.DB, query, username string) error { + rows, err := d.Query(query, username) + if err != nil { + return err + } + defer rows.Close() + exact := make(map[string]aclEntry) + var pattern []aclEntry + for rows.Next() { + var topic string + var read, write bool + if err := rows.Scan(&topic, &read, &write); err != nil { + return err + } + entry, isWildcard, err := newACLEntry(topic, read, write) + if err != nil { + return err + } + if isWildcard { + pattern = append(pattern, entry) + } else { + exact[topic] = entry + } + } + if err := rows.Err(); err != nil { + return err + } + c.mu.Lock() + if len(exact) == 0 { + delete(c.exact, username) + } else { + c.exact[username] = exact + } + if len(pattern) == 0 { + delete(c.pattern, username) + } else { + c.pattern[username] = pattern + } + c.mu.Unlock() + return nil +} + +// newACLEntry builds an aclEntry from one user_access row's values. The +// isWildcard return tells the caller which storage slot the entry belongs in: +// the per-user wildcard slice if true, the per-user exact map if false. +// Wildcards have their LIKE pattern pre-compiled into entry.pattern; exact +// entries leave entry.pattern nil. +func newACLEntry(topic string, read, write bool) (entry aclEntry, isWildcard bool, err error) { + entry = aclEntry{length: len(topic), read: read, write: write} + if !strings.Contains(topic, "%") { + return entry, false, nil + } + re, err := compileLikeToRegex(topic) + if err != nil { + return entry, true, err + } + entry.pattern = re + return entry, true, nil +} + // Lookup returns the effective (read, write, found) permission for the given // (username, topic), preserving the priority ordering of the original SQL query: // 1. specific user beats Everyone diff --git a/user/manager.go b/user/manager.go index f9995e84..4626d8e6 100644 --- a/user/manager.go +++ b/user/manager.go @@ -40,9 +40,8 @@ const ( DefaultUserPasswordBcryptCost = 10 // DefaultAccessCacheReloadInterval bounds how stale the in-memory ACL snapshot // can be relative to writes made by *other* processes (e.g. a separate `ntfy - // access` CLI invocation modifying the same database). Mutations performed - // by this Manager refresh the cache synchronously and do not depend on this. - DefaultAccessCacheReloadInterval = 5 * time.Second + // access` CLI invocation modifying the same database) + DefaultAccessCacheReloadInterval = 60 * time.Second ) var ( @@ -87,8 +86,6 @@ func newManager(d *db.DB, queries queries, config *Config) (*Manager, error) { if err := manager.maybeProvisionUsersAccessAndTokens(); err != nil { return nil, err } - // Populate the ACL cache after provisioning so the initial snapshot includes - // any provisioned access rules. Subsequent mutations call reloadAccessCache. if err := manager.reloadAccessCache(); err != nil { return nil, err } @@ -99,18 +96,22 @@ func newManager(d *db.DB, queries queries, config *Config) (*Manager, error) { return manager, nil } -// reloadAccessCache rebuilds the in-memory ACL snapshot from the primary -// database. Called once at startup (after provisioning) and after every method -// that mutates user_access (directly or by cascade from "user" deletion). +// reloadAccessCache rebuilds the in-memory access cache from the primary database func (a *Manager) reloadAccessCache() error { return a.accessCache.reload(a.db, a.queries.selectAllAccessForCache) } -// asyncAccessCacheReloader periodically refreshes the ACL snapshot so that -// writes made by other processes against the same database (most notably the -// `ntfy access` CLI subcommand running while a server holds the cache) become -// visible within the configured interval. This Manager's own writes do not -// depend on the poller -- they refresh the cache synchronously. +// reloadAccessCacheUsers refreshes the cache slices for the given usernames +func (a *Manager) reloadAccessCacheUsers(usernames ...string) error { + for _, username := range usernames { + if err := a.accessCache.reloadUser(a.db, a.queries.selectAccessForCacheByUser, username); err != nil { + return err + } + } + return nil +} + +// asyncAccessCacheReloader periodically refreshes the access cache func (a *Manager) asyncAccessCacheReloader(interval time.Duration) { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -198,13 +199,16 @@ func (a *Manager) RemoveUser(username string) error { if err := a.CanChangeUser(username); err != nil { return err } - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { return a.removeUserTx(tx, username) - }); err != nil { + }) + if err != nil { return err } - // user_access rows are cascade-deleted along with the user; refresh the snapshot. - return a.reloadAccessCache() + // user_access rows are cascade-deleted along with the user (both by user_id + // and by owner_user_id). Refresh this user's own slice (now empty) and + // Everyone's slice, since reservations owned by this user landed there too. + return a.reloadAccessCacheUsers(username, Everyone) } // removeUserTx deletes the user with the given username @@ -225,7 +229,7 @@ func (a *Manager) MarkUserRemoved(user *User) error { if !AllowedUsername(user.Name) { return ErrInvalidArgument } - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { if err := a.resetUserAccessTx(tx, user.Name); err != nil { return err } @@ -237,11 +241,14 @@ func (a *Manager) MarkUserRemoved(user *User) error { return err } return nil - }); err != nil { + }) + if err != nil { return err } - // resetUserAccessTx wiped this user's user_access rows; refresh the snapshot. - return a.reloadAccessCache() + // resetUserAccessTx deleted this user's rows AND any row owned by this user + // (typically the matching Everyone rows from their reservations). Refresh + // both slices to mirror the DB exactly. + return a.reloadAccessCacheUsers(user.Name, Everyone) } // RemoveDeletedUsers deletes all users that have been marked deleted @@ -281,15 +288,17 @@ func (a *Manager) ChangeRole(username string, role Role) error { if err := a.CanChangeUser(username); err != nil { return err } - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { return a.changeRoleTx(tx, username, role) - }); err != nil { + }) + if err != nil { return err } - // Promotion to admin clears user_access rows for the user; refresh the snapshot. - // Other role changes are no-ops for the cache but reloading is cheap and keeps - // the code path uniform. - return a.reloadAccessCache() + // Promotion to admin clears user_access rows for this user AND rows owned + // by this user (Everyone rows from their reservations). Other role changes + // are no-ops for the cache but reloading the two affected slices is cheap + // and keeps the code path uniform. + return a.reloadAccessCacheUsers(username, Everyone) } // changeRoleTx changes a user's role @@ -650,12 +659,14 @@ func (a *Manager) resolvePerms(base, perm Permission) error { // read/write access to a topic. The parameter topicPattern may include wildcards (*). The ACL entry // owner may either be a user (username), or the system (empty). func (a *Manager) AllowAccess(username string, topicPattern string, permission Permission) error { - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { return a.allowAccessTx(tx, username, topicPattern, permission, false) - }); err != nil { + }) + if err != nil { return err } - return a.reloadAccessCache() + // Only this user's row set changed; refresh their slice only. + return a.reloadAccessCacheUsers(username) } func (a *Manager) allowAccessTx(tx *sql.Tx, username string, topicPattern string, permission Permission, provisioned bool) error { @@ -671,12 +682,20 @@ func (a *Manager) allowAccessTx(tx *sql.Tx, username string, topicPattern string // ResetAccess removes an access control list entry for a specific username/topic, or (if topic is // empty) for an entire user. The parameter topicPattern may include wildcards (*). func (a *Manager) ResetAccess(username string, topicPattern string) error { - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { return a.resetAccessTx(tx, username, topicPattern) - }); err != nil { + }) + if err != nil { return err } - return a.reloadAccessCache() + // "Delete all access" affects every user; do the bulk reload. + // Otherwise refresh the named user plus Everyone, since resetUserAccessTx + // and deleteTopicAccess both touch rows owned by the user (typically + // Everyone rows from their reservations). + if username == "" { + return a.reloadAccessCache() + } + return a.reloadAccessCacheUsers(username, Everyone) } func (a *Manager) resetAccessTx(tx *sql.Tx, username string, topicPattern string) error { @@ -790,7 +809,7 @@ func (a *Manager) AddReservation(username string, topic string, everyone Permiss if !AllowedUsername(username) || username == Everyone || !AllowedTopic(topic) { return ErrInvalidArgument } - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { if limit > 0 { hasReservation, err := a.hasReservationTx(tx, username, topic) if err != nil { @@ -813,10 +832,12 @@ func (a *Manager) AddReservation(username string, topic string, everyone Permiss return err } return nil - }); err != nil { + }) + if err != nil { return err } - return a.reloadAccessCache() + // Both user's and Everyone's rows changed. + return a.reloadAccessCacheUsers(username, Everyone) } // RemoveReservations deletes the access control entries associated with the given username/topic, @@ -831,17 +852,20 @@ func (a *Manager) RemoveReservations(username string, topics ...string) error { return ErrInvalidArgument } } - if err := db.ExecTx(a.db, func(tx *sql.Tx) error { + err := db.ExecTx(a.db, func(tx *sql.Tx) error { for _, topic := range topics { if err := a.removeReservationAccessTx(tx, username, topic); err != nil { return err } } return nil - }); err != nil { + }) + if err != nil { return err } - return a.reloadAccessCache() + // Mirror the DB: rows for this user and any Everyone rows owned by this + // user are gone. Refresh both slices. + return a.reloadAccessCacheUsers(username, Everyone) } // Reservations returns all user-owned topics, and the associated everyone-access diff --git a/user/manager_postgres.go b/user/manager_postgres.go index d3bb6f5f..8dec4c7d 100644 --- a/user/manager_postgres.go +++ b/user/manager_postgres.go @@ -75,6 +75,12 @@ const ( FROM user_access a JOIN "user" u ON u.id = a.user_id ` + postgresSelectAccessForCacheByUserQuery = ` + SELECT a.topic, a.read, a.write + FROM user_access a + JOIN "user" u ON u.id = a.user_id + WHERE u.user_name = $1 + ` postgresSelectUserAllAccessQuery = ` SELECT user_id, topic, read, write, provisioned FROM user_access @@ -243,6 +249,7 @@ var postgresQueries = queries{ deleteUsersMarked: postgresDeleteUsersMarkedQuery, deleteUsersProvisioned: postgresDeleteUsersProvisionedQuery, selectAllAccessForCache: postgresSelectAllAccessForCacheQuery, + selectAccessForCacheByUser: postgresSelectAccessForCacheByUserQuery, selectUserAllAccess: postgresSelectUserAllAccessQuery, selectUserAccess: postgresSelectUserAccessQuery, selectUserReservations: postgresSelectUserReservationsQuery, diff --git a/user/manager_sqlite.go b/user/manager_sqlite.go index 4d498ce2..26e22b36 100644 --- a/user/manager_sqlite.go +++ b/user/manager_sqlite.go @@ -81,6 +81,12 @@ const ( FROM user_access a JOIN user u ON u.id = a.user_id ` + sqliteSelectAccessForCacheByUserQuery = ` + SELECT a.topic, a.read, a.write + FROM user_access a + JOIN user u ON u.id = a.user_id + WHERE u.user = ? + ` sqliteSelectUserAllAccessQuery = ` SELECT user_id, topic, read, write, provisioned FROM user_access @@ -241,6 +247,7 @@ var sqliteQueries = queries{ deleteUsersMarked: sqliteDeleteUsersMarkedQuery, deleteUsersProvisioned: sqliteDeleteUsersProvisionedQuery, selectAllAccessForCache: sqliteSelectAllAccessForCacheQuery, + selectAccessForCacheByUser: sqliteSelectAccessForCacheByUserQuery, selectUserAllAccess: sqliteSelectUserAllAccessQuery, selectUserAccess: sqliteSelectUserAccessQuery, selectUserReservations: sqliteSelectUserReservationsQuery, diff --git a/user/types.go b/user/types.go index 6d2650ed..3af01f5d 100644 --- a/user/types.go +++ b/user/types.go @@ -314,6 +314,7 @@ type queries struct { // Access queries selectAllAccessForCache string // Bulk load: (user_name, topic, read, write) for the in-memory ACL cache + selectAccessForCacheByUser string // Per-user load: (topic, read, write) for one username; used to refresh just one user's slice of the cache after mutation selectUserAllAccess string selectUserAccess string selectUserReservations string