fix: address all PR #10 review findings
All checks were successful
check / check (push) Successful in 2m19s
All checks were successful
check / check (push) Successful in 2m19s
Security: - Add channel membership check before PRIVMSG (prevents non-members from sending) - Add membership check on history endpoint (channels require membership, DMs scoped to own nick) - Enforce MaxBytesReader on all POST request bodies - Fix rand.Read error being silently ignored in token generation Data integrity: - Fix TOCTOU race in GetOrCreateChannel using INSERT OR IGNORE + SELECT Build: - Add CGO_ENABLED=0 to golangci-lint install in Dockerfile (fixes alpine build) Linting: - Strict .golangci.yml: only wsl disabled (deprecated in v2) - Re-enable exhaustruct, depguard, godot, wrapcheck, varnamelen - Fix linters-settings -> linters.settings for v2 config format - Fix ALL lint findings in actual code (no linter config weakening) - Wrap all external package errors (wrapcheck) - Fill struct fields or add targeted nolint:exhaustruct where appropriate - Rename short variables (ts->timestamp, n->bufIndex, etc.) - Add depguard deny policy for io/ioutil and math/rand - Exclude G704 (SSRF) in gosec config (CLI client takes user-configured URLs) Tests: - Add security tests (TestNonMemberCannotSend, TestHistoryNonMember) - Split TestInsertAndPollMessages for reduced complexity - Fix parallel test safety (viper global state prevents parallelism) - Use t.Context() instead of context.Background() in tests Docker build verified passing locally.
This commit is contained in:
clawbot
@@ -37,93 +37,93 @@ type Params struct {
|
||||
|
||||
// Database manages the SQLite connection and migrations.
|
||||
type Database struct {
|
||||
db *sql.DB
|
||||
conn *sql.DB
|
||||
log *slog.Logger
|
||||
params *Params
|
||||
}
|
||||
|
||||
// New creates a new Database and registers lifecycle hooks.
|
||||
func New(
|
||||
lc fx.Lifecycle,
|
||||
lifecycle fx.Lifecycle,
|
||||
params Params,
|
||||
) (*Database, error) {
|
||||
s := new(Database)
|
||||
s.params = ¶ms
|
||||
s.log = params.Logger.Get()
|
||||
database := &Database{ //nolint:exhaustruct // conn set in OnStart
|
||||
params: ¶ms,
|
||||
log: params.Logger.Get(),
|
||||
}
|
||||
|
||||
s.log.Info("Database instantiated")
|
||||
database.log.Info("Database instantiated")
|
||||
|
||||
lc.Append(fx.Hook{
|
||||
lifecycle.Append(fx.Hook{
|
||||
OnStart: func(ctx context.Context) error {
|
||||
s.log.Info("Database OnStart Hook")
|
||||
database.log.Info("Database OnStart Hook")
|
||||
|
||||
return s.connect(ctx)
|
||||
return database.connect(ctx)
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
s.log.Info("Database OnStop Hook")
|
||||
database.log.Info("Database OnStop Hook")
|
||||
|
||||
if s.db != nil {
|
||||
return s.db.Close()
|
||||
if database.conn != nil {
|
||||
closeErr := database.conn.Close()
|
||||
if closeErr != nil {
|
||||
return fmt.Errorf(
|
||||
"close db: %w", closeErr,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
|
||||
return s, nil
|
||||
return database, nil
|
||||
}
|
||||
|
||||
// GetDB returns the underlying sql.DB connection.
|
||||
func (s *Database) GetDB() *sql.DB {
|
||||
return s.db
|
||||
func (database *Database) GetDB() *sql.DB {
|
||||
return database.conn
|
||||
}
|
||||
|
||||
func (s *Database) connect(ctx context.Context) error {
|
||||
dbURL := s.params.Config.DBURL
|
||||
func (database *Database) connect(ctx context.Context) error {
|
||||
dbURL := database.params.Config.DBURL
|
||||
if dbURL == "" {
|
||||
dbURL = "file:./data.db?_journal_mode=WAL&_busy_timeout=5000"
|
||||
}
|
||||
|
||||
s.log.Info("connecting to database", "url", dbURL)
|
||||
database.log.Info(
|
||||
"connecting to database", "url", dbURL,
|
||||
)
|
||||
|
||||
d, err := sql.Open("sqlite", dbURL)
|
||||
conn, err := sql.Open("sqlite", dbURL)
|
||||
if err != nil {
|
||||
s.log.Error(
|
||||
"failed to open database", "error", err,
|
||||
)
|
||||
|
||||
return err
|
||||
return fmt.Errorf("open database: %w", err)
|
||||
}
|
||||
|
||||
err = d.PingContext(ctx)
|
||||
err = conn.PingContext(ctx)
|
||||
if err != nil {
|
||||
s.log.Error(
|
||||
"failed to ping database", "error", err,
|
||||
)
|
||||
|
||||
return err
|
||||
return fmt.Errorf("ping database: %w", err)
|
||||
}
|
||||
|
||||
d.SetMaxOpenConns(1)
|
||||
conn.SetMaxOpenConns(1)
|
||||
|
||||
s.db = d
|
||||
s.log.Info("database connected")
|
||||
database.conn = conn
|
||||
database.log.Info("database connected")
|
||||
|
||||
_, err = s.db.ExecContext(
|
||||
_, err = database.conn.ExecContext(
|
||||
ctx, "PRAGMA foreign_keys = ON",
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("enable foreign keys: %w", err)
|
||||
}
|
||||
|
||||
_, err = s.db.ExecContext(
|
||||
_, err = database.conn.ExecContext(
|
||||
ctx, "PRAGMA busy_timeout = 5000",
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("set busy timeout: %w", err)
|
||||
}
|
||||
|
||||
return s.runMigrations(ctx)
|
||||
return database.runMigrations(ctx)
|
||||
}
|
||||
|
||||
type migration struct {
|
||||
@@ -132,10 +132,10 @@ type migration struct {
|
||||
sql string
|
||||
}
|
||||
|
||||
func (s *Database) runMigrations(
|
||||
func (database *Database) runMigrations(
|
||||
ctx context.Context,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version INTEGER PRIMARY KEY,
|
||||
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP)`)
|
||||
@@ -145,37 +145,37 @@ func (s *Database) runMigrations(
|
||||
)
|
||||
}
|
||||
|
||||
migrations, err := s.loadMigrations()
|
||||
migrations, err := database.loadMigrations()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, m := range migrations {
|
||||
err = s.applyMigration(ctx, m)
|
||||
for _, mig := range migrations {
|
||||
err = database.applyMigration(ctx, mig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
s.log.Info("database migrations complete")
|
||||
database.log.Info("database migrations complete")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Database) applyMigration(
|
||||
func (database *Database) applyMigration(
|
||||
ctx context.Context,
|
||||
m migration,
|
||||
mig migration,
|
||||
) error {
|
||||
var exists int
|
||||
|
||||
err := s.db.QueryRowContext(ctx,
|
||||
err := database.conn.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM schema_migrations
|
||||
WHERE version = ?`,
|
||||
m.version,
|
||||
mig.version,
|
||||
).Scan(&exists)
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"check migration %d: %w", m.version, err,
|
||||
"check migration %d: %w", mig.version, err,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -183,55 +183,63 @@ func (s *Database) applyMigration(
|
||||
return nil
|
||||
}
|
||||
|
||||
s.log.Info(
|
||||
database.log.Info(
|
||||
"applying migration",
|
||||
"version", m.version,
|
||||
"name", m.name,
|
||||
"version", mig.version,
|
||||
"name", mig.name,
|
||||
)
|
||||
|
||||
return s.execMigration(ctx, m)
|
||||
return database.execMigration(ctx, mig)
|
||||
}
|
||||
|
||||
func (s *Database) execMigration(
|
||||
func (database *Database) execMigration(
|
||||
ctx context.Context,
|
||||
m migration,
|
||||
mig migration,
|
||||
) error {
|
||||
tx, err := s.db.BeginTx(ctx, nil)
|
||||
transaction, err := database.conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"begin tx for migration %d: %w",
|
||||
m.version, err,
|
||||
mig.version, err,
|
||||
)
|
||||
}
|
||||
|
||||
_, err = tx.ExecContext(ctx, m.sql)
|
||||
_, err = transaction.ExecContext(ctx, mig.sql)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
_ = transaction.Rollback()
|
||||
|
||||
return fmt.Errorf(
|
||||
"apply migration %d (%s): %w",
|
||||
m.version, m.name, err,
|
||||
mig.version, mig.name, err,
|
||||
)
|
||||
}
|
||||
|
||||
_, err = tx.ExecContext(ctx,
|
||||
_, err = transaction.ExecContext(ctx,
|
||||
`INSERT INTO schema_migrations (version)
|
||||
VALUES (?)`,
|
||||
m.version,
|
||||
mig.version,
|
||||
)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
_ = transaction.Rollback()
|
||||
|
||||
return fmt.Errorf(
|
||||
"record migration %d: %w",
|
||||
m.version, err,
|
||||
mig.version, err,
|
||||
)
|
||||
}
|
||||
|
||||
return tx.Commit()
|
||||
err = transaction.Commit()
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"commit migration %d: %w",
|
||||
mig.version, err,
|
||||
)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Database) loadMigrations() (
|
||||
func (database *Database) loadMigrations() (
|
||||
[]migration,
|
||||
error,
|
||||
) {
|
||||
|
||||
@@ -13,35 +13,48 @@ var testDBCounter atomic.Int64
|
||||
|
||||
// NewTestDatabase creates an in-memory database for testing.
|
||||
func NewTestDatabase() (*Database, error) {
|
||||
n := testDBCounter.Add(1)
|
||||
counter := testDBCounter.Add(1)
|
||||
|
||||
dsn := fmt.Sprintf(
|
||||
"file:testdb%d?mode=memory"+
|
||||
"&cache=shared&_pragma=foreign_keys(1)",
|
||||
n,
|
||||
counter,
|
||||
)
|
||||
|
||||
d, err := sql.Open("sqlite", dsn)
|
||||
conn, err := sql.Open("sqlite", dsn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("open test db: %w", err)
|
||||
}
|
||||
|
||||
database := &Database{db: d, log: slog.Default()}
|
||||
database := &Database{ //nolint:exhaustruct // test helper, params not needed
|
||||
conn: conn,
|
||||
log: slog.Default(),
|
||||
}
|
||||
|
||||
err = database.runMigrations(context.Background())
|
||||
if err != nil {
|
||||
closeErr := d.Close()
|
||||
closeErr := conn.Close()
|
||||
if closeErr != nil {
|
||||
return nil, closeErr
|
||||
return nil, fmt.Errorf(
|
||||
"close after migration failure: %w",
|
||||
closeErr,
|
||||
)
|
||||
}
|
||||
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"run test migrations: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return database, nil
|
||||
}
|
||||
|
||||
// Close closes the underlying database connection.
|
||||
func (s *Database) Close() error {
|
||||
return s.db.Close()
|
||||
func (database *Database) Close() error {
|
||||
err := database.conn.Close()
|
||||
if err != nil {
|
||||
return fmt.Errorf("close database: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -18,11 +18,15 @@ const (
|
||||
defaultHistLimit = 50
|
||||
)
|
||||
|
||||
func generateToken() string {
|
||||
b := make([]byte, tokenBytes)
|
||||
_, _ = rand.Read(b)
|
||||
func generateToken() (string, error) {
|
||||
buf := make([]byte, tokenBytes)
|
||||
|
||||
return hex.EncodeToString(b)
|
||||
_, err := rand.Read(buf)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("generate token: %w", err)
|
||||
}
|
||||
|
||||
return hex.EncodeToString(buf), nil
|
||||
}
|
||||
|
||||
// IRCMessage is the IRC envelope for all messages.
|
||||
@@ -52,14 +56,18 @@ type MemberInfo struct {
|
||||
}
|
||||
|
||||
// CreateUser registers a new user with the given nick.
|
||||
func (s *Database) CreateUser(
|
||||
func (database *Database) CreateUser(
|
||||
ctx context.Context,
|
||||
nick string,
|
||||
) (int64, string, error) {
|
||||
token := generateToken()
|
||||
token, err := generateToken()
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
|
||||
res, err := s.db.ExecContext(ctx,
|
||||
res, err := database.conn.ExecContext(ctx,
|
||||
`INSERT INTO users
|
||||
(nick, token, created_at, last_seen)
|
||||
VALUES (?, ?, ?, ?)`,
|
||||
@@ -68,90 +76,88 @@ func (s *Database) CreateUser(
|
||||
return 0, "", fmt.Errorf("create user: %w", err)
|
||||
}
|
||||
|
||||
id, _ := res.LastInsertId()
|
||||
userID, _ := res.LastInsertId()
|
||||
|
||||
return id, token, nil
|
||||
return userID, token, nil
|
||||
}
|
||||
|
||||
// GetUserByToken returns user id and nick for a token.
|
||||
func (s *Database) GetUserByToken(
|
||||
func (database *Database) GetUserByToken(
|
||||
ctx context.Context,
|
||||
token string,
|
||||
) (int64, string, error) {
|
||||
var id int64
|
||||
var userID int64
|
||||
|
||||
var nick string
|
||||
|
||||
err := s.db.QueryRowContext(
|
||||
err := database.conn.QueryRowContext(
|
||||
ctx,
|
||||
"SELECT id, nick FROM users WHERE token = ?",
|
||||
token,
|
||||
).Scan(&id, &nick)
|
||||
).Scan(&userID, &nick)
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
return 0, "", fmt.Errorf("get user by token: %w", err)
|
||||
}
|
||||
|
||||
_, _ = s.db.ExecContext(
|
||||
_, _ = database.conn.ExecContext(
|
||||
ctx,
|
||||
"UPDATE users SET last_seen = ? WHERE id = ?",
|
||||
time.Now(), id,
|
||||
time.Now(), userID,
|
||||
)
|
||||
|
||||
return id, nick, nil
|
||||
return userID, nick, nil
|
||||
}
|
||||
|
||||
// GetUserByNick returns user id for a given nick.
|
||||
func (s *Database) GetUserByNick(
|
||||
func (database *Database) GetUserByNick(
|
||||
ctx context.Context,
|
||||
nick string,
|
||||
) (int64, error) {
|
||||
var id int64
|
||||
var userID int64
|
||||
|
||||
err := s.db.QueryRowContext(
|
||||
err := database.conn.QueryRowContext(
|
||||
ctx,
|
||||
"SELECT id FROM users WHERE nick = ?",
|
||||
nick,
|
||||
).Scan(&id)
|
||||
).Scan(&userID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get user by nick: %w", err)
|
||||
}
|
||||
|
||||
return id, err
|
||||
return userID, nil
|
||||
}
|
||||
|
||||
// GetChannelByName returns the channel ID for a name.
|
||||
func (s *Database) GetChannelByName(
|
||||
func (database *Database) GetChannelByName(
|
||||
ctx context.Context,
|
||||
name string,
|
||||
) (int64, error) {
|
||||
var id int64
|
||||
var channelID int64
|
||||
|
||||
err := s.db.QueryRowContext(
|
||||
err := database.conn.QueryRowContext(
|
||||
ctx,
|
||||
"SELECT id FROM channels WHERE name = ?",
|
||||
name,
|
||||
).Scan(&id)
|
||||
).Scan(&channelID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf(
|
||||
"get channel by name: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return id, err
|
||||
return channelID, nil
|
||||
}
|
||||
|
||||
// GetOrCreateChannel returns channel id, creating if needed.
|
||||
func (s *Database) GetOrCreateChannel(
|
||||
// Uses INSERT OR IGNORE to avoid TOCTOU races.
|
||||
func (database *Database) GetOrCreateChannel(
|
||||
ctx context.Context,
|
||||
name string,
|
||||
) (int64, error) {
|
||||
var id int64
|
||||
|
||||
err := s.db.QueryRowContext(
|
||||
ctx,
|
||||
"SELECT id FROM channels WHERE name = ?",
|
||||
name,
|
||||
).Scan(&id)
|
||||
if err == nil {
|
||||
return id, nil
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
|
||||
res, err := s.db.ExecContext(ctx,
|
||||
`INSERT INTO channels
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`INSERT OR IGNORE INTO channels
|
||||
(name, created_at, updated_at)
|
||||
VALUES (?, ?, ?)`,
|
||||
name, now, now)
|
||||
@@ -159,51 +165,71 @@ func (s *Database) GetOrCreateChannel(
|
||||
return 0, fmt.Errorf("create channel: %w", err)
|
||||
}
|
||||
|
||||
id, _ = res.LastInsertId()
|
||||
var channelID int64
|
||||
|
||||
return id, nil
|
||||
err = database.conn.QueryRowContext(
|
||||
ctx,
|
||||
"SELECT id FROM channels WHERE name = ?",
|
||||
name,
|
||||
).Scan(&channelID)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("get channel: %w", err)
|
||||
}
|
||||
|
||||
return channelID, nil
|
||||
}
|
||||
|
||||
// JoinChannel adds a user to a channel.
|
||||
func (s *Database) JoinChannel(
|
||||
func (database *Database) JoinChannel(
|
||||
ctx context.Context,
|
||||
channelID, userID int64,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`INSERT OR IGNORE INTO channel_members
|
||||
(channel_id, user_id, joined_at)
|
||||
VALUES (?, ?, ?)`,
|
||||
channelID, userID, time.Now())
|
||||
if err != nil {
|
||||
return fmt.Errorf("join channel: %w", err)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// PartChannel removes a user from a channel.
|
||||
func (s *Database) PartChannel(
|
||||
func (database *Database) PartChannel(
|
||||
ctx context.Context,
|
||||
channelID, userID int64,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`DELETE FROM channel_members
|
||||
WHERE channel_id = ? AND user_id = ?`,
|
||||
channelID, userID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("part channel: %w", err)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteChannelIfEmpty removes a channel with no members.
|
||||
func (s *Database) DeleteChannelIfEmpty(
|
||||
func (database *Database) DeleteChannelIfEmpty(
|
||||
ctx context.Context,
|
||||
channelID int64,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`DELETE FROM channels WHERE id = ?
|
||||
AND NOT EXISTS
|
||||
(SELECT 1 FROM channel_members
|
||||
WHERE channel_id = ?)`,
|
||||
channelID, channelID)
|
||||
if err != nil {
|
||||
return fmt.Errorf(
|
||||
"delete channel if empty: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// scanChannels scans rows into a ChannelInfo slice.
|
||||
@@ -215,19 +241,21 @@ func scanChannels(
|
||||
var out []ChannelInfo
|
||||
|
||||
for rows.Next() {
|
||||
var ch ChannelInfo
|
||||
var chanInfo ChannelInfo
|
||||
|
||||
err := rows.Scan(&ch.ID, &ch.Name, &ch.Topic)
|
||||
err := rows.Scan(
|
||||
&chanInfo.ID, &chanInfo.Name, &chanInfo.Topic,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("scan channel: %w", err)
|
||||
}
|
||||
|
||||
out = append(out, ch)
|
||||
out = append(out, chanInfo)
|
||||
}
|
||||
|
||||
err := rows.Err()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("rows error: %w", err)
|
||||
}
|
||||
|
||||
if out == nil {
|
||||
@@ -238,11 +266,11 @@ func scanChannels(
|
||||
}
|
||||
|
||||
// ListChannels returns channels the user has joined.
|
||||
func (s *Database) ListChannels(
|
||||
func (database *Database) ListChannels(
|
||||
ctx context.Context,
|
||||
userID int64,
|
||||
) ([]ChannelInfo, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT c.id, c.name, c.topic
|
||||
FROM channels c
|
||||
INNER JOIN channel_members cm
|
||||
@@ -250,32 +278,34 @@ func (s *Database) ListChannels(
|
||||
WHERE cm.user_id = ?
|
||||
ORDER BY c.name`, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("list channels: %w", err)
|
||||
}
|
||||
|
||||
return scanChannels(rows)
|
||||
}
|
||||
|
||||
// ListAllChannels returns every channel.
|
||||
func (s *Database) ListAllChannels(
|
||||
func (database *Database) ListAllChannels(
|
||||
ctx context.Context,
|
||||
) ([]ChannelInfo, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT id, name, topic
|
||||
FROM channels ORDER BY name`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"list all channels: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return scanChannels(rows)
|
||||
}
|
||||
|
||||
// ChannelMembers returns all members of a channel.
|
||||
func (s *Database) ChannelMembers(
|
||||
func (database *Database) ChannelMembers(
|
||||
ctx context.Context,
|
||||
channelID int64,
|
||||
) ([]MemberInfo, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT u.id, u.nick, u.last_seen
|
||||
FROM users u
|
||||
INNER JOIN channel_members cm
|
||||
@@ -283,7 +313,9 @@ func (s *Database) ChannelMembers(
|
||||
WHERE cm.channel_id = ?
|
||||
ORDER BY u.nick`, channelID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"query channel members: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
@@ -291,19 +323,23 @@ func (s *Database) ChannelMembers(
|
||||
var members []MemberInfo
|
||||
|
||||
for rows.Next() {
|
||||
var m MemberInfo
|
||||
var member MemberInfo
|
||||
|
||||
err = rows.Scan(&m.ID, &m.Nick, &m.LastSeen)
|
||||
err = rows.Scan(
|
||||
&member.ID, &member.Nick, &member.LastSeen,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"scan member: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
members = append(members, m)
|
||||
members = append(members, member)
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("rows error: %w", err)
|
||||
}
|
||||
|
||||
if members == nil {
|
||||
@@ -313,6 +349,27 @@ func (s *Database) ChannelMembers(
|
||||
return members, nil
|
||||
}
|
||||
|
||||
// IsChannelMember checks if a user belongs to a channel.
|
||||
func (database *Database) IsChannelMember(
|
||||
ctx context.Context,
|
||||
channelID, userID int64,
|
||||
) (bool, error) {
|
||||
var count int
|
||||
|
||||
err := database.conn.QueryRowContext(ctx,
|
||||
`SELECT COUNT(*) FROM channel_members
|
||||
WHERE channel_id = ? AND user_id = ?`,
|
||||
channelID, userID,
|
||||
).Scan(&count)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf(
|
||||
"check membership: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return count > 0, nil
|
||||
}
|
||||
|
||||
// scanInt64s scans rows into an int64 slice.
|
||||
func scanInt64s(rows *sql.Rows) ([]int64, error) {
|
||||
defer func() { _ = rows.Close() }()
|
||||
@@ -320,58 +377,64 @@ func scanInt64s(rows *sql.Rows) ([]int64, error) {
|
||||
var ids []int64
|
||||
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
var val int64
|
||||
|
||||
err := rows.Scan(&id)
|
||||
err := rows.Scan(&val)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"scan int64: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
ids = append(ids, id)
|
||||
ids = append(ids, val)
|
||||
}
|
||||
|
||||
err := rows.Err()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf("rows error: %w", err)
|
||||
}
|
||||
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// GetChannelMemberIDs returns user IDs in a channel.
|
||||
func (s *Database) GetChannelMemberIDs(
|
||||
func (database *Database) GetChannelMemberIDs(
|
||||
ctx context.Context,
|
||||
channelID int64,
|
||||
) ([]int64, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT user_id FROM channel_members
|
||||
WHERE channel_id = ?`, channelID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"get channel member ids: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return scanInt64s(rows)
|
||||
}
|
||||
|
||||
// GetUserChannelIDs returns channel IDs the user is in.
|
||||
func (s *Database) GetUserChannelIDs(
|
||||
func (database *Database) GetUserChannelIDs(
|
||||
ctx context.Context,
|
||||
userID int64,
|
||||
) ([]int64, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT channel_id FROM channel_members
|
||||
WHERE user_id = ?`, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"get user channel ids: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return scanInt64s(rows)
|
||||
}
|
||||
|
||||
// InsertMessage stores a message and returns its DB ID.
|
||||
func (s *Database) InsertMessage(
|
||||
func (database *Database) InsertMessage(
|
||||
ctx context.Context,
|
||||
command, from, to string,
|
||||
command, from, target string,
|
||||
body json.RawMessage,
|
||||
meta json.RawMessage,
|
||||
) (int64, string, error) {
|
||||
@@ -386,38 +449,43 @@ func (s *Database) InsertMessage(
|
||||
meta = json.RawMessage("{}")
|
||||
}
|
||||
|
||||
res, err := s.db.ExecContext(ctx,
|
||||
res, err := database.conn.ExecContext(ctx,
|
||||
`INSERT INTO messages
|
||||
(uuid, command, msg_from, msg_to,
|
||||
body, meta, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)`,
|
||||
msgUUID, command, from, to,
|
||||
msgUUID, command, from, target,
|
||||
string(body), string(meta), now)
|
||||
if err != nil {
|
||||
return 0, "", err
|
||||
return 0, "", fmt.Errorf(
|
||||
"insert message: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
id, _ := res.LastInsertId()
|
||||
dbID, _ := res.LastInsertId()
|
||||
|
||||
return id, msgUUID, nil
|
||||
return dbID, msgUUID, nil
|
||||
}
|
||||
|
||||
// EnqueueMessage adds a message to a user's queue.
|
||||
func (s *Database) EnqueueMessage(
|
||||
func (database *Database) EnqueueMessage(
|
||||
ctx context.Context,
|
||||
userID, messageID int64,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`INSERT OR IGNORE INTO client_queues
|
||||
(user_id, message_id, created_at)
|
||||
VALUES (?, ?, ?)`,
|
||||
userID, messageID, time.Now())
|
||||
if err != nil {
|
||||
return fmt.Errorf("enqueue message: %w", err)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// PollMessages returns queued messages for a user.
|
||||
func (s *Database) PollMessages(
|
||||
func (database *Database) PollMessages(
|
||||
ctx context.Context,
|
||||
userID, afterQueueID int64,
|
||||
limit int,
|
||||
@@ -426,7 +494,7 @@ func (s *Database) PollMessages(
|
||||
limit = defaultPollLimit
|
||||
}
|
||||
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT cq.id, m.uuid, m.command,
|
||||
m.msg_from, m.msg_to,
|
||||
m.body, m.meta, m.created_at
|
||||
@@ -437,7 +505,9 @@ func (s *Database) PollMessages(
|
||||
ORDER BY cq.id ASC LIMIT ?`,
|
||||
userID, afterQueueID, limit)
|
||||
if err != nil {
|
||||
return nil, afterQueueID, err
|
||||
return nil, afterQueueID, fmt.Errorf(
|
||||
"poll messages: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
msgs, lastQID, scanErr := scanMessages(
|
||||
@@ -451,7 +521,7 @@ func (s *Database) PollMessages(
|
||||
}
|
||||
|
||||
// GetHistory returns message history for a target.
|
||||
func (s *Database) GetHistory(
|
||||
func (database *Database) GetHistory(
|
||||
ctx context.Context,
|
||||
target string,
|
||||
beforeID int64,
|
||||
@@ -461,7 +531,7 @@ func (s *Database) GetHistory(
|
||||
limit = defaultHistLimit
|
||||
}
|
||||
|
||||
rows, err := s.queryHistory(
|
||||
rows, err := database.queryHistory(
|
||||
ctx, target, beforeID, limit,
|
||||
)
|
||||
if err != nil {
|
||||
@@ -482,14 +552,14 @@ func (s *Database) GetHistory(
|
||||
return msgs, nil
|
||||
}
|
||||
|
||||
func (s *Database) queryHistory(
|
||||
func (database *Database) queryHistory(
|
||||
ctx context.Context,
|
||||
target string,
|
||||
beforeID int64,
|
||||
limit int,
|
||||
) (*sql.Rows, error) {
|
||||
if beforeID > 0 {
|
||||
return s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT id, uuid, command, msg_from,
|
||||
msg_to, body, meta, created_at
|
||||
FROM messages
|
||||
@@ -497,9 +567,16 @@ func (s *Database) queryHistory(
|
||||
AND command = 'PRIVMSG'
|
||||
ORDER BY id DESC LIMIT ?`,
|
||||
target, beforeID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf(
|
||||
"query history: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
return s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT id, uuid, command, msg_from,
|
||||
msg_to, body, meta, created_at
|
||||
FROM messages
|
||||
@@ -507,6 +584,11 @@ func (s *Database) queryHistory(
|
||||
AND command = 'PRIVMSG'
|
||||
ORDER BY id DESC LIMIT ?`,
|
||||
target, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query history: %w", err)
|
||||
}
|
||||
|
||||
return rows, nil
|
||||
}
|
||||
|
||||
func scanMessages(
|
||||
@@ -521,33 +603,37 @@ func scanMessages(
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
m IRCMessage
|
||||
msg IRCMessage
|
||||
qID int64
|
||||
body, meta string
|
||||
ts time.Time
|
||||
createdAt time.Time
|
||||
)
|
||||
|
||||
err := rows.Scan(
|
||||
&qID, &m.ID, &m.Command,
|
||||
&m.From, &m.To,
|
||||
&body, &meta, &ts,
|
||||
&qID, &msg.ID, &msg.Command,
|
||||
&msg.From, &msg.To,
|
||||
&body, &meta, &createdAt,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fallbackQID, err
|
||||
return nil, fallbackQID, fmt.Errorf(
|
||||
"scan message: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
m.Body = json.RawMessage(body)
|
||||
m.Meta = json.RawMessage(meta)
|
||||
m.TS = ts.Format(time.RFC3339Nano)
|
||||
m.DBID = qID
|
||||
msg.Body = json.RawMessage(body)
|
||||
msg.Meta = json.RawMessage(meta)
|
||||
msg.TS = createdAt.Format(time.RFC3339Nano)
|
||||
msg.DBID = qID
|
||||
lastQID = qID
|
||||
|
||||
msgs = append(msgs, m)
|
||||
msgs = append(msgs, msg)
|
||||
}
|
||||
|
||||
err := rows.Err()
|
||||
if err != nil {
|
||||
return nil, fallbackQID, err
|
||||
return nil, fallbackQID, fmt.Errorf(
|
||||
"rows error: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
if msgs == nil {
|
||||
@@ -564,59 +650,70 @@ func reverseMessages(msgs []IRCMessage) {
|
||||
}
|
||||
|
||||
// ChangeNick updates a user's nickname.
|
||||
func (s *Database) ChangeNick(
|
||||
func (database *Database) ChangeNick(
|
||||
ctx context.Context,
|
||||
userID int64,
|
||||
newNick string,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
"UPDATE users SET nick = ? WHERE id = ?",
|
||||
newNick, userID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("change nick: %w", err)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetTopic sets the topic for a channel.
|
||||
func (s *Database) SetTopic(
|
||||
func (database *Database) SetTopic(
|
||||
ctx context.Context,
|
||||
channelName, topic string,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(ctx,
|
||||
_, err := database.conn.ExecContext(ctx,
|
||||
`UPDATE channels SET topic = ?,
|
||||
updated_at = ? WHERE name = ?`,
|
||||
topic, time.Now(), channelName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("set topic: %w", err)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteUser removes a user and all their data.
|
||||
func (s *Database) DeleteUser(
|
||||
func (database *Database) DeleteUser(
|
||||
ctx context.Context,
|
||||
userID int64,
|
||||
) error {
|
||||
_, err := s.db.ExecContext(
|
||||
_, err := database.conn.ExecContext(
|
||||
ctx,
|
||||
"DELETE FROM users WHERE id = ?",
|
||||
userID,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete user: %w", err)
|
||||
}
|
||||
|
||||
return err
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetAllChannelMembershipsForUser returns channels
|
||||
// a user belongs to.
|
||||
func (s *Database) GetAllChannelMembershipsForUser(
|
||||
func (database *Database) GetAllChannelMembershipsForUser(
|
||||
ctx context.Context,
|
||||
userID int64,
|
||||
) ([]ChannelInfo, error) {
|
||||
rows, err := s.db.QueryContext(ctx,
|
||||
rows, err := database.conn.QueryContext(ctx,
|
||||
`SELECT c.id, c.name, c.topic
|
||||
FROM channels c
|
||||
INNER JOIN channel_members cm
|
||||
ON cm.channel_id = c.id
|
||||
WHERE cm.user_id = ?`, userID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, fmt.Errorf(
|
||||
"get memberships: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
return scanChannels(rows)
|
||||
|
||||
@@ -12,19 +12,19 @@ import (
|
||||
func setupTestDB(t *testing.T) *db.Database {
|
||||
t.Helper()
|
||||
|
||||
d, err := db.NewTestDatabase()
|
||||
database, err := db.NewTestDatabase()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() {
|
||||
closeErr := d.Close()
|
||||
closeErr := database.Close()
|
||||
if closeErr != nil {
|
||||
t.Logf("close db: %v", closeErr)
|
||||
}
|
||||
})
|
||||
|
||||
return d
|
||||
return database
|
||||
}
|
||||
|
||||
func TestCreateUser(t *testing.T) {
|
||||
@@ -349,12 +349,30 @@ func TestSetTopic(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func insertTestMessage(
|
||||
t *testing.T,
|
||||
database *db.Database,
|
||||
) (int64, int64) {
|
||||
t.Helper()
|
||||
func TestInsertMessage(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
database := setupTestDB(t)
|
||||
ctx := t.Context()
|
||||
|
||||
body := json.RawMessage(`["hello"]`)
|
||||
|
||||
dbID, msgUUID, err := database.InsertMessage(
|
||||
ctx, "PRIVMSG", "poller", "#test", body, nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if dbID == 0 || msgUUID == "" {
|
||||
t.Fatal("expected valid id and uuid")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPollMessages(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
database := setupTestDB(t)
|
||||
ctx := t.Context()
|
||||
|
||||
uid, _, err := database.CreateUser(ctx, "poller")
|
||||
@@ -364,11 +382,11 @@ func insertTestMessage(
|
||||
|
||||
body := json.RawMessage(`["hello"]`)
|
||||
|
||||
dbID, msgUUID, err := database.InsertMessage(
|
||||
dbID, _, err := database.InsertMessage(
|
||||
ctx, "PRIVMSG", "poller", "#test", body, nil,
|
||||
)
|
||||
if err != nil || dbID == 0 || msgUUID == "" {
|
||||
t.Fatal("insert failed")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = database.EnqueueMessage(ctx, uid, dbID)
|
||||
@@ -376,19 +394,10 @@ func insertTestMessage(
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
return uid, dbID
|
||||
}
|
||||
|
||||
func TestInsertAndPollMessages(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
database := setupTestDB(t)
|
||||
uid, _ := insertTestMessage(t, database)
|
||||
|
||||
const batchSize = 10
|
||||
|
||||
msgs, lastQID, err := database.PollMessages(
|
||||
t.Context(), uid, 0, batchSize,
|
||||
ctx, uid, 0, batchSize,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -411,7 +420,7 @@ func TestInsertAndPollMessages(t *testing.T) {
|
||||
}
|
||||
|
||||
msgs, _, _ = database.PollMessages(
|
||||
t.Context(), uid, lastQID, batchSize,
|
||||
ctx, uid, lastQID, batchSize,
|
||||
)
|
||||
|
||||
if len(msgs) != 0 {
|
||||
|
||||
Reference in New Issue
Block a user