root
8bbb3179a1
Why: collaborative KeePass use needs one arbiter of "who wrote last" that never sees a master password, plus access control that follows an org chart. What: stdlib-only Go server. Metadata is a single JSON document with one writer and atomic commit; KDBX payloads are versioned blobs on disk. Access comes from a group tree where a grant on a group reaches every member below it. Login is password+TOTP, invite links, or OIDC through a browser-approved device flow. Pushes carry If-Match and are arbitrated by a per-database merge strategy (server-wins by default). Updates are pushed to clients over SSE. Proved: go test ./internal/... covers RFC 6238 vectors, ACL inheritance and isolation, store rollback, blob pruning, every push verdict, and an end-to-end HTTP pass over login, conflict, force, read-only grants, invite redemption, TOTP enrolment, device login and the event stream.
95 lines
2.1 KiB
Go
95 lines
2.1 KiB
Go
package server
|
|
|
|
import (
|
|
"encoding/json"
|
|
"sync"
|
|
)
|
|
|
|
// Event is one push notification sent to clients over SSE.
|
|
type Event struct {
|
|
Type string `json:"type"`
|
|
DatabaseID string `json:"database_id,omitempty"`
|
|
Name string `json:"name,omitempty"`
|
|
Version int64 `json:"version,omitempty"`
|
|
SHA256 string `json:"sha256,omitempty"`
|
|
By string `json:"by,omitempty"`
|
|
}
|
|
|
|
const (
|
|
EventUpdated = "db-updated"
|
|
EventChanged = "db-list-changed"
|
|
)
|
|
|
|
// subBuffer is how many events a slow subscriber may fall behind before its
|
|
// stream is dropped. A dropped client reconnects and resyncs from scratch,
|
|
// so losing events is safe; blocking the publisher would not be.
|
|
const subBuffer = 16
|
|
|
|
type subscriber struct {
|
|
userID string
|
|
ch chan Event
|
|
}
|
|
|
|
// Hub fans events out to the connected clients of specific users.
|
|
type Hub struct {
|
|
mu sync.Mutex
|
|
next int
|
|
subs map[int]*subscriber
|
|
}
|
|
|
|
func NewHub() *Hub { return &Hub{subs: map[int]*subscriber{}} }
|
|
|
|
// Subscribe registers a stream for one user and returns its id and channel.
|
|
func (h *Hub) Subscribe(userID string) (int, <-chan Event) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
h.next++
|
|
s := &subscriber{userID: userID, ch: make(chan Event, subBuffer)}
|
|
h.subs[h.next] = s
|
|
return h.next, s.ch
|
|
}
|
|
|
|
// Unsubscribe removes a stream and closes its channel.
|
|
func (h *Hub) Unsubscribe(id int) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if s, ok := h.subs[id]; ok {
|
|
delete(h.subs, id)
|
|
close(s.ch)
|
|
}
|
|
}
|
|
|
|
// Publish delivers ev to every stream of every listed user. A stream that is
|
|
// not keeping up is dropped rather than waited for.
|
|
func (h *Hub) Publish(userIDs []string, ev Event) {
|
|
want := make(map[string]bool, len(userIDs))
|
|
for _, id := range userIDs {
|
|
want[id] = true
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for id, s := range h.subs {
|
|
if !want[s.userID] {
|
|
continue
|
|
}
|
|
select {
|
|
case s.ch <- ev:
|
|
default:
|
|
delete(h.subs, id)
|
|
close(s.ch)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Count reports the number of live streams, for the health endpoint.
|
|
func (h *Hub) Count() int {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
return len(h.subs)
|
|
}
|
|
|
|
func (e Event) encode() []byte {
|
|
b, _ := json.Marshal(e)
|
|
return b
|
|
}
|