| 1 | diff --git a/internal/daemon/http.go b/internal/daemon/http.go |
| 2 | new file mode 100644 |
| 3 | index 0000000..b73a333 |
| 4 | --- /dev/null |
| 5 | +++ b/internal/daemon/http.go |
| 6 | @@ -0,0 +1,379 @@ |
| 7 | +package daemon |
| 8 | + |
| 9 | +import ( |
| 10 | + "context" |
| 11 | + "encoding/json" |
| 12 | + "errors" |
| 13 | + "html/template" |
| 14 | + "io" |
| 15 | + "net" |
| 16 | + "net/http" |
| 17 | + "strings" |
| 18 | + "time" |
| 19 | + |
| 20 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 21 | + "github.com/FelineStateMachine/tinyrelay/internal/relay" |
| 22 | + "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 23 | +) |
| 24 | + |
| 25 | +func (t *Tenant) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 26 | + w.Header().Set("Access-Control-Allow-Origin", "*") |
| 27 | + w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, X-SHA-256, X-Content-Length, X-Content-Type") |
| 28 | + w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, DELETE, OPTIONS") |
| 29 | + w.Header().Set("X-Content-Type-Options", "nosniff") |
| 30 | + if r.Method == http.MethodOptions { |
| 31 | + w.WriteHeader(http.StatusNoContent) |
| 32 | + return |
| 33 | + } |
| 34 | + if strings.EqualFold(r.Header.Get("Upgrade"), "websocket") { |
| 35 | + t.router.HandleHTTP(w, r) |
| 36 | + return |
| 37 | + } |
| 38 | + if t.sites != nil && t.sites.MatchesHost(r.Host) { |
| 39 | + t.sites.Handler().ServeHTTP(w, r) |
| 40 | + return |
| 41 | + } |
| 42 | + if t.git != nil && (strings.HasPrefix(r.URL.Path, "/npub1") || strings.HasPrefix(r.URL.Path, "/prs/")) { |
| 43 | + t.git.ServeHTTP(w, r) |
| 44 | + return |
| 45 | + } |
| 46 | + if t.blobs != nil && (r.URL.Path == "/upload" || r.URL.Path == "/mirror" || r.URL.Path == "/report" || r.URL.Path == "/nip96" || strings.HasPrefix(r.URL.Path, "/.well-known/nostr/nip96") || strings.HasPrefix(r.URL.Path, "/list/") || len(r.URL.Path) == 65 || len(r.URL.Path) == 69) { |
| 47 | + t.blobs.Handler().ServeHTTP(w, r) |
| 48 | + return |
| 49 | + } |
| 50 | + if strings.Contains(r.Header.Get("Accept"), "application/nostr+json") { |
| 51 | + t.information(w, r) |
| 52 | + return |
| 53 | + } |
| 54 | + if r.Method == http.MethodPost && strings.Contains(r.Header.Get("Content-Type"), "application/nostr+json+rpc") { |
| 55 | + t.manageHTTP(w, r) |
| 56 | + return |
| 57 | + } |
| 58 | + if r.Method == http.MethodPost && (r.URL.Path == "/events" || r.URL.Path == "/query" || r.URL.Path == "/count") { |
| 59 | + t.bridge(w, r) |
| 60 | + return |
| 61 | + } |
| 62 | + if r.URL.Path == "/.well-known/nostr.json" { |
| 63 | + t.nip05(w, r) |
| 64 | + return |
| 65 | + } |
| 66 | + if r.URL.Path == "/healthz" { |
| 67 | + writeJSON(w, 200, map[string]any{"status": "ready"}) |
| 68 | + return |
| 69 | + } |
| 70 | + if t.ui != nil { |
| 71 | + t.ui.ServeHTTP(w, r) |
| 72 | + return |
| 73 | + } |
| 74 | + if r.URL.Path == "/" { |
| 75 | + t.home(w, r) |
| 76 | + return |
| 77 | + } |
| 78 | + http.NotFound(w, r) |
| 79 | +} |
| 80 | + |
| 81 | +func writeJSON(w http.ResponseWriter, status int, value any) { |
| 82 | + raw, err := json.Marshal(value) |
| 83 | + if err != nil { |
| 84 | + http.Error(w, "could not encode response", 500) |
| 85 | + return |
| 86 | + } |
| 87 | + w.Header().Set("Content-Type", "application/json; charset=utf-8") |
| 88 | + w.WriteHeader(status) |
| 89 | + if _, err := w.Write(raw); err != nil { |
| 90 | + return |
| 91 | + } |
| 92 | +} |
| 93 | + |
| 94 | +func (t *Tenant) requestBody(w http.ResponseWriter, r *http.Request) ([]byte, error) { |
| 95 | + var reader io.Reader = r.Body |
| 96 | + if t.app.cfg.MaxMessageBytes > 0 { |
| 97 | + reader = http.MaxBytesReader(w, r.Body, t.app.cfg.MaxMessageBytes) |
| 98 | + } |
| 99 | + body, err := io.ReadAll(reader) |
| 100 | + if err != nil { |
| 101 | + return nil, err |
| 102 | + } |
| 103 | + return body, nil |
| 104 | +} |
| 105 | + |
| 106 | +func (t *Tenant) session(r *http.Request, body []byte) (relay.Session, error) { |
| 107 | + ip, _, err := net.SplitHostPort(r.RemoteAddr) |
| 108 | + if err != nil { |
| 109 | + ip = r.RemoteAddr |
| 110 | + } |
| 111 | + s := relay.Session{RelayURL: t.RelayURL(), RemoteIP: ip} |
| 112 | + url := t.publicURL + r.URL.RequestURI() |
| 113 | + e, err := t.auth.VerifyNIP98(r.Header.Get("Authorization"), url, r.Method, string(body)) |
| 114 | + if err != nil { |
| 115 | + return s, err |
| 116 | + } |
| 117 | + s.PubKeys = []string{e.PubKey} |
| 118 | + return s, nil |
| 119 | +} |
| 120 | + |
| 121 | +func (t *Tenant) bridge(w http.ResponseWriter, r *http.Request) { |
| 122 | + body, err := t.requestBody(w, r) |
| 123 | + if err != nil { |
| 124 | + writeJSON(w, 400, map[string]string{"error": err.Error()}) |
| 125 | + return |
| 126 | + } |
| 127 | + s, err := t.session(r, body) |
| 128 | + if err != nil { |
| 129 | + writeJSON(w, 401, map[string]string{"error": err.Error()}) |
| 130 | + return |
| 131 | + } |
| 132 | + if r.URL.Path == "/events" { |
| 133 | + t.httpPublish(w, r, s, body) |
| 134 | + return |
| 135 | + } |
| 136 | + var raw []json.RawMessage |
| 137 | + if err := json.Unmarshal(body, &raw); err != nil || len(raw) == 0 { |
| 138 | + writeJSON(w, 400, map[string]string{"error": "invalid: body must be a non-empty array of filters"}) |
| 139 | + return |
| 140 | + } |
| 141 | + filters := make([]event.Filter, len(raw)) |
| 142 | + for i, b := range raw { |
| 143 | + f, err := event.ParseFilter(b) |
| 144 | + if err != nil { |
| 145 | + writeJSON(w, 400, map[string]string{"error": err.Error()}) |
| 146 | + return |
| 147 | + } |
| 148 | + filters[i] = f |
| 149 | + } |
| 150 | + var result any |
| 151 | + if r.URL.Path == "/count" { |
| 152 | + result, err = t.Count(r.Context(), filters, s) |
| 153 | + } else { |
| 154 | + result, err = t.Query(r.Context(), filters, s) |
| 155 | + } |
| 156 | + if err != nil { |
| 157 | + writeJSON(w, statusFor(err), map[string]string{"error": err.Error()}) |
| 158 | + return |
| 159 | + } |
| 160 | + writeJSON(w, 200, result) |
| 161 | +} |
| 162 | + |
| 163 | +func (t *Tenant) httpPublish(w http.ResponseWriter, r *http.Request, s relay.Session, body []byte) { |
| 164 | + e, err := event.Parse(body) |
| 165 | + if err != nil { |
| 166 | + writeJSON(w, 400, map[string]string{"error": err.Error()}) |
| 167 | + return |
| 168 | + } |
| 169 | + message, err := t.router.Publish(r.Context(), e, s) |
| 170 | + status := 200 |
| 171 | + if err != nil { |
| 172 | + message = err.Error() |
| 173 | + status = 400 |
| 174 | + } |
| 175 | + writeJSON(w, status, map[string]any{"event_id": e.ID, "accepted": err == nil, "message": message}) |
| 176 | +} |
| 177 | + |
| 178 | +func statusFor(err error) int { |
| 179 | + if strings.HasPrefix(err.Error(), "auth-required:") { |
| 180 | + return 401 |
| 181 | + } |
| 182 | + if strings.HasPrefix(err.Error(), "restricted:") || strings.HasPrefix(err.Error(), "blocked:") { |
| 183 | + return 403 |
| 184 | + } |
| 185 | + return 400 |
| 186 | +} |
| 187 | + |
| 188 | +func (t *Tenant) information(w http.ResponseWriter, r *http.Request) { |
| 189 | + p := t.Policy() |
| 190 | + nips := []int{1, 9, 11, 13, 17, 40, 42, 62, 67, 70, 86, 98} |
| 191 | + if p.Features.Count { |
| 192 | + nips = append(nips, 45) |
| 193 | + } |
| 194 | + if p.Features.Search != "off" { |
| 195 | + nips = append(nips, 50) |
| 196 | + } |
| 197 | + if p.Features.Sync { |
| 198 | + nips = append(nips, 77) |
| 199 | + } |
| 200 | + if p.Features.Names { |
| 201 | + nips = append(nips, 5) |
| 202 | + } |
| 203 | + if p.Features.Signer { |
| 204 | + nips = append(nips, 46) |
| 205 | + } |
| 206 | + name := p.Name |
| 207 | + if name == "" { |
| 208 | + name = t.meta.Name |
| 209 | + } |
| 210 | + limits := map[string]any{"max_subid_length": 64, "auth_required": p.Reads != "open", "payment_required": false, "restricted_writes": p.Writes != "open"} |
| 211 | + if t.app.cfg.MaxMessageBytes > 0 { |
| 212 | + limits["max_message_length"] = t.app.cfg.MaxMessageBytes |
| 213 | + } |
| 214 | + if p.MaxFuture > 0 { |
| 215 | + limits["created_at_upper_limit"] = p.MaxFuture |
| 216 | + } |
| 217 | + if p.MinPow > 0 { |
| 218 | + limits["min_pow_difficulty"] = p.MinPow |
| 219 | + } |
| 220 | + info := map[string]any{"name": name, "description": p.Description, "pubkey": p.Owner, "supported_nips": nips, "software": "https://github.com/FelineStateMachine/tinyrelay", "version": t.app.cfg.Version, "self_url": t.RelayURL(), "limitation": limits} |
| 221 | + for key, value := range map[string]string{"icon": p.Icon, "banner": p.Banner, "contact": p.Contact, "posting_policy": p.PostingPolicy, "privacy_policy": p.PrivacyPolicy} { |
| 222 | + if value != "" { |
| 223 | + info[key] = value |
| 224 | + } |
| 225 | + } |
| 226 | + if p.JoinTerms != "" { |
| 227 | + info["terms_of_service"] = t.publicURL + "/terms" |
| 228 | + } |
| 229 | + if len(p.Tags) > 0 { |
| 230 | + info["tags"] = p.Tags |
| 231 | + } |
| 232 | + if len(p.LanguageTags) > 0 { |
| 233 | + info["language_tags"] = p.LanguageTags |
| 234 | + } |
| 235 | + if len(p.RelayCountries) > 0 { |
| 236 | + info["relay_countries"] = p.RelayCountries |
| 237 | + } |
| 238 | + raw, err := json.Marshal(info) |
| 239 | + if err != nil { |
| 240 | + http.Error(w, err.Error(), 500) |
| 241 | + return |
| 242 | + } |
| 243 | + w.Header().Set("Content-Type", "application/nostr+json") |
| 244 | + if _, err := w.Write(raw); err != nil { |
| 245 | + return |
| 246 | + } |
| 247 | +} |
| 248 | + |
| 249 | +func (t *Tenant) nip05(w http.ResponseWriter, r *http.Request) { |
| 250 | + if !t.Policy().Features.Names { |
| 251 | + http.NotFound(w, r) |
| 252 | + return |
| 253 | + } |
| 254 | + result, err := t.community.NIP05(r.Context(), r.URL.Query().Get("name"), t.RelayURL()) |
| 255 | + if err != nil { |
| 256 | + writeJSON(w, 404, map[string]string{"error": "name not found"}) |
| 257 | + return |
| 258 | + } |
| 259 | + writeJSON(w, 200, result) |
| 260 | +} |
| 261 | + |
| 262 | +func (t *Tenant) home(w http.ResponseWriter, r *http.Request) { |
| 263 | + tmpl, err := template.New("home").Parse(`<!doctype html><html lang="en"><meta charset="utf-8"><meta name="viewport" content="width=device-width"><title>{{.Name}}</title><h1>{{.Name}}</h1><p>{{.Description}}</p><p>Relay: <code>{{.URL}}</code></p><p><a href="./people">People</a> · <a href="./manage">Manage relay</a></p></html>`) |
| 264 | + if err != nil { |
| 265 | + http.Error(w, "page unavailable", 500) |
| 266 | + return |
| 267 | + } |
| 268 | + p := t.Policy() |
| 269 | + name := p.Name |
| 270 | + if name == "" { |
| 271 | + name = t.meta.Name |
| 272 | + } |
| 273 | + w.Header().Set("Content-Type", "text/html; charset=utf-8") |
| 274 | + if err := tmpl.Execute(w, map[string]string{"Name": name, "Description": p.Description, "URL": t.RelayURL()}); err != nil { |
| 275 | + t.app.telemetry.Logger().Error("render page", "error", err) |
| 276 | + } |
| 277 | +} |
| 278 | + |
| 279 | +func (t *Tenant) manageHTTP(w http.ResponseWriter, r *http.Request) { |
| 280 | + body, err := t.requestBody(w, r) |
| 281 | + if err != nil { |
| 282 | + writeJSON(w, 400, map[string]string{"error": err.Error()}) |
| 283 | + return |
| 284 | + } |
| 285 | + var call struct { |
| 286 | + Method string `json:"method"` |
| 287 | + Params []json.RawMessage `json:"params"` |
| 288 | + } |
| 289 | + if err := json.Unmarshal(body, &call); err != nil { |
| 290 | + writeJSON(w, 400, map[string]string{"error": "invalid: malformed management request"}) |
| 291 | + return |
| 292 | + } |
| 293 | + actor := "" |
| 294 | + if call.Method != "supportedmethods" { |
| 295 | + s, err := t.session(r, body) |
| 296 | + if err != nil { |
| 297 | + writeJSON(w, 401, map[string]string{"error": err.Error()}) |
| 298 | + return |
| 299 | + } |
| 300 | + actor = s.PubKeys[0] |
| 301 | + } |
| 302 | + result, err := t.Execute(r.Context(), actor, call.Method, call.Params) |
| 303 | + if err != nil { |
| 304 | + writeJSON(w, statusFor(err), map[string]string{"error": err.Error()}) |
| 305 | + return |
| 306 | + } |
| 307 | + writeJSON(w, 200, map[string]any{"result": result}) |
| 308 | +} |
| 309 | + |
| 310 | +func (t *Tenant) Execute(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) { |
| 311 | + if method == "supportedmethods" { |
| 312 | + methods := append([]string{"supportedmethods", "getpolicy", "setpolicy", "stats", "listlisthistory", "restorelist"}, t.community.Methods()...) |
| 313 | + if t.config != nil { |
| 314 | + methods = append(methods, "exportconfig", "importconfig", "planconfig", "applypreset", "listpresets") |
| 315 | + } |
| 316 | + if t.replication != nil { |
| 317 | + methods = append(methods, t.replication.Methods()...) |
| 318 | + } |
| 319 | + return methods, nil |
| 320 | + } |
| 321 | + role, err := t.community.Role(ctx, actor) |
| 322 | + if err != nil { |
| 323 | + return nil, err |
| 324 | + } |
| 325 | + if method == "listlisthistory" { |
| 326 | + return t.store.ListHistory(ctx, actor, time.Now().Unix()) |
| 327 | + } |
| 328 | + if method == "restorelist" { |
| 329 | + if len(params) == 0 { |
| 330 | + return nil, errors.New("invalid: event id required") |
| 331 | + } |
| 332 | + var id string |
| 333 | + if err := json.Unmarshal(params[0], &id); err != nil { |
| 334 | + return nil, err |
| 335 | + } |
| 336 | + return t.store.RestoreHistory(ctx, storage.RestoreOptions{Owner: actor, EventID: id, Now: time.Now().Unix()}) |
| 337 | + } |
| 338 | + if role != "owner" && role != "moderator" { |
| 339 | + return nil, errors.New("restricted: owner or moderator required") |
| 340 | + } |
| 341 | + switch method { |
| 342 | + case "getpolicy": |
| 343 | + return t.Policy(), nil |
| 344 | + case "setpolicy": |
| 345 | + if role != "owner" { |
| 346 | + return nil, errors.New("restricted: only the owner sets policy") |
| 347 | + } |
| 348 | + if len(params) == 0 { |
| 349 | + return nil, errors.New("invalid: policy patch required") |
| 350 | + } |
| 351 | + var patch map[string]json.RawMessage |
| 352 | + if err := json.Unmarshal(params[0], &patch); err != nil { |
| 353 | + return nil, err |
| 354 | + } |
| 355 | + delete(patch, "owner") |
| 356 | + return t.setPolicy(ctx, patch) |
| 357 | + case "stats": |
| 358 | + return t.store.Stats(ctx) |
| 359 | + default: |
| 360 | + if t.config != nil && containsString([]string{"exportconfig", "importconfig", "planconfig", "applypreset", "listpresets"}, method) { |
| 361 | + return t.config.Execute(ctx, method, params) |
| 362 | + } |
| 363 | + if t.replication != nil && containsString(t.replication.Methods(), method) { |
| 364 | + values := make([]any, len(params)) |
| 365 | + for i, raw := range params { |
| 366 | + var value any |
| 367 | + if err := json.Unmarshal(raw, &value); err != nil { |
| 368 | + return nil, err |
| 369 | + } |
| 370 | + values[i] = value |
| 371 | + } |
| 372 | + return t.replication.Execute(ctx, method, values...) |
| 373 | + } |
| 374 | + return t.community.Execute(ctx, actor, method, params) |
| 375 | + } |
| 376 | +} |
| 377 | + |
| 378 | +func containsString(values []string, wanted string) bool { |
| 379 | + for _, value := range values { |
| 380 | + if value == wanted { |
| 381 | + return true |
| 382 | + } |
| 383 | + } |
| 384 | + return false |
| 385 | +} |
| 386 | diff --git a/internal/daemon/services.go b/internal/daemon/services.go |
| 387 | new file mode 100644 |
| 388 | index 0000000..cdcf712 |
| 389 | --- /dev/null |
| 390 | +++ b/internal/daemon/services.go |
| 391 | @@ -0,0 +1,159 @@ |
| 392 | +package daemon |
| 393 | + |
| 394 | +import ( |
| 395 | + "context" |
| 396 | + "encoding/json" |
| 397 | + "io" |
| 398 | + "net/http" |
| 399 | + "net/url" |
| 400 | + "strings" |
| 401 | + |
| 402 | + "github.com/FelineStateMachine/tinyrelay/internal/blob" |
| 403 | + "github.com/FelineStateMachine/tinyrelay/internal/configport" |
| 404 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 405 | + "github.com/FelineStateMachine/tinyrelay/internal/gitrelay" |
| 406 | + "github.com/FelineStateMachine/tinyrelay/internal/policy" |
| 407 | + "github.com/FelineStateMachine/tinyrelay/internal/records" |
| 408 | + "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 409 | + "github.com/FelineStateMachine/tinyrelay/internal/sites" |
| 410 | + "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 411 | + "github.com/FelineStateMachine/tinyrelay/internal/webui" |
| 412 | +) |
| 413 | + |
| 414 | +// initServices constructs the doors around a tenant's single durable store. |
| 415 | +// Each service receives the same policy snapshot function so policy changes |
| 416 | +// take effect without reopening the tenant. |
| 417 | +func (t *Tenant) initServices(ctx context.Context) error { |
| 418 | + var err error |
| 419 | + t.blobs, err = blob.New(ctx, blob.Config{ |
| 420 | + Root: t.meta.Paths.Root, Store: t.store, Authorize: t.authorizeBlob, |
| 421 | + CanRead: func(ctx context.Context, pubkey string, hashes []string) bool { |
| 422 | + p := t.Policy() |
| 423 | + if p.Reads == "open" { |
| 424 | + return true |
| 425 | + } |
| 426 | + role, err := t.community.Role(ctx, pubkey) |
| 427 | + return err == nil && role != "" |
| 428 | + }, IsOwner: func(pubkey string) bool { return pubkey == t.Policy().Owner }, |
| 429 | + }) |
| 430 | + if err != nil { |
| 431 | + return err |
| 432 | + } |
| 433 | + t.sites, err = sites.New(sites.Config{ |
| 434 | + Store: t.store, |
| 435 | + GetBlob: func(ctx context.Context, hash string) (sites.Blob, error) { |
| 436 | + entry, body, err := t.blobs.Get(ctx, hash) |
| 437 | + if err != nil { |
| 438 | + return sites.Blob{}, err |
| 439 | + } |
| 440 | + return sites.Blob{Body: body, Type: entry.Type, Size: entry.Size}, nil |
| 441 | + }, |
| 442 | + PutBlob: func(ctx context.Context, hash, typ string, body io.Reader) error { |
| 443 | + _, err := t.blobs.Put(ctx, blob.PutOptions{Reader: body, Type: typ, Uploader: t.Policy().Owner, Hash: hash}) |
| 444 | + return err |
| 445 | + }, Policy: t.Policy, BaseDomain: hostName(t.publicURL), ReadAccess: func(r *http.Request) bool { return t.siteReadAccess(r) }, |
| 446 | + }) |
| 447 | + if err != nil { |
| 448 | + return err |
| 449 | + } |
| 450 | + t.records, err = records.New(ctx, records.Config{Store: t.store, Policy: t.Policy, RelayURL: t.RelayURL(), OnGenerated: t.generatedRecord}) |
| 451 | + if err != nil { |
| 452 | + return err |
| 453 | + } |
| 454 | + t.config = configport.New(configport.ConfigStore{Store: t.store, Community: t.community, Policy: t.Policy, OnApplied: func(p policy.Policy) { t.replacePolicy(p) }}) |
| 455 | + transport := &replication.NostrTransport{Dialer: replication.WebsocketDialer{}} |
| 456 | + t.replication, err = replication.NewService(replication.Config{Store: t.store, Policy: replication.Policy{Enabled: t.Policy().Delivery.Enabled, SelfPubKey: t.Policy().Owner}, CurrentPolicy: func() replication.Policy { |
| 457 | + p := t.Policy() |
| 458 | + return replication.Policy{Enabled: p.Delivery.Enabled, SelfPubKey: p.Owner} |
| 459 | + }, Delivery: transport, Pull: transport, Push: transport, DataDir: t.meta.Paths.Root, Ingest: t.ingest}) |
| 460 | + if err != nil { |
| 461 | + return err |
| 462 | + } |
| 463 | + t.git, err = gitrelay.New(gitrelay.Config{Store: t.store, Root: t.meta.Paths.Git, Policy: t.Policy, PublicURL: t.publicURL, AuthorizeHTTP: t.authorizeGit}) |
| 464 | + if err != nil { |
| 465 | + return err |
| 466 | + } |
| 467 | + t.ui, err = webui.New(backend{tenant: t}, webui.Options{Actor: t.resolveUIActor}) |
| 468 | + if err != nil { |
| 469 | + return err |
| 470 | + } |
| 471 | + t.workCtx, t.workCancel = context.WithCancel(context.Background()) |
| 472 | + t.workWG.Add(1) |
| 473 | + go func() { defer t.workWG.Done(); _ = t.replication.Run(t.workCtx) }() |
| 474 | + return nil |
| 475 | +} |
| 476 | + |
| 477 | +// backend adapts the management method signature to webui's form-oriented API. |
| 478 | +type backend struct{ tenant *Tenant } |
| 479 | + |
| 480 | +func (b backend) Query(ctx context.Context, method string, params []json.RawMessage, actor string) (any, error) { |
| 481 | + return b.tenant.Execute(ctx, actor, method, params) |
| 482 | +} |
| 483 | +func (b backend) Policy() policy.Policy { return b.tenant.Policy() } |
| 484 | +func (b backend) URL() string { return b.tenant.publicURL } |
| 485 | +func (b backend) Slug() string { return b.tenant.meta.Name } |
| 486 | +func (b backend) Identity() string { return b.tenant.records.PublicKey() } |
| 487 | + |
| 488 | +func (t *Tenant) closeServices(ctx context.Context) error { |
| 489 | + if t.workCancel != nil { |
| 490 | + t.workCancel() |
| 491 | + } |
| 492 | + t.workWG.Wait() |
| 493 | + return nil |
| 494 | +} |
| 495 | + |
| 496 | +func hostName(raw string) string { |
| 497 | + u, err := url.Parse(raw) |
| 498 | + if err != nil { |
| 499 | + return "localhost" |
| 500 | + } |
| 501 | + if u.Hostname() == "" { |
| 502 | + return "localhost" |
| 503 | + } |
| 504 | + return u.Hostname() |
| 505 | +} |
| 506 | +func (t *Tenant) siteReadAccess(r *http.Request) bool { |
| 507 | + if t.Policy().Reads == "open" { |
| 508 | + return true |
| 509 | + } |
| 510 | + actor, err := t.resolveUIActor(r) |
| 511 | + return err == nil && actor != "" |
| 512 | +} |
| 513 | +func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, error) { |
| 514 | + e, err := t.auth.WhoAsks(r.Header.Get("Authorization"), t.publicURL+r.URL.RequestURI(), r.Method, "", string(action), strings.TrimPrefix(r.URL.Path, "/")) |
| 515 | + return first(e), err |
| 516 | +} |
| 517 | +func (t *Tenant) authorizeGit(ctx context.Context, r *http.Request, repo gitrelay.Repository) error { |
| 518 | + _, err := t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.publicURL+r.URL.RequestURI(), r.Method, "") |
| 519 | + return err |
| 520 | +} |
| 521 | +func first(v []string) string { |
| 522 | + if len(v) == 0 { |
| 523 | + return "" |
| 524 | + } |
| 525 | + return v[0] |
| 526 | +} |
| 527 | +func (t *Tenant) resolveUIActor(r *http.Request) (string, error) { |
| 528 | + e, err := t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.publicURL+r.URL.RequestURI(), r.Method, "") |
| 529 | + if err != nil { |
| 530 | + return "", err |
| 531 | + } |
| 532 | + return e.PubKey, nil |
| 533 | +} |
| 534 | +func (t *Tenant) replacePolicy(p policy.Policy) { |
| 535 | + t.mu.Lock() |
| 536 | + t.policy = p |
| 537 | + t.mu.Unlock() |
| 538 | + t.router.CloseSubscriptions("blocked: relay policy changed; subscribe again") |
| 539 | +} |
| 540 | +func (t *Tenant) generatedRecord(ctx context.Context, e event.Event) error { |
| 541 | + _, err := t.store.Save(ctx, e, storage.SaveOptions{Now: e.CreatedAt, SearchMode: t.Policy().Features.Search}) |
| 542 | + if err == nil { |
| 543 | + t.router.BroadcastGenerated(e) |
| 544 | + } |
| 545 | + return err |
| 546 | +} |
| 547 | +func (t *Tenant) ingest(ctx context.Context, e event.Event, _ replication.Origin) error { |
| 548 | + _, err := t.store.Save(ctx, e, storage.SaveOptions{Now: e.CreatedAt, SearchMode: t.Policy().Features.Search}) |
| 549 | + return err |
| 550 | +} |
| 551 | diff --git a/internal/daemon/tenant.go b/internal/daemon/tenant.go |
| 552 | new file mode 100644 |
| 553 | index 0000000..0636b1c |
| 554 | --- /dev/null |
| 555 | +++ b/internal/daemon/tenant.go |
| 556 | @@ -0,0 +1,258 @@ |
| 557 | +package daemon |
| 558 | + |
| 559 | +import ( |
| 560 | + "context" |
| 561 | + "encoding/json" |
| 562 | + "errors" |
| 563 | + "sort" |
| 564 | + "strings" |
| 565 | + "sync" |
| 566 | + "time" |
| 567 | + |
| 568 | + "github.com/FelineStateMachine/tinyrelay/internal/auth" |
| 569 | + "github.com/FelineStateMachine/tinyrelay/internal/blob" |
| 570 | + "github.com/FelineStateMachine/tinyrelay/internal/catalog" |
| 571 | + "github.com/FelineStateMachine/tinyrelay/internal/community" |
| 572 | + "github.com/FelineStateMachine/tinyrelay/internal/configport" |
| 573 | + "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 574 | + "github.com/FelineStateMachine/tinyrelay/internal/gates" |
| 575 | + "github.com/FelineStateMachine/tinyrelay/internal/gitrelay" |
| 576 | + "github.com/FelineStateMachine/tinyrelay/internal/policy" |
| 577 | + "github.com/FelineStateMachine/tinyrelay/internal/records" |
| 578 | + "github.com/FelineStateMachine/tinyrelay/internal/relay" |
| 579 | + "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 580 | + "github.com/FelineStateMachine/tinyrelay/internal/sites" |
| 581 | + "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 582 | + "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol" |
| 583 | + "github.com/FelineStateMachine/tinyrelay/internal/webui" |
| 584 | +) |
| 585 | + |
| 586 | +type tenantConfig struct { |
| 587 | + app *App |
| 588 | + meta catalog.Tenant |
| 589 | + store *storage.Store |
| 590 | + policy policy.Policy |
| 591 | + publicURL string |
| 592 | +} |
| 593 | + |
| 594 | +type Tenant struct { |
| 595 | + app *App |
| 596 | + meta catalog.Tenant |
| 597 | + store *storage.Store |
| 598 | + community *community.Service |
| 599 | + gate *gates.Gate |
| 600 | + router *relay.Router |
| 601 | + auth *auth.Validator |
| 602 | + blobs *blob.Service |
| 603 | + sites *sites.Service |
| 604 | + records *records.Service |
| 605 | + config *configport.ConfigStore |
| 606 | + replication *replication.Service |
| 607 | + git *gitrelay.GitRelay |
| 608 | + ui *webui.App |
| 609 | + workCtx context.Context |
| 610 | + workCancel context.CancelFunc |
| 611 | + workWG sync.WaitGroup |
| 612 | + publicURL string |
| 613 | + mu sync.RWMutex |
| 614 | + policy policy.Policy |
| 615 | +} |
| 616 | + |
| 617 | +func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) { |
| 618 | + t := &Tenant{app: cfg.app, meta: cfg.meta, store: cfg.store, policy: cfg.policy, publicURL: cfg.publicURL, auth: auth.NewValidator(time.Now)} |
| 619 | + var err error |
| 620 | + t.community, err = community.New(ctx, t.store, t.policy.Owner) |
| 621 | + if err != nil { |
| 622 | + return nil, err |
| 623 | + } |
| 624 | + t.gate, err = gates.New(gates.Config{Store: t.store, Community: t.community, Policy: t.Policy, Slug: t.meta.Name}) |
| 625 | + if err != nil { |
| 626 | + return nil, err |
| 627 | + } |
| 628 | + t.router = relay.New(t, relay.Config{RelayURL: t.RelayURL(), MaxMessageBytes: t.app.cfg.MaxMessageBytes, MaxPendingBytes: t.app.cfg.MaxPendingBytes, OriginPatterns: []string{"*"}}) |
| 629 | + if err := t.initServices(ctx); err != nil { |
| 630 | + return nil, err |
| 631 | + } |
| 632 | + return t, nil |
| 633 | +} |
| 634 | + |
| 635 | +func (t *Tenant) Policy() policy.Policy { |
| 636 | + t.mu.RLock() |
| 637 | + defer t.mu.RUnlock() |
| 638 | + // Policy patches produce a fresh value; callers only inspect this snapshot. |
| 639 | + return t.policy |
| 640 | +} |
| 641 | +func (t *Tenant) RelayURL() string { return strings.Replace(t.publicURL, "http", "ws", 1) } |
| 642 | +func (t *Tenant) Close(ctx context.Context) error { |
| 643 | + return errors.Join(t.closeServices(ctx), t.router.Close(ctx), t.store.Close()) |
| 644 | +} |
| 645 | + |
| 646 | +func (t *Tenant) Publish(ctx context.Context, e event.Event, s relay.Session) (string, error) { |
| 647 | + ctx, finish := t.app.telemetry.Start(ctx, "publish") |
| 648 | + outcome := "error" |
| 649 | + defer func() { finish(outcome) }() |
| 650 | + now := time.Now().Unix() |
| 651 | + if err := t.gate.Write(ctx, e, s, now); err != nil { |
| 652 | + return "", err |
| 653 | + } |
| 654 | + if e.Kind == event.KIND_VANISH { |
| 655 | + if err := t.vanish(ctx, e); err != nil { |
| 656 | + return "", err |
| 657 | + } |
| 658 | + outcome = "ok" |
| 659 | + return "", nil |
| 660 | + } |
| 661 | + opts := storage.SaveOptions{Now: now, SearchMode: t.Policy().Features.Search} |
| 662 | + if t.sites != nil { |
| 663 | + if err := sites.ValidateManifest(e); err != nil { |
| 664 | + return "", err |
| 665 | + } |
| 666 | + opts = t.sites.SaveOptions(e, now) |
| 667 | + opts.SearchMode = t.Policy().Features.Search |
| 668 | + opts.Intents = append(opts.Intents, t.replication.Prepare(e, replication.OriginClient)...) |
| 669 | + } |
| 670 | + _, err := t.store.Save(ctx, e, opts) |
| 671 | + if errors.Is(err, storage.ErrDuplicate) { |
| 672 | + outcome = "duplicate" |
| 673 | + return storage.ErrDuplicate.Error(), nil |
| 674 | + } |
| 675 | + if err != nil { |
| 676 | + return "", err |
| 677 | + } |
| 678 | + outcome = "ok" |
| 679 | + return "", nil |
| 680 | +} |
| 681 | + |
| 682 | +func (t *Tenant) vanish(ctx context.Context, e event.Event) error { |
| 683 | + for _, tag := range e.Tags { |
| 684 | + if len(tag) > 1 && tag[0] == "relay" && (tag[1] == "ALL_RELAYS" || strings.TrimSuffix(tag[1], "/") == strings.TrimSuffix(t.RelayURL(), "/")) { |
| 685 | + return t.store.Vanish(ctx, e.PubKey, e.CreatedAt) |
| 686 | + } |
| 687 | + } |
| 688 | + return errors.New("invalid: vanish request does not name this relay") |
| 689 | +} |
| 690 | + |
| 691 | +func (t *Tenant) Query(ctx context.Context, filters []event.Filter, s relay.Session) ([]event.Event, error) { |
| 692 | + if _, err := t.gate.Read(ctx, filters, s); err != nil { |
| 693 | + return nil, err |
| 694 | + } |
| 695 | + result := []event.Event{} |
| 696 | + seen := make(map[string]bool) |
| 697 | + for _, f := range filters { |
| 698 | + rows, err := t.store.Query(ctx, f, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: s.PubKeys}}) |
| 699 | + if err != nil { |
| 700 | + return nil, err |
| 701 | + } |
| 702 | + for _, e := range rows.Events { |
| 703 | + if !seen[e.ID] && t.gate.CanSee(ctx, e, s, &f) { |
| 704 | + seen[e.ID] = true |
| 705 | + result = append(result, e) |
| 706 | + } |
| 707 | + } |
| 708 | + } |
| 709 | + sort.Slice(result, func(i, j int) bool { |
| 710 | + if result[i].CreatedAt == result[j].CreatedAt { |
| 711 | + return result[i].ID < result[j].ID |
| 712 | + } |
| 713 | + return result[i].CreatedAt > result[j].CreatedAt |
| 714 | + }) |
| 715 | + return result, nil |
| 716 | +} |
| 717 | + |
| 718 | +func (t *Tenant) Count(ctx context.Context, filters []event.Filter, s relay.Session) (any, error) { |
| 719 | + if !t.Policy().Features.Count { |
| 720 | + return nil, errors.New("unsupported: COUNT is switched off on this relay") |
| 721 | + } |
| 722 | + // Count the same authorized union as REQ, with NIP-45 ignoring limit. |
| 723 | + unlimited := make([]event.Filter, len(filters)) |
| 724 | + copy(unlimited, filters) |
| 725 | + for i := range unlimited { |
| 726 | + unlimited[i].Limit = nil |
| 727 | + } |
| 728 | + events, err := t.Query(ctx, unlimited, s) |
| 729 | + if err != nil { |
| 730 | + return nil, err |
| 731 | + } |
| 732 | + result := map[string]any{"count": len(events)} |
| 733 | + if len(filters) == 1 { |
| 734 | + if offset, ok := syncprotocol.HLLFilterOffset(filters[0]); ok { |
| 735 | + hll := syncprotocol.NewHLL(offset) |
| 736 | + for _, e := range events { |
| 737 | + if err := hll.Add(e.PubKey); err != nil { |
| 738 | + return nil, err |
| 739 | + } |
| 740 | + } |
| 741 | + result["hll"] = hll.Hex() |
| 742 | + } |
| 743 | + } |
| 744 | + return result, nil |
| 745 | +} |
| 746 | + |
| 747 | +func (t *Tenant) Sync(ctx context.Context, f event.Filter, s relay.Session) ([]syncprotocol.Item, error) { |
| 748 | + if !t.Policy().Features.Sync { |
| 749 | + return nil, errors.New("unsupported: sync is switched off on this relay") |
| 750 | + } |
| 751 | + f.Limit = nil |
| 752 | + events, err := t.Query(ctx, []event.Filter{f}, s) |
| 753 | + if err != nil { |
| 754 | + return nil, err |
| 755 | + } |
| 756 | + items := make([]syncprotocol.Item, len(events)) |
| 757 | + for i, e := range events { |
| 758 | + items[i] = syncprotocol.Item{ID: e.ID, Timestamp: e.CreatedAt} |
| 759 | + } |
| 760 | + return items, nil |
| 761 | +} |
| 762 | + |
| 763 | +func (t *Tenant) CanRead(e event.Event, s relay.Session) bool { |
| 764 | + return t.gate.CanSee(context.Background(), e, s, nil) |
| 765 | +} |
| 766 | + |
| 767 | +func (t *Tenant) CanReadFilter(e event.Event, s relay.Session, f *event.Filter) bool { |
| 768 | + return t.gate.CanSee(context.Background(), e, s, f) |
| 769 | +} |
| 770 | + |
| 771 | +func (t *Tenant) QueryHints(ctx context.Context, filters []event.Filter, s relay.Session) ([]event.Event, []string, error) { |
| 772 | + authHint, err := t.gate.Read(ctx, filters, s) |
| 773 | + if err != nil { |
| 774 | + return nil, nil, err |
| 775 | + } |
| 776 | + events, err := t.Query(ctx, filters, s) |
| 777 | + if err != nil { |
| 778 | + return nil, nil, err |
| 779 | + } |
| 780 | + hints := []string{"finish"} |
| 781 | + if authHint { |
| 782 | + hints = append(hints, "auth") |
| 783 | + } |
| 784 | + for _, f := range filters { |
| 785 | + if f.Limit == nil { |
| 786 | + continue |
| 787 | + } |
| 788 | + rows, err := t.store.Query(ctx, f, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{PubKeys: s.PubKeys}}) |
| 789 | + if err != nil { |
| 790 | + return nil, nil, err |
| 791 | + } |
| 792 | + if rows.More { |
| 793 | + hints[0] = "more" |
| 794 | + break |
| 795 | + } |
| 796 | + } |
| 797 | + return events, hints, nil |
| 798 | +} |
| 799 | + |
| 800 | +func (t *Tenant) setPolicy(ctx context.Context, patch map[string]json.RawMessage) (policy.Policy, error) { |
| 801 | + t.mu.Lock() |
| 802 | + next, err := policy.Patch(t.policy, patch) |
| 803 | + if err == nil { |
| 804 | + err = t.store.PutSetting(ctx, "policy", next) |
| 805 | + } |
| 806 | + if err == nil { |
| 807 | + t.policy = next |
| 808 | + } |
| 809 | + t.mu.Unlock() |
| 810 | + if err == nil { |
| 811 | + t.router.CloseSubscriptions("blocked: relay policy changed; subscribe again") |
| 812 | + } |
| 813 | + return next, err |
| 814 | +} |