| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "net" |
| 10 | "net/http" |
| 11 | "net/url" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | |
| 15 | "github.com/FelineStateMachine/tinyrelay/internal/catalog" |
| 16 | "github.com/FelineStateMachine/tinyrelay/internal/community" |
| 17 | "github.com/FelineStateMachine/tinyrelay/internal/policy" |
| 18 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 19 | "github.com/FelineStateMachine/tinyrelay/internal/telemetry" |
| 20 | "github.com/FelineStateMachine/tinyrelay/internal/templates" |
| 21 | ) |
| 22 | |
| 23 | // Config supplies process-level paths, public addresses, transport limits and |
| 24 | // operator permissions. Tenant policy supplies each relay's access rules. |
| 25 | // DataDir is required; DefaultTenant defaults to "main". PublicURL, when set, |
| 26 | // is an absolute HTTP or HTTPS address used for externally visible URLs. |
| 27 | type Config struct { |
| 28 | DataDir string |
| 29 | PublicURL string |
| 30 | DefaultTenant string |
| 31 | Version string |
| 32 | Revision string |
| 33 | Telemetry telemetry.Config |
| 34 | MaxMessageBytes int64 |
| 35 | MaxPendingBytes int |
| 36 | AllowPrivateRelays bool |
| 37 | // ProvisionOwners controls browser-created tenants. An empty list closes |
| 38 | // public provisioning while direct operator creation remains available. |
| 39 | ProvisionOwners []string |
| 40 | DiscoveryRelays []string |
| 41 | // PushCallbackOrigins is the operator-approved HTTPS origin allowlist for |
| 42 | // host-wide push registrations. Tenant policy PushCallbacks is a separate |
| 43 | // owner-controlled allowlist; both approvals are required. |
| 44 | PushCallbackOrigins []string |
| 45 | // PeerMonitor configures NIP-66 liveness probes for its listed peers. |
| 46 | PeerMonitor *PeerMonitorConfig |
| 47 | } |
| 48 | |
| 49 | type CreateOptions = catalog.CreateOptions |
| 50 | |
| 51 | // App owns the catalog, active tenants and process-wide service lifecycle. |
| 52 | // New constructs it, Start starts serving work, and Close releases resources. |
| 53 | // An HTTP server uses App as its handler. |
| 54 | type App struct { |
| 55 | cfg Config |
| 56 | catalog *catalog.Catalog |
| 57 | telemetry *telemetry.Telemetry |
| 58 | peerMonitor *PeerMonitor |
| 59 | mu sync.Mutex |
| 60 | tenants map[string]*Tenant |
| 61 | closed bool |
| 62 | lifecycle lifecycleState |
| 63 | } |
| 64 | |
| 65 | // New opens catalog and telemetry resources and attempts to resume incomplete |
| 66 | // tenant creation. The caller owns the returned App and calls Close to release |
| 67 | // it. Start begins the serving lifecycle. |
| 68 | func New(ctx context.Context, cfg Config) (*App, error) { |
| 69 | if cfg.DataDir == "" { |
| 70 | return nil, errors.New("data directory is required" ) |
| 71 | } |
| 72 | if cfg.DefaultTenant == "" { |
| 73 | cfg.DefaultTenant = "main" |
| 74 | } |
| 75 | if cfg.PublicURL != "" { |
| 76 | u, err := url.Parse(cfg.PublicURL) |
| 77 | if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https" ) { |
| 78 | return nil, errors.New("public URL must be an absolute http or https URL" ) |
| 79 | } |
| 80 | } |
| 81 | cat, err := catalog.Open(ctx, cfg.DataDir) |
| 82 | if err != nil { |
| 83 | return nil, err |
| 84 | } |
| 85 | cfg.Telemetry.Version = cfg.Version |
| 86 | cfg.Telemetry.Revision = cfg.Revision |
| 87 | metrics, err := telemetry.New(ctx, cfg.Telemetry) |
| 88 | if err != nil { |
| 89 | return nil, errors.Join(err, cat.Close()) |
| 90 | } |
| 91 | var peerMonitor *PeerMonitor |
| 92 | if cfg.PeerMonitor != nil { |
| 93 | peerMonitor, err = NewPeerMonitor(*cfg.PeerMonitor) |
| 94 | if err != nil { |
| 95 | return nil, errors.Join(err, cat.Close(), metrics.Close(ctx)) |
| 96 | } |
| 97 | } |
| 98 | app := &App{cfg: cfg, catalog: cat, telemetry: metrics, peerMonitor: peerMonitor, tenants: make(map[string]*Tenant)} |
| 99 | if err := metrics.Register(app.QueueCollector()); err != nil { |
| 100 | return nil, errors.Join(err, cat.Close(), metrics.Close(ctx)) |
| 101 | } |
| 102 | if err := app.RecoverCreating(ctx); err != nil { |
| 103 | metrics.Logger().Error("recover unfinished tenant creation" , "error" , err) |
| 104 | } |
| 105 | return app, nil |
| 106 | } |
| 107 | |
| 108 | func (a *App) Catalog() *catalog.Catalog { return a.catalog } |
| 109 | func (a *App) Diagnostics() http.Handler { return a.telemetry.Handler() } |
| 110 | func (a *App) PeerMonitor() *PeerMonitor { return a.peerMonitor } |
| 111 | |
| 112 | // ConfigurePeerPublication binds the explicitly selected tenant to the |
| 113 | // opt-in NIP-66 publisher. The CLI operator is the authorization boundary; |
| 114 | // tenant selection is required so a private tenant is never implicit. |
| 115 | func (a *App) ConfigurePeerPublication(ctx context.Context, tenantName string) error { |
| 116 | if a.peerMonitor == nil { |
| 117 | return errors.New("daemon: peer monitor is not configured" ) |
| 118 | } |
| 119 | if strings.TrimSpace(tenantName) == "" { |
| 120 | return errors.New("daemon: peer publication tenant is required" ) |
| 121 | } |
| 122 | meta, err := a.catalog.GetByName(ctx, tenantName) |
| 123 | if err != nil { |
| 124 | return err |
| 125 | } |
| 126 | a.lifecycle.mu.Lock() |
| 127 | base, started := a.lifecycle.baseURL, a.lifecycle.started |
| 128 | a.lifecycle.mu.Unlock() |
| 129 | if !started { |
| 130 | return errors.New("daemon: start the app before configuring peer publication" ) |
| 131 | } |
| 132 | publicURL := base |
| 133 | if meta.Name != a.cfg.DefaultTenant { |
| 134 | publicURL += "/r/" + meta.Name |
| 135 | } |
| 136 | t, err := a.tenant(ctx, meta, publicURL) |
| 137 | if err != nil { |
| 138 | return err |
| 139 | } |
| 140 | a.peerMonitor.SetPublisher(t.PublishPeerStatus) |
| 141 | return nil |
| 142 | } |
| 143 | |
| 144 | // PeerMonitorRun runs configured NIP-66 probes under the caller's lifecycle. |
| 145 | // Publication remains opt-in in PeerMonitorConfig and is only a capability |
| 146 | // status until an operator supplies a signed publication path. |
| 147 | func (a *App) PeerMonitorRun(ctx context.Context) error { |
| 148 | if a.peerMonitor == nil { |
| 149 | return errors.New("daemon: peer monitor is not configured" ) |
| 150 | } |
| 151 | return a.peerMonitor.Run(ctx) |
| 152 | } |
| 153 | |
| 154 | // Create provisions a tenant from a template, initializes its durable state |
| 155 | // and marks it ready. A running App also starts its services. The returned |
| 156 | // metadata identifies the creation record even if a later setup step fails; |
| 157 | // RecoverCreating resumes incomplete creation from that durable record. |
| 158 | func (a *App) Create(ctx context.Context, opts CreateOptions) (catalog.Tenant, error) { |
| 159 | if template, ok := templates.Find(opts.Template); ok && template.Source == "required" && strings.TrimSpace(opts.Source) == "" { |
| 160 | return catalog.Tenant{}, fmt.Errorf("template %q requires a source relay" , opts.Template) |
| 161 | } |
| 162 | p, err := templates.ApplyTemplate(opts.Template, opts.Owner) |
| 163 | if err != nil { |
| 164 | return catalog.Tenant{}, err |
| 165 | } |
| 166 | meta, err := a.catalog.Create(ctx, opts) |
| 167 | if err != nil { |
| 168 | return meta, err |
| 169 | } |
| 170 | store, err := storage.Open(ctx, meta.Paths.Database) |
| 171 | if err != nil { |
| 172 | return meta, err |
| 173 | } |
| 174 | if err := initializeTenant(ctx, store, p, opts.Template); err != nil { |
| 175 | return meta, errors.Join(err, store.Close()) |
| 176 | } |
| 177 | if err := store.Close(); err != nil { |
| 178 | return meta, err |
| 179 | } |
| 180 | if err := a.AfterCreate(ctx, meta, opts); err != nil { |
| 181 | return meta, err |
| 182 | } |
| 183 | if err := a.catalog.MarkReady(ctx, meta.ID); err != nil { |
| 184 | return meta, err |
| 185 | } |
| 186 | meta.Status = catalog.StatusReady |
| 187 | if err := a.startCreatedTenant(ctx, meta); err != nil { |
| 188 | return meta, fmt.Errorf("relay created but services could not start: %w" , err) |
| 189 | } |
| 190 | return meta, nil |
| 191 | } |
| 192 | |
| 193 | func initializeTenant(ctx context.Context, s *storage.Store, p policy.Policy, name string) error { |
| 194 | if _, err := community.New(ctx, s, p.Owner); err != nil { |
| 195 | return err |
| 196 | } |
| 197 | return s.WithTx(ctx, func(tx *sql.Tx) error { |
| 198 | if err := storage.PutSetting(ctx, tx, "policy" , p); err != nil { |
| 199 | return err |
| 200 | } |
| 201 | communityConfig := community.ConfigSnapshot{} |
| 202 | for _, rule := range []struct { |
| 203 | kinds []int |
| 204 | name string |
| 205 | }{{p.AllowedKinds, "allow" }, {p.BlockedKinds, "block" }} { |
| 206 | for _, kind := range rule.kinds { |
| 207 | communityConfig.KindRules = append(communityConfig.KindRules, community.ConfigKindRule{Kind: kind, Rule: rule.name}) |
| 208 | } |
| 209 | } |
| 210 | for _, r := range p.Retention { |
| 211 | kind := r.Kind |
| 212 | communityConfig.Retention = append(communityConfig.Retention, community.ConfigRetention{Kind: &kind, Days: r.Days}) |
| 213 | } |
| 214 | if err := community.ApplyConfigTx(ctx, tx, communityConfig, false, false, false, false, true, true); err != nil { |
| 215 | return err |
| 216 | } |
| 217 | if _, err := tx.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS config_connections(position INTEGER PRIMARY KEY,value TEXT NOT NULL)" ); err != nil { |
| 218 | return err |
| 219 | } |
| 220 | template, ok := templates.Find(name) |
| 221 | if !ok { |
| 222 | return fmt.Errorf("template %q disappeared" , name) |
| 223 | } |
| 224 | connections := template.Connections |
| 225 | if connections == nil { |
| 226 | connections = []string{"notes" , "find-me" , "group" } |
| 227 | } |
| 228 | for i, name := range connections { |
| 229 | visibility := "public" |
| 230 | for _, connection := range templates.Connections() { |
| 231 | if connection.Name == name { |
| 232 | visibility = connection.Visibility |
| 233 | break |
| 234 | } |
| 235 | } |
| 236 | b, err := json.Marshal(map[string]string{"template" : name, "visibility" : visibility}) |
| 237 | if err != nil { |
| 238 | return err |
| 239 | } |
| 240 | if _, err := tx.ExecContext(ctx, "INSERT INTO config_connections(position,value) VALUES(?,?)" , i, string(b)); err != nil { |
| 241 | return err |
| 242 | } |
| 243 | } |
| 244 | return nil |
| 245 | }) |
| 246 | } |
| 247 | |
| 248 | func (a *App) tenant(ctx context.Context, meta catalog.Tenant, publicURL string) (*Tenant, error) { |
| 249 | a.mu.Lock() |
| 250 | defer a.mu.Unlock() |
| 251 | if a.closed { |
| 252 | return nil, errors.New("daemon is shutting down" ) |
| 253 | } |
| 254 | current, err := a.catalog.GetByID(ctx, meta.ID) |
| 255 | if err != nil { |
| 256 | return nil, err |
| 257 | } |
| 258 | if current.Status != catalog.StatusReady { |
| 259 | return nil, errors.New("relay is not ready" ) |
| 260 | } |
| 261 | meta = current |
| 262 | if t := a.tenants[meta.ID]; t != nil { |
| 263 | return t, nil |
| 264 | } |
| 265 | s, err := storage.Open(ctx, meta.Paths.Database) |
| 266 | if err != nil { |
| 267 | return nil, err |
| 268 | } |
| 269 | s.SetObserver(a.telemetry) |
| 270 | var p policy.Policy |
| 271 | err = s.GetSetting(ctx, "policy" , &p) |
| 272 | if errors.Is(err, sql.ErrNoRows) { |
| 273 | p, err = templates.ApplyTemplate(meta.Template, meta.Owner) |
| 274 | } |
| 275 | if err != nil { |
| 276 | return nil, errors.Join(err, s.Close()) |
| 277 | } |
| 278 | t, err := newTenant(ctx, tenantConfig{app: a, meta: meta, store: s, policy: p, publicURL: publicURL}) |
| 279 | if err != nil { |
| 280 | return nil, errors.Join(err, s.Close()) |
| 281 | } |
| 282 | a.tenants[meta.ID] = t |
| 283 | return t, nil |
| 284 | } |
| 285 | |
| 286 | // Close stops catalog reconciliation and tenant services, then closes shared |
| 287 | // resources and releases the process lock. It joins cleanup errors and returns |
| 288 | // nil on subsequent calls. The caller shuts down its HTTP servers separately. |
| 289 | func (a *App) Close(ctx context.Context) error { |
| 290 | a.mu.Lock() |
| 291 | if a.closed { |
| 292 | a.mu.Unlock() |
| 293 | return nil |
| 294 | } |
| 295 | a.closed = true |
| 296 | a.mu.Unlock() |
| 297 | lifecycleErr := a.stopLifecycle() |
| 298 | a.mu.Lock() |
| 299 | var tenants []*Tenant |
| 300 | for _, t := range a.tenants { |
| 301 | tenants = append(tenants, t) |
| 302 | } |
| 303 | a.mu.Unlock() |
| 304 | errs := []error{lifecycleErr} |
| 305 | for _, t := range tenants { |
| 306 | errs = append(errs, t.Close(ctx)) |
| 307 | } |
| 308 | errs = append(errs, a.catalog.Close(), a.telemetry.Close(ctx), a.releaseProcessLock()) |
| 309 | return errors.Join(errs...) |
| 310 | } |
| 311 | |
| 312 | // ServeHTTP resolves the request's tenant or hosted site and delegates to its |
| 313 | // handler. Health and relay discovery routes are served at the process level. |
| 314 | func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 315 | if r.URL.Path == "/relays" && a.ServeLanding(w, r) { |
| 316 | return |
| 317 | } |
| 318 | if r.URL.Path == "/healthz" || r.URL.Path == "/readyz" { |
| 319 | a.healthHTTP(w, r) |
| 320 | return |
| 321 | } |
| 322 | ctx, finish := a.telemetry.Start(r.Context(), "http" ) |
| 323 | r = r.WithContext(ctx) |
| 324 | if meta, label, isSite, siteErr := a.resolveHostedSite(r); isSite { |
| 325 | if siteErr != nil { |
| 326 | finish("error" ) |
| 327 | http.NotFound(w, r) |
| 328 | return |
| 329 | } |
| 330 | if meta.Status != catalog.StatusReady { |
| 331 | finish("unavailable" ) |
| 332 | http.Error(w, "relay is " +string(meta.Status), http.StatusServiceUnavailable) |
| 333 | return |
| 334 | } |
| 335 | t, err := a.tenant(ctx, meta, a.publicURL(r, "" )) |
| 336 | if err != nil { |
| 337 | finish("error" ) |
| 338 | http.Error(w, "relay could not be opened" , http.StatusServiceUnavailable) |
| 339 | return |
| 340 | } |
| 341 | if t.sites == nil { |
| 342 | finish("error" ) |
| 343 | http.NotFound(w, r) |
| 344 | return |
| 345 | } |
| 346 | ctx, done, err := t.beginOperation(r.Context()) |
| 347 | if err != nil { |
| 348 | finish("unavailable" ) |
| 349 | http.Error(w, err.Error(), http.StatusServiceUnavailable) |
| 350 | return |
| 351 | } |
| 352 | defer done() |
| 353 | r = r.WithContext(ctx) |
| 354 | t.sites.HandlerForLabel(label).ServeHTTP(w, r) |
| 355 | finish("ok" ) |
| 356 | return |
| 357 | } |
| 358 | meta, prefix, err := a.resolve(r) |
| 359 | if err != nil { |
| 360 | finish("error" ) |
| 361 | if errors.Is(err, catalog.ErrNotFound) && r.Method == http.MethodGet && r.URL.Path == "/" && !strings.Contains(r.Header.Get("Accept" ), "application/nostr+json" ) && !strings.EqualFold(r.Header.Get("Upgrade" ), "websocket" ) && a.ServeLanding(w, r) { |
| 362 | return |
| 363 | } |
| 364 | http.Error(w, "relay not found" , http.StatusNotFound) |
| 365 | return |
| 366 | } |
| 367 | if meta.Status != catalog.StatusReady { |
| 368 | finish("unavailable" ) |
| 369 | http.Error(w, "relay is " +string(meta.Status), http.StatusServiceUnavailable) |
| 370 | return |
| 371 | } |
| 372 | publicURL := a.publicURL(r, prefix) |
| 373 | t, err := a.tenant(ctx, meta, publicURL) |
| 374 | if err != nil { |
| 375 | finish("error" ) |
| 376 | a.telemetry.Logger().Error("open tenant" , "error" , err) |
| 377 | http.Error(w, "relay could not be opened" , http.StatusServiceUnavailable) |
| 378 | return |
| 379 | } |
| 380 | if prefix != "" { |
| 381 | r = r.Clone(ctx) |
| 382 | u := *r.URL |
| 383 | r.URL = &u |
| 384 | r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix) |
| 385 | if r.URL.Path == "" { |
| 386 | r.URL.Path = "/" |
| 387 | } |
| 388 | } |
| 389 | t.ServeHTTP(w, r) |
| 390 | finish("ok" ) |
| 391 | } |
| 392 | |
| 393 | func (a *App) resolve(r *http.Request) (catalog.Tenant, string, error) { |
| 394 | if strings.HasPrefix(r.URL.Path, "/r/" ) { |
| 395 | parts := strings.SplitN(strings.TrimPrefix(r.URL.Path, "/r/" ), "/" , 2) |
| 396 | meta, err := a.catalog.GetByName(r.Context(), parts[0]) |
| 397 | return meta, "/r/" + parts[0], err |
| 398 | } |
| 399 | host := r.Host |
| 400 | if hostname, _, err := net.SplitHostPort(host); err == nil { |
| 401 | host = hostname |
| 402 | } |
| 403 | meta, err := a.catalog.ResolveHost(r.Context(), host) |
| 404 | if err == nil { |
| 405 | return meta, "" , nil |
| 406 | } |
| 407 | if !errors.Is(err, catalog.ErrNotFound) { |
| 408 | return meta, "" , err |
| 409 | } |
| 410 | meta, err = a.catalog.GetByName(r.Context(), a.cfg.DefaultTenant) |
| 411 | return meta, "" , err |
| 412 | } |
| 413 | |
| 414 | func (a *App) publicURL(r *http.Request, prefix string) string { |
| 415 | if a.cfg.PublicURL != "" { |
| 416 | return strings.TrimSuffix(a.cfg.PublicURL, "/" ) + prefix |
| 417 | } |
| 418 | scheme := "http" |
| 419 | if r.TLS != nil { |
| 420 | scheme = "https" |
| 421 | } |
| 422 | return fmt.Sprintf("%s://%s%s" , scheme, r.Host, prefix) |
| 423 | } |