amadan

rastrillo / aviso Public

Add the subscription store

Ownership rules live in one transaction: an endpoint held by another
subject is refused rather than reassigned, and previousEndpoint only
deletes the caller's own row. confirm and prune match on the revision
the send captured, so a slow send cannot delete a subscription the
browser refreshed meanwhile. Test subscriptions carry real P-256
points and 16-byte auth secrets because webpush-go decodes both before
any HTTP happens.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev de0ae528cd7fb452e45f2a9a60619df5d041d20a parent c3ed31a
2 files changed, +349 −0
  • store.go +131 −0
  • store_test.go +218 −0
diff --git a/store.go b/store.go
new file mode 100644
index 0000000..1772747
--- /dev/null
+++ b/store.go
@@ -0,0 +1,131 @@
+package aviso
+
+import (
+ "context"
+ "crypto/rand"
+ "database/sql"
+ "encoding/base64"
+ "errors"
+ "fmt"
+ "time"
+)
+
+func newID() (string, error) {
+ b := make([]byte, 16)
+ if _, err := rand.Read(b); err != nil {
+ return "", err
+ }
+ return base64.RawURLEncoding.EncodeToString(b), nil
+}
+
+const selectCols = `id, endpoint, subject, p256dh, auth, vapid_key_id, revision`
+
+func scanStored(rows *sql.Rows) ([]Stored, error) {
+ var out []Stored
+ for rows.Next() {
+ var st Stored
+ if err := rows.Scan(&st.ID, &st.Endpoint, &st.Subject, &st.P256dh, &st.Auth, &st.VAPIDKeyID, &st.Revision); err != nil {
+ return nil, err
+ }
+ out = append(out, st)
+ }
+ return out, rows.Err()
+}
+
+// List returns every device subject has enrolled, oldest first.
+func (s *Service) List(ctx context.Context, subject string) ([]Stored, error) {
+ rows, err := s.cfg.DB.QueryContext(ctx,
+ `SELECT `+selectCols+` FROM aviso_subscriptions WHERE subject = ? ORDER BY created_at, id`, subject)
+ if err != nil {
+ return nil, fmt.Errorf("aviso: list: %w", err)
+ }
+ defer rows.Close()
+ return scanStored(rows)
+}
+
+// DeleteSubject removes every device subject enrolled — account
+// deletion's hook.
+func (s *Service) DeleteSubject(ctx context.Context, subject string) error {
+ _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE subject = ?`, subject)
+ if err != nil {
+ return fmt.Errorf("aviso: delete subject: %w", err)
+ }
+ return nil
+}
+
+// Sweep deletes subscriptions not confirmed since t. Confirmation
+// moves on reconcile and on an accepted send, so this measures whether
+// the subscription is alive, not whether the person still wants it.
+func (s *Service) Sweep(ctx context.Context, t time.Time) error {
+ _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE last_confirmed_at < ?`, t.Unix())
+ if err != nil {
+ return fmt.Errorf("aviso: sweep: %w", err)
+ }
+ return nil
+}
+
+// put is Subscribe's write: insert, or update when the same subject
+// already holds the endpoint, in one transaction with the optional
+// previousEndpoint delete — which only removes the caller's own row,
+// so naming someone else's endpoint is a no-op rather than a weapon.
+func (s *Service) put(ctx context.Context, subject string, sub Subscription, previousEndpoint string) error {
+ tx, err := s.cfg.DB.BeginTx(ctx, nil)
+ if err != nil {
+ return fmt.Errorf("aviso: put: %w", err)
+ }
+ defer tx.Rollback()
+ now := s.now().Unix()
+ var owner string
+ err = tx.QueryRowContext(ctx, `SELECT subject FROM aviso_subscriptions WHERE endpoint = ?`, sub.Endpoint).Scan(&owner)
+ switch {
+ case err == nil && owner != subject:
+ return ErrOwnedElsewhere
+ case err == nil:
+ _, err = tx.ExecContext(ctx, `UPDATE aviso_subscriptions
+ SET p256dh = ?, auth = ?, vapid_key_id = ?, revision = revision + 1, last_confirmed_at = ?
+ WHERE endpoint = ?`, sub.P256dh, sub.Auth, s.keyID, now, sub.Endpoint)
+ case errors.Is(err, sql.ErrNoRows):
+ var id string
+ if id, err = newID(); err != nil {
+ return err
+ }
+ _, err = tx.ExecContext(ctx, `INSERT INTO aviso_subscriptions
+ (id, endpoint, subject, p256dh, auth, vapid_key_id, revision, created_at, last_confirmed_at)
+ VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)`, id, sub.Endpoint, subject, sub.P256dh, sub.Auth, s.keyID, now, now)
+ }
+ if err != nil {
+ return fmt.Errorf("aviso: put: %w", err)
+ }
+ if previousEndpoint != "" && previousEndpoint != sub.Endpoint {
+ if _, err := tx.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE endpoint = ? AND subject = ?`, previousEndpoint, subject); err != nil {
+ return fmt.Errorf("aviso: put: %w", err)
+ }
+ }
+ return tx.Commit()
+}
+
+// deleteOwn removes endpoint only if subject holds it; nil either way,
+// because Unsubscribe is idempotent and must not confirm whether an
+// endpoint exists to someone who does not own it.
+func (s *Service) deleteOwn(ctx context.Context, subject, endpoint string) error {
+ _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE endpoint = ? AND subject = ?`, endpoint, subject)
+ if err != nil {
+ return fmt.Errorf("aviso: unsubscribe: %w", err)
+ }
+ return nil
+}
+
+// confirm bumps last_confirmed_at for an accepted send, only if the
+// row is still at the revision the send captured.
+func (s *Service) confirm(ctx context.Context, id string, revision int64) error {
+ _, err := s.cfg.DB.ExecContext(ctx, `UPDATE aviso_subscriptions SET last_confirmed_at = ? WHERE id = ? AND revision = ?`, s.now().Unix(), id, revision)
+ return err
+}
+
+// prune deletes a row the push service reported gone, only if it is
+// still at the revision the send captured: a browser that refreshed
+// meanwhile has a live subscription under the same id.
+func (s *Service) prune(ctx context.Context, id string, revision int64) error {
+ _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE id = ? AND revision = ?`, id, revision)
+ return err
+}
diff --git a/store_test.go b/store_test.go
new file mode 100644
index 0000000..987537b
--- /dev/null
+++ b/store_test.go
@@ -0,0 +1,218 @@
+package aviso
+
+import (
+ "context"
+ "crypto/ecdh"
+ "crypto/rand"
+ "database/sql"
+ "encoding/base64"
+ "errors"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "amadan.net/rastrillo/rastrillo/db"
+ "amadan.net/rastrillo/rastrillo/migrate"
+)
+
+func openInternalDB(t *testing.T) *sql.DB {
+ t.Helper()
+ d, err := db.Open(filepath.Join(t.TempDir(), "aviso.db"), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { d.Close() })
+ if _, err := migrate.Apply(context.Background(), d, Schema); err != nil {
+ t.Fatal(err)
+ }
+ return d.Writer()
+}
+
+func newInternalService(t *testing.T) *Service {
+ t.Helper()
+ key, _ := GenerateKey()
+ s, err := New(Config{DB: openInternalDB(t), PrivateKey: key, Contact: "mailto:x@y", Origin: "https://a"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return s
+}
+
+// sub builds a subscription with real keys: webpush-go decodes p256dh
+// into a P-256 point and auth into 16 bytes before any HTTP happens,
+// so a placeholder string would fail in the encryptor, not the test.
+func sub(endpoint string) Subscription {
+ k, err := ecdh.P256().GenerateKey(rand.Reader)
+ if err != nil {
+ panic(err)
+ }
+ auth := make([]byte, 16)
+ if _, err := rand.Read(auth); err != nil {
+ panic(err)
+ }
+ return Subscription{
+ Endpoint: endpoint,
+ P256dh: base64.RawURLEncoding.EncodeToString(k.PublicKey().Bytes()),
+ Auth: base64.RawURLEncoding.EncodeToString(auth),
+ }
+}
+
+func TestPutInsertsThenUpdatesSameOwner(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ if err := s.put(ctx, "alice", sub("https://push.example/1"), ""); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ := s.List(ctx, "alice")
+ if len(rows) != 1 || rows[0].Revision != 1 || rows[0].VAPIDKeyID != s.keyID || rows[0].Subject != "alice" {
+ t.Fatalf("after insert: %+v", rows)
+ }
+ again := sub("https://push.example/1")
+ if err := s.put(ctx, "alice", again, ""); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ = s.List(ctx, "alice")
+ if len(rows) != 1 || rows[0].Revision != 2 || rows[0].Auth != again.Auth {
+ t.Fatalf("after update: %+v", rows)
+ }
+}
+
+func TestPutRefusesCrossOwner(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub("https://push.example/1"), "")
+ err := s.put(ctx, "bob", sub("https://push.example/1"), "")
+ if !errors.Is(err, ErrOwnedElsewhere) {
+ t.Fatalf("got %v, want ErrOwnedElsewhere", err)
+ }
+ rows, _ := s.List(ctx, "alice")
+ if len(rows) != 1 {
+ t.Fatal("alice lost her row")
+ }
+ if rows, _ := s.List(ctx, "bob"); len(rows) != 0 {
+ t.Fatal("bob gained a row")
+ }
+}
+
+func TestPutDeletesPreviousOnlyWhenOwned(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub("https://push.example/old"), "")
+ _ = s.put(ctx, "bob", sub("https://push.example/bobs"), "")
+ // alice re-subscribes and names her old endpoint: gone.
+ if err := s.put(ctx, "alice", sub("https://push.example/new"), "https://push.example/old"); err != nil {
+ t.Fatal(err)
+ }
+ rows, _ := s.List(ctx, "alice")
+ if len(rows) != 1 || rows[0].Endpoint != "https://push.example/new" {
+ t.Fatalf("alice rows: %+v", rows)
+ }
+ // alice names bob's endpoint as previous: bob keeps it.
+ _ = s.put(ctx, "alice", sub("https://push.example/new2"), "https://push.example/bobs")
+ rows, _ = s.List(ctx, "bob")
+ if len(rows) != 1 {
+ t.Fatal("bob's row deleted by alice's previousEndpoint")
+ }
+}
+
+func TestPutPreviousEqualToNewIsNotADelete(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub("https://push.example/same"), "")
+ if err := s.put(ctx, "alice", sub("https://push.example/same"), "https://push.example/same"); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 1 || rows[0].Revision != 2 {
+ t.Fatalf("rows: %+v", rows)
+ }
+}
+
+func TestDeleteOwnIsOwnerScoped(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub("https://push.example/1"), "")
+ if err := s.deleteOwn(ctx, "bob", "https://push.example/1"); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 1 {
+ t.Fatal("bob deleted alice's row")
+ }
+ _ = s.deleteOwn(ctx, "alice", "https://push.example/1")
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 0 {
+ t.Fatal("own delete did nothing")
+ }
+}
+
+func TestConfirmAndPruneAreRevisionConditional(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ _ = s.put(ctx, "alice", sub("https://push.example/1"), "")
+ before, _ := s.List(ctx, "alice")
+ id := before[0].ID
+ // Browser refreshes: revision 2.
+ _ = s.put(ctx, "alice", sub("https://push.example/1"), "")
+ // A send that captured revision 1 comes back 410: must not prune.
+ if err := s.prune(ctx, id, 1); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 1 {
+ t.Fatal("stale prune deleted a refreshed subscription")
+ }
+ // Stale confirm must not touch last_confirmed_at.
+ s.now = func() time.Time { return time.Unix(1_800_000_000, 0) }
+ _ = s.confirm(ctx, id, 1)
+ var got int64
+ _ = s.cfg.DB.QueryRow(`SELECT last_confirmed_at FROM aviso_subscriptions WHERE id=?`, id).Scan(&got)
+ if got == 1_800_000_000 {
+ t.Fatal("stale confirm bumped last_confirmed_at")
+ }
+ _ = s.confirm(ctx, id, 2)
+ _ = s.cfg.DB.QueryRow(`SELECT last_confirmed_at FROM aviso_subscriptions WHERE id=?`, id).Scan(&got)
+ if got != 1_800_000_000 {
+ t.Fatalf("current confirm did not bump: %d", got)
+ }
+ if err := s.prune(ctx, id, 2); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 0 {
+ t.Fatal("current prune did not delete")
+ }
+}
+
+func TestSweepAndDeleteSubject(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ s.now = func() time.Time { return time.Unix(1000, 0) }
+ _ = s.put(ctx, "alice", sub("https://push.example/old"), "")
+ s.now = func() time.Time { return time.Unix(2000, 0) }
+ _ = s.put(ctx, "alice", sub("https://push.example/new"), "")
+ _ = s.put(ctx, "bob", sub("https://push.example/bob"), "")
+ if err := s.Sweep(ctx, time.Unix(1500, 0)); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 1 || rows[0].Endpoint != "https://push.example/new" {
+ t.Fatalf("sweep: %+v", rows)
+ }
+ if err := s.DeleteSubject(ctx, "bob"); err != nil {
+ t.Fatal(err)
+ }
+ if rows, _ := s.List(ctx, "bob"); len(rows) != 0 {
+ t.Fatal("DeleteSubject left rows")
+ }
+ if rows, _ := s.List(ctx, "alice"); len(rows) != 1 {
+ t.Fatal("DeleteSubject touched another subject")
+ }
+}
+
+func TestListOrdersOldestFirst(t *testing.T) {
+ s := newInternalService(t)
+ ctx := context.Background()
+ s.now = func() time.Time { return time.Unix(2000, 0) }
+ _ = s.put(ctx, "alice", sub("https://push.example/second"), "")
+ s.now = func() time.Time { return time.Unix(1000, 0) }
+ _ = s.put(ctx, "alice", sub("https://push.example/first"), "")
+ rows, _ := s.List(ctx, "alice")
+ if len(rows) != 2 || rows[0].Endpoint != "https://push.example/first" {
+ t.Fatalf("order: %+v", rows)
+ }
+}