t5de-hub/internal/store/store.go

// Package store is the hub's data in Postgres: people, their linked IMVU accounts, sessions, proxy sign-ins and
// the audit log. Plain SQL with pgx.
package store

import (
	"context"
	"crypto/sha256"
	"embed"
	"errors"
	"fmt"
	"io/fs"
	"sort"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
)

//go:embed migrations/*.sql
var migrations embed.FS

// Store is the hub's database.
type Store struct {
	db *pgxpool.Pool
}

// ErrNotFound means there's no such row.
var ErrNotFound = errors.New("not found")

// ErrTaken means a unique value (a username) is in use.
var ErrTaken = errors.New("taken")

// Open connects to Postgres and brings its schema up to date.
func Open(ctx context.Context, url string) (*Store, error) {
	db, err := pgxpool.New(ctx, url)
	if err != nil {
		return nil, err
	}
	s := &Store{db: db}
	if err := s.migrate(ctx); err != nil {
		db.Close()
		return nil, err
	}
	return s, nil
}

func (s *Store) Close() { s.db.Close() }

// migrate applies the migrations not applied yet, in name order, each in a transaction.
func (s *Store) migrate(ctx context.Context) error {
	if _, err := s.db.Exec(ctx, `create table if not exists schema_migrations (name text primary key, applied_at timestamptz not null default now())`); err != nil {
		return err
	}
	names, err := fs.Glob(migrations, "migrations/*.sql")
	if err != nil {
		return err
	}
	sort.Strings(names)
	for _, name := range names {
		var done bool
		if err := s.db.QueryRow(ctx, `select exists (select 1 from schema_migrations where name = $1)`, name).Scan(&done); err != nil {
			return err
		}
		if done {
			continue
		}
		sql, _ := migrations.ReadFile(name)
		err := pgx.BeginFunc(ctx, s.db, func(tx pgx.Tx) error {
			if _, err := tx.Exec(ctx, string(sql)); err != nil {
				return err
			}
			_, err := tx.Exec(ctx, `insert into schema_migrations (name) values ($1)`, name)
			return err
		})
		if err != nil {
			return fmt.Errorf("migration %s: %w", name, err)
		}
	}
	return nil
}

// Hash is how tokens are stored: never the token itself.
func Hash(token string) []byte {
	h := sha256.Sum256([]byte(token))
	return h[:]
}

func isUniqueViolation(err error) bool {
	var pg *pgconn.PgError
	return errors.As(err, &pg) && pg.Code == "23505"
}

// Person is a hub account: a person.
type Person struct {
	ID           int64
	Username     string
	PasswordHash string
	Role         string
	LinkCap      int
	CreatedAt    time.Time
	BannedAt     *time.Time
	BanReason    string
}

func (p *Person) IsModerator() bool { return p.Role == "moderator" || p.Role == "admin" }
func (p *Person) IsAdmin() bool     { return p.Role == "admin" }
func (p *Person) Banned() bool      { return p.BannedAt != nil }

const personColumns = `id, username, password_hash, role, link_cap, created_at, banned_at, ban_reason`

func scanPerson(row pgx.Row) (*Person, error) {
	var p Person
	err := row.Scan(&p.ID, &p.Username, &p.PasswordHash, &p.Role, &p.LinkCap, &p.CreatedAt, &p.BannedAt, &p.BanReason)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNotFound
	}
	return &p, err
}

// CreatePerson makes an account. The first account is an admin.
func (s *Store) CreatePerson(ctx context.Context, username, passwordHash string) (*Person, error) {
	p, err := scanPerson(s.db.QueryRow(ctx, `insert into persons (username, password_hash, role)
		values ($1, $2, case when exists (select 1 from persons) then 'user' else 'admin' end)
		returning `+personColumns, username, passwordHash))
	if isUniqueViolation(err) {
		return nil, ErrTaken
	}
	return p, err
}

func (s *Store) PersonByID(ctx context.Context, id int64) (*Person, error) {
	return scanPerson(s.db.QueryRow(ctx, `select `+personColumns+` from persons where id = $1`, id))
}

func (s *Store) PersonByUsername(ctx context.Context, username string) (*Person, error) {
	return scanPerson(s.db.QueryRow(ctx, `select `+personColumns+` from persons where lower(username) = lower($1)`, username))
}

// People lists accounts, newest first, optionally filtered by a username search.
func (s *Store) People(ctx context.Context, search string, limit int) ([]Person, error) {
	rows, err := s.db.Query(ctx, `select `+personColumns+` from persons
		where $1 = '' or lower(username) like '%' || lower($1) || '%'
		   or id in (select person_id from imvu_links where lower(username) like '%' || lower($1) || '%')
		order by id desc limit $2`, search, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Person
	for rows.Next() {
		p, err := scanPerson(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, *p)
	}
	return out, rows.Err()
}

func (s *Store) SetRole(ctx context.Context, id int64, role string) error {
	_, err := s.db.Exec(ctx, `update persons set role = $2 where id = $1`, id, role)
	return err
}

func (s *Store) SetLinkCap(ctx context.Context, id int64, cap int) error {
	_, err := s.db.Exec(ctx, `update persons set link_cap = $2 where id = $1`, id, cap)
	return err
}

// SetBanned bans (with a reason) or unbans a person; a ban also ends their sessions and sign-ins.
func (s *Store) SetBanned(ctx context.Context, id int64, banned bool, reason string) error {
	return pgx.BeginFunc(ctx, s.db, func(tx pgx.Tx) error {
		if !banned {
			_, err := tx.Exec(ctx, `update persons set banned_at = null, ban_reason = '' where id = $1`, id)
			return err
		}
		if _, err := tx.Exec(ctx, `update persons set banned_at = now(), ban_reason = $2 where id = $1`, id, reason); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx, `delete from sessions where person_id = $1`, id); err != nil {
			return err
		}
		_, err := tx.Exec(ctx, `update refresh_tokens set revoked_at = now() where person_id = $1 and revoked_at is null`, id)
		return err
	})
}

// Sessions

func (s *Store) CreateSession(ctx context.Context, token string, personID int64, csrf string, expires time.Time) error {
	_, err := s.db.Exec(ctx, `insert into sessions (token_hash, person_id, csrf, expires_at) values ($1, $2, $3, $4)`,
		Hash(token), personID, csrf, expires)
	return err
}

// Session is the person and CSRF token of a live session.
func (s *Store) Session(ctx context.Context, token string) (*Person, string, error) {
	var csrf string
	var p Person
	err := s.db.QueryRow(ctx, `select s.csrf, `+prefixed("p", personColumns)+` from sessions s join persons p on p.id = s.person_id
		where s.token_hash = $1 and s.expires_at > now()`, Hash(token)).
		Scan(&csrf, &p.ID, &p.Username, &p.PasswordHash, &p.Role, &p.LinkCap, &p.CreatedAt, &p.BannedAt, &p.BanReason)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, "", ErrNotFound
	}
	return &p, csrf, err
}

func (s *Store) DeleteSession(ctx context.Context, token string) error {
	_, err := s.db.Exec(ctx, `delete from sessions where token_hash = $1`, Hash(token))
	return err
}

func prefixed(alias, columns string) string {
	parts := strings.Split(columns, ", ")
	for i := range parts {
		parts[i] = alias + "." + parts[i]
	}
	return strings.Join(parts, ", ")
}

// Linked IMVU accounts

// Link is a linked IMVU account.
type Link struct {
	IMVUID       int64
	PersonID     int64
	PairwiseID   string
	Username     string
	LinkedAt     time.Time
	AgeVerified  bool
	AccessPass   bool
	CheckedAt    time.Time
	ShowBadge    bool
	ShowPresence bool
}

// Adult reports whether the account may see adult content: age verified by IMVU, or with Access Pass.
func (l *Link) Adult() bool { return l.AgeVerified || l.AccessPass }

const linkColumns = `imvu_id, person_id, pairwise_id, username, linked_at, age_verified, access_pass, checked_at, show_badge, show_presence`

func scanLink(row pgx.Row) (*Link, error) {
	var l Link
	err := row.Scan(&l.IMVUID, &l.PersonID, &l.PairwiseID, &l.Username, &l.LinkedAt, &l.AgeVerified, &l.AccessPass, &l.CheckedAt, &l.ShowBadge, &l.ShowPresence)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNotFound
	}
	return &l, err
}

func (s *Store) Links(ctx context.Context, personID int64) ([]Link, error) {
	rows, err := s.db.Query(ctx, `select `+linkColumns+` from imvu_links where person_id = $1 order by linked_at`, personID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Link
	for rows.Next() {
		l, err := scanLink(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, *l)
	}
	return out, rows.Err()
}

func (s *Store) LinkByIMVUID(ctx context.Context, imvuID int64) (*Link, error) {
	return scanLink(s.db.QueryRow(ctx, `select `+linkColumns+` from imvu_links where imvu_id = $1`, imvuID))
}

// ErrLinkCap means the person has linked as many accounts as they may.
var ErrLinkCap = errors.New("link cap reached")

// LinkAccount links an IMVU account to a person, moving it from whoever had it (the most recent proof wins).
// It returns the person it was moved from, if any.
func (s *Store) LinkAccount(ctx context.Context, personID, imvuID int64, username, pairwiseID string, ageVerified, accessPass bool) (movedFrom int64, err error) {
	err = pgx.BeginFunc(ctx, s.db, func(tx pgx.Tx) error {
		var cap, count int
		if err := tx.QueryRow(ctx, `select link_cap, (select count(*) from imvu_links where person_id = $1 and imvu_id <> $2)
			from persons where id = $1 for update`, personID, imvuID).Scan(&cap, &count); err != nil {
			return err
		}
		if count >= cap {
			return ErrLinkCap
		}
		err := tx.QueryRow(ctx, `select person_id from imvu_links where imvu_id = $1`, imvuID).Scan(&movedFrom)
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return err
		}
		if movedFrom == personID {
			movedFrom = 0
		}
		// A moved account gets a new pairwise ID: nothing connects its new owner to the old one.
		_, err = tx.Exec(ctx, `insert into imvu_links (imvu_id, person_id, pairwise_id, username, age_verified, access_pass)
			values ($1, $2, $3, $4, $5, $6)
			on conflict (imvu_id) do update set person_id = excluded.person_id,
				pairwise_id = case when imvu_links.person_id = excluded.person_id then imvu_links.pairwise_id else excluded.pairwise_id end,
				username = excluded.username, age_verified = excluded.age_verified, access_pass = excluded.access_pass,
				checked_at = now(),
				linked_at = case when imvu_links.person_id = excluded.person_id then imvu_links.linked_at else now() end,
				show_badge = case when imvu_links.person_id = excluded.person_id then imvu_links.show_badge else false end,
				show_presence = case when imvu_links.person_id = excluded.person_id then imvu_links.show_presence else false end`,
			imvuID, personID, pairwiseID, username, ageVerified, accessPass)
		if err != nil {
			return err
		}
		_, err = tx.Exec(ctx, `delete from link_challenges where imvu_id = $1`, imvuID)
		return err
	})
	return movedFrom, err
}

func (s *Store) Unlink(ctx context.Context, personID, imvuID int64) error {
	_, err := s.db.Exec(ctx, `delete from imvu_links where person_id = $1 and imvu_id = $2`, personID, imvuID)
	return err
}

// UpdateAge records a fresh age check of a linked account.
func (s *Store) UpdateAge(ctx context.Context, imvuID int64, username string, ageVerified, accessPass bool) error {
	_, err := s.db.Exec(ctx, `update imvu_links set username = $2, age_verified = $3, access_pass = $4, checked_at = now() where imvu_id = $1`,
		imvuID, username, ageVerified, accessPass)
	return err
}

func (s *Store) SetLinkSettings(ctx context.Context, personID, imvuID int64, badge, presence bool) error {
	_, err := s.db.Exec(ctx, `update imvu_links set show_badge = $3, show_presence = $4 where person_id = $1 and imvu_id = $2`,
		personID, imvuID, badge, presence)
	return err
}

// Challenge is a pending tagline code.
type Challenge struct {
	IMVUID    int64
	Username  string
	Code      string
	CreatedAt time.Time
}

// SetChallenge stores (or replaces) the code for linking an account.
func (s *Store) SetChallenge(ctx context.Context, personID, imvuID int64, username, code string) error {
	_, err := s.db.Exec(ctx, `insert into link_challenges (person_id, imvu_id, username, code) values ($1, $2, $3, $4)
		on conflict (person_id, imvu_id) do update set code = excluded.code, username = excluded.username, created_at = now()`,
		personID, imvuID, username, code)
	return err
}

func (s *Store) Challenges(ctx context.Context, personID int64) ([]Challenge, error) {
	rows, err := s.db.Query(ctx, `select imvu_id, username, code, created_at from link_challenges
		where person_id = $1 and created_at > now() - interval '1 day' order by created_at`, personID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Challenge
	for rows.Next() {
		var c Challenge
		if err := rows.Scan(&c.IMVUID, &c.Username, &c.Code, &c.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	return out, rows.Err()
}

func (s *Store) Challenge(ctx context.Context, personID, imvuID int64) (*Challenge, error) {
	var c Challenge
	err := s.db.QueryRow(ctx, `select imvu_id, username, code, created_at from link_challenges
		where person_id = $1 and imvu_id = $2 and created_at > now() - interval '1 day'`, personID, imvuID).
		Scan(&c.IMVUID, &c.Username, &c.Code, &c.CreatedAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNotFound
	}
	return &c, err
}

func (s *Store) DeleteChallenge(ctx context.Context, personID, imvuID int64) error {
	_, err := s.db.Exec(ctx, `delete from link_challenges where person_id = $1 and imvu_id = $2`, personID, imvuID)
	return err
}

// Proxy sign-in (device codes) and refresh tokens

func (s *Store) CreateDeviceCode(ctx context.Context, deviceCode, userCode string, expires time.Time) error {
	_, err := s.db.Exec(ctx, `insert into device_codes (device_hash, user_code, expires_at) values ($1, $2, $3)`,
		Hash(deviceCode), userCode, expires)
	if isUniqueViolation(err) {
		return ErrTaken
	}
	return err
}

// ApproveDeviceCode lets the person's proxy waiting with userCode sign in.
func (s *Store) ApproveDeviceCode(ctx context.Context, userCode string, personID int64) error {
	tag, err := s.db.Exec(ctx, `update device_codes set person_id = $2
		where user_code = $1 and person_id is null and consumed_at is null and expires_at > now()`, userCode, personID)
	if err == nil && tag.RowsAffected() == 0 {
		return ErrNotFound
	}
	return err
}

// DeviceStatus is a waiting proxy's sign-in: approved (with the person), pending, or gone.
type DeviceStatus int

const (
	DevicePending DeviceStatus = iota
	DeviceApproved
	DeviceGone
)

// ConsumeDeviceCode reports a device code's state; an approved one is used up by this call.
func (s *Store) ConsumeDeviceCode(ctx context.Context, deviceCode string) (DeviceStatus, int64, error) {
	var personID *int64
	var expires time.Time
	var consumed *time.Time
	err := s.db.QueryRow(ctx, `select person_id, expires_at, consumed_at from device_codes where device_hash = $1`, Hash(deviceCode)).
		Scan(&personID, &expires, &consumed)
	if errors.Is(err, pgx.ErrNoRows) || err == nil && (consumed != nil || time.Now().After(expires)) {
		return DeviceGone, 0, nil
	}
	if err != nil {
		return 0, 0, err
	}
	if personID == nil {
		return DevicePending, 0, nil
	}
	tag, err := s.db.Exec(ctx, `update device_codes set consumed_at = now() where device_hash = $1 and consumed_at is null`, Hash(deviceCode))
	if err != nil {
		return 0, 0, err
	}
	if tag.RowsAffected() == 0 {
		return DeviceGone, 0, nil
	}
	return DeviceApproved, *personID, nil
}

func (s *Store) CreateRefreshToken(ctx context.Context, token string, personID int64) error {
	_, err := s.db.Exec(ctx, `insert into refresh_tokens (token_hash, person_id) values ($1, $2)`, Hash(token), personID)
	return err
}

// RotateRefreshToken trades a live refresh token for a new one, returning the person it's for.
func (s *Store) RotateRefreshToken(ctx context.Context, old, replacement string) (int64, error) {
	var personID int64
	err := pgx.BeginFunc(ctx, s.db, func(tx pgx.Tx) error {
		err := tx.QueryRow(ctx, `update refresh_tokens set revoked_at = now()
			where token_hash = $1 and revoked_at is null returning person_id`, Hash(old)).Scan(&personID)
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		_, err = tx.Exec(ctx, `insert into refresh_tokens (token_hash, person_id) values ($1, $2)`, Hash(replacement), personID)
		return err
	})
	return personID, err
}

// Signins counts a person's signed-in proxies.
func (s *Store) Signins(ctx context.Context, personID int64) (int, error) {
	var n int
	err := s.db.QueryRow(ctx, `select count(*) from refresh_tokens where person_id = $1 and revoked_at is null`, personID).Scan(&n)
	return n, err
}

// SignOutProxies ends all of a person's proxy sign-ins.
func (s *Store) SignOutProxies(ctx context.Context, personID int64) error {
	_, err := s.db.Exec(ctx, `update refresh_tokens set revoked_at = now() where person_id = $1 and revoked_at is null`, personID)
	return err
}

// Audit log

// Audit records something a person (or the hub, actor 0) did.
func (s *Store) Audit(ctx context.Context, actorID int64, action, target string, detail map[string]any) error {
	var actor *int64
	if actorID != 0 {
		actor = &actorID
	}
	if detail == nil {
		detail = map[string]any{}
	}
	_, err := s.db.Exec(ctx, `insert into audit (actor_id, action, target, detail) values ($1, $2, $3, $4)`, actor, action, target, detail)
	return err
}

// AuditEntry is a line of the audit log.
type AuditEntry struct {
	At     time.Time
	Actor  string
	Action string
	Target string
	Detail map[string]any
}

func (s *Store) AuditLog(ctx context.Context, limit int) ([]AuditEntry, error) {
	rows, err := s.db.Query(ctx, `select a.at, coalesce(p.username, ''), a.action, a.target, a.detail
		from audit a left join persons p on p.id = a.actor_id order by a.id desc limit $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []AuditEntry
	for rows.Next() {
		var e AuditEntry
		if err := rows.Scan(&e.At, &e.Actor, &e.Action, &e.Target, &e.Detail); err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

// Counts are totals for the console.
type Counts struct {
	People, Links, Signins int
}

func (s *Store) Counts(ctx context.Context) (Counts, error) {
	var c Counts
	err := s.db.QueryRow(ctx, `select (select count(*) from persons), (select count(*) from imvu_links),
		(select count(*) from refresh_tokens where revoked_at is null)`).Scan(&c.People, &c.Links, &c.Signins)
	return c, err
}