Add FluentBit pipeline for Keycloak account events.

fluent-bit/filesystem/etc/fluent-bit/fluent-bit.yaml

@@ -27,6 +27,13 @@ pipeline:
       match: telemetry_stalwart
       script: /etc/fluent-bit/lua/stalwart_telemetry_events.lua
       call: stalwart_telemetry_callback
+
+    # Run a Lua script that processes Keycloak account events (logins, token introspections, etc.)
+    # polled by the Accounts Celery worker and POSTed to `/telemetry/keycloak`.
+    - name: lua
+      match: telemetry_keycloak
+      script: /etc/fluent-bit/lua/keycloak_account_events.lua
+      call: keycloak_account_callback
 
   outputs:
     # Throw away health checks
@@ -45,3 +52,12 @@ pipeline:
       tls: on
       uri: /batch
       format: json_lines
+
+    # Send Keycloak account events to Posthog
+    - name: http
+      match: telemetry_keycloak
+      host: us.i.posthog.com
+      port: 443
+      tls: on
+      uri: /batch
+      format: json_lines

fluent-bit/filesystem/etc/fluent-bit/lua/keycloak_account_events.lua

@@ -0,0 +1,76 @@
+--[[
+    keycloak_account_events.lua
+
+This module acts as a fluent-bit event processor for raw Keycloak user events. The Accounts
+application's Celery worker periodically polls the Keycloak Admin API for recent events
+(logins, token introspections, registrations, etc.) and POSTs the raw event JSON to
+fluent-bit's HTTP input at /telemetry/keycloak.
+
+This filter handles all transformation:
+
+  - Event type mapping: Keycloak types are collapsed into PostHog event names
+    (e.g. INTROSPECT_TOKEN, REFRESH_TOKEN, CODE_TO_TOKEN → accounts.activity)
+  - Identity hashing: userId is SHA-256 hashed for use as PostHog distinct_id
+  - Field stripping: Only analytically useful fields are kept (clientId, environment, etc.)
+  - PII removal: ipAddress, details map, sessionId, and other sensitive fields are dropped
+  - PostHog formatting: Events are reformatted into the PostHog batch API payload
+
+This script needs:
+
+  - POSTHOG_API_KEY (required)
+  - ENV (optional, defaults to "dev")
+
+]]--
+
+local sha256 = require('sha2').sha256
+
+-- Keycloak event types mapped to collapsed PostHog event names
+local EVENT_MAP = {
+    LOGIN = 'accounts.login',
+    LOGIN_ERROR = 'accounts.login_error',
+    REGISTER = 'accounts.register',
+    REGISTER_ERROR = 'accounts.register_error',
+    LOGOUT = 'accounts.logout',
+    CODE_TO_TOKEN = 'accounts.activity',
+    CODE_TO_TOKEN_ERROR = 'accounts.activity',
+    INTROSPECT_TOKEN = 'accounts.activity',
+    REFRESH_TOKEN = 'accounts.activity',
+}
+
+function keycloak_account_callback(tag, timestamp, record)
+    local os_env = os.getenv('ENV') or 'dev'
+    local batch = {}
+
+    for idx, event in ipairs(record.events) do
+        local kc_type = event.type or 'UNKNOWN'
+        local user_id = event.userId or ''
+        local is_error = string.sub(kc_type, -6) == '_ERROR'
+
+        local distinct_id = 'n/a'
+        if user_id ~= '' then
+            distinct_id = sha256(user_id)
+        end
+
+        batch[idx] = {
+            event = EVENT_MAP[kc_type] or 'accounts.activity',
+            properties = {
+                distinct_id = distinct_id,
+                clientId = event.clientId or '',
+                environment = os_env,
+                service = 'accounts',
+                keycloak_event_type = kc_type,
+                keycloak_event_id = event.id or '',
+                is_error = is_error,
+                ['$ip'] = nil,
+            },
+        }
+    end
+
+    local payload = {
+        api_key = os.getenv('POSTHOG_API_KEY'),
+        historical_migration = false,
+        batch = batch,
+    }
+
+    return 2, timestamp, payload
+end

pulumi/config.prod.yaml

@@ -42,6 +42,9 @@ resources:
       additional_routes:
         - destination_cidr_block: 10.0.0.0/16  # mailstrom-prod
           vpc_peering_connection_id: pcx-0d581e66566a6e00b
+        # TODO: Add route for accounts-prod VPC peering once the peering connection is created
+        # - destination_cidr_block: 10.10.0.0/16  # accounts-prod
+        #   vpc_peering_connection_id: pcx-PLACEHOLDER
 
   tb:fargate:AutoscalingFargateCluster:
     fluentbit:
@@ -120,6 +123,12 @@ resources:
                 cidr_blocks:
                   - 10.0.0.0/16  # stalwart-prod
                 description: Allow access from stalwart-prod
+              - from_port: 443
+                to_port: 443
+                protocol: tcp
+                cidr_blocks:
+                  - 10.10.0.0/16  # accounts-prod
+                description: Allow access from accounts-prod
             egress:
               - from_port: 0
                 to_port: 65535

pulumi/config.stage.yaml

@@ -41,6 +41,9 @@ resources:
       additional_routes:
         - destination_cidr_block: 10.1.0.0/16  # mailstrom-stage
           vpc_peering_connection_id: pcx-03dd1daa0b700ab37
+        # TODO: Add route for accounts-stage VPC peering once the peering connection is created
+        # - destination_cidr_block: 10.11.0.0/16  # accounts-stage
+        #   vpc_peering_connection_id: pcx-PLACEHOLDER
 
   tb:fargate:AutoscalingFargateCluster:
     fluentbit:
@@ -120,6 +123,12 @@ resources:
                 cidr_blocks:
                   - 10.1.0.0/16  # stalwart-stage
                 description: Allow access from stalwart-stage
+              - from_port: 443
+                to_port: 443
+                protocol: tcp
+                cidr_blocks:
+                  - 10.11.0.0/16  # accounts-stage
+                description: Allow access from accounts-stage
             egress:
               - from_port: 0
                 to_port: 65535