feat(openframe): stamp team_id on CDC-captured tables for shared-DB Debezium

openframe/docs/fork-file-manifest.md

@@ -212,6 +212,7 @@ openframe/scripts/test_host_assignments.sh
 openframe/scripts/verify.sh
 server/datastore/mysql/migrations/openframe/20260301000001_AddPolicyHostsJoinTable.go
 server/datastore/mysql/migrations/openframe/20260301000002_AddQueryHostsJoinTable.go
+server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go
 server/datastore/mysql/migrations/openframe/migration.go
 server/datastore/mysql/migrations_openframe_test.go
 server/datastore/redis/keyprefix.go
@@ -247,6 +248,7 @@ go.mod
 go.sum
 orbit/cmd/orbit/orbit.go
 orbit/pkg/osquery/osquery.go
+server/activity/internal/mysql/new_activity.go
 server/archtest/README.md
 server/archtest/test_files/dependency/dependency.go
 server/config/config.go

openframe/docs/mysql-multitenancy-feature.md

@@ -48,6 +48,7 @@ Applied by `fleet prepare db`. Idempotent (`columnExists`/`indexExists` guards).
 | `20260629000001_AddTeamsOpenframeTenantUUID` | `teams.openframe_tenant_uuid CHAR(36)` + unique key | the UUID→team bridge |
 | `20260626000001_ScopeHostIdentityUniqueToTeam` | `hosts` virtual col `openframe_team_key = IFNULL(team_id,0)` + `UNIQUE(osquery_host_id, openframe_team_key)`, drop global `UNIQUE(osquery_host_id)` | host identity unique **per team** — the same device can exist in two tenants |
 | `20260620000001_ScopeLabelUniqueNameToTeam` | same generated-column pattern for `labels.name` | label names unique per team; built-ins stay global |
+| `20260722000001_AddTeamIdToCdcTables` | nullable `team_id` on `activity_past`, `activity_host_past`, `query_results`, `policy_membership` (no index, no FK) | the Debezium CDC tables must be self-describing on a shared DB — see "CDC team stamping" below |
 
 The `IFNULL(team_id,0)` sentinel collapses all NULL-team rows onto key `0` (team ids start at 1), so
 **flag-off / pre-backfill the uniqueness is bit-for-bit the old global uniqueness**, and the
@@ -102,6 +103,30 @@ clusters' migration Jobs pointed at one DB, the first wins and migrates; the res
 the schema already applied and no-op. Session-scoped (auto-released if the Job dies). Flag-off runs
 are untouched.
 
+## CDC team stamping (Debezium on the shared DB)
+
+The OpenFrame Debezium pipeline captures `activity_past`, `activity_host_past`, `query_results`
+and `policy_membership`. On a shared DB one connector serves every tenant, and its SMTs are
+stateless — a CDC record must therefore carry its own tenant discriminator. The fork stamps the
+`team_id` column (added by `20260722000001`) at write time:
+
+| Table | Stamp source | Where |
+|---|---|---|
+| `query_results` | request team pin (`fleet.OpenframeTeamID(ctx)` — the agent plane is always pinned in shared mode) | `OverwriteQueryResultRows` |
+| `policy_membership` (sync) | request team pin | `RecordPolicyQueryExecutions` |
+| `policy_membership` (async collector) | the row's own host, `(SELECT team_id FROM hosts WHERE id = ?)` — the cron context is never pinned | `AsyncBatchInsertPolicyMembership` |
+| `activity_past` | request team pin — the only tenant signal for host-less (user/team-level) activities | `server/activity/internal/mysql/new_activity.go` |
+| `activity_host_past` | the row's own host via subselect (host activities can come from unpinned crons; the host is authoritative anyway) | same |
+
+Unpinned writes (flag off, or background crons writing host-less activities) keep the original
+statement **byte-identical** and leave `team_id` NULL; the downstream consumer drops NULL-team
+events (fail closed). The `teams` table (with the `openframe_tenant_uuid` bridge) is also added
+to the connector's capture list platform-side, so consumers can resolve `team_id` → tenant UUID
+without touching Fleet's API. Platform side of this pipeline: shared connector registration in
+`openframe-saas-shared`, and shared-plane consumption in that repo's `openframe-saas-stream`
+(the MeshCentral pattern — per-event tenant resolution, gated by
+`openframe.fleet.multi-tenancy.enabled`).
+
 ## Helm / config wiring (`charts/fleet/`)
 
 `values.yaml` adds `fleet.openframe.multiTenancy` (`enabled: false` default; `tenantUuid` /

server/activity/internal/mysql/new_activity.go

@@ -9,6 +9,9 @@ import (
 	"github.com/fleetdm/fleet/v4/server/activity/api"
 	"github.com/fleetdm/fleet/v4/server/activity/internal/types"
 	"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
+	// >>> OPENFRAME(mysql-multitenancy): team pin helpers for CDC team_id stamping
+	"github.com/fleetdm/fleet/v4/server/fleet"
+	// <<< OPENFRAME(mysql-multitenancy)
 	platform_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql"
 	"github.com/jmoiron/sqlx"
 )
@@ -72,6 +75,18 @@ func (ds *Datastore) NewActivity(
 		cols = append(cols, "user_email")
 	}
 
+	// >>> OPENFRAME(mysql-multitenancy): stamp the tenant team onto CDC-captured rows so the
+	// shared-DB Debezium pipeline can resolve the tenant per event — openframe/docs/mysql-multitenancy-feature.md.
+	// activity_past has no host reference, so the request's team pin is the only tenant signal
+	// (user/team-level activities). Unpinned contexts (flag off, or background crons) keep the
+	// original statement byte-identical and leave team_id NULL.
+	teamID, teamPinned := fleet.OpenframeTeamID(ctx)
+	if teamPinned {
+		cols = append(cols, "team_id")
+		args = append(args, teamID)
+	}
+	// <<< OPENFRAME(mysql-multitenancy)
+
 	return platform_mysql.WithRetryTxx(ctx, ds.primary, func(tx sqlx.ExtContext) error {
 		const insertActStmt = `INSERT INTO activity_past (%s) VALUES (%s)`
 		sqlStmt := fmt.Sprintf(insertActStmt, strings.Join(cols, ","), strings.Repeat("?,", len(cols)-1)+"?")
@@ -84,13 +99,28 @@ func (ds *Datastore) NewActivity(
 		// This supposes a reasonable amount of hosts per activity, to revisit if we
 		// get in the 10K+.
 		if ah, ok := activity.(types.ActivityHosts); ok {
-			const insertActHostStmt = `INSERT INTO activity_host_past (host_id, activity_id) VALUES `
+			// >>> OPENFRAME(mysql-multitenancy): stamp each row with its host's team via a scalar
+			// subselect (host activities can be written from unpinned cron contexts, and the host is
+			// the authoritative tenant signal anyway) — engaged when the multitenancy flag is on,
+			// or when the caller context is team-pinned. Flag off keeps the statement byte-identical.
+			stampTeam := teamPinned || fleet.IsOpenframeMultitenancy()
+			insertActHostStmt := `INSERT INTO activity_host_past (host_id, activity_id) VALUES `
+			if stampTeam {
+				insertActHostStmt = `INSERT INTO activity_host_past (host_id, activity_id, team_id) VALUES `
+			}
+			// <<< OPENFRAME(mysql-multitenancy)
 
 			var sb strings.Builder
 			if hostIDs := ah.HostIDs(); len(hostIDs) > 0 {
 				sb.WriteString(insertActHostStmt)
 				actID, _ := res.LastInsertId()
 				for _, hid := range hostIDs {
+					// >>> OPENFRAME(mysql-multitenancy)
+					if stampTeam {
+						sb.WriteString(fmt.Sprintf("(%d, %d, (SELECT team_id FROM hosts WHERE id = %d)),", hid, actID, hid))
+						continue
+					}
+					// <<< OPENFRAME(mysql-multitenancy)
 					sb.WriteString(fmt.Sprintf("(%d, %d),", hid, actID))
 				}
 

server/activity/internal/mysql/new_activity_test.go

@@ -2,13 +2,18 @@ package mysql
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"testing"
 	"time"
 
 	"github.com/fleetdm/fleet/v4/server/activity/api"
 	"github.com/fleetdm/fleet/v4/server/activity/internal/testutils"
 	"github.com/fleetdm/fleet/v4/server/activity/internal/types"
+	// >>> OPENFRAME(mysql-multitenancy)
+	"github.com/fleetdm/fleet/v4/server/datastore/mysql/migrations/openframe"
+	"github.com/fleetdm/fleet/v4/server/fleet"
+	// <<< OPENFRAME(mysql-multitenancy)
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -29,6 +34,9 @@ func TestNewActivity(t *testing.T) {
 		{"HostAssociation", testNewActivityHostAssociation},
 		{"HostOnly", testNewActivityHostOnly},
 		{"DeletedUser", testNewActivityDeletedUser},
+		// >>> OPENFRAME(mysql-multitenancy)
+		{"OpenframeTeamStamping", testNewActivityOpenframeTeamStamping},
+		// <<< OPENFRAME(mysql-multitenancy)
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
@@ -244,3 +252,63 @@ func mustJSON(t *testing.T, v any) []byte {
 	require.NoError(t, err)
 	return b
 }
+
+// >>> OPENFRAME(mysql-multitenancy): CDC team_id stamping — openframe/docs/mysql-multitenancy-feature.md
+// testNewActivityOpenframeTeamStamping verifies that with the multitenancy flag on:
+//   - a team-pinned context stamps activity_past.team_id (the only tenant signal for host-less
+//     activities);
+//   - activity_host_past rows are stamped with their host's own team via the scalar subselect
+//     (host activities may be written from unpinned cron contexts).
+//
+// The test schema comes from schema.sql, which does not include the OpenFrame migration
+// pipeline, so the CDC migration is applied directly first.
+func testNewActivityOpenframeTeamStamping(t *testing.T, env *testEnv) {
+	// No t.Setenv (the harness runs parallel): the host-activity stamping gate is
+	// (multitenancy env || ctx pin), and the pinned context below engages it.
+	ctx := webhookCtx(t)
+
+	// Apply the openframe CDC migration (idempotent) — adds team_id to the captured tables.
+	tx, err := env.DB.Begin()
+	require.NoError(t, err)
+	require.NoError(t, openframe.Up_20260722000001(tx))
+	require.NoError(t, tx.Commit())
+
+	// A real team row (hosts.team_id has an FK to teams).
+	res, err := env.DB.ExecContext(t.Context(), `INSERT INTO teams (name) VALUES ('cdc-team')`)
+	require.NoError(t, err)
+	teamID64, err := res.LastInsertId()
+	require.NoError(t, err)
+	hostTeamID := uint(teamID64) //nolint:gosec // test value from LastInsertId
+	hostID := env.InsertHost(t, "cdc.local", &hostTeamID)
+
+	userID := env.InsertUser(t, "cdcuser", "cdc@example.com")
+	user := &api.User{ID: userID, Name: "cdcuser", Email: "cdc@example.com"}
+
+	// Pin the request to a different team than the host's to tell the two stamps apart.
+	const pinnedTeam = uint(77)
+	pinnedCtx := fleet.NewOpenframeTeamContext(ctx, pinnedTeam)
+
+	activity := hostActivity{
+		dummyActivity: dummyActivity{name: "ran_script", details: map[string]any{"host_id": float64(hostID)}},
+		hostIDs:       []uint{hostID},
+	}
+	require.NoError(t, env.ds.NewActivity(pinnedCtx, user, activity, mustJSON(t, activity.details), time.Now()))
+
+	// activity_past carries the request pin.
+	var actTeam sql.NullInt64
+	var actID uint
+	require.NoError(t, env.DB.QueryRowContext(t.Context(),
+		`SELECT id, team_id FROM activity_past WHERE activity_type = 'ran_script' AND user_id = ? ORDER BY id DESC LIMIT 1`,
+		userID).Scan(&actID, &actTeam))
+	require.True(t, actTeam.Valid)
+	require.EqualValues(t, pinnedTeam, actTeam.Int64)
+
+	// activity_host_past carries the HOST's team (subselect), not the pin.
+	var hostActTeam sql.NullInt64
+	require.NoError(t, env.DB.QueryRowContext(t.Context(),
+		`SELECT team_id FROM activity_host_past WHERE activity_id = ? AND host_id = ?`, actID, hostID).Scan(&hostActTeam))
+	require.True(t, hostActTeam.Valid)
+	require.EqualValues(t, hostTeamID, hostActTeam.Int64)
+}
+
+// <<< OPENFRAME(mysql-multitenancy)

server/datastore/mysql/migrations/openframe/20260722000001_AddTeamIdToCdcTables.go

@@ -0,0 +1,47 @@
+package openframe
+
+import (
+	"database/sql"
+	"fmt"
+)
+
+func init() {
+	MigrationClient.AddMigration(Up_20260722000001, Down_20260722000001)
+}
+
+// Up_20260722000001 adds a nullable `team_id` column to the four tables captured by the
+// OpenFrame Debezium CDC pipeline: activity_past, activity_host_past, query_results and
+// policy_membership. Under shared-database multitenancy one MySQL serves every tenant, so a
+// CDC record must carry its own tenant discriminator — the connector's SMTs are stateless and
+// cannot join host_id → hosts.team_id, and activity_past rows have no host reference at all.
+// The column is stamped at write time from the request's team pin (fleet.OpenframeTeamID) or,
+// for writes without a pinned context (async policy membership, host activities), from the
+// host's own team via a scalar subselect.
+//
+// Rows written with the multitenancy flag off (or by unpinned background jobs) keep team_id
+// NULL — byte-identical to pre-migration behavior. No index: nothing queries these tables by
+// team; the column exists solely so the change-stream consumer can resolve the tenant.
+//
+// Intentionally NO foreign key to teams(id): a team deletion must not rewrite or cascade over
+// millions of historical CDC rows.
+//
+// Idempotent.
+func Up_20260722000001(tx *sql.Tx) error {
+	for _, table := range []string{"activity_past", "activity_host_past", "query_results", "policy_membership"} {
+		hasCol, err := columnExists(tx, table, "team_id")
+		if err != nil {
+			return fmt.Errorf("checking %s.team_id column: %w", table, err)
+		}
+		if hasCol {
+			continue
+		}
+		if _, err := tx.Exec(fmt.Sprintf("ALTER TABLE %s ADD COLUMN team_id INT UNSIGNED NULL", table)); err != nil {
+			return fmt.Errorf("adding team_id column to %s: %w", table, err)
+		}
+	}
+	return nil
+}
+
+func Down_20260722000001(tx *sql.Tx) error {
+	return nil
+}

server/datastore/mysql/migrations_openframe_test.go

@@ -12,10 +12,12 @@ package mysql
 
 import (
 	"context"
+	"database/sql"
 	"testing"
 	"time"
 
 	"github.com/fleetdm/fleet/v4/server/fleet"
+	"github.com/fleetdm/fleet/v4/server/test"
 	"github.com/stretchr/testify/require"
 )
 
@@ -260,3 +262,70 @@ func TestOpenframeMigrationLock(t *testing.T) {
 	require.NoError(t, err)
 	release2()
 }
+
+// TestMigrateOpenframeCdcTeamIdStamping verifies 20260722000001 (team_id on the Debezium
+// CDC-captured tables) plus the write-path stamping:
+//   - unpinned context (flag off / fork-main behavior) leaves team_id NULL;
+//   - a team-pinned context stamps query_results rows;
+//   - the async policy-membership collector (no pinned context) stamps each row from its
+//     host's own team via the flag-gated scalar subselect.
+func TestMigrateOpenframeCdcTeamIdStamping(t *testing.T) {
+	ds := CreateMySQLDS(t)
+	ctx := context.Background()
+
+	require.NoError(t, ds.MigrateOpenframe(ctx))
+	for _, table := range []string{"activity_past", "activity_host_past", "query_results", "policy_membership"} {
+		var n int
+		require.NoError(t, ds.writer(ctx).GetContext(ctx, &n,
+			"SELECT COUNT(*) FROM information_schema.COLUMNS WHERE table_schema = DATABASE() AND table_name = ? AND column_name = 'team_id'",
+			table))
+		require.Equal(t, 1, n, "expected %s.team_id after MigrateOpenframe", table)
+	}
+
+	user := test.NewUser(t, ds, "CDC User", "cdc@example.com", true)
+	query := test.NewQuery(t, ds, nil, "CDC Query", "SELECT 1", user.ID, true)
+	host := test.NewHost(t, ds, "cdc-host", "192.168.1.10", "cdc-key", "cdc-uuid", time.Now())
+
+	rows := []*fleet.ScheduledQueryResultRow{
+		{QueryID: query.ID, HostID: host.ID, LastFetched: time.Now().UTC().Truncate(time.Second)},
+	}
+
+	// Unpinned: team_id stays NULL — byte-identical to fork-main.
+	_, err := ds.OverwriteQueryResultRows(ctx, rows, fleet.DefaultMaxQueryReportRows)
+	require.NoError(t, err)
+	var teamIDs []sql.NullInt64
+	require.NoError(t, ds.writer(ctx).SelectContext(ctx, &teamIDs,
+		"SELECT team_id FROM query_results WHERE host_id = ?", host.ID))
+	require.Len(t, teamIDs, 1)
+	require.False(t, teamIDs[0].Valid)
+
+	// Pinned: stamped with the pin (overwrite deletes + reinserts the host's rows).
+	pinned := fleet.NewOpenframeTeamContext(ctx, 42)
+	_, err = ds.OverwriteQueryResultRows(pinned, rows, fleet.DefaultMaxQueryReportRows)
+	require.NoError(t, err)
+	require.NoError(t, ds.writer(ctx).SelectContext(ctx, &teamIDs,
+		"SELECT team_id FROM query_results WHERE host_id = ?", host.ID))
+	require.Len(t, teamIDs, 1)
+	require.True(t, teamIDs[0].Valid)
+	require.EqualValues(t, 42, teamIDs[0].Int64)
+
+	// Async policy membership: stamping comes from the host row via the flag-gated subselect.
+	// The gate is (multitenancy env || ctx pin); the pinned context triggers it here without
+	// t.Setenv (forbidden — CreateMySQLDS forces t.Parallel), and the SQL ignores the pin value
+	// and reads hosts.team_id, exactly as the unpinned production cron path does.
+	team, err := ds.NewTeam(ctx, &fleet.Team{Name: "cdc-team"})
+	require.NoError(t, err)
+	require.NoError(t, ds.AddHostsToTeam(ctx, fleet.NewAddHostsToTeamParams(&team.ID, []uint{host.ID})))
+	pol, err := ds.NewGlobalPolicy(ctx, &user.ID, fleet.PolicyPayload{Name: "cdc-policy", Query: "SELECT 1"})
+	require.NoError(t, err)
+
+	passes := true
+	require.NoError(t, ds.AsyncBatchInsertPolicyMembership(pinned, []fleet.PolicyMembershipResult{
+		{PolicyID: pol.ID, HostID: host.ID, Passes: &passes},
+	}))
+	var memberTeam sql.NullInt64
+	require.NoError(t, ds.writer(ctx).GetContext(ctx, &memberTeam,
+		"SELECT team_id FROM policy_membership WHERE policy_id = ? AND host_id = ?", pol.ID, host.ID))
+	require.True(t, memberTeam.Valid)
+	require.EqualValues(t, team.ID, memberTeam.Int64)
+}