- commit
- cef34c70332a419493c111ed56539339936e907e
- parent
- fb3940d0224778bbab72bdf259ce1caca311fcb4
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 6 | + |
| A | internal/ | 587 | ++++++++++++++++++++++++++ |
| A | internal/ | 205 | +++++++++ |
| A | internal/ | 679 | ++++++++++++++++++++++++++++++ |
| M | internal/ | 8 | +- |
| M | internal/ | 2 | +- |
| M | internal/ | 12 | +- |
| M | internal/ | 67 | ++- |
| M | internal/ | 21 | +- |
| M | internal/ | 6 | +- |
@@ -682,6 +682,12 @@ func (s *Service) audit(ctx context.Context) (any, error) {
682682 }683683 return out, rows.Err()684684 }685+686+// Record writes one audit row for an action another service carried out.687+func (s *Service) Record(ctx context.Context, actor, action, target, detail string) error {688+ return s.store.WithTx(ctx, func(tx *sql.Tx) error { return s.recordTx(ctx, tx, actor, action, target, detail) })689+}690+685691 func (s *Service) recordTx(ctx context.Context, tx *sql.Tx, actor, action, target, detail string) error {686692 _, err := tx.ExecContext(ctx, `INSERT INTO community_audit(at,actor,action,target,detail) VALUES(?,?,?,?,?)`, now(), actor, action, target, detail)687693 return err@@ -0,0 +1,587 @@
1+package daemon2+3+// Callbacks wake agents that cannot hold a relay connection open. A key4+// registers an https URL with a filter, and the relay POSTs each matching5+// event the key may see to that URL as it arrives. Registration is a6+// management method; delivery is durable work with bounded retries. This7+// 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 = 833+ callbackKindMax = 3234+ callbackURLMax = 204835+ callbackSecretMin = 1636+ callbackSecretMax = 12837+ callbackTimeout = 10 * time.Second38+ callbackMaxAttempts = 339+ callbackPauseFailures = 2040+)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 and49+// 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.Authors64+ }65+ for _, name := range callbackFilterTags {66+ if values := f.Tags[name]; len(values) > 0 {67+ out["#"+name] = values68+ }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] = values78+ }79+ return event.Filter{Kinds: f.Kinds, Authors: f.Authors, Tags: tags}80+}81+82+// parseCallbackFilter validates the filter subset. Unknown keys, empty83+// lists, malformed values and lists longer than the cap are refused so a84+// 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.RawMessage87+ 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 string90+ 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 whole117+ // filters; a callback only sees events that arrive after it.118+ case len(key) == 2 && key[0] == '#' && containsString(callbackFilterTags, key[1:]):119+ var list []string120+ 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{}, err126+ }127+ }128+ f.Tags[key[1:]] = list129+ 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, nil138+}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 nil157+}158+159+// callbackRecord is one stored callback. The secret never leaves the160+// 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 string172+}173+174+// summary is the management view: the host stands in for the URL so a175+// 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.Host186+}187+188+func (t *Tenant) initCallbacks(ctx context.Context) error {189+ _, err := t.store.DB().ExecContext(ctx, callbackSchema)190+ return err191+}192+193+// checkCallbackURL accepts https URLs to public hosts. When the relay itself194+// runs on a loopback address, as it does in tests, private targets are195+// allowed so a local receiver can be exercised.196+func (t *Tenant) checkCallbackURL(raw 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 t.loopbackRelay() {205+ return nil206+ }207+ host := strings.ToLower(parsed.Hostname())208+ if host == "localhost" || strings.HasSuffix(host, ".localhost") || strings.HasSuffix(host, ".local") || strings.HasSuffix(host, ".internal") || !strings.Contains(host, ".") {209+ return errors.New("invalid: url must name a public host")210+ }211+ if ip := net.ParseIP(host); ip != nil && privateAddress(ip) {212+ return errors.New("invalid: url must not name a private or loopback address")213+ }214+ return nil215+}216+217+func privateAddress(ip net.IP) bool {218+ if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified() {219+ return true220+ }221+ _, shared, _ := net.ParseCIDR("100.64.0.0/10")222+ return shared.Contains(ip)223+}224+225+// loopbackRelay reports whether the relay's own public URL is a loopback or226+// private address, which only happens in tests and local trials.227+func (t *Tenant) loopbackRelay() bool {228+ parsed, err := url.Parse(t.publicURL)229+ if err != nil {230+ return false231+ }232+ host := strings.ToLower(parsed.Hostname())233+ if host == "localhost" {234+ return true235+ }236+ ip := net.ParseIP(host)237+ return ip != nil && privateAddress(ip)238+}239+240+func callbackMethod(method string) bool {241+ return containsString([]string{"listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"}, method)242+}243+244+// callbackExecute serves the callback management methods. Members and agents245+// manage callbacks for their own key; the owner and moderators see and246+// control every callback. Each call runs under the callback operation.247+func (t *Tenant) callbackExecute(ctx context.Context, actor, method string, params []json.RawMessage) (result any, err error) {248+ ctx, finish := t.app.telemetry.Start(ctx, "callback")249+ outcome := "error"250+ defer func() {251+ if err != nil {252+ outcome = callbackOutcome(err)253+ }254+ finish(outcome)255+ }()256+ role, err := t.community.Role(ctx, actor)257+ if err != nil {258+ return nil, err259+ }260+ if role == "" {261+ return nil, fmt.Errorf("restricted: guest may not %s", method)262+ }263+ operator := role == "owner" || role == "moderator"264+ switch method {265+ case "listcallbacks":266+ var rows []callbackRecord267+ if operator {268+ rows, err = t.callbackRows(ctx, "")269+ } else {270+ rows, err = t.callbackRows(ctx, actor)271+ }272+ if err != nil {273+ return nil, err274+ }275+ out := make([]map[string]any, 0, len(rows))276+ for _, row := range rows {277+ item := row.summary()278+ if row.Owner != actor {279+ // Operators see whose callback it is and where it goes,280+ // not the path, which may carry a token.281+ delete(item, "url")282+ }283+ out = append(out, item)284+ }285+ outcome = "ok"286+ return out, nil287+ case "addcallback":288+ result, err = t.addCallback(ctx, actor, role, params)289+ default:290+ result, err = t.changeCallback(ctx, actor, operator, method, params)291+ }292+ if err == nil {293+ outcome = "ok"294+ }295+ return result, err296+}297+298+func callbackOutcome(err error) string {299+ switch {300+ case strings.HasPrefix(err.Error(), "invalid:"), strings.HasPrefix(err.Error(), "not found:"):301+ return "invalid"302+ case strings.HasPrefix(err.Error(), "restricted:"):303+ return "unauthorized"304+ default:305+ return "error"306+ }307+}308+309+func (t *Tenant) addCallback(ctx context.Context, actor, role string, params []json.RawMessage) (any, error) {310+ if len(params) != 1 {311+ return nil, errors.New("invalid: addcallback expects one object with url, filter and an optional secret")312+ }313+ var options struct {314+ URL string `json:"url"`315+ Filter json.RawMessage `json:"filter"`316+ Secret string `json:"secret"`317+ }318+ if err := json.Unmarshal(params[0], &options); err != nil {319+ return nil, errors.New("invalid: addcallback expects one object with url, filter and an optional secret")320+ }321+ options.URL = strings.TrimSpace(options.URL)322+ if err := t.checkCallbackURL(options.URL); err != nil {323+ return nil, err324+ }325+ filter, err := parseCallbackFilter(options.Filter)326+ if err != nil {327+ return nil, err328+ }329+ if role == "agent" {330+ if err := t.checkCallbackScope(ctx, actor, filter); err != nil {331+ return nil, err332+ }333+ }334+ generated := false335+ secret := strings.TrimSpace(options.Secret)336+ if secret == "" {337+ secret, err = randomHex(32)338+ if err != nil {339+ return nil, err340+ }341+ generated = true342+ } else if len(secret) < callbackSecretMin || len(secret) > callbackSecretMax || strings.Trim(secret, "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~") != "" {343+ return nil, fmt.Errorf("invalid: secret must be %d to %d printable ASCII characters", callbackSecretMin, callbackSecretMax)344+ }345+ id, err := randomHex(16)346+ if err != nil {347+ return nil, err348+ }349+ encoded, err := json.Marshal(filter)350+ if err != nil {351+ return nil, err352+ }353+ limit := t.Policy().Callbacks354+ now := time.Now().Unix()355+ err = t.store.WithTx(ctx, func(tx *sql.Tx) error {356+ if role != "owner" {357+ var count int358+ if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM callbacks WHERE owner=?`, actor).Scan(&count); err != nil {359+ return err360+ }361+ if count >= limit {362+ return fmt.Errorf("invalid: at most %d callbacks per key", limit)363+ }364+ }365+ _, err := tx.ExecContext(ctx, `INSERT INTO callbacks(id,owner,url,filter,secret,created_at) VALUES(?,?,?,?,?,?)`, id, actor, options.URL, string(encoded), secret, now)366+ return err367+ })368+ if err != nil {369+ return nil, err370+ }371+ t.invalidateCallbacks()372+ if err := t.community.Record(ctx, actor, "addcallback", id, ""); err != nil {373+ return nil, err374+ }375+ // Counts only: the host, path and secret stay out of the log.376+ t.app.telemetry.Logger().Info("callback registered", "tenant", t.meta.Name, "kinds", len(filter.Kinds), "generated", generated)377+ record := callbackRecord{ID: id, Owner: actor, URL: options.URL, Filter: filter, CreatedAt: now}378+ out := record.summary()379+ out["secret"] = secret380+ return out, nil381+}382+383+// checkCallbackScope keeps an agent's callback inside its grant: a room it384+// names must be one it may post in and a repository it names must be one385+// the grant covers. Kinds are not restricted, so an agent can listen for386+// answers it may not publish itself.387+func (t *Tenant) checkCallbackScope(ctx context.Context, actor string, filter callbackFilter) error {388+ grant, ok, err := t.community.AgentGrant(ctx, actor)389+ if err != nil {390+ return err391+ }392+ if !ok || !grant.Active(time.Now().Unix()) {393+ return errors.New("restricted: agent grant is not active")394+ }395+ for _, room := range filter.Tags["h"] {396+ if !grant.AllowsRoom(room) {397+ return fmt.Errorf("restricted: agent grant does not cover room %s", room)398+ }399+ }400+ for _, coordinate := range filter.Tags["a"] {401+ owner, identifier, ok := repoCoordinate(coordinate)402+ if !ok {403+ continue404+ }405+ if grant.RepoLevel(owner, identifier) == "" {406+ return fmt.Errorf("restricted: agent grant does not cover repository %s", identifier)407+ }408+ }409+ return nil410+}411+412+func (t *Tenant) changeCallback(ctx context.Context, actor string, operator bool, method string, params []json.RawMessage) (any, error) {413+ id := callbackIDParam(params)414+ if id == "" {415+ return nil, errors.New("invalid: callback id required")416+ }417+ record, err := t.callback(ctx, id)418+ if err != nil {419+ return nil, err420+ }421+ if !operator && record.Owner != actor {422+ return nil, fmt.Errorf("restricted: only the callback's owner, the relay owner or a moderator may %s", method)423+ }424+ switch method {425+ case "removecallback":426+ _, err = t.store.DB().ExecContext(ctx, `DELETE FROM callbacks WHERE id=?`, id)427+ case "pausecallback":428+ _, err = t.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=1, last_status='paused' WHERE id=?`, id)429+ record.Paused, record.LastStatus = true, "paused"430+ case "resumecallback":431+ _, err = t.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=0, failures=0, last_status='' WHERE id=?`, id)432+ record.Paused, record.Failures, record.LastStatus = false, 0, ""433+ default:434+ return nil, fmt.Errorf("unsupported: management method %q", method)435+ }436+ if err != nil {437+ return nil, err438+ }439+ t.invalidateCallbacks()440+ if err := t.community.Record(ctx, actor, method, id, ""); err != nil {441+ return nil, err442+ }443+ if method == "removecallback" {444+ return map[string]any{"id": id, "removed": true}, nil445+ }446+ return record.summary(), nil447+}448+449+// callbackIDParam reads the id from a bare string or from {"id": ...}.450+func callbackIDParam(params []json.RawMessage) string {451+ if len(params) == 0 {452+ return ""453+ }454+ if id := stringParam(params, 0); id != "" {455+ return id456+ }457+ var options struct {458+ ID string `json:"id"`459+ }460+ _ = json.Unmarshal(params[0], &options)461+ return strings.TrimSpace(options.ID)462+}463+464+func (t *Tenant) callback(ctx context.Context, id string) (callbackRecord, error) {465+ rows, err := t.callbackRows(ctx, "", id)466+ if err != nil {467+ return callbackRecord{}, err468+ }469+ if len(rows) == 0 {470+ return callbackRecord{}, errors.New("not found: callback")471+ }472+ return rows[0], nil473+}474+475+// callbackRows lists callbacks, narrowed to an owner or to one id.476+func (t *Tenant) callbackRows(ctx context.Context, owner string, ids ...string) ([]callbackRecord, error) {477+ query := `SELECT id,owner,url,filter,secret,created_at,last_delivery_at,last_status,failures,paused FROM callbacks`478+ args := []any{}479+ switch {480+ case len(ids) > 0:481+ query += ` WHERE id=?`482+ args = append(args, ids[0])483+ case owner != "":484+ query += ` WHERE owner=?`485+ args = append(args, owner)486+ }487+ query += ` ORDER BY created_at,id`488+ rows, err := t.store.DB().QueryContext(ctx, query, args...)489+ if err != nil {490+ return nil, fmt.Errorf("list callbacks: %w", err)491+ }492+ defer rows.Close()493+ var out []callbackRecord494+ for rows.Next() {495+ var item callbackRecord496+ var filter string497+ var paused int498+ if err := rows.Scan(&item.ID, &item.Owner, &item.URL, &filter, &item.secret, &item.CreatedAt, &item.LastAt, &item.LastStatus, &item.Failures, &paused); err != nil {499+ return nil, err500+ }501+ item.Paused = paused != 0502+ if item.Filter, err = parseCallbackFilter(json.RawMessage(filter)); err != nil {503+ return nil, fmt.Errorf("callback filter: %w", err)504+ }505+ out = append(out, item)506+ }507+ return out, rows.Err()508+}509+510+// callbackRules returns the active callbacks indexed by kind. The index is511+// built from the table on first use and dropped whenever a callback512+// changes, so matching an event costs one map lookup.513+func (t *Tenant) callbackRules(ctx context.Context) map[int][]callbackRecord {514+ t.callbackMu.Lock()515+ defer t.callbackMu.Unlock()516+ if t.callbackIndex != nil {517+ return t.callbackIndex518+ }519+ index := map[int][]callbackRecord{}520+ rows, err := t.callbackRows(ctx, "")521+ if err != nil {522+ t.app.telemetry.Logger().Error("load callbacks", "tenant", t.meta.Name, "error", err)523+ return index524+ }525+ for _, row := range rows {526+ if row.Paused {527+ continue528+ }529+ for _, kind := range row.Filter.Kinds {530+ index[kind] = append(index[kind], row)531+ }532+ }533+ t.callbackIndex = index534+ return index535+}536+537+func (t *Tenant) invalidateCallbacks() {538+ t.callbackMu.Lock()539+ t.callbackIndex = nil540+ t.callbackMu.Unlock()541+}542+543+// notifyCallbacks runs after an event is stored. Each active callback whose544+// filter matches and whose key may see the event gets one delivery intent.545+// It never fails the publish; a callback is best effort on top of the546+// durable event.547+func (t *Tenant) notifyCallbacks(ctx context.Context, e event.Event) {548+ candidates := t.callbackRules(ctx)[e.Kind]549+ if len(candidates) == 0 {550+ return551+ }552+ var intents []storage.Intent553+ unauthorized := 0554+ for _, candidate := range candidates {555+ if !event.Matches(candidate.Filter.eventFilter(), e) {556+ continue557+ }558+ if !t.gate.CanSee(ctx, e, relay.Session{PubKeys: []string{candidate.Owner}, RelayURL: t.RelayURL()}, nil) {559+ unauthorized++560+ continue561+ }562+ payload, err := json.Marshal(callbackPayload{Event: e, Attempt: 1})563+ if err != nil {564+ continue565+ }566+ intents = append(intents, storage.Intent{Kind: callbackDelivery, EventID: e.ID, Target: candidate.ID, Payload: string(payload)})567+ }568+ if len(intents) == 0 {569+ return570+ }571+ err := t.store.WithTx(ctx, func(tx *sql.Tx) error {572+ return storage.AddIntents(ctx, tx, intents, time.Now().Unix())573+ })574+ if err != nil {575+ t.app.telemetry.Logger().Error("callback deliveries not queued", "tenant", t.meta.Name, "error", err)576+ return577+ }578+ t.app.telemetry.Logger().Debug("callback deliveries queued", "tenant", t.meta.Name, "queued", len(intents), "unauthorized", unauthorized)579+}580+581+func randomHex(size int) (string, error) {582+ buffer := make([]byte, size)583+ if _, err := rand.Read(buffer); err != nil {584+ return "", fmt.Errorf("callback secret: %w", err)585+ }586+ return hex.EncodeToString(buffer), nil587+}@@ -0,0 +1,205 @@
1+package daemon2+3+// Callback delivery. Each matching event is one POST to the callback URL,4+// signed with the callback's secret. A failed POST is tried again after one5+// minute and then after five; after 20 failures in a row the callback is6+// paused until its owner resumes it.7+8+import (9+ "context"10+ "crypto/hmac"11+ "crypto/sha256"12+ "crypto/tls"13+ "database/sql"14+ "encoding/hex"15+ "encoding/json"16+ "errors"17+ "fmt"18+ "io"19+ "net/http"20+ "strings"21+ "sync"22+ "time"23+24+ "github.com/FelineStateMachine/tinyrelay/internal/event"25+ "github.com/FelineStateMachine/tinyrelay/internal/relay"26+ "github.com/FelineStateMachine/tinyrelay/internal/replication"27+ "github.com/FelineStateMachine/tinyrelay/internal/work"28+)29+30+// callbackPayload is the work intent body: the event itself, so delivery31+// does not depend on the event still being stored, and the attempt number.32+type callbackPayload struct {33+ Event event.Event `json:"event"`34+ Attempt int `json:"attempt"`35+}36+37+// callbackSignature is the X-Tiny-Signature value for a body: sha256= and38+// the hex HMAC-SHA256 of the body under the callback secret.39+func callbackSignature(secret string, body []byte) string {40+ mac := hmac.New(sha256.New, []byte(secret))41+ mac.Write(body)42+ return "sha256=" + hex.EncodeToString(mac.Sum(nil))43+}44+45+// callbackHTTPClient refuses private addresses and redirects unless the46+// relay itself runs on a loopback address.47+func (t *Tenant) callbackHTTPClient() *http.Client {48+ if t.callbackClient != nil {49+ return t.callbackClient50+ }51+ if t.loopbackRelay() {52+ return &http.Client{Timeout: callbackTimeout, Transport: &http.Transport{TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}}, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}53+ }54+ return replication.NewPinnedClient(callbackTimeout)55+}56+57+// callbackLock bounds delivery to one POST per callback at a time.58+func (t *Tenant) callbackLock(id string) *sync.Mutex {59+ t.callbackMu.Lock()60+ defer t.callbackMu.Unlock()61+ if t.callbackLocks == nil {62+ t.callbackLocks = map[string]*sync.Mutex{}63+ }64+ lock := t.callbackLocks[id]65+ if lock == nil {66+ lock = &sync.Mutex{}67+ t.callbackLocks[id] = lock68+ }69+ return lock70+}71+72+// handleCallbackDelivery serves one callback-delivery intent. It rechecks73+// the callback and the key's access right before the POST, so a callback74+// removed or paused after the event arrived delivers nothing.75+func (t *Tenant) handleCallbackDelivery(ctx context.Context, intent work.Intent) (err error) {76+ ctx, finish := t.app.telemetry.Start(ctx, "callback")77+ outcome := "error"78+ defer func() { finish(outcome) }()79+ var payload callbackPayload80+ if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil || payload.Event.ID == "" {81+ outcome = "invalid"82+ return errors.New("callback-delivery: invalid payload")83+ }84+ if payload.Attempt < 1 {85+ payload.Attempt = 186+ }87+ lock := t.callbackLock(intent.Target)88+ lock.Lock()89+ defer lock.Unlock()90+ record, err := t.callback(ctx, intent.Target)91+ if err != nil {92+ if strings.HasPrefix(err.Error(), "not found:") {93+ outcome = "paused"94+ return nil95+ }96+ return err97+ }98+ if record.Paused {99+ outcome = "paused"100+ return nil101+ }102+ role, err := t.community.Role(ctx, record.Owner)103+ if err != nil {104+ return err105+ }106+ if role == "" && record.Owner != t.Policy().Owner {107+ // The key lost its standing since it registered; stop until a108+ // person resumes the callback.109+ outcome = "unauthorized"110+ return t.pauseCallback(ctx, record.ID, record.Failures, "paused: the key is no longer a member")111+ }112+ if !t.gate.CanSee(ctx, payload.Event, relay.Session{PubKeys: []string{record.Owner}, RelayURL: t.RelayURL()}, nil) {113+ outcome = "unauthorized"114+ return nil115+ }116+ status, postErr := t.postCallback(ctx, record, payload.Event)117+ now := time.Now().Unix()118+ if postErr == nil {119+ outcome = "ok"120+ _, err = t.store.DB().ExecContext(ctx, `UPDATE callbacks SET last_delivery_at=?, last_status='ok', failures=0 WHERE id=?`, now, record.ID)121+ t.app.telemetry.Logger().Info("callback delivered", "tenant", t.meta.Name, "attempt", payload.Attempt, "status", status)122+ return err123+ }124+ // Counts only: the reason names the HTTP status or the failure class,125+ // never the URL.126+ reason := postErr.Error()127+ failures := record.Failures + 1128+ paused := failures >= callbackPauseFailures129+ if paused {130+ outcome = "paused"131+ if err := t.pauseCallback(ctx, record.ID, failures, fmt.Sprintf("paused after %d failures: %s", failures, reason)); err != nil {132+ return err133+ }134+ } else if _, err := t.store.DB().ExecContext(ctx, `UPDATE callbacks SET failures=?, last_status=? WHERE id=?`, failures, reason, record.ID); err != nil {135+ return err136+ }137+ t.app.telemetry.Logger().Info("callback delivery failed", "tenant", t.meta.Name, "attempt", payload.Attempt, "status", status, "failures", failures, "paused", paused)138+ if paused || payload.Attempt >= callbackMaxAttempts {139+ return nil140+ }141+ return t.retryCallback(ctx, intent, payload)142+}143+144+// pauseCallback stops deliveries and records the count and reason, then145+// drops the index so no new intents are queued for the callback.146+func (t *Tenant) pauseCallback(ctx context.Context, id string, failures int, status string) error {147+ if _, err := t.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=1, failures=?, last_status=? WHERE id=?`, failures, status, id); err != nil {148+ return err149+ }150+ t.invalidateCallbacks()151+ return nil152+}153+154+// retryCallback queues the next attempt after its backoff. Each attempt is155+// its own intent so the durable queue's ordering and fencing still apply.156+func (t *Tenant) retryCallback(ctx context.Context, intent work.Intent, payload callbackPayload) error {157+ delay := callbackBackoff[len(callbackBackoff)-1]158+ if payload.Attempt-1 < len(callbackBackoff) {159+ delay = callbackBackoff[payload.Attempt-1]160+ }161+ payload.Attempt++162+ encoded, err := json.Marshal(payload)163+ if err != nil {164+ return err165+ }166+ now := time.Now().Unix()167+ key := sha256.Sum256([]byte(callbackDelivery + "\x00" + payload.Event.ID + "\x00" + intent.Target + "\x00" + fmt.Sprint(payload.Attempt)))168+ return t.store.WithTx(ctx, func(tx *sql.Tx) error {169+ _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO work_intents(id,kind,event_id,target,payload,next_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)`, hex.EncodeToString(key[:]), callbackDelivery, payload.Event.ID+"#"+fmt.Sprint(payload.Attempt), intent.Target, string(encoded), now+int64(delay/time.Second), now, now)170+ return err171+ })172+}173+174+// postCallback sends the event and returns the HTTP status with a bounded175+// reason on failure.176+func (t *Tenant) postCallback(ctx context.Context, record callbackRecord, e event.Event) (int, error) {177+ body, err := json.Marshal(e)178+ if err != nil {179+ return 0, errors.New("encode event")180+ }181+ ctx, cancel := context.WithTimeout(ctx, callbackTimeout)182+ defer cancel()183+ request, err := http.NewRequestWithContext(ctx, http.MethodPost, record.URL, strings.NewReader(string(body)))184+ if err != nil {185+ return 0, errors.New("bad url")186+ }187+ request.Header.Set("Content-Type", "application/json")188+ request.Header.Set("X-Tiny-Callback", record.ID)189+ request.Header.Set("X-Tiny-Signature", callbackSignature(record.secret, body))190+ request.Header.Set("X-Tiny-Relay", t.publicURL)191+ request.Header.Set("User-Agent", "tinyrelay")192+ response, err := t.callbackHTTPClient().Do(request)193+ if err != nil {194+ if errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {195+ return 0, errors.New("timeout")196+ }197+ return 0, errors.New("connection failed")198+ }199+ defer response.Body.Close()200+ _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096))201+ if response.StatusCode < 200 || response.StatusCode >= 300 {202+ return response.StatusCode, fmt.Errorf("HTTP %d", response.StatusCode)203+ }204+ return response.StatusCode, nil205+}@@ -0,0 +1,679 @@
1+package daemon2+3+import (4+ "context"5+ "crypto/hmac"6+ "crypto/sha256"7+ "encoding/hex"8+ "encoding/json"9+ "io"10+ "net/http"11+ "net/http/httptest"12+ "strconv"13+ "strings"14+ "sync"15+ "testing"16+ "time"17+18+ "github.com/FelineStateMachine/tinyrelay/internal/community"19+ "github.com/FelineStateMachine/tinyrelay/internal/event"20+ "github.com/FelineStateMachine/tinyrelay/internal/work"21+)22+23+// loopbackTenant is a tenant whose public URL is a loopback address, which24+// lets a callback point at a local test server.25+func loopbackTenant(t *testing.T) (*App, *Tenant) {26+ t.Helper()27+ ctx := context.Background()28+ a, err := New(ctx, Config{DataDir: t.TempDir(), DefaultTenant: "main"})29+ if err != nil {30+ t.Fatal(err)31+ }32+ t.Cleanup(func() {33+ if err := a.Close(ctx); err != nil {34+ t.Error(err)35+ }36+ })37+ owner, _ := event.PublicKey(testOwnerSecret)38+ meta, err := a.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "default"})39+ if err != nil {40+ t.Fatal(err)41+ }42+ tenant, err := a.tenant(ctx, meta, "http://127.0.0.1:8080")43+ if err != nil {44+ t.Fatal(err)45+ }46+ // Fence the production worker so the tests drive each delivery intent47+ // themselves.48+ tenant.workCancel()49+ tenant.workWG.Wait()50+ return a, tenant51+}52+53+func setRole(t *testing.T, tenant *Tenant, pubkey, role string) {54+ t.Helper()55+ raw, _ := json.Marshal(map[string]string{"role": role})56+ if _, err := tenant.Execute(context.Background(), tenant.Policy().Owner, "setmember", []json.RawMessage{json.RawMessage(strconv.Quote(pubkey)), raw}); err != nil {57+ t.Fatal(err)58+ }59+}60+61+func callbackParams(url string, filter string, extra ...string) []json.RawMessage {62+ options := map[string]any{"url": url, "filter": json.RawMessage(filter)}63+ if len(extra) > 0 {64+ options["secret"] = extra[0]65+ }66+ raw, _ := json.Marshal(options)67+ return []json.RawMessage{raw}68+}69+70+func addCallback(t *testing.T, tenant *Tenant, actor, url, filter string) map[string]any {71+ t.Helper()72+ result, err := tenant.Execute(context.Background(), actor, "addcallback", callbackParams(url, filter))73+ if err != nil {74+ t.Fatalf("addcallback: %v", err)75+ }76+ return result.(map[string]any)77+}78+79+func TestCallbackRegistrationValidatesURLFilterScopeAndCap(t *testing.T) {80+ _, tenant := testTenant(t)81+ ctx := context.Background()82+ owner := tenant.Policy().Owner83+ member, _ := event.PublicKey(testMemberSecret)84+ agent, _ := event.PublicKey(testAgentSecret)85+ setRole(t, tenant, member, "member")86+ now := time.Now().Unix()87+ if err := publishAs(t, tenant, agentGrantEvent(t, agent, now, now+3600, []string{"k", "1"}, []string{"room", "build"}, []string{"repo", owner + ":tinyrelay:read"})); err != nil {88+ t.Fatal(err)89+ }90+ ids := make([]string, 9)91+ for i := range ids {92+ ids[i] = strings.Repeat(strconv.Itoa(i), 64)93+ }94+ nineIDs, _ := json.Marshal(ids)95+ for _, tc := range []struct {96+ name, actor, url, filter, want string97+ }{98+ {"guest", strings.Repeat("e", 64), "https://hooks.example/wake", `{"kinds":[1]}`, "restricted:"},99+ {"http url", member, "http://hooks.example/wake", `{"kinds":[1]}`, "invalid: url must be https"},100+ {"credentials", member, "https://user:pw@hooks.example/wake", `{"kinds":[1]}`, "invalid: url must be https"},101+ {"private address", member, "https://10.0.0.1/wake", `{"kinds":[1]}`, "invalid: url must not name a private"},102+ {"loopback address", member, "https://127.0.0.1/wake", `{"kinds":[1]}`, "invalid: url must not name a private"},103+ {"localhost", member, "https://localhost/wake", `{"kinds":[1]}`, "invalid: url must name a public host"},104+ {"bare host", member, "https://relay/wake", `{"kinds":[1]}`, "invalid: url must name a public host"},105+ {"no kinds", member, "https://hooks.example/wake", `{"#p":["` + member + `"]}`, "invalid: filter must name at least one kind"},106+ {"unsupported key", member, "https://hooks.example/wake", `{"kinds":[1],"ids":["` + strings.Repeat("a", 64) + `"]}`, "invalid: filter key \"ids\""},107+ {"unsupported tag", member, "https://hooks.example/wake", `{"kinds":[1],"#t":["x"]}`, "invalid: filter key \"#t\""},108+ {"too many values", member, "https://hooks.example/wake", `{"kinds":[1],"#e":` + string(nineIDs) + `}`, "invalid: #e must list 1 to 8 values"},109+ {"bad pubkey", member, "https://hooks.example/wake", `{"kinds":[1],"#p":["nope"]}`, "invalid: #p values"},110+ {"bad coordinate", member, "https://hooks.example/wake", `{"kinds":[1621],"#a":["tinyrelay"]}`, "invalid: #a values"},111+ {"bad room", member, "https://hooks.example/wake", `{"kinds":[9],"#h":["Not A Room"]}`, "invalid: #h values"},112+ {"not an object", member, "https://hooks.example/wake", `[1]`, "invalid: filter must be a JSON object"},113+ {"agent room outside grant", agent, "https://hooks.example/wake", `{"kinds":[9],"#h":["general"]}`, "restricted: agent grant does not cover room general"},114+ {"agent repo outside grant", agent, "https://hooks.example/wake", `{"kinds":[1621],"#a":["30617:` + owner + `:other"]}`, "restricted: agent grant does not cover repository other"},115+ } {116+ _, err := tenant.Execute(ctx, tc.actor, "addcallback", callbackParams(tc.url, tc.filter))117+ if err == nil || !strings.HasPrefix(err.Error(), tc.want) {118+ t.Errorf("%s: err = %v, want prefix %q", tc.name, err, tc.want)119+ }120+ }121+ if _, err := tenant.Execute(ctx, member, "addcallback", callbackParams("https://hooks.example/wake", `{"kinds":[1]}`, "short")); err == nil || !strings.HasPrefix(err.Error(), "invalid: secret must be") {122+ t.Fatalf("short secret: %v", err)123+ }124+ // An agent may watch the room and repository its grant covers, and any125+ // kind, including ones it may not publish.126+ inside := addCallback(t, tenant, agent, "https://hooks.example/wake", `{"kinds":[7,9,1621],"#h":["build"],"#a":["30617:`+owner+`:tinyrelay"],"since":1}`)127+ if inside["owner"] != agent || inside["host"] != "hooks.example" || inside["paused"] != false {128+ t.Fatalf("agent callback = %v", inside)129+ }130+ secret, _ := inside["secret"].(string)131+ if len(secret) != 64 {132+ t.Fatalf("generated secret = %q", secret)133+ }134+ filter, _ := json.Marshal(inside["filter"])135+ if string(filter) != `{"#a":["30617:`+owner+`:tinyrelay"],"#h":["build"],"kinds":[7,9,1621]}` {136+ t.Fatalf("stored filter = %s", filter)137+ }138+ // A member gets four callbacks; the owner has no cap.139+ for i := 0; i < 4; i++ {140+ addCallback(t, tenant, member, "https://hooks.example/wake/"+strconv.Itoa(i), `{"kinds":[1]}`)141+ }142+ if _, err := tenant.Execute(ctx, member, "addcallback", callbackParams("https://hooks.example/wake/5", `{"kinds":[1]}`)); err == nil || err.Error() != "invalid: at most 4 callbacks per key" {143+ t.Fatalf("fifth callback: %v", err)144+ }145+ for i := 0; i < 6; i++ {146+ addCallback(t, tenant, owner, "https://hooks.example/owner/"+strconv.Itoa(i), `{"kinds":[1]}`)147+ }148+ // The secret shows once: listings and the table row never repeat it.149+ listed, err := tenant.Execute(ctx, agent, "listcallbacks", nil)150+ if err != nil {151+ t.Fatal(err)152+ }153+ rows := listed.([]map[string]any)154+ if len(rows) != 1 || rows[0]["id"] != inside["id"] || rows[0]["url"] != "https://hooks.example/wake" {155+ t.Fatalf("agent listing = %v", rows)156+ }157+ if _, present := rows[0]["secret"]; present {158+ t.Fatal("listing repeats the secret")159+ }160+ encoded, _ := json.Marshal(rows[0])161+ if strings.Contains(string(encoded), secret) {162+ t.Fatal("listing carries the secret")163+ }164+ // Setting the policy allowance to zero closes registration for165+ // everyone but the owner.166+ if _, err := tenant.Execute(ctx, owner, "setpolicy", []json.RawMessage{json.RawMessage(`{"callbacks":0}`)}); err != nil {167+ t.Fatal(err)168+ }169+ if _, err := tenant.Execute(ctx, agent, "addcallback", callbackParams("https://hooks.example/again", `{"kinds":[1]}`)); err == nil || err.Error() != "invalid: at most 0 callbacks per key" {170+ t.Fatalf("closed registration: %v", err)171+ }172+ if got := tenant.Policy().Callbacks; got != 0 {173+ t.Fatalf("policy callbacks = %d", got)174+ }175+}176+177+func TestCallbackManagementPermissionMatrix(t *testing.T) {178+ _, tenant := testTenant(t)179+ ctx := context.Background()180+ owner := tenant.Policy().Owner181+ member, _ := event.PublicKey(testMemberSecret)182+ moderator, _ := event.PublicKey(testModSecret)183+ agent, _ := event.PublicKey(testAgentSecret)184+ guest := strings.Repeat("e", 64)185+ setRole(t, tenant, member, "member")186+ setRole(t, tenant, moderator, "moderator")187+ now := time.Now().Unix()188+ if err := publishAs(t, tenant, agentGrantEvent(t, agent, now, now+3600, []string{"k", "1"})); err != nil {189+ t.Fatal(err)190+ }191+ methods, err := tenant.Execute(ctx, owner, "supportedmethods", nil)192+ if err != nil {193+ t.Fatal(err)194+ }195+ for _, name := range []string{"listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"} {196+ if !containsString(methods.([]string), name) {197+ t.Fatalf("supportedmethods lacks %s", name)198+ }199+ }200+ mine := addCallback(t, tenant, member, "https://hooks.example/member?token=abc", `{"kinds":[1]}`)201+ theirs := addCallback(t, tenant, agent, "https://hooks.example/agent", `{"kinds":[1]}`)202+ id := func(values map[string]any) json.RawMessage {203+ return json.RawMessage(strconv.Quote(values["id"].(string)))204+ }205+ for _, method := range []string{"listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"} {206+ if _, err := tenant.Execute(ctx, guest, method, []json.RawMessage{id(mine)}); err == nil || !strings.HasPrefix(err.Error(), "restricted:") {207+ t.Fatalf("%s by guest: %v", method, err)208+ }209+ }210+ // A member sees and controls only its own callbacks.211+ listed, _ := tenant.Execute(ctx, member, "listcallbacks", nil)212+ if rows := listed.([]map[string]any); len(rows) != 1 || rows[0]["id"] != mine["id"] {213+ t.Fatalf("member listing = %v", rows)214+ }215+ for _, method := range []string{"pausecallback", "resumecallback", "removecallback"} {216+ if _, err := tenant.Execute(ctx, member, method, []json.RawMessage{id(theirs)}); err == nil || !strings.HasPrefix(err.Error(), "restricted:") {217+ t.Fatalf("%s on another key's callback: %v", method, err)218+ }219+ }220+ paused, err := tenant.Execute(ctx, member, "pausecallback", []json.RawMessage{json.RawMessage(`{"id":"` + mine["id"].(string) + `"}`)})221+ if err != nil || paused.(map[string]any)["paused"] != true {222+ t.Fatalf("pause own: %v %v", paused, err)223+ }224+ if resumed, err := tenant.Execute(ctx, member, "resumecallback", []json.RawMessage{id(mine)}); err != nil || resumed.(map[string]any)["paused"] != false {225+ t.Fatalf("resume own: %v %v", resumed, err)226+ }227+ // Moderators and the owner see everything, with the host but not the228+ // path of other keys' URLs, and may control any callback.229+ for _, actor := range []string{moderator, owner} {230+ listed, err := tenant.Execute(ctx, actor, "listcallbacks", nil)231+ if err != nil {232+ t.Fatalf("listcallbacks by %s: %v", actor, err)233+ }234+ rows := listed.([]map[string]any)235+ byOwner := map[string]map[string]any{}236+ for _, row := range rows {237+ byOwner[row["owner"].(string)] = row238+ }239+ if len(rows) != 2 || byOwner[member]["host"] != "hooks.example" || byOwner[agent]["id"] != theirs["id"] {240+ t.Fatalf("operator listing = %v", rows)241+ }242+ if _, present := byOwner[member]["url"]; present {243+ t.Fatalf("operator listing carries another key's URL path: %v", byOwner[member])244+ }245+ if _, err := tenant.Execute(ctx, actor, "pausecallback", []json.RawMessage{id(theirs)}); err != nil {246+ t.Fatalf("pause by %s: %v", actor, err)247+ }248+ if _, err := tenant.Execute(ctx, actor, "resumecallback", []json.RawMessage{id(theirs)}); err != nil {249+ t.Fatalf("resume by %s: %v", actor, err)250+ }251+ }252+ if _, err := tenant.Execute(ctx, owner, "pausecallback", []json.RawMessage{json.RawMessage(`"missing"`)}); err == nil || !strings.HasPrefix(err.Error(), "not found:") {253+ t.Fatalf("unknown id: %v", err)254+ }255+ if _, err := tenant.Execute(ctx, owner, "pausecallback", nil); err == nil || !strings.HasPrefix(err.Error(), "invalid:") {256+ t.Fatalf("missing id: %v", err)257+ }258+ removed, err := tenant.Execute(ctx, moderator, "removecallback", []json.RawMessage{id(theirs)})259+ if err != nil || removed.(map[string]any)["removed"] != true {260+ t.Fatalf("remove by moderator: %v %v", removed, err)261+ }262+ if _, err := tenant.Execute(ctx, agent, "removecallback", []json.RawMessage{id(mine)}); err == nil || !strings.HasPrefix(err.Error(), "restricted:") {263+ t.Fatalf("agent removing a member's callback: %v", err)264+ }265+ if _, err := tenant.Execute(ctx, member, "removecallback", []json.RawMessage{id(mine)}); err != nil {266+ t.Fatalf("remove own: %v", err)267+ }268+ audit, err := tenant.Execute(ctx, owner, "listaudit", nil)269+ if err != nil {270+ t.Fatal(err)271+ }272+ seen := map[string]bool{}273+ for _, row := range audit.([]community.AuditRow) {274+ seen[row.Action] = true275+ if strings.Contains(row.Detail, "hooks.example") || strings.Contains(row.Target, "hooks.example") {276+ t.Fatalf("audit carries a callback URL: %+v", row)277+ }278+ }279+ for _, action := range []string{"addcallback", "pausecallback", "resumecallback", "removecallback"} {280+ if !seen[action] {281+ t.Fatalf("audit lacks %s: %v", action, seen)282+ }283+ }284+}285+286+func TestCallbackMatchesFilterAndKeepsPrivateEventsBehindTheGate(t *testing.T) {287+ _, tenant := loopbackTenant(t)288+ ctx := context.Background()289+ member, _ := event.PublicKey(testMemberSecret)290+ moderator, _ := event.PublicKey(testModSecret)291+ setRole(t, tenant, member, "member")292+ setRole(t, tenant, moderator, "moderator")293+ now := time.Now().Unix()294+ notes := addCallback(t, tenant, member, "https://127.0.0.1:1/notes", `{"kinds":[1],"#p":["`+member+`"]}`)295+ messages := addCallback(t, tenant, member, "https://127.0.0.1:1/messages", `{"kinds":[4]}`)296+ intents := func() map[string]int {297+ rows, err := tenant.store.DB().QueryContext(ctx, "SELECT target,count(*) FROM work_intents WHERE kind=? GROUP BY target", callbackDelivery)298+ if err != nil {299+ t.Fatal(err)300+ }301+ defer rows.Close()302+ counts := map[string]int{}303+ for rows.Next() {304+ var target string305+ var count int306+ if err := rows.Scan(&target, &count); err != nil {307+ t.Fatal(err)308+ }309+ counts[target] = count310+ }311+ return counts312+ }313+ // Kind and tag both have to match.314+ if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 1, now, [][]string{{"p", moderator}}, "not for the member")); err != nil {315+ t.Fatal(err)316+ }317+ if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 7, now, [][]string{{"p", member}}, "+")); err != nil {318+ t.Fatal(err)319+ }320+ if counts := intents(); len(counts) != 0 {321+ t.Fatalf("intents after unmatched events = %v", counts)322+ }323+ if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 1, now+1, [][]string{{"p", member}}, "for the member")); err != nil {324+ t.Fatal(err)325+ }326+ if counts := intents(); counts[notes["id"].(string)] != 1 {327+ t.Fatalf("intents after matched note = %v", counts)328+ }329+ // A private message the key may not read is matched by the filter but330+ // stopped by the gate; one addressed to the key goes through.331+ if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 4, now+2, [][]string{{"p", moderator}}, "secret")); err != nil {332+ t.Fatal(err)333+ }334+ if counts := intents(); counts[messages["id"].(string)] != 0 {335+ t.Fatalf("private message leaked to the callback: %v", counts)336+ }337+ if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 4, now+3, [][]string{{"p", member}}, "for you")); err != nil {338+ t.Fatal(err)339+ }340+ if counts := intents(); counts[messages["id"].(string)] != 1 {341+ t.Fatalf("addressed message not queued: %v", counts)342+ }343+ // A paused callback leaves the index at once.344+ if _, err := tenant.Execute(ctx, member, "pausecallback", []json.RawMessage{json.RawMessage(strconv.Quote(notes["id"].(string)))}); err != nil {345+ t.Fatal(err)346+ }347+ if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 1, now+4, [][]string{{"p", member}}, "while paused")); err != nil {348+ t.Fatal(err)349+ }350+ if counts := intents(); counts[notes["id"].(string)] != 1 {351+ t.Fatalf("paused callback still queued: %v", counts)352+ }353+ if got := tenant.Policy().Callbacks; got != 4 {354+ t.Fatalf("default callbacks allowance = %d", got)355+ }356+}357+358+type callbackReceiver struct {359+ mu sync.Mutex360+ status int361+ headers []http.Header362+ bodies [][]byte363+}364+365+func (r *callbackReceiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {366+ body, _ := io.ReadAll(req.Body)367+ r.mu.Lock()368+ r.headers = append(r.headers, req.Header.Clone())369+ r.bodies = append(r.bodies, body)370+ status := r.status371+ r.mu.Unlock()372+ w.WriteHeader(status)373+}374+375+func (r *callbackReceiver) count() int {376+ r.mu.Lock()377+ defer r.mu.Unlock()378+ return len(r.bodies)379+}380+381+// pendingCallbackIntents lists queued deliveries of one event to one382+// callback, first attempt first.383+func pendingCallbackIntents(t *testing.T, tenant *Tenant, id, eventID string) []work.Intent {384+ t.Helper()385+ rows, err := tenant.store.DB().QueryContext(context.Background(), "SELECT id,event_id,payload,next_at FROM work_intents WHERE kind=? AND target=? AND state='pending' AND (event_id=? OR event_id LIKE ?) ORDER BY next_at,created_at,id", callbackDelivery, id, eventID, eventID+"#%")386+ if err != nil {387+ t.Fatal(err)388+ }389+ defer rows.Close()390+ var out []work.Intent391+ for rows.Next() {392+ var item work.Intent393+ var next int64394+ if err := rows.Scan(&item.ID, &item.EventID, &item.Payload, &next); err != nil {395+ t.Fatal(err)396+ }397+ item.Kind, item.Target, item.NextAt = callbackDelivery, id, time.Unix(next, 0)398+ out = append(out, item)399+ }400+ return out401+}402+403+func TestCallbackDeliverySignsBodyRetriesAndPauses(t *testing.T) {404+ _, tenant := loopbackTenant(t)405+ ctx := context.Background()406+ member, _ := event.PublicKey(testMemberSecret)407+ setRole(t, tenant, member, "member")408+ receiver := &callbackReceiver{status: http.StatusAccepted}409+ service := httptest.NewTLSServer(receiver)410+ defer service.Close()411+ tenant.callbackClient = service.Client()412+ registered := addCallback(t, tenant, member, service.URL+"/wake?token=t0k3n", `{"kinds":[1621]}`)413+ id, secret := registered["id"].(string), registered["secret"].(string)414+ now := time.Now().Unix()415+ issue := signedEvent(t, testOwnerSecret, 1621, now, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Broken build"}}, "The build fails.")416+ if err := publishAs(t, tenant, issue); err != nil {417+ t.Fatal(err)418+ }419+ queued := pendingCallbackIntents(t, tenant, id, issue.ID)420+ if len(queued) != 1 || queued[0].EventID != issue.ID {421+ t.Fatalf("queued = %+v", queued)422+ }423+ if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {424+ t.Fatal(err)425+ }426+ if receiver.count() != 1 {427+ t.Fatalf("deliveries = %d", receiver.count())428+ }429+ headers, body := receiver.headers[0], receiver.bodies[0]430+ mac := hmac.New(sha256.New, []byte(secret))431+ mac.Write(body)432+ if headers.Get("Content-Type") != "application/json" || headers.Get("X-Tiny-Callback") != id || headers.Get("X-Tiny-Relay") != "http://127.0.0.1:8080" || headers.Get("X-Tiny-Signature") != "sha256="+hex.EncodeToString(mac.Sum(nil)) {433+ t.Fatalf("headers = %v", headers)434+ }435+ delivered, err := event.Parse(body)436+ if err != nil || delivered.ID != issue.ID || delivered.Content != issue.Content {437+ t.Fatalf("body = %s (%v)", body, err)438+ }439+ var lastAt int64440+ var lastStatus string441+ var failures, paused int442+ state := func() {443+ t.Helper()444+ if err := tenant.store.DB().QueryRowContext(ctx, "SELECT last_delivery_at,last_status,failures,paused FROM callbacks WHERE id=?", id).Scan(&lastAt, &lastStatus, &failures, &paused); err != nil {445+ t.Fatal(err)446+ }447+ }448+ state()449+ if lastAt < now || lastStatus != "ok" || failures != 0 || paused != 0 {450+ t.Fatalf("after delivery: at=%d status=%q failures=%d paused=%d", lastAt, lastStatus, failures, paused)451+ }452+ listed, _ := tenant.Execute(ctx, member, "listcallbacks", nil)453+ if rows := listed.([]map[string]any); rows[0]["lastStatus"] != "ok" || rows[0]["lastDelivery"].(int64) < now {454+ t.Fatalf("listing after delivery = %v", rows)455+ }456+457+ // A failing receiver: attempt one queues attempt two a minute out,458+ // attempt two queues attempt three five minutes out, and attempt three459+ // is the last.460+ receiver.status = http.StatusInternalServerError461+ second := signedEvent(t, testOwnerSecret, 1621, now+1, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Again"}}, "Again.")462+ if err := publishAs(t, tenant, second); err != nil {463+ t.Fatal(err)464+ }465+ queued = pendingCallbackIntents(t, tenant, id, second.ID)466+ if len(queued) != 1 {467+ t.Fatalf("queued = %+v", queued)468+ }469+ if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {470+ t.Fatalf("attempt 1 should not fail the intent: %v", err)471+ }472+ state()473+ if failures != 1 || lastStatus != "HTTP 500" || paused != 0 {474+ t.Fatalf("after attempt 1: status=%q failures=%d paused=%d", lastStatus, failures, paused)475+ }476+ retries := pendingCallbackIntents(t, tenant, id, second.ID)477+ if len(retries) != 2 || retries[1].EventID != second.ID+"#2" || retries[1].NextAt.Unix() < now+55 || retries[1].NextAt.Unix() > now+70 {478+ t.Fatalf("retry after attempt 1 = %+v", retries)479+ }480+ var payload callbackPayload481+ if err := json.Unmarshal([]byte(retries[1].Payload), &payload); err != nil || payload.Attempt != 2 || payload.Event.ID != second.ID {482+ t.Fatalf("retry payload = %s", retries[1].Payload)483+ }484+ if err := tenant.handleCallbackDelivery(ctx, retries[1]); err != nil {485+ t.Fatal(err)486+ }487+ retries = pendingCallbackIntents(t, tenant, id, second.ID)488+ if len(retries) != 3 || retries[2].EventID != second.ID+"#3" || retries[2].NextAt.Unix() < now+295 || retries[2].NextAt.Unix() > now+310 {489+ t.Fatalf("retry after attempt 2 = %+v", retries)490+ }491+ if err := tenant.handleCallbackDelivery(ctx, retries[2]); err != nil {492+ t.Fatal(err)493+ }494+ if again := pendingCallbackIntents(t, tenant, id, second.ID); len(again) != 3 {495+ t.Fatalf("attempt 3 queued another attempt: %+v", again)496+ }497+ state()498+ if failures != 3 || receiver.count() != 4 {499+ t.Fatalf("after three attempts: failures=%d deliveries=%d", failures, receiver.count())500+ }501+ // Twenty failures in a row pause the callback with the reason, after502+ // which nothing more is queued or sent until it is resumed.503+ for failures < callbackPauseFailures {504+ if err := tenant.handleCallbackDelivery(ctx, retries[2]); err != nil {505+ t.Fatal(err)506+ }507+ state()508+ }509+ if paused != 1 || !strings.HasPrefix(lastStatus, "paused after 20 failures: HTTP 500") {510+ t.Fatalf("after 20 failures: status=%q paused=%d", lastStatus, paused)511+ }512+ sent := receiver.count()513+ third := signedEvent(t, testOwnerSecret, 1621, now+2, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Third"}}, "Third.")514+ if err := publishAs(t, tenant, third); err != nil {515+ t.Fatal(err)516+ }517+ if queued := pendingCallbackIntents(t, tenant, id, third.ID); len(queued) != 0 {518+ t.Fatalf("paused callback queued a delivery: %+v", queued)519+ }520+ if err := tenant.handleCallbackDelivery(ctx, retries[2]); err != nil || receiver.count() != sent {521+ t.Fatalf("paused callback delivered: %v %d", err, receiver.count())522+ }523+ receiver.status = http.StatusOK524+ if _, err := tenant.Execute(ctx, member, "resumecallback", []json.RawMessage{json.RawMessage(strconv.Quote(id))}); err != nil {525+ t.Fatal(err)526+ }527+ state()528+ if paused != 0 || failures != 0 {529+ t.Fatalf("after resume: failures=%d paused=%d", failures, paused)530+ }531+ fourth := signedEvent(t, testOwnerSecret, 1621, now+3, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Fourth"}}, "Fourth.")532+ if err := publishAs(t, tenant, fourth); err != nil {533+ t.Fatal(err)534+ }535+ queued = pendingCallbackIntents(t, tenant, id, fourth.ID)536+ if len(queued) != 1 {537+ t.Fatalf("queued after resume = %+v", queued)538+ }539+ if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != sent+1 {540+ t.Fatalf("delivery after resume: %v %d", err, receiver.count())541+ }542+ // A removed callback consumes its intents without a POST.543+ if _, err := tenant.Execute(ctx, member, "removecallback", []json.RawMessage{json.RawMessage(strconv.Quote(id))}); err != nil {544+ t.Fatal(err)545+ }546+ if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != sent+1 {547+ t.Fatalf("removed callback delivered: %v %d", err, receiver.count())548+ }549+}550+551+func TestCallbackDeliveryRechecksTheGateAndMembership(t *testing.T) {552+ _, tenant := loopbackTenant(t)553+ ctx := context.Background()554+ owner := tenant.Policy().Owner555+ member, _ := event.PublicKey(testMemberSecret)556+ setRole(t, tenant, member, "member")557+ receiver := &callbackReceiver{status: http.StatusOK}558+ service := httptest.NewTLSServer(receiver)559+ defer service.Close()560+ tenant.callbackClient = service.Client()561+ registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`)562+ id := registered["id"].(string)563+ now := time.Now().Unix()564+ note := signedEvent(t, testOwnerSecret, 1, now, nil, "hidden later")565+ if err := publishAs(t, tenant, note); err != nil {566+ t.Fatal(err)567+ }568+ queued := pendingCallbackIntents(t, tenant, id, note.ID)569+ if len(queued) != 1 {570+ t.Fatalf("queued = %+v", queued)571+ }572+ // Hidden between arrival and delivery: nothing is sent.573+ if _, err := tenant.Execute(ctx, owner, "banevent", []json.RawMessage{json.RawMessage(strconv.Quote(note.ID))}); err != nil {574+ t.Fatal(err)575+ }576+ if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != 0 {577+ t.Fatalf("hidden event delivered: %v %d", err, receiver.count())578+ }579+ // A key that lost its membership has its callback paused.580+ if _, err := tenant.Execute(ctx, owner, "removemember", []json.RawMessage{json.RawMessage(strconv.Quote(member))}); err != nil {581+ t.Fatal(err)582+ }583+ if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != 0 {584+ t.Fatalf("former member delivered: %v %d", err, receiver.count())585+ }586+ var paused int587+ var status string588+ if err := tenant.store.DB().QueryRowContext(ctx, "SELECT paused,last_status FROM callbacks WHERE id=?", id).Scan(&paused, &status); err != nil || paused != 1 || status != "paused: the key is no longer a member" {589+ t.Fatalf("former member callback: paused=%d status=%q %v", paused, status, err)590+ }591+}592+593+func TestMCPCallbackTools(t *testing.T) {594+ app, tenant := testTenant(t)595+ member, _ := event.PublicKey(testMemberSecret)596+ setRole(t, tenant, member, "member")597+ call := func(name string, arguments map[string]any, secret string) (map[string]any, bool) {598+ t.Helper()599+ w, response := mcpCall{method: "tools/call", name: name, arguments: arguments, sign: true, secret: secret}.do(t, app, "/mcp")600+ if w.Code != http.StatusOK {601+ t.Fatalf("%s: %d %s", name, w.Code, w.Body.String())602+ }603+ return mcpToolResult(t, response)604+ }605+ structured := func(result map[string]any) map[string]any {606+ value, _ := result["structuredContent"].(map[string]any)607+ inner, _ := value["result"].(map[string]any)608+ if inner == nil {609+ return value610+ }611+ return inner612+ }613+ text := func(result map[string]any) string {614+ content, _ := result["content"].([]any)615+ if len(content) == 0 {616+ return ""617+ }618+ return content[0].(map[string]any)["text"].(string)619+ }620+ _, response := mcpCall{method: "tools/list", sign: true}.do(t, app, "/mcp")621+ names := map[string]bool{}622+ for _, tool := range response.Result.(map[string]any)["tools"].([]any) {623+ names[tool.(map[string]any)["name"].(string)] = true624+ }625+ for _, want := range []string{"list_callbacks", "add_callback", "remove_callback", "pause_callback", "resume_callback"} {626+ if !names[want] {627+ t.Fatalf("missing tool %s", want)628+ }629+ }630+ filter := map[string]any{"kinds": []any{1621.0, 1111.0}, "#p": []any{member}}631+ result, isError := call("add_callback", map[string]any{"url": "https://hooks.example/wake", "filter": filter}, strings.Repeat("7", 64))632+ if !isError || !strings.Contains(text(result), "restricted") {633+ t.Fatalf("add_callback as guest: %v %s", isError, text(result))634+ }635+ result, isError = call("add_callback", map[string]any{"url": "http://hooks.example/wake", "filter": filter}, testMemberSecret)636+ if !isError || !strings.Contains(text(result), "https") {637+ t.Fatalf("add_callback with http: %v %s", isError, text(result))638+ }639+ result, isError = call("add_callback", map[string]any{"url": "https://hooks.example/wake", "filter": filter, "secret": "a-shared-secret-of-my-own"}, testMemberSecret)640+ added := structured(result)641+ if isError || added["secret"] != "a-shared-secret-of-my-own" || added["host"] != "hooks.example" || added["owner"] != member {642+ t.Fatalf("add_callback: %s", text(result))643+ }644+ id, _ := added["id"].(string)645+ result, isError = call("list_callbacks", map[string]any{}, testMemberSecret)646+ listed, _ := result["structuredContent"].(map[string]any)["result"].([]any)647+ if isError || len(listed) != 1 || listed[0].(map[string]any)["id"] != id || listed[0].(map[string]any)["paused"] != false {648+ t.Fatalf("list_callbacks: %s", text(result))649+ }650+ if strings.Contains(text(result), "a-shared-secret-of-my-own") {651+ t.Fatal("list_callbacks repeats the secret")652+ }653+ result, isError = call("pause_callback", map[string]any{"id": id}, "")654+ if isError || structured(result)["paused"] != true {655+ t.Fatalf("pause_callback as owner: %s", text(result))656+ }657+ result, isError = call("resume_callback", map[string]any{"id": id}, testMemberSecret)658+ if isError || structured(result)["paused"] != false {659+ t.Fatalf("resume_callback: %s", text(result))660+ }661+ result, isError = call("remove_callback", map[string]any{"id": id}, strings.Repeat("7", 64))662+ if !isError {663+ t.Fatalf("remove_callback as guest: %s", text(result))664+ }665+ result, isError = call("remove_callback", map[string]any{"id": id}, testMemberSecret)666+ if isError || structured(result)["removed"] != true {667+ t.Fatalf("remove_callback: %s", text(result))668+ }669+ result, _ = call("list_callbacks", map[string]any{}, "")670+ if listed, _ := result["structuredContent"].(map[string]any)["result"].([]any); len(listed) != 0 {671+ t.Fatalf("list_callbacks after remove: %s", text(result))672+ }673+ r := httptest.NewRequest(http.MethodGet, "http://relay.test/llms.txt", nil)674+ w := httptest.NewRecorder()675+ app.ServeHTTP(w, r)676+ if !strings.Contains(w.Body.String(), "list_callbacks") || !strings.Contains(w.Body.String(), "add_callback") {677+ t.Fatalf("llms.txt lacks the callback tools: %s", w.Body.String())678+ }679+}@@ -44,6 +44,12 @@ func (t *Tenant) executeManagement(ctx context.Context, actor, method string, pa
4444 if publicManagementMethod(method) {4545 return t.executePublicManagement(ctx, actor, method, params)4646 }47+ if callbackMethod(method) {48+ // Members and agents manage their own callbacks, so the role rule49+ // lives with the methods.50+ result, err = t.callbackExecute(ctx, actor, method, params)51+ return result, true, err52+ }4753 if !managementAdapterMethod(method) {4854 return nil, false, nil4955 }@@ -132,7 +138,7 @@ func publicManagementMethod(method string) bool {
132138 // ManagementMethods is the source-compatible NIP-86 registry. The hosted133139 // lease, trial, fuel, and entitlement controls are intentionally absent.134140 func ManagementMethods() []string {135- return []string{"supportedmethods", "listaudit", "stats", "getpolicy", "setpolicy", "listviews", "banpubkey", "setblockedwords", "allowpubkey", "setmember", "unrulepubkey", "removemember", "listbannedpubkeys", "listallowedpubkeys", "listmembers", "listpeople", "createinvite", "listinvites", "revokeinvite", "listclaims", "listlisthistory", "restorelist", "createclaim", "deleteclaim", "removesubtree", "banevent", "allowevent", "listeventsneedingmoderation", "blockip", "unblockip", "listblockedips", "listreports", "resolvereport", "exportconfig", "importconfig", "deleterelay", "listblobs", "listsites", "deleteblob", "deleteevent", "listrecentevents", "searchevents", "pinevent", "unpinevent", "listpins", "allowkind", "disallowkind", "unrulekind", "storagestats", "gitstorage", "setretention", "listretention", "purgekind", "listallowedkinds", "listblockedkinds", "notifytest", "resetrules", "listpresets", "listconnectiontemplates", "listconnections", "setconnections", "applypreset", "forkrelay", "pullfrom", "pullstatus", "listjobs", "deliverystatus", "addjob", "removejob", "runjob", "backfill", "transferowner", "listdumps", "deletedump", "dumpnow", "backupnow", "listbackups", "deletebackup", "setsuccession", "clearsuccession", "successionstatus", "changerelayname", "changerelaydescription", "changerelayicon", "adddomain", "setdomainsite", "checkdomain", "removedomain", "listdomains", "listagents", "pauseagent", "resumeagent", "revokeagent", "pauseallagents", "resumeallagents"}141+ return []string{"supportedmethods", "listaudit", "stats", "getpolicy", "setpolicy", "listviews", "banpubkey", "setblockedwords", "allowpubkey", "setmember", "unrulepubkey", "removemember", "listbannedpubkeys", "listallowedpubkeys", "listmembers", "listpeople", "createinvite", "listinvites", "revokeinvite", "listclaims", "listlisthistory", "restorelist", "createclaim", "deleteclaim", "removesubtree", "banevent", "allowevent", "listeventsneedingmoderation", "blockip", "unblockip", "listblockedips", "listreports", "resolvereport", "exportconfig", "importconfig", "deleterelay", "listblobs", "listsites", "deleteblob", "deleteevent", "listrecentevents", "searchevents", "pinevent", "unpinevent", "listpins", "allowkind", "disallowkind", "unrulekind", "storagestats", "gitstorage", "setretention", "listretention", "purgekind", "listallowedkinds", "listblockedkinds", "notifytest", "resetrules", "listpresets", "listconnectiontemplates", "listconnections", "setconnections", "applypreset", "forkrelay", "pullfrom", "pullstatus", "listjobs", "deliverystatus", "addjob", "removejob", "runjob", "backfill", "transferowner", "listdumps", "deletedump", "dumpnow", "backupnow", "listbackups", "deletebackup", "setsuccession", "clearsuccession", "successionstatus", "changerelayname", "changerelaydescription", "changerelayicon", "adddomain", "setdomainsite", "checkdomain", "removedomain", "listdomains", "listagents", "pauseagent", "resumeagent", "revokeagent", "pauseallagents", "resumeallagents", "listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"}136142 }137143 138144 func managementRoleAllows(method, role string) bool {@@ -8,7 +8,7 @@ import (
88 "github.com/prometheus/client_golang/prometheus"99 )1010 11-var queueMetricKinds = []string{"delivery", "push", "replication", "webhook", "job", "records-projection", "site-mirror", "git-metadata", "catalog-owner", "callback", "delivery-discovery", "notification-delivery", "notification-broadcast", "notification", "other"}11+var queueMetricKinds = []string{"delivery", "push", "replication", "webhook", "job", "records-projection", "site-mirror", "git-metadata", "catalog-owner", "callback", "delivery-discovery", "notification-delivery", "notification-broadcast", "notification", "callback-delivery", "other"}1212 1313 type queueCollector struct {1414 app *App@@ -43,6 +43,9 @@ func (t *Tenant) initServices(ctx context.Context) error {
4343 if err := t.initPush(ctx); err != nil {4444 return err4545 }46+ if err := t.initCallbacks(ctx); err != nil {47+ return err48+ }4649 t.blobs, err = blob.New(ctx, blob.Config{4750 Root: t.meta.Paths.Root, PublicURL: t.publicURL, Store: t.store, Authorize: t.authorizeBlob,4851 Limits: func() blob.Limits {@@ -506,11 +509,12 @@ func (t *Tenant) workHandlers() map[string]work.Handler {
506509 handlers[kind] = handler507510 }508511 handlers[notificationPush] = t.handleNotificationPush512+ handlers[callbackDelivery] = t.handleCallbackDelivery509513 return handlers510514 }511515 512516 func (t *Tenant) releaseGit(ctx context.Context, id string) error {513- _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {517+ released, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {514518 var promoted event.Event515519 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {516520 result, err := tx.ExecContext(ctx, "DELETE FROM pending_events WHERE id=?", id)@@ -537,6 +541,11 @@ func (t *Tenant) releaseGit(ctx context.Context, id string) error {
537541 })538542 return promoted, err539543 })544+ if err == nil && released.ID != "" {545+ // A repository state is hidden until its objects arrive, so agents546+ // that wait for pushes are woken here rather than at publish time.547+ t.notifyCallbacks(ctx, released)548+ }540549 return err541550 }542551 @@ -674,6 +683,7 @@ func (t *Tenant) commitImported(ctx context.Context, e event.Event, origin repli
674683 _, err := t.store.Save(ctx, e, opts)675684 if err == nil {676685 t.notifyDevices(ctx, e)686+ t.notifyCallbacks(ctx, e)677687 t.notifyWikiMerge(ctx, e)678688 }679689 return err@@ -41,37 +41,41 @@ type tenantConfig struct {
4141 }4242 4343 type Tenant struct {44- app *App45- meta catalog.Tenant46- store *storage.Store47- community *community.Service48- gate *gates.Gate49- router *relay.Router50- auth *auth.Validator51- blobs *blob.Service52- sites *sites.Service53- records *records.Service54- config *configport.ConfigStore55- replication *replication.Service56- git *gitrelay.GitRelay57- ui *webui.App58- mcp *mcp.Server59- schedulerWake chan struct{}60- workCtx context.Context61- workCancel context.CancelFunc62- workWG sync.WaitGroup63- workErrMu sync.Mutex64- workErr error65- publicURL string66- mu sync.RWMutex67- policyWrite sync.Mutex68- policy policy.Policy69- maintenance maintenanceGate70- gitLegacy *replication.LegacyCache71- pushMu sync.Mutex72- pushVAPID *webpush.Keys73- pushClient *http.Client74- pushRecent map[string]time.Time44+ app *App45+ meta catalog.Tenant46+ store *storage.Store47+ community *community.Service48+ gate *gates.Gate49+ router *relay.Router50+ auth *auth.Validator51+ blobs *blob.Service52+ sites *sites.Service53+ records *records.Service54+ config *configport.ConfigStore55+ replication *replication.Service56+ git *gitrelay.GitRelay57+ ui *webui.App58+ mcp *mcp.Server59+ schedulerWake chan struct{}60+ workCtx context.Context61+ workCancel context.CancelFunc62+ workWG sync.WaitGroup63+ workErrMu sync.Mutex64+ workErr error65+ publicURL string66+ mu sync.RWMutex67+ policyWrite sync.Mutex68+ policy policy.Policy69+ maintenance maintenanceGate70+ gitLegacy *replication.LegacyCache71+ pushMu sync.Mutex72+ pushVAPID *webpush.Keys73+ pushClient *http.Client74+ pushRecent map[string]time.Time75+ callbackMu sync.Mutex76+ callbackIndex map[int][]callbackRecord77+ callbackLocks map[string]*sync.Mutex78+ callbackClient *http.Client7579 }7680 7781 func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) {@@ -381,6 +385,7 @@ func (t *Tenant) Publish(ctx context.Context, e event.Event, s relay.Session) (s
381385 return "", err382386 }383387 t.notifyDevices(ctx, e)388+ t.notifyCallbacks(ctx, e)384389 t.notifyWikiMerge(ctx, e)385390 // Stage Git metadata before acknowledging the event. This closes the386391 // publish-ACK/receive-pack race: the signed pending refs and hook exist@@ -126,7 +126,10 @@ type Policy struct {
126126 DumpsKeep int `json:"dumpsKeep"`127127 // Rooms caps the chat rooms a tenant may create beside its main group.128128 // Zero switches room creation off.129- Rooms int `json:"rooms"`129+ Rooms int `json:"rooms"`130+ // Callbacks caps the event callbacks one key may register. The owner is131+ // not limited. Zero switches registration off for everyone else.132+ Callbacks int `json:"callbacks"`130133 BlockedKinds []int `json:"-"`131134 AllowedKinds []int `json:"-"`132135 Retention []RetentionRule `json:"-"`@@ -135,12 +138,16 @@ type Policy struct {
135138 // DefaultRooms is the room allowance applied when a policy does not name one.136139 const DefaultRooms = 64137140 138-// UnmarshalJSON keeps the room allowance and the long task switch at their139-// defaults when a stored policy predates the fields, so existing tenants do140-// not lose room creation or job traffic.141+// DefaultCallbacks is the callback allowance per key when a policy does not142+// name one.143+const DefaultCallbacks = 4144+145+// UnmarshalJSON keeps the room and callback allowances and the long task146+// switch at their defaults when a stored policy predates the fields, so147+// existing tenants do not lose room creation, callbacks or job traffic.141148 func (p *Policy) UnmarshalJSON(data []byte) error {142149 type plain Policy143- decoded := plain{Rooms: DefaultRooms}150+ decoded := plain{Rooms: DefaultRooms, Callbacks: DefaultCallbacks}144151 decoded.Features.Jobs = true145152 if err := json.Unmarshal(data, &decoded); err != nil {146153 return err@@ -150,7 +157,7 @@ func (p *Policy) UnmarshalJSON(data []byte) error {
150157 }151158 152159 func Defaults(owner string) Policy {153- return Policy{Owner: owner, CustomHosts: []CustomHost{}, Writes: "open", Reads: "open", DirectoryPublic: true, MaxFuture: 900, Dumps: "off", DumpsKeep: 7, Rooms: DefaultRooms,160+ return Policy{Owner: owner, CustomHosts: []CustomHost{}, Writes: "open", Reads: "open", DirectoryPublic: true, MaxFuture: 900, Dumps: "off", DumpsKeep: 7, Rooms: DefaultRooms, Callbacks: DefaultCallbacks,154161 Features: Features{Search: "prose", Sync: true, Count: true, Discovery: true, Names: true, Files: true, Pages: true, Signer: true, Sites: Sites{Enabled: true, Mirror: true}, Jobs: true},155162 Notify: Notify{}, Views: map[string]string{}, Tags: []string{}, LanguageTags: []string{}, RelayCountries: []string{}, OpenKinds: []int{}, BlockedWords: []string{}, PushCallbacks: []string{}, Delivery: Delivery{}}156163 }@@ -212,7 +219,7 @@ func Validate(p Policy) error {
212219 }213220 }214221 }215- if p.ReportThreshold < 0 || p.MinPow < 0 || p.MaxFuture < 0 || p.MemberInvites.Depth < 0 || p.MemberInvites.Quota < 0 || p.Rooms < 0 {222+ if p.ReportThreshold < 0 || p.MinPow < 0 || p.MaxFuture < 0 || p.MemberInvites.Depth < 0 || p.MemberInvites.Quota < 0 || p.Rooms < 0 || p.Callbacks < 0 {216223 return errors.New("policy: numeric values cannot be negative")217224 }218225 if p.Succession != nil && (p.Succession.Heir == "" || p.Succession.AfterDays <= 0) {@@ -236,10 +236,10 @@ func latencyBuckets() []float64 {
236236 return []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10}237237 }238238 239-var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browsejobs": {}, "browsejob": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}}240-var outcomeNames = map[string]struct{}{"success": {}, "ok": {}, "invalid": {}, "unauthorized": {}, "timeout": {}, "busy": {}, "closed": {}, "error": {}}239+var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}}240+var outcomeNames = map[string]struct{}{"success": {}, "ok": {}, "invalid": {}, "unauthorized": {}, "timeout": {}, "busy": {}, "closed": {}, "error": {}, "paused": {}}241241 var queueNames = map[string]struct{}{"inbox": {}, "outbox": {}, "write": {}, "delivery": {}}242-var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification-push": {}, "notification": {}}242+var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification-push": {}, "notification": {}, "callback-delivery": {}}243243 244244 func bounded(value string, allowed map[string]struct{}) string {245245 if _, ok := allowed[value]; ok {