migrating repo from old host server

2026-08-22 18:58:55 -04:00
commit 4968d7f575
153 changed files with 20009 additions and 0 deletions
--- a/backend/internal/models/address.go
+++ b/backend/internal/models/address.go
@@ -0,0 +1,148 @@
+package models
+
+import (
+	"context"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+)
+
+type AddressModel struct {
+	pool *pgxpool.Pool
+}
+
+func NewAddressModel(pool *pgxpool.Pool) *AddressModel {
+	return &AddressModel{pool: pool}
+}
+
+func (m *AddressModel) Create(ctx context.Context, userID, address string, label *string) (*domain.Address, error) {
+	var a domain.Address
+	err := m.pool.QueryRow(ctx,
+		`INSERT INTO addresses (user_id, address, label)
+		 VALUES ($1, $2, $3)
+		 RETURNING id, user_id, address, label, created_at`,
+		userID, address, label,
+	).Scan(&a.ID, &a.UserID, &a.Address, &a.Label, &a.CreatedAt)
+	if err != nil {
+		return nil, err
+	}
+	return &a, nil
+}
+
+func (m *AddressModel) ListByUser(ctx context.Context, userID string) ([]domain.Address, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT id, user_id, address, label, created_at
+		 FROM addresses
+		 WHERE user_id = $1
+		 ORDER BY created_at DESC`,
+		userID,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var addresses []domain.Address
+	for rows.Next() {
+		var a domain.Address
+		if err := rows.Scan(&a.ID, &a.UserID, &a.Address, &a.Label, &a.CreatedAt); err != nil {
+			return nil, err
+		}
+		addresses = append(addresses, a)
+	}
+	return addresses, rows.Err()
+}
+
+// ListAll returns addresses for users with an active subscription (used by the poller).
+func (m *AddressModel) ListAll(ctx context.Context) ([]domain.Address, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT a.id, a.user_id, a.address, a.label, a.created_at
+		 FROM addresses a
+		 JOIN users u ON u.id = a.user_id
+		 WHERE u.subscription_status IN ('active', 'trialing')
+		 ORDER BY a.created_at DESC`,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var addresses []domain.Address
+	for rows.Next() {
+		var a domain.Address
+		if err := rows.Scan(&a.ID, &a.UserID, &a.Address, &a.Label, &a.CreatedAt); err != nil {
+			return nil, err
+		}
+		addresses = append(addresses, a)
+	}
+	return addresses, rows.Err()
+}
+
+func (m *AddressModel) FindByID(ctx context.Context, id int, userID *string) (*domain.Address, error) {
+	var a domain.Address
+	var err error
+
+	if userID != nil {
+		err = m.pool.QueryRow(ctx,
+			`SELECT id, user_id, address, label, created_at
+			 FROM addresses
+			 WHERE id = $1 AND user_id = $2`,
+			id, *userID,
+		).Scan(&a.ID, &a.UserID, &a.Address, &a.Label, &a.CreatedAt)
+	} else {
+		err = m.pool.QueryRow(ctx,
+			`SELECT id, user_id, address, label, created_at
+			 FROM addresses
+			 WHERE id = $1`,
+			id,
+		).Scan(&a.ID, &a.UserID, &a.Address, &a.Label, &a.CreatedAt)
+	}
+
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &a, nil
+}
+
+// UpdateLabel updates the label for an address owned by userID.
+// Returns nil, nil if no row matched (address not found or not owned by user).
+func (m *AddressModel) UpdateLabel(ctx context.Context, id int, userID string, label *string) (*domain.Address, error) {
+	var a domain.Address
+	err := m.pool.QueryRow(ctx,
+		`UPDATE addresses SET label = $3 WHERE id = $1 AND user_id = $2
+		 RETURNING id, user_id, address, label, created_at`,
+		id, userID, label,
+	).Scan(&a.ID, &a.UserID, &a.Address, &a.Label, &a.CreatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &a, nil
+}
+
+func (m *AddressModel) CountByUser(ctx context.Context, userID string) (int, error) {
+	var count int
+	err := m.pool.QueryRow(ctx,
+		`SELECT COUNT(*) FROM addresses WHERE user_id = $1`,
+		userID,
+	).Scan(&count)
+	return count, err
+}
+
+func (m *AddressModel) Remove(ctx context.Context, id int, userID string) (bool, error) {
+	tag, err := m.pool.Exec(ctx,
+		`DELETE FROM addresses WHERE id = $1 AND user_id = $2`,
+		id, userID,
+	)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() > 0, nil
+}
--- a/backend/internal/models/alert_event.go
+++ b/backend/internal/models/alert_event.go
@@ -0,0 +1,88 @@
+package models
+
+import (
+	"context"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+)
+
+type AlertEventModel struct {
+	pool *pgxpool.Pool
+}
+
+func NewAlertEventModel(pool *pgxpool.Pool) *AlertEventModel {
+	return &AlertEventModel{pool: pool}
+}
+
+func (m *AlertEventModel) ListRecentByUser(ctx context.Context, userID string, limit int) ([]domain.AlertEvent, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT ae.id, ae.alert_rule_id, ae.message, ae.address_label, ae.tx_hash, ae.timestamp
+		 FROM alert_events ae
+		 JOIN alert_rules ar ON ar.id = ae.alert_rule_id
+		 JOIN addresses a ON a.id = ar.address_id
+		 WHERE a.user_id = $1
+		 ORDER BY ae.timestamp DESC
+		 LIMIT $2`,
+		userID, limit,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var events []domain.AlertEvent
+	for rows.Next() {
+		var e domain.AlertEvent
+		if err := rows.Scan(&e.ID, &e.AlertRuleID, &e.Message, &e.AddressLabel, &e.TxHash, &e.Timestamp); err != nil {
+			return nil, err
+		}
+		events = append(events, e)
+	}
+	return events, rows.Err()
+}
+
+func (m *AlertEventModel) ListRecent(ctx context.Context, limit int) ([]domain.AlertEvent, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT id, alert_rule_id, message, address_label, tx_hash, timestamp
+		 FROM alert_events
+		 ORDER BY timestamp DESC
+		 LIMIT $1`,
+		limit,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var events []domain.AlertEvent
+	for rows.Next() {
+		var e domain.AlertEvent
+		if err := rows.Scan(&e.ID, &e.AlertRuleID, &e.Message, &e.AddressLabel, &e.TxHash, &e.Timestamp); err != nil {
+			return nil, err
+		}
+		events = append(events, e)
+	}
+	return events, rows.Err()
+}
+
+func (m *AlertEventModel) Create(ctx context.Context, alertRuleID int, message string, addressLabel *string, txHash *string) (*domain.AlertEvent, error) {
+	var e domain.AlertEvent
+	err := m.pool.QueryRow(ctx,
+		`INSERT INTO alert_events (alert_rule_id, message, address_label, tx_hash)
+		 VALUES ($1, $2, $3, $4)
+		 ON CONFLICT DO NOTHING
+		 RETURNING id, alert_rule_id, message, address_label, tx_hash, timestamp`,
+		alertRuleID, message, addressLabel, txHash,
+	).Scan(&e.ID, &e.AlertRuleID, &e.Message, &e.AddressLabel, &e.TxHash, &e.Timestamp)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			// Duplicate silently skipped by ON CONFLICT DO NOTHING
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &e, nil
+}
--- a/backend/internal/models/alert_rule.go
+++ b/backend/internal/models/alert_rule.go
@@ -0,0 +1,172 @@
+package models
+
+import (
+	"context"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+)
+
+type AlertRuleModel struct {
+	pool *pgxpool.Pool
+}
+
+func NewAlertRuleModel(pool *pgxpool.Pool) *AlertRuleModel {
+	return &AlertRuleModel{pool: pool}
+}
+
+func (m *AlertRuleModel) Create(ctx context.Context, addressID int, alertType domain.AlertType, threshold, minimum, maximum *float64) (*domain.AlertRule, error) {
+	var r domain.AlertRule
+	err := m.pool.QueryRow(ctx,
+		`INSERT INTO alert_rules (address_id, type, threshold, minimum, maximum, enabled)
+		 VALUES ($1, $2, $3, $4, $5, TRUE)
+		 RETURNING id, address_id, type, threshold, minimum, maximum, enabled, created_at`,
+		addressID, alertType.String(), threshold, minimum, maximum,
+	).Scan(&r.ID, &r.AddressID, &r.Type, &r.Threshold, &r.Minimum, &r.Maximum, &r.Enabled, &r.CreatedAt)
+	if err != nil {
+		return nil, err
+	}
+	return &r, nil
+}
+
+func (m *AlertRuleModel) ListByAddress(ctx context.Context, addressID int) ([]domain.AlertRule, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT id, address_id, type, threshold, minimum, maximum, enabled, created_at
+		 FROM alert_rules
+		 WHERE address_id = $1
+		 ORDER BY created_at DESC`,
+		addressID,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var rules []domain.AlertRule
+	for rows.Next() {
+		var r domain.AlertRule
+		if err := rows.Scan(&r.ID, &r.AddressID, &r.Type, &r.Threshold, &r.Minimum, &r.Maximum, &r.Enabled, &r.CreatedAt); err != nil {
+			return nil, err
+		}
+		rules = append(rules, r)
+	}
+	return rules, rows.Err()
+}
+
+func (m *AlertRuleModel) FindByID(ctx context.Context, id int, userID *string) (*domain.AlertRule, error) {
+	var r domain.AlertRule
+	var err error
+
+	if userID != nil {
+		err = m.pool.QueryRow(ctx,
+			`SELECT ar.id, ar.address_id, ar.type, ar.threshold, ar.minimum, ar.maximum, ar.enabled, ar.created_at
+			 FROM alert_rules ar
+			 JOIN addresses a ON a.id = ar.address_id
+			 WHERE ar.id = $1 AND a.user_id = $2`,
+			id, *userID,
+		).Scan(&r.ID, &r.AddressID, &r.Type, &r.Threshold, &r.Minimum, &r.Maximum, &r.Enabled, &r.CreatedAt)
+	} else {
+		err = m.pool.QueryRow(ctx,
+			`SELECT id, address_id, type, threshold, minimum, maximum, enabled, created_at
+			 FROM alert_rules
+			 WHERE id = $1`,
+			id,
+		).Scan(&r.ID, &r.AddressID, &r.Type, &r.Threshold, &r.Minimum, &r.Maximum, &r.Enabled, &r.CreatedAt)
+	}
+
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &r, nil
+}
+
+func (m *AlertRuleModel) UpdateEnabled(ctx context.Context, id int, enabled bool) (*domain.AlertRule, error) {
+	var r domain.AlertRule
+	err := m.pool.QueryRow(ctx,
+		`UPDATE alert_rules
+		 SET enabled = $2
+		 WHERE id = $1
+		 RETURNING id, address_id, type, threshold, minimum, maximum, enabled, created_at`,
+		id, enabled,
+	).Scan(&r.ID, &r.AddressID, &r.Type, &r.Threshold, &r.Minimum, &r.Maximum, &r.Enabled, &r.CreatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &r, nil
+}
+
+func (m *AlertRuleModel) UpdateThresholds(ctx context.Context, id int, minimum, maximum *float64) (*domain.AlertRule, error) {
+	var r domain.AlertRule
+	err := m.pool.QueryRow(ctx,
+		`UPDATE alert_rules
+		 SET minimum = $2, maximum = $3
+		 WHERE id = $1
+		 RETURNING id, address_id, type, threshold, minimum, maximum, enabled, created_at`,
+		id, minimum, maximum,
+	).Scan(&r.ID, &r.AddressID, &r.Type, &r.Threshold, &r.Minimum, &r.Maximum, &r.Enabled, &r.CreatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &r, nil
+}
+
+func (m *AlertRuleModel) CountDistinctTypesByAddress(ctx context.Context, addressID int) (int, error) {
+	var count int
+	err := m.pool.QueryRow(ctx,
+		`SELECT COUNT(DISTINCT type) FROM alert_rules WHERE address_id = $1`,
+		addressID,
+	).Scan(&count)
+	return count, err
+}
+
+func (m *AlertRuleModel) Remove(ctx context.Context, id int) (bool, error) {
+	tag, err := m.pool.Exec(ctx,
+		`DELETE FROM alert_rules WHERE id = $1`,
+		id,
+	)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() > 0, nil
+}
+
+// DisableAllForUser sets enabled = false on every alert rule owned by addresses of userID.
+func (m *AlertRuleModel) DisableAllForUser(ctx context.Context, userID string) (int64, error) {
+	tag, err := m.pool.Exec(ctx,
+		`UPDATE alert_rules ar
+		 SET enabled = FALSE
+		 FROM addresses a
+		 WHERE ar.address_id = a.id AND a.user_id = $1`,
+		userID,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
+
+// EnableAllForUser sets enabled = true on every alert rule owned by addresses of userID.
+func (m *AlertRuleModel) EnableAllForUser(ctx context.Context, userID string) (int64, error) {
+	tag, err := m.pool.Exec(ctx,
+		`UPDATE alert_rules ar
+		 SET enabled = TRUE
+		 FROM addresses a
+		 WHERE ar.address_id = a.id AND a.user_id = $1`,
+		userID,
+	)
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
--- a/backend/internal/models/checkpoint.go
+++ b/backend/internal/models/checkpoint.go
@@ -0,0 +1,101 @@
+package models
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+)
+
+type CheckpointModel struct {
+	pool *pgxpool.Pool
+}
+
+func NewCheckpointModel(pool *pgxpool.Pool) *CheckpointModel {
+	return &CheckpointModel{pool: pool}
+}
+
+// GetLatestBlock returns the highest last_checked_block and its timestamp across all addresses.
+// Returns nil, nil, nil when no checkpoints exist yet.
+func (m *CheckpointModel) GetLatestBlock(ctx context.Context) (*int, *time.Time, error) {
+	var block *int
+	var checkedAt *time.Time
+	err := m.pool.QueryRow(ctx,
+		`SELECT MAX(last_checked_block), MAX(last_checked_at) FROM address_checkpoints`,
+	).Scan(&block, &checkedAt)
+	if err != nil {
+		return nil, nil, err
+	}
+	return block, checkedAt, nil
+}
+
+// GetLastCheckedBlock returns the last checked block for an address, or -1 if never checked.
+func (m *CheckpointModel) GetLastCheckedBlock(ctx context.Context, addressID int) (int, bool, error) {
+	var block int
+	err := m.pool.QueryRow(ctx,
+		`SELECT last_checked_block FROM address_checkpoints WHERE address_id = $1`,
+		addressID,
+	).Scan(&block)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, false, nil
+		}
+		return 0, false, err
+	}
+	return block, true, nil
+}
+
+func (m *CheckpointModel) UpdateLastCheckedBlock(ctx context.Context, addressID, blockNumber int) (*domain.AddressCheckpoint, error) {
+	var cp domain.AddressCheckpoint
+	err := m.pool.QueryRow(ctx,
+		`INSERT INTO address_checkpoints (address_id, last_checked_block, last_checked_at)
+		 VALUES ($1, $2, NOW())
+		 ON CONFLICT (address_id)
+		 DO UPDATE SET
+		   last_checked_block = $2,
+		   last_checked_at = NOW()
+		 RETURNING address_id, last_checked_block, last_checked_at`,
+		addressID, blockNumber,
+	).Scan(&cp.AddressID, &cp.LastCheckedBlock, &cp.LastCheckedAt)
+	if err != nil {
+		return nil, err
+	}
+	return &cp, nil
+}
+
+func (m *CheckpointModel) ListAll(ctx context.Context) ([]domain.CheckpointDetail, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT ac.address_id, a.address, a.label, ac.last_checked_block, ac.last_checked_at
+		 FROM address_checkpoints ac
+		 JOIN addresses a ON a.id = ac.address_id
+		 ORDER BY ac.last_checked_at DESC`,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var details []domain.CheckpointDetail
+	for rows.Next() {
+		var d domain.CheckpointDetail
+		if err := rows.Scan(&d.AddressID, &d.Address, &d.Label, &d.LastCheckedBlock, &d.LastCheckedAt); err != nil {
+			return nil, err
+		}
+		details = append(details, d)
+	}
+	return details, rows.Err()
+}
+
+func (m *CheckpointModel) Remove(ctx context.Context, addressID int) (bool, error) {
+	tag, err := m.pool.Exec(ctx,
+		`DELETE FROM address_checkpoints WHERE address_id = $1`,
+		addressID,
+	)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() > 0, nil
+}
--- a/backend/internal/models/notification_config.go
+++ b/backend/internal/models/notification_config.go
@@ -0,0 +1,103 @@
+package models
+
+import (
+	"context"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+)
+
+type NotificationConfigModel struct {
+	pool *pgxpool.Pool
+}
+
+func NewNotificationConfigModel(pool *pgxpool.Pool) *NotificationConfigModel {
+	return &NotificationConfigModel{pool: pool}
+}
+
+func (m *NotificationConfigModel) GetConfig(ctx context.Context, userID string) (*domain.NotificationConfig, error) {
+	var c domain.NotificationConfig
+	err := m.pool.QueryRow(ctx,
+		`SELECT user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token,
+		        email, slack_webhook_url, notification_enabled, created_at, updated_at
+		 FROM user_notification_configs
+		 WHERE user_id = $1`,
+		userID,
+	).Scan(&c.UserID, &c.DiscordWebhookURL, &c.TelegramChatID, &c.TelegramBotToken,
+		&c.Email, &c.SlackWebhookURL, &c.NotificationEnabled, &c.CreatedAt, &c.UpdatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &c, nil
+}
+
+func (m *NotificationConfigModel) UpsertConfig(ctx context.Context, userID string, cfg domain.NotificationConfig) (*domain.NotificationConfig, error) {
+	var c domain.NotificationConfig
+	err := m.pool.QueryRow(ctx,
+		`INSERT INTO user_notification_configs
+		   (user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token,
+		    email, slack_webhook_url, notification_enabled, updated_at)
+		 VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
+		 ON CONFLICT (user_id)
+		 DO UPDATE SET
+		   discord_webhook_url = $2,
+		   telegram_chat_id = $3,
+		   telegram_bot_token = $4,
+		   email = $5,
+		   slack_webhook_url = $6,
+		   notification_enabled = $7,
+		   updated_at = NOW()
+		 RETURNING user_id, discord_webhook_url, telegram_chat_id, telegram_bot_token,
+		           email, slack_webhook_url, notification_enabled, created_at, updated_at`,
+		userID, cfg.DiscordWebhookURL, cfg.TelegramChatID, cfg.TelegramBotToken,
+		cfg.Email, cfg.SlackWebhookURL, cfg.NotificationEnabled,
+	).Scan(&c.UserID, &c.DiscordWebhookURL, &c.TelegramChatID, &c.TelegramBotToken,
+		&c.Email, &c.SlackWebhookURL, &c.NotificationEnabled, &c.CreatedAt, &c.UpdatedAt)
+	if err != nil {
+		return nil, err
+	}
+	return &c, nil
+}
+
+func (m *NotificationConfigModel) Remove(ctx context.Context, userID string) (bool, error) {
+	tag, err := m.pool.Exec(ctx,
+		`DELETE FROM user_notification_configs WHERE user_id = $1`,
+		userID,
+	)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() > 0, nil
+}
+
+func (m *NotificationConfigModel) ListEnabled(ctx context.Context) ([]domain.NotificationConfig, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT nc.user_id, nc.discord_webhook_url, nc.telegram_chat_id, nc.telegram_bot_token,
+		        nc.email, nc.slack_webhook_url
+		 FROM user_notification_configs nc
+		 JOIN users u ON u.id = nc.user_id
+		 WHERE nc.notification_enabled = TRUE
+		   AND u.subscription_status IN ('active', 'trialing')`,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var configs []domain.NotificationConfig
+	for rows.Next() {
+		var c domain.NotificationConfig
+		if err := rows.Scan(&c.UserID, &c.DiscordWebhookURL, &c.TelegramChatID,
+			&c.TelegramBotToken, &c.Email, &c.SlackWebhookURL); err != nil {
+			return nil, err
+		}
+		c.NotificationEnabled = true
+		configs = append(configs, c)
+	}
+	return configs, rows.Err()
+}
--- a/backend/internal/models/user.go
+++ b/backend/internal/models/user.go
@@ -0,0 +1,223 @@
+package models
+
+import (
+	"context"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/kjannette/koin-ping/backend/internal/domain"
+)
+
+type UserModel struct {
+	pool *pgxpool.Pool
+}
+
+func NewUserModel(pool *pgxpool.Pool) *UserModel {
+	return &UserModel{pool: pool}
+}
+
+const userColumns = `id, firebase_uid, email, display_name,
+	stripe_customer_id, stripe_subscription_id, subscription_status,
+	subscription_tier, subscription_created_at, created_at, updated_at`
+
+func scanUser(row pgx.Row) (*domain.User, error) {
+	var u domain.User
+	err := row.Scan(
+		&u.ID, &u.FirebaseUID, &u.Email, &u.DisplayName,
+		&u.StripeCustomerID, &u.StripeSubscriptionID, &u.SubscriptionStatus,
+		&u.SubscriptionTier, &u.SubscriptionCreatedAt, &u.CreatedAt, &u.UpdatedAt,
+	)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil //nolint:nilnil
+		}
+		return nil, err
+	}
+	return &u, nil
+}
+
+func (m *UserModel) getByFirebaseUID(ctx context.Context, firebaseUID string) (*domain.User, error) {
+	row := m.pool.QueryRow(ctx,
+		`SELECT `+userColumns+` FROM users WHERE firebase_uid = $1`,
+		firebaseUID,
+	)
+	return scanUser(row)
+}
+
+func (m *UserModel) getByEmail(ctx context.Context, email string) (*domain.User, error) {
+	row := m.pool.QueryRow(ctx,
+		`SELECT `+userColumns+` FROM users WHERE email = $1`,
+		email,
+	)
+	return scanUser(row)
+}
+
+func isUniqueViolation(err error) bool {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		return pgErr.Code == "23505"
+	}
+	return false
+}
+
+// FindOrCreateByFirebaseUID returns the local user for a Firebase UID,
+// creating one if it doesn't exist yet.
+//
+// For onboarding races or legacy duplicate-identity states, this method
+// gracefully falls back to an existing row by email instead of failing
+// with a unique-email violation.
+func (m *UserModel) FindOrCreateByFirebaseUID(ctx context.Context, firebaseUID, email string) (*domain.User, error) {
+	existingByUID, err := m.getByFirebaseUID(ctx, firebaseUID)
+	if err != nil {
+		return nil, err
+	}
+	if existingByUID != nil {
+		return existingByUID, nil
+	}
+
+	if email != "" {
+		existingByEmail, err := m.getByEmail(ctx, email)
+		if err != nil {
+			return nil, err
+		}
+		if existingByEmail != nil {
+			return existingByEmail, nil
+		}
+	}
+
+	row := m.pool.QueryRow(ctx,
+		`INSERT INTO users (firebase_uid, email)
+		 VALUES ($1, $2)
+		 ON CONFLICT (firebase_uid) DO UPDATE SET updated_at = NOW()
+		 RETURNING `+userColumns,
+		firebaseUID, email,
+	)
+	user, err := scanUser(row)
+	if err == nil {
+		return user, nil
+	}
+
+	// If a concurrent request inserted by email or firebase_uid first,
+	// read the existing record and continue without surfacing a 500.
+	if isUniqueViolation(err) {
+		existingByUID, readErr := m.getByFirebaseUID(ctx, firebaseUID)
+		if readErr != nil {
+			return nil, readErr
+		}
+		if existingByUID != nil {
+			return existingByUID, nil
+		}
+		if email != "" {
+			existingByEmail, readErr := m.getByEmail(ctx, email)
+			if readErr != nil {
+				return nil, readErr
+			}
+			if existingByEmail != nil {
+				return existingByEmail, nil
+			}
+		}
+	}
+
+	return nil, err
+}
+
+func (m *UserModel) GetByID(ctx context.Context, id string) (*domain.User, error) {
+	row := m.pool.QueryRow(ctx,
+		`SELECT `+userColumns+` FROM users WHERE id = $1`, id,
+	)
+	return scanUser(row)
+}
+
+// GetByStripeCustomerID loads a user by their Stripe Customer ID if set.
+func (m *UserModel) GetByStripeCustomerID(ctx context.Context, stripeCustomerID string) (*domain.User, error) {
+	row := m.pool.QueryRow(ctx,
+		`SELECT `+userColumns+` FROM users WHERE stripe_customer_id = $1`,
+		stripeCustomerID,
+	)
+	return scanUser(row)
+}
+
+// ListPaidUsersWithoutActiveSubscription finds paid-tier rows whose Stripe
+// subscription is not active or trialing. Used by the periodic billing sweep.
+func (m *UserModel) ListPaidUsersWithoutActiveSubscription(ctx context.Context) ([]domain.User, error) {
+	rows, err := m.pool.Query(ctx,
+		`SELECT `+userColumns+` FROM users
+		 WHERE subscription_tier IN ('premium', 'pro')
+		   AND subscription_status NOT IN ('active', 'trialing')
+		   AND COALESCE(trim(firebase_uid), '') <> ''`,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	out := []domain.User{}
+	for rows.Next() {
+		var u domain.User
+		rowErr := rows.Scan(
+			&u.ID, &u.FirebaseUID, &u.Email, &u.DisplayName,
+			&u.StripeCustomerID, &u.StripeSubscriptionID, &u.SubscriptionStatus,
+			&u.SubscriptionTier, &u.SubscriptionCreatedAt, &u.CreatedAt, &u.UpdatedAt,
+		)
+		if rowErr != nil {
+			return nil, rowErr
+		}
+		out = append(out, u)
+	}
+
+	return out, rows.Err()
+}
+
+func (m *UserModel) UpdateStripeCustomer(ctx context.Context, userID, stripeCustomerID string) error {
+	_, err := m.pool.Exec(ctx,
+		`UPDATE users SET stripe_customer_id = $2, updated_at = NOW() WHERE id = $1`,
+		userID, stripeCustomerID,
+	)
+	return err
+}
+
+func (m *UserModel) ActivateSubscription(ctx context.Context, stripeCustomerID, subscriptionID, status string, tier domain.SubscriptionTier) error {
+	_, err := m.pool.Exec(ctx,
+		`UPDATE users
+		 SET stripe_subscription_id = $2,
+		     subscription_status = $3,
+		     subscription_tier = $4,
+		     subscription_created_at = COALESCE(subscription_created_at, NOW()),
+		     updated_at = NOW()
+		 WHERE stripe_customer_id = $1`,
+		stripeCustomerID, subscriptionID, status, string(tier),
+	)
+	return err
+}
+
+func (m *UserModel) UpdateSubscriptionTier(ctx context.Context, userID string, tier domain.SubscriptionTier) error {
+	_, err := m.pool.Exec(ctx,
+		`UPDATE users SET subscription_tier = $2, updated_at = NOW() WHERE id = $1`,
+		userID, string(tier),
+	)
+	return err
+}
+
+func (m *UserModel) ActivateFreeTier(ctx context.Context, userID string) error {
+	_, err := m.pool.Exec(ctx,
+		`UPDATE users
+		 SET subscription_status = 'active',
+		     subscription_tier = 'free',
+		     subscription_created_at = COALESCE(subscription_created_at, NOW()),
+		     updated_at = NOW()
+		 WHERE id = $1`,
+		userID,
+	)
+	return err
+}
+
+func (m *UserModel) UpdateSubscriptionStatus(ctx context.Context, stripeCustomerID, status string) error {
+	_, err := m.pool.Exec(ctx,
+		`UPDATE users SET subscription_status = $2, updated_at = NOW()
+		 WHERE stripe_customer_id = $1`,
+		stripeCustomerID, status,
+	)
+	return err
+}