Pete/internal/storage/push.go

package storage

import "fmt"

// PushSubscription is one browser/device endpoint a signed-in user has opted in
// for Web Push digests. See the push_subscriptions schema for the field roles.
type PushSubscription struct {
	Endpoint          string
	UserSub           string
	Localpart         string
	P256dh            string
	Auth              string
	CreatedAt         int64
	LastNotifiedAt    int64
	LastAdvNotifiedAt int64
}

// AddPushSubscription records (or refreshes) a push endpoint for a user. The
// endpoint is the primary key, so a re-subscribe from the same browser updates
// the keys and resets both watermarks to now — the user shouldn't be paged for
// everything published before they opted in.
//
// localpart is the session's Matrix handle, refreshed on every re-subscribe so a
// row stored by a build that predated adventure alerts heals itself the first
// time that browser subscribes again. It may legitimately be empty (a session
// minted before the game economy existed carries no username); such a row simply
// never matches an owner-scoped alert.
func AddPushSubscription(sub, localpart, endpoint, p256dh, auth string) error {
	now := nowUnix()
	_, err := Get().Exec(`
		INSERT INTO push_subscriptions
			(endpoint, user_sub, user_localpart, p256dh, auth, created_at, last_notified_at, last_adv_notified_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(endpoint) DO UPDATE SET
			user_sub = excluded.user_sub,
			user_localpart = excluded.user_localpart,
			p256dh = excluded.p256dh,
			auth = excluded.auth,
			last_notified_at = excluded.last_notified_at,
			last_adv_notified_at = excluded.last_adv_notified_at`,
		endpoint, sub, localpart, p256dh, auth, now, now, now)
	if err != nil {
		return fmt.Errorf("add push subscription: %w", err)
	}
	return nil
}

// RemovePushSubscription drops one endpoint regardless of owner. Reserved for
// the digest sender's prune path, where a push service has reported the endpoint
// gone (404/410) and there's no caller identity to scope by. User-initiated
// opt-outs must use RemovePushSubscriptionForUser.
func RemovePushSubscription(endpoint string) error {
	_, err := Get().Exec(`DELETE FROM push_subscriptions WHERE endpoint = ?`, endpoint)
	if err != nil {
		return fmt.Errorf("remove push subscription: %w", err)
	}
	return nil
}

// RemovePushSubscriptionForUser drops an endpoint only if it belongs to sub, so
// a signed-in user can't unsubscribe another account's device by presenting its
// endpoint string. A no-op (no matching row) is not an error.
func RemovePushSubscriptionForUser(sub, endpoint string) error {
	_, err := Get().Exec(
		`DELETE FROM push_subscriptions WHERE endpoint = ? AND user_sub = ?`, endpoint, sub)
	if err != nil {
		return fmt.Errorf("remove push subscription: %w", err)
	}
	return nil
}

// ListPushSubscriptions returns every stored subscription, for the digest sender.
func ListPushSubscriptions() ([]PushSubscription, error) {
	rows, err := Get().Query(
		`SELECT endpoint, user_sub, user_localpart, p256dh, auth,
		        created_at, last_notified_at, last_adv_notified_at
		 FROM push_subscriptions`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []PushSubscription
	for rows.Next() {
		var p PushSubscription
		if err := rows.Scan(&p.Endpoint, &p.UserSub, &p.Localpart, &p.P256dh, &p.Auth,
			&p.CreatedAt, &p.LastNotifiedAt, &p.LastAdvNotifiedAt); err != nil {
			return nil, err
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

// TouchAdvPushSubscription advances an endpoint's *adventure alert* watermark, so
// the next pass only considers dispatches that occurred after ts. Kept separate
// from TouchPushSubscription for the reason the schema gives: the digest and the
// alerts must not be able to consume each other's backlog.
func TouchAdvPushSubscription(endpoint string, ts int64) error {
	_, err := Get().Exec(
		`UPDATE push_subscriptions SET last_adv_notified_at = ? WHERE endpoint = ?`, ts, endpoint)
	if err != nil {
		return fmt.Errorf("touch adventure push watermark: %w", err)
	}
	return nil
}

// TouchPushSubscription advances an endpoint's digest watermark so its next
// digest only considers stories seen after ts.
func TouchPushSubscription(endpoint string, ts int64) error {
	_, err := Get().Exec(
		`UPDATE push_subscriptions SET last_notified_at = ? WHERE endpoint = ?`, ts, endpoint)
	if err != nil {
		return fmt.Errorf("touch push subscription: %w", err)
	}
	return nil
}

// NewClassifiedSince returns classified stories first seen after sinceUnix,
// newest first, capped at limit. The digest sender uses it to count and preview
// what's new for a subscriber; it carries just the fields a digest needs.
func NewClassifiedSince(sinceUnix int64, limit int) ([]Story, error) {
	rows, err := Get().Query(
		`SELECT id, headline, source, channel, seen_at
		 FROM stories
		 WHERE classified = 1 AND channel NOT IN ('_discarded', '_duplicate') AND seen_at > ?
		 ORDER BY seen_at DESC
		 LIMIT ?`, sinceUnix, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Story
	for rows.Next() {
		var s Story
		if err := rows.Scan(&s.ID, &s.Headline, &s.Source, &s.Channel, &s.SeenAt); err != nil {
			return nil, err
		}
		out = append(out, s)
	}
	return out, rows.Err()
}