diff --git a/README.md b/README.md index 61753fc..487d0bf 100644 --- a/README.md +++ b/README.md @@ -87,14 +87,31 @@ Available columns: See `addresses.csv.example` for a complete example with all columns. +## Quota pools + +Addresses can share one quota. The nullable `address.quota_pool` column holds the `address_lower` (the case folded address) of the pool owner; `NULL` means the address is its own pool. For example linked addresses such as `@alice_bot@example.com` can share the quota of `@alice@example.com`: + +```sql +update address set quota_pool = '@alice@example.com' +where address_lower = '@alice_bot@example.com'; +``` + +- An address in a pool is checked against the pool owner's limits (`limit_*`) and the combined usage of the owner and every address whose `quota_pool` is the owner. Its own limits are kept but not used while it is pooled. +- The pool owner reports the same combined usage, so a pool member's usage counts against the owner too. +- `accepting_new` and `display_name` stay per address. +- Pools are one level deep: a pool owner's own `quota_pool` should be `NULL`. `POST /fmsgid` enforces that; anything writing the column directly should too. +- `quota_pool` references `address (address_lower)`, so the owner must exist first; deleting the owner sets its members' `quota_pool` back to `NULL`. + +Set `quota_pool` with `POST /fmsgid`, or have the system syncing your identity provider update the column directly. A role that does so needs `UPDATE (quota_pool)` on `address` besides `SELECT`. + ## API Routes All routes are served over HTTPS under the `/fmsgid` path. | Method | Route | Description | |--------|-------|-------------| -| `GET` | `/fmsgid/:address` | Lookup an fmsg address and return its details including display name, quotas, and usage. The address must be in fmsg format (`@user@example.com`). Returns `AddressDetail` JSON on success, `400` if the address is invalid, `404` if not found. | -| `POST` | `/fmsgid` | Register a new address with default quotas (the `address` table's column defaults in `dd.sql`), idempotently. Accepts a JSON body with `address` and optional `display_name`. Returns `201` if the address was created, `200` if it already existed (never modifies an existing row — use CSV sync to change quotas or `accepting_new` on an existing address), `400` if the address is invalid. | +| `GET` | `/fmsgid/:address` | Lookup an fmsg address and return its details including display name, quotas, and usage. The address must be in fmsg format (`@user@example.com`). Returns `AddressDetail` JSON on success, `400` if the address is invalid, `404` if not found. For an address in a [quota pool](#quota-pools) the limits are the pool owner's, the usage is the whole pool's and `quotaPool` is the pool owner's address; `quotaPool` is omitted otherwise. `acceptingNew` is always the address's own. | +| `POST` | `/fmsgid` | Register an address, idempotently. Accepts a JSON body with `address` and optional `display_name`, `accepting_new` (bool) and `quota_pool` (an fmsg address, see [Quota pools](#quota-pools)). A new address takes the `address` table's column defaults in `dd.sql` except for `accepting_new` and `quota_pool` when given. An existing address is left as it is except for `accepting_new` and `quota_pool` when given (`display_name` and quotas are never changed — use CSV sync for those). Setting `accepting_new` `false` stops the address accepting new messages; setting it `true` again reinstates it. `quota_pool` `""` or the address itself takes the address out of any pool. Returns `201` if the address was created, `200` if it already existed, both with `{"created":bool,"reinstated":bool}` where `reinstated` is `true` when the address existed, wasn't accepting new messages and now is. Returns `400` if the body or an address is invalid, `422` with `{"error":"..."}` if `quota_pool` names an address that doesn't exist or is in another pool, or the address owns a pool itself. | | `POST` | `/fmsgid/send` | Record a send transaction. Accepts an `AddressTx` JSON body with `address`, `ts` (timestamp), and `size`. The address is case folded, and every call is recorded, including several with the same timestamp. Returns `200`, or `400` if the body or address is invalid. | | `POST` | `/fmsgid/recv` | Record a receive transaction. Accepts an `AddressTx` JSON body with `address`, `ts` (timestamp), and `size`. The address is case folded, and every call is recorded, including several with the same timestamp. Returns `200`, or `400` if the body or address is invalid. | diff --git a/dd.sql b/dd.sql index c341c76..9cb5703 100644 --- a/dd.sql +++ b/dd.sql @@ -32,6 +32,26 @@ create table if not exists address ( limit_send_count_per_1d bigint not null default 1000 ); +-- quota pools: an address whose quota_pool is set (to the address_lower of +-- another address, the pool owner) shares the pool owner's quota: it is +-- checked against the owner's limits and the combined usage of the owner and +-- every address whose quota_pool is the owner. NULL means the address is its +-- own pool; the owner's usage includes its pool members' too. Pools are one +-- level deep, so a pool owner's own quota_pool should be NULL. A nullable +-- column without a default is added without rewriting the table. +alter table address add column if not exists quota_pool text; + +do $$ +begin + if not exists (select 1 from pg_constraint + where conrelid = 'address'::regclass and conname = 'address_quota_pool_fkey') then + alter table address add constraint address_quota_pool_fkey + foreign key (quota_pool) references address (address_lower) on delete set null; + end if; +end $$; + +create index if not exists address_quota_pool on address (quota_pool); + -- one row per message sent or received; several may share a timestamp create table if not exists address_tx ( id bigserial primary key, diff --git a/src/fmsgid.go b/src/fmsgid.go index c8492a6..ee407f9 100644 --- a/src/fmsgid.go +++ b/src/fmsgid.go @@ -2,12 +2,15 @@ package main import ( "context" + "errors" + "fmt" "log" "net/http" "os" "strings" "github.com/gin-gonic/gin" + "github.com/jackc/pgx/v4" "github.com/jackc/pgx/v4/pgxpool" "github.com/joho/godotenv" "golang.org/x/text/cases" @@ -39,8 +42,32 @@ type AddressTx struct { type CreateAddressReq struct { Address string `json:"address"` DisplayName string `json:"display_name"` + // AcceptingNew, when given, is set on the address whether it is new or + // already exists. When omitted a new address takes the column default and + // an existing one is left as it is. + AcceptingNew *bool `json:"accepting_new"` + // QuotaPool, when given, puts the address in the quota pool of the + // address it names (which must exist and be in no other pool) whether the + // address is new or already exists. "" or the address itself takes the + // address out of any pool. When omitted quota_pool is left as it is. + QuotaPool *string `json:"quota_pool"` } +// CreateAddressResp is the result of POST /fmsgid. +type CreateAddressResp struct { + // Created is true when the address didn't exist before. + Created bool `json:"created"` + // Reinstated is true when the address already existed, wasn't accepting + // new messages and now is, because the request set accepting_new true. + Reinstated bool `json:"reinstated"` +} + +// quotaPoolError is a quota_pool the address can't join; its text is returned +// to the caller. +type quotaPoolError string + +func (e quotaPoolError) Error() string { return string(e) } + type AddressDetail struct { Address string `json:"address"` DisplayName string `json:"displayName"` @@ -60,6 +87,9 @@ type AddressDetail struct { SendSizePer1d int64 `json:"sendSizePer1d"` SendCountPer1d int64 `json:"sendCountPer1d"` Tags []string `json:"tags"` + // QuotaPool is the address of the pool owner whose limits and combined + // usage are reported, omitted when the address is in no pool. + QuotaPool *string `json:"quotaPool,omitempty"` } func initPool() error { @@ -111,19 +141,23 @@ func getAddressDetail(c *gin.Context) { } var ad AddressDetail + var poolOwner string + // the limits are the quota pool's: the pool owner's when the address is in + // a pool, else its own err = rows.Scan(&ad.Address, &ad.DisplayName, &ad.AcceptingNew, &ad.LimitRecvSizeTotal, &ad.LimitRecvSizePerMsg, &ad.LimitRecvSizePer1d, &ad.LimitRecvCountPer1d, &ad.LimitSendSizeTotal, &ad.LimitSendSizePerMsg, &ad.LimitSendSizePer1d, - &ad.LimitSendCountPer1d) + &ad.LimitSendCountPer1d, &poolOwner, &ad.QuotaPool) if err != nil { c.AbortWithError(500, err) return } + rows.Close() - // select actuals; an aggregate always returns one row, zeros when the - // address hasn't sent or received anything - err = pool.QueryRow(ctx, sqlActuals, addr).Scan( + // select actuals summed over the quota pool; an aggregate always returns + // one row, zeros when the pool hasn't sent or received anything + err = pool.QueryRow(ctx, sqlActuals, poolOwner).Scan( &ad.RecvSizeTotal, &ad.RecvSizePer1d, &ad.RecvCountPer1d, &ad.SendSizeTotal, &ad.SendSizePer1d, &ad.SendCountPer1d) if err != nil { @@ -134,9 +168,11 @@ func getAddressDetail(c *gin.Context) { c.JSON(http.StatusOK, ad) } -// postCreateAddress registers a new address with default quotas, idempotently. -// It never modifies an address that already exists — use CSV sync to change -// quotas or accepting_new on an existing address. +// postCreateAddress registers an address, idempotently. A new address takes +// the address table's column defaults except for any accepting_new and +// quota_pool given. An existing address is left as it is except for any +// accepting_new and quota_pool given; display_name and the limits are never +// changed — use CSV sync for those. func postCreateAddress(c *gin.Context) { ctx := c.Request.Context() @@ -151,20 +187,122 @@ func postCreateAddress(c *gin.Context) { c.AbortWithStatus(http.StatusBadRequest) return } + if req.QuotaPool != nil && *req.QuotaPool != "" && !isValidFmsgAddr(*req.QuotaPool) { + c.AbortWithStatus(http.StatusBadRequest) + return + } - addrLower := cases.Fold().String(req.Address) - - tag, err := pool.Exec(ctx, sqlInsertAddressIfNotExists, addrLower, req.Address, req.DisplayName) + resp, err := registerAddress(ctx, req) + var qpErr quotaPoolError + if errors.As(err, &qpErr) { + c.AbortWithStatusJSON(http.StatusUnprocessableEntity, gin.H{"error": qpErr.Error()}) + return + } if err != nil { c.AbortWithError(500, err) return } - if tag.RowsAffected() == 0 { - c.Status(http.StatusOK) + if resp.Created { + c.JSON(http.StatusCreated, resp) return } - c.Status(http.StatusCreated) + c.JSON(http.StatusOK, resp) +} + +// registerAddress inserts or updates the requested address in one transaction. +func registerAddress(ctx context.Context, req CreateAddressReq) (CreateAddressResp, error) { + var resp CreateAddressResp + addrLower := cases.Fold().String(req.Address) + + // quotaPool is the case folded pool owner to set when req.QuotaPool is + // given, nil to take the address out of any pool + var quotaPool *string + if req.QuotaPool != nil && *req.QuotaPool != "" { + if p := cases.Fold().String(*req.QuotaPool); p != addrLower { + quotaPool = &p + } + } + + tx, err := pool.Begin(ctx) + if err != nil { + return resp, err + } + defer tx.Rollback(ctx) //nolint:errcheck // a no-op after commit + + // pools are one level deep: the pool owner must be in no other pool, and + // an address that owns a pool can't join another + if quotaPool != nil { + var ownerPool *string + err := tx.QueryRow(ctx, sqlSelectQuotaPoolForShare, *quotaPool).Scan(&ownerPool) + if errors.Is(err, pgx.ErrNoRows) { + return resp, quotaPoolError("quota_pool address not found") + } + if err != nil { + return resp, err + } + if ownerPool != nil && *ownerPool != *quotaPool { + return resp, quotaPoolError("quota_pool address is in another quota pool") + } + var isOwner bool + if err := tx.QueryRow(ctx, sqlIsPoolOwner, addrLower).Scan(&isOwner); err != nil { + return resp, err + } + if isOwner { + return resp, quotaPoolError("address owns a quota pool") + } + } + + // only the columns given are inserted so the rest take their defaults + cols := []string{"address_lower", "address", "display_name"} + args := []any{addrLower, req.Address, req.DisplayName} + if req.AcceptingNew != nil { + cols = append(cols, "accepting_new") + args = append(args, *req.AcceptingNew) + } + if req.QuotaPool != nil { + cols = append(cols, "quota_pool") + args = append(args, quotaPool) + } + tag, err := tx.Exec(ctx, insertAddressSQL(cols), args...) + if err != nil { + return resp, err + } + resp.Created = tag.RowsAffected() == 1 + + if !resp.Created && (req.AcceptingNew != nil || req.QuotaPool != nil) { + var wasAccepting bool + if err := tx.QueryRow(ctx, sqlSelectAcceptingNewForUpdate, addrLower).Scan(&wasAccepting); err != nil { + return resp, err + } + var set []string + args := []any{addrLower} + if req.AcceptingNew != nil { + args = append(args, *req.AcceptingNew) + set = append(set, fmt.Sprintf("accepting_new = $%d", len(args))) + } + if req.QuotaPool != nil { + args = append(args, quotaPool) + set = append(set, fmt.Sprintf("quota_pool = $%d", len(args))) + } + if _, err := tx.Exec(ctx, "update address set "+strings.Join(set, ", ")+" where address_lower = $1;", args...); err != nil { + return resp, err + } + resp.Reinstated = !wasAccepting && req.AcceptingNew != nil && *req.AcceptingNew + } + + return resp, tx.Commit(ctx) +} + +// insertAddressSQL returns an insert of the given address columns, in order as +// $1..$n, that does nothing when the address already exists. +func insertAddressSQL(cols []string) string { + params := make([]string, len(cols)) + for i := range cols { + params[i] = fmt.Sprintf("$%d", i+1) + } + return "insert into address (" + strings.Join(cols, ", ") + ") values (" + + strings.Join(params, ", ") + ") on conflict (address_lower) do nothing;" } func postAddressTxSend(c *gin.Context) { diff --git a/src/pool_test.go b/src/pool_test.go new file mode 100644 index 0000000..fe24b2d --- /dev/null +++ b/src/pool_test.go @@ -0,0 +1,371 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v4/pgxpool" +) + +// poolDomain is used only by these tests; its rows are deleted before and +// after each test. +const poolDomain = "pooltest.example" + +func deletePoolRows(t *testing.T, p *pgxpool.Pool) { + t.Helper() + mustExec(t, p, `delete from address_tx where address_lower like '%@`+poolDomain+`'`) + mustExec(t, p, `update address set quota_pool = null where address_lower like '%@`+poolDomain+`'`) + mustExec(t, p, `delete from address where address_lower like '%@`+poolDomain+`'`) +} + +func poolTestDB(t *testing.T) { + t.Helper() + testDB(t) + deletePoolRows(t, pool) + t.Cleanup(func() { deletePoolRows(t, pool) }) +} + +// postAddress POSTs body to /fmsgid and checks the status and response. +func postAddress(t *testing.T, body string, wantCode int, want CreateAddressResp) { + t.Helper() + w := serveRoute("POST", "/fmsgid", body) + if w.Code != wantCode { + t.Fatalf("POST /fmsgid %s: status %d, want %d (%s)", body, w.Code, wantCode, w.Body.String()) + } + if wantCode != http.StatusOK && wantCode != http.StatusCreated { + return + } + var got CreateAddressResp + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatalf("POST /fmsgid %s: %s: %q", body, err, w.Body.String()) + } + if got != want { + t.Fatalf("POST /fmsgid %s: response %+v, want %+v", body, got, want) + } +} + +func getDetail(t *testing.T, addr string) AddressDetail { + t.Helper() + w := serveRoute("GET", "/fmsgid/"+addr, "") + if w.Code != http.StatusOK { + t.Fatalf("GET %s: status %d, want 200", addr, w.Code) + } + var ad AddressDetail + if err := json.Unmarshal(w.Body.Bytes(), &ad); err != nil { + t.Fatal(err) + } + return ad +} + +// setLimits gives addr every limit as base plus its position, so the limits of +// two addresses never coincide. +func setLimits(t *testing.T, addr string, base int64) { + t.Helper() + mustExec(t, pool, `update address set + limit_recv_size_total = $2::bigint, limit_recv_size_per_msg = $2::bigint + 1, + limit_recv_size_per_1d = $2::bigint + 2, limit_recv_count_per_1d = $2::bigint + 3, + limit_send_size_total = $2::bigint + 4, limit_send_size_per_msg = $2::bigint + 5, + limit_send_size_per_1d = $2::bigint + 6, limit_send_count_per_1d = $2::bigint + 7 + where address_lower = $1`, addr, base) +} + +func checkLimits(t *testing.T, ad AddressDetail, base int64) { + t.Helper() + got := []int64{ad.LimitRecvSizeTotal, ad.LimitRecvSizePerMsg, ad.LimitRecvSizePer1d, ad.LimitRecvCountPer1d, + ad.LimitSendSizeTotal, ad.LimitSendSizePerMsg, ad.LimitSendSizePer1d, ad.LimitSendCountPer1d} + for i, v := range got { + if v != base+int64(i) { + t.Errorf("%s: limits %v, want %d..%d", ad.Address, got, base, base+7) + return + } + } +} + +// usage is the six actuals of an AddressDetail, in field order. +type usage [6]int64 + +func usageOf(ad AddressDetail) usage { + return usage{ad.RecvSizeTotal, ad.RecvSizePer1d, ad.RecvCountPer1d, + ad.SendSizeTotal, ad.SendSizePer1d, ad.SendCountPer1d} +} + +func recordTx(t *testing.T, route, addr string, ts float64, size int) { + t.Helper() + body := fmt.Sprintf(`{"address":%q,"ts":%f,"size":%d}`, addr, ts, size) + if w := serveRoute("POST", route, body); w.Code != http.StatusOK { + t.Fatalf("POST %s %s: status %d, want 200", route, body, w.Code) + } +} + +func quotaPoolOf(t *testing.T, addr string) *string { + t.Helper() + var qp *string + if err := pool.QueryRow(context.Background(), + `select quota_pool from address where address_lower = $1`, addr).Scan(&qp); err != nil { + t.Fatal(err) + } + return qp +} + +func acceptingNewOf(t *testing.T, addr string) bool { + t.Helper() + var a bool + if err := pool.QueryRow(context.Background(), + `select accepting_new from address where address_lower = $1`, addr).Scan(&a); err != nil { + t.Fatal(err) + } + return a +} + +// TestAddressDetail_QuotaPool checks a pooled address reports its pool +// owner's limits and the pool's combined usage, the owner reports the same +// usage, and an address in no pool is unaffected. +func TestAddressDetail_QuotaPool(t *testing.T) { + poolTestDB(t) + + postAddress(t, `{"address":"@Alice@PoolTest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@alice_bot@pooltest.example","quota_pool":"@ALICE@pooltest.example"}`, + http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@alice_cli@pooltest.example","quota_pool":"@alice@pooltest.example"}`, + http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@bob@pooltest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + setLimits(t, "@alice@pooltest.example", 1000) + setLimits(t, "@alice_bot@pooltest.example", 2000) + setLimits(t, "@alice_cli@pooltest.example", 3000) + setLimits(t, "@bob@pooltest.example", 4000) + + now := float64(time.Now().UnixMicro()) / 1e6 + old := now - 2*24*60*60 + recordTx(t, "/fmsgid/recv", "@alice@pooltest.example", now, 100) + recordTx(t, "/fmsgid/recv", "@alice@pooltest.example", old, 1000) + recordTx(t, "/fmsgid/recv", "@alice_bot@pooltest.example", now, 20) + recordTx(t, "/fmsgid/recv", "@ALICE_CLI@pooltest.example", now, 3) + recordTx(t, "/fmsgid/send", "@alice_bot@pooltest.example", now, 40) + recordTx(t, "/fmsgid/send", "@alice_cli@pooltest.example", old, 500) + recordTx(t, "/fmsgid/recv", "@bob@pooltest.example", now, 7) + recordTx(t, "/fmsgid/send", "@bob@pooltest.example", now, 9) + + poolUsage := usage{1123, 123, 3, 540, 40, 1} + + for _, addr := range []string{"@alice_bot@pooltest.example", "@Alice_CLI@pooltest.example"} { + ad := getDetail(t, addr) + checkLimits(t, ad, 1000) + if got := usageOf(ad); got != poolUsage { + t.Errorf("%s: usage %v, want pool usage %v", addr, got, poolUsage) + } + if ad.QuotaPool == nil || *ad.QuotaPool != "@Alice@PoolTest.example" { + t.Errorf("%s: quotaPool %v, want @Alice@PoolTest.example", addr, ad.QuotaPool) + } + if !ad.AcceptingNew { + t.Errorf("%s: not accepting new", addr) + } + } + + owner := getDetail(t, "@alice@pooltest.example") + checkLimits(t, owner, 1000) + if got := usageOf(owner); got != poolUsage { + t.Errorf("owner: usage %v, want pool usage %v", got, poolUsage) + } + if owner.QuotaPool != nil { + t.Errorf("owner: quotaPool %q, want omitted", *owner.QuotaPool) + } + + bob := getDetail(t, "@bob@pooltest.example") + checkLimits(t, bob, 4000) + if got, want := usageOf(bob), (usage{7, 7, 1, 9, 9, 1}); got != want { + t.Errorf("unpooled: usage %v, want %v", got, want) + } + if bob.QuotaPool != nil { + t.Errorf("unpooled: quotaPool %q, want omitted", *bob.QuotaPool) + } + + // accepting_new stays per address + mustExec(t, pool, `update address set accepting_new = false where address_lower = '@alice_bot@pooltest.example'`) + if getDetail(t, "@alice_bot@pooltest.example").AcceptingNew { + t.Error("pooled address accepting new after disabling it") + } + if !getDetail(t, "@alice@pooltest.example").AcceptingNew { + t.Error("owner not accepting new after disabling a pool member") + } + + // the response keeps every pre-pool field and adds only quotaPool + w := serveRoute("GET", "/fmsgid/@bob@pooltest.example", "") + var fields map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &fields); err != nil { + t.Fatal(err) + } + if _, ok := fields["quotaPool"]; ok { + t.Error("unpooled response has quotaPool") + } +} + +// TestAddressDetail_QuotaPoolOwnerMissing checks quota_pool can't name an +// address that doesn't exist, and that deleting a pool owner leaves its +// members as their own pools. +func TestAddressDetail_QuotaPoolOwnerMissing(t *testing.T) { + poolTestDB(t) + ctx := context.Background() + + postAddress(t, `{"address":"@solo@pooltest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + if _, err := pool.Exec(ctx, `update address set quota_pool = '@nobody@pooltest.example' + where address_lower = '@solo@pooltest.example'`); err == nil { + t.Error("quota_pool naming a missing address was accepted") + } + postAddress(t, `{"address":"@solo@pooltest.example","quota_pool":"@nobody@pooltest.example"}`, + http.StatusUnprocessableEntity, CreateAddressResp{}) + postAddress(t, `{"address":"@new@pooltest.example","quota_pool":"@nobody@pooltest.example"}`, + http.StatusUnprocessableEntity, CreateAddressResp{}) + if w := serveRoute("GET", "/fmsgid/@new@pooltest.example", ""); w.Code != http.StatusNotFound { + t.Errorf("address created despite a missing quota_pool: status %d", w.Code) + } + + postAddress(t, `{"address":"@gone@pooltest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@solo@pooltest.example","quota_pool":"@gone@pooltest.example"}`, + http.StatusOK, CreateAddressResp{}) + setLimits(t, "@gone@pooltest.example", 1000) + setLimits(t, "@solo@pooltest.example", 2000) + checkLimits(t, getDetail(t, "@solo@pooltest.example"), 1000) + + mustExec(t, pool, `delete from address where address_lower = '@gone@pooltest.example'`) + ad := getDetail(t, "@solo@pooltest.example") + checkLimits(t, ad, 2000) + if ad.QuotaPool != nil { + t.Errorf("quotaPool %q after its owner was deleted, want omitted", *ad.QuotaPool) + } + if qp := quotaPoolOf(t, "@solo@pooltest.example"); qp != nil { + t.Errorf("quota_pool %q after its owner was deleted, want NULL", *qp) + } +} + +// TestPostCreateAddress_QuotaPool checks setting, clearing and refusing +// quota_pool through POST /fmsgid. +func TestPostCreateAddress_QuotaPool(t *testing.T) { + poolTestDB(t) + + postAddress(t, `{"address":"@owner@pooltest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@other@pooltest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@member@pooltest.example","quota_pool":"@Owner@pooltest.example"}`, + http.StatusCreated, CreateAddressResp{Created: true}) + if qp := quotaPoolOf(t, "@member@pooltest.example"); qp == nil || *qp != "@owner@pooltest.example" { + t.Fatalf("quota_pool %v, want @owner@pooltest.example (case folded)", qp) + } + + // omitting quota_pool leaves it as it is + postAddress(t, `{"address":"@member@pooltest.example"}`, http.StatusOK, CreateAddressResp{}) + if quotaPoolOf(t, "@member@pooltest.example") == nil { + t.Fatal("quota_pool cleared by a request without one") + } + + // pools are one level deep + postAddress(t, `{"address":"@nested@pooltest.example","quota_pool":"@member@pooltest.example"}`, + http.StatusUnprocessableEntity, CreateAddressResp{}) + postAddress(t, `{"address":"@owner@pooltest.example","quota_pool":"@other@pooltest.example"}`, + http.StatusUnprocessableEntity, CreateAddressResp{}) + if qp := quotaPoolOf(t, "@owner@pooltest.example"); qp != nil { + t.Fatalf("pool owner joined another pool: %q", *qp) + } + + postAddress(t, `{"address":"@member@pooltest.example","quota_pool":"alice"}`, + http.StatusBadRequest, CreateAddressResp{}) + + // the address itself, or "", takes it out of any pool + postAddress(t, `{"address":"@member@pooltest.example","quota_pool":"@MEMBER@pooltest.example"}`, + http.StatusOK, CreateAddressResp{}) + if qp := quotaPoolOf(t, "@member@pooltest.example"); qp != nil { + t.Fatalf("quota_pool %q after setting it to the address itself, want NULL", *qp) + } + postAddress(t, `{"address":"@member@pooltest.example","quota_pool":"@other@pooltest.example"}`, + http.StatusOK, CreateAddressResp{}) + postAddress(t, `{"address":"@member@pooltest.example","quota_pool":""}`, http.StatusOK, CreateAddressResp{}) + if qp := quotaPoolOf(t, "@member@pooltest.example"); qp != nil { + t.Fatalf("quota_pool %q after clearing it, want NULL", *qp) + } + + // owner and member may now both be pooled + postAddress(t, `{"address":"@owner@pooltest.example","quota_pool":"@other@pooltest.example"}`, + http.StatusOK, CreateAddressResp{}) +} + +// TestPostCreateAddress_AcceptingNew checks disabling and reinstating an +// address through POST /fmsgid, and that the response says when an address +// was reinstated. +func TestPostCreateAddress_AcceptingNew(t *testing.T) { + poolTestDB(t) + const addr = "@agent@pooltest.example" + + postAddress(t, `{"address":"@Agent@pooltest.example","display_name":"Agent"}`, + http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@agent@pooltest.example"}`, http.StatusOK, CreateAddressResp{}) + postAddress(t, `{"address":"@agent@pooltest.example","accepting_new":true}`, http.StatusOK, CreateAddressResp{}) + + postAddress(t, `{"address":"@AGENT@pooltest.example","accepting_new":false}`, http.StatusOK, CreateAddressResp{}) + if acceptingNewOf(t, addr) { + t.Fatal("accepting_new still true after setting it false") + } + postAddress(t, `{"address":"@agent@pooltest.example","accepting_new":false}`, http.StatusOK, CreateAddressResp{}) + + // registering without accepting_new leaves a disabled address disabled + postAddress(t, `{"address":"@agent@pooltest.example"}`, http.StatusOK, CreateAddressResp{}) + if acceptingNewOf(t, addr) { + t.Fatal("accepting_new set true by a request without it") + } + + postAddress(t, `{"address":"@agent@pooltest.example","accepting_new":true}`, + http.StatusOK, CreateAddressResp{Reinstated: true}) + if !acceptingNewOf(t, addr) { + t.Fatal("accepting_new still false after reinstating") + } + postAddress(t, `{"address":"@agent@pooltest.example","accepting_new":true}`, http.StatusOK, CreateAddressResp{}) + + // display_name and the original spelling are never changed + ad := getDetail(t, addr) + if ad.Address != "@Agent@pooltest.example" || ad.DisplayName != "Agent" { + t.Errorf("address %q display name %q changed", ad.Address, ad.DisplayName) + } + + // a new address may be created not accepting new messages + postAddress(t, `{"address":"@held@pooltest.example","accepting_new":false}`, + http.StatusCreated, CreateAddressResp{Created: true}) + if acceptingNewOf(t, "@held@pooltest.example") { + t.Error("new address accepting new despite accepting_new false") + } + + // reinstating and pooling in one request + postAddress(t, `{"address":"@owner@pooltest.example"}`, http.StatusCreated, CreateAddressResp{Created: true}) + postAddress(t, `{"address":"@held@pooltest.example","accepting_new":true,"quota_pool":"@owner@pooltest.example"}`, + http.StatusOK, CreateAddressResp{Reinstated: true}) + if quotaPoolOf(t, "@held@pooltest.example") == nil || !acceptingNewOf(t, "@held@pooltest.example") { + t.Error("reinstate with quota_pool didn't apply both") + } +} + +// TestDDL_QuotaPoolIdempotent re-applies dd.sql and checks quota_pool has +// exactly one foreign key and its index. +func TestDDL_QuotaPoolIdempotent(t *testing.T) { + testDB(t) + ctx := context.Background() + ddl, err := os.ReadFile("../dd.sql") + if err != nil { + t.Fatal(err) + } + mustExec(t, pool, string(ddl)) + + var fks, idx int + if err := pool.QueryRow(ctx, `select count(*) from pg_constraint c + join pg_attribute a on a.attrelid = c.conrelid and a.attnum = any (c.conkey) + where c.conrelid = 'address'::regclass and c.contype = 'f' and a.attname = 'quota_pool'`).Scan(&fks); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, `select count(*) from pg_indexes + where tablename = 'address' and indexname = 'address_quota_pool'`).Scan(&idx); err != nil { + t.Fatal(err) + } + if fks != 1 || idx != 1 { + t.Errorf("quota_pool foreign keys %d, indexes %d, want 1 and 1", fks, idx) + } +} diff --git a/src/sql.go b/src/sql.go index f0c73e4..13f6b1a 100644 --- a/src/sql.go +++ b/src/sql.go @@ -1,21 +1,29 @@ package main +// sqlSelectAddressDetail selects an address with the limits of its quota pool: +// the pool owner's when quota_pool is set, else its own. The left join keeps +// the address's own limits should its pool owner be missing. pool_owner is the +// address_lower whose pool the usage is summed over (see sqlActuals); +// quota_pool is the pool owner's address, NULL when the address isn't pooled. const sqlSelectAddressDetail string = `select - address - , display_name - , accepting_new - , limit_recv_size_total - , limit_recv_size_per_msg - , limit_recv_size_per_1d - , limit_recv_count_per_1d - , limit_send_size_total - , limit_send_size_per_msg - , limit_send_size_per_1d - , limit_send_count_per_1d + a.address + , a.display_name + , a.accepting_new + , coalesce(o.limit_recv_size_total, a.limit_recv_size_total) + , coalesce(o.limit_recv_size_per_msg, a.limit_recv_size_per_msg) + , coalesce(o.limit_recv_size_per_1d, a.limit_recv_size_per_1d) + , coalesce(o.limit_recv_count_per_1d, a.limit_recv_count_per_1d) + , coalesce(o.limit_send_size_total, a.limit_send_size_total) + , coalesce(o.limit_send_size_per_msg, a.limit_send_size_per_msg) + , coalesce(o.limit_send_size_per_1d, a.limit_send_size_per_1d) + , coalesce(o.limit_send_count_per_1d, a.limit_send_count_per_1d) + , coalesce(o.address_lower, a.address_lower) as pool_owner + , o.address as quota_pool from - address + address a + left join address o on o.address_lower = a.quota_pool where - address_lower = $1;` + a.address_lower = $1;` const sqlInsertTx string = `insert into address_tx (address_lower, ts, type, size) VALUES ($1, to_timestamp($2), $3, $4);` @@ -40,17 +48,21 @@ on conflict (address_lower) do update set // sqlDisableAbsentAddresses disables addresses not present in the provided parameter array. const sqlDisableAbsentAddresses string = `update address set accepting_new = false where address_lower != ALL($1);` -// sqlInsertAddressIfNotExists registers a new address. Only the identity -// columns are given so accepting_new and every limit take the address table's -// column defaults (see dd.sql). It never modifies an existing row — callers -// that need to update quotas or accepting_new should use sqlUpsertAddress -// (e.g. the CSV sync path) instead. -const sqlInsertAddressIfNotExists string = `insert into address (address_lower, address, display_name) -values ($1, $2, $3) -on conflict (address_lower) do nothing;` +// sqlSelectAcceptingNewForUpdate reads an existing address's accepting_new, +// locking the row until the transaction ends. +const sqlSelectAcceptingNewForUpdate string = `select accepting_new from address where address_lower = $1 for update;` -// sqlActuals sums an address's recorded messages. Columns are in the order -// getAddressDetail scans them into AddressDetail. +// sqlSelectQuotaPoolForShare reads a prospective pool owner's own quota_pool, +// locking the row so it can't join another pool until the transaction ends. +const sqlSelectQuotaPoolForShare string = `select quota_pool from address where address_lower = $1 for share;` + +// sqlIsPoolOwner reports whether any address is in the given address's pool. +const sqlIsPoolOwner string = `select exists (select 1 from address where quota_pool = $1 and address_lower <> $1);` + +// sqlActuals sums the recorded messages of a quota pool: the pool owner $1 +// and every address whose quota_pool is $1. For an address in no pool, and +// with no pool members, that is just its own messages. Columns are in the +// order getAddressDetail scans them into AddressDetail. const sqlActuals string = `select coalesce(sum(size) filter (where type = 1), 0) as recv_size_total , coalesce(sum(size) filter (where type = 1 and ts > now() - interval '1 day'), 0) as recv_size_1d @@ -61,4 +73,6 @@ const sqlActuals string = `select from address_tx where - address_lower = $1;` + address_lower = any (array( + select address_lower from address where address_lower = $1 or quota_pool = $1 + ));`