- commit
- da03ad396f629daaeff2332a94db7d4923f54ca8
- parent
- 758f7de874554b1aa08ff5b6ec1537d9d2c179dd
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 4 | +- |
| M | internal/ | 4 | + |
| M | internal/ | 18 | +++ |
| M | internal/ | 4 | + |
| A | internal/ | 64 | ++++++++++ |
| A | internal/ | 194 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 151 | +++++++++++++++++++++++ |
| M | internal/ | 12 | ++ |
| M | internal/ | 28 | ++++- |
| M | internal/ | 24 | ++++- |
| M | internal/ | 49 | ++++++-- |
| M | internal/ | 2 | +- |
@@ -97,7 +97,9 @@ func RoomAdminKind(kind int) bool {
9797 }9898 9999 // RoomNoticeKind reports whether kind is a relay-signed member notice.100-func RoomNoticeKind(kind int) bool { return kind == event.KIND_ROOM_MEMBER_ADDED || kind == event.KIND_ROOM_MEMBER_REMOVED }100+func RoomNoticeKind(kind int) bool {101+ return kind == event.KIND_ROOM_MEMBER_ADDED || kind == event.KIND_ROOM_MEMBER_REMOVED102+}101103 102104 // RoomAccessFor maps the tenant read rule onto the slug room's access.103105 func RoomAccessFor(reads string) string {@@ -48,6 +48,10 @@ func (t *Tenant) ServeHTTP(w http.ResponseWriter, r *http.Request) {
4848 t.router.HandleHTTP(w, r)4949 return5050 }51+ if isRoomStreamPath(r.URL.Path) {52+ t.roomStreamHTTP(w, r)53+ return54+ }5155 if r.URL.Path != "/backups/restore" && r.URL.Path != "/manage/jobs/status" {5256 ctx, done, err := t.beginOperation(r.Context())5357 if err != nil {@@ -163,6 +163,24 @@ func (g *maintenanceGate) endMaintenance() func() {
163163 }164164 }165165 166+// watch returns a channel that closes on the next gate change, so a167+// long-lived reader can stop when maintenance or shutdown begins.168+func (g *maintenanceGate) watch() <-chan struct{} {169+ g.mu.Lock()170+ defer g.mu.Unlock()171+ if g.changed == nil {172+ g.changed = make(chan struct{})173+ }174+ return g.changed175+}176+177+// blocked reports whether new operations are being refused.178+func (g *maintenanceGate) blocked() bool {179+ g.mu.Lock()180+ defer g.mu.Unlock()181+ return g.maintenance || g.closing182+}183+166184 func (g *maintenanceGate) signalLocked() {167185 if g.changed != nil {168186 close(g.changed)@@ -37,6 +37,10 @@ func (t *Tenant) executeManagement(ctx context.Context, actor, method string, pa
3737 result, err = t.executeBrowse(ctx, actor, method, params)3838 return result, true, err3939 }40+ if roomBrowseMethod(method) {41+ result, err = t.executeRoomBrowse(ctx, actor, method, params)42+ return result, true, err43+ }4044 if publicManagementMethod(method) {4145 return t.executePublicManagement(ctx, actor, method, params)4246 }@@ -0,0 +1,64 @@
1+package daemon2+3+import (4+ "context"5+ "time"6+7+ "github.com/FelineStateMachine/tinyrelay/internal/community"8+ "github.com/FelineStateMachine/tinyrelay/internal/event"9+)10+11+// roomMessageKinds are the events a room timeline shows: chat, threads and12+// their replies, reactions, Buzz rich content and edits, and the relay's13+// member notices.14+var roomMessageKinds = []int{event.KIND_CHAT, event.KIND_THREAD, event.KIND_THREAD_REPLY, 7, event.KIND_RICH_CONTENT, event.KIND_CONTENT_EDIT, event.KIND_ROOM_MEMBER_ADDED, event.KIND_ROOM_MEMBER_REMOVED}15+16+// projectRoom refreshes the relay's records for a room after a durable17+// change. The slug room's 39000 to 39002 records come from the tenant18+// projection; here it only receives member notices. A room that has been19+// deleted since the change was queued has its records retired instead.20+func (t *Tenant) projectRoom(ctx context.Context, roomID, pubkey string, removed, deleted bool) error {21+ now := time.Now().Unix()22+ room, err := t.community.Room(ctx, roomID)23+ if err != nil {24+ return err25+ }26+ if deleted || !room.Live() {27+ return t.records.RetireRoom(ctx, roomID)28+ }29+ if roomID != t.meta.Name {30+ if _, err := t.records.PublishRoom(ctx, roomID, now); err != nil {31+ return err32+ }33+ }34+ if pubkey == "" {35+ return nil36+ }37+ _, err = t.records.RoomNotice(ctx, roomID, pubkey, !removed, now)38+ return err39+}40+41+// roomFor loads a live room the actor may see, or explains why not.42+func (t *Tenant) roomFor(ctx context.Context, actor, id string) (community.Room, string, error) {43+ if !community.ValidRoomID(id) {44+ return community.Room{}, "", errRoomNotFound45+ }46+ room, err := t.community.Room(ctx, id)47+ if err != nil {48+ return community.Room{}, "", err49+ }50+ if !room.Live() {51+ return community.Room{}, "", errRoomNotFound52+ }53+ role, err := t.community.RoomRole(ctx, id, actor)54+ if err != nil {55+ return community.Room{}, "", err56+ }57+ if room.Access == community.RoomMembers && role == "" {58+ if actor == "" {59+ return community.Room{}, "", errRoomAuth60+ }61+ return community.Room{}, "", errRoomRestricted62+ }63+ return room, role, nil64+}@@ -0,0 +1,194 @@
1+package daemon2+3+import (4+ "context"5+ "encoding/json"6+ "errors"7+ "fmt"8+ "time"9+10+ "github.com/FelineStateMachine/tinyrelay/internal/community"11+ "github.com/FelineStateMachine/tinyrelay/internal/event"12+ "github.com/FelineStateMachine/tinyrelay/internal/storage"13+)14+15+var (16+ errRoomNotFound = errors.New("not found: room")17+ errRoomAuth = errors.New("auth-required: this room is members-only")18+ errRoomRestricted = errors.New("restricted: this room is members-only")19+)20+21+type roomBrowseRequest struct {22+ ID string `json:"id"`23+ Event string `json:"event"`24+ Cursor string `json:"cursor"`25+ Limit int `json:"limit"`26+}27+28+type roomSummary struct {29+ community.Room30+ Members int `json:"members"`31+ LastMessageAt int64 `json:"last_message_at"`32+ Role string `json:"role"`33+}34+35+func roomBrowseMethod(method string) bool {36+ return method == "browserooms" || method == "browseroom" || method == "browsethread"37+}38+39+// executeRoomBrowse serves the room reads used by the web pages and agents.40+// Access follows the tenant read rule and then the room's own access.41+func (t *Tenant) executeRoomBrowse(ctx context.Context, actor, method string, params []json.RawMessage) (result any, err error) {42+ ctx, finish := t.app.telemetry.Start(ctx, "rooms")43+ defer func() {44+ if err != nil {45+ finish("error")46+ } else {47+ finish("success")48+ }49+ }()50+ if err := t.browseRead(ctx, actor); err != nil {51+ return nil, err52+ }53+ q := roomBrowseRequest{}54+ if len(params) > 0 {55+ if err := json.Unmarshal(params[0], &q); err != nil {56+ return nil, fmt.Errorf("invalid: room parameters: %w", err)57+ }58+ }59+ if q.Limit <= 0 || q.Limit > 100 {60+ q.Limit = 10061+ }62+ switch method {63+ case "browserooms":64+ return t.browseRooms(ctx, actor, q)65+ case "browseroom":66+ return t.browseRoom(ctx, actor, q)67+ case "browsethread":68+ return t.browseThread(ctx, actor, q)69+ }70+ return nil, errors.New("unsupported: room operation")71+}72+73+func (t *Tenant) browseRooms(ctx context.Context, actor string, q roomBrowseRequest) (any, error) {74+ rooms, err := t.community.Rooms(ctx)75+ if err != nil {76+ return nil, err77+ }78+ items := []roomSummary{}79+ next := ""80+ for _, room := range rooms {81+ if room.ID <= q.Cursor {82+ continue83+ }84+ _, role, err := t.roomFor(ctx, actor, room.ID)85+ if err != nil {86+ continue87+ }88+ if len(items) == q.Limit {89+ next = items[len(items)-1].ID90+ break91+ }92+ summary, err := t.roomSummary(ctx, actor, room, role)93+ if err != nil {94+ return nil, err95+ }96+ items = append(items, summary)97+ }98+ return map[string]any{"items": items, "next_cursor": next}, nil99+}100+101+func (t *Tenant) roomSummary(ctx context.Context, actor string, room community.Room, role string) (roomSummary, error) {102+ count, err := t.community.RoomMemberCount(ctx, room.ID)103+ if err != nil {104+ return roomSummary{}, err105+ }106+ summary := roomSummary{Room: room, Members: count, Role: role}107+ latest, err := t.store.Query(ctx, event.Filter{Kinds: []int{event.KIND_CHAT, event.KIND_THREAD, event.KIND_THREAD_REPLY, event.KIND_RICH_CONTENT}, Tags: map[string][]string{"h": {room.ID}}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: browseSession(t, actor).PubKeys}, Limit: 1})108+ if err != nil {109+ return roomSummary{}, err110+ }111+ if len(latest.Events) > 0 {112+ summary.LastMessageAt = latest.Events[0].CreatedAt113+ }114+ return summary, nil115+}116+117+func (t *Tenant) browseRoom(ctx context.Context, actor string, q roomBrowseRequest) (any, error) {118+ room, role, err := t.roomFor(ctx, actor, q.ID)119+ if err != nil {120+ return nil, err121+ }122+ cursor, err := parseCollaborationCursor(q.Cursor)123+ if err != nil {124+ return nil, errors.New("invalid: room cursor")125+ }126+ summary, err := t.roomSummary(ctx, actor, room, role)127+ if err != nil {128+ return nil, err129+ }130+ members := []community.RoomMember{}131+ if role != "" || t.Policy().DirectoryPublic {132+ members, err = t.community.RoomMembers(ctx, room.ID)133+ if err != nil {134+ return nil, err135+ }136+ }137+ messages, next, err := t.roomMessages(ctx, actor, event.Filter{Kinds: roomMessageKinds, Tags: map[string][]string{"h": {room.ID}}}, cursor, q.Limit)138+ if err != nil {139+ return nil, err140+ }141+ return map[string]any{"room": summary, "members": members, "messages": messages, "next_cursor": next}, nil142+}143+144+func (t *Tenant) browseThread(ctx context.Context, actor string, q roomBrowseRequest) (any, error) {145+ room, _, err := t.roomFor(ctx, actor, q.ID)146+ if err != nil {147+ return nil, err148+ }149+ if len(q.Event) != 64 || !hexLower(q.Event) {150+ return nil, errors.New("invalid: thread root")151+ }152+ cursor, err := parseCollaborationCursor(q.Cursor)153+ if err != nil {154+ return nil, errors.New("invalid: thread cursor")155+ }156+ roots, err := t.Query(ctx, []event.Filter{{IDs: []string{q.Event}, Limit: intPtr(1)}}, browseSession(t, actor))157+ if err != nil {158+ return nil, err159+ }160+ if len(roots) != 1 || event.Tag(roots[0], "h") != room.ID || (roots[0].Kind != event.KIND_THREAD && roots[0].Kind != event.KIND_CHAT && roots[0].Kind != event.KIND_RICH_CONTENT) {161+ return nil, errors.New("not found: thread")162+ }163+ replies, next, err := t.roomMessages(ctx, actor, event.Filter{Kinds: []int{event.KIND_THREAD_REPLY}, Tags: map[string][]string{"h": {room.ID}, "e": {q.Event}}}, cursor, q.Limit)164+ if err != nil {165+ return nil, err166+ }167+ return map[string]any{"room": room, "root": roots[0], "replies": replies, "next_cursor": next}, nil168+}169+170+// roomMessages pages a room filter newest first through the read gate. The171+// cursor is the created_at and id of the last event on the previous page.172+func (t *Tenant) roomMessages(ctx context.Context, actor string, filter event.Filter, cursor *storage.EventCursor, limit int) ([]event.Event, string, error) {173+ session := browseSession(t, actor)174+ if _, err := t.gate.Read(ctx, []event.Filter{filter}, session); err != nil {175+ return nil, "", err176+ }177+ page, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: session.PubKeys}, Limit: limit, Before: cursor})178+ if err != nil {179+ return nil, "", err180+ }181+ rows := make([]event.Event, 0, len(page.Events))182+ for _, row := range page.Events {183+ if t.gate.CanSee(ctx, row, session, nil) {184+ rows = append(rows, row)185+ }186+ }187+ sortEventsNewestFirst(rows)188+ next := ""189+ if page.More && len(page.Events) > 0 {190+ last := page.Events[len(page.Events)-1]191+ next = collaborationCursor(collaborationItemFrom(last))192+ }193+ return rows, next, nil194+}@@ -0,0 +1,151 @@
1+package daemon2+3+import (4+ "encoding/json"5+ "net/http"6+ "net/url"7+ "regexp"8+ "strings"9+ "time"10+11+ "github.com/FelineStateMachine/tinyrelay/internal/event"12+)13+14+const (15+ roomStreamKeepAlive = 25 * time.Second16+ roomStreamBuffer = 6417+)18+19+var roomStreamPath = regexp.MustCompile(`^/rooms/([a-z0-9_-]{1,64})/stream$`)20+21+func isRoomStreamPath(path string) bool { return roomStreamPath.MatchString(path) }22+23+// roomStreamHTTP serves GET /rooms/<id>/stream as server-sent events. Each24+// event the viewer may see in the room arrives as "event: message" with the25+// JSON event as data; a comment keeps the connection alive. The viewer is26+// identified by the browser session cookie or a NIP-98 signature.27+func (t *Tenant) roomStreamHTTP(w http.ResponseWriter, r *http.Request) {28+ ctx, finish := t.app.telemetry.Start(r.Context(), "rooms")29+ outcome := "error"30+ defer func() { finish(outcome) }()31+ if r.Method != http.MethodGet {32+ w.Header().Set("Allow", "GET")33+ http.Error(w, "method not allowed", http.StatusMethodNotAllowed)34+ outcome = "invalid"35+ return36+ }37+ id := roomStreamPath.FindStringSubmatch(r.URL.Path)[1]38+ if origin := r.Header.Get("Origin"); origin != "" && r.Header.Get("Authorization") == "" {39+ base, err := url.Parse(t.requestURL(r))40+ if err != nil || origin != base.Scheme+"://"+base.Host {41+ http.Error(w, "auth-required: sign this request", http.StatusUnauthorized)42+ outcome = "unauthorized"43+ return44+ }45+ }46+ actor, err := t.resolveUIActor(r)47+ if err != nil {48+ browseHTTPError(w, err)49+ outcome = "unauthorized"50+ return51+ }52+ // Hold an operation only while checking access. A stream may stay open53+ // for hours and must not keep maintenance waiting; it ends instead when54+ // maintenance or shutdown begins.55+ opCtx, done, err := t.beginOperation(ctx)56+ if err != nil {57+ http.Error(w, err.Error(), http.StatusServiceUnavailable)58+ outcome = "busy"59+ return60+ }61+ err = t.browseRead(opCtx, actor)62+ if err == nil {63+ _, _, err = t.roomFor(opCtx, actor, id)64+ }65+ done()66+ if err != nil {67+ browseHTTPError(w, err)68+ outcome = "unauthorized"69+ return70+ }71+ flusher, ok := w.(http.Flusher)72+ if !ok {73+ http.Error(w, "streaming unsupported", http.StatusInternalServerError)74+ return75+ }76+ session := browseSession(t, actor)77+ events := make(chan event.Event, roomStreamBuffer)78+ overflow := make(chan struct{}, 1)79+ stop := t.router.Listen(func(e event.Event) {80+ if e.Kind == event.KIND_MARMOT_GROUP || event.Tag(e, "h") != id {81+ return82+ }83+ select {84+ case events <- e:85+ default:86+ select {87+ case overflow <- struct{}{}:88+ default:89+ }90+ }91+ })92+ defer stop()93+ w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")94+ w.Header().Set("Cache-Control", "no-store, private")95+ w.Header().Set("X-Accel-Buffering", "no")96+ w.WriteHeader(http.StatusOK)97+ if _, err := w.Write([]byte(": room " + id + "\n\n")); err != nil {98+ return99+ }100+ flusher.Flush()101+ keepAlive := time.NewTicker(roomStreamKeepAlive)102+ defer keepAlive.Stop()103+ delivered := 0104+ defer func() {105+ t.app.telemetry.Logger().Debug("room stream closed", "events", delivered, "outcome", outcome)106+ }()107+ paused := t.maintenance.watch()108+ for {109+ select {110+ case <-r.Context().Done():111+ outcome = "closed"112+ return113+ case <-t.workCtx.Done():114+ outcome = "closed"115+ return116+ case <-paused:117+ if t.maintenance.blocked() {118+ outcome = "closed"119+ return120+ }121+ paused = t.maintenance.watch()122+ case <-overflow:123+ outcome = "busy"124+ return125+ case <-keepAlive.C:126+ if _, err := w.Write([]byte(": keep-alive\n\n")); err != nil {127+ outcome = "closed"128+ return129+ }130+ flusher.Flush()131+ case e := <-events:132+ if !t.gate.CanSee(r.Context(), e, session, nil) {133+ continue134+ }135+ raw, err := json.Marshal(e)136+ if err != nil {137+ continue138+ }139+ var frame strings.Builder140+ frame.WriteString("event: message\ndata: ")141+ frame.Write(raw)142+ frame.WriteString("\n\n")143+ if _, err := w.Write([]byte(frame.String())); err != nil {144+ outcome = "closed"145+ return146+ }147+ flusher.Flush()148+ delivered++149+ }150+ }151+}@@ -18,6 +18,7 @@ import (
1818 1919 "github.com/FelineStateMachine/tinyrelay/internal/auth"2020 "github.com/FelineStateMachine/tinyrelay/internal/blob"21+ "github.com/FelineStateMachine/tinyrelay/internal/community"2122 "github.com/FelineStateMachine/tinyrelay/internal/configport"2223 "github.com/FelineStateMachine/tinyrelay/internal/event"2324 "github.com/FelineStateMachine/tinyrelay/internal/gitrelay"@@ -475,13 +476,19 @@ func (t *Tenant) workHandlers() map[string]work.Handler {
475476 },476477 "records-projection": func(ctx context.Context, intent work.Intent) error {477478 var payload struct {479+ Action string `json:"action"`480+ Room string `json:"room"`478481 PubKey string `json:"pubkey"`479482 Role string `json:"role"`480483 Removed int `json:"removed"`484+ Deleted int `json:"deleted"`481485 }482486 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil {483487 return err484488 }489+ if payload.Action == "room" {490+ return t.projectRoom(ctx, payload.Room, payload.PubKey, payload.Removed != 0, payload.Deleted != 0)491+ }485492 var changes []records.MembershipChange486493 if payload.PubKey != "" && (payload.Role != "" || payload.Removed != 0) {487494 added := payload.Removed == 0@@ -578,6 +585,11 @@ func (t *Tenant) replacePolicy(p policy.Policy) {
578585 previous := t.policy579586 t.policy = p580587 t.mu.Unlock()588+ if community.RoomAccessFor(previous.Reads) != community.RoomAccessFor(p.Reads) {589+ if err := t.community.EnsureSlugRoom(context.Background(), t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {590+ t.app.telemetry.Logger().Error("refresh main room", "tenant", t.meta.Name, "error", err)591+ }592+ }581593 if err := t.reconcileAutomaticInbox(context.Background(), previous, p); err != nil {582594 t.app.telemetry.Logger().Error("reconcile automatic inbox job", "tenant", t.meta.Name, "error", err)583595 }@@ -95,6 +95,9 @@ func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) {
9595 if err != nil {9696 return nil, err9797 }98+ if err := t.community.EnsureSlugRoom(ctx, t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {99+ return nil, err100+ }98101 t.gate, err = gates.New(gates.Config{Store: t.store, Community: t.community, Policy: t.Policy, Slug: t.meta.Name})99102 if err != nil {100103 return nil, err@@ -227,7 +230,7 @@ func (t *Tenant) Publish(ctx context.Context, e event.Event, s relay.Session) (s
227230 }228231 var metadataNext policy.Policy229232 metadataChanged := false230- if e.Kind == event.KIND_EDIT_METADATA || e.Kind == event.KIND_PINS {233+ if (e.Kind == event.KIND_EDIT_METADATA || e.Kind == event.KIND_PINS) && t.roomScope(e) == "" {231234 role, roleErr := t.community.Role(ctx, e.PubKey)232235 if roleErr != nil {233236 return "", roleErr@@ -354,10 +357,15 @@ func (t *Tenant) Publish(ctx context.Context, e event.Event, s relay.Session) (s
354357 _, saveErr := storage.SaveTx(ctx, tx, e, opts)355358 return saveErr356359 }357- switch e.Kind {358- case event.KIND_JOIN, event.KIND_LEAVE, event.KIND_NIP43_JOIN, event.KIND_NIP43_LEAVE:360+ room := t.roomScope(e)361+ switch {362+ case room != "" && community.RoomAdminKind(e.Kind):363+ _, err = t.community.HandleRoomEventTx(ctx, e, persist)364+ case room != "" && !event.IsEphemeral(e.Kind):365+ err = t.community.HandleRoomMessageTx(ctx, e, persist)366+ case e.Kind == event.KIND_JOIN, e.Kind == event.KIND_LEAVE, e.Kind == event.KIND_NIP43_JOIN, e.Kind == event.KIND_NIP43_LEAVE:359367 _, err = t.community.HandleMembershipEventTx(ctx, e, persist)360- case event.KIND_PUT_USER, event.KIND_REMOVE_USER, event.KIND_DELETE_EVENT, event.KIND_CREATE_INVITE:368+ case e.Kind == event.KIND_PUT_USER, e.Kind == event.KIND_REMOVE_USER, e.Kind == event.KIND_DELETE_EVENT, e.Kind == event.KIND_CREATE_INVITE:361369 _, err = t.community.HandleModerationEventTx(ctx, e, persist)362370 default:363371 _, err = t.store.Save(ctx, e, opts)@@ -412,6 +420,18 @@ func (t *Tenant) agentBeforeCommit(e event.Event, now int64, before func(context
412420 }413421 }414422 423+// roomScope names the room an event is addressed to when that room is not424+// the tenant's own group, which keeps its established handling.425+func (t *Tenant) roomScope(e event.Event) string {426+ if e.Kind == event.KIND_MARMOT_GROUP {427+ return ""428+ }429+ if id := event.Tag(e, "h"); id != "" && id != t.meta.Name {430+ return id431+ }432+ return ""433+}434+415435 func (t *Tenant) vanish(ctx context.Context, e event.Event) error {416436 for _, tag := range e.Tags {417437 if len(tag) > 1 && tag[0] == "relay" && (tag[1] == "ALL_RELAYS" || strings.TrimSuffix(tag[1], "/") == strings.TrimSuffix(t.RelayURL(), "/")) {@@ -111,6 +111,28 @@ func (t *Tenant) pushNotices(ctx context.Context, e event.Event) []pushNotice {
111111 }112112 notices = append(notices, pushNotice{recipient: recipient, category: category, body: body + excerpt(e.Content), url: base + "/e/" + e.ID})113113 }114+ case event.KIND_CHAT, event.KIND_THREAD, event.KIND_THREAD_REPLY:115+ // Room messages name the room so the summary reads as a conversation.116+ roomID := event.Tag(e, "h")117+ if roomID == "" || e.Kind == event.KIND_MARMOT_GROUP {118+ return nil119+ }120+ room, err := t.community.Room(ctx, roomID)121+ if err != nil || !room.Live() {122+ return nil123+ }124+ name := room.Name125+ if name == "" {126+ name = room.ID127+ }128+ authored := t.referencedAuthors(ctx, e)129+ for _, recipient := range pushRecipients(e) {130+ category, body := pushMentions, "Mentioned you in "+name+": "131+ if authored[recipient] {132+ category, body = pushReplies, "Replied to you in "+name+": "133+ }134+ notices = append(notices, pushNotice{recipient: recipient, category: category, body: body + excerpt(e.Content), url: base + "/rooms/" + room.ID})135+ }114136 }115137 return notices116138 }@@ -199,7 +221,7 @@ func (t *Tenant) inboxUnread(ctx context.Context, pubkey string) int {
199221 since := t.inboxSeen(ctx, pubkey)200222 // Count conversation kinds only, so relay records addressed to the201223 // owner do not inflate the badge.202- filter := event.Filter{Kinds: []int{1, 4, 1111, event.KIND_WRAP, event.KIND_GIT_ISSUE, event.KIND_GIT_PR, 30023}, Tags: map[string][]string{"p": {pubkey}}}224+ filter := event.Filter{Kinds: []int{1, 4, event.KIND_CHAT, event.KIND_THREAD, event.KIND_THREAD_REPLY, 1111, event.KIND_WRAP, event.KIND_GIT_ISSUE, event.KIND_GIT_PR, 30023}, Tags: map[string][]string{"p": {pubkey}}}203225 if since > 0 {204226 filter.Since = &since205227 }@@ -95,18 +95,38 @@ func (c Config) withDefaults() Config {
9595 }9696 9797 type Router struct {98- backend Backend99- cfg Config100- mu sync.RWMutex // publish is exclusive with REQ registration + query101- clientsMu sync.Mutex102- clients map[*client]struct{}103- closed chan struct{}104- closeOnce sync.Once105- clientsWG sync.WaitGroup98+ backend Backend99+ cfg Config100+ mu sync.RWMutex // publish is exclusive with REQ registration + query101+ clientsMu sync.Mutex102+ clients map[*client]struct{}103+ listenersMu sync.Mutex104+ listeners map[*listener]struct{}105+ closed chan struct{}106+ closeOnce sync.Once107+ clientsWG sync.WaitGroup106108 }107109 110+type listener struct{ fn func(event.Event) }111+108112 func New(backend Backend, cfg Config) *Router {109- return &Router{backend: backend, cfg: cfg.withDefaults(), clients: make(map[*client]struct{}), closed: make(chan struct{})}113+ return &Router{backend: backend, cfg: cfg.withDefaults(), clients: make(map[*client]struct{}), listeners: make(map[*listener]struct{}), closed: make(chan struct{})}114+}115+116+// Listen registers an in-process observer of live fan-out. It sees every117+// event delivered to websocket subscribers, before the reader's own access118+// check. The callback runs under the publish fence and must not block; the119+// returned function removes it.120+func (r *Router) Listen(fn func(event.Event)) func() {121+ l := &listener{fn: fn}122+ r.listenersMu.Lock()123+ r.listeners[l] = struct{}{}124+ r.listenersMu.Unlock()125+ return func() {126+ r.listenersMu.Lock()127+ delete(r.listeners, l)128+ r.listenersMu.Unlock()129+ }110130 }111131 112132 func (r *Router) HandleHTTP(w http.ResponseWriter, req *http.Request) {@@ -689,6 +709,15 @@ func (r *Router) fanout(e event.Event) {
689709 }690710 }691711 }712+ r.listenersMu.Lock()713+ ls := make([]*listener, 0, len(r.listeners))714+ for l := range r.listeners {715+ ls = append(ls, l)716+ }717+ r.listenersMu.Unlock()718+ for _, l := range ls {719+ l.fn(e)720+ }692721 }693722 694723 func (c *client) canRead(e event.Event, f *event.Filter) bool {@@ -885,7 +914,7 @@ func accessEvent(e event.Event) bool {
885914 switch e.Kind {886915 case event.KIND_REPORT, event.KIND_VANISH, event.KIND_JOIN, event.KIND_LEAVE,887916 event.KIND_NIP43_JOIN, event.KIND_NIP43_LEAVE, event.KIND_PUT_USER,888- event.KIND_REMOVE_USER, event.KIND_DELETE_EVENT:917+ event.KIND_REMOVE_USER, event.KIND_DELETE_EVENT, event.KIND_DELETE_GROUP:889918 return true890919 default:891920 return false@@ -26,7 +26,7 @@ func (a *App) webMCPQuery(writer http.ResponseWriter, request *http.Request) {
2626 return2727 }2828 }29- allowed := map[string]bool{"browserepos": true, "browserepo": true, "browsefiles": true, "browsefile": true, "browsestatus": true, "browseissues": true, "browsepulls": true, "browseissue": true, "browsepull": true, "browsewiki": true, "browsewikipage": true, "browsewikimerge": true}29+ allowed := map[string]bool{"browserepos": true, "browserepo": true, "browsefiles": true, "browsefile": true, "browsestatus": true, "browseissues": true, "browsepulls": true, "browseissue": true, "browsepull": true, "browsewiki": true, "browsewikipage": true, "browsewikimerge": true, "browserooms": true, "browseroom": true, "browsethread": true}3030 if !allowed[method] {3131 writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "unsupported WebMCP query"})3232 return