diff --git a/.changeset/growth-signals-core.md b/.changeset/growth-signals-core.md new file mode 100644 index 00000000000..972243350a1 --- /dev/null +++ b/.changeset/growth-signals-core.md @@ -0,0 +1,5 @@ +--- +"server": patch +--- + +Add the `growthsignals` package, which describes notable moments in Gram — organizations and projects created, MCP servers deployed, security policies written, members invited and joining — as a single PostHog `gram_activity` event with a stable property shape. It carries the activity taxonomy, the map from audit action to activity (including the pass-through name that gives uncurated actions coverage and the exclusion list that keeps high-volume noise out), the event builder, a repo-backed enricher behind a TTL cache, and the emitter that skips the demo organization and logs rather than returns capture failures. Nothing calls it yet, so no events are emitted and no behaviour changes. diff --git a/docs/superpowers/specs/2026-09-03-posthog-slack-notifications-design.md b/docs/superpowers/specs/2026-09-03-posthog-slack-notifications-design.md new file mode 100644 index 00000000000..70dc4e7cd5f --- /dev/null +++ b/docs/superpowers/specs/2026-09-03-posthog-slack-notifications-design.md @@ -0,0 +1,396 @@ +# Revamp PostHog Slack notifications (GRW-66) + +## Context + +Gram's Slack notifications in `#ops-significant-events`, `#ops-aicp-events` and +`#ops-all-events` grew one destination at a time over eighteen months. Eleven +PostHog destinations now post Gram events across those three channels, with +overlapping filters, three different message formats, and two different Slack +workspace integrations. + +The bigger problem is coverage. The events Growth actually wants to see do not +exist in PostHog at all. There is no event for a project being created, an MCP +server being deployed, a security policy being written, or a member joining an +existing organization. Signup fires before we know whether the user was invited +or arrived organically. + +Meanwhile every one of those mutations is already recorded by Gram's audit +logger and published to Pub/Sub as an `audit_log.*_event_v1` outbox event. The +`gram streams` process already consumes that stream. The signal exists; nothing +forwards it to PostHog. + +This design replaces the eleven destinations with a single well-shaped event +and three purpose-built destinations, and closes the coverage gap by deriving +activity from the audit log. + +## Goals + +- One PostHog event, `gram_activity`, describing everything notable that happens + in Gram, with a stable property shape. +- A significant-events channel that carries only the six moments Growth cares + about, and a firehose channel that carries everything. +- Distinguish invited signups from organic ones. +- No new Temporal actions and no work in any request hot path. + +## Non-goals + +- Changing the dashboard-side `telemetry.capture` events (`mcp_event`, + `toolset_event`, `onboarding_event`). They stay as they are and remain + available for funnels. This design does not migrate them. +- Building a PostHog dashboard. Alex's existing dashboard can be extended + against the new event separately. +- Customer-facing notifications. This is internal ops signal only. + +## The event + +A single event named `gram_activity`, emitted server-side. + +Distinct id is the actor's email when one is resolvable, so the event attaches +to the same PostHog person as the existing signup and onboarding events. It +falls back to the organization id for system- and role-actor events that have +no human behind them. + +Base properties, present on every emission: + +| Property | Meaning | +| ------------------- | ------------------------------------------------------ | +| `activity` | The taxonomy name, e.g. `project_created` | +| `organization_id` | Gram organization id | +| `organization_slug` | Organization slug | +| `organization_name` | Organization display name | +| `project_id` | Project id, when the activity is project-scoped | +| `project_slug` | Project slug, when project-scoped | +| `actor_email` | Acting user's email, when resolvable | +| `actor_name` | Acting user's display name | +| `subject_name` | Display name of the thing acted on | +| `acting_surface` | `dashboard`, `api_key`, `platform_mcp`, `assistant`, … | +| `dashboard_url` | Deep link to the subject in the Gram dashboard | +| `audit_action` | The raw audit action, e.g. `mcp-server:create` | + +Deliberately absent: any "is this the first of its kind" flag. An earlier draft +carried `is_first_in_project` and `is_first_in_organization` to gate the +significant channel to the first MCP server per project. Every MCP creation is +now significant, so nothing consumes those flags, and each would have cost a +count query per event on a path that is otherwise pure lookups. Where first-ness +genuinely matters it is baked into the activity name instead, as with +`device_first_seen` and `agent_first_detected`. + +Per-activity extras are added on top: `signup_source` (`invited` / `organic`), +`mcp_kind` (`hosted`, `remote`, `tunneled`, `unproxied`, `meta`), `role` on +member joins, and `policy_name` on security policies. + +The taxonomy lives in Go. Which activities count as _significant_ lives in the +PostHog destination filter, so Growth can retune the significant channel without +a deploy. + +## Architecture + +``` +service tx ──> audit logger ──> outbox row (same tx) + │ + ▼ + Pub/Sub webhook events topic + │ + ▼ + gram streams: webhookEventHandler + │ + ┌──────────────┬───────┴────────┬──────────────────┐ + ▼ ▼ ▼ ▼ + svix relay payg key refresh billing notif growthsignals (new) + │ + ▼ + PostHog gram_activity +``` + +`growthsignals` is a new package at `server/internal/growthsignals/`. It joins +the existing fan-out in `server/cmd/gram/streams.go` rather than opening its own +subscription, so it adds no new infrastructure. + +**Cost line.** `Temporal actions/month: 0 (outbox → existing streams handler, no +Temporal).` Scales with audit-log writes, which already flow through this +subscription. + +### Matching on action, not event type + +The handler decodes `event.Payload` into `events.AuditLogCreatedPayloadV1` and +switches on `payload.Action`, not on `event.EventType`. The event type is a +coarse bucket — MCP metadata updates and MCP server creates both publish under +`audit_log.mcp_server_event_v1` — while the action is the precise discriminator. +This also means the handler covers all 71 audit event types uniformly. + +### Activity map + +A curated map gives friendly names to the actions Growth named: + +| Audit action | Activity | +| --------------------------------- | -------------------------------------------- | +| `project:create` | `project_created` | +| `mcp-server:create` | `mcp_server_created` (`mcp_kind: hosted`) | +| `remote-mcp:create` | `mcp_server_created` (`mcp_kind: remote`) | +| `tunneled-mcp:create` | `mcp_server_created` (`mcp_kind: tunneled`) | +| `unproxied-mcp:create` | `mcp_server_created` (`mcp_kind: unproxied`) | +| `meta-mcp:create` | `mcp_server_created` (`mcp_kind: meta`) | +| `risk_policy:create` | `security_policy_created` | +| `risk_policy:update` | `security_policy_updated` | +| `mcp-server:update` | `mcp_server_updated` | +| `mcp_metadata:update` | `mcp_server_updated` | +| `mcp-server:update-tool-metadata` | `mcp_server_updated` | +| `organization_invitation:create` | `member_invited` | + +Every other audit action passes through with `activity` set to a normalized form +of the raw action, so the firehose has full coverage without an allowlist to +maintain. A small exclusion list drops known high-volume noise that has no ops +value: assistant tool calls, assistant wakes, chat session access, and platform +MCP diagnostics reads. + +### Activities the audit log does not cover + +Three moments are not audited today and are emitted directly, in-process, by +calling the same `growthsignals` emitter: + +- **`organization_created`** — from both org-provisioning paths in the auth + service (signup with a company name, and platform-admin invite). +- **`user_signed_up`** — at first-time user creation, carrying + `signup_source: invited` when a pending invitation exists for that email at + that moment, and `organic` otherwise. +- **`member_joined_organization`** — when a pending invite is accepted, at login + or via the invite callback, carrying the granted `role`. + +These reuse the existing "telemetry failures are logged, never returned" rule +already followed by `captureSignupTelemetry`: a dropped analytics event must +never fail the request that produced it. + +### Devices and agents + +New devices seen and new agents detected are in scope for the firehose. They are +a materially different shape from everything above, so they ship as their own +PRs at the end of the stack rather than as a later phase. + +Neither is audited. The only device-related audit actions cover _integration +config_ (`device_integration:upsert` and friends), and there is no +agent-detection action at all. So neither can ride the audit path. + +They also do not arrive as single-row mutations: + +- **Devices** are not one table. There are four, written by three unrelated + ingest paths: `mdm_devices` from bulk MDM snapshot syncs + (`server/internal/deviceintegrations/sync.go`), the three + `device_agent_*_syncs` tables from the device agent's ~60s heartbeat + (`server/internal/agent/impl.go`), and `device_owners` from AI-provider + account attribution (`server/internal/hooks/account_attribution.go`). None of + them carries a project id. + + None can currently tell that a row is new. The MDM upsert is `:execrows` with + `ON CONFLICT DO UPDATE`, so it returns 1 on both insert and update; detecting + newness needs `RETURNING (xmax = 0)`. The heartbeat writes are worse: they are + `:exec` with a one-minute throttle guard in the `WHERE`, so a row count would + conflate "new" with "throttled". + + Even with newness solved, one MDM sync inserts every device it sees, so + emitting per new row would post an organization's entire fleet to Slack the + first time an integration is connected. + +- **Agents** are not stored as detection rows at all. They are derived at read + time by aggregating telemetry in ClickHouse + (`server/internal/access/ai_detections.go` via + `telemetryrepo.ListAIDetectionSummariesParams`). There is no insert to hook. + +The device and agent PRs therefore add: + +- `device_first_seen`, emitted from the MDM sync path for rows the sync newly + inserted, and **suppressed for a config's first successful sync**, which is a + backfill rather than a stream of new devices. A per-sync cap guards against a + large fleet expansion posting hundreds of messages. +- `agent_first_detected`, emitted from the AI scan write path + (`server/internal/telemetry/repo/ai_detections.go`). That repo already reads + existing rows before upserting so it can preserve `first_seen`, which means a + detection with no prior row is identifiable without a schema change. The + existing lookup is keyed per device and user, so an organization-first signal + needs one extra ClickHouse query keyed on organization and target only. No new + table is required. + +Both route to `#ops-aicp-events` only, never to the significant channel. The +firehose destination needs no change to pick them up, because they are the same +`gram_activity` event. + +## PR stack + +The work ships as five stacked PRs, each branched from the one before, so review +stays tractable and the risky parts land last. + +| PR | Scope | Touches | +| --- | ------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------- | +| 1 | Core `growthsignals` package: activity taxonomy, action map, event builder, emitter and enricher interfaces, TTL cache. No wiring, no behaviour change. | new package only | +| 2 | Audit-event stream handler plus fan-out wiring. Turns on every audit-derived activity. | `growthsignals/`, `cmd/gram/streams.go` | +| 3 | Direct emits: `organization_created`, `user_signed_up` with invited-vs-organic, `member_joined_organization`. | `auth/`, `organizations/`, `auth/identity/` | +| 4 | Devices: newness detection across the MDM and heartbeat write paths, backfill suppression, per-sync cap. | `deviceintegrations/`, `agent/`, `hooks/` | +| 5 | Agents: org-scoped first-detection query and emit. | `telemetry/repo/ai_detections.go` | + +PR 1 is the contract every later PR codes against, so it merges first. PRs 2 and +3 are independent of each other. PRs 4 and 5 are the highest-risk and land last, +behind signal that the pipeline already works. + +Each PR carries a changeset with `"server": patch`. PR 4 changes SQL queries, so +it must commit the regenerated sqlc output. + +### Enrichment and caching + +The audit payload carries organization id, project id, actor id, and display +names, but not the organization slug or name, the project slug, or the actor's +email. Those need lookups. + +The handler keeps a small in-process TTL cache keyed by organization id and by +project id, so a burst of events from one org costs one query rather than one +per event. Actor email is resolved from the user repo, also cached. A failed +lookup degrades the event rather than dropping it: the property is omitted and +the event still ships. + +### Filtering + +The demo organization is skipped entirely, so the daily reseed does not spam the +channels. Internal Speakeasy users are excluded at the PostHog destination level +via `filter_test_accounts: true`, which already carries the maintained list of +internal email patterns. + +## PostHog rebuild + +### Teardown + +These eleven Gram destinations are disabled, verified quiet, then deleted: + +| Destination | Channel | +| -------------------------------------------------- | --------------------------------------- | +| Gram - sign up -> sig-events | `#ops-significant-events` | +| Gram - first time functions -> sig-events | `#ops-significant-events` | +| Gram - Subscription Changes -> Sig Events | `#ops-significant-events` | +| Gram - feature request -> sig-events | `#ops-significant-events` | +| Gram - enterprise gate viewed -> sig-events | `#ops-significant-events` | +| Gram - Elements actions -> #ops-significant-events | `#ops-significant-events` | +| Gram - book demo page view -> sig-events | `#ops-significant-events` (already off) | +| Gram - overage reporting -> #significant-events | `#ops-significant-events` (already off) | +| Gram - all actions -> #gram-events | `#ops-aicp-events` | +| Gram - Elements actions -> #gram-events | `#ops-aicp-events` | +| Gram - all actions -> #all-events | `#ops-all-events` | + +`Gram - identity provider interest` is left alone: it is already disabled and +points at a different channel. + +### New destinations + +All six are created on Slack workspace integration 57009, which already posts +successfully to all three channels. Each is created disabled, tested with a +sample invocation, then enabled. + +| Name | Channel | Fires on | +| ------------------------- | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Gram significant | `#ops-significant-events` | `gram_activity` where `activity` is one of `organization_created`, `user_signed_up`, `member_joined_organization`, `project_created`, `mcp_server_created`, `security_policy_created` | +| Gram firehose | `#ops-aicp-events` | every `gram_activity` | +| Gram MCP settings | `#ops-all-events` | `gram_activity` where `activity` is `mcp_server_updated` | +| Gram subscription changes | `#ops-significant-events` | `gram_subscription_changed` | +| Gram feature requests | `#ops-significant-events` | `feature_requested` | +| Gram enterprise gate | `#ops-significant-events` | `enterprise_gate_viewed` | + +Every MCP server creation reaches the significant channel, not only the first per +project. MCP settings changes reach both `#ops-aicp-events` (via the firehose) +and `#ops-all-events` (via its own destination). + +The last three carry over today's behaviour on their own events; they are +recreated cleanly rather than folded into `gram_activity` because they are +genuinely different events with different properties. + +All six share one message template: actor, activity, org and project, and a +button linking to the subject in the Gram dashboard. + +## Testing + +- Unit tests for the action-to-activity mapping, including the pass-through and + exclusion behaviour. +- Handler tests following the `billingnotifications` handler test pattern, with + a fake PostHog client and a fake enricher capturing forwarded payloads. These + need no database and run in parallel. Envelopes are built with + `webhooksv1.Event_builder{...}.Build()`. +- One integration test against the test database covering real enrichment. +- Tests for the three direct emits, in particular that `signup_source` is + `invited` when a pending invitation exists and `organic` when it does not. +- A test that the demo organization is skipped. +- Manual end-to-end run on the local stack: create a project and an MCP server, + confirm `gram_activity` arrives in the PostHog Dev project (96574) with the + expected properties. + +## Rollout + +1. Ship the Go change. Events begin flowing to PostHog with no destinations + consuming them yet, so Slack stays quiet. +2. Confirm `gram_activity` volume and shape in PostHog over a day. +3. Create the six new destinations disabled, test-invoke each, enable them. +4. Disable the eleven old destinations in the same sitting, so there is no + window of doubled notifications. +5. After a week of clean signal, delete the eleven disabled destinations. +6. PRs 4 and 5 land devices and agents. No destination + change is needed, since both are `gram_activity` and the firehose already + accepts every activity. + +## Open risks + +- **Firehose volume is unknown until step 2.** If `#ops-aicp-events` proves too + loud, the fix is a filter change on one destination, not a deploy. +- **Enrichment adds queries to the streams handler.** The TTL cache should keep + this to a handful of queries per burst; if it does not, the fallback is to + drop enrichment to the ids the payload already carries. + +## Appendix: the shared Slack message template + +All six destinations use one template, so every Gram notification reads the same +way regardless of channel. PostHog's Slack destination templates support hog +expressions, including `??` and ternaries, which today's destinations already use. + +Plain-text fallback: + +``` +{event.properties.activity} in {event.properties.organization_name} by {event.properties.actor_email ?? 'system'} +``` + +Blocks: + +```json +[ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "*{event.properties.activity}* in *{event.properties.organization_name}*\n{event.properties.actor_email ?? 'system'} · {event.properties.subject_name}" + } + }, + { + "type": "context", + "elements": [ + { + "type": "mrkdwn", + "text": "`{event.properties.organization_slug}{event.properties.project_slug ? concat('/', event.properties.project_slug) : ''}` · via {event.properties.acting_surface}" + } + ] + }, + { + "type": "actions", + "elements": [ + { + "type": "button", + "text": { "type": "plain_text", "text": "Open in Gram" }, + "url": "{event.properties.dashboard_url}" + }, + { + "type": "button", + "text": { "type": "plain_text", "text": "View Event" }, + "url": "{event.url}" + } + ] + } +] +``` + +**This imposes a requirement on the emitter.** Slack rejects a button whose `url` +is empty, which would fail the whole message. So `dashboard_url` must be present +on _every_ `gram_activity` emission, never omitted. Where an activity has no +natural subject page, it falls back to the organization's dashboard page, and +where even the organization slug is unresolvable it falls back to the Gram site +root. This is the one property exempt from the "omit empty properties" rule. diff --git a/server/internal/growthsignals/actionmap.go b/server/internal/growthsignals/actionmap.go new file mode 100644 index 00000000000..0f75fcce0d3 --- /dev/null +++ b/server/internal/growthsignals/actionmap.go @@ -0,0 +1,158 @@ +package growthsignals + +import ( + "strings" + + "github.com/speakeasy-api/gram/server/internal/audit" +) + +// McpKind names the flavour of MCP server an ActivityMcpServerCreated event +// describes. Five audit actions create MCP servers and Growth reads them as one +// moment, so the flavour survives as a property instead of as five activities. +type McpKind string + +const ( + // McpKindHosted is a server Gram builds and hosts from a deployment. + McpKindHosted McpKind = "hosted" + + // McpKindRemote is a third-party server Gram proxies. + McpKindRemote McpKind = "remote" + + // McpKindTunneled is a server reachable through a customer-run tunnel. + McpKindTunneled McpKind = "tunneled" + + // McpKindUnproxied is a server registered for visibility but not proxied. + McpKindUnproxied McpKind = "unproxied" + + // McpKindMeta is a server that aggregates other servers. + McpKindMeta McpKind = "meta" +) + +// ActionMapping is everything one audit action contributes to an event. +type ActionMapping struct { + // Activity is the taxonomy name the action is reported as. ActivitySkip + // means the action is excluded and nothing should be emitted for it. + Activity Activity + + // Extra holds the static properties the mapping itself contributes, such + // as the mcp_kind of an MCP server creation, and is nil when it + // contributes none. Each call returns a freshly allocated map, so callers + // may add their own per-event extras to it. + Extra map[string]string +} + +// curatedActivities gives the moments Growth named a friendly activity, so the +// significant-events channel can filter on a stable name rather than on the +// shape of whichever audit action happened to produce it. +// +// It is deliberately short. Every other audited action still reaches PostHog +// through the pass-through name, so this map is a renaming layer rather than an +// allowlist that has to be extended for new coverage. +// +//nolint:exhaustive // a renaming layer over a subset of actions, not a case analysis +var curatedActivities = map[audit.Action]Activity{ + audit.ActionProjectCreate: ActivityProjectCreated, + + audit.ActionMcpServerCreate: ActivityMcpServerCreated, + audit.ActionRemoteMcpServerCreate: ActivityMcpServerCreated, + audit.ActionTunneledMcpServerCreate: ActivityMcpServerCreated, + audit.ActionUnproxiedMcpServerCreate: ActivityMcpServerCreated, + audit.ActionMetaMcpServerCreate: ActivityMcpServerCreated, + + audit.ActionRiskPolicyCreate: ActivitySecurityPolicyCreated, + audit.ActionRiskPolicyUpdate: ActivitySecurityPolicyUpdated, + + audit.ActionMcpServerUpdate: ActivityMcpServerUpdated, + audit.ActionMCPMetadataUpdate: ActivityMcpServerUpdated, + audit.ActionMcpServerToolMetadataUpdate: ActivityMcpServerUpdated, + + audit.ActionOrganizationInviteCreate: ActivityMemberInvited, +} + +// mcpCreateKinds records which MCP server flavour each creation action makes. +// +//nolint:exhaustive // only MCP creation actions have a server flavour +var mcpCreateKinds = map[audit.Action]McpKind{ + audit.ActionMcpServerCreate: McpKindHosted, + audit.ActionRemoteMcpServerCreate: McpKindRemote, + audit.ActionTunneledMcpServerCreate: McpKindTunneled, + audit.ActionUnproxiedMcpServerCreate: McpKindUnproxied, + audit.ActionMetaMcpServerCreate: McpKindMeta, +} + +// excludedActions are audited actions with no ops value that would otherwise +// dominate the firehose: an assistant's tool calls and wake timers, chat +// session reads, and the diagnostics call Platform MCP clients poll. +// +// They are dropped here rather than at the PostHog destination because the +// volume is the problem, and a destination filter still pays for every event +// that reaches it. +// +//nolint:exhaustive // an exclusion list is partial by definition +var excludedActions = map[audit.Action]struct{}{ + audit.ActionAssistantToolCall: {}, + audit.ActionWakeScheduled: {}, + audit.ActionWakeFired: {}, + audit.ActionWakeCancelled: {}, + audit.ActionChatSessionAccess: {}, + audit.ActionPlatformMcpDiagnosticsUserStatusRead: {}, +} + +// ActivityForAction resolves an audit action to what it contributes to an +// event. +// +// An action nobody curated still maps to an activity, derived from the action +// itself, so the firehose covers every audited mutation without an allowlist to +// maintain as services add audit coverage. +func ActivityForAction(action audit.Action) ActionMapping { + if _, excluded := excludedActions[action]; excluded { + return ActionMapping{Activity: ActivitySkip, Extra: nil} + } + + activity, curated := curatedActivities[action] + if !curated { + return ActionMapping{Activity: passThroughActivity(action), Extra: nil} + } + + if kind, isMcpCreate := mcpCreateKinds[action]; isMcpCreate { + return ActionMapping{ + Activity: activity, + Extra: map[string]string{PropertyMcpKind: string(kind)}, + } + } + + return ActionMapping{Activity: activity, Extra: nil} +} + +// passThroughActivity derives an activity name from a raw audit action: +// "toolset:create" becomes "toolset_create". +// +// Audit actions use several separators (":", "-", "_") and the activity +// property is read by humans in Slack and grouped on in PostHog, so every +// non-alphanumeric run collapses to a single underscore. An action with no +// alphanumeric content at all names nothing and is skipped rather than emitted +// as a blank activity. +func passThroughActivity(action audit.Action) Activity { + var name strings.Builder + name.Grow(len(action)) + + separatorPending := false + for _, r := range strings.ToLower(string(action)) { + if (r < 'a' || r > 'z') && (r < '0' || r > '9') { + separatorPending = true + continue + } + + if separatorPending && name.Len() > 0 { + name.WriteByte('_') + } + separatorPending = false + name.WriteRune(r) + } + + if name.Len() == 0 { + return ActivitySkip + } + + return Activity(name.String()) +} diff --git a/server/internal/growthsignals/actionmap_test.go b/server/internal/growthsignals/actionmap_test.go new file mode 100644 index 00000000000..7fd7ed0893f --- /dev/null +++ b/server/internal/growthsignals/actionmap_test.go @@ -0,0 +1,142 @@ +package growthsignals_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/speakeasy-api/gram/server/internal/audit" + "github.com/speakeasy-api/gram/server/internal/growthsignals" +) + +func TestActivityForActionCuratedMoments(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + action audit.Action + want growthsignals.Activity + }{ + {name: "project create", action: audit.ActionProjectCreate, want: growthsignals.ActivityProjectCreated}, + {name: "risk policy create", action: audit.ActionRiskPolicyCreate, want: growthsignals.ActivitySecurityPolicyCreated}, + {name: "risk policy update", action: audit.ActionRiskPolicyUpdate, want: growthsignals.ActivitySecurityPolicyUpdated}, + {name: "mcp server update", action: audit.ActionMcpServerUpdate, want: growthsignals.ActivityMcpServerUpdated}, + {name: "mcp metadata update", action: audit.ActionMCPMetadataUpdate, want: growthsignals.ActivityMcpServerUpdated}, + {name: "mcp tool metadata update", action: audit.ActionMcpServerToolMetadataUpdate, want: growthsignals.ActivityMcpServerUpdated}, + {name: "organization invite create", action: audit.ActionOrganizationInviteCreate, want: growthsignals.ActivityMemberInvited}, + } + + for _, tt := range tests { + mapping := growthsignals.ActivityForAction(tt.action) + + require.Equal(t, tt.want, mapping.Activity, "activity for %s (%s)", tt.name, tt.action) + require.Empty(t, mapping.Extra, "extras for %s (%s)", tt.name, tt.action) + } +} + +// The five MCP creation actions collapse to one activity, so the flavour has to +// survive as a property or the distinction is lost entirely. +func TestActivityForActionCarriesMcpKind(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + action audit.Action + want growthsignals.McpKind + }{ + {name: "hosted", action: audit.ActionMcpServerCreate, want: growthsignals.McpKindHosted}, + {name: "remote", action: audit.ActionRemoteMcpServerCreate, want: growthsignals.McpKindRemote}, + {name: "tunneled", action: audit.ActionTunneledMcpServerCreate, want: growthsignals.McpKindTunneled}, + {name: "unproxied", action: audit.ActionUnproxiedMcpServerCreate, want: growthsignals.McpKindUnproxied}, + {name: "meta", action: audit.ActionMetaMcpServerCreate, want: growthsignals.McpKindMeta}, + } + + for _, tt := range tests { + mapping := growthsignals.ActivityForAction(tt.action) + + require.Equal(t, growthsignals.ActivityMcpServerCreated, mapping.Activity, "activity for %s mcp create", tt.name) + require.Equal(t, map[string]string{ + growthsignals.PropertyMcpKind: string(tt.want), + }, mapping.Extra, "extras for %s mcp create", tt.name) + } +} + +func TestActivityForActionPassesThroughUncuratedActions(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + action audit.Action + want growthsignals.Activity + }{ + {name: "colon separator", action: audit.ActionToolsetCreate, want: "toolset_create"}, + {name: "hyphenated subject", action: audit.ActionMcpEndpointCreate, want: "mcp_endpoint_create"}, + {name: "already underscored", action: audit.ActionEnvironmentCreate, want: "environment_create"}, + {name: "mixed separators", action: audit.ActionRemoteSessionClientAttachJsonWebKeySet, want: "remote_session_client_attach_json_web_key_set"}, + {name: "deleted mcp server", action: audit.ActionMcpServerDelete, want: "mcp_server_delete"}, + {name: "unknown action", action: audit.Action("widget:frobnicate"), want: "widget_frobnicate"}, + {name: "repeated separators", action: audit.Action("widget::--frobnicate"), want: "widget_frobnicate"}, + {name: "leading and trailing separators", action: audit.Action(":widget:"), want: "widget"}, + {name: "uppercase", action: audit.Action("Widget:Frobnicate"), want: "widget_frobnicate"}, + } + + for _, tt := range tests { + mapping := growthsignals.ActivityForAction(tt.action) + + require.Equal(t, tt.want, mapping.Activity, "activity for %s (%s)", tt.name, tt.action) + require.Empty(t, mapping.Extra, "extras for %s (%s)", tt.name, tt.action) + } +} + +// An action with nothing to name is skipped rather than emitted as a blank +// activity, which would be indistinguishable from a bug in the taxonomy. +func TestActivityForActionSkipsUnnamableActions(t *testing.T) { + t.Parallel() + + tests := []audit.Action{"", ":", "---", "::--"} + + for _, action := range tests { + mapping := growthsignals.ActivityForAction(action) + + require.Equal(t, growthsignals.ActivitySkip, mapping.Activity, "activity for %q", action) + } +} + +func TestActivityForActionSkipsHighVolumeNoise(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + action audit.Action + }{ + {name: "assistant tool call", action: audit.ActionAssistantToolCall}, + {name: "assistant wake scheduled", action: audit.ActionWakeScheduled}, + {name: "assistant wake fired", action: audit.ActionWakeFired}, + {name: "assistant wake cancelled", action: audit.ActionWakeCancelled}, + {name: "chat session access", action: audit.ActionChatSessionAccess}, + {name: "platform mcp diagnostics read", action: audit.ActionPlatformMcpDiagnosticsUserStatusRead}, + } + + for _, tt := range tests { + mapping := growthsignals.ActivityForAction(tt.action) + + require.Equal(t, growthsignals.ActivitySkip, mapping.Activity, "activity for %s (%s)", tt.name, tt.action) + require.Empty(t, mapping.Extra, "extras for %s (%s)", tt.name, tt.action) + } +} + +// Callers merge their own per-event extras into the returned map, so it must +// not be the map the package keeps. +func TestActivityForActionReturnsFreshExtras(t *testing.T) { + t.Parallel() + + first := growthsignals.ActivityForAction(audit.ActionMcpServerCreate) + first.Extra[growthsignals.PropertyMcpKind] = "tampered" + first.Extra["added"] = "by caller" + + second := growthsignals.ActivityForAction(audit.ActionMcpServerCreate) + + require.Equal(t, map[string]string{ + growthsignals.PropertyMcpKind: string(growthsignals.McpKindHosted), + }, second.Extra) +} diff --git a/server/internal/growthsignals/activity.go b/server/internal/growthsignals/activity.go new file mode 100644 index 00000000000..534faf1d82c --- /dev/null +++ b/server/internal/growthsignals/activity.go @@ -0,0 +1,77 @@ +// Package growthsignals turns notable moments in Gram — projects created, MCP +// servers deployed, members joining, security policies written — into a single +// PostHog event that internal Slack destinations render as ops signal. +// +// Everything is reported as one event name with a stable property shape, so +// which moments count as significant is a filter on the PostHog destination +// rather than a deploy. The taxonomy of what happened lives here; the judgement +// of what matters does not. +// +// Nothing in this package sits on a request's critical path. An activity whose +// lookups fail still ships with the properties that did resolve, and an +// activity that cannot be captured is logged and dropped: analytics must never +// fail the work that produced it. +package growthsignals + +// EventName is the single PostHog event every activity is captured as. One +// name with a stable property shape keeps destination filters expressible as +// property matches instead of a growing list of event names. +const EventName = "gram_activity" + +// Activity is the taxonomy name an event carries in its `activity` property. +// It answers what happened in terms Growth reads, rather than in terms of the +// audit action or table that produced it. +type Activity string + +const ( + // ActivityOrganizationCreated is a new Gram organization being provisioned. + ActivityOrganizationCreated Activity = "organization_created" + + // ActivityUserSignedUp is a person's first Gram user record being created. + // It carries a signup_source distinguishing an invited arrival from an + // organic one. + ActivityUserSignedUp Activity = "user_signed_up" + + // ActivityMemberJoinedOrganization is a pending invitation being accepted. + // It carries the role the new member was granted. + ActivityMemberJoinedOrganization Activity = "member_joined_organization" + + // ActivityProjectCreated is a new project inside an organization. + ActivityProjectCreated Activity = "project_created" + + // ActivityMcpServerCreated is any flavour of MCP server being created. The + // flavour survives as the mcp_kind property, so the five creation paths + // read as one moment. + ActivityMcpServerCreated Activity = "mcp_server_created" + + // ActivitySecurityPolicyCreated is a new risk policy. + ActivitySecurityPolicyCreated Activity = "security_policy_created" + + // ActivitySecurityPolicyUpdated is a change to an existing risk policy. + ActivitySecurityPolicyUpdated Activity = "security_policy_updated" + + // ActivityMcpServerUpdated is a change to an MCP server, its metadata, or + // its tool metadata. + ActivityMcpServerUpdated Activity = "mcp_server_updated" + + // ActivityMemberInvited is an invitation being sent to join an + // organization. The join itself is ActivityMemberJoinedOrganization. + ActivityMemberInvited Activity = "member_invited" + + // ActivityDeviceFirstSeen is a device appearing in an organization's fleet + // for the first time. + ActivityDeviceFirstSeen Activity = "device_first_seen" + + // ActivityAgentFirstDetected is an AI agent being attributed to an + // organization for the first time. + ActivityAgentFirstDetected Activity = "agent_first_detected" +) + +// ActivitySkip marks an audit action that carries no ops value and must not be +// emitted. It is a decision, not a taxonomy name: it never reaches PostHog, and +// the emitter drops any activity carrying it. +// +// Excluding at the source rather than at the destination filter matters because +// the excluded actions are the high-volume ones — an assistant's tool calls +// alone would dwarf everything else on the topic. +const ActivitySkip Activity = "skip" diff --git a/server/internal/growthsignals/cache.go b/server/internal/growthsignals/cache.go new file mode 100644 index 00000000000..f9717c5bb8c --- /dev/null +++ b/server/internal/growthsignals/cache.go @@ -0,0 +1,112 @@ +package growthsignals + +import ( + "context" + "fmt" + "time" + + "github.com/hashicorp/golang-lru/v2/expirable" + "golang.org/x/sync/singleflight" +) + +const ( + // lookupLoadTimeout bounds one detached enrichment query. + lookupLoadTimeout = 5 * time.Second + + // lookupTTL bounds how stale a resolved name may be. A burst of events from + // one organization then costs one query rather than one per event, and a + // renamed organization or project catches up within the window. + lookupTTL = 5 * time.Minute + + // lookupCacheSize caps each cache so the long tail of organizations that + // emit a single event cannot grow it without bound. It sits well above the + // number of organizations active in any TTL window, so eviction is not part + // of the steady state. + lookupCacheSize = 4096 +) + +// lookupCache memoizes one kind of enrichment lookup. +// +// The value is cached whatever it holds, so an id that resolves to nothing is +// as cheap the second time as one that resolves to a name — the misses are what +// a firehose produces most of. Errors are never cached, so a database blip +// costs one failed event rather than a TTL of them. +type lookupCache[K comparable, V any] struct { + entries *expirable.LRU[K, V] + load func(context.Context, K) (V, error) + + // inflight collapses concurrent misses for the same key into one load. A + // burst of stream messages from one organization all miss together on the + // first event, and without this each message issues its own query for the + // same row. + inflight singleflight.Group +} + +func newLookupCache[K comparable, V any](load func(context.Context, K) (V, error)) *lookupCache[K, V] { + return &lookupCache[K, V]{ + entries: expirable.NewLRU[K, V](lookupCacheSize, nil, lookupTTL), + load: load, + inflight: singleflight.Group{}, + } +} + +// resolve returns the cached value for key, loading it on a miss. +func (c *lookupCache[K, V]) resolve(ctx context.Context, key K) (V, error) { + if cached, ok := c.entries.Get(key); ok { + return cached, nil + } + + // The key is stringified for the flight group only; the cache itself stays + // typed. Callers that arrive while a load is in progress wait for it rather + // than issuing their own. + flight := c.inflight.DoChan(fmt.Sprint(key), func() (any, error) { + // Re-read inside the flight. A caller that missed the cache before an + // earlier flight finished can still reach this point after that flight + // was removed from the group, and without this it would issue a second + // query for a row already cached. + if cached, ok := c.entries.Get(key); ok { + return cached, nil + } + + // The flight is shared, so it must not inherit the cancellation of + // whichever caller happened to start it: that caller going away would + // fail the lookup for everyone still waiting on it. Values are kept, so + // the load stays inside the originating trace, and the timeout keeps a + // detached query from outliving its usefulness. + loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), lookupLoadTimeout) + defer cancel() + + value, err := c.load(loadCtx, key) + if err != nil { + return nil, err + } + + c.entries.Add(key, value) + + return value, nil + }) + + var result singleflight.Result + select { + case <-ctx.Done(): + // Waiting on somebody else's query must not outlive this caller. The + // flight continues for whoever else is waiting on it. + var zero V + return zero, fmt.Errorf("growth signal lookup for %v: %w", key, ctx.Err()) + case result = <-flight: + } + + loaded, err := result.Val, result.Err + if err != nil { + var zero V + return zero, err + } + + value, ok := loaded.(V) + if !ok { + var zero V + return zero, fmt.Errorf("growth signal cache loaded %T for key %v", loaded, key) + } + + return value, nil +} diff --git a/server/internal/growthsignals/cache_internal_test.go b/server/internal/growthsignals/cache_internal_test.go new file mode 100644 index 00000000000..e56ffc9f68f --- /dev/null +++ b/server/internal/growthsignals/cache_internal_test.go @@ -0,0 +1,122 @@ +package growthsignals + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/hashicorp/golang-lru/v2/expirable" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The point of the cache is that a burst of events from one organization costs +// one query, so a repeat key must not reach the loader again. +func TestLookupCacheServesRepeatsFromMemory(t *testing.T) { + t.Parallel() + + var loads atomic.Int64 + cache := newLookupCache(func(_ context.Context, key string) (string, error) { + loads.Add(1) + return "resolved " + key, nil + }) + + for range 5 { + value, err := cache.resolve(t.Context(), "org_placeholder") + + require.NoError(t, err) + require.Equal(t, "resolved org_placeholder", value) + } + + require.Equal(t, int64(1), loads.Load()) +} + +func TestLookupCacheKeepsKeysApart(t *testing.T) { + t.Parallel() + + var loads atomic.Int64 + cache := newLookupCache(func(_ context.Context, key string) (string, error) { + loads.Add(1) + return "resolved " + key, nil + }) + + tests := []string{"org_first", "org_second", "org_first", "org_second"} + for _, key := range tests { + value, err := cache.resolve(t.Context(), key) + + require.NoError(t, err) + require.Equal(t, "resolved "+key, value, "value for %s", key) + } + + require.Equal(t, int64(2), loads.Load()) +} + +// An id that resolves to nothing is the common case on a firehose, so the empty +// answer has to be as cheap the second time as a found one. +func TestLookupCacheRemembersEmptyResults(t *testing.T) { + t.Parallel() + + var loads atomic.Int64 + cache := newLookupCache(func(_ context.Context, _ string) (string, error) { + loads.Add(1) + return "", nil + }) + + for range 3 { + value, err := cache.resolve(t.Context(), "org_deleted") + + require.NoError(t, err) + require.Empty(t, value) + } + + require.Equal(t, int64(1), loads.Load()) +} + +// Caching a failure would turn a database blip into a TTL of blank events. +func TestLookupCacheDoesNotRememberFailures(t *testing.T) { + t.Parallel() + + var loads atomic.Int64 + failure := errors.New("lookup unavailable") + cache := newLookupCache(func(_ context.Context, key string) (string, error) { + if loads.Add(1) == 1 { + return "", failure + } + + return "resolved " + key, nil + }) + + _, err := cache.resolve(t.Context(), "org_placeholder") + require.ErrorIs(t, err, failure) + + value, err := cache.resolve(t.Context(), "org_placeholder") + require.NoError(t, err) + require.Equal(t, "resolved org_placeholder", value) + require.Equal(t, int64(2), loads.Load()) +} + +// A renamed organization has to catch up, so entries must not be permanent. +func TestLookupCacheExpiresEntries(t *testing.T) { + t.Parallel() + + var loads atomic.Int64 + cache := &lookupCache[string, string]{ + entries: expirable.NewLRU[string, string](lookupCacheSize, nil, 10*time.Millisecond), + load: func(_ context.Context, _ string) (string, error) { + loads.Add(1) + return "resolved", nil + }, + } + + _, err := cache.resolve(t.Context(), "org_placeholder") + require.NoError(t, err) + require.Equal(t, int64(1), loads.Load()) + + require.EventuallyWithT(t, func(collect *assert.CollectT) { + _, err := cache.resolve(t.Context(), "org_placeholder") + assert.NoError(collect, err) + assert.Greater(collect, loads.Load(), int64(1)) + }, 5*time.Second, 5*time.Millisecond) +} diff --git a/server/internal/growthsignals/emitter.go b/server/internal/growthsignals/emitter.go new file mode 100644 index 00000000000..e8205a97182 --- /dev/null +++ b/server/internal/growthsignals/emitter.go @@ -0,0 +1,153 @@ +package growthsignals + +import ( + "context" + "log/slog" + "net/url" + + "github.com/google/uuid" + + "github.com/speakeasy-api/gram/server/internal/attr" + "github.com/speakeasy-api/gram/server/internal/constants" + "github.com/speakeasy-api/gram/server/internal/urn" +) + +// Emitter ships activities to PostHog. +// +// It is the single door every producer goes through, so the rules that must +// hold for all of them — the demo organization never reports, a failed lookup +// narrows an event instead of dropping it, a failed capture never reaches the +// caller — are enforced once here rather than at each emission. +type Emitter struct { + logger *slog.Logger + client PostHogClient + enricher Enricher + + // siteURL is the dashboard's base URL, used to build the fallback + // dashboard_url for activities that carry no subject page of their own. + siteURL *url.URL +} + +func NewEmitter(logger *slog.Logger, client PostHogClient, enricher Enricher, siteURL *url.URL) *Emitter { + componentLogger := logger.With(attr.SlogComponent("growth-signals")) + if siteURL == nil { + // Every activity would then report no dashboard link, and a Slack + // destination that renders one as a button omits it. Worth saying once + // at startup rather than leaving it to be noticed in the channel. + componentLogger.WarnContext(context.Background(), "growth signals have no site url; activities will carry no dashboard link") + } + + return &Emitter{ + logger: componentLogger, + client: client, + enricher: enricher, + siteURL: siteURL, + } +} + +// Emit enriches one activity and captures it. +// +// It returns nothing. Producers call this from paths whose real work has +// already succeeded — a project was created, a member joined — and a dropped +// analytics event must never be able to fail that work. Everything that goes +// wrong is logged here instead. +func (e *Emitter) Emit(ctx context.Context, event ActivityEvent) { + if event.Activity == "" || event.Activity == ActivitySkip { + return + } + + // The demo organization is reseeded daily from a fixture, so every one of + // its mutations is a scripted one. Reporting them would bury the real + // signal under a burst of identical activity every morning. + if event.OrganizationID == constants.DemoOrganizationID { + return + } + + captured := BuildEvent(event, e.enrich(ctx, event), e.siteURL) + + if err := e.client.CaptureEvent(ctx, captured.Name, captured.DistinctID, captured.Properties); err != nil { + e.logger.ErrorContext(ctx, "capture growth activity", + attr.SlogError(err), + attr.SlogEvent(string(event.Activity)), + attr.SlogOrganizationID(event.OrganizationID), + ) + } +} + +// enrich resolves the ids an event carries into the names it reports. Each +// lookup stands alone: one that fails costs its own properties and nothing +// else, so a degraded event still carries everything that did resolve. +func (e *Emitter) enrich(ctx context.Context, event ActivityEvent) Enrichment { + enrichment := Enrichment{ + Organization: OrganizationDetails{Slug: "", Name: ""}, + Project: ProjectDetails{Slug: "", Name: ""}, + ActorEmail: event.ActorEmail, + } + + if event.OrganizationID != "" { + organization, err := e.enricher.Organization(ctx, event.OrganizationID) + if err != nil { + e.logger.WarnContext(ctx, "resolve organization for growth activity", + attr.SlogError(err), + attr.SlogOrganizationID(event.OrganizationID), + ) + } else { + enrichment.Organization = organization + } + } + + if event.ProjectID != uuid.Nil { + project, err := e.enricher.Project(ctx, event.ProjectID) + if err != nil { + e.logger.WarnContext(ctx, "resolve project for growth activity", + attr.SlogError(err), + attr.SlogProjectID(event.ProjectID.String()), + ) + } else { + enrichment.Project = project + } + } + + if enrichment.ActorEmail == "" { + enrichment.ActorEmail = e.resolveActorEmail(ctx, event) + } + + return enrichment +} + +// resolveActorEmail finds the email address of the person behind an activity, +// or returns empty when there is none. +// +// An email principal already is the address. A user principal needs a lookup, +// except for the reserved subject-set id, which stands for every user rather +// than for one. A role principal has no person behind it at all, and the event +// falls back to the organization as its distinct id. +func (e *Emitter) resolveActorEmail(ctx context.Context, event ActivityEvent) string { + if event.ActorID == "" { + return "" + } + + switch event.ActorType { + case urn.PrincipalTypeEmail: + return event.ActorID + case urn.PrincipalTypeUser: + if event.ActorID == urn.AllUsersPrincipalID { + return "" + } + + email, err := e.enricher.UserEmail(ctx, event.ActorID) + if err != nil { + e.logger.WarnContext(ctx, "resolve actor email for growth activity", + attr.SlogError(err), + attr.SlogUserID(event.ActorID), + ) + return "" + } + + return email + case urn.PrincipalTypeRole: + return "" + default: + return "" + } +} diff --git a/server/internal/growthsignals/emitter_test.go b/server/internal/growthsignals/emitter_test.go new file mode 100644 index 00000000000..cd17cb1228a --- /dev/null +++ b/server/internal/growthsignals/emitter_test.go @@ -0,0 +1,244 @@ +package growthsignals_test + +import ( + "bytes" + "errors" + "log/slog" + "net/url" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/speakeasy-api/gram/server/internal/audit" + "github.com/speakeasy-api/gram/server/internal/constants" + "github.com/speakeasy-api/gram/server/internal/growthsignals" + "github.com/speakeasy-api/gram/server/internal/testenv" + "github.com/speakeasy-api/gram/server/internal/urn" +) + +func TestEmitterCapturesEnrichedActivity(t *testing.T) { + t.Parallel() + + projectID := uuid.MustParse("11111111-2222-3333-4444-555555555555") + client := &capturePostHog{} + enricher := &fakeEnricher{ + organization: growthsignals.OrganizationDetails{Slug: "acme", Name: "Acme Incorporated"}, + project: growthsignals.ProjectDetails{Slug: "widgets", Name: "Widgets"}, + userEmails: map[string]string{"user_placeholder": "person@example.test"}, + } + + growthsignals.NewEmitter(testenv.NewLogger(t), client, enricher, emitterSiteURL()).Emit(t.Context(), growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + ProjectID: projectID, + ActorID: "user_placeholder", + ActorType: urn.PrincipalTypeUser, + SubjectName: "Widgets", + ActingSurface: string(audit.SurfaceDashboard), + AuditAction: audit.ActionProjectCreate, + }) + + captured := client.Captured() + require.Len(t, captured, 1) + require.Equal(t, growthsignals.EventName, captured[0].Name) + require.Equal(t, "person@example.test", captured[0].DistinctID) + require.Equal(t, "acme", captured[0].Properties["organization_slug"]) + require.Equal(t, "widgets", captured[0].Properties["project_slug"]) + require.Equal(t, "person@example.test", captured[0].Properties["actor_email"]) +} + +// The demo organization is reseeded daily from a fixture, so every one of its +// mutations would post to Slack every morning. +func TestEmitterSkipsDemoOrganization(t *testing.T) { + t.Parallel() + + tests := []growthsignals.Activity{ + growthsignals.ActivityOrganizationCreated, + growthsignals.ActivityProjectCreated, + growthsignals.ActivityMcpServerCreated, + growthsignals.ActivityDeviceFirstSeen, + } + + client := &capturePostHog{} + emitter := growthsignals.NewEmitter(testenv.NewLogger(t), client, &fakeEnricher{}, emitterSiteURL()) + + for _, activity := range tests { + emitter.Emit(t.Context(), growthsignals.ActivityEvent{ + Activity: activity, + OrganizationID: constants.DemoOrganizationID, + }) + } + + require.Empty(t, client.Captured()) +} + +func TestEmitterSkipsExcludedActivities(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + activity growthsignals.Activity + }{ + {name: "explicitly skipped", activity: growthsignals.ActivitySkip}, + {name: "unset", activity: ""}, + } + + client := &capturePostHog{} + emitter := growthsignals.NewEmitter(testenv.NewLogger(t), client, &fakeEnricher{}, emitterSiteURL()) + + for _, tt := range tests { + emitter.Emit(t.Context(), growthsignals.ActivityEvent{ + Activity: tt.activity, + OrganizationID: "org_placeholder", + }) + + require.Empty(t, client.Captured(), "captured a %s activity", tt.name) + } +} + +// A lookup that fails narrows the event. Dropping it instead would lose the +// moment entirely over a property nobody filters on. +func TestEmitterShipsActivityWhenEnrichmentFails(t *testing.T) { + t.Parallel() + + client := &capturePostHog{} + enricher := &fakeEnricher{ + organizationErr: errors.New("organization lookup unavailable"), + projectErr: errors.New("project lookup unavailable"), + userEmailErr: errors.New("user lookup unavailable"), + } + + growthsignals.NewEmitter(testenv.NewLogger(t), client, enricher, emitterSiteURL()).Emit(t.Context(), growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + ProjectID: uuid.MustParse("11111111-2222-3333-4444-555555555555"), + ActorID: "user_placeholder", + ActorType: urn.PrincipalTypeUser, + AuditAction: audit.ActionProjectCreate, + }) + + captured := client.Captured() + require.Len(t, captured, 1) + require.Equal(t, "org_placeholder", captured[0].DistinctID) + require.Equal(t, "project_created", captured[0].Properties["activity"]) + require.Equal(t, "project:create", captured[0].Properties["audit_action"]) + require.NotContains(t, captured[0].Properties, "organization_slug") + require.NotContains(t, captured[0].Properties, "project_slug") + require.NotContains(t, captured[0].Properties, "actor_email") +} + +// Emission failures are logged rather than returned, because the work that +// produced the activity has already succeeded. +func TestEmitterLogsCaptureFailure(t *testing.T) { + t.Parallel() + + logs := &bytes.Buffer{} + logger := slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{ + AddSource: false, + Level: slog.LevelDebug, + ReplaceAttr: nil, + })) + client := &capturePostHog{failWith: errors.New("posthog unavailable")} + + growthsignals.NewEmitter(logger, client, &fakeEnricher{}, emitterSiteURL()).Emit(t.Context(), growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + }) + + require.Len(t, client.Captured(), 1) + require.Contains(t, logs.String(), "capture growth activity") + require.Contains(t, logs.String(), "posthog unavailable") +} + +func TestEmitterResolvesActorEmailByPrincipalType(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + actorID string + actorType urn.PrincipalType + suppliedEmail string + wantEmail string + wantLookups []string + }{ + { + name: "user principal is looked up", + actorID: "user_placeholder", + actorType: urn.PrincipalTypeUser, + wantEmail: "person@example.test", + wantLookups: []string{"user_placeholder"}, + }, + { + name: "email principal is already an address", + actorID: "person@example.test", + actorType: urn.PrincipalTypeEmail, + wantEmail: "person@example.test", + wantLookups: nil, + }, + { + name: "role principal has no person behind it", + actorID: "admin", + actorType: urn.PrincipalTypeRole, + wantEmail: "", + wantLookups: nil, + }, + { + name: "subject set principal stands for everyone", + actorID: urn.AllUsersPrincipalID, + actorType: urn.PrincipalTypeUser, + wantEmail: "", + wantLookups: nil, + }, + { + name: "supplied email skips the lookup", + actorID: "user_placeholder", + actorType: urn.PrincipalTypeUser, + suppliedEmail: "known@example.test", + wantEmail: "known@example.test", + wantLookups: nil, + }, + { + name: "no actor at all", + actorID: "", + actorType: "", + wantEmail: "", + wantLookups: nil, + }, + } + + for _, tt := range tests { + client := &capturePostHog{} + enricher := &fakeEnricher{userEmails: map[string]string{"user_placeholder": "person@example.test"}} + + growthsignals.NewEmitter(testenv.NewLogger(t), client, enricher, emitterSiteURL()).Emit(t.Context(), growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + ActorID: tt.actorID, + ActorType: tt.actorType, + ActorEmail: tt.suppliedEmail, + }) + + captured := client.Captured() + require.Len(t, captured, 1, "captures for %s", tt.name) + require.Equal(t, tt.wantEmail, propertyString(captured[0].Properties, "actor_email"), "actor email for %s", tt.name) + require.Equal(t, tt.wantLookups, enricher.UserEmailCalls(), "user lookups for %s", tt.name) + } +} + +// propertyString reads a property that may legitimately be absent, so a missing +// one compares as empty instead of as nil. +func propertyString(properties map[string]any, key string) string { + value, ok := properties[key].(string) + if !ok { + return "" + } + + return value +} + +// emitterSiteURL is the dashboard base URL used to build the fallback +// dashboard_url property. +func emitterSiteURL() *url.URL { + return &url.URL{Scheme: "https", Host: "app.example.test"} +} diff --git a/server/internal/growthsignals/enricher.go b/server/internal/growthsignals/enricher.go new file mode 100644 index 00000000000..c475908bdc8 --- /dev/null +++ b/server/internal/growthsignals/enricher.go @@ -0,0 +1,82 @@ +package growthsignals + +import ( + "context" + "errors" + "fmt" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + orgrepo "github.com/speakeasy-api/gram/server/internal/organizations/repo" + projectsrepo "github.com/speakeasy-api/gram/server/internal/projects/repo" + usersrepo "github.com/speakeasy-api/gram/server/internal/users/repo" +) + +// DatabaseEnricher resolves ids against Gram's own tables, behind a per-id TTL +// cache. +// +// The events this serves arrive in bursts — one organization's audit log is +// mostly one organization — so caching per id rather than loading a set keeps +// the work proportional to who is actually active. A row that no longer exists +// is cached as readily as one that does, because a deleted project is a +// permanent answer for the length of a TTL rather than a reason to retry on +// every event. +type DatabaseEnricher struct { + organizations *lookupCache[string, OrganizationDetails] + projects *lookupCache[uuid.UUID, ProjectDetails] + userEmails *lookupCache[string, string] +} + +var _ Enricher = (*DatabaseEnricher)(nil) + +func NewDatabaseEnricher(db *pgxpool.Pool) *DatabaseEnricher { + return &DatabaseEnricher{ + organizations: newLookupCache(func(ctx context.Context, organizationID string) (OrganizationDetails, error) { + organization, err := orgrepo.New(db).GetOrganizationMetadata(ctx, organizationID) + switch { + case errors.Is(err, pgx.ErrNoRows): + return OrganizationDetails{Slug: "", Name: ""}, nil + case err != nil: + return OrganizationDetails{Slug: "", Name: ""}, fmt.Errorf("get organization metadata: %w", err) + } + + return OrganizationDetails{Slug: organization.Slug, Name: organization.Name}, nil + }), + projects: newLookupCache(func(ctx context.Context, projectID uuid.UUID) (ProjectDetails, error) { + project, err := projectsrepo.New(db).GetProjectByID(ctx, projectID) + switch { + case errors.Is(err, pgx.ErrNoRows): + return ProjectDetails{Slug: "", Name: ""}, nil + case err != nil: + return ProjectDetails{Slug: "", Name: ""}, fmt.Errorf("get project by id: %w", err) + } + + return ProjectDetails{Slug: project.Slug, Name: project.Name}, nil + }), + userEmails: newLookupCache(func(ctx context.Context, userID string) (string, error) { + user, err := usersrepo.New(db).GetUser(ctx, userID) + switch { + case errors.Is(err, pgx.ErrNoRows): + return "", nil + case err != nil: + return "", fmt.Errorf("get user: %w", err) + } + + return user.Email, nil + }), + } +} + +func (e *DatabaseEnricher) Organization(ctx context.Context, organizationID string) (OrganizationDetails, error) { + return e.organizations.resolve(ctx, organizationID) +} + +func (e *DatabaseEnricher) Project(ctx context.Context, projectID uuid.UUID) (ProjectDetails, error) { + return e.projects.resolve(ctx, projectID) +} + +func (e *DatabaseEnricher) UserEmail(ctx context.Context, userID string) (string, error) { + return e.userEmails.resolve(ctx, userID) +} diff --git a/server/internal/growthsignals/event.go b/server/internal/growthsignals/event.go new file mode 100644 index 00000000000..1f09025a539 --- /dev/null +++ b/server/internal/growthsignals/event.go @@ -0,0 +1,248 @@ +package growthsignals + +import ( + "net/url" + + "github.com/google/uuid" + "github.com/speakeasy-api/gram/server/internal/audit" + "github.com/speakeasy-api/gram/server/internal/conv" + "github.com/speakeasy-api/gram/server/internal/urn" +) + +// Property keys every event may carry. They are the stable contract PostHog +// destinations filter and template against, so they are written once here +// rather than at each emission. +const ( + propertyActivity = "activity" + propertyOrganizationID = "organization_id" + propertyOrganizationSlug = "organization_slug" + propertyOrganizationName = "organization_name" + propertyProjectID = "project_id" + propertyProjectSlug = "project_slug" + propertyProjectName = "project_name" + propertyActorEmail = "actor_email" + propertyActorName = "actor_name" + propertySubjectName = "subject_name" + propertyActingSurface = "acting_surface" + propertyDashboardURL = "dashboard_url" + propertyAuditAction = "audit_action" +) + +// PropertyMcpKind is the flavour of MCP server an ActivityMcpServerCreated +// event describes. +const PropertyMcpKind = "mcp_kind" + +// PropertySignupSource distinguishes an invited arrival from an organic one on +// an ActivityUserSignedUp event. +const PropertySignupSource = "signup_source" + +// PropertyRole is the organization role granted on an +// ActivityMemberJoinedOrganization event. +const PropertyRole = "role" + +// PropertyPolicyName is the name of the risk policy a security policy event +// describes. +const PropertyPolicyName = "policy_name" + +const ( + // SignupSourceInvited marks a signup that had a pending invitation waiting + // for its email address. + SignupSourceInvited = "invited" + + // SignupSourceOrganic marks a signup that arrived without an invitation. + SignupSourceOrganic = "organic" +) + +// ActivityEvent is one notable moment, described in the terms its source +// already has. Ids here are resolved to names by an Enricher before the event +// is built, so producers never have to query for display values themselves. +// +// Every field beyond Activity and OrganizationID is optional: a producer sets +// what it knows and the resulting event carries what was set. +type ActivityEvent struct { + // Activity is what happened. An empty activity or ActivitySkip is not + // emitted. + Activity Activity + + // OrganizationID is the Gram organization the activity belongs to. It is + // the distinct id when no actor email resolves. + OrganizationID string + + // ProjectID is the project the activity belongs to, or uuid.Nil for + // organization-scoped activities. + ProjectID uuid.UUID + + // ActorID identifies the acting principal, and is interpreted according to + // ActorType: a Gram user id for a user, an email address for an email + // principal, a role name for a role. + ActorID string + + // ActorType is the kind of principal that acted. Only a user principal is + // worth a user lookup, and only user and email principals have a person + // behind them. + ActorType urn.PrincipalType + + // ActorEmail is the acting user's email when the producer already knows it. + // Setting it skips the lookup that would otherwise resolve ActorID. + ActorEmail string + + // ActorName is the acting principal's display name. + ActorName string + + // SubjectName is the display name of the thing acted on — the project that + // was created, the MCP server that was updated. + SubjectName string + + // ActingSurface is how the change was made: a dashboard session, an API + // key, Platform MCP, an assistant. + ActingSurface string + + // AuditAction is the audit action the activity was derived from, and is + // empty for activities that are emitted directly rather than from the + // audit log. It is reported so a surprising activity can be traced back to + // the record that produced it. + AuditAction audit.Action + + // DashboardURL deep links to the subject in the Gram dashboard. Leaving it + // empty is allowed: BuildEvent falls back to the organization's page and + // then to the site root, because the property must never be absent. + DashboardURL string + + // Extra holds per-activity properties such as mcp_kind, signup_source, + // role and policy_name. Blank values and the base property keys are + // ignored, so an extra can never rewrite the event's identity. + Extra map[string]string +} + +// Enrichment is what an ActivityEvent's ids resolved to. Any part of it may be +// zero: a lookup that failed or found nothing narrows the event rather than +// dropping it. +type Enrichment struct { + // Organization is what OrganizationID resolved to. + Organization OrganizationDetails + + // Project is what ProjectID resolved to. + Project ProjectDetails + + // ActorEmail is the acting user's email address. + ActorEmail string +} + +// CapturedEvent is the PostHog capture one activity produces. +type CapturedEvent struct { + // Name is always EventName. + Name string + + // DistinctID is the PostHog person the event attaches to. + DistinctID string + + // Properties is the event's property map. + Properties map[string]any +} + +// BuildEvent assembles the PostHog capture for one activity. +// +// The distinct id is the actor's email whenever one resolved, so an activity +// lands on the same PostHog person as that user's signup and onboarding events. +// It falls back to the organization id, because role and system actors have no +// person behind them and an organization is still a useful thing to group by. +// +// reservedProperties are the keys the event shape owns. Per-activity extras may +// not write them, so an activity can never rename an organization or claim a +// project it does not belong to. +var reservedProperties = map[string]struct{}{ + propertyActivity: {}, + propertyOrganizationID: {}, + propertyOrganizationSlug: {}, + propertyOrganizationName: {}, + propertyProjectID: {}, + propertyProjectSlug: {}, + propertyProjectName: {}, + propertyActorEmail: {}, + propertyActorName: {}, + propertySubjectName: {}, + propertyActingSurface: {}, + propertyDashboardURL: {}, + propertyAuditAction: {}, +} + +// Empty properties are omitted rather than sent blank. A blank organization +// name in Slack reads as an organization with no name; an absent one reads as +// what it is. +func BuildEvent(event ActivityEvent, enrichment Enrichment, siteURL *url.URL) CapturedEvent { + properties := make(map[string]any, len(event.Extra)+15) + + // Extras go in first, but never on a key the event shape owns. Writing them + // first is not enough on its own: a base property whose value is empty is + // omitted rather than written, and the project keys are skipped entirely on + // an organization-scoped activity, so without this filter an extra could + // occupy a reserved key and change what the event appears to say. + for key, value := range event.Extra { + if key == "" || value == "" { + continue + } + if _, reserved := reservedProperties[key]; reserved { + continue + } + properties[key] = value + } + + properties[propertyActivity] = string(event.Activity) + + setProperty(properties, propertyOrganizationID, event.OrganizationID) + setProperty(properties, propertyOrganizationSlug, enrichment.Organization.Slug) + setProperty(properties, propertyOrganizationName, enrichment.Organization.Name) + + if event.ProjectID != uuid.Nil { + properties[propertyProjectID] = event.ProjectID.String() + setProperty(properties, propertyProjectSlug, enrichment.Project.Slug) + setProperty(properties, propertyProjectName, enrichment.Project.Name) + } + + setProperty(properties, propertyActorEmail, enrichment.ActorEmail) + setProperty(properties, propertyActorName, event.ActorName) + setProperty(properties, propertySubjectName, event.SubjectName) + setProperty(properties, propertyActingSurface, event.ActingSurface) + setProperty(properties, propertyDashboardURL, dashboardURL(event, enrichment, siteURL)) + setProperty(properties, propertyAuditAction, string(event.AuditAction)) + + return CapturedEvent{ + Name: EventName, + DistinctID: conv.Default(enrichment.ActorEmail, event.OrganizationID), + Properties: properties, + } +} + +func setProperty(properties map[string]any, key string, value string) { + if value == "" { + return + } + + properties[key] = value +} + +// dashboardURL is the one property that is never omitted. Slack rejects a +// button whose url is empty and fails the whole message, so a destination +// template that links the subject would break every notification that happens +// to lack a link. An activity with no subject page falls back to the +// organization's page, and one whose organization did not resolve falls back to +// the site root, which is always a valid URL. +func dashboardURL(event ActivityEvent, enrichment Enrichment, siteURL *url.URL) string { + if event.DashboardURL != "" { + return event.DashboardURL + } + + // No site URL configured. Reporting an empty string would be worse than + // reporting nothing: a Slack destination that renders this as a button link + // fails the whole message on a blank url, while an absent property lets the + // template omit the button. The emitter warns about this at construction. + if siteURL == nil { + return "" + } + + if enrichment.Organization.Slug != "" { + return siteURL.JoinPath(enrichment.Organization.Slug).String() + } + + return siteURL.String() +} diff --git a/server/internal/growthsignals/event_test.go b/server/internal/growthsignals/event_test.go new file mode 100644 index 00000000000..900fbbeb3e3 --- /dev/null +++ b/server/internal/growthsignals/event_test.go @@ -0,0 +1,259 @@ +package growthsignals_test + +import ( + "net/url" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/speakeasy-api/gram/server/internal/audit" + "github.com/speakeasy-api/gram/server/internal/growthsignals" + "github.com/speakeasy-api/gram/server/internal/urn" +) + +func TestBuildEventCarriesEveryProperty(t *testing.T) { + t.Parallel() + + projectID := uuid.MustParse("11111111-2222-3333-4444-555555555555") + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityMcpServerCreated, + OrganizationID: "org_placeholder", + ProjectID: projectID, + ActorID: "user_placeholder", + ActorType: urn.PrincipalTypeUser, + ActorEmail: "", + ActorName: "Acting Person", + SubjectName: "Widget Server", + ActingSurface: string(audit.SurfaceDashboard), + AuditAction: audit.ActionMcpServerCreate, + DashboardURL: "https://app.example.test/acme/widgets/mcp/widget-server", + Extra: map[string]string{growthsignals.PropertyMcpKind: string(growthsignals.McpKindHosted)}, + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "acme", Name: "Acme Incorporated"}, + Project: growthsignals.ProjectDetails{Slug: "widgets", Name: "Widgets"}, + ActorEmail: "person@example.test", + }, testSiteURL()) + + require.Equal(t, growthsignals.EventName, built.Name) + require.Equal(t, "person@example.test", built.DistinctID) + require.Equal(t, map[string]any{ + "activity": "mcp_server_created", + "organization_id": "org_placeholder", + "organization_slug": "acme", + "organization_name": "Acme Incorporated", + "project_id": projectID.String(), + "project_slug": "widgets", + "project_name": "Widgets", + "actor_email": "person@example.test", + "actor_name": "Acting Person", + "subject_name": "Widget Server", + "acting_surface": "dashboard", + "dashboard_url": "https://app.example.test/acme/widgets/mcp/widget-server", + "audit_action": "mcp-server:create", + "mcp_kind": "hosted", + }, built.Properties) +} + +// A blank property in Slack reads as a thing with no name; an absent one reads +// as what it is. +func TestBuildEventOmitsUnresolvedProperties(t *testing.T) { + t.Parallel() + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + ProjectID: uuid.Nil, + ActorID: "", + ActorType: "", + ActorEmail: "", + ActorName: "", + SubjectName: "", + ActingSurface: "", + AuditAction: "", + DashboardURL: "", + Extra: map[string]string{growthsignals.PropertyRole: ""}, + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "", Name: ""}, + Project: growthsignals.ProjectDetails{Slug: "", Name: ""}, + ActorEmail: "", + }, testSiteURL()) + + require.Equal(t, map[string]any{ + "activity": "project_created", + "organization_id": "org_placeholder", + // dashboard_url is never omitted: with no subject page and no resolved + // organization slug it falls back to the site root. + "dashboard_url": "https://app.example.test", + }, built.Properties) +} + +// Project slug and name describe a project, so they must not appear on an +// organization-scoped activity even when a stale enrichment carries them. +func TestBuildEventOmitsProjectPropertiesWhenNotProjectScoped(t *testing.T) { + t.Parallel() + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityMemberInvited, + OrganizationID: "org_placeholder", + ProjectID: uuid.Nil, + ActorID: "", + ActorType: "", + ActorEmail: "", + ActorName: "", + SubjectName: "", + ActingSurface: "", + AuditAction: "", + DashboardURL: "", + Extra: nil, + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "acme", Name: "Acme Incorporated"}, + Project: growthsignals.ProjectDetails{Slug: "widgets", Name: "Widgets"}, + ActorEmail: "", + }, testSiteURL()) + + require.NotContains(t, built.Properties, "project_id") + require.NotContains(t, built.Properties, "project_slug") + require.NotContains(t, built.Properties, "project_name") +} + +// Role and system actors have no person behind them, so the organization is the +// only sensible thing left to attach the event to. +func TestBuildEventDistinctIDFallsBackToOrganization(t *testing.T) { + t.Parallel() + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivitySecurityPolicyUpdated, + OrganizationID: "org_placeholder", + ProjectID: uuid.Nil, + ActorID: "admin", + ActorType: urn.PrincipalTypeRole, + ActorEmail: "", + ActorName: "", + SubjectName: "", + ActingSurface: "", + AuditAction: "", + DashboardURL: "", + Extra: nil, + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "", Name: ""}, + Project: growthsignals.ProjectDetails{Slug: "", Name: ""}, + ActorEmail: "", + }, testSiteURL()) + + require.Equal(t, "org_placeholder", built.DistinctID) + require.NotContains(t, built.Properties, "actor_email") +} + +// Extras are supplied per activity, so a stray key must not be able to rewrite +// the identity of the event carrying it. +func TestBuildEventBasePropertiesWinOverExtras(t *testing.T) { + t.Parallel() + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityUserSignedUp, + OrganizationID: "org_placeholder", + ProjectID: uuid.Nil, + ActorID: "", + ActorType: "", + ActorEmail: "", + ActorName: "", + SubjectName: "", + ActingSurface: "", + AuditAction: "", + DashboardURL: "", + Extra: map[string]string{ + "activity": "something_else", + "organization_id": "org_other", + growthsignals.PropertySignupSource: growthsignals.SignupSourceInvited, + "": "blank key", + }, + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "", Name: ""}, + Project: growthsignals.ProjectDetails{Slug: "", Name: ""}, + ActorEmail: "", + }, testSiteURL()) + + require.Equal(t, map[string]any{ + "activity": "user_signed_up", + "organization_id": "org_placeholder", + "signup_source": "invited", + "dashboard_url": "https://app.example.test", + }, built.Properties) +} + +// testSiteURL is the dashboard base URL the property builder falls back to when +// an activity carries no subject page of its own. +func testSiteURL() *url.URL { + return &url.URL{Scheme: "https", Host: "app.example.test"} +} + +// dashboard_url is the one property that is never omitted. A Slack destination +// that renders it as a button link fails the entire message when the url is +// empty, so an activity with no subject page of its own must still resolve to +// something valid. +func TestBuildEventAlwaysSetsDashboardURL(t *testing.T) { + t.Parallel() + + subject := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + DashboardURL: "https://app.example.test/acme/widgets", + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "acme"}, + }, testSiteURL()) + require.Equal(t, "https://app.example.test/acme/widgets", subject.Properties["dashboard_url"]) + + organization := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + }, growthsignals.Enrichment{ + Organization: growthsignals.OrganizationDetails{Slug: "acme"}, + }, testSiteURL()) + require.Equal(t, "https://app.example.test/acme", organization.Properties["dashboard_url"]) + + root := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + }, growthsignals.Enrichment{}, testSiteURL()) + require.Equal(t, "https://app.example.test", root.Properties["dashboard_url"]) +} + +// An extra may never occupy a key the event shape owns. Writing extras first is +// not enough on its own, because a base property whose value is empty is +// omitted rather than written, and the project keys are skipped entirely on an +// organization-scoped activity. +func TestBuildEventExtrasCannotOccupyReservedKeys(t *testing.T) { + t.Parallel() + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityUserSignedUp, + OrganizationID: "org_placeholder", + Extra: map[string]string{ + "organization_slug": "attacker-owned", + "project_slug": "claimed", + "activity": "something_else", + growthsignals.PropertySignupSource: growthsignals.SignupSourceOrganic, + }, + }, growthsignals.Enrichment{}, testSiteURL()) + + require.Equal(t, "user_signed_up", built.Properties["activity"]) + require.NotContains(t, built.Properties, "organization_slug") + require.NotContains(t, built.Properties, "project_slug") + require.Equal(t, growthsignals.SignupSourceOrganic, built.Properties[growthsignals.PropertySignupSource]) +} + +// With no site URL configured there is no link to report. Reporting an empty +// string would be worse than reporting nothing, because a Slack destination +// that renders it as a button link fails the whole message on a blank url. +func TestBuildEventOmitsDashboardURLWithoutSiteURL(t *testing.T) { + t.Parallel() + + built := growthsignals.BuildEvent(growthsignals.ActivityEvent{ + Activity: growthsignals.ActivityProjectCreated, + OrganizationID: "org_placeholder", + }, growthsignals.Enrichment{}, nil) + + require.NotContains(t, built.Properties, "dashboard_url") +} diff --git a/server/internal/growthsignals/interfaces.go b/server/internal/growthsignals/interfaces.go new file mode 100644 index 00000000000..0c5ea173660 --- /dev/null +++ b/server/internal/growthsignals/interfaces.go @@ -0,0 +1,58 @@ +package growthsignals + +import ( + "context" + + "github.com/google/uuid" +) + +// PostHogClient is the slice of the PostHog client this package uses. Declaring +// it here rather than depending on the concrete client keeps the vendor types +// out of the emitter and lets callers and tests supply a capture that records +// instead of ships. +type PostHogClient interface { + // CaptureEvent records one event against a distinct id. Implementations + // enqueue rather than block, and a disabled client reports success. + CaptureEvent(ctx context.Context, eventName string, distinctID string, eventProperties map[string]any) error +} + +// OrganizationDetails is what an organization id resolves to. +type OrganizationDetails struct { + // Slug is the organization's URL slug, and is empty when the organization + // could not be resolved. + Slug string + + // Name is the organization's display name, and is empty when the + // organization could not be resolved. + Name string +} + +// ProjectDetails is what a project id resolves to. +type ProjectDetails struct { + // Slug is the project's URL slug, and is empty when the project could not + // be resolved. + Slug string + + // Name is the project's display name, and is empty when the project could + // not be resolved. + Name string +} + +// Enricher resolves the ids an activity carries into the values an event +// reports. Audit payloads carry ids and little else, and Slack readers need +// names. +// +// A caller that cannot find a row returns zero details and no error: an +// organization that no longer exists is a fact about the event, not a failure +// to resolve it. An error means the lookup itself failed, and the emitter +// degrades the event rather than dropping it. +type Enricher interface { + // Organization resolves an organization id to its slug and name. + Organization(ctx context.Context, organizationID string) (OrganizationDetails, error) + + // Project resolves a project id to its slug and name. + Project(ctx context.Context, projectID uuid.UUID) (ProjectDetails, error) + + // UserEmail resolves a Gram user id to that user's email address. + UserEmail(ctx context.Context, userID string) (string, error) +} diff --git a/server/internal/growthsignals/setup_test.go b/server/internal/growthsignals/setup_test.go new file mode 100644 index 00000000000..40aeaf63335 --- /dev/null +++ b/server/internal/growthsignals/setup_test.go @@ -0,0 +1,98 @@ +package growthsignals_test + +import ( + "context" + "slices" + "sync" + + "github.com/google/uuid" + + "github.com/speakeasy-api/gram/server/internal/growthsignals" +) + +// capturedEvent is one call the fake PostHog client recorded. +type capturedEvent struct { + Name string + DistinctID string + Properties map[string]any +} + +// capturePostHog records what would have been sent to PostHog. PostHogClient is +// our own interface rather than a vendor type, so a capture client is the right +// double here: it lets a test assert on the exact payload. +type capturePostHog struct { + mu sync.Mutex + events []capturedEvent + failWith error +} + +func (c *capturePostHog) CaptureEvent(_ context.Context, eventName string, distinctID string, eventProperties map[string]any) error { + c.mu.Lock() + defer c.mu.Unlock() + + c.events = append(c.events, capturedEvent{ + Name: eventName, + DistinctID: distinctID, + Properties: eventProperties, + }) + + return c.failWith +} + +func (c *capturePostHog) Captured() []capturedEvent { + c.mu.Lock() + defer c.mu.Unlock() + + return slices.Clone(c.events) +} + +// fakeEnricher answers lookups from fixed values, and can fail any of the three +// independently so a test can prove one failed lookup does not cost the others. +type fakeEnricher struct { + organization growthsignals.OrganizationDetails + organizationErr error + + project growthsignals.ProjectDetails + projectErr error + + userEmails map[string]string + userEmailErr error + + mu sync.Mutex + userEmailCalls []string +} + +func (f *fakeEnricher) Organization(_ context.Context, _ string) (growthsignals.OrganizationDetails, error) { + if f.organizationErr != nil { + return growthsignals.OrganizationDetails{}, f.organizationErr + } + + return f.organization, nil +} + +func (f *fakeEnricher) Project(_ context.Context, _ uuid.UUID) (growthsignals.ProjectDetails, error) { + if f.projectErr != nil { + return growthsignals.ProjectDetails{}, f.projectErr + } + + return f.project, nil +} + +func (f *fakeEnricher) UserEmail(_ context.Context, userID string) (string, error) { + f.mu.Lock() + f.userEmailCalls = append(f.userEmailCalls, userID) + f.mu.Unlock() + + if f.userEmailErr != nil { + return "", f.userEmailErr + } + + return f.userEmails[userID], nil +} + +func (f *fakeEnricher) UserEmailCalls() []string { + f.mu.Lock() + defer f.mu.Unlock() + + return slices.Clone(f.userEmailCalls) +}