so much

internal/data/device_tokens.go

@@ -0,0 +1,87 @@
+package data
+
+import (
+	"context"
+	"database/sql"
+	"time"
+)
+
+type DeviceToken struct {
+	ID        int64     `json:"id"`
+	UserID    int64     `json:"user_id"`
+	Token     string    `json:"token"`
+	CreatedAt time.Time `json:"created_at"`
+}
+
+type DeviceTokenModel struct {
+	DB *sql.DB
+}
+
+func (m DeviceTokenModel) Insert(userID int64, token string) error {
+	query := `
+		INSERT INTO device_tokens (user_id, token)
+		VALUES ($1, $2)
+		ON CONFLICT (token) DO UPDATE SET user_id = EXCLUDED.user_id`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	_, err := m.DB.ExecContext(ctx, query, userID, token)
+	return err
+}
+
+func (m DeviceTokenModel) GetForUser(userID int64) ([]DeviceToken, error) {
+	query := `SELECT id, user_id, token, created_at FROM device_tokens WHERE user_id = $1`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	rows, err := m.DB.QueryContext(ctx, query, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var tokens []DeviceToken
+	for rows.Next() {
+		var dt DeviceToken
+		if err := rows.Scan(&dt.ID, &dt.UserID, &dt.Token, &dt.CreatedAt); err != nil {
+			return nil, err
+		}
+		tokens = append(tokens, dt)
+	}
+	return tokens, rows.Err()
+}
+
+func (m DeviceTokenModel) GetAllTokens() ([]DeviceToken, error) {
+	query := `SELECT id, user_id, token, created_at FROM device_tokens`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	rows, err := m.DB.QueryContext(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var tokens []DeviceToken
+	for rows.Next() {
+		var dt DeviceToken
+		if err := rows.Scan(&dt.ID, &dt.UserID, &dt.Token, &dt.CreatedAt); err != nil {
+			return nil, err
+		}
+		tokens = append(tokens, dt)
+	}
+	return tokens, rows.Err()
+}
+
+func (m DeviceTokenModel) Delete(token string) error {
+	query := `DELETE FROM device_tokens WHERE token = $1`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	_, err := m.DB.ExecContext(ctx, query, token)
+	return err
+}

internal/data/models.go

@@ -20,6 +20,8 @@ type Models struct {
 	BlindSigns        BlindSignModel
 	Votes             VoteModel
 	Options           OptionModel
+	ParlWatcher       ParlWatcherModel
+	DeviceTokens      DeviceTokenModel
 }
 
 func NewModels(db *sql.DB) Models {
@@ -33,5 +35,7 @@ func NewModels(db *sql.DB) Models {
 		BlindSigns:        BlindSignModel{DB: db},
 		Votes:             VoteModel{DB: db},
 		Options:           OptionModel{DB: db},
+		ParlWatcher:       ParlWatcherModel{DB: db},
+		DeviceTokens:      DeviceTokenModel{DB: db},
 	}
 }

internal/data/parlwatcher.go

@@ -0,0 +1,82 @@
+package data
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"github.com/lib/pq"
+)
+
+type ParlWatcherModel struct {
+	DB *sql.DB
+}
+
+// FilterNew returns only the numbers from the input that are not yet recorded.
+func (m ParlWatcherModel) FilterNew(numbers []string) ([]string, error) {
+	query := `
+		SELECT unnest($1::text[]) AS num
+		EXCEPT
+		SELECT number FROM parl_notified_votes`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	rows, err := m.DB.QueryContext(ctx, query, pq.Array(numbers))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var result []string
+	for rows.Next() {
+		var n string
+		if err := rows.Scan(&n); err != nil {
+			return nil, err
+		}
+		result = append(result, n)
+	}
+	return result, rows.Err()
+}
+
+// MarkNotified records the given numbers so they are not notified again.
+func (m ParlWatcherModel) MarkNotified(numbers []string) error {
+	query := `
+		INSERT INTO parl_notified_votes (number)
+		SELECT unnest($1::text[])
+		ON CONFLICT (number) DO NOTHING`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := m.DB.ExecContext(ctx, query, pq.Array(numbers))
+	return err
+}
+
+// GetAllNotifiedNumbers returns every number that has already been notified.
+// Unused in the hot path but useful for debugging/admin.
+func (m ParlWatcherModel) GetAllNotifiedNumbers() ([]string, error) {
+	query := `SELECT number FROM parl_notified_votes ORDER BY notified_at DESC`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	rows, err := m.DB.QueryContext(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var result []string
+	for rows.Next() {
+		var n string
+		if err := rows.Scan(&n); err != nil {
+			return nil, err
+		}
+		result = append(result, n)
+	}
+	return result, rows.Err()
+}
+
+// Ensure ParlWatcherModel satisfies the interface even if DB is nil (e.g. in tests).
+var _ = sql.ErrNoRows

internal/data/users.go

@@ -392,3 +392,26 @@ WHERE id = $1`
 
 	return nil
 }
+
+func (m UserModel) GetAllActivated() ([]*User, error) {
+	query := `SELECT id, email, name FROM users WHERE activated = true ORDER BY id`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	rows, err := m.DB.QueryContext(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var users []*User
+	for rows.Next() {
+		u := &User{}
+		if err := rows.Scan(&u.ID, &u.Email, &u.Name); err != nil {
+			return nil, err
+		}
+		users = append(users, u)
+	}
+	return users, rows.Err()
+}