Fix usage counters, address_tx key, tx address folding, insert defaults

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
 

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);
 

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 {

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
+}

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

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;`