diff --git a/README.md b/README.md index d08ff66..61753fc 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,14 @@ go build . This will produce an executable named `fmsgid` (or `fmsgid.exe` on Windows). +### Test + +``` +go test ./... +``` + +Tests that need PostgreSQL are skipped unless `FMSGID_TEST_DATABASE_URL` is set to a connection string, e.g. `postgres://postgres:pass@127.0.0.1:5432/fmsgid_test?sslmode=disable`. They apply `dd.sql` and write and delete rows, so point them at a throwaway database only. + ## Running PostgreSQL database with tables created from `dd.sql` is required. The database connection details are got from the environment and the user must have write access to these tables. Run the executable which will update environment with any variables in `.env` file alongside if present. @@ -86,9 +94,9 @@ 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, 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. | -| `POST` | `/fmsgid/send` | Record a send transaction. Accepts an `AddressTx` JSON body with `address`, `ts` (timestamp), and `size`. | -| `POST` | `/fmsgid/recv` | Record a receive transaction. Accepts an `AddressTx` JSON body with `address`, `ts` (timestamp), and `size`. | +| `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. | +| `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. | ### systemd diff --git a/dd.sql b/dd.sql index 8df754d..c341c76 100644 --- a/dd.sql +++ b/dd.sql @@ -3,8 +3,9 @@ * PostgreSQL database objects data definition for fmsgid * * This script is IDEMPOTENT: every statement is safe to re-run - * (create table/index if not exists; alter table set storage - * parameters simply reapplies). Migrating an existing database is + * (create table/index if not exists; do blocks that check the + * catalog first; alter table set storage parameters simply + * reapplies). Migrating an existing database is * therefore just re-running the whole script, e.g.: * * psql -d fmsgid -v ON_ERROR_STOP=1 -f dd.sql @@ -31,14 +32,43 @@ create table if not exists address ( limit_send_count_per_1d bigint not null default 1000 ); +-- one row per message sent or received; several may share a timestamp create table if not exists address_tx ( + id bigserial primary key, address_lower text not null references address (address_lower) , ts timestamptz not null, type smallint not null, -- 1 for recv, 2 for send - size int not null, - primary key (address_lower, ts) + size int not null ); +-- migrate address_tx created with primary key (address_lower, ts), which +-- rejected a second message to an address with the same timestamp: give it a +-- surrogate id key instead. The column is added inside the do block, not with +-- add column if not exists, because older PostgreSQL versions create an +-- orphan sequence for a serial column even when the column already exists. +do $$ +declare + pk name; +begin + if not exists (select 1 from pg_attribute + where attrelid = 'address_tx'::regclass and attname = 'id' and not attisdropped) then + alter table address_tx add column id bigserial; + end if; + select c.conname into pk + from pg_constraint c + join pg_attribute a on a.attrelid = c.conrelid and a.attname = 'id' + where c.conrelid = 'address_tx'::regclass + and c.contype = 'p' + and c.conkey <> array[a.attnum]; + if pk is not null then + execute format('alter table address_tx drop constraint %I', pk); + end if; + if not exists (select 1 from pg_constraint + where conrelid = 'address_tx'::regclass and contype = 'p') then + alter table address_tx add primary key (id); + end if; +end $$; + create index if not exists address_tx_addr_type_ts on address_tx (address_lower, type, ts desc); diff --git a/src/csv.go b/src/csv.go index ccd2de1..bfa0988 100644 --- a/src/csv.go +++ b/src/csv.go @@ -35,8 +35,9 @@ type addressRow struct { // parseCSV reads a CSV file and returns address rows. The CSV must have a header // row with column names matching the address table columns. Only the "address" -// column is required; others use defaults (accepting_new=true, limits=-1). -// Malformed rows are logged and skipped. +// column is required; a missing or empty column gets accepting_new=true and the +// same limits as the address table's column defaults in dd.sql. Malformed rows +// are logged and skipped. func parseCSV(filePath string) ([]addressRow, error) { f, err := os.Open(filePath) if err != nil { diff --git a/src/db_test.go b/src/db_test.go new file mode 100644 index 0000000..347dec9 --- /dev/null +++ b/src/db_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "context" + "net/http/httptest" + "os" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/jackc/pgx/v4/pgxpool" +) + +// Tests that need a PostgreSQL database are skipped unless +// FMSGID_TEST_DATABASE_URL is set to its connection string. They apply +// ../dd.sql and write and delete rows: point them at a throwaway database only. + +func init() { + gin.SetMode(gin.TestMode) +} + +// testDB connects the package pool to the test database with dd.sql applied. +func testDB(t *testing.T) { + t.Helper() + dsn := os.Getenv("FMSGID_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("FMSGID_TEST_DATABASE_URL not set") + } + ctx := context.Background() + p, err := pgxpool.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + ddl, err := os.ReadFile("../dd.sql") + if err != nil { + t.Fatal(err) + } + if _, err := p.Exec(ctx, string(ddl)); err != nil { + t.Fatalf("applying dd.sql: %s", err) + } + + prev := pool + pool = p + t.Cleanup(func() { + pool = prev + p.Close() + }) +} + +func mustExec(t *testing.T, p *pgxpool.Pool, sql string, args ...interface{}) { + t.Helper() + if _, err := p.Exec(context.Background(), sql, args...); err != nil { + t.Fatal(err) + } +} + +// serveRoute serves one request through the same routes main registers. +func serveRoute(method, path, body string) *httptest.ResponseRecorder { + r := gin.New() + r.GET("/fmsgid/:address", getAddressDetail) + r.POST("/fmsgid", postCreateAddress) + r.POST("/fmsgid/send", postAddressTxSend) + r.POST("/fmsgid/recv", postAddressTxRecv) + + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(method, path, strings.NewReader(body))) + return w +} diff --git a/src/fmsgid.go b/src/fmsgid.go index 3c31c8b..c8492a6 100644 --- a/src/fmsgid.go +++ b/src/fmsgid.go @@ -121,23 +121,15 @@ func getAddressDetail(c *gin.Context) { return } - // select actuals - rows, err = pool.Query(ctx, sqlActuals, addr) + // 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( + &ad.RecvSizeTotal, &ad.RecvSizePer1d, &ad.RecvCountPer1d, + &ad.SendSizeTotal, &ad.SendSizePer1d, &ad.SendCountPer1d) if err != nil { c.AbortWithError(500, err) return } - defer rows.Close() - - // no rows is possible when haven't sent or recieved anything! - if rows.Next() { - err = rows.Scan(&ad.SendSizeTotal, &ad.SendCountPer1d, &ad.SendSizePer1d, - &ad.RecvSizeTotal, &ad.RecvCountPer1d, &ad.RecvSizePer1d) - if err != nil { - c.AbortWithError(500, err) - return - } - } c.JSON(http.StatusOK, ad) } @@ -194,7 +186,16 @@ func postAddressTx(c *gin.Context, typ int) { return } - _, err = pool.Exec(ctx, sqlInsertTx, tx.Address, tx.Timestamp, typ, tx.Size) + if !isValidFmsgAddr(tx.Address) { + c.AbortWithStatus(http.StatusBadRequest) + return + } + + // address_lower references address rows, which are keyed by the case + // folded address + addrLower := cases.Fold().String(tx.Address) + + _, err = pool.Exec(ctx, sqlInsertTx, addrLower, tx.Timestamp, typ, tx.Size) if err != nil { c.AbortWithError(500, err) return diff --git a/src/sql.go b/src/sql.go index 18576cd..f0c73e4 100644 --- a/src/sql.go +++ b/src/sql.go @@ -1,63 +1,64 @@ -package main - -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 -from - address -where - address_lower = $1;` - -const sqlInsertTx string = `insert into address_tx (address_lower, ts, type, size) VALUES ($1, to_timestamp($2), $3, $4);` - -const sqlUpsertAddress string = `insert into address ( - address_lower, 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 -) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) -on conflict (address_lower) do update set - address = excluded.address, - display_name = excluded.display_name, - accepting_new = excluded.accepting_new, - limit_recv_size_total = excluded.limit_recv_size_total, - limit_recv_size_per_msg = excluded.limit_recv_size_per_msg, - limit_recv_size_per_1d = excluded.limit_recv_size_per_1d, - limit_recv_count_per_1d = excluded.limit_recv_count_per_1d, - limit_send_size_total = excluded.limit_send_size_total, - limit_send_size_per_msg = excluded.limit_send_size_per_msg, - limit_send_size_per_1d = excluded.limit_send_size_per_1d, - limit_send_count_per_1d = excluded.limit_send_count_per_1d;` - -// 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 with default quotas. -// 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, 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 -) values ($1, $2, $3, true, 102400000, 10240, 102400, 1000, 102400000, 10240, 102400, 1000) -on conflict (address_lower) do nothing;` - -const sqlActuals string = `select - coalesce(sum(size) filter (where type = 2), 0) as sent_size_total - , coalesce(sum(size) filter (where type = 2 and ts > now() - interval '1 day'), 0) as sent_size_1d - , 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 - , count(*) filter (where type = 2 and ts > now() - interval '1 day') as sent_count_1d - , count(*) filter (where type = 1 and ts > now() - interval '1 day') as recv_count_1d -from - address_tx -where - address_lower = $1;` +package main + +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 +from + address +where + address_lower = $1;` + +const sqlInsertTx string = `insert into address_tx (address_lower, ts, type, size) VALUES ($1, to_timestamp($2), $3, $4);` + +const sqlUpsertAddress string = `insert into address ( + address_lower, 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 +) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) +on conflict (address_lower) do update set + address = excluded.address, + display_name = excluded.display_name, + accepting_new = excluded.accepting_new, + limit_recv_size_total = excluded.limit_recv_size_total, + limit_recv_size_per_msg = excluded.limit_recv_size_per_msg, + limit_recv_size_per_1d = excluded.limit_recv_size_per_1d, + limit_recv_count_per_1d = excluded.limit_recv_count_per_1d, + limit_send_size_total = excluded.limit_send_size_total, + limit_send_size_per_msg = excluded.limit_send_size_per_msg, + limit_send_size_per_1d = excluded.limit_send_size_per_1d, + limit_send_count_per_1d = excluded.limit_send_count_per_1d;` + +// 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;` + +// sqlActuals sums an address's recorded 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 + , count(*) filter (where type = 1 and ts > now() - interval '1 day') as recv_count_1d + , coalesce(sum(size) filter (where type = 2), 0) as send_size_total + , coalesce(sum(size) filter (where type = 2 and ts > now() - interval '1 day'), 0) as send_size_1d + , count(*) filter (where type = 2 and ts > now() - interval '1 day') as send_count_1d +from + address_tx +where + address_lower = $1;` diff --git a/src/usage_test.go b/src/usage_test.go new file mode 100644 index 0000000..3d6c5e4 --- /dev/null +++ b/src/usage_test.go @@ -0,0 +1,199 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v4/pgxpool" +) + +// usageDomain is used only by these tests; its rows are deleted before and +// after each test. +const usageDomain = "usagetest.example" + +func deleteUsageRows(t *testing.T, p *pgxpool.Pool) { + t.Helper() + mustExec(t, p, `delete from address_tx where address_lower like '%@`+usageDomain+`'`) + mustExec(t, p, `delete from address where address_lower like '%@`+usageDomain+`'`) +} + +// TestAddressDetail_Usage records messages of both types, some older than a +// day, some sharing a timestamp and some under a differently cased address, +// and checks every usage field of the address detail. +func TestAddressDetail_Usage(t *testing.T) { + testDB(t) + deleteUsageRows(t, pool) + t.Cleanup(func() { deleteUsageRows(t, pool) }) + + if w := serveRoute("POST", "/fmsgid", `{"address":"@Usage@UsageTest.example"}`); w.Code != http.StatusCreated { + t.Fatalf("create address: status %d, want 201", w.Code) + } + + now := float64(time.Now().UnixMicro()) / 1e6 + old := now - 2*24*60*60 + tx := func(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) + } + } + + tx("/fmsgid/recv", "@usage@usagetest.example", old, 5000) + tx("/fmsgid/recv", "@usage@usagetest.example", now, 100) + tx("/fmsgid/recv", "@USAGE@usagetest.example", now, 200) // same timestamp, other case + tx("/fmsgid/recv", "@Usage@UsageTest.example", now-60, 300) + tx("/fmsgid/send", "@usage@usagetest.example", old, 1000) + tx("/fmsgid/send", "@usage@usagetest.example", now, 10) + tx("/fmsgid/send", "@usage@USAGETEST.EXAMPLE", now, 20) // same timestamp, other case + + w := serveRoute("GET", "/fmsgid/@usage@usagetest.example", "") + if w.Code != http.StatusOK { + t.Fatalf("GET: status %d, want 200", w.Code) + } + var ad AddressDetail + if err := json.Unmarshal(w.Body.Bytes(), &ad); err != nil { + t.Fatal(err) + } + + got := map[string]int64{ + "RecvSizeTotal": ad.RecvSizeTotal, + "RecvSizePer1d": ad.RecvSizePer1d, + "RecvCountPer1d": ad.RecvCountPer1d, + "SendSizeTotal": ad.SendSizeTotal, + "SendSizePer1d": ad.SendSizePer1d, + "SendCountPer1d": ad.SendCountPer1d, + } + want := map[string]int64{ + "RecvSizeTotal": 5600, + "RecvSizePer1d": 600, + "RecvCountPer1d": 3, + "SendSizeTotal": 1030, + "SendSizePer1d": 30, + "SendCountPer1d": 2, + } + for k, v := range want { + if got[k] != v { + t.Errorf("%s = %d, want %d", k, got[k], v) + } + } +} + +// TestAddressDetail_NoUsage checks an address that has sent and received +// nothing reports zero usage. +func TestAddressDetail_NoUsage(t *testing.T) { + testDB(t) + deleteUsageRows(t, pool) + t.Cleanup(func() { deleteUsageRows(t, pool) }) + + if w := serveRoute("POST", "/fmsgid", `{"address":"@idle@usagetest.example"}`); w.Code != http.StatusCreated { + t.Fatalf("create address: status %d, want 201", w.Code) + } + w := serveRoute("GET", "/fmsgid/@idle@usagetest.example", "") + if w.Code != http.StatusOK { + t.Fatalf("GET: status %d, want 200", w.Code) + } + var ad AddressDetail + if err := json.Unmarshal(w.Body.Bytes(), &ad); err != nil { + t.Fatal(err) + } + if ad.RecvSizeTotal != 0 || ad.RecvSizePer1d != 0 || ad.RecvCountPer1d != 0 || + ad.SendSizeTotal != 0 || ad.SendSizePer1d != 0 || ad.SendCountPer1d != 0 { + t.Errorf("usage not zero: %+v", ad) + } +} + +func TestPostAddressTx_InvalidAddress(t *testing.T) { + for _, route := range []string{"/fmsgid/send", "/fmsgid/recv"} { + for _, addr := range []string{"", "alice@example.com", "@alice"} { + body := fmt.Sprintf(`{"address":%q,"ts":1,"size":1}`, addr) + if w := serveRoute("POST", route, body); w.Code != http.StatusBadRequest { + t.Errorf("POST %s %s: status %d, want 400", route, body, w.Code) + } + } + } +} + +// TestPostCreateAddress_ColumnDefaults checks a new address takes +// accepting_new and every limit from the address table's column defaults, +// using a copy of the table whose defaults differ from dd.sql's. +func TestPostCreateAddress_ColumnDefaults(t *testing.T) { + testDB(t) + ctx := context.Background() + dsn := os.Getenv("FMSGID_TEST_DATABASE_URL") + const schema = "fmsgid_test_defaults" + mustExec(t, pool, `drop schema if exists `+schema+` cascade`) + mustExec(t, pool, `create schema `+schema) + mustExec(t, pool, `create table `+schema+`.address (like public.address including all)`) + t.Cleanup(func() { + p, err := pgxpool.Connect(ctx, dsn) + if err == nil { + _, _ = p.Exec(ctx, `drop schema if exists `+schema+` cascade`) + p.Close() + } + }) + + defaults := []struct { + col string + val int64 + }{ + {"limit_recv_size_total", 11}, + {"limit_recv_size_per_msg", 12}, + {"limit_recv_size_per_1d", 13}, + {"limit_recv_count_per_1d", 14}, + {"limit_send_size_total", 21}, + {"limit_send_size_per_msg", 22}, + {"limit_send_size_per_1d", 23}, + {"limit_send_count_per_1d", 24}, + } + mustExec(t, pool, `alter table `+schema+`.address alter column accepting_new set default false`) + for _, d := range defaults { + mustExec(t, pool, fmt.Sprintf(`alter table %s.address alter column %s set default %d`, schema, d.col, d.val)) + } + + cfg, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + cfg.ConnConfig.RuntimeParams["search_path"] = schema + p, err := pgxpool.ConnectConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer p.Close() + prev := pool + pool = p + defer func() { pool = prev }() + + w := serveRoute("POST", "/fmsgid", `{"address":"@Dflt@Example.com","display_name":"Default"}`) + if w.Code != http.StatusCreated { + t.Fatalf("POST: status %d, want 201", w.Code) + } + + var addr, displayName string + var acceptingNew bool + err = pool.QueryRow(ctx, `select address, display_name, accepting_new from address where address_lower = $1`, + "@dflt@example.com").Scan(&addr, &displayName, &acceptingNew) + if err != nil { + t.Fatal(err) + } + if addr != "@Dflt@Example.com" || displayName != "Default" || acceptingNew { + t.Errorf("address %q display_name %q accepting_new %v", addr, displayName, acceptingNew) + } + for _, d := range defaults { + var v int64 + err := pool.QueryRow(ctx, `select `+d.col+` from address where address_lower = $1`, + "@dflt@example.com").Scan(&v) + if err != nil { + t.Fatal(err) + } + if v != d.val { + t.Errorf("%s = %d, want column default %d", d.col, v, d.val) + } + } +}