tinyrelayfile
tinyrelay/internal/daemon/tenant.go/

internal/daemon/tenant.go raw
1package daemon
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "net/http"
10 "sort"
11 "strings"
12 "sync"
13 "time"
14
15 "github.com/FelineStateMachine/tinyrelay/internal/blob"
16 "github.com/FelineStateMachine/tinyrelay/internal/catalog"
17 "github.com/FelineStateMachine/tinyrelay/internal/community"
18 "github.com/FelineStateMachine/tinyrelay/internal/configport"
19 "github.com/FelineStateMachine/tinyrelay/internal/event"
20 "github.com/FelineStateMachine/tinyrelay/internal/gates"
21 "github.com/FelineStateMachine/tinyrelay/internal/mcp"
22 "github.com/FelineStateMachine/tinyrelay/internal/policy"
23 "github.com/FelineStateMachine/tinyrelay/internal/records"
24 "github.com/FelineStateMachine/tinyrelay/internal/relay"
25 "github.com/FelineStateMachine/tinyrelay/internal/replication"
26 "github.com/FelineStateMachine/tinyrelay/internal/sites"
27 "github.com/FelineStateMachine/tinyrelay/internal/storage"
28 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"
29 "github.com/FelineStateMachine/tinyrelay/internal/webpush"
30 "github.com/FelineStateMachine/tinyrelay/protocol/auth"
31 webui "github.com/FelineStateMachine/tinyrelay/tinyclient"
32 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"
33)
34
35type tenantConfig struct {
36 app *App
37 meta catalog.Tenant
38 store *storage.Store
39 policy policy.Policy
40 publicURL string
41}
42
43// Tenant composes one relay's policy, storage, protocol adapters and services.
44// App creates and owns tenants. Request handlers and workers enter the tenant's
45// maintenance gate before accessing shared services.
46type Tenant struct {
47 app *App
48 meta catalog.Tenant
49 store *storage.Store
50 community *community.Service
51 gate *gates.Gate
52 router *relay.Router
53 auth *auth.Validator
54 blobs *blob.Service
55 sites *sites.Service
56 records *records.Service
57 config *configport.ConfigStore
58 replication *replication.Service
59 git *gitrelay.GitRelay
60 ui *webui.App
61 mcp *mcp.Server
62 schedulerWake chan struct{}
63 workCtx context.Context
64 workCancel context.CancelFunc
65 workWG sync.WaitGroup
66 workErrMu sync.Mutex
67 workErr error
68 publicURL string
69 mu sync.RWMutex
70 policyWrite sync.Mutex
71 policy policy.Policy
72 maintenance maintenanceGate
73 gitLegacy *replication.LegacyCache
74 pushMu sync.Mutex
75 pushVAPID *webpush.Keys
76 pushClient *http.Client
77 pushRecent map[string]time.Time
78 callbacks *callbackService
79 customViews *customViewService
80 followups *eventFollowups
81}
82
83func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) {
84 p := cfg.policy
85 legacyPrivate, err := hasLegacyPrivateRepository(ctx, cfg.store)
86 if err != nil {
87 return nil, err
88 }
89 if legacyPrivate && (!p.Features.Grasp08 || p.Reads != "members") {
90 // Keep the private boundary durable even if an announcement is later
91 // removed while its repository or collaboration data remains.
92 p.Features.Grasp = true
93 p.Features.Grasp08 = true
94 p.Reads = "members"
95 if err := cfg.store.WithTx(ctx, func(tx *sql.Tx) error { return storage.PutSetting(ctx, tx, "policy", p) }); err != nil {
96 return nil, fmt.Errorf("preserve legacy repository privacy: %w", err)
97 }
98 }
99 t := &Tenant{app: cfg.app, meta: cfg.meta, store: cfg.store, policy: p, publicURL: cfg.publicURL, auth: auth.NewValidator(time.Now), schedulerWake: make(chan struct{}, 1), gitLegacy: &replication.LegacyCache{}}
100 t.community, err = community.New(ctx, t.store, t.policy.Owner)
101 if err != nil {
102 return nil, err
103 }
104 if err := t.community.EnsureSlugRoom(ctx, t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {
105 return nil, err
106 }
107 t.gate, err = gates.New(gates.Config{Store: t.store, Community: t.community, Policy: t.Policy, Slug: t.meta.Name, ArtifactVisible: func(ctx context.Context, e event.Event, s relay.Session) bool {
108 return t.customViews != nil && t.customViews.artifactEventVisible(ctx, e, s)
109 }})
110 if err != nil {
111 return nil, err
112 }
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 if err := t.initServices(ctx); err != nil {
115 return nil, err
116 }
117 return t, nil
118}
119
120func (t *Tenant) authenticated(ctx context.Context, s relay.Session) error {
121 ctx, done, admissionErr := t.beginOperation(ctx)
122 if admissionErr != nil {
123 return admissionErr
124 }
125 defer done()
126 for _, pubkey := range s.PubKeys {
127 if privatePolicy(t.Policy()) {
128 // Keep NIP-42 and NIP-98 at the same private-service boundary.
129 // Membership is resolved for every AUTH, so revocation applies to
130 // already connected clients when they authenticate again.
131 if err := t.requirePrivateAccess(ctx, pubkey); err != nil {
132 return err
133 }
134 }
135 banned, err := t.community.IsBanned(ctx, pubkey)
136 if err != nil {
137 return err
138 }
139 if banned {
140 return errors.New("blocked: this pubkey is banned")
141 }
142 if err := t.records.NotePresence(ctx, pubkey, time.Now().Unix()); err != nil {
143 return err
144 }
145 if pubkey == t.Policy().Owner {
146 if err := t.records.Heartbeat(ctx, pubkey, time.Now().Unix()); err != nil {
147 return err
148 }
149 }
150 }
151 return nil
152}
153
154// Policy returns the current read-only policy snapshot. Its maps and slices
155// share the snapshot's backing data; callers preserve them and submit changes
156// through the policy management path.
157func (t *Tenant) Policy() policy.Policy {
158 t.mu.RLock()
159 defer t.mu.RUnlock()
160 // Policy patches produce a fresh value; callers only inspect this snapshot.
161 return t.policy
162}
163func (t *Tenant) RelayURL() string { return strings.Replace(t.publicURL, "http", "ws", 1) }
164
165// Close stops new operations, joins background services, closes live relay
166// connections and waits for active operations before closing the store. If
167// that wait exceeds ctx, the store stays open for the remaining operations.
168func (t *Tenant) Close(ctx context.Context) error {
169 t.maintenance.markClosing()
170 servicesErr := t.closeServices(ctx)
171 routerErr := t.router.Close(ctx)
172 if err := t.maintenance.waitIdle(ctx); err != nil {
173 return errors.Join(servicesErr, routerErr, err)
174 }
175 return errors.Join(servicesErr, routerErr, t.store.Close())
176}
177
178// Publish applies write admission and the event kind's persistence or control
179// action for session s. Its string result is the relay OK message; duplicate
180// events produce a duplicate message with a nil error. Durable event writes
181// include projections and follow-up planning intents in their transaction.
182// Git metadata staging runs after that commit and before a successful reply.
183// The relay router handles live fan-out of accepted events.
184func (t *Tenant) Publish(ctx context.Context, e event.Event, s relay.Session) (string, error) {
185 ctx, done, admissionErr := t.beginOperation(ctx)
186 if admissionErr != nil {
187 return "", admissionErr
188 }
189 defer done()
190 ctx, finish := t.app.telemetry.Start(ctx, "publish")
191 outcome := "error"
192 defer func() { finish(outcome) }()
193 now := time.Now().Unix()
194 if err := t.gate.Write(ctx, e, s, now); err != nil {
195 return "", err
196 }
197 if err := t.validateStoredEvent(ctx, e); err != nil {
198 return "", err
199 }
200 if err := t.validatePushRegistration(ctx, e, s); err != nil {
201 return "", err
202 }
203 if e.Kind == event.KIND_VANISH {
204 if err := t.vanish(ctx, e); err != nil {
205 return "", err
206 }
207 outcome = "ok"
208 return "", nil
209 }
210 if e.Kind == event.KIND_REPORT {
211 reason, err := t.publishReport(ctx, e)
212 if err == nil {
213 outcome = "ok"
214 }
215 return reason, err
216 }
217 plan, metadata, err := t.prepareClientEvent(ctx, e, s, now)
218 if err != nil {
219 return "", err
220 }
221 reason, err := t.persistClientEvent(ctx, e, plan.options)
222 if errors.Is(err, storage.ErrDuplicate) {
223 outcome = "duplicate"
224 return storage.ErrDuplicate.Error(), nil
225 }
226 if err != nil {
227 return "", err
228 }
229 t.afterStoredEvent(ctx, e, plan.followups)
230 // Stage Git metadata before acknowledging the event. This closes the
231 // publish-ACK/receive-pack race: the signed pending refs and hook exist
232 // before a client can push objects for the state.
233 if plan.gitMetadata {
234 if err := t.git.CommitAfterStoreNoNotify(ctx, e, plan.repository); err != nil {
235 return "", err
236 }
237 }
238 if metadata.changed {
239 t.mu.Lock()
240 t.policy = metadata.next
241 t.mu.Unlock()
242 }
243 if err := t.records.NotePresence(ctx, e.PubKey, now); err != nil {
244 t.app.telemetry.Logger().Error("publish presence", "error", err)
245 }
246 if e.PubKey == t.Policy().Owner {
247 if err := t.records.Heartbeat(ctx, e.PubKey, now); err != nil {
248 t.app.telemetry.Logger().Error("owner heartbeat", "error", err)
249 }
250 }
251 outcome = "ok"
252 return reason, nil
253}
254
255// roomScope returns the named room for a room event. An empty result selects
256// the tenant-wide group path, including Marmot group events.
257func (t *Tenant) roomScope(e event.Event) string {
258 if e.Kind == event.KIND_MARMOT_GROUP {
259 return ""
260 }
261 if id := event.Tag(e, "h"); id != "" && id != t.meta.Name {
262 return id
263 }
264 return ""
265}
266
267func (t *Tenant) vanish(ctx context.Context, e event.Event) error {
268 for _, tag := range e.Tags {
269 if len(tag) > 1 && tag[0] == "relay" && (tag[1] == "ALL_RELAYS" || strings.TrimSuffix(tag[1], "/") == strings.TrimSuffix(t.RelayURL(), "/")) {
270 return t.store.Vanish(ctx, e.PubKey, e.CreatedAt)
271 }
272 }
273 return errors.New("invalid: vanish request does not name this relay")
274}
275
276func parsePinTags(tags [][]string) ([]string, error) {
277 seen := map[string]bool{}
278 out := []string{}
279 for _, tag := range tags {
280 if len(tag) < 2 || (tag[0] != "e" && tag[0] != "a") {
281 continue
282 }
283 ref := tag[1]
284 if tag[0] == "e" && (len(ref) != 64 || !hexLower(ref)) {
285 continue
286 }
287 if tag[0] == "a" && !strings.Contains(ref, ":") {
288 continue
289 }
290 if !seen[ref] {
291 seen[ref] = true
292 out = append(out, ref)
293 }
294 }
295 if len(out) > 20 {
296 return nil, errors.New("invalid: pin list has more than 20 entries")
297 }
298 return out, nil
299}
300
301func hexLower(value string) bool {
302 for _, c := range value {
303 if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) {
304 return false
305 }
306 }
307 return true
308}
309
310// Query returns the authorized union of the filters for session s. It applies
311// per-filter limits, removes duplicate IDs and sorts by descending creation
312// time, then ascending ID. Matching NIP-43 invitations are generated for the
313// requesting session alongside the stored results.
314func (t *Tenant) Query(ctx context.Context, filters []event.Filter, s relay.Session) ([]event.Event, error) {
315 ctx, done, admissionErr := t.beginOperation(ctx)
316 if admissionErr != nil {
317 return nil, admissionErr
318 }
319 defer done()
320 if _, err := t.gate.Read(ctx, filters, s); err != nil {
321 return nil, err
322 }
323 result := []event.Event{}
324 seen := make(map[string]bool)
325 for _, f := range filters {
326 invite, includeInvite, err := t.nip43Invite(ctx, f, s)
327 if err != nil {
328 return nil, err
329 }
330 if includeInvite && !seen[invite.ID] {
331 seen[invite.ID] = true
332 result = append(result, invite)
333 }
334 queryFilter, queryStored := withoutNIP43Invite(f)
335 if !queryStored {
336 continue
337 }
338 rows, err := t.store.Query(ctx, queryFilter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: s.PubKeys}})
339 if err != nil {
340 return nil, err
341 }
342 for _, e := range rows.Events {
343 if !seen[e.ID] && t.gate.CanSee(ctx, e, s, &f) {
344 seen[e.ID] = true
345 result = append(result, e)
346 }
347 }
348 }
349 sort.Slice(result, func(i, j int) bool {
350 if result[i].CreatedAt == result[j].CreatedAt {
351 return result[i].ID < result[j].ID
352 }
353 return result[i].CreatedAt > result[j].CreatedAt
354 })
355 return result, nil
356}
357
358// Count returns a NIP-45 result with the size of Query's authorized union,
359// using filters with their limits cleared. A supported single-filter request
360// also receives an HLL sketch. The tenant's count feature must be enabled.
361func (t *Tenant) Count(ctx context.Context, filters []event.Filter, s relay.Session) (any, error) {
362 ctx, done, admissionErr := t.beginOperation(ctx)
363 if admissionErr != nil {
364 return nil, admissionErr
365 }
366 defer done()
367 if !t.Policy().Features.Count {
368 return nil, errors.New("unsupported: COUNT is switched off on this relay")
369 }
370 // Count the same authorized union as REQ, with NIP-45 ignoring limit.
371 unlimited := make([]event.Filter, len(filters))
372 copy(unlimited, filters)
373 for i := range unlimited {
374 unlimited[i].Limit = nil
375 }
376 events, err := t.Query(ctx, unlimited, s)
377 if err != nil {
378 return nil, err
379 }
380 result := map[string]any{"count": len(events)}
381 if len(filters) == 1 {
382 if offset, ok := syncprotocol.HLLFilterOffset(filters[0]); ok {
383 hll := syncprotocol.NewHLL(offset)
384 for _, e := range events {
385 if err := hll.Add(e.PubKey); err != nil {
386 return nil, err
387 }
388 }
389 result["hll"] = hll.Hex()
390 }
391 }
392 return result, nil
393}
394
395// Sync supplies visible event IDs and timestamps for a reconciliation session.
396// It clears the filter limit and requires the tenant's sync feature.
397func (t *Tenant) Sync(ctx context.Context, f event.Filter, s relay.Session) ([]syncprotocol.Item, error) {
398 ctx, done, admissionErr := t.beginOperation(ctx)
399 if admissionErr != nil {
400 return nil, admissionErr
401 }
402 defer done()
403 if !t.Policy().Features.Sync {
404 return nil, errors.New("unsupported: sync is switched off on this relay")
405 }
406 f.Limit = nil
407 events, err := t.Query(ctx, []event.Filter{f}, s)
408 if err != nil {
409 return nil, err
410 }
411 items := make([]syncprotocol.Item, len(events))
412 for i, e := range events {
413 items[i] = syncprotocol.Item{ID: e.ID, Timestamp: e.CreatedAt}
414 }
415 return items, nil
416}
417
418// CanRead checks an event against current policy and session authority for
419// live delivery, including pending state, moderation and room visibility.
420func (t *Tenant) CanRead(e event.Event, s relay.Session) bool {
421 return t.gate.CanSee(context.Background(), e, s, nil)
422}
423
424// CanReadFilter checks live visibility with the subscription filter's context,
425// including addressed signer messages available during authentication.
426func (t *Tenant) CanReadFilter(e event.Event, s relay.Session, f *event.Filter) bool {
427 return t.gate.CanSee(context.Background(), e, s, f)
428}
429
430// QueryHints returns Query's events and relay completion hints: "finish" or
431// "more", plus "auth" when authentication can affect the query's visibility.
432func (t *Tenant) QueryHints(ctx context.Context, filters []event.Filter, s relay.Session) ([]event.Event, []string, error) {
433 ctx, done, admissionErr := t.beginOperation(ctx)
434 if admissionErr != nil {
435 return nil, nil, admissionErr
436 }
437 defer done()
438 authHint, err := t.gate.Read(ctx, filters, s)
439 if err != nil {
440 return nil, nil, err
441 }
442 events, err := t.Query(ctx, filters, s)
443 if err != nil {
444 return nil, nil, err
445 }
446 hints := []string{"finish"}
447 if authHint {
448 hints = append(hints, "auth")
449 }
450 for _, f := range filters {
451 if f.Limit == nil {
452 continue
453 }
454 rows, err := t.store.Query(ctx, f, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: s.PubKeys}})
455 if err != nil {
456 return nil, nil, err
457 }
458 if rows.More {
459 hints[0] = "more"
460 break
461 }
462 }
463 return events, hints, nil
464}
465
466func (t *Tenant) setPolicy(ctx context.Context, patch map[string]json.RawMessage) (policy.Policy, error) {
467 t.policyWrite.Lock()
468 defer t.policyWrite.Unlock()
469 next, err := policy.Patch(t.Policy(), patch)
470 if err != nil {
471 return next, err
472 }
473 return next, t.persistPolicy(ctx, next)
474}
475
476func (t *Tenant) applyPolicy(ctx context.Context, next policy.Policy) error {
477 t.policyWrite.Lock()
478 defer t.policyWrite.Unlock()
479 return t.persistPolicy(ctx, next)
480}
481
482func (t *Tenant) validatePolicyTransition(next policy.Policy) error {
483 previous := t.Policy()
484 if previous.Features.Grasp08 && previous.Reads == "members" && (!next.Features.Grasp08 || next.Reads != "members") {
485 return errors.New("blocked: opening a private GRASP-08 tenant requires an explicit data migration")
486 }
487 return nil
488}
489
490func (t *Tenant) persistPolicy(ctx context.Context, next policy.Policy) error {
491 previous := t.Policy()
492 if err := t.validatePolicyTransition(next); err != nil {
493 return err
494 }
495 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {
496 if next.Owner != previous.Owner {
497 if err := t.community.ApplyOwnerTx(ctx, tx, previous.Owner, next.Owner); err != nil {
498 return err
499 }
500 if err := storage.AddIntents(ctx, tx, []storage.Intent{{Kind: "catalog-owner", EventID: next.Owner + time.Now().Format(time.RFC3339Nano), Target: t.meta.ID, Payload: next.Owner}}, time.Now().Unix()); err != nil {
501 return err
502 }
503 }
504 return storage.PutSetting(ctx, tx, "policy", next)
505 })
506 if err != nil {
507 return err
508 }
509 t.community.SetOwner(next.Owner)
510 t.replacePolicy(next)
511 return nil
512}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs