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
	P256dh         string
	Auth           string
	CreatedAt      int64
	LastNotifiedAt 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 the digest watermark to now — the user shouldn't be
// paged for everything published before they opted in.
func AddPushSubscription(sub, endpoint, p256dh, auth string) error {
	now := nowUnix()
	_, err := Get().Exec(`
		INSERT INTO push_subscriptions (endpoint, user_sub, p256dh, auth, created_at, last_notified_at)
		VALUES (?, ?, ?, ?, ?, ?)
		ON CONFLICT(endpoint) DO UPDATE SET
			user_sub = excluded.user_sub,
			p256dh = excluded.p256dh,
			auth = excluded.auth,
			last_notified_at = excluded.last_notified_at`,
		endpoint, sub, p256dh, auth, now, now)
	if err != nil {
		return fmt.Errorf("add push subscription: %w", err)
	}
	return nil
}

// RemovePushSubscription drops one endpoint. Used both for an explicit opt-out
// and when a push service reports the endpoint is gone (404/410).
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
}

// ListPushSubscriptions returns every stored subscription, for the digest sender.
func ListPushSubscriptions() ([]PushSubscription, error) {
	rows, err := Get().Query(
		`SELECT endpoint, user_sub, p256dh, auth, created_at, last_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.P256dh, &p.Auth, &p.CreatedAt, &p.LastNotifiedAt); err != nil {
			return nil, err
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

// 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 <> '_discarded' 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()
}