From 6565c7c6147a945c328e90221c995c6d63cb2f70 Mon Sep 17 00:00:00 2001 From: Benjamin Knofe-Vider Date: Mon, 6 Jul 2026 16:56:21 +0200 Subject: [PATCH 1/2] default_team_id is an integer end-to-end (migration to BIGINT) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PostHog team ids are integers (Team.id), so PostHog naturally sends `"default_team_id": 12345` — the string-typed field 400ed that obvious payload at provision time ('json: cannot unmarshal number into Go struct field provisionRequest.default_team_id of type string'), breaking a beta customer's warehouse provisioning. Rather than stringify on the caller side, make the type match the source of truth everywhere: - Migration 000017: duckgres_orgs.default_team_id VARCHAR → BIGINT (NULL stays NULL; all stored values are numeric strings, a non-numeric one fails the migration loudly) and the compute-usage bucket key's team_id TEXT → BIGINT (0 = 'no default team' sentinel, it sits in the PK so it can't be NULL). USING clauses go through ::text so a goose replay onto an already-converted schema still type-checks. - Org.DefaultTeamID *int64, OrgConfig/OrgDefaultTeamID int64 (0=unset); meter + bucket key + pull-API response carry team_id as an integer (JSON number). - Provisioning API: default_team_id is a JSON number; a quoted string is now a decode-time 400 naming the field (contract test pins it). - Admin API: *int64 — nil preserves, 0 (or explicit JSON null) clears, n sets; audit detail renders (unset) for nil/0. Admin UI input keeps the text field but validates digits-only and sends number/0. - e2e: provision bodies send numbers, PUT set/clear/restore uses 424242/0/number, the billing pull round-trip compares team_id numerically. No PostHog-side change needed — its current integer payload becomes correct once this deploys. --- CLAUDE.md | 10 ++-- controlplane/admin/api.go | 28 +++++++++-- controlplane/admin/ui/src/pages/OrgDetail.tsx | 13 +++-- controlplane/admin/ui/src/pages/Orgs.tsx | 2 +- controlplane/admin/ui/src/types/api.ts | 5 +- controlplane/compute_billing_api_test.go | 2 +- controlplane/compute_meter.go | 16 +++---- controlplane/compute_meter_test.go | 36 +++++++------- controlplane/configstore/compute_usage.go | 4 +- .../000017_default_team_id_bigint.sql | 31 ++++++++++++ controlplane/configstore/models.go | 16 +++---- controlplane/configstore/store.go | 14 +++--- .../provisioning/analytics_events_test.go | 2 +- controlplane/provisioning/api.go | 13 ++--- controlplane/provisioning/api_test.go | 48 ++++++++++++++----- controlplane/provisioning/store.go | 21 ++++---- docs/design/billing-pull-api.md | 11 +++-- tests/configstore/migrations_postgres_test.go | 36 ++++++++++++-- tests/e2e-mw-dev/harness.sh | 25 +++++----- 19 files changed, 224 insertions(+), 109 deletions(-) create mode 100644 controlplane/configstore/migrations/000017_default_team_id_bigint.sql diff --git a/CLAUDE.md b/CLAUDE.md index e5426fe2..6c4fba9f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -460,8 +460,10 @@ worker size: `cpu_seconds = vCPU × ceil(conn_secs)`, `memory_seconds = GiB × ceil(conn_secs)`. Counted internally in integer **millicore-seconds** / **MiB-seconds** (`compute_meter.go`) to avoid truncating a fractional-core / sub-GiB worker; worker size is stored in the bucket key as exact NUMERIC -decimals (vCPU / GiB). `team_id` is the org\'s `default_team_id` (resolved from -the config snapshot at connection end); `query_source` is the +decimals (vCPU / GiB). `team_id` is the org's `default_team_id` (an integer — +PostHog's `Team.id`; a JSON NUMBER on every API surface, BIGINT in the config +store, 0 = "no default team"; resolved from the config snapshot at connection +end); `query_source` is the `duckgres.query_source` session GUC (`standard` unless set; a mid-connection change bills the whole connection under the final value). Invariants for anyone touching this path: @@ -483,10 +485,10 @@ touching this path: is UPSERT-increment so all CP pods sum into one row per key. - **Serve only closed buckets.** `watermark_high` = the newest bucket with `bucket_start ≤ now − 60s − 30s grace` (grace > flush interval, so every - CP\'s contribution has landed before a minute is served). The GET aggregates + CP's contribution has landed before a minute is served). The GET aggregates the window `(cursor, watermark_high]` into one row per `(org, team, query_source, cpu, mem_gib)` per **UTC day** — response size is - bounded by active keys × days, so billing downtime can\'t make it explode. + bounded by active keys × days, so billing downtime can't make it explode. - **Ack is the only deletion path (plus the 30d GC).** `POST /billing/ack` advances the single global cursor monotonically and deletes buckets `≤ watermark_high` in one TXN (`AckComputeUsage`). Idempotent — re-acks and diff --git a/controlplane/admin/api.go b/controlplane/admin/api.go index cb8cc7ec..6c44d29f 100644 --- a/controlplane/admin/api.go +++ b/controlplane/admin/api.go @@ -201,10 +201,11 @@ func (s *gormAPIStore) UpdateOrg(name string, updates configstore.Org) (*configs fields["hostname_alias"] = *updates.HostnameAlias } } - // DefaultTeamID is *string with the same semantics: nil = preserve, "" = - // clear (NULL), "x" = set. Nullable/optional — NULL is always valid. + // DefaultTeamID is *int64: nil = preserve, 0 = clear (NULL), n = set. + // (PostHog team ids start at 1, so 0 is a safe clear sentinel; the + // handler also maps an explicit JSON null onto it.) if updates.DefaultTeamID != nil { - if *updates.DefaultTeamID == "" { + if *updates.DefaultTeamID == 0 { fields["default_team_id"] = nil } else { fields["default_team_id"] = *updates.DefaultTeamID @@ -610,7 +611,15 @@ func (h *apiHandler) updateOrg(c *gin.Context) { merged.HostnameAlias = updates.HostnameAlias } if _, ok := fields["default_team_id"]; ok { - merged.DefaultTeamID = updates.DefaultTeamID + // Key present: a number sets, and an explicit JSON null (which + // unmarshals to a nil pointer) clears — normalize null onto the 0 + // clear-sentinel so the store layer sees one shape. + if updates.DefaultTeamID == nil { + zero := int64(0) + merged.DefaultTeamID = &zero + } else { + merged.DefaultTeamID = updates.DefaultTeamID + } } // Audit detail: which fields changed and their old → new values, so the @@ -629,7 +638,7 @@ func (h *apiHandler) updateOrg(c *gin.Context) { addChange("default_worker_ttl", orgStr(existing.DefaultWorkerTTL), orgStr(merged.DefaultWorkerTTL)) addChange("default_worker_min_hot_idle", existing.DefaultWorkerMinHotIdle, merged.DefaultWorkerMinHotIdle) addChange("hostname_alias", orgStrPtr(existing.HostnameAlias), orgStrPtr(merged.HostnameAlias)) - addChange("default_team_id", orgStrPtr(existing.DefaultTeamID), orgStrPtr(merged.DefaultTeamID)) + addChange("default_team_id", orgInt64Ptr(existing.DefaultTeamID), orgInt64Ptr(merged.DefaultTeamID)) if len(changes) > 0 { setAuditDetail(c, strings.Join(changes, ", ")) } @@ -698,6 +707,15 @@ func orgStrPtr(s *string) string { return orgStr(*s) } +// orgInt64Ptr renders an optional org config integer (nil or 0 == unset) for +// audit detail. +func orgInt64Ptr(v *int64) string { + if v == nil || *v == 0 { + return "(unset)" + } + return fmt.Sprintf("%d", *v) +} + func validateOrgMutationPayload(org *configstore.Org) error { if org == nil { return nil diff --git a/controlplane/admin/ui/src/pages/OrgDetail.tsx b/controlplane/admin/ui/src/pages/OrgDetail.tsx index 43392f76..4eff9a7d 100644 --- a/controlplane/admin/ui/src/pages/OrgDetail.tsx +++ b/controlplane/admin/ui/src/pages/OrgDetail.tsx @@ -53,7 +53,7 @@ function orgToForm(o: { default_worker_ttl: string; default_worker_min_hot_idle: number; hostname_alias: string | null; - default_team_id: string | null; + default_team_id: number | null; }): FormState { return { max_workers: String(o.max_workers), @@ -63,7 +63,7 @@ function orgToForm(o: { default_worker_ttl: o.default_worker_ttl, default_worker_min_hot_idle: String(o.default_worker_min_hot_idle), hostname_alias: o.hostname_alias ?? "", - default_team_id: o.default_team_id ?? "", + default_team_id: o.default_team_id == null ? "" : String(o.default_team_id), }; } @@ -107,6 +107,13 @@ export function OrgDetail() { const save = async () => { setMsg(null); + // default_team_id is an integer on the wire (PostHog team id); the text + // input needs a digits-only guard so junk can't silently clear it. + const teamIdText = form.default_team_id.trim(); + if (teamIdText !== "" && !/^\d+$/.test(teamIdText)) { + setMsg({ kind: "err", text: "Default team id must be a number (or empty to clear)." }); + return; + } const body: OrgUpdate = { max_workers: Number(form.max_workers) || 0, max_vcpus: Number(form.max_vcpus) || 0, @@ -115,7 +122,7 @@ export function OrgDetail() { default_worker_ttl: form.default_worker_ttl, default_worker_min_hot_idle: Number(form.default_worker_min_hot_idle) || 0, hostname_alias: form.hostname_alias === "" ? "" : form.hostname_alias, - default_team_id: form.default_team_id === "" ? "" : form.default_team_id, + default_team_id: teamIdText === "" ? 0 : Number(teamIdText), }; try { await update.mutateAsync(body); diff --git a/controlplane/admin/ui/src/pages/Orgs.tsx b/controlplane/admin/ui/src/pages/Orgs.tsx index 881d7019..a07d8f9a 100644 --- a/controlplane/admin/ui/src/pages/Orgs.tsx +++ b/controlplane/admin/ui/src/pages/Orgs.tsx @@ -57,7 +57,7 @@ export function Orgs() { accessorKey: "default_team_id", header: "Default team", cell: ({ getValue }) => { - const v = getValue() as string | null; + const v = getValue() as number | null; return v ? {v} : —; }, }, diff --git a/controlplane/admin/ui/src/types/api.ts b/controlplane/admin/ui/src/types/api.ts index 55a1c1ae..a45e0419 100644 --- a/controlplane/admin/ui/src/types/api.ts +++ b/controlplane/admin/ui/src/types/api.ts @@ -44,7 +44,7 @@ export interface Org { name: string; database_name: string; hostname_alias: string | null; - default_team_id: string | null; + default_team_id: number | null; max_workers: number; max_vcpus: number; default_worker_cpu: string; @@ -66,7 +66,8 @@ export interface OrgUpdate { default_worker_ttl?: string; default_worker_min_hot_idle?: number; hostname_alias?: string | null; - default_team_id?: string | null; + // Number sets, 0 clears (backend maps 0 → NULL), absent preserves. + default_team_id?: number; } // ---- Users (confirmed) ---- diff --git a/controlplane/compute_billing_api_test.go b/controlplane/compute_billing_api_test.go index 43202381..985d2f4c 100644 --- a/controlplane/compute_billing_api_test.go +++ b/controlplane/compute_billing_api_test.go @@ -124,7 +124,7 @@ func TestBillingUsageWindowAndRows(t *testing.T) { store := &fakeBillingStore{ cursor: cursor, hasAck: true, rows: []configstore.ComputeUsageRow{{ - Date: "2026-07-01", OrgID: "org_abc", TeamID: "12345", QuerySource: "standard", + Date: "2026-07-01", OrgID: "org_abc", TeamID: 12345, QuerySource: "standard", CPU: "8", MemGiB: "16", CPUSeconds: 4800, MemorySeconds: 9600, }}, } diff --git a/controlplane/compute_meter.go b/controlplane/compute_meter.go index 2b3c0715..3711cf04 100644 --- a/controlplane/compute_meter.go +++ b/controlplane/compute_meter.go @@ -26,7 +26,7 @@ const ( // bill) separately. type computeUsageKey struct { orgID string - teamID string + teamID int64 querySource string millicores int64 mib int64 @@ -81,7 +81,7 @@ func computeConnectionUsage(millicores, mib int64, dur time.Duration) (milliCPUS // Record adds one connection's usage to the in-process counter, bucketed by // connection-end time. Best-effort: a zero/unknown size is a no-op. Never // errors. -func (c *computeUsageCounter) Record(orgID, teamID, querySource string, millicores, mib int64, endTime time.Time, dur time.Duration) { +func (c *computeUsageCounter) Record(orgID, querySource string, teamID, millicores, mib int64, endTime time.Time, dur time.Duration) { if c == nil || orgID == "" || millicores <= 0 { return } @@ -152,15 +152,15 @@ type computeUsageStore interface { // computeMeter wires the per-org counter to a flusher. Nil-safe: a disabled // meter (non-remote backend) leaves this nil and every call site no-ops. // resolveTeam maps an org to its default PostHog team id at record time (a -// config-snapshot read — no I/O); empty is tolerated per the OrgDefaultTeamID -// contract and the bucket then carries an empty team_id. +// config-snapshot read — no I/O); 0 is tolerated per the OrgDefaultTeamID +// contract and the bucket then carries team_id 0 ("no default team"). type computeMeter struct { counter *computeUsageCounter store computeUsageStore - resolveTeam func(orgID string) string + resolveTeam func(orgID string) int64 } -func newComputeMeter(store computeUsageStore, resolveTeam func(orgID string) string) *computeMeter { +func newComputeMeter(store computeUsageStore, resolveTeam func(orgID string) int64) *computeMeter { return &computeMeter{counter: newComputeUsageCounter(), store: store, resolveTeam: resolveTeam} } @@ -169,11 +169,11 @@ func (m *computeMeter) Record(orgID, querySource string, millicores, mib int64, if m == nil { return } - teamID := "" + teamID := int64(0) if m.resolveTeam != nil { teamID = m.resolveTeam(orgID) } - m.counter.Record(orgID, teamID, querySource, millicores, mib, endTime, dur) + m.counter.Record(orgID, querySource, teamID, millicores, mib, endTime, dur) } // Flush drains the in-process counter into the durable buffer once. Best-effort: diff --git a/controlplane/compute_meter_test.go b/controlplane/compute_meter_test.go index 5601b8b7..b8367cf1 100644 --- a/controlplane/compute_meter_test.go +++ b/controlplane/compute_meter_test.go @@ -58,23 +58,23 @@ func TestComputeConnectionUsage(t *testing.T) { // keyA is the billing key most tests share: orgA's default team, standard // source, an 8-vCPU/16-GiB worker. func keyA(bucket time.Time) computeUsageKey { - return computeUsageKey{orgID: "orgA", teamID: "42", querySource: "standard", millicores: 8000, mib: 16 * 1024, bucket: bucket} + return computeUsageKey{orgID: "orgA", teamID: 42, querySource: "standard", millicores: 8000, mib: 16 * 1024, bucket: bucket} } func TestComputeUsageCounterBucketing(t *testing.T) { c := newComputeUsageCounter() base := time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) // Two connections in the same minute-bucket for the same key accumulate. - c.Record("orgA", "42", "standard", 8000, 16*1024, base.Add(5*time.Second), 10*time.Second) // 80 cpu-s, 160 gib-s - c.Record("orgA", "42", "standard", 8000, 16*1024, base.Add(40*time.Second), 10*time.Second) // +80, +160 + c.Record("orgA", "standard", 42, 8000, 16*1024, base.Add(5*time.Second), 10*time.Second) // 80 cpu-s, 160 gib-s + c.Record("orgA", "standard", 42, 8000, 16*1024, base.Add(40*time.Second), 10*time.Second) // +80, +160 // A connection in the next bucket is separate. - c.Record("orgA", "42", "standard", 8000, 16*1024, base.Add(65*time.Second), 5*time.Second) // 40 cpu-s, 80 gib-s + c.Record("orgA", "standard", 42, 8000, 16*1024, base.Add(65*time.Second), 5*time.Second) // 40 cpu-s, 80 gib-s // Different org, same bucket. - c.Record("orgB", "7", "standard", 1000, 1024, base.Add(5*time.Second), 2*time.Second) // 2 cpu-s, 2 gib-s + c.Record("orgB", "standard", 7, 1000, 1024, base.Add(5*time.Second), 2*time.Second) // 2 cpu-s, 2 gib-s // Same org+bucket but a different query source is a separate key. - c.Record("orgA", "42", "endpoints", 8000, 16*1024, base.Add(10*time.Second), 10*time.Second) // 80, 160 + c.Record("orgA", "endpoints", 42, 8000, 16*1024, base.Add(10*time.Second), 10*time.Second) // 80, 160 // Same org+bucket but a different worker size is a separate key. - c.Record("orgA", "42", "standard", 2000, 4*1024, base.Add(10*time.Second), 10*time.Second) // 20, 40 + c.Record("orgA", "standard", 42, 2000, 4*1024, base.Add(10*time.Second), 10*time.Second) // 20, 40 deltas := c.drainWholeUnits() got := map[computeUsageKey]configstore.ComputeUsageDelta{} @@ -92,7 +92,7 @@ func TestComputeUsageCounterBucketing(t *testing.T) { if d := got[keyA(bucket1)]; d.CPUSeconds != 40 || d.MemorySeconds != 80 { t.Errorf("orgA bucket1 = (%d,%d), want (40,80)", d.CPUSeconds, d.MemorySeconds) } - if d := got[computeUsageKey{orgID: "orgB", teamID: "7", querySource: "standard", millicores: 1000, mib: 1024, bucket: bucket0}]; d.CPUSeconds != 2 || d.MemorySeconds != 2 { + if d := got[computeUsageKey{orgID: "orgB", teamID: 7, querySource: "standard", millicores: 1000, mib: 1024, bucket: bucket0}]; d.CPUSeconds != 2 || d.MemorySeconds != 2 { t.Errorf("orgB bucket0 = (%d,%d), want (2,2)", d.CPUSeconds, d.MemorySeconds) } endpointsKey := keyA(bucket0) @@ -112,12 +112,12 @@ func TestComputeUsageCounterRemainderCarry(t *testing.T) { base := time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) // 500 millicores × 1s = 500 milli-cpu-seconds = 0 whole vCPU-seconds yet, // and 512 MiB × 1s = 512 MiB-seconds = 0 whole GiB-seconds yet. - c.Record("orgA", "42", "standard", 500, 512, base, 1*time.Second) + c.Record("orgA", "standard", 42, 500, 512, base, 1*time.Second) if deltas := c.drainWholeUnits(); len(deltas) != 0 { t.Fatalf("expected no whole units yet, got %v", deltas) } // Add another 500 milli-cpu-s (→ 1000 = 1 vCPU-s) and 512 MiB-s (→ 1024 = 1 GiB-s). - c.Record("orgA", "42", "standard", 500, 512, base, 1*time.Second) + c.Record("orgA", "standard", 42, 500, 512, base, 1*time.Second) deltas := c.drainWholeUnits() if len(deltas) != 1 || deltas[0].CPUSeconds != 1 || deltas[0].MemorySeconds != 1 { t.Fatalf("expected exactly 1 vCPU-s + 1 GiB-s after carry, got %v", deltas) @@ -146,11 +146,11 @@ func (f *fakeFlushStore) FlushComputeUsage(d []configstore.ComputeUsageDelta) er return nil } -func teamResolverA(orgID string) string { +func teamResolverA(orgID string) int64 { if orgID == "orgA" { - return "42" + return 42 } - return "" + return 0 } func TestComputeMeterFlush(t *testing.T) { @@ -166,8 +166,8 @@ func TestComputeMeterFlush(t *testing.T) { t.Fatalf("flushed = %v, want one (80,160)", store.flushed) } // The meter resolves the org's default team at record time. - if store.flushed[0].TeamID != "42" || store.flushed[0].QuerySource != "standard" { - t.Fatalf("flushed key = (team=%q, source=%q), want (42, standard)", store.flushed[0].TeamID, store.flushed[0].QuerySource) + if store.flushed[0].TeamID != 42 || store.flushed[0].QuerySource != "standard" { + t.Fatalf("flushed key = (team=%d, source=%q), want (42, standard)", store.flushed[0].TeamID, store.flushed[0].QuerySource) } // Nothing left to flush. if n := m.Flush(); n != 0 { @@ -177,15 +177,15 @@ func TestComputeMeterFlush(t *testing.T) { func TestComputeMeterUnknownTeamTolerated(t *testing.T) { // An org with no default_team_id (or a nil resolver) still meters — the - // bucket carries an empty team_id rather than dropping usage. + // bucket carries team_id 0 rather than dropping usage. store := &fakeFlushStore{} m := newComputeMeter(store, teamResolverA) m.Record("orgB", "standard", 1000, 1024, time.Now(), 2*time.Second) if n := m.Flush(); n != 1 { t.Fatalf("Flush rows = %d, want 1", n) } - if store.flushed[0].TeamID != "" { - t.Fatalf("team = %q, want empty for unknown org", store.flushed[0].TeamID) + if store.flushed[0].TeamID != 0 { + t.Fatalf("team = %d, want 0 for unknown org", store.flushed[0].TeamID) } } diff --git a/controlplane/configstore/compute_usage.go b/controlplane/configstore/compute_usage.go index ae1946e1..05b60cfe 100644 --- a/controlplane/configstore/compute_usage.go +++ b/controlplane/configstore/compute_usage.go @@ -20,7 +20,7 @@ import ( // float-equality issues. type ComputeUsageDelta struct { OrgID string - TeamID string + TeamID int64 QuerySource string Millicores int64 MiB int64 @@ -36,7 +36,7 @@ type ComputeUsageDelta struct { type ComputeUsageRow struct { Date string `json:"date"` OrgID string `json:"org_id"` - TeamID string `json:"team_id"` + TeamID int64 `json:"team_id"` QuerySource string `json:"query_source"` CPU json.Number `json:"cpu"` MemGiB json.Number `json:"mem_gib"` diff --git a/controlplane/configstore/migrations/000017_default_team_id_bigint.sql b/controlplane/configstore/migrations/000017_default_team_id_bigint.sql new file mode 100644 index 00000000..9c490bb0 --- /dev/null +++ b/controlplane/configstore/migrations/000017_default_team_id_bigint.sql @@ -0,0 +1,31 @@ +-- +goose Up + +-- default_team_id becomes a BIGINT end-to-end. PostHog team ids are integers +-- (Team.id), so callers naturally send JSON numbers — the string-typed column +-- forced a string JSON field, which 400ed the obvious `"default_team_id": +-- 12345` provision payload. All stored values are numeric strings (set via the +-- provisioning API / admin UI), so the cast is safe; a non-numeric value would +-- fail this migration loudly, which is the correct outcome (fix the row, not +-- the type). NULL stays NULL (empty string normalizes to NULL). The ::text in +-- the USING clauses makes the rewrite idempotent — a replay onto an +-- already-BIGINT column (goose re-run in tests) still type-checks. +ALTER TABLE duckgres_orgs + ALTER COLUMN default_team_id TYPE BIGINT USING NULLIF(default_team_id::text, '')::bigint; + +-- The compute-usage bucket key mirrors the org column. team_id sits in the +-- PRIMARY KEY so it must stay NOT NULL: 0 = "org had no default team" (PostHog +-- team ids start at 1), replacing the old '' sentinel. +ALTER TABLE duckgres_org_compute_usage + ALTER COLUMN team_id TYPE BIGINT USING COALESCE(NULLIF(team_id::text, ''), '0')::bigint, + ALTER COLUMN team_id SET DEFAULT 0; + +-- +goose Down + +-- Best-effort reversal: the 0 sentinel downgrades to '0' (not the old ''), +-- which is fine for a transient buffer. +ALTER TABLE duckgres_org_compute_usage + ALTER COLUMN team_id TYPE TEXT USING team_id::text, + ALTER COLUMN team_id SET DEFAULT ''; + +ALTER TABLE duckgres_orgs + ALTER COLUMN default_team_id TYPE VARCHAR(255) USING default_team_id::text; diff --git a/controlplane/configstore/models.go b/controlplane/configstore/models.go index 6f398f5e..aa1ffaa1 100644 --- a/controlplane/configstore/models.go +++ b/controlplane/configstore/models.go @@ -25,13 +25,13 @@ type Org struct { DefaultWorkerMemory string `gorm:"size:32" json:"default_worker_memory"` DefaultWorkerTTL string `gorm:"size:32" json:"default_worker_ttl"` DefaultWorkerMinHotIdle int `gorm:"default:0" json:"default_worker_min_hot_idle"` - // DefaultTeamID links the org to its default PostHog team (a team id, kept - // as a string). It is a prerequisite for pull-based compute billing — - // usage buckets are keyed by team_id = the org's default team. NULLABLE / - // optional everywhere: NULL means "unset" (existing orgs are backfilled - // separately, a follow-up makes it required). *string so the column is a - // nullable VARCHAR; callers must tolerate an empty/absent value. - DefaultTeamID *string `gorm:"size:255" json:"default_team_id,omitempty"` + // DefaultTeamID links the org to its default PostHog team id (an integer, + // matching PostHog's Team.id). It is a prerequisite for pull-based compute + // billing — usage buckets are keyed by team_id = the org's default team. + // *int64 so the column is a nullable BIGINT: NULL means "unset" (required + // at provision time for new orgs; the admin API can still clear it). + // Callers must tolerate an absent value. + DefaultTeamID *int64 `json:"default_team_id,omitempty"` Users []OrgUser `gorm:"foreignKey:OrgID;references:Name" json:"users,omitempty"` Warehouse *ManagedWarehouse `gorm:"foreignKey:OrgID;references:Name;constraint:OnDelete:CASCADE" json:"warehouse,omitempty"` CreatedAt time.Time `json:"created_at"` @@ -480,7 +480,7 @@ type OrgConfig struct { DefaultWorkerMemory string // org default worker profile: pod memory quantity ("" = unset) DefaultWorkerTTL string // org default worker profile: hot-idle TTL, Go duration string ("" = unset) DefaultWorkerMinHotIdle int // minimum default-profile hot-idle workers to retain for this org - DefaultTeamID string // org's default PostHog team id ("" = unset); prereq for pull-based compute billing + DefaultTeamID int64 // org's default PostHog team id (0 = unset); prereq for pull-based compute billing Users map[string]string // username -> password Warehouse *ManagedWarehouseConfig } diff --git a/controlplane/configstore/store.go b/controlplane/configstore/store.go index 29880b9f..2cd3086f 100644 --- a/controlplane/configstore/store.go +++ b/controlplane/configstore/store.go @@ -197,7 +197,7 @@ func (cs *ConfigStore) load() (*Snapshot, error) { if o.HostnameAlias != nil { alias = *o.HostnameAlias } - defaultTeamID := "" + defaultTeamID := int64(0) if o.DefaultTeamID != nil { defaultTeamID = *o.DefaultTeamID } @@ -485,19 +485,19 @@ func (cs *ConfigStore) OrgDefaultWorkerMinHotIdle(orgID string) int { } // OrgDefaultTeamID returns the org's default PostHog team id from the current -// snapshot, or "" when unset (including unknown orgs and a not-yet-loaded -// snapshot). NULLABLE/optional: an empty return is a valid, non-error state — -// callers must tolerate it and MUST NOT fail a connection or activation on it. +// snapshot, or 0 when unset (including unknown orgs and a not-yet-loaded +// snapshot). A zero return is a valid, non-error state — callers must +// tolerate it and MUST NOT fail a connection or activation on it. // Prerequisite for pull-based compute billing (usage keyed by team id). -func (cs *ConfigStore) OrgDefaultTeamID(orgID string) string { +func (cs *ConfigStore) OrgDefaultTeamID(orgID string) int64 { cs.mu.RLock() defer cs.mu.RUnlock() if cs.snapshot == nil { - return "" + return 0 } oc, ok := cs.snapshot.Orgs[orgID] if !ok { - return "" + return 0 } return oc.DefaultTeamID } diff --git a/controlplane/provisioning/analytics_events_test.go b/controlplane/provisioning/analytics_events_test.go index 14d55652..6834b11c 100644 --- a/controlplane/provisioning/analytics_events_test.go +++ b/controlplane/provisioning/analytics_events_test.go @@ -55,7 +55,7 @@ func TestProvisionEmitsAnalyticsEvent(t *testing.T) { store := newFakeStore() router := newTestRouter(store) - body := []byte(`{"database_name": "acme-db", "default_team_id": "1", "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + body := []byte(`{"database_name": "acme-db", "default_team_id": 1, "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/acme/provision", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() diff --git a/controlplane/provisioning/api.go b/controlplane/provisioning/api.go index fedfc290..79237711 100644 --- a/controlplane/provisioning/api.go +++ b/controlplane/provisioning/api.go @@ -135,12 +135,13 @@ type connectionDetails struct { type provisionRequest struct { DatabaseName string `json:"database_name"` - // DefaultTeamID links the org to its default PostHog team id. REQUIRED - // when the provision creates a NEW org (400 otherwise — every org carries - // its team id from birth; pull-based compute billing keys usage buckets by - // it). Optional on re-provision of an existing org: absent/empty keeps the - // stored value, never wipes it. - DefaultTeamID string `json:"default_team_id,omitempty"` + // DefaultTeamID links the org to its default PostHog team id — a JSON + // NUMBER, matching PostHog's integer Team.id (a quoted string is a 400 at + // decode time). REQUIRED when the provision creates a NEW org (400 + // otherwise — every org carries its team id from birth; pull-based compute + // billing keys usage buckets by it). Optional on re-provision of an + // existing org: absent/0 keeps the stored value, never wipes it. + DefaultTeamID int64 `json:"default_team_id,omitempty"` MetadataStore *provisionMetadataReq `json:"metadata_store,omitempty"` DataStore *provisionDataStoreReq `json:"data_store,omitempty"` DuckLake *provisionDuckLakeReq `json:"ducklake,omitempty"` diff --git a/controlplane/provisioning/api_test.go b/controlplane/provisioning/api_test.go index 725185a3..13e15f8e 100644 --- a/controlplane/provisioning/api_test.go +++ b/controlplane/provisioning/api_test.go @@ -112,12 +112,12 @@ func (s *fakeStore) Provision(req ProvisionRequest) error { // requires default_team_id (rejected before any write); an existing org // keeps its stored value when the field is omitted (set-only, never wipe). if _, ok := s.orgs[req.OrgID]; !ok { - if req.DefaultTeamID == "" { + if req.DefaultTeamID == 0 { return ErrDefaultTeamIDRequired } s.orgs[req.OrgID] = &configstore.Org{Name: req.OrgID, DatabaseName: req.DatabaseName} } - if req.DefaultTeamID != "" { + if req.DefaultTeamID != 0 { teamID := req.DefaultTeamID s.orgs[req.OrgID].DefaultTeamID = &teamID } @@ -215,7 +215,7 @@ func TestProvisionAutoCreatesOrg(t *testing.T) { store := newFakeStore() router := newTestRouter(store) - body := []byte(`{"database_name": "test-db", "default_team_id": "1", "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + body := []byte(`{"database_name": "test-db", "default_team_id": 1, "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/new-org/provision", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() @@ -238,7 +238,7 @@ func TestProvisionPersistsDefaultTeamID(t *testing.T) { store := newFakeStore() router := newTestRouter(store) - body := []byte(`{"database_name": "team-db", "default_team_id": "12345", "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + body := []byte(`{"database_name": "team-db", "default_team_id": 12345, "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/team-org/provision", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() @@ -254,8 +254,8 @@ func TestProvisionPersistsDefaultTeamID(t *testing.T) { if org.DefaultTeamID == nil { t.Fatal("expected default_team_id to be set, got nil") } - if *org.DefaultTeamID != "12345" { - t.Fatalf("default_team_id = %q, want %q", *org.DefaultTeamID, "12345") + if *org.DefaultTeamID != 12345 { + t.Fatalf("default_team_id = %d, want %d", *org.DefaultTeamID, 12345) } } @@ -292,7 +292,7 @@ func TestProvisionNewOrgRequiresDefaultTeamID(t *testing.T) { // stored value — the field is set-only, never wiped by omission. func TestReprovisionExistingOrgKeepsDefaultTeamID(t *testing.T) { store := newFakeStore() - teamID := "777" + teamID := int64(777) store.orgs["backfilled"] = &configstore.Org{Name: "backfilled", DefaultTeamID: &teamID} router := newTestRouter(store) @@ -526,7 +526,7 @@ func TestProvisionTransactionRollsBackOnUserFailure(t *testing.T) { store.setProvisionUserFailHook(errors.New("simulated DB write failure")) router := newTestRouter(store) - body := []byte(`{"database_name": "team-7-db", "default_team_id": "7", "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + body := []byte(`{"database_name": "team-7-db", "default_team_id": 7, "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/7/provision", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() @@ -648,7 +648,7 @@ func TestProvisionDuckLakeExternal(t *testing.T) { body := []byte(`{ "database_name": "extdl-db", - "default_team_id": "1", + "default_team_id": 1, "metadata_store": {"type": "external", "external": { "endpoint": "rds.example.us-east-1.rds.amazonaws.com", "password_aws_secret": "duckling-example-rds-password", @@ -699,7 +699,7 @@ func TestProvisionComputesS3BucketName(t *testing.T) { wantBucket := "posthog-duckling-0194d6405db400006cde48d6114c0f99-mw-prod-us" body := []byte(`{ "database_name": "db", - "default_team_id": "1", + "default_team_id": 1, "metadata_store": {"type": "cnpg-shard"}, "data_store": {"type": "s3bucket"}, "ducklake": {"enabled": true} @@ -738,7 +738,7 @@ func TestProvisionNoBucketSuffixLeavesNameEmpty(t *testing.T) { body := []byte(`{ "database_name": "db", - "default_team_id": "1", + "default_team_id": 1, "metadata_store": {"type": "cnpg-shard"}, "data_store": {"type": "s3bucket"}, "ducklake": {"enabled": true} @@ -862,7 +862,7 @@ func TestProvisionRejectsOverlongSlugOrgID(t *testing.T) { func TestProvisionAcceptsHyphenatedOrgID(t *testing.T) { store := newFakeStore() router := newTestRouter(store) - body := []byte(`{"database_name":"d","default_team_id":"1","metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}`) + body := []byte(`{"database_name":"d","default_team_id":1,"metadata_store":{"type":"cnpg-shard"},"ducklake":{"enabled":true}}`) req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/my-org-cnpg/provision", bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") rec := httptest.NewRecorder() @@ -871,3 +871,27 @@ func TestProvisionAcceptsHyphenatedOrgID(t *testing.T) { t.Fatalf("hyphenated org id should be accepted, got %d: %s", rec.Code, rec.Body.String()) } } + +// TestProvisionRejectsStringDefaultTeamID pins the wire contract: +// default_team_id is a JSON NUMBER (PostHog's integer Team.id). A quoted +// string is a decode-time 400 naming the field, and creates nothing. +func TestProvisionRejectsStringDefaultTeamID(t *testing.T) { + store := newFakeStore() + router := newTestRouter(store) + + body := []byte(`{"database_name": "strteam-db", "default_team_id": "12345", "metadata_store": {"type": "cnpg-shard"}, "ducklake": {"enabled": true}}`) + req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/strteam-org/provision", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "default_team_id") { + t.Fatalf("error should name default_team_id: %s", rec.Body.String()) + } + if _, ok := store.orgs["strteam-org"]; ok { + t.Fatal("org must not be created on a malformed default_team_id") + } +} diff --git a/controlplane/provisioning/store.go b/controlplane/provisioning/store.go index 31a5f89d..12d927a8 100644 --- a/controlplane/provisioning/store.go +++ b/controlplane/provisioning/store.go @@ -37,12 +37,13 @@ var ErrDefaultTeamIDRequired = errors.New("default_team_id is required when crea type ProvisionRequest struct { OrgID string DatabaseName string - // DefaultTeamID links the org to its default PostHog team id. REQUIRED - // when the org does not exist yet (Provision returns - // ErrDefaultTeamIDRequired otherwise); optional on re-provision of an - // existing org, where empty keeps the stored value (never a wipe). - // Prerequisite for pull-based compute billing. - DefaultTeamID string + // DefaultTeamID links the org to its default PostHog team id (an + // integer, matching PostHog's Team.id). REQUIRED when the org does not + // exist yet (Provision returns ErrDefaultTeamIDRequired otherwise); + // optional (0) on re-provision of an existing org, where it keeps the + // stored value (never a wipe). Prerequisite for pull-based compute + // billing. + DefaultTeamID int64 Warehouse *configstore.ManagedWarehouse // RootUserHash is the bcrypt hash of the freshly-generated root // password. Plaintext stays in the handler (returned to the @@ -89,7 +90,7 @@ func (s *gormStore) CreatePendingWarehouse(orgID, databaseName string, warehouse // No default_team_id on this standalone path — an existing org keeps // its column as-is; creating a NEW org through here now fails with // ErrDefaultTeamIDRequired (same invariant as Provision). - return createPendingWarehouseTx(tx, orgID, databaseName, "", warehouse) + return createPendingWarehouseTx(tx, orgID, databaseName, 0, warehouse) }) } @@ -102,7 +103,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, defaultTeamID string, warehouse *configstore.ManagedWarehouse) error { +func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName string, defaultTeamID int64, 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 (pull-based // compute billing keys usage buckets by it; all pre-existing orgs are @@ -112,7 +113,7 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName, defaultTeamID st err := tx.Where("name = ?", orgID).First(&org).Error switch { case errors.Is(err, gorm.ErrRecordNotFound): - if defaultTeamID == "" { + if defaultTeamID == 0 { return ErrDefaultTeamIDRequired } org = configstore.Org{Name: orgID, DatabaseName: databaseName, DefaultTeamID: &defaultTeamID} @@ -131,7 +132,7 @@ func createPendingWarehouseTx(tx *gorm.DB, orgID, databaseName, defaultTeamID st // If a default_team_id was supplied, persist it even when the org already // existed (the create branch above only runs on insert). Only ever set, // never cleared here, so an omitted value is a no-op rather than a wipe. - if defaultTeamID != "" { + if defaultTeamID != 0 { if err := tx.Model(&org).Update("default_team_id", defaultTeamID).Error; err != nil { return err } diff --git a/docs/design/billing-pull-api.md b/docs/design/billing-pull-api.md index b5aff9a1..6d88b22e 100644 --- a/docs/design/billing-pull-api.md +++ b/docs/design/billing-pull-api.md @@ -36,8 +36,9 @@ Internally, one row per unique key, values accumulated: - **Values:** `cpu_seconds` (vCPU-seconds), `memory_seconds` (GiB-seconds) — summed over every connection that falls in that key + minute. - `team_id` is the org's **default team** for now — a fixed value per org (from the - config-store org→team default), reported so the shape is right, but **no per-team - attribution logic** yet. A "team" is really a schema and one connection can span + config-store org→team default; an **integer**, matching PostHog's `Team.id` — + the provisioning/admin APIs and this response all carry it as a JSON number), + reported so the shape is right, but **no per-team attribution logic** yet. A "team" is really a schema and one connection can span several, so true per-team split is future work; today every bucket carries the org's default team. `query_source` (`standard` | `endpoints`) is set by a session GUC (`duckgres.query_source`), defaulting to `standard` when unset; the meter @@ -80,7 +81,7 @@ per day; a same-day window yields one). Size is reported as `cpu` (vCPU) and "date": "2026-07-01", "query_source": "endpoints" | "standard", "org_id": "org_abc", - "team_id": "12345", + "team_id": 12345, "cpu": 8, "mem_gib": 16, "cpu_seconds": 4800, @@ -162,8 +163,8 @@ sizes; stored as `NUMERIC` so grouping is exact.) (`NUMERIC`) in the key (new migration); add a single `last_acked` cursor row; add the HTTP API (aggregate-on-read into one row per key per UTC day + watermark ack) + safety GC. -- **Add:** a `default_team_id` column on the org (used as the bucket `team_id` — - fixed per org, no per-team logic yet); a `duckgres.query_source` session GUC +- **Add:** a `default_team_id` column on the org (BIGINT; used as the bucket + `team_id` — fixed per org, no per-team logic yet); a `duckgres.query_source` session GUC (`standard` | `endpoints`, default `standard`) read by the meter; a bearer secret for the API. diff --git a/tests/configstore/migrations_postgres_test.go b/tests/configstore/migrations_postgres_test.go index ca60b9f8..256c7438 100644 --- a/tests/configstore/migrations_postgres_test.go +++ b/tests/configstore/migrations_postgres_test.go @@ -34,7 +34,8 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) { requireGooseMigrationRecorded(t, db, 14) requireGooseMigrationRecorded(t, db, 15) requireGooseMigrationRecorded(t, db, 16) - requireGooseLatestVersion(t, db, 16) + requireGooseMigrationRecorded(t, db, 17) + requireGooseLatestVersion(t, db, 17) requireTableAbsent(t, db, "duckgres_schema_migrations") // Migration 000016 added the worker spawn log that feeds dynamic headroom @@ -44,8 +45,12 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) { requireColumnPresent(t, db, "duckgres_worker_spawn_log", "mem_bytes") requireColumnPresent(t, db, "duckgres_worker_spawn_log", "spawned_at") - // Migration 000013 added the nullable per-org default_team_id column. + // Migration 000013 added the nullable per-org default_team_id column; + // 000017 converted it (and the usage-bucket team_id below) to BIGINT to + // match PostHog's integer team ids. requireColumnPresent(t, db, "duckgres_orgs", "default_team_id") + requireColumnType(t, db, "duckgres_orgs", "default_team_id", "bigint") + requireColumnType(t, db, "duckgres_org_compute_usage", "team_id", "bigint") // Migration 000007 added the compute-usage billing buffer; 000015 widened // its key for pull-based billing (team_id, query_source, worker size), @@ -116,7 +121,7 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { ALTER TABLE duckgres_managed_warehouses ADD COLUMN IF NOT EXISTS iceberg_state VARCHAR(32); ALTER TABLE duckgres_org_users ADD COLUMN IF NOT EXISTS default_catalog VARCHAR(255); DROP TABLE duckgres_worker_spawn_log; - DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16); + DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17); `).Error; err != nil { t.Fatalf("downgrade baseline schema to pre-v9 shape: %v", err) } @@ -146,7 +151,8 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { requireGooseMigrationRecorded(t, upgradedDB, 14) requireGooseMigrationRecorded(t, upgradedDB, 15) requireGooseMigrationRecorded(t, upgradedDB, 16) - requireGooseLatestVersion(t, upgradedDB, 16) + requireGooseMigrationRecorded(t, upgradedDB, 17) + requireGooseLatestVersion(t, upgradedDB, 17) requireTablePresent(t, upgradedDB, "duckgres_worker_spawn_log") requireColumnDefault(t, upgradedDB, "duckgres_orgs", "max_vcpus", "0") requireColumnDefault(t, upgradedDB, "duckgres_org_users", "max_vcpus", "0") @@ -504,6 +510,28 @@ type foreignKeyMetadata struct { DeleteRule string } +func requireColumnType(t *testing.T, db *sql.DB, tableName, columnName, wantType string) { + t.Helper() + + var got string + err := db.QueryRow(` + SELECT data_type + FROM information_schema.columns + WHERE table_schema = current_schema() + AND table_name = $1 + AND column_name = $2 + `, tableName, columnName).Scan(&got) + if err != nil { + if err == sql.ErrNoRows { + t.Fatalf("%s.%s column missing", tableName, columnName) + } + t.Fatalf("query %s.%s column type: %v", tableName, columnName, err) + } + if got != wantType { + t.Fatalf("%s.%s type = %q, want %q", tableName, columnName, got, wantType) + } +} + func requireColumnDefault(t *testing.T, db *sql.DB, tableName, columnName, wantDefault string) { t.Helper() diff --git a/tests/e2e-mw-dev/harness.sh b/tests/e2e-mw-dev/harness.sh index 41dc2b60..3cc72d1c 100755 --- a/tests/e2e-mw-dev/harness.sh +++ b/tests/e2e-mw-dev/harness.sh @@ -479,7 +479,7 @@ compute_usage_pull_api() { # org password a=0 body="" while [ "$a" -lt 30 ]; do body="$(curl -fsS -H "$H" "$API/api/v1/billing/usage")" || body="" - if [ -n "$body" ] && echo "$body" | jq -e --arg o "$org" --arg t "$CNPG_DEFAULT_TEAM_ID" ' + if [ -n "$body" ] && echo "$body" | jq -e --arg o "$org" --argjson t "$CNPG_DEFAULT_TEAM_ID" ' (.usage | map(select(.org_id==$o and .team_id==$t and .query_source=="standard" and .cpu_seconds>0 and .cpu>0 and .mem_gib>0)) | length >= 1) and (.usage | map(select(.org_id==$o and .team_id==$t and .query_source=="endpoints" and .cpu_seconds>0)) | length >= 1)' >/dev/null 2>&1; then @@ -1065,9 +1065,10 @@ org_default_profile() { # org password catalog } # ---- org default_team_id (mandatory on new orgs) ---------------------------- -# default_team_id links an org to its default PostHog team id (config-store -# column, prereq for pull-based compute billing where usage buckets are keyed by -# team_id). Contract: MANDATORY when a provision creates a NEW org (400, nothing +# default_team_id links an org to its default PostHog team id (a BIGINT +# config-store column — a JSON NUMBER on the wire, matching PostHog's integer +# Team.id; prereq for pull-based compute billing where usage buckets are keyed +# by team_id). Contract: MANDATORY when a provision creates a NEW org (400, nothing # created); optional on re-provision of an existing org, where omission keeps # the stored value (set-only, never a wipe — the keep path is covered by # TestReprovisionExistingOrgKeepsDefaultTeamID, since same-id re-provision @@ -1077,7 +1078,7 @@ org_default_profile() { # org password catalog # bodies (mandatory now); GET /orgs/:id must round-trip exactly each value. # 2. Reject path: provisioning a brand-new org WITHOUT default_team_id must be # 400 naming the field, and must create nothing (org GET stays 404). -# 3. Mutate path: PUT /orgs/:id can set and clear it (""=NULL) on the EXT org +# 3. Mutate path: PUT /orgs/:id can set and clear it (0=NULL) on the EXT org # (admin escape hatch), round-tripping on GET; the provisioned value is # restored afterwards so the org stays contract-conformant. # get_org_default_team_id prints the raw JSON value ("null" when NULL) so the @@ -1113,15 +1114,15 @@ default_team_id_mandatory() { # cnpg_org ext_org log "default_team_id OK: new org without it rejected with 400, nothing created" # 3. Mutate path: PUT can set, clear ("" -> NULL), then restore on the ext org. - code="$(put_org "$ext_org" '{"default_team_id":"424242"}')" + code="$(put_org "$ext_org" '{"default_team_id":424242}')" [ "$code" = "200" ] || fail "default_team_id: PUT set -> HTTP $code: $(cat /tmp/put_org_out)" got="$(get_org_default_team_id "$ext_org")" [ "$got" = "424242" ] || fail "default_team_id: after PUT set, GET = '$got' want '424242'" - code="$(put_org "$ext_org" '{"default_team_id":""}')" + code="$(put_org "$ext_org" '{"default_team_id":0}')" [ "$code" = "200" ] || fail "default_team_id: PUT clear -> HTTP $code: $(cat /tmp/put_org_out)" got="$(get_org_default_team_id "$ext_org")" - [ "$got" = "null" ] || fail "default_team_id: after PUT clear, GET = '$got' want 'null' (empty must clear to NULL)" - code="$(put_org "$ext_org" "{\"default_team_id\":\"$EXT_DEFAULT_TEAM_ID\"}")" + [ "$got" = "null" ] || fail "default_team_id: after PUT clear, GET = '$got' want 'null' (0 must clear to NULL)" + code="$(put_org "$ext_org" "{\"default_team_id\":$EXT_DEFAULT_TEAM_ID}")" [ "$code" = "200" ] || fail "default_team_id: PUT restore -> HTTP $code: $(cat /tmp/put_org_out)" log "default_team_id OK: PUT set/clear/restore round-trips on $ext_org" } @@ -2153,13 +2154,13 @@ lifecycle_teardown_cnpg() { # org # asserts the round-trips and that omitting it on a new org is rejected. CNPG_DEFAULT_TEAM_ID='90210' CNPG_BODY='{"database_name":"'"$CNPG"'","metadata_store":{"type":"cnpg-shard"}, - "default_team_id":"'"$CNPG_DEFAULT_TEAM_ID"'", + "default_team_id":'"$CNPG_DEFAULT_TEAM_ID"', "data_store":{"type":"s3bucket"},"ducklake":{"enabled":true}}' # ---- ext duckling: external RDS metadata + DuckLake ----------------------- EXT_DEFAULT_TEAM_ID='31337' EXT_BODY='{"database_name":"'"$EXT"'", - "default_team_id":"'"$EXT_DEFAULT_TEAM_ID"'", + "default_team_id":'"$EXT_DEFAULT_TEAM_ID"', "metadata_store":{"type":"external","external":{ "endpoint":"'"$EXT_RDS_ENDPOINT"'","password_aws_secret":"'"$EXT_RDS_SECRET"'", "user":"ducklingexample","database":"ducklingexample"}}, @@ -2170,7 +2171,7 @@ EXT_BODY='{"database_name":"'"$EXT"'", # DuckLake-only: these orgs exist purely to host the worker-churn-heavy # resilience lanes, so keep their provision footprint small. res_body() { # org - printf '{"database_name":"%s","default_team_id":"1","metadata_store":{"type":"cnpg-shard"},"data_store":{"type":"s3bucket"},"ducklake":{"enabled":true}}' "$1" + printf '{"database_name":"%s","default_team_id":1,"metadata_store":{"type":"cnpg-shard"},"data_store":{"type":"s3bucket"},"ducklake":{"enabled":true}}' "$1" } # ---- per-user kill switch --------------------------------------------------- From 4414712c1793df7dd0a3757ec3fa33e95de60f9f Mon Sep 17 00:00:00 2001 From: Benjamin Knofe-Vider Date: Mon, 6 Jul 2026 17:06:55 +0200 Subject: [PATCH 2/2] Fix migration 000017 on a live schema + v8-replay test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ALTER TYPE cannot auto-cast the usage table's '' column default to BIGINT (42804) — drop the default before the type change (both directions). The v8-upgrade test now also reverses 000015/000017 on the compute-usage buffer (back to the v7 (org_id, bucket_start) shape + drain-state table) before replaying, mirroring what it already does for the org columns — replaying 000015's ''-keyed backfill onto an already-BIGINT team_id column cannot work and a real v8 database never has those columns anyway. Verified against a real local Postgres. --- .../migrations/000017_default_team_id_bigint.sql | 4 ++++ tests/configstore/migrations_postgres_test.go | 14 ++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/controlplane/configstore/migrations/000017_default_team_id_bigint.sql b/controlplane/configstore/migrations/000017_default_team_id_bigint.sql index 9c490bb0..ce24447f 100644 --- a/controlplane/configstore/migrations/000017_default_team_id_bigint.sql +++ b/controlplane/configstore/migrations/000017_default_team_id_bigint.sql @@ -15,7 +15,10 @@ ALTER TABLE duckgres_orgs -- The compute-usage bucket key mirrors the org column. team_id sits in the -- PRIMARY KEY so it must stay NOT NULL: 0 = "org had no default team" (PostHog -- team ids start at 1), replacing the old '' sentinel. +-- Drop the old '' default before the type change — Postgres refuses to +-- auto-cast a column default during ALTER TYPE (42804). ALTER TABLE duckgres_org_compute_usage + ALTER COLUMN team_id DROP DEFAULT, ALTER COLUMN team_id TYPE BIGINT USING COALESCE(NULLIF(team_id::text, ''), '0')::bigint, ALTER COLUMN team_id SET DEFAULT 0; @@ -24,6 +27,7 @@ ALTER TABLE duckgres_org_compute_usage -- Best-effort reversal: the 0 sentinel downgrades to '0' (not the old ''), -- which is fine for a transient buffer. ALTER TABLE duckgres_org_compute_usage + ALTER COLUMN team_id DROP DEFAULT, ALTER COLUMN team_id TYPE TEXT USING team_id::text, ALTER COLUMN team_id SET DEFAULT ''; diff --git a/tests/configstore/migrations_postgres_test.go b/tests/configstore/migrations_postgres_test.go index 256c7438..cc4ba1c7 100644 --- a/tests/configstore/migrations_postgres_test.go +++ b/tests/configstore/migrations_postgres_test.go @@ -121,6 +121,20 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) { ALTER TABLE duckgres_managed_warehouses ADD COLUMN IF NOT EXISTS iceberg_state VARCHAR(32); ALTER TABLE duckgres_org_users ADD COLUMN IF NOT EXISTS default_catalog VARCHAR(255); DROP TABLE duckgres_worker_spawn_log; + -- Reverse 000015 + 000017 on the compute-usage buffer (widened + -- pull-billing key → the original v7 (org_id, bucket_start) shape; + -- dropping the PK columns drops the composite PK with them), and + -- restore the push-drain state table 000015 removed. Without this + -- the replay of 000015's backfill hits a BIGINT team_id with a '' + -- comparison and fails. + ALTER TABLE duckgres_org_compute_usage + DROP COLUMN team_id, DROP COLUMN query_source, DROP COLUMN cpu, DROP COLUMN mem_gib; + ALTER TABLE duckgres_org_compute_usage ADD PRIMARY KEY (org_id, bucket_start); + DROP TABLE IF EXISTS duckgres_compute_billing_cursor; + CREATE TABLE IF NOT EXISTS duckgres_org_compute_drain_state ( + org_id TEXT PRIMARY KEY, + last_drained_bucket TIMESTAMPTZ NOT NULL + ); DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17); `).Error; err != nil { t.Fatalf("downgrade baseline schema to pre-v9 shape: %v", err)