diff --git a/claimrelay/claimrelay.go b/claimrelay/claimrelay.go new file mode 100644 index 0000000..2b1ee91 --- /dev/null +++ b/claimrelay/claimrelay.go @@ -0,0 +1,42 @@ +package claimrelay + +import ( + "context" + "encoding/json" + + "github.com/TicketsBot-cloud/common/utils" + "github.com/go-redis/redis/v8" +) + +type TicketClaim struct { + GuildId uint64 `json:"guild_id"` + TicketId int `json:"ticket_id"` + UserId uint64 `json:"user_id"` // who performed the action + Claim bool `json:"claim"` // true = claim, false = unclaim +} + +const key = "tickets:claim" + +func Publish(redis *redis.Client, data TicketClaim) error { + marshalled, err := json.Marshal(data) + if err != nil { + return err + } + return redis.RPush(utils.DefaultContext(), key, string(marshalled)).Err() +} + +func Listen(redis *redis.Client, ch chan TicketClaim) { + for { + res, err := redis.BLPop(context.Background(), 0, key).Result() + if err != nil { + continue + } + + var data TicketClaim + if err := json.Unmarshal([]byte(res[1]), &data); err != nil { + continue + } + + ch <- data + } +} diff --git a/closerelay/closerelay.go b/closerelay/closerelay.go index 9f230d7..a93e47e 100644 --- a/closerelay/closerelay.go +++ b/closerelay/closerelay.go @@ -13,6 +13,11 @@ type TicketClose struct { TicketId int `json:"ticket_id"` UserId uint64 `json:"user_id"` Reason string `json:"reason"` + // CausationId propagates a workflow-run causation when the close is triggered + // by an automation. Empty for user/dashboard-initiated closes. The worker's + // close listener stamps this onto the synthetic worker.Context so downstream + // trigger emissions carry it forward, letting the recursion guard catch loops. + CausationId string `json:"causation_id,omitempty"` } const key = "tickets:close" diff --git a/experiments/experiments.go b/experiments/experiments.go index 89da340..cdb1f31 100644 --- a/experiments/experiments.go +++ b/experiments/experiments.go @@ -17,12 +17,10 @@ import ( type Experiment string const ( - COMPONENTS_V2_STATISTICS Experiment = "COMPONENTS_V2_STATISTICS" - API_BASED_FORM_INPUTS Experiment = "API_BASED_FORM_INPUTS" + API_BASED_FORM_INPUTS Experiment = "API_BASED_FORM_INPUTS" ) var List = []Experiment{ - COMPONENTS_V2_STATISTICS, API_BASED_FORM_INPUTS, } diff --git a/featureflags/attributes.go b/featureflags/attributes.go new file mode 100644 index 0000000..6884fb9 --- /dev/null +++ b/featureflags/attributes.go @@ -0,0 +1,180 @@ +package featureflags + +import ( + "strconv" + + gb "github.com/growthbook/growthbook-golang" +) + +// Identifier attribute names. These are the values a GrowthBook experiment's +// "assignment attribute" must be set to, and they double as the identifier_type +// recorded against an exposure. +const ( + AttrGuild = "guild_id" + AttrUser = "user_id" + AttrDashboardUser = "dashboard_user_id" +) + +// Targeting attributes that are not assignment units. Rules match on these but +// never bucket on them. +const ( + AttrPremiumTier = "premium_tier" + AttrEntitlementSource = "entitlement_source" + AttrShard = "shard" + AttrGuildSize = "guild_size" + AttrStaffTier = "staff_tier" +) + +// Attributes describes the entity a flag is being evaluated for. Build one with +// ForGuild, ForUser or ForDashboardUser so the bucketing unit is always explicit +// rather than inferred, then add optional targeting data with the With helpers. +// +// The old implementation bucketed on guildId % 100 with no per-flag salt, which +// meant every experiment enrolled the same guilds. GrowthBook hashes the primary +// identifier together with a per-flag seed, so cohorts are independent by +// construction. That only holds if the primary identifier is set, hence the +// constructors. +type Attributes struct { + // primary is the attribute name GrowthBook buckets on by default. + primary string + + guildId uint64 + userId uint64 + dashboardUserId uint64 + + premiumTier *int8 + entitlementSource string + shard *int + guildSize *int + staffTier string + + // extra carries attributes that do not warrant a field here. It exists so a + // new targeting dimension does not require a tagged release of common and a + // version bump in every consuming service. + extra map[string]any +} + +// ForGuild buckets on the guild, which is the right unit for most bot features. +func ForGuild(guildId uint64) Attributes { + return Attributes{primary: AttrGuild, guildId: guildId} +} + +// ForUser buckets on the Discord user. The guild is still recorded so rules can +// target both, but assignment follows the user across guilds. +func ForUser(guildId, userId uint64) Attributes { + return Attributes{primary: AttrUser, guildId: guildId, userId: userId} +} + +// ForDashboardUser buckets on the logged-in web user, for dashboard-only rollouts. +func ForDashboardUser(userId uint64) Attributes { + return Attributes{primary: AttrDashboardUser, dashboardUserId: userId} +} + +func (a Attributes) WithPremiumTier(tier int8) Attributes { + a.premiumTier = &tier + return a +} + +func (a Attributes) WithEntitlementSource(source string) Attributes { + a.entitlementSource = source + return a +} + +func (a Attributes) WithShard(shard int) Attributes { + a.shard = &shard + return a +} + +func (a Attributes) WithGuildSize(size int) Attributes { + a.guildSize = &size + return a +} + +// WithGuild attaches a guild ID as a targeting attribute without changing the +// bucketing unit. Use this when the primary unit is something else (e.g. a +// dashboard user) but guild-targeted rules ("Specific servers", "Percentage of +// servers" bucketed on guild) still need to match. +func (a Attributes) WithGuild(guildId uint64) Attributes { + a.guildId = guildId + return a +} + +// WithStaffTier marks the evaluation as being for a member of bot staff, so flags +// can be dogfooded internally before any customer sees them. Expects one of +// "helper", "admin" or "owner"; an empty string is treated as not staff. +// +// Callers must supply this for staff-targeted rules to match. A rule targeting +// staff at a call site that never sets it will silently never fire. +func (a Attributes) WithStaffTier(tier string) Attributes { + a.staffTier = tier + return a +} + +// WithExtra attaches an arbitrary targeting attribute. Prefer a typed helper for +// anything used more than once. +func (a Attributes) WithExtra(key string, value any) Attributes { + // Copy on write: Attributes is passed by value, so mutating a shared map + // would leak between callers that derived from the same base. + next := make(map[string]any, len(a.extra)+1) + for k, v := range a.extra { + next[k] = v + } + next[key] = value + a.extra = next + return a +} + +// toGrowthBook converts to the SDK's attribute map. +// +// Snowflake IDs are always emitted as strings. They exceed the 2^53 integer +// range a JSON number can represent exactly, so passing them as numbers risks +// silent precision loss, which would move a guild between buckets. +func (a Attributes) toGrowthBook() gb.Attributes { + attrs := make(gb.Attributes, 8+len(a.extra)) + + for k, v := range a.extra { + attrs[k] = v + } + + if a.guildId != 0 { + attrs[AttrGuild] = strconv.FormatUint(a.guildId, 10) + } + + if a.userId != 0 { + attrs[AttrUser] = strconv.FormatUint(a.userId, 10) + } + + if a.dashboardUserId != 0 { + attrs[AttrDashboardUser] = strconv.FormatUint(a.dashboardUserId, 10) + } + + if a.premiumTier != nil { + attrs[AttrPremiumTier] = int(*a.premiumTier) + } + + if a.entitlementSource != "" { + attrs[AttrEntitlementSource] = a.entitlementSource + } + + if a.shard != nil { + attrs[AttrShard] = *a.shard + } + + if a.guildSize != nil { + attrs[AttrGuildSize] = *a.guildSize + } + + if a.staffTier != "" { + attrs[AttrStaffTier] = a.staffTier + } + + // GrowthBook falls back to the "id" attribute when a feature does not name an + // assignment attribute of its own, so mirror the chosen primary onto it. This + // makes a flag created in the UI with default settings bucket on the unit the + // caller intended instead of silently not bucketing at all. + if primary, ok := attrs[a.primary]; ok { + attrs["id"] = primary + } + + return attrs +} diff --git a/featureflags/cache.go b/featureflags/cache.go new file mode 100644 index 0000000..77b1d02 --- /dev/null +++ b/featureflags/cache.go @@ -0,0 +1,43 @@ +package featureflags + +import ( + "context" + "errors" + "fmt" + + "github.com/go-redis/redis/v8" +) + +// payloadKey holds the last-known-good GrowthBook payload. +// +// Deliberately has no TTL. A stale ruleset is far better than none: if +// GrowthBook has been down for longer than any TTL we would have picked, a +// restarting worker should still come up with the rules it had before rather +// than with every flag off. +const payloadKey = "featureflags:payload" + +type payloadCache struct { + redis *redis.Client +} + +// get returns the cached payload, or an empty string if nothing is cached. +func (p *payloadCache) get(ctx context.Context) (string, error) { + payload, err := p.redis.Get(ctx, payloadKey).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return "", nil + } + + return "", fmt.Errorf("featureflags: reading %s: %w", payloadKey, err) + } + + return payload, nil +} + +func (p *payloadCache) set(ctx context.Context, payload string) error { + if err := p.redis.Set(ctx, payloadKey, payload, 0).Err(); err != nil { + return fmt.Errorf("featureflags: writing %s: %w", payloadKey, err) + } + + return nil +} diff --git a/featureflags/config.go b/featureflags/config.go new file mode 100644 index 0000000..878d556 --- /dev/null +++ b/featureflags/config.go @@ -0,0 +1,98 @@ +package featureflags + +import ( + "fmt" + "time" +) + +// Config describes how to reach GrowthBook. When both ApiHost and ClientKey are +// empty the whole subsystem degrades to compiled defaults rather than failing, +// so self-hosted deployments and local development work with no configuration. +// Setting only one of the two is rejected by validate as a broken config, +// rather than being treated the same as neither being set. +type Config struct { + // ApiHost is the GrowthBook backend as reached from inside the cluster, so + // http://growthbook-backend.growthbook.svc:3100 rather than the Twingate + // alias. The alias exists for staff browsers; pointing services at it would + // send in-cluster traffic out through the tunnel and back. + ApiHost string `env:"GROWTHBOOK_API_HOST"` + + // ClientKey is an SDK connection key from GrowthBook, not an API key. It only + // grants read access to the flag payload for one environment. + ClientKey string `env:"GROWTHBOOK_CLIENT_KEY"` + + // DecryptionKey is only needed if the SDK connection in GrowthBook has + // encrypted payloads enabled. + DecryptionKey string `env:"GROWTHBOOK_DECRYPTION_KEY"` + + // UseSSE opts into streaming flag updates instead of polling. + // + // Defaults to false because a stock self-hosted GrowthBook does not serve + // streams: the payload response carries no x-sse-support header and + // /sub/ returns 401. Streaming is a GrowthBook Proxy feature, so + // only enable this once the proxy is deployed, otherwise startup fails to + // load any rules and every flag evaluates to off. + UseSSE bool `env:"GROWTHBOOK_USE_SSE" envDefault:"false"` + + // PollInterval is used when UseSSE is false. The SDK sends a conditional + // request and gets a 304 when nothing changed, so a short interval is cheap. + // This is also the upper bound on how long a kill switch takes to reach a + // running process. + PollInterval time.Duration `env:"GROWTHBOOK_POLL_INTERVAL" envDefault:"30s"` + + // LoadTimeout bounds how long startup waits for the first payload before + // falling back to the cached copy in Redis. Keep it short: a process must + // never block on GrowthBook being reachable. + LoadTimeout time.Duration `env:"GROWTHBOOK_LOAD_TIMEOUT" envDefault:"5s"` + + // CacheRefreshInterval is how often the last-known-good payload is written + // to Redis so restarting processes have something to boot from. + CacheRefreshInterval time.Duration `env:"GROWTHBOOK_CACHE_REFRESH_INTERVAL" envDefault:"5m"` +} + +// Enabled reports whether enough configuration is present to talk to GrowthBook. +func (c Config) Enabled() bool { + return c.ApiHost != "" && c.ClientKey != "" +} + +// Attempted reports whether any GrowthBook configuration was supplied at all, +// as opposed to neither ApiHost nor ClientKey being set. +// +// This is deliberately looser than Enabled: it is true for a valid config and +// also for a broken, partially-set one (only one of the two present). Callers +// deciding whether to fail closed on a construction error must use this, not +// Enabled, otherwise a config missing just one of the two env vars (a routine +// secret-rotation or ConfigMap mistake) is indistinguishable from "GrowthBook +// was never configured" and the caller fails open instead of closed. validate +// rejects a partial config outright, so in practice Attempted only needs to +// cover the error path from New; a config that reaches Enabled() has already +// passed validate and cannot be partial. +func (c Config) Attempted() bool { + return c.ApiHost != "" || c.ClientKey != "" +} + +func (c Config) validate() error { + // Half-configured is worse than unconfigured: unconfigured deliberately + // defaults every flag to enabled (see the package doc comment), but a config + // missing just one of ApiHost/ClientKey would otherwise fall into that same + // branch by accident, silently turning every kill switch on. Reject it + // outright instead of letting Enabled() treat "one set" and "none set" the + // same way. + if (c.ApiHost == "") != (c.ClientKey == "") { + return fmt.Errorf("featureflags: partial GrowthBook configuration (ApiHost set=%v, ClientKey set=%v): both must be set, or neither", c.ApiHost != "", c.ClientKey != "") + } + + if c.LoadTimeout <= 0 { + return fmt.Errorf("featureflags: LoadTimeout must be positive, got %s", c.LoadTimeout) + } + + if !c.UseSSE && c.PollInterval <= 0 { + return fmt.Errorf("featureflags: PollInterval must be positive when SSE is disabled, got %s", c.PollInterval) + } + + if c.CacheRefreshInterval <= 0 { + return fmt.Errorf("featureflags: CacheRefreshInterval must be positive, got %s", c.CacheRefreshInterval) + } + + return nil +} diff --git a/featureflags/featureflags.go b/featureflags/featureflags.go new file mode 100644 index 0000000..8eacc3c --- /dev/null +++ b/featureflags/featureflags.go @@ -0,0 +1,446 @@ +// Package featureflags evaluates feature flags and experiments against +// GrowthBook. +// +// Flag definitions live in GrowthBook, not in Go code, so adding a flag needs no +// deploy and no tagged release of this module. Evaluation is entirely in-process: +// the SDK holds the ruleset in memory and performs no network call per call to +// IsEnabled, which is a hard requirement for the worker's event hot path. +// +// The package draws a hard line between two kinds of failure. +// +// GrowthBook simply not being configured (empty ApiHost/ClientKey, which is +// what a self-hosted installation gets by default) is not a failure at all: with +// no ruleset to consult, every flag evaluates to enabled, so a self-hosted +// operator who never sets up GrowthBook is not silently missing shipped +// features behind kill switches. +// +// GrowthBook being configured but unreachable is a real failure, and is where +// this package fails closed instead. If GrowthBook is unreachable at startup +// the last-known-good payload is read from Redis; if that is missing too, every +// flag evaluates to its zero value (off) and the process still boots. An +// unknown flag key against an otherwise-healthy client also evaluates to off. +// This fail-closed behaviour is the entire point of a kill switch: a SaaS +// instance whose GrowthBook connection drops mid-incident must keep treating +// every flag it can no longer resolve as off, not fall back to on. +package featureflags + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + "github.com/go-redis/redis/v8" + gb "github.com/growthbook/growthbook-golang" + "go.uber.org/zap" +) + +// Exposure records that a unit was genuinely enrolled in an experiment. It is +// the input to the assignment table GrowthBook reads when computing results. +type Exposure struct { + // ExperimentKey identifies the experiment. + ExperimentKey string + // VariationId is the zero-based index of the assigned variation. + VariationId int + // IdentifierType is the attribute assignment was keyed on, one of AttrGuild, + // AttrUser or AttrDashboardUser. + IdentifierType string + // Identifier is the value of that attribute. + Identifier string + // FeatureKey is the flag the experiment was reached through, if any. + FeatureKey string +} + +// ExposureRecorder persists exposures. Implementations must be non-blocking: +// Record is called on the evaluation path, which for the worker means inside +// Discord event handling. +type ExposureRecorder interface { + Record(ctx context.Context, exposure Exposure) +} + +// Result is the outcome of evaluating a flag. +type Result struct { + // On is the flag's truthiness, for boolean rollouts and kill switches. + On bool + // Value is the raw value, for multivariate flags. + Value any + // InExperiment reports whether this evaluation enrolled the unit in an + // experiment, as opposed to matching a plain targeting or rollout rule. + InExperiment bool + // VariationId is the assigned variation index when InExperiment is true. + VariationId int + // Source describes which rule produced the value, for debugging. When On was + // defaulted to true because GrowthBook is not configured at all, Source is + // SourceUnconfiguredDefaultOn rather than a GrowthBook rule source, so a + // caller or log can tell "we don't actually know, we defaulted" apart from + // "GrowthBook told us this is genuinely on". + Source string +} + +// SourceUnconfiguredDefaultOn is the Result.Source value used when a flag was +// defaulted to on because GrowthBook is not configured, rather than actually +// evaluated against a ruleset. +const SourceUnconfiguredDefaultOn = "unconfigured-default-on" + +// Client evaluates flags. It is safe for concurrent use. +type Client struct { + cfg Config + logger *zap.Logger + gb *gb.Client + cache *payloadCache + recorder ExposureRecorder + + // unconfigured is true when New was given no ApiHost/ClientKey. It is the + // only condition that flips evaluation from fail-closed (off) to + // default-on: gb is nil in that case, but gb being nil never happens for any + // other reason on a *Client returned without error, so this field exists + // mainly to say so at the call site rather than lean on that invariant. + unconfigured bool + + stop chan struct{} + stopOnce sync.Once + wg sync.WaitGroup +} + +// New builds a Client. +// +// redisClient may be nil, in which case the last-known-good payload is not +// cached and a process starting while GrowthBook is down has no rules to fall +// back to. recorder may be nil, in which case exposures are not recorded and +// experiments cannot be analysed, though flags still evaluate normally. +// +// New never returns an error because GrowthBook is unreachable. It only fails on +// misconfiguration. +func New( + ctx context.Context, + cfg Config, + logger *zap.Logger, + redisClient *redis.Client, + recorder ExposureRecorder, +) (*Client, error) { + if err := cfg.validate(); err != nil { + return nil, err + } + + c := &Client{ + cfg: cfg, + logger: logger, + recorder: recorder, + stop: make(chan struct{}), + } + + if !cfg.Enabled() { + logger.Warn("featureflags: GrowthBook not configured, every flag evaluates to enabled") + c.unconfigured = true + return c, nil + } + + if redisClient != nil { + c.cache = &payloadCache{redis: redisClient} + } + + opts := []gb.ClientOption{ + gb.WithApiHost(cfg.ApiHost), + gb.WithClientKey(cfg.ClientKey), + gb.WithLogger(newSlogAdapter(logger)), + gb.WithExperimentCallback(c.onExperimentViewed), + } + + if cfg.DecryptionKey != "" { + opts = append(opts, gb.WithDecryptionKey(cfg.DecryptionKey)) + } + + if cfg.UseSSE { + opts = append(opts, gb.WithSseDataSource()) + } else { + opts = append(opts, gb.WithPollDataSource(cfg.PollInterval)) + } + + client, err := gb.NewClient(ctx, opts...) + if err != nil { + // A construction failure is a configuration problem, not a transient one. + return nil, fmt.Errorf("featureflags: creating GrowthBook client: %w", err) + } + + c.gb = client + c.primeRuleset(ctx) + + c.wg.Add(1) + go c.cacheRefreshLoop() + + return c, nil +} + +// NewOffline builds a Client that evaluates a fixed ruleset with no network +// access, no Redis and no background work. +// +// featuresJSON is a GrowthBook features map, the inner object of an SDK payload: +// {"my-flag": {"defaultValue": false, "rules": [...]}}. Use this for tests, and +// to ship compiled-in defaults where a deployment has no GrowthBook at all. +func NewOffline( + ctx context.Context, + logger *zap.Logger, + featuresJSON string, + recorder ExposureRecorder, +) (*Client, error) { + c := &Client{ + cfg: Config{}, + logger: logger, + recorder: recorder, + stop: make(chan struct{}), + } + + client, err := gb.NewClient( + ctx, + gb.WithJsonFeatures(featuresJSON), + gb.WithLogger(newSlogAdapter(logger)), + gb.WithExperimentCallback(c.onExperimentViewed), + ) + if err != nil { + return nil, fmt.Errorf("featureflags: creating offline client: %w", err) + } + + c.gb = client + + return c, nil +} + +// primeRuleset gets a usable ruleset in place before New returns. +// +// The data source started asynchronously. Give it a bounded moment to deliver +// the first payload, and if it does not, fall back to the copy in Redis. The +// ordering matters: seeding from Redis unconditionally would race the live +// payload and could overwrite fresh rules with stale ones. +func (c *Client) primeRuleset(ctx context.Context) { + loadCtx, cancel := context.WithTimeout(ctx, c.cfg.LoadTimeout) + defer cancel() + + loadErr := c.gb.EnsureLoaded(loadCtx) + if loadErr == nil { + c.persistPayload(ctx) + return + } + + // Distinguish an exhausted deadline from a real failure. Reporting everything + // as a timeout hides the actual cause, which is usually an unreachable host or + // a rejected client key. + if errors.Is(loadErr, context.DeadlineExceeded) { + c.logger.Warn("featureflags: GrowthBook payload not loaded within timeout, falling back to cache", + zap.Duration("timeout", c.cfg.LoadTimeout), + zap.String("api_host", c.cfg.ApiHost)) + } else { + c.logger.Warn("featureflags: loading GrowthBook payload failed, falling back to cache", + zap.String("api_host", c.cfg.ApiHost), + zap.Error(loadErr)) + } + + if c.cache == nil { + c.logger.Error("featureflags: no Redis cache configured, starting with no rules") + return + } + + payload, err := c.cache.get(ctx) + if err != nil { + c.logger.Error("featureflags: reading cached payload failed, starting with no rules", zap.Error(err)) + return + } + + if payload == "" { + c.logger.Error("featureflags: no cached payload available, starting with no rules") + return + } + + if err := c.gb.UpdateFromApiResponseJSON(payload); err != nil { + c.logger.Error("featureflags: cached payload rejected, starting with no rules", zap.Error(err)) + return + } + + c.logger.Info("featureflags: loaded rules from cached payload") +} + +// cacheRefreshLoop keeps the Redis copy of the payload current so a restarting +// process has something recent to boot from. +func (c *Client) cacheRefreshLoop() { + defer c.wg.Done() + + if c.cache == nil { + return + } + + ticker := time.NewTicker(c.cfg.CacheRefreshInterval) + defer ticker.Stop() + + for { + select { + case <-c.stop: + return + case <-ticker.C: + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + c.persistPayload(ctx) + cancel() + } + } +} + +func (c *Client) persistPayload(ctx context.Context) { + if c.cache == nil { + return + } + + // Fetch through the SDK rather than reconstructing the payload from + // Features(), so what gets cached is byte-compatible with what + // UpdateFromApiResponseJSON expects on the way back in. + resp, err := c.gb.CallFeatureApi(ctx, "") + if err != nil { + c.logger.Warn("featureflags: fetching payload for cache failed", zap.Error(err)) + return + } + + encoded, err := json.Marshal(resp) + if err != nil { + c.logger.Warn("featureflags: encoding payload for cache failed", zap.Error(err)) + return + } + + if err := c.cache.set(ctx, string(encoded)); err != nil { + c.logger.Warn("featureflags: writing cached payload failed", zap.Error(err)) + } +} + +func (c *Client) onExperimentViewed(ctx context.Context, exp *gb.Experiment, result *gb.ExperimentResult, _ any) { + if c.recorder == nil || exp == nil || result == nil { + return + } + + // Only genuine enrolment counts. Plain rollout rules and kill switches reach + // this callback without putting the unit in an experiment, and recording + // those would inflate the assignment table with rows no analysis wants. + if !result.InExperiment || result.HashAttribute == "" || result.HashValue == "" { + return + } + + c.recorder.Record(ctx, Exposure{ + ExperimentKey: exp.Key, + VariationId: result.VariationId, + IdentifierType: result.HashAttribute, + Identifier: result.HashValue, + FeatureKey: result.FeatureId, + }) +} + +// Eval evaluates key for attrs. It never returns an error, but the two ways it +// can fail to get a real answer are handled differently: +// +// - GrowthBook is not configured at all (c is nil, or c was built with no +// ApiHost/ClientKey, the self-hosted default): every flag defaults to +// enabled. Result.On is true and Result.Source is +// SourceUnconfiguredDefaultOn, so a caller or log can tell this apart from a +// genuine GrowthBook "on". +// - GrowthBook is configured but the answer is unavailable for any other +// reason (unreachable at startup with nothing usable in the Redis cache, or +// an unknown flag key against an otherwise-healthy client): the zero Result +// is returned, so callers read that as "off unless explicitly turned on". +// This is the fail-closed behaviour a kill switch exists for, and it holds +// even when GrowthBook was reachable a moment ago and has since dropped out +// mid-incident. +// +// A nil receiver is valid. Services hold this as a package-level variable +// assigned during startup, and the old implementation's settable global could be +// read before it was set and panic. Treating nil the same as "GrowthBook not +// configured" removes that failure mode entirely and keeps both observations of +// the same underlying reality consistent. +func (c *Client) Eval(ctx context.Context, key string, attrs Attributes) Result { + if c == nil || c.unconfigured { + return Result{On: true, Source: SourceUnconfiguredDefaultOn} + } + + if c.gb == nil { + // Not expected to happen outside the unconfigured branch above, but stay + // defensive rather than risk a nil-pointer dereference below. + return Result{} + } + + scoped, err := c.gb.WithAttributes(attrs.toGrowthBook()) + if err != nil { + c.logger.Error("featureflags: applying attributes failed", + zap.String("flag", key), zap.Error(err)) + return Result{} + } + + res := scoped.EvalFeature(ctx, key) + if res == nil { + return Result{} + } + + out := Result{ + On: res.On, + Value: res.Value, + Source: string(res.Source), + } + + if res.ExperimentResult != nil { + out.InExperiment = res.ExperimentResult.InExperiment + out.VariationId = res.ExperimentResult.VariationId + } + + return out +} + +// IsEnabled reports whether a boolean flag is on. This is the direct replacement +// for the old HasFeature. +func (c *Client) IsEnabled(ctx context.Context, key string, attrs Attributes) bool { + return c.Eval(ctx, key, attrs).On +} + +// StringValue returns a string-valued flag, or def if the flag is missing or +// holds a non-string value. +func (c *Client) StringValue(ctx context.Context, key string, attrs Attributes, def string) string { + value, ok := c.Eval(ctx, key, attrs).Value.(string) + if !ok { + return def + } + + return value +} + +// IntValue returns a numeric flag, or def if the flag is missing or holds a +// non-numeric value. +// +// GrowthBook payloads arrive as JSON, so whole numbers decode to float64. Both +// are accepted to keep callers from having to know that. +func (c *Client) IntValue(ctx context.Context, key string, attrs Attributes, def int) int { + switch value := c.Eval(ctx, key, attrs).Value.(type) { + case float64: + return int(value) + case int: + return value + default: + return def + } +} + +// Close stops background work and releases the SDK's connections. A nil receiver +// is a no-op. +func (c *Client) Close() error { + if c == nil { + return nil + } + + c.stopOnce.Do(func() { + close(c.stop) + }) + + c.wg.Wait() + + if c.gb == nil { + return nil + } + + if err := c.gb.Close(); err != nil { + return fmt.Errorf("featureflags: closing GrowthBook client: %w", err) + } + + return nil +} diff --git a/featureflags/featureflags_test.go b/featureflags/featureflags_test.go new file mode 100644 index 0000000..7ef5786 --- /dev/null +++ b/featureflags/featureflags_test.go @@ -0,0 +1,846 @@ +package featureflags + +import ( + "context" + "net/http" + "net/http/httptest" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +// rolloutRuleset defines two flags at the same 10% coverage, keyed on guild_id. +// GrowthBook seeds the rollout hash with the feature key, so the two are expected +// to enrol different guilds. The old implementation used guildId % 100 with no +// seed, which made every flag select the identical cohort. +const rolloutRuleset = `{ + "flag-a": { + "defaultValue": false, + "rules": [{"force": true, "coverage": 0.1, "hashAttribute": "guild_id"}] + }, + "flag-b": { + "defaultValue": false, + "rules": [{"force": true, "coverage": 0.1, "hashAttribute": "guild_id"}] + }, + "flag-none": { + "defaultValue": false, + "rules": [{"force": true, "coverage": 0, "hashAttribute": "guild_id"}] + }, + "flag-all": { + "defaultValue": false, + "rules": [{"force": true, "coverage": 1, "hashAttribute": "guild_id"}] + } +}` + +func testLogger(t *testing.T) *zap.Logger { + t.Helper() + return zap.NewNop() +} + +// snowflakes builds IDs shaped like real Discord snowflakes: a millisecond +// timestamp in the high bits, then worker, process and a sequence counter in the +// low 22 bits. The low bits are the reason a plain modulo is not a fair hash. +func snowflakes(n int) []uint64 { + ids := make([]uint64, 0, n) + base := uint64(1_700_000_000_000) + + for i := 0; i < n; i++ { + ms := base + uint64(i*37) + worker := uint64(i % 8) + process := uint64(i % 4) + seq := uint64(i % 4096) + ids = append(ids, (ms<<22)|(worker<<17)|(process<<12)|seq) + } + + return ids +} + +func TestConfigEnabled(t *testing.T) { + tests := []struct { + name string + cfg Config + expected bool + }{ + {"both set", Config{ApiHost: "https://gb", ClientKey: "sdk-1"}, true}, + {"host missing", Config{ClientKey: "sdk-1"}, false}, + {"key missing", Config{ApiHost: "https://gb"}, false}, + {"empty", Config{}, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.expected, tt.cfg.Enabled()) + }) + } +} + +func TestConfigValidate(t *testing.T) { + valid := Config{LoadTimeout: time.Second, PollInterval: time.Minute, CacheRefreshInterval: time.Minute, UseSSE: true} + + tests := []struct { + name string + mutate func(Config) Config + wantErr bool + }{ + {"valid", func(c Config) Config { return c }, false}, + {"zero load timeout", func(c Config) Config { c.LoadTimeout = 0; return c }, true}, + {"negative load timeout", func(c Config) Config { c.LoadTimeout = -time.Second; return c }, true}, + {"zero cache refresh", func(c Config) Config { c.CacheRefreshInterval = 0; return c }, true}, + {"poll interval irrelevant while SSE on", func(c Config) Config { c.PollInterval = 0; return c }, false}, + {"poll interval required without SSE", func(c Config) Config { c.UseSSE = false; c.PollInterval = 0; return c }, true}, + {"only ApiHost set", func(c Config) Config { c.ApiHost = "https://gb"; return c }, true}, + {"only ClientKey set", func(c Config) Config { c.ClientKey = "sdk-1"; return c }, true}, + {"both ApiHost and ClientKey set", func(c Config) Config { c.ApiHost = "https://gb"; c.ClientKey = "sdk-1"; return c }, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.mutate(valid).validate() + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + }) + } +} + +func TestConfigAttempted(t *testing.T) { + tests := []struct { + name string + cfg Config + expected bool + }{ + {"both set", Config{ApiHost: "https://gb", ClientKey: "sdk-1"}, true}, + {"host only", Config{ApiHost: "https://gb"}, true}, + {"key only", Config{ClientKey: "sdk-1"}, true}, + {"empty", Config{}, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.expected, tt.cfg.Attempted()) + }) + } +} + +// TestNewRejectsPartialGrowthBookConfig is the regression test for the HIGH +// finding: a config with only one of ApiHost/ClientKey set used to fall +// through New's `!cfg.Enabled()` check exactly like a genuinely unconfigured +// deployment, silently defaulting every 202608_FEATURE_* kill switch to on. It +// must now be rejected as an error from New, the same as any other invalid +// Config, and each case must also report Attempted() as true - that is the +// exact condition dash-api's and the worker's main.go now gate their +// NewOffline fail-closed fallback on, so this also demonstrates that gate +// would fire for this error rather than leaving the fallback un-triggered. +func TestNewRejectsPartialGrowthBookConfig(t *testing.T) { + base := Config{LoadTimeout: time.Second, PollInterval: time.Minute, CacheRefreshInterval: time.Minute} + + tests := []struct { + name string + mutate func(Config) Config + }{ + {"only ApiHost set", func(c Config) Config { c.ApiHost = "https://gb"; return c }}, + {"only ClientKey set", func(c Config) Config { c.ClientKey = "sdk-1"; return c }}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := tt.mutate(base) + + client, err := New(context.Background(), cfg, testLogger(t), nil, nil) + require.Error(t, err) + require.Nil(t, client) + + // The gate in main.go must route this error to NewOffline, not to a + // silently-open client, so it must key on Attempted (true here) rather + // than Enabled (false here, since the config is partial). + require.True(t, cfg.Attempted()) + require.False(t, cfg.Enabled()) + }) + } +} + +func TestAttributesSnowflakesStayExactStrings(t *testing.T) { + // Above 2^53 a float64 can no longer represent consecutive integers, so a + // snowflake passed as a number would be rounded and could change bucket. + const guildId uint64 = 1328073426023219221 + require.Greater(t, guildId, uint64(1)<<53) + + attrs := ForGuild(guildId).toGrowthBook() + + require.Equal(t, "1328073426023219221", attrs[AttrGuild]) + require.Equal(t, "1328073426023219221", attrs["id"]) + require.IsType(t, "", attrs[AttrGuild]) +} + +func TestAttributesPrimaryUnit(t *testing.T) { + tests := []struct { + name string + attrs Attributes + wantId string + wantPresent []string + wantAbsent []string + }{ + { + name: "guild", + attrs: ForGuild(10), + wantId: "10", + wantPresent: []string{AttrGuild}, + wantAbsent: []string{AttrUser, AttrDashboardUser}, + }, + { + name: "user carries guild but buckets on user", + attrs: ForUser(10, 20), + wantId: "20", + wantPresent: []string{AttrGuild, AttrUser}, + wantAbsent: []string{AttrDashboardUser}, + }, + { + name: "dashboard user", + attrs: ForDashboardUser(30), + wantId: "30", + wantPresent: []string{AttrDashboardUser}, + wantAbsent: []string{AttrGuild, AttrUser}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tt.attrs.toGrowthBook() + require.Equal(t, tt.wantId, got["id"]) + + for _, key := range tt.wantPresent { + require.Contains(t, got, key) + } + for _, key := range tt.wantAbsent { + require.NotContains(t, got, key) + } + }) + } +} + +func TestAttributesOptionalTargeting(t *testing.T) { + attrs := ForGuild(1). + WithPremiumTier(1). + WithEntitlementSource("patreon"). + WithShard(7). + WithGuildSize(4200). + WithExtra("region", "eu"). + toGrowthBook() + + require.Equal(t, 1, attrs["premium_tier"]) + require.Equal(t, "patreon", attrs["entitlement_source"]) + require.Equal(t, 7, attrs["shard"]) + require.Equal(t, 4200, attrs["guild_size"]) + require.Equal(t, "eu", attrs["region"]) +} + +func TestAttributesStaffTier(t *testing.T) { + attrs := ForDashboardUser(1).WithStaffTier("admin").toGrowthBook() + require.Equal(t, "admin", attrs[AttrStaffTier]) + + // Empty means not staff, and must be absent rather than "" so a rule matching + // on presence does not fire for everyone. + require.NotContains(t, ForDashboardUser(1).WithStaffTier("").toGrowthBook(), AttrStaffTier) +} + +// WithGuild lets a dashboard-user evaluation carry a guild ID, so "Specific +// servers" and "Percentage of servers" rules can match on a dashboard flag. It +// must not change which attribute is the bucketing unit. +func TestAttributesWithGuildOnDashboardUser(t *testing.T) { + attrs := ForDashboardUser(30).WithGuild(10).toGrowthBook() + + require.Equal(t, "30", attrs[AttrDashboardUser]) + require.Equal(t, "10", attrs[AttrGuild]) + + // The bucketing fallback must still mirror the primary unit, not the guild + // that was merely attached for targeting. + require.Equal(t, "30", attrs["id"]) +} + +// Staff targeting is the one attribute a caller must remember to supply. Prove a +// staff rule does not match when it is absent, since a silent non-match is the +// failure mode. +func TestStaffRuleRequiresStaffAttribute(t *testing.T) { + const ruleset = `{ + "staff-only": { + "defaultValue": false, + "rules": [{ + "condition": {"staff_tier": {"$in": ["helper", "admin", "owner"]}}, + "force": true + }] + } + }` + + client, err := NewOffline(context.Background(), testLogger(t), ruleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + + require.True(t, client.IsEnabled(ctx, "staff-only", ForDashboardUser(1).WithStaffTier("helper"))) + require.False(t, client.IsEnabled(ctx, "staff-only", ForDashboardUser(1))) +} + +func TestAttributesOmitsUnsetOptionalTargeting(t *testing.T) { + // Absent must differ from zero: a guild with no premium must not match a + // premium_tier == 0 rule, which is the Premium tier. + attrs := ForGuild(1).toGrowthBook() + + require.NotContains(t, attrs, "premium_tier") + require.NotContains(t, attrs, "shard") + require.NotContains(t, attrs, "guild_size") + require.NotContains(t, attrs, "entitlement_source") +} + +func TestAttributesWithExtraDoesNotMutateBase(t *testing.T) { + base := ForGuild(1).WithExtra("a", 1) + first := base.WithExtra("b", 2) + second := base.WithExtra("c", 3) + + require.NotContains(t, base.toGrowthBook(), "b") + require.NotContains(t, first.toGrowthBook(), "c") + require.NotContains(t, second.toGrowthBook(), "b") + require.Contains(t, first.toGrowthBook(), "b") + require.Contains(t, second.toGrowthBook(), "c") +} + +// A client with no GrowthBook configured must evaluate every boolean flag as +// enabled and never fail. This is the self-hosted and local-development path, +// and is the behaviour that distinguishes "not configured" from "configured but +// unreachable", which must still fail closed (see +// TestFeatureFlagFailureModesDistinguishUnconfiguredFromUnreachable). A +// multivariate flag still has no explicit value to offer, so StringValue and +// IntValue keep returning the caller's fallback. +func TestUnconfiguredClientDefaultsEveryFlagOn(t *testing.T) { + client, err := New( + context.Background(), + Config{LoadTimeout: time.Second, CacheRefreshInterval: time.Minute, UseSSE: true}, + testLogger(t), + nil, + nil, + ) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + attrs := ForGuild(1) + + require.True(t, client.IsEnabled(ctx, "anything", attrs)) + require.Equal(t, "fallback", client.StringValue(ctx, "anything", attrs, "fallback")) + require.Equal(t, 42, client.IntValue(ctx, "anything", attrs, 42)) + require.Equal(t, Result{On: true, Source: SourceUnconfiguredDefaultOn}, client.Eval(ctx, "anything", attrs)) +} + +// TestFeatureFlagFailureModesDistinguishUnconfiguredFromUnreachable is the +// regression test for scoping the "default every flag on" behaviour narrowly to +// "GrowthBook is not configured". It must not bleed into any of the other ways +// evaluation can fail to get a real answer: an unreachable backend with an empty +// cache, or an unknown key against a healthy client, both need to keep failing +// closed to off exactly as before, because that fail-closed behaviour is what a +// kill switch exists to guarantee, including mid-incident on a SaaS instance +// whose GrowthBook connection has just dropped. +func TestFeatureFlagFailureModesDistinguishUnconfiguredFromUnreachable(t *testing.T) { + unreachable := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "nope", http.StatusInternalServerError) + })) + t.Cleanup(unreachable.Close) + + tests := []struct { + name string + buildClient func(t *testing.T) *Client + key string + wantOn bool + wantSource string // empty means the test does not assert on Source + }{ + { + name: "nil client is treated as unconfigured and defaults on", + buildClient: func(t *testing.T) *Client { + return nil + }, + key: "anything", + wantOn: true, + wantSource: SourceUnconfiguredDefaultOn, + }, + { + name: "empty ApiHost and ClientKey is unconfigured and defaults on", + buildClient: func(t *testing.T) *Client { + client, err := New( + context.Background(), + Config{LoadTimeout: time.Second, CacheRefreshInterval: time.Minute, UseSSE: true}, + testLogger(t), + nil, + nil, + ) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + return client + }, + key: "anything", + wantOn: true, + wantSource: SourceUnconfiguredDefaultOn, + }, + { + name: "configured but unreachable at startup with an empty cache stays off", + buildClient: func(t *testing.T) *Client { + client, err := New( + context.Background(), + Config{ + ApiHost: unreachable.URL, + ClientKey: "sdk-test", + UseSSE: false, + PollInterval: time.Minute, + LoadTimeout: time.Second, + CacheRefreshInterval: time.Hour, + }, + testLogger(t), + nil, + nil, + ) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + return client + }, + key: "anything", + wantOn: false, + }, + { + // This is the shape the two main.go startup paths (dash-api, worker) + // fall back to when featureflags.New itself returns an error (a + // genuine misconfiguration, e.g. a bad duration) but GrowthBook was + // configured. It must stay fail-closed rather than joining the + // unconfigured-default-on path, since that deployment did intend to + // use GrowthBook. + name: "empty offline ruleset is the startup fail-closed fallback and stays off", + buildClient: func(t *testing.T) *Client { + client, err := NewOffline(context.Background(), testLogger(t), "{}", nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + return client + }, + key: "202608_FEATURE_TICKETS", + wantOn: false, + }, + { + name: "configured and populated but an unknown key stays off", + buildClient: func(t *testing.T) *Client { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + return client + }, + key: "no-such-flag", + wantOn: false, + }, + { + name: "configured and populated with a known key evaluates normally", + buildClient: func(t *testing.T) *Client { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + return client + }, + key: "flag-all", + wantOn: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client := tt.buildClient(t) + result := client.Eval(context.Background(), tt.key, ForGuild(1)) + + require.Equal(t, tt.wantOn, result.On) + if tt.wantSource != "" { + require.Equal(t, tt.wantSource, result.Source) + } else { + require.NotEqual(t, SourceUnconfiguredDefaultOn, result.Source, + "a fail-closed result must never be mistaken for an unconfigured default") + } + }) + } +} + +// Polling is the default data source because a stock self-hosted GrowthBook does +// not serve SSE. This covers the real path end to end against a stub serving the +// documented payload shape. +func TestPollDataSourceLoadsPayload(t *testing.T) { + var requested string + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requested = r.URL.Path + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "status": 200, + "features": { + "202608_NEW_PRICING": {"defaultValue": true}, + "off-by-default": {"defaultValue": false} + }, + "dateUpdated": "2026-08-06T07:59:44.699Z" + }`)) + })) + t.Cleanup(server.Close) + + client, err := New( + context.Background(), + Config{ + ApiHost: server.URL, + ClientKey: "sdk-test", + UseSSE: false, + PollInterval: time.Minute, + LoadTimeout: 5 * time.Second, + CacheRefreshInterval: time.Hour, + }, + testLogger(t), + nil, + nil, + ) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + require.True(t, client.IsEnabled(ctx, "202608_NEW_PRICING", ForGuild(1))) + require.False(t, client.IsEnabled(ctx, "off-by-default", ForGuild(1))) + require.Contains(t, requested, "sdk-test", "SDK should request the payload for the client key") +} + +// An unreachable backend must not stop startup, and must not return an error from +// New: the process boots with every flag off. +func TestUnreachableBackendStillStarts(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "nope", http.StatusInternalServerError) + })) + t.Cleanup(server.Close) + + client, err := New( + context.Background(), + Config{ + ApiHost: server.URL, + ClientKey: "sdk-test", + UseSSE: false, + PollInterval: time.Minute, + LoadTimeout: time.Second, + CacheRefreshInterval: time.Hour, + }, + testLogger(t), + nil, + nil, + ) + require.NoError(t, err, "an unreachable backend must not fail startup") + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + require.False(t, client.IsEnabled(context.Background(), "anything", ForGuild(1))) +} + +func TestNewRejectsInvalidConfig(t *testing.T) { + _, err := New(context.Background(), Config{}, testLogger(t), nil, nil) + require.Error(t, err) +} + +func TestUnknownFlagIsOff(t *testing.T) { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + require.False(t, client.IsEnabled(context.Background(), "no-such-flag", ForGuild(1))) +} + +func TestRolloutBoundaries(t *testing.T) { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + ids := snowflakes(2000) + + tests := []struct { + flag string + expected bool + }{ + {"flag-none", false}, + {"flag-all", true}, + } + + for _, tt := range tests { + t.Run(tt.flag, func(t *testing.T) { + for _, id := range ids { + require.Equal(t, tt.expected, client.IsEnabled(ctx, tt.flag, ForGuild(id)), + "guild %d", id) + } + }) + } +} + +// The distribution must actually be near the configured percentage over realistic +// snowflakes. guildId % 100 was both biased by the snowflake's low bits and off +// by one, so a nominal 10% was neither 10% nor uniform. +func TestRolloutHitsTargetPercentage(t *testing.T) { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + ids := snowflakes(20000) + + var enrolled int + for _, id := range ids { + if client.IsEnabled(ctx, "flag-a", ForGuild(id)) { + enrolled++ + } + } + + ratio := float64(enrolled) / float64(len(ids)) + // 3 sigma for n=20000, p=0.1 is about 0.6 percentage points; allow 1.5. + require.InDelta(t, 0.1, ratio, 0.015, "enrolled %d of %d", enrolled, len(ids)) +} + +// Two flags at the same percentage must not select the same guilds, otherwise +// concurrent experiments contaminate each other. +func TestConcurrentFlagsEnrolIndependentCohorts(t *testing.T) { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + ids := snowflakes(20000) + + cohortA := map[uint64]struct{}{} + cohortB := map[uint64]struct{}{} + + for _, id := range ids { + if client.IsEnabled(ctx, "flag-a", ForGuild(id)) { + cohortA[id] = struct{}{} + } + if client.IsEnabled(ctx, "flag-b", ForGuild(id)) { + cohortB[id] = struct{}{} + } + } + + require.NotEmpty(t, cohortA) + require.NotEmpty(t, cohortB) + + var overlap int + for id := range cohortA { + if _, ok := cohortB[id]; ok { + overlap++ + } + } + + // Independent 10% samples should overlap on roughly 10% of cohort A. Insist + // it is well under half, which a shared-cohort implementation could never do: + // the old modulo scheme would have produced complete overlap. + require.Less(t, float64(overlap)/float64(len(cohortA)), 0.5, + "cohorts overlap on %d of %d guilds", overlap, len(cohortA)) +} + +// Assignment must be a pure function of flag and unit, so a guild does not flip +// between variants across evaluations, pods or restarts. +func TestAssignmentIsStableAcrossClients(t *testing.T) { + ctx := context.Background() + ids := snowflakes(500) + + first, err := NewOffline(ctx, testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, first.Close()) }) + + second, err := NewOffline(ctx, testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, second.Close()) }) + + for _, id := range ids { + want := first.IsEnabled(ctx, "flag-a", ForGuild(id)) + + // Repeated within the same client. + require.Equal(t, want, first.IsEnabled(ctx, "flag-a", ForGuild(id)), "guild %d", id) + // And in a separately constructed client, standing in for another pod. + require.Equal(t, want, second.IsEnabled(ctx, "flag-a", ForGuild(id)), "guild %d", id) + } +} + +func TestEvalIsConcurrencySafe(t *testing.T) { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + const goroutines = 16 + const perGoroutine = 200 + + var wg sync.WaitGroup + for g := 0; g < goroutines; g++ { + wg.Add(1) + go func(offset int) { + defer wg.Done() + for i := 0; i < perGoroutine; i++ { + guildId := uint64(offset*perGoroutine + i + 1) + client.IsEnabled(ctx, "flag-a", ForGuild(guildId)) + } + }(g) + } + wg.Wait() +} + +func TestMultivariateValues(t *testing.T) { + const ruleset = `{ + "greeting": {"defaultValue": "hello"}, + "limit": {"defaultValue": 25}, + "switch": {"defaultValue": true} + }` + + client, err := NewOffline(context.Background(), testLogger(t), ruleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + attrs := ForGuild(1) + + require.Equal(t, "hello", client.StringValue(ctx, "greeting", attrs, "fallback")) + // Whole numbers arrive from JSON as float64; IntValue hides that. + require.Equal(t, 25, client.IntValue(ctx, "limit", attrs, 0)) + require.True(t, client.IsEnabled(ctx, "switch", attrs)) + + // Type mismatches fall back rather than panicking. + require.Equal(t, "fallback", client.StringValue(ctx, "limit", attrs, "fallback")) + require.Equal(t, 99, client.IntValue(ctx, "greeting", attrs, 99)) +} + +type recordingRecorder struct { + mu sync.Mutex + exposures []Exposure +} + +func (r *recordingRecorder) Record(_ context.Context, exposure Exposure) { + r.mu.Lock() + defer r.mu.Unlock() + r.exposures = append(r.exposures, exposure) +} + +func (r *recordingRecorder) all() []Exposure { + r.mu.Lock() + defer r.mu.Unlock() + return append([]Exposure(nil), r.exposures...) +} + +// A plain percentage rollout is not an experiment. Recording exposures for one +// would fill the assignment table with rows no analysis wants. +func TestRolloutDoesNotRecordExposures(t *testing.T) { + recorder := &recordingRecorder{} + + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, recorder) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + for _, id := range snowflakes(500) { + client.IsEnabled(ctx, "flag-a", ForGuild(id)) + } + + require.Empty(t, recorder.all()) +} + +// An experiment rule does record, and carries the identifier the assignment +// query needs. +func TestExperimentRecordsExposure(t *testing.T) { + const ruleset = `{ + "checkout-copy": { + "defaultValue": "control", + "rules": [{ + "key": "checkout-copy-test", + "hashAttribute": "guild_id", + "variations": ["control", "treatment"], + "weights": [0.5, 0.5], + "coverage": 1 + }] + } + }` + + recorder := &recordingRecorder{} + + client, err := NewOffline(context.Background(), testLogger(t), ruleset, recorder) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + const guildId uint64 = 1328073426023219221 + + result := client.Eval(ctx, "checkout-copy", ForGuild(guildId)) + require.True(t, result.InExperiment) + + exposures := recorder.all() + require.Len(t, exposures, 1) + + exposure := exposures[0] + require.Equal(t, "checkout-copy-test", exposure.ExperimentKey) + require.Equal(t, AttrGuild, exposure.IdentifierType) + require.Equal(t, strconv.FormatUint(guildId, 10), exposure.Identifier) + require.Equal(t, "checkout-copy", exposure.FeatureKey) + require.Contains(t, []int{0, 1}, exposure.VariationId) + require.Equal(t, result.VariationId, exposure.VariationId) +} + +func TestNilRecorderIsSafe(t *testing.T) { + const ruleset = `{ + "exp": { + "defaultValue": "control", + "rules": [{ + "key": "exp-test", + "hashAttribute": "guild_id", + "variations": ["control", "treatment"], + "weights": [0.5, 0.5], + "coverage": 1 + }] + } + }` + + client, err := NewOffline(context.Background(), testLogger(t), ruleset, nil) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + require.NotPanics(t, func() { + client.Eval(context.Background(), "exp", ForGuild(1)) + }) +} + +func TestCloseIsIdempotent(t *testing.T) { + client, err := NewOffline(context.Background(), testLogger(t), rolloutRuleset, nil) + require.NoError(t, err) + + require.NoError(t, client.Close()) + require.NoError(t, client.Close()) +} + +func TestSlogAdapterForwardsToZap(t *testing.T) { + // Exercises the bridge with attributes and groups so a malformed field cannot + // panic inside the SDK's logging path. + logger := newSlogAdapter(zap.NewNop()) + + require.NotPanics(t, func() { + logger.Info("plain") + logger.With("key", "value").Warn("with attr") + logger.WithGroup("outer").With("inner", 1).Error("grouped") + logger.Info("mixed", "count", 3, "ok", true) + }) +} + +func BenchmarkIsEnabled(b *testing.B) { + client, err := NewOffline(context.Background(), zap.NewNop(), rolloutRuleset, nil) + if err != nil { + b.Fatal(err) + } + defer func() { _ = client.Close() }() + + ctx := context.Background() + ids := snowflakes(1024) + + b.ReportAllocs() + b.ResetTimer() + + for i := 0; i < b.N; i++ { + client.IsEnabled(ctx, "flag-a", ForGuild(ids[i%len(ids)])) + } +} diff --git a/featureflags/logger.go b/featureflags/logger.go new file mode 100644 index 0000000..f59fb74 --- /dev/null +++ b/featureflags/logger.go @@ -0,0 +1,115 @@ +package featureflags + +import ( + "context" + "log/slog" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +// The GrowthBook SDK logs through log/slog, while every service here logs through +// zap. Bridging the two keeps SDK diagnostics in the same structured stream +// Promtail already scrapes, which matters because the failure we most need to see +// is a silently dead data source connection. +type slogAdapter struct { + logger *zap.Logger + fields []zap.Field + group string +} + +func newSlogAdapter(logger *zap.Logger) *slog.Logger { + return slog.New(&slogAdapter{logger: logger.Named("growthbook")}) +} + +func (s *slogAdapter) Enabled(_ context.Context, level slog.Level) bool { + return s.logger.Core().Enabled(zapLevel(level)) +} + +func (s *slogAdapter) Handle(_ context.Context, record slog.Record) error { + fields := make([]zap.Field, 0, len(s.fields)+record.NumAttrs()) + fields = append(fields, s.fields...) + + record.Attrs(func(attr slog.Attr) bool { + fields = appendAttr(fields, s.group, attr) + return true + }) + + if entry := s.logger.Check(zapLevel(record.Level), record.Message); entry != nil { + entry.Write(fields...) + } + + return nil +} + +func (s *slogAdapter) WithAttrs(attrs []slog.Attr) slog.Handler { + next := &slogAdapter{ + logger: s.logger, + group: s.group, + fields: make([]zap.Field, len(s.fields), len(s.fields)+len(attrs)), + } + copy(next.fields, s.fields) + + for _, attr := range attrs { + next.fields = appendAttr(next.fields, s.group, attr) + } + + return next +} + +func (s *slogAdapter) WithGroup(name string) slog.Handler { + if name == "" { + return s + } + + group := name + if s.group != "" { + group = s.group + "." + name + } + + next := &slogAdapter{ + logger: s.logger, + group: group, + fields: make([]zap.Field, len(s.fields)), + } + copy(next.fields, s.fields) + + return next +} + +// appendAttr flattens a slog attribute into zap fields, joining group names with +// dots since zap fields are a flat namespace. +func appendAttr(fields []zap.Field, group string, attr slog.Attr) []zap.Field { + if attr.Equal(slog.Attr{}) { + return fields + } + + key := attr.Key + if group != "" { + key = group + "." + key + } + + value := attr.Value.Resolve() + if value.Kind() == slog.KindGroup { + for _, nested := range value.Group() { + fields = appendAttr(fields, key, nested) + } + + return fields + } + + return append(fields, zap.Any(key, value.Any())) +} + +func zapLevel(level slog.Level) zapcore.Level { + switch { + case level >= slog.LevelError: + return zapcore.ErrorLevel + case level >= slog.LevelWarn: + return zapcore.WarnLevel + case level >= slog.LevelInfo: + return zapcore.InfoLevel + default: + return zapcore.DebugLevel + } +} diff --git a/featureflags/recorder.go b/featureflags/recorder.go new file mode 100644 index 0000000..64a78cf --- /dev/null +++ b/featureflags/recorder.go @@ -0,0 +1,389 @@ +package featureflags + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/go-redis/redis/v8" + "go.uber.org/zap" +) + +// RecordedExposure is an Exposure stamped with the moment it happened, rather +// than the moment it reached the database. Assignment analysis cares when the +// unit saw the variation. +type RecordedExposure struct { + Exposure + ExposedAt time.Time +} + +// ExposureSink persists a batch of exposures. Kept as an interface so this +// package does not depend on the database module, which matters because common +// consumes database as a pinned version. +type ExposureSink interface { + InsertExposures(ctx context.Context, exposures []RecordedExposure) error +} + +// SinkFunc adapts a plain function to ExposureSink, so a service can supply the +// Postgres write at its own call site without this package importing database: +// +// sink := featureflags.SinkFunc(func(ctx context.Context, exposures []featureflags.RecordedExposure) error { +// rows := make([]database.ExperimentExposure, 0, len(exposures)) +// for _, e := range exposures { +// rows = append(rows, database.ExperimentExposure{ +// ExperimentKey: e.ExperimentKey, +// VariationId: e.VariationId, +// IdentifierType: e.IdentifierType, +// Identifier: e.Identifier, +// FeatureKey: e.FeatureKey, +// ExposedAt: e.ExposedAt, +// }) +// } +// return dbclient.Client.ExperimentExposures.InsertBatch(ctx, rows) +// }) +type SinkFunc func(ctx context.Context, exposures []RecordedExposure) error + +func (f SinkFunc) InsertExposures(ctx context.Context, exposures []RecordedExposure) error { + return f(ctx, exposures) +} + +// ExposureDeduper claims exposures across processes, returning for each key +// whether this caller won the claim. Keys already claimed by another pod return +// false and are dropped rather than written. +type ExposureDeduper interface { + Claim(ctx context.Context, keys []string, ttl time.Duration) ([]bool, error) +} + +// RecorderConfig tunes the exposure pipeline. Zero values take the defaults +// applied by withDefaults. +type RecorderConfig struct { + // QueueSize bounds memory and, more importantly, bounds how far behind the + // writer can fall before shedding load. + QueueSize int + // BatchSize is how many rows one COPY carries. + BatchSize int + // FlushInterval caps how long a partial batch waits. + FlushInterval time.Duration + // WriteTimeout bounds one flush. + WriteTimeout time.Duration + // DedupeTTL is how long a claimed exposure suppresses further rows for the + // same unit and experiment. + DedupeTTL time.Duration + // LocalCacheEntries caps the per-generation size of the in-process set. + LocalCacheEntries int + // LocalCacheRotation is how often the in-process set drops a generation. + LocalCacheRotation time.Duration +} + +func (c RecorderConfig) withDefaults() RecorderConfig { + if c.QueueSize <= 0 { + c.QueueSize = 4096 + } + if c.BatchSize <= 0 { + c.BatchSize = 500 + } + if c.FlushInterval <= 0 { + c.FlushInterval = 5 * time.Second + } + if c.WriteTimeout <= 0 { + c.WriteTimeout = 10 * time.Second + } + if c.DedupeTTL <= 0 { + c.DedupeTTL = 24 * time.Hour + } + if c.LocalCacheEntries <= 0 { + c.LocalCacheEntries = 100_000 + } + if c.LocalCacheRotation <= 0 { + c.LocalCacheRotation = 30 * time.Minute + } + + return c +} + +// RecorderStats is a snapshot of pipeline counters. Services expose these as +// metrics; this package does not depend on any metrics library. +type RecorderStats struct { + // Enqueued exposures accepted onto the queue. + Enqueued uint64 + // SuppressedLocally were already in the in-process set, so cost no I/O. + SuppressedLocally uint64 + // Dropped were shed because the queue was full. Non-zero means experiment + // data is being lost and the writer cannot keep up. + Dropped uint64 + // SuppressedRemotely lost the cross-process claim, meaning another pod + // already recorded this unit. + SuppressedRemotely uint64 + // Written rows reached the sink. + Written uint64 + // FailedWrites are flushes that errored. Those exposures are lost. + FailedWrites uint64 + // LocalCacheEntries currently held in the in-process set. + LocalCacheEntries uint64 +} + +// Recorder implements ExposureRecorder. It keeps all I/O off the caller's +// goroutine: Record does a read-locked map lookup and a non-blocking channel +// send, nothing more. +type Recorder struct { + logger *zap.Logger + sink ExposureSink + deduper ExposureDeduper + cfg RecorderConfig + + queue chan RecordedExposure + seen *rotatingSet + + enqueued atomic.Uint64 + suppressedLocally atomic.Uint64 + dropped atomic.Uint64 + suppressedRemotely atomic.Uint64 + written atomic.Uint64 + failedWrites atomic.Uint64 + + stop chan struct{} + stopOnce sync.Once + done chan struct{} + + // now is overridable for tests. + now func() time.Time +} + +// NewRecorder starts the background writer. deduper may be nil, in which case +// deduplication is in-process only and duplicate rows across pods are expected. +func NewRecorder(logger *zap.Logger, sink ExposureSink, deduper ExposureDeduper, cfg RecorderConfig) *Recorder { + cfg = cfg.withDefaults() + + r := &Recorder{ + logger: logger, + sink: sink, + deduper: deduper, + cfg: cfg, + queue: make(chan RecordedExposure, cfg.QueueSize), + seen: newRotatingSet(cfg.LocalCacheEntries, cfg.LocalCacheRotation), + stop: make(chan struct{}), + done: make(chan struct{}), + now: time.Now, + } + + go r.run() + + return r +} + +// Record is called from the evaluation path, which for the worker means inside +// Discord event handling. It must never block and never perform I/O. A nil +// receiver is a no-op, so a service can wire flags without exposure recording. +func (r *Recorder) Record(_ context.Context, exposure Exposure) { + if r == nil { + return + } + + key := dedupeKey(exposure) + + if r.seen.contains(key) { + r.suppressedLocally.Add(1) + return + } + + select { + case r.queue <- RecordedExposure{Exposure: exposure, ExposedAt: r.now()}: + // Marked seen only after a successful send. Marking before would mean a + // shed exposure was never retried, silently losing that unit for a whole + // rotation window. + r.seen.add(key) + r.enqueued.Add(1) + default: + r.dropped.Add(1) + } +} + +func (r *Recorder) run() { + defer close(r.done) + + ticker := time.NewTicker(r.cfg.FlushInterval) + defer ticker.Stop() + + batch := make([]RecordedExposure, 0, r.cfg.BatchSize) + + for { + select { + case <-r.stop: + // Drain whatever is already queued so a graceful shutdown does not + // discard exposures that were accepted. + for { + select { + case exposure := <-r.queue: + batch = append(batch, exposure) + if len(batch) >= r.cfg.BatchSize { + r.flush(batch) + batch = batch[:0] + } + default: + r.flush(batch) + return + } + } + case exposure := <-r.queue: + batch = append(batch, exposure) + if len(batch) >= r.cfg.BatchSize { + r.flush(batch) + batch = batch[:0] + } + case <-ticker.C: + if len(batch) > 0 { + r.flush(batch) + batch = batch[:0] + } + } + } +} + +func (r *Recorder) flush(batch []RecordedExposure) { + if len(batch) == 0 { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), r.cfg.WriteTimeout) + defer cancel() + + batch = r.claim(ctx, batch) + if len(batch) == 0 { + return + } + + if err := r.sink.InsertExposures(ctx, batch); err != nil { + r.failedWrites.Add(uint64(len(batch))) + r.logger.Error("featureflags: writing exposures failed", + zap.Int("count", len(batch)), zap.Error(err)) + return + } + + r.written.Add(uint64(len(batch))) +} + +// claim filters the batch down to exposures this process won the claim for. +// +// On deduper failure it deliberately fails open and keeps everything. Losing +// exposures biases an experiment's results, whereas duplicate rows are harmless +// provided the assignment query takes the first exposure per unit, which it must +// do anyway to handle deduplication-window boundaries. +func (r *Recorder) claim(ctx context.Context, batch []RecordedExposure) []RecordedExposure { + if r.deduper == nil { + return batch + } + + keys := make([]string, 0, len(batch)) + for _, exposure := range batch { + keys = append(keys, dedupeKey(exposure.Exposure)) + } + + claimed, err := r.deduper.Claim(ctx, keys, r.cfg.DedupeTTL) + if err != nil { + r.logger.Warn("featureflags: claiming exposures failed, writing without deduplication", + zap.Int("count", len(batch)), zap.Error(err)) + return batch + } + + if len(claimed) != len(batch) { + r.logger.Warn("featureflags: deduper returned mismatched results, writing without deduplication", + zap.Int("want", len(batch)), zap.Int("got", len(claimed))) + return batch + } + + kept := batch[:0] + for i, ok := range claimed { + if ok { + kept = append(kept, batch[i]) + continue + } + + r.suppressedRemotely.Add(1) + } + + return kept +} + +func (r *Recorder) Stats() RecorderStats { + return RecorderStats{ + Enqueued: r.enqueued.Load(), + SuppressedLocally: r.suppressedLocally.Load(), + Dropped: r.dropped.Load(), + SuppressedRemotely: r.suppressedRemotely.Load(), + Written: r.written.Load(), + FailedWrites: r.failedWrites.Load(), + LocalCacheEntries: uint64(r.seen.len()), + } +} + +// Close stops accepting work, flushes what is queued, and waits for the writer. +func (r *Recorder) Close() error { + r.stopOnce.Do(func() { + close(r.stop) + }) + + <-r.done + + return nil +} + +// dedupeKey identifies one unit's enrolment in one experiment. The variation is +// deliberately excluded: assignment is stable, so including it would let a +// bucketing change produce a second row for the same unit. +func dedupeKey(exposure Exposure) string { + var b strings.Builder + b.Grow(len(exposure.ExperimentKey) + len(exposure.IdentifierType) + len(exposure.Identifier) + 2) + b.WriteString(exposure.ExperimentKey) + b.WriteByte(':') + b.WriteString(exposure.IdentifierType) + b.WriteByte(':') + b.WriteString(exposure.Identifier) + + return b.String() +} + +// redisDeduper claims exposures in Redis so that all pods together write one row +// per unit per experiment per DedupeTTL. +type redisDeduper struct { + redis *redis.Client +} + +// NewRedisDeduper returns an ExposureDeduper backed by SETNX. +func NewRedisDeduper(client *redis.Client) ExposureDeduper { + return &redisDeduper{redis: client} +} + +func (d *redisDeduper) Claim(ctx context.Context, keys []string, ttl time.Duration) ([]bool, error) { + if len(keys) == 0 { + return nil, nil + } + + pipe := d.redis.Pipeline() + + cmds := make([]*redis.BoolCmd, 0, len(keys)) + for _, key := range keys { + cmds = append(cmds, pipe.SetNX(ctx, "featureflags:exposure:"+key, 1, ttl)) + } + + // Exec reports the first command error. redis.Nil is not meaningful for + // SETNX, so it is not treated as a failure. + if _, err := pipe.Exec(ctx); err != nil && !errors.Is(err, redis.Nil) { + return nil, fmt.Errorf("featureflags: claiming exposures: %w", err) + } + + claimed := make([]bool, 0, len(cmds)) + for _, cmd := range cmds { + ok, err := cmd.Result() + if err != nil && !errors.Is(err, redis.Nil) { + return nil, fmt.Errorf("featureflags: claiming exposures: %w", err) + } + + claimed = append(claimed, ok) + } + + return claimed, nil +} diff --git a/featureflags/recorder_test.go b/featureflags/recorder_test.go new file mode 100644 index 0000000..9d5973f --- /dev/null +++ b/featureflags/recorder_test.go @@ -0,0 +1,614 @@ +package featureflags + +import ( + "context" + "errors" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +type fakeSink struct { + mu sync.Mutex + batches [][]RecordedExposure + err error + inserted []RecordedExposure +} + +func (f *fakeSink) InsertExposures(_ context.Context, exposures []RecordedExposure) error { + f.mu.Lock() + defer f.mu.Unlock() + + if f.err != nil { + return f.err + } + + batch := append([]RecordedExposure(nil), exposures...) + f.batches = append(f.batches, batch) + f.inserted = append(f.inserted, batch...) + + return nil +} + +func (f *fakeSink) all() []RecordedExposure { + f.mu.Lock() + defer f.mu.Unlock() + + return append([]RecordedExposure(nil), f.inserted...) +} + +func (f *fakeSink) batchCount() int { + f.mu.Lock() + defer f.mu.Unlock() + + return len(f.batches) +} + +// fakeDeduper claims a key the first time it is seen, mimicking SETNX. +type fakeDeduper struct { + mu sync.Mutex + seen map[string]struct{} + err error + badLen bool + calls int +} + +func newFakeDeduper() *fakeDeduper { + return &fakeDeduper{seen: map[string]struct{}{}} +} + +func (f *fakeDeduper) Claim(_ context.Context, keys []string, _ time.Duration) ([]bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + + f.calls++ + + if f.err != nil { + return nil, f.err + } + + if f.badLen { + return []bool{true}, nil + } + + claimed := make([]bool, 0, len(keys)) + for _, key := range keys { + _, exists := f.seen[key] + claimed = append(claimed, !exists) + f.seen[key] = struct{}{} + } + + return claimed, nil +} + +func exposure(experiment string, guildId uint64) Exposure { + return Exposure{ + ExperimentKey: experiment, + VariationId: 1, + IdentifierType: AttrGuild, + Identifier: strconv.FormatUint(guildId, 10), + FeatureKey: "feature-" + experiment, + } +} + +// eventually polls rather than sleeping a fixed duration, so the tests are not +// timing-fragile on a loaded machine. +func eventually(t *testing.T, condition func() bool) { + t.Helper() + require.Eventually(t, condition, 3*time.Second, 5*time.Millisecond) +} + +func TestRecorderConfigDefaults(t *testing.T) { + cfg := RecorderConfig{}.withDefaults() + + require.Positive(t, cfg.QueueSize) + require.Positive(t, cfg.BatchSize) + require.Positive(t, cfg.FlushInterval) + require.Positive(t, cfg.WriteTimeout) + require.Equal(t, 24*time.Hour, cfg.DedupeTTL) + require.Positive(t, cfg.LocalCacheEntries) + require.Positive(t, cfg.LocalCacheRotation) +} + +func TestRecorderConfigKeepsExplicitValues(t *testing.T) { + cfg := RecorderConfig{ + QueueSize: 7, + BatchSize: 3, + FlushInterval: time.Second, + WriteTimeout: 2 * time.Second, + DedupeTTL: time.Hour, + LocalCacheEntries: 11, + LocalCacheRotation: time.Minute, + }.withDefaults() + + require.Equal(t, 7, cfg.QueueSize) + require.Equal(t, 3, cfg.BatchSize) + require.Equal(t, time.Hour, cfg.DedupeTTL) + require.Equal(t, 11, cfg.LocalCacheEntries) +} + +func TestDedupeKeyIgnoresVariation(t *testing.T) { + // Assignment is stable, so a variation change for the same unit means the + // bucketing changed. That must not produce a second row. + a := exposure("exp", 1) + b := exposure("exp", 1) + b.VariationId = 0 + + require.Equal(t, dedupeKey(a), dedupeKey(b)) +} + +func TestDedupeKeyDistinguishesUnitsExperimentsAndTypes(t *testing.T) { + base := exposure("exp", 1) + + otherGuild := exposure("exp", 2) + otherExperiment := exposure("exp2", 1) + + otherType := base + otherType.IdentifierType = AttrUser + + keys := map[string]struct{}{ + dedupeKey(base): {}, + dedupeKey(otherGuild): {}, + dedupeKey(otherExperiment): {}, + dedupeKey(otherType): {}, + } + + require.Len(t, keys, 4) +} + +// The whole point of the local set: repeat traffic for a unit already queued must +// not reach the queue at all. +func TestRecordSuppressesRepeatsLocally(t *testing.T) { + sink := &fakeSink{} + deduper := newFakeDeduper() + + recorder := NewRecorder(zap.NewNop(), sink, deduper, RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + ctx := context.Background() + for i := 0; i < 100; i++ { + recorder.Record(ctx, exposure("exp", 42)) + } + + eventually(t, func() bool { return len(sink.all()) == 1 }) + + stats := recorder.Stats() + require.Equal(t, uint64(1), stats.Enqueued) + require.Equal(t, uint64(99), stats.SuppressedLocally) + require.Zero(t, stats.Dropped) +} + +func TestRecordStampsExposureTime(t *testing.T) { + sink := &fakeSink{} + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + frozen := time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC) + recorder.now = func() time.Time { return frozen } + + recorder.Record(context.Background(), exposure("exp", 1)) + eventually(t, func() bool { return len(sink.all()) == 1 }) + + require.Equal(t, frozen, sink.all()[0].ExposedAt) +} + +// A full queue must shed load rather than block the caller, and a shed exposure +// must remain eligible so a later event retries it. +func TestRecordShedsLoadWithoutBlockingAndRetriesLater(t *testing.T) { + sink := &fakeSink{} + blocked := make(chan struct{}) + + recorder := &Recorder{ + logger: zap.NewNop(), + sink: sink, + cfg: RecorderConfig{QueueSize: 2}.withDefaults(), + queue: make(chan RecordedExposure, 2), + seen: newRotatingSet(1000, time.Hour), + stop: make(chan struct{}), + done: blocked, + now: time.Now, + } + + ctx := context.Background() + + // Writer is not running, so the queue fills and stays full. + recorder.Record(ctx, exposure("exp", 1)) + recorder.Record(ctx, exposure("exp", 2)) + + done := make(chan struct{}) + go func() { + defer close(done) + recorder.Record(ctx, exposure("exp", 3)) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Record blocked when the queue was full") + } + + require.Equal(t, uint64(2), recorder.Stats().Enqueued) + require.Equal(t, uint64(1), recorder.Stats().Dropped) + + // Guild 3 was shed, so it must not be marked seen; otherwise the unit would + // be silently lost for a whole rotation window. + require.False(t, recorder.seen.contains(dedupeKey(exposure("exp", 3)))) + require.True(t, recorder.seen.contains(dedupeKey(exposure("exp", 1)))) +} + +func TestRecorderDedupesAcrossProcesses(t *testing.T) { + sink := &fakeSink{} + // One shared deduper stands in for Redis seen by two pods. + deduper := newFakeDeduper() + + cfg := RecorderConfig{BatchSize: 1, FlushInterval: 10 * time.Millisecond} + + first := NewRecorder(zap.NewNop(), sink, deduper, cfg) + t.Cleanup(func() { require.NoError(t, first.Close()) }) + second := NewRecorder(zap.NewNop(), sink, deduper, cfg) + t.Cleanup(func() { require.NoError(t, second.Close()) }) + + ctx := context.Background() + first.Record(ctx, exposure("exp", 99)) + eventually(t, func() bool { return len(sink.all()) == 1 }) + + // The second pod has its own empty local set, so it reaches the shared claim + // and must lose. + second.Record(ctx, exposure("exp", 99)) + eventually(t, func() bool { return second.Stats().SuppressedRemotely == 1 }) + + require.Len(t, sink.all(), 1) +} + +// Losing exposures biases a result; duplicates do not, provided the assignment +// query takes the first per unit. So a deduper outage must fail open. +func TestClaimFailureFailsOpen(t *testing.T) { + sink := &fakeSink{} + deduper := newFakeDeduper() + deduper.err = errors.New("redis down") + + recorder := NewRecorder(zap.NewNop(), sink, deduper, RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + recorder.Record(context.Background(), exposure("exp", 1)) + + eventually(t, func() bool { return len(sink.all()) == 1 }) + require.Zero(t, recorder.Stats().SuppressedRemotely) +} + +func TestClaimLengthMismatchFailsOpen(t *testing.T) { + sink := &fakeSink{} + deduper := newFakeDeduper() + deduper.badLen = true + + recorder := NewRecorder(zap.NewNop(), sink, deduper, RecorderConfig{ + BatchSize: 2, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + ctx := context.Background() + recorder.Record(ctx, exposure("exp", 1)) + recorder.Record(ctx, exposure("exp", 2)) + + eventually(t, func() bool { return len(sink.all()) == 2 }) +} + +func TestBatchesFillToBatchSize(t *testing.T) { + sink := &fakeSink{} + + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 10, + // Long enough that only the size trigger can fire. + FlushInterval: time.Hour, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + ctx := context.Background() + for i := 0; i < 10; i++ { + recorder.Record(ctx, exposure("exp", uint64(i+1))) + } + + eventually(t, func() bool { return len(sink.all()) == 10 }) + require.Equal(t, 1, sink.batchCount(), "expected a single batched write") +} + +func TestPartialBatchFlushesOnInterval(t *testing.T) { + sink := &fakeSink{} + + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 1000, + FlushInterval: 20 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + recorder.Record(context.Background(), exposure("exp", 1)) + + eventually(t, func() bool { return len(sink.all()) == 1 }) +} + +// Accepted exposures must not be discarded by shutdown. +func TestCloseDrainsQueue(t *testing.T) { + sink := &fakeSink{} + + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 1000, + FlushInterval: time.Hour, + }) + + ctx := context.Background() + for i := 0; i < 50; i++ { + recorder.Record(ctx, exposure("exp", uint64(i+1))) + } + + require.NoError(t, recorder.Close()) + require.Len(t, sink.all(), 50) +} + +func TestCloseIsIdempotentForRecorder(t *testing.T) { + recorder := NewRecorder(zap.NewNop(), &fakeSink{}, nil, RecorderConfig{}) + + require.NoError(t, recorder.Close()) + require.NoError(t, recorder.Close()) +} + +func TestSinkFailureIsCountedNotFatal(t *testing.T) { + sink := &fakeSink{err: errors.New("postgres down")} + + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + recorder.Record(context.Background(), exposure("exp", 1)) + + eventually(t, func() bool { return recorder.Stats().FailedWrites == 1 }) + require.Zero(t, recorder.Stats().Written) +} + +func TestRecordIsConcurrencySafe(t *testing.T) { + sink := &fakeSink{} + recorder := NewRecorder(zap.NewNop(), sink, newFakeDeduper(), RecorderConfig{ + QueueSize: 8192, + BatchSize: 100, + FlushInterval: 5 * time.Millisecond, + }) + + ctx := context.Background() + const goroutines = 16 + const perGoroutine = 200 + + var wg sync.WaitGroup + for g := 0; g < goroutines; g++ { + wg.Add(1) + go func(offset int) { + defer wg.Done() + for i := 0; i < perGoroutine; i++ { + recorder.Record(ctx, exposure("exp", uint64(offset*perGoroutine+i+1))) + } + }(g) + } + wg.Wait() + + require.NoError(t, recorder.Close()) + + stats := recorder.Stats() + require.Equal(t, uint64(goroutines*perGoroutine), stats.Enqueued+stats.Dropped+stats.SuppressedLocally) + require.Len(t, sink.all(), int(stats.Written)) +} + +// End-to-end wiring: an experiment evaluated through the Client must land in the +// sink, while a plain rollout must not. +func TestClientToRecorderIntegration(t *testing.T) { + const ruleset = `{ + "exp-flag": { + "defaultValue": "control", + "rules": [{ + "key": "exp-key", + "hashAttribute": "guild_id", + "variations": ["control", "treatment"], + "weights": [0.5, 0.5], + "coverage": 1 + }] + }, + "rollout-flag": { + "defaultValue": false, + "rules": [{"force": true, "coverage": 1, "hashAttribute": "guild_id"}] + } + }` + + sink := &fakeSink{} + recorder := NewRecorder(zap.NewNop(), sink, newFakeDeduper(), RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + client, err := NewOffline(context.Background(), zap.NewNop(), ruleset, recorder) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, client.Close()) }) + + ctx := context.Background() + client.IsEnabled(ctx, "rollout-flag", ForGuild(1234)) + client.Eval(ctx, "exp-flag", ForGuild(1234)) + + eventually(t, func() bool { return len(sink.all()) == 1 }) + + written := sink.all()[0] + require.Equal(t, "exp-key", written.ExperimentKey) + require.Equal(t, AttrGuild, written.IdentifierType) + require.Equal(t, "1234", written.Identifier) + require.False(t, written.ExposedAt.IsZero()) +} + +// SinkFunc is how services supply the Postgres write without this package +// importing the database module. +func TestSinkFuncAdaptsAFunction(t *testing.T) { + var got []RecordedExposure + + sink := SinkFunc(func(_ context.Context, exposures []RecordedExposure) error { + got = append(got, exposures...) + return nil + }) + + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + + recorder.Record(context.Background(), exposure("exp", 7)) + require.NoError(t, recorder.Close()) + + require.Len(t, got, 1) + require.Equal(t, "7", got[0].Identifier) +} + +func TestSinkFuncPropagatesErrors(t *testing.T) { + sink := SinkFunc(func(_ context.Context, _ []RecordedExposure) error { + return errors.New("write failed") + }) + + recorder := NewRecorder(zap.NewNop(), sink, nil, RecorderConfig{ + BatchSize: 1, + FlushInterval: 10 * time.Millisecond, + }) + t.Cleanup(func() { require.NoError(t, recorder.Close()) }) + + recorder.Record(context.Background(), exposure("exp", 1)) + + eventually(t, func() bool { return recorder.Stats().FailedWrites == 1 }) +} + +// Services hold these as package-level variables assigned during startup, so a +// read before assignment must degrade rather than panic. This is the failure mode +// the old settable singleton had. +// +// A nil Client is treated the same as "GrowthBook not configured": every flag +// defaults to enabled rather than off, since both are the same self-hosted +// reality of having no ruleset to consult, just observed one level up. +func TestNilReceiversAreSafe(t *testing.T) { + var client *Client + var recorder *Recorder + + ctx := context.Background() + + require.NotPanics(t, func() { + require.True(t, client.IsEnabled(ctx, "flag", ForGuild(1))) + require.Equal(t, "def", client.StringValue(ctx, "flag", ForGuild(1), "def")) + require.Equal(t, 5, client.IntValue(ctx, "flag", ForGuild(1), 5)) + require.Equal(t, Result{On: true, Source: SourceUnconfiguredDefaultOn}, client.Eval(ctx, "flag", ForGuild(1))) + require.NoError(t, client.Close()) + + recorder.Record(ctx, exposure("exp", 1)) + }) +} + +func TestRotatingSetBasics(t *testing.T) { + set := newRotatingSet(1000, time.Hour) + + require.False(t, set.contains("a")) + set.add("a") + require.True(t, set.contains("a")) + require.False(t, set.contains("b")) +} + +// Size-triggered rotation is what bounds memory when a high-coverage experiment +// is running across 500k guilds on a 256Mi pod. +func TestRotatingSetRotatesOnSize(t *testing.T) { + set := newRotatingSet(10, time.Hour) + + for i := 0; i < 10; i++ { + set.add(strconv.Itoa(i)) + } + + // The 11th insert rotates: current becomes previous, a fresh map starts. + set.add("trigger") + + // Still visible via the previous generation. + require.True(t, set.contains("0")) + require.True(t, set.contains("trigger")) + + // Filling a second generation evicts the first. + for i := 0; i < 10; i++ { + set.add("second-" + strconv.Itoa(i)) + } + set.add("trigger-2") + + require.False(t, set.contains("0"), "first generation should have been evicted") + require.True(t, set.contains("trigger-2")) + require.LessOrEqual(t, set.len(), 22) +} + +func TestRotatingSetRotatesOnAge(t *testing.T) { + set := newRotatingSet(1_000_000, time.Minute) + + now := time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC) + set.now = func() time.Time { return now } + set.lastRotate = now + + set.add("first") + require.True(t, set.contains("first")) + + now = now.Add(2 * time.Minute) + set.add("second") + require.True(t, set.contains("first"), "one rotation keeps the previous generation") + + now = now.Add(2 * time.Minute) + set.add("third") + require.False(t, set.contains("first"), "two rotations evict") + require.True(t, set.contains("second")) + require.True(t, set.contains("third")) +} + +func TestRotatingSetIsConcurrencySafe(t *testing.T) { + set := newRotatingSet(500, 10*time.Millisecond) + + var wg sync.WaitGroup + for g := 0; g < 8; g++ { + wg.Add(1) + go func(offset int) { + defer wg.Done() + for i := 0; i < 500; i++ { + key := strconv.Itoa(offset*500 + i) + set.add(key) + set.contains(key) + } + }(g) + } + wg.Wait() +} + +func BenchmarkRecordSuppressed(b *testing.B) { + recorder := NewRecorder(zap.NewNop(), &fakeSink{}, nil, RecorderConfig{ + QueueSize: 1024, + BatchSize: 1000, + FlushInterval: time.Hour, + }) + defer func() { _ = recorder.Close() }() + + ctx := context.Background() + e := exposure("exp", 1) + recorder.Record(ctx, e) + + b.ReportAllocs() + b.ResetTimer() + + // The steady-state hot path: a unit already recorded, so no I/O and no queue. + for i := 0; i < b.N; i++ { + recorder.Record(ctx, e) + } +} diff --git a/featureflags/rotatingset.go b/featureflags/rotatingset.go new file mode 100644 index 0000000..a7d26c4 --- /dev/null +++ b/featureflags/rotatingset.go @@ -0,0 +1,88 @@ +package featureflags + +import ( + "hash/maphash" + "sync" + "time" +) + +// rotatingSet is a bounded, approximate "have I seen this recently" set. It sits +// in front of the cross-process deduplication check so repeat traffic for a unit +// already recorded costs nothing but a read lock. +// +// Two design choices are driven by the worker's 256Mi memory limit: +// +// Keys are stored as 64-bit hashes rather than strings. A worker pod consumes +// gateway events from a Redis Stream consumer group, so any pod can see any +// guild; at 500k guilds across several live experiments the string form would run +// to tens of megabytes per pod. Hashing costs a false-positive rate of roughly +// n^2/2^65, which at these sizes is around one in ten billion. A false positive +// skips one exposure for one unit, which cannot meaningfully move an experiment +// result, and no false negatives are possible. +// +// Entries expire by generation rather than per-key TTL: a full map is dropped +// wholesale, which is cheap and needs no timestamps. Effective lifetime is +// therefore between one and two rotation intervals. +type rotatingSet struct { + mu sync.RWMutex + seed maphash.Seed + current map[uint64]struct{} + previous map[uint64]struct{} + + maxEntries int + rotateEvery time.Duration + lastRotate time.Time + + // now is overridable so rotation can be tested without sleeping. + now func() time.Time +} + +func newRotatingSet(maxEntries int, rotateEvery time.Duration) *rotatingSet { + return &rotatingSet{ + seed: maphash.MakeSeed(), + current: make(map[uint64]struct{}), + previous: make(map[uint64]struct{}), + maxEntries: maxEntries, + rotateEvery: rotateEvery, + now: time.Now, + lastRotate: time.Now(), + } +} + +func (s *rotatingSet) contains(key string) bool { + hashed := maphash.String(s.seed, key) + + s.mu.RLock() + defer s.mu.RUnlock() + + if _, ok := s.current[hashed]; ok { + return true + } + + _, ok := s.previous[hashed] + return ok +} + +func (s *rotatingSet) add(key string) { + hashed := maphash.String(s.seed, key) + + s.mu.Lock() + defer s.mu.Unlock() + + // Rotate on age or on size, whichever comes first. The size trigger is what + // keeps the footprint bounded when a high-coverage experiment is running. + if s.now().Sub(s.lastRotate) >= s.rotateEvery || len(s.current) >= s.maxEntries { + s.previous = s.current + s.current = make(map[uint64]struct{}) + s.lastRotate = s.now() + } + + s.current[hashed] = struct{}{} +} + +func (s *rotatingSet) len() int { + s.mu.RLock() + defer s.mu.RUnlock() + + return len(s.current) + len(s.previous) +} diff --git a/go.mod b/go.mod index c722df5..cbc8ad8 100644 --- a/go.mod +++ b/go.mod @@ -12,11 +12,12 @@ require ( github.com/getsentry/sentry-go v0.21.0 github.com/go-redis/redis/v8 v8.11.3 github.com/google/uuid v1.6.0 + github.com/growthbook/growthbook-golang v0.2.9 github.com/jackc/pgx/v4 v4.18.3 github.com/klauspost/compress v1.17.8 github.com/panjf2000/ants/v2 v2.10.0 github.com/sirupsen/logrus v1.9.0 - github.com/stretchr/testify v1.8.2 + github.com/stretchr/testify v1.9.0 go.uber.org/atomic v1.6.0 go.uber.org/zap v1.13.0 golang.org/x/sync v0.9.0 @@ -45,6 +46,7 @@ require ( github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/tatsuworks/czlib v0.0.0-20190916144400-8a51758ea0d9 // indirect + github.com/tmaxmax/go-sse v0.10.0 // indirect go.uber.org/multierr v1.5.0 // indirect golang.org/x/crypto v0.28.0 // indirect golang.org/x/exp v0.0.0-20241108190413-2d47ceb2692f // indirect diff --git a/go.sum b/go.sum index a7e6cd6..3ac7cd3 100644 --- a/go.sum +++ b/go.sum @@ -68,6 +68,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM= github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/growthbook/growthbook-golang v0.2.9 h1:J/HGjxhHFgGtpEu/VJmGnxbTJjAeYZtb2qsLClrtdjo= +github.com/growthbook/growthbook-golang v0.2.9/go.mod h1:mY8oBSateRALL7hMwr8UaPmsdm+10ffmgWIT1N5iQZE= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9QQ7T3kePo= github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk= @@ -198,10 +200,13 @@ github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8= github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/tatsuworks/czlib v0.0.0-20190916144400-8a51758ea0d9 h1:i2aD44Moa5N5pt/WNwHLvIklzPymtr8vkkBlVdNElUE= github.com/tatsuworks/czlib v0.0.0-20190916144400-8a51758ea0d9/go.mod h1:6HrfShlf4bKeQEFdWn4JP/yet/mHW2RhxOQf0e3HWA0= +github.com/tmaxmax/go-sse v0.10.0 h1:j9F93WB4Hxt8wUf6oGffMm4dutALvUPoDDxfuDQOSqA= +github.com/tmaxmax/go-sse v0.10.0/go.mod h1:u/2kZQR1tyngo1lKaNCj1mJmhXGZWS1Zs5yiSOD+Eg8= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= diff --git a/model/entitlements.go b/model/entitlements.go index b7c6264..410ba2d 100644 --- a/model/entitlements.go +++ b/model/entitlements.go @@ -21,7 +21,9 @@ const ( EntitlementSourceDiscord EntitlementSource = "discord" EntitlementSourcePatreon EntitlementSource = "patreon" EntitlementSourceVoting EntitlementSource = "voting" - EntitlementSourceKey EntitlementSource = "key" + EntitlementSourceKey EntitlementSource = "key" + EntitlementSourcePolar EntitlementSource = "polar" + EntitlementSourceAffiliate EntitlementSource = "affiliate" ) type EntitlementTier string diff --git a/permission/permissionlevel.go b/permission/permissionlevel.go index c7e3df1..8f132e1 100644 --- a/permission/permissionlevel.go +++ b/permission/permissionlevel.go @@ -2,7 +2,7 @@ package permission type PermissionLevel int -const( +const ( Everyone PermissionLevel = iota Support Admin @@ -11,3 +11,18 @@ const( func (l PermissionLevel) Int() int { return int(l) } + +type PermissionSource string + +const ( + SourceNone PermissionSource = "" + SourceBotAdmin PermissionSource = "bot_admin" + SourceBotStaff PermissionSource = "bot_staff" + SourceAdministrator PermissionSource = "administrator" + SourceGuildOwner PermissionSource = "guild_owner" + SourceAddsupport PermissionSource = "addsupport" + SourceAdminRole PermissionSource = "admin_role" + SourceStaffTeam PermissionSource = "staff_team" + SourceSupportRole PermissionSource = "support_role" + SourceTeamRole PermissionSource = "team_role" +) diff --git a/permission/rediscache.go b/permission/rediscache.go index 079dbaf..8cf6157 100644 --- a/permission/rediscache.go +++ b/permission/rediscache.go @@ -52,3 +52,27 @@ func (c *RedisCache) DeleteCachedPermissionLevel(ctx context.Context, guildId, u key := fmt.Sprintf("permissions:%d:%d", guildId, userId) return c.client.Del(ctx, key).Err() } + +func (c *RedisCache) DeleteGuildPermissionCache(ctx context.Context, guildId uint64) error { + pattern := fmt.Sprintf("permissions:%d:*", guildId) + var cursor uint64 + for { + keys, nextCursor, err := c.client.Scan(ctx, cursor, pattern, 100).Result() + if err != nil { + return err + } + + if len(keys) > 0 { + if err := c.client.Del(ctx, keys...).Err(); err != nil { + return err + } + } + + cursor = nextCursor + if cursor == 0 { + break + } + } + + return nil +} diff --git a/permission/retriever.go b/permission/retriever.go index a314d46..4e69aa9 100644 --- a/permission/retriever.go +++ b/permission/retriever.go @@ -111,3 +111,89 @@ func GetPermissionLevel(ctx context.Context, retriever Retriever, member member. return Everyone, nil } + +// GetPermissionLevelWithSource mirrors the logic of GetPermissionLevel but also +// returns a PermissionSource indicating which check granted the permission. This +// function bypasses the cache because the cache only stores the level, not the source. +func GetPermissionLevelWithSource(ctx context.Context, retriever Retriever, member member.Member, guildId uint64) (PermissionLevel, PermissionSource, error) { + // Check if the user is a bot admin user + if retriever.IsBotAdmin(ctx, member.User.Id) { + return Admin, SourceBotAdmin, nil + } + + // Check if user has Administrator permission + if member.Permissions > 0 && permission.HasPermissionRaw(member.Permissions, permission.Administrator) { + return Admin, SourceAdministrator, nil + } + + // Check if user is guild owner + if guildOwner, err := retriever.GetGuildOwner(ctx, guildId); err == nil { + if member.User.Id == guildOwner { + return Admin, SourceGuildOwner, nil + } + } else { + return Everyone, SourceNone, err + } + + // Check user perms for admin + if adminUser, err := retriever.Db().Permissions.IsAdmin(ctx, guildId, member.User.Id); err == nil { + if adminUser { + return Admin, SourceAddsupport, nil + } + } else { + return Everyone, SourceNone, err + } + + // Check roles from DB + adminRoles, err := retriever.Db().RolePermissions.GetAdminRoles(ctx, guildId) + if err != nil { + return Everyone, SourceNone, err + } + + for _, adminRoleId := range adminRoles { + if member.HasRole(adminRoleId) { + return Admin, SourceAdminRole, nil + } + } + + // Check user perms for support + if isSupport, err := retriever.Db().Permissions.IsSupport(ctx, guildId, member.User.Id); err == nil { + if isSupport { + return Support, SourceAddsupport, nil + } + } else { + return Everyone, SourceNone, err + } + + // Check if user is a member of a support team + if isSupport, err := retriever.Db().SupportTeamMembers.IsSupport(ctx, guildId, member.User.Id); err == nil { + if isSupport { + return Support, SourceStaffTeam, nil + } + } else { + return Everyone, SourceNone, err + } + + // Check DB for support roles + supportRoles, err := retriever.Db().RolePermissions.GetSupportRoles(ctx, guildId) + if err != nil { + return Everyone, SourceNone, err + } + + for _, supportRoleId := range supportRoles { + if member.HasRole(supportRoleId) { + return Support, SourceSupportRole, nil + } + } + + // Check if user has a role assigned to a support team + if isSupport, err := retriever.Db().SupportTeamRoles.IsSupportAny(ctx, guildId, member.Roles); err == nil { + if isSupport { + return Support, SourceTeamRole, nil + } + } else { + return Everyone, SourceNone, err + } + + return Everyone, SourceNone, nil +} diff --git a/whitelabel/whitelabel.go b/whitelabel/whitelabel.go new file mode 100644 index 0000000..f963872 --- /dev/null +++ b/whitelabel/whitelabel.go @@ -0,0 +1,149 @@ +package whitelabel + +import ( + "context" + "errors" + + "github.com/TicketsBot-cloud/database" + "github.com/TicketsBot-cloud/gdl/objects/application" + "github.com/TicketsBot-cloud/gdl/rest" + "github.com/TicketsBot-cloud/gdl/rest/request" +) + +// Discord only accepts the three "limited" intent flags on PATCH /applications/@me. The +// non-limited counterparts are granted by intents review, and asking to turn a limited bit +// on for an application that is already approved (or is over the exposure threshold) is +// rejected with APPLICATION_MAX_INTENTS_EXPOSURE_REACHED. +const writableFlags = application.FlagIntentGatewayPresenceLimited | + application.FlagIntentGatewayGuildMembersLimited | + application.FlagGatewayMessageContentLimited + +const invalidFormBodyCode = 50035 + +const intentsExposureErrorCode = "APPLICATION_MAX_INTENTS_EXPOSURE_REACHED" + +const IntentsRejectedMessage = "Discord refused to enable the Server Members and Message " + + "Content intents for your bot: the application is exposed to too many users and must be " + + "reviewed for privileged intents first. Apply for them in the Discord Developer Portal, " + + "then try again." + +// DesiredIntentFlags returns the flags value to send for an application whose flags are +// currently current, or nil if current already grants the intents the bot needs and the +// field should be omitted from the request. +func DesiredIntentFlags(current application.Flag) *application.Flag { + desired := current & writableFlags + + if !current.Has(application.FlagIntentGatewayGuildMembers) { + desired |= application.FlagIntentGatewayGuildMembersLimited + } + + if !current.Has(application.FlagGatewayMessageContent) { + desired |= application.FlagGatewayMessageContentLimited + } + + if desired == current&writableFlags { + return nil + } + + return &desired +} + +// IsIntentsRejection reports whether err is Discord refusing to enable a privileged intent +// because the application is exposed to too many users and has not been reviewed. +func IsIntentsRejection(err error) bool { + var restError request.RestError + if !errors.As(err, &restError) || restError.ApiError.Code != invalidFormBodyCode { + return false + } + + for _, fieldError := range restError.ApiError.Errors { + if code, ok := fieldError.Code.(string); ok && code == intentsExposureErrorCode { + return true + } + } + + return false +} + +// ReapplyIntents reapplies the gateway intents to the whitelabel application, without +// touching the interactions endpoint URL. Used when resyncing a bot that is already set up. +func ReapplyIntents(ctx context.Context, token string) error { + app, err := rest.GetCurrentApplication(ctx, token, nil) + if err != nil { + return err + } + + var currentFlags application.Flag + if app.Flags != nil { + currentFlags = *app.Flags + } + + flags := DesiredIntentFlags(currentFlags) + if flags == nil { + return nil + } + + _, err = rest.EditCurrentApplication(ctx, token, nil, rest.EditCurrentApplicationData{ + Flags: flags, + }) + return err +} + +// SyncGuilds reconciles the whitelabel_guilds table for botId against the guilds the bot is +// actually a member of, fetched from Discord using its token. Guilds present on Discord but +// missing from the DB are added; guilds in the DB the bot is no longer in are removed. +// Deletion only happens after the full guild list has been enumerated successfully, so a +// partial fetch never purges valid rows. +func SyncGuilds(ctx context.Context, db *database.Database, token string, botId uint64) error { + discord := make(map[uint64]struct{}) + + var after uint64 + for { + guilds, err := rest.GetCurrentUserGuilds(ctx, token, nil, rest.CurrentUserGuildsData{ + After: after, + Limit: 200, + }) + if err != nil { + return err + } + + for _, g := range guilds { + discord[g.Id] = struct{}{} + after = g.Id + } + + if len(guilds) < 200 { + break + } + } + + stored, err := db.WhitelabelGuilds.GetGuilds(ctx, botId) + if err != nil { + return err + } + + storedSet := make(map[uint64]struct{}, len(stored)) + for _, id := range stored { + storedSet[id] = struct{}{} + } + + // Add guilds the bot is in that we don't have stored + for id := range discord { + if _, ok := storedSet[id]; !ok { + if err := db.WhitelabelGuilds.Add(ctx, botId, id); err != nil { + return err + } + } + } + + // Remove stored guilds the bot is no longer in + for _, id := range stored { + if _, ok := discord[id]; !ok { + if err := db.WhitelabelGuilds.Delete(ctx, botId, id); err != nil { + return err + } + } + } + + return nil +} diff --git a/workflowbus/producer.go b/workflowbus/producer.go new file mode 100644 index 0000000..6d1f339 --- /dev/null +++ b/workflowbus/producer.go @@ -0,0 +1,112 @@ +package workflowbus + +import ( + "context" + "encoding/json" + "sync" + "time" + + "github.com/go-redis/redis/v8" + "github.com/google/uuid" + "go.uber.org/zap" +) + +const maxLenApproxWorkflows int64 = 50000 + +type Producer struct { + redis *redis.Client + logger *zap.Logger + signer *Signer +} + +func NewProducer(redisClient *redis.Client, logger *zap.Logger, signer *Signer) (*Producer, error) { + if redisClient == nil { + return &Producer{logger: logger, signer: signer}, nil + } + + return &Producer{redis: redisClient, logger: logger, signer: signer}, nil +} + +func (p *Producer) Close() {} + +func (p *Producer) Emit(_ context.Context, triggerType string, guildId uint64, causationId string, payload any) { + if p == nil || p.redis == nil { + return + } + + payloadBytes, err := json.Marshal(payload) + if err != nil { + if p.logger != nil { + p.logger.Error("workflowbus: failed to marshal trigger payload", zap.String("trigger", triggerType), zap.Error(err)) + } + return + } + + if causationId == "" { + causationId = uuid.NewString() + } + + env := Envelope{ + Version: EnvelopeVersion, + TriggerType: triggerType, + GuildId: guildId, + CausationId: causationId, + OccurredAt: time.Now().UTC(), + Payload: payloadBytes, + } + + if err := p.signer.Sign(&env); err != nil { + if p.logger != nil { + p.logger.Error("workflowbus: failed to sign envelope", zap.Error(err)) + } + return + } + + envBytes, err := json.Marshal(env) + if err != nil { + if p.logger != nil { + p.logger.Error("workflowbus: failed to marshal envelope", zap.Error(err)) + } + return + } + + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + err := p.redis.XAdd(ctx, &redis.XAddArgs{ + Stream: TopicWorkflowTriggers, + MaxLenApprox: maxLenApproxWorkflows, + ID: "*", + Values: map[string]interface{}{"data": string(envBytes)}, + }).Err() + + if err != nil && p.logger != nil { + p.logger.Error("workflowbus: produce failed", + zap.String("trigger", triggerType), + zap.Uint64("guild_id", guildId), + zap.Error(err)) + } + }() +} + +var ( + globalProducer *Producer + globalMu sync.RWMutex +) + +func SetGlobal(p *Producer) { + globalMu.Lock() + defer globalMu.Unlock() + globalProducer = p +} + +func Emit(ctx context.Context, triggerType string, guildId uint64, causationId string, payload any) { + globalMu.RLock() + p := globalProducer + globalMu.RUnlock() + if p == nil { + return + } + p.Emit(ctx, triggerType, guildId, causationId, payload) +} diff --git a/workflowbus/signer.go b/workflowbus/signer.go new file mode 100644 index 0000000..e9ee0c7 --- /dev/null +++ b/workflowbus/signer.go @@ -0,0 +1,115 @@ +package workflowbus + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" +) + +// SecretEnvVar is the environment variable every service in this system reads +// to find the shared HMAC secret for workflowbus envelopes. Using the same name +// in every service prevents the fiddly asymmetric configuration failure mode +// where producer and consumer have different secrets and every envelope fails +// verification with no obvious cause. +const SecretEnvVar = "WORKFLOWBUS_HMAC_SECRET" + +// ErrInvalidSignature is returned by Verify when the envelope's signature +// doesn't match the re-computed HMAC. Consumers typically drop the envelope +// and log when this happens - never execute. +var ErrInvalidSignature = errors.New("workflowbus: envelope signature invalid") + +// ErrMissingSignature is returned when a signer is configured but the envelope +// arrived unsigned. Consumers can choose to reject strictly (recommended once +// the rollout has completed) or log and accept (during rollout). +var ErrMissingSignature = errors.New("workflowbus: envelope missing signature") + +// Signer signs and verifies envelopes with HMAC-SHA256. Nil-safe: a Signer +// created with an empty secret acts as a no-op - Sign leaves Signature empty +// and Verify accepts any envelope. Services read the secret from SecretEnvVar +// and construct a Signer once at startup. +type Signer struct { + secret []byte +} + +// NewSigner returns a Signer for the given secret. If secret is empty the +// returned Signer is a pass-through: Sign does nothing, Verify accepts all. +// This is the safe default during gradual rollout - add the secret to one side +// at a time, then flip the strict flag once both sides have it. +func NewSigner(secret []byte) *Signer { + if len(secret) == 0 { + return &Signer{} + } + // Defensive copy - callers may reuse the backing slice. + s := make([]byte, len(secret)) + copy(s, secret) + return &Signer{secret: s} +} + +// Active reports whether this Signer will produce / require signatures. +func (s *Signer) Active() bool { + return s != nil && len(s.secret) > 0 +} + +// Sign computes the envelope's HMAC and writes it to env.Signature. +// No-op when the Signer is inactive. +func (s *Signer) Sign(env *Envelope) error { + if !s.Active() { + return nil + } + mac, err := s.compute(env) + if err != nil { + return err + } + env.Signature = base64.StdEncoding.EncodeToString(mac) + return nil +} + +// Verify checks the envelope's signature against a freshly-computed HMAC. +// Returns nil when: +// - the Signer is inactive (no secret configured; accept all), OR +// - the envelope's signature matches the computed HMAC. +// +// When strictOnMissing is true, envelopes with an empty Signature are rejected +// with ErrMissingSignature - use this after the rollout is complete to lock +// out unsigned traffic entirely. +func (s *Signer) Verify(env *Envelope, strictOnMissing bool) error { + if !s.Active() { + return nil + } + if env.Signature == "" { + if strictOnMissing { + return ErrMissingSignature + } + return nil + } + sig, err := base64.StdEncoding.DecodeString(env.Signature) + if err != nil { + return ErrInvalidSignature + } + mac, err := s.compute(env) + if err != nil { + return err + } + if !hmac.Equal(sig, mac) { + return ErrInvalidSignature + } + return nil +} + +// compute serialises the envelope with Signature empty and HMACs the result. +// The empty-Signature canonicalisation means the signed blob and the +// to-be-signed blob agree byte-for-byte - without this, Sign on an already- +// signed envelope would be ambiguous. +func (s *Signer) compute(env *Envelope) ([]byte, error) { + clone := *env + clone.Signature = "" + raw, err := json.Marshal(clone) + if err != nil { + return nil, err + } + mac := hmac.New(sha256.New, s.secret) + mac.Write(raw) + return mac.Sum(nil), nil +} diff --git a/workflowbus/signer_test.go b/workflowbus/signer_test.go new file mode 100644 index 0000000..bcd14e7 --- /dev/null +++ b/workflowbus/signer_test.go @@ -0,0 +1,80 @@ +package workflowbus + +import ( + "encoding/json" + "testing" + "time" +) + +func mkEnvelope() Envelope { + return Envelope{ + Version: EnvelopeVersion, + TriggerType: TriggerTicketCreated, + GuildId: 777, + CausationId: "abc", + OccurredAt: time.Date(2026, 4, 19, 10, 0, 0, 0, time.UTC), + Payload: json.RawMessage(`{"ticket_id": 123}`), + } +} + +func TestSigner_Inactive_AcceptsAnything(t *testing.T) { + s := NewSigner(nil) + if s.Active() { + t.Fatal("empty-secret signer should be inactive") + } + env := mkEnvelope() + if err := s.Sign(&env); err != nil { + t.Fatalf("Sign on inactive signer should be no-op: %v", err) + } + if env.Signature != "" { + t.Fatal("inactive Sign should leave Signature empty") + } + if err := s.Verify(&env, true); err != nil { + t.Fatalf("inactive Verify should accept unsigned: %v", err) + } +} + +func TestSigner_SignAndVerify(t *testing.T) { + s := NewSigner([]byte("shared-secret")) + env := mkEnvelope() + if err := s.Sign(&env); err != nil { + t.Fatalf("Sign failed: %v", err) + } + if env.Signature == "" { + t.Fatal("active Sign should populate Signature") + } + if err := s.Verify(&env, true); err != nil { + t.Fatalf("freshly-signed envelope failed verify: %v", err) + } +} + +func TestSigner_RejectsTamperedPayload(t *testing.T) { + s := NewSigner([]byte("shared-secret")) + env := mkEnvelope() + _ = s.Sign(&env) + env.Payload = json.RawMessage(`{"ticket_id": 999}`) + if err := s.Verify(&env, true); err == nil { + t.Fatal("expected tampered payload to fail verification") + } +} + +func TestSigner_RejectsMissingSignatureInStrictMode(t *testing.T) { + s := NewSigner([]byte("shared-secret")) + env := mkEnvelope() // no Sign called + if err := s.Verify(&env, true); err == nil { + t.Fatal("strict mode should reject unsigned envelope") + } + if err := s.Verify(&env, false); err != nil { + t.Fatalf("non-strict mode should accept unsigned: %v", err) + } +} + +func TestSigner_RejectsWrongSecret(t *testing.T) { + produced := NewSigner([]byte("producer-secret")) + consumed := NewSigner([]byte("consumer-secret")) + env := mkEnvelope() + _ = produced.Sign(&env) + if err := consumed.Verify(&env, true); err == nil { + t.Fatal("verify with mismatched secret should fail") + } +} diff --git a/workflowbus/triggers.go b/workflowbus/triggers.go new file mode 100644 index 0000000..4faa6cf --- /dev/null +++ b/workflowbus/triggers.go @@ -0,0 +1,81 @@ +package workflowbus + +import ( + "encoding/json" + "time" +) + +const ( + TopicWorkflowTriggers = "stream:rpc:workflows" + + TriggerTicketCreated = "ticket.created" + TriggerTicketClaimed = "ticket.claimed" + TriggerTicketClosed = "ticket.closed" + TriggerTicketReopened = "ticket.reopened" + TriggerTicketTransferred = "ticket.transferred" + TriggerCron = "cron" + TriggerWebhook = "webhook" + + EnvelopeVersion = 1 +) + +// Envelope is the wire format for every message on TopicWorkflowTriggers. +// Payload is a trigger-type-specific JSON document. +// +// Signature is a base64 HMAC-SHA256 of the canonical envelope bytes (i.e. the +// envelope with Signature field empty, marshalled deterministically). It's +// populated by the Signer and verified by the executor when a shared secret is +// configured. When the secret is unset on both sides, Signature stays empty and +// is ignored - preserves backward compatibility during rollout. +type Envelope struct { + Version int `json:"version"` + TriggerType string `json:"trigger_type"` + GuildId uint64 `json:"guild_id,string"` + CausationId string `json:"causation_id"` + WorkflowId int64 `json:"workflow_id,string,omitempty"` + OccurredAt time.Time `json:"occurred_at"` + Payload json.RawMessage `json:"payload"` + Signature string `json:"signature,omitempty"` +} + +type TicketCreatedPayload struct { + TicketId int `json:"ticket_id"` + OpenerId uint64 `json:"opener_id,string"` + PanelId *int `json:"panel_id,omitempty"` + ChannelId *uint64 `json:"channel_id,string,omitempty"` + IsThread bool `json:"is_thread"` + Form map[string]string `json:"form,omitempty"` +} + +type TicketClaimedPayload struct { + TicketId int `json:"ticket_id"` + ClaimedBy uint64 `json:"claimed_by,string"` +} + +type TicketClosedPayload struct { + TicketId int `json:"ticket_id"` + ClosedBy uint64 `json:"closed_by,string"` + Reason *string `json:"reason,omitempty"` +} + +type TicketReopenedPayload struct { + TicketId int `json:"ticket_id"` + ReopenedBy uint64 `json:"reopened_by,string"` +} + +type TicketTransferredPayload struct { + TicketId int `json:"ticket_id"` + FromUserId uint64 `json:"from_user_id,string"` + ToUserId uint64 `json:"to_user_id,string"` + TransferredBy uint64 `json:"transferred_by,string"` +} + +type CronPayload struct { + AutomationId int64 `json:"automation_id,string"` +} + +type WebhookPayload struct { + AutomationId int64 `json:"automation_id,string"` + Headers map[string]string `json:"headers,omitempty"` + Body json.RawMessage `json:"body,omitempty"` +}