| 1 | package aviso |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "database/sql" |
| 7 | "encoding/base64" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | func newID() (string, error) { |
| 14 | b := make([]byte, 16) |
| 15 | if _, err := rand.Read(b); err != nil { |
| 16 | return "", err |
| 17 | } |
| 18 | return base64.RawURLEncoding.EncodeToString(b), nil |
| 19 | } |
| 20 | |
| 21 | const selectCols = `id, endpoint, subject, p256dh, auth, vapid_key_id, revision` |
| 22 | |
| 23 | func scanStored(rows *sql.Rows) ([]Stored, error) { |
| 24 | var out []Stored |
| 25 | for rows.Next() { |
| 26 | var st Stored |
| 27 | if err := rows.Scan(&st.ID, &st.Endpoint, &st.Subject, &st.P256dh, &st.Auth, &st.VAPIDKeyID, &st.Revision); err != nil { |
| 28 | return nil, err |
| 29 | } |
| 30 | out = append(out, st) |
| 31 | } |
| 32 | return out, rows.Err() |
| 33 | } |
| 34 | |
| 35 | // List returns every device subject has enrolled, oldest first. |
| 36 | func (s *Service) List(ctx context.Context, subject string) ([]Stored, error) { |
| 37 | rows, err := s.cfg.DB.QueryContext(ctx, |
| 38 | `SELECT `+selectCols+` FROM aviso_subscriptions WHERE subject = ? ORDER BY created_at, id`, subject) |
| 39 | if err != nil { |
| 40 | return nil, fmt.Errorf("aviso: list: %w", err) |
| 41 | } |
| 42 | defer rows.Close() |
| 43 | return scanStored(rows) |
| 44 | } |
| 45 | |
| 46 | // DeleteSubject removes every device subject enrolled — account |
| 47 | // deletion's hook. |
| 48 | func (s *Service) DeleteSubject(ctx context.Context, subject string) error { |
| 49 | _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE subject = ?`, subject) |
| 50 | if err != nil { |
| 51 | return fmt.Errorf("aviso: delete subject: %w", err) |
| 52 | } |
| 53 | return nil |
| 54 | } |
| 55 | |
| 56 | // Sweep deletes subscriptions not confirmed since t. Confirmation |
| 57 | // moves on reconcile and on an accepted send, so this measures whether |
| 58 | // the subscription is alive, not whether the person still wants it. |
| 59 | func (s *Service) Sweep(ctx context.Context, t time.Time) error { |
| 60 | _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE last_confirmed_at < ?`, t.Unix()) |
| 61 | if err != nil { |
| 62 | return fmt.Errorf("aviso: sweep: %w", err) |
| 63 | } |
| 64 | return nil |
| 65 | } |
| 66 | |
| 67 | // put is Subscribe's write: insert, or update when the same subject |
| 68 | // already holds the endpoint, in one transaction with the optional |
| 69 | // previousEndpoint delete — which only removes the caller's own row, |
| 70 | // so naming someone else's endpoint is a no-op rather than a weapon. |
| 71 | func (s *Service) put(ctx context.Context, subject string, sub Subscription, previousEndpoint string) error { |
| 72 | tx, err := s.cfg.DB.BeginTx(ctx, nil) |
| 73 | if err != nil { |
| 74 | return fmt.Errorf("aviso: put: %w", err) |
| 75 | } |
| 76 | defer tx.Rollback() |
| 77 | now := s.now().Unix() |
| 78 | var owner string |
| 79 | err = tx.QueryRowContext(ctx, `SELECT subject FROM aviso_subscriptions WHERE endpoint = ?`, sub.Endpoint).Scan(&owner) |
| 80 | switch { |
| 81 | case err == nil && owner != subject: |
| 82 | return ErrOwnedElsewhere |
| 83 | case err == nil: |
| 84 | _, err = tx.ExecContext(ctx, `UPDATE aviso_subscriptions |
| 85 | SET p256dh = ?, auth = ?, vapid_key_id = ?, revision = revision + 1, last_confirmed_at = ? |
| 86 | WHERE endpoint = ?`, sub.P256dh, sub.Auth, s.keyID, now, sub.Endpoint) |
| 87 | case errors.Is(err, sql.ErrNoRows): |
| 88 | var id string |
| 89 | if id, err = newID(); err != nil { |
| 90 | return err |
| 91 | } |
| 92 | _, err = tx.ExecContext(ctx, `INSERT INTO aviso_subscriptions |
| 93 | (id, endpoint, subject, p256dh, auth, vapid_key_id, revision, created_at, last_confirmed_at) |
| 94 | VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?)`, id, sub.Endpoint, subject, sub.P256dh, sub.Auth, s.keyID, now, now) |
| 95 | } |
| 96 | if err != nil { |
| 97 | return fmt.Errorf("aviso: put: %w", err) |
| 98 | } |
| 99 | if previousEndpoint != "" && previousEndpoint != sub.Endpoint { |
| 100 | if _, err := tx.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE endpoint = ? AND subject = ?`, previousEndpoint, subject); err != nil { |
| 101 | return fmt.Errorf("aviso: put: %w", err) |
| 102 | } |
| 103 | } |
| 104 | return tx.Commit() |
| 105 | } |
| 106 | |
| 107 | // deleteOwn removes endpoint only if subject holds it; nil either way, |
| 108 | // because Unsubscribe is idempotent and must not confirm whether an |
| 109 | // endpoint exists to someone who does not own it. |
| 110 | func (s *Service) deleteOwn(ctx context.Context, subject, endpoint string) error { |
| 111 | _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE endpoint = ? AND subject = ?`, endpoint, subject) |
| 112 | if err != nil { |
| 113 | return fmt.Errorf("aviso: unsubscribe: %w", err) |
| 114 | } |
| 115 | return nil |
| 116 | } |
| 117 | |
| 118 | // confirm bumps last_confirmed_at for an accepted send, only if the |
| 119 | // row is still at the revision the send captured. |
| 120 | func (s *Service) confirm(ctx context.Context, id string, revision int64) error { |
| 121 | _, err := s.cfg.DB.ExecContext(ctx, `UPDATE aviso_subscriptions SET last_confirmed_at = ? WHERE id = ? AND revision = ?`, s.now().Unix(), id, revision) |
| 122 | return err |
| 123 | } |
| 124 | |
| 125 | // prune deletes a row the push service reported gone, only if it is |
| 126 | // still at the revision the send captured: a browser that refreshed |
| 127 | // meanwhile has a live subscription under the same id. |
| 128 | func (s *Service) prune(ctx context.Context, id string, revision int64) error { |
| 129 | _, err := s.cfg.DB.ExecContext(ctx, `DELETE FROM aviso_subscriptions WHERE id = ? AND revision = ?`, id, revision) |
| 130 | return err |
| 131 | } |
| 132 | |
amadan