codegirl007/plumber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Address PR review: graceful shutdown, Role/NewUser, drop SQLite.

Browse Source
This commit is contained in:
codegirl007
2026-08-21 23:40:59 -07:00
parent d167b9216a
commit 3391cce7bd
13 changed files with 152 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/blob/spaces.go
+13 -13
View File
@@ -21,12 +21,6 @@ type Uploader interface {
// Disabled is a no-op uploader used when Spaces is not configured.
type Disabled struct{}
func (Disabled) Enabled() bool { return false }
func (Disabled) Upload(context.Context, string, io.Reader, string, int64) (string, error) {
return "", fmt.Errorf("avatar uploads are not configured")
}
// SpacesConfig holds DigitalOcean Spaces settings.
type SpacesConfig struct {
Key string
@@ -37,6 +31,17 @@ type SpacesConfig struct {
CDNBase string // optional public base URL without trailing slash
}
type spaces struct {
client *s3.Client
cfg SpacesConfig
}
func (Disabled) Enabled() bool { return false }
func (Disabled) Upload(context.Context, string, io.Reader, string, int64) (string, error) {
return "", fmt.Errorf("avatar uploads are not configured")
}
// NewSpaces returns an Uploader when required env is present; otherwise Disabled.
func NewSpaces(cfg SpacesConfig) Uploader {
cfg.Key = strings.TrimSpace(cfg.Key)
@@ -49,18 +54,13 @@ func NewSpaces(cfg SpacesConfig) Uploader {
return Disabled{}
}
client := s3.New(s3.Options{
Region: cfg.Region,
Credentials: credentials.NewStaticCredentialsProvider(cfg.Key, cfg.Secret, ""),
Region: cfg.Region,
Credentials: credentials.NewStaticCredentialsProvider(cfg.Key, cfg.Secret, ""),
BaseEndpoint: aws.String(cfg.Endpoint),
})
return &spaces{client: client, cfg: cfg}
}
type spaces struct {
client *s3.Client
cfg SpacesConfig
}
func (s *spaces) Enabled() bool { return true }
func (s *spaces) Upload(ctx context.Context, key string, body io.Reader, contentType string, size int64) (string, error) {
internal/store/db.go
+17 -2
View File
@@ -8,15 +8,30 @@ import (
// ErrLastAdmin is returned when demoting the only remaining admin.
var ErrLastAdmin = errors.New("cannot demote the last admin")
// Role is a user privilege level stored in users.role.
type Role string
const (
RoleUser Role = "user"
RoleAdmin Role = "admin"
)
// NewUser is the input for CreateUser.
type NewUser struct {
Username string
PasswordHash string
Role Role
}
// DB is the persistence API used by the web layer.
// Named DB to avoid colliding with scs.Store.
type DB interface {
CreateUser(ctx context.Context, username, passwordHash string, asAdmin bool) (*User, error)
CreateUser(ctx context.Context, user NewUser) (*User, error)
UserByID(ctx context.Context, id string) (*User, error)
UserByUsername(ctx context.Context, username string) (*User, error)
CountAdmins(ctx context.Context) (int, error)
ListUsers(ctx context.Context) ([]User, error)
SetRole(ctx context.Context, userID, role string) error
SetRole(ctx context.Context, userID string, role Role) error
CreateQuestion(ctx context.Context, authorID, title, body, city string) (*RankedQuestion, error)
ListHunt(ctx context.Context, huntDate, viewerID string) ([]RankedQuestion, error)
GetQuestion(ctx context.Context, id, viewerID string) (*RankedQuestion, error)
internal/store/migrate.go
+3 -13
View File
@@ -3,24 +3,14 @@ package store
import (
"database/sql"
"fmt"
"strings"
)
func migrateUserProfileColumns(db *sql.DB, dialect string) error {
// migrateUserProfileColumns adds avatar_url and state when missing (existing DBs).
func migrateUserProfileColumns(db *sql.DB) error {
cols := []string{"avatar_url", "state"}
for _, col := range cols {
var stmt string
switch dialect {
case dialectPostgres:
stmt = fmt.Sprintf(`ALTER TABLE users ADD COLUMN IF NOT EXISTS %s TEXT NOT NULL DEFAULT ''`, col)
default:
stmt = fmt.Sprintf(`ALTER TABLE users ADD COLUMN %s TEXT NOT NULL DEFAULT ''`, col)
}
stmt := fmt.Sprintf(`ALTER TABLE users ADD COLUMN IF NOT EXISTS %s TEXT NOT NULL DEFAULT ''`, col)
if _, err := db.Exec(stmt); err != nil {
// SQLite errors when the column already exists.
if dialect == dialectSQLite && strings.Contains(strings.ToLower(err.Error()), "duplicate column") {
continue
}
return fmt.Errorf("add column %s: %w", col, err)
}
}
internal/store/postgres.go
+10 -33
View File
@@ -11,11 +11,7 @@ import (
_ "github.com/jackc/pgx/v5/stdlib"
)
const (
dialectSQLite = "sqlite"
dialectPostgres = "postgres"
)
// rebind converts ? placeholders to Postgres $1, $2, ... form.
func rebind(query string) string {
n := 0
var b strings.Builder
@@ -31,13 +27,12 @@ func rebind(query string) string {
return b.String()
}
// q rebinds SQL placeholders for Postgres.
func (s *Store) q(query string) string {
if s.dialect == dialectPostgres {
return rebind(query)
}
return query
return rebind(query)
}
// applySchema runs semicolon-separated DDL statements, skipping PRAGMA lines.
func applySchema(db *sql.DB, schema string) error {
for _, stmt := range strings.Split(schema, ";") {
stmt = strings.TrimSpace(stmt)
@@ -55,6 +50,7 @@ func applySchema(db *sql.DB, schema string) error {
return nil
}
// postgresDSN normalizes DATABASE_URL for pgx (sslmode default, strip unsupported params).
func postgresDSN(raw string) (string, error) {
u, err := url.Parse(raw)
if err != nil {
@@ -77,16 +73,8 @@ func postgresDSN(raw string) (string, error) {
return u.String(), nil
}
// OpenPostgres connects to Postgres, applies schema/migrations, and starts session cleanup.
func OpenPostgres(databaseURL, schema string) (*Store, error) {
return openPostgres(databaseURL, schema, 5*time.Minute)
}
// OpenPostgresWithoutSessionCleanup opens Postgres without a session cleanup goroutine (for tests).
func OpenPostgresWithoutSessionCleanup(databaseURL, schema string) (*Store, error) {
return openPostgres(databaseURL, schema, 0)
}
func openPostgres(databaseURL, schema string, sessionCleanup time.Duration) (*Store, error) {
dsn, err := postgresDSN(databaseURL)
if err != nil {
return nil, err
@@ -105,26 +93,15 @@ func openPostgres(databaseURL, schema string, sessionCleanup time.Duration) (*St
_ = db.Close()
return nil, fmt.Errorf("apply schema: %w", err)
}
if err := applySessionsSchema(db, dialectPostgres); err != nil {
if err := applySessionsSchema(db); err != nil {
_ = db.Close()
return nil, fmt.Errorf("apply sessions schema: %w", err)
}
if err := migrateUserProfileColumns(db, dialectPostgres); err != nil {
if err := migrateUserProfileColumns(db); err != nil {
_ = db.Close()
return nil, fmt.Errorf("migrate profile columns: %w", err)
}
st := &Store{db: db, dialect: dialectPostgres}
st.initSessionStore(sessionCleanup)
st := &Store{db: db}
st.initSessionStore(5 * time.Minute)
return st, nil
}
// Connect uses PlanetScale Postgres when DATABASE_URL is set, otherwise SQLite.
func Connect(databaseURL, sqlitePath, schema string) (*Store, error) {
if strings.TrimSpace(databaseURL) != "" {
return OpenPostgres(databaseURL, schema)
}
if sqlitePath == "" {
sqlitePath = "data.db"
}
return Open(sqlitePath, schema)
}
internal/store/sessions.go
+6 -122
View File
@@ -2,22 +2,12 @@ package store
import (
"database/sql"
"log"
"time"
"github.com/alexedwards/scs/postgresstore"
"github.com/alexedwards/scs/v2"
)
const sessionsSchemaSQLite = `
CREATE TABLE IF NOT EXISTS sessions (
token TEXT PRIMARY KEY,
data BLOB NOT NULL,
expiry REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS sessions_expiry_idx ON sessions(expiry);
`
const sessionsSchemaPostgres = `
CREATE TABLE IF NOT EXISTS sessions (
token TEXT PRIMARY KEY,
@@ -27,12 +17,9 @@ CREATE TABLE IF NOT EXISTS sessions (
CREATE INDEX IF NOT EXISTS sessions_expiry_idx ON sessions (expiry);
`
func applySessionsSchema(db *sql.DB, dialect string) error {
schema := sessionsSchemaSQLite
if dialect == dialectPostgres {
schema = sessionsSchemaPostgres
}
return applySchema(db, schema)
// applySessionsSchema creates the scs sessions table if missing.
func applySessionsSchema(db *sql.DB) error {
return applySchema(db, sessionsSchemaPostgres)
}
type sessionStopper interface {
@@ -45,110 +32,7 @@ func (s *Store) SessionStore() scs.Store {
}
func (s *Store) initSessionStore(cleanupInterval time.Duration) {
switch s.dialect {
case dialectPostgres:
ps := postgresstore.NewWithCleanupInterval(s.db, cleanupInterval)
s.sessionStore = ps
s.sessionStopper = ps
default:
ss := newSQLiteSessionStore(s.db, cleanupInterval)
s.sessionStore = ss
s.sessionStopper = ss
}
ps := postgresstore.NewWithCleanupInterval(s.db, cleanupInterval)
s.sessionStore = ps
s.sessionStopper = ps
}
// sqliteSessionStore is a modernc-safe scs.Store (uses ? placeholders).
type sqliteSessionStore struct {
db *sql.DB
stopCleanup chan bool
}
func newSQLiteSessionStore(db *sql.DB, cleanupInterval time.Duration) *sqliteSessionStore {
s := &sqliteSessionStore{db: db}
if cleanupInterval > 0 {
s.stopCleanup = make(chan bool)
go s.startCleanup(cleanupInterval)
}
return s
}
func (s *sqliteSessionStore) Find(token string) ([]byte, bool, error) {
var b []byte
err := s.db.QueryRow(
`SELECT data FROM sessions WHERE token = ? AND julianday('now') < expiry`,
token,
).Scan(&b)
if err == sql.ErrNoRows {
return nil, false, nil
}
if err != nil {
return nil, false, err
}
return b, true, nil
}
func (s *sqliteSessionStore) Commit(token string, b []byte, expiry time.Time) error {
_, err := s.db.Exec(
`REPLACE INTO sessions (token, data, expiry) VALUES (?, ?, julianday(?))`,
token,
b,
expiry.UTC().Format("2006-01-02T15:04:05.999"),
)
return err
}
func (s *sqliteSessionStore) Delete(token string) error {
_, err := s.db.Exec(`DELETE FROM sessions WHERE token = ?`, token)
return err
}
func (s *sqliteSessionStore) All() (map[string][]byte, error) {
rows, err := s.db.Query(`SELECT token, data FROM sessions WHERE julianday('now') < expiry`)
if err != nil {
return nil, err
}
defer rows.Close()
out := make(map[string][]byte)
for rows.Next() {
var token string
var data []byte
if err := rows.Scan(&token, &data); err != nil {
return nil, err
}
out[token] = data
}
return out, rows.Err()
}
func (s *sqliteSessionStore) startCleanup(interval time.Duration) {
ticker := time.NewTicker(interval)
for {
select {
case <-ticker.C:
if err := s.deleteExpired(); err != nil {
log.Println(err)
}
case <-s.stopCleanup:
ticker.Stop()
return
}
}
}
func (s *sqliteSessionStore) StopCleanup() {
if s.stopCleanup != nil {
s.stopCleanup <- true
}
}
func (s *sqliteSessionStore) deleteExpired() error {
_, err := s.db.Exec(`DELETE FROM sessions WHERE expiry < julianday('now')`)
return err
}
// Ensure interface compliance.
var (
_ scs.Store = (*sqliteSessionStore)(nil)
_ scs.IterableStore = (*sqliteSessionStore)(nil)
_ sessionStopper = (*sqliteSessionStore)(nil)
)
internal/store/store.go
+22 -57
View File
@@ -9,14 +9,12 @@ import (
"github.com/alexedwards/scs/v2"
"github.com/google/uuid"
_ "modernc.org/sqlite"
"plumber/internal/pacific"
)
type Store struct {
db *sql.DB
dialect string
sessionStore scs.Store
sessionStopper sessionStopper
}
@@ -25,7 +23,7 @@ type User struct {
ID string
Username string
Name string
Role string
Role Role
AvatarURL string
State string
CreatedAt string
@@ -33,7 +31,7 @@ type User struct {
}
func (u *User) Admin() bool {
return u != nil && u.Role == "admin"
return u != nil && u.Role == RoleAdmin
}
type RankedQuestion struct {
@@ -60,42 +58,6 @@ type Answer struct {
UpdatedAt string
}
func Open(path, schema string) (*Store, error) {
return openSQLite(path, schema, 5*time.Minute)
}
// OpenWithoutSessionCleanup opens SQLite without a session cleanup goroutine (for tests).
func OpenWithoutSessionCleanup(path, schema string) (*Store, error) {
return openSQLite(path, schema, 0)
}
func openSQLite(path, schema string, sessionCleanup time.Duration) (*Store, error) {
dsn := path
if !strings.Contains(dsn, "?") {
dsn += "?_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)"
}
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(1)
if _, err := db.Exec(schema); err != nil {
_ = db.Close()
return nil, fmt.Errorf("apply schema: %w", err)
}
if err := applySessionsSchema(db, dialectSQLite); err != nil {
_ = db.Close()
return nil, fmt.Errorf("apply sessions schema: %w", err)
}
if err := migrateUserProfileColumns(db, dialectSQLite); err != nil {
_ = db.Close()
return nil, fmt.Errorf("migrate profile columns: %w", err)
}
st := &Store{db: db, dialect: dialectSQLite}
st.initSessionStore(sessionCleanup)
return st, nil
}
func (s *Store) Close() error {
if s.sessionStopper != nil {
s.sessionStopper.StopCleanup()
@@ -104,22 +66,21 @@ func (s *Store) Close() error {
return s.db.Close()
}
func (s *Store) CreateUser(ctx context.Context, username, passwordHash string, asAdmin bool) (*User, error) {
username = NormalizeUsername(username)
role := "user"
if asAdmin {
role = "admin"
func (s *Store) CreateUser(ctx context.Context, nu NewUser) (*User, error) {
if nu.Role != RoleUser && nu.Role != RoleAdmin {
return nil, fmt.Errorf("invalid role")
}
username := NormalizeUsername(nu.Username)
u := &User{
ID: uuid.NewString(),
Username: username,
Name: username,
Role: role,
PasswordHash: passwordHash,
Role: nu.Role,
PasswordHash: nu.PasswordHash,
CreatedAt: time.Now().UTC().Format(time.RFC3339),
}
_, err := s.db.ExecContext(ctx, s.q(`INSERT INTO users (id, username, name, password_hash, role, avatar_url, state, created_at) VALUES (?, ?, ?, ?, ?, '', '', ?)`),
u.ID, u.Username, u.Name, u.PasswordHash, u.Role, u.CreatedAt)
u.ID, u.Username, u.Name, u.PasswordHash, string(u.Role), u.CreatedAt)
if err != nil {
return nil, err
}
@@ -128,7 +89,7 @@ func (s *Store) CreateUser(ctx context.Context, username, passwordHash string, a
func (s *Store) CountAdmins(ctx context.Context) (int, error) {
var n int
err := s.db.QueryRowContext(ctx, s.q(`SELECT COUNT(*) FROM users WHERE role = ?`), "admin").Scan(&n)
err := s.db.QueryRowContext(ctx, s.q(`SELECT COUNT(*) FROM users WHERE role = ?`), string(RoleAdmin)).Scan(&n)
return n, err
}
@@ -141,16 +102,18 @@ func (s *Store) ListUsers(ctx context.Context) ([]User, error) {
var out []User
for rows.Next() {
var u User
if err := rows.Scan(&u.ID, &u.Username, &u.Name, &u.Role, &u.AvatarURL, &u.State, &u.CreatedAt); err != nil {
var role string
if err := rows.Scan(&u.ID, &u.Username, &u.Name, &role, &u.AvatarURL, &u.State, &u.CreatedAt); err != nil {
return nil, err
}
u.Role = Role(role)
out = append(out, u)
}
return out, rows.Err()
}
func (s *Store) SetRole(ctx context.Context, userID, role string) error {
if role != "user" && role != "admin" {
func (s *Store) SetRole(ctx context.Context, userID string, role Role) error {
if role != RoleUser && role != RoleAdmin {
return fmt.Errorf("invalid role")
}
tx, err := s.db.BeginTx(ctx, nil)
@@ -164,16 +127,16 @@ func (s *Store) SetRole(ctx context.Context, userID, role string) error {
if err != nil {
return err
}
if current == "admin" && role == "user" {
if Role(current) == RoleAdmin && role == RoleUser {
var n int
if err := tx.QueryRowContext(ctx, s.q(`SELECT COUNT(*) FROM users WHERE role = ?`), "admin").Scan(&n); err != nil {
if err := tx.QueryRowContext(ctx, s.q(`SELECT COUNT(*) FROM users WHERE role = ?`), string(RoleAdmin)).Scan(&n); err != nil {
return err
}
if n <= 1 {
return ErrLastAdmin
}
}
res, err := tx.ExecContext(ctx, s.q(`UPDATE users SET role = ? WHERE id = ?`), role, userID)
res, err := tx.ExecContext(ctx, s.q(`UPDATE users SET role = ? WHERE id = ?`), string(role), userID)
if err != nil {
return err
}
@@ -197,15 +160,17 @@ func (s *Store) UserByUsername(ctx context.Context, username string) (*User, err
func scanUser(row *sql.Row, withSecrets bool) (*User, error) {
var u User
var role string
var err error
if withSecrets {
err = row.Scan(&u.ID, &u.Username, &u.Name, &u.Role, &u.AvatarURL, &u.State, &u.CreatedAt, &u.PasswordHash)
err = row.Scan(&u.ID, &u.Username, &u.Name, &role, &u.AvatarURL, &u.State, &u.CreatedAt, &u.PasswordHash)
} else {
err = row.Scan(&u.ID, &u.Username, &u.Name, &u.Role, &u.AvatarURL, &u.State, &u.CreatedAt)
err = row.Scan(&u.ID, &u.Username, &u.Name, &role, &u.AvatarURL, &u.State, &u.CreatedAt)
}
if err != nil {
return nil, err
}
u.Role = Role(role)
return &u, nil
}
internal/web/admin.go
+1 -1
View File
@@ -47,7 +47,7 @@ func (s *Server) handleAdminSetRole(w http.ResponseWriter, r *http.Request) {
return
}
id := chi.URLParam(r, "id")
role := r.PostFormValue("role")
role := store.Role(r.PostFormValue("role"))
err := s.store.SetRole(r.Context(), id, role)
if errors.Is(err, store.ErrLastAdmin) {
users, listErr := s.store.ListUsers(r.Context())
internal/web/auth.go
+9 -3
View File
@@ -88,16 +88,22 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
http.Error(w, "could not save password", http.StatusInternalServerError)
return
}
asAdmin := false
role := store.RoleUser
if s.cfg.AdminUsername != "" && store.NormalizeUsername(username) == store.NormalizeUsername(s.cfg.AdminUsername) {
n, err := s.store.CountAdmins(r.Context())
if err != nil {
http.Error(w, "could not create account", http.StatusInternalServerError)
return
}
asAdmin = n == 0
if n == 0 {
role = store.RoleAdmin
}
}
u, err := s.store.CreateUser(r.Context(), username, string(hash), asAdmin)
u, err := s.store.CreateUser(r.Context(), store.NewUser{
Username: username,
PasswordHash: string(hash),
Role: role,
})
if err != nil {
p.Error = "That username is taken."
s.exec(w, "register", p)
internal/web/memstore_test.go
+16 -17
View File
@@ -18,11 +18,11 @@ import (
// memDB is an in-memory store.DB for tests.
type memDB struct {
mu sync.Mutex
users map[string]*store.User // id -> user
byName map[string]string // username -> id
questions map[string]*store.RankedQuestion // id -> question
votes map[string]int // userID|questionID -> value
answers map[string]*store.Answer // questionID -> answer
users map[string]*store.User // id -> user
byName map[string]string // username -> id
questions map[string]*store.RankedQuestion // id -> question
votes map[string]int // userID|questionID -> value
answers map[string]*store.Answer // questionID -> answer
}
func newMemDB() *memDB {
@@ -39,23 +39,22 @@ func voteKey(userID, questionID string) string {
return userID + "|" + questionID
}
func (m *memDB) CreateUser(_ context.Context, username, passwordHash string, asAdmin bool) (*store.User, error) {
func (m *memDB) CreateUser(_ context.Context, nu store.NewUser) (*store.User, error) {
m.mu.Lock()
defer m.mu.Unlock()
username = store.NormalizeUsername(username)
username := store.NormalizeUsername(nu.Username)
if _, ok := m.byName[username]; ok {
return nil, fmt.Errorf("username taken")
}
role := "user"
if asAdmin {
role = "admin"
if nu.Role != store.RoleUser && nu.Role != store.RoleAdmin {
return nil, fmt.Errorf("invalid role")
}
u := &store.User{
ID: uuid.NewString(),
Username: username,
Name: username,
Role: role,
PasswordHash: passwordHash,
Role: nu.Role,
PasswordHash: nu.PasswordHash,
CreatedAt: time.Now().UTC().Format(time.RFC3339),
}
m.users[u.ID] = u
@@ -92,7 +91,7 @@ func (m *memDB) CountAdmins(_ context.Context) (int, error) {
defer m.mu.Unlock()
n := 0
for _, u := range m.users {
if u.Role == "admin" {
if u.Role == store.RoleAdmin {
n++
}
}
@@ -114,8 +113,8 @@ func (m *memDB) ListUsers(_ context.Context) ([]store.User, error) {
return out, nil
}
func (m *memDB) SetRole(_ context.Context, userID, role string) error {
if role != "user" && role != "admin" {
func (m *memDB) SetRole(_ context.Context, userID string, role store.Role) error {
if role != store.RoleUser && role != store.RoleAdmin {
return fmt.Errorf("invalid role")
}
m.mu.Lock()
@@ -124,10 +123,10 @@ func (m *memDB) SetRole(_ context.Context, userID, role string) error {
if !ok {
return sql.ErrNoRows
}
if u.Role == "admin" && role == "user" {
if u.Role == store.RoleAdmin && role == store.RoleUser {
n := 0
for _, x := range m.users {
if x.Role == "admin" {
if x.Role == store.RoleAdmin {
n++
}
}