amadan

carlos / platform Public

console: account plan assignments

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude cb194225f801325e8a322ccdf06078086ea5f137 parent bec0c78
4 files changed, +334 −0
  • internal/console/accountplans.go +149 −0
  • internal/console/accountplans_test.go +167 −0
  • internal/console/store.go +17 −0
  • internal/console/store_gate.go +1 −0
diff --git a/internal/console/accountplans.go b/internal/console/accountplans.go
new file mode 100644
index 000000000..4d34b2cfc
--- /dev/null
+++ b/internal/console/accountplans.go
@@ -0,0 +1,149 @@
+package console
+
+import (
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "time"
+
+ "github.com/carlosframework/platform/internal/plans"
+)
+
+// The sources an account's plan row can have. "default" rows are the ones
+// EnsureDefaultPlans inserts for accounts nobody has assigned.
+const (
+ PlanSourceDefault = "default"
+ PlanSourceOperator = "operator"
+ PlanSourceSponsored = "sponsored"
+ PlanSourceBilling = "billing"
+)
+
+// planDefaultActor is who a default row names as its writer.
+const planDefaultActor = "plans"
+
+// AccountPlanRow is one account's plan assignment.
+type AccountPlanRow struct {
+ AccountID int64
+ Sqid string
+ Plan string
+ Packs map[string]int
+ Source string
+ Sponsored *plans.Limits
+ Reason string
+ Revision int64
+ PublishedRevision int64
+ PublishedCatalog string
+ UpdatedBy string
+ UpdatedAt time.Time
+}
+
+const accountPlanColumns = `p.account_id, a.sqid, p.plan, p.packs, p.source, p.sponsored, p.reason,
+ p.revision, p.published_revision, p.published_catalog, p.updated_by, p.updated_at`
+
+func scanAccountPlan(sc interface{ Scan(...any) error }) (*AccountPlanRow, error) {
+ var r AccountPlanRow
+ var packs, sponsored string
+ var updated int64
+ if err := sc.Scan(&r.AccountID, &r.Sqid, &r.Plan, &packs, &r.Source, &sponsored, &r.Reason,
+ &r.Revision, &r.PublishedRevision, &r.PublishedCatalog, &r.UpdatedBy, &updated); err != nil {
+ return nil, err
+ }
+ if err := json.Unmarshal([]byte(packs), &r.Packs); err != nil {
+ return nil, err
+ }
+ if sponsored != "" {
+ r.Sponsored = &plans.Limits{}
+ if err := json.Unmarshal([]byte(sponsored), r.Sponsored); err != nil {
+ return nil, err
+ }
+ }
+ r.UpdatedAt = time.Unix(updated, 0)
+ return &r, nil
+}
+
+// AccountPlan is the account's plan row, or nil when it has none.
+func (s *Store) AccountPlan(accountID int64) (*AccountPlanRow, error) {
+ r, err := scanAccountPlan(s.db.QueryRow(`SELECT `+accountPlanColumns+`
+ FROM account_plans p JOIN accounts a ON a.id = p.account_id WHERE p.account_id = ?`, accountID))
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ return r, err
+}
+
+// SetAccountPlan writes the account's plan and returns its new revision.
+// It validates nothing: callers check the plan against the catalog first.
+func (s *Store) SetAccountPlan(accountID int64, plan string, packs map[string]int, source string,
+ sponsored *plans.Limits, reason, by string, now time.Time) (int64, error) {
+ if packs == nil {
+ packs = map[string]int{}
+ }
+ pj, err := json.Marshal(packs)
+ if err != nil {
+ return 0, err
+ }
+ sj := ""
+ if sponsored != nil {
+ b, err := json.Marshal(sponsored)
+ if err != nil {
+ return 0, err
+ }
+ sj = string(b)
+ }
+ var rev int64
+ err = s.db.QueryRow(`INSERT INTO account_plans
+ (account_id, plan, packs, source, sponsored, reason, revision, updated_by, updated_at)
+ VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)
+ ON CONFLICT(account_id) DO UPDATE SET plan = excluded.plan, packs = excluded.packs,
+ source = excluded.source, sponsored = excluded.sponsored, reason = excluded.reason,
+ updated_by = excluded.updated_by, updated_at = excluded.updated_at, revision = revision + 1
+ RETURNING revision`,
+ accountID, plan, string(pj), source, sj, reason, by, now.Unix()).Scan(&rev)
+ return rev, err
+}
+
+// EnsureDefaultPlans gives every live account without a plan row the
+// default plan, and reports how many rows it inserted.
+func (s *Store) EnsureDefaultPlans(defaultPlan string, now time.Time) (int, error) {
+ res, err := s.db.Exec(`INSERT INTO account_plans
+ (account_id, plan, packs, source, revision, updated_by, updated_at)
+ SELECT id, ?, '{}', ?, 1, ?, ? FROM accounts
+ WHERE deleted_at = 0 AND id NOT IN (SELECT account_id FROM account_plans)`,
+ defaultPlan, PlanSourceDefault, planDefaultActor, now.Unix())
+ if err != nil {
+ return 0, err
+ }
+ n, err := res.RowsAffected()
+ return int(n), err
+}
+
+// PlansToPublish lists live accounts whose bucket record is behind their
+// row: an unpublished revision, or a record from another catalog.
+func (s *Store) PlansToPublish(catalogID string, limit int) ([]AccountPlanRow, error) {
+ rows, err := s.db.Query(`SELECT `+accountPlanColumns+`
+ FROM account_plans p JOIN accounts a ON a.id = p.account_id
+ WHERE a.deleted_at = 0 AND (p.published_revision < p.revision OR p.published_catalog != ?)
+ ORDER BY p.account_id LIMIT ?`, catalogID, limit)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []AccountPlanRow
+ for rows.Next() {
+ r, err := scanAccountPlan(rows)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, *r)
+ }
+ return out, rows.Err()
+}
+
+// MarkPlanPublished records that the bucket carries revision under
+// catalogID. A revision that is no longer current marks nothing: the newer
+// one still needs publishing.
+func (s *Store) MarkPlanPublished(accountID, revision int64, catalogID string) error {
+ _, err := s.db.Exec(`UPDATE account_plans SET published_revision = ?, published_catalog = ?
+ WHERE account_id = ? AND revision = ?`, revision, catalogID, accountID, revision)
+ return err
+}
diff --git a/internal/console/accountplans_test.go b/internal/console/accountplans_test.go
new file mode 100644
index 000000000..3aea77d54
--- /dev/null
+++ b/internal/console/accountplans_test.go
@@ -0,0 +1,167 @@
+package console
+
+import (
+ "path/filepath"
+ "reflect"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/plans"
+)
+
+func planStore(t *testing.T, accounts ...string) (*Store, map[string]int64) {
+ t.Helper()
+ st, err := Open(filepath.Join(t.TempDir(), "console.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { st.Close() })
+ ids := map[string]int64{}
+ for i, sqid := range accounts {
+ id := int64(100 + i)
+ if err := st.CreateAccount(id, sqid, sqid, sqid+"@example.com", time.Unix(1_700_000_000, 0)); err != nil {
+ t.Fatal(err)
+ }
+ ids[sqid] = id
+ }
+ return st, ids
+}
+
+var planNow = time.Unix(1_800_000_000, 0)
+
+func TestAccountPlanAbsentIsNil(t *testing.T) {
+ st, ids := planStore(t, "aaa")
+ row, err := st.AccountPlan(ids["aaa"])
+ if err != nil || row != nil {
+ t.Fatalf("got %+v, %v; want nil, nil", row, err)
+ }
+}
+
+func TestSetAccountPlanBumpsRevision(t *testing.T) {
+ st, ids := planStore(t, "aaa")
+ id := ids["aaa"]
+ sp := &plans.Limits{RunningThings: 12}
+ rev, err := st.SetAccountPlan(id, "personal", map[string]int{"apps": 2}, "sponsored", sp, "existing user", "ops@example.com", planNow)
+ if err != nil || rev != 1 {
+ t.Fatalf("first set: rev %d, %v", rev, err)
+ }
+ rev, err = st.SetAccountPlan(id, "personal", map[string]int{"apps": 3}, "sponsored", sp, "more apps", "ops@example.com", planNow)
+ if err != nil || rev != 2 {
+ t.Fatalf("second set: rev %d, %v", rev, err)
+ }
+ row, err := st.AccountPlan(id)
+ if err != nil || row == nil {
+ t.Fatalf("read back: %+v, %v", row, err)
+ }
+ if row.Sqid != "aaa" || row.Plan != "personal" || row.Source != "sponsored" || row.Reason != "more apps" ||
+ row.UpdatedBy != "ops@example.com" || row.Revision != 2 || !row.UpdatedAt.Equal(planNow) {
+ t.Fatalf("row %+v", row)
+ }
+ if !reflect.DeepEqual(row.Packs, map[string]int{"apps": 3}) {
+ t.Fatalf("packs %v", row.Packs)
+ }
+ if row.Sponsored == nil || *row.Sponsored != *sp {
+ t.Fatalf("sponsored %+v", row.Sponsored)
+ }
+ if _, err := st.SetAccountPlan(id, "free", nil, "operator", nil, "back to free", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ row, _ = st.AccountPlan(id)
+ if row.Sponsored != nil || len(row.Packs) != 0 {
+ t.Fatalf("cleared row kept sponsored %+v / packs %v", row.Sponsored, row.Packs)
+ }
+}
+
+func TestEnsureDefaultPlansInsertsOnlyMissing(t *testing.T) {
+ st, ids := planStore(t, "aaa", "bbb", "ccc")
+ if _, err := st.SetAccountPlan(ids["bbb"], "personal", nil, "operator", nil, "paid", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ n, err := st.EnsureDefaultPlans("free", planNow)
+ if err != nil || n != 2 {
+ t.Fatalf("inserted %d, %v; want 2", n, err)
+ }
+ b, _ := st.AccountPlan(ids["bbb"])
+ if b.Plan != "personal" || b.Revision != 1 {
+ t.Fatalf("existing row changed: %+v", b)
+ }
+ a, _ := st.AccountPlan(ids["aaa"])
+ if a == nil || a.Plan != "free" || a.Source != "default" || a.Revision != 1 || a.UpdatedBy != "plans" {
+ t.Fatalf("default row %+v", a)
+ }
+ if n, err := st.EnsureDefaultPlans("free", planNow); err != nil || n != 0 {
+ t.Fatalf("second ensure inserted %d, %v", n, err)
+ }
+}
+
+func TestEnsureDefaultPlansSkipsDeletedAccounts(t *testing.T) {
+ st, ids := planStore(t, "aaa", "gone")
+ if _, err := st.db.Exec(`UPDATE accounts SET deleted_at = ? WHERE id = ?`, planNow.Unix(), ids["gone"]); err != nil {
+ t.Fatal(err)
+ }
+ if n, err := st.EnsureDefaultPlans("free", planNow); err != nil || n != 1 {
+ t.Fatalf("inserted %d, %v; want 1", n, err)
+ }
+ if row, _ := st.AccountPlan(ids["gone"]); row != nil {
+ t.Fatalf("deleted account got a row: %+v", row)
+ }
+}
+
+func TestPlansToPublish(t *testing.T) {
+ st, ids := planStore(t, "aaa", "bbb", "ccc")
+ if _, err := st.EnsureDefaultPlans("free", planNow); err != nil {
+ t.Fatal(err)
+ }
+ rows, err := st.PlansToPublish("v1@abc", 10)
+ if err != nil || len(rows) != 3 {
+ t.Fatalf("got %d rows, %v; want 3", len(rows), err)
+ }
+ if rows, _ := st.PlansToPublish("v1@abc", 2); len(rows) != 2 {
+ t.Fatalf("limit ignored: %d rows", len(rows))
+ }
+ if err := st.MarkPlanPublished(ids["aaa"], 1, "v1@abc"); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ = st.PlansToPublish("v1@abc", 10)
+ for _, r := range rows {
+ if r.Sqid == "aaa" {
+ t.Fatal("published row still due for the same catalog")
+ }
+ }
+ if len(rows) != 2 {
+ t.Fatalf("got %d rows, want 2", len(rows))
+ }
+ rows, _ = st.PlansToPublish("v2@def", 10)
+ if len(rows) != 3 {
+ t.Fatalf("catalog change: got %d rows, want 3", len(rows))
+ }
+ if _, err := st.db.Exec(`UPDATE accounts SET deleted_at = ? WHERE id = ?`, planNow.Unix(), ids["bbb"]); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ = st.PlansToPublish("v2@def", 10)
+ if len(rows) != 2 {
+ t.Fatalf("deleted account still due: got %d rows, want 2", len(rows))
+ }
+}
+
+func TestMarkPlanPublishedIgnoresAStaleRevision(t *testing.T) {
+ st, ids := planStore(t, "aaa")
+ id := ids["aaa"]
+ for range 2 {
+ if _, err := st.SetAccountPlan(id, "free", nil, "operator", nil, "r", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := st.MarkPlanPublished(id, 1, "c"); err != nil {
+ t.Fatal(err)
+ }
+ if row, _ := st.AccountPlan(id); row.PublishedRevision != 0 {
+ t.Fatalf("stale publish marked: %d", row.PublishedRevision)
+ }
+ if err := st.MarkPlanPublished(id, 2, "c"); err != nil {
+ t.Fatal(err)
+ }
+ if row, _ := st.AccountPlan(id); row.PublishedRevision != 2 || row.PublishedCatalog != "c" {
+ t.Fatalf("published %d / %q", row.PublishedRevision, row.PublishedCatalog)
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index 7bfe00f00..21523c90e 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -381,6 +381,23 @@ func (s *Store) migrate() error {
warned INTEGER NOT NULL DEFAULT 0,
held INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (host, op, name)
+);`,
+ // Plans (plans-caps-packs spec §4, §7): one row per account naming
+ // its plan and packs. revision moves on every change;
+ // published_revision/published_catalog say which revision, under
+ // which catalog, the bucket's limits record already carries.
+ `CREATE TABLE IF NOT EXISTS account_plans (
+ account_id INTEGER PRIMARY KEY,
+ plan TEXT NOT NULL,
+ packs TEXT NOT NULL DEFAULT '{}',
+ source TEXT NOT NULL,
+ sponsored TEXT NOT NULL DEFAULT '',
+ reason TEXT NOT NULL DEFAULT '',
+ revision INTEGER NOT NULL,
+ published_revision INTEGER NOT NULL DEFAULT 0,
+ published_catalog TEXT NOT NULL DEFAULT '',
+ updated_by TEXT NOT NULL,
+ updated_at INTEGER NOT NULL
);`,
} {
if _, err := s.db.Exec(stmt); err != nil {
diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go
index bda729a56..6fe9ae657 100644
--- a/internal/console/store_gate.go
+++ b/internal/console/store_gate.go
@@ -208,6 +208,7 @@ var claimSQLTables = map[string]claimSQLTable{
"accounts": {NonApp: "the tenancy primitive itself; names no app"},
"memberships": {NonApp: "account membership, keyed by account and address; names no app"},
"invites": {NonApp: "account invites, keyed by account and address; names no app"},
+ "account_plans": {NonApp: "an account's plan and packs, keyed by account; names no app"},
"alert_seen": {NonApp: "operator alert memory, keyed by box, rule and address"},
"claim_loss_pages": {NonApp: "claim recovery page attempts per loss record and condition, and scan progress; no app state or lifecycle authority"},
"move_pages": {NonApp: "the move pass's last page attempt per finding (rule names the host and the move); no app column"},