Expose embedding contracts backed by the existing connection, publication and synchronization engine. Keep tenant access-change classification in daemon and retain its transaction hooks and observability.
- commit
- eb8e94f31eb336798a436e53264af9fde2b39657
- parent
- 56c5070c50fac6258b214c8e2812d4aa84dbcaa4
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 2 | +- |
| M | internal/ | 2 | +- |
| M | internal/ | 2 | +- |
| A | internal/ | 19 | ++++ |
| M | internal/ | 8 | +- |
| M | internal/ | 36 | ++----- |
| M | internal/ | 2 | +- |
| A | tinyrelay/ | 66 | ++++++++++++++ |
| A | tinyrelay/ | 137 | ++++++++++++++++++++++++++++++ |
@@ -45,7 +45,7 @@ func (t *Tenant) ServeHTTP(w http.ResponseWriter, r *http.Request) {
4545 return4646 }4747 if strings.EqualFold(r.Header.Get("Upgrade"), "websocket") {48- t.router.HandleHTTP(w, r)48+ t.router.ServeHTTP(w, r)4949 return5050 }5151 if isRoomStreamPath(r.URL.Path) {@@ -1,8 +1,8 @@
11 package daemon22 33 import (4- "github.com/FelineStateMachine/tinyrelay/internal/relay"54 "github.com/FelineStateMachine/tinyrelay/internal/telemetry"5+ relay "github.com/FelineStateMachine/tinyrelay/tinyrelay"66 )77 88 // InstrumentRelayConfig attaches process-level connection and subscription@@ -8,8 +8,8 @@ import (
88 "strings"99 "testing"1010 11- "github.com/FelineStateMachine/tinyrelay/internal/relay"1211 "github.com/FelineStateMachine/tinyrelay/internal/telemetry"12+ relay "github.com/FelineStateMachine/tinyrelay/tinyrelay"1313 )1414 1515 func TestInstrumentRelayConfigRecordsConnectionLifecycle(t *testing.T) {@@ -0,0 +1,19 @@
1+package daemon2+3+import "github.com/FelineStateMachine/tinyrelay/internal/event"4+5+// relayAccessChange identifies tenant membership and moderation events whose6+// projections can change the visibility of existing subscriptions.7+func relayAccessChange(e event.Event) bool {8+ if e.Kind == event.KIND_NIP43_JOIN && event.Tag(e, "claim") == "" {9+ return false10+ }11+ switch e.Kind {12+ case event.KIND_REPORT, event.KIND_VANISH, event.KIND_JOIN, event.KIND_LEAVE,13+ event.KIND_NIP43_JOIN, event.KIND_NIP43_LEAVE, event.KIND_PUT_USER,14+ event.KIND_REMOVE_USER, event.KIND_DELETE_EVENT, event.KIND_DELETE_GROUP:15+ return true16+ default:17+ return false18+ }19+}@@ -30,6 +30,7 @@ import (
3030 "github.com/FelineStateMachine/tinyrelay/protocol/auth"3131 webui "github.com/FelineStateMachine/tinyrelay/tinyclient"3232 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"33+ "github.com/FelineStateMachine/tinyrelay/tinyrelay"3334 )3435 3536 type tenantConfig struct {@@ -49,7 +50,7 @@ type Tenant struct {
4950 store *storage.Store5051 community *community.Service5152 gate *gates.Gate52- router *relay.Router53+ router *tinyrelay.Relay5354 auth *auth.Validator5455 blobs *blob.Service5556 sites *sites.Service@@ -110,7 +111,10 @@ func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) {
110111 if err != nil {111112 return nil, err112113 }113- t.router = relay.New(t, InstrumentRelayConfig(relay.Config{RelayURL: t.RelayURL(), RequestRelayURL: func(r *http.Request) string { return strings.Replace(t.requestURL(r), "http", "ws", 1) }, OnAuthenticate: t.authenticated, MaxMessageBytes: t.app.cfg.MaxMessageBytes, MaxPendingBytes: t.app.cfg.MaxPendingBytes, OriginPatterns: []string{"*"}}, t.app.telemetry))114+ t.router, err = tinyrelay.New(t, InstrumentRelayConfig(tinyrelay.Config{RelayURL: t.RelayURL(), RequestRelayURL: func(r *http.Request) string { return strings.Replace(t.requestURL(r), "http", "ws", 1) }, OnAuthenticate: t.authenticated, MaxMessageBytes: t.app.cfg.MaxMessageBytes, MaxPendingBytes: t.app.cfg.MaxPendingBytes, OriginPatterns: []string{"*"}, ChangesAccess: relayAccessChange}, t.app.telemetry))115+ if err != nil {116+ return nil, err117+ }114118 if err := t.initServices(ctx); err != nil {115119 return nil, err116120 }@@ -16,17 +16,11 @@ import (
1616 1717 "github.com/FelineStateMachine/tinyrelay/internal/event"1818 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"19+ "github.com/FelineStateMachine/tinyrelay/protocol/auth"1920 "github.com/coder/websocket"2021 )2122 22-type Session struct {23- // PubKeys contains public keys established by the connection's AUTH flow.24- PubKeys []string25- // RelayURL is the relay identity used when authenticating and publishing.26- RelayURL string27- // RemoteIP is the peer address observed by the HTTP server.28- RemoteIP string29-}23+type Session = auth.Session3024 3125 // Backend publishes while the Router's write gate is held. Publish performs3226 // the backend's admission, persistence, duplicate handling or control action@@ -94,6 +88,9 @@ type Config struct {
9488 // OnConnection and OnSubscription receive +1 and -1 lifecycle deltas.9589 OnConnection func(int)9690 OnSubscription func(int)91+ // ChangesAccess identifies host-specific events that invalidate existing92+ // subscriptions. A nil value leaves subscriptions intact.93+ ChangesAccess func(event.Event) bool9794 }9895 9996 func (c Config) withDefaults() Config {@@ -184,7 +181,7 @@ func (r *Router) Publish(ctx context.Context, e event.Event, s Session) (string,
184181 defer r.mu.Unlock()185182 reason, err := r.backend.Publish(ctx, e, s)186183 if err == nil && !duplicateReason(reason) {187- if accessEvent(e) {184+ if r.changesAccess(e) {188185 r.closeSubscriptionsLocked("blocked: relay access policy changed; subscribe again")189186 }190187 r.fanout(e)@@ -201,7 +198,7 @@ func (r *Router) publishClient(c *client, e event.Event, s Session) (string, err
201198 }202199 // Queue the sender's OK before any live EVENT fanout on that socket.203200 c.replyOK(e.ID, true, reason)204- if accessEvent(e) {201+ if r.changesAccess(e) {205202 c.r.closeSubscriptionsLocked("blocked: relay access policy changed; subscribe again")206203 }207204 if reason == "" {@@ -931,21 +928,10 @@ func knownProtocolReason(reason string) bool {
931928 return false932929 }933930 934-func accessEvent(e event.Event) bool {935- if e.Kind == event.KIND_NIP43_JOIN && event.Tag(e, "claim") == "" {936- // An access request waits for review and changes nobody's access.937- return false938- }939- switch e.Kind {940- case event.KIND_REPORT, event.KIND_VANISH, event.KIND_JOIN, event.KIND_LEAVE,941- event.KIND_NIP43_JOIN, event.KIND_NIP43_LEAVE, event.KIND_PUT_USER,942- event.KIND_REMOVE_USER, event.KIND_DELETE_EVENT, event.KIND_DELETE_GROUP:943- return true944- default:945- return false946- }947-}948-949931 func duplicateReason(reason string) bool {950932 return strings.HasPrefix(reason, "duplicate:")951933 }934+935+func (r *Router) changesAccess(e event.Event) bool {936+ return r.cfg.ChangesAccess != nil && r.cfg.ChangesAccess(e)937+}@@ -146,7 +146,7 @@ func TestApplyACLChangeInvalidatesSyncUnderPublicationFence(t *testing.T) {
146146 }147147 148148 func TestAcceptedMembershipEventInvalidatesSyncBeforeFanout(t *testing.T) {149- r := New(testBackend{}, Config{})149+ r := New(testBackend{}, Config{ChangesAccess: func(e event.Event) bool { return e.Kind == event.KIND_LEAVE }})150150 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))151151 c.syncs["sync"] = &syncprotocol.Session{}152152 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{event.KIND_LEAVE}}}}@@ -0,0 +1,66 @@
1+package tinyrelay_test2+3+import (4+ "context"5+ "errors"6+ "testing"7+8+ "github.com/FelineStateMachine/tinyrelay/protocol/nostr"9+ "github.com/FelineStateMachine/tinyrelay/tinyrelay"10+)11+12+type hostBackend struct {13+ publishErr error14+ published int15+}16+17+func (b *hostBackend) Publish(context.Context, nostr.Event, tinyrelay.Session) (string, error) {18+ b.published++19+ return "", b.publishErr20+}21+22+func (*hostBackend) Query(context.Context, []nostr.Filter, tinyrelay.Session) ([]nostr.Event, error) {23+ return nil, nil24+}25+26+func (*hostBackend) Count(context.Context, []nostr.Filter, tinyrelay.Session) (any, error) {27+ return map[string]any{"count": 0}, nil28+}29+30+func TestEmbeddedRelayWaitsForHostCommit(t *testing.T) {31+ ctx := context.Background()32+ rejected := errors.New("host transaction failed")33+ backend := &hostBackend{publishErr: rejected}34+ app, err := tinyrelay.New(backend, tinyrelay.Config{RelayURL: "wss://relay.example"})35+ if err != nil {36+ t.Fatal(err)37+ }38+ t.Cleanup(func() {39+ if err := app.Close(ctx); err != nil {40+ t.Error(err)41+ }42+ })43+ delivered := 044+ stop := app.Listen(func(nostr.Event) { delivered++ })45+ defer stop()46+ e := nostr.Event{ID: "host-accepted-event", Kind: 1}47+ if _, err := app.Publish(ctx, e, tinyrelay.Session{}); !errors.Is(err, rejected) {48+ t.Fatalf("publication error: %v", err)49+ }50+ if delivered != 0 {51+ t.Fatal("failed transaction reached live delivery")52+ }53+ backend.publishErr = nil54+ if _, err := app.Publish(ctx, e, tinyrelay.Session{}); err != nil {55+ t.Fatal(err)56+ }57+ if backend.published != 2 || delivered != 1 {58+ t.Fatalf("commits=%d, deliveries=%d", backend.published, delivered)59+ }60+}61+62+func TestEmbeddedRelayRequiresBackend(t *testing.T) {63+ if _, err := tinyrelay.New(nil, tinyrelay.Config{}); err == nil {64+ t.Fatal("accepted a relay without a backend")65+ }66+}@@ -0,0 +1,137 @@
1+// Package tinyrelay provides an embeddable Nostr relay and a standalone server.2+// OpenServer owns a durable event store. New uses a host backend so an integrated3+// application can commit event projections and required work atomically before4+// publication. Both use the same authentication, subscription and synchronization5+// transport. Neither constructor starts an HTTP listener.6+package tinyrelay7+8+import (9+ "context"10+ "errors"11+ "net/http"12+ "time"13+14+ "github.com/FelineStateMachine/tinyrelay/internal/relay"15+ "github.com/FelineStateMachine/tinyrelay/protocol/auth"16+ "github.com/FelineStateMachine/tinyrelay/protocol/nostr"17+)18+19+type Session = auth.Session20+type SyncItem = nostr.EventRef21+22+// Backend owns admission, persistence and authorized query results. Publish23+// must commit required durable work before returning success. A successful24+// duplicate returns a reason beginning with "duplicate:" and is not broadcast.25+// New borrows the backend; its caller retains lifecycle ownership.26+type Backend interface {27+ Publish(context.Context, nostr.Event, Session) (string, error)28+ Query(context.Context, []nostr.Filter, Session) ([]nostr.Event, error)29+ Count(context.Context, []nostr.Filter, Session) (any, error)30+}31+32+// Reader applies current access policy to historical and live event delivery.33+type Reader interface {34+ CanRead(nostr.Event, Session) bool35+}36+37+// FilterReader additionally supplies the subscription filter to access policy.38+type FilterReader interface {39+ CanReadFilter(nostr.Event, Session, *nostr.Filter) bool40+}41+42+// QueryHints adds completion and authentication hints to a historical query.43+type QueryHints interface {44+ QueryHints(context.Context, []nostr.Filter, Session) ([]nostr.Event, []string, error)45+}46+47+// PolicyChanges releases host subscription state after access is invalidated.48+type PolicyChanges interface {49+ CloseSubscriptions(Session)50+}51+52+// SyncBackend supplies an authorized snapshot for NIP-77 reconciliation.53+type SyncBackend interface {54+ Sync(context.Context, nostr.Filter, Session) ([]SyncItem, error)55+}56+57+// Config controls connection identity, limits, authorization and observability.58+// Zero MaxMessageBytes leaves input unbounded; a negative value selects the59+// transport default. OpenServer supplies a bounded standalone default instead.60+type Config struct {61+ RelayURL string62+ RequestRelayURL func(*http.Request) string63+ OnAuthenticate func(context.Context, Session) error64+ MaxMessageBytes int6465+ MaxPendingBytes int66+ ReadLimit time.Duration67+ WriteLimit time.Duration68+ PingInterval time.Duration69+ OriginPatterns []string70+ OnConnection func(int)71+ OnSubscription func(int)72+ // ChangesAccess identifies host-specific events that invalidate active73+ // subscriptions. Standalone relays leave this unset.74+ ChangesAccess func(nostr.Event) bool75+}76+77+// Relay owns live connections and serializes publication with query snapshots.78+// Stop accepting HTTP requests before Close, which drains its connections.79+type Relay struct {80+ router *relay.Router81+}82+83+var _ http.Handler = (*Relay)(nil)84+85+// New embeds the relay protocol engine around a caller-owned backend.86+func New(backend Backend, cfg Config) (*Relay, error) {87+ if backend == nil {88+ return nil, errors.New("tinyrelay: backend is required")89+ }90+ router := relay.New(backend, relay.Config{91+ RelayURL: cfg.RelayURL, RequestRelayURL: cfg.RequestRelayURL,92+ OnAuthenticate: cfg.OnAuthenticate, MaxMessageBytes: cfg.MaxMessageBytes,93+ MaxPendingBytes: cfg.MaxPendingBytes, ReadLimit: cfg.ReadLimit,94+ WriteLimit: cfg.WriteLimit, PingInterval: cfg.PingInterval,95+ OriginPatterns: append([]string(nil), cfg.OriginPatterns...),96+ OnConnection: cfg.OnConnection, OnSubscription: cfg.OnSubscription,97+ ChangesAccess: cfg.ChangesAccess,98+ })99+ return &Relay{router: router}, nil100+}101+102+// ServeHTTP upgrades a request to the Nostr WebSocket transport.103+func (r *Relay) ServeHTTP(w http.ResponseWriter, req *http.Request) {104+ r.router.HandleHTTP(w, req)105+}106+107+// Publish commits through the backend before delivering to live subscribers.108+func (r *Relay) Publish(ctx context.Context, e nostr.Event, s Session) (string, error) {109+ return r.router.Publish(ctx, e, s)110+}111+112+// Listen observes accepted events before per-reader access checks. The callback113+// runs under the publication fence and must not block or publish recursively.114+// Call the returned function to unregister the observer.115+func (r *Relay) Listen(fn func(nostr.Event)) func() { return r.router.Listen(fn) }116+117+// BroadcastGenerated delivers an event that has already been durably accepted.118+// Use CommitGenerated when persistence must share the historical query fence.119+func (r *Relay) BroadcastGenerated(e nostr.Event) { r.router.BroadcastGenerated(e) }120+121+// CommitGenerated commits an imported/generated event under the publication122+// fence, then broadcasts it. Failed commits never reach subscribers.123+func (r *Relay) CommitGenerated(ctx context.Context, commit func(context.Context) (nostr.Event, error)) (nostr.Event, error) {124+ return r.router.CommitGenerated(ctx, commit)125+}126+127+// ApplyACLChange applies an access mutation and invalidates subscriptions while128+// holding the same fence used by publication and reconciliation snapshots.129+func (r *Relay) ApplyACLChange(ctx context.Context, change func(context.Context) (any, error), reason string) (any, error) {130+ return r.router.ApplyACLChange(ctx, change, reason)131+}132+133+// CloseSubscriptions invalidates current subscriptions after a policy revision.134+func (r *Relay) CloseSubscriptions(reason string) { r.router.CloseSubscriptions(reason) }135+136+// Close stops and joins live connections without closing the borrowed backend.137+func (r *Relay) Close(ctx context.Context) error { return r.router.Close(ctx) }