feat: forward audited mutations to PostHog

.changeset/growth-signals-stream-handler.md

@@ -0,0 +1,5 @@
+---
+"server": patch
+---
+
+Forward audited mutations to PostHog as `gram_activity` events. The `gram streams` process now runs a growth-signals handler alongside its existing webhook consumers, so every audited mutation — projects and MCP servers created, security policies written, members invited — reaches PostHog without any service emitting analytics of its own. Uncurated actions pass through under a normalized name so coverage is automatic, and a small exclusion list keeps high-volume noise out.

server/cmd/gram/deps.go

@@ -6,6 +6,7 @@ import (
 	"encoding/csv"
 	"errors"
 	"fmt"
+	"github.com/speakeasy-api/gram/server/internal/growthsignals"
 	"io/fs"
 	"log/slog"
 	"net/url"
@@ -1407,3 +1408,15 @@ func newKMSSigningClients(ctx context.Context, logger *slog.Logger, c *cli.Conte
 		return client, nil
 	}, nil
 }
+
+// newGrowthSignalsEmitter builds the emitter that reports notable moments to
+// PostHog.
+//
+// It enriches against the read replica. The lookups are display names for an
+// ops notification, not authority for anything, and a miss degrades the event
+// by omitting a property rather than dropping it — so replication lag costs at
+// most a missing slug on a very fresh row, which is not worth the primary's
+// capacity.
+func newGrowthSignalsEmitter(logger *slog.Logger, posthogClient *posthog.Posthog, replicaDB *pgxpool.Pool, siteURL *url.URL) *growthsignals.Emitter {
+	return growthsignals.NewEmitter(logger, posthogClient, growthsignals.NewDatabaseEnricher(replicaDB), siteURL)
+}

server/cmd/gram/streams.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"net/url"
 	"os"
 	"os/signal"
 	"runtime/debug"
@@ -44,6 +45,7 @@ import (
 	"github.com/speakeasy-api/gram/server/internal/control"
 	"github.com/speakeasy-api/gram/server/internal/encryption"
 	"github.com/speakeasy-api/gram/server/internal/feature"
+	"github.com/speakeasy-api/gram/server/internal/growthsignals"
 	"github.com/speakeasy-api/gram/server/internal/metering"
 	meteringchrepo "github.com/speakeasy-api/gram/server/internal/metering/chrepo"
 	"github.com/speakeasy-api/gram/server/internal/modelkeys"
@@ -230,6 +232,11 @@ func newStreamsCommand() *cli.Command {
 			EnvVars:  []string{"GRAM_DISALLOWED_CIDR_BLOCKS"},
 			Required: false,
 		},
+		&cli.StringFlag{
+			Name:    "site-url",
+			Usage:   "The URL of the dashboard site, used to deep link from growth activity events",
+			EnvVars: []string{"GRAM_SITE_URL"},
+		},
 		&cli.PathFlag{
 			Name:     "config-file",
 			Usage:    "Path to a config file to load. Supported formats are JSON, TOML and YAML.",
@@ -492,6 +499,23 @@ func newStreamsCommand() *cli.Command {
 			paygKeyRefreshHandler := usage.NewPaygKeyRefreshHandler(logger, openRouterKeyRefresher)
 			trialConversionKeyReconcileHandler := usage.NewEnterpriseTrialConversionKeyReconcileHandler(logger, openRouterKeyRefresher)
 			billingNotificationHandler := billingnotifications.NewEventHandler(logger, &background.TemporalBillingEmailScheduler{TemporalEnv: temporalEnv})
+
+			var siteURL *url.URL
+			if raw := c.String("site-url"); raw != "" {
+				siteURL, err = url.Parse(raw)
+				if err != nil {
+					return fmt.Errorf("parse site url: %w", err)
+				}
+				// url.Parse accepts a bare path or a custom scheme, and either
+				// would produce a link Slack rejects. A site URL that cannot
+				// address the dashboard is a misconfiguration worth failing on
+				// rather than discovering one dead button at a time.
+				if (siteURL.Scheme != "http" && siteURL.Scheme != "https") || siteURL.Host == "" {
+					return fmt.Errorf("site url must be an absolute http(s) URL, got %q", raw)
+				}
+			}
+			growthSignalHandler := growthsignals.NewEventHandler(logger, newGrowthSignalsEmitter(logger, posthogClient, replicaDB, siteURL))
+
 			webhookEventHandler := streams.HandlerFunc[*webhooksv1.Event](func(ctx context.Context, event *webhooksv1.Event, metadata gcp.MessageMetadata) error {
 				var handlerErrors []error
 				if err := svixRelayHandler.Handle(ctx, event, metadata); err != nil {
@@ -506,6 +530,9 @@ func newStreamsCommand() *cli.Command {
 				if err := billingNotificationHandler.Handle(ctx, event, metadata); err != nil {
 					handlerErrors = append(handlerErrors, fmt.Errorf("schedule billing notification: %w", err))
 				}
+				if err := growthSignalHandler.Handle(ctx, event, metadata); err != nil {
+					handlerErrors = append(handlerErrors, fmt.Errorf("report growth activity: %w", err))
+				}
 				return errors.Join(handlerErrors...)
 			})
 

server/internal/growthsignals/handler.go

@@ -0,0 +1,143 @@
+package growthsignals
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"strings"
+
+	"github.com/google/uuid"
+
+	webhooksv1 "github.com/speakeasy-api/gram/infra/gen/gram/webhooks/v1"
+	"github.com/speakeasy-api/gram/infra/pkg/gcp"
+	"github.com/speakeasy-api/gram/server/internal/attr"
+	"github.com/speakeasy-api/gram/server/internal/audit"
+	"github.com/speakeasy-api/gram/server/internal/outbox/events"
+	"github.com/speakeasy-api/gram/server/internal/urn"
+)
+
+// auditEventTypePrefix selects the audit-derived events out of the webhook
+// topic. Every audited subject publishes under its own `audit_log.*_event_v1`
+// type, so matching the prefix covers all of them and keeps covering the ones
+// services add later, without a list here to extend.
+const auditEventTypePrefix = "audit_log."
+
+// propertyInsertID is PostHog's deduplication key. Two captures carrying the
+// same value describe one occurrence, which is what makes an at-least-once
+// topic safe to report from.
+const propertyInsertID = "$insert_id"
+
+// EventHandler turns audited mutations into growth activities.
+//
+// It reads the webhook stream the audit logger already publishes, so every
+// mutation Gram records reaches PostHog without any service having to emit
+// analytics of its own.
+type EventHandler struct {
+	logger  *slog.Logger
+	emitter *Emitter
+}
+
+func NewEventHandler(logger *slog.Logger, emitter *Emitter) *EventHandler {
+	return &EventHandler{
+		logger:  logger.With(attr.SlogComponent("growth-signal-events")),
+		emitter: emitter,
+	}
+}
+
+// Handle emits one activity for an audited mutation.
+//
+// It returns an error only for a failure worth redelivering. Every other
+// outcome — an envelope that makes no sense, a payload that will not decode, an
+// action with no ops value — acks the message, because a message this handler
+// can never make sense of would otherwise redeliver until the subscription's
+// retention runs out. Emission itself cannot fail the message: an activity that
+// could not be captured is logged by the emitter and dropped, since analytics
+// must never hold up the stream.
+func (h *EventHandler) Handle(ctx context.Context, event *webhooksv1.Event, _ gcp.MessageMetadata) error {
+	if event == nil {
+		h.logger.ErrorContext(ctx, "dropping nil growth signal event")
+		return nil
+	}
+
+	eventID := event.GetEventId()
+	organizationID := event.GetOrganizationId()
+	if eventID == "" || organizationID == "" {
+		h.logger.ErrorContext(ctx, "dropping invalid growth signal event",
+			attr.SlogOrganizationID(organizationID),
+			attr.SlogOutboxPublicID(eventID),
+			attr.SlogEvent(event.GetEventType()),
+		)
+		return nil
+	}
+
+	// Every webhook event on the topic reaches this handler, so anything that
+	// is not an audit record must leave before the payload is decoded.
+	if !strings.HasPrefix(event.GetEventType(), auditEventTypePrefix) {
+		return nil
+	}
+
+	var payload events.AuditLogCreatedPayloadV1
+	if err := json.Unmarshal(event.GetPayload(), &payload); err != nil {
+		h.logger.ErrorContext(ctx, "dropping unreadable growth signal event",
+			attr.SlogOutboxPublicID(eventID),
+			attr.SlogEvent(event.GetEventType()),
+			attr.SlogError(err),
+		)
+		return nil
+	}
+
+	// The event type is a coarse bucket — an MCP server creation and a metadata
+	// edit share one — so the action is what says which activity this is.
+	action := audit.Action(payload.Action)
+	mapping := ActivityForAction(action)
+	if mapping.Activity == ActivitySkip {
+		h.logger.DebugContext(ctx, "growth signal excluded by action",
+			attr.SlogOutboxPublicID(eventID),
+			attr.SlogEvent(event.GetEventType()),
+		)
+		return nil
+	}
+
+	projectID := uuid.Nil
+	if payload.ProjectID.Valid {
+		projectID = payload.ProjectID.UUID
+	}
+
+	h.emitter.Emit(ctx, ActivityEvent{
+		Activity:       mapping.Activity,
+		OrganizationID: organizationID,
+		ProjectID:      projectID,
+		ActorID:        payload.ActorID,
+		ActorType:      urn.PrincipalType(payload.ActorType),
+		// The audit record carries the actor's id, not their address, so the
+		// emitter resolves the email itself behind its cache.
+		ActorEmail:    "",
+		ActorName:     payload.ActorDisplayName,
+		SubjectName:   payload.SubjectDisplayName,
+		ActingSurface: payload.ActingSurface,
+		AuditAction:   action,
+		// Audit records address subjects by id and type rather than by page, so
+		// the event takes the emitter's organization-level fallback link.
+		DashboardURL: "",
+		Extra:        withInsertID(mapping.Extra, eventID),
+	})
+
+	return nil
+}
+
+// withInsertID stamps the outbox event id as PostHog's deduplication key.
+//
+// The topic is at-least-once and this handler shares a message with its
+// siblings: when any of them fails, the whole message nacks and every handler
+// sees the event again. Without a stable key that is a duplicate Slack line for
+// something that happened once. The outbox id is stable across redeliveries,
+// which is exactly what the key needs to be.
+func withInsertID(extra map[string]string, eventID string) map[string]string {
+	stamped := make(map[string]string, len(extra)+1)
+	for key, value := range extra {
+		stamped[key] = value
+	}
+	stamped[propertyInsertID] = eventID
+
+	return stamped
+}