1package daemon23import (4 "context"5 "database/sql"6 "encoding/json"7 "errors"8 "fmt"9 "net/http"10 "sort"11 "strings"12 "sync"13 "time"1415 "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 "github.com/FelineStateMachine/tinyrelay/tinyrelay"34)3536type tenantConfig struct {37 app *App38 meta catalog.Tenant39 store *storage.Store40 policy policy.Policy41 publicURL string42}4344// Tenant composes one relay's policy, storage, protocol adapters and services.45// App creates and owns tenants. Request handlers and workers enter the tenant's46// maintenance gate before accessing shared services.47type Tenant struct {48 app *App49 meta catalog.Tenant50 store *storage.Store51 community *community.Service52 gate *gates.Gate53 router *tinyrelay.Relay54 auth *auth.Validator55 blobs *blob.Service56 sites *sites.Service57 records *records.Service58 config *configport.ConfigStore59 replication *replication.Service60 git *gitrelay.GitRelay61 ui *webui.App62 mcp *mcp.Server63 schedulerWake chan struct{}64 workCtx context.Context65 workCancel context.CancelFunc66 workWG sync.WaitGroup67 workErrMu sync.Mutex68 workErr error69 publicURL string70 mu sync.RWMutex71 policyWrite sync.Mutex72 policy policy.Policy73 maintenance maintenanceGate74 gitLegacy *replication.LegacyCache75 pushMu sync.Mutex76 pushVAPID *webpush.Keys77 pushClient *http.Client78 pushRecent map[string]time.Time79 callbacks *callbackService80 customViews *customViewService81 followups *eventFollowups82}8384func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) {85 p := cfg.policy86 legacyPrivate, err := hasLegacyPrivateRepository(ctx, cfg.store)87 if err != nil {88 return nil, err89 }90 if legacyPrivate && (!p.Features.Grasp08 || p.Reads != "members") {91 // Keep the private boundary durable even if an announcement is later92 // removed while its repository or collaboration data remains.93 p.Features.Grasp = true94 p.Features.Grasp08 = true95 p.Reads = "members"96 if err := cfg.store.WithTx(ctx, func(tx *sql.Tx) error { return storage.PutSetting(ctx, tx, "policy", p) }); err != nil {97 return nil, fmt.Errorf("preserve legacy repository privacy: %w", err)98 }99 }100 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{}}101 t.community, err = community.New(ctx, t.store, t.policy.Owner)102 if err != nil {103 return nil, err104 }105 if err := t.community.EnsureSlugRoom(ctx, t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {106 return nil, err107 }108 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 {109 return t.customViews != nil && t.customViews.artifactEventVisible(ctx, e, s)110 }})111 if err != nil {112 return nil, err113 }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 }118 if err := t.initServices(ctx); err != nil {119 return nil, err120 }121 return t, nil122}123124func (t *Tenant) authenticated(ctx context.Context, s relay.Session) error {125 ctx, done, admissionErr := t.beginOperation(ctx)126 if admissionErr != nil {127 return admissionErr128 }129 defer done()130 for _, pubkey := range s.PubKeys {131 if privatePolicy(t.Policy()) {132 // Keep NIP-42 and NIP-98 at the same private-service boundary.133 // Membership is resolved for every AUTH, so revocation applies to134 // already connected clients when they authenticate again.135 if err := t.requirePrivateAccess(ctx, pubkey); err != nil {136 return err137 }138 }139 banned, err := t.community.IsBanned(ctx, pubkey)140 if err != nil {141 return err142 }143 if banned {144 return errors.New("blocked: this pubkey is banned")145 }146 if err := t.records.NotePresence(ctx, pubkey, time.Now().Unix()); err != nil {147 return err148 }149 if pubkey == t.Policy().Owner {150 if err := t.records.Heartbeat(ctx, pubkey, time.Now().Unix()); err != nil {151 return err152 }153 }154 }155 return nil156}157158// Policy returns the current read-only policy snapshot. Its maps and slices159// share the snapshot's backing data; callers preserve them and submit changes160// through the policy management path.161func (t *Tenant) Policy() policy.Policy {162 t.mu.RLock()163 defer t.mu.RUnlock()164 // Policy patches produce a fresh value; callers only inspect this snapshot.165 return t.policy166}167func (t *Tenant) RelayURL() string { return strings.Replace(t.publicURL, "http", "ws", 1) }168169// Close stops new operations, joins background services, closes live relay170// connections and waits for active operations before closing the store. If171// that wait exceeds ctx, the store stays open for the remaining operations.172func (t *Tenant) Close(ctx context.Context) error {173 t.maintenance.markClosing()174 servicesErr := t.closeServices(ctx)175 routerErr := t.router.Close(ctx)176 if err := t.maintenance.waitIdle(ctx); err != nil {177 return errors.Join(servicesErr, routerErr, err)178 }179 return errors.Join(servicesErr, routerErr, t.store.Close())180}181182// Publish applies write admission and the event kind's persistence or control183// action for session s. Its string result is the relay OK message; duplicate184// events produce a duplicate message with a nil error. Durable event writes185// include projections and follow-up planning intents in their transaction.186// Git metadata staging runs after that commit and before a successful reply.187// The relay router handles live fan-out of accepted events.188func (t *Tenant) Publish(ctx context.Context, e event.Event, s relay.Session) (string, error) {189 ctx, done, admissionErr := t.beginOperation(ctx)190 if admissionErr != nil {191 return "", admissionErr192 }193 defer done()194 ctx, finish := t.app.telemetry.Start(ctx, "publish")195 outcome := "error"196 defer func() { finish(outcome) }()197 now := time.Now().Unix()198 if err := t.gate.Write(ctx, e, s, now); err != nil {199 return "", err200 }201 if err := t.validateStoredEvent(ctx, e); err != nil {202 return "", err203 }204 if err := t.validatePushRegistration(ctx, e, s); err != nil {205 return "", err206 }207 if e.Kind == event.KIND_VANISH {208 if err := t.vanish(ctx, e); err != nil {209 return "", err210 }211 outcome = "ok"212 return "", nil213 }214 if e.Kind == event.KIND_REPORT {215 reason, err := t.publishReport(ctx, e)216 if err == nil {217 outcome = "ok"218 }219 return reason, err220 }221 plan, metadata, err := t.prepareClientEvent(ctx, e, s, now)222 if err != nil {223 return "", err224 }225 reason, err := t.persistClientEvent(ctx, e, plan.options)226 if errors.Is(err, storage.ErrDuplicate) {227 outcome = "duplicate"228 return storage.ErrDuplicate.Error(), nil229 }230 if err != nil {231 return "", err232 }233 t.afterStoredEvent(ctx, e, plan.followups)234 // Stage Git metadata before acknowledging the event. This closes the235 // publish-ACK/receive-pack race: the signed pending refs and hook exist236 // before a client can push objects for the state.237 if plan.gitMetadata {238 if err := t.git.CommitAfterStoreNoNotify(ctx, e, plan.repository); err != nil {239 return "", err240 }241 }242 if metadata.changed {243 t.mu.Lock()244 t.policy = metadata.next245 t.mu.Unlock()246 }247 if err := t.records.NotePresence(ctx, e.PubKey, now); err != nil {248 t.app.telemetry.Logger().Error("publish presence", "error", err)249 }250 if e.PubKey == t.Policy().Owner {251 if err := t.records.Heartbeat(ctx, e.PubKey, now); err != nil {252 t.app.telemetry.Logger().Error("owner heartbeat", "error", err)253 }254 }255 outcome = "ok"256 return reason, nil257}258259// roomScope returns the named room for a room event. An empty result selects260// the tenant-wide group path, including Marmot group events.261func (t *Tenant) roomScope(e event.Event) string {262 if e.Kind == event.KIND_MARMOT_GROUP {263 return ""264 }265 if id := event.Tag(e, "h"); id != "" && id != t.meta.Name {266 return id267 }268 return ""269}270271func (t *Tenant) vanish(ctx context.Context, e event.Event) error {272 for _, tag := range e.Tags {273 if len(tag) > 1 && tag[0] == "relay" && (tag[1] == "ALL_RELAYS" || strings.TrimSuffix(tag[1], "/") == strings.TrimSuffix(t.RelayURL(), "/")) {274 return t.store.Vanish(ctx, e.PubKey, e.CreatedAt)275 }276 }277 return errors.New("invalid: vanish request does not name this relay")278}279280func parsePinTags(tags [][]string) ([]string, error) {281 seen := map[string]bool{}282 out := []string{}283 for _, tag := range tags {284 if len(tag) < 2 || (tag[0] != "e" && tag[0] != "a") {285 continue286 }287 ref := tag[1]288 if tag[0] == "e" && (len(ref) != 64 || !hexLower(ref)) {289 continue290 }291 if tag[0] == "a" && !strings.Contains(ref, ":") {292 continue293 }294 if !seen[ref] {295 seen[ref] = true296 out = append(out, ref)297 }298 }299 if len(out) > 20 {300 return nil, errors.New("invalid: pin list has more than 20 entries")301 }302 return out, nil303}304305func hexLower(value string) bool {306 for _, c := range value {307 if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) {308 return false309 }310 }311 return true312}313314// Query returns the authorized union of the filters for session s. It applies315// per-filter limits, removes duplicate IDs and sorts by descending creation316// time, then ascending ID. Matching NIP-43 invitations are generated for the317// requesting session alongside the stored results.318func (t *Tenant) Query(ctx context.Context, filters []event.Filter, s relay.Session) ([]event.Event, error) {319 ctx, done, admissionErr := t.beginOperation(ctx)320 if admissionErr != nil {321 return nil, admissionErr322 }323 defer done()324 if _, err := t.gate.Read(ctx, filters, s); err != nil {325 return nil, err326 }327 result := []event.Event{}328 seen := make(map[string]bool)329 for _, f := range filters {330 invite, includeInvite, err := t.nip43Invite(ctx, f, s)331 if err != nil {332 return nil, err333 }334 if includeInvite && !seen[invite.ID] {335 seen[invite.ID] = true336 result = append(result, invite)337 }338 queryFilter, queryStored := withoutNIP43Invite(f)339 if !queryStored {340 continue341 }342 rows, err := t.store.Query(ctx, queryFilter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: s.PubKeys}})343 if err != nil {344 return nil, err345 }346 for _, e := range rows.Events {347 if !seen[e.ID] && t.gate.CanSee(ctx, e, s, &f) {348 seen[e.ID] = true349 result = append(result, e)350 }351 }352 }353 sort.Slice(result, func(i, j int) bool {354 if result[i].CreatedAt == result[j].CreatedAt {355 return result[i].ID < result[j].ID356 }357 return result[i].CreatedAt > result[j].CreatedAt358 })359 return result, nil360}361362// Count returns a NIP-45 result with the size of Query's authorized union,363// using filters with their limits cleared. A supported single-filter request364// also receives an HLL sketch. The tenant's count feature must be enabled.365func (t *Tenant) Count(ctx context.Context, filters []event.Filter, s relay.Session) (any, error) {366 ctx, done, admissionErr := t.beginOperation(ctx)367 if admissionErr != nil {368 return nil, admissionErr369 }370 defer done()371 if !t.Policy().Features.Count {372 return nil, errors.New("unsupported: COUNT is switched off on this relay")373 }374 // Count the same authorized union as REQ, with NIP-45 ignoring limit.375 unlimited := make([]event.Filter, len(filters))376 copy(unlimited, filters)377 for i := range unlimited {378 unlimited[i].Limit = nil379 }380 events, err := t.Query(ctx, unlimited, s)381 if err != nil {382 return nil, err383 }384 result := map[string]any{"count": len(events)}385 if len(filters) == 1 {386 if offset, ok := syncprotocol.HLLFilterOffset(filters[0]); ok {387 hll := syncprotocol.NewHLL(offset)388 for _, e := range events {389 if err := hll.Add(e.PubKey); err != nil {390 return nil, err391 }392 }393 result["hll"] = hll.Hex()394 }395 }396 return result, nil397}398399// Sync supplies visible event IDs and timestamps for a reconciliation session.400// It clears the filter limit and requires the tenant's sync feature.401func (t *Tenant) Sync(ctx context.Context, f event.Filter, s relay.Session) ([]syncprotocol.Item, error) {402 ctx, done, admissionErr := t.beginOperation(ctx)403 if admissionErr != nil {404 return nil, admissionErr405 }406 defer done()407 if !t.Policy().Features.Sync {408 return nil, errors.New("unsupported: sync is switched off on this relay")409 }410 f.Limit = nil411 events, err := t.Query(ctx, []event.Filter{f}, s)412 if err != nil {413 return nil, err414 }415 items := make([]syncprotocol.Item, len(events))416 for i, e := range events {417 items[i] = syncprotocol.Item{ID: e.ID, Timestamp: e.CreatedAt}418 }419 return items, nil420}421422// CanRead checks an event against current policy and session authority for423// live delivery, including pending state, moderation and room visibility.424func (t *Tenant) CanRead(e event.Event, s relay.Session) bool {425 return t.gate.CanSee(context.Background(), e, s, nil)426}427428// CanReadFilter checks live visibility with the subscription filter's context,429// including addressed signer messages available during authentication.430func (t *Tenant) CanReadFilter(e event.Event, s relay.Session, f *event.Filter) bool {431 return t.gate.CanSee(context.Background(), e, s, f)432}433434// QueryHints returns Query's events and relay completion hints: "finish" or435// "more", plus "auth" when authentication can affect the query's visibility.436func (t *Tenant) QueryHints(ctx context.Context, filters []event.Filter, s relay.Session) ([]event.Event, []string, error) {437 ctx, done, admissionErr := t.beginOperation(ctx)438 if admissionErr != nil {439 return nil, nil, admissionErr440 }441 defer done()442 authHint, err := t.gate.Read(ctx, filters, s)443 if err != nil {444 return nil, nil, err445 }446 events, err := t.Query(ctx, filters, s)447 if err != nil {448 return nil, nil, err449 }450 hints := []string{"finish"}451 if authHint {452 hints = append(hints, "auth")453 }454 for _, f := range filters {455 if f.Limit == nil {456 continue457 }458 rows, err := t.store.Query(ctx, f, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: s.PubKeys}})459 if err != nil {460 return nil, nil, err461 }462 if rows.More {463 hints[0] = "more"464 break465 }466 }467 return events, hints, nil468}469470func (t *Tenant) setPolicy(ctx context.Context, patch map[string]json.RawMessage) (policy.Policy, error) {471 t.policyWrite.Lock()472 defer t.policyWrite.Unlock()473 next, err := policy.Patch(t.Policy(), patch)474 if err != nil {475 return next, err476 }477 return next, t.persistPolicy(ctx, next)478}479480func (t *Tenant) applyPolicy(ctx context.Context, next policy.Policy) error {481 t.policyWrite.Lock()482 defer t.policyWrite.Unlock()483 return t.persistPolicy(ctx, next)484}485486func (t *Tenant) validatePolicyTransition(next policy.Policy) error {487 previous := t.Policy()488 if previous.Features.Grasp08 && previous.Reads == "members" && (!next.Features.Grasp08 || next.Reads != "members") {489 return errors.New("blocked: opening a private GRASP-08 tenant requires an explicit data migration")490 }491 return nil492}493494func (t *Tenant) persistPolicy(ctx context.Context, next policy.Policy) error {495 previous := t.Policy()496 if err := t.validatePolicyTransition(next); err != nil {497 return err498 }499 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {500 if next.Owner != previous.Owner {501 if err := t.community.ApplyOwnerTx(ctx, tx, previous.Owner, next.Owner); err != nil {502 return err503 }504 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 {505 return err506 }507 }508 return storage.PutSetting(ctx, tx, "policy", next)509 })510 if err != nil {511 return err512 }513 t.community.SetOwner(next.Owner)514 t.replacePolicy(next)515 return nil516}