diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 76f8166..adbf7e2 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -2668,7 +2668,7 @@ func CountWebhooks(db DBTX, userDID string) (int, error) { func ListWebhooks(db DBTX, userDID string) ([]Webhook, error) { rows, err := db.Query(` SELECT id, user_did, url, secret, triggers, created_at, last_fired_at - FROM webhooks WHERE user_did = ? ORDER BY created_at ASC + FROM webhooks WHERE user_did = ? ORDER BY created_at ASC, id ASC `, userDID) if err != nil { return nil, err @@ -2743,6 +2743,7 @@ func GetWebhooksForUser(db DBTX, userDID string) ([]Webhook, error) { rows, err := db.Query(` SELECT id, user_did, url, secret, triggers, created_at, last_fired_at FROM webhooks WHERE user_did = ? + ORDER BY created_at ASC, id ASC `, userDID) if err != nil { return nil, err diff --git a/pkg/appview/handlers/webhooks.go b/pkg/appview/handlers/webhooks.go index 94a9bc0..e23b581 100644 --- a/pkg/appview/handlers/webhooks.go +++ b/pkg/appview/handlers/webhooks.go @@ -146,11 +146,10 @@ func (h *AddWebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // Trigger bitmask enforcement: free users can only set TriggerFirst, - // TriggerPush, and TriggerQuota. Check only the flag bits — the packed - // threshold percent (bits 8-15) must not be compared as a "trigger". - freeMask := webhooks.TriggerFirst | webhooks.TriggerPush | webhooks.TriggerQuota - if !limits.AllTriggers && webhooks.TriggerFlags(triggers) & ^freeMask != 0 { + // Trigger bitmask enforcement: free users can only set the free trigger + // flags. Check only the flag bits — the packed threshold percent (bits 8-15) + // must not be compared as a "trigger". Same mask the dispatch gate uses. + if !limits.AllTriggers && webhooks.TriggerFlags(triggers) & ^webhooks.FreeTriggerMask != 0 { h.renderWebhookError(w, "Additional trigger types require a paid plan") return } diff --git a/pkg/appview/server.go b/pkg/appview/server.go index ca6e256..cc6ca6a 100644 --- a/pkg/appview/server.go +++ b/pkg/appview/server.go @@ -307,7 +307,10 @@ func NewAppViewServer(cfg *Config, branding *BrandingOverrides) (*AppViewServer, FaviconURL: cfg.Server.BaseURL + "/favicon-96x96.png", RegistryDomains: cfg.Server.RegistryDomains, } - s.WebhookDispatcher = webhooks.NewDispatcher(s.Database, appviewMeta) + // Gate dispatch on the same entitlement the creation form uses, so webhooks + // stop firing paid behavior when a user loses entitlement (switches to a + // self-hosted hold or downgrades) after creating them. + s.WebhookDispatcher = webhooks.NewDispatcher(s.Database, appviewMeta, s.BillingManager.GetWebhookLimits) middleware.SetGlobalWebhookDispatcher(s.WebhookDispatcher) // Initialize Jetstream workers diff --git a/pkg/appview/webhooks/dispatch.go b/pkg/appview/webhooks/dispatch.go index 5486e2f..5714bab 100644 --- a/pkg/appview/webhooks/dispatch.go +++ b/pkg/appview/webhooks/dispatch.go @@ -11,6 +11,7 @@ import ( "log/slog" "math/rand/v2" "net/http" + "slices" "strings" "time" @@ -19,6 +20,13 @@ import ( "atcr.io/pkg/atproto" ) +// WebhookLimiter reports a user's current webhook allowance: the max number of +// webhooks (-1 = unlimited) and whether paid trigger types are allowed. It is +// consulted at dispatch time so losing entitlement (switching to a self-hosted +// hold, or a plan downgrade) immediately caps delivery — the creation-time gate +// alone can't catch a transition after the webhook already exists. +type WebhookLimiter func(userDID string) (maxWebhooks int, allTriggers bool) + // Dispatcher handles webhook delivery for push and scan notifications. // It reads webhooks from the appview DB and delivers payloads // with Discord/Slack formatting and HMAC signing. @@ -26,17 +34,44 @@ type Dispatcher struct { db db.DBTX meta atproto.AppviewMetadata httpClient *http.Client + limits WebhookLimiter } -// NewDispatcher creates a new webhook dispatcher -func NewDispatcher(database db.DBTX, meta atproto.AppviewMetadata) *Dispatcher { +// NewDispatcher creates a new webhook dispatcher. limits may be nil (treated as +// unlimited), but production wires it to the billing manager's GetWebhookLimits. +func NewDispatcher(database db.DBTX, meta atproto.AppviewMetadata, limits WebhookLimiter) *Dispatcher { return &Dispatcher{ db: database, meta: meta, httpClient: http.DefaultClient, + limits: limits, } } +// entitledWebhooks caps a user's webhook list to their current allowance and +// reports whether paid triggers are allowed right now. The list must already be +// ordered oldest-first (GetWebhooksForUser) so the cap keeps the oldest N — the +// same ones the creation-time count gate would have permitted. A nil limiter +// (e.g. tests, billing disabled) means unlimited. +func (d *Dispatcher) entitledWebhooks(userDID string, hooks []db.Webhook) ([]db.Webhook, bool) { + if d.limits == nil { + return hooks, true + } + max, allTriggers := d.limits(userDID) + if max >= 0 && len(hooks) > max { + // Log it: the webhooks still show in settings, so from the user's side + // delivery just stops. This is also the only signal if the limiter itself + // degraded (a billing lookup failure falls back to free-tier limits), + // which would silently demote a paying user mid-dispatch. + slog.Info("Capping webhook delivery to current entitlement", + "userDID", userDID, + "configured", len(hooks), + "allowed", max) + hooks = hooks[:max] + } + return hooks, allTriggers +} + // DispatchForScan fires matching webhooks after a scan record arrives via Jetstream. // previousScan is nil for first-time scans. userHandle is used for payload enrichment. func (d *Dispatcher) DispatchForScan(ctx context.Context, scan, previousScan *db.Scan, userHandle, tag, holdEndpoint string) { @@ -44,6 +79,7 @@ func (d *Dispatcher) DispatchForScan(ctx context.Context, scan, previousScan *db if err != nil || len(webhooks) == 0 { return } + webhooks, allTriggers := d.entitledWebhooks(scan.UserDID, webhooks) isFirst := previousScan == nil isChanged := previousScan != nil && vulnCountsChanged(scan, previousScan) @@ -71,7 +107,11 @@ func (d *Dispatcher) DispatchForScan(ctx context.Context, scan, previousScan *db for _, wh := range webhooks { // Check each trigger condition against bitmask. Read flag bits via // TriggerFlags so packed threshold bits (quota webhooks) don't leak in. + // Drop paid trigger bits when the user isn't currently entitled. flags := TriggerFlags(wh.Triggers) + if !allTriggers { + flags &= FreeTriggerMask + } var triggers []string if flags&TriggerFirst != 0 && isFirst { triggers = append(triggers, "scan:first") @@ -130,6 +170,20 @@ func (d *Dispatcher) DispatchForQuota(ctx context.Context, event storage.QuotaWe if err != nil || len(hooks) == 0 { return } + // Most pushes have no quota webhook configured at all, and the entitlement + // lookup can reach the billing backend, so check cheaply whether there is any + // quota work before paying for it. This preserves the original "do no work on + // the common path" property that fetching the allowance up front would lose. + if !slices.ContainsFunc(hooks, func(wh db.Webhook) bool { + return TriggerFlags(wh.Triggers)&TriggerQuota != 0 + }) { + return + } + + // Cap to the user's current allowance before thresholding. The cap is applied + // over the full list, not the quota subset, because the count limit spans all + // webhook types, matching the creation-time gate. + hooks, _ = d.entitledWebhooks(event.UserDID, hooks) // Filter to quota-trigger webhooks before doing any work; most pushes // will have no quota webhooks configured. @@ -210,6 +264,7 @@ func (d *Dispatcher) DispatchForPush(ctx context.Context, event storage.PushWebh if err != nil || len(webhooks) == 0 { return } + webhooks, _ = d.entitledWebhooks(event.OwnerDID, webhooks) // Fetch star/pull counts for payload enrichment var starCount, pullCount int diff --git a/pkg/appview/webhooks/dispatch_entitlement_test.go b/pkg/appview/webhooks/dispatch_entitlement_test.go new file mode 100644 index 0000000..3c84299 --- /dev/null +++ b/pkg/appview/webhooks/dispatch_entitlement_test.go @@ -0,0 +1,82 @@ +package webhooks + +import ( + "context" + "net/http/httptest" + "testing" + "time" + + "atcr.io/pkg/appview/db" + "atcr.io/pkg/atproto" +) + +// scanForUser builds a minimal first-time scan record for dispatch tests. +func scanForUser(userDID string) *db.Scan { + return &db.Scan{ + UserDID: userDID, + HoldDID: "did:web:hold", + Repository: "app", + ManifestDigest: "sha256:deadbeef", + ScannedAt: time.Now().UTC(), + ScannerVersion: "test", + } +} + +// TestDispatchForScan_EntitlementGate verifies the dispatch-time backstop: +// paid triggers are dropped and the webhook count is capped when the limiter +// reports a non-entitled (e.g. self-hosted / downgraded) user. +func TestDispatchForScan_EntitlementGate(t *testing.T) { + conn, err := db.InitDB(":memory:", db.LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + defer conn.Close() + + const userDID = "did:plc:scanent" + if err := db.UpsertUser(conn, &db.User{ + DID: userDID, Handle: "se.test", PDSEndpoint: "https://pds", LastSeen: time.Now(), + }); err != nil { + t.Fatalf("upsert user: %v", err) + } + + receiver := newFakeReceiver(8) + recvSrv := httptest.NewServer(receiver.handler()) + defer recvSrv.Close() + + // Two webhooks, both with the paid TriggerAll set, created oldest-first. + for i, id := range []string{"wh-old", "wh-new"} { + hook := &db.Webhook{ + ID: id, + UserDID: userDID, + URL: recvSrv.URL, + Triggers: PackTriggers(TriggerFirst|TriggerAll, 0), + CreatedAt: time.Now().UTC().Add(time.Duration(i) * time.Second), + } + if err := db.InsertWebhook(conn, hook); err != nil { + t.Fatalf("insert hook %s: %v", id, err) + } + } + + meta := atproto.AppviewMetadata{ClientShortName: "ATCR", BaseURL: "https://atcr.test"} + + // Free tier: max 1 webhook, no paid triggers. The cap keeps only the oldest + // webhook, and scan:all (paid) is masked out — leaving just scan:first. + free := NewDispatcher(conn, meta, func(string) (int, bool) { return 1, false }) + free.DispatchForScan(context.Background(), scanForUser(userDID), nil, "se.test", "latest", "https://hold") + if !receiver.waitFor(1, 2*time.Second) { + t.Fatalf("free tier: expected 1 delivery (scan:first on oldest hook), got %d", receiver.count()) + } + time.Sleep(150 * time.Millisecond) // allow any erroneous extra deliveries to land + if got := receiver.count(); got != 1 { + t.Fatalf("free tier: expected exactly 1 delivery, got %d", got) + } + + // Entitled: unlimited + all triggers. Both webhooks fire, each delivering + // scan:first AND scan:all = 4 deliveries (regression guard that the gate + // doesn't over-suppress). + entitled := NewDispatcher(conn, meta, func(string) (int, bool) { return -1, true }) + entitled.DispatchForScan(context.Background(), scanForUser(userDID), nil, "se.test", "latest", "https://hold") + if !receiver.waitFor(1+4, 2*time.Second) { + t.Fatalf("entitled: expected 4 more deliveries (2 hooks x scan:first+scan:all), total got %d", receiver.count()) + } +} diff --git a/pkg/appview/webhooks/dispatch_quota_test.go b/pkg/appview/webhooks/dispatch_quota_test.go index 30ebbe4..4a6d950 100644 --- a/pkg/appview/webhooks/dispatch_quota_test.go +++ b/pkg/appview/webhooks/dispatch_quota_test.go @@ -136,7 +136,7 @@ func TestDispatchForQuotaEdgeTriggered(t *testing.T) { t.Fatalf("insert push hook: %v", err) } - d := NewDispatcher(conn, atproto.AppviewMetadata{ClientShortName: "ATCR", BaseURL: "https://atcr.test"}) + d := NewDispatcher(conn, atproto.AppviewMetadata{ClientShortName: "ATCR", BaseURL: "https://atcr.test"}, nil) event := storage.QuotaWebhookEvent{ UserDID: userDID, @@ -252,7 +252,7 @@ func TestDispatchForQuotaUnlimitedHold(t *testing.T) { t.Fatal(err) } - d := NewDispatcher(conn, atproto.AppviewMetadata{}) + d := NewDispatcher(conn, atproto.AppviewMetadata{}, nil) d.DispatchForQuota(context.Background(), storage.QuotaWebhookEvent{ UserDID: userDID, HoldDID: holdSrv.URL, HoldEndpoint: holdSrv.URL, }) diff --git a/pkg/appview/webhooks/types.go b/pkg/appview/webhooks/types.go index b148c06..5754fca 100644 --- a/pkg/appview/webhooks/types.go +++ b/pkg/appview/webhooks/types.go @@ -19,6 +19,11 @@ const ( TriggerQuota = 0x40 // Storage usage crossed the configured threshold (per webhook) ) +// FreeTriggerMask is the set of trigger flags available without a paid plan. +// TriggerAll and TriggerChanged are paid-only. Shared by the webhook creation +// gate and the dispatch-time entitlement backstop so they stay in lockstep. +const FreeTriggerMask = TriggerFirst | TriggerPush | TriggerQuota + const triggerFlagsMask = 0xFF // PackTriggers combines a set of flag bits with a threshold percent (0-100) diff --git a/pkg/billing/billing.go b/pkg/billing/billing.go index b345228..8b10a6d 100644 --- a/pkg/billing/billing.go +++ b/pkg/billing/billing.go @@ -801,8 +801,8 @@ func (m *Manager) getOrCreateCustomer(userDID, userHandle string) (*stripe.Custo } m.customerCacheMu.RUnlock() - // Search Stripe - cust, err := m.findCustomerByDID(userDID) + // Search Stripe. Uses the uncached variant: the cache was just checked. + cust, err := m.searchCustomerByDID(userDID) if err == nil { m.cacheCustomer(userDID, cust) return cust, nil @@ -829,8 +829,33 @@ func (m *Manager) getOrCreateCustomer(userDID, userHandle string) (*stripe.Custo return cust, nil } -// findCustomerByDID searches Stripe for a customer with matching DID metadata. +// findCustomerByDID returns the Stripe customer whose metadata carries this DID, +// preferring the shared customer cache over a live search. +// +// The cache matters because this is now on a hot path: the webhook dispatcher +// consults GetWebhookLimits on every delivery, which lands here via +// GetSubscriptionInfo. Searching Stripe per dispatch would mean a rate-limited +// Search API call for every push and every scan record of every user who has a +// webhook configured. func (m *Manager) findCustomerByDID(userDID string) (*stripe.Customer, error) { + m.customerCacheMu.RLock() + if cached, ok := m.customerCache[userDID]; ok && time.Now().Before(cached.expiresAt) { + m.customerCacheMu.RUnlock() + return cached.customer, nil + } + m.customerCacheMu.RUnlock() + + cust, err := m.searchCustomerByDID(userDID) + if err != nil { + return nil, err + } + m.cacheCustomer(userDID, cust) + return cust, nil +} + +// searchCustomerByDID searches Stripe for a customer with matching DID metadata, +// bypassing the cache. Callers should prefer findCustomerByDID. +func (m *Manager) searchCustomerByDID(userDID string) (*stripe.Customer, error) { // DIDs reaching here are OAuth-validated (the DID grammar forbids quotes), // but escape defensively so the query's safety doesn't silently depend on a // validator several layers away. Stripe search escapes ' and \ with a