amadan

carlos / platform Public

Record inert operator-reviewed sponsorship enrollment

claude 4a9649218ce5d6f0c009adf1de2f2b9277cae8bc parent 9fa0af0
3 files changed, +1353 −1
  • internal/console/sponsoredenrollment.go +573 −0
  • internal/console/sponsoredenrollment_test.go +776 −0
  • internal/console/store.go +4 −1

CI passed — run details

diff --git a/internal/console/sponsoredenrollment.go b/internal/console/sponsoredenrollment.go
new file mode 100644
index 000000000..9593c1870
--- /dev/null
+++ b/internal/console/sponsoredenrollment.go
@@ -0,0 +1,573 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "database/sql"
+ "encoding/binary"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "io"
+ "sort"
+ "strings"
+ "time"
+ "unicode/utf8"
+
+ "github.com/carlosframework/platform/internal/domains"
+)
+
+const sponsoredEnrollmentDecision = "carloku-existing-2026-10-07"
+const sponsoredEnrollmentMaxBody = 4 << 20
+
+var (
+ errSponsoredEnrollmentInvalid = errors.New("sponsored_enrollment_invalid")
+ errSponsoredEnrollmentDenied = errors.New("sponsored_enrollment_denied")
+ errSponsoredEnrollmentConflict = errors.New("sponsored_enrollment_conflict")
+ errSponsoredEnrollmentNotFound = errors.New("sponsored_enrollment_not_found")
+ errSponsoredEnrollmentCorrupt = errors.New("sponsored_enrollment_corrupt")
+ errSponsoredEnrollmentTooLarge = errors.New("sponsored_enrollment_too_large")
+)
+
+type sponsoredEnrollmentChoiceKind int
+
+const (
+ sponsoredEnroll sponsoredEnrollmentChoiceKind = iota + 1
+ sponsoredExclude
+ sponsoredNeedsReview
+)
+
+type sponsoredEnrollmentOutcome int
+
+const (
+ sponsoredAdded sponsoredEnrollmentOutcome = iota + 1
+ sponsoredAlreadyRecorded
+ sponsoredExcludedByOperator
+ sponsoredAwaitingReview
+ sponsoredExcludedMissing
+ sponsoredExcludedRetired
+)
+
+type sponsoredEnrollmentRequest struct {
+ OperationID, PreviewDigest, DecisionReference, Note string
+ CoverageAcknowledged bool
+ Items []sponsoredEnrollmentChoice
+}
+type sponsoredEnrollmentChoice struct {
+ AccountID int64
+ Sqid string
+ Choice sponsoredEnrollmentChoiceKind
+ Reason string
+ ConfirmSystemAccount bool
+}
+type sponsoredEnrollmentResult struct {
+ Historical sponsoredEnrollmentReceipt
+ Replayed bool
+}
+type sponsoredEnrollmentReceipt struct {
+ Version int
+ Actor string
+ Request sponsoredEnrollmentRequest
+ RecordedAt int64
+ Rows []sponsoredEnrollmentObservation
+}
+type sponsoredEnrollmentObservation struct {
+ AccountID int64
+ Sqid string
+ Outcome sponsoredEnrollmentOutcome
+ Facts *sponsoredEnrollmentFactsV1
+ FirstEnrollmentOperationID string
+}
+
+// These fields pin the durable v1 encoding independently of inventory changes.
+type sponsoredEnrollmentFactsV1 struct {
+ ID int64
+ Sqid, Name string
+ StoredCreatedAt, DeletedAt int64
+ Owners []string
+ Billing sponsoredEnrollmentBillingV1
+ CheckoutPresent, BillingRetirement bool
+ Outbox sponsoredEnrollmentOutboxV1
+ OpenBillingReviews, EntitlementReviews int64
+}
+type sponsoredEnrollmentBillingV1 struct {
+ Present, House, LiveMode, CustomerPresent bool
+ Subscription, Status string
+}
+type sponsoredEnrollmentOutboxV1 struct {
+ Present, LiveMode, SerialOwed bool
+ Revision, PublishedRevision int64
+ Digest, State string
+}
+
+const sponsoredEnrollmentReceiptSQL = `CREATE TABLE sponsored_enrollment_receipts (operation_id TEXT PRIMARY KEY CHECK(length(operation_id)=64), actor TEXT NOT NULL, request_digest TEXT NOT NULL CHECK(length(request_digest)=64), body BLOB NOT NULL, body_digest TEXT NOT NULL CHECK(length(body_digest)=64), recorded_at INTEGER NOT NULL CHECK(recorded_at>0)) WITHOUT ROWID`
+const sponsoredEnrollmentMemberSQL = `CREATE TABLE sponsored_cohort_memberships (account_id INTEGER PRIMARY KEY CHECK(account_id>0), sqid TEXT NOT NULL, first_operation_id TEXT NOT NULL CHECK(length(first_operation_id)=64)) WITHOUT ROWID`
+
+var sponsoredEnrollmentSchema = []struct{ name, kind, table, definition string }{
+ {"sponsored_enrollment_receipts", "table", "sponsored_enrollment_receipts", sponsoredEnrollmentReceiptSQL},
+ {"sponsored_cohort_memberships", "table", "sponsored_cohort_memberships", sponsoredEnrollmentMemberSQL},
+ {"sponsored_receipts_no_update", "trigger", "sponsored_enrollment_receipts", `CREATE TRIGGER sponsored_receipts_no_update BEFORE UPDATE ON sponsored_enrollment_receipts BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_immutable'); END`},
+ {"sponsored_receipts_no_delete", "trigger", "sponsored_enrollment_receipts", `CREATE TRIGGER sponsored_receipts_no_delete BEFORE DELETE ON sponsored_enrollment_receipts BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_immutable'); END`},
+ {"sponsored_receipts_no_replace", "trigger", "sponsored_enrollment_receipts", `CREATE TRIGGER sponsored_receipts_no_replace BEFORE INSERT ON sponsored_enrollment_receipts WHEN EXISTS(SELECT 1 FROM sponsored_enrollment_receipts WHERE operation_id=NEW.operation_id) BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_immutable'); END`},
+ {"sponsored_members_no_update", "trigger", "sponsored_cohort_memberships", `CREATE TRIGGER sponsored_members_no_update BEFORE UPDATE ON sponsored_cohort_memberships BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_immutable'); END`},
+ {"sponsored_members_no_delete", "trigger", "sponsored_cohort_memberships", `CREATE TRIGGER sponsored_members_no_delete BEFORE DELETE ON sponsored_cohort_memberships BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_immutable'); END`},
+ {"sponsored_members_no_replace", "trigger", "sponsored_cohort_memberships", `CREATE TRIGGER sponsored_members_no_replace BEFORE INSERT ON sponsored_cohort_memberships WHEN EXISTS(SELECT 1 FROM sponsored_cohort_memberships WHERE account_id=NEW.account_id) BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_immutable'); END`},
+ {"sponsored_members_parent", "trigger", "sponsored_cohort_memberships", `CREATE TRIGGER sponsored_members_parent BEFORE INSERT ON sponsored_cohort_memberships WHEN NOT EXISTS(SELECT 1 FROM sponsored_enrollment_receipts WHERE operation_id=NEW.first_operation_id) BEGIN SELECT RAISE(ABORT,'sponsored_enrollment_corrupt'); END`},
+}
+
+func (s *Store) migrateSponsoredEnrollment() error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ present, err := sponsoredEnrollmentSchemaPresent(tx)
+ if err != nil {
+ return err
+ }
+ if !present {
+ if err := createSponsoredEnrollmentSchemaTx(tx); err != nil {
+ return err
+ }
+ }
+ return tx.Commit()
+}
+
+func createSponsoredEnrollmentSchemaTx(tx *sql.Tx) error {
+ for _, item := range sponsoredEnrollmentSchema {
+ if _, err := tx.Exec(item.definition); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func sponsoredEnrollmentSchemaPresent(tx *sql.Tx) (bool, error) {
+ expected := make(map[string]int)
+ for i, item := range sponsoredEnrollmentSchema {
+ expected[item.name] = i
+ }
+ protectedTable := func(name string) bool {
+ return strings.EqualFold(name, "sponsored_enrollment_receipts") || strings.EqualFold(name, "sponsored_cohort_memberships")
+ }
+ seen := make(map[string]bool)
+ namedTables := 0
+ invalid := false
+ for _, catalog := range []string{"sqlite_master", "sqlite_temp_master"} {
+ rows, err := tx.Query(`SELECT type,name,tbl_name,sql FROM ` + catalog)
+ if err != nil {
+ return false, err
+ }
+ for rows.Next() {
+ var kind, name, table string
+ var definition sql.NullString
+ if err := rows.Scan(&kind, &name, &table, &definition); err != nil {
+ rows.Close()
+ return false, err
+ }
+ index, named := expected[strings.ToLower(name)]
+ if !named && !protectedTable(table) {
+ continue
+ }
+ if catalog == "sqlite_temp_master" {
+ rows.Close()
+ return false, errSponsoredEnrollmentCorrupt
+ }
+ if protectedTable(name) {
+ namedTables++
+ }
+ if !named {
+ if kind == "index" && strings.HasPrefix(name, "sqlite_autoindex_") && !definition.Valid {
+ continue
+ }
+ invalid = true
+ continue
+ }
+ want := sponsoredEnrollmentSchema[index]
+ if name != want.name || kind != want.kind || table != want.table || !definition.Valid || definition.String != want.definition {
+ invalid = true
+ continue
+ }
+ seen[name] = true
+ }
+ if err := closeSponsoredRows(rows); err != nil {
+ return false, err
+ }
+ }
+ rows, err := tx.Query(`SELECT schema,name,wr FROM pragma_table_list`)
+ if err != nil {
+ return false, err
+ }
+ tables := 0
+ for rows.Next() {
+ var schema, name string
+ var wr int
+ if err := rows.Scan(&schema, &name, &wr); err != nil {
+ rows.Close()
+ return false, err
+ }
+ if protectedTable(name) {
+ if schema != "main" || wr != 1 || !seen[name] {
+ rows.Close()
+ return false, errSponsoredEnrollmentCorrupt
+ }
+ tables++
+ }
+ }
+ if err := closeSponsoredRows(rows); err != nil {
+ return false, err
+ }
+ if namedTables == 0 && tables == 0 && !invalid {
+ return false, nil
+ }
+ if invalid || len(seen) != len(sponsoredEnrollmentSchema) || tables != 2 {
+ return false, errSponsoredEnrollmentCorrupt
+ }
+ return true, nil
+}
+
+func sponsoredEnrollmentHex(v string) bool {
+ if len(v) != 64 {
+ return false
+ }
+ for _, c := range v {
+ if !(c >= '0' && c <= '9' || c >= 'a' && c <= 'f') {
+ return false
+ }
+ }
+ return true
+}
+func sponsoredEnrollmentActor(actor string) bool {
+ return actor == domains.NormalizeAddress(actor) && domains.ValidAddress(actor) && validReviewText(actor, billingReviewActorMax)
+}
+func validateSponsoredEnrollmentRequest(req sponsoredEnrollmentRequest) error {
+ if !sponsoredEnrollmentHex(req.OperationID) || !sponsoredEnrollmentHex(req.PreviewDigest) || req.DecisionReference != sponsoredEnrollmentDecision || !req.CoverageAcknowledged || !validReviewText(req.Note, billingReviewNoteMax) || len(req.Items) < 1 || len(req.Items) > 1000 {
+ return errSponsoredEnrollmentInvalid
+ }
+ var previous int64
+ for _, item := range req.Items {
+ if item.AccountID <= previous || item.Sqid != domains.Sqid(item.AccountID) || !validReviewText(item.Reason, billingReviewNoteMax) || item.Choice < sponsoredEnroll || item.Choice > sponsoredNeedsReview {
+ return errSponsoredEnrollmentInvalid
+ }
+ if item.ConfirmSystemAccount != (item.AccountID == domains.ReservedAccountID && item.Choice == sponsoredEnroll) {
+ return errSponsoredEnrollmentInvalid
+ }
+ previous = item.AccountID
+ }
+ return nil
+}
+func sponsoredEnrollmentDigest(domain string, fields ...[]byte) string {
+ h := sha256.New()
+ h.Write([]byte("carlos-sponsored-enrollment-" + domain + "-v1\x00"))
+ var size [8]byte
+ for _, field := range fields {
+ binary.BigEndian.PutUint64(size[:], uint64(len(field)))
+ h.Write(size[:])
+ h.Write(field)
+ }
+ return hex.EncodeToString(h.Sum(nil))
+}
+func sponsoredEnrollmentRequestDigest(actor string, req sponsoredEnrollmentRequest) (string, error) {
+ body, err := json.Marshal(req)
+ if err != nil {
+ return "", err
+ }
+ return sponsoredEnrollmentDigest("request", []byte(actor), body), nil
+}
+func authorizeSponsoredEnrollment(ctx context.Context, tx *sql.Tx, actor string) error {
+ var allowed bool
+ if err := tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM users WHERE address=? AND operator<>0)`, actor).Scan(&allowed); err != nil {
+ return err
+ }
+ if !allowed {
+ return errSponsoredEnrollmentDenied
+ }
+ return nil
+}
+
+func (s *Store) enrollSponsoredCandidates(ctx context.Context, actor string, req sponsoredEnrollmentRequest, now time.Time) (sponsoredEnrollmentResult, error) {
+ var zero sponsoredEnrollmentResult
+ if s == nil || s.db == nil || ctx == nil || !sponsoredEnrollmentActor(actor) {
+ return zero, errSponsoredEnrollmentInvalid
+ }
+ if err := validateSponsoredEnrollmentRequest(req); err != nil {
+ return zero, err
+ }
+ tx, err := s.db.BeginTx(ctx, nil)
+ if err != nil {
+ return zero, err
+ }
+ defer tx.Rollback()
+ result, err := enrollSponsoredCandidatesTx(ctx, tx, actor, req, now, nil)
+ if err != nil {
+ return zero, err
+ }
+ if err := tx.Commit(); err != nil {
+ return zero, err
+ }
+ return result, nil
+}
+func enrollSponsoredCandidatesTx(ctx context.Context, tx *sql.Tx, actor string, req sponsoredEnrollmentRequest, now time.Time, afterReserve func() error) (sponsoredEnrollmentResult, error) {
+ var zero sponsoredEnrollmentResult
+ if ctx == nil || tx == nil || !sponsoredEnrollmentActor(actor) {
+ return zero, errSponsoredEnrollmentInvalid
+ }
+ if err := validateSponsoredEnrollmentRequest(req); err != nil {
+ return zero, err
+ }
+ reserved, err := tx.ExecContext(ctx, `UPDATE sponsored_enrollment_receipts SET operation_id=operation_id WHERE 0`)
+ if err != nil {
+ return zero, err
+ }
+ count, err := reserved.RowsAffected()
+ if err != nil {
+ return zero, err
+ }
+ if count != 0 {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ if afterReserve != nil {
+ if err := afterReserve(); err != nil {
+ return zero, err
+ }
+ }
+ if err := authorizeSponsoredEnrollment(ctx, tx, actor); err != nil {
+ return zero, err
+ }
+ digest, err := sponsoredEnrollmentRequestDigest(actor, req)
+ if err != nil {
+ return zero, err
+ }
+ old, err := readSponsoredEnrollmentTx(ctx, tx, req.OperationID)
+ if err == nil {
+ prior, err := sponsoredEnrollmentRequestDigest(old.Actor, old.Request)
+ if err != nil {
+ return zero, err
+ }
+ if prior != digest {
+ return zero, errSponsoredEnrollmentConflict
+ }
+ return sponsoredEnrollmentResult{Historical: old, Replayed: true}, nil
+ }
+ if !errors.Is(err, errSponsoredEnrollmentNotFound) {
+ return zero, err
+ }
+ if now.IsZero() || now.Unix() <= 0 {
+ return zero, errSponsoredEnrollmentInvalid
+ }
+ inventory, err := readSponsoredCandidatesTx(ctx, tx, nil)
+ if err != nil {
+ return zero, err
+ }
+ byID := make(map[int64]sponsoredCandidateRow, len(inventory.Rows))
+ for _, row := range inventory.Rows {
+ byID[row.ID] = row
+ }
+ receipt := sponsoredEnrollmentReceipt{Version: 1, Actor: actor, Request: req, RecordedAt: now.UTC().Truncate(time.Second).Unix(), Rows: make([]sponsoredEnrollmentObservation, 0, len(req.Items))}
+ for _, item := range req.Items {
+ first, err := sponsoredEnrollmentMember(ctx, tx, item.AccountID, item.Sqid)
+ if err != nil {
+ return zero, err
+ }
+ row := sponsoredEnrollmentObservation{AccountID: item.AccountID, Sqid: item.Sqid, FirstEnrollmentOperationID: first}
+ if fact, ok := byID[item.AccountID]; ok {
+ if fact.Sqid != item.Sqid {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ row.Facts = sponsoredEnrollmentCopyFacts(fact)
+ if !validSponsoredEnrollmentFacts(row.Facts) {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ }
+ row.Outcome = sponsoredEnrollmentClassify(item, row.Facts, first)
+ if row.Outcome == sponsoredAdded {
+ row.FirstEnrollmentOperationID = req.OperationID
+ }
+ receipt.Rows = append(receipt.Rows, row)
+ }
+ body, err := json.Marshal(receipt)
+ if err != nil {
+ return zero, err
+ }
+ if len(body) > sponsoredEnrollmentMaxBody {
+ return zero, errSponsoredEnrollmentTooLarge
+ }
+ result, err := tx.ExecContext(ctx, `INSERT INTO sponsored_enrollment_receipts(operation_id,actor,request_digest,body,body_digest,recorded_at) VALUES(?,?,?,?,?,?)`, req.OperationID, actor, digest, body, sponsoredEnrollmentDigest("receipt", body), receipt.RecordedAt)
+ if err := sponsoredEnrollmentInserted(result, err); err != nil {
+ return zero, err
+ }
+ for _, row := range receipt.Rows {
+ if row.Outcome != sponsoredAdded {
+ continue
+ }
+ result, err := tx.ExecContext(ctx, `INSERT INTO sponsored_cohort_memberships(account_id,sqid,first_operation_id) VALUES(?,?,?)`, row.AccountID, row.Sqid, req.OperationID)
+ if err := sponsoredEnrollmentInserted(result, err); err != nil {
+ return zero, err
+ }
+ }
+ return sponsoredEnrollmentResult{Historical: receipt}, nil
+}
+func sponsoredEnrollmentInserted(result sql.Result, err error) error {
+ if err != nil {
+ return err
+ }
+ n, err := result.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n != 1 {
+ return errSponsoredEnrollmentCorrupt
+ }
+ return nil
+}
+func sponsoredEnrollmentMember(ctx context.Context, tx *sql.Tx, id int64, sqid string) (string, error) {
+ var got, first string
+ err := tx.QueryRowContext(ctx, `SELECT sqid,first_operation_id FROM sponsored_cohort_memberships WHERE account_id=?`, id).Scan(&got, &first)
+ if errors.Is(err, sql.ErrNoRows) {
+ return "", nil
+ }
+ if err != nil {
+ return "", err
+ }
+ if id < 1 || got != domains.Sqid(id) || got != sqid || !sponsoredEnrollmentHex(first) {
+ return "", errSponsoredEnrollmentCorrupt
+ }
+ var parent bool
+ if err := tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM sponsored_enrollment_receipts WHERE operation_id=?)`, first).Scan(&parent); err != nil {
+ return "", err
+ }
+ if !parent {
+ return "", errSponsoredEnrollmentCorrupt
+ }
+ return first, nil
+}
+func sponsoredEnrollmentClassify(item sponsoredEnrollmentChoice, facts *sponsoredEnrollmentFactsV1, first string) sponsoredEnrollmentOutcome {
+ switch {
+ case facts == nil:
+ return sponsoredExcludedMissing
+ case facts.DeletedAt != 0 || facts.BillingRetirement:
+ return sponsoredExcludedRetired
+ case item.Choice == sponsoredExclude:
+ return sponsoredExcludedByOperator
+ case item.Choice == sponsoredNeedsReview:
+ return sponsoredAwaitingReview
+ case first != "":
+ return sponsoredAlreadyRecorded
+ default:
+ return sponsoredAdded
+ }
+}
+func sponsoredEnrollmentCopyFacts(in sponsoredCandidateRow) *sponsoredEnrollmentFactsV1 {
+ return &sponsoredEnrollmentFactsV1{ID: in.ID, Sqid: in.Sqid, Name: in.Name, StoredCreatedAt: in.StoredCreatedAt, DeletedAt: in.DeletedAt, Owners: append([]string{}, in.Owners...), Billing: sponsoredEnrollmentBillingV1{in.Billing.Present, in.Billing.House, in.Billing.LiveMode, in.Billing.CustomerPresent, in.Billing.Subscription, in.Billing.Status}, CheckoutPresent: in.CheckoutPresent, BillingRetirement: in.BillingRetirement, Outbox: sponsoredEnrollmentOutboxV1{in.Outbox.Present, in.Outbox.LiveMode, in.Outbox.SerialOwed, in.Outbox.Revision, in.Outbox.PublishedRevision, in.Outbox.Digest, in.Outbox.State}, OpenBillingReviews: in.OpenBillingReviews, EntitlementReviews: in.EntitlementReviews}
+}
+func validSponsoredEnrollmentFacts(f *sponsoredEnrollmentFactsV1) bool {
+ if f.ID < 1 || f.Sqid != domains.Sqid(f.ID) || f.Owners == nil || !sort.StringsAreSorted(f.Owners) {
+ return false
+ }
+ for _, v := range append([]string{f.Name, f.Billing.Subscription, f.Billing.Status, f.Outbox.Digest, f.Outbox.State}, f.Owners...) {
+ if !utf8.ValidString(v) {
+ return false
+ }
+ }
+ return true
+}
+func (s *Store) readSponsoredEnrollment(ctx context.Context, actor, operationID string) (sponsoredEnrollmentReceipt, error) {
+ var zero sponsoredEnrollmentReceipt
+ if s == nil || s.db == nil || ctx == nil || !sponsoredEnrollmentActor(actor) || !sponsoredEnrollmentHex(operationID) {
+ return zero, errSponsoredEnrollmentInvalid
+ }
+ tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
+ if err != nil {
+ return zero, err
+ }
+ defer tx.Rollback()
+ if err := authorizeSponsoredEnrollment(ctx, tx, actor); err != nil {
+ return zero, err
+ }
+ out, err := readSponsoredEnrollmentTx(ctx, tx, operationID)
+ if err != nil {
+ return zero, err
+ }
+ if err := tx.Commit(); err != nil {
+ return zero, err
+ }
+ return out, nil
+}
+func readSponsoredEnrollmentTx(ctx context.Context, tx *sql.Tx, operationID string) (sponsoredEnrollmentReceipt, error) {
+ var zero, out sponsoredEnrollmentReceipt
+ var actor, digest, bodyDigest, bodyType string
+ var body []byte
+ var recorded int64
+ err := tx.QueryRowContext(ctx, `SELECT actor,request_digest,substr(body,1,?),body_digest,recorded_at,typeof(body) FROM sponsored_enrollment_receipts WHERE operation_id=?`, sponsoredEnrollmentMaxBody+1, operationID).Scan(&actor, &digest, &body, &bodyDigest, &recorded, &bodyType)
+ if errors.Is(err, sql.ErrNoRows) {
+ return zero, errSponsoredEnrollmentNotFound
+ }
+ if err != nil {
+ return zero, err
+ }
+ if bodyType != "blob" || len(body) > sponsoredEnrollmentMaxBody || sponsoredEnrollmentDigest("receipt", body) != bodyDigest {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ dec := json.NewDecoder(bytes.NewReader(body))
+ dec.DisallowUnknownFields()
+ if err := dec.Decode(&out); err != nil {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ var trailing any
+ if err := dec.Decode(&trailing); err != io.EOF {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ encoded, err := json.Marshal(out)
+ if err != nil || !bytes.Equal(encoded, body) {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ if out.Version != 1 || out.Actor != actor || !sponsoredEnrollmentActor(actor) || out.Request.OperationID != operationID || out.RecordedAt != recorded || recorded <= 0 || validateSponsoredEnrollmentRequest(out.Request) != nil || len(out.Rows) != len(out.Request.Items) {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ computed, err := sponsoredEnrollmentRequestDigest(out.Actor, out.Request)
+ if err != nil || computed != digest {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ added := 0
+ for i, row := range out.Rows {
+ item := out.Request.Items[i]
+ if row.AccountID != item.AccountID || row.Sqid != item.Sqid {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ if row.Facts != nil && (!validSponsoredEnrollmentFacts(row.Facts) || row.Facts.ID != row.AccountID || row.Facts.Sqid != row.Sqid) {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ first := row.FirstEnrollmentOperationID
+ classificationFirst := first
+ if row.Outcome == sponsoredAdded {
+ classificationFirst = ""
+ added++
+ if first != operationID {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ } else if first == operationID {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ if row.Outcome != sponsoredEnrollmentClassify(item, row.Facts, classificationFirst) {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ if first != "" {
+ actual, err := sponsoredEnrollmentMember(ctx, tx, row.AccountID, row.Sqid)
+ if err != nil {
+ return zero, err
+ }
+ if actual != first {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ }
+ }
+ var count int
+ if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM sponsored_cohort_memberships WHERE first_operation_id=?`, operationID).Scan(&count); err != nil {
+ return zero, err
+ }
+ if count != added {
+ return zero, errSponsoredEnrollmentCorrupt
+ }
+ return out, nil
+}
diff --git a/internal/console/sponsoredenrollment_test.go b/internal/console/sponsoredenrollment_test.go
new file mode 100644
index 000000000..8dfc8341c
--- /dev/null
+++ b/internal/console/sponsoredenrollment_test.go
@@ -0,0 +1,776 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+)
+
+const enrollmentActor = "sponsor@example.test"
+
+var enrollmentNow = time.Date(2026, 10, 9, 18, 0, 0, 0, time.UTC)
+
+func enrollmentContext(t *testing.T) context.Context {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+ t.Cleanup(cancel)
+ return ctx
+}
+func enrollmentStore(t *testing.T) *Store {
+ t.Helper()
+ s := openSponsoredTestStore(t, "enrollment")
+ if err := s.SetOperator(enrollmentActor, true); err != nil {
+ t.Fatal(err)
+ }
+ return s
+}
+func enrollmentRequest(op int, ids ...int64) sponsoredEnrollmentRequest {
+ r := sponsoredEnrollmentRequest{OperationID: fmt.Sprintf("%064x", op), PreviewDigest: strings.Repeat("a", 64), DecisionReference: sponsoredEnrollmentDecision, Note: "reviewed", CoverageAcknowledged: true, Items: []sponsoredEnrollmentChoice{}}
+ for _, id := range ids {
+ r.Items = append(r.Items, sponsoredEnrollmentChoice{AccountID: id, Sqid: domains.Sqid(id), Choice: sponsoredEnroll, Reason: "existing account", ConfirmSystemAccount: id == 1})
+ }
+ return r
+}
+func enrollmentRun(t *testing.T, s *Store, r sponsoredEnrollmentRequest) sponsoredEnrollmentResult {
+ t.Helper()
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, r, enrollmentNow)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return v
+}
+func enrollmentExistingSnapshot(t *testing.T, s *Store) []sponsoredTableSnapshot {
+ t.Helper()
+ var out []sponsoredTableSnapshot
+ for _, v := range sponsoredLogicalSnapshot(t, s.db) {
+ if v.Name != "sponsored_enrollment_receipts" && v.Name != "sponsored_cohort_memberships" {
+ out = append(out, v)
+ }
+ }
+ return out
+}
+func enrollmentZero(t *testing.T, v sponsoredEnrollmentResult, err error) {
+ t.Helper()
+ if err == nil || !reflect.DeepEqual(v, sponsoredEnrollmentResult{}) {
+ t.Fatalf("result=%#v error=%v", v, err)
+ }
+}
+
+func TestSponsoredEnrollmentMixedRosterAndEvidenceOnly(t *testing.T) {
+ s := enrollmentStore(t)
+ seedSponsoredCoverage(t, s, false)
+ sponsoredInsertAccount(t, s.db, 8, "review", 0, 0)
+ first := enrollmentRun(t, s, enrollmentRequest(1, 2))
+ if first.Replayed {
+ t.Fatal("first replayed")
+ }
+ before := enrollmentExistingSnapshot(t, s)
+ req := enrollmentRequest(2, 1, 2, 3, 4, 5, 6, 7, 8)
+ req.Items[2].Choice = sponsoredExclude
+ req.Items[7].Choice = sponsoredNeedsReview
+ got := enrollmentRun(t, s, req)
+ want := []sponsoredEnrollmentOutcome{sponsoredAdded, sponsoredAlreadyRecorded, sponsoredExcludedByOperator, sponsoredExcludedRetired, sponsoredExcludedRetired, sponsoredAdded, sponsoredExcludedMissing, sponsoredAwaitingReview}
+ for i, row := range got.Historical.Rows {
+ if row.Outcome != want[i] || row.AccountID != req.Items[i].AccountID {
+ t.Fatalf("row %d: %#v", i, row)
+ }
+ }
+ f := got.Historical.Rows[5].Facts
+ if !f.CheckoutPresent || !f.Billing.CustomerPresent || f.Billing.Subscription != "sub_full" || f.OpenBillingReviews != 2 || f.EntitlementReviews != 2 || !f.Outbox.SerialOwed || f.Outbox.PublishedRevision != 5 {
+ t.Fatalf("facts %#v", f)
+ }
+ if after := enrollmentExistingSnapshot(t, s); !reflect.DeepEqual(before, after) {
+ t.Fatal("modified existing tables")
+ }
+ read, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, req.OperationID)
+ if err != nil || !reflect.DeepEqual(read, got.Historical) {
+ t.Fatalf("read %v %#v", err, read)
+ }
+ if err := s.migrate(); err != nil {
+ t.Fatal(err)
+ }
+ read, err = s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, req.OperationID)
+ if err != nil || !reflect.DeepEqual(read, got.Historical) {
+ t.Fatal("reopen-chain changed receipt", err)
+ }
+}
+
+func TestSponsoredEnrollmentReplayRetirementMissingAndAddOnly(t *testing.T) {
+ s := enrollmentStore(t)
+ for _, id := range []int64{2, 3, 4} {
+ sponsoredInsertAccount(t, s.db, id, "account", 0, 0)
+ }
+ req := enrollmentRequest(10, 2, 3)
+ got := enrollmentRun(t, s, req)
+ sponsoredExec(t, s.db, `UPDATE accounts SET deleted_at=9 WHERE id=2`)
+ sponsoredExec(t, s.db, `INSERT INTO billing_retirements(account_id,started_at) VALUES(3,9)`)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ replay, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, req, time.Time{})
+ if err != nil || !replay.Replayed || !reflect.DeepEqual(replay.Historical, got.Historical) {
+ t.Fatalf("replay %#v %v", replay, err)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("replay wrote")
+ }
+ newer := enrollmentRun(t, s, enrollmentRequest(11, 2, 3))
+ for _, row := range newer.Historical.Rows {
+ if row.Outcome != sponsoredExcludedRetired || row.FirstEnrollmentOperationID != req.OperationID {
+ t.Fatalf("retired %#v", row)
+ }
+ }
+ sponsoredExec(t, s.db, `DELETE FROM accounts WHERE id=2`)
+ missing := enrollmentRun(t, s, enrollmentRequest(12, 2))
+ if missing.Historical.Rows[0].Outcome != sponsoredExcludedMissing || missing.Historical.Rows[0].FirstEnrollmentOperationID != req.OperationID {
+ t.Fatal("lost missing history")
+ }
+ excluded := enrollmentRequest(13, 4)
+ excluded.Items[0].Choice = sponsoredExclude
+ old := enrollmentRun(t, s, excluded)
+ if old.Historical.Rows[0].FirstEnrollmentOperationID != "" {
+ t.Fatal("unexpected link")
+ }
+ enrollmentRun(t, s, enrollmentRequest(14, 4))
+ read, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, excluded.OperationID)
+ if err != nil || !reflect.DeepEqual(read, old.Historical) {
+ t.Fatal("blank historical link invalidated", err)
+ }
+ for _, choice := range []sponsoredEnrollmentChoiceKind{sponsoredExclude, sponsoredNeedsReview} {
+ r := enrollmentRequest(20+int(choice), 4)
+ r.Items[0].Choice = choice
+ v := enrollmentRun(t, s, r)
+ if v.Historical.Rows[0].FirstEnrollmentOperationID != fmt.Sprintf("%064x", 14) {
+ t.Fatal("replaced first membership")
+ }
+ }
+}
+
+func TestSponsoredEnrollmentRequestValidationAndReplayConflict(t *testing.T) {
+ s := enrollmentStore(t)
+ sponsoredInsertAccount(t, s.db, 2, "account", 0, 0)
+ req := enrollmentRequest(30, 2)
+ enrollmentRun(t, s, req)
+ mutations := []func(*sponsoredEnrollmentRequest){
+ func(r *sponsoredEnrollmentRequest) { r.Note = "changed" }, func(r *sponsoredEnrollmentRequest) { r.PreviewDigest = strings.Repeat("b", 64) }, func(r *sponsoredEnrollmentRequest) { r.Items[0].Reason = "changed" }, func(r *sponsoredEnrollmentRequest) { r.Items[0].Choice = sponsoredExclude },
+ }
+ for _, mutate := range mutations {
+ r := req
+ r.Items = append([]sponsoredEnrollmentChoice{}, req.Items...)
+ mutate(&r)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, r, enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !errors.Is(err, errSponsoredEnrollmentConflict) {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("conflict wrote")
+ }
+ }
+ invalid := []sponsoredEnrollmentRequest{enrollmentRequest(31), enrollmentRequest(32, 2, 2), enrollmentRequest(33, 3, 2), enrollmentRequest(34, 0)}
+ for _, mutate := range []func(*sponsoredEnrollmentRequest){func(r *sponsoredEnrollmentRequest) { r.OperationID = "x" }, func(r *sponsoredEnrollmentRequest) { r.PreviewDigest = "x" }, func(r *sponsoredEnrollmentRequest) { r.DecisionReference = "other" }, func(r *sponsoredEnrollmentRequest) { r.CoverageAcknowledged = false }, func(r *sponsoredEnrollmentRequest) { r.Items[0].Sqid = "wrong" }, func(r *sponsoredEnrollmentRequest) { r.Items[0].Choice = 99 }, func(r *sponsoredEnrollmentRequest) { r.Items[0].Reason = "" }, func(r *sponsoredEnrollmentRequest) { r.Note = strings.Repeat("x", 257) }, func(r *sponsoredEnrollmentRequest) { r.Items[0].ConfirmSystemAccount = true }} {
+ r := enrollmentRequest(35, 2)
+ mutate(&r)
+ invalid = append(invalid, r)
+ }
+ ops := enrollmentRequest(36, 1)
+ ops.Items[0].ConfirmSystemAccount = false
+ invalid = append(invalid, ops)
+ many := enrollmentRequest(37)
+ for i := int64(1); i <= 1001; i++ {
+ many.Items = append(many.Items, enrollmentRequest(1, i).Items[0])
+ }
+ invalid = append(invalid, many)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ for _, r := range invalid {
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, r, enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !errors.Is(err, errSponsoredEnrollmentInvalid) {
+ t.Fatalf("want invalid: %v", err)
+ }
+ }
+ for _, actor := range []string{"Sponsor@example.test", " sponsor@example.test", "bad", ""} {
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), actor, enrollmentRequest(40, 2), enrollmentNow)
+ enrollmentZero(t, v, err)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("invalid request wrote")
+ }
+}
+
+func TestSponsoredEnrollmentEscapingAndReceiptLimit(t *testing.T) {
+ s := enrollmentStore(t)
+ for k, reason := range []string{strings.Repeat("<&>", 85) + "<", strings.Repeat("\u2028", 85) + "<"} {
+ r := enrollmentRequest(50 + k)
+ for i := int64(2); i <= 1001; i++ {
+ item := enrollmentRequest(1, i).Items[0]
+ item.Reason = reason
+ r.Items = append(r.Items, item)
+ }
+ v := enrollmentRun(t, s, r)
+ for _, row := range v.Historical.Rows {
+ if row.Outcome != sponsoredExcludedMissing {
+ t.Fatal("all-excluded result", row)
+ }
+ }
+ var members int
+ if err := s.db.QueryRow(`SELECT count(*) FROM sponsored_cohort_memberships`).Scan(&members); err != nil || members != 0 {
+ t.Fatal("all-excluded wrote membership", members, err)
+ }
+ }
+ sponsoredInsertAccount(t, s.db, 2, strings.Repeat("x", sponsoredEnrollmentMaxBody), 0, 0)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, enrollmentRequest(55, 2), enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !errors.Is(err, errSponsoredEnrollmentTooLarge) {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("oversize wrote")
+ }
+}
+
+func TestSponsoredEnrollmentAtomicRollback(t *testing.T) {
+ for _, fail := range []string{"receipt", "first", "later"} {
+ t.Run(fail, func(t *testing.T) {
+ s := enrollmentStore(t)
+ for _, id := range []int64{2, 3} {
+ sponsoredInsertAccount(t, s.db, id, "account", 0, 0)
+ }
+ q := `CREATE TRIGGER fail_enrollment BEFORE INSERT ON sponsored_enrollment_receipts BEGIN SELECT RAISE(ABORT,'injected'); END`
+ if fail != "receipt" {
+ id := 2
+ if fail == "later" {
+ id = 3
+ }
+ q = fmt.Sprintf(`CREATE TRIGGER fail_enrollment BEFORE INSERT ON sponsored_cohort_memberships WHEN NEW.account_id=%d BEGIN SELECT RAISE(ABORT,'injected'); END`, id)
+ }
+ sponsoredExec(t, s.db, q)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, enrollmentRequest(60, 2, 3), enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("partial commit")
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentWriterReservationDirect(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "writers.db")
+ reader := openSponsoredStoreAt(t, path)
+ writer := openSponsoredStoreAt(t, path)
+ if err := reader.SetOperator(enrollmentActor, true); err != nil {
+ t.Fatal(err)
+ }
+ sponsoredInsertAccount(t, reader.db, 2, "account", 0, 0)
+ enrollmentRun(t, reader, enrollmentRequest(70, 2))
+ before := sponsoredLogicalSnapshot(t, reader.db)
+ ctx := enrollmentContext(t)
+ tx, err := reader.db.BeginTx(ctx, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tx.Rollback()
+ got, err := enrollSponsoredCandidatesTx(ctx, tx, enrollmentActor, enrollmentRequest(70, 2), enrollmentNow, func() error {
+ bounded, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
+ defer cancel()
+ _, err := writer.db.ExecContext(bounded, `INSERT INTO users (address,created_at,operator) VALUES(?,?,?) ON CONFLICT(address) DO UPDATE SET operator=excluded.operator`, enrollmentActor, enrollmentNow.Unix(), 0)
+ if err == nil {
+ return errors.New("second writer succeeded under reservation")
+ }
+ if !errors.Is(err, context.DeadlineExceeded) && !strings.Contains(strings.ToLower(err.Error()), "locked") && !strings.Contains(err.Error(), "SQLITE_BUSY") {
+ return err
+ }
+ return nil
+ })
+ if err != nil || !got.Replayed {
+ t.Fatalf("reservation: result=%#v err=%v", got, err)
+ }
+ if err := tx.Commit(); err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, reader.db)) {
+ t.Fatal("reservation or replay changed bytes")
+ }
+ if err := writer.SetOperator(enrollmentActor, false); err != nil {
+ t.Fatal(err)
+ }
+ v, err := reader.enrollSponsoredCandidates(ctx, enrollmentActor, enrollmentRequest(70, 2), enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !errors.Is(err, errSponsoredEnrollmentDenied) {
+ t.Fatal(err)
+ }
+ if _, err := reader.readSponsoredEnrollment(ctx, enrollmentActor, strings.Repeat("f", 64)); !errors.Is(err, errSponsoredEnrollmentDenied) {
+ t.Fatal("operation disclosed", err)
+ }
+ if err := writer.SetOperator("other@example.test", true); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := reader.readSponsoredEnrollment(ctx, "other@example.test", fmt.Sprintf("%064x", 70)); err != nil {
+ t.Fatal(err)
+ }
+ v, err = reader.enrollSponsoredCandidates(ctx, "other@example.test", enrollmentRequest(70, 2), enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !errors.Is(err, errSponsoredEnrollmentConflict) {
+ t.Fatal(err)
+ }
+}
+
+func TestSponsoredEnrollmentConcurrentOperations(t *testing.T) {
+ for _, kind := range []string{"same", "different", "changed"} {
+ t.Run(kind, func(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "race.db")
+ a := openSponsoredStoreAt(t, path)
+ b := openSponsoredStoreAt(t, path)
+ if err := a.SetOperator(enrollmentActor, true); err != nil {
+ t.Fatal(err)
+ }
+ sponsoredInsertAccount(t, a.db, 2, "account", 0, 0)
+ reqs := []sponsoredEnrollmentRequest{enrollmentRequest(80, 2), enrollmentRequest(80, 2)}
+ if kind == "different" {
+ reqs[1].OperationID = fmt.Sprintf("%064x", 81)
+ }
+ if kind == "changed" {
+ reqs[1].Note = "different"
+ }
+ ctx := enrollmentContext(t)
+ start := make(chan struct{})
+ var wg sync.WaitGroup
+ results := make([]sponsoredEnrollmentResult, 2)
+ errs := make([]error, 2)
+ for i, s := range []*Store{a, b} {
+ wg.Add(1)
+ go func(i int, s *Store) {
+ defer wg.Done()
+ <-start
+ results[i], errs[i] = s.enrollSponsoredCandidates(ctx, enrollmentActor, reqs[i], enrollmentNow)
+ }(i, s)
+ }
+ close(start)
+ wg.Wait()
+ var count int
+ if err := a.db.QueryRow(`SELECT count(*) FROM sponsored_cohort_memberships`).Scan(&count); err != nil || count != 1 {
+ t.Fatal("membership count", count, err)
+ }
+ switch kind {
+ case "same":
+ if errs[0] != nil || errs[1] != nil || results[0].Replayed == results[1].Replayed {
+ t.Fatal(results, errs)
+ }
+ case "different":
+ if errs[0] != nil || errs[1] != nil {
+ t.Fatal(errs)
+ }
+ x, y := results[0].Historical.Rows[0].Outcome, results[1].Historical.Rows[0].Outcome
+ if !((x == sponsoredAdded && y == sponsoredAlreadyRecorded) || (y == sponsoredAdded && x == sponsoredAlreadyRecorded)) {
+ t.Fatal(x, y)
+ }
+ case "changed":
+ if !((errs[0] == nil && errors.Is(errs[1], errSponsoredEnrollmentConflict)) || (errs[1] == nil && errors.Is(errs[0], errSponsoredEnrollmentConflict))) {
+ t.Fatal(errs)
+ }
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentSchemaRefusals(t *testing.T) {
+ cases := map[string]string{
+ "partial": `DROP TABLE sponsored_cohort_memberships`,
+ "different trigger": `DROP TRIGGER sponsored_receipts_no_update; CREATE TRIGGER sponsored_receipts_no_update BEFORE UPDATE ON sponsored_enrollment_receipts BEGIN SELECT RAISE(ABORT,'different'); END`,
+ "extra trigger": `CREATE TRIGGER extra_sponsored BEFORE INSERT ON sponsored_cohort_memberships BEGIN SELECT 1; END`,
+ "extra index": `CREATE INDEX extra_sponsored ON sponsored_cohort_memberships(sqid)`,
+ "missing trigger": `DROP TRIGGER sponsored_members_parent`,
+ "wrong table": `DROP TRIGGER sponsored_members_parent; CREATE TRIGGER sponsored_members_parent BEFORE INSERT ON accounts BEGIN SELECT 1; END`,
+ "rowid": `DROP TABLE sponsored_cohort_memberships; CREATE TABLE sponsored_cohort_memberships(account_id INTEGER PRIMARY KEY,sqid TEXT NOT NULL,first_operation_id TEXT NOT NULL)`,
+ }
+ for name, q := range cases {
+ t.Run(name, func(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "schema.db")
+ s, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sponsoredExec(t, s.db, q)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ if err := s.Close(); err != nil {
+ t.Fatal(err)
+ }
+ reopened, err := Open(path)
+ if reopened != nil {
+ reopened.Close()
+ }
+ if !errors.Is(err, errSponsoredEnrollmentCorrupt) {
+ t.Fatal("did not refuse", err)
+ }
+ dsn, err := consoleSQLiteDSN(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ db, err := sql.Open("sqlite", dsn)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, db)) {
+ t.Fatal("repaired corrupt schema")
+ }
+ })
+ }
+ for _, shadow := range []string{"sponsored_enrollment_receipts", "SpOnSoReD_CoHoRt_MeMbErShIpS", "SPONSORED_MEMBERS_PARENT"} {
+ t.Run("temp "+shadow, func(t *testing.T) {
+ s := enrollmentStore(t)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ sponsoredExec(t, s.db, `CREATE TEMP TABLE "`+shadow+`"(x)`)
+ if err := s.migrate(); !errors.Is(err, errSponsoredEnrollmentCorrupt) {
+ t.Fatal("temp shadow accepted", err)
+ }
+ sponsoredExec(t, s.db, `DROP TABLE temp."`+shadow+`"`)
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("temp refusal changed main")
+ }
+ if err := s.migrate(); err != nil {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentImmutableRoutes(t *testing.T) {
+ s := enrollmentStore(t)
+ for _, id := range []int64{2, 3} {
+ sponsoredInsertAccount(t, s.db, id, "account", 0, 0)
+ }
+ enrollmentRun(t, s, enrollmentRequest(90, 2))
+ enrollmentRun(t, s, enrollmentRequest(91, 3))
+ sponsoredExec(t, s.db, `PRAGMA recursive_triggers=OFF`)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ queries := []string{
+ `UPDATE sponsored_enrollment_receipts SET actor='changed'`, `DELETE FROM sponsored_enrollment_receipts`,
+ `INSERT OR REPLACE INTO sponsored_enrollment_receipts SELECT * FROM sponsored_enrollment_receipts`, `REPLACE INTO sponsored_enrollment_receipts SELECT * FROM sponsored_enrollment_receipts`,
+ `UPDATE OR REPLACE sponsored_enrollment_receipts SET operation_id=(SELECT min(operation_id) FROM sponsored_enrollment_receipts)`,
+ `INSERT INTO sponsored_enrollment_receipts SELECT * FROM sponsored_enrollment_receipts WHERE true ON CONFLICT(operation_id) DO UPDATE SET actor='changed'`,
+ `UPDATE sponsored_cohort_memberships SET sqid='changed'`, `DELETE FROM sponsored_cohort_memberships`,
+ `INSERT OR REPLACE INTO sponsored_cohort_memberships SELECT * FROM sponsored_cohort_memberships`, `REPLACE INTO sponsored_cohort_memberships SELECT * FROM sponsored_cohort_memberships`,
+ `UPDATE OR REPLACE sponsored_cohort_memberships SET account_id=2`,
+ `INSERT INTO sponsored_cohort_memberships SELECT * FROM sponsored_cohort_memberships WHERE true ON CONFLICT(account_id) DO UPDATE SET sqid='changed'`,
+ `INSERT INTO sponsored_cohort_memberships VALUES(4,'bad','ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff')`,
+ }
+ for _, table := range []string{"sponsored_enrollment_receipts", "sponsored_cohort_memberships"} {
+ for _, alias := range []string{"rowid", "_rowid_", "oid"} {
+ queries = append(queries, `UPDATE OR REPLACE `+table+` SET `+alias+`=1`, `INSERT OR REPLACE INTO `+table+`(`+alias+`) VALUES(1)`)
+ }
+ }
+ for _, q := range queries {
+ if _, err := s.db.Exec(q); err == nil {
+ t.Fatalf("mutation succeeded: %s", q)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("immutable evidence changed", q)
+ }
+ }
+ if _, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, fmt.Sprintf("%064x", 90)); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestSponsoredEnrollmentCanonicalCorruption(t *testing.T) {
+ for _, name := range []string{"case", "duplicate", "newline", "digest", "columns", "version", "order", "added link", "count", "nonadded self"} {
+ t.Run(name, func(t *testing.T) {
+ s := enrollmentStore(t)
+ for _, id := range []int64{2, 3} {
+ sponsoredInsertAccount(t, s.db, id, "account", 0, 0)
+ }
+ req := enrollmentRequest(100, 2, 3)
+ original := enrollmentRun(t, s, req).Historical
+ var body []byte
+ if err := s.db.QueryRow(`SELECT body FROM sponsored_enrollment_receipts WHERE operation_id=?`, req.OperationID).Scan(&body); err != nil {
+ t.Fatal(err)
+ }
+ sponsoredExec(t, s.db, `DROP TRIGGER sponsored_receipts_no_update`)
+ switch name {
+ case "case":
+ body = bytes.Replace(body, []byte(`"Version"`), []byte(`"version"`), 1)
+ case "duplicate":
+ body = bytes.Replace(body, []byte(`"Version":1`), []byte(`"Version":1,"Version":1`), 1)
+ case "newline":
+ body = append(body, '\n')
+ case "version":
+ original.Version = 2
+ body, _ = json.Marshal(original)
+ case "order":
+ original.Rows[0], original.Rows[1] = original.Rows[1], original.Rows[0]
+ body, _ = json.Marshal(original)
+ case "added link":
+ original.Rows[0].FirstEnrollmentOperationID = strings.Repeat("e", 64)
+ body, _ = json.Marshal(original)
+ case "count":
+ sponsoredExec(t, s.db, `DROP TRIGGER sponsored_members_no_delete`)
+ sponsoredExec(t, s.db, `DELETE FROM sponsored_cohort_memberships WHERE account_id=2`)
+ case "nonadded self":
+ original.Rows[0].Outcome = sponsoredAlreadyRecorded
+ body, _ = json.Marshal(original)
+ }
+ sponsoredExec(t, s.db, `UPDATE sponsored_enrollment_receipts SET body=?,body_digest=? WHERE operation_id=?`, body, sponsoredEnrollmentDigest("receipt", body), req.OperationID)
+ if name == "digest" {
+ sponsoredExec(t, s.db, `UPDATE sponsored_enrollment_receipts SET request_digest=?`, strings.Repeat("0", 64))
+ }
+ if name == "columns" {
+ sponsoredExec(t, s.db, `UPDATE sponsored_enrollment_receipts SET recorded_at=recorded_at+1`)
+ }
+ got, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, req.OperationID)
+ if !errors.Is(err, errSponsoredEnrollmentCorrupt) || !reflect.DeepEqual(got, sponsoredEnrollmentReceipt{}) {
+ t.Fatalf("corrupt accepted %#v %v", got, err)
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentContextAndSourceCorruption(t *testing.T) {
+ s := enrollmentStore(t)
+ sponsoredInsertAccount(t, s.db, 2, "account", 0, 0)
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ v, err := s.enrollSponsoredCandidates(ctx, enrollmentActor, enrollmentRequest(110, 2), enrollmentNow)
+ enrollmentZero(t, v, err)
+ tx, err := s.db.BeginTx(enrollmentContext(t), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel = context.WithCancel(context.Background())
+ v, err = enrollSponsoredCandidatesTx(ctx, tx, enrollmentActor, enrollmentRequest(111, 2), enrollmentNow, func() error { cancel(); return nil })
+ enrollmentZero(t, v, err)
+ tx.Rollback()
+ for _, q := range []string{`UPDATE accounts SET created_at='x' WHERE id=2`, `UPDATE accounts SET created_at=0,sqid='wrong' WHERE id=2`} {
+ sponsoredExec(t, s.db, q)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, enrollmentRequest(112, 2), enrollmentNow)
+ enrollmentZero(t, v, err)
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("source corruption wrote")
+ }
+ }
+}
+
+func TestSponsoredEnrollmentMigrationRollsBackNaturalDDLCollision(t *testing.T) {
+ for _, kind := range []string{"trigger", "table"} {
+ t.Run(kind, func(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "ddl.db")
+ s, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sponsoredExec(t, s.db, `DROP TABLE sponsored_cohort_memberships`)
+ sponsoredExec(t, s.db, `DROP TABLE sponsored_enrollment_receipts`)
+ sponsoredExec(t, s.db, `CREATE TABLE enrollment_unrelated(x)`)
+ sponsoredExec(t, s.db, `INSERT INTO enrollment_unrelated VALUES(7)`)
+ if kind == "trigger" {
+ sponsoredExec(t, s.db, `CREATE TRIGGER sponsored_members_parent BEFORE INSERT ON enrollment_unrelated BEGIN SELECT 1; END`)
+ } else {
+ sponsoredExec(t, s.db, `CREATE TABLE sponsored_receipts_no_update(x)`)
+ sponsoredExec(t, s.db, `INSERT INTO sponsored_receipts_no_update VALUES(9)`)
+ }
+ before := sponsoredLogicalSnapshot(t, s.db)
+ if err := s.Close(); err != nil {
+ t.Fatal(err)
+ }
+ opened, err := Open(path)
+ if opened != nil {
+ opened.Close()
+ }
+ if !errors.Is(err, errSponsoredEnrollmentCorrupt) {
+ t.Fatalf("reserved-name collision must be corrupt on first Open: %v", err)
+ }
+ dsn, err := consoleSQLiteDSN(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ db, err := sql.Open("sqlite", dsn)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, db)) {
+ t.Fatal("collision changed database")
+ }
+ var count int
+ if err := db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE tbl_name IN ('sponsored_enrollment_receipts','sponsored_cohort_memberships')`).Scan(&count); err != nil || count != 0 {
+ t.Fatal("new objects survived", count, err)
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentSchemaCreationRollsBackMidDDL(t *testing.T) {
+ s := enrollmentStore(t)
+ sponsoredExec(t, s.db, `DROP TABLE sponsored_cohort_memberships`)
+ sponsoredExec(t, s.db, `DROP TABLE sponsored_enrollment_receipts`)
+ sponsoredExec(t, s.db, `CREATE TABLE enrollment_unrelated(x)`)
+ sponsoredExec(t, s.db, `INSERT INTO enrollment_unrelated VALUES(7)`)
+ sponsoredExec(t, s.db, `CREATE TRIGGER sponsored_members_parent BEFORE INSERT ON enrollment_unrelated BEGIN SELECT 1; END`)
+ before := sponsoredLogicalSnapshot(t, s.db)
+ tx, err := s.db.Begin()
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer tx.Rollback()
+ if err := createSponsoredEnrollmentSchemaTx(tx); err == nil {
+ t.Fatal("late DDL collision accepted")
+ }
+ var count int
+ if err := tx.QueryRow(`SELECT count(*) FROM sqlite_master WHERE tbl_name IN ('sponsored_enrollment_receipts','sponsored_cohort_memberships')`).Scan(&count); err != nil || count != len(sponsoredEnrollmentSchema)-1 {
+ t.Fatal("collision did not occur after earlier DDL", count, err)
+ }
+ if err := tx.Rollback(); err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(before, sponsoredLogicalSnapshot(t, s.db)) {
+ t.Fatal("partial DDL committed")
+ }
+ if err := s.db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE tbl_name IN ('sponsored_enrollment_receipts','sponsored_cohort_memberships')`).Scan(&count); err != nil || count != 0 {
+ t.Fatal("new objects survived", count, err)
+ }
+}
+
+func TestSponsoredEnrollmentParentExistenceWithoutParentDecode(t *testing.T) {
+ s := enrollmentStore(t)
+ sponsoredInsertAccount(t, s.db, 2, "account", 0, 0)
+ first := enrollmentRequest(120, 2)
+ enrollmentRun(t, s, first)
+ sponsoredExec(t, s.db, `DROP TRIGGER sponsored_receipts_no_update`)
+ sponsoredExec(t, s.db, `UPDATE sponsored_enrollment_receipts SET body=? WHERE operation_id=?`, []byte("not canonical"), first.OperationID)
+ next := enrollmentRequest(121, 2)
+ v := enrollmentRun(t, s, next)
+ if v.Historical.Rows[0].Outcome != sponsoredAlreadyRecorded {
+ t.Fatal("not already recorded")
+ }
+ if _, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, next.OperationID); err != nil {
+ t.Fatal("traversed parent body", err)
+ }
+ if _, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, first.OperationID); !errors.Is(err, errSponsoredEnrollmentCorrupt) {
+ t.Fatal("parent self-validation absent", err)
+ }
+ sponsoredExec(t, s.db, `DROP TRIGGER sponsored_receipts_no_delete`)
+ sponsoredExec(t, s.db, `DELETE FROM sponsored_enrollment_receipts WHERE operation_id=?`, first.OperationID)
+ if _, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, next.OperationID); !errors.Is(err, errSponsoredEnrollmentCorrupt) {
+ t.Fatal("missing parent accepted", err)
+ }
+}
+
+func TestSponsoredEnrollmentRetirementBetweenPreviewAndEnrollment(t *testing.T) {
+ for _, kind := range []string{"deleted", "pending billing", "committed billing"} {
+ t.Run(kind, func(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "retirement.db")
+ reader := openSponsoredStoreAt(t, path)
+ writer := openSponsoredStoreAt(t, path)
+ if err := reader.SetOperator(enrollmentActor, true); err != nil {
+ t.Fatal(err)
+ }
+ for _, id := range []int64{2, 3} {
+ sponsoredInsertAccount(t, reader.db, id, "account", 0, 0)
+ }
+ ctx := enrollmentContext(t)
+ before, err := reader.readSponsoredCandidates(ctx)
+ if err != nil || before.Rows[0].DeletedAt != 0 || before.Rows[0].BillingRetirement {
+ t.Fatal("preview", err)
+ }
+ switch kind {
+ case "deleted":
+ _, err = writer.db.ExecContext(ctx, `UPDATE accounts SET deleted_at=1 WHERE id=2`)
+ case "pending billing":
+ _, err = writer.db.ExecContext(ctx, `INSERT INTO billing_retirements(account_id,started_at) VALUES(2,1)`)
+ case "committed billing":
+ _, err = writer.db.ExecContext(ctx, `INSERT INTO billing_retirements(account_id,started_at,committed_at) VALUES(2,1,2)`)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ v, err := reader.enrollSponsoredCandidates(ctx, enrollmentActor, enrollmentRequest(130, 2, 3), enrollmentNow)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if v.Historical.Rows[0].Outcome != sponsoredExcludedRetired || v.Historical.Rows[1].Outcome != sponsoredAdded {
+ t.Fatalf("outcomes %#v", v.Historical.Rows)
+ }
+ var n int
+ if err := reader.db.QueryRowContext(ctx, `SELECT count(*) FROM sponsored_cohort_memberships WHERE account_id=2`).Scan(&n); err != nil || n != 0 {
+ t.Fatal("retired enrolled", n, err)
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentReceiptExactSizeBoundaries(t *testing.T) {
+ for _, delta := range []int{-1, 0, 1} {
+ t.Run(fmt.Sprint(delta), func(t *testing.T) {
+ s := enrollmentStore(t)
+ sponsoredInsertAccount(t, s.db, 2, "", 0, 0)
+ req := enrollmentRequest(140, 2)
+ fact := sponsoredEnrollmentCopyFacts(readSponsoredTestInventory(t, s).Rows[0])
+ want := sponsoredEnrollmentReceipt{Version: 1, Actor: enrollmentActor, Request: req, RecordedAt: enrollmentNow.Unix(), Rows: []sponsoredEnrollmentObservation{{AccountID: 2, Sqid: domains.Sqid(2), Outcome: sponsoredAdded, Facts: fact, FirstEnrollmentOperationID: req.OperationID}}}
+ empty, err := json.Marshal(want)
+ if err != nil {
+ t.Fatal(err)
+ }
+ name := strings.Repeat("x", sponsoredEnrollmentMaxBody-len(empty)+delta)
+ sponsoredExec(t, s.db, `UPDATE accounts SET name=? WHERE id=2`, name)
+ v, err := s.enrollSponsoredCandidates(enrollmentContext(t), enrollmentActor, req, enrollmentNow)
+ if delta > 0 {
+ enrollmentZero(t, v, err)
+ if !errors.Is(err, errSponsoredEnrollmentTooLarge) {
+ t.Fatal(err)
+ }
+ var count int
+ if err := s.db.QueryRow(`SELECT count(*) FROM sponsored_enrollment_receipts`).Scan(&count); err != nil || count != 0 {
+ t.Fatal("oversize receipt stored", count, err)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ var size int
+ if err := s.db.QueryRow(`SELECT length(body) FROM sponsored_enrollment_receipts WHERE operation_id=?`, req.OperationID).Scan(&size); err != nil || size != sponsoredEnrollmentMaxBody+delta {
+ t.Fatal(size, err)
+ }
+ read, err := s.readSponsoredEnrollment(enrollmentContext(t), enrollmentActor, req.OperationID)
+ if err != nil || !reflect.DeepEqual(read, v.Historical) {
+ t.Fatal("boundary read", err)
+ }
+ })
+ }
+}
+
+func TestSponsoredEnrollmentFreshMigrationPreservesExistingData(t *testing.T) {
+ s := enrollmentStore(t)
+ seedSponsoredCoverage(t, s, false)
+ if err := s.migrate(); err != nil {
+ t.Fatal(err)
+ }
+ sponsoredExec(t, s.db, `DROP TABLE sponsored_cohort_memberships`)
+ sponsoredExec(t, s.db, `DROP TABLE sponsored_enrollment_receipts`)
+ before := enrollmentExistingSnapshot(t, s)
+ if err := s.migrate(); err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(before, enrollmentExistingSnapshot(t, s)) {
+ t.Fatal("migration changed existing data")
+ }
+ for _, table := range []string{"sponsored_cohort_memberships", "sponsored_enrollment_receipts"} {
+ var n int
+ if err := s.db.QueryRow(`SELECT count(*) FROM ` + table).Scan(&n); err != nil || n != 0 {
+ t.Fatal("migration enrolled", table, n, err)
+ }
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index 57118b659..6fe7e68db 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -441,7 +441,10 @@ func (s *Store) migrate() error {
if err := s.migrateOveragePreference(); err != nil {
return err
}
- return s.migrateSignupRecords()
+ if err := s.migrateSignupRecords(); err != nil {
+ return err
+ }
+ return s.migrateSponsoredEnrollment()
}
// migrateAppsKey moves the apps table from PRIMARY KEY(app) — a GLOBAL name