| 1 | package daemon |
| 2 | |
| 3 | // Callbacks wake agents that cannot hold a relay connection open. A key |
| 4 | // registers an https URL with a filter, and the relay POSTs each matching |
| 5 | // event the key may see to that URL as it arrives. Registration is a |
| 6 | // management method; delivery is durable work with bounded retries. This |
| 7 | // file holds storage, validation, matching and the management methods; |
| 8 | // callbacks_delivery.go holds the POST. |
| 9 | |
| 10 | import ( |
| 11 | "context" |
| 12 | "crypto/rand" |
| 13 | "database/sql" |
| 14 | "encoding/hex" |
| 15 | "encoding/json" |
| 16 | "errors" |
| 17 | "fmt" |
| 18 | "net" |
| 19 | "net/url" |
| 20 | "sort" |
| 21 | "strings" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/FelineStateMachine/tinyrelay/internal/community" |
| 25 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 26 | "github.com/FelineStateMachine/tinyrelay/internal/relay" |
| 27 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 28 | ) |
| 29 | |
| 30 | const ( |
| 31 | callbackDelivery = "callback-delivery" |
| 32 | callbackTagMax = 8 |
| 33 | callbackKindMax = 32 |
| 34 | callbackURLMax = 2048 |
| 35 | callbackSecretMin = 16 |
| 36 | callbackSecretMax = 128 |
| 37 | callbackTimeout = 10 * time.Second |
| 38 | callbackMaxAttempts = 3 |
| 39 | callbackPauseFailures = 20 |
| 40 | ) |
| 41 | |
| 42 | // callbackBackoff is the wait before the second and third attempt. |
| 43 | var callbackBackoff = []time.Duration{time.Minute, 5 * time.Minute} |
| 44 | |
| 45 | const callbackSchema = `CREATE TABLE IF NOT EXISTS callbacks(id TEXT PRIMARY KEY, owner TEXT NOT NULL, url TEXT NOT NULL, filter TEXT NOT NULL, secret TEXT NOT NULL, created_at INTEGER NOT NULL, last_delivery_at INTEGER NOT NULL DEFAULT 0, last_status TEXT NOT NULL DEFAULT '', failures INTEGER NOT NULL DEFAULT 0, paused INTEGER NOT NULL DEFAULT 0); |
| 46 | CREATE INDEX IF NOT EXISTS callbacks_owner ON callbacks(owner);` |
| 47 | |
| 48 | // callbackFilter is the NIP-01 subset a callback may use: kinds, authors and |
| 49 | // the a, e, p and h tags. since is accepted and ignored. |
| 50 | type callbackFilter struct { |
| 51 | Kinds []int `json:"kinds"` |
| 52 | Authors []string `json:"authors,omitempty"` |
| 53 | Tags map[string][]string `json:"-"` |
| 54 | } |
| 55 | |
| 56 | var callbackFilterTags = []string{"a" , "e" , "p" , "h" } |
| 57 | |
| 58 | // MarshalJSON writes the filter in NIP-01 form, with tag lists as #a, #e, |
| 59 | // #p and #h. |
| 60 | func (f callbackFilter) MarshalJSON() ([]byte, error) { |
| 61 | out := map[string]any{"kinds" : f.Kinds} |
| 62 | if len(f.Authors) > 0 { |
| 63 | out["authors" ] = f.Authors |
| 64 | } |
| 65 | for _, name := range callbackFilterTags { |
| 66 | if values := f.Tags[name]; len(values) > 0 { |
| 67 | out["#" +name] = values |
| 68 | } |
| 69 | } |
| 70 | return json.Marshal(out) |
| 71 | } |
| 72 | |
| 73 | // eventFilter is the storage filter that matches the same events. |
| 74 | func (f callbackFilter) eventFilter() event.Filter { |
| 75 | tags := map[string][]string{} |
| 76 | for name, values := range f.Tags { |
| 77 | tags[name] = values |
| 78 | } |
| 79 | return event.Filter{Kinds: f.Kinds, Authors: f.Authors, Tags: tags} |
| 80 | } |
| 81 | |
| 82 | // parseCallbackFilter validates the filter subset. Unknown keys, empty |
| 83 | // lists, malformed values and lists longer than the cap are refused so a |
| 84 | // registration that will never match is caught when it is made. |
| 85 | func parseCallbackFilter(raw json.RawMessage) (callbackFilter, error) { |
| 86 | var values map[string]json.RawMessage |
| 87 | if len(raw) == 0 || json.Unmarshal(raw, &values) != nil { |
| 88 | // A filter may arrive as a JSON string from a plain form. |
| 89 | var text string |
| 90 | if json.Unmarshal(raw, &text) != nil || json.Unmarshal([]byte(text), &values) != nil { |
| 91 | return callbackFilter{}, errors.New("invalid: filter must be a JSON object" ) |
| 92 | } |
| 93 | } |
| 94 | f := callbackFilter{Tags: map[string][]string{}} |
| 95 | for key, value := range values { |
| 96 | switch { |
| 97 | case key == "kinds" : |
| 98 | if json.Unmarshal(value, &f.Kinds) != nil || len(f.Kinds) == 0 || len(f.Kinds) > callbackKindMax { |
| 99 | return callbackFilter{}, fmt.Errorf("invalid: kinds must list 1 to %d kinds" , callbackKindMax) |
| 100 | } |
| 101 | for _, kind := range f.Kinds { |
| 102 | if kind < 0 || kind > 65535 { |
| 103 | return callbackFilter{}, errors.New("invalid: kinds must be between 0 and 65535" ) |
| 104 | } |
| 105 | } |
| 106 | case key == "authors" : |
| 107 | if json.Unmarshal(value, &f.Authors) != nil || len(f.Authors) == 0 || len(f.Authors) > callbackTagMax { |
| 108 | return callbackFilter{}, fmt.Errorf("invalid: authors must list 1 to %d public keys" , callbackTagMax) |
| 109 | } |
| 110 | for _, author := range f.Authors { |
| 111 | if !hex64(author) { |
| 112 | return callbackFilter{}, errors.New("invalid: authors must be 64 lowercase hex characters" ) |
| 113 | } |
| 114 | } |
| 115 | case key == "since" : |
| 116 | // Accepted for compatibility with clients that send whole |
| 117 | // filters; a callback only sees events that arrive after it. |
| 118 | case len(key) == 2 && key[0] == '#' && containsString(callbackFilterTags, key[1:]): |
| 119 | var list []string |
| 120 | if json.Unmarshal(value, &list) != nil || len(list) == 0 || len(list) > callbackTagMax { |
| 121 | return callbackFilter{}, fmt.Errorf("invalid: %s must list 1 to %d values" , key, callbackTagMax) |
| 122 | } |
| 123 | for _, item := range list { |
| 124 | if err := checkCallbackTagValue(key[1:], item); err != nil { |
| 125 | return callbackFilter{}, err |
| 126 | } |
| 127 | } |
| 128 | f.Tags[key[1:]] = list |
| 129 | default: |
| 130 | return callbackFilter{}, fmt.Errorf("invalid: filter key %q is not supported; use kinds, authors, #a, #e, #p or #h" , key) |
| 131 | } |
| 132 | } |
| 133 | if len(f.Kinds) == 0 { |
| 134 | return callbackFilter{}, errors.New("invalid: filter must name at least one kind" ) |
| 135 | } |
| 136 | sort.Ints(f.Kinds) |
| 137 | return f, nil |
| 138 | } |
| 139 | |
| 140 | func checkCallbackTagValue(name, value string) error { |
| 141 | switch name { |
| 142 | case "e" , "p" : |
| 143 | if !hex64(value) { |
| 144 | return fmt.Errorf("invalid: #%s values must be 64 lowercase hex characters" , name) |
| 145 | } |
| 146 | case "a" : |
| 147 | parts := strings.SplitN(value, ":" , 3) |
| 148 | if len(parts) != 3 || !hex64(parts[1]) || strings.Trim(parts[0], "0123456789" ) != "" || parts[0] == "" { |
| 149 | return errors.New("invalid: #a values must be <kind>:<pubkey>:<identifier>" ) |
| 150 | } |
| 151 | case "h" : |
| 152 | if !community.ValidRoomID(value) { |
| 153 | return errors.New("invalid: #h values must be room ids" ) |
| 154 | } |
| 155 | } |
| 156 | return nil |
| 157 | } |
| 158 | |
| 159 | // callbackRecord is one stored callback. The secret never leaves the |
| 160 | // tenant except once, in the addcallback answer. |
| 161 | type callbackRecord struct { |
| 162 | ID string `json:"id"` |
| 163 | Owner string `json:"owner"` |
| 164 | URL string `json:"url"` |
| 165 | Filter callbackFilter `json:"filter"` |
| 166 | CreatedAt int64 `json:"created"` |
| 167 | LastAt int64 `json:"lastDelivery"` |
| 168 | LastStatus string `json:"lastStatus"` |
| 169 | Failures int `json:"failures"` |
| 170 | Paused bool `json:"paused"` |
| 171 | secret string |
| 172 | } |
| 173 | |
| 174 | // summary is the management view: the host stands in for the URL so a |
| 175 | // listing shown to moderators does not expose the path. |
| 176 | func (c callbackRecord) summary() map[string]any { |
| 177 | return map[string]any{"id" : c.ID, "owner" : c.Owner, "url" : c.URL, "host" : callbackHost(c.URL), "filter" : c.Filter, "created" : c.CreatedAt, "lastDelivery" : c.LastAt, "lastStatus" : c.LastStatus, "failures" : c.Failures, "paused" : c.Paused} |
| 178 | } |
| 179 | |
| 180 | func callbackHost(raw string) string { |
| 181 | parsed, err := url.Parse(raw) |
| 182 | if err != nil { |
| 183 | return "" |
| 184 | } |
| 185 | return parsed.Host |
| 186 | } |
| 187 | |
| 188 | func (s *callbackService) initCallbacks(ctx context.Context) error { |
| 189 | _, err := s.store.DB().ExecContext(ctx, callbackSchema) |
| 190 | return err |
| 191 | } |
| 192 | |
| 193 | // validateWebhookURL accepts https URLs to public hosts. When the relay |
| 194 | // itself runs on a loopback address, as it does in tests, private targets are |
| 195 | // allowed so a local receiver can be exercised. |
| 196 | func validateWebhookURL(raw, publicURL string) error { |
| 197 | if len(raw) > callbackURLMax { |
| 198 | return errors.New("invalid: url is too long" ) |
| 199 | } |
| 200 | parsed, err := url.Parse(raw) |
| 201 | if err != nil || parsed.Scheme != "https" || parsed.Hostname() == "" || parsed.User != nil || parsed.Fragment != "" { |
| 202 | return errors.New("invalid: url must be https without credentials or a fragment" ) |
| 203 | } |
| 204 | if loopbackPublicURL(publicURL) { |
| 205 | return nil |
| 206 | } |
| 207 | host := strings.ToLower(parsed.Hostname()) |
| 208 | if ip := net.ParseIP(host); ip != nil { |
| 209 | if privateAddress(ip) { |
| 210 | return errors.New("invalid: url must not name a private or loopback address" ) |
| 211 | } |
| 212 | return nil |
| 213 | } |
| 214 | if host == "localhost" || strings.HasSuffix(host, ".localhost" ) || strings.HasSuffix(host, ".local" ) || strings.HasSuffix(host, ".internal" ) || !strings.Contains(host, "." ) { |
| 215 | return errors.New("invalid: url must name a public host" ) |
| 216 | } |
| 217 | return nil |
| 218 | } |
| 219 | |
| 220 | func privateAddress(ip net.IP) bool { |
| 221 | if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified() { |
| 222 | return true |
| 223 | } |
| 224 | _, shared, _ := net.ParseCIDR("100.64.0.0/10" ) |
| 225 | return shared.Contains(ip) |
| 226 | } |
| 227 | |
| 228 | // loopbackPublicURL reports whether the relay's own public URL is a loopback |
| 229 | // or private address, which only happens in tests and local trials. |
| 230 | func loopbackPublicURL(publicURL string) bool { |
| 231 | parsed, err := url.Parse(publicURL) |
| 232 | if err != nil { |
| 233 | return false |
| 234 | } |
| 235 | host := strings.ToLower(parsed.Hostname()) |
| 236 | if host == "localhost" { |
| 237 | return true |
| 238 | } |
| 239 | ip := net.ParseIP(host) |
| 240 | return ip != nil && privateAddress(ip) |
| 241 | } |
| 242 | |
| 243 | func callbackMethod(method string) bool { |
| 244 | return containsString([]string{"listcallbacks" , "addcallback" , "removecallback" , "pausecallback" , "resumecallback" }, method) |
| 245 | } |
| 246 | |
| 247 | // callbackExecute serves the callback management methods. Members and agents |
| 248 | // manage callbacks for their own key; the owner and moderators see and |
| 249 | // control every callback. Each call runs under the callback operation. |
| 250 | func (s *callbackService) callbackExecute(ctx context.Context, actor, method string, params []json.RawMessage) (result any, err error) { |
| 251 | ctx, finish := s.telemetry.Start(ctx, "callback" ) |
| 252 | outcome := "error" |
| 253 | defer func() { |
| 254 | if err != nil { |
| 255 | outcome = callbackOutcome(err) |
| 256 | } |
| 257 | finish(outcome) |
| 258 | }() |
| 259 | role, err := s.community.Role(ctx, actor) |
| 260 | if err != nil { |
| 261 | return nil, err |
| 262 | } |
| 263 | if role == "" { |
| 264 | return nil, fmt.Errorf("restricted: guest may not %s" , method) |
| 265 | } |
| 266 | operator := role == "owner" || role == "moderator" |
| 267 | switch method { |
| 268 | case "listcallbacks" : |
| 269 | var rows []callbackRecord |
| 270 | if operator { |
| 271 | rows, err = s.callbackRows(ctx, "" ) |
| 272 | } else { |
| 273 | rows, err = s.callbackRows(ctx, actor) |
| 274 | } |
| 275 | if err != nil { |
| 276 | return nil, err |
| 277 | } |
| 278 | out := make([]map[string]any, 0, len(rows)) |
| 279 | for _, row := range rows { |
| 280 | item := row.summary() |
| 281 | if row.Owner != actor { |
| 282 | // Operators see whose callback it is and where it goes, |
| 283 | // not the path, which may carry a token. |
| 284 | delete(item, "url" ) |
| 285 | } |
| 286 | out = append(out, item) |
| 287 | } |
| 288 | outcome = "ok" |
| 289 | return out, nil |
| 290 | case "addcallback" : |
| 291 | result, err = s.addCallback(ctx, actor, role, params) |
| 292 | default: |
| 293 | result, err = s.changeCallback(ctx, actor, operator, method, params) |
| 294 | } |
| 295 | if err == nil { |
| 296 | outcome = "ok" |
| 297 | } |
| 298 | return result, err |
| 299 | } |
| 300 | |
| 301 | func callbackOutcome(err error) string { |
| 302 | switch { |
| 303 | case strings.HasPrefix(err.Error(), "invalid:" ), strings.HasPrefix(err.Error(), "not found:" ): |
| 304 | return "invalid" |
| 305 | case strings.HasPrefix(err.Error(), "restricted:" ): |
| 306 | return "unauthorized" |
| 307 | default: |
| 308 | return "error" |
| 309 | } |
| 310 | } |
| 311 | |
| 312 | func (s *callbackService) addCallback(ctx context.Context, actor, role string, params []json.RawMessage) (any, error) { |
| 313 | if len(params) != 1 { |
| 314 | return nil, errors.New("invalid: addcallback expects one object with url, filter and an optional secret" ) |
| 315 | } |
| 316 | var options struct { |
| 317 | URL string `json:"url"` |
| 318 | Filter json.RawMessage `json:"filter"` |
| 319 | Secret string `json:"secret"` |
| 320 | } |
| 321 | if err := json.Unmarshal(params[0], &options); err != nil { |
| 322 | return nil, errors.New("invalid: addcallback expects one object with url, filter and an optional secret" ) |
| 323 | } |
| 324 | options.URL = strings.TrimSpace(options.URL) |
| 325 | if err := validateWebhookURL(options.URL, s.publicURL); err != nil { |
| 326 | return nil, err |
| 327 | } |
| 328 | filter, err := parseCallbackFilter(options.Filter) |
| 329 | if err != nil { |
| 330 | return nil, err |
| 331 | } |
| 332 | if role == "agent" { |
| 333 | if err := s.checkCallbackScope(ctx, actor, filter); err != nil { |
| 334 | return nil, err |
| 335 | } |
| 336 | } |
| 337 | generated := false |
| 338 | secret := strings.TrimSpace(options.Secret) |
| 339 | if secret == "" { |
| 340 | secret, err = randomHex(32) |
| 341 | if err != nil { |
| 342 | return nil, err |
| 343 | } |
| 344 | generated = true |
| 345 | } else if len(secret) < callbackSecretMin || len(secret) > callbackSecretMax || strings.Trim(secret, "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~" ) != "" { |
| 346 | return nil, fmt.Errorf("invalid: secret must be %d to %d printable ASCII characters" , callbackSecretMin, callbackSecretMax) |
| 347 | } |
| 348 | id, err := randomHex(16) |
| 349 | if err != nil { |
| 350 | return nil, err |
| 351 | } |
| 352 | encoded, err := json.Marshal(filter) |
| 353 | if err != nil { |
| 354 | return nil, err |
| 355 | } |
| 356 | limit := s.policy().Callbacks |
| 357 | now := time.Now().Unix() |
| 358 | err = s.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 359 | if role != "owner" { |
| 360 | var count int |
| 361 | if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM callbacks WHERE owner=?` , actor).Scan(&count); err != nil { |
| 362 | return err |
| 363 | } |
| 364 | if count >= limit { |
| 365 | return fmt.Errorf("invalid: at most %d callbacks per key" , limit) |
| 366 | } |
| 367 | } |
| 368 | _, err := tx.ExecContext(ctx, `INSERT INTO callbacks(id,owner,url,filter,secret,created_at) VALUES(?,?,?,?,?,?)` , id, actor, options.URL, string(encoded), secret, now) |
| 369 | return err |
| 370 | }) |
| 371 | if err != nil { |
| 372 | return nil, err |
| 373 | } |
| 374 | s.invalidateCallbacks() |
| 375 | if err := s.community.Record(ctx, actor, "addcallback" , id, "" ); err != nil { |
| 376 | return nil, err |
| 377 | } |
| 378 | // Counts only: the host, path and secret stay out of the log. |
| 379 | s.telemetry.Logger().Info("callback registered" , "tenant" , s.tenant.Name, "kinds" , len(filter.Kinds), "generated" , generated) |
| 380 | record := callbackRecord{ID: id, Owner: actor, URL: options.URL, Filter: filter, CreatedAt: now} |
| 381 | out := record.summary() |
| 382 | out["secret" ] = secret |
| 383 | return out, nil |
| 384 | } |
| 385 | |
| 386 | // checkCallbackScope keeps an agent's callback inside its grant: a room it |
| 387 | // names must be one it may post in and a repository it names must be one |
| 388 | // the grant covers. Kinds are not restricted, so an agent can listen for |
| 389 | // answers it may not publish itself. |
| 390 | func (s *callbackService) checkCallbackScope(ctx context.Context, actor string, filter callbackFilter) error { |
| 391 | grant, ok, err := s.community.AgentGrant(ctx, actor) |
| 392 | if err != nil { |
| 393 | return err |
| 394 | } |
| 395 | if !ok || !grant.Active(time.Now().Unix()) { |
| 396 | return errors.New("restricted: agent grant is not active" ) |
| 397 | } |
| 398 | for _, room := range filter.Tags["h" ] { |
| 399 | if !grant.AllowsRoom(room) { |
| 400 | return fmt.Errorf("restricted: agent grant does not cover room %s" , room) |
| 401 | } |
| 402 | } |
| 403 | for _, coordinate := range filter.Tags["a" ] { |
| 404 | owner, identifier, ok := repoCoordinate(coordinate) |
| 405 | if !ok { |
| 406 | continue |
| 407 | } |
| 408 | if grant.RepoLevel(owner, identifier) == "" { |
| 409 | return fmt.Errorf("restricted: agent grant does not cover repository %s" , identifier) |
| 410 | } |
| 411 | } |
| 412 | return nil |
| 413 | } |
| 414 | |
| 415 | func (s *callbackService) changeCallback(ctx context.Context, actor string, operator bool, method string, params []json.RawMessage) (any, error) { |
| 416 | id := callbackIDParam(params) |
| 417 | if id == "" { |
| 418 | return nil, errors.New("invalid: callback id required" ) |
| 419 | } |
| 420 | record, err := s.callback(ctx, id) |
| 421 | if err != nil { |
| 422 | return nil, err |
| 423 | } |
| 424 | if !operator && record.Owner != actor { |
| 425 | return nil, fmt.Errorf("restricted: only the callback's owner, the relay owner or a moderator may %s" , method) |
| 426 | } |
| 427 | switch method { |
| 428 | case "removecallback" : |
| 429 | _, err = s.store.DB().ExecContext(ctx, `DELETE FROM callbacks WHERE id=?` , id) |
| 430 | case "pausecallback" : |
| 431 | _, err = s.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=1, last_status='paused' WHERE id=?` , id) |
| 432 | record.Paused, record.LastStatus = true, "paused" |
| 433 | case "resumecallback" : |
| 434 | _, err = s.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=0, failures=0, last_status='' WHERE id=?` , id) |
| 435 | record.Paused, record.Failures, record.LastStatus = false, 0, "" |
| 436 | default: |
| 437 | return nil, fmt.Errorf("unsupported: management method %q" , method) |
| 438 | } |
| 439 | if err != nil { |
| 440 | return nil, err |
| 441 | } |
| 442 | s.invalidateCallbacks() |
| 443 | if err := s.community.Record(ctx, actor, method, id, "" ); err != nil { |
| 444 | return nil, err |
| 445 | } |
| 446 | if method == "removecallback" { |
| 447 | return map[string]any{"id" : id, "removed" : true}, nil |
| 448 | } |
| 449 | result := record.summary() |
| 450 | if operator && record.Owner != actor { |
| 451 | delete(result, "url" ) |
| 452 | } |
| 453 | return result, nil |
| 454 | } |
| 455 | |
| 456 | // callbackIDParam reads the id from a bare string or from {"id": ...}. |
| 457 | func callbackIDParam(params []json.RawMessage) string { |
| 458 | if len(params) == 0 { |
| 459 | return "" |
| 460 | } |
| 461 | if id := stringParam(params, 0); id != "" { |
| 462 | return id |
| 463 | } |
| 464 | var options struct { |
| 465 | ID string `json:"id"` |
| 466 | } |
| 467 | _ = json.Unmarshal(params[0], &options) |
| 468 | return strings.TrimSpace(options.ID) |
| 469 | } |
| 470 | |
| 471 | func (s *callbackService) callback(ctx context.Context, id string) (callbackRecord, error) { |
| 472 | rows, err := s.callbackRows(ctx, "" , id) |
| 473 | if err != nil { |
| 474 | return callbackRecord{}, err |
| 475 | } |
| 476 | if len(rows) == 0 { |
| 477 | return callbackRecord{}, errors.New("not found: callback" ) |
| 478 | } |
| 479 | return rows[0], nil |
| 480 | } |
| 481 | |
| 482 | // callbackRows lists callbacks, narrowed to an owner or to one id. |
| 483 | func (s *callbackService) callbackRows(ctx context.Context, owner string, ids ...string) ([]callbackRecord, error) { |
| 484 | query := `SELECT id,owner,url,filter,secret,created_at,last_delivery_at,last_status,failures,paused FROM callbacks` |
| 485 | args := []any{} |
| 486 | switch { |
| 487 | case len(ids) > 0: |
| 488 | query += ` WHERE id=?` |
| 489 | args = append(args, ids[0]) |
| 490 | case owner != "" : |
| 491 | query += ` WHERE owner=?` |
| 492 | args = append(args, owner) |
| 493 | } |
| 494 | query += ` ORDER BY created_at,id` |
| 495 | rows, err := s.store.DB().QueryContext(ctx, query, args...) |
| 496 | if err != nil { |
| 497 | return nil, fmt.Errorf("list callbacks: %w" , err) |
| 498 | } |
| 499 | defer rows.Close() |
| 500 | var out []callbackRecord |
| 501 | for rows.Next() { |
| 502 | var item callbackRecord |
| 503 | var filter string |
| 504 | var paused int |
| 505 | if err := rows.Scan(&item.ID, &item.Owner, &item.URL, &filter, &item.secret, &item.CreatedAt, &item.LastAt, &item.LastStatus, &item.Failures, &paused); err != nil { |
| 506 | return nil, err |
| 507 | } |
| 508 | item.Paused = paused != 0 |
| 509 | if item.Filter, err = parseCallbackFilter(json.RawMessage(filter)); err != nil { |
| 510 | return nil, fmt.Errorf("callback filter: %w" , err) |
| 511 | } |
| 512 | out = append(out, item) |
| 513 | } |
| 514 | return out, rows.Err() |
| 515 | } |
| 516 | |
| 517 | // callbackRules returns the active callbacks indexed by kind. The index is |
| 518 | // built from the table on first use and dropped whenever a callback |
| 519 | // changes, so matching an event costs one map lookup. |
| 520 | func (s *callbackService) callbackRules(ctx context.Context) map[int][]callbackRecord { |
| 521 | s.callbackMu.Lock() |
| 522 | defer s.callbackMu.Unlock() |
| 523 | if s.callbackIndex != nil { |
| 524 | return s.callbackIndex |
| 525 | } |
| 526 | index := map[int][]callbackRecord{} |
| 527 | rows, err := s.callbackRows(ctx, "" ) |
| 528 | if err != nil { |
| 529 | s.telemetry.Logger().Error("load callbacks" , "tenant" , s.tenant.Name, "error" , err) |
| 530 | return index |
| 531 | } |
| 532 | for _, row := range rows { |
| 533 | if row.Paused { |
| 534 | continue |
| 535 | } |
| 536 | for _, kind := range row.Filter.Kinds { |
| 537 | index[kind] = append(index[kind], row) |
| 538 | } |
| 539 | } |
| 540 | s.callbackIndex = index |
| 541 | return index |
| 542 | } |
| 543 | |
| 544 | func (s *callbackService) invalidateCallbacks() { |
| 545 | s.callbackMu.Lock() |
| 546 | s.callbackIndex = nil |
| 547 | s.callbackMu.Unlock() |
| 548 | } |
| 549 | |
| 550 | // CandidateIDs captures the active callback registrations that match an |
| 551 | // event. The caller must persist these IDs with the event so later planning |
| 552 | // cannot include callbacks registered after the event was accepted. |
| 553 | func (s *callbackService) CandidateIDs(ctx context.Context, e event.Event) ([]string, error) { |
| 554 | rows, err := s.callbackRows(ctx, "" ) |
| 555 | if err != nil { |
| 556 | return nil, err |
| 557 | } |
| 558 | ids := make([]string, 0, len(rows)) |
| 559 | for _, candidate := range rows { |
| 560 | if candidate.Paused || !event.Matches(candidate.Filter.eventFilter(), e) { |
| 561 | continue |
| 562 | } |
| 563 | ids = append(ids, candidate.ID) |
| 564 | } |
| 565 | return ids, nil |
| 566 | } |
| 567 | |
| 568 | // PrepareFor creates delivery intents for captured registration IDs. Each |
| 569 | // registration is re-read so deletion, pausing, membership changes and |
| 570 | // visibility changes take effect before an intent is created. |
| 571 | func (s *callbackService) PrepareFor(ctx context.Context, e event.Event, ids []string) ([]storage.Intent, error) { |
| 572 | if len(ids) == 0 { |
| 573 | return nil, nil |
| 574 | } |
| 575 | intents := make([]storage.Intent, 0, len(ids)) |
| 576 | seen := make(map[string]struct{}, len(ids)) |
| 577 | for _, id := range ids { |
| 578 | if id == "" { |
| 579 | continue |
| 580 | } |
| 581 | if _, ok := seen[id]; ok { |
| 582 | continue |
| 583 | } |
| 584 | seen[id] = struct{}{} |
| 585 | candidate, err := s.callback(ctx, id) |
| 586 | if err != nil { |
| 587 | if strings.HasPrefix(err.Error(), "not found:" ) { |
| 588 | continue |
| 589 | } |
| 590 | return nil, err |
| 591 | } |
| 592 | if candidate.Paused || !event.Matches(candidate.Filter.eventFilter(), e) { |
| 593 | continue |
| 594 | } |
| 595 | if !s.callbackOwnerCanSee(ctx, candidate.Owner, e) { |
| 596 | continue |
| 597 | } |
| 598 | payload, err := json.Marshal(callbackPayload{Event: e, Attempt: 1}) |
| 599 | if err != nil { |
| 600 | return nil, fmt.Errorf("encode callback delivery: %w" , err) |
| 601 | } |
| 602 | intents = append(intents, storage.Intent{Kind: callbackDelivery, EventID: e.ID, Target: candidate.ID, Payload: string(payload)}) |
| 603 | } |
| 604 | return intents, nil |
| 605 | } |
| 606 | |
| 607 | func (s *callbackService) callbackOwnerCanSee(ctx context.Context, owner string, e event.Event) bool { |
| 608 | return s.gate.CanSee(ctx, e, relay.Session{PubKeys: []string{owner}, RelayURL: s.relayURL}, nil) |
| 609 | } |
| 610 | |
| 611 | // Prepare matches current callback registrations and rechecks their access |
| 612 | // before returning delivery intents. The caller persists the returned work. |
| 613 | func (s *callbackService) Prepare(ctx context.Context, e event.Event) ([]storage.Intent, error) { |
| 614 | ids, err := s.CandidateIDs(ctx, e) |
| 615 | if err != nil { |
| 616 | return nil, err |
| 617 | } |
| 618 | return s.PrepareFor(ctx, e, ids) |
| 619 | } |
| 620 | |
| 621 | // notifyCallbacks plans and queues delivery in its own transaction, logging |
| 622 | // planning or storage failures. Ephemeral publication uses this path because |
| 623 | // its event is carried by the delivery payload. |
| 624 | func (s *callbackService) notifyCallbacks(ctx context.Context, e event.Event) { |
| 625 | intents, err := s.Prepare(ctx, e) |
| 626 | if err != nil { |
| 627 | s.telemetry.Logger().Error("callback deliveries not prepared" , "tenant" , s.tenant.Name, "error" , err) |
| 628 | return |
| 629 | } |
| 630 | if len(intents) == 0 { |
| 631 | return |
| 632 | } |
| 633 | err = s.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 634 | return storage.AddIntents(ctx, tx, intents, time.Now().Unix()) |
| 635 | }) |
| 636 | if err != nil { |
| 637 | s.telemetry.Logger().Error("callback deliveries not queued" , "tenant" , s.tenant.Name, "error" , err) |
| 638 | return |
| 639 | } |
| 640 | s.telemetry.Logger().Debug("callback deliveries queued" , "tenant" , s.tenant.Name, "queued" , len(intents)) |
| 641 | } |
| 642 | |
| 643 | func randomHex(size int) (string, error) { |
| 644 | buffer := make([]byte, size) |
| 645 | if _, err := rand.Read(buffer); err != nil { |
| 646 | return "" , fmt.Errorf("callback secret: %w" , err) |
| 647 | } |
| 648 | return hex.EncodeToString(buffer), nil |
| 649 | } |
| 650 | |
| 651 | // Tenant wrappers keep management and work dispatch independent of the |
| 652 | // callback service's storage and delivery state. |
| 653 | func (t *Tenant) initCallbacks(ctx context.Context) error { |
| 654 | return t.callbacks.initCallbacks(ctx) |
| 655 | } |
| 656 | |
| 657 | func (t *Tenant) callbackExecute(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) { |
| 658 | return t.callbacks.callbackExecute(ctx, actor, method, params) |
| 659 | } |
| 660 | |
| 661 | func (t *Tenant) notifyCallbacks(ctx context.Context, e event.Event) { |
| 662 | t.callbacks.notifyCallbacks(ctx, e) |
| 663 | } |