From 3085e1c7aa9f13dd4bbeb513ed4bbb68ed48d70d Mon Sep 17 00:00:00 2001 From: Benjamin Knofe-Vider Date: Mon, 20 Jul 2026 15:14:38 +0200 Subject: [PATCH] feat(configstore): org team CRUD APIs, admin UI, and legacy table-name grandfathering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Managed warehouses now carry many PostHog teams per org (duckgres_org_teams, migration 000024); this adds the surfaces to manage those rows. Migration 000025 adds three nullable legacy table-name overrides to duckgres_org_teams (events_table_name, persons_table_name, schema_data_imports_name — NULL means derive from schema_name: .events, .persons, _data_imports; non-NULL values are grandfathered explicit names for pre-existing teams) and a unique (org_id, schema_name) index so two teams in one org can never share a schema. Provisioning API (internal secret, for the PostHog backend): - GET /api/v1/orgs/:id/teams — list the org's team rows - POST /api/v1/orgs/:id/teams — create/upsert. This IS the grandfather path: upserting an existing (org, team) row may overwrite schema_name and the legacy table names, because the PostHog backfill replaces the migration's 'team_' placeholder through it. schema_name is validated as a safe lowercase identifier; a duplicate within the org is a 409. - DELETE /api/v1/orgs/:id/teams/:team_id — config-only delete (never warehouse data). Deleting the billing team promotes the remaining team with the oldest created_at and re-attributes unacked usage buckets in the same transaction (ReattributeUsageTeamTx); the org's last team is refused with a 409 (deleting the org is the only way). - The provision body gains optional team_id + schema_name that create the first (billing) team row with an explicit schema. The legacy default_team_id field keeps working (this ships before the PostHog-side change); when both are given they must agree or 400. Admin API + console (user-facing surface — schema_name is immutable here, unlike the provisioning grandfather path): GET /teams (cross-org list), POST /teams (create-only, org_id in the body), and PUT /orgs/:id/teams/:team_id (enabled / backfill_enabled / billing repoint with atomic usage re-attribution; billing can never be cleared, only repointed). Per-org list and delete reuse the provisioning routes on the same audited router group — gin refuses duplicate paths. The UI gains an 'Org teams' sidebar page below Organizations (full CRUD with last-team and billing-handover warnings), a Teams card on the org detail page, and an Org Teams entry in the config-store explorer after Orgs. Shared rules live in configstore (UpsertOrgTeamTx, DeleteOrgTeamTx, ValidateOrgTeamSchemaName). Tests cover the migration shape, the grandfather overwrite, per-org schema 409 (and cross-org reuse), billing handover ordering by created_at with usage re-attribution, last-team refusal, and both provision paths; the e2e harness gains an org_teams_crud leg. --- CLAUDE.md | 20 + controlplane/admin/api.go | 282 ++++++++++++ controlplane/admin/api_postgres_test.go | 74 ++++ controlplane/admin/api_test.go | 248 +++++++++++ controlplane/admin/audit.go | 6 + controlplane/admin/audit_test.go | 4 + controlplane/admin/models_api.go | 1 + controlplane/admin/models_api_test.go | 5 +- controlplane/admin/ui/src/App.tsx | 2 + .../ui/src/components/OrgTeamDialogs.tsx | 360 +++++++++++++++ controlplane/admin/ui/src/components/nav.ts | 2 + controlplane/admin/ui/src/hooks/useApi.ts | 56 +++ controlplane/admin/ui/src/lib/api.ts | 15 + controlplane/admin/ui/src/lib/audit.ts | 4 + controlplane/admin/ui/src/pages/OrgDetail.tsx | 127 +++++- .../admin/ui/src/pages/OrgTeams.test.tsx | 114 +++++ controlplane/admin/ui/src/pages/OrgTeams.tsx | 194 ++++++++ controlplane/admin/ui/src/types/api.ts | 38 +- .../000025_org_teams_table_names.sql | 25 ++ controlplane/configstore/models.go | 33 +- controlplane/configstore/org_teams.go | 264 +++++++++++ controlplane/provisioning/api.go | 162 ++++++- controlplane/provisioning/api_test.go | 419 ++++++++++++++++++ controlplane/provisioning/store.go | 81 +++- tests/configstore/migrations_postgres_test.go | 68 ++- tests/configstore/org_teams_postgres_test.go | 245 ++++++++++ tests/mw-dev/e2e/harness.sh | 66 +++ 27 files changed, 2891 insertions(+), 24 deletions(-) create mode 100644 controlplane/admin/ui/src/components/OrgTeamDialogs.tsx create mode 100644 controlplane/admin/ui/src/pages/OrgTeams.test.tsx create mode 100644 controlplane/admin/ui/src/pages/OrgTeams.tsx create mode 100644 controlplane/configstore/migrations/000025_org_teams_table_names.sql create mode 100644 tests/configstore/org_teams_postgres_test.go diff --git a/CLAUDE.md b/CLAUDE.md index e4480593..a7da1989 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -535,6 +535,26 @@ touching this path: `controlplane/admin/api_test.go` + `api_postgres_test.go` (`TestUpdateOrg*Reattribut*`), and the re-attribution leg of `compute_usage_pull_api` in the e2e harness. +- **Org team CRUD (`duckgres_org_teams`)**: the PostHog backend manages an + org's team rows via `GET/POST /api/v1/orgs/:id/teams` + + `DELETE /api/v1/orgs/:id/teams/:team_id` (internal secret, + `controlplane/provisioning`). The POST is the **grandfather upsert**: it MAY + overwrite an existing row's `schema_name` and the legacy + `events_table_name`/`persons_table_name`/`schema_data_imports_name` + overrides (NULL = derive from `schema_name`: `.events`, + `.persons`, `_data_imports`), because the PostHog backfill + replaces migration 000024's `team_` placeholder through it. Two teams in + one org can never share a schema (unique `(org_id, schema_name)`, migration + 000025 → 409). DELETE removes CONFIG only (never warehouse data); deleting + the billing team promotes the remaining team with the OLDEST `created_at` + and re-attributes unacked usage in the same TXN; the org's LAST team is + undeletable (409 — delete the org). The admin console mirrors this on a + user-facing surface (`GET /teams`, `POST /teams`, + `PUT /orgs/:id/teams/:team_id`) where `schema_name` is immutable and + billing can only be repointed, never cleared. Shared rules live in + `configstore.UpsertOrgTeamTx` / `DeleteOrgTeamTx`; tests: + `tests/configstore/org_teams_postgres_test.go`, the provisioning/admin API + tests, and `org_teams_crud` in the e2e harness. - **Storage metric** (`managed_warehouse_storage_gib_seconds`, `storage_meter.go`): a LEADER-ONLY sampler (double writers would double-bill — the UPSERT is additive) visits each Ready warehouse's DuckLake diff --git a/controlplane/admin/api.go b/controlplane/admin/api.go index 439a23b0..c2ee80c5 100644 --- a/controlplane/admin/api.go +++ b/controlplane/admin/api.go @@ -11,6 +11,7 @@ import ( "log/slog" "net/http" "sort" + "strconv" "strings" "time" @@ -112,6 +113,18 @@ func registerAPIWithStore(r *gin.RouterGroup, store apiStore, info OrgStackInfo, // run without ever touching the config-store DB directly. r.PATCH("/orgs/:id/warehouse/pinning", h.patchTenantPinning) + // Org teams (duckgres_org_teams). The per-org list/upsert/delete routes + // (GET/POST /orgs/:id/teams, DELETE /orgs/:id/teams/:team_id) are + // registered by the provisioning API on this same group — gin refuses + // duplicate routes, so the admin surface adds only what provisioning + // doesn't have: the cross-org list, a CREATE-ONLY POST (org_id in the + // body, mirroring POST /users), and the update. Unlike the internal + // provisioning upsert, this is a user-facing surface: schema_name is set + // once at create time and immutable afterwards (the PUT rejects it). + r.GET("/teams", h.listAllOrgTeams) + r.POST("/teams", h.createOrgTeam) + r.PUT("/orgs/:id/teams/:team_id", h.updateOrgTeam) + // Users CRUD r.GET("/users", h.listUsers) r.POST("/users", h.createUser) @@ -143,6 +156,18 @@ type apiStore interface { UpdateOrg(name string, updates configstore.Org, reattributeUsageTeam *int64) (*configstore.Org, bool, error) DeleteOrg(name string) (bool, error) + // Org teams. CreateOrgTeam is create-only (the admin surface never + // overwrites an existing row — schema_name is immutable here); it returns + // errOrgTeamExists / configstore.ErrOrgTeamSchemaConflict for the two + // conflict shapes, gorm.ErrRecordNotFound for an unknown org. + // UpdateOrgTeam mutates enabled / backfill_enabled and can repoint the + // billing team (usage buckets re-attributed in the same transaction). + // Per-org list and delete are served by the provisioning API's routes on + // the same router group (identical rules — configstore.DeleteOrgTeamTx). + ListAllOrgTeams() ([]configstore.OrgTeam, error) + CreateOrgTeam(orgID string, team *configstore.OrgTeam) error + UpdateOrgTeam(orgID string, teamID int64, upd orgTeamUpdate) (*configstore.OrgTeam, error) + ListUsers() ([]configstore.OrgUser, error) CreateUser(user *configstore.OrgUser) error GetUser(orgID, username string) (*configstore.OrgUser, error) @@ -320,6 +345,119 @@ func (s *gormAPIStore) DeleteOrg(name string) (bool, error) { return returnRows > 0, nil } +// errOrgTeamExists distinguishes the (org, team) primary-key conflict from +// the schema-name conflict on the admin create endpoint — the admin surface +// never overwrites an existing row (that's the internal provisioning +// grandfather path), so an existing row is a 409, not an upsert. +var errOrgTeamExists = errors.New("team already exists in this org") + +// orgTeamUpdate carries the admin-editable fields of one org team. Pointer +// fields are presence-aware (nil = preserve). schema_name is deliberately NOT +// here: it is immutable on the admin surface. BackfillSet distinguishes an +// explicit `"backfill_enabled": null` (clear to unset) from an absent key. +type orgTeamUpdate struct { + Enabled *bool + BackfillSet bool + Backfill *bool + // MakeBilling repoints the org's billing team to this row (the buffered + // usage buckets follow atomically). There is no "unset billing" — every + // org must keep a billing team; repoint by marking another team. + MakeBilling bool +} + +func (s *gormAPIStore) ListAllOrgTeams() ([]configstore.OrgTeam, error) { + var teams []configstore.OrgTeam + if err := s.db().Order("org_id, team_id").Find(&teams).Error; err != nil { + return nil, err + } + return teams, nil +} + +func (s *gormAPIStore) CreateOrgTeam(orgID string, team *configstore.OrgTeam) error { + return s.db().Transaction(func(tx *gorm.DB) error { + var orgCount int64 + if err := tx.Model(&configstore.Org{}).Where("name = ?", orgID).Count(&orgCount).Error; err != nil { + return err + } + if orgCount == 0 { + return gorm.ErrRecordNotFound + } + var dup int64 + if err := tx.Model(&configstore.OrgTeam{}). + Where("org_id = ? AND team_id = ?", orgID, team.TeamID).Count(&dup).Error; err != nil { + return err + } + if dup > 0 { + return errOrgTeamExists + } + var schemaClash int64 + if err := tx.Model(&configstore.OrgTeam{}). + Where("org_id = ? AND schema_name = ?", orgID, team.SchemaName).Count(&schemaClash).Error; err != nil { + return err + } + if schemaClash > 0 { + return configstore.ErrOrgTeamSchemaConflict + } + team.OrgID = orgID + return tx.Create(team).Error + }) +} + +func (s *gormAPIStore) UpdateOrgTeam(orgID string, teamID int64, upd orgTeamUpdate) (*configstore.OrgTeam, error) { + var stored configstore.OrgTeam + err := s.db().Transaction(func(tx *gorm.DB) error { + var team configstore.OrgTeam + err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + First(&team, "org_id = ? AND team_id = ?", orgID, teamID).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return configstore.ErrOrgTeamNotFound + } + if err != nil { + return err + } + + fields := map[string]interface{}{"updated_at": gorm.Expr("now()")} + if upd.Enabled != nil { + fields["enabled"] = *upd.Enabled + } + if upd.BackfillSet { + fields["backfill_enabled"] = upd.Backfill + } + if len(fields) > 1 { + if err := tx.Model(&configstore.OrgTeam{}). + Where("org_id = ? AND team_id = ?", orgID, teamID). + Updates(fields).Error; err != nil { + return err + } + } + + if upd.MakeBilling && (team.IsBillingTeam == nil || !*team.IsBillingTeam) { + // Repointing billing carries the org's buffered usage buckets + // along in the SAME transaction — identical to the + // default_team_id repoint on the org endpoints. + oldTeam, err := configstore.OrgBillingTeamIDTx(tx, orgID) + if err != nil { + return err + } + if err := configstore.SetOrgBillingTeamTx(tx, orgID, teamID); err != nil { + return err + } + moved, err := configstore.ReattributeUsageTeamTx(tx, orgID, teamID) + if err != nil { + return err + } + slog.Info("Re-attributed org usage buckets to new billing team.", + "org", orgID, "old_team", oldTeam, "new_team", teamID, "rows", moved) + } + + return tx.First(&stored, "org_id = ? AND team_id = ?", orgID, teamID).Error + }) + if err != nil { + return nil, err + } + return &stored, nil +} + func (s *gormAPIStore) ListUsers() ([]configstore.OrgUser, error) { var users []configstore.OrgUser if err := s.db().Find(&users).Error; err != nil { @@ -759,6 +897,150 @@ func (h *apiHandler) deleteOrg(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"deleted": name}) } +// --- Org teams --- + +func (h *apiHandler) listAllOrgTeams(c *gin.Context) { + teams, err := h.store.ListAllOrgTeams() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"teams": teams}) +} + +// createOrgTeamRequest is the admin create body (POST /teams — the org rides +// in the body like POST /users, because the /orgs/:id/teams POST route +// belongs to the provisioning grandfather upsert). schema_name is set here +// ONCE and immutable afterwards on this surface (the update endpoint rejects +// it) — only the internal provisioning grandfather path may overwrite it. +type createOrgTeamRequest struct { + OrgID string `json:"org_id"` + TeamID int64 `json:"team_id"` + SchemaName string `json:"schema_name"` + Enabled *bool `json:"enabled,omitempty"` + BackfillEnabled *bool `json:"backfill_enabled,omitempty"` +} + +func (h *apiHandler) createOrgTeam(c *gin.Context) { + var req createOrgTeamRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + orgID := req.OrgID + if orgID == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "org_id is required"}) + return + } + if req.TeamID <= 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "team_id is required (a positive PostHog team id)"}) + return + } + if err := configstore.ValidateOrgTeamSchemaName(req.SchemaName); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + team := configstore.OrgTeam{ + OrgID: orgID, + TeamID: req.TeamID, + SchemaName: req.SchemaName, + Enabled: req.Enabled == nil || *req.Enabled, + BackfillEnabled: req.BackfillEnabled, + } + // POST /teams has no :id param, so the audit org column is blank — record + // the target org here (mirrors POST /users). + setAuditDetail(c, fmt.Sprintf("created team %d (schema %s) in org %s", req.TeamID, req.SchemaName, orgID)) + if err := h.store.CreateOrgTeam(orgID, &team); err != nil { + switch { + case errors.Is(err, gorm.ErrRecordNotFound): + c.JSON(http.StatusNotFound, gin.H{"error": "org not found"}) + case errors.Is(err, errOrgTeamExists), errors.Is(err, configstore.ErrOrgTeamSchemaConflict): + c.JSON(http.StatusConflict, gin.H{"error": err.Error()}) + default: + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + } + return + } + c.JSON(http.StatusCreated, team) +} + +// updateOrgTeamRequest is the admin PUT body: enabled / backfill_enabled / +// is_billing_team only. schema_name is immutable on this surface, and billing +// can only be pointed AT a team (is_billing_team: true) — never cleared. +type updateOrgTeamRequest struct { + Enabled *bool `json:"enabled,omitempty"` + BackfillEnabled *bool `json:"backfill_enabled,omitempty"` + IsBillingTeam *bool `json:"is_billing_team,omitempty"` +} + +func (h *apiHandler) updateOrgTeam(c *gin.Context) { + orgID := c.Param("id") + teamID, perr := strconv.ParseInt(c.Param("team_id"), 10, 64) + if perr != nil || teamID <= 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "team_id must be a positive integer"}) + return + } + body, err := io.ReadAll(c.Request.Body) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + var req updateOrgTeamRequest + if err := json.Unmarshal(body, &req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(body, &fields); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if _, ok := fields["schema_name"]; ok { + c.JSON(http.StatusBadRequest, gin.H{"error": "schema_name is immutable — it is set once when the team is created"}) + return + } + if _, ok := fields["is_billing_team"]; ok && (req.IsBillingTeam == nil || !*req.IsBillingTeam) { + c.JSON(http.StatusBadRequest, gin.H{"error": "is_billing_team can only be set to true (repoint billing by marking another team); every org must keep a billing team"}) + return + } + _, backfillSet := fields["backfill_enabled"] + upd := orgTeamUpdate{ + Enabled: req.Enabled, + BackfillSet: backfillSet, + Backfill: req.BackfillEnabled, + MakeBilling: req.IsBillingTeam != nil && *req.IsBillingTeam, + } + + var changes []string + if req.Enabled != nil { + changes = append(changes, fmt.Sprintf("enabled=%v", *req.Enabled)) + } + if backfillSet { + if req.BackfillEnabled == nil { + changes = append(changes, "backfill_enabled=(unset)") + } else { + changes = append(changes, fmt.Sprintf("backfill_enabled=%v", *req.BackfillEnabled)) + } + } + if upd.MakeBilling { + changes = append(changes, "billing team") + } + if len(changes) > 0 { + setAuditDetail(c, fmt.Sprintf("team %d: %s", teamID, strings.Join(changes, ", "))) + } + + team, err := h.store.UpdateOrgTeam(orgID, teamID, upd) + if err != nil { + if errors.Is(err, configstore.ErrOrgTeamNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "org team not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, team) +} + // topLevelJSONKeys returns the sorted top-level object keys of a JSON body, or // nil if it isn't a JSON object. Used to audit WHICH warehouse sections a PUT // touched without recording their (possibly secret) values. diff --git a/controlplane/admin/api_postgres_test.go b/controlplane/admin/api_postgres_test.go index e233ba07..20b7af90 100644 --- a/controlplane/admin/api_postgres_test.go +++ b/controlplane/admin/api_postgres_test.go @@ -3,6 +3,7 @@ package admin import ( + "errors" "fmt" "os" "os/exec" @@ -360,3 +361,76 @@ func TestMutateManagedWarehouseSerializesConcurrentWriters(t *testing.T) { t.Fatalf("status_message = %q, want %q (lost updates under concurrency)", final.StatusMessage, want) } } + +// TestAdminOrgTeamStorePostgres exercises the real gormAPIStore team methods: +// billing repoint carries the buffered usage buckets, and create enforces +// per-org schema uniqueness. (Delete rules live in configstore.DeleteOrgTeamTx +// and are covered by tests/configstore/org_teams_postgres_test.go — the admin +// UI deletes through the provisioning-registered route.) +func TestAdminOrgTeamStorePostgres(t *testing.T) { + store := newPostgresConfigStore(t) + apiStore := newGormAPIStore(store).(*gormAPIStore) + + if err := store.DB().Create(&configstore.Org{Name: "teamsorg", DatabaseName: "teamsorgdb"}).Error; err != nil { + t.Fatalf("create org: %v", err) + } + base := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + if err := store.DB().Exec(` + INSERT INTO duckgres_org_teams (org_id, team_id, schema_name, enabled, is_billing_team, created_at, updated_at) + VALUES ('teamsorg', 1, 'team_1', TRUE, TRUE, ?, ?), ('teamsorg', 2, 'team_2', TRUE, NULL, ?, ?)`, + base, base, base.Add(time.Hour), base.Add(time.Hour)).Error; err != nil { + t.Fatalf("seed teams: %v", err) + } + bucket := time.Date(2026, 7, 14, 10, 0, 0, 0, time.UTC) + if err := store.FlushComputeUsage([]configstore.ComputeUsageDelta{{ + OrgID: "teamsorg", TeamID: 1, QuerySource: "standard", + Millicores: 2000, MiB: 4096, BucketStart: bucket, + CPUSeconds: 10, MemorySeconds: 20, + }}); err != nil { + t.Fatalf("seed usage: %v", err) + } + + // Repoint billing to team 2: the buffered bucket must follow atomically. + updated, err := apiStore.UpdateOrgTeam("teamsorg", 2, orgTeamUpdate{MakeBilling: true}) + if err != nil { + t.Fatalf("repoint billing: %v", err) + } + if updated.IsBillingTeam == nil || !*updated.IsBillingTeam { + t.Fatalf("team 2 must be billing, got %+v", updated) + } + var usageTeams []int64 + if err := store.DB().Raw(`SELECT team_id FROM duckgres_org_compute_usage WHERE org_id = 'teamsorg'`).Scan(&usageTeams).Error; err != nil { + t.Fatalf("read usage teams: %v", err) + } + if len(usageTeams) != 1 || usageTeams[0] != 2 { + t.Fatalf("usage teams after repoint = %v, want [2]", usageTeams) + } + + // Create with a schema another team already holds: refused. + err = apiStore.CreateOrgTeam("teamsorg", &configstore.OrgTeam{TeamID: 3, SchemaName: "team_1", Enabled: true}) + if !errors.Is(err, configstore.ErrOrgTeamSchemaConflict) { + t.Fatalf("duplicate schema create err = %v, want ErrOrgTeamSchemaConflict", err) + } + // A duplicate (org, team) is the other conflict shape. + err = apiStore.CreateOrgTeam("teamsorg", &configstore.OrgTeam{TeamID: 1, SchemaName: "fresh", Enabled: true}) + if !errors.Is(err, errOrgTeamExists) { + t.Fatalf("duplicate team create err = %v, want errOrgTeamExists", err) + } + // A fresh schema works and appears in the cross-org list. + if err := apiStore.CreateOrgTeam("teamsorg", &configstore.OrgTeam{TeamID: 3, SchemaName: "team_3", Enabled: true}); err != nil { + t.Fatalf("create team 3: %v", err) + } + teams, err := apiStore.ListAllOrgTeams() + if err != nil { + t.Fatalf("list all teams: %v", err) + } + var got []int64 + for _, tm := range teams { + if tm.OrgID == "teamsorg" { + got = append(got, tm.TeamID) + } + } + if len(got) != 3 || got[0] != 1 || got[1] != 2 || got[2] != 3 { + t.Fatalf("teamsorg teams = %v, want [1 2 3]", got) + } +} diff --git a/controlplane/admin/api_test.go b/controlplane/admin/api_test.go index b89729b9..b354feec 100644 --- a/controlplane/admin/api_test.go +++ b/controlplane/admin/api_test.go @@ -10,6 +10,7 @@ import ( "net/http" "net/http/httptest" "slices" + "sort" "strings" "testing" @@ -22,10 +23,14 @@ type fakeAPIStore struct { orgs map[string]*configstore.Org users map[string]*configstore.OrgUser warehouses map[string]*configstore.ManagedWarehouse + teams map[string]map[int64]*configstore.OrgTeam // reattributeCalls records every non-nil reattributeUsageTeam passed to // UpdateOrg (org → new team ids, in order), so tests can assert the handler // requests bucket re-attribution exactly when default_team_id changes. reattributeCalls []fakeReattributeCall + // teamBillingRepoints records every billing repoint requested through + // UpdateOrgTeam (usage re-attribution rides along in the real store). + teamBillingRepoints []fakeReattributeCall } type fakeReattributeCall struct { @@ -38,6 +43,7 @@ func newFakeAPIStore() *fakeAPIStore { orgs: make(map[string]*configstore.Org), users: make(map[string]*configstore.OrgUser), warehouses: make(map[string]*configstore.ManagedWarehouse), + teams: make(map[string]map[int64]*configstore.OrgTeam), } } @@ -114,6 +120,70 @@ func (s *fakeAPIStore) DeleteOrg(name string) (bool, error) { return true, nil } +func (s *fakeAPIStore) seedTeam(team configstore.OrgTeam) { + if s.teams[team.OrgID] == nil { + s.teams[team.OrgID] = make(map[int64]*configstore.OrgTeam) + } + clone := team + s.teams[team.OrgID][team.TeamID] = &clone +} + +func (s *fakeAPIStore) ListAllOrgTeams() ([]configstore.OrgTeam, error) { + var out []configstore.OrgTeam + for _, orgTeams := range s.teams { + for _, t := range orgTeams { + out = append(out, *t) + } + } + sort.Slice(out, func(i, j int) bool { + if out[i].OrgID != out[j].OrgID { + return out[i].OrgID < out[j].OrgID + } + return out[i].TeamID < out[j].TeamID + }) + return out, nil +} + +func (s *fakeAPIStore) CreateOrgTeam(orgID string, team *configstore.OrgTeam) error { + if _, ok := s.orgs[orgID]; !ok { + return gorm.ErrRecordNotFound + } + if _, ok := s.teams[orgID][team.TeamID]; ok { + return errOrgTeamExists + } + for _, t := range s.teams[orgID] { + if t.SchemaName == team.SchemaName { + return configstore.ErrOrgTeamSchemaConflict + } + } + team.OrgID = orgID + s.seedTeam(*team) + return nil +} + +func (s *fakeAPIStore) UpdateOrgTeam(orgID string, teamID int64, upd orgTeamUpdate) (*configstore.OrgTeam, error) { + team, ok := s.teams[orgID][teamID] + if !ok { + return nil, configstore.ErrOrgTeamNotFound + } + if upd.Enabled != nil { + team.Enabled = *upd.Enabled + } + if upd.BackfillSet { + team.BackfillEnabled = upd.Backfill + } + if upd.MakeBilling && (team.IsBillingTeam == nil || !*team.IsBillingTeam) { + for _, t := range s.teams[orgID] { + t.IsBillingTeam = nil + } + billing := true + team.IsBillingTeam = &billing + s.teamBillingRepoints = append(s.teamBillingRepoints, fakeReattributeCall{org: orgID, teamID: teamID}) + } + clone := *team + return &clone, nil +} + func (s *fakeAPIStore) ListUsers() ([]configstore.OrgUser, error) { users := make([]configstore.OrgUser, 0, len(s.users)) for _, user := range s.users { @@ -2163,3 +2233,181 @@ func TestCreateOrgRejectsInvalidDefaultWorkerTTL(t *testing.T) { t.Fatal("org should NOT have been created") } } + +// --- Org teams (admin surface) --- + +func adminJSON(t *testing.T, router *gin.Engine, method, path, body string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(method, path, bytes.NewReader([]byte(body))) + if body != "" { + req.Header.Set("Content-Type", "application/json") + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +func TestAdminCreateOrgTeam(t *testing.T) { + store := newFakeAPIStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + router := newTestAPIRouter(store) + + rec := adminJSON(t, router, http.MethodPost, "/api/v1/teams", + `{"org_id":"acme","team_id":5,"schema_name":"team_5","backfill_enabled":false}`) + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d, want 201: %s", rec.Code, rec.Body.String()) + } + stored := store.teams["acme"][5] + if stored == nil || stored.SchemaName != "team_5" || !stored.Enabled { + t.Fatalf("stored team = %+v, want schema team_5, enabled default", stored) + } + if stored.BackfillEnabled == nil || *stored.BackfillEnabled { + t.Fatalf("backfill_enabled = %v, want false", stored.BackfillEnabled) + } +} + +func TestAdminCreateOrgTeamConflicts(t *testing.T) { + store := newFakeAPIStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true}) + router := newTestAPIRouter(store) + + // Existing (org, team) is a 409 — the admin surface never overwrites + // (that's the internal provisioning grandfather path). + rec := adminJSON(t, router, http.MethodPost, "/api/v1/teams", + `{"org_id":"acme","team_id":1,"schema_name":"other"}`) + if rec.Code != http.StatusConflict { + t.Fatalf("existing team: status = %d, want 409: %s", rec.Code, rec.Body.String()) + } + if store.teams["acme"][1].SchemaName != "team_1" { + t.Fatal("admin create must never overwrite an existing row") + } + + // Duplicate schema within the org is a 409 too. + rec = adminJSON(t, router, http.MethodPost, "/api/v1/teams", + `{"org_id":"acme","team_id":2,"schema_name":"team_1"}`) + if rec.Code != http.StatusConflict { + t.Fatalf("duplicate schema: status = %d, want 409: %s", rec.Code, rec.Body.String()) + } + + // Unknown org is a 404. + rec = adminJSON(t, router, http.MethodPost, "/api/v1/teams", + `{"org_id":"nope","team_id":2,"schema_name":"team_2"}`) + if rec.Code != http.StatusNotFound { + t.Fatalf("unknown org: status = %d, want 404: %s", rec.Code, rec.Body.String()) + } + + // Missing org_id is a 400. + rec = adminJSON(t, router, http.MethodPost, "/api/v1/teams", + `{"team_id":2,"schema_name":"team_2"}`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("missing org_id: status = %d, want 400: %s", rec.Code, rec.Body.String()) + } +} + +func TestAdminUpdateOrgTeamRejectsSchemaChange(t *testing.T) { + store := newFakeAPIStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true}) + router := newTestAPIRouter(store) + + rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/acme/teams/1", + `{"schema_name":"renamed"}`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "immutable") { + t.Fatalf("error should say schema_name is immutable: %s", rec.Body.String()) + } + if store.teams["acme"][1].SchemaName != "team_1" { + t.Fatal("schema_name must not change through the admin update") + } +} + +func TestAdminUpdateOrgTeamFields(t *testing.T) { + store := newFakeAPIStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + backfill := true + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true, BackfillEnabled: &backfill}) + router := newTestAPIRouter(store) + + // enabled=false + explicit backfill null clears the tri-state. + rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/acme/teams/1", + `{"enabled":false,"backfill_enabled":null}`) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + stored := store.teams["acme"][1] + if stored.Enabled { + t.Fatal("enabled must be cleared") + } + if stored.BackfillEnabled != nil { + t.Fatalf("backfill_enabled = %v, want NULL after explicit null", *stored.BackfillEnabled) + } +} + +func TestAdminUpdateOrgTeamBillingRepoint(t *testing.T) { + store := newFakeAPIStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + billing := true + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true, IsBillingTeam: &billing}) + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 2, SchemaName: "team_2", Enabled: true}) + router := newTestAPIRouter(store) + + // Clearing billing is rejected — repoint only. + rec := adminJSON(t, router, http.MethodPut, "/api/v1/orgs/acme/teams/1", + `{"is_billing_team":false}`) + if rec.Code != http.StatusBadRequest { + t.Fatalf("clear billing: status = %d, want 400: %s", rec.Code, rec.Body.String()) + } + + rec = adminJSON(t, router, http.MethodPut, "/api/v1/orgs/acme/teams/2", + `{"is_billing_team":true}`) + if rec.Code != http.StatusOK { + t.Fatalf("repoint: status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + if got := store.teams["acme"][2]; got.IsBillingTeam == nil || !*got.IsBillingTeam { + t.Fatalf("team 2 must be billing after repoint, got %+v", got) + } + if len(store.teamBillingRepoints) != 1 || store.teamBillingRepoints[0] != (fakeReattributeCall{org: "acme", teamID: 2}) { + t.Fatalf("billing repoint (with usage re-attribution) not requested: %+v", store.teamBillingRepoints) + } + + // Marking the current billing team again is a no-op, not a re-attribution. + rec = adminJSON(t, router, http.MethodPut, "/api/v1/orgs/acme/teams/2", + `{"is_billing_team":true}`) + if rec.Code != http.StatusOK { + t.Fatalf("idempotent repoint: status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + if len(store.teamBillingRepoints) != 1 { + t.Fatalf("idempotent repoint must not re-attribute again: %+v", store.teamBillingRepoints) + } +} + +func TestAdminListOrgTeams(t *testing.T) { + store := newFakeAPIStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + store.orgs["zeta"] = &configstore.Org{Name: "zeta"} + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 2, SchemaName: "team_2", Enabled: true}) + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true}) + store.seedTeam(configstore.OrgTeam{OrgID: "zeta", TeamID: 9, SchemaName: "team_9", Enabled: true}) + router := newTestAPIRouter(store) + + rec := adminJSON(t, router, http.MethodGet, "/api/v1/teams", "") + if rec.Code != http.StatusOK { + t.Fatalf("global list: status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + var global struct { + Teams []configstore.OrgTeam `json:"teams"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &global); err != nil { + t.Fatalf("decode: %v", err) + } + if len(global.Teams) != 3 { + t.Fatalf("global teams = %d rows, want 3", len(global.Teams)) + } + + if global.Teams[0].OrgID != "acme" || global.Teams[0].TeamID != 1 || global.Teams[2].OrgID != "zeta" { + t.Fatalf("global teams order = %+v, want acme[1,2] then zeta[9]", global.Teams) + } +} diff --git a/controlplane/admin/audit.go b/controlplane/admin/audit.go index 390c3367..ef93c0fd 100644 --- a/controlplane/admin/audit.go +++ b/controlplane/admin/audit.go @@ -177,6 +177,9 @@ func auditActionFor(method, path string) string { case "users": // Top-level user create (POST /users) — not nested under an org. return "user." + verb + case "teams": + // Top-level team create (POST /teams) — the org rides in the body. + return "team." + verb case "sessions": // /sessions/:pid/cancel and /sessions/by-worker/:wid/cancel. if last == "cancel" { @@ -198,6 +201,9 @@ func auditActionFor(method, path string) string { case hasSeg(segs, "warehouse"): // PUT /warehouse and PATCH /warehouse/pinning both map here. return "warehouse." + verb + case hasSeg(segs, "teams"): + // /orgs/:id/teams and /orgs/:id/teams/:team_id. + return "team." + verb case hasSeg(segs, "impersonate"): // Impersonation records its own richer row; this is a fallback. return "impersonate." + verb diff --git a/controlplane/admin/audit_test.go b/controlplane/admin/audit_test.go index 9f6cb940..0a52a15f 100644 --- a/controlplane/admin/audit_test.go +++ b/controlplane/admin/audit_test.go @@ -27,6 +27,10 @@ func TestAuditActionFor(t *testing.T) { {"org user disable", http.MethodPost, "/api/v1/orgs/acme/users/bob/disable", "user.disable"}, {"org user enable", http.MethodPost, "/api/v1/orgs/acme/users/bob/enable", "user.enable"}, {"warehouse put", http.MethodPut, "/api/v1/orgs/acme/warehouse", "warehouse.update"}, + {"team create (top-level)", http.MethodPost, "/api/v1/teams", "team.create"}, + {"org team upsert", http.MethodPost, "/api/v1/orgs/acme/teams", "team.create"}, + {"org team update", http.MethodPut, "/api/v1/orgs/acme/teams/7", "team.update"}, + {"org team delete", http.MethodDelete, "/api/v1/orgs/acme/teams/7", "team.delete"}, {"warehouse pinning patch", http.MethodPatch, "/api/v1/orgs/acme/warehouse/pinning", "warehouse.update"}, {"org create", http.MethodPost, "/api/v1/orgs", "org.create"}, {"org update", http.MethodPut, "/api/v1/orgs/acme", "org.update"}, diff --git a/controlplane/admin/models_api.go b/controlplane/admin/models_api.go index a6ad2f65..71fc8a6e 100644 --- a/controlplane/admin/models_api.go +++ b/controlplane/admin/models_api.go @@ -90,6 +90,7 @@ func modelDescriptors() []modelDescriptor { } return []modelDescriptor{ mk("orgs", "Orgs", modelGroupTenants, false, configstore.Org{}), + mk("org-teams", "Org Teams", modelGroupTenants, false, configstore.OrgTeam{}), mk("org-users", "Org Users", modelGroupTenants, false, configstore.OrgUser{}), mk("org-user-secrets", "Org User Secrets", modelGroupTenants, false, configstore.OrgUserSecret{}), mk("managed-warehouses", "Managed Warehouses", modelGroupTenants, false, configstore.ManagedWarehouse{}), diff --git a/controlplane/admin/models_api_test.go b/controlplane/admin/models_api_test.go index f73b4594..8fbb30b0 100644 --- a/controlplane/admin/models_api_test.go +++ b/controlplane/admin/models_api_test.go @@ -25,7 +25,7 @@ func TestModelDescriptorsRedaction(t *testing.T) { // Registry must cover exactly the persisted models we expect. wantKeys := map[string]bool{ - "orgs": true, "org-users": true, "org-user-secrets": true, "managed-warehouses": true, + "orgs": true, "org-teams": true, "org-users": true, "org-user-secrets": true, "managed-warehouses": true, "cp-instances": true, "worker-records": true, "flight-session-records": true, "org-connection-queue": true, "org-connection-leases": true, "operators": true, @@ -155,7 +155,10 @@ func TestModelsAPIPostgres(t *testing.T) { // list. Tables with no typed descriptor (goose bookkeeping, migrations // added later) must show up under "Other" with information_schema columns, // and credential-shaped column VALUES must come back redacted. + // IF-NOT-EXISTS-proof: the shared schema persists across local runs, so a + // previous run's probe table must not fail the seed. if err := store.DB().Exec(` + DROP TABLE IF EXISTS explorer_probe; CREATE TABLE explorer_probe ( id BIGSERIAL PRIMARY KEY, note TEXT NOT NULL, diff --git a/controlplane/admin/ui/src/App.tsx b/controlplane/admin/ui/src/App.tsx index 36ed261c..490594e3 100644 --- a/controlplane/admin/ui/src/App.tsx +++ b/controlplane/admin/ui/src/App.tsx @@ -6,6 +6,7 @@ import { NotFound } from "@/pages/NotFound"; import { Overview } from "@/pages/Overview"; import { Orgs } from "@/pages/Orgs"; import { OrgDetail } from "@/pages/OrgDetail"; +import { OrgTeams } from "@/pages/OrgTeams"; import { ReshardForm } from "@/pages/ReshardForm"; import { Reshards } from "@/pages/Reshards"; import { ReshardOperation } from "@/pages/ReshardOperation"; @@ -35,6 +36,7 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/controlplane/admin/ui/src/components/OrgTeamDialogs.tsx b/controlplane/admin/ui/src/components/OrgTeamDialogs.tsx new file mode 100644 index 00000000..56bb88ab --- /dev/null +++ b/controlplane/admin/ui/src/components/OrgTeamDialogs.tsx @@ -0,0 +1,360 @@ +// Create / edit / delete dialogs for org teams (duckgres_org_teams rows), +// shared by the Org teams page and the org detail page. The dialogs encode the +// surface rules: schema_name is set once at create time and immutable +// afterwards; billing is repointed (never cleared) and carries the org's +// buffered usage with it; the org's last team cannot be deleted. +import { useState } from "react"; +import { AlertTriangle } from "lucide-react"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; +import { Switch } from "@/components/ui/switch"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { useCreateOrgTeam, useDeleteOrgTeam, useUpdateOrgTeam } from "@/hooks/useApi"; +import type { OrgTeam, OrgTeamUpdateBody } from "@/types/api"; + +// Tri-state backfill select value ↔ wire value. +type BackfillChoice = "unset" | "on" | "off"; + +function backfillChoice(v: boolean | null | undefined): BackfillChoice { + if (v == null) return "unset"; + return v ? "on" : "off"; +} + +export function BackfillBadge({ value }: { value: boolean | null | undefined }) { + if (value == null) return —; + return value ? on : off; +} + +function errMsg(e: unknown): string { + return e instanceof Error ? e.message : String(e); +} + +function FieldRow({ label, children }: { label: string; children: React.ReactNode }) { + return ( +
+ + {children} +
+ ); +} + +// CreateTeamDialog: org is either fixed (org detail page) or picked from +// `orgs` (the global Org teams page). +export function CreateTeamDialog({ + open, + onClose, + org, + orgs, +}: { + open: boolean; + onClose: () => void; + org?: string; + orgs?: string[]; +}) { + const create = useCreateOrgTeam(); + const [orgChoice, setOrgChoice] = useState(org ?? ""); + const [teamId, setTeamId] = useState(""); + const [schemaName, setSchemaName] = useState(""); + const [enabled, setEnabled] = useState(true); + const [backfill, setBackfill] = useState("unset"); + const [error, setError] = useState(null); + + const targetOrg = org ?? orgChoice; + const teamIdOk = /^\d+$/.test(teamId.trim()) && Number(teamId.trim()) > 0; + + const close = () => { + setOrgChoice(org ?? ""); + setTeamId(""); + setSchemaName(""); + setEnabled(true); + setBackfill("unset"); + setError(null); + onClose(); + }; + + const submit = async () => { + setError(null); + try { + await create.mutateAsync({ + org_id: targetOrg, + team_id: Number(teamId.trim()), + schema_name: schemaName.trim(), + enabled, + ...(backfill === "unset" ? {} : { backfill_enabled: backfill === "on" }), + }); + close(); + } catch (e) { + setError(errMsg(e)); + } + }; + + return ( + !o && close()}> + + + Add team + + Map a PostHog team to {org ? {org} : "an org"} and the + warehouse schema its data lives in. + + +
+ {!org && ( + + + + )} + + setTeamId(e.target.value)} + placeholder="PostHog team id, e.g. 12345" + className="font-mono text-xs" + /> + + + setSchemaName(e.target.value)} + placeholder="e.g. team_12345" + className="font-mono text-xs" + /> + +

+ + + The schema name cannot be changed later — it is where the team's warehouse data lives. + Lowercase letters, digits and underscores only. + +

+
+ + +
+ + + + {error &&

{error}

} +
+ + + + + + + ); +} + +// EditTeamDialog: enabled / backfill / billing repoint. The schema name is +// shown read-only; billing can only be pointed at this team, never cleared. +export function EditTeamDialog({ team, onClose }: { team: OrgTeam; onClose: () => void }) { + const update = useUpdateOrgTeam(); + const isBilling = Boolean(team.is_billing_team); + const [enabled, setEnabled] = useState(team.enabled); + const [backfill, setBackfill] = useState(backfillChoice(team.backfill_enabled)); + const [makeBilling, setMakeBilling] = useState(false); + const [error, setError] = useState(null); + + const submit = async () => { + setError(null); + const body: OrgTeamUpdateBody = {}; + if (enabled !== team.enabled) body.enabled = enabled; + if (backfill !== backfillChoice(team.backfill_enabled)) { + body.backfill_enabled = backfill === "unset" ? null : backfill === "on"; + } + if (makeBilling && !isBilling) body.is_billing_team = true; + if (Object.keys(body).length === 0) { + onClose(); + return; + } + try { + await update.mutateAsync({ org: team.org_id, teamId: team.team_id, body }); + onClose(); + } catch (e) { + setError(errMsg(e)); + } + }; + + return ( + !o && onClose()}> + + + + Edit team {team.team_id} in "{team.org_id}" + + + Schema {team.schema_name} (immutable). + + +
+
+ + +
+ + + + {isBilling ? ( +

+ This is the org's billing team. To change it, mark another team as billing. +

+ ) : ( +
+
+ + +
+ {makeBilling && ( +

+ + + Repoints the org's billing to this team and moves its buffered usage buckets + along with it. + +

+ )} +
+ )} + {error &&

{error}

} +
+ + + + + + + ); +} + +// DeleteTeamDialog: config-only delete with the two rules surfaced up front — +// last-team deletion is refused (delete the org instead), and deleting the +// billing team hands billing to the oldest remaining team. +export function DeleteTeamDialog({ + team, + teamCount, + onClose, +}: { + team: OrgTeam; + // How many teams the org currently has, so the last-team refusal shows + // before the request instead of as a raw 409. + teamCount: number; + onClose: () => void; +}) { + const del = useDeleteOrgTeam(); + const [error, setError] = useState(null); + const isLast = teamCount <= 1; + const isBilling = Boolean(team.is_billing_team); + + const submit = async () => { + setError(null); + try { + await del.mutateAsync({ org: team.org_id, teamId: team.team_id }); + onClose(); + } catch (e) { + setError(errMsg(e)); + } + }; + + return ( + !o && onClose()}> + + + + Delete team {team.team_id} from "{team.org_id}"? + + + Removes only the team's config row. The warehouse schema{" "} + {team.schema_name} and its data are not touched. + + +
+ {isLast ? ( +

+ + + This is the org's last team and cannot be deleted. The only way to remove it is + deleting the org. + +

+ ) : ( + isBilling && ( +

+ + + This is the billing team. The org's oldest remaining team automatically becomes + billing, and buffered usage buckets move to it. + +

+ ) + )} + {error &&

{error}

} +
+ + + + + + + ); +} diff --git a/controlplane/admin/ui/src/components/nav.ts b/controlplane/admin/ui/src/components/nav.ts index a07979a5..334b178e 100644 --- a/controlplane/admin/ui/src/components/nav.ts +++ b/controlplane/admin/ui/src/components/nav.ts @@ -4,6 +4,7 @@ import { ArrowLeftRight, Building2, Database, + Layers, LayoutDashboard, LineChart, Network, @@ -26,6 +27,7 @@ export interface NavItem { export const NAV: NavItem[] = [ { to: "/", label: "Overview", icon: LayoutDashboard, end: true }, { to: "/orgs", label: "Organizations", icon: Building2 }, + { to: "/org-teams", label: "Org teams", icon: Layers }, { to: "/users", label: "Org Users", icon: Users }, { to: "/operators", label: "Operators", icon: ShieldCheck, adminOnly: true }, { to: "/live", label: "Live", icon: Activity }, diff --git a/controlplane/admin/ui/src/hooks/useApi.ts b/controlplane/admin/ui/src/hooks/useApi.ts index 7b1b9587..de37b638 100644 --- a/controlplane/admin/ui/src/hooks/useApi.ts +++ b/controlplane/admin/ui/src/hooks/useApi.ts @@ -32,6 +32,9 @@ import type { ModelSummary, Operator, Org, + OrgTeam, + OrgTeamCreateBody, + OrgTeamUpdateBody, OrgUpdate, OrgUser, OrgUserSecret, @@ -171,6 +174,59 @@ export function useDeprovisionWarehouse(id: string) { }); } +// ---- org teams ---- + +// All teams across every org (the Org teams nav page). 404-tolerant so a +// pre-rollout backend renders an empty state instead of an error. +export function useAllOrgTeams() { + return useQuery({ + queryKey: ["org-teams"], + queryFn: () => tolerate404([])(api.listAllOrgTeams()), + }); +} + +export function useOrgTeams(org: string | undefined) { + return useQuery({ + queryKey: ["org-teams", org], + queryFn: () => tolerate404([])(api.listOrgTeams(org!)), + enabled: !!org, + }); +} + +// Team mutations invalidate the global + per-org team lists, the org queries +// (the billing team surfaces there as default_team_id) and the models sidebar +// counts. +function invalidateOrgTeams(qc: ReturnType) { + qc.invalidateQueries({ queryKey: ["org-teams"] }); + qc.invalidateQueries({ queryKey: ["orgs"] }); + qc.invalidateQueries({ queryKey: ["models"] }); +} + +export function useCreateOrgTeam() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (body: OrgTeamCreateBody) => api.createOrgTeam(body), + onSuccess: () => invalidateOrgTeams(qc), + }); +} + +export function useUpdateOrgTeam() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (v: { org: string; teamId: number; body: OrgTeamUpdateBody }) => + api.updateOrgTeam(v.org, v.teamId, v.body), + onSuccess: () => invalidateOrgTeams(qc), + }); +} + +export function useDeleteOrgTeam() { + const qc = useQueryClient(); + return useMutation({ + mutationFn: (v: { org: string; teamId: number }) => api.deleteOrgTeam(v.org, v.teamId), + onSuccess: () => invalidateOrgTeams(qc), + }); +} + // ---- ducklings (admin-only) ---- const EMPTY_DRIFT: DucklingDriftResponse = { available: false, checked: 0, entries: [] }; diff --git a/controlplane/admin/ui/src/lib/api.ts b/controlplane/admin/ui/src/lib/api.ts index e2cbdf93..1f69b5a1 100644 --- a/controlplane/admin/ui/src/lib/api.ts +++ b/controlplane/admin/ui/src/lib/api.ts @@ -24,6 +24,10 @@ import type { ModelSummary, Operator, Org, + OrgTeam, + OrgTeamCreateBody, + OrgTeamDeleteResult, + OrgTeamUpdateBody, OrgUpdate, OrgUser, OrgUserSecret, @@ -138,6 +142,17 @@ export const api = { deprovisionWarehouse: (id: string) => post<{ status: string; org: string }>(`/orgs/${enc(id)}/deprovision`, {}), + // org teams (duckgres_org_teams). Schema names are immutable after create + // on this surface; delete removes config only (never warehouse data). + listAllOrgTeams: () => get<{ teams: OrgTeam[] }>("/teams").then((r) => r.teams ?? []), + listOrgTeams: (org: string) => + get<{ teams: OrgTeam[] }>(`/orgs/${enc(org)}/teams`).then((r) => r.teams ?? []), + createOrgTeam: (body: OrgTeamCreateBody) => post("/teams", body), + updateOrgTeam: (org: string, teamId: number, body: OrgTeamUpdateBody) => + put(`/orgs/${enc(org)}/teams/${teamId}`, body), + deleteOrgTeam: (org: string, teamId: number) => + del(`/orgs/${enc(org)}/teams/${teamId}`), + // ducklings (admin-only) getDucklingDrift: () => get("/ducklings/drift"), getDucklingsMetadata: () => get("/ducklings/metadata"), diff --git a/controlplane/admin/ui/src/lib/audit.ts b/controlplane/admin/ui/src/lib/audit.ts index 7f427f9f..bd873dda 100644 --- a/controlplane/admin/ui/src/lib/audit.ts +++ b/controlplane/admin/ui/src/lib/audit.ts @@ -20,6 +20,10 @@ const ACTION_LABELS: Record = { "warehouse.update": "Updated warehouse config", "warehouse.delete": "Deleted warehouse config", + "team.create": "Created org team", + "team.update": "Updated org team", + "team.delete": "Deleted org team", + "user.create": "Created user", "user.update": "Updated user", "user.delete": "Deleted user", diff --git a/controlplane/admin/ui/src/pages/OrgDetail.tsx b/controlplane/admin/ui/src/pages/OrgDetail.tsx index c43c9d93..89a5ca79 100644 --- a/controlplane/admin/ui/src/pages/OrgDetail.tsx +++ b/controlplane/admin/ui/src/pages/OrgDetail.tsx @@ -1,6 +1,6 @@ import { useEffect, useState } from "react"; import { Link, useNavigate, useParams } from "react-router-dom"; -import { AlertTriangle, ArrowLeft, Database, Save, Trash2, Warehouse } from "lucide-react"; +import { AlertTriangle, ArrowLeft, Database, Layers, Pencil, Plus, Save, Trash2, Warehouse } from "lucide-react"; import { PageBody, PageHeader } from "@/components/AppShell"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; import { Button } from "@/components/ui/button"; @@ -29,11 +29,19 @@ import { useDucklingsMetadata, useOrg, useOrgReshards, + useOrgTeams, useUpdateOrg, useUpdateWarehouse, useWarehouse, } from "@/hooks/useApi"; -import type { ManagedWarehouse, OrgUpdate } from "@/types/api"; +import { + BackfillBadge, + CreateTeamDialog, + DeleteTeamDialog, + EditTeamDialog, +} from "@/components/OrgTeamDialogs"; +import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table"; +import type { ManagedWarehouse, OrgTeam, OrgUpdate } from "@/types/api"; interface FormState { max_workers: string; @@ -250,12 +258,15 @@ export function OrgDetail() { onChange={(e) => set("hostname_alias", e.target.value)} /> - + set("default_team_id", e.target.value)} /> +

+ Repoints the org's billing team. All of the org's teams are listed below. +

@@ -275,6 +286,7 @@ export function OrgDetail() {
+ (open ? setConfirmDelete(true) : closeDelete())}> @@ -664,6 +676,115 @@ function ReshardHistory({ orgId }: { orgId: string }) { ); } +// OrgTeamsCard lists the org's duckgres_org_teams rows with full CRUD. +// The billing team is also editable as "default_team_id" in the config form +// above (the legacy wire field) — both paths repoint the same row. +function OrgTeamsCard({ orgId }: { orgId: string }) { + const teams = useOrgTeams(orgId); + const [creating, setCreating] = useState(false); + const [editing, setEditing] = useState(null); + const [deleting, setDeleting] = useState(null); + const rows = teams.data ?? []; + + return ( + + + + Teams + + + + + + + {teams.isLoading ? ( + + ) : teams.isError ? ( + teams.refetch()} /> + ) : rows.length === 0 ? ( +

+ No PostHog teams are mapped to this org. +

+ ) : ( + + + + Team id + Schema + Enabled + Billing + Backfill + Created + Actions + + + + {rows.map((t) => ( + + {t.team_id} + {t.schema_name} + + {t.enabled ? ( + enabled + ) : ( + disabled + )} + + + {t.is_billing_team ? ( + billing + ) : ( + — + )} + + + + + {fmtTime(t.created_at)} + +
+ + + + + + +
+ + + ))} + +
+ )} + + + setCreating(false)} org={orgId} /> + {editing && setEditing(null)} />} + {deleting && ( + setDeleting(null)} /> + )} + + ); +} + function ReadRow({ label, value }: { label: string; value: unknown }) { return (
diff --git a/controlplane/admin/ui/src/pages/OrgTeams.test.tsx b/controlplane/admin/ui/src/pages/OrgTeams.test.tsx new file mode 100644 index 00000000..ec991c99 --- /dev/null +++ b/controlplane/admin/ui/src/pages/OrgTeams.test.tsx @@ -0,0 +1,114 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import { render, screen } from "@testing-library/react"; +import userEvent from "@testing-library/user-event"; +import { MemoryRouter } from "react-router-dom"; +import type { OrgTeam } from "@/types/api"; + +// Mock the data + identity hooks so the page renders with a controlled team +// list and role, asserting the table columns and the delete-dialog rules. +const hooks = vi.hoisted(() => ({ + useAllOrgTeams: vi.fn(), + useOrgs: vi.fn(), + useCreateOrgTeam: vi.fn(), + useUpdateOrgTeam: vi.fn(), + useDeleteOrgTeam: vi.fn(), +})); +vi.mock("@/hooks/useApi", () => hooks); + +const identity = vi.hoisted(() => ({ useIdentity: vi.fn() })); +vi.mock("@/components/IdentityProvider", () => identity); + +import { OrgTeams } from "./OrgTeams"; + +const ok = (data: T) => ({ data, isSuccess: true, isLoading: false, isError: false, refetch: vi.fn() }); +const mut = () => ({ mutateAsync: vi.fn(), isPending: false }); + +const TEAMS: OrgTeam[] = [ + { + org_id: "acme", + team_id: 1, + schema_name: "team_1", + enabled: true, + is_billing_team: true, + backfill_enabled: null, + created_at: "2026-07-01T00:00:00Z", + updated_at: "2026-07-01T00:00:00Z", + }, + { + org_id: "acme", + team_id: 2, + schema_name: "team_2", + enabled: false, + is_billing_team: null, + backfill_enabled: true, + created_at: "2026-07-02T00:00:00Z", + updated_at: "2026-07-02T00:00:00Z", + }, + { + org_id: "solo", + team_id: 7, + schema_name: "custom_schema", + enabled: true, + is_billing_team: true, + backfill_enabled: false, + created_at: "2026-07-03T00:00:00Z", + updated_at: "2026-07-03T00:00:00Z", + }, +]; + +function renderPage() { + render( + + + , + ); +} + +describe("Org teams page", () => { + beforeEach(() => { + vi.clearAllMocks(); + identity.useIdentity.mockReturnValue({ isAdmin: true, me: { email: "a@posthog.com", role: "admin", source: "sso" } }); + hooks.useAllOrgTeams.mockReturnValue(ok(TEAMS)); + hooks.useOrgs.mockReturnValue(ok([{ name: "acme" }, { name: "solo" }])); + hooks.useCreateOrgTeam.mockReturnValue(mut()); + hooks.useUpdateOrgTeam.mockReturnValue(mut()); + hooks.useDeleteOrgTeam.mockReturnValue(mut()); + }); + + it("lists every team with schema, enabled, billing and backfill state", () => { + renderPage(); + + expect(screen.getAllByText("acme")).toHaveLength(2); + expect(screen.getByText("solo")).toBeInTheDocument(); + expect(screen.getByText("team_1")).toBeInTheDocument(); + expect(screen.getByText("team_2")).toBeInTheDocument(); + expect(screen.getByText("custom_schema")).toBeInTheDocument(); + // Both billing rows carry the badge; the disabled team is flagged. + expect(screen.getAllByText("billing")).toHaveLength(2); + expect(screen.getByText("disabled")).toBeInTheDocument(); + // Admin affordance present. + expect(screen.getByRole("button", { name: /add team/i })).toBeInTheDocument(); + }); + + it("refuses deleting an org's last team up front", async () => { + renderPage(); + + // "solo" has exactly one team → the delete dialog must refuse. + const deleteButtons = screen.getAllByTitle("Delete"); + await userEvent.click(deleteButtons[2]); + + expect(screen.getByText(/last team and cannot be deleted/i)).toBeInTheDocument(); + expect(screen.getByRole("button", { name: /delete team/i })).toBeDisabled(); + }); + + it("warns that deleting a billing team hands billing to the oldest remaining team", async () => { + renderPage(); + + // acme team 1 is billing and acme has two teams → warning, not refusal. + const deleteButtons = screen.getAllByTitle("Delete"); + await userEvent.click(deleteButtons[0]); + + expect(screen.getByText(/oldest remaining team automatically becomes billing/i)).toBeInTheDocument(); + expect(screen.getByRole("button", { name: /delete team/i })).toBeEnabled(); + }); +}); diff --git a/controlplane/admin/ui/src/pages/OrgTeams.tsx b/controlplane/admin/ui/src/pages/OrgTeams.tsx new file mode 100644 index 00000000..47300986 --- /dev/null +++ b/controlplane/admin/ui/src/pages/OrgTeams.tsx @@ -0,0 +1,194 @@ +import { useMemo, useState } from "react"; +import { type ColumnDef } from "@tanstack/react-table"; +import { Layers, Pencil, Plus, Search, Trash2 } from "lucide-react"; +import { Link } from "react-router-dom"; +import { PageBody, PageHeader } from "@/components/AppShell"; +import { DataTable } from "@/components/DataTable"; +import { Card } from "@/components/ui/card"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Badge } from "@/components/ui/badge"; +import { AdminGate } from "@/components/AdminOnly"; +import { + BackfillBadge, + CreateTeamDialog, + DeleteTeamDialog, + EditTeamDialog, +} from "@/components/OrgTeamDialogs"; +import { EmptyState, ErrorState, TableSkeleton } from "@/components/states"; +import { useAllOrgTeams, useOrgs } from "@/hooks/useApi"; +import { fmtTime } from "@/lib/format"; +import type { OrgTeam } from "@/types/api"; + +export function OrgTeams() { + const teams = useAllOrgTeams(); + const orgs = useOrgs(); + const [filter, setFilter] = useState(""); + const [creating, setCreating] = useState(false); + const [editing, setEditing] = useState(null); + const [deleting, setDeleting] = useState(null); + + // Per-org row counts feed the delete dialog's last-team refusal. + const countByOrg = useMemo(() => { + const m = new Map(); + for (const t of teams.data ?? []) { + m.set(t.org_id, (m.get(t.org_id) ?? 0) + 1); + } + return m; + }, [teams.data]); + + const columns = useMemo[]>( + () => [ + { + accessorKey: "org_id", + header: "Org", + cell: ({ getValue }) => { + const org = String(getValue()); + return ( + e.stopPropagation()} + > + {org} + + ); + }, + }, + { + accessorKey: "team_id", + header: "Team id", + cell: ({ getValue }) => {String(getValue())}, + }, + { + accessorKey: "schema_name", + header: "Schema", + cell: ({ getValue }) => {String(getValue())}, + }, + { + accessorKey: "enabled", + header: "Enabled", + cell: ({ getValue }) => + getValue() ? enabled : disabled, + }, + { + id: "billing", + header: "Billing", + accessorFn: (t) => Boolean(t.is_billing_team), + cell: ({ getValue }) => + getValue() ? billing : —, + }, + { + id: "backfill", + header: "Backfill", + accessorFn: (t) => t.backfill_enabled ?? null, + cell: ({ row }) => , + }, + { + accessorKey: "created_at", + header: "Created", + cell: ({ getValue }) => {fmtTime(getValue() as string)}, + }, + { + id: "actions", + header: "", + enableSorting: false, + cell: ({ row }) => ( +
+ + + + + + +
+ ), + }, + ], + [], + ); + + return ( + <> + +
+ + setFilter(e.target.value)} + placeholder="Filter teams…" + className="w-64 pl-8" + /> +
+ + + + + } + /> + + + {teams.isLoading ? ( + + ) : teams.isError ? ( + teams.refetch()} /> + ) : ( + } + title="No org teams" + description="No PostHog teams are mapped to any org yet." + /> + } + /> + )} + + + + setCreating(false)} + orgs={(orgs.data ?? []).map((o) => o.name)} + /> + {editing && setEditing(null)} />} + {deleting && ( + setDeleting(null)} + /> + )} + + ); +} diff --git a/controlplane/admin/ui/src/types/api.ts b/controlplane/admin/ui/src/types/api.ts index 2ac6e0b5..eba4a071 100644 --- a/controlplane/admin/ui/src/types/api.ts +++ b/controlplane/admin/ui/src/types/api.ts @@ -78,7 +78,10 @@ export interface OrgUpdate { // One duckgres_org_teams row: a PostHog team mapped to this org and the // warehouse schema its data lives in. At most one row per org is the billing -// team. +// team. The *_name fields are legacy overrides for grandfathered pre-existing +// teams; null means "derive from schema_name" (events at +// .events, persons at .persons, data imports under +// _data_imports). export interface OrgTeam { org_id: string; team_id: number; @@ -86,10 +89,43 @@ export interface OrgTeam { enabled: boolean; is_billing_team?: boolean | null; backfill_enabled?: boolean | null; + events_table_name?: string | null; + persons_table_name?: string | null; + schema_data_imports_name?: string | null; created_at: string; updated_at: string; } +// POST /api/v1/teams — admin create (org_id in the body, mirroring +// POST /users). schema_name is set once here and immutable afterwards on the +// admin surface. +export interface OrgTeamCreateBody { + org_id: string; + team_id: number; + schema_name: string; + enabled?: boolean; + backfill_enabled?: boolean; +} + +// PUT /api/v1/orgs/:id/teams/:team_id — enabled / backfill / billing repoint +// only (never schema_name). is_billing_team may only be true: billing is +// repointed by marking another team, never cleared. An explicit +// `backfill_enabled: null` clears the tri-state back to unset. +export interface OrgTeamUpdateBody { + enabled?: boolean; + backfill_enabled?: boolean | null; + is_billing_team?: true; +} + +// DELETE /api/v1/orgs/:id/teams/:team_id response. new_billing_team_id is set +// when the deleted row was the billing team — the remaining team with the +// oldest created_at took over (usage buckets re-attributed server-side). +export interface OrgTeamDeleteResult { + deleted: number; + org: string; + new_billing_team_id?: number; +} + // ---- Users (confirmed) ---- export interface OrgUser { diff --git a/controlplane/configstore/migrations/000025_org_teams_table_names.sql b/controlplane/configstore/migrations/000025_org_teams_table_names.sql new file mode 100644 index 00000000..d9708b08 --- /dev/null +++ b/controlplane/configstore/migrations/000025_org_teams_table_names.sql @@ -0,0 +1,25 @@ +-- +goose Up + +-- Per-team legacy table-name overrides. NULL (the default, and the value for +-- every row created going forward) means "derive from schema_name": events +-- live at .events, persons at .persons, and data +-- imports under the _data_imports schema. Non-NULL values are +-- grandfathered explicit names for pre-existing teams whose warehouse tables +-- predate the schema-per-team convention — the PostHog-side backfill sets +-- them via the provisioning team upsert. +ALTER TABLE duckgres_org_teams ADD COLUMN IF NOT EXISTS events_table_name VARCHAR(255); +ALTER TABLE duckgres_org_teams ADD COLUMN IF NOT EXISTS persons_table_name VARCHAR(255); +ALTER TABLE duckgres_org_teams ADD COLUMN IF NOT EXISTS schema_data_imports_name VARCHAR(255); + +-- Two teams in one org must never share a schema name: the schema is where a +-- team's warehouse data lives, so a collision would silently interleave two +-- teams' tables. Safe on existing data — migration 000024 backfilled at most +-- one row per org ('team_', keyed by the unique default_team_id). +CREATE UNIQUE INDEX IF NOT EXISTS idx_duckgres_org_teams_org_schema ON duckgres_org_teams (org_id, schema_name); + +-- +goose Down + +DROP INDEX IF EXISTS idx_duckgres_org_teams_org_schema; +ALTER TABLE duckgres_org_teams DROP COLUMN IF EXISTS schema_data_imports_name; +ALTER TABLE duckgres_org_teams DROP COLUMN IF EXISTS persons_table_name; +ALTER TABLE duckgres_org_teams DROP COLUMN IF EXISTS events_table_name; diff --git a/controlplane/configstore/models.go b/controlplane/configstore/models.go index b1af8e42..d58fd33e 100644 --- a/controlplane/configstore/models.go +++ b/controlplane/configstore/models.go @@ -60,19 +60,32 @@ func (o *Org) BillingTeamID() *int64 { // lives in (conventionally "team_"). An org can carry many teams; exactly // one may be the billing team — the team pull-based billing keys the org's // usage buckets by (enforced by a partial unique index, migration 000024). -// IsBillingTeam and BackfillEnabled are tri-state (*bool): NULL means "not the -// billing team" / "backfill preference unset". +// Two teams in one org must never share a schema name (unique index on +// (org_id, schema_name), migration 000025). IsBillingTeam and BackfillEnabled +// are tri-state (*bool): NULL means "not the billing team" / "backfill +// preference unset". +// +// The *TableName fields are legacy overrides. NULL — the value for every team +// created going forward — means "derive from schema_name": events live at +// .events, persons at .persons, and data imports +// under the _data_imports schema. Non-NULL values are +// grandfathered explicit names for pre-existing teams whose warehouse tables +// predate the schema-per-team convention; the PostHog-side backfill sets them +// via the provisioning team upsert. type OrgTeam struct { - OrgID string `gorm:"primaryKey;size:255;index:idx_duckgres_org_teams_billing_org,unique,where:is_billing_team IS TRUE" json:"org_id"` + OrgID string `gorm:"primaryKey;size:255;index:idx_duckgres_org_teams_billing_org,unique,where:is_billing_team IS TRUE;index:idx_duckgres_org_teams_org_schema,unique,priority:1" json:"org_id"` // autoIncrement:false — int fields in a composite primary key default to // auto-increment in gorm, which would make this a bigserial. - TeamID int64 `gorm:"primaryKey;autoIncrement:false" json:"team_id"` - SchemaName string `gorm:"size:255;not null" json:"schema_name"` - Enabled bool `gorm:"not null;default:true" json:"enabled"` - IsBillingTeam *bool `json:"is_billing_team,omitempty"` - BackfillEnabled *bool `json:"backfill_enabled,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + TeamID int64 `gorm:"primaryKey;autoIncrement:false" json:"team_id"` + SchemaName string `gorm:"size:255;not null;index:idx_duckgres_org_teams_org_schema,unique,priority:2" json:"schema_name"` + Enabled bool `gorm:"not null;default:true" json:"enabled"` + IsBillingTeam *bool `json:"is_billing_team,omitempty"` + BackfillEnabled *bool `json:"backfill_enabled,omitempty"` + EventsTableName *string `gorm:"size:255" json:"events_table_name,omitempty"` + PersonsTableName *string `gorm:"size:255" json:"persons_table_name,omitempty"` + SchemaDataImportsName *string `gorm:"size:255" json:"schema_data_imports_name,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } func (OrgTeam) TableName() string { return "duckgres_org_teams" } diff --git a/controlplane/configstore/org_teams.go b/controlplane/configstore/org_teams.go index d970d6ce..0db11b60 100644 --- a/controlplane/configstore/org_teams.go +++ b/controlplane/configstore/org_teams.go @@ -1,11 +1,58 @@ package configstore import ( + "errors" "fmt" + "regexp" "gorm.io/gorm" + "gorm.io/gorm/clause" ) +// ErrOrgTeamNotFound is returned by DeleteOrgTeamTx when the (org, team) row +// does not exist. HTTP handlers map it to 404. +var ErrOrgTeamNotFound = errors.New("org team not found") + +// ErrLastOrgTeam is returned by DeleteOrgTeamTx when the row is the org's LAST +// team. An org must always keep at least one team (the billing team is the +// billing bucket key); the only way to remove the last team is deleting the +// org itself. HTTP handlers map it to 409. +var ErrLastOrgTeam = errors.New("cannot delete the org's last team; deleting the org is the only way to remove it") + +// ErrOrgTeamSchemaConflict is returned by UpsertOrgTeamTx when the requested +// schema_name is already used by a DIFFERENT team in the same org (two teams +// must never share a schema — their warehouse tables would interleave). +// HTTP handlers map it to 409. +var ErrOrgTeamSchemaConflict = errors.New("schema_name is already used by another team in this org") + +// orgTeamSchemaNamePattern constrains team schema names to safe lowercase SQL +// identifiers, mirroring the DNS-label strictness of provisioning org IDs: +// these names are interpolated into warehouse DDL and object paths, so no +// quoting-dependent characters are allowed. +var orgTeamSchemaNamePattern = regexp.MustCompile(`^[a-z_][a-z0-9_]*$`) + +// maxOrgTeamSchemaNameLength caps schema names at the Postgres identifier +// limit (NAMEDATALEN-1). The derived "_data_imports" schema must +// also fit, but grandfathered pre-existing names take priority over reserving +// suffix headroom — the derived-name path only applies to new conventional +// names ("team_"), which are far below the cap. +const maxOrgTeamSchemaNameLength = 63 + +// ValidateOrgTeamSchemaName rejects team schema names that are not safe +// lowercase identifiers ([a-z0-9_], not starting with a digit, ≤63 chars). +func ValidateOrgTeamSchemaName(name string) error { + if name == "" { + return errors.New("schema_name is required") + } + if len(name) > maxOrgTeamSchemaNameLength { + return fmt.Errorf("schema_name must be at most %d characters", maxOrgTeamSchemaNameLength) + } + if !orgTeamSchemaNamePattern.MatchString(name) { + return errors.New("schema_name must be a lowercase identifier: [a-z0-9_], not starting with a digit") + } + return nil +} + // OrgBillingTeamIDTx returns the org's current billing team id (the // duckgres_org_teams row with is_billing_team = TRUE) inside the caller's // transaction, or 0 when the org has none. @@ -51,3 +98,220 @@ func SetOrgBillingTeamTx(tx *gorm.DB, orgID string, teamID int64) error { } return nil } + +// OrgTeamUpsert is the provisioning-API input for creating or overwriting one +// duckgres_org_teams row. Pointer fields are presence-aware: nil preserves the +// stored value on an existing row (and takes the documented default on +// insert); a non-nil pointer sets the field, with the empty string clearing a +// legacy table-name override back to NULL ("derive from schema_name"). +type OrgTeamUpsert struct { + TeamID int64 + SchemaName string + // Enabled: nil = TRUE on insert / preserve on update. + Enabled *bool + // BackfillEnabled: nil = NULL on insert / preserve on update. + BackfillEnabled *bool + // Legacy explicit table names for grandfathered pre-existing teams + // (NULL = derive from schema_name). nil = preserve; "" = clear to NULL. + EventsTableName *string + PersonsTableName *string + SchemaDataImportsName *string +} + +// UpsertOrgTeamTx creates or overwrites the (org, team) row inside the +// caller's transaction and returns the stored row. +// +// This is deliberately a full upsert, NOT an immutable-create: it is the +// PostHog-side grandfather path. The migration seeds every existing team with +// the conventional "team_" schema placeholder, and the PostHog backfill +// then replaces it (and sets the legacy explicit table names) through this +// call — so schema_name immutability CANNOT be enforced here. It IS enforced +// on the user-facing surfaces (the admin API update rejects schema changes). +// +// Returns gorm.ErrRecordNotFound when the org doesn't exist and +// ErrOrgTeamSchemaConflict when schema_name is already used by a different +// team in the same org (backed by the unique (org_id, schema_name) index, +// migration 000025 — a concurrent insert that slips past the pre-check still +// fails with a 23505 on that index). +func UpsertOrgTeamTx(tx *gorm.DB, orgID string, up OrgTeamUpsert) (*OrgTeam, error) { + if err := ValidateOrgTeamSchemaName(up.SchemaName); err != nil { + return nil, err + } + + var orgCount int64 + if err := tx.Model(&Org{}).Where("name = ?", orgID).Count(&orgCount).Error; err != nil { + return nil, fmt.Errorf("check org (org=%s): %w", orgID, err) + } + if orgCount == 0 { + return nil, gorm.ErrRecordNotFound + } + + var schemaClash int64 + if err := tx.Model(&OrgTeam{}). + Where("org_id = ? AND schema_name = ? AND team_id <> ?", orgID, up.SchemaName, up.TeamID). + Count(&schemaClash).Error; err != nil { + return nil, fmt.Errorf("check schema conflict (org=%s schema=%s): %w", orgID, up.SchemaName, err) + } + if schemaClash > 0 { + return nil, ErrOrgTeamSchemaConflict + } + + // Lock the existing row (if any) so the presence-aware merge below can't + // interleave with a concurrent upsert of the same team. + var existing OrgTeam + err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + First(&existing, "org_id = ? AND team_id = ?", orgID, up.TeamID).Error + switch { + case errors.Is(err, gorm.ErrRecordNotFound): + row := OrgTeam{ + OrgID: orgID, + TeamID: up.TeamID, + SchemaName: up.SchemaName, + Enabled: up.Enabled == nil || *up.Enabled, + BackfillEnabled: up.BackfillEnabled, + } + applyOrgTeamTableNames(&row, up) + if err := tx.Create(&row).Error; err != nil { + return nil, fmt.Errorf("create org team (org=%s team=%d): %w", orgID, up.TeamID, err) + } + return &row, nil + case err != nil: + return nil, fmt.Errorf("read org team (org=%s team=%d): %w", orgID, up.TeamID, err) + } + + existing.SchemaName = up.SchemaName + if up.Enabled != nil { + existing.Enabled = *up.Enabled + } + if up.BackfillEnabled != nil { + existing.BackfillEnabled = up.BackfillEnabled + } + applyOrgTeamTableNames(&existing, up) + // Save with explicit column selection: gorm's Save would skip NULLing the + // pointer fields via zero-value pruning on composite-PK updates. + if err := tx.Model(&OrgTeam{}). + Where("org_id = ? AND team_id = ?", orgID, up.TeamID). + Select("schema_name", "enabled", "backfill_enabled", + "events_table_name", "persons_table_name", "schema_data_imports_name", "updated_at"). + Updates(map[string]interface{}{ + "schema_name": existing.SchemaName, + "enabled": existing.Enabled, + "backfill_enabled": existing.BackfillEnabled, + "events_table_name": existing.EventsTableName, + "persons_table_name": existing.PersonsTableName, + "schema_data_imports_name": existing.SchemaDataImportsName, + "updated_at": gorm.Expr("now()"), + }).Error; err != nil { + return nil, fmt.Errorf("update org team (org=%s team=%d): %w", orgID, up.TeamID, err) + } + var stored OrgTeam + if err := tx.First(&stored, "org_id = ? AND team_id = ?", orgID, up.TeamID).Error; err != nil { + return nil, fmt.Errorf("reload org team (org=%s team=%d): %w", orgID, up.TeamID, err) + } + return &stored, nil +} + +// applyOrgTeamTableNames folds the presence-aware legacy table-name fields of +// an upsert into row: nil preserves, "" clears to NULL, a value sets. +func applyOrgTeamTableNames(row *OrgTeam, up OrgTeamUpsert) { + for _, f := range []struct { + src *string + dst **string + }{ + {up.EventsTableName, &row.EventsTableName}, + {up.PersonsTableName, &row.PersonsTableName}, + {up.SchemaDataImportsName, &row.SchemaDataImportsName}, + } { + if f.src == nil { + continue + } + if *f.src == "" { + *f.dst = nil + continue + } + v := *f.src + *f.dst = &v + } +} + +// OrgTeamDeleteResult reports what DeleteOrgTeamTx did beyond the delete +// itself, so callers can surface the billing handover. +type OrgTeamDeleteResult struct { + // WasBilling is true when the deleted row carried the billing mark. + WasBilling bool + // NewBillingTeamID is the remaining team that automatically became the + // billing team (the one with the OLDEST created_at). 0 when the deleted + // row was not the billing team. + NewBillingTeamID int64 + // UsageRowsMoved counts the buffered usage-bucket rows re-attributed to + // the new billing team. + UsageRowsMoved int64 +} + +// DeleteOrgTeamTx deletes the (org, team) CONFIG row inside the caller's +// transaction. It never touches warehouse data — the team's schema and tables +// stay untouched; only the mapping goes away. +// +// Rules, atomic with the delete: +// - The org's LAST team cannot be deleted (ErrLastOrgTeam) — deleting the +// org is the only way to remove it. Every org must keep a billing team. +// - Deleting the billing team automatically promotes the remaining team +// with the OLDEST created_at (ties broken by team_id) and re-attributes +// the org's buffered usage buckets to it (ReattributeUsageTeamTx), the +// same handover a billing repoint does. +// +// All of the org's team rows are locked up front so two concurrent deletes +// can't each see the other's row as "remaining" and empty the org. +func DeleteOrgTeamTx(tx *gorm.DB, orgID string, teamID int64) (*OrgTeamDeleteResult, error) { + var teams []OrgTeam + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("org_id = ?", orgID). + Order("created_at, team_id"). + Find(&teams).Error; err != nil { + return nil, fmt.Errorf("lock org teams (org=%s): %w", orgID, err) + } + + var target *OrgTeam + var successor *OrgTeam + for i := range teams { + if teams[i].TeamID == teamID { + target = &teams[i] + } else if successor == nil { + // First non-target in (created_at, team_id) order = the oldest + // remaining team. + successor = &teams[i] + } + } + if target == nil { + return nil, ErrOrgTeamNotFound + } + if successor == nil { + return nil, ErrLastOrgTeam + } + + if err := tx.Where("org_id = ? AND team_id = ?", orgID, teamID). + Delete(&OrgTeam{}).Error; err != nil { + return nil, fmt.Errorf("delete org team (org=%s team=%d): %w", orgID, teamID, err) + } + + res := &OrgTeamDeleteResult{ + WasBilling: target.IsBillingTeam != nil && *target.IsBillingTeam, + } + if !res.WasBilling { + return res, nil + } + + // The billing team is gone: the oldest remaining team takes over, and the + // org's buffered usage buckets follow it in the same transaction — exactly + // the handover a billing repoint performs. + if err := SetOrgBillingTeamTx(tx, orgID, successor.TeamID); err != nil { + return nil, err + } + moved, err := ReattributeUsageTeamTx(tx, orgID, successor.TeamID) + if err != nil { + return nil, err + } + res.NewBillingTeamID = successor.TeamID + res.UsageRowsMoved = moved + return res, nil +} diff --git a/controlplane/provisioning/api.go b/controlplane/provisioning/api.go index 5c504f07..a7913e1f 100644 --- a/controlplane/provisioning/api.go +++ b/controlplane/provisioning/api.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "regexp" + "strconv" "time" "github.com/gin-gonic/gin" @@ -82,6 +83,16 @@ type Store interface { UpdateOrgUserPassword(orgID, username, passwordHash string) error SetWarehouseDeleting(orgID string, expectedState configstore.ManagedWarehouseProvisioningState) error IsDatabaseNameAvailable(name string) (bool, error) + // Team CRUD for the PostHog backend (duckgres_org_teams rows — config + // only, never warehouse data). ListOrgTeams returns + // gorm.ErrRecordNotFound for an unknown org. UpsertOrgTeam is the + // grandfather path: it MAY overwrite schema_name and the legacy table + // names of an existing row (see configstore.UpsertOrgTeamTx). + // DeleteOrgTeam enforces the last-team refusal and the billing handover + // (see configstore.DeleteOrgTeamTx). + ListOrgTeams(orgID string) ([]configstore.OrgTeam, error) + UpsertOrgTeam(orgID string, up configstore.OrgTeamUpsert) (*configstore.OrgTeam, error) + DeleteOrgTeam(orgID string, teamID int64) (*configstore.OrgTeamDeleteResult, error) } // RegisterAPI registers provisioning endpoints on the given router group. @@ -95,6 +106,12 @@ func RegisterAPI(r *gin.RouterGroup, store Store, bucketSuffix string) { r.GET("/orgs/:id/warehouse/status", h.getWarehouseStatus) r.POST("/orgs/:id/reset-password", h.resetPassword) r.GET("/database-name/check", h.checkDatabaseName) + // Team CRUD: the PostHog backend manages the org's duckgres_org_teams + // rows through these (config only — deleting a team never touches + // warehouse data). + r.GET("/orgs/:id/teams", h.listOrgTeams) + r.POST("/orgs/:id/teams", h.upsertOrgTeam) + r.DELETE("/orgs/:id/teams/:team_id", h.deleteOrgTeam) } type handler struct { @@ -142,7 +159,15 @@ type provisionRequest struct { // its billing team from birth; pull-based compute billing keys usage // buckets by it). Optional on re-provision of an existing org: absent/0 // keeps the stored billing team, never wipes it. - DefaultTeamID int64 `json:"default_team_id,omitempty"` + DefaultTeamID int64 `json:"default_team_id,omitempty"` + // TeamID + SchemaName are the successor of DefaultTeamID: they create the + // org's first duckgres_org_teams row (billing = TRUE) with an explicit + // warehouse schema instead of the conventional "team_". They COEXIST + // with the legacy field (this ships before the PostHog-side change): when + // both team ids are given they must agree (400 otherwise), and + // schema_name requires a team id via either field. + TeamID int64 `json:"team_id,omitempty"` + SchemaName string `json:"schema_name,omitempty"` MetadataStore *provisionMetadataReq `json:"metadata_store,omitempty"` DataStore *provisionDataStoreReq `json:"data_store,omitempty"` DuckLake *provisionDuckLakeReq `json:"ducklake,omitempty"` @@ -226,6 +251,28 @@ func (h *handler) provisionWarehouse(c *gin.Context) { return } + // Resolve the legacy default_team_id and the new team_id/schema_name pair + // into one effective billing team. Both fields keep working — this ships + // before the PostHog-side switch — but disagreement is a caller bug. + if req.TeamID != 0 && req.DefaultTeamID != 0 && req.TeamID != req.DefaultTeamID { + c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("team_id (%d) and default_team_id (%d) disagree; send one, or the same value in both", req.TeamID, req.DefaultTeamID)}) + return + } + effectiveTeamID := req.DefaultTeamID + if req.TeamID != 0 { + effectiveTeamID = req.TeamID + } + if req.SchemaName != "" { + if effectiveTeamID == 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "schema_name requires team_id"}) + return + } + if err := configstore.ValidateOrgTeamSchemaName(req.SchemaName); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + } + // The catalog is decoupled from the metadata backend: a duckling runs // DuckLake on any of the metadata stores. It must be enabled (a warehouse // without a catalog has nothing to attach). @@ -315,7 +362,8 @@ func (h *handler) provisionWarehouse(c *gin.Context) { if err := h.store.Provision(ProvisionRequest{ OrgID: orgID, DatabaseName: req.DatabaseName, - DefaultTeamID: req.DefaultTeamID, + DefaultTeamID: effectiveTeamID, + SchemaName: req.SchemaName, Warehouse: warehouse, RootUserHash: hash, }); err != nil { @@ -501,3 +549,113 @@ func (h *handler) checkDatabaseName(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"name": name, "available": available}) } + +// --- Org teams (duckgres_org_teams CRUD for the PostHog backend) --- + +// orgTeamUpsertRequest is the POST /orgs/:id/teams body. Pointer fields are +// presence-aware: absent preserves the stored value on an existing row (and +// takes the documented default on a new one). The legacy *_name fields exist +// for grandfathering pre-existing teams whose warehouse tables predate the +// schema-per-team convention; NULL means "derive from schema_name". +type orgTeamUpsertRequest struct { + TeamID int64 `json:"team_id"` + SchemaName string `json:"schema_name"` + Enabled *bool `json:"enabled,omitempty"` + BackfillEnabled *bool `json:"backfill_enabled,omitempty"` + EventsTableName *string `json:"events_table_name,omitempty"` + PersonsTableName *string `json:"persons_table_name,omitempty"` + SchemaDataImportsName *string `json:"schema_data_imports_name,omitempty"` +} + +func (h *handler) listOrgTeams(c *gin.Context) { + orgID := c.Param("id") + teams, err := h.store.ListOrgTeams(orgID) + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + c.JSON(http.StatusNotFound, gin.H{"error": "org not found"}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"teams": teams}) +} + +// upsertOrgTeam creates or overwrites one (org, team) row. This endpoint IS +// the grandfather path: upserting an existing (org, team) MAY overwrite +// schema_name and the legacy table names, because the PostHog-side backfill +// needs to replace the migration's conventional "team_" placeholder with +// the team's real pre-existing names. Schema immutability is therefore NOT +// enforced here — it is enforced on the user-facing surfaces (the admin API +// update rejects schema changes). +func (h *handler) upsertOrgTeam(c *gin.Context) { + orgID := c.Param("id") + + var req orgTeamUpsertRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if req.TeamID <= 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "team_id is required (a positive PostHog team id)"}) + return + } + if err := configstore.ValidateOrgTeamSchemaName(req.SchemaName); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + team, err := h.store.UpsertOrgTeam(orgID, configstore.OrgTeamUpsert{ + TeamID: req.TeamID, + SchemaName: req.SchemaName, + Enabled: req.Enabled, + BackfillEnabled: req.BackfillEnabled, + EventsTableName: req.EventsTableName, + PersonsTableName: req.PersonsTableName, + SchemaDataImportsName: req.SchemaDataImportsName, + }) + if err != nil { + switch { + case errors.Is(err, gorm.ErrRecordNotFound): + c.JSON(http.StatusNotFound, gin.H{"error": "org not found"}) + case errors.Is(err, configstore.ErrOrgTeamSchemaConflict), isUniqueViolation(err): + // The pre-check catches the common case with a clean message; the + // unique (org_id, schema_name) index catches the concurrent one. + c.JSON(http.StatusConflict, gin.H{"error": configstore.ErrOrgTeamSchemaConflict.Error()}) + default: + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + } + return + } + c.JSON(http.StatusOK, team) +} + +// deleteOrgTeam removes one (org, team) CONFIG row — never warehouse data. +// Deleting the billing team promotes the oldest remaining team (usage buckets +// re-attributed atomically); deleting the org's last team is refused. +func (h *handler) deleteOrgTeam(c *gin.Context) { + orgID := c.Param("id") + teamID, err := strconv.ParseInt(c.Param("team_id"), 10, 64) + if err != nil || teamID <= 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "team_id must be a positive integer"}) + return + } + + res, err := h.store.DeleteOrgTeam(orgID, teamID) + if err != nil { + switch { + case errors.Is(err, configstore.ErrOrgTeamNotFound): + c.JSON(http.StatusNotFound, gin.H{"error": "org team not found"}) + case errors.Is(err, configstore.ErrLastOrgTeam): + c.JSON(http.StatusConflict, gin.H{"error": configstore.ErrLastOrgTeam.Error()}) + default: + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + } + return + } + resp := gin.H{"deleted": teamID, "org": orgID} + if res.WasBilling { + resp["new_billing_team_id"] = res.NewBillingTeamID + } + c.JSON(http.StatusOK, resp) +} diff --git a/controlplane/provisioning/api_test.go b/controlplane/provisioning/api_test.go index c2d63168..87f09b02 100644 --- a/controlplane/provisioning/api_test.go +++ b/controlplane/provisioning/api_test.go @@ -7,8 +7,10 @@ import ( "fmt" "net/http" "net/http/httptest" + "sort" "strings" "testing" + "time" "github.com/gin-gonic/gin" "github.com/posthog/duckgres/controlplane/configstore" @@ -19,7 +21,12 @@ type fakeStore struct { orgs map[string]*configstore.Org users map[configstore.OrgUserKey]string warehouses map[string]*configstore.ManagedWarehouse + teams map[string]map[int64]*configstore.OrgTeam provisionUserFailHook error // set non-nil to simulate user-step failure inside Provision + // lastProvision records the ProvisionRequest the handler dispatched, so + // tests can assert the request-field → store-request threading (the real + // billing-row/schema writes are covered by the Postgres-backed tests). + lastProvision *ProvisionRequest } func newFakeStore() *fakeStore { @@ -27,6 +34,7 @@ func newFakeStore() *fakeStore { orgs: make(map[string]*configstore.Org), users: make(map[configstore.OrgUserKey]string), warehouses: make(map[string]*configstore.ManagedWarehouse), + teams: make(map[string]map[int64]*configstore.OrgTeam), } } @@ -93,7 +101,118 @@ func (s *fakeStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // handler treats partial failure as a complete rollback. func (s *fakeStore) setProvisionUserFailHook(err error) { s.provisionUserFailHook = err } +// seedTeam plants one team row (nil billing/backfill unless set by the caller). +func (s *fakeStore) seedTeam(team configstore.OrgTeam) { + if s.teams[team.OrgID] == nil { + s.teams[team.OrgID] = make(map[int64]*configstore.OrgTeam) + } + clone := team + s.teams[team.OrgID][team.TeamID] = &clone +} + +func (s *fakeStore) ListOrgTeams(orgID string) ([]configstore.OrgTeam, error) { + if _, ok := s.orgs[orgID]; !ok { + return nil, gorm.ErrRecordNotFound + } + var out []configstore.OrgTeam + for _, t := range s.teams[orgID] { + out = append(out, *t) + } + sort.Slice(out, func(i, j int) bool { return out[i].TeamID < out[j].TeamID }) + return out, nil +} + +// UpsertOrgTeam mirrors configstore.UpsertOrgTeamTx's contract in memory: +// unknown org → gorm.ErrRecordNotFound, schema held by another team → +// ErrOrgTeamSchemaConflict, presence-aware merge otherwise. +func (s *fakeStore) UpsertOrgTeam(orgID string, up configstore.OrgTeamUpsert) (*configstore.OrgTeam, error) { + if _, ok := s.orgs[orgID]; !ok { + return nil, gorm.ErrRecordNotFound + } + for _, t := range s.teams[orgID] { + if t.SchemaName == up.SchemaName && t.TeamID != up.TeamID { + return nil, configstore.ErrOrgTeamSchemaConflict + } + } + if s.teams[orgID] == nil { + s.teams[orgID] = make(map[int64]*configstore.OrgTeam) + } + existing, ok := s.teams[orgID][up.TeamID] + if !ok { + row := &configstore.OrgTeam{ + OrgID: orgID, + TeamID: up.TeamID, + SchemaName: up.SchemaName, + Enabled: up.Enabled == nil || *up.Enabled, + BackfillEnabled: up.BackfillEnabled, + } + applyFakeTableNames(row, up) + s.teams[orgID][up.TeamID] = row + clone := *row + return &clone, nil + } + existing.SchemaName = up.SchemaName + if up.Enabled != nil { + existing.Enabled = *up.Enabled + } + if up.BackfillEnabled != nil { + existing.BackfillEnabled = up.BackfillEnabled + } + applyFakeTableNames(existing, up) + clone := *existing + return &clone, nil +} + +func applyFakeTableNames(row *configstore.OrgTeam, up configstore.OrgTeamUpsert) { + set := func(dst **string, src *string) { + if src == nil { + return + } + if *src == "" { + *dst = nil + return + } + v := *src + *dst = &v + } + set(&row.EventsTableName, up.EventsTableName) + set(&row.PersonsTableName, up.PersonsTableName) + set(&row.SchemaDataImportsName, up.SchemaDataImportsName) +} + +// DeleteOrgTeam mirrors configstore.DeleteOrgTeamTx's rules in memory: 404 for +// a missing row, refusal on the last team, billing handover to the remaining +// team with the oldest created_at. +func (s *fakeStore) DeleteOrgTeam(orgID string, teamID int64) (*configstore.OrgTeamDeleteResult, error) { + target, ok := s.teams[orgID][teamID] + if !ok { + return nil, configstore.ErrOrgTeamNotFound + } + if len(s.teams[orgID]) == 1 { + return nil, configstore.ErrLastOrgTeam + } + delete(s.teams[orgID], teamID) + res := &configstore.OrgTeamDeleteResult{ + WasBilling: target.IsBillingTeam != nil && *target.IsBillingTeam, + } + if res.WasBilling { + var successor *configstore.OrgTeam + for _, t := range s.teams[orgID] { + if successor == nil || t.CreatedAt.Before(successor.CreatedAt) || + (t.CreatedAt.Equal(successor.CreatedAt) && t.TeamID < successor.TeamID) { + successor = t + } + } + billing := true + successor.IsBillingTeam = &billing + res.NewBillingTeamID = successor.TeamID + } + return res, nil +} + func (s *fakeStore) Provision(req ProvisionRequest) error { + reqCopy := req + s.lastProvision = &reqCopy // Pre-check: warehouse already exists in non-terminal state? Mirrors // createPendingWarehouseTx so the handler's 409 branch is exercised. if existing, ok := s.warehouses[req.OrgID]; ok && @@ -901,3 +1020,303 @@ func TestProvisionRejectsStringDefaultTeamID(t *testing.T) { t.Fatal("org must not be created on a malformed default_team_id") } } + +// --- Org team CRUD (provisioning API) --- + +func doJSON(t *testing.T, router *gin.Engine, method, path string, body string) *httptest.ResponseRecorder { + t.Helper() + var reader *bytes.Reader + if body == "" { + reader = bytes.NewReader(nil) + } else { + reader = bytes.NewReader([]byte(body)) + } + req := httptest.NewRequest(method, path, reader) + if body != "" { + req.Header.Set("Content-Type", "application/json") + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec +} + +func TestOrgTeamUpsertValidation(t *testing.T) { + for name, tc := range map[string]struct { + body string + wantStatus int + wantSubstr string + }{ + "missing team_id": {`{"schema_name":"team_1"}`, http.StatusBadRequest, "team_id"}, + "negative team_id": {`{"team_id":-1,"schema_name":"team_1"}`, http.StatusBadRequest, "team_id"}, + "missing schema_name": {`{"team_id":1}`, http.StatusBadRequest, "schema_name"}, + "uppercase schema_name": {`{"team_id":1,"schema_name":"Team_1"}`, http.StatusBadRequest, "lowercase"}, + "hyphenated schema_name": {`{"team_id":1,"schema_name":"team-1"}`, http.StatusBadRequest, "lowercase"}, + "digit-led schema_name": {`{"team_id":1,"schema_name":"1team"}`, http.StatusBadRequest, "lowercase"}, + "overlong schema_name": {`{"team_id":1,"schema_name":"` + strings.Repeat("a", 64) + `"}`, http.StatusBadRequest, "63"}, + } { + t.Run(name, func(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + router := newTestRouter(store) + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/acme/teams", tc.body) + if rec.Code != tc.wantStatus { + t.Fatalf("status = %d, want %d: %s", rec.Code, tc.wantStatus, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), tc.wantSubstr) { + t.Fatalf("error should mention %q: %s", tc.wantSubstr, rec.Body.String()) + } + }) + } +} + +func TestOrgTeamUpsertCreatesAndLists(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + router := newTestRouter(store) + + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/acme/teams", + `{"team_id":7,"schema_name":"team_7","backfill_enabled":true}`) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + var team configstore.OrgTeam + if err := json.Unmarshal(rec.Body.Bytes(), &team); err != nil { + t.Fatalf("decode: %v", err) + } + if team.TeamID != 7 || team.SchemaName != "team_7" || !team.Enabled { + t.Fatalf("created team = %+v, want team 7, schema team_7, enabled default true", team) + } + if team.BackfillEnabled == nil || !*team.BackfillEnabled { + t.Fatalf("backfill_enabled = %v, want true", team.BackfillEnabled) + } + + rec = doJSON(t, router, http.MethodGet, "/api/v1/orgs/acme/teams", "") + if rec.Code != http.StatusOK { + t.Fatalf("list status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + var listing struct { + Teams []configstore.OrgTeam `json:"teams"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &listing); err != nil { + t.Fatalf("decode listing: %v", err) + } + if len(listing.Teams) != 1 || listing.Teams[0].TeamID != 7 { + t.Fatalf("listing = %+v, want the created team", listing.Teams) + } +} + +func TestOrgTeamListUnknownOrg404(t *testing.T) { + store := newFakeStore() + router := newTestRouter(store) + rec := doJSON(t, router, http.MethodGet, "/api/v1/orgs/nope/teams", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404: %s", rec.Code, rec.Body.String()) + } +} + +// TestOrgTeamUpsertGrandfathersExistingRow pins the grandfather contract: an +// upsert of an existing (org, team) MAY overwrite schema_name and the legacy +// table names — the PostHog backfill replaces the migration's "team_" +// placeholder through this exact path. +func TestOrgTeamUpsertGrandfathersExistingRow(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 7, SchemaName: "team_7", Enabled: true}) + router := newTestRouter(store) + + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/acme/teams", + `{"team_id":7,"schema_name":"legacy_wh","events_table_name":"legacy_events","persons_table_name":"legacy_persons","schema_data_imports_name":"legacy_imports"}`) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + stored := store.teams["acme"][7] + if stored.SchemaName != "legacy_wh" { + t.Fatalf("schema_name = %q, want overwritten legacy_wh", stored.SchemaName) + } + if stored.EventsTableName == nil || *stored.EventsTableName != "legacy_events" || + stored.PersonsTableName == nil || *stored.PersonsTableName != "legacy_persons" || + stored.SchemaDataImportsName == nil || *stored.SchemaDataImportsName != "legacy_imports" { + t.Fatalf("legacy table names not stored: %+v", stored) + } + if !stored.Enabled { + t.Fatal("omitted enabled must preserve the stored value on update") + } +} + +func TestOrgTeamUpsertDuplicateSchema409(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true}) + router := newTestRouter(store) + + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/acme/teams", + `{"team_id":2,"schema_name":"team_1"}`) + if rec.Code != http.StatusConflict { + t.Fatalf("status = %d, want 409: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "schema_name") { + t.Fatalf("error should name schema_name: %s", rec.Body.String()) + } +} + +func TestOrgTeamDeleteRules(t *testing.T) { + billing := true + base := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + + t.Run("last team refused", func(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true, IsBillingTeam: &billing}) + router := newTestRouter(store) + + rec := doJSON(t, router, http.MethodDelete, "/api/v1/orgs/acme/teams/1", "") + if rec.Code != http.StatusConflict { + t.Fatalf("status = %d, want 409: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "deleting the org") { + t.Fatalf("error should state that deleting the org is the only way: %s", rec.Body.String()) + } + if store.teams["acme"][1] == nil { + t.Fatal("last team must not be deleted") + } + }) + + t.Run("billing team delete promotes oldest remaining", func(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + b := billing + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true, IsBillingTeam: &b, CreatedAt: base}) + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 3, SchemaName: "team_3", Enabled: true, CreatedAt: base.Add(2 * time.Hour)}) + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 2, SchemaName: "team_2", Enabled: true, CreatedAt: base.Add(time.Hour)}) + router := newTestRouter(store) + + rec := doJSON(t, router, http.MethodDelete, "/api/v1/orgs/acme/teams/1", "") + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + var resp map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v", err) + } + // Team 2 has the oldest created_at among the remaining rows. + if resp["new_billing_team_id"] != float64(2) { + t.Fatalf("new_billing_team_id = %v, want 2", resp["new_billing_team_id"]) + } + if got := store.teams["acme"][2]; got.IsBillingTeam == nil || !*got.IsBillingTeam { + t.Fatalf("team 2 must carry the billing mark, got %+v", got) + } + }) + + t.Run("non-billing delete has no handover", func(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + b := billing + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 1, SchemaName: "team_1", Enabled: true, IsBillingTeam: &b, CreatedAt: base}) + store.seedTeam(configstore.OrgTeam{OrgID: "acme", TeamID: 2, SchemaName: "team_2", Enabled: true, CreatedAt: base.Add(time.Hour)}) + router := newTestRouter(store) + + rec := doJSON(t, router, http.MethodDelete, "/api/v1/orgs/acme/teams/2", "") + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + if strings.Contains(rec.Body.String(), "new_billing_team_id") { + t.Fatalf("non-billing delete must not report a handover: %s", rec.Body.String()) + } + }) + + t.Run("unknown team 404", func(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + router := newTestRouter(store) + rec := doJSON(t, router, http.MethodDelete, "/api/v1/orgs/acme/teams/9", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404: %s", rec.Code, rec.Body.String()) + } + }) + + t.Run("non-numeric team id 400", func(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + router := newTestRouter(store) + rec := doJSON(t, router, http.MethodDelete, "/api/v1/orgs/acme/teams/abc", "") + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body.String()) + } + }) +} + +// --- provision team_id/schema_name resolution --- + +func TestProvisionTeamIDAndSchemaName(t *testing.T) { + store := newFakeStore() + router := newTestRouter(store) + + body := `{"database_name":"d","team_id":42,"schema_name":"custom_wh","metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}` + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/neworg/provision", body) + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body.String()) + } + if store.lastProvision == nil { + t.Fatal("Provision not dispatched") + } + if store.lastProvision.DefaultTeamID != 42 || store.lastProvision.SchemaName != "custom_wh" { + t.Fatalf("provision request = %+v, want team 42 schema custom_wh", store.lastProvision) + } +} + +func TestProvisionTeamIDFieldsMustAgree(t *testing.T) { + store := newFakeStore() + router := newTestRouter(store) + + body := `{"database_name":"d","team_id":42,"default_team_id":7,"metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}` + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/neworg/provision", body) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "disagree") { + t.Fatalf("error should state the fields disagree: %s", rec.Body.String()) + } + if store.lastProvision != nil { + t.Fatal("nothing must be provisioned on disagreement") + } +} + +func TestProvisionAgreeingTeamIDFieldsAccepted(t *testing.T) { + store := newFakeStore() + router := newTestRouter(store) + + body := `{"database_name":"d","team_id":42,"default_team_id":42,"metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}` + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/neworg/provision", body) + if rec.Code != http.StatusAccepted { + t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body.String()) + } + if store.lastProvision.DefaultTeamID != 42 { + t.Fatalf("effective team = %d, want 42", store.lastProvision.DefaultTeamID) + } +} + +func TestProvisionSchemaNameRequiresTeamID(t *testing.T) { + store := newFakeStore() + store.orgs["acme"] = &configstore.Org{Name: "acme"} + router := newTestRouter(store) + + body := `{"database_name":"d","schema_name":"custom_wh","metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}` + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/acme/provision", body) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "schema_name requires team_id") { + t.Fatalf("error should say schema_name requires team_id: %s", rec.Body.String()) + } +} + +func TestProvisionInvalidSchemaNameRejected(t *testing.T) { + store := newFakeStore() + router := newTestRouter(store) + + body := `{"database_name":"d","team_id":42,"schema_name":"Bad-Name","metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}` + rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/neworg/provision", body) + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body.String()) + } +} diff --git a/controlplane/provisioning/store.go b/controlplane/provisioning/store.go index 883381ab..70c3a036 100644 --- a/controlplane/provisioning/store.go +++ b/controlplane/provisioning/store.go @@ -46,7 +46,13 @@ type ProvisionRequest struct { // re-provision of an existing org, where it keeps the stored billing // team (never a wipe). Prerequisite for pull-based compute billing. DefaultTeamID int64 - Warehouse *configstore.ManagedWarehouse + // SchemaName optionally overrides the conventional "team_" warehouse + // schema for the billing team row created/updated by this provision. Only + // meaningful when DefaultTeamID is set (the handler resolves the new + // team_id/schema_name request fields into this pair). Empty = keep the + // conventional name. + SchemaName string + Warehouse *configstore.ManagedWarehouse // RootUserHash is the bcrypt hash of the freshly-generated root // password. Plaintext stays in the handler (returned to the // caller); only the hash is persisted, same as before. @@ -93,7 +99,7 @@ func (s *gormStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // No default_team_id on this standalone path — an existing org keeps // its billing team as-is; creating a NEW org through here now fails // with ErrDefaultTeamIDRequired (same invariant as Provision). - return createPendingWarehouseTx(tx, orgID, databaseName, 0, warehouse) + return createPendingWarehouseTx(tx, orgID, databaseName, 0, "", warehouse) }) } @@ -106,7 +112,7 @@ func (s *gormStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // Returns a sentinel-comparable error string ("warehouse already // exists in non-terminal state") so HTTP handlers can map to 409 // without an extra error type. -func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTeamID int64, warehouse *configstore.ManagedWarehouse) error { +func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTeamID int64, schemaName string, warehouse *configstore.ManagedWarehouse) error { // Auto-create org if it doesn't exist (PostHog calls provision, duckgres // creates everything). A NEW org MUST carry default_team_id — it becomes // the org's first duckgres_org_teams row, marked as the billing team @@ -151,6 +157,18 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTe if err := configstore.SetOrgBillingTeamTx(tx, orgID, defaultTeamID); err != nil { return err } + // An explicit schema_name (the new provision fields) replaces the + // conventional "team_" on the billing row. The unique + // (org_id, schema_name) index rejects a name another team already + // holds (23505 → 409 at the handler). + if schemaName != "" { + if err := tx.Exec(` + UPDATE duckgres_org_teams SET schema_name = ?, updated_at = now() + WHERE org_id = ? AND team_id = ?`, + schemaName, orgID, defaultTeamID).Error; err != nil { + return fmt.Errorf("set billing team schema (org=%s team=%d): %w", orgID, defaultTeamID, err) + } + } // The org's billing team changed on a re-provision: re-attribute its // buffered (unacked) billing buckets to the new team in this same // transaction, so the next billing pull reports them under the new @@ -225,7 +243,7 @@ func (s *gormStore) Provision(req ProvisionRequest) error { } return s.cs.DB().Transaction(func(tx *gorm.DB) error { // 1. Warehouse + Org (extracted helper). - if err := createPendingWarehouseTx(tx, req.OrgID, req.DatabaseName, req.DefaultTeamID, req.Warehouse); err != nil { + if err := createPendingWarehouseTx(tx, req.OrgID, req.DatabaseName, req.DefaultTeamID, req.SchemaName, req.Warehouse); err != nil { return err } @@ -285,3 +303,58 @@ func (s *gormStore) SetWarehouseDeleting(orgID string, expectedState configstore } return nil } + +// ListOrgTeams returns every duckgres_org_teams row for the org, or +// gorm.ErrRecordNotFound when the org doesn't exist. +func (s *gormStore) ListOrgTeams(orgID string) ([]configstore.OrgTeam, error) { + var count int64 + if err := s.cs.DB().Model(&configstore.Org{}).Where("name = ?", orgID).Count(&count).Error; err != nil { + return nil, err + } + if count == 0 { + return nil, gorm.ErrRecordNotFound + } + var teams []configstore.OrgTeam + if err := s.cs.DB().Where("org_id = ?", orgID).Order("team_id").Find(&teams).Error; err != nil { + return nil, err + } + return teams, nil +} + +// UpsertOrgTeam creates or overwrites one (org, team) row in its own +// transaction. See configstore.UpsertOrgTeamTx for the grandfather semantics +// (schema_name IS overwritable here, by design). +func (s *gormStore) UpsertOrgTeam(orgID string, up configstore.OrgTeamUpsert) (*configstore.OrgTeam, error) { + var stored *configstore.OrgTeam + err := s.cs.DB().Transaction(func(tx *gorm.DB) error { + var err error + stored, err = configstore.UpsertOrgTeamTx(tx, orgID, up) + return err + }) + if err != nil { + return nil, err + } + return stored, nil +} + +// DeleteOrgTeam deletes one (org, team) CONFIG row in its own transaction, +// with the billing handover / last-team rules of configstore.DeleteOrgTeamTx. +func (s *gormStore) DeleteOrgTeam(orgID string, teamID int64) (*configstore.OrgTeamDeleteResult, error) { + var res *configstore.OrgTeamDeleteResult + err := s.cs.DB().Transaction(func(tx *gorm.DB) error { + var err error + res, err = configstore.DeleteOrgTeamTx(tx, orgID, teamID) + if err != nil { + return err + } + if res.WasBilling { + slog.Info("Deleted billing team; promoted oldest remaining team and re-attributed usage.", + "org", orgID, "deleted_team", teamID, "new_billing_team", res.NewBillingTeamID, "rows", res.UsageRowsMoved) + } + return nil + }) + if err != nil { + return nil, err + } + return res, nil +} diff --git a/tests/configstore/migrations_postgres_test.go b/tests/configstore/migrations_postgres_test.go index 4079853c..61570a05 100644 --- a/tests/configstore/migrations_postgres_test.go +++ b/tests/configstore/migrations_postgres_test.go @@ -43,7 +43,8 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) { requireGooseMigrationRecorded(t, db, 22) requireGooseMigrationRecorded(t, db, 23) requireGooseMigrationRecorded(t, db, 24) - requireGooseLatestVersion(t, db, 24) + requireGooseMigrationRecorded(t, db, 25) + requireGooseLatestVersion(t, db, 25) requireTableAbsent(t, db, "duckgres_schema_migrations") // Migration 000018 added the reshard operation + verbose log tables. @@ -84,6 +85,13 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) { requireColumnDefault(t, db, "duckgres_org_teams", "enabled", "true") requireColumnType(t, db, "duckgres_org_compute_usage", "team_id", "bigint") + // Migration 000025 added the legacy table-name overrides (nullable — NULL + // means "derive from schema_name") and the per-org schema uniqueness. + requireColumnNullable(t, db, "duckgres_org_teams", "events_table_name") + requireColumnNullable(t, db, "duckgres_org_teams", "persons_table_name") + requireColumnNullable(t, db, "duckgres_org_teams", "schema_data_imports_name") + requireUniqueIndex(t, db, "duckgres_org_teams", "org_id,schema_name") + // Migration 000007 added the compute-usage billing buffer; 000015 widened // its key for pull-based billing (team_id, query_source, worker size), // added the single ack cursor and dropped the push-drain state table. @@ -175,7 +183,7 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { ); DROP TABLE IF EXISTS duckgres_reshard_operation_log; DROP TABLE IF EXISTS duckgres_reshard_operations; - DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24); + DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25); `).Error; err != nil { t.Fatalf("downgrade baseline schema to pre-v9 shape: %v", err) } @@ -213,7 +221,8 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { requireGooseMigrationRecorded(t, upgradedDB, 22) requireGooseMigrationRecorded(t, upgradedDB, 23) requireGooseMigrationRecorded(t, upgradedDB, 24) - requireGooseLatestVersion(t, upgradedDB, 24) + requireGooseMigrationRecorded(t, upgradedDB, 25) + requireGooseLatestVersion(t, upgradedDB, 25) requireColumnPresent(t, upgradedDB, "duckgres_reshard_operations", "password_url") requireTablePresent(t, upgradedDB, "duckgres_worker_spawn_log") requireColumnDefault(t, upgradedDB, "duckgres_orgs", "max_vcpus", "0") @@ -221,6 +230,8 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { requireColumnDefault(t, upgradedDB, "duckgres_org_users", "disabled", "false") requireColumnAbsent(t, upgradedDB, "duckgres_orgs", "default_team_id") requireTablePresent(t, upgradedDB, "duckgres_org_teams") + requireColumnNullable(t, upgradedDB, "duckgres_org_teams", "events_table_name") + requireUniqueIndex(t, upgradedDB, "duckgres_org_teams", "org_id,schema_name") requireColumnAbsent(t, upgradedDB, "duckgres_orgs", "max_connections") requireColumnAbsent(t, upgradedDB, "duckgres_managed_warehouses", "iceberg_enabled") requireColumnAbsent(t, upgradedDB, "duckgres_org_users", "default_catalog") @@ -685,6 +696,57 @@ func requireColumnNotNull(t *testing.T, db *sql.DB, tableName, columnName string } } +func requireColumnNullable(t *testing.T, db *sql.DB, tableName, columnName string) { + t.Helper() + + var isNullable string + err := db.QueryRow(` + SELECT is_nullable + FROM information_schema.columns + WHERE table_schema = current_schema() + AND table_name = $1 + AND column_name = $2 + `, tableName, columnName).Scan(&isNullable) + if err != nil { + if err == sql.ErrNoRows { + t.Fatalf("%s.%s column missing", tableName, columnName) + } + t.Fatalf("query %s.%s nullability: %v", tableName, columnName, err) + } + if isNullable != "YES" { + t.Fatalf("%s.%s is_nullable = %q, want %q (column must be nullable)", tableName, columnName, isNullable, "YES") + } +} + +func requireUniqueIndex(t *testing.T, db *sql.DB, tableName, columnNames string) { + t.Helper() + + var count int + if err := db.QueryRow(` + SELECT COUNT(*) + FROM ( + SELECT idx.indexrelid, + string_agg(src_col.attname, ',' ORDER BY cols.ord) AS column_names + FROM pg_index idx + JOIN pg_class src ON src.oid = idx.indrelid + JOIN pg_namespace src_ns ON src_ns.oid = src.relnamespace + JOIN LATERAL unnest(idx.indkey) WITH ORDINALITY AS cols(attnum, ord) ON true + JOIN pg_attribute src_col ON src_col.attrelid = src.oid AND src_col.attnum = cols.attnum + WHERE idx.indisunique + AND NOT idx.indisprimary + AND src_ns.nspname = current_schema() + AND src.relname = $1 + GROUP BY idx.indexrelid + ) i + WHERE i.column_names = $2 + `, tableName, columnNames).Scan(&count); err != nil { + t.Fatalf("query unique index on %s(%s): %v", tableName, columnNames, err) + } + if count != 1 { + t.Fatalf("unique index on %s(%s): count = %d, want 1", tableName, columnNames, count) + } +} + func requireColumnDefault(t *testing.T, db *sql.DB, tableName, columnName, wantDefault string) { t.Helper() diff --git a/tests/configstore/org_teams_postgres_test.go b/tests/configstore/org_teams_postgres_test.go new file mode 100644 index 00000000..b42ffe91 --- /dev/null +++ b/tests/configstore/org_teams_postgres_test.go @@ -0,0 +1,245 @@ +//go:build linux || darwin + +package configstore_test + +import ( + "errors" + "testing" + "time" + + "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/controlplane/provisioning" + "gorm.io/gorm" +) + +func seedOrg(t *testing.T, store *configstore.ConfigStore, name string) { + t.Helper() + if err := store.DB().Create(&configstore.Org{Name: name, DatabaseName: name + "db"}).Error; err != nil { + t.Fatalf("create org %s: %v", name, err) + } +} + +func readTeam(t *testing.T, store *configstore.ConfigStore, orgID string, teamID int64) configstore.OrgTeam { + t.Helper() + var team configstore.OrgTeam + if err := store.DB().First(&team, "org_id = ? AND team_id = ?", orgID, teamID).Error; err != nil { + t.Fatalf("read team (org=%s team=%d): %v", orgID, teamID, err) + } + return team +} + +func strPtr(s string) *string { return &s } +func boolPtr(b bool) *bool { return &b } + +// TestUpsertOrgTeamGrandfatherPostgres pins the grandfather contract of the +// provisioning upsert: an existing row's schema_name and legacy table names +// ARE overwritable (the PostHog backfill replaces the migration's "team_" +// placeholder through this path), while omitted presence-aware fields +// preserve the stored values. +func TestUpsertOrgTeamGrandfatherPostgres(t *testing.T) { + store := newIsolatedConfigStore(t) + seedOrg(t, store, "acme") + pstore := provisioning.NewGormStore(store) + + created, err := pstore.UpsertOrgTeam("acme", configstore.OrgTeamUpsert{ + TeamID: 7, + SchemaName: "team_7", + }) + if err != nil { + t.Fatalf("create upsert: %v", err) + } + if !created.Enabled || created.BackfillEnabled != nil || created.EventsTableName != nil { + t.Fatalf("created row = %+v, want enabled default true, backfill/legacy NULL", created) + } + + // Grandfather: replace the placeholder schema and set explicit names; + // disable backfill; enabled omitted must be preserved. + updated, err := pstore.UpsertOrgTeam("acme", configstore.OrgTeamUpsert{ + TeamID: 7, + SchemaName: "legacy_wh", + BackfillEnabled: boolPtr(false), + EventsTableName: strPtr("legacy_events"), + PersonsTableName: strPtr("legacy_persons"), + SchemaDataImportsName: strPtr("legacy_imports"), + }) + if err != nil { + t.Fatalf("grandfather upsert: %v", err) + } + if updated.SchemaName != "legacy_wh" { + t.Fatalf("schema_name = %q, want overwritten legacy_wh", updated.SchemaName) + } + if updated.EventsTableName == nil || *updated.EventsTableName != "legacy_events" || + updated.PersonsTableName == nil || *updated.PersonsTableName != "legacy_persons" || + updated.SchemaDataImportsName == nil || *updated.SchemaDataImportsName != "legacy_imports" { + t.Fatalf("legacy names not stored: %+v", updated) + } + if updated.BackfillEnabled == nil || *updated.BackfillEnabled { + t.Fatalf("backfill_enabled = %v, want false", updated.BackfillEnabled) + } + if !updated.Enabled { + t.Fatal("omitted enabled must preserve the stored value") + } + + // Explicit "" clears a legacy override back to NULL (derive again). + cleared, err := pstore.UpsertOrgTeam("acme", configstore.OrgTeamUpsert{ + TeamID: 7, + SchemaName: "legacy_wh", + EventsTableName: strPtr(""), + }) + if err != nil { + t.Fatalf("clearing upsert: %v", err) + } + if cleared.EventsTableName != nil { + t.Fatalf("events_table_name = %v, want NULL after explicit empty", *cleared.EventsTableName) + } + if cleared.PersonsTableName == nil || *cleared.PersonsTableName != "legacy_persons" { + t.Fatalf("omitted persons_table_name must be preserved, got %+v", cleared) + } +} + +func TestUpsertOrgTeamSchemaConflictPostgres(t *testing.T) { + store := newIsolatedConfigStore(t) + seedOrg(t, store, "acme") + seedOrg(t, store, "other") + pstore := provisioning.NewGormStore(store) + + if _, err := pstore.UpsertOrgTeam("acme", configstore.OrgTeamUpsert{TeamID: 1, SchemaName: "shared"}); err != nil { + t.Fatalf("seed team: %v", err) + } + + // Same org, different team, same schema → conflict. + if _, err := pstore.UpsertOrgTeam("acme", configstore.OrgTeamUpsert{TeamID: 2, SchemaName: "shared"}); !errors.Is(err, configstore.ErrOrgTeamSchemaConflict) { + t.Fatalf("duplicate schema in org: err = %v, want ErrOrgTeamSchemaConflict", err) + } + + // The same schema in a DIFFERENT org is fine (uniqueness is per org). + if _, err := pstore.UpsertOrgTeam("other", configstore.OrgTeamUpsert{TeamID: 9, SchemaName: "shared"}); err != nil { + t.Fatalf("same schema across orgs must be allowed: %v", err) + } + + // Unknown org → not found. + if _, err := pstore.UpsertOrgTeam("ghost", configstore.OrgTeamUpsert{TeamID: 1, SchemaName: "x"}); !errors.Is(err, gorm.ErrRecordNotFound) { + t.Fatalf("unknown org: err = %v, want gorm.ErrRecordNotFound", err) + } + + // The unique index backs the pre-check: a direct insert that bypasses the + // helper still fails at the database. + err := store.DB().Exec(` + INSERT INTO duckgres_org_teams (org_id, team_id, schema_name, enabled, created_at, updated_at) + VALUES ('acme', 3, 'shared', TRUE, now(), now())`).Error + if err == nil { + t.Fatal("direct duplicate-schema insert must violate the unique index") + } +} + +// TestDeleteOrgTeamPostgres covers the transactional delete rules: last-team +// refusal, billing handover to the OLDEST remaining team, and the usage +// bucket re-attribution riding in the same transaction. +func TestDeleteOrgTeamPostgres(t *testing.T) { + store := newIsolatedConfigStore(t) + seedOrg(t, store, "acme") + pstore := provisioning.NewGormStore(store) + + // Billing team 1 (oldest), then 3, then 2 — created_at decides succession, + // NOT team id, so the successor must be team 3. + base := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + for _, seed := range []struct { + team int64 + schema string + createdAt time.Time + billing bool + }{ + {1, "team_1", base, true}, + {3, "team_3", base.Add(time.Hour), false}, + {2, "team_2", base.Add(2 * time.Hour), false}, + } { + var billing interface{} + if seed.billing { + billing = true + } + if err := store.DB().Exec(` + INSERT INTO duckgres_org_teams (org_id, team_id, schema_name, enabled, is_billing_team, created_at, updated_at) + VALUES ('acme', ?, ?, TRUE, ?, ?, ?)`, + seed.team, seed.schema, billing, seed.createdAt, seed.createdAt).Error; err != nil { + t.Fatalf("seed team %d: %v", seed.team, err) + } + } + bucket := time.Date(2026, 7, 14, 10, 0, 0, 0, time.UTC) + seedUsage(t, store, "acme", 1, bucket, 10, 20, 1000) + + // Deleting a non-billing team: no handover, billing untouched. + res, err := pstore.DeleteOrgTeam("acme", 2) + if err != nil { + t.Fatalf("delete non-billing team: %v", err) + } + if res.WasBilling || res.NewBillingTeamID != 0 { + t.Fatalf("non-billing delete result = %+v, want no handover", res) + } + + // Deleting the billing team: the OLDEST remaining team (3) takes over and + // the buffered usage moves to it atomically. + res, err = pstore.DeleteOrgTeam("acme", 1) + if err != nil { + t.Fatalf("delete billing team: %v", err) + } + if !res.WasBilling || res.NewBillingTeamID != 3 { + t.Fatalf("billing delete result = %+v, want handover to team 3", res) + } + successor := readTeam(t, store, "acme", 3) + if successor.IsBillingTeam == nil || !*successor.IsBillingTeam { + t.Fatalf("team 3 must carry the billing mark, got %+v", successor) + } + compute := computeRowsForOrg(t, store, "acme") + if len(compute) != 1 || compute[0].TeamID != 3 || compute[0].CPUSeconds != 10 { + t.Fatalf("compute usage not re-attributed to team 3: %+v", compute) + } + storage := storageRowsForOrg(t, store, "acme") + if len(storage) != 1 || storage[0].TeamID != 3 || storage[0].ByteSeconds != 1000 { + t.Fatalf("storage usage not re-attributed to team 3: %+v", storage) + } + + // The last team cannot be deleted. + if _, err := pstore.DeleteOrgTeam("acme", 3); !errors.Is(err, configstore.ErrLastOrgTeam) { + t.Fatalf("last-team delete: err = %v, want ErrLastOrgTeam", err) + } + if got := readTeam(t, store, "acme", 3); got.TeamID != 3 { + t.Fatal("last team must survive the refused delete") + } + + // Unknown team → not found. + if _, err := pstore.DeleteOrgTeam("acme", 99); !errors.Is(err, configstore.ErrOrgTeamNotFound) { + t.Fatalf("unknown team delete: err = %v, want ErrOrgTeamNotFound", err) + } +} + +// TestProvisionWithTeamIDAndSchemaNamePostgres: the org-create path with the +// new team_id+schema_name pair creates the first team row as the billing team +// with the EXPLICIT schema instead of the conventional "team_". (The +// legacy default_team_id path — conventional schema — is pinned by +// TestProvisionReattributesUsageOnTeamChangePostgres.) +func TestProvisionWithTeamIDAndSchemaNamePostgres(t *testing.T) { + store := newIsolatedConfigStore(t) + pstore := provisioning.NewGormStore(store) + + if err := pstore.Provision(provisioning.ProvisionRequest{ + OrgID: "schemaorg", + DatabaseName: "schemaorgdb", + DefaultTeamID: 42, + SchemaName: "custom_wh", + Warehouse: &configstore.ManagedWarehouse{DucklingName: "schemaorg"}, + RootUserHash: "hash", + }); err != nil { + t.Fatalf("provision with schema: %v", err) + } + + team := readTeam(t, store, "schemaorg", 42) + if team.SchemaName != "custom_wh" { + t.Fatalf("schema_name = %q, want explicit custom_wh", team.SchemaName) + } + if team.IsBillingTeam == nil || !*team.IsBillingTeam { + t.Fatalf("first team must be the billing team, got %+v", team) + } + if !team.Enabled { + t.Fatal("first team must be enabled") + } +} diff --git a/tests/mw-dev/e2e/harness.sh b/tests/mw-dev/e2e/harness.sh index dd105256..ee186aad 100755 --- a/tests/mw-dev/e2e/harness.sh +++ b/tests/mw-dev/e2e/harness.sh @@ -1356,6 +1356,69 @@ default_team_id_mandatory() { # cnpg_org ext_org log "default_team_id OK: PUT set round-trips, clear (0/null) rejected 400, value preserved on $ext_org" } +# ---- org teams CRUD (duckgres_org_teams via the provisioning API) ----------- +# The PostHog backend manages an org's team rows through +# GET/POST /orgs/:id/teams + DELETE /orgs/:id/teams/:team_id. Contract asserted +# on a real org against the real config store, additively and self-cleaning +# (the org ends the function with exactly its provisioned billing team): +# 1. List: the provisioned billing team row exists with the conventional +# "team_" schema and is_billing_team=true. +# 2. Upsert: a second team is created (enabled defaults true); re-POSTing it +# is the GRANDFATHER path — schema_name and the legacy table-name +# overrides are overwritten (the PostHog backfill replaces the +# migration's placeholder through this route). +# 3. Conflict: a third team claiming an existing schema_name is a 409. +# 4. Delete: removing the second team is 200 with no billing handover; +# deleting the org's LAST team (the billing one) is refused with a 409 +# naming org deletion as the only way. +org_teams_crud() { # org billing_team_id + org="$1"; team="$2"; extra=$((team + 1)); clash=$((team + 2)) + log "org teams CRUD on $org (billing team $team)" + + # Earlier default_team_id repoints (set → restore) leave DEMOTED team rows + # behind (SetOrgBillingTeamTx demotes, never deletes). Delete every + # non-billing row first so the last-team refusal below is deterministic. + for tid in $(curl -fsS -H "$H" "$API/api/v1/orgs/$org/teams" | jq -r '.teams[] | select(.is_billing_team != true) | .team_id'); do + code="$(curl -s -o /tmp/team_out -w '%{http_code}' -X DELETE -H "$H" "$API/api/v1/orgs/$org/teams/$tid")" + [ "$code" = "200" ] || fail "org teams: cleanup delete of leftover team $tid -> HTTP $code: $(cat /tmp/team_out)" + done + + # 1. Provisioned billing row present. + curl -fsS -H "$H" "$API/api/v1/orgs/$org/teams" \ + | jq -e --argjson t "$team" '.teams | map(select(.team_id==$t and .schema_name=="team_\($t)" and .is_billing_team==true and .enabled==true)) | length == 1' >/dev/null \ + || fail "org teams: provisioned billing row for team $team missing/wrong: $(curl -fsS -H "$H" "$API/api/v1/orgs/$org/teams" | head -c 400)" + + # 2. Create a second team, then grandfather-overwrite its schema + legacy names. + code="$(curl -s -o /tmp/team_out -w '%{http_code}' -X POST -H "$H" -H 'Content-Type: application/json' \ + -d "{\"team_id\":$extra,\"schema_name\":\"team_$extra\"}" "$API/api/v1/orgs/$org/teams")" + [ "$code" = "200" ] || fail "org teams: create team $extra -> HTTP $code: $(cat /tmp/team_out)" + code="$(curl -s -o /tmp/team_out -w '%{http_code}' -X POST -H "$H" -H 'Content-Type: application/json' \ + -d "{\"team_id\":$extra,\"schema_name\":\"e2e_legacy_wh\",\"events_table_name\":\"legacy_events\"}" "$API/api/v1/orgs/$org/teams")" + [ "$code" = "200" ] || fail "org teams: grandfather upsert of team $extra -> HTTP $code: $(cat /tmp/team_out)" + jq -e '.schema_name=="e2e_legacy_wh" and .events_table_name=="legacy_events" and .is_billing_team==null' /tmp/team_out >/dev/null \ + || fail "org teams: grandfather upsert did not overwrite schema/legacy names: $(cat /tmp/team_out)" + + # 3. Duplicate schema within the org -> 409. + code="$(curl -s -o /tmp/team_out -w '%{http_code}' -X POST -H "$H" -H 'Content-Type: application/json' \ + -d "{\"team_id\":$clash,\"schema_name\":\"e2e_legacy_wh\"}" "$API/api/v1/orgs/$org/teams")" + [ "$code" = "409" ] || fail "org teams: duplicate schema -> HTTP $code want 409: $(cat /tmp/team_out)" + + # 4. Delete the extra team (non-billing: no handover), then assert the LAST + # (billing) team refuses deletion — a refused delete leaves state untouched. + code="$(curl -s -o /tmp/team_out -w '%{http_code}' -X DELETE -H "$H" "$API/api/v1/orgs/$org/teams/$extra")" + [ "$code" = "200" ] || fail "org teams: delete team $extra -> HTTP $code: $(cat /tmp/team_out)" + jq -e 'has("new_billing_team_id") | not' /tmp/team_out >/dev/null \ + || fail "org teams: non-billing delete must not hand billing over: $(cat /tmp/team_out)" + code="$(curl -s -o /tmp/team_out -w '%{http_code}' -X DELETE -H "$H" "$API/api/v1/orgs/$org/teams/$team")" + [ "$code" = "409" ] || fail "org teams: last-team delete -> HTTP $code want 409: $(cat /tmp/team_out)" + grep -q "deleting the org" /tmp/team_out \ + || fail "org teams: last-team refusal should name org deletion as the only way: $(cat /tmp/team_out)" + curl -fsS -H "$H" "$API/api/v1/orgs/$org/teams" \ + | jq -e --argjson t "$team" '.teams | length == 1 and (.[0].team_id==$t) and (.[0].is_billing_team==true)' >/dev/null \ + || fail "org teams: org must end with exactly its billing team $team" + log "org teams OK: list/upsert/grandfather/409/delete rules on $org" +} + # ---- user persistent secrets ------------------------------------------------ # CREATE PERSISTENT SECRET must survive across sessions: worker pods are # ephemeral, so the CP intercepts the statement, stores it encrypted in the @@ -3175,6 +3238,9 @@ lane_ext() { # external-RDS metadata backend + org default profile # restore on EXT. Runs here because both orgs are provisioned by now; it only # touches the provisioning + admin APIs, no worker/DB. default_team_id_mandatory "$CNPG" "$EXT" + # Team CRUD on the ext org (no other test mutates its team rows); ends with + # the org back at exactly its provisioned billing team. + org_teams_crud "$EXT" "$EXT_DEFAULT_TEAM_ID" } main() {