package storage import ( "context" "database/sql" "fmt" "time" "github.com/imjasonh/webpush" _ "github.com/mattn/go-sqlite3" // SQLite driver ) // SQLite implements storage using SQLite. type SQLite struct { db *sql.DB } // NewSQLite creates a new SQLite storage. // dsn is the data source name, e.g., "webpush.db" or ":memory:". func NewSQLite(dsn string) (*SQLite, error) { db, err := sql.Open("sqlite3", dsn) if err != nil { return nil, fmt.Errorf("opening database: %w", err) } // Create table if it doesn't exist _, err = db.Exec(` CREATE TABLE IF NOT EXISTS subscriptions ( id TEXT PRIMARY KEY, user_id TEXT, endpoint TEXT NOT NULL UNIQUE, p256dh TEXT NOT NULL, auth TEXT NOT NULL, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ); CREATE INDEX IF NOT EXISTS idx_user_id ON subscriptions(user_id); CREATE INDEX IF NOT EXISTS idx_endpoint ON subscriptions(endpoint); `) if err != nil { db.Close() return nil, fmt.Errorf("creating table: %w", err) } return &SQLite{db: db}, nil } // Save stores or updates a subscription. func (s *SQLite) Save(ctx context.Context, record *Record) error { now := time.Now() if record.CreatedAt.IsZero() { record.CreatedAt = now } record.UpdatedAt = now _, err := s.db.ExecContext(ctx, ` INSERT INTO subscriptions (id, user_id, endpoint, p256dh, auth, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET user_id = excluded.user_id, endpoint = excluded.endpoint, p256dh = excluded.p256dh, auth = excluded.auth, updated_at = excluded.updated_at `, record.ID, record.UserID, record.Subscription.Endpoint, record.Subscription.Keys.P256dh, record.Subscription.Keys.Auth, record.CreatedAt, record.UpdatedAt, ) if err != nil { return fmt.Errorf("saving subscription: %w", err) } return nil } // Get retrieves a subscription by ID. func (s *SQLite) Get(ctx context.Context, id string) (*Record, error) { row := s.db.QueryRowContext(ctx, ` SELECT id, user_id, endpoint, p256dh, auth, created_at, updated_at FROM subscriptions WHERE id = ? `, id) return scanRecord(row) } // GetByEndpoint retrieves a subscription by its endpoint URL. func (s *SQLite) GetByEndpoint(ctx context.Context, endpoint string) (*Record, error) { row := s.db.QueryRowContext(ctx, ` SELECT id, user_id, endpoint, p256dh, auth, created_at, updated_at FROM subscriptions WHERE endpoint = ? `, endpoint) return scanRecord(row) } // GetByUserID retrieves all subscriptions for a user. func (s *SQLite) GetByUserID(ctx context.Context, userID string) ([]*Record, error) { rows, err := s.db.QueryContext(ctx, ` SELECT id, user_id, endpoint, p256dh, auth, created_at, updated_at FROM subscriptions WHERE user_id = ? `, userID) if err != nil { return nil, fmt.Errorf("querying subscriptions: %w", err) } defer rows.Close() return scanRecords(rows) } // Delete removes a subscription by ID. func (s *SQLite) Delete(ctx context.Context, id string) error { result, err := s.db.ExecContext(ctx, "DELETE FROM subscriptions WHERE id = ?", id) if err != nil { return fmt.Errorf("deleting subscription: %w", err) } n, err := result.RowsAffected() if err != nil { return fmt.Errorf("checking rows affected: %w", err) } if n == 0 { return ErrNotFound } return nil } // DeleteByEndpoint removes a subscription by its endpoint URL. func (s *SQLite) DeleteByEndpoint(ctx context.Context, endpoint string) error { result, err := s.db.ExecContext(ctx, "DELETE FROM subscriptions WHERE endpoint = ?", endpoint) if err != nil { return fmt.Errorf("deleting subscription: %w", err) } n, err := result.RowsAffected() if err != nil { return fmt.Errorf("checking rows affected: %w", err) } if n == 0 { return ErrNotFound } return nil } // List returns all subscriptions with pagination. func (s *SQLite) List(ctx context.Context, limit, offset int) ([]*Record, error) { rows, err := s.db.QueryContext(ctx, ` SELECT id, user_id, endpoint, p256dh, auth, created_at, updated_at FROM subscriptions ORDER BY created_at DESC LIMIT ? OFFSET ? `, limit, offset) if err != nil { return nil, fmt.Errorf("querying subscriptions: %w", err) } defer rows.Close() return scanRecords(rows) } // Close closes the database connection. func (s *SQLite) Close() error { return s.db.Close() } type scanner interface { Scan(dest ...interface{}) error } func scanRecord(row scanner) (*Record, error) { var ( id string userID sql.NullString endpoint string p256dh string auth string createdAt time.Time updatedAt time.Time ) err := row.Scan(&id, &userID, &endpoint, &p256dh, &auth, &createdAt, &updatedAt) if err == sql.ErrNoRows { return nil, ErrNotFound } if err != nil { return nil, fmt.Errorf("scanning row: %w", err) } return &Record{ ID: id, UserID: userID.String, CreatedAt: createdAt, UpdatedAt: updatedAt, Subscription: &webpush.Subscription{ Endpoint: endpoint, Keys: webpush.Keys{ P256dh: p256dh, Auth: auth, }, }, }, nil } func scanRecords(rows *sql.Rows) ([]*Record, error) { var records []*Record for rows.Next() { record, err := scanRecord(rows) if err != nil { return nil, err } records = append(records, record) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterating rows: %w", err) } return records, nil }